use actix_web::body::{BoxBody, MessageBody};
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::middleware::Next;
use actix_web::{Error, HttpResponse, Responder, post, web};
use reqwest::StatusCode as ReqwestStatusCode;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use sqlx::{PgPool, Row};
use std::collections::HashMap;
use std::env;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use tracing::{info, warn};
use uuid::Uuid;

use crate::AppState;
use crate::api::client_context::logging_pool;
use crate::api::gateway::auth::{require_admin_or_gateway, typesense_proxy_right};
use crate::api::response::{
    api_success, bad_request, conflict, internal_error, not_found, service_unavailable,
};
use crate::parser::query_builder::{sanitize_identifier, sanitize_qualified_table_identifier};

static TYPESENSE_SCHEMA_WARNED: AtomicBool = AtomicBool::new(false);

const DEFAULT_TYPESENSE_TIMEOUT_MS: u64 = 10_000;
const DEFAULT_TYPESENSE_WORKER_POLL_MS: u64 = 30_000;
const SYNC_LOCK_TTL_MINS: i32 = 15;

#[derive(Debug, Deserialize)]
struct TypesenseConnectionRequest {
    base_url: String,
    api_key: String,
    #[serde(default)]
    timeout_ms: Option<u64>,
}

#[derive(Debug, Deserialize)]
struct TypesenseSearchRequest {
    #[serde(default)]
    job_id: Option<i64>,
    #[serde(default)]
    source_client_name: Option<String>,
    #[serde(default)]
    source_schema_name: Option<String>,
    #[serde(default)]
    source_table_name: Option<String>,
    q: String,
    #[serde(default)]
    query_by: Option<Vec<String>>,
    #[serde(default)]
    filter_by: Option<String>,
    #[serde(default)]
    sort_by: Option<String>,
    #[serde(default)]
    page: Option<u32>,
    #[serde(default)]
    per_page: Option<u32>,
}

#[derive(Debug, Clone)]
struct TypesenseConnection {
    base_url: String,
    api_key: String,
    timeout_ms: u64,
}

#[derive(Debug, Serialize)]
struct TypesenseCollectionSummary {
    name: String,
    num_documents: Option<u64>,
    fields: Vec<String>,
}

#[derive(Debug, Serialize)]
struct TypesenseSyncJobRunResponse {
    job_id: i64,
    run_id: i64,
    status: String,
    collection_name: String,
    source_rows: usize,
    imported_count: usize,
    failed_count: usize,
    duration_ms: u128,
    result_preview: Value,
}

#[derive(Debug, Clone, sqlx::FromRow)]
struct TypesenseFieldMetadata {
    column_name: String,
    data_type: String,
    udt_name: String,
    is_nullable: String,
}

#[allow(dead_code)]
#[derive(Debug, sqlx::FromRow)]
struct TypesenseJobWithProfile {
    id: i64,
    name: String,
    enabled: bool,
    auto_sync_enabled: bool,
    sync_strategy: String,
    source_client_name: String,
    source_schema_name: String,
    source_table_name: String,
    collection_name: String,
    document_id_column: String,
    selected_columns: Value,
    query_by_fields: Value,
    facet_by_fields: Value,
    sort_by_fields: Value,
    field_config: Value,
    default_sort_field: Option<String>,
    batch_size: i32,
    sync_interval_minutes: i32,
    stale_after_minutes: i32,
    typesense_profile_id: i64,
    profile_name: String,
    profile_base_url: String,
    profile_api_key: String,
    profile_search_api_key: Option<String>,
    profile_timeout_ms: i32,
}

#[derive(Debug)]
struct TypesensePreparedSchema {
    schema: Value,
    selected_columns: Vec<String>,
    field_types: HashMap<String, String>,
}

#[derive(Debug)]
struct TypesenseImportSummary {
    imported_count: usize,
    failed_count: usize,
    result_preview: Value,
}

#[derive(Debug, Clone)]
struct TypesenseSyncProgressUpdate {
    source_row_count: usize,
    estimated_total_rows: Option<usize>,
    processed_row_count: usize,
    imported_count: usize,
    failed_count: usize,
    progress_pct: i32,
    progress_stage: String,
    progress_message: Option<String>,
    batches_completed: usize,
}

fn typesense_allow_http() -> bool {
    env::var("ATHENA_TYPESENSE_ALLOW_HTTP")
        .map(|value| matches!(value.trim(), "1" | "true" | "TRUE" | "yes" | "YES"))
        .unwrap_or(false)
}

fn typesense_worker_enabled() -> bool {
    env::var("ATHENA_TYPESENSE_SYNC_WORKER_ENABLED")
        .map(|value| !matches!(value.trim(), "0" | "false" | "FALSE" | "off" | "OFF"))
        .unwrap_or(true)
}

fn typesense_worker_poll_ms() -> u64 {
    env::var("ATHENA_TYPESENSE_SYNC_WORKER_POLL_MS")
        .ok()
        .and_then(|value| value.trim().parse::<u64>().ok())
        .filter(|value| *value >= 1_000)
        .unwrap_or(DEFAULT_TYPESENSE_WORKER_POLL_MS)
}

fn clamp_progress_pct(value: i32) -> i32 {
    value.clamp(0, 100)
}

fn normalize_catalog_row_estimate(raw_estimate: Option<i64>) -> Option<usize> {
    raw_estimate
        .and_then(|value| (value > 0).then_some(value))
        .and_then(|value| usize::try_from(value).ok())
}

fn calculate_import_progress_pct(
    processed_row_count: usize,
    estimated_total_rows: Option<usize>,
) -> i32 {
    let Some(estimated_total_rows) = estimated_total_rows else {
        if processed_row_count == 0 {
            return 15;
        }

        let processed = processed_row_count as f64;
        let heuristic = 15.0 + (75.0 * (1.0 - (-processed / 5_000.0).exp()));
        return heuristic.round().clamp(15.0, 90.0) as i32;
    };

    if estimated_total_rows == 0 {
        return 95;
    }

    let ratio = (processed_row_count as f64 / estimated_total_rows as f64).clamp(0.0, 1.0);
    let pct = 15.0 + ratio * 80.0;
    clamp_progress_pct(pct.round() as i32)
}

fn sqlx_relation_missing(err: &sqlx::Error) -> bool {
    if let sqlx::Error::Database(db_err) = err {
        return matches!(db_err.code().as_deref(), Some("42P01") | Some("3F000"));
    }
    err.to_string().contains("does not exist")
}

