//! Render gateway webhook sink projections and insert them into target Athena schemas.

use serde_json::Value;
use sha256::digest;
use std::collections::BTreeSet;
use std::future::Future;
use std::pin::Pin;
use std::time::Instant;
use tracing::{error, warn};

use crate::dispatch::{
    GatewayWebhookTrigger, WebhookRuntime, build_context_snapshot_from_trigger, parse_schema_table,
    render_template_string, validate_sql_ident,
};
use crate::sinks::{
    GatewayWebhookSinkExecutionDefinition, GatewayWebhookSinkRecord,
    GatewayWebhookSinkStateResolutionMode, GatewayWebhookSinkTargetWriteMode,
    gate_webhook_sink_delivery, get_gateway_webhook_sink,
    insert_webhook_sink_delivery_terminal_failure, list_webhook_sinks_for_dispatch,
    update_webhook_sink_delivery_outcome,
};

fn sink_delivery_idempotency_key(
    sink_id: i64,
    trigger: &GatewayWebhookTrigger,
    fingerprint: &str,
) -> String {
    let rid = trigger.request_id.as_deref().unwrap_or("");
    let table = trigger.table_name.as_deref().unwrap_or("");
    digest(format!(
        "sink|{}|{}|{}|{}|{}",
        sink_id, rid, trigger.route_key, table, fingerprint
    ))
}

fn fingerprint_from_response(response: &Option<Value>) -> String {
    let Some(value) = response else {
        return String::new();
    };
    if let Some(resource_id) = value.get("resource_id").and_then(Value::as_str) {
        return resource_id.to_string();
    }
    if let Some(data) = value.get("data")
        && let Some(resource_id) = data.get("resource_id").and_then(Value::as_str)
    {
        return resource_id.to_string();
    }
    value.to_string().chars().take(512).collect()
}

fn render_projection_value<'a>(
    value: &'a Value,
) -> impl std::future::Future<Output = Result<Value, String>> + 'a {
    async move { Ok(value.clone()) }
}

fn render_projection_template<'a, R: WebhookRuntime + Sync + ?Sized + 'a>(
    runtime: &'a R,
    sink: &'a GatewayWebhookSinkRecord,
    sink_definition: &'a GatewayWebhookSinkExecutionDefinition,
    trigger: &'a GatewayWebhookTrigger,
    context: &'a Value,
    value: &'a Value,
) -> Pin<Box<dyn Future<Output = Result<Value, String>> + Send + 'a>> {
    Box::pin(async move {
        let _ = render_projection_value(value).await?;
        match value {
            Value::String(template) => Ok(Value::String(
                render_template_string(
                    template,
                    runtime,
                    &sink_definition.target_client_name,
                    trigger,
                    context,
                    &sink_definition.lookup_keys,
                )
                .await?,
            )),
            Value::Array(items) => {
                let mut rendered = Vec::with_capacity(items.len());
                for item in items {
                    rendered.push(
                        render_projection_template(
                            runtime,
                            sink,
                            sink_definition,
                            trigger,
                            context,
                            item,
                        )
                        .await?,
                    );
                }
                Ok(Value::Array(rendered))
            }
            Value::Object(map) => {
                let mut rendered = serde_json::Map::with_capacity(map.len());
                for (key, item) in map {
                    rendered.insert(
                        key.clone(),
                        render_projection_template(
                            runtime,
                            sink,
                            sink_definition,
                            trigger,
                            context,
                            item,
                        )
                        .await?,
                    );
                }
                Ok(Value::Object(rendered))
            }
            _ => Ok(value.clone()),
        }
    })
}

async fn resolve_sink_key_source(
    source: &str,
    trigger: &GatewayWebhookTrigger,
    context: &Value,
) -> Option<String> {
    let trimmed = source.trim();
    if let Some(rest) = trimmed.strip_prefix("header:") {
        let key = rest.trim().to_ascii_lowercase().replace('-', "_");
        let headers = context.get("headers")?;
        return headers.get(&key)?.as_str().map(ToString::to_string);
    }
    if let Some(rest) = trimmed.strip_prefix("payload:") {
        return path_string_value(context.get("payload")?, rest);
    }
    if let Some(rest) = trimmed.strip_prefix("response:") {
        return path_string_value(context.get("response")?, rest);
    }
    let _ = trigger;
    None
}

