//! Inbound webhook sink registry and portable insert-row contracts.
//!
//! This module defines boot-time webhook sink templates that Athena runtime
//! surfaces through a generic `/webhook-sinks/{name}` ingress route. The route
//! remains thin while the sink contract lives in this reusable workspace crate.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use sha256::digest;
use std::collections::HashMap;
use std::error::Error;
use std::fs;
use std::path::Path;

#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
#[serde(tag = "from", rename_all = "snake_case")]
pub enum InboundWebhookValueSelector {
    Payload { pointer: String },
    QueryParam { name: String },
    Header { name: String },
    Context { key: String },
    Literal { value: Value },
}

#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "snake_case", default)]
pub struct InboundWebhookInsertTargetDefinition {
    pub client: Option<String>,
    pub schema_name: Option<String>,
    pub table_name: String,
    pub row: Value,
}

#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "snake_case", default)]
pub struct InboundWebhookBillingEnrichmentDefinition {
    pub provider: String,
    pub client: Option<String>,
    pub connection_id: Option<InboundWebhookValueSelector>,
    pub mode: Option<InboundWebhookValueSelector>,
    pub provider_profile_id: Option<InboundWebhookValueSelector>,
}

#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "snake_case", default)]
pub struct InboundWebhookSinkDefinition {
    pub client: Option<String>,
    pub athena_base_url: Option<String>,
    pub schema_name: Option<String>,
    pub table_name: String,
    pub secret_header: Option<String>,
    pub secret_query_param: Option<String>,
    pub shared_secret_env_var: Option<String>,
    pub disable_shared_secret_verification: bool,
    pub acknowledge_without_persistence: bool,
    pub event_type_json_pointer: Option<String>,
    pub external_id_json_pointer: Option<String>,
    pub metadata: Option<Value>,
    pub billing_enrichment: Option<InboundWebhookBillingEnrichmentDefinition>,
    pub insert_targets: Vec<InboundWebhookInsertTargetDefinition>,
}

#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct InboundWebhookSinkTemplateSummary {
    pub name: String,
    pub client: Option<String>,
    pub athena_base_url: Option<String>,
    pub schema_name: Option<String>,
    pub table_name: String,
    pub qualified_table_name: String,
    pub secret_header: String,
    pub secret_query_param: String,
    pub shared_secret_env_var: Option<String>,
    pub disable_shared_secret_verification: bool,
    pub acknowledge_without_persistence: bool,
    pub event_type_json_pointer: Option<String>,
    pub external_id_json_pointer: Option<String>,
    pub insert_targets: Vec<InboundWebhookInsertTargetSummary>,
}

#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct InboundWebhookInsertTargetSummary {
    pub client: Option<String>,
    pub schema_name: Option<String>,
    pub table_name: String,
    pub qualified_table_name: String,
    pub column_count: usize,
}

#[derive(Debug, Clone)]
pub struct InboundWebhookSinkPayload<'a> {
    pub sink_name: &'a str,
    pub received_at: DateTime<Utc>,
    pub request_path: &'a str,
    pub http_method: &'a str,
    pub source_ip: Option<&'a str>,
    pub content_type: Option<&'a str>,
    pub query_params_json: Value,
    pub headers_json: Value,
    pub payload_json: Option<Value>,
    pub payload_text: Option<&'a str>,
    pub verified: bool,
    pub event_type: Option<String>,
    pub external_id: Option<String>,
    pub metadata: Value,
}

#[derive(Debug, Clone)]
pub struct InboundWebhookSinkRuntimeContext<'a> {
    pub sink_name: &'a str,
    pub received_at: DateTime<Utc>,
    pub request_path: &'a str,
    pub http_method: &'a str,
    pub source_ip: Option<&'a str>,
    pub content_type: Option<&'a str>,
    pub verified: bool,
    pub event_type: Option<&'a str>,
    pub external_id: Option<&'a str>,
    pub query_params_json: &'a Value,
    pub headers_json: &'a Value,
    pub payload_json: Option<&'a Value>,
    pub payload_text: Option<&'a str>,
    pub payload_sha256: &'a str,
    pub metadata: &'a Value,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
