crank/apps/admin-api/src/service/observability.rs

use crank_core::{
    AgentId, ApprovalRequestId, ApprovalRequestStatus, InvocationLogId, OperationId, UsagePeriod,
    WorkspaceId,
};
use crank_registry::{
    ApprovalRequestRecord, ExpireApprovalRequest, InvocationLogRecord, ListApprovalRequestsQuery,
    ListInvocationLogsQuery, UsageQuery, UsageRollupRecord,
};
use serde_json::json;
use time::OffsetDateTime;
use tracing::instrument;

use crate::{
    error::ApiError,
    service::{
        AdminService, ApprovalsQuery, LogsQuery, UsageOverviewResponse, UsageRequestQuery,
        usage_window,
    },
};

impl AdminService {
    #[instrument(skip(self))]
    pub async fn list_logs(
        &self,
        workspace_id: &WorkspaceId,
        query: LogsQuery,
    ) -> Result<Vec<InvocationLogRecord>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let operation_id = query.operation_id.as_deref().map(OperationId::new);
        let agent_id = query.agent_id.as_deref().map(AgentId::new);
        let (_, created_after, _) = usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;

        self.registry
            .list_invocation_logs(ListInvocationLogsQuery {
                workspace_id,
                level: query.level,
                search_text: query.search.as_deref(),
                source: query.source,
                operation_id: operation_id.as_ref(),
                agent_id: agent_id.as_ref(),
                created_after: Some(&created_after),
                limit: query.limit.unwrap_or(100),
            })
            .await
            .map_err(ApiError::from)
    }

    #[instrument(skip(self))]
    pub async fn get_log(
        &self,
        workspace_id: &WorkspaceId,
        log_id: &InvocationLogId,
    ) -> Result<InvocationLogRecord, ApiError> {
        self.registry
            .get_invocation_log(workspace_id, log_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found_with_context(
                    format!("invocation log {} was not found", log_id.as_str()),
                    json!({ "log_id": log_id.as_str() }),
                )
            })
    }

    #[instrument(skip(self))]
    pub async fn list_approvals(
        &self,
        workspace_id: &WorkspaceId,
        query: ApprovalsQuery,
    ) -> Result<Vec<ApprovalRequestRecord>, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let records = self
            .registry
            .list_approval_requests(ListApprovalRequestsQuery {
                workspace_id,
                status: query.status,
                limit: query.limit.unwrap_or(50).clamp(1, 200),
            })
            .await?;

        let mut normalized = Vec::with_capacity(records.len());
        for record in records {
            let record = self.normalize_approval_record(record).await?;
            if query.status.is_none() || record.approval.status == query.status.unwrap() {
                normalized.push(record);
            }
        }

        Ok(normalized)
    }

    #[instrument(skip(self))]
    pub async fn get_approval(
        &self,
        workspace_id: &WorkspaceId,
        approval_id: &ApprovalRequestId,
    ) -> Result<ApprovalRequestRecord, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let record = self
            .registry
            .get_approval_request(workspace_id, approval_id)
            .await?
            .ok_or_else(|| {
                ApiError::not_found_with_context(
                    format!("approval request {} was not found", approval_id.as_str()),
                    json!({ "approval_id": approval_id.as_str() }),
                )
            })?;

        self.normalize_approval_record(record).await
    }

    async fn normalize_approval_record(
        &self,
        record: ApprovalRequestRecord,
    ) -> Result<ApprovalRequestRecord, ApiError> {
        if record.approval.status != ApprovalRequestStatus::Pending
            || record.approval.expires_at > OffsetDateTime::now_utc()
        {
            return Ok(record);
        }

        Ok(self
            .registry
            .expire_approval_request(ExpireApprovalRequest {
                workspace_id: &record.approval.workspace_id,
                agent_id: &record.approval.agent_id,
                approval_id: &record.approval.id,
                expired_at: OffsetDateTime::now_utc(),
            })
            .await?
            .unwrap_or(record))
    }

    #[instrument(skip(self))]
    pub async fn get_usage_overview(
        &self,
        workspace_id: &WorkspaceId,
        query: UsageRequestQuery,
    ) -> Result<UsageOverviewResponse, ApiError> {
        self.ensure_workspace_exists(workspace_id).await?;
        let (period, created_after, bucket) =
            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;
        let usage_query = UsageQuery {
            workspace_id,
            period,
            source: query.source,
            created_after: &created_after,
            bucket,
        };

        let summary = self.registry.summarize_usage(usage_query.clone()).await?;
        let timeline = self
            .registry
            .list_usage_timeline(usage_query.clone())
            .await?;
        let operations = self
            .registry
            .list_usage_by_operation(usage_query.clone())
            .await?;
        let agents = self.registry.list_usage_by_agent(usage_query).await?;

        Ok(UsageOverviewResponse {
            summary,
            timeline,
            operations,
            agents,
        })
    }

    #[instrument(skip(self))]
    pub async fn get_operation_usage(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        query: UsageRequestQuery,
    ) -> Result<UsageRollupRecord, ApiError> {
        self.get_operation(workspace_id, operation_id).await?;
        let (period, created_after, bucket) =
            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;

        self.registry
            .get_usage_for_operation(
                UsageQuery {
                    workspace_id,
                    period,
                    source: query.source,
                    created_after: &created_after,
                    bucket,
                },
                operation_id,
            )
            .await?
            .ok_or_else(|| {
                ApiError::not_found_with_context(
                    format!(
                        "usage for operation {} was not found",
                        operation_id.as_str()
                    ),
                    json!({ "operation_id": operation_id.as_str() }),
                )
            })
    }

    #[instrument(skip(self))]
    pub async fn get_agent_usage(
        &self,
        workspace_id: &WorkspaceId,
        agent_id: &AgentId,
        query: UsageRequestQuery,
    ) -> Result<UsageRollupRecord, ApiError> {
        self.get_agent(workspace_id, agent_id).await?;
        let (period, created_after, bucket) =
            usage_window(query.period.unwrap_or(UsagePeriod::Last7Days))?;

        self.registry
            .get_usage_for_agent(
                UsageQuery {
                    workspace_id,
                    period,
                    source: query.source,
                    created_after: &created_after,
                    bucket,
                },
                agent_id,
            )
            .await?
            .ok_or_else(|| {
                ApiError::not_found_with_context(
                    format!("usage for agent {} was not found", agent_id.as_str()),
                    json!({ "agent_id": agent_id.as_str() }),
                )
            })
    }
}