diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index def377a..567ef6e 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/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; diff --git a/crates/crank-registry/src/postgres/operation.rs b/crates/crank-registry/src/postgres/operation.rs index f5735ce..db0191f 100644 --- a/crates/crank-registry/src/postgres/operation.rs +++ b/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, 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, 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() - } } diff --git a/crates/crank-registry/src/postgres/operation_artifact.rs b/crates/crank-registry/src/postgres/operation_artifact.rs new file mode 100644 index 0000000..e01c687 --- /dev/null +++ b/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, 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, 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() + } +}