athena_rs 3.12.3

//! Gateway webhook 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 thiserror::Error;

#[derive(Debug, Error)]
pub enum WebhookStoreError {
    #[error("database error: {0}")]
    Database(#[from] sqlx::Error),
    #[error("webhook not found")]
    NotFound,
}

#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct GatewayWebhookRecord {
    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 http_method: String,
    pub url_template: String,
    pub headers_templates: Value,
    pub body_template: Option<String>,
    pub timeout_ms: i32,
    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,
}

#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct GatewayWebhookDeliveryRecord {
    pub id: i64,
    pub created_at: DateTime<Utc>,
    pub webhook_id: i64,
    pub trigger_route_key: String,
    pub request_id: Option<String>,
    pub idempotency_key: String,
    pub status: String,
    pub http_status: Option<i32>,
    pub response_headers_json: Option<Value>,
    pub response_body_snippet: Option<String>,
    pub error_message: Option<String>,
    pub duration_ms: Option<i64>,
    pub resolved_url: Option<String>,
    pub context_snapshot: Value,
}

#[derive(Debug, Clone)]
pub struct CreateGatewayWebhookParams {
    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 http_method: String,
    pub url_template: String,
    pub headers_templates: Value,
    pub body_template: Option<String>,
    pub timeout_ms: i32,
    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 PatchGatewayWebhookParams {
    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 http_method: Option<String>,
    pub url_template: Option<String>,
    pub headers_templates: Option<Value>,
    pub body_template: Option<Option<String>>,
    pub timeout_ms: Option<i32>,
    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>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WebhookGateDecision {
    Allowed {
        delivery_id: i64,
    },
    Duplicate {
        delivery_id: Option<i64>,
    },
    OverLimit {
        delivery_id: i64,
        observed_count: i64,
    },
}

/// Webhooks that should run for this gateway event.
pub async fn list_webhooks_for_dispatch(
    pool: &PgPool,
    client_name: &str,
    route_key: &str,
    table_from_event: Option<&str>,
    instance_base_url: Option<&str>,
) -> Result<Vec<GatewayWebhookRecord>, WebhookStoreError> {
    let rows: Vec<GatewayWebhookRecord> = sqlx::query_as::<_, GatewayWebhookRecord>(
        r#"
        SELECT
           id, created_at, updated_at, enabled, name, description, athena_base_url,
            client_name, table_name, route_key, http_method, url_template,
            headers_templates, body_template, timeout_ms, 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 gateway_webhook
        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_webhooks(
    pool: &PgPool,
    limit: i64,
    offset: i64,
) -> Result<Vec<GatewayWebhookRecord>, WebhookStoreError> {
    list_gateway_webhooks_filtered(pool, None, limit, offset).await
}

/// List webhook definitions, optionally restricted to one `X-Athena-Client` / `client_name`.
pub async fn list_gateway_webhooks_filtered(
    pool: &PgPool,
    client_name: Option<&str>,
    limit: i64,
    offset: i64,
) -> Result<Vec<GatewayWebhookRecord>, WebhookStoreError> {
    let rows: Vec<GatewayWebhookRecord> = if let Some(cn) = client_name.filter(|s| !s.is_empty()) {
        sqlx::query_as::<_, GatewayWebhookRecord>(
            r#"
            SELECT
                id, created_at, updated_at, enabled, name, description, athena_base_url,
                client_name, table_name, route_key, http_method, url_template,
                headers_templates, body_template, timeout_ms, 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 gateway_webhook
            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::<_, GatewayWebhookRecord>(
            r#"
            SELECT
                id, created_at, updated_at, enabled, name, description, athena_base_url,
                client_name, table_name, route_key, http_method, url_template,
                headers_templates, body_template, timeout_ms, 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 gateway_webhook
            ORDER BY id DESC
            LIMIT $1 OFFSET $2
            "#,
        )
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await?
    };
    Ok(rows)
}

pub async fn get_gateway_webhook(
    pool: &PgPool,
    id: i64,
) -> Result<GatewayWebhookRecord, WebhookStoreError> {
    sqlx::query_as::<_, GatewayWebhookRecord>(
        r#"
        SELECT
            id, created_at, updated_at, enabled, name, description, athena_base_url,
            client_name, table_name, route_key, http_method, url_template,
            headers_templates, body_template, timeout_ms, 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 gateway_webhook
        WHERE id = $1
        "#,
    )
    .bind(id)
    .fetch_optional(pool)
    .await?
    .ok_or(WebhookStoreError::NotFound)
}

pub async fn upsert_gateway_webhook(
    pool: &PgPool,
    params: CreateGatewayWebhookParams,
) -> Result<GatewayWebhookRecord, WebhookStoreError> {
    let record: GatewayWebhookRecord = sqlx::query_as::<_, GatewayWebhookRecord>(
        r#"
        INSERT INTO gateway_webhook (
            enabled, name, description, athena_base_url, client_name, table_name,
            route_key, http_method, url_template, headers_templates, body_template,
            timeout_ms, 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, http_method, url_template,
            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,
            headers_templates = EXCLUDED.headers_templates,
            body_template = EXCLUDED.body_template,
            timeout_ms = EXCLUDED.timeout_ms,
            include_request_body_in_context = EXCLUDED.include_request_body_in_context,
            lookup_keys = EXCLUDED.lookup_keys,
            slug = COALESCE(EXCLUDED.slug, gateway_webhook.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, http_method, url_template,
            headers_templates, body_template, timeout_ms, 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.http_method)
    .bind(&params.url_template)
    .bind(&params.headers_templates)
    .bind(&params.body_template)
    .bind(params.timeout_ms)
    .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(
    pool: &PgPool,
    id: i64,
    patch: PatchGatewayWebhookParams,
) -> Result<GatewayWebhookRecord, WebhookStoreError> {
    let existing: GatewayWebhookRecord = get_gateway_webhook(pool, id).await?;
    let name: String = patch.name.unwrap_or(existing.name);
    let description: Option<String> = match &patch.description {
        Some(d) => d.clone(),
        None => existing.description.clone(),
    };
    let enabled: bool = patch.enabled.unwrap_or(existing.enabled);
    let athena_base_url: Option<String> = match &patch.athena_base_url {
        Some(u) => u.clone(),
        None => existing.athena_base_url.clone(),
    };
    let client_name: String = patch.client_name.unwrap_or(existing.client_name);
    let table_name: Option<String> = match &patch.table_name {
        Some(t) => t.clone(),
        None => existing.table_name.clone(),
    };
    let route_key: String = patch.route_key.unwrap_or(existing.route_key);
    let http_method: String = patch.http_method.unwrap_or(existing.http_method);
    let url_template: String = patch.url_template.unwrap_or(existing.url_template);
    let headers_templates: Value = patch
        .headers_templates
        .unwrap_or(existing.headers_templates.clone());
    let body_template: Option<String> = match &patch.body_template {
        Some(b) => b.clone(),
        None => existing.body_template.clone(),
    };
    let timeout_ms: i32 = patch.timeout_ms.unwrap_or(existing.timeout_ms);
    let include_request_body_in_context: bool = patch
        .include_request_body_in_context
        .unwrap_or(existing.include_request_body_in_context);
    let lookup_keys: Value = patch.lookup_keys.unwrap_or(existing.lookup_keys.clone());
    let slug: Option<String> = match &patch.slug {
        Some(s) => s.clone(),
        None => existing.slug.clone(),
    };
    let state_key_template: Option<String> = match &patch.state_key_template {
        Some(v) => v.clone(),
        None => existing.state_key_template.clone(),
    };
    let state_cap_max_fires: Option<i32> = match patch.state_cap_max_fires {
        Some(v) => v,
        None => existing.state_cap_max_fires,
    };
    let state_cap_window_seconds: Option<i32> = match patch.state_cap_window_seconds {
        Some(v) => v,
        None => existing.state_cap_window_seconds,
    };
    let idempotency_template: Option<String> = match &patch.idempotency_template {
        Some(v) => v.clone(),
        None => existing.idempotency_template.clone(),
    };
    let state_resolution_mode: String = patch
        .state_resolution_mode
        .unwrap_or(existing.state_resolution_mode);
    let state_resolution_query_builder: Value = patch
        .state_resolution_query_builder
        .unwrap_or(existing.state_resolution_query_builder);

    sqlx::query(
        r#"
        UPDATE gateway_webhook SET
            updated_at = now(),
            name = $1,
            description = $2,
            enabled = $3,
            athena_base_url = $4,
            client_name = $5,
            table_name = $6,
            route_key = $7,
            http_method = $8,
            url_template = $9,
            headers_templates = $10,
            body_template = $11,
            timeout_ms = $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(&http_method)
    .bind(&url_template)
    .bind(&headers_templates)
    .bind(&body_template)
    .bind(timeout_ms)
    .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(pool, id).await
}

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

/// Insert a pending delivery row; returns `None` if `idempotency_key` already exists.
pub async fn insert_webhook_delivery_pending(
    pool: &PgPool,
    webhook_id: i64,
    trigger_route_key: &str,
    request_id: Option<&str>,
    idempotency_key: &str,
    context_snapshot: Value,
) -> Result<Option<i64>, WebhookStoreError> {
    let row: Option<(i64,)> = sqlx::query_as(
        r#"
        INSERT INTO gateway_webhook_delivery (
            webhook_id, trigger_route_key, request_id, idempotency_key,
            status, context_snapshot
        )
        VALUES ($1, $2, $3, $4, 'pending', $5)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id
        "#,
    )
    .bind(webhook_id)
    .bind(trigger_route_key)
    .bind(request_id)
    .bind(idempotency_key)
    .bind(&context_snapshot)
    .fetch_optional(pool)
    .await?;
    Ok(row.map(|t| t.0))
}

pub async fn update_webhook_delivery_outcome(
    pool: &PgPool,
    delivery_id: i64,
    status: &str,
    http_status: Option<i32>,
    response_headers_json: Option<Value>,
    response_body_snippet: Option<&str>,
    error_message: Option<&str>,
    duration_ms: Option<i64>,
    resolved_url: Option<&str>,
) -> Result<(), WebhookStoreError> {
    sqlx::query(
        r#"
        UPDATE gateway_webhook_delivery SET
            status = $1,
            http_status = $2,
            response_headers_json = $3,
            response_body_snippet = $4,
            error_message = $5,
            duration_ms = $6,
            resolved_url = $7
        WHERE id = $8
        "#,
    )
    .bind(status)
    .bind(http_status)
    .bind(response_headers_json)
    .bind(response_body_snippet)
    .bind(error_message)
    .bind(duration_ms)
    .bind(resolved_url)
    .bind(delivery_id)
    .execute(pool)
    .await?;
    Ok(())
}

/// Insert a terminal failed delivery row for resolution/gate errors; returns `None` on idempotency conflict.
pub async fn insert_webhook_delivery_terminal_failure(
    pool: &PgPool,
    webhook_id: i64,
    trigger_route_key: &str,
    request_id: Option<&str>,
    idempotency_key: &str,
    context_snapshot: Value,
    error_code: &str,
    detail: Option<&str>,
) -> Result<Option<i64>, WebhookStoreError> {
    const DETAIL_MAX: usize = 2000;
    let error_message: String = match detail {
        Some(d) if !d.trim().is_empty() => {
            let t = d.trim();
            let suffix = if t.len() > DETAIL_MAX {
                format!("{}…", &t[..DETAIL_MAX])
            } else {
                t.to_string()
            };
            format!("{}: {}", error_code, suffix)
        }
        _ => error_code.to_string(),
    };
    let row: Option<(i64,)> = sqlx::query_as(
        r#"
        INSERT INTO gateway_webhook_delivery (
            webhook_id, trigger_route_key, request_id, idempotency_key,
            status, context_snapshot, error_message
        )
        VALUES ($1, $2, $3, $4, 'failed', $5, $6)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id
        "#,
    )
    .bind(webhook_id)
    .bind(trigger_route_key)
    .bind(request_id)
    .bind(idempotency_key)
    .bind(&context_snapshot)
    .bind(&error_message)
    .fetch_optional(pool)
    .await?;
    Ok(row.map(|t| t.0))
}

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

    // Serialize cap checks per (webhook, state key) so concurrent deliveries cannot exceed N in one window.
    if let (Some(_max_fires), Some(_window_seconds), 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!("{}|{}", webhook_id, key_hash))
        .execute(&mut *tx)
        .await?;
    }

    let inserted_delivery: Option<(i64,)> = sqlx::query_as(
        r#"
        INSERT INTO gateway_webhook_delivery (
            webhook_id, trigger_route_key, request_id, idempotency_key,
            status, context_snapshot
        )
        VALUES ($1, $2, $3, $4, 'pending', $5)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id
        "#,
    )
    .bind(webhook_id)
    .bind(trigger_route_key)
    .bind(request_id)
    .bind(idempotency_key)
    .bind(&context_snapshot)
    .fetch_optional(&mut *tx)
    .await?;

    let delivery_id: i64 = match inserted_delivery {
        Some((id,)) => id,
        None => {
            let existing: Option<(i64,)> = sqlx::query_as(
                r#"SELECT id FROM gateway_webhook_delivery WHERE idempotency_key = $1"#,
            )
            .bind(idempotency_key)
            .fetch_optional(&mut *tx)
            .await?;
            tx.commit().await?;
            return Ok(WebhookGateDecision::Duplicate {
                delivery_id: existing.map(|t| t.0),
            });
        }
    };

    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 gateway_webhook_state_ledger
            WHERE webhook_id = $1
              AND state_key_hash = $2
              AND created_at >= (now() - ($3::bigint * interval '1 second'))
            "#,
        )
        .bind(webhook_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 gateway_webhook_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(WebhookGateDecision::OverLimit {
                delivery_id,
                observed_count: observed.0,
            });
        }

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

    tx.commit().await?;
    Ok(WebhookGateDecision::Allowed { delivery_id })
}

pub async fn cleanup_webhook_state_ledger_older_than(
    pool: &PgPool,
    older_than_seconds: i64,
) -> Result<u64, WebhookStoreError> {
    let result = sqlx::query(
        r#"
        DELETE FROM gateway_webhook_state_ledger
        WHERE created_at < (now() - ($1::bigint * interval '1 second'))
        "#,
    )
    .bind(older_than_seconds)
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}

pub async fn list_webhook_deliveries(
    pool: &PgPool,
    webhook_id: i64,
    limit: i64,
    offset: i64,
) -> Result<Vec<GatewayWebhookDeliveryRecord>, WebhookStoreError> {
    let rows: Vec<GatewayWebhookDeliveryRecord> =
        sqlx::query_as::<_, GatewayWebhookDeliveryRecord>(
            r#"
        SELECT
            id, created_at, webhook_id, trigger_route_key, request_id, idempotency_key,
            status, http_status, response_headers_json, response_body_snippet,
            error_message, duration_ms, resolved_url, context_snapshot
        FROM gateway_webhook_delivery
        WHERE webhook_id = $1
        ORDER BY id DESC
        LIMIT $2 OFFSET $3
        "#,
        )
        .bind(webhook_id)
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await?;
    Ok(rows)
}

/// Count webhooks (admin list pagination).
pub async fn count_gateway_webhooks(pool: &PgPool) -> Result<i64, WebhookStoreError> {
    count_gateway_webhooks_filtered(pool, None).await
}

pub async fn count_gateway_webhooks_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 gateway_webhook WHERE client_name = $1")
            .bind(cn)
            .fetch_optional(pool)
            .await?
    } else {
        sqlx::query_as("SELECT COUNT(*)::bigint FROM gateway_webhook")
            .fetch_optional(pool)
            .await?
    }
    .unwrap_or((0,));
    Ok(c.0)
}

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