use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

use crate::storage::BillingSqlDocumentTable;

pub const DEFAULT_BILLING_SINK_SOURCE_CLIENT_NAME: &str = "athena-main";
pub const DEFAULT_BILLING_SINK_TARGET_CLIENT_NAME: &str = "athena-replica";
pub const DEFAULT_BILLING_SINK_TARGET_SCHEMA: &str = "billing";

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BillingSinkResiliencePolicy {
    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, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BillingSinkBlueprint {
    pub slug: String,
    pub document_kind: String,
    pub name: String,
    pub description: String,
    pub table_name: String,
    pub default_target_schema: String,
    pub target_table_basename: String,
    pub target_table: String,
    pub lookup_keys: Value,
    pub resilience_policy: BillingSinkResiliencePolicy,
    pub target_write_mode: String,
    pub target_conflict_columns: Value,
    pub insert_mode: String,
    pub projection_template: Value,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BillingSinkHelperDefinition {
    pub slug: String,
    pub document_kind: String,
    pub name: String,
    pub description: String,
    pub target_schema: String,
    pub target_table: String,
    pub lookup_keys: Value,
    pub resilience_policy: BillingSinkResiliencePolicy,
    pub target_write_mode: String,
    pub target_conflict_columns: Value,
    pub insert_mode: String,
    pub create_body: Value,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BillingSinkCreateParams {
    pub name: String,
    pub description: Option<String>,
    pub enabled: bool,
    pub athena_base_url: Option<String>,
    pub client_name: String,
    pub table_name: String,
    pub route_key: String,
    pub target_client_name: String,
    pub target_schema: 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: 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, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BillingSinkResolvedDefinition {
    pub blueprint: BillingSinkBlueprint,
    pub helper: BillingSinkHelperDefinition,
    pub create_params: BillingSinkCreateParams,
}

pub fn is_valid_sql_identifier(value: &str) -> bool {
    let mut chars = value.chars();
    match chars.next() {
        Some(first) if first == '_' || first.is_ascii_alphabetic() => {}
        _ => return false,
    }
    chars.all(|ch| ch == '_' || ch.is_ascii_alphanumeric())
}

pub fn billing_sink_target_table(table: BillingSqlDocumentTable, target_schema: &str) -> String {
    table.qualified_table_name(target_schema)
}

fn canonical_billing_sink_lookup_keys() -> Value {
    json!({
        "provider_ref": "response:upsert.provider_ref",
        "table": "response:upsert.table"
    })
}

fn canonical_billing_sink_resilience_policy() -> BillingSinkResiliencePolicy {
    BillingSinkResiliencePolicy {
        state_key_template: Some(
            "{{ response.upsert.provider_ref }}|{{ response.sql_write.table }}".to_string(),
        ),
        state_cap_max_fires: None,
        state_cap_window_seconds: None,
        idempotency_template: None,
        state_resolution_mode: Some("template".to_string()),
        state_resolution_query_builder: Some(json!({})),
    }
}

fn canonical_payment_projection_template() -> Value {
    json!({
        "provider": "{{ response.sql_write.row.provider }}",
        "provider_payment_id": "{{ response.sql_write.row.provider_payment_id }}",
        "provider_customer_id": "{{ response.sql_write.row.provider_customer_id }}",
        "provider_profile_id": "{{ response.sql_write.row.provider_profile_id }}",
        "provider_subscription_id": "{{ response.sql_write.row.provider_subscription_id }}",
        "provider_payment_link_id": "{{ response.sql_write.row.provider_payment_link_id }}",
        "status": "{{ response.sql_write.row.status }}",
        "amount_currency": "{{ response.sql_write.row.amount_currency }}",
        "amount_value": "{{ response.sql_write.row.amount_value }}",
        "amount_refunded_currency": "{{ response.sql_write.row.amount_refunded_currency }}",
        "amount_refunded_value": "{{ response.sql_write.row.amount_refunded_value }}",
        "description": "{{ response.sql_write.row.description }}",
        "metadata": "{{ response.sql_write.row.metadata }}",
        "raw": "{{ response.sql_write.row.raw }}",
        "paid_at": "{{ response.sql_write.row.paid_at }}",
        "created_at": "{{ response.sql_write.row.created_at }}"
    })
}

fn canonical_subscription_projection_template() -> Value {
    json!({
        "provider": "{{ response.sql_write.row.provider }}",
        "provider_subscription_id": "{{ response.sql_write.row.provider_subscription_id }}",
        "provider_customer_id": "{{ response.sql_write.row.provider_customer_id }}",
        "provider_profile_id": "{{ response.sql_write.row.provider_profile_id }}",
        "status": "{{ response.sql_write.row.status }}",
        "amount_currency": "{{ response.sql_write.row.amount_currency }}",
        "amount_value": "{{ response.sql_write.row.amount_value }}",
        "interval": "{{ response.sql_write.row.interval }}",
        "description": "{{ response.sql_write.row.description }}",
        "metadata": "{{ response.sql_write.row.metadata }}",
        "raw": "{{ response.sql_write.row.raw }}",
        "next_payment_date": "{{ response.sql_write.row.next_payment_date }}",
        "created_at": "{{ response.sql_write.row.created_at }}",
        "canceled_at": "{{ response.sql_write.row.canceled_at }}"
    })
}

fn canonical_invoice_projection_template() -> Value {
    json!({
        "provider": "{{ response.sql_write.row.provider }}",
        "provider_invoice_id": "{{ response.sql_write.row.provider_invoice_id }}",
        "provider_profile_id": "{{ response.sql_write.row.provider_profile_id }}",
        "provider_customer_id": "{{ response.sql_write.row.provider_customer_id }}",
        "status": "{{ response.sql_write.row.status }}",
        "amount_currency": "{{ response.sql_write.row.amount_currency }}",
        "amount_value": "{{ response.sql_write.row.amount_value }}",
        "amount_paid_currency": "{{ response.sql_write.row.amount_paid_currency }}",
        "amount_paid_value": "{{ response.sql_write.row.amount_paid_value }}",
        "description": "{{ response.sql_write.row.description }}",
        "metadata": "{{ response.sql_write.row.metadata }}",
        "raw": "{{ response.sql_write.row.raw }}",
        "issued_at": "{{ response.sql_write.row.issued_at }}",
        "paid_at": "{{ response.sql_write.row.paid_at }}",
        "created_at": "{{ response.sql_write.row.created_at }}"
    })
}

fn conflict_columns_for_table(table: BillingSqlDocumentTable) -> Value {
    match table {
        BillingSqlDocumentTable::Payments => json!(["provider", "provider_payment_id"]),
        BillingSqlDocumentTable::Subscriptions => {
            json!(["provider", "provider_subscription_id"])
        }
        BillingSqlDocumentTable::Invoices => json!(["provider", "provider_invoice_id"]),
    }
}

fn blueprint_for_table(
    slug: &str,
    document_kind: &str,
    name: &str,
    table: BillingSqlDocumentTable,
    projection_template: Value,
) -> BillingSinkBlueprint {
    let table_name = table.table_name().to_string();
    BillingSinkBlueprint {
        slug: slug.to_string(),
        document_kind: document_kind.to_string(),
        name: name.to_string(),
        description: format!(
            "Mirror canonical billing {} upserts into {} on another Athena client.",
            table_name
                .trim_start_matches("billing.billing_")
                .trim_end_matches('s'),
            table_name
        ),
        table_name: table_name.clone(),
        default_target_schema: DEFAULT_BILLING_SINK_TARGET_SCHEMA.to_string(),
        target_table_basename: table.table_basename().to_string(),
        target_table: billing_sink_target_table(table, DEFAULT_BILLING_SINK_TARGET_SCHEMA),
        lookup_keys: canonical_billing_sink_lookup_keys(),
        resilience_policy: canonical_billing_sink_resilience_policy(),
        target_write_mode: "upsert".to_string(),
        target_conflict_columns: conflict_columns_for_table(table),
        insert_mode: "upsert".to_string(),
        projection_template,
    }
}

pub fn billing_sink_blueprints() -> Vec<BillingSinkBlueprint> {
    vec![
        blueprint_for_table(
            "billing-payments-canonical-mirror",
            "payment",
            "Billing payments canonical mirror",
            BillingSqlDocumentTable::Payments,
            canonical_payment_projection_template(),
        ),
        blueprint_for_table(
            "billing-subscriptions-canonical-mirror",
            "subscription",
            "Billing subscriptions canonical mirror",
            BillingSqlDocumentTable::Subscriptions,
            canonical_subscription_projection_template(),
        ),
        blueprint_for_table(
            "billing-invoices-canonical-mirror",
            "invoice",
            "Billing invoices canonical mirror",
            BillingSqlDocumentTable::Invoices,
            canonical_invoice_projection_template(),
        ),
    ]
}

pub fn billing_sink_helpers(
    client_name: &str,
    target_client_name: &str,
    route_key: &str,
    target_schema: &str,
    athena_base_url: Option<&str>,
) -> Vec<BillingSinkHelperDefinition> {
    billing_sink_blueprints()
        .into_iter()
        .map(|blueprint| {
            let target_table = format!("{target_schema}.{}", blueprint.target_table_basename);
            let athena_base_url = athena_base_url
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToString::to_string);
            BillingSinkHelperDefinition {
                slug: blueprint.slug.clone(),
                document_kind: blueprint.document_kind.clone(),
                name: blueprint.name.clone(),
                description: blueprint.description.clone(),
                target_schema: target_schema.to_string(),
                target_table: target_table.clone(),
                lookup_keys: blueprint.lookup_keys.clone(),
                resilience_policy: blueprint.resilience_policy.clone(),
                target_write_mode: blueprint.target_write_mode.clone(),
                target_conflict_columns: blueprint.target_conflict_columns.clone(),
                insert_mode: blueprint.insert_mode.clone(),
                create_body: json!({
                "name": blueprint.name,
                "description": format!("Mirrors {} emitted by provider webhook ingestion.", blueprint.table_name.trim_start_matches("billing.").replace('_', " ")),
                "enabled": true,
                "athenaBaseUrl": athena_base_url,
                "clientName": client_name,
                "tableName": blueprint.table_name,
                "routeKey": route_key,
                "targetClientName": target_client_name,
                "targetSchema": target_schema,
                "targetTable": target_table,
                "targetWriteMode": blueprint.target_write_mode,
                "targetConflictColumns": blueprint.target_conflict_columns,
                "projectionTemplate": blueprint.projection_template,
                "includeRequestBodyInContext": true,
                "lookupKeys": blueprint.lookup_keys,
                "resiliencePolicy": {
                    "stateKeyTemplate": blueprint.resilience_policy.state_key_template,
                    "stateCapMaxFires": blueprint.resilience_policy.state_cap_max_fires,
                    "stateCapWindowSeconds": blueprint.resilience_policy.state_cap_window_seconds,
                    "idempotencyTemplate": blueprint.resilience_policy.idempotency_template,
                    "stateResolutionMode": blueprint.resilience_policy.state_resolution_mode,
                    "stateResolutionQueryBuilder": blueprint.resilience_policy.state_resolution_query_builder
                }
            }),
            }
        })
        .collect()
}

pub fn billing_sink_create_params(
    client_name: &str,
    target_client_name: &str,
    route_key: &str,
    target_schema: &str,
    athena_base_url: Option<&str>,
    enabled: bool,
) -> Vec<BillingSinkCreateParams> {
    billing_sink_blueprints()
        .into_iter()
        .map(|blueprint| BillingSinkCreateParams {
            name: blueprint.name.clone(),
            description: Some(format!(
                "Mirrors {} emitted by provider webhook ingestion.",
                blueprint
                    .table_name
                    .trim_start_matches("billing.")
                    .replace('_', " ")
            )),
            enabled,
            athena_base_url: athena_base_url.map(ToString::to_string),
            client_name: client_name.to_string(),
            table_name: blueprint.table_name.clone(),
            route_key: route_key.to_string(),
            target_client_name: target_client_name.to_string(),
            target_schema: target_schema.to_string(),
            target_table: format!("{target_schema}.{}", blueprint.target_table_basename),
            target_write_mode: blueprint.target_write_mode.clone(),
            target_conflict_columns: blueprint.target_conflict_columns.clone(),
            projection_template: blueprint.projection_template.clone(),
            include_request_body_in_context: true,
            lookup_keys: blueprint.lookup_keys.clone(),
            slug: blueprint.slug.clone(),
            state_key_template: blueprint.resilience_policy.state_key_template.clone(),
            state_cap_max_fires: blueprint.resilience_policy.state_cap_max_fires,
            state_cap_window_seconds: blueprint.resilience_policy.state_cap_window_seconds,
            idempotency_template: blueprint.resilience_policy.idempotency_template.clone(),
            state_resolution_mode: blueprint.resilience_policy.state_resolution_mode.clone(),
            state_resolution_query_builder: blueprint
                .resilience_policy
                .state_resolution_query_builder
                .clone(),
        })
        .collect()
}

pub fn billing_sink_resolved_definitions(
    client_name: &str,
    target_client_name: &str,
    route_key: &str,
    target_schema: &str,
    athena_base_url: Option<&str>,
    enabled: bool,
) -> Vec<BillingSinkResolvedDefinition> {
    let blueprints = billing_sink_blueprints();
    let helpers = billing_sink_helpers(
        client_name,
        target_client_name,
        route_key,
        target_schema,
        athena_base_url,
    );
    let create_params = billing_sink_create_params(
        client_name,
        target_client_name,
        route_key,
        target_schema,
        athena_base_url,
        enabled,
    );

    blueprints
        .into_iter()
        .zip(helpers)
        .zip(create_params)
        .map(
            |((blueprint, helper), create_params)| BillingSinkResolvedDefinition {
                blueprint,
                helper,
                create_params,
            },
        )
        .collect()
}

pub fn render_billing_target_schema_sql(target_schema: &str) -> String {
    format!(
        r#"-- Generated canonical billing mirror tables for a downstream Athena schema.
-- Apply this to the target client before provisioning the billing webhook sinks.

CREATE SCHEMA IF NOT EXISTS {schema};

CREATE TABLE IF NOT EXISTS {schema}.billing_payments (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    provider text NOT NULL,
    provider_payment_id text NOT NULL,
    provider_customer_id text,
    provider_profile_id text,
    provider_subscription_id text,
    provider_payment_link_id text,
    status text NOT NULL,
    amount_currency text NOT NULL,
    amount_value numeric(18, 6) NOT NULL,
    amount_refunded_currency text,
    amount_refunded_value numeric(18, 6),
    description text,
    metadata jsonb NOT NULL DEFAULT '{{}}'::jsonb,
    raw jsonb NOT NULL DEFAULT '{{}}'::jsonb,
    paid_at timestamptz,
    created_at timestamptz,
    ingested_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT billing_payments_provider_check CHECK (provider IN ('mollie', 'stripe')),
    CONSTRAINT billing_payments_status_check CHECK (
        status IN ('pending', 'authorized', 'paid', 'failed', 'canceled', 'refunded')
    )
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_{schema}_billing_payments_provider_payment_id
    ON {schema}.billing_payments (provider, provider_payment_id);

CREATE TABLE IF NOT EXISTS {schema}.billing_subscriptions (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    provider text NOT NULL,
    provider_subscription_id text NOT NULL,
    provider_customer_id text NOT NULL,
    provider_profile_id text,
    status text NOT NULL,
    amount_currency text,
    amount_value numeric(18, 6),
    interval text,
    description text,
    metadata jsonb NOT NULL DEFAULT '{{}}'::jsonb,
    raw jsonb NOT NULL DEFAULT '{{}}'::jsonb,
    next_payment_date text,
    created_at timestamptz,
    canceled_at timestamptz,
    ingested_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT billing_subscriptions_provider_check CHECK (provider IN ('mollie', 'stripe')),
    CONSTRAINT billing_subscriptions_status_check CHECK (
        status IN ('trialing', 'active', 'past_due', 'canceled', 'paused')
    )
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_{schema}_billing_subscriptions_provider_subscription_id
    ON {schema}.billing_subscriptions (provider, provider_subscription_id);

CREATE TABLE IF NOT EXISTS {schema}.billing_invoices (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    provider text NOT NULL,
    provider_invoice_id text NOT NULL,
    provider_profile_id text,
    provider_customer_id text,
    status text NOT NULL,
    amount_currency text,
    amount_value numeric(18, 6),
    amount_paid_currency text,
    amount_paid_value numeric(18, 6),
    description text,
    metadata jsonb NOT NULL DEFAULT '{{}}'::jsonb,
    raw jsonb NOT NULL DEFAULT '{{}}'::jsonb,
    issued_at timestamptz,
    paid_at timestamptz,
    created_at timestamptz,
    ingested_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT billing_invoices_provider_check CHECK (provider IN ('mollie', 'stripe')),
    CONSTRAINT billing_invoices_status_check CHECK (
        status IN ('draft', 'open', 'paid', 'void', 'uncollectible')
    )
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_{schema}_billing_invoices_provider_invoice_id
    ON {schema}.billing_invoices (provider, provider_invoice_id);
"#,
        schema = target_schema
    )
}

pub fn render_default_billing_target_schema_sql() -> String {
    render_billing_target_schema_sql("billing")
}

pub fn render_billing_sink_seed_sql(
    source_client_name: &str,
    target_client_name: &str,
    route_key: &str,
    target_schema: &str,
    athena_base_url: Option<&str>,
) -> String {
    let rows = billing_sink_create_params(
        source_client_name,
        target_client_name,
        route_key,
        target_schema,
        athena_base_url,
        true,
    );

    let mut sql = String::from(
        "-- Generated canonical billing webhook sink seed.\n\
         -- Replace the client names or athena_base_url if you need environment-specific routing.\n\n\
         INSERT INTO athena.gateway_webhook_sink (\n\
             enabled,\n\
             name,\n\
             description,\n\
             athena_base_url,\n\
             client_name,\n\
             table_name,\n\
             route_key,\n\
             target_client_name,\n\
             target_table,\n\
             target_write_mode,\n\
             target_conflict_columns,\n\
             projection_template,\n\
             include_request_body_in_context,\n\
             lookup_keys,\n\
             slug,\n\
             state_key_template,\n\
             state_cap_max_fires,\n\
             state_cap_window_seconds,\n\
             idempotency_template,\n\
             state_resolution_mode,\n\
             state_resolution_query_builder\n\
         )\n\
         VALUES\n",
    );

    for (index, row) in rows.iter().enumerate() {
        let projection_template =
            serde_json::to_string_pretty(&row.projection_template).unwrap_or_else(|_| "{}".into());
        let lookup_keys = serde_json::to_string(&row.lookup_keys).unwrap_or_else(|_| "{}".into());
        let target_conflict_columns =
            serde_json::to_string(&row.target_conflict_columns).unwrap_or_else(|_| "[]".into());
        let athena_base_url = row
            .athena_base_url
            .as_ref()
            .map(|value| format!("'{}'", value.replace('\'', "''")))
            .unwrap_or_else(|| "NULL".to_string());
        let state_key_template = row
            .state_key_template
            .as_ref()
            .map(|value| format!("'{}'", value.replace('\'', "''")))
            .unwrap_or_else(|| "NULL".to_string());
        let state_cap_max_fires = row
            .state_cap_max_fires
            .map(|value| value.to_string())
            .unwrap_or_else(|| "NULL".to_string());
        let state_cap_window_seconds = row
            .state_cap_window_seconds
            .map(|value| value.to_string())
            .unwrap_or_else(|| "NULL".to_string());
        let idempotency_template = row
            .idempotency_template
            .as_ref()
            .map(|value| format!("'{}'", value.replace('\'', "''")))
            .unwrap_or_else(|| "NULL".to_string());
        let state_resolution_mode = row
            .state_resolution_mode
            .as_ref()
            .map(|value| format!("'{}'", value.replace('\'', "''")))
            .unwrap_or_else(|| "NULL".to_string());
        let state_resolution_query_builder = row
            .state_resolution_query_builder
            .as_ref()
            .map(|value| format!("'{}'::jsonb", value.to_string().replace('\'', "''")))
            .unwrap_or_else(|| "NULL".to_string());
        let separator = if index + 1 == rows.len() { "\n" } else { ",\n" };

        sql.push_str(&format!(
            "(\n    true,\n    '{}',\n    '{}',\n    {},\n    '{}',\n    '{}',\n    '{}',\n    '{}',\n    '{}',\n    '{}',\n    '{}'::jsonb,\n    '{}'::jsonb,\n    true,\n    '{}'::jsonb,\n    '{}',\n    {},\n    {},\n    {},\n    {},\n    {},\n    {}\n){}",
            row.name.replace('\'', "''"),
            row.description
                .as_deref()
                .unwrap_or_default()
                .replace('\'', "''"),
            athena_base_url,
            row.client_name.replace('\'', "''"),
            row.table_name.replace('\'', "''"),
            row.route_key.replace('\'', "''"),
            row.target_client_name.replace('\'', "''"),
            row.target_table.replace('\'', "''"),
            row.target_write_mode.replace('\'', "''"),
            target_conflict_columns.replace('\'', "''"),
            projection_template.replace('\'', "''"),
            lookup_keys.replace('\'', "''"),
            row.slug.replace('\'', "''"),
            state_key_template,
            state_cap_max_fires,
            state_cap_window_seconds,
            idempotency_template,
            state_resolution_mode,
            state_resolution_query_builder,
            separator,
        ));
    }

    sql.push_str(
        "ON CONFLICT (\n    client_name,\n    route_key,\n    target_client_name,\n    target_table,\n    athena_base_url_norm,\n    table_name_norm\n)\nDO UPDATE SET\n    updated_at = now(),\n    enabled = EXCLUDED.enabled,\n    name = EXCLUDED.name,\n    description = EXCLUDED.description,\n    athena_base_url = EXCLUDED.athena_base_url,\n    table_name = EXCLUDED.table_name,\n    target_write_mode = EXCLUDED.target_write_mode,\n    target_conflict_columns = EXCLUDED.target_conflict_columns,\n    projection_template = EXCLUDED.projection_template,\n    include_request_body_in_context = EXCLUDED.include_request_body_in_context,\n    lookup_keys = EXCLUDED.lookup_keys,\n    slug = COALESCE(EXCLUDED.slug, gateway_webhook_sink.slug),\n    state_key_template = EXCLUDED.state_key_template,\n    state_cap_max_fires = EXCLUDED.state_cap_max_fires,\n    state_cap_window_seconds = EXCLUDED.state_cap_window_seconds,\n    idempotency_template = EXCLUDED.idempotency_template,\n    state_resolution_mode = EXCLUDED.state_resolution_mode,\n    state_resolution_query_builder = EXCLUDED.state_resolution_query_builder;\n",
    );

    sql
}

pub fn render_default_billing_sink_seed_sql(route_key: &str) -> String {
    render_billing_sink_seed_sql(
        DEFAULT_BILLING_SINK_SOURCE_CLIENT_NAME,
        DEFAULT_BILLING_SINK_TARGET_CLIENT_NAME,
        route_key,
        DEFAULT_BILLING_SINK_TARGET_SCHEMA,
        None,
    )
}

#[cfg(test)]
mod tests {
    use super::{
        DEFAULT_BILLING_SINK_SOURCE_CLIENT_NAME, DEFAULT_BILLING_SINK_TARGET_CLIENT_NAME,
        DEFAULT_BILLING_SINK_TARGET_SCHEMA, billing_sink_blueprints, billing_sink_create_params,
        billing_sink_helpers, billing_sink_resolved_definitions, is_valid_sql_identifier,
        render_billing_sink_seed_sql, render_billing_target_schema_sql,
        render_default_billing_sink_seed_sql, render_default_billing_target_schema_sql,
    };
    use serde_json::json;

    #[test]
    fn sink_blueprints_match_canonical_table_shapes() {
        let blueprints = billing_sink_blueprints();
        assert_eq!(blueprints.len(), 3);
        assert_eq!(blueprints[0].table_name, "billing.billing_payments");
        assert_eq!(blueprints[1].table_name, "billing.billing_subscriptions");
        assert_eq!(blueprints[2].table_name, "billing.billing_invoices");
        assert_eq!(blueprints[0].document_kind, "payment");
        assert_eq!(
            blueprints[2].default_target_schema,
            DEFAULT_BILLING_SINK_TARGET_SCHEMA
        );
        assert_eq!(blueprints[1].target_table_basename, "billing_subscriptions");
        assert_eq!(
            blueprints[0]
                .resilience_policy
                .state_resolution_mode
                .as_deref(),
            Some("template")
        );
        assert_eq!(
            blueprints[0]
                .resilience_policy
                .state_resolution_query_builder,
            Some(json!({}))
        );
        assert_eq!(blueprints[0].target_write_mode, "upsert");
        assert_eq!(
            blueprints[0].target_conflict_columns,
            json!(["provider", "provider_payment_id"])
        );
        assert_eq!(
            blueprints[1].projection_template["interval"].as_str(),
            Some("{{ response.sql_write.row.interval }}")
        );
        assert!(
            blueprints[1]
                .projection_template
                .get("interval_unit")
                .is_none()
        );
    }

    #[test]
    fn sink_helpers_and_create_params_share_route_key_and_targets() {
        let helpers = billing_sink_helpers(
            "athena-main",
            "athena-replica",
            "billing_document_upserted",
            "billing",
            Some("https://athena.example"),
        );
        let create = billing_sink_create_params(
            "athena-main",
            "athena-replica",
            "billing_document_upserted",
            "billing",
            Some("https://athena.example"),
            true,
        );

        assert_eq!(helpers.len(), 3);
        assert_eq!(create.len(), 3);
        assert_eq!(
            helpers[0].create_body["routeKey"],
            "billing_document_upserted"
        );
        assert_eq!(helpers[0].target_schema, "billing");
        assert_eq!(helpers[0].target_table, "billing.billing_payments");
        assert_eq!(helpers[0].target_write_mode, "upsert");
        assert_eq!(
            helpers[0].create_body["athenaBaseUrl"],
            json!("https://athena.example")
        );
        assert_eq!(
            helpers[0].target_conflict_columns,
            json!(["provider", "provider_payment_id"])
        );
        assert_eq!(
            helpers[0].create_body["resiliencePolicy"]["stateResolutionMode"],
            "template"
        );
        assert_eq!(helpers[0].create_body["targetWriteMode"], "upsert");
        assert_eq!(create[1].target_table, "billing.billing_subscriptions");
        assert_eq!(create[0].target_schema, "billing");
        assert_eq!(create[0].target_write_mode, "upsert");
        assert_eq!(
            create[0].target_conflict_columns,
            json!(["provider", "provider_payment_id"])
        );
        assert_eq!(
            create[0].state_key_template.as_deref(),
            Some("{{ response.upsert.provider_ref }}|{{ response.sql_write.table }}")
        );
        assert_eq!(create[0].state_resolution_query_builder, Some(json!({})));
        assert_eq!(
            create[0].athena_base_url.as_deref(),
            Some("https://athena.example")
        );
    }

    #[test]
    fn sink_resolved_definitions_keep_blueprints_helpers_and_create_params_aligned() {
        let definitions = billing_sink_resolved_definitions(
            "athena-main",
            "athena-replica",
            "billing_document_upserted",
            "billing",
            Some("https://athena.example"),
            true,
        );

        assert_eq!(definitions.len(), 3);
        assert_eq!(
            definitions[0].blueprint.slug,
            "billing-payments-canonical-mirror"
        );
        assert_eq!(definitions[0].helper.slug, definitions[0].blueprint.slug);
        assert_eq!(
            definitions[0].create_params.slug,
            definitions[0].blueprint.slug
        );
        assert_eq!(
            definitions[0].helper.target_table,
            definitions[0].create_params.target_table
        );
        assert_eq!(
            definitions[1].create_params.target_table,
            "billing.billing_subscriptions"
        );
    }

    #[test]
    fn rendered_seed_sql_uses_current_subscription_projection_shape() {
        let sql = render_billing_sink_seed_sql(
            "athena-main",
            "athena-replica",
            "billing_document_upserted",
            "billing",
            None,
        );
        assert!(sql.contains("\"interval\": \"{{ response.sql_write.row.interval }}\""));
        assert!(!sql.contains("interval_unit"));
        assert!(!sql.contains("\"currency\": \"{{ response.sql_write.row.currency }}\""));
        assert!(sql.contains("'billing.billing_payments'"));
        assert!(sql.contains("state_key_template"));
        assert!(sql.contains("'{}'::jsonb"));
    }

    #[test]
    fn rendered_default_seed_sql_uses_checked_in_defaults() {
        let sql = render_default_billing_sink_seed_sql("billing_document_upserted");
        assert!(sql.contains(DEFAULT_BILLING_SINK_SOURCE_CLIENT_NAME));
        assert!(sql.contains(DEFAULT_BILLING_SINK_TARGET_CLIENT_NAME));
        assert!(sql.contains(DEFAULT_BILLING_SINK_TARGET_SCHEMA));
    }

    #[test]
    fn rendered_target_schema_sql_uses_requested_schema_and_canonical_tables() {
        let sql = render_billing_target_schema_sql("billing");
        assert!(sql.contains("CREATE SCHEMA IF NOT EXISTS billing;"));
        assert!(sql.contains("CREATE TABLE IF NOT EXISTS billing.billing_payments"));
        assert!(sql.contains("CREATE TABLE IF NOT EXISTS billing.billing_subscriptions"));
        assert!(sql.contains("CREATE TABLE IF NOT EXISTS billing.billing_invoices"));
        assert!(sql.contains("idx_billing_billing_payments_provider_payment_id"));

        let default_sql = render_default_billing_target_schema_sql();
        assert!(default_sql.contains("CREATE SCHEMA IF NOT EXISTS billing;"));
    }

    #[test]
    fn sql_identifier_validator_matches_target_schema_contract() {
        assert!(is_valid_sql_identifier("public"));
        assert!(is_valid_sql_identifier("billing"));
        assert!(!is_valid_sql_identifier("app-billing"));
        assert!(!is_valid_sql_identifier("123billing"));
    }
}
