//! Gateway webhook sink definitions and delivery log persistence.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sqlx::FromRow;
use sqlx::postgres::PgPool;

use crate::store::WebhookStoreError;

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GatewayWebhookSinkStateResolutionMode {
    Template,
    GatewayQueryBuilder,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GatewayWebhookSinkTargetWriteMode {
    Insert,
    Upsert { conflict_columns: Vec<String> },
}

#[derive(Debug, Clone, PartialEq)]
pub struct GatewayWebhookSinkResiliencePolicyDefinition {
    pub state_key_template: Option<String>,
    pub state_cap_max_fires: Option<i32>,
    pub state_cap_window_seconds: Option<i32>,
    pub idempotency_template: Option<String>,
    pub state_resolution_mode: GatewayWebhookSinkStateResolutionMode,
    pub state_resolution_query_builder: Value,
}

#[derive(Debug, Clone, PartialEq)]
pub struct GatewayWebhookSinkExecutionDefinition {
    pub target_client_name: String,
    pub target_table: String,
    pub target_write_mode: GatewayWebhookSinkTargetWriteMode,
    pub projection_template: Value,
    pub include_request_body_in_context: bool,
    pub lookup_keys: Value,
    pub resilience_policy: GatewayWebhookSinkResiliencePolicyDefinition,
}

#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct GatewayWebhookSinkRecord {
    pub id: i64,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub enabled: bool,
    pub name: String,
    pub description: Option<String>,
    pub athena_base_url: Option<String>,
    pub client_name: String,
    pub table_name: Option<String>,
    pub route_key: String,
    pub target_client_name: String,
    pub target_table: String,
    pub target_write_mode: String,
    pub target_conflict_columns: Value,
    pub projection_template: Value,
    pub include_request_body_in_context: bool,
    pub lookup_keys: Value,
    pub slug: Option<String>,
    pub state_key_template: Option<String>,
    pub state_cap_max_fires: Option<i32>,
    pub state_cap_window_seconds: Option<i32>,
    pub idempotency_template: Option<String>,
    pub state_resolution_mode: String,
    pub state_resolution_query_builder: Value,
}

impl GatewayWebhookSinkRecord {
    pub fn execution_definition(&self) -> Result<GatewayWebhookSinkExecutionDefinition, String> {
        Ok(GatewayWebhookSinkExecutionDefinition {
            target_client_name: self.target_client_name.clone(),
            target_table: self.target_table.clone(),
            target_write_mode: parse_sink_target_write_mode(
                &self.target_write_mode,
                &self.target_conflict_columns,
            )?,
            projection_template: self.projection_template.clone(),
            include_request_body_in_context: self.include_request_body_in_context,
            lookup_keys: self.lookup_keys.clone(),
            resilience_policy: GatewayWebhookSinkResiliencePolicyDefinition {
                state_key_template: self.state_key_template.clone(),
                state_cap_max_fires: self.state_cap_max_fires,
                state_cap_window_seconds: self.state_cap_window_seconds,
                idempotency_template: self.idempotency_template.clone(),
                state_resolution_mode: parse_sink_state_resolution_mode(
                    &self.state_resolution_mode,
                )?,
                state_resolution_query_builder: self.state_resolution_query_builder.clone(),
            },
        })
    }
}

fn parse_sink_state_resolution_mode(
    mode: &str,
) -> Result<GatewayWebhookSinkStateResolutionMode, String> {
    match mode.trim().to_ascii_lowercase().as_str() {
        "" | "template" => Ok(GatewayWebhookSinkStateResolutionMode::Template),
        "gateway_query_builder" => Ok(GatewayWebhookSinkStateResolutionMode::GatewayQueryBuilder),
        _ => Err("invalid sink state_resolution_mode".to_string()),
    }
}