struct WebhookSinkEntry {
    name: String,
    #[serde(flatten)]
    definition: InboundWebhookSinkDefinition,
}

#[derive(Debug, Deserialize)]
struct WebhookSinksFile {
    webhook_sinks: Vec<WebhookSinkEntry>,
}

pub fn load_inbound_sink_registry_from_path<P: AsRef<Path>>(
    path: P,
) -> Result<HashMap<String, InboundWebhookSinkDefinition>, Box<dyn Error + Send + Sync>> {
    let content = fs::read_to_string(path.as_ref())?;
    let file: WebhookSinksFile = serde_yaml::from_str(&content)?;
    let mut registry = HashMap::new();
    for entry in file.webhook_sinks {
        validate_inbound_webhook_sink_definition(&entry.definition)?;
        registry.insert(entry.name, entry.definition);
    }
    Ok(registry)
}

pub fn validate_inbound_webhook_sink_definition(
    definition: &InboundWebhookSinkDefinition,
) -> Result<(), String> {
    qualify_webhook_sink_table_name(definition.schema_name.as_deref(), &definition.table_name)?;
    for target in &definition.insert_targets {
        validate_inbound_insert_target_definition(target)?;
    }
    if let Some(enrichment) = &definition.billing_enrichment {
        validate_inbound_billing_enrichment_definition(enrichment)?;
    }

    if let Some(pointer) = &definition.event_type_json_pointer
        && !pointer.starts_with('/')
    {
        return Err("event_type_json_pointer must be a JSON pointer starting with '/'".to_string());
    }

    if let Some(pointer) = &definition.external_id_json_pointer
        && !pointer.starts_with('/')
    {
        return Err(
            "external_id_json_pointer must be a JSON pointer starting with '/'".to_string(),
        );
    }

    if let Some(athena_base_url) = definition
        .athena_base_url
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        let uri = athena_base_url
            .parse::<actix_web::http::Uri>()
            .map_err(|_| "athena_base_url must be an absolute URL".to_string())?;
        if uri.scheme_str().is_none() || uri.authority().is_none() {
            return Err("athena_base_url must be an absolute URL".to_string());
        }
    }

    Ok(())
}

pub fn validate_inbound_insert_target_definition(
    definition: &InboundWebhookInsertTargetDefinition,
) -> Result<(), String> {
    qualify_webhook_sink_table_name(definition.schema_name.as_deref(), &definition.table_name)
        .map_err(|err| format!("insert_targets[].{err}"))?;
    if !definition.row.is_object() {
        return Err("insert_targets[].row must be a JSON object".to_string());
    }

    validate_selector_tree(&definition.row)
}

pub fn validate_inbound_billing_enrichment_definition(
    definition: &InboundWebhookBillingEnrichmentDefinition,
) -> Result<(), String> {
    if definition.provider.trim().is_empty() {
        return Err("billing_enrichment.provider is required".to_string());
    }

    for (field_name, selector) in [
        ("connection_id", definition.connection_id.as_ref()),
        ("mode", definition.mode.as_ref()),
        (
            "provider_profile_id",
            definition.provider_profile_id.as_ref(),
        ),
    ] {
        if let Some(selector) = selector {
            validate_selector(selector.clone())
                .map_err(|error| format!("billing_enrichment.{field_name}: {error}"))?;
        }
    }

    Ok(())
}

pub fn resolve_inbound_sink_client_name(
    override_client: Option<&str>,
    fallback: Option<&str>,
) -> Option<String> {
    override_client
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .or_else(|| {
            fallback
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(str::to_string)
        })
}

pub fn default_secret_header_name(definition: &InboundWebhookSinkDefinition) -> String {
    definition
        .secret_header
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("X-Athena-Webhook-Secret")
        .to_string()
}

