Split operation artifact persistence

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

@@ -4,6 +4,7 @@ mod auth;
 mod connection;
 mod observability;
 mod operation;
+mod operation_artifact;
 mod pool_config;
 mod secret;
 mod upstream;

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

@@ -851,172 +851,4 @@ impl PostgresRegistry {
             })
             .collect()
     }
-
-    pub async fn save_sample_metadata(
-        &self,
-        request: SaveSampleMetadataRequest<'_>,
-    ) -> Result<(), RegistryError> {
-        sqlx::query(
-            "insert into operation_samples (
-                id,
-                operation_id,
-                version,
-                sample_kind,
-                storage_ref,
-                content_type,
-                file_name,
-                created_at
-             ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
-             on conflict(id) do update set
-                operation_id = excluded.operation_id,
-                version = excluded.version,
-                sample_kind = excluded.sample_kind,
-                storage_ref = excluded.storage_ref,
-                content_type = excluded.content_type,
-                file_name = excluded.file_name,
-                created_at = excluded.created_at",
-        )
-        .bind(request.sample.id.as_str())
-        .bind(request.sample.operation_id.as_str())
-        .bind(to_db_version(request.sample.version))
-        .bind(serialize_enum_text(
-            &request.sample.sample_kind,
-            "sample_kind",
-        )?)
-        .bind(&request.sample.storage_ref)
-        .bind(&request.sample.content_type)
-        .bind(request.sample.file_name.as_deref())
-        .bind(request.sample.created_at)
-        .execute(&self.pool)
-        .await?;
-
-        Ok(())
-    }
-
-    pub async fn list_sample_metadata(
-        &self,
-        operation_id: &OperationId,
-        version: u32,
-    ) -> Result<Vec<OperationSampleMetadata>, RegistryError> {
-        let rows = sqlx::query!(
-            "select
-                id,
-                operation_id,
-                version,
-                sample_kind,
-                storage_ref,
-                content_type,
-                file_name,
-                created_at as \"created_at!: time::OffsetDateTime\"
-             from operation_samples
-             where operation_id = $1 and version = $2
-             order by created_at asc",
-            operation_id.as_str(),
-            to_db_version(version),
-        )
-        .fetch_all(&self.pool)
-        .await?;
-
-        rows.into_iter()
-            .map(|row| {
-                build_sample_metadata(
-                    row.id,
-                    row.operation_id,
-                    row.version,
-                    row.sample_kind,
-                    row.storage_ref,
-                    row.content_type,
-                    row.file_name,
-                    row.created_at,
-                )
-            })
-            .collect()
-    }
-
-    pub async fn save_descriptor_metadata(
-        &self,
-        request: SaveDescriptorMetadataRequest<'_>,
-    ) -> Result<(), RegistryError> {
-        sqlx::query(
-            "insert into descriptors (
-                id,
-                operation_id,
-                version,
-                descriptor_kind,
-                storage_ref,
-                source_name,
-                package_index_json,
-                created_at
-             ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
-             on conflict(id) do update set
-                operation_id = excluded.operation_id,
-                version = excluded.version,
-                descriptor_kind = excluded.descriptor_kind,
-                storage_ref = excluded.storage_ref,
-                source_name = excluded.source_name,
-                package_index_json = excluded.package_index_json,
-                created_at = excluded.created_at",
-        )
-        .bind(request.descriptor.id.as_str())
-        .bind(
-            request
-                .descriptor
-                .operation_id
-                .as_ref()
-                .map(|value| value.as_str()),
-        )
-        .bind(request.descriptor.version.map(to_db_version))
-        .bind(serialize_enum_text(
-            &request.descriptor.descriptor_kind,
-            "descriptor_kind",
-        )?)
-        .bind(&request.descriptor.storage_ref)
-        .bind(request.descriptor.source_name.as_deref())
-        .bind(serialize_option_json_value(&request.descriptor.package_index)?.map(Json))
-        .bind(request.descriptor.created_at)
-        .execute(&self.pool)
-        .await?;
-
-        Ok(())
-    }
-
-    pub async fn list_descriptor_metadata(
-        &self,
-        operation_id: &OperationId,
-        version: u32,
-    ) -> Result<Vec<DescriptorMetadata>, RegistryError> {
-        let rows = sqlx::query!(
-            "select
-                id,
-                operation_id,
-                version,
-                descriptor_kind,
-                storage_ref,
-                source_name,
-                package_index_json,
-                created_at as \"created_at!: time::OffsetDateTime\"
-             from descriptors
-             where operation_id = $1 and version = $2
-             order by created_at asc",
-            operation_id.as_str(),
-            to_db_version(version),
-        )
-        .fetch_all(&self.pool)
-        .await?;
-
-        rows.into_iter()
-            .map(|row| {
-                build_descriptor_metadata(
-                    row.id,
-                    row.operation_id,
-                    row.version,
-                    row.descriptor_kind,
-                    row.storage_ref,
-                    row.source_name,
-                    row.package_index_json,
-                    row.created_at,
-                )
-            })
-            .collect()
-    }
 }

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