fn typesense_not_provisioned_response() -> HttpResponse {
    service_unavailable(
        "Typesense integration not provisioned",
        "Run sql/typesense.sql on the Athena logging database, then retry.",
    )
}

fn normalize_typesense_connection(
    body: &TypesenseConnectionRequest,
) -> Result<TypesenseConnection, HttpResponse> {
    let trimmed_base = body.base_url.trim().trim_end_matches('/');
    if trimmed_base.is_empty() {
        return Err(bad_request("Invalid request", "base_url is required"));
    }
    if body.api_key.trim().is_empty() {
        return Err(bad_request("Invalid request", "api_key is required"));
    }

    let parsed = reqwest::Url::parse(trimmed_base)
        .map_err(|err| bad_request("Invalid request", format!("Invalid base_url: {err}")))?;

    match parsed.scheme() {
        "https" => {}
        "http" if typesense_allow_http() => {}
        "http" => {
            return Err(bad_request(
                "Invalid request",
                "Typesense connections must use HTTPS unless ATHENA_TYPESENSE_ALLOW_HTTP=1 is set for local development.",
            ));
        }
        _ => {
            return Err(bad_request(
                "Invalid request",
                "Typesense base_url must start with http:// or https://.",
            ));
        }
    }

    if parsed.host_str().is_none() || parsed.username() != "" || parsed.password().is_some() {
        return Err(bad_request(
            "Invalid request",
            "Typesense base_url must contain only scheme, host, and optional port.",
        ));
    }

    if parsed.query().is_some() || parsed.fragment().is_some() {
        return Err(bad_request(
            "Invalid request",
            "Typesense base_url must not contain a query string or fragment.",
        ));
    }

    if parsed.path() != "/" && !parsed.path().is_empty() {
        return Err(bad_request(
            "Invalid request",
            "Typesense base_url must not contain a path. Use the root server URL.",
        ));
    }

    Ok(TypesenseConnection {
        base_url: trimmed_base.to_string(),
        api_key: body.api_key.trim().to_string(),
        timeout_ms: body
            .timeout_ms
            .filter(|value| *value >= 1_000)
            .unwrap_or(DEFAULT_TYPESENSE_TIMEOUT_MS),
    })
}

fn string_vec_from_json(value: &Value) -> Vec<String> {
    value
        .as_array()
        .map(|items| {
            items
                .iter()
                .filter_map(Value::as_str)
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToString::to_string)
                .collect::<Vec<String>>()
        })
        .unwrap_or_default()
}

fn stringify_value(value: &Value) -> String {
    match value {
        Value::Null => String::new(),
        Value::String(text) => text.clone(),
        Value::Bool(flag) => flag.to_string(),
        Value::Number(number) => number.to_string(),
        _ => value.to_string(),
    }
}

fn document_id_from_value(value: &Value) -> Option<String> {
    let text = stringify_value(value);
    if text.trim().is_empty() {
        None
    } else {
        Some(text)
    }
}

fn field_override<'a>(field_config: &'a Value, name: &str) -> Option<&'a Map<String, Value>> {
    field_config.as_object()?.get(name)?.as_object()
}

fn map_postgres_type_to_typesense(field: &TypesenseFieldMetadata) -> String {
    match field.data_type.as_str() {
        "smallint" | "integer" => "int32".to_string(),
        "bigint" => "int64".to_string(),
        "real" | "double precision" | "numeric" => "float".to_string(),
        "boolean" => "bool".to_string(),
        "ARRAY" => match field.udt_name.as_str() {
            "_int2" | "_int4" => "int32[]".to_string(),
            "_int8" => "int64[]".to_string(),
            "_float4" | "_float8" | "_numeric" => "float[]".to_string(),
            "_bool" => "bool[]".to_string(),
            _ => "string[]".to_string(),
        },
        _ => "string".to_string(),
    }
}

fn normalize_document_value(value: Value, typesense_type: &str) -> Value {
    match typesense_type {
        "string" => Value::String(stringify_value(&value)),
        "string[]" => match value {
            Value::Array(items) => Value::Array(
                items
                    .into_iter()
                    .map(|item| Value::String(stringify_value(&item)))
                    .collect(),
            ),
            Value::Null => Value::Array(Vec::new()),
            other => Value::Array(vec![Value::String(stringify_value(&other))]),
        },
        _ => value,
    }
}

async fn typesense_get_json(
    state: &AppState,
    connection: &TypesenseConnection,
    path: &str,
    query: &[(&str, String)],
) -> Result<(ReqwestStatusCode, Value), String> {
    let mut request = state
        .client
        .get(format!("{}{}", connection.base_url, path))
        .timeout(Duration::from_millis(connection.timeout_ms))
        .header("X-TYPESENSE-API-KEY", &connection.api_key);

    if !query.is_empty() {
        request = request.query(query);
    }

    let response = request
        .send()
        .await
        .map_err(|err| format!("Typesense request failed: {err}"))?;
    let status = response.status();
    let payload = response.json::<Value>().await.unwrap_or_else(|_| json!({}));
    Ok((status, payload))
}

async fn typesense_post_json(
    state: &AppState,
    connection: &TypesenseConnection,
    path: &str,
    query: &[(&str, String)],
    body: &Value,
) -> Result<(ReqwestStatusCode, Value), String> {
    let mut request = state
        .client
        .post(format!("{}{}", connection.base_url, path))
        .timeout(Duration::from_millis(connection.timeout_ms))
        .header("X-TYPESENSE-API-KEY", &connection.api_key)
        .json(body);

    if !query.is_empty() {
        request = request.query(query);
    }

    let response = request
        .send()
        .await
        .map_err(|err| format!("Typesense request failed: {err}"))?;
    let status = response.status();
    let payload = response.json::<Value>().await.unwrap_or_else(|_| json!({}));
    Ok((status, payload))
}

async fn typesense_post_text(
    state: &AppState,
    connection: &TypesenseConnection,
    path: &str,
    query: &[(&str, String)],
    body: String,
) -> Result<(ReqwestStatusCode, String), String> {
    let mut request = state
        .client
        .post(format!("{}{}", connection.base_url, path))
        .timeout(Duration::from_millis(connection.timeout_ms))
        .header("X-TYPESENSE-API-KEY", &connection.api_key)
        .header("Content-Type", "text/plain")
        .body(body);

    if !query.is_empty() {
        request = request.query(query);
    }

    let response = request
        .send()
        .await
        .map_err(|err| format!("Typesense request failed: {err}"))?;
    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|err| format!("Failed to read Typesense response: {err}"))?;
    Ok((status, text))
}