pub fn default_secret_query_param(definition: &InboundWebhookSinkDefinition) -> String {
    definition
        .secret_query_param
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("secret")
        .to_string()
}

pub fn summary_from_inbound_sink(
    name: &str,
    definition: &InboundWebhookSinkDefinition,
) -> InboundWebhookSinkTemplateSummary {
    let qualified_table_name =
        qualify_webhook_sink_table_name(definition.schema_name.as_deref(), &definition.table_name)
            .unwrap_or_else(|_| definition.table_name.clone());
    InboundWebhookSinkTemplateSummary {
        name: name.to_string(),
        client: definition.client.clone(),
        athena_base_url: definition.athena_base_url.clone(),
        schema_name: definition.schema_name.clone(),
        table_name: definition.table_name.clone(),
        qualified_table_name,
        secret_header: default_secret_header_name(definition),
        secret_query_param: default_secret_query_param(definition),
        shared_secret_env_var: definition.shared_secret_env_var.clone(),
        disable_shared_secret_verification: definition.disable_shared_secret_verification,
        acknowledge_without_persistence: definition.acknowledge_without_persistence,
        event_type_json_pointer: definition.event_type_json_pointer.clone(),
        external_id_json_pointer: definition.external_id_json_pointer.clone(),
        insert_targets: definition
            .insert_targets
            .iter()
            .map(|target| InboundWebhookInsertTargetSummary {
                client: target.client.clone(),
                schema_name: target.schema_name.clone(),
                table_name: target.table_name.clone(),
                qualified_table_name: qualify_webhook_sink_table_name(
                    target.schema_name.as_deref(),
                    &target.table_name,
                )
                .unwrap_or_else(|_| target.table_name.clone()),
                column_count: target
                    .row
                    .as_object()
                    .map(|row| row.len())
                    .unwrap_or_default(),
            })
            .collect(),
    }
}

pub fn qualify_webhook_sink_table_name(
    schema_name: Option<&str>,
    table_name: &str,
) -> Result<String, String> {
    let table_name = table_name.trim();
    if table_name.is_empty() {
        return Err("table_name is required".to_string());
    }

    let schema_name = schema_name.map(str::trim).filter(|value| !value.is_empty());

    if let Some(schema_name) = schema_name {
        if table_name.contains('.') {
            return Err(
                "schema_name must not be combined with a schema-qualified table_name".to_string(),
            );
        }
        return Ok(format!("{schema_name}.{table_name}"));
    }

    Ok(table_name.to_string())
}

pub fn build_default_inbound_sink_row(payload: InboundWebhookSinkPayload<'_>) -> Value {
    let payload_text = payload.payload_text.unwrap_or_default();

    json!({
        "sink_name": payload.sink_name,
        "received_at": payload.received_at.to_rfc3339(),
        "request_path": payload.request_path,
        "http_method": payload.http_method,
        "source_ip": payload.source_ip,
        "content_type": payload.content_type,
        "verified": payload.verified,
        "event_type": payload.event_type,
        "external_id": payload.external_id,
        "query_params_json": payload.query_params_json,
        "headers_json": payload.headers_json,
        "payload_json": payload.payload_json,
        "payload_text": if payload_text.is_empty() { Value::Null } else { Value::String(payload_text.to_string()) },
        "payload_sha256": digest(payload_text),
        "metadata": payload.metadata,
    })
}

pub fn json_pointer_string(value: Option<&Value>, pointer: Option<&str>) -> Option<String> {
    let pointer = pointer?.trim();
    if pointer.is_empty() {
        return None;
    }

    match value?.pointer(pointer) {
        Some(Value::String(text)) => Some(text.clone()),
        Some(Value::Number(number)) => Some(number.to_string()),
        Some(Value::Bool(flag)) => Some(flag.to_string()),
        Some(other) if !other.is_null() => Some(other.to_string()),
        _ => None,
    }
}

