Add OpenAPI import backend

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

@@ -0,0 +1,107 @@
+use super::*;
+
+impl PostgresRegistry {
+    pub async fn create_import_job(
+        &self,
+        request: CreateImportJobRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        sqlx::query(
+            "insert into import_jobs (
+                id,
+                workspace_id,
+                kind,
+                source_format,
+                source_version,
+                status,
+                preview_payload,
+                created_operation_ids,
+                error_text,
+                created_at,
+                expires_at,
+                finished_at
+             ) values (
+                $1, $2, $3, $4, $5, $6, $7, '[]'::jsonb, null, $8::timestamptz, $9::timestamptz, null
+             )",
+        )
+        .bind(request.id.as_str())
+        .bind(request.workspace_id.as_str())
+        .bind(serialize_enum_text(&request.kind, "kind")?)
+        .bind(request.source_format)
+        .bind(request.source_version)
+        .bind(serialize_enum_text(&request.status, "status")?)
+        .bind(request.preview_payload)
+        .bind(request.created_at)
+        .bind(request.expires_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn get_import_job(
+        &self,
+        workspace_id: &WorkspaceId,
+        job_id: &ImportJobId,
+    ) -> Result<Option<ImportJob>, RegistryError> {
+        let row = sqlx::query(
+            "select
+                id,
+                workspace_id,
+                kind,
+                source_format,
+                source_version,
+                status,
+                preview_payload,
+                created_operation_ids,
+                error_text,
+                created_at,
+                expires_at,
+                finished_at
+             from import_jobs
+             where id = $1 and workspace_id = $2",
+        )
+        .bind(job_id.as_str())
+        .bind(workspace_id.as_str())
+        .fetch_optional(&self.pool)
+        .await?;
+
+        row.as_ref().map(map_import_job).transpose()
+    }
+
+    pub async fn finish_import_job(
+        &self,
+        request: FinishImportJobRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        let result = sqlx::query(
+            "update import_jobs
+                set status = $2,
+                    created_operation_ids = $3,
+                    error_text = $4,
+                    finished_at = $5::timestamptz
+              where id = $1",
+        )
+        .bind(request.id.as_str())
+        .bind(serialize_enum_text(&request.status, "status")?)
+        .bind(request.created_operation_ids)
+        .bind(request.error_text)
+        .bind(request.finished_at)
+        .execute(&self.pool)
+        .await?;
+
+        if result.rows_affected() == 0 {
+            return Err(RegistryError::ImportJobNotFound {
+                job_id: request.id.as_str().to_owned(),
+            });
+        }
+
+        Ok(())
+    }
+
+    pub async fn delete_expired_import_jobs(&self) -> Result<u64, RegistryError> {
+        let result = sqlx::query("delete from import_jobs where expires_at < now()")
+            .execute(&self.pool)
+            .await?;
+
+        Ok(result.rows_affected())
+    }
+}

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

@@ -2,6 +2,7 @@ mod agent;
 mod api_key;
 mod auth;
 mod connection;
+mod import_job;
 mod observability;
 mod operation;
 mod operation_artifact;
@@ -28,9 +29,10 @@ use crate::{
     error::RegistryError,
     model::{
         AgentSummary, AgentVersionRecord, AuthUserRecord, CreateAgentDraftVersionRequest,
-        CreateAgentRequest, CreateInvitationRequest, CreateInvocationLogRequest,
-        CreatePlatformApiKeyRequest, CreateSecretRequest, CreateVersionRequest,
-        CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorMetadata, InvitationRecord,
+        CreateAgentRequest, CreateImportJobRequest, CreateInvitationRequest,
+        CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateSecretRequest,
+        CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest,
+        DescriptorMetadata, FinishImportJobRequest, ImportJob, ImportJobId, InvitationRecord,
         InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationAgentRef,
         OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord,
         PlatformApiKeyRecord, PublishAgentRequest, PublishRequest, PublishedAgentTool,
@@ -661,6 +663,23 @@ fn map_yaml_import_job(row: &PgRow) -> Result<YamlImportJob, RegistryError> {
     })
 }
 
+fn map_import_job(row: &PgRow) -> Result<ImportJob, RegistryError> {
+    Ok(ImportJob {
+        id: ImportJobId::new(row.try_get::<String, _>("id")?),
+        workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
+        kind: deserialize_enum_text(&row.try_get::<String, _>("kind")?, "kind")?,
+        source_format: row.try_get("source_format")?,
+        source_version: row.try_get("source_version")?,
+        status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
+        preview_payload: row.try_get("preview_payload")?,
+        created_operation_ids: row.try_get("created_operation_ids")?,
+        error_text: row.try_get("error_text")?,
+        created_at: row.try_get("created_at")?,
+        expires_at: row.try_get("expires_at")?,
+        finished_at: row.try_get("finished_at")?,
+    })
+}
+
 fn map_usage_operation_breakdown(row: &PgRow) -> Result<UsageOperationBreakdown, RegistryError> {
     Ok(UsageOperationBreakdown {
         operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),