fn parse_sink_target_write_mode(
    target_write_mode: &str,
    target_conflict_columns: &Value,
) -> Result<GatewayWebhookSinkTargetWriteMode, String> {
    let conflict_columns = target_conflict_columns
        .as_array()
        .ok_or_else(|| "sink target_conflict_columns must be a JSON array".to_string())?
        .iter()
        .map(|value| {
            value
                .as_str()
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToString::to_string)
                .ok_or_else(|| "sink target_conflict_columns entries must be strings".to_string())
        })
        .collect::<Result<Vec<_>, _>>()?;

    match target_write_mode.trim().to_ascii_lowercase().as_str() {
        "" | "insert" => Ok(GatewayWebhookSinkTargetWriteMode::Insert),
        "upsert" => {
            if conflict_columns.is_empty() {
                Err("sink target_conflict_columns is required for upsert mode".to_string())
            } else {
                Ok(GatewayWebhookSinkTargetWriteMode::Upsert { conflict_columns })
            }
        }
        _ => Err("invalid sink target_write_mode".to_string()),
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct GatewayWebhookSinkDeliveryRecord {
    pub id: i64,
    pub created_at: DateTime<Utc>,
    pub sink_id: i64,
    pub trigger_route_key: String,
    pub request_id: Option<String>,
    pub idempotency_key: String,
    pub status: String,
    pub inserted_row_count: Option<i32>,
    pub error_message: Option<String>,
    pub duration_ms: Option<i64>,
    pub target_client_name: String,
    pub target_table: String,
    pub projected_payload: Option<Value>,
    pub context_snapshot: Value,
}

#[derive(Debug, Clone)]
pub struct CreateGatewayWebhookSinkParams {
    pub name: String,
    pub description: Option<String>,
    pub enabled: bool,
    pub athena_base_url: Option<String>,
    pub client_name: String,
    pub table_name: Option<String>,
    pub route_key: String,
    pub target_client_name: String,
    pub target_table: String,
    pub target_write_mode: Option<String>,
    pub target_conflict_columns: Option<Value>,
    pub projection_template: Value,
    pub include_request_body_in_context: bool,
    pub lookup_keys: Value,
    pub slug: Option<String>,
    pub state_key_template: Option<String>,
    pub state_cap_max_fires: Option<i32>,
    pub state_cap_window_seconds: Option<i32>,
    pub idempotency_template: Option<String>,
    pub state_resolution_mode: Option<String>,
    pub state_resolution_query_builder: Option<Value>,
}

#[derive(Debug, Clone, Default)]
pub struct PatchGatewayWebhookSinkParams {
    pub name: Option<String>,
    pub description: Option<Option<String>>,
    pub enabled: Option<bool>,
    pub athena_base_url: Option<Option<String>>,
    pub client_name: Option<String>,
    pub table_name: Option<Option<String>>,
    pub route_key: Option<String>,
    pub target_client_name: Option<String>,
    pub target_table: Option<String>,
    pub target_write_mode: Option<String>,
    pub target_conflict_columns: Option<Value>,
    pub projection_template: Option<Value>,
    pub include_request_body_in_context: Option<bool>,
    pub lookup_keys: Option<Value>,
    pub slug: Option<Option<String>>,
    pub state_key_template: Option<Option<String>>,
    pub state_cap_max_fires: Option<Option<i32>>,
    pub state_cap_window_seconds: Option<Option<i32>>,
    pub idempotency_template: Option<Option<String>>,
    pub state_resolution_mode: Option<String>,
    pub state_resolution_query_builder: Option<Value>,
}

pub async fn list_webhook_sinks_for_dispatch(
    pool: &PgPool,
    client_name: &str,
    route_key: &str,
    table_from_event: Option<&str>,
    instance_base_url: Option<&str>,
) -> Result<Vec<GatewayWebhookSinkRecord>, WebhookStoreError> {
    let rows = sqlx::query_as::<_, GatewayWebhookSinkRecord>(
        r#"
        SELECT
            id, created_at, updated_at, enabled, name, description, athena_base_url,
            client_name, table_name, route_key, target_client_name, target_table,
            target_write_mode, target_conflict_columns,
            projection_template, include_request_body_in_context, lookup_keys, slug,
            state_key_template, state_cap_max_fires, state_cap_window_seconds,
            idempotency_template, state_resolution_mode, state_resolution_query_builder
        FROM athena.gateway_webhook_sink
        WHERE enabled = true
          AND client_name = $1
          AND route_key = $2
          AND (
            table_name IS NULL
            OR ($3::text IS NOT NULL AND table_name = $3)
          )
          AND (
            athena_base_url IS NULL
            OR (
                $4::text IS NOT NULL
                AND lower(rtrim(athena_base_url, '/')) = lower(rtrim($4::text, '/'))
            )
          )
        ORDER BY id ASC
        "#,
    )
    .bind(client_name)
    .bind(route_key)
    .bind(table_from_event)
    .bind(instance_base_url)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

pub async fn list_gateway_webhook_sinks_filtered(
    pool: &PgPool,
    client_name: Option<&str>,
    limit: i64,
    offset: i64,
) -> Result<Vec<GatewayWebhookSinkRecord>, WebhookStoreError> {
    let rows = if let Some(cn) = client_name.filter(|s| !s.is_empty()) {
        sqlx::query_as::<_, GatewayWebhookSinkRecord>(
            r#"
            SELECT
                id, created_at, updated_at, enabled, name, description, athena_base_url,
                client_name, table_name, route_key, target_client_name, target_table,
                target_write_mode, target_conflict_columns,
                projection_template, include_request_body_in_context, lookup_keys, slug,
                state_key_template, state_cap_max_fires, state_cap_window_seconds,
                idempotency_template, state_resolution_mode, state_resolution_query_builder
            FROM athena.gateway_webhook_sink
            WHERE client_name = $1
            ORDER BY id DESC
            LIMIT $2 OFFSET $3
            "#,
        )
        .bind(cn)
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await?
    } else {
        sqlx::query_as::<_, GatewayWebhookSinkRecord>(
            r#"
            SELECT
                id, created_at, updated_at, enabled, name, description, athena_base_url,
                client_name, table_name, route_key, target_client_name, target_table,
                target_write_mode, target_conflict_columns,
                projection_template, include_request_body_in_context, lookup_keys, slug,
                state_key_template, state_cap_max_fires, state_cap_window_seconds,
                idempotency_template, state_resolution_mode, state_resolution_query_builder
            FROM athena.gateway_webhook_sink
            ORDER BY id DESC
            LIMIT $1 OFFSET $2
            "#,
        )
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await?
    };
    Ok(rows)
}

pub async fn count_gateway_webhook_sinks_filtered(
    pool: &PgPool,
    client_name: Option<&str>,
) -> Result<i64, WebhookStoreError> {
    let c: (i64,) = if let Some(cn) = client_name.filter(|s| !s.is_empty()) {
        sqlx::query_as(
            "SELECT COUNT(*)::bigint FROM athena.gateway_webhook_sink WHERE client_name = $1",
        )
        .bind(cn)
        .fetch_optional(pool)
        .await?
    } else {
        sqlx::query_as("SELECT COUNT(*)::bigint FROM athena.gateway_webhook_sink")
            .fetch_optional(pool)
            .await?
    }
    .unwrap_or((0,));
    Ok(c.0)
}

pub async fn get_gateway_webhook_sink(
    pool: &PgPool,
    id: i64,
) -> Result<GatewayWebhookSinkRecord, WebhookStoreError> {
    sqlx::query_as::<_, GatewayWebhookSinkRecord>(
        r#"
        SELECT
            id, created_at, updated_at, enabled, name, description, athena_base_url,
            client_name, table_name, route_key, target_client_name, target_table,
            target_write_mode, target_conflict_columns,
            projection_template, include_request_body_in_context, lookup_keys, slug,
            state_key_template, state_cap_max_fires, state_cap_window_seconds,
            idempotency_template, state_resolution_mode, state_resolution_query_builder
        FROM athena.gateway_webhook_sink
        WHERE id = $1
        "#,
    )
    .bind(id)
    .fetch_optional(pool)
    .await?
    .ok_or(WebhookStoreError::NotFound)
}

pub async fn upsert_gateway_webhook_sink(
    pool: &PgPool,
    params: CreateGatewayWebhookSinkParams,
) -> Result<GatewayWebhookSinkRecord, WebhookStoreError> {
    let record = sqlx::query_as::<_, GatewayWebhookSinkRecord>(
        r#"
        INSERT INTO athena.gateway_webhook_sink (
            enabled, name, description, athena_base_url, client_name, table_name, route_key,
            target_client_name, target_table, target_write_mode, target_conflict_columns, projection_template,
            include_request_body_in_context, lookup_keys, slug,
            state_key_template, state_cap_max_fires, state_cap_window_seconds,
            idempotency_template, state_resolution_mode, state_resolution_query_builder
        )
        VALUES (
            $1, $2, $3, $4, $5, $6, $7,
            $8, $9, $10, $11, $12,
            $13, $14, $15,
            $16, $17, $18,
            $19, $20, $21
        )
        ON CONFLICT (
            client_name, route_key, target_client_name, target_table,
            athena_base_url_norm, table_name_norm
        )
        DO UPDATE SET
            updated_at = now(),
            enabled = EXCLUDED.enabled,
            name = EXCLUDED.name,
            description = EXCLUDED.description,
            athena_base_url = EXCLUDED.athena_base_url,
            table_name = EXCLUDED.table_name,
            target_write_mode = EXCLUDED.target_write_mode,
            target_conflict_columns = EXCLUDED.target_conflict_columns,
            projection_template = EXCLUDED.projection_template,
            include_request_body_in_context = EXCLUDED.include_request_body_in_context,
            lookup_keys = EXCLUDED.lookup_keys,
            slug = COALESCE(EXCLUDED.slug, gateway_webhook_sink.slug),
            state_key_template = EXCLUDED.state_key_template,
            state_cap_max_fires = EXCLUDED.state_cap_max_fires,
            state_cap_window_seconds = EXCLUDED.state_cap_window_seconds,
            idempotency_template = EXCLUDED.idempotency_template,
            state_resolution_mode = EXCLUDED.state_resolution_mode,
            state_resolution_query_builder = EXCLUDED.state_resolution_query_builder
        RETURNING
            id, created_at, updated_at, enabled, name, description, athena_base_url,
            client_name, table_name, route_key, target_client_name, target_table,
            target_write_mode, target_conflict_columns,
            projection_template, include_request_body_in_context, lookup_keys, slug,
            state_key_template, state_cap_max_fires, state_cap_window_seconds,
            idempotency_template, state_resolution_mode, state_resolution_query_builder
        "#,
    )
    .bind(params.enabled)
    .bind(&params.name)
    .bind(&params.description)
    .bind(&params.athena_base_url)
    .bind(&params.client_name)
    .bind(&params.table_name)
    .bind(&params.route_key)
    .bind(&params.target_client_name)
    .bind(&params.target_table)
    .bind(
        params
            .target_write_mode
            .unwrap_or_else(|| "insert".to_string()),
    )
    .bind(params.target_conflict_columns.unwrap_or_else(|| json!([])))
    .bind(&params.projection_template)
    .bind(params.include_request_body_in_context)
    .bind(&params.lookup_keys)
    .bind(&params.slug)
    .bind(&params.state_key_template)
    .bind(params.state_cap_max_fires)
    .bind(params.state_cap_window_seconds)
    .bind(&params.idempotency_template)
    .bind(
        params
            .state_resolution_mode
            .unwrap_or_else(|| "template".to_string()),
    )
    .bind(
        params
            .state_resolution_query_builder
            .unwrap_or_else(|| json!({})),
    )
    .fetch_one(pool)
    .await?;
    Ok(record)
}

pub async fn patch_gateway_webhook_sink(
    pool: &PgPool,
    id: i64,
    patch: PatchGatewayWebhookSinkParams,
) -> Result<GatewayWebhookSinkRecord, WebhookStoreError> {
    let existing = get_gateway_webhook_sink(pool, id).await?;
    let name = patch.name.unwrap_or(existing.name);
    let description = patch.description.unwrap_or(existing.description);
    let enabled = patch.enabled.unwrap_or(existing.enabled);
    let athena_base_url = patch.athena_base_url.unwrap_or(existing.athena_base_url);
    let client_name = patch.client_name.unwrap_or(existing.client_name);
    let table_name = patch.table_name.unwrap_or(existing.table_name);
    let route_key = patch.route_key.unwrap_or(existing.route_key);
    let target_client_name = patch
        .target_client_name
        .unwrap_or(existing.target_client_name);
    let target_table = patch.target_table.unwrap_or(existing.target_table);
    let target_write_mode = patch
        .target_write_mode
        .unwrap_or(existing.target_write_mode);
    let target_conflict_columns = patch
        .target_conflict_columns
        .unwrap_or(existing.target_conflict_columns);
    let projection_template = patch
        .projection_template
        .unwrap_or(existing.projection_template);
    let include_request_body_in_context = patch
        .include_request_body_in_context
        .unwrap_or(existing.include_request_body_in_context);
    let lookup_keys = patch.lookup_keys.unwrap_or(existing.lookup_keys);
    let slug = patch.slug.unwrap_or(existing.slug);
    let state_key_template = patch
        .state_key_template
        .unwrap_or(existing.state_key_template);
    let state_cap_max_fires = patch
        .state_cap_max_fires
        .unwrap_or(existing.state_cap_max_fires);
    let state_cap_window_seconds = patch
        .state_cap_window_seconds
        .unwrap_or(existing.state_cap_window_seconds);
    let idempotency_template = patch
        .idempotency_template
        .unwrap_or(existing.idempotency_template);
    let state_resolution_mode = patch
        .state_resolution_mode
        .unwrap_or(existing.state_resolution_mode);
    let state_resolution_query_builder = patch
        .state_resolution_query_builder
        .unwrap_or(existing.state_resolution_query_builder);

    sqlx::query(
        r#"
        UPDATE athena.gateway_webhook_sink SET
            updated_at = now(),
            name = $1,
            description = $2,
            enabled = $3,
            athena_base_url = $4,
            client_name = $5,
            table_name = $6,
            route_key = $7,
            target_client_name = $8,
            target_table = $9,
            target_write_mode = $10,
            target_conflict_columns = $11,
            projection_template = $12,
            include_request_body_in_context = $13,
            lookup_keys = $14,
            slug = $15,
            state_key_template = $16,
            state_cap_max_fires = $17,
            state_cap_window_seconds = $18,
            idempotency_template = $19,
            state_resolution_mode = $20,
            state_resolution_query_builder = $21
        WHERE id = $22
        "#,
    )
    .bind(&name)
    .bind(&description)
    .bind(enabled)
    .bind(&athena_base_url)
    .bind(&client_name)
    .bind(&table_name)
    .bind(&route_key)
    .bind(&target_client_name)
    .bind(&target_table)
    .bind(&target_write_mode)
    .bind(&target_conflict_columns)
    .bind(&projection_template)
    .bind(include_request_body_in_context)
    .bind(&lookup_keys)
    .bind(&slug)
    .bind(&state_key_template)
    .bind(state_cap_max_fires)
    .bind(state_cap_window_seconds)
    .bind(&idempotency_template)
    .bind(&state_resolution_mode)
    .bind(&state_resolution_query_builder)
    .bind(id)
    .execute(pool)
    .await?;

    get_gateway_webhook_sink(pool, id).await
}

pub async fn delete_gateway_webhook_sink(
    pool: &PgPool,
    id: i64,
) -> Result<bool, WebhookStoreError> {
    let result = sqlx::query("DELETE FROM athena.gateway_webhook_sink WHERE id = $1")
        .bind(id)
        .execute(pool)
        .await?;
    Ok(result.rows_affected() > 0)
}

pub async fn gate_webhook_sink_delivery(
    pool: &PgPool,
    sink_id: i64,
    trigger_route_key: &str,
    request_id: Option<&str>,
    idempotency_key: &str,
    context_snapshot: Value,
    target_client_name: &str,
    target_table: &str,
    projected_payload: Option<&Value>,
    state_key_hash: Option<&str>,
    state_key_raw: Option<&str>,
    state_cap_max_fires: Option<i32>,
    state_cap_window_seconds: Option<i32>,
) -> Result<i64, WebhookStoreError> {
    let mut tx = pool.begin().await?;

    if let (Some(_), Some(_), Some(key_hash)) = (
        state_cap_max_fires,
        state_cap_window_seconds,
        state_key_hash,
    ) {
        sqlx::query(r#"SELECT pg_advisory_xact_lock(hashtext($1::text)::bigint)"#)
            .bind(format!("sink|{}|{}", sink_id, key_hash))
            .execute(&mut *tx)
            .await?;
    }

    let inserted: Option<(i64,)> = sqlx::query_as(
        r#"
        INSERT INTO athena.gateway_webhook_sink_delivery (
            sink_id, trigger_route_key, request_id, idempotency_key, status,
            target_client_name, target_table, projected_payload, context_snapshot
        )
        VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7, $8)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id
        "#,
    )
    .bind(sink_id)
    .bind(trigger_route_key)
    .bind(request_id)
    .bind(idempotency_key)
    .bind(target_client_name)
    .bind(target_table)
    .bind(projected_payload)
    .bind(&context_snapshot)
    .fetch_optional(&mut *tx)
    .await?;

    let delivery_id = match inserted {
        Some((id,)) => id,
        None => {
            tx.commit().await?;
            return Err(WebhookStoreError::Database(sqlx::Error::Protocol(
                "duplicate idempotency key".to_string(),
            )));
        }
    };

    if let (Some(max_fires), Some(window_seconds), Some(key_hash)) = (
        state_cap_max_fires,
        state_cap_window_seconds,
        state_key_hash,
    ) {
        let observed: (i64,) = sqlx::query_as(
            r#"
            SELECT COUNT(*)::bigint
            FROM athena.gateway_webhook_sink_state_ledger
            WHERE sink_id = $1
              AND state_key_hash = $2
              AND created_at >= (now() - ($3::bigint * interval '1 second'))
            "#,
        )
        .bind(sink_id)
        .bind(key_hash)
        .bind(window_seconds as i64)
        .fetch_one(&mut *tx)
        .await?;

        if observed.0 >= max_fires as i64 {
            sqlx::query(
                r#"
                UPDATE athena.gateway_webhook_sink_delivery
                SET status = 'failed', error_message = $1
                WHERE id = $2
                "#,
            )
            .bind("OVER_LIMIT")
            .bind(delivery_id)
            .execute(&mut *tx)
            .await?;
            tx.commit().await?;
            return Ok(delivery_id);
        }

        sqlx::query(
            r#"
            INSERT INTO athena.gateway_webhook_sink_state_ledger (
                sink_id, delivery_id, state_key_hash, state_key_raw, window_started_at
            )
            VALUES ($1, $2, $3, $4, (now() - ($5::bigint * interval '1 second')))
            "#,
        )
        .bind(sink_id)
        .bind(delivery_id)
        .bind(key_hash)
        .bind(state_key_raw)
        .bind(window_seconds as i64)
        .execute(&mut *tx)
        .await?;
    }

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

pub async fn update_webhook_sink_delivery_outcome(
    pool: &PgPool,
    delivery_id: i64,
    status: &str,
    inserted_row_count: Option<i32>,
    error_message: Option<&str>,
    duration_ms: Option<i64>,
) -> Result<(), WebhookStoreError> {
    sqlx::query(
        r#"
        UPDATE athena.gateway_webhook_sink_delivery SET
            status = $1,
            inserted_row_count = $2,
            error_message = $3,
            duration_ms = $4
        WHERE id = $5
        "#,
    )
    .bind(status)
    .bind(inserted_row_count)
    .bind(error_message)
    .bind(duration_ms)
    .bind(delivery_id)
    .execute(pool)
    .await?;
    Ok(())
}

pub async fn insert_webhook_sink_delivery_terminal_failure(
    pool: &PgPool,
    sink_id: i64,
    trigger_route_key: &str,
    request_id: Option<&str>,
    idempotency_key: &str,
    target_client_name: &str,
    target_table: &str,
    projected_payload: Option<&Value>,
    context_snapshot: Value,
    error_code: &str,
    detail: Option<&str>,
) -> Result<Option<i64>, WebhookStoreError> {
    const DETAIL_MAX: usize = 2000;
    let error_message = match detail {
        Some(d) if !d.trim().is_empty() => {
            let trimmed = d.trim();
            let suffix = if trimmed.len() > DETAIL_MAX {
                format!("{}...", &trimmed[..DETAIL_MAX])
            } else {
                trimmed.to_string()
            };
            format!("{}: {}", error_code, suffix)
        }
        _ => error_code.to_string(),
    };

    let row: Option<(i64,)> = sqlx::query_as(
        r#"
        INSERT INTO athena.gateway_webhook_sink_delivery (
            sink_id, trigger_route_key, request_id, idempotency_key, status,
            target_client_name, target_table, projected_payload, context_snapshot, error_message
        )
        VALUES ($1, $2, $3, $4, 'failed', $5, $6, $7, $8, $9)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id
        "#,
    )
    .bind(sink_id)
    .bind(trigger_route_key)
    .bind(request_id)
    .bind(idempotency_key)
    .bind(target_client_name)
    .bind(target_table)
    .bind(projected_payload)
    .bind(&context_snapshot)
    .bind(&error_message)
    .fetch_optional(pool)
    .await?;
    Ok(row.map(|tuple| tuple.0))
}

pub async fn list_webhook_sink_deliveries(
    pool: &PgPool,
    sink_id: i64,
    limit: i64,
    offset: i64,
) -> Result<Vec<GatewayWebhookSinkDeliveryRecord>, WebhookStoreError> {
    let rows = sqlx::query_as::<_, GatewayWebhookSinkDeliveryRecord>(
        r#"
        SELECT
            id, created_at, sink_id, trigger_route_key, request_id, idempotency_key,
            status, inserted_row_count, error_message, duration_ms,
            target_client_name, target_table, projected_payload, context_snapshot
        FROM athena.gateway_webhook_sink_delivery
        WHERE sink_id = $1
        ORDER BY id DESC
        LIMIT $2 OFFSET $3
        "#,
    )
    .bind(sink_id)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

pub async fn count_webhook_sink_deliveries(
    pool: &PgPool,
    sink_id: i64,
) -> Result<i64, WebhookStoreError> {
    let c: (i64,) = sqlx::query_as(
        "SELECT COUNT(*)::bigint FROM athena.gateway_webhook_sink_delivery WHERE sink_id = $1",
    )
    .bind(sink_id)
    .fetch_optional(pool)
    .await?
    .unwrap_or((0,));
    Ok(c.0)
}

#[cfg(test)]
mod tests {
    use super::{
        GatewayWebhookSinkRecord, GatewayWebhookSinkStateResolutionMode,
        GatewayWebhookSinkTargetWriteMode,
    };
    use chrono::Utc;
    use serde_json::json;

    fn sample_sink_record() -> GatewayWebhookSinkRecord {
        GatewayWebhookSinkRecord {
            id: 1,
            created_at: Utc::now(),
            updated_at: Utc::now(),
            enabled: true,
            name: "billing payments".to_string(),
            description: Some("project Mollie events".to_string()),
            athena_base_url: None,
            client_name: "athena".to_string(),
            table_name: Some("billing_payment".to_string()),
            route_key: "billing_payment_upserted".to_string(),
            target_client_name: "billing".to_string(),
            target_table: "billing.billing_payments".to_string(),
            target_write_mode: "upsert".to_string(),
            target_conflict_columns: json!(["provider", "provider_payment_id"]),
            projection_template: json!({
                "provider": "{{payload.provider}}",
                "provider_payment_id": "{{payload.paymentId}}"
            }),
            include_request_body_in_context: true,
            lookup_keys: json!({
                "tenantId": "payload.tenant_id"
            }),
            slug: Some("billing-payments".to_string()),
            state_key_template: Some("payment:{{payload.paymentId}}".to_string()),
            state_cap_max_fires: Some(1),
            state_cap_window_seconds: Some(3600),
            idempotency_template: Some("{{payload.paymentId}}".to_string()),
            state_resolution_mode: "gateway_query_builder".to_string(),
            state_resolution_query_builder: json!({
                "table": "billing.billing_payments",
                "select_column": "id",
                "filters": [{
                    "column": "provider_payment_id",
                    "source": "payload.paymentId"
                }]
            }),
        }
    }

    #[test]
    fn execution_definition_parses_runtime_contract() {
        let record = sample_sink_record();
        let definition = record.execution_definition().expect("execution definition");

        assert_eq!(definition.target_client_name, "billing");
        assert_eq!(definition.target_table, "billing.billing_payments");
        assert_eq!(
            definition.target_write_mode,
            GatewayWebhookSinkTargetWriteMode::Upsert {
                conflict_columns: vec!["provider".to_string(), "provider_payment_id".to_string(),],
            }
        );
        assert!(definition.include_request_body_in_context);
        assert_eq!(
            definition.resilience_policy.state_resolution_mode,
            GatewayWebhookSinkStateResolutionMode::GatewayQueryBuilder
        );
        assert_eq!(
            definition.resilience_policy.state_resolution_query_builder["select_column"],
            json!("id")
        );
    }

    #[test]
    fn execution_definition_rejects_invalid_write_mode() {
        let mut record = sample_sink_record();
        record.target_write_mode = "merge".to_string();

        let error = record
            .execution_definition()
            .expect_err("invalid write mode should fail");
        assert_eq!(error, "invalid sink target_write_mode");
    }
}
