use super::*;

impl PostgresRegistry {
    pub async fn create_operation(
        &self,
        workspace_id: &WorkspaceId,
        snapshot: &RegistryOperation,
        created_by: Option<&str>,
    ) -> Result<(), RegistryError> {
        if snapshot.version != 1 {
            return Err(RegistryError::InvalidInitialVersion {
                operation_id: snapshot.id.as_str().to_owned(),
                version: snapshot.version,
            });
        }

        if self
            .get_operation_summary(workspace_id, &snapshot.id)
            .await?
            .is_some()
        {
            return Err(RegistryError::OperationAlreadyExists {
                operation_id: snapshot.id.as_str().to_owned(),
            });
        }

        let mut tx = self.pool.begin().await?;

        sqlx::query(
            "insert into operations (
                id,
                workspace_id,
                name,
                display_name,
                category,
                protocol,
                security_level,
                status,
                current_draft_version,
                latest_published_version,
                created_at,
                updated_at,
                published_at
            ) values (
                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
                $11::timestamptz,
                $12::timestamptz,
                $13::timestamptz
            )",
        )
        .bind(snapshot.id.as_str())
        .bind(workspace_id.as_str())
        .bind(&snapshot.name)
        .bind(&snapshot.display_name)
        .bind(&snapshot.category)
        .bind(serialize_enum_text(&snapshot.protocol, "protocol")?)
        .bind(serialize_enum_text(
            &snapshot.security_level,
            "security_level",
        )?)
        .bind(serialize_enum_text(&snapshot.status, "status")?)
        .bind(to_db_version(snapshot.version))
        .bind(
            snapshot
                .published_at
                .as_ref()
                .map(|_| to_db_version(snapshot.version)),
        )
        .bind(snapshot.created_at)
        .bind(snapshot.updated_at)
        .bind(snapshot.published_at)
        .execute(&mut *tx)
        .await?;

        insert_version_row(&mut tx, snapshot, None, created_by).await?;