fn path_string_value(root: &Value, path: &str) -> Option<String> {
    let mut current = root;
    for segment in path
        .split('.')
        .map(str::trim)
        .filter(|segment| !segment.is_empty())
    {
        current = current.get(segment)?;
    }
    match current {
        Value::String(value) => Some(value.clone()),
        Value::Null => None,
        _ => Some(current.to_string()),
    }
}

async fn resolve_sink_state_via_gateway_builder<R: WebhookRuntime + ?Sized>(
    runtime: &R,
    sink_definition: &GatewayWebhookSinkExecutionDefinition,
    trigger: &GatewayWebhookTrigger,
    context: &Value,
    definition: &Value,
) -> Result<Option<String>, String> {
    let table = definition
        .get("table")
        .and_then(Value::as_str)
        .ok_or_else(|| "state_resolution_query_builder.table is required".to_string())?;
    let select_column = definition
        .get("select_column")
        .and_then(Value::as_str)
        .ok_or_else(|| "state_resolution_query_builder.select_column is required".to_string())?;
    let (schema, table_name) = parse_schema_table(table)?;
    if !validate_sql_ident(schema)
        || !validate_sql_ident(table_name)
        || !validate_sql_ident(select_column)
    {
        return Err("invalid gateway query builder identifiers".to_string());
    }

    let filters = definition
        .get("filters")
        .and_then(Value::as_array)
        .ok_or_else(|| "state_resolution_query_builder.filters must be an array".to_string())?;
    let limit = definition
        .get("limit")
        .and_then(Value::as_i64)
        .unwrap_or(1)
        .clamp(1, 10);
    let pool = runtime
        .pool_for_client(&sink_definition.target_client_name)
        .ok_or_else(|| {
            format!(
                "client '{}' pool unavailable for gateway builder lookup",
                sink_definition.target_client_name
            )
        })?;

    let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new(format!(
        r#"SELECT "{}"::text AS value FROM "{}"."{}""#,
        select_column, schema, table_name
    ));
    if !filters.is_empty() {
        query.push(" WHERE ");
        for (index, filter) in filters.iter().enumerate() {
            if index > 0 {
                query.push(" AND ");
            }
            let column = filter
                .get("column")
                .and_then(Value::as_str)
                .ok_or_else(|| "filter.column is required".to_string())?;
            let source = filter
                .get("source")
                .and_then(Value::as_str)
                .ok_or_else(|| "filter.source is required".to_string())?;
            if !validate_sql_ident(column) {
                return Err("invalid gateway query builder filter column".to_string());
            }
            let value = resolve_sink_key_source(source, trigger, context)
                .await
                .ok_or_else(|| format!("unable to resolve filter source '{}'", source))?;
            query.push(format!(r#""{}"::text = "#, column));
            query.push_bind(value);
        }
    }
    query.push(" LIMIT ");
    query.push_bind(limit);
    let row = query
        .build()
        .fetch_optional(&pool)
        .await
        .map_err(|error| format!("gateway query builder execution failed: {}", error))?;
    Ok(row.and_then(|record| {
        sqlx::Row::try_get::<Option<String>, _>(&record, "value")
            .ok()
            .flatten()
    }))
}

async fn resolve_sink_state_key<R: WebhookRuntime + ?Sized>(
    runtime: &R,
    _sink: &GatewayWebhookSinkRecord,
    sink_definition: &GatewayWebhookSinkExecutionDefinition,
    trigger: &GatewayWebhookTrigger,
    context: &Value,
    policy: &crate::sinks::GatewayWebhookSinkResiliencePolicyDefinition,
) -> Result<Option<String>, String> {
    let base = if let Some(template) = policy.state_key_template.as_ref() {
        let rendered = render_template_string(
            template,
            runtime,
            &sink_definition.target_client_name,
            trigger,
            context,
            &sink_definition.lookup_keys,
        )
        .await?;
        let trimmed = rendered.trim().to_string();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed)
        }
    } else {
        None
    };

    match &policy.state_resolution_mode {
        GatewayWebhookSinkStateResolutionMode::Template => Ok(base),
        GatewayWebhookSinkStateResolutionMode::GatewayQueryBuilder => {
            let resolved = resolve_sink_state_via_gateway_builder(
                runtime,
                sink_definition,
                trigger,
                context,
                &policy.state_resolution_query_builder,
            )
            .await?;
            Ok(match (base, resolved) {
                (Some(a), Some(b)) => Some(format!("{}|{}", a, b)),
                (Some(a), None) => Some(a),
                (None, Some(b)) => Some(b),
                (None, None) => None,
            })
        }
    }
}