pub fn build_projected_inbound_sink_rows(
    definition: &InboundWebhookSinkDefinition,
    ctx: &InboundWebhookSinkRuntimeContext<'_>,
) -> Result<Vec<(Option<String>, String, Value)>, String> {
    definition
        .insert_targets
        .iter()
        .map(|target| {
            let qualified_table_name =
                qualify_webhook_sink_table_name(target.schema_name.as_deref(), &target.table_name)?;
            resolve_selector_tree(&target.row, ctx)
                .map(|row| (target.client.clone(), qualified_table_name, row))
        })
        .collect()
}

fn validate_selector_tree(value: &Value) -> Result<(), String> {
    match value {
        Value::Object(object) => {
            if let Some(selector) = parse_selector(value) {
                validate_selector(selector)
            } else {
                for nested in object.values() {
                    validate_selector_tree(nested)?;
                }
                Ok(())
            }
        }
        Value::Array(items) => {
            for item in items {
                validate_selector_tree(item)?;
            }
            Ok(())
        }
        _ => Ok(()),
    }
}

fn validate_selector(selector: InboundWebhookValueSelector) -> Result<(), String> {
    match selector {
        InboundWebhookValueSelector::Payload { pointer } => {
            if !pointer.trim().starts_with('/') {
                Err("payload selector pointer must start with '/'".to_string())
            } else {
                Ok(())
            }
        }
        InboundWebhookValueSelector::QueryParam { name }
        | InboundWebhookValueSelector::Header { name } => {
            if name_or_key_is_blank(&name) {
                Err("selector field must not be blank".to_string())
            } else {
                Ok(())
            }
        }
        InboundWebhookValueSelector::Context { key } => {
            if name_or_key_is_blank(&key) {
                Err("selector field must not be blank".to_string())
            } else {
                Ok(())
            }
        }
        InboundWebhookValueSelector::Literal { .. } => Ok(()),
    }
}

fn name_or_key_is_blank(value: &str) -> bool {
    value.trim().is_empty()
}

fn parse_selector(value: &Value) -> Option<InboundWebhookValueSelector> {
    serde_json::from_value::<InboundWebhookValueSelector>(value.clone()).ok()
}

fn resolve_selector_tree(
    value: &Value,
    ctx: &InboundWebhookSinkRuntimeContext<'_>,
) -> Result<Value, String> {
    if let Some(selector) = parse_selector(value) {
        return resolve_selector(selector, ctx);
    }

    match value {
        Value::Object(object) => {
            let mut resolved = Map::new();
            for (key, nested) in object {
                resolved.insert(key.clone(), resolve_selector_tree(nested, ctx)?);
            }
            Ok(Value::Object(resolved))
        }
        Value::Array(items) => {
            let mut resolved = Vec::with_capacity(items.len());
            for item in items {
                resolved.push(resolve_selector_tree(item, ctx)?);
            }
            Ok(Value::Array(resolved))
        }
        _ => Ok(value.clone()),
    }
}

fn resolve_selector(
    selector: InboundWebhookValueSelector,
    ctx: &InboundWebhookSinkRuntimeContext<'_>,
) -> Result<Value, String> {
    match selector {
        InboundWebhookValueSelector::Payload { pointer } => Ok(ctx
            .payload_json
            .and_then(|payload| payload.pointer(pointer.trim()))
            .cloned()
            .unwrap_or(Value::Null)),
        InboundWebhookValueSelector::QueryParam { name } => Ok(ctx
            .query_params_json
            .as_object()
            .and_then(|query| query.get(name.trim()))
            .cloned()
            .unwrap_or(Value::Null)),
        InboundWebhookValueSelector::Header { name } => {
            let normalized = name.trim().to_ascii_lowercase();
            Ok(ctx
                .headers_json
                .as_object()
                .and_then(|headers| headers.get(&normalized))
                .cloned()
                .unwrap_or(Value::Null))
        }
        InboundWebhookValueSelector::Context { key } => resolve_context_value(key.trim(), ctx),
        InboundWebhookValueSelector::Literal { value } => Ok(value),
    }
}

