use mcpaas_core::{AuthProfile, OperationId, OperationStatus};
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value;
use sqlx::{
    PgPool, Postgres, Row, Transaction,
    postgres::{PgPoolOptions, PgRow},
    types::Json,
};

use crate::{
    error::RegistryError,
    migrations,
    model::{
        CreateVersionRequest, CreateYamlImportJobRequest, DescriptorMetadata,
        OperationSampleMetadata, OperationSummary, OperationVersionRecord, PublishRequest,
        RegistryOperation, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
        SaveSampleMetadataRequest, YamlImportJob, YamlImportJobCompletion, YamlImportJobId,
        YamlImportJobStatus,
    },
};

#[derive(Clone, Debug)]
pub struct PostgresRegistry {
    pool: PgPool,
}

impl PostgresRegistry {
    pub async fn connect(database_url: &str) -> Result<Self, RegistryError> {
        Self::connect_in_schema(database_url, None).await
    }

    pub fn pool(&self) -> &PgPool {
        &self.pool
    }

    pub async fn migrate(&self) -> Result<(), RegistryError> {
        migrations::apply_postgres(&self.pool).await?;
        Ok(())
    }

    async fn connect_in_schema(
        database_url: &str,
        schema: Option<&str>,
    ) -> Result<Self, RegistryError> {
        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(database_url)
            .await?;

        if let Some(schema) = schema {
            sqlx::query(&format!("set search_path to {schema}"))
                .execute(&pool)
                .await?;
        }

        let registry = Self { pool };
        registry.migrate().await?;
        Ok(registry)
    }