async fn resolve_sink_idempotency_key<R: WebhookRuntime + ?Sized>(
    runtime: &R,
    sink: &GatewayWebhookSinkRecord,
    sink_definition: &GatewayWebhookSinkExecutionDefinition,
    trigger: &GatewayWebhookTrigger,
    context: &Value,
    policy: &crate::sinks::GatewayWebhookSinkResiliencePolicyDefinition,
    fingerprint: &str,
) -> Result<String, String> {
    if let Some(template) = policy.idempotency_template.as_ref() {
        let rendered = render_template_string(
            template,
            runtime,
            &sink_definition.target_client_name,
            trigger,
            context,
            &sink_definition.lookup_keys,
        )
        .await?;
        let trimmed = rendered.trim();
        if !trimmed.is_empty() {
            return Ok(digest(trimmed));
        }
    }
    Ok(sink_delivery_idempotency_key(sink.id, trigger, fingerprint))
}

async fn execute_sink_insert<R: WebhookRuntime + ?Sized>(
    runtime: &R,
    sink_definition: &GatewayWebhookSinkExecutionDefinition,
    projected_payload: &Value,
) -> Result<i32, String> {
    let (schema, table) = parse_schema_table(&sink_definition.target_table)?;
    if !validate_sql_ident(schema) || !validate_sql_ident(table) {
        return Err("invalid sink target_table identifiers".to_string());
    }
    let pool = runtime
        .pool_for_client(&sink_definition.target_client_name)
        .ok_or_else(|| {
            format!(
                "client '{}' pool unavailable for sink target insert",
                sink_definition.target_client_name
            )
        })?;

    let projected_columns = projected_payload_object_columns(projected_payload)?;
    let sql = build_sink_insert_sql(
        &schema,
        &table,
        &projected_columns,
        &sink_definition.target_write_mode,
    )?;

    let result = sqlx::query(&sql)
        .bind(projected_payload)
        .execute(&pool)
        .await
        .map_err(|error| format!("sink insert failed: {}", error))?;
    Ok(result.rows_affected() as i32)
}

fn projected_payload_object_columns(projected_payload: &Value) -> Result<Vec<String>, String> {
    let object = projected_payload
        .as_object()
        .ok_or_else(|| "sink projected payload must be a JSON object".to_string())?;
    let mut columns = object
        .keys()
        .cloned()
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect::<Vec<_>>();
    if columns.is_empty() {
        return Err("sink projected payload must include at least one target column".to_string());
    }
    if columns.iter().any(|column| !validate_sql_ident(column)) {
        return Err(
            "sink projected payload contains invalid target column identifiers".to_string(),
        );
    }
    columns.sort();
    Ok(columns)
}