async fn typesense_healthcheck(
    state: &AppState,
    connection: &TypesenseConnection,
) -> Result<(u128, Value), String> {
    let started = Instant::now();
    let (status, payload) = typesense_get_json(state, connection, "/health", &[]).await?;
    if !status.is_success() {
        return Err(format!(
            "Typesense health check failed with HTTP {}: {}",
            status.as_u16(),
            payload
        ));
    }
    Ok((started.elapsed().as_millis(), payload))
}

fn typesense_connection_from_job(job: &TypesenseJobWithProfile) -> TypesenseConnection {
    TypesenseConnection {
        base_url: job.profile_base_url.clone(),
        api_key: job.profile_api_key.clone(),
        timeout_ms: u64::try_from(job.profile_timeout_ms.max(1))
            .unwrap_or(DEFAULT_TYPESENSE_TIMEOUT_MS),
    }
}

fn logging_pool_from_state(state: &AppState) -> Option<PgPool> {
    let client_name = state.logging_client_name.as_ref()?;
    state.pg_registry.get_pool(client_name)
}

async fn load_typesense_job(
    pool: &PgPool,
    job_id: i64,
) -> Result<Option<TypesenseJobWithProfile>, sqlx::Error> {
    sqlx::query_as::<_, TypesenseJobWithProfile>(
        r#"
        SELECT
            j.id,
            j.name,
            j.enabled,
            j.auto_sync_enabled,
            j.sync_strategy,
            j.source_client_name,
            j.source_schema_name,
            j.source_table_name,
            j.collection_name,
            j.document_id_column,
            j.selected_columns,
            j.query_by_fields,
            j.facet_by_fields,
            j.sort_by_fields,
            j.field_config,
            j.default_sort_field,
            j.batch_size,
            j.sync_interval_minutes,
            j.stale_after_minutes,
            j.typesense_profile_id,
            p.name AS profile_name,
            p.base_url AS profile_base_url,
            p.api_key AS profile_api_key,
            p.search_api_key AS profile_search_api_key,
            p.timeout_ms AS profile_timeout_ms
        FROM public.typesense_sync_jobs j
        JOIN public.typesense_profiles p
            ON p.id = j.typesense_profile_id
        WHERE j.id = $1
        "#,
    )
    .bind(job_id)
    .fetch_optional(pool)
    .await
}

async fn due_job_ids(pool: &PgPool) -> Result<Vec<i64>, sqlx::Error> {
    let rows = sqlx::query(
        r#"
        SELECT id
        FROM public.typesense_sync_jobs
        WHERE enabled = true
          AND auto_sync_enabled = true
          AND (next_sync_at IS NULL OR next_sync_at <= now())
          AND (sync_lock_expires_at IS NULL OR sync_lock_expires_at < now())
        ORDER BY COALESCE(next_sync_at, created_at) ASC
        LIMIT 5
        "#,
    )
    .fetch_all(pool)
    .await?;

    Ok(rows
        .into_iter()
        .filter_map(|row| row.try_get::<i64, _>("id").ok())
        .collect())
}

async fn claim_sync_job(pool: &PgPool, job_id: i64) -> Result<bool, sqlx::Error> {
    let lock_token = Uuid::new_v4();
    let claimed = sqlx::query(
        r#"
        UPDATE public.typesense_sync_jobs
        SET
            updated_at = now(),
            last_sync_status = 'running',
            last_sync_started_at = now(),
            sync_lock_token = $2,
            sync_locked_at = now(),
            sync_lock_expires_at = now() + make_interval(mins => $3::int)
        WHERE id = $1
          AND enabled = true
          AND (sync_lock_expires_at IS NULL OR sync_lock_expires_at < now())
        RETURNING id
        "#,
    )
    .bind(job_id)
    .bind(lock_token)
    .bind(SYNC_LOCK_TTL_MINS)
    .fetch_optional(pool)
    .await?;

    Ok(claimed.is_some())
}

async fn create_sync_run(
    pool: &PgPool,
    job_id: i64,
    trigger_source: &str,
) -> Result<i64, sqlx::Error> {
    let row = sqlx::query(
        r#"
        INSERT INTO public.typesense_sync_runs (
            job_id,
            trigger_source,
            status,
            progress_pct,
            progress_stage,
            progress_message
        )
        VALUES ($1, $2, 'running', 0, 'starting', 'Claimed sync job and preparing run.')
        RETURNING id
        "#,
    )
    .bind(job_id)
    .bind(trigger_source)
    .fetch_one(pool)
    .await?;
    row.try_get("id")
}