        tx.commit().await?;
        Ok(())
    }

    pub async fn create_version(
        &self,
        request: CreateVersionRequest<'_>,
    ) -> Result<(), RegistryError> {
        let Some(summary) = self
            .get_operation_summary(request.workspace_id, &request.snapshot.id)
            .await?
        else {
            return Err(RegistryError::OperationNotFound {
                operation_id: request.snapshot.id.as_str().to_owned(),
            });
        };

        assert_immutable_fields(&summary, request.snapshot)?;

        let expected = summary.current_draft_version + 1;
        if request.snapshot.version != expected {
            return Err(RegistryError::InvalidVersionSequence {
                operation_id: request.snapshot.id.as_str().to_owned(),
                expected,
                actual: request.snapshot.version,
            });
        }

        let mut tx = self.pool.begin().await?;

        insert_version_row(
            &mut tx,
            request.snapshot,
            request.change_note,
            request.created_by,
        )
        .await?;

        sqlx::query(
            "update operations
                set status = $1,
                    current_draft_version = $2,
                    updated_at = $3::timestamptz
              where id = $4",
        )
        .bind(serialize_enum_text(&request.snapshot.status, "status")?)
        .bind(to_db_version(request.snapshot.version))
        .bind(request.snapshot.updated_at)
        .bind(request.snapshot.id.as_str())
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;
        Ok(())
    }

    pub async fn update_operation_draft(
        &self,
        workspace_id: &WorkspaceId,
        snapshot: &RegistryOperation,
    ) -> Result<(), RegistryError> {
        let Some(summary) = self
            .get_operation_summary(workspace_id, &snapshot.id)
            .await?
        else {
            return Err(RegistryError::OperationNotFound {
                operation_id: snapshot.id.as_str().to_owned(),
            });
        };

        if snapshot.version != summary.current_draft_version {
            return Err(RegistryError::InvalidVersionSequence {
                operation_id: snapshot.id.as_str().to_owned(),
                expected: summary.current_draft_version,
                actual: snapshot.version,
            });
        }

        assert_immutable_fields(&summary, snapshot)?;

        let mut tx = self.pool.begin().await?;

        sqlx::query(
            "update operations
                set display_name = $1,
                    category = $2,
                    security_level = $3,
                    status = $4,
                    updated_at = $5::timestamptz
              where workspace_id = $6 and id = $7",
        )
        .bind(&snapshot.display_name)
        .bind(&snapshot.category)
        .bind(serialize_enum_text(
            &snapshot.security_level,
            "security_level",
        )?)
        .bind(serialize_enum_text(&snapshot.status, "status")?)
        .bind(snapshot.updated_at)
        .bind(workspace_id.as_str())
        .bind(snapshot.id.as_str())
        .execute(&mut *tx)
        .await?;

        sqlx::query(
            "update operation_versions
                set status = $1,
                    target_json = $2,
                    input_schema_json = $3,
                    output_schema_json = $4,
                    input_mapping_json = $5,
                    output_mapping_json = $6,
                    execution_config_json = $7,
                    tool_description_json = $8,
                    samples_json = $9,
                    generated_draft_json = $10,
                    config_export_json = $11
              where operation_id = $12 and version = $13",
        )
        .bind(serialize_enum_text(&snapshot.status, "status")?)
        .bind(Json(serialize_json_value(&snapshot.target)?))
        .bind(Json(serialize_json_value(&snapshot.input_schema)?))
        .bind(Json(serialize_json_value(&snapshot.output_schema)?))
        .bind(Json(serialize_json_value(&snapshot.input_mapping)?))
        .bind(Json(serialize_json_value(&snapshot.output_mapping)?))
        .bind(Json(serialize_json_value(&snapshot.execution_config)?))
        .bind(Json(serialize_json_value(&snapshot.tool_description)?))
        .bind(serialize_option_json_value(&snapshot.samples)?.map(Json))
        .bind(serialize_option_json_value(&snapshot.generated_draft)?.map(Json))
        .bind(serialize_option_json_value(&snapshot.config_export)?.map(Json))
        .bind(snapshot.id.as_str())
        .bind(to_db_version(snapshot.version))
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;
        Ok(())
    }

    pub async fn archive_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        archived_at: &time::OffsetDateTime,
    ) -> Result<(), RegistryError> {
        let Some(summary) = self
            .get_operation_summary(workspace_id, operation_id)
            .await?
        else {
            return Err(RegistryError::OperationNotFound {
                operation_id: operation_id.as_str().to_owned(),
            });
        };

        let mut tx = self.pool.begin().await?;

        sqlx::query(
            "update operations
                set status = $1,
                    updated_at = $2::timestamptz
              where workspace_id = $3 and id = $4",
        )
        .bind(serialize_enum_text(&OperationStatus::Archived, "status")?)
        .bind(archived_at)
        .bind(workspace_id.as_str())
        .bind(operation_id.as_str())
        .execute(&mut *tx)
        .await?;

        sqlx::query(
            "update operation_versions
                set status = $1
              where operation_id = $2 and version = $3",
        )
        .bind(serialize_enum_text(&OperationStatus::Archived, "status")?)
        .bind(operation_id.as_str())
        .bind(to_db_version(summary.current_draft_version))
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;
        Ok(())
    }

    pub async fn delete_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<(), RegistryError> {
        if self
            .has_published_agent_bindings_for_operation(workspace_id, operation_id)
            .await?
        {
            return Err(RegistryError::OperationHasPublishedAgentBindings {
                operation_id: operation_id.as_str().to_owned(),
            });
        }

        let deleted = sqlx::query(
            "delete from operations
              where workspace_id = $1 and id = $2",
        )
        .bind(workspace_id.as_str())
        .bind(operation_id.as_str())
        .execute(&self.pool)
        .await?
        .rows_affected();

        if deleted == 0 {
            return Err(RegistryError::OperationNotFound {
                operation_id: operation_id.as_str().to_owned(),
            });
        }

        Ok(())
    }

    pub async fn has_published_agent_bindings_for_operation(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<bool, RegistryError> {
        let row = sqlx::query!(
            "select 1 as \"present!\"
             from agents a
             join published_agents pa on pa.agent_id = a.id
             join agent_operation_bindings b
               on b.agent_id = a.id and b.agent_version = pa.version
             where a.workspace_id = $1
               and b.operation_id = $2
             limit 1",
            workspace_id.as_str(),
            operation_id.as_str(),
        )
        .fetch_optional(&self.pool)
        .await?;

        Ok(row.is_some())
    }

    pub async fn list_operations(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<OperationSummary>, RegistryError> {
        let rows = sqlx::query!(
            "select
                operations.id,
                operations.workspace_id,
                operations.name,
                operations.display_name,
                operations.category,
                operations.protocol,
                operations.security_level,
                ov.target_json,
                operations.status,
                operations.current_draft_version,
                operations.latest_published_version,
                operations.created_at as \"created_at!: time::OffsetDateTime\",
                operations.updated_at as \"updated_at!: time::OffsetDateTime\",
                operations.published_at as \"published_at: time::OffsetDateTime\"
             from operations
             join operation_versions ov
               on ov.operation_id = operations.id
              and ov.version = operations.current_draft_version
             where operations.workspace_id = $1
             order by operations.name asc",
            workspace_id.as_str(),
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(|row| {
                build_operation_summary(
                    row.id,
                    row.workspace_id,
                    row.name,
                    row.display_name,
                    row.category,
                    row.protocol,
                    row.security_level,
                    row.target_json,
                    row.status,
                    row.current_draft_version,
                    row.latest_published_version,
                    row.created_at,
                    row.updated_at,
                    row.published_at,
                )
            })
            .collect()
    }

    pub async fn list_operation_usage_summaries(
        &self,
        workspace_id: &WorkspaceId,
        created_after: &str,
    ) -> Result<Vec<OperationUsageSummary>, RegistryError> {
        let rows = sqlx::query(
            "select
                operation_id,
                count(*)::bigint as calls_today,
                coalesce(
                    round((sum(case when status = 'error' then 1 else 0 end)::numeric / nullif(count(*), 0)) * 100, 2),
                    0
                )::float8 as error_rate_pct,
                coalesce(round(avg(duration_ms))::bigint, 0) as avg_latency_ms
             from invocation_logs
             where workspace_id = $1
               and created_at >= $2::timestamptz
             group by operation_id",
        )
        .bind(workspace_id.as_str())
        .bind(created_after)
        .fetch_all(&self.pool)
        .await?;

        rows.iter().map(map_operation_usage_summary).collect()
    }

    pub async fn list_operation_agent_refs(
        &self,
        workspace_id: &WorkspaceId,
    ) -> Result<Vec<OperationAgentRef>, RegistryError> {
        let rows = sqlx::query!(
            "select
                b.operation_id,
                a.id as agent_id,
                a.slug as agent_slug,
                a.display_name
             from agents a
             join published_agents pa on pa.agent_id = a.id
             join agent_operation_bindings b
               on b.agent_id = a.id and b.agent_version = pa.version
             where a.workspace_id = $1
             order by a.display_name asc, b.tool_name asc",
            workspace_id.as_str(),
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(|row| {
                build_operation_agent_ref(
                    row.operation_id,
                    row.agent_id,
                    row.agent_slug,
                    row.display_name,
                )
            })
            .collect()
    }

    pub async fn get_operation_summary(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<Option<OperationSummary>, RegistryError> {
        let row = sqlx::query!(
            "select
                operations.id,
                operations.workspace_id,
                operations.name,
                operations.display_name,
                operations.category,
                operations.protocol,
                operations.security_level,
                ov.target_json,
                operations.status,
                operations.current_draft_version,
                operations.latest_published_version,
                operations.created_at as \"created_at!: time::OffsetDateTime\",
                operations.updated_at as \"updated_at!: time::OffsetDateTime\",
                operations.published_at as \"published_at: time::OffsetDateTime\"
             from operations
             join operation_versions ov
               on ov.operation_id = operations.id
              and ov.version = operations.current_draft_version
             where operations.workspace_id = $1 and operations.id = $2",
            workspace_id.as_str(),
            operation_id.as_str(),
        )
        .fetch_optional(&self.pool)
        .await?;

        row.map(|row| {
            build_operation_summary(
                row.id,
                row.workspace_id,
                row.name,
                row.display_name,
                row.category,
                row.protocol,
                row.security_level,
                row.target_json,
                row.status,
                row.current_draft_version,
                row.latest_published_version,
                row.created_at,
                row.updated_at,
                row.published_at,
            )
        })
        .transpose()
    }

    pub async fn get_operation_version(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
        version: u32,
    ) -> Result<Option<OperationVersionRecord>, RegistryError> {
        let row = sqlx::query!(
            "select
                o.id,
                o.workspace_id,
                o.name,
                o.display_name,
                o.category,
                o.protocol,
                o.security_level,
                o.created_at as \"operation_created_at!: time::OffsetDateTime\",
                o.updated_at as \"operation_updated_at!: time::OffsetDateTime\",
                o.published_at as \"operation_published_at: time::OffsetDateTime\",
                ov.version,
                ov.status,
                ov.target_json,
                ov.input_schema_json,
                ov.output_schema_json,
                ov.input_mapping_json,
                ov.output_mapping_json,
                ov.execution_config_json,
                ov.tool_description_json,
                ov.samples_json,
                ov.generated_draft_json,
                ov.config_export_json,
                ov.change_note,
                ov.created_at as \"created_at!: time::OffsetDateTime\",
                ov.created_by
             from operation_versions ov
             join operations o on o.id = ov.operation_id
             where o.workspace_id = $1 and ov.operation_id = $2 and ov.version = $3",
            workspace_id.as_str(),
            operation_id.as_str(),
            to_db_version(version),
        )
        .fetch_optional(&self.pool)
        .await?;

        row.map(|row| {
            build_operation_version_record(
                row.id,
                row.workspace_id,
                row.name,
                row.display_name,
                row.category,
                row.protocol,
                row.security_level,
                row.operation_created_at,
                row.operation_updated_at,
                row.operation_published_at,
                row.version,
                row.status,
                row.target_json,
                row.input_schema_json,
                row.output_schema_json,
                row.input_mapping_json,
                row.output_mapping_json,
                row.execution_config_json,
                row.tool_description_json,
                row.samples_json,
                row.generated_draft_json,
                row.config_export_json,
                row.change_note,
                row.created_at,
                row.created_by,
            )
        })
        .transpose()
    }

    pub async fn list_operation_versions(
        &self,
        workspace_id: &WorkspaceId,
        operation_id: &OperationId,
    ) -> Result<Vec<OperationVersionRecord>, RegistryError> {
        let rows = sqlx::query!(
            "select
                o.id,
                o.workspace_id,
                o.name,
                o.display_name,
                o.category,
                o.protocol,
                o.security_level,
                o.created_at as \"operation_created_at!: time::OffsetDateTime\",
                o.updated_at as \"operation_updated_at!: time::OffsetDateTime\",
                o.published_at as \"operation_published_at: time::OffsetDateTime\",
                ov.version,
                ov.status,
                ov.target_json,
                ov.input_schema_json,
                ov.output_schema_json,
                ov.input_mapping_json,
                ov.output_mapping_json,
                ov.execution_config_json,
                ov.tool_description_json,
                ov.samples_json,
                ov.generated_draft_json,
                ov.config_export_json,
                ov.change_note,
                ov.created_at as \"created_at!: time::OffsetDateTime\",
                ov.created_by
             from operation_versions ov
             join operations o on o.id = ov.operation_id
             where o.workspace_id = $1 and ov.operation_id = $2
             order by ov.version asc",
            workspace_id.as_str(),
            operation_id.as_str(),
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(|row| {
                build_operation_version_record(
                    row.id,
                    row.workspace_id,
                    row.name,
                    row.display_name,
                    row.category,
                    row.protocol,
                    row.security_level,
                    row.operation_created_at,
                    row.operation_updated_at,
                    row.operation_published_at,
                    row.version,
                    row.status,
                    row.target_json,
                    row.input_schema_json,
                    row.output_schema_json,
                    row.input_mapping_json,
                    row.output_mapping_json,
                    row.execution_config_json,
                    row.tool_description_json,
                    row.samples_json,
                    row.generated_draft_json,
                    row.config_export_json,
                    row.change_note,
                    row.created_at,
                    row.created_by,
                )
            })
            .collect()
    }

    pub async fn publish_operation(
        &self,
        request: PublishRequest<'_>,
    ) -> Result<(), RegistryError> {
        if self
            .get_operation_version(request.workspace_id, request.operation_id, request.version)
            .await?
            .is_none()
        {
            return Err(RegistryError::OperationVersionNotFound {
                operation_id: request.operation_id.as_str().to_owned(),
                version: request.version,
            });
        }

        let mut tx = self.pool.begin().await?;

        sqlx::query(
            "insert into published_operations (
                operation_id,
                version,
                published_at,
                published_by
             ) values ($1, $2, $3::timestamptz, $4)
             on conflict(operation_id) do update set
                version = excluded.version,
                published_at = excluded.published_at,
                published_by = excluded.published_by",
        )
        .bind(request.operation_id.as_str())
        .bind(to_db_version(request.version))
        .bind(request.published_at)
        .bind(request.published_by)
        .execute(&mut *tx)
        .await?;

        sqlx::query(
            "update operation_versions
                set status = $1
              where operation_id = $2 and version = $3",
        )
        .bind(serialize_enum_text(&OperationStatus::Published, "status")?)
        .bind(request.operation_id.as_str())
        .bind(to_db_version(request.version))
        .execute(&mut *tx)
        .await?;

        sqlx::query(
            "update operations
                set status = $1,
                    latest_published_version = $2,
                    published_at = $3::timestamptz,
                    updated_at = $4::timestamptz
              where id = $5",
        )
        .bind(serialize_enum_text(&OperationStatus::Published, "status")?)
        .bind(to_db_version(request.version))
        .bind(request.published_at)
        .bind(request.published_at)
        .bind(request.operation_id.as_str())
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;
        Ok(())
    }

    pub async fn get_published_operation(
        &self,
        operation_id: &OperationId,
    ) -> Result<Option<RegistryOperation>, RegistryError> {
        let row = sqlx::query!(
            "select
                o.id,
                o.workspace_id,
                o.name,
                o.display_name,
                o.category,
                o.protocol,
                o.security_level,
                o.created_at as \"operation_created_at!: time::OffsetDateTime\",
                o.updated_at as \"operation_updated_at!: time::OffsetDateTime\",
                o.published_at as \"operation_published_at: time::OffsetDateTime\",
                ov.version,
                ov.status,
                ov.target_json,
                ov.input_schema_json,
                ov.output_schema_json,
                ov.input_mapping_json,
                ov.output_mapping_json,
                ov.execution_config_json,
                ov.tool_description_json,
                ov.samples_json,
                ov.generated_draft_json,
                ov.config_export_json,
                ov.change_note,
                ov.created_at as \"created_at!: time::OffsetDateTime\",
                ov.created_by
             from published_operations po
             join operation_versions ov
               on ov.operation_id = po.operation_id and ov.version = po.version
             join operations o on o.id = po.operation_id
             where po.operation_id = $1",
            operation_id.as_str(),
        )
        .fetch_optional(&self.pool)
        .await?;

        row.map(|row| {
            build_operation_version_record(
                row.id,
                row.workspace_id,
                row.name,
                row.display_name,
                row.category,
                row.protocol,
                row.security_level,
                row.operation_created_at,
                row.operation_updated_at,
                row.operation_published_at,
                row.version,
                row.status,
                row.target_json,
                row.input_schema_json,
                row.output_schema_json,
                row.input_mapping_json,
                row.output_mapping_json,
                row.execution_config_json,
                row.tool_description_json,
                row.samples_json,
                row.generated_draft_json,
                row.config_export_json,
                row.change_note,
                row.created_at,
                row.created_by,
            )
            .map(|record| record.snapshot)
        })
        .transpose()
    }

    pub async fn list_published_operations(&self) -> Result<Vec<RegistryOperation>, RegistryError> {
        let rows = sqlx::query!(
            "select
                o.id,
                o.workspace_id,
                o.name,
                o.display_name,
                o.category,
                o.protocol,
                o.security_level,
                o.created_at as \"operation_created_at!: time::OffsetDateTime\",
                o.updated_at as \"operation_updated_at!: time::OffsetDateTime\",
                o.published_at as \"operation_published_at: time::OffsetDateTime\",
                ov.version,
                ov.status,
                ov.target_json,
                ov.input_schema_json,
                ov.output_schema_json,
                ov.input_mapping_json,
                ov.output_mapping_json,
                ov.execution_config_json,
                ov.tool_description_json,
                ov.samples_json,
                ov.generated_draft_json,
                ov.config_export_json,
                ov.change_note,
                ov.created_at as \"created_at!: time::OffsetDateTime\",
                ov.created_by
             from published_operations po
             join operation_versions ov
               on ov.operation_id = po.operation_id and ov.version = po.version
             join operations o on o.id = po.operation_id
             order by o.name asc",
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(|row| {
                build_operation_version_record(
                    row.id,
                    row.workspace_id,
                    row.name,
                    row.display_name,
                    row.category,
                    row.protocol,
                    row.security_level,
                    row.operation_created_at,
                    row.operation_updated_at,
                    row.operation_published_at,
                    row.version,
                    row.status,
                    row.target_json,
                    row.input_schema_json,
                    row.output_schema_json,
                    row.input_mapping_json,
                    row.output_mapping_json,
                    row.execution_config_json,
                    row.tool_description_json,
                    row.samples_json,
                    row.generated_draft_json,
                    row.config_export_json,
                    row.change_note,
                    row.created_at,
                    row.created_by,
                )
                .map(|record| record.snapshot)
            })
            .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()
    }
}