fn build_sink_insert_sql(
    schema: &str,
    table: &str,
    projected_columns: &[String],
    write_mode: &GatewayWebhookSinkTargetWriteMode,
) -> Result<String, String> {
    let column_list = projected_columns
        .iter()
        .map(|column| format!(r#""{column}""#))
        .collect::<Vec<_>>()
        .join(", ");
    let select_list = projected_columns
        .iter()
        .map(|column| format!(r#"record."{column}""#))
        .collect::<Vec<_>>()
        .join(", ");
    let base = format!(
        r#"
        INSERT INTO "{schema}"."{table}" ({column_list})
        SELECT {select_list}
        FROM jsonb_populate_record(NULL::"{schema}"."{table}", $1::jsonb) AS record
        "#
    );
    match write_mode {
        GatewayWebhookSinkTargetWriteMode::Insert => Ok(base),
        GatewayWebhookSinkTargetWriteMode::Upsert { conflict_columns } => {
            if conflict_columns
                .iter()
                .any(|column| !validate_sql_ident(column))
            {
                return Err("sink target_conflict_columns contains invalid identifiers".to_string());
            }
            if conflict_columns.iter().any(|column| {
                !projected_columns
                    .iter()
                    .any(|candidate| candidate == column)
            }) {
                return Err(
                    "sink target_conflict_columns must be present in the projected payload"
                        .to_string(),
                );
            }
            let update_columns = projected_columns
                .iter()
                .filter(|column| !conflict_columns.iter().any(|conflict| conflict == *column))
                .cloned()
                .collect::<Vec<_>>();
            let conflict_list = conflict_columns
                .iter()
                .map(|column| format!(r#""{column}""#))
                .collect::<Vec<_>>()
                .join(", ");
            if update_columns.is_empty() {
                Ok(format!("{base}\nON CONFLICT ({conflict_list}) DO NOTHING"))
            } else {
                let update_set = update_columns
                    .iter()
                    .map(|column| format!(r#""{column}" = EXCLUDED."{column}""#))
                    .collect::<Vec<_>>()
                    .join(", ");
                Ok(format!(
                    "{base}\nON CONFLICT ({conflict_list}) DO UPDATE SET {update_set}"
                ))
            }
        }
    }
}

pub async fn dispatch_webhook_sink_test<R: WebhookRuntime + Sync + ?Sized>(
    runtime: &R,
    catalog_pool: &sqlx::postgres::PgPool,
    sink: &GatewayWebhookSinkRecord,
    trigger: GatewayWebhookTrigger,
) -> Result<(i64, String, String, Option<i32>, Value), String> {
    let sink_definition = sink.execution_definition()?;
    let policy = &sink_definition.resilience_policy;
    let context = build_context_snapshot_from_trigger(
        &trigger,
        sink_definition.include_request_body_in_context,
    );
    let projection = render_projection_template(
        runtime,
        sink,
        &sink_definition,
        &trigger,
        &context,
        &sink_definition.projection_template,
    )
    .await?;
    let fingerprint = fingerprint_from_response(&trigger.response);
    let idempotency_key = resolve_sink_idempotency_key(
        runtime,
        sink,
        &sink_definition,
        &trigger,
        &context,
        policy,
        &fingerprint,
    )
    .await?;
    let state_key =
        resolve_sink_state_key(runtime, sink, &sink_definition, &trigger, &context, policy).await?;

    let delivery_id = gate_webhook_sink_delivery(
        catalog_pool,
        sink.id,
        &trigger.route_key,
        trigger.request_id.as_deref(),
        &idempotency_key,
        context.clone(),
        &sink_definition.target_client_name,
        &sink_definition.target_table,
        Some(&projection),
        state_key
            .as_ref()
            .map(|value| digest(value).to_string())
            .as_deref(),
        state_key.as_deref(),
        policy.state_cap_max_fires,
        policy.state_cap_window_seconds,
    )
    .await
    .map_err(|error| error.to_string())?;

    match execute_sink_insert(runtime, &sink_definition, &projection).await {
        Ok(inserted_row_count) => {
            update_webhook_sink_delivery_outcome(
                catalog_pool,
                delivery_id,
                "success",
                Some(inserted_row_count),
                None,
                None,
            )
            .await
            .map_err(|error| error.to_string())?;
            Ok((
                delivery_id,
                sink_definition.target_client_name.clone(),
                sink_definition.target_table.clone(),
                Some(inserted_row_count),
                projection,
            ))
        }
        Err(error_message) => {
            update_webhook_sink_delivery_outcome(
                catalog_pool,
                delivery_id,
                "failed",
                None,
                Some(error_message.as_str()),
                None,
            )
            .await
            .map_err(|error| error.to_string())?;
            Err(error_message)
        }
    }
}

pub fn spawn_gateway_webhook_sink_dispatch<R>(runtime: R, trigger: GatewayWebhookTrigger)
where
    R: WebhookRuntime + Clone + Send + Sync + 'static,
{
    tokio::spawn(async move {
        run_gateway_webhook_sink_dispatch(&runtime, &trigger).await;
    });
}

pub(crate) async fn run_gateway_webhook_sink_dispatch<R: WebhookRuntime + Sync + ?Sized>(
    runtime: &R,
    trigger: &GatewayWebhookTrigger,
) {
    let Some(catalog_pool) = runtime.pool_for_client(&trigger.client_name) else {
        warn!(
            client_name = %trigger.client_name,
            "gateway webhook sinks: request-bound catalog pool unavailable; skipping"
        );
        return;
    };
    let instance_url = std::env::var("ATHENA_WEBHOOK_PUBLIC_BASE_URL")
        .ok()
        .map(|value| value.trim().trim_end_matches('/').to_string());
    let sinks = match list_webhook_sinks_for_dispatch(
        &catalog_pool,
        &trigger.client_name,
        &trigger.route_key,
        trigger.table_name.as_deref(),
        instance_url.as_deref(),
    )
    .await
    {
        Ok(sinks) => sinks,
        Err(error) => {
            error!(error = %error, "gateway webhook sinks: list failed");
            return;
        }
    };
    if sinks.is_empty() {
        return;
    }

    let fingerprint = fingerprint_from_response(&trigger.response);
    for sink in sinks {
        let sink_definition = match sink.execution_definition() {
            Ok(definition) => definition,
            Err(error_message) => {
                let fallback_key = sink_delivery_idempotency_key(sink.id, trigger, &fingerprint);
                let context = build_context_snapshot_from_trigger(
                    trigger,
                    sink.include_request_body_in_context,
                );
                let _ = insert_webhook_sink_delivery_terminal_failure(
                    &catalog_pool,
                    sink.id,
                    &trigger.route_key,
                    trigger.request_id.as_deref(),
                    &fallback_key,
                    &sink.target_client_name,
                    &sink.target_table,
                    None,
                    context,
                    "SINK_DEFINITION_INVALID",
                    Some(error_message.as_str()),
                )
                .await;
                continue;
            }
        };
        let policy = &sink_definition.resilience_policy;
        let context = build_context_snapshot_from_trigger(
            trigger,
            sink_definition.include_request_body_in_context,
        );
        let projection = match render_projection_template(
            runtime,
            &sink,
            &sink_definition,
            trigger,
            &context,
            &sink_definition.projection_template,
        )
        .await
        {
            Ok(projection) => projection,
            Err(error_message) => {
                let fallback_key = sink_delivery_idempotency_key(sink.id, trigger, &fingerprint);
                let _ = insert_webhook_sink_delivery_terminal_failure(
                    &catalog_pool,
                    sink.id,
                    &trigger.route_key,
                    trigger.request_id.as_deref(),
                    &fallback_key,
                    &sink.target_client_name,
                    &sink.target_table,
                    None,
                    context.clone(),
                    "PROJECTION_RENDER_FAILED",
                    Some(error_message.as_str()),
                )
                .await;
                continue;
            }
        };

        let idempotency_key = match resolve_sink_idempotency_key(
            runtime,
            &sink,
            &sink_definition,
            trigger,
            &context,
            policy,
            &fingerprint,
        )
        .await
        {
            Ok(value) => value,
            Err(error_message) => {
                let fallback_key = sink_delivery_idempotency_key(sink.id, trigger, &fingerprint);
                let _ = insert_webhook_sink_delivery_terminal_failure(
                    &catalog_pool,
                    sink.id,
                    &trigger.route_key,
                    trigger.request_id.as_deref(),
                    &fallback_key,
                    &sink.target_client_name,
                    &sink.target_table,
                    Some(&projection),
                    context.clone(),
                    "IDEMPOTENCY_RESOLUTION_FAILED",
                    Some(error_message.as_str()),
                )
                .await;
                continue;
            }
        };

        let state_key = match resolve_sink_state_key(
            runtime,
            &sink,
            &sink_definition,
            trigger,
            &context,
            policy,
        )
        .await
        {
            Ok(value) => value,
            Err(error_message) => {
                let _ = insert_webhook_sink_delivery_terminal_failure(
                    &catalog_pool,
                    sink.id,
                    &trigger.route_key,
                    trigger.request_id.as_deref(),
                    &idempotency_key,
                    &sink.target_client_name,
                    &sink.target_table,
                    Some(&projection),
                    context.clone(),
                    "STATE_RESOLUTION_FAILED",
                    Some(error_message.as_str()),
                )
                .await;
                continue;
            }
        };

        let delivery_id = match gate_webhook_sink_delivery(
            &catalog_pool,
            sink.id,
            &trigger.route_key,
            trigger.request_id.as_deref(),
            &idempotency_key,
            context.clone(),
            &sink.target_client_name,
            &sink.target_table,
            Some(&projection),
            state_key
                .as_ref()
                .map(|value| digest(value).to_string())
                .as_deref(),
            state_key.as_deref(),
            policy.state_cap_max_fires,
            policy.state_cap_window_seconds,
        )
        .await
        {
            Ok(delivery_id) => delivery_id,
            Err(error) => {
                error!(error = %error, sink_id = sink.id, "gateway webhook sinks: insert delivery failed");
                continue;
            }
        };

        let started = Instant::now();
        match execute_sink_insert(runtime, &sink_definition, &projection).await {
            Ok(inserted_row_count) => {
                let duration_ms = started.elapsed().as_millis() as i64;
                if let Err(error) = update_webhook_sink_delivery_outcome(
                    &catalog_pool,
                    delivery_id,
                    "success",
                    Some(inserted_row_count),
                    None,
                    Some(duration_ms),
                )
                .await
                {
                    error!(error = %error, "gateway webhook sinks: finalize success delivery failed");
                }
            }
            Err(error_message) => {
                let duration_ms = started.elapsed().as_millis() as i64;
                if let Err(error) = update_webhook_sink_delivery_outcome(
                    &catalog_pool,
                    delivery_id,
                    "failed",
                    None,
                    Some(error_message.as_str()),
                    Some(duration_ms),
                )
                .await
                {
                    error!(error = %error, "gateway webhook sinks: finalize failed delivery failed");
                }
            }
        }
    }
}

pub async fn dispatch_webhook_sink_test_by_id<R: WebhookRuntime + Sync + ?Sized>(
    runtime: &R,
    catalog_pool: &sqlx::postgres::PgPool,
    sink_id: i64,
    trigger: GatewayWebhookTrigger,
) -> Result<(i64, String, String, Option<i32>, Value), String> {
    let sink = get_gateway_webhook_sink(catalog_pool, sink_id)
        .await
        .map_err(|error| error.to_string())?;
    dispatch_webhook_sink_test(runtime, catalog_pool, &sink, trigger).await
}

#[cfg(test)]
mod tests {
    use super::{build_sink_insert_sql, projected_payload_object_columns};
    use crate::sinks::GatewayWebhookSinkTargetWriteMode;
    use serde_json::json;

    #[test]
    fn projected_payload_columns_are_sorted_and_validated() {
        let columns = projected_payload_object_columns(&json!({
            "provider_payment_id": "pay_123",
            "provider": "mollie"
        }))
        .expect("columns should resolve");
        assert_eq!(
            columns,
            vec!["provider".to_string(), "provider_payment_id".to_string()]
        );
        assert!(projected_payload_object_columns(&json!(["bad"])).is_err());
    }

    #[test]
    fn sink_insert_sql_uses_explicit_columns_for_defaults() {
        let sql = build_sink_insert_sql(
            "public",
            "billing_payments",
            &["provider".to_string(), "provider_payment_id".to_string()],
            &GatewayWebhookSinkTargetWriteMode::Insert,
        )
        .expect("insert sql");
        assert!(sql.contains(
            r#"INSERT INTO "public"."billing_payments" ("provider", "provider_payment_id")"#
        ));
        assert!(sql.contains(
            r#"FROM jsonb_populate_record(NULL::"public"."billing_payments", $1::jsonb) AS record"#
        ));
        assert!(!sql.contains("ON CONFLICT"));
    }

    #[test]
    fn sink_upsert_sql_uses_conflict_columns_and_updates_non_conflicts() {
        let sql = build_sink_insert_sql(
            "billing",
            "billing_invoices",
            &[
                "provider".to_string(),
                "provider_invoice_id".to_string(),
                "status".to_string(),
            ],
            &GatewayWebhookSinkTargetWriteMode::Upsert {
                conflict_columns: vec!["provider".to_string(), "provider_invoice_id".to_string()],
            },
        )
        .expect("upsert sql");
        assert!(sql.contains(r#"ON CONFLICT ("provider", "provider_invoice_id") DO UPDATE SET "status" = EXCLUDED."status""#));
    }
}
