crank/crates/crank-registry/src/postgres/approval.rs

use super::*;

impl PostgresRegistry {
    pub async fn create_approval_request(
        &self,
        request: CreateApprovalRequest<'_>,
    ) -> Result<(), RegistryError> {
        sqlx::query(
            "insert into approval_requests (
                id,
                workspace_id,
                agent_id,
                operation_id,
                operation_version,
                status,
                risk_level,
                confirmation_title,
                confirmation_body,
                request_payload_json,
                response_payload_json,
                created_at,
                expires_at,
                decided_at,
                decided_by_key_id,
                decision_note
            ) values (
                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
                $11, $12::timestamptz, $13::timestamptz, $14::timestamptz,
                $15, $16
            )",
        )
        .bind(request.approval.id.as_str())
        .bind(request.approval.workspace_id.as_str())
        .bind(request.approval.agent_id.as_str())
        .bind(request.approval.operation_id.as_str())
        .bind(to_db_version(request.approval.operation_version))
        .bind(serialize_enum_text(
            &request.approval.status,
            "approval_status",
        )?)
        .bind(serialize_enum_text(
            &request.approval.risk_level,
            "approval_risk_level",
        )?)
        .bind(&request.approval.confirmation_title)
        .bind(&request.approval.confirmation_body)
        .bind(Json(&request.approval.request_payload))
        .bind(request.approval.response_payload.as_ref().map(Json))
        .bind(request.approval.created_at)
        .bind(request.approval.expires_at)
        .bind(request.approval.decided_at)
        .bind(
            request
                .approval
                .decided_by_key_id
                .as_ref()
                .map(PlatformApiKeyId::as_str),
        )
        .bind(request.approval.decision_note.as_deref())
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    pub async fn list_pending_approval_requests_for_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
    ) -> Result<Vec<ApprovalRequestRecord>, RegistryError> {
        let rows = sqlx::query(
            "select
                id,
                workspace_id,
                agent_id,
                operation_id,
                operation_version,
                status,
                risk_level,
                confirmation_title,
                confirmation_body,
                request_payload_json,
                response_payload_json,
                created_at,
                expires_at,
                decided_at,
                decided_by_key_id,
                decision_note
             from approval_requests
             where workspace_id = $1
               and agent_id = $2
               and status = 'pending'
               and expires_at > now()
             order by created_at asc",
        )
        .bind(workspace_id.as_str())
        .bind(agent_id.as_str())
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(map_approval_request_row).collect()
    }

    pub async fn get_approval_request_for_agent(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        approval_id: &ApprovalRequestId,
    ) -> Result<Option<ApprovalRequestRecord>, RegistryError> {
        let row = sqlx::query(
            "select
                id,
                workspace_id,
                agent_id,
                operation_id,
                operation_version,
                status,
                risk_level,
                confirmation_title,
                confirmation_body,
                request_payload_json,
                response_payload_json,
                created_at,
                expires_at,
                decided_at,
                decided_by_key_id,
                decision_note
             from approval_requests
             where workspace_id = $1
               and agent_id = $2
               and id = $3
             limit 1",
        )
        .bind(workspace_id.as_str())
        .bind(agent_id.as_str())
        .bind(approval_id.as_str())
        .fetch_optional(&self.pool)
        .await?;

        row.map(map_approval_request_row).transpose()
    }

    pub async fn decide_approval_request(
        &self,
        request: DecideApprovalRequest<'_>,
    ) -> Result<Option<ApprovalRequestRecord>, RegistryError> {
        let row = sqlx::query(
            "update approval_requests
                set status = $1,
                    response_payload_json = $2,
                    decided_at = $3::timestamptz,
                    decided_by_key_id = $4,
                    decision_note = $5
              where workspace_id = $6
                and agent_id = $7
                and id = $8
                and status = 'pending'
                and expires_at > $3::timestamptz
              returning
                id,
                workspace_id,
                agent_id,
                operation_id,
                operation_version,
                status,
                risk_level,
                confirmation_title,
                confirmation_body,
                request_payload_json,
                response_payload_json,
                created_at,
                expires_at,
                decided_at,
                decided_by_key_id,
                decision_note",
        )
        .bind(serialize_enum_text(&request.status, "approval_status")?)
        .bind(request.response_payload.as_ref().map(Json))
        .bind(request.decided_at)
        .bind(request.decided_by_key_id.as_str())
        .bind(request.decision_note)
        .bind(request.workspace_id.as_str())
        .bind(request.agent_id.as_str())
        .bind(request.approval_id.as_str())
        .fetch_optional(&self.pool)
        .await?;

        row.map(map_approval_request_row).transpose()
    }
}

fn map_approval_request_row(row: PgRow) -> Result<ApprovalRequestRecord, RegistryError> {
    Ok(ApprovalRequestRecord {
        approval: ApprovalRequest {
            id: ApprovalRequestId::new(row.get::<String, _>("id")),
            workspace_id: WorkspaceId::new(row.get::<String, _>("workspace_id")),
            agent_id: AgentId::new(row.get::<String, _>("agent_id")),
            operation_id: OperationId::new(row.get::<String, _>("operation_id")),
            operation_version: from_db_version(row.get("operation_version"), "operation_version")?,
            status: deserialize_enum_text(&row.get::<String, _>("status"), "approval_status")?,
            risk_level: deserialize_enum_text(
                &row.get::<String, _>("risk_level"),
                "approval_risk_level",
            )?,
            confirmation_title: row.get("confirmation_title"),
            confirmation_body: row.get("confirmation_body"),
            request_payload: row.get::<Value, _>("request_payload_json"),
            response_payload: row.get::<Option<Value>, _>("response_payload_json"),
            created_at: row.get("created_at"),
            expires_at: row.get("expires_at"),
            decided_at: row.get("decided_at"),
            decided_by_key_id: row
                .get::<Option<String>, _>("decided_by_key_id")
                .map(PlatformApiKeyId::new),
            decision_note: row.get("decision_note"),
        },
    })
}