async fn update_sync_run_progress(
    pool: &PgPool,
    run_id: i64,
    progress: &TypesenseSyncProgressUpdate,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            source_row_count = $2,
            estimated_total_rows = $3,
            processed_row_count = $4,
            imported_count = $5,
            failed_count = $6,
            progress_pct = $7,
            progress_stage = $8,
            progress_message = $9,
            batches_completed = $10
        WHERE id = $1
        "#,
    )
    .bind(run_id)
    .bind(i32::try_from(progress.source_row_count).unwrap_or(i32::MAX))
    .bind(
        progress
            .estimated_total_rows
            .map(|value| i32::try_from(value).unwrap_or(i32::MAX)),
    )
    .bind(i32::try_from(progress.processed_row_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(progress.imported_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(progress.failed_count).unwrap_or(i32::MAX))
    .bind(clamp_progress_pct(progress.progress_pct))
    .bind(progress.progress_stage.as_str())
    .bind(progress.progress_message.as_deref())
    .bind(i32::try_from(progress.batches_completed).unwrap_or(i32::MAX))
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_sync_run_success(
    pool: &PgPool,
    run_id: i64,
    source_row_count: usize,
    imported_count: usize,
    failed_count: usize,
    batches_completed: usize,
    duration_ms: u128,
    result_preview: &Value,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            status = 'success',
            finished_at = now(),
            source_row_count = $2,
            estimated_total_rows = $3,
            processed_row_count = $2,
            imported_count = $4,
            failed_count = $5,
            progress_pct = 100,
            progress_stage = 'completed',
            progress_message = CASE
                WHEN $2 = 0 THEN 'Completed sync with no source rows to import.'
                WHEN $5 > 0 THEN format('Completed sync with %s imported and %s failed rows.', $4, $5)
                ELSE format('Completed sync with %s imported rows.', $4)
            END,
            batches_completed = CASE
                WHEN COALESCE($6, 0) <= 0 THEN batches_completed
                ELSE GREATEST(batches_completed, 1)
            END,
            duration_ms = $7,
            result_preview = $8
        WHERE id = $1
        "#,
    )
    .bind(run_id)
    .bind(i32::try_from(source_row_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(source_row_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(imported_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(failed_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(batches_completed).unwrap_or(i32::MAX))
    .bind(i64::try_from(duration_ms).unwrap_or(i64::MAX))
    .bind(result_preview)
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_sync_run_failure(
    pool: &PgPool,
    run_id: i64,
    error_message: &str,
    duration_ms: u128,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            status = 'failed',
            finished_at = now(),
            duration_ms = $2,
            progress_stage = 'failed',
            progress_message = $3,
            error_message = $3
        WHERE id = $1
        "#,
    )
    .bind(run_id)
    .bind(i64::try_from(duration_ms).unwrap_or(i64::MAX))
    .bind(error_message)
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_job_success(
    pool: &PgPool,
    job: &TypesenseJobWithProfile,
    source_row_count: usize,
    imported_count: usize,
    failed_count: usize,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_jobs
        SET
            updated_at = now(),
            last_sync_status = CASE WHEN enabled THEN 'success' ELSE 'paused' END,
            last_sync_finished_at = now(),
            last_synced_at = now(),
            last_sync_row_count = $2,
            last_sync_success_count = $3,
            last_sync_error_count = $4,
            last_sync_error = NULL,
            next_sync_at = CASE
                WHEN auto_sync_enabled THEN now() + make_interval(mins => GREATEST(sync_interval_minutes, 1)::int)
                ELSE next_sync_at
            END,
            sync_lock_token = NULL,
            sync_locked_at = NULL,
            sync_lock_expires_at = NULL
        WHERE id = $1
        "#,
    )
    .bind(job.id)
    .bind(i32::try_from(source_row_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(imported_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(failed_count).unwrap_or(i32::MAX))
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_job_failure(
    pool: &PgPool,
    job_id: i64,
    error_message: &str,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_jobs
        SET
            updated_at = now(),
            last_sync_status = 'failed',
            last_sync_finished_at = now(),
            last_sync_error = $2,
            next_sync_at = CASE
                WHEN auto_sync_enabled THEN now() + make_interval(mins => GREATEST(sync_interval_minutes, 1)::int)
                ELSE next_sync_at
            END,
            sync_lock_token = NULL,
            sync_locked_at = NULL,
            sync_lock_expires_at = NULL
        WHERE id = $1
        "#,
    )
    .bind(job_id)
    .bind(error_message)
    .execute(pool)
    .await?;
    Ok(())
}

async fn fetch_source_field_metadata(
    source_pool: &PgPool,
    schema_name: &str,
    table_name: &str,
) -> Result<Vec<TypesenseFieldMetadata>, sqlx::Error> {
    sqlx::query_as::<_, TypesenseFieldMetadata>(
        r#"
        SELECT
            column_name,
            data_type,
            udt_name,
            is_nullable
        FROM information_schema.columns
        WHERE table_schema = $1
          AND table_name = $2
        ORDER BY ordinal_position ASC
        "#,
    )
    .bind(schema_name)
    .bind(table_name)
    .fetch_all(source_pool)
    .await
}

fn build_typesense_schema(
    job: &TypesenseJobWithProfile,
    fields: &[TypesenseFieldMetadata],
) -> Result<TypesensePreparedSchema, String> {
    let mut field_by_name: HashMap<String, TypesenseFieldMetadata> = HashMap::new();
    for field in fields {
        field_by_name.insert(field.column_name.clone(), field.clone());
    }

    let mut selected_columns = string_vec_from_json(&job.selected_columns);
    if selected_columns.is_empty() {
        selected_columns = fields
            .iter()
            .map(|field| field.column_name.clone())
            .collect();
    }

    if !selected_columns
        .iter()
        .any(|column| column == &job.document_id_column)
    {
        selected_columns.push(job.document_id_column.clone());
    }

    let mut typesense_fields = Vec::new();
    let mut field_types = HashMap::new();

    typesense_fields.push(json!({
        "name": "id",
        "type": "string"
    }));
    field_types.insert("id".to_string(), "string".to_string());

    for column in &selected_columns {
        let field = field_by_name.get(column).ok_or_else(|| {
            format!(
                "Selected column '{}' does not exist on the source table.",
                column
            )
        })?;
        let override_map = field_override(&job.field_config, column);
        let field_type = override_map
            .and_then(|map| map.get("type"))
            .and_then(Value::as_str)
            .map(ToString::to_string)
            .unwrap_or_else(|| map_postgres_type_to_typesense(field));

        let mut entry = Map::new();
        entry.insert("name".to_string(), Value::String(column.clone()));
        entry.insert("type".to_string(), Value::String(field_type.clone()));
        entry.insert(
            "optional".to_string(),
            Value::Bool(
                override_map
                    .and_then(|map| map.get("optional"))
                    .and_then(Value::as_bool)
                    .unwrap_or(field.is_nullable.eq_ignore_ascii_case("YES")),
            ),
        );
        if let Some(facet) = override_map
            .and_then(|map| map.get("facet"))
            .and_then(Value::as_bool)
        {
            entry.insert("facet".to_string(), Value::Bool(facet));
        }
        if let Some(sort) = override_map
            .and_then(|map| map.get("sort"))
            .and_then(Value::as_bool)
        {
            entry.insert("sort".to_string(), Value::Bool(sort));
        }
        typesense_fields.push(Value::Object(entry));
        field_types.insert(column.clone(), field_type);
    }

    let mut schema = Map::new();
    schema.insert(
        "name".to_string(),
        Value::String(job.collection_name.clone()),
    );
    schema.insert("fields".to_string(), Value::Array(typesense_fields));
    if let Some(default_sort_field) = job.default_sort_field.as_ref() {
        schema.insert(
            "default_sorting_field".to_string(),
            Value::String(default_sort_field.clone()),
        );
    }

    Ok(TypesensePreparedSchema {
        schema: Value::Object(schema),
        selected_columns,
        field_types,
    })
}

async fn ensure_collection_exists(
    state: &AppState,
    connection: &TypesenseConnection,
    prepared_schema: &TypesensePreparedSchema,
    collection_name: &str,
) -> Result<(), String> {
    let (status, _) = typesense_get_json(
        state,
        connection,
        &format!("/collections/{collection_name}"),
        &[],
    )
    .await?;

    if status == ReqwestStatusCode::NOT_FOUND {
        let (create_status, payload) = typesense_post_json(
            state,
            connection,
            "/collections",
            &[],
            &prepared_schema.schema,
        )
        .await?;
        if !create_status.is_success() {
            return Err(format!(
                "Failed to create Typesense collection '{}': {}",
                collection_name, payload
            ));
        }
    } else if !status.is_success() {
        return Err(format!(
            "Failed to inspect Typesense collection '{}': HTTP {}",
            collection_name,
            status.as_u16()
        ));
    }

    Ok(())
}

async fn fetch_source_documents(
    source_pool: &PgPool,
    schema_name: &str,
    table_name: &str,
    selected_columns: &[String],
    document_id_column: &str,
    batch_size: i32,
    offset: i64,
) -> Result<Vec<Value>, String> {
    let qualified_table =
        sanitize_qualified_table_identifier(&format!("{schema_name}.{table_name}"))
            .ok_or_else(|| "Invalid source schema/table binding.".to_string())?;
    let mut quoted_columns = Vec::new();
    for column in selected_columns {
        quoted_columns.push(
            sanitize_identifier(column)
                .ok_or_else(|| format!("Invalid selected column '{}'.", column))?,
        );
    }
    let document_id_identifier = sanitize_identifier(document_id_column)
        .ok_or_else(|| format!("Invalid document id column '{}'.", document_id_column))?;

    let sql = format!(
        "SELECT to_jsonb(t) AS row FROM (SELECT {} FROM {} ORDER BY {} LIMIT $1 OFFSET $2) t",
        quoted_columns.join(", "),
        qualified_table,
        document_id_identifier,
    );

    let rows = sqlx::query(&sql)
        .bind(i64::from(batch_size.max(1)))
        .bind(offset)
        .fetch_all(source_pool)
        .await
        .map_err(|err| format!("Failed to read source rows: {err}"))?;

    let mut decoded_rows = Vec::with_capacity(rows.len());
    for row in rows {
        let value: Value = row
            .try_get("row")
            .map_err(|err| format!("Failed to decode source rows: {err}"))?;
        decoded_rows.push(value);
    }

    Ok(decoded_rows)
}

async fn estimate_source_documents(
    source_pool: &PgPool,
    schema_name: &str,
    table_name: &str,
) -> Result<Option<usize>, String> {
    let row = sqlx::query(
        r#"
        SELECT GREATEST(c.reltuples, 0)::bigint AS estimated_total_rows
        FROM pg_class c
        JOIN pg_namespace n
            ON n.oid = c.relnamespace
        WHERE n.nspname = $1
          AND c.relname = $2
          AND c.relkind IN ('r', 'p', 'm')
        LIMIT 1
        "#,
    )
    .bind(schema_name)
    .bind(table_name)
    .fetch_optional(source_pool)
    .await
    .map_err(|err| format!("Failed to estimate source rows from catalog statistics: {err}"))?;

    let Some(row) = row else {
        return Ok(None);
    };

    let estimate: i64 = row
        .try_get("estimated_total_rows")
        .map_err(|err| format!("Failed to decode catalog row estimate: {err}"))?;
    Ok(normalize_catalog_row_estimate(Some(estimate)))
}

fn rows_to_ndjson(
    rows: Vec<Value>,
    document_id_column: &str,
    field_types: &HashMap<String, String>,
) -> Result<String, String> {
    let mut lines = Vec::new();

    for row in rows {
        let Value::Object(mut map) = row else {
            continue;
        };

        let id_value = map.get(document_id_column).ok_or_else(|| {
            format!(
                "Document id column '{}' is missing from a source row.",
                document_id_column
            )
        })?;
        let id = document_id_from_value(id_value).ok_or_else(|| {
            format!(
                "Document id column '{}' produced an empty id for at least one row.",
                document_id_column
            )
        })?;
        map.insert("id".to_string(), Value::String(id));

        for (field_name, field_type) in field_types {
            if let Some(existing) = map.get(field_name).cloned() {
                map.insert(
                    field_name.clone(),
                    normalize_document_value(existing, field_type),
                );
            }
        }

        lines.push(
            serde_json::to_string(&Value::Object(map))
                .map_err(|err| format!("Failed to serialize Typesense document: {err}"))?,
        );
    }

    Ok(lines.join("\n"))
}

async fn import_documents(
    state: &AppState,
    connection: &TypesenseConnection,
    collection_name: &str,
    ndjson: String,
) -> Result<TypesenseImportSummary, String> {
    if ndjson.trim().is_empty() {
        return Ok(TypesenseImportSummary {
            imported_count: 0,
            failed_count: 0,
            result_preview: json!([]),
        });
    }

    let (status, text) = typesense_post_text(
        state,
        connection,
        &format!("/collections/{collection_name}/documents/import"),
        &[("action", "upsert".to_string())],
        ndjson,
    )
    .await?;

    if !status.is_success() {
        return Err(format!(
            "Typesense import failed with HTTP {}: {}",
            status.as_u16(),
            text
        ));
    }

    let mut imported_count = 0usize;
    let mut failed_count = 0usize;
    let mut preview = Vec::new();

    for (index, line) in text.lines().enumerate() {
        if line.trim().is_empty() {
            continue;
        }
        let parsed = serde_json::from_str::<Value>(line).unwrap_or_else(|_| json!({ "raw": line }));
        if parsed
            .get("success")
            .and_then(Value::as_bool)
            .unwrap_or(false)
        {
            imported_count += 1;
        } else {
            failed_count += 1;
        }
        if index < 25 {
            preview.push(parsed);
        }
    }

    Ok(TypesenseImportSummary {
        imported_count,
        failed_count,
        result_preview: Value::Array(preview),
    })
}

async fn execute_typesense_sync_job(
    state: &AppState,
    job_id: i64,
    trigger_source: &str,
) -> Result<TypesenseSyncJobRunResponse, String> {
    let logging = logging_pool_from_state(state)
        .ok_or_else(|| "Logging store unavailable for Typesense sync jobs.".to_string())?;
    if !claim_sync_job(&logging, job_id)
        .await
        .map_err(|err| err.to_string())?
    {
        return Err("Sync job is already running or is disabled.".to_string());
    }

    let run_id = create_sync_run(&logging, job_id, trigger_source)
        .await
        .map_err(|err| err.to_string())?;
    let started = Instant::now();

    let execute_result = async {
        let job = load_typesense_job(&logging, job_id)
            .await
            .map_err(|err| err.to_string())?
            .ok_or_else(|| "Sync job not found.".to_string())?;
        update_sync_run_progress(
            &logging,
            run_id,
            &TypesenseSyncProgressUpdate {
                source_row_count: 0,
                estimated_total_rows: None,
                processed_row_count: 0,
                imported_count: 0,
                failed_count: 0,
                progress_pct: 5,
                progress_stage: "loading_source".to_string(),
                progress_message: Some(format!(
                    "Loading source client '{}' and inspecting sync job definition.",
                    job.source_client_name
                )),
                batches_completed: 0,
            },
        )
        .await
        .map_err(|err| err.to_string())?;

        let source_pool = state
            .pg_registry
            .get_pool(&job.source_client_name)
            .ok_or_else(|| {
                format!(
                    "Source Athena client '{}' is not currently connected.",
                    job.source_client_name
                )
            })?;
        let field_metadata = fetch_source_field_metadata(
            &source_pool,
            &job.source_schema_name,
            &job.source_table_name,
        )
        .await
        .map_err(|err| format!("Failed to inspect source table metadata: {err}"))?;
        update_sync_run_progress(
            &logging,
            run_id,
            &TypesenseSyncProgressUpdate {
                source_row_count: 0,
                estimated_total_rows: None,
                processed_row_count: 0,
                imported_count: 0,
                failed_count: 0,
                progress_pct: 10,
                progress_stage: "preparing_schema".to_string(),
                progress_message: Some(format!(
                    "Loaded {} source columns from {}.{}.",
                    field_metadata.len(),
                    job.source_schema_name,
                    job.source_table_name
                )),
                batches_completed: 0,
            },
        )
        .await
        .map_err(|err| err.to_string())?;
        let prepared_schema = build_typesense_schema(&job, &field_metadata)?;
        let estimated_total_rows = estimate_source_documents(
            &source_pool,
            &job.source_schema_name,
            &job.source_table_name,
        )
        .await?;
        update_sync_run_progress(
            &logging,
            run_id,
            &TypesenseSyncProgressUpdate {
                source_row_count: estimated_total_rows.unwrap_or(0),
                estimated_total_rows,
                processed_row_count: 0,
                imported_count: 0,
                failed_count: 0,
                progress_pct: 12,
                progress_stage: "estimating_source_rows".to_string(),
                progress_message: Some(match estimated_total_rows {
                    Some(value) => format!(
                        "Estimated about {} source rows from catalog statistics and prepared the collection schema.",
                        value
                    ),
                    None => "Source row estimate unavailable from catalog statistics; continuing without an estimated total."
                        .to_string(),
                }),
                batches_completed: 0,
            },
        )
        .await
        .map_err(|err| err.to_string())?;
        let connection = typesense_connection_from_job(&job);
        ensure_collection_exists(state, &connection, &prepared_schema, &job.collection_name)
            .await?;
        update_sync_run_progress(
            &logging,
            run_id,
            &TypesenseSyncProgressUpdate {
                source_row_count: estimated_total_rows.unwrap_or(0),
                estimated_total_rows,
                processed_row_count: 0,
                imported_count: 0,
                failed_count: 0,
                progress_pct: 15,
                progress_stage: "importing_documents".to_string(),
                progress_message: Some(format!(
                    "Collection '{}' is ready. Starting document import.",
                    job.collection_name
                )),
                batches_completed: 0,
            },
        )
        .await
        .map_err(|err| err.to_string())?;

        let batch_size = job.batch_size.max(1);
        let mut offset = 0_i64;
        let mut source_row_count = 0usize;
        let mut imported_count = 0usize;
        let mut failed_count = 0usize;
        let mut preview = Vec::new();
        let mut batches_completed = 0usize;
        let total_batches = estimated_total_rows.map(|rows| {
            if rows == 0 {
                0usize
            } else {
                rows.div_ceil(usize::try_from(batch_size).unwrap_or(1))
            }
        });

        loop {
            let documents = fetch_source_documents(
                &source_pool,
                &job.source_schema_name,
                &job.source_table_name,
                &prepared_schema.selected_columns,
                &job.document_id_column,
                batch_size,
                offset,
            )
            .await?;

            if documents.is_empty() {
                break;
            }

            let current_batch_len = documents.len();
            source_row_count += current_batch_len;
            let ndjson = rows_to_ndjson(
                documents,
                &job.document_id_column,
                &prepared_schema.field_types,
            )?;
            let batch_summary =
                import_documents(state, &connection, &job.collection_name, ndjson).await?;
            imported_count += batch_summary.imported_count;
            failed_count += batch_summary.failed_count;
            batches_completed += 1;

            if preview.len() < 25 {
                let remaining = 25 - preview.len();
                if let Some(items) = batch_summary.result_preview.as_array() {
                    preview.extend(items.iter().take(remaining).cloned());
                }
            }

            update_sync_run_progress(
                &logging,
                run_id,
                &TypesenseSyncProgressUpdate {
                    source_row_count,
                    estimated_total_rows,
                    processed_row_count: source_row_count,
                    imported_count,
                    failed_count,
                    progress_pct: calculate_import_progress_pct(source_row_count, estimated_total_rows),
                    progress_stage: "importing_documents".to_string(),
                    progress_message: Some(match total_batches {
                        Some(value) if value > 0 => format!(
                            "Imported batch {} of {} into '{}'.",
                            batches_completed, value, job.collection_name
                        ),
                        _ => format!(
                            "Imported {} source rows into '{}'.",
                            source_row_count, job.collection_name
                        ),
                    }),
                    batches_completed,
                },
            )
            .await
            .map_err(|err| err.to_string())?;

            if current_batch_len < usize::try_from(batch_size).unwrap_or(usize::MAX) {
                break;
            }

            offset += i64::from(batch_size);
        }

        let duration_ms = started.elapsed().as_millis();
        let result_preview = Value::Array(preview);

        update_sync_run_success(
            &logging,
            run_id,
            source_row_count,
            imported_count,
            failed_count,
            batches_completed,
            duration_ms,
            &result_preview,
        )
        .await
        .map_err(|err| err.to_string())?;
        update_job_success(
            &logging,
            &job,
            source_row_count,
            imported_count,
            failed_count,
        )
        .await
        .map_err(|err| err.to_string())?;

        Ok::<TypesenseSyncJobRunResponse, String>(TypesenseSyncJobRunResponse {
            job_id: job.id,
            run_id,
            status: "success".to_string(),
            collection_name: job.collection_name,
            source_rows: source_row_count,
            imported_count,
            failed_count,
            duration_ms,
            result_preview,
        })
    }
    .await;

    if let Err(err) = &execute_result {
        let duration_ms = started.elapsed().as_millis();
        let _ = update_sync_run_failure(&logging, run_id, err, duration_ms).await;
        let _ = update_job_failure(&logging, job_id, err).await;
    }

    execute_result
}

#[cfg(test)]
mod tests {
    use super::{calculate_import_progress_pct, normalize_catalog_row_estimate};

    #[test]
    fn import_progress_pct_handles_zero_total_rows() {
        assert_eq!(calculate_import_progress_pct(0, Some(0)), 95);
    }

    #[test]
    fn import_progress_pct_respects_known_total_range() {
        assert_eq!(calculate_import_progress_pct(0, Some(100)), 15);
        assert_eq!(calculate_import_progress_pct(50, Some(100)), 55);
        assert_eq!(calculate_import_progress_pct(100, Some(100)), 95);
        assert_eq!(calculate_import_progress_pct(150, Some(100)), 95);
    }

    #[test]
    fn import_progress_pct_with_unknown_total_is_bounded() {
        let empty = calculate_import_progress_pct(0, None);
        let in_progress = calculate_import_progress_pct(1_000, None);
        let large = calculate_import_progress_pct(5_000_000, None);
        assert_eq!(empty, 15);
        assert!(in_progress > empty);
        assert!(in_progress <= 90);
        assert!(large <= 90);
    }

    #[test]
    fn normalize_catalog_row_estimate_maps_expected_values() {
        assert_eq!(normalize_catalog_row_estimate(None), None);
        assert_eq!(normalize_catalog_row_estimate(Some(-1)), None);
        assert_eq!(normalize_catalog_row_estimate(Some(0)), None);
        assert_eq!(normalize_catalog_row_estimate(Some(42)), Some(42));
    }
}

async fn resolve_job_for_search(
    pool: &PgPool,
    request: &TypesenseSearchRequest,
) -> Result<Option<TypesenseJobWithProfile>, sqlx::Error> {
    if let Some(job_id) = request.job_id {
        return load_typesense_job(pool, job_id).await;
    }

    let Some(source_client_name) = request.source_client_name.as_ref() else {
        return Ok(None);
    };
    let Some(source_table_name) = request.source_table_name.as_ref() else {
        return Ok(None);
    };
    let schema_name = request.source_schema_name.as_deref().unwrap_or("public");

    sqlx::query_as::<_, TypesenseJobWithProfile>(
        r#"
        SELECT
            j.id,
            j.name,
            j.enabled,
            j.auto_sync_enabled,
            j.sync_strategy,
            j.source_client_name,
            j.source_schema_name,
            j.source_table_name,
            j.collection_name,
            j.document_id_column,
            j.selected_columns,
            j.query_by_fields,
            j.facet_by_fields,
            j.sort_by_fields,
            j.field_config,
            j.default_sort_field,
            j.batch_size,
            j.sync_interval_minutes,
            j.stale_after_minutes,
            j.typesense_profile_id,
            p.name AS profile_name,
            p.base_url AS profile_base_url,
            p.api_key AS profile_api_key,
            p.search_api_key AS profile_search_api_key,
            p.timeout_ms AS profile_timeout_ms
        FROM public.typesense_sync_jobs j
        JOIN public.typesense_profiles p
            ON p.id = j.typesense_profile_id
        WHERE j.enabled = true
          AND j.source_client_name = $1
          AND j.source_schema_name = $2
          AND j.source_table_name = $3
        ORDER BY j.auto_sync_enabled DESC, j.updated_at DESC, j.created_at DESC
        LIMIT 1
        "#,
    )
    .bind(source_client_name)
    .bind(schema_name)
    .bind(source_table_name)
    .fetch_optional(pool)
    .await
}

async fn typesense_auth_middleware<B>(
    req: ServiceRequest,
    next: Next<B>,
) -> Result<ServiceResponse<BoxBody>, Error>
where
    B: MessageBody + 'static,
{
    if req.path().starts_with("/typesense/") {
        let Some(data) = req.app_data::<web::Data<AppState>>() else {
            return Err(actix_web::error::ErrorInternalServerError(
                "AppState not configured",
            ));
        };
        let state: &AppState = data.get_ref();
        if let Err(http_resp) =
            require_admin_or_gateway(req.request(), state, None, vec![typesense_proxy_right()])
                .await
        {
            return Ok(req.into_response(http_resp.map_into_boxed_body()));
        }
    }

    Ok(next.call(req).await?.map_into_boxed_body())
}

#[post("/backends/test")]
async fn test_typesense_backend(
    body: web::Json<TypesenseConnectionRequest>,
    app_state: web::Data<AppState>,
) -> impl Responder {
    let connection = match normalize_typesense_connection(&body) {
        Ok(connection) => connection,
        Err(resp) => return resp,
    };

    match typesense_healthcheck(app_state.get_ref(), &connection).await {
        Ok((latency_ms, payload)) => api_success(
            "Typesense backend is reachable",
            json!({
                "ok": true,
                "latency_ms": latency_ms,
                "base_url": connection.base_url,
                "health": payload,
            }),
        ),
        Err(err) => internal_error("Typesense backend check failed", err),
    }
}

#[post("/collections/list")]
async fn list_typesense_collections(
    body: web::Json<TypesenseConnectionRequest>,
    app_state: web::Data<AppState>,
) -> impl Responder {
    let connection = match normalize_typesense_connection(&body) {
        Ok(connection) => connection,
        Err(resp) => return resp,
    };

    let started = Instant::now();
    let (status, payload) =
        match typesense_get_json(app_state.get_ref(), &connection, "/collections", &[]).await {
            Ok(result) => result,
            Err(err) => return internal_error("Failed to list Typesense collections", err),
        };

    if !status.is_success() {
        return internal_error(
            "Failed to list Typesense collections",
            format!("Typesense returned HTTP {}: {}", status.as_u16(), payload),
        );
    }

    let collections = payload
        .as_array()
        .cloned()
        .unwrap_or_default()
        .into_iter()
        .map(|entry| TypesenseCollectionSummary {
            name: entry
                .get("name")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_string(),
            num_documents: entry.get("num_documents").and_then(Value::as_u64),
            fields: entry
                .get("fields")
                .and_then(Value::as_array)
                .map(|items| {
                    items
                        .iter()
                        .filter_map(|field| field.get("name").and_then(Value::as_str))
                        .map(ToString::to_string)
                        .collect::<Vec<String>>()
                })
                .unwrap_or_default(),
        })
        .collect::<Vec<TypesenseCollectionSummary>>();

    api_success(
        "Typesense collections loaded",
        json!({
            "count": collections.len(),
            "latency_ms": started.elapsed().as_millis(),
            "collections": collections,
        }),
    )
}

#[post("/sync-jobs/{job_id}/run")]
async fn run_typesense_sync_job_route(
    path: web::Path<i64>,
    app_state: web::Data<AppState>,
) -> impl Responder {
    let Some(logging) = logging_pool_from_state(app_state.get_ref()) else {
        return service_unavailable(
            "Logging store unavailable",
            "Configure gateway.logging_client and ensure it is connected before running Typesense sync jobs.",
        );
    };

    match load_typesense_job(&logging, *path).await {
        Ok(Some(_)) => {}
        Ok(None) => {
            return not_found(
                "Typesense sync job not found",
                format!("No Typesense sync job exists for id {}", *path),
            );
        }
        Err(err) if sqlx_relation_missing(&err) => return typesense_not_provisioned_response(),
        Err(err) => return internal_error("Failed to load Typesense sync job", err.to_string()),
    }

    match execute_typesense_sync_job(app_state.get_ref(), *path, "manual").await {
        Ok(result) => api_success("Typesense sync completed", json!(result)),
        Err(err) if err.contains("already running") => {
            conflict("Typesense sync job is already running", err)
        }
        Err(err) => internal_error("Typesense sync failed", err),
    }
}

#[post("/search")]
async fn typesense_search_route(
    body: web::Json<TypesenseSearchRequest>,
    app_state: web::Data<AppState>,
) -> impl Responder {
    if body.q.trim().is_empty() {
        return bad_request("Invalid request", "q is required");
    }

    let pool = match logging_pool(&app_state) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };
    let job = match resolve_job_for_search(&pool, &body).await {
        Ok(Some(job)) => job,
        Ok(None) => {
            return not_found(
                "Typesense sync job not found",
                "Provide job_id or a source_client_name + source_table_name binding that matches an enabled Typesense sync job.",
            );
        }
        Err(err) if sqlx_relation_missing(&err) => return typesense_not_provisioned_response(),
        Err(err) => return internal_error("Failed to resolve Typesense sync job", err.to_string()),
    };

    let query_by = body
        .query_by
        .clone()
        .filter(|fields| !fields.is_empty())
        .unwrap_or_else(|| string_vec_from_json(&job.query_by_fields));
    if query_by.is_empty() {
        return bad_request(
            "Typesense search is not configured",
            "The sync job does not define any query_by fields yet.",
        );
    }

    let connection = typesense_connection_from_job(&job);
    let mut query = vec![
        ("q", body.q.trim().to_string()),
        ("query_by", query_by.join(",")),
        ("page", body.page.unwrap_or(1).max(1).to_string()),
        (
            "per_page",
            body.per_page.unwrap_or(20).clamp(1, 250).to_string(),
        ),
    ];
    if let Some(filter_by) = body
        .filter_by
        .as_ref()
        .map(|value| value.trim())
        .filter(|value| !value.is_empty())
    {
        query.push(("filter_by", filter_by.to_string()));
    }
    if let Some(sort_by) = body
        .sort_by
        .as_ref()
        .map(|value| value.trim())
        .filter(|value| !value.is_empty())
    {
        query.push(("sort_by", sort_by.to_string()));
    }

    let started = Instant::now();
    let (status, payload) = match typesense_get_json(
        app_state.get_ref(),
        &connection,
        &format!("/collections/{}/documents/search", job.collection_name),
        &query,
    )
    .await
    {
        Ok(result) => result,
        Err(err) => return internal_error("Typesense search failed", err),
    };

    if !status.is_success() {
        return internal_error(
            "Typesense search failed",
            format!("Typesense returned HTTP {}: {}", status.as_u16(), payload),
        );
    }

    api_success(
        "Typesense search completed",
        json!({
            "job_id": job.id,
            "job_name": job.name,
            "profile_name": job.profile_name,
            "collection_name": job.collection_name,
            "source_client_name": job.source_client_name,
            "source_schema_name": job.source_schema_name,
            "source_table_name": job.source_table_name,
            "query_by": query_by,
            "latency_ms": started.elapsed().as_millis(),
            "result": payload,
        }),
    )
}

async fn run_due_sync_jobs_once(state: &AppState) -> Result<bool, String> {
    let Some(logging) = logging_pool_from_state(state) else {
        return Ok(false);
    };

    let job_ids = due_job_ids(&logging).await.map_err(|err| err.to_string())?;
    if job_ids.is_empty() {
        return Ok(false);
    }

    for job_id in job_ids {
        if let Err(err) = execute_typesense_sync_job(state, job_id, "scheduled").await {
            warn!(job_id, error = %err, "Typesense scheduled sync job failed");
        }
    }

    Ok(true)
}

pub fn spawn_typesense_sync_worker(state: web::Data<AppState>) {
    if !typesense_worker_enabled() {
        info!("Typesense sync worker disabled");
        return;
    }

    let poll_ms = typesense_worker_poll_ms();
    tokio::spawn(async move {
        info!(poll_ms, "Typesense sync worker started");
        loop {
            match run_due_sync_jobs_once(state.get_ref()).await {
                Ok(ran_jobs) => {
                    TYPESENSE_SCHEMA_WARNED.store(false, Ordering::Relaxed);
                    let sleep_for = if ran_jobs { 1_000 } else { poll_ms };
                    tokio::time::sleep(Duration::from_millis(sleep_for)).await;
                }
                Err(err) => {
                    let should_warn = !TYPESENSE_SCHEMA_WARNED.swap(true, Ordering::Relaxed);
                    if should_warn {
                        warn!(error = %err, "Typesense sync worker tick failed");
                    }
                    tokio::time::sleep(Duration::from_millis(poll_ms)).await;
                }
            }
        }
    });
}

pub fn services(cfg: &mut web::ServiceConfig) {
    cfg.service(
        web::scope("/typesense")
            .wrap(actix_web::middleware::from_fn(typesense_auth_middleware))
            .service(test_typesense_backend)
            .service(list_typesense_collections)
            .service(run_typesense_sync_job_route)
            .service(typesense_search_route),
    );
}