    pub async fn create_operation(
        &self,
        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(&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,
                name,
                display_name,
                protocol,
                status,
                current_draft_version,
                latest_published_version,
                created_at,
                updated_at,
                published_at
            ) values (
                $1, $2, $3, $4, $5, $6, $7,
                $8::timestamptz,
                $9::timestamptz,
                $10::timestamptz
            )",
        )
        .bind(snapshot.id.as_str())
        .bind(&snapshot.name)
        .bind(&snapshot.display_name)
        .bind(serialize_enum_text(&snapshot.protocol, "protocol")?)
        .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.as_deref())
        .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.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 list_operations(&self) -> Result<Vec<OperationSummary>, RegistryError> {
        let rows = sqlx::query(
            "select
                id,
                name,
                display_name,
                protocol,
                status,
                current_draft_version,
                latest_published_version,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
                to_char(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
             from operations
             order by name asc",
        )
        .fetch_all(&self.pool)
        .await?;

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

    pub async fn get_operation_summary(
        &self,
        operation_id: &OperationId,
    ) -> Result<Option<OperationSummary>, RegistryError> {
        let row = sqlx::query(
            "select
                id,
                name,
                display_name,
                protocol,
                status,
                current_draft_version,
                latest_published_version,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
                to_char(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at
             from operations
             where id = $1",
        )
        .bind(operation_id.as_str())
        .fetch_optional(&self.pool)
        .await?;

        row.as_ref().map(map_operation_summary).transpose()
    }

    pub async fn get_operation_version(
        &self,
        operation_id: &OperationId,
        version: u32,
    ) -> Result<Option<OperationVersionRecord>, RegistryError> {
        let row = sqlx::query(
            "select
                o.id,
                o.name,
                o.display_name,
                o.protocol,
                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
                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,
                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                ov.created_by
             from operation_versions ov
             join operations o on o.id = ov.operation_id
             where ov.operation_id = $1 and ov.version = $2",
        )
        .bind(operation_id.as_str())
        .bind(to_db_version(version))
        .fetch_optional(&self.pool)
        .await?;

        row.as_ref().map(map_operation_version_record).transpose()
    }

    pub async fn list_operation_versions(
        &self,
        operation_id: &OperationId,
    ) -> Result<Vec<OperationVersionRecord>, RegistryError> {
        let rows = sqlx::query(
            "select
                o.id,
                o.name,
                o.display_name,
                o.protocol,
                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
                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,
                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                ov.created_by
             from operation_versions ov
             join operations o on o.id = ov.operation_id
             where ov.operation_id = $1
             order by ov.version asc",
        )
        .bind(operation_id.as_str())
        .fetch_all(&self.pool)
        .await?;

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

    pub async fn publish_operation(
        &self,
        request: PublishRequest<'_>,
    ) -> Result<(), RegistryError> {
        if self
            .get_operation_version(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.name,
                o.display_name,
                o.protocol,
                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
                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,
                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                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",
        )
        .bind(operation_id.as_str())
        .fetch_optional(&self.pool)
        .await?;

        row.as_ref()
            .map(|value| map_operation_version_record(value).map(|record| record.snapshot))
            .transpose()
    }

    pub async fn list_published_operations(&self) -> Result<Vec<RegistryOperation>, RegistryError> {
        let rows = sqlx::query(
            "select
                o.id,
                o.name,
                o.display_name,
                o.protocol,
                to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at,
                to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at,
                to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at,
                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,
                to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                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.iter()
            .map(|row| map_operation_version_record(row).map(|record| record.snapshot))
            .collect()
    }

    pub async fn save_auth_profile(
        &self,
        request: SaveAuthProfileRequest<'_>,
    ) -> Result<(), RegistryError> {
        sqlx::query(
            "insert into auth_profiles (
                id,
                name,
                kind,
                config_json,
                created_at,
                updated_at
             ) values ($1, $2, $3, $4, $5::timestamptz, $6::timestamptz)
             on conflict(id) do update set
                name = excluded.name,
                kind = excluded.kind,
                config_json = excluded.config_json,
                updated_at = excluded.updated_at",
        )
        .bind(request.profile.id.as_str())
        .bind(&request.profile.name)
        .bind(serialize_enum_text(&request.profile.kind, "kind")?)
        .bind(Json(serialize_json_value(&request.profile.config)?))
        .bind(&request.profile.created_at)
        .bind(&request.profile.updated_at)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    pub async fn get_auth_profile(
        &self,
        auth_profile_id: &mcpaas_core::AuthProfileId,
    ) -> Result<Option<AuthProfile>, RegistryError> {
        let row = sqlx::query(
            "select
                id,
                name,
                kind,
                config_json,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at
             from auth_profiles
             where id = $1",
        )
        .bind(auth_profile_id.as_str())
        .fetch_optional(&self.pool)
        .await?;

        row.as_ref().map(map_auth_profile).transpose()
    }

    pub async fn list_auth_profiles(&self) -> Result<Vec<AuthProfile>, RegistryError> {
        let rows = sqlx::query(
            "select
                id,
                name,
                kind,
                config_json,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at
             from auth_profiles
             order by name asc",
        )
        .fetch_all(&self.pool)
        .await?;

        rows.iter().map(map_auth_profile).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,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
             from operation_samples
             where operation_id = $1 and version = $2
             order by created_at asc",
        )
        .bind(operation_id.as_str())
        .bind(to_db_version(version))
        .fetch_all(&self.pool)
        .await?;

        rows.iter().map(map_sample_metadata).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,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
             from descriptors
             where operation_id = $1 and version = $2
             order by created_at asc",
        )
        .bind(operation_id.as_str())
        .bind(to_db_version(version))
        .fetch_all(&self.pool)
        .await?;

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

    pub async fn create_yaml_import_job(
        &self,
        request: CreateYamlImportJobRequest<'_>,
    ) -> Result<(), RegistryError> {
        sqlx::query(
            "insert into yaml_import_jobs (
                id,
                source_sample_id,
                status,
                format_version,
                mode,
                result_operation_id,
                result_version,
                error_text,
                created_at,
                finished_at
             ) values ($1, $2, $3, $4, $5, null, null, null, $6::timestamptz, null)",
        )
        .bind(request.id.as_str())
        .bind(request.source_sample_id.map(|value| value.as_str()))
        .bind(serialize_enum_text(
            &YamlImportJobStatus::Pending,
            "status",
        )?)
        .bind(request.format_version)
        .bind(serialize_enum_text(&request.mode, "mode")?)
        .bind(request.created_at)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    pub async fn finish_yaml_import_job(
        &self,
        job_id: &YamlImportJobId,
        completion: &YamlImportJobCompletion,
    ) -> Result<(), RegistryError> {
        let result = sqlx::query(
            "update yaml_import_jobs
                set status = $1,
                    result_operation_id = $2,
                    result_version = $3,
                    error_text = $4,
                    finished_at = $5::timestamptz
              where id = $6",
        )
        .bind(serialize_enum_text(&completion.status, "status")?)
        .bind(
            completion
                .result_operation_id
                .as_ref()
                .map(|value| value.as_str()),
        )
        .bind(completion.result_version.map(to_db_version))
        .bind(completion.error_text.as_deref())
        .bind(&completion.finished_at)
        .bind(job_id.as_str())
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            return Err(RegistryError::YamlImportJobNotFound {
                job_id: job_id.as_str().to_owned(),
            });
        }

        Ok(())
    }

    pub async fn get_yaml_import_job(
        &self,
        job_id: &YamlImportJobId,
    ) -> Result<Option<YamlImportJob>, RegistryError> {
        let row = sqlx::query(
            "select
                id,
                source_sample_id,
                status,
                format_version,
                mode,
                result_operation_id,
                result_version,
                error_text,
                to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
                to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at
             from yaml_import_jobs
             where id = $1",
        )
        .bind(job_id.as_str())
        .fetch_optional(&self.pool)
        .await?;

        row.as_ref().map(map_yaml_import_job).transpose()
    }
}

async fn insert_version_row(
    tx: &mut Transaction<'_, Postgres>,
    snapshot: &RegistryOperation,
    change_note: Option<&str>,
    created_by: Option<&str>,
) -> Result<(), RegistryError> {
    sqlx::query(
        "insert into operation_versions (
            operation_id,
            version,
            status,
            target_json,
            input_schema_json,
            output_schema_json,
            input_mapping_json,
            output_mapping_json,
            execution_config_json,
            tool_description_json,
            samples_json,
            generated_draft_json,
            config_export_json,
            change_note,
            created_at,
            created_by
         ) values (
            $1, $2, $3, $4, $5, $6, $7, $8,
            $9, $10, $11, $12, $13, $14, $15::timestamptz, $16
         )",
    )
    .bind(snapshot.id.as_str())
    .bind(to_db_version(snapshot.version))
    .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(change_note)
    .bind(&snapshot.updated_at)
    .bind(created_by)
    .execute(&mut **tx)
    .await?;

    Ok(())
}

fn assert_immutable_fields(
    summary: &OperationSummary,
    snapshot: &RegistryOperation,
) -> Result<(), RegistryError> {
    if summary.name != snapshot.name {
        return Err(RegistryError::ImmutableOperationFieldChanged {
            operation_id: snapshot.id.as_str().to_owned(),
            field: "name",
        });
    }

    if summary.display_name != snapshot.display_name {
        return Err(RegistryError::ImmutableOperationFieldChanged {
            operation_id: snapshot.id.as_str().to_owned(),
            field: "display_name",
        });
    }

    if summary.protocol != snapshot.protocol {
        return Err(RegistryError::ImmutableOperationFieldChanged {
            operation_id: snapshot.id.as_str().to_owned(),
            field: "protocol",
        });
    }

    Ok(())
}

fn map_operation_summary(row: &PgRow) -> Result<OperationSummary, RegistryError> {
    Ok(OperationSummary {
        id: OperationId::new(row.try_get::<String, _>("id")?),
        name: row.try_get("name")?,
        display_name: row.try_get("display_name")?,
        protocol: deserialize_enum_text(&row.try_get::<String, _>("protocol")?, "protocol")?,
        status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
        current_draft_version: from_db_version(
            row.try_get("current_draft_version")?,
            "current_draft_version",
        )?,
        latest_published_version: row
            .try_get::<Option<i32>, _>("latest_published_version")?
            .map(|value| from_db_version(value, "latest_published_version"))
            .transpose()?,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
        published_at: row.try_get("published_at")?,
    })
}

fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, RegistryError> {
    let operation_id = OperationId::new(row.try_get::<String, _>("id")?);
    let version = from_db_version(row.try_get("version")?, "version")?;
    let status = deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?;

    Ok(OperationVersionRecord {
        operation_id: operation_id.clone(),
        version,
        status,
        change_note: row.try_get("change_note")?,
        created_at: row.try_get("created_at")?,
        created_by: row.try_get("created_by")?,
        snapshot: RegistryOperation {
            id: operation_id,
            name: row.try_get("name")?,
            display_name: row.try_get("display_name")?,
            protocol: deserialize_enum_text(&row.try_get::<String, _>("protocol")?, "protocol")?,
            status,
            version,
            target: deserialize_json_value(row.try_get::<Json<Value>, _>("target_json")?.0)?,
            input_schema: deserialize_json_value(
                row.try_get::<Json<Value>, _>("input_schema_json")?.0,
            )?,
            output_schema: deserialize_json_value(
                row.try_get::<Json<Value>, _>("output_schema_json")?.0,
            )?,
            input_mapping: deserialize_json_value(
                row.try_get::<Json<Value>, _>("input_mapping_json")?.0,
            )?,
            output_mapping: deserialize_json_value(
                row.try_get::<Json<Value>, _>("output_mapping_json")?.0,
            )?,
            execution_config: deserialize_json_value(
                row.try_get::<Json<Value>, _>("execution_config_json")?.0,
            )?,
            tool_description: deserialize_json_value(
                row.try_get::<Json<Value>, _>("tool_description_json")?.0,
            )?,
            samples: row
                .try_get::<Option<Json<Value>>, _>("samples_json")?
                .map(|value| deserialize_json_value(value.0))
                .transpose()?,
            generated_draft: row
                .try_get::<Option<Json<Value>>, _>("generated_draft_json")?
                .map(|value| deserialize_json_value(value.0))
                .transpose()?,
            config_export: row
                .try_get::<Option<Json<Value>>, _>("config_export_json")?
                .map(|value| deserialize_json_value(value.0))
                .transpose()?,
            created_at: row.try_get("operation_created_at")?,
            updated_at: row.try_get("operation_updated_at")?,
            published_at: row.try_get("operation_published_at")?,
        },
    })
}

fn map_auth_profile(row: &PgRow) -> Result<AuthProfile, RegistryError> {
    Ok(AuthProfile {
        id: mcpaas_core::AuthProfileId::new(row.try_get::<String, _>("id")?),
        name: row.try_get("name")?,
        kind: deserialize_enum_text(&row.try_get::<String, _>("kind")?, "kind")?,
        config: deserialize_json_value(row.try_get::<Json<Value>, _>("config_json")?.0)?,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
    })
}

fn map_sample_metadata(row: &PgRow) -> Result<OperationSampleMetadata, RegistryError> {
    Ok(OperationSampleMetadata {
        id: mcpaas_core::SampleId::new(row.try_get::<String, _>("id")?),
        operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
        version: from_db_version(row.try_get("version")?, "version")?,
        sample_kind: deserialize_enum_text(
            &row.try_get::<String, _>("sample_kind")?,
            "sample_kind",
        )?,
        storage_ref: row.try_get("storage_ref")?,
        content_type: row.try_get("content_type")?,
        file_name: row.try_get("file_name")?,
        created_at: row.try_get("created_at")?,
    })
}

fn map_descriptor_metadata(row: &PgRow) -> Result<DescriptorMetadata, RegistryError> {
    Ok(DescriptorMetadata {
        id: mcpaas_core::DescriptorId::new(row.try_get::<String, _>("id")?),
        operation_id: row
            .try_get::<Option<String>, _>("operation_id")?
            .map(OperationId::new),
        version: row
            .try_get::<Option<i32>, _>("version")?
            .map(|value| from_db_version(value, "version"))
            .transpose()?,
        descriptor_kind: deserialize_enum_text(
            &row.try_get::<String, _>("descriptor_kind")?,
            "descriptor_kind",
        )?,
        storage_ref: row.try_get("storage_ref")?,
        source_name: row.try_get("source_name")?,
        package_index: row
            .try_get::<Option<Json<Value>>, _>("package_index_json")?
            .map(|value| value.0),
        created_at: row.try_get("created_at")?,
    })
}

fn map_yaml_import_job(row: &PgRow) -> Result<YamlImportJob, RegistryError> {
    Ok(YamlImportJob {
        id: YamlImportJobId::new(row.try_get::<String, _>("id")?),
        source_sample_id: row
            .try_get::<Option<String>, _>("source_sample_id")?
            .map(mcpaas_core::SampleId::new),
        status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
        format_version: row.try_get("format_version")?,
        mode: deserialize_enum_text(&row.try_get::<String, _>("mode")?, "mode")?,
        result_operation_id: row
            .try_get::<Option<String>, _>("result_operation_id")?
            .map(OperationId::new),
        result_version: row
            .try_get::<Option<i32>, _>("result_version")?
            .map(|value| from_db_version(value, "result_version"))
            .transpose()?,
        error_text: row.try_get("error_text")?,
        created_at: row.try_get("created_at")?,
        finished_at: row.try_get("finished_at")?,
    })
}

fn serialize_json_value<T: Serialize>(value: &T) -> Result<Value, RegistryError> {
    Ok(serde_json::to_value(value)?)
}

fn serialize_option_json_value<T: Serialize>(
    value: &Option<T>,
) -> Result<Option<Value>, RegistryError> {
    value.as_ref().map(serialize_json_value).transpose()
}

fn deserialize_json_value<T: DeserializeOwned>(value: Value) -> Result<T, RegistryError> {
    Ok(serde_json::from_value(value)?)
}

fn serialize_enum_text<T: Serialize>(
    value: &T,
    field: &'static str,
) -> Result<String, RegistryError> {
    serde_json::to_value(value)?
        .as_str()
        .map(ToOwned::to_owned)
        .ok_or(RegistryError::InvalidEnumRepresentation { field })
}

fn deserialize_enum_text<T: DeserializeOwned>(
    value: &str,
    field: &'static str,
) -> Result<T, RegistryError> {
    serde_json::from_value(Value::String(value.to_owned()))
        .map_err(|_| RegistryError::InvalidEnumRepresentation { field })
}

fn to_db_version(value: u32) -> i32 {
    value as i32
}

fn from_db_version(value: i32, field: &'static str) -> Result<u32, RegistryError> {
    u32::try_from(value).map_err(|_| RegistryError::InvalidNumericValue {
        field,
        value: i64::from(value),
    })
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        env,
        time::{SystemTime, UNIX_EPOCH},
    };

    use mcpaas_core::{
        ApiKeyHeaderAuthConfig, AuthConfig, AuthKind, AuthProfile, ConfigExport, ExecutionConfig,
        ExportMode, GeneratedDraft, GeneratedDraftStatus, HttpMethod, OperationId, OperationStatus,
        Protocol, RestTarget, RetryPolicy, Samples, SecretRef, Target, ToolDescription,
        ToolExample,
    };
    use mcpaas_mapping::{MappingRule, MappingSet};
    use mcpaas_schema::{Schema, SchemaKind};
    use serde_json::json;
    use sqlx::{Executor, PgPool, postgres::PgPoolOptions};

    use crate::{
        PostgresRegistry, RegistryError,
        model::{
            CreateVersionRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata,
            OperationSampleMetadata, PublishRequest, RegistryOperation, SampleKind,
            SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest,
            YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus,
        },
    };

    #[tokio::test]
    async fn stores_versions_and_published_operations() {
        let database = TestDatabase::new().await;
        let registry = database.registry().await;
        let operation_v1 = test_operation("op_rest_01", 1, OperationStatus::Draft);

        registry
            .create_operation(&operation_v1, Some("alice"))
            .await
            .unwrap();

        let operation_v2 = test_operation("op_rest_01", 2, OperationStatus::Draft);

        registry
            .create_version(CreateVersionRequest {
                snapshot: &operation_v2,
                change_note: Some("add output mapping"),
                created_by: Some("alice"),
            })
            .await
            .unwrap();

        registry
            .publish_operation(PublishRequest {
                operation_id: &operation_v2.id,
                version: operation_v2.version,
                published_at: "2026-03-25T12:10:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();

        let summary = registry
            .get_operation_summary(&operation_v2.id)
            .await
            .unwrap()
            .unwrap();
        let versions = registry
            .list_operation_versions(&operation_v2.id)
            .await
            .unwrap();
        let published = registry
            .get_published_operation(&operation_v2.id)
            .await
            .unwrap()
            .unwrap();

        assert_eq!(summary.current_draft_version, 2);
        assert_eq!(summary.latest_published_version, Some(2));
        assert_eq!(summary.status, OperationStatus::Published);
        assert_eq!(versions.len(), 2);
        assert_eq!(
            versions[1].change_note.as_deref(),
            Some("add output mapping")
        );
        assert_eq!(published.version, 2);
        assert!(published.is_published());

        database.cleanup().await;
    }

    #[tokio::test]
    async fn rejects_out_of_order_versions() {
        let database = TestDatabase::new().await;
        let registry = database.registry().await;
        let operation = test_operation("op_rest_02", 1, OperationStatus::Draft);

        registry.create_operation(&operation, None).await.unwrap();

        let invalid = test_operation("op_rest_02", 3, OperationStatus::Draft);
        let error = registry
            .create_version(CreateVersionRequest {
                snapshot: &invalid,
                change_note: None,
                created_by: None,
            })
            .await
            .unwrap_err();

        assert!(matches!(
            error,
            RegistryError::InvalidVersionSequence {
                expected: 2,
                actual: 3,
                ..
            }
        ));

        database.cleanup().await;
    }

    #[tokio::test]
    async fn stores_auth_profiles_and_artifact_metadata() {
        let database = TestDatabase::new().await;
        let registry = database.registry().await;
        let operation = test_operation("op_rest_03", 1, OperationStatus::Draft);

        registry.create_operation(&operation, None).await.unwrap();

        let auth_profile = AuthProfile {
            id: "auth_rmcp".into(),
            name: "RMCP API key".to_owned(),
            kind: AuthKind::ApiKeyHeader,
            config: AuthConfig::ApiKeyHeader(ApiKeyHeaderAuthConfig {
                header_name: "X-Api-Key".to_owned(),
                secret_ref: SecretRef::new("vault://rmcp/api-key"),
            }),
            created_at: "2026-03-25T12:00:00Z".to_owned(),
            updated_at: "2026-03-25T12:00:00Z".to_owned(),
        };

        registry
            .save_auth_profile(SaveAuthProfileRequest {
                profile: &auth_profile,
            })
            .await
            .unwrap();

        let input_sample = OperationSampleMetadata {
            id: "sample_input".into(),
            operation_id: operation.id.clone(),
            version: 1,
            sample_kind: SampleKind::InputJson,
            storage_ref: "file:///tmp/input.json".to_owned(),
            content_type: "application/json".to_owned(),
            file_name: Some("input.json".to_owned()),
            created_at: "2026-03-25T12:01:00Z".to_owned(),
        };
        let descriptor = DescriptorMetadata {
            id: "descriptor_01".into(),
            operation_id: Some(operation.id.clone()),
            version: Some(1),
            descriptor_kind: DescriptorKind::DescriptorSet,
            storage_ref: "file:///tmp/schema.desc".to_owned(),
            source_name: Some("schema.desc".to_owned()),
            package_index: Some(json!({ "crm.v1": ["LeadService"] })),
            created_at: "2026-03-25T12:02:00Z".to_owned(),
        };

        registry
            .save_sample_metadata(SaveSampleMetadataRequest {
                sample: &input_sample,
            })
            .await
            .unwrap();
        registry
            .save_descriptor_metadata(SaveDescriptorMetadataRequest {
                descriptor: &descriptor,
            })
            .await
            .unwrap();

        let auth_profiles = registry.list_auth_profiles().await.unwrap();
        let samples = registry
            .list_sample_metadata(&operation.id, 1)
            .await
            .unwrap();
        let descriptors = registry
            .list_descriptor_metadata(&operation.id, 1)
            .await
            .unwrap();

        assert_eq!(auth_profiles, vec![auth_profile]);
        assert_eq!(samples, vec![input_sample]);
        assert_eq!(descriptors, vec![descriptor]);

        database.cleanup().await;
    }

    #[tokio::test]
    async fn stores_and_finishes_yaml_import_jobs() {
        let database = TestDatabase::new().await;
        let registry = database.registry().await;
        let job_id = YamlImportJobId::new("job_yaml_01");
        let operation = test_operation("op_rest_04", 1, OperationStatus::Draft);

        registry.create_operation(&operation, None).await.unwrap();

        registry
            .create_yaml_import_job(CreateYamlImportJobRequest {
                id: &job_id,
                source_sample_id: None,
                format_version: "v1",
                mode: ExportMode::Portable,
                created_at: "2026-03-25T12:00:00Z",
            })
            .await
            .unwrap();

        registry
            .finish_yaml_import_job(
                &job_id,
                &YamlImportJobCompletion {
                    status: YamlImportJobStatus::Completed,
                    result_operation_id: Some(operation.id.clone()),
                    result_version: Some(2),
                    error_text: None,
                    finished_at: "2026-03-25T12:05:00Z".to_owned(),
                },
            )
            .await
            .unwrap();

        let job = registry
            .get_yaml_import_job(&job_id)
            .await
            .unwrap()
            .unwrap();

        assert_eq!(job.status, YamlImportJobStatus::Completed);
        assert_eq!(job.result_version, Some(2));
        assert_eq!(job.mode, ExportMode::Portable);

        database.cleanup().await;
    }

    struct TestDatabase {
        admin_pool: PgPool,
        database_url: String,
        schema: String,
    }

    impl TestDatabase {
        async fn new() -> Self {
            let database_url = env::var("TEST_DATABASE_URL")
                .expect("TEST_DATABASE_URL must point to a reachable PostgreSQL database");
            let admin_pool = PgPoolOptions::new()
                .max_connections(1)
                .connect(&database_url)
                .await
                .unwrap();
            let schema = format!(
                "test_registry_{}_{}",
                std::process::id(),
                SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap()
                    .as_nanos()
            );

            admin_pool
                .execute(sqlx::query(&format!("create schema {schema}")))
                .await
                .unwrap();

            Self {
                admin_pool,
                database_url,
                schema,
            }
        }

        async fn registry(&self) -> PostgresRegistry {
            PostgresRegistry::connect_in_schema(&self.database_url, Some(&self.schema))
                .await
                .unwrap()
        }

        async fn cleanup(&self) {
            self.admin_pool
                .execute(sqlx::query(&format!(
                    "drop schema if exists {} cascade",
                    self.schema
                )))
                .await
                .unwrap();
        }
    }

    fn test_operation(id: &str, version: u32, status: OperationStatus) -> RegistryOperation {
        RegistryOperation {
            id: OperationId::new(id),
            name: format!("{id}_tool"),
            display_name: format!("Display {id}"),
            protocol: Protocol::Rest,
            status,
            version,
            target: Target::Rest(RestTarget {
                base_url: "https://api.example.com".to_owned(),
                method: HttpMethod::Post,
                path_template: "/v1/leads".to_owned(),
                static_headers: BTreeMap::from([("X-Static".to_owned(), "true".to_owned())]),
            }),
            input_schema: Schema {
                kind: SchemaKind::Object,
                description: Some("input".to_owned()),
                required: true,
                nullable: false,
                default_value: None,
                fields: BTreeMap::from([(
                    "email".to_owned(),
                    Schema {
                        kind: SchemaKind::String,
                        description: None,
                        required: true,
                        nullable: false,
                        default_value: None,
                        fields: BTreeMap::new(),
                        items: None,
                        enum_values: Vec::new(),
                        variants: Vec::new(),
                    },
                )]),
                items: None,
                enum_values: Vec::new(),
                variants: Vec::new(),
            },
            output_schema: Schema {
                kind: SchemaKind::Object,
                description: Some("output".to_owned()),
                required: true,
                nullable: false,
                default_value: None,
                fields: BTreeMap::from([(
                    "id".to_owned(),
                    Schema {
                        kind: SchemaKind::String,
                        description: None,
                        required: true,
                        nullable: false,
                        default_value: None,
                        fields: BTreeMap::new(),
                        items: None,
                        enum_values: Vec::new(),
                        variants: Vec::new(),
                    },
                )]),
                items: None,
                enum_values: Vec::new(),
                variants: Vec::new(),
            },
            input_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.mcp.email".to_owned(),
                    target: "$.request.body.email".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            output_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.response.body.id".to_owned(),
                    target: "$.output.id".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            execution_config: ExecutionConfig {
                timeout_ms: 10_000,
                retry_policy: Some(RetryPolicy { max_attempts: 3 }),
                auth_profile_ref: Some("auth_rmcp".into()),
                headers: BTreeMap::from([("X-Request-Id".to_owned(), "static".to_owned())]),
                protocol_options: None,
            },
            tool_description: ToolDescription {
                title: "Create lead".to_owned(),
                description: "Creates CRM lead".to_owned(),
                tags: vec!["crm".to_owned()],
                examples: vec![ToolExample {
                    input: json!({ "email": "a@example.com" }),
                }],
            },
            samples: Some(Samples {
                input_json_sample_ref: Some("sample_input".into()),
                output_json_sample_ref: Some("sample_output".into()),
                proto_file_ref: None,
                descriptor_ref: None,
            }),
            generated_draft: Some(GeneratedDraft {
                status: GeneratedDraftStatus::Available,
                source_types: vec!["input_json".to_owned(), "output_json".to_owned()],
                generated_at: Some("2026-03-25T11:59:00Z".to_owned()),
                input_schema_generated: true,
                output_schema_generated: true,
                input_mapping_generated: true,
                output_mapping_generated: true,
                warnings: Vec::new(),
            }),
            config_export: Some(ConfigExport {
                format_version: "v1".to_owned(),
                export_mode: ExportMode::Portable,
            }),
            created_at: "2026-03-25T11:58:00Z".to_owned(),
            updated_at: format!("2026-03-25T12:{version:02}:00Z"),
            published_at: None,
        }
    }
}