pub fn resolve_inbound_selector_value(
    selector: &InboundWebhookValueSelector,
    ctx: &InboundWebhookSinkRuntimeContext<'_>,
) -> Result<Value, String> {
    resolve_selector(selector.clone(), ctx)
}

pub fn resolve_inbound_selector_string(
    selector: &InboundWebhookValueSelector,
    ctx: &InboundWebhookSinkRuntimeContext<'_>,
) -> Result<Option<String>, String> {
    let value = resolve_inbound_selector_value(selector, ctx)?;
    selector_value_to_optional_string(&value)
}

fn selector_value_to_optional_string(value: &Value) -> Result<Option<String>, String> {
    match value {
        Value::Null => Ok(None),
        Value::String(text) => {
            let trimmed = text.trim();
            if trimmed.is_empty() {
                Ok(None)
            } else {
                Ok(Some(trimmed.to_string()))
            }
        }
        Value::Number(number) => Ok(Some(number.to_string())),
        Value::Bool(boolean) => Ok(Some(boolean.to_string())),
        Value::Array(_) | Value::Object(_) => Err(
            "selector resolved to a JSON container but billing enrichment requires a scalar value"
                .to_string(),
        ),
    }
}

fn resolve_context_value(
    key: &str,
    ctx: &InboundWebhookSinkRuntimeContext<'_>,
) -> Result<Value, String> {
    match key {
        "sink_name" => Ok(Value::String(ctx.sink_name.to_string())),
        "received_at" => Ok(Value::String(ctx.received_at.to_rfc3339())),
        "request_path" => Ok(Value::String(ctx.request_path.to_string())),
        "http_method" => Ok(Value::String(ctx.http_method.to_string())),
        "source_ip" => Ok(ctx
            .source_ip
            .map(|value| Value::String(value.to_string()))
            .unwrap_or(Value::Null)),
        "content_type" => Ok(ctx
            .content_type
            .map(|value| Value::String(value.to_string()))
            .unwrap_or(Value::Null)),
        "verified" => Ok(Value::Bool(ctx.verified)),
        "event_type" => Ok(ctx
            .event_type
            .map(|value| Value::String(value.to_string()))
            .unwrap_or(Value::Null)),
        "external_id" => Ok(ctx
            .external_id
            .map(|value| Value::String(value.to_string()))
            .unwrap_or(Value::Null)),
        "query_params_json" => Ok(ctx.query_params_json.clone()),
        "headers_json" => Ok(ctx.headers_json.clone()),
        "payload_json" => Ok(ctx.payload_json.cloned().unwrap_or(Value::Null)),
        "payload_text" => Ok(ctx
            .payload_text
            .map(|value| Value::String(value.to_string()))
            .unwrap_or(Value::Null)),
        "payload_sha256" => Ok(Value::String(ctx.payload_sha256.to_string())),
        "metadata" => Ok(ctx.metadata.clone()),
        other => Err(format!("unsupported context selector key '{other}'")),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::{SystemTime, UNIX_EPOCH};

    #[test]
    fn validate_inbound_sink_requires_table_name() {
        let definition = InboundWebhookSinkDefinition::default();
        assert_eq!(
            validate_inbound_webhook_sink_definition(&definition),
            Err("table_name is required".to_string())
        );
    }

    #[test]
    fn build_default_inbound_sink_row_hashes_payload_text() {
        let row = build_default_inbound_sink_row(InboundWebhookSinkPayload {
            sink_name: "mollie_raw",
            received_at: Utc::now(),
            request_path: "/webhook-sinks/mollie_raw",
            http_method: "POST",
            source_ip: Some("127.0.0.1"),
            content_type: Some("application/json"),
            query_params_json: json!({"foo": "bar"}),
            headers_json: json!({"x-test": "1"}),
            payload_json: Some(json!({"id": "evt_123", "type": "payment-link.paid"})),
            payload_text: Some("{\"id\":\"evt_123\"}"),
            verified: true,
            event_type: Some("payment-link.paid".to_string()),
            external_id: Some("evt_123".to_string()),
            metadata: json!({"provider": "mollie"}),
        });

        assert_eq!(row["sink_name"], "mollie_raw");
        assert_eq!(row["payload_sha256"], digest("{\"id\":\"evt_123\"}"));
        assert_eq!(row["event_type"], "payment-link.paid");
    }

    #[test]
    fn load_inbound_sink_registry_reads_yaml_templates() {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system time")
            .as_nanos();
        let temp_path = std::env::temp_dir().join(format!("athena-webhook-sinks-{unique}.yaml"));

        fs::write(
            &temp_path,
            r#"
webhook_sinks:
  - name: mollie_raw
    client: billing
    table_name: athena.webhook_sink_events
    shared_secret_env_var: MOLLIE_WEBHOOK_SECRET
    event_type_json_pointer: /type
"#,
        )
        .expect("write temp yaml");

        let registry =
            load_inbound_sink_registry_from_path(&temp_path).expect("registry should load");
        let definition = registry.get("mollie_raw").expect("template entry");
        assert_eq!(definition.table_name, "athena.webhook_sink_events");
        assert_eq!(definition.schema_name.as_deref(), None);
        assert_eq!(definition.event_type_json_pointer.as_deref(), Some("/type"));
        assert!(!definition.disable_shared_secret_verification);

        let _ = fs::remove_file(temp_path);
    }

    #[test]
    fn json_pointer_string_extracts_scalar_values() {
        let payload = json!({
            "type": "payment-link.paid",
            "data": { "id": 42 }
        });

        assert_eq!(
            json_pointer_string(Some(&payload), Some("/type")),
            Some("payment-link.paid".to_string())
        );
        assert_eq!(
            json_pointer_string(Some(&payload), Some("/data/id")),
            Some("42".to_string())
        );
    }

    #[test]
    fn validate_insert_target_requires_object_row() {
        let definition = InboundWebhookInsertTargetDefinition {
            client: Some("billing".to_string()),
            schema_name: Some("billing".to_string()),
            table_name: "billing_projection_events".to_string(),
            row: json!(["nope"]),
        };

        assert_eq!(
            validate_inbound_insert_target_definition(&definition),
            Err("insert_targets[].row must be a JSON object".to_string())
        );
    }

    #[test]
    fn validate_billing_enrichment_requires_provider() {
        let definition = InboundWebhookSinkDefinition {
            table_name: "athena.webhook_sink_events".to_string(),
            billing_enrichment: Some(InboundWebhookBillingEnrichmentDefinition {
                provider: String::new(),
                ..Default::default()
            }),
            ..Default::default()
        };

        assert_eq!(
            validate_inbound_webhook_sink_definition(&definition),
            Err("billing_enrichment.provider is required".to_string())
        );
    }

    #[test]
    fn resolve_inbound_selector_string_rejects_json_containers() {
        let payload = json!({
            "context": {
                "payment": {
                    "id": "tr_123"
                }
            }
        });
        let query = json!({});
        let headers = json!({});
        let metadata = json!({});
        let ctx = InboundWebhookSinkRuntimeContext {
            sink_name: "mollie",
            received_at: Utc::now(),
            request_path: "/webhook-sinks/mollie",
            http_method: "POST",
            source_ip: None,
            content_type: Some("application/json"),
            verified: false,
            event_type: Some("balance-transaction.created"),
            external_id: Some("baltr_123"),
            query_params_json: &query,
            headers_json: &headers,
            payload_json: Some(&payload),
            payload_text: None,
            payload_sha256: "hash_123",
            metadata: &metadata,
        };

        let error = resolve_inbound_selector_string(
            &InboundWebhookValueSelector::Payload {
                pointer: "/context/payment".to_string(),
            },
            &ctx,
        )
        .expect_err("container payload should be rejected");
        assert!(error.contains("requires a scalar value"));
    }

    #[test]
    fn build_projected_inbound_sink_rows_resolves_selectors() {
        let definition = InboundWebhookSinkDefinition {
            client: Some("billing".to_string()),
            schema_name: Some("billing".to_string()),
            table_name: "webhook_sink_events".to_string(),
            insert_targets: vec![InboundWebhookInsertTargetDefinition {
                client: Some("tenant_a".to_string()),
                schema_name: Some("billing".to_string()),
                table_name: "billing_projection_events".to_string(),
                row: json!({
                    "provider": { "from": "literal", "value": "mollie" },
                    "event_type": { "from": "payload", "pointer": "/type" },
                    "external_id": { "from": "context", "key": "external_id" },
                    "verified": { "from": "context", "key": "verified" },
                    "x_request_id": { "from": "header", "name": "x-request-id" },
                    "secret": { "from": "query_param", "name": "secret" },
                    "raw_payload": { "from": "context", "key": "payload_json" }
                }),
            }],
            ..Default::default()
        };
        let query = json!({ "secret": "abc123" });
        let headers = json!({ "x-request-id": "req_123" });
        let payload = json!({
            "id": "evt_123",
            "type": "payment-link.paid"
        });
        let ctx = InboundWebhookSinkRuntimeContext {
            sink_name: "mollie_raw",
            received_at: Utc::now(),
            request_path: "/webhook-sinks/mollie_raw",
            http_method: "POST",
            source_ip: Some("127.0.0.1"),
            content_type: Some("application/json"),
            verified: true,
            event_type: Some("payment-link.paid"),
            external_id: Some("evt_123"),
            query_params_json: &query,
            headers_json: &headers,
            payload_json: Some(&payload),
            payload_text: Some("{\"id\":\"evt_123\"}"),
            payload_sha256: "hash_123",
            metadata: &json!({ "provider": "mollie" }),
        };

        let rows = build_projected_inbound_sink_rows(&definition, &ctx).expect("rows");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].0.as_deref(), Some("tenant_a"));
        assert_eq!(rows[0].1, "billing.billing_projection_events");
        assert_eq!(rows[0].2["provider"], "mollie");
        assert_eq!(rows[0].2["event_type"], "payment-link.paid");
        assert_eq!(rows[0].2["external_id"], "evt_123");
        assert_eq!(rows[0].2["verified"], true);
        assert_eq!(rows[0].2["x_request_id"], "req_123");
        assert_eq!(rows[0].2["secret"], "abc123");
        assert_eq!(rows[0].2["raw_payload"]["id"], "evt_123");
    }

    #[test]
    fn qualify_webhook_sink_table_name_supports_schema_name() {
        assert_eq!(
            qualify_webhook_sink_table_name(Some("athena"), "webhook_sink_events"),
            Ok("athena.webhook_sink_events".to_string())
        );
        assert_eq!(
            qualify_webhook_sink_table_name(None, "billing.billing_projection_events"),
            Ok("billing.billing_projection_events".to_string())
        );
    }

    #[test]
    fn qualify_webhook_sink_table_name_rejects_ambiguous_schema_input() {
        assert_eq!(
            qualify_webhook_sink_table_name(Some("public"), "billing.billing_projection_events"),
            Err("schema_name must not be combined with a schema-qualified table_name".to_string())
        );
    }

    #[test]
    fn validate_inbound_sink_rejects_non_absolute_athena_base_url() {
        let definition = InboundWebhookSinkDefinition {
            table_name: "athena.webhook_sink_events".to_string(),
            athena_base_url: Some("mirror2.athena-cluster.com".to_string()),
            ..Default::default()
        };

        assert_eq!(
            validate_inbound_webhook_sink_definition(&definition),
            Err("athena_base_url must be an absolute URL".to_string())
        );
    }
}
