From 92af1552a0bfd1ff94dcd82fb9c373e036ffcc41 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Wed, 26 Aug 2026 14:53:47 +0000 Subject: [PATCH] refactor(db): move workflow run and approval store ownership Signed-off-by: OpenAI Codex --- crates/buzz-db/src/lib.rs | 163 ------------------------------- crates/buzz-db/src/workflow.rs | 172 ++++++++++++++++++++++++++++++++- 2 files changed, 171 insertions(+), 164 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 99dab12979d..2f5300af171 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1308,169 +1308,6 @@ impl Db { } } - /// Create a new workflow run. - #[datastore_span(name = "create_workflow_run", system = "postgresql")] - pub async fn create_workflow_run( - &self, - community_id: CommunityId, - workflow_id: Uuid, - trigger_event_id: Option<&[u8]>, - trigger_context: Option<&serde_json::Value>, - ) -> Result { - workflow::create_workflow_run( - &self.pool, - community_id, - workflow_id, - trigger_event_id, - trigger_context, - ) - .await - } - - /// Fetch a single workflow run, scoped to its community. - #[datastore_span(name = "get_workflow_run", system = "postgresql")] - pub async fn get_workflow_run( - &self, - community_id: CommunityId, - id: Uuid, - ) -> Result { - workflow::get_workflow_run(&self.pool, community_id, id).await - } - - /// List runs for a workflow. - #[datastore_span(name = "list_workflow_runs", system = "postgresql")] - pub async fn list_workflow_runs( - &self, - community_id: CommunityId, - workflow_id: Uuid, - limit: i64, - ) -> Result> { - workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await - } - - /// List one keyset-paginated page of workflow runs. - #[datastore_span(name = "list_workflow_runs_page", system = "postgresql")] - pub async fn list_workflow_runs_page( - &self, - community_id: CommunityId, - workflow_id: Uuid, - before: Option>, - before_id: Option, - limit: i64, - ) -> Result> { - workflow::list_workflow_runs_page( - &self.pool, - community_id, - workflow_id, - before, - before_id, - limit, - ) - .await - } - - /// Update a workflow run's status. - #[datastore_span(name = "update_workflow_run", system = "postgresql")] - pub async fn update_workflow_run( - &self, - community_id: CommunityId, - id: Uuid, - status: workflow::RunStatus, - current_step: i32, - trace: &serde_json::Value, - failure: Option>, - ) -> Result<()> { - workflow::update_workflow_run( - &self.pool, - community_id, - id, - status, - current_step, - trace, - failure, - ) - .await - } - - /// Create an approval request. - #[datastore_span(name = "create_approval", system = "postgresql")] - pub async fn create_approval(&self, params: workflow::CreateApprovalParams<'_>) -> Result<()> { - workflow::create_approval(&self.pool, params).await - } - - /// Fetch an approval by raw token. - #[datastore_span(name = "get_approval", system = "postgresql")] - pub async fn get_approval( - &self, - community_id: CommunityId, - token: &str, - ) -> Result { - workflow::get_approval(&self.pool, community_id, token).await - } - - /// Fetch an approval by its already-hashed token (no re-hashing). - #[datastore_span(name = "get_approval_by_stored_hash", system = "postgresql")] - pub async fn get_approval_by_stored_hash( - &self, - community_id: CommunityId, - token_hash: &[u8], - ) -> Result { - workflow::get_approval_by_stored_hash(&self.pool, community_id, token_hash).await - } - - /// Fetch all approvals for a workflow run. - #[datastore_span(name = "get_run_approvals", system = "postgresql")] - pub async fn get_run_approvals( - &self, - community_id: CommunityId, - workflow_id: uuid::Uuid, - run_id: uuid::Uuid, - ) -> Result> { - workflow::get_run_approvals(&self.pool, community_id, workflow_id, run_id).await - } - - /// Update an approval's status. - #[datastore_span(name = "update_approval", system = "postgresql")] - pub async fn update_approval( - &self, - community_id: CommunityId, - token: &str, - status: workflow::ApprovalStatus, - approver_pubkey: Option<&[u8]>, - note: Option<&str>, - ) -> Result { - workflow::update_approval( - &self.pool, - community_id, - token, - status, - approver_pubkey, - note, - ) - .await - } - - /// Update an approval by its already-hashed token (no re-hashing). - #[datastore_span(name = "update_approval_by_stored_hash", system = "postgresql")] - pub async fn update_approval_by_stored_hash( - &self, - community_id: CommunityId, - token_hash: &[u8], - status: workflow::ApprovalStatus, - approver_pubkey: Option<&[u8]>, - note: Option<&str>, - ) -> Result { - workflow::update_approval_by_stored_hash( - &self.pool, - community_id, - token_hash, - status, - approver_pubkey, - note, - ) - .await - } - /// Ensures monthly partitions exist for the next N months. #[datastore_span(name = "ensure_future_partitions", system = "postgresql")] pub async fn ensure_future_partitions(&self, months_ahead: u32) -> Result<()> { diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 657f1502802..0ae1b623764 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -1268,7 +1268,177 @@ pub async fn find_by_owner_and_name( } } -// -- Db API ------------------------------------------------------------------- +// -- Run and approval Db API -------------------------------------------------- + +impl Db { + /// Create a new workflow run. + #[datastore_span(name = "create_workflow_run", system = "postgresql")] + pub async fn create_workflow_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + trigger_event_id: Option<&[u8]>, + trigger_context: Option<&serde_json::Value>, + ) -> Result { + crate::workflow::create_workflow_run( + &self.pool, + community_id, + workflow_id, + trigger_event_id, + trigger_context, + ) + .await + } + + /// Fetch a single workflow run, scoped to its community. + #[datastore_span(name = "get_workflow_run", system = "postgresql")] + pub async fn get_workflow_run( + &self, + community_id: CommunityId, + id: Uuid, + ) -> Result { + crate::workflow::get_workflow_run(&self.pool, community_id, id).await + } + + /// List runs for a workflow. + #[datastore_span(name = "list_workflow_runs", system = "postgresql")] + pub async fn list_workflow_runs( + &self, + community_id: CommunityId, + workflow_id: Uuid, + limit: i64, + ) -> Result> { + crate::workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await + } + + /// List one keyset-paginated page of workflow runs. + #[datastore_span(name = "list_workflow_runs_page", system = "postgresql")] + pub async fn list_workflow_runs_page( + &self, + community_id: CommunityId, + workflow_id: Uuid, + before: Option>, + before_id: Option, + limit: i64, + ) -> Result> { + crate::workflow::list_workflow_runs_page( + &self.pool, + community_id, + workflow_id, + before, + before_id, + limit, + ) + .await + } + + /// Update a workflow run's status. + #[datastore_span(name = "update_workflow_run", system = "postgresql")] + pub async fn update_workflow_run( + &self, + community_id: CommunityId, + id: Uuid, + status: crate::workflow::RunStatus, + current_step: i32, + trace: &serde_json::Value, + failure: Option>, + ) -> Result<()> { + crate::workflow::update_workflow_run( + &self.pool, + community_id, + id, + status, + current_step, + trace, + failure, + ) + .await + } + + /// Create an approval request. + #[datastore_span(name = "create_approval", system = "postgresql")] + pub async fn create_approval( + &self, + params: crate::workflow::CreateApprovalParams<'_>, + ) -> Result<()> { + crate::workflow::create_approval(&self.pool, params).await + } + + /// Fetch an approval by raw token. + #[datastore_span(name = "get_approval", system = "postgresql")] + pub async fn get_approval( + &self, + community_id: CommunityId, + token: &str, + ) -> Result { + crate::workflow::get_approval(&self.pool, community_id, token).await + } + + /// Fetch an approval by its already-hashed token (no re-hashing). + #[datastore_span(name = "get_approval_by_stored_hash", system = "postgresql")] + pub async fn get_approval_by_stored_hash( + &self, + community_id: CommunityId, + token_hash: &[u8], + ) -> Result { + crate::workflow::get_approval_by_stored_hash(&self.pool, community_id, token_hash).await + } + + /// Fetch all approvals for a workflow run. + #[datastore_span(name = "get_run_approvals", system = "postgresql")] + pub async fn get_run_approvals( + &self, + community_id: CommunityId, + workflow_id: uuid::Uuid, + run_id: uuid::Uuid, + ) -> Result> { + crate::workflow::get_run_approvals(&self.pool, community_id, workflow_id, run_id).await + } + + /// Update an approval's status. + #[datastore_span(name = "update_approval", system = "postgresql")] + pub async fn update_approval( + &self, + community_id: CommunityId, + token: &str, + status: crate::workflow::ApprovalStatus, + approver_pubkey: Option<&[u8]>, + note: Option<&str>, + ) -> Result { + crate::workflow::update_approval( + &self.pool, + community_id, + token, + status, + approver_pubkey, + note, + ) + .await + } + + /// Update an approval by its already-hashed token (no re-hashing). + #[datastore_span(name = "update_approval_by_stored_hash", system = "postgresql")] + pub async fn update_approval_by_stored_hash( + &self, + community_id: CommunityId, + token_hash: &[u8], + status: crate::workflow::ApprovalStatus, + approver_pubkey: Option<&[u8]>, + note: Option<&str>, + ) -> Result { + crate::workflow::update_approval_by_stored_hash( + &self.pool, + community_id, + token_hash, + status, + approver_pubkey, + note, + ) + .await + } +} + +// -- Workflow lifecycle Db API ------------------------------------------------ impl Db { /// Create a new workflow.