@@ -0,0 +1,171 @@
+use super::*;
+
+impl PostgresRegistry {
+    pub async fn save_sample_metadata(
+        &self,
+        request: SaveSampleMetadataRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        sqlx::query(
+            "insert into operation_samples (
+                id,
+                operation_id,
+                version,
+                sample_kind,
+                storage_ref,
+                content_type,
+                file_name,
+                created_at
+             ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
+             on conflict(id) do update set
+                operation_id = excluded.operation_id,
+                version = excluded.version,
+                sample_kind = excluded.sample_kind,
+                storage_ref = excluded.storage_ref,
+                content_type = excluded.content_type,
+                file_name = excluded.file_name,
+                created_at = excluded.created_at",
+        )
+        .bind(request.sample.id.as_str())
+        .bind(request.sample.operation_id.as_str())
+        .bind(to_db_version(request.sample.version))
+        .bind(serialize_enum_text(
+            &request.sample.sample_kind,
+            "sample_kind",
+        )?)
+        .bind(&request.sample.storage_ref)
+        .bind(&request.sample.content_type)
+        .bind(request.sample.file_name.as_deref())
+        .bind(request.sample.created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn list_sample_metadata(
+        &self,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<Vec<OperationSampleMetadata>, RegistryError> {
+        let rows = sqlx::query!(
+            "select
+                id,
+                operation_id,
+                version,
+                sample_kind,
+                storage_ref,
+                content_type,
+                file_name,
+                created_at as \"created_at!: time::OffsetDateTime\"
+             from operation_samples
+             where operation_id = $1 and version = $2
+             order by created_at asc",
+            operation_id.as_str(),
+            to_db_version(version),
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.into_iter()
+            .map(|row| {
+                build_sample_metadata(
+                    row.id,
+                    row.operation_id,
+                    row.version,
+                    row.sample_kind,
+                    row.storage_ref,
+                    row.content_type,
+                    row.file_name,
+                    row.created_at,
+                )
+            })
+            .collect()
+    }
+
+    pub async fn save_descriptor_metadata(
+        &self,
+        request: SaveDescriptorMetadataRequest<'_>,
+    ) -> Result<(), RegistryError> {
+        sqlx::query(
+            "insert into descriptors (
+                id,
+                operation_id,
+                version,
+                descriptor_kind,
+                storage_ref,
+                source_name,
+                package_index_json,
+                created_at
+             ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz)
+             on conflict(id) do update set
+                operation_id = excluded.operation_id,
+                version = excluded.version,
+                descriptor_kind = excluded.descriptor_kind,
+                storage_ref = excluded.storage_ref,
+                source_name = excluded.source_name,
+                package_index_json = excluded.package_index_json,
+                created_at = excluded.created_at",
+        )
+        .bind(request.descriptor.id.as_str())
+        .bind(
+            request
+                .descriptor
+                .operation_id
+                .as_ref()
+                .map(|value| value.as_str()),
+        )
+        .bind(request.descriptor.version.map(to_db_version))
+        .bind(serialize_enum_text(
+            &request.descriptor.descriptor_kind,
+            "descriptor_kind",
+        )?)
+        .bind(&request.descriptor.storage_ref)
+        .bind(request.descriptor.source_name.as_deref())
+        .bind(serialize_option_json_value(&request.descriptor.package_index)?.map(Json))
+        .bind(request.descriptor.created_at)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(())
+    }
+
+    pub async fn list_descriptor_metadata(
+        &self,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<Vec<DescriptorMetadata>, RegistryError> {
+        let rows = sqlx::query!(
+            "select
+                id,
+                operation_id,
+                version,
+                descriptor_kind,
+                storage_ref,
+                source_name,
+                package_index_json,
+                created_at as \"created_at!: time::OffsetDateTime\"
+             from descriptors
+             where operation_id = $1 and version = $2
+             order by created_at asc",
+            operation_id.as_str(),
+            to_db_version(version),
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        rows.into_iter()
+            .map(|row| {
+                build_descriptor_metadata(
+                    row.id,
+                    row.operation_id,
+                    row.version,
+                    row.descriptor_kind,
+                    row.storage_ref,
+                    row.source_name,
+                    row.package_index_json,
+                    row.created_at,
+                )
+            })
+            .collect()
+    }
+}