use actix_web::HttpResponse;
use athena_control_plane::{normalize_daemon_id, resolve_runtime_daemon_id};
use chrono::{DateTime, Utc};
use once_cell::sync::Lazy;
use reqwest::StatusCode as ReqwestStatusCode;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use sqlx::{PgPool, QueryBuilder, Row};
use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard};
use std::time::{Duration, Instant};
use tokio::sync::OnceCell;
use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};
use uuid::Uuid;

use crate::TypesenseRuntime;

static TYPESENSE_RUN_CANCEL_TOKENS: Lazy<Mutex<HashMap<i64, CancellationToken>>> =
    Lazy::new(|| Mutex::new(HashMap::new()));
static TYPESENSE_SCHEMA_COMPAT_ONCE: OnceCell<Result<(), String>> = OnceCell::const_new();

// TODO: These are typesense config values that dont belong here
const DEFAULT_TYPESENSE_TIMEOUT_MS: u64 = 10_000;
const MAX_TYPESENSE_IMPORT_RETRY_DELAY_MS: u64 = 5_000;
const MAX_TYPESENSE_BATCH_DELAY_MS: i32 = 600_000;
const TYPESENSE_SAGA_PREVIEW_LIMIT: usize = 25;
const TYPESENSE_MANIFEST_BUILD_BATCH_SIZE: i64 = 1_000;
const SYNC_LOCK_TTL_MINS: i32 = 15;
const TYPESENSE_MAX_PARALLEL_RUNS: i64 = 1;
const TYPESENSE_GLOBAL_RUN_ADVISORY_LOCK: i64 = 4_222_001_001;
const TYPESENSE_CANCELLED_ERROR: &str = "Typesense sync run cancelled by operator.";
const TYPESENSE_WORKER_BUSY_ERROR_PREFIX: &str = "Typesense worker is busy.";
const TYPESENSE_PROGRESS_TOTAL_KIND_MISSING_ERROR: &str =
    "column \"progress_total_kind\" does not exist";
const TYPESENSE_DAEMON_AFFINITY_ERROR_PREFIX: &str = "Typesense sync job is pinned to daemon";

#[derive(Debug, Serialize)]
struct ApiSuccessResponse<T: Serialize> {
    status: &'static str,
    message: String,
    data: T,
}

#[derive(Debug, Serialize)]
struct ApiErrorResponse {
    status: &'static str,
    message: String,
    error: String,
}

fn api_success<T: Serialize>(message: impl Into<String>, data: T) -> HttpResponse {
    HttpResponse::Ok().json(ApiSuccessResponse {
        status: "success",
        message: message.into(),
        data,
    })
}

fn bad_request(message: impl Into<String>, error: impl Into<String>) -> HttpResponse {
    HttpResponse::BadRequest().json(ApiErrorResponse {
        status: "error",
        message: message.into(),
        error: error.into(),
    })
}

fn conflict(message: impl Into<String>, error: impl Into<String>) -> HttpResponse {
    HttpResponse::Conflict().json(ApiErrorResponse {
        status: "error",
        message: message.into(),
        error: error.into(),
    })
}

fn internal_error(message: impl Into<String>, error: impl Into<String>) -> HttpResponse {
    HttpResponse::InternalServerError().json(ApiErrorResponse {
        status: "error",
        message: message.into(),
        error: error.into(),
    })
}

fn not_found(message: impl Into<String>, error: impl Into<String>) -> HttpResponse {
    HttpResponse::NotFound().json(ApiErrorResponse {
        status: "error",
        message: message.into(),
        error: error.into(),
    })
}

fn service_unavailable(message: impl Into<String>, error: impl Into<String>) -> HttpResponse {
    HttpResponse::ServiceUnavailable().json(ApiErrorResponse {
        status: "error",
        message: message.into(),
        error: error.into(),
    })
}

fn sanitize_identifier(identifier: &str) -> Option<String> {
    let trimmed = identifier.trim();
    let mut chars = trimmed.chars();
    let first = chars.next()?;
    if !(first.is_ascii_alphabetic() || first == '_') {
        return None;
    }
    if !chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_') {
        return None;
    }
    Some(format!("\"{trimmed}\""))
}

fn sanitize_qualified_table_identifier(table_name: &str) -> Option<String> {
    let parts: Vec<&str> = table_name.split('.').collect();
    if parts.is_empty() || parts.len() > 2 {
        return None;
    }

    let mut sanitized_parts = Vec::with_capacity(parts.len());
    for part in parts {
        sanitized_parts.push(sanitize_identifier(part)?);
    }

    Some(sanitized_parts.join("."))
}

fn is_typesense_worker_busy_error(error: &str) -> bool {
    error.contains(TYPESENSE_WORKER_BUSY_ERROR_PREFIX)
}

fn is_typesense_progress_total_kind_missing_error(error: &str) -> bool {
    error.contains(TYPESENSE_PROGRESS_TOTAL_KIND_MISSING_ERROR)
}

fn current_daemon_id() -> String {
    resolve_runtime_daemon_id("api_inline", None)
        .expect("typesense runtime daemon id resolution should not fail")
        .to_string()
}

fn daemon_affinity_matches(affinity: Option<&str>, daemon_id: &str) -> bool {
    match normalize_daemon_id(affinity) {
        Some(target) => target == daemon_id,
        None => true,
    }
}

/// Request body for testing a direct Typesense backend connection.
#[derive(Debug, Deserialize)]
pub struct TypesenseConnectionRequest {
    base_url: String,
    api_key: String,
    #[serde(default)]
    timeout_ms: Option<u64>,
}

/// Request body for searching through a configured Athena -> Typesense binding.
#[derive(Debug, Deserialize)]
pub 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,
    retry_attempt_count: usize,
    rollback_attempted: bool,
    rollback_status: String,
    rollback_restored_count: usize,
    rollback_deleted_count: usize,
    result_preview: Value,
    progress_total_kind: String,
    manifest_total_rows: Option<usize>,
    manifest_processed_rows: usize,
    resumed: bool,
}

#[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,
    batch_delay_ms: i32,
    sync_interval_minutes: i32,
    stale_after_minutes: i32,
    execution_daemon_id: Option<String>,
    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)]
struct TypesenseImportFailure {
    message: String,
    status: Option<ReqwestStatusCode>,
    retryable: bool,
}

#[derive(Debug, Clone)]
struct TypesenseNormalizedDocument {
    id: String,
    document: Value,
}

#[derive(Debug, Default)]
struct TypesenseSagaBackupState {
    original_documents: HashMap<String, Option<Value>>,
}

#[derive(Debug, Default)]
struct TypesenseSagaCompensationResult {
    restored_count: usize,
    deleted_count: usize,
    error_message: Option<String>,
}

#[derive(Debug, Clone)]
struct TypesenseSyncProgressUpdate {
    source_row_count: usize,
    estimated_total_rows: Option<usize>,
    progress_total_kind: Option<String>,
    manifest_total_rows: Option<usize>,
    manifest_processed_rows: Option<usize>,
    next_manifest_seq: Option<usize>,
    resume_attempt_count: Option<usize>,
    processed_row_count: usize,
    imported_count: usize,
    failed_count: usize,
    progress_pct: i32,
    progress_stage: String,
    progress_message: Option<String>,
    inter_batch_delay_ms: usize,
    next_batch_at: Option<DateTime<Utc>>,
    batches_completed: usize,
}

#[derive(Debug, sqlx::FromRow)]
struct TypesenseRunningRunState {
    id: i64,
    progress_total_kind: String,
    manifest_total_rows: Option<i32>,
    manifest_processed_rows: i32,
    next_manifest_seq: i64,
    source_row_count: Option<i32>,
    imported_count: Option<i32>,
    failed_count: Option<i32>,
    batches_completed: i32,
    retry_attempt_count: i32,
    rollback_attempted: bool,
    rollback_status: String,
    rollback_restored_count: Option<i32>,
    rollback_deleted_count: Option<i32>,
}

#[derive(Debug, sqlx::FromRow)]
struct DueTypesenseJobCandidate {
    id: i64,
    source_client_name: String,
    execution_daemon_id: Option<String>,
}

#[derive(Debug)]
struct TypesenseManifestBatchRow {
    manifest_seq: i64,
    row_data: Value,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum TypesenseClaimResult {
    Claimed(Uuid),
    BusyOrDisabled,
    WorkerAtCapacity,
}

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

fn clamp_batch_delay_ms(value: i32) -> i32 {
    value.clamp(0, MAX_TYPESENSE_BATCH_DELAY_MS)
}

fn normalize_progress_total_kind(value: &str) -> &'static str {
    match value {
        "exact" => "exact",
        "estimated" => "estimated",
        _ => "unknown",
    }
}

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 typesense_retry_delay(import_retry_base_ms: u64, attempt_number: usize) -> Duration {
    let exponent = u32::try_from(attempt_number.saturating_sub(1)).unwrap_or(0);
    let base = import_retry_base_ms.clamp(100, 5_000);
    let multiplier = 2u64.saturating_pow(exponent).max(1);
    let delay_ms = base
        .saturating_mul(multiplier)
        .min(MAX_TYPESENSE_IMPORT_RETRY_DELAY_MS);
    Duration::from_millis(delay_ms)
}

fn is_retryable_typesense_status(status: ReqwestStatusCode) -> bool {
    status.is_server_error()
        || matches!(
            status,
            ReqwestStatusCode::REQUEST_TIMEOUT
                | ReqwestStatusCode::TOO_MANY_REQUESTS
                | ReqwestStatusCode::BAD_GATEWAY
                | ReqwestStatusCode::SERVICE_UNAVAILABLE
                | ReqwestStatusCode::GATEWAY_TIMEOUT
        )
}

fn is_retryable_typesense_request_error(error: &reqwest::Error) -> bool {
    error.is_timeout() || error.is_connect() || error.is_request() || error.is_body()
}

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.",
    )
}

async fn ensure_typesense_sync_runs_schema_compat(pool: &PgPool) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        ALTER TABLE public.typesense_sync_jobs
            ADD COLUMN IF NOT EXISTS execution_daemon_id text
        "#,
    )
    .execute(pool)
    .await?;
    sqlx::query(
        r#"
        CREATE INDEX IF NOT EXISTS idx_typesense_sync_jobs_execution_daemon
            ON public.typesense_sync_jobs(execution_daemon_id)
        "#,
    )
    .execute(pool)
    .await?;

    // TODO: Raw sql
    sqlx::query(
        r#"
        ALTER TABLE public.typesense_sync_runs
            ADD COLUMN IF NOT EXISTS inter_batch_delay_ms integer NOT NULL DEFAULT 0,
            ADD COLUMN IF NOT EXISTS next_batch_at timestamptz,
            ADD COLUMN IF NOT EXISTS progress_total_kind text NOT NULL DEFAULT 'unknown',
            ADD COLUMN IF NOT EXISTS retry_attempt_count integer NOT NULL DEFAULT 0,
            ADD COLUMN IF NOT EXISTS manifest_total_rows integer,
            ADD COLUMN IF NOT EXISTS manifest_processed_rows integer NOT NULL DEFAULT 0,
            ADD COLUMN IF NOT EXISTS next_manifest_seq bigint NOT NULL DEFAULT 0,
            ADD COLUMN IF NOT EXISTS resume_attempt_count integer NOT NULL DEFAULT 0,
            ADD COLUMN IF NOT EXISTS rollback_attempted boolean NOT NULL DEFAULT false,
            ADD COLUMN IF NOT EXISTS rollback_status text NOT NULL DEFAULT 'not_needed',
            ADD COLUMN IF NOT EXISTS rollback_restored_count integer,
            ADD COLUMN IF NOT EXISTS rollback_deleted_count integer,
            ADD COLUMN IF NOT EXISTS rollback_error_message text
        "#,
    )
    .execute(pool)
    .await?;
    Ok(())
}

async fn ensure_typesense_sync_runs_schema_compat_once(pool: &PgPool) -> Result<(), String> {
    TYPESENSE_SCHEMA_COMPAT_ONCE
        .get_or_init(|| async {
            ensure_typesense_sync_runs_schema_compat(pool)
                .await
                .map_err(|err| format!("Failed to reconcile Typesense sync run schema: {err}"))
        })
        .await
        .clone()
}

fn register_typesense_run_cancel_token(run_id: i64, token: CancellationToken) {
    let mut guard: MutexGuard<'_, HashMap<i64, CancellationToken>> =
        TYPESENSE_RUN_CANCEL_TOKENS.lock().unwrap();
    guard.insert(run_id, token);
}

fn unregister_typesense_run_cancel_token(run_id: i64) {
    let mut guard: MutexGuard<'_, HashMap<i64, CancellationToken>> =
        TYPESENSE_RUN_CANCEL_TOKENS.lock().unwrap();
    guard.remove(&run_id);
}

fn trigger_typesense_run_cancel_token(run_id: i64) {
    let guard: MutexGuard<'_, HashMap<i64, CancellationToken>> =
        TYPESENSE_RUN_CANCEL_TOKENS.lock().unwrap();
    if let Some(token) = guard.get(&run_id) {
        token.cancel();
    }
}

struct TypesenseRunCancelGuard(i64);

impl Drop for TypesenseRunCancelGuard {
    fn drop(&mut self) {
        unregister_typesense_run_cancel_token(self.0);
    }
}

fn normalize_typesense_connection(
    body: &TypesenseConnectionRequest,
    allow_http: bool,
) -> 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 allow_http => {}
        "http" => {
            return Err(bad_request(
                "Invalid request",
                "Typesense connections must use HTTPS unless typesense.allow_http is set to true in config.yaml 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<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    path: &str,
    query: &[(&str, String)],
) -> Result<(ReqwestStatusCode, Value), String> {
    let mut request = state
        .http_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<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    path: &str,
    query: &[(&str, String)],
    body: &Value,
) -> Result<(ReqwestStatusCode, Value), String> {
    let mut request = state
        .http_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_healthcheck<R: TypesenseRuntime>(
    state: &R,
    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<R: TypesenseRuntime>(state: &R) -> Option<PgPool> {
    state.logging_pool()
}

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.batch_delay_ms,
            j.sync_interval_minutes,
            j.stale_after_minutes,
            j.execution_daemon_id,
            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_jobs_for_daemon(
    pool: &PgPool,
    daemon_id: &str,
) -> Result<Vec<DueTypesenseJobCandidate>, sqlx::Error> {
    sqlx::query_as::<_, DueTypesenseJobCandidate>(
        r#"
        SELECT
            id,
            source_client_name,
            execution_daemon_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())
          AND (
            NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') IS NULL
            OR NULLIF(BTRIM(COALESCE(execution_daemon_id, '')), '') = $1
          )
        ORDER BY COALESCE(next_sync_at, created_at) ASC
        LIMIT 5
        "#,
    )
    .bind(daemon_id)
    .fetch_all(pool)
    .await
}

async fn claim_sync_job(pool: &PgPool, job_id: i64) -> Result<TypesenseClaimResult, sqlx::Error> {
    let lock_token = Uuid::new_v4();
    let mut tx = pool.begin().await?;
    let advisory_lock_acquired =
        sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_xact_lock($1)")
            .bind(TYPESENSE_GLOBAL_RUN_ADVISORY_LOCK)
            .fetch_one(&mut *tx)
            .await?;
    if !advisory_lock_acquired {
        tx.commit().await?;
        return Ok(TypesenseClaimResult::WorkerAtCapacity);
    }

    let running_count = sqlx::query_scalar::<_, i64>(
        r#"
        SELECT COUNT(*)::bigint
        FROM public.typesense_sync_jobs
        WHERE last_sync_status = 'running'
          AND sync_lock_expires_at IS NOT NULL
          AND sync_lock_expires_at > now()
        "#,
    )
    .fetch_one(&mut *tx)
    .await?;
    if running_count >= TYPESENSE_MAX_PARALLEL_RUNS {
        tx.commit().await?;
        return Ok(TypesenseClaimResult::WorkerAtCapacity);
    }

    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(&mut *tx)
    .await?;
    tx.commit().await?;

    if claimed.is_some() {
        Ok(TypesenseClaimResult::Claimed(lock_token))
    } else {
        Ok(TypesenseClaimResult::BusyOrDisabled)
    }
}

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

    Ok(renewed.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,
            inter_batch_delay_ms
        )
        VALUES (
            $1,
            $2,
            'running',
            0,
            'starting',
            'Claimed sync job and preparing run.',
            (SELECT COALESCE(batch_delay_ms, 0) FROM public.typesense_sync_jobs WHERE id = $1)
        )
        RETURNING id
        "#,
    )
    .bind(job_id)
    .bind(trigger_source)
    .fetch_one(pool)
    .await?;
    row.try_get("id")
}

async fn load_running_run_for_job(pool: &PgPool, job_id: i64) -> Result<Option<i64>, sqlx::Error> {
    sqlx::query_scalar::<_, i64>(
        r#"
        SELECT id
        FROM public.typesense_sync_runs
        WHERE job_id = $1
          AND status = 'running'
        ORDER BY started_at DESC, id DESC
        LIMIT 1
        "#,
    )
    .bind(job_id)
    .fetch_optional(pool)
    .await
}

async fn load_running_sync_run_state(
    pool: &PgPool,
    job_id: i64,
) -> Result<Option<TypesenseRunningRunState>, sqlx::Error> {
    sqlx::query_as::<_, TypesenseRunningRunState>(
        r#"
        SELECT
            id,
            progress_total_kind,
            manifest_total_rows,
            manifest_processed_rows,
            next_manifest_seq,
            source_row_count,
            imported_count,
            failed_count,
            batches_completed,
            retry_attempt_count,
            rollback_attempted,
            rollback_status,
            rollback_restored_count,
            rollback_deleted_count
        FROM public.typesense_sync_runs
        WHERE job_id = $1
          AND status = 'running'
        ORDER BY started_at DESC, id DESC
        LIMIT 1
        "#,
    )
    .bind(job_id)
    .fetch_optional(pool)
    .await
}

async fn bump_sync_run_resume_attempt(pool: &PgPool, run_id: i64) -> Result<usize, sqlx::Error> {
    let row = sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            resume_attempt_count = resume_attempt_count + 1,
            progress_stage = 'resuming',
            progress_message = 'Worker resumed an in-flight run after restart/lease handoff.'
        WHERE id = $1
          AND status = 'running'
        RETURNING resume_attempt_count
        "#,
    )
    .bind(run_id)
    .fetch_one(pool)
    .await?;
    let value: i32 = row.try_get("resume_attempt_count")?;
    Ok(usize::try_from(value.max(0)).unwrap_or(0))
}

async fn clear_manifest_rows(pool: &PgPool, run_id: i64) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        DELETE FROM public.typesense_sync_run_manifest_rows
        WHERE run_id = $1
        "#,
    )
    .bind(run_id)
    .execute(pool)
    .await?;
    Ok(())
}

async fn build_manifest_snapshot(
    source_pool: &PgPool,
    logging_pool: &PgPool,
    run_id: i64,
    schema_name: &str,
    table_name: &str,
    selected_columns: &[String],
    document_id_column: &str,
) -> Result<usize, 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_data, ({document_id})::text AS document_id \
         FROM (SELECT {columns} FROM {table} \
               WHERE ($2::text IS NULL OR ({document_id})::text > $2::text) \
               ORDER BY ({document_id})::text \
               LIMIT $1) t",
        document_id = document_id_identifier,
        columns = quoted_columns.join(", "),
        table = qualified_table,
    );

    let mut source_tx = source_pool
        .begin()
        .await
        .map_err(|err| format!("Failed to start source snapshot transaction: {err}"))?;
    sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY")
        .execute(&mut *source_tx)
        .await
        .map_err(|err| format!("Failed to set source snapshot isolation level: {err}"))?;

    let mut last_document_id: Option<String> = None;
    let mut manifest_seq: i64 = 0;
    let mut total_rows: usize = 0;

    loop {
        let rows = sqlx::query(&sql)
            .bind(TYPESENSE_MANIFEST_BUILD_BATCH_SIZE)
            .bind(last_document_id.as_deref())
            .fetch_all(&mut *source_tx)
            .await
            .map_err(|err| format!("Failed to build manifest snapshot from source rows: {err}"))?;

        if rows.is_empty() {
            break;
        }

        let mut payloads = Vec::with_capacity(rows.len());
        for row in rows {
            let document_id: String = row
                .try_get("document_id")
                .map_err(|err| format!("Failed to decode manifest document id: {err}"))?;
            let row_data: Value = row
                .try_get("row_data")
                .map_err(|err| format!("Failed to decode manifest source row payload: {err}"))?;
            payloads.push((manifest_seq, document_id, row_data));
            manifest_seq += 1;
        }

        last_document_id = payloads
            .last()
            .map(|(_, document_id, _)| document_id.clone());
        total_rows += payloads.len();

        let mut builder = QueryBuilder::<sqlx::Postgres>::new(
            "INSERT INTO public.typesense_sync_run_manifest_rows \
             (run_id, manifest_seq, document_id, row_data) ",
        );
        builder.push_values(payloads.iter(), |mut b, (seq, document_id, row_data)| {
            b.push_bind(run_id)
                .push_bind(*seq)
                .push_bind(document_id)
                .push_bind(row_data);
        });
        builder
            .build()
            .execute(logging_pool)
            .await
            .map_err(|err| format!("Failed to persist manifest snapshot rows: {err}"))?;
    }

    source_tx
        .commit()
        .await
        .map_err(|err| format!("Failed to finalize source manifest snapshot transaction: {err}"))?;

    Ok(total_rows)
}

async fn load_manifest_batch(
    pool: &PgPool,
    run_id: i64,
    from_seq: usize,
    batch_size: usize,
) -> Result<Vec<TypesenseManifestBatchRow>, String> {
    let rows = sqlx::query(
        r#"
        SELECT manifest_seq, row_data
        FROM public.typesense_sync_run_manifest_rows
        WHERE run_id = $1
          AND manifest_seq >= $2
        ORDER BY manifest_seq ASC
        LIMIT $3
        "#,
    )
    .bind(run_id)
    .bind(i64::try_from(from_seq).unwrap_or(i64::MAX))
    .bind(i64::try_from(batch_size.max(1)).unwrap_or(i64::MAX))
    .fetch_all(pool)
    .await
    .map_err(|err| format!("Failed to load manifest batch rows: {err}"))?;

    let mut batch_rows = Vec::with_capacity(rows.len());
    for row in rows {
        let manifest_seq: i64 = row
            .try_get("manifest_seq")
            .map_err(|err| format!("Failed to decode manifest sequence: {err}"))?;
        let row_data: Value = row
            .try_get("row_data")
            .map_err(|err| format!("Failed to decode manifest row payload: {err}"))?;
        batch_rows.push(TypesenseManifestBatchRow {
            manifest_seq,
            row_data,
        });
    }
    Ok(batch_rows)
}

async fn load_current_batch_delay_ms(pool: &PgPool, job_id: i64) -> Result<i32, sqlx::Error> {
    let delay = sqlx::query_scalar::<_, i32>(
        r#"
        SELECT COALESCE(batch_delay_ms, 0)
        FROM public.typesense_sync_jobs
        WHERE id = $1
        "#,
    )
    .bind(job_id)
    .fetch_optional(pool)
    .await?
    .unwrap_or(0);
    Ok(clamp_batch_delay_ms(delay))
}

async fn run_is_cancelled(pool: &PgPool, run_id: i64) -> Result<bool, sqlx::Error> {
    let status = sqlx::query_scalar::<_, String>(
        r#"
        SELECT status
        FROM public.typesense_sync_runs
        WHERE id = $1
        "#,
    )
    .bind(run_id)
    .fetch_optional(pool)
    .await?;
    Ok(matches!(status.as_deref(), Some("cancelled")))
}

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,
            progress_total_kind = COALESCE($4, progress_total_kind),
            manifest_total_rows = COALESCE($5, manifest_total_rows),
            manifest_processed_rows = COALESCE($6, manifest_processed_rows),
            next_manifest_seq = COALESCE($7, next_manifest_seq),
            resume_attempt_count = COALESCE($8, resume_attempt_count),
            processed_row_count = $9,
            imported_count = $10,
            failed_count = $11,
            progress_pct = $12,
            progress_stage = $13,
            progress_message = $14,
            inter_batch_delay_ms = $15,
            next_batch_at = $16,
            batches_completed = $17
        WHERE id = $1
          AND status = 'running'
        "#,
    )
    .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(
        progress
            .progress_total_kind
            .as_deref()
            .map(normalize_progress_total_kind),
    )
    .bind(
        progress
            .manifest_total_rows
            .map(|value| i32::try_from(value).unwrap_or(i32::MAX)),
    )
    .bind(
        progress
            .manifest_processed_rows
            .map(|value| i32::try_from(value).unwrap_or(i32::MAX)),
    )
    .bind(
        progress
            .next_manifest_seq
            .map(|value| i64::try_from(value).unwrap_or(i64::MAX)),
    )
    .bind(
        progress
            .resume_attempt_count
            .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.inter_batch_delay_ms).unwrap_or(i32::MAX))
    .bind(progress.next_batch_at)
    .bind(i32::try_from(progress.batches_completed).unwrap_or(i32::MAX))
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_sync_run_retry_attempts(
    pool: &PgPool,
    run_id: i64,
    retry_attempt_count: usize,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET retry_attempt_count = $2
        WHERE id = $1
          AND status = 'running'
        "#,
    )
    .bind(run_id)
    .bind(i32::try_from(retry_attempt_count).unwrap_or(i32::MAX))
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_sync_run_rollback_state(
    pool: &PgPool,
    run_id: i64,
    rollback_attempted: bool,
    rollback_status: &str,
    rollback_restored_count: usize,
    rollback_deleted_count: usize,
    rollback_error_message: Option<&str>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            rollback_attempted = $2,
            rollback_status = $3,
            rollback_restored_count = $4,
            rollback_deleted_count = $5,
            rollback_error_message = $6
        WHERE id = $1
          AND status = 'running'
        "#,
    )
    .bind(run_id)
    .bind(rollback_attempted)
    .bind(rollback_status)
    .bind(i32::try_from(rollback_restored_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(rollback_deleted_count).unwrap_or(i32::MAX))
    .bind(rollback_error_message)
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_sync_run_success(
    pool: &PgPool,
    run_id: i64,
    source_row_count: usize,
    estimated_total_rows: Option<usize>,
    progress_total_kind: &str,
    manifest_total_rows: Option<usize>,
    manifest_processed_rows: usize,
    next_manifest_seq: usize,
    resume_attempt_count: usize,
    imported_count: usize,
    failed_count: usize,
    batches_completed: usize,
    duration_ms: u128,
    retry_attempt_count: usize,
    rollback_attempted: bool,
    rollback_status: &str,
    rollback_restored_count: usize,
    rollback_deleted_count: usize,
    rollback_error_message: Option<&str>,
    result_preview: &Value,
) -> Result<bool, sqlx::Error> {
    let updated = sqlx::query(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            status = 'success',
            finished_at = now(),
            source_row_count = $2,
            estimated_total_rows = $3,
            progress_total_kind = $4,
            manifest_total_rows = $5,
            manifest_processed_rows = $6,
            next_manifest_seq = $7,
            resume_attempt_count = $8,
            processed_row_count = $6,
            imported_count = $9,
            failed_count = $10,
            progress_pct = 100,
            progress_stage = 'completed',
            progress_message = CASE
                WHEN $2 = 0 THEN 'Completed sync with no source rows to import.'
                WHEN $10 > 0 THEN format('Completed sync with %s imported and %s failed rows.', $9, $10)
                ELSE format('Completed sync with %s imported rows.', $9)
            END,
            next_batch_at = NULL,
            batches_completed = CASE
                WHEN COALESCE($11, 0) <= 0 THEN batches_completed
                ELSE GREATEST(batches_completed, 1)
            END,
            duration_ms = $12,
            retry_attempt_count = $13,
            rollback_attempted = $14,
            rollback_status = $15,
            rollback_restored_count = $16,
            rollback_deleted_count = $17,
            rollback_error_message = $18,
            result_preview = $19
        WHERE id = $1
          AND status = 'running'
        "#,
    )
    .bind(run_id)
    .bind(i32::try_from(source_row_count).unwrap_or(i32::MAX))
    .bind(estimated_total_rows.map(|value| i32::try_from(value).unwrap_or(i32::MAX)))
    .bind(normalize_progress_total_kind(progress_total_kind))
    .bind(manifest_total_rows.map(|value| i32::try_from(value).unwrap_or(i32::MAX)))
    .bind(i32::try_from(manifest_processed_rows).unwrap_or(i32::MAX))
    .bind(i64::try_from(next_manifest_seq).unwrap_or(i64::MAX))
    .bind(i32::try_from(resume_attempt_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(i32::try_from(retry_attempt_count).unwrap_or(i32::MAX))
    .bind(rollback_attempted)
    .bind(rollback_status)
    .bind(i32::try_from(rollback_restored_count).unwrap_or(i32::MAX))
    .bind(i32::try_from(rollback_deleted_count).unwrap_or(i32::MAX))
    .bind(rollback_error_message)
    .bind(result_preview)
    .execute(pool)
    .await?;
    Ok(updated.rows_affected() > 0)
}

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,
            next_batch_at = NULL,
            error_message = $3
        WHERE id = $1
          AND status = 'running'
        "#,
    )
    .bind(run_id)
    .bind(i64::try_from(duration_ms).unwrap_or(i64::MAX))
    .bind(error_message)
    .execute(pool)
    .await?;
    Ok(())
}

async fn update_sync_run_cancelled(
    pool: &PgPool,
    run_id: i64,
    reason: &str,
) -> Result<Option<i64>, sqlx::Error> {
    sqlx::query_scalar::<_, i64>(
        r#"
        UPDATE public.typesense_sync_runs
        SET
            status = 'cancelled',
            finished_at = COALESCE(finished_at, now()),
            progress_stage = 'cancelled',
            progress_message = $2,
            next_batch_at = NULL,
            error_message = $2
        WHERE id = $1
          AND status = 'running'
        RETURNING job_id
        "#,
    )
    .bind(run_id)
    .bind(reason)
    .fetch_optional(pool)
    .await
}

async fn update_job_success(
    pool: &PgPool,
    job: &TypesenseJobWithProfile,
    source_row_count: usize,
    imported_count: usize,
    failed_count: usize,
    lock_token: Uuid,
) -> Result<bool, sqlx::Error> {
    let updated = 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
          AND sync_lock_token = $5
        "#,
    )
    .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))
    .bind(lock_token)
    .execute(pool)
    .await?;
    Ok(updated.rows_affected() > 0)
}

async fn update_job_failure(
    pool: &PgPool,
    job_id: i64,
    error_message: &str,
    lock_token: Uuid,
) -> Result<bool, sqlx::Error> {
    let updated = 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
          AND sync_lock_token = $3
        "#,
    )
    .bind(job_id)
    .bind(error_message)
    .bind(lock_token)
    .execute(pool)
    .await?;
    Ok(updated.rows_affected() > 0)
}

async fn update_job_cancelled(pool: &PgPool, job_id: i64, reason: &str) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE public.typesense_sync_jobs
        SET
            updated_at = now(),
            last_sync_status = 'cancelled',
            last_sync_finished_at = now(),
            last_sync_error = $2,
            sync_lock_token = NULL,
            sync_locked_at = NULL,
            sync_lock_expires_at = NULL
        WHERE id = $1
        "#,
    )
    .bind(job_id)
    .bind(reason)
    .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<R: TypesenseRuntime>(
    state: &R,
    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 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 normalize_rows_for_import(
    rows: Vec<Value>,
    document_id_column: &str,
    field_types: &HashMap<String, String>,
) -> Result<Vec<TypesenseNormalizedDocument>, String> {
    let mut documents = 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.clone()));

        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),
                );
            }
        }

        documents.push(TypesenseNormalizedDocument {
            id,
            document: Value::Object(map),
        });
    }

    Ok(documents)
}

fn documents_to_ndjson(documents: &[TypesenseNormalizedDocument]) -> Result<String, String> {
    let mut lines = Vec::with_capacity(documents.len());
    for document in documents {
        lines.push(
            serde_json::to_string(&document.document)
                .map_err(|err| format!("Failed to serialize Typesense document: {err}"))?,
        );
    }
    Ok(lines.join("\n"))
}

fn typesense_document_url(
    connection: &TypesenseConnection,
    collection_name: &str,
    document_id: &str,
) -> Result<reqwest::Url, String> {
    let mut url = reqwest::Url::parse(&connection.base_url).map_err(|err| {
        format!(
            "Invalid Typesense base URL '{}': {err}",
            connection.base_url
        )
    })?;
    {
        let mut segments = url
            .path_segments_mut()
            .map_err(|_| "Typesense base URL does not support path segments.".to_string())?;
        segments.push("collections");
        segments.push(collection_name);
        segments.push("documents");
        segments.push(document_id);
    }
    Ok(url)
}

async fn fetch_document_backup_once<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    document_id: &str,
) -> Result<Option<Value>, TypesenseImportFailure> {
    let url =
        typesense_document_url(connection, collection_name, document_id).map_err(|message| {
            TypesenseImportFailure {
                message,
                status: None,
                retryable: false,
            }
        })?;

    let response = state
        .http_client()
        .get(url)
        .timeout(Duration::from_millis(connection.timeout_ms))
        .header("X-TYPESENSE-API-KEY", &connection.api_key)
        .send()
        .await
        .map_err(|err| TypesenseImportFailure {
            message: format!(
                "Typesense backup read failed for id '{}': {err}",
                document_id
            ),
            status: None,
            retryable: is_retryable_typesense_request_error(&err),
        })?;

    let status = response.status();
    if status == ReqwestStatusCode::NOT_FOUND {
        return Ok(None);
    }
    if !status.is_success() {
        let body = response.text().await.unwrap_or_default();
        return Err(TypesenseImportFailure {
            message: format!(
                "Typesense backup read failed for id '{}' with HTTP {}: {}",
                document_id,
                status.as_u16(),
                body
            ),
            status: Some(status),
            retryable: is_retryable_typesense_status(status),
        });
    }

    let payload = response
        .json::<Value>()
        .await
        .map_err(|err| TypesenseImportFailure {
            message: format!(
                "Typesense backup read returned invalid JSON for id '{}': {err}",
                document_id
            ),
            status: Some(status),
            retryable: false,
        })?;
    Ok(Some(payload))
}

async fn fetch_document_backup_with_retry<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    document_id: &str,
) -> Result<Option<Value>, TypesenseImportFailure> {
    let max_attempts = state.typesense_import_max_attempts();
    for attempt in 1..=max_attempts {
        match fetch_document_backup_once(state, connection, collection_name, document_id).await {
            Ok(snapshot) => return Ok(snapshot),
            Err(err) if err.retryable && attempt < max_attempts => {
                tokio::time::sleep(typesense_retry_delay(
                    state.typesense_import_retry_base_ms(),
                    attempt,
                ))
                .await;
            }
            Err(err) => return Err(err),
        }
    }
    Err(TypesenseImportFailure {
        message: "Typesense backup read exhausted all retries.".to_string(),
        status: None,
        retryable: false,
    })
}

async fn delete_document_once<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    document_id: &str,
) -> Result<bool, TypesenseImportFailure> {
    let url =
        typesense_document_url(connection, collection_name, document_id).map_err(|message| {
            TypesenseImportFailure {
                message,
                status: None,
                retryable: false,
            }
        })?;

    let response = state
        .http_client()
        .delete(url)
        .timeout(Duration::from_millis(connection.timeout_ms))
        .header("X-TYPESENSE-API-KEY", &connection.api_key)
        .send()
        .await
        .map_err(|err| TypesenseImportFailure {
            message: format!(
                "Typesense rollback delete failed for id '{}': {err}",
                document_id
            ),
            status: None,
            retryable: is_retryable_typesense_request_error(&err),
        })?;

    let status = response.status();
    if status == ReqwestStatusCode::NOT_FOUND {
        return Ok(false);
    }
    if !status.is_success() {
        let body = response.text().await.unwrap_or_default();
        return Err(TypesenseImportFailure {
            message: format!(
                "Typesense rollback delete failed for id '{}' with HTTP {}: {}",
                document_id,
                status.as_u16(),
                body
            ),
            status: Some(status),
            retryable: is_retryable_typesense_status(status),
        });
    }
    Ok(true)
}

async fn delete_document_with_retry<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    document_id: &str,
) -> Result<bool, TypesenseImportFailure> {
    let max_attempts = state.typesense_import_max_attempts();
    for attempt in 1..=max_attempts {
        match delete_document_once(state, connection, collection_name, document_id).await {
            Ok(deleted) => return Ok(deleted),
            Err(err) if err.retryable && attempt < max_attempts => {
                tokio::time::sleep(typesense_retry_delay(
                    state.typesense_import_retry_base_ms(),
                    attempt,
                ))
                .await;
            }
            Err(err) => return Err(err),
        }
    }
    Err(TypesenseImportFailure {
        message: "Typesense rollback delete exhausted all retries.".to_string(),
        status: None,
        retryable: false,
    })
}

async fn apply_saga_backups_for_batch<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    batch_documents: &[TypesenseNormalizedDocument],
    backup_state: &mut TypesenseSagaBackupState,
) -> Result<(), TypesenseImportFailure> {
    for document in batch_documents {
        if backup_state.original_documents.contains_key(&document.id) {
            continue;
        }
        let snapshot =
            fetch_document_backup_with_retry(state, connection, collection_name, &document.id)
                .await?;
        backup_state
            .original_documents
            .insert(document.id.clone(), snapshot);
    }
    Ok(())
}

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

    let response = state
        .http_client()
        .post(format!(
            "{}/collections/{collection_name}/documents/import",
            connection.base_url
        ))
        .timeout(Duration::from_millis(connection.timeout_ms))
        .header("X-TYPESENSE-API-KEY", &connection.api_key)
        .header("Content-Type", "text/plain")
        .query(&[
            ("action", "upsert".to_string()),
            ("return_id", "true".to_string()),
        ])
        .body(ndjson)
        .send()
        .await
        .map_err(|err| TypesenseImportFailure {
            message: format!("Typesense import request failed: {err}"),
            status: None,
            retryable: is_retryable_typesense_request_error(&err),
        })?;

    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|err| TypesenseImportFailure {
            message: format!("Failed to read Typesense import response: {err}"),
            status: Some(status),
            retryable: false,
        })?;

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

    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 < TYPESENSE_SAGA_PREVIEW_LIMIT {
            preview.push(parsed);
        }
    }

    if failed_count > 0 && imported_count == 0 {
        return Err(TypesenseImportFailure {
            message: format!(
                "Typesense import returned {} failed documents and no successful imports.",
                failed_count
            ),
            status: Some(status),
            retryable: false,
        });
    }

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

async fn import_documents_with_retry<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    ndjson: String,
) -> Result<(TypesenseImportSummary, usize), TypesenseImportFailure> {
    let max_attempts = state.typesense_import_max_attempts();
    for attempt in 1..=max_attempts {
        match import_documents_once(state, connection, collection_name, ndjson.clone()).await {
            Ok(summary) => return Ok((summary, attempt.saturating_sub(1))),
            Err(err) if err.retryable && attempt < max_attempts => {
                tokio::time::sleep(typesense_retry_delay(
                    state.typesense_import_retry_base_ms(),
                    attempt,
                ))
                .await;
            }
            Err(err) => return Err(err),
        }
    }
    Err(TypesenseImportFailure {
        message: "Typesense import exhausted all retries.".to_string(),
        status: None,
        retryable: false,
    })
}

async fn compensate_failed_import<R: TypesenseRuntime>(
    state: &R,
    connection: &TypesenseConnection,
    collection_name: &str,
    backup_state: &TypesenseSagaBackupState,
) -> TypesenseSagaCompensationResult {
    let mut restored_documents = Vec::new();
    let mut created_only_ids = Vec::new();

    for (document_id, original) in &backup_state.original_documents {
        match original {
            Some(document) => restored_documents.push(TypesenseNormalizedDocument {
                id: document_id.clone(),
                document: document.clone(),
            }),
            None => created_only_ids.push(document_id.clone()),
        }
    }

    let mut result = TypesenseSagaCompensationResult::default();

    if !restored_documents.is_empty() {
        match documents_to_ndjson(&restored_documents) {
            Ok(ndjson) => {
                let restore_result =
                    import_documents_with_retry(state, connection, collection_name, ndjson).await;
                match restore_result {
                    Ok((summary, _)) => {
                        result.restored_count = summary.imported_count;
                    }
                    Err(err) => {
                        result.error_message =
                            Some(format!("Rollback restore failed ({}).", err.message));
                    }
                }
            }
            Err(err) => {
                result.error_message =
                    Some(format!("Rollback restore serialization failed: {err}"));
            }
        }
    }

    for document_id in created_only_ids {
        match delete_document_with_retry(state, connection, collection_name, &document_id).await {
            Ok(deleted) => {
                if deleted {
                    result.deleted_count += 1;
                }
            }
            Err(err) => {
                if result.error_message.is_none() {
                    result.error_message = Some(format!(
                        "Rollback delete failed for id '{}': {}",
                        document_id, err.message
                    ));
                }
            }
        }
    }

    result
}

async fn execute_typesense_sync_job<R: TypesenseRuntime>(
    state: &R,
    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())?;
    ensure_typesense_sync_runs_schema_compat_once(&logging).await?;
    let job = load_typesense_job(&logging, job_id)
        .await
        .map_err(|err| err.to_string())?
        .ok_or_else(|| "Sync job not found.".to_string())?;
    let daemon_id = current_daemon_id();
    if !daemon_affinity_matches(job.execution_daemon_id.as_deref(), &daemon_id) {
        let target = normalize_daemon_id(job.execution_daemon_id.as_deref()).unwrap_or_default();
        return Err(format!(
            "{} '{}'; current daemon is '{}'.",
            TYPESENSE_DAEMON_AFFINITY_ERROR_PREFIX, target, daemon_id
        ));
    }
    let source_pool = state
        .postgres_pool_for_client(&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}"))?;
    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?;
    let connection = typesense_connection_from_job(&job);
    ensure_collection_exists(state, &connection, &prepared_schema, &job.collection_name).await?;
    let lock_token = match claim_sync_job(&logging, job_id)
        .await
        .map_err(|err| err.to_string())?
    {
        TypesenseClaimResult::Claimed(token) => token,
        TypesenseClaimResult::BusyOrDisabled => {
            return Err("Sync job is already running or is disabled.".to_string());
        }
        TypesenseClaimResult::WorkerAtCapacity => {
            return Err(format!(
                "Typesense worker is busy. At most {} run can execute at a time.",
                TYPESENSE_MAX_PARALLEL_RUNS
            ));
        }
    };

    let mut resumed = false;
    let mut resume_attempt_count = 0usize;
    let mut manifest_total_rows = None;
    let mut manifest_processed_rows = 0usize;
    let mut next_manifest_seq = 0usize;
    let mut source_row_count = 0usize;
    let mut imported_count = 0usize;
    let mut failed_count = 0usize;
    let mut batches_completed = 0usize;
    let mut retry_attempt_count = 0usize;
    let mut rollback_attempted = false;
    let mut rollback_status = "not_needed".to_string();
    let mut rollback_result = TypesenseSagaCompensationResult::default();

    let run_id = match async {
        if let Some(existing_run) = load_running_sync_run_state(&logging, job_id)
            .await
            .map_err(|err| err.to_string())?
        {
            if normalize_progress_total_kind(&existing_run.progress_total_kind) == "exact"
                && existing_run.manifest_total_rows.unwrap_or(0) >= 0
            {
                resumed = true;
                manifest_total_rows = existing_run
                    .manifest_total_rows
                    .and_then(|value| usize::try_from(value.max(0)).ok());
                manifest_processed_rows =
                    usize::try_from(existing_run.manifest_processed_rows.max(0)).unwrap_or(0);
                next_manifest_seq =
                    usize::try_from(existing_run.next_manifest_seq.max(0)).unwrap_or(0);
                source_row_count = existing_run
                    .source_row_count
                    .and_then(|value| usize::try_from(value.max(0)).ok())
                    .unwrap_or(manifest_total_rows.unwrap_or(manifest_processed_rows));
                imported_count = existing_run
                    .imported_count
                    .and_then(|value| usize::try_from(value.max(0)).ok())
                    .unwrap_or(0);
                failed_count = existing_run
                    .failed_count
                    .and_then(|value| usize::try_from(value.max(0)).ok())
                    .unwrap_or(0);
                batches_completed =
                    usize::try_from(existing_run.batches_completed.max(0)).unwrap_or(0);
                retry_attempt_count =
                    usize::try_from(existing_run.retry_attempt_count.max(0)).unwrap_or(0);
                rollback_attempted = existing_run.rollback_attempted;
                rollback_status = existing_run.rollback_status.clone();
                rollback_result.restored_count = existing_run
                    .rollback_restored_count
                    .and_then(|value| usize::try_from(value.max(0)).ok())
                    .unwrap_or(0);
                rollback_result.deleted_count = existing_run
                    .rollback_deleted_count
                    .and_then(|value| usize::try_from(value.max(0)).ok())
                    .unwrap_or(0);
                resume_attempt_count = bump_sync_run_resume_attempt(&logging, existing_run.id)
                    .await
                    .map_err(|err| err.to_string())?;
                Ok(existing_run.id)
            } else {
                let _ = update_sync_run_failure(
                    &logging,
                    existing_run.id,
                    "Existing running run had no resumable manifest snapshot metadata; restarted with a fresh run.",
                    0,
                )
                .await;
                create_sync_run(&logging, job_id, trigger_source)
                    .await
                    .map_err(|err| err.to_string())
            }
        } else {
            create_sync_run(&logging, job_id, trigger_source)
                .await
                .map_err(|err| err.to_string())
        }
    }
    .await
    {
        Ok(run_id) => run_id,
        Err(err) => {
            let _ = update_job_failure(&logging, job_id, &err, lock_token).await;
            return Err(err);
        }
    };

    if let Err(err) = async {
        if !resumed {
            clear_manifest_rows(&logging, run_id)
                .await
                .map_err(|err| err.to_string())?;
            update_sync_run_progress(
                &logging,
                run_id,
                &TypesenseSyncProgressUpdate {
                    source_row_count: estimated_total_rows.unwrap_or(0),
                    estimated_total_rows,
                    progress_total_kind: Some(if estimated_total_rows.is_some() {
                        "estimated".to_string()
                    } else {
                        "unknown".to_string()
                    }),
                    manifest_total_rows: None,
                    manifest_processed_rows: Some(0),
                    next_manifest_seq: Some(0),
                    resume_attempt_count: Some(0),
                    processed_row_count: 0,
                    imported_count: 0,
                    failed_count: 0,
                    progress_pct: 10,
                    progress_stage: "building_manifest".to_string(),
                    progress_message: Some(match estimated_total_rows {
                        Some(value) => format!(
                            "Preparing run manifest snapshot from {}.{} (catalog estimate around {} rows).",
                            job.source_schema_name, job.source_table_name, value
                        ),
                        None => format!(
                            "Preparing run manifest snapshot from {}.{}.",
                            job.source_schema_name, job.source_table_name
                        ),
                    }),
                    inter_batch_delay_ms: usize::try_from(clamp_batch_delay_ms(job.batch_delay_ms))
                        .unwrap_or(0),
                    next_batch_at: None,
                    batches_completed: 0,
                },
            )
            .await
            .map_err(|err| err.to_string())?;

            let manifest_count = build_manifest_snapshot(
                &source_pool,
                &logging,
                run_id,
                &job.source_schema_name,
                &job.source_table_name,
                &prepared_schema.selected_columns,
                &job.document_id_column,
            )
            .await?;
            manifest_total_rows = Some(manifest_count);
            source_row_count = manifest_count;
        }
        Ok::<(), String>(())
    }
    .await
    {
        let _ = update_sync_run_failure(&logging, run_id, &err, 0).await;
        let _ = update_job_failure(&logging, job_id, &err, lock_token).await;
        return Err(err);
    }

    let cancel_token = CancellationToken::new();
    register_typesense_run_cancel_token(run_id, cancel_token.clone());
    let _cancel_guard = TypesenseRunCancelGuard(run_id);
    let started = Instant::now();
    let mut preview = Vec::new();
    let mut backup_state = TypesenseSagaBackupState::default();
    let mut current_batch_delay_ms =
        usize::try_from(clamp_batch_delay_ms(job.batch_delay_ms)).unwrap_or(0);

    let execute_result = async {
        if cancel_token.is_cancelled() || run_is_cancelled(&logging, run_id).await.unwrap_or(false) {
            return Err(TYPESENSE_CANCELLED_ERROR.to_string());
        }
        if !renew_sync_job_lock(&logging, job_id, lock_token)
            .await
            .map_err(|err| err.to_string())?
        {
            return Err(format!(
                "Sync lock lost before job initialization for job {}. Another mirror is running this sync.",
                job_id
            ));
        }

        update_sync_run_progress(
            &logging,
            run_id,
            &TypesenseSyncProgressUpdate {
                source_row_count: manifest_total_rows.unwrap_or(source_row_count),
                estimated_total_rows,
                progress_total_kind: Some("exact".to_string()),
                manifest_total_rows,
                manifest_processed_rows: Some(manifest_processed_rows),
                next_manifest_seq: Some(next_manifest_seq),
                resume_attempt_count: Some(resume_attempt_count),
                processed_row_count: manifest_processed_rows,
                imported_count,
                failed_count,
                progress_pct: calculate_import_progress_pct(
                    manifest_processed_rows,
                    manifest_total_rows.or(estimated_total_rows),
                ),
                progress_stage: if resumed {
                    "resuming".to_string()
                } else {
                    "importing_documents".to_string()
                },
                progress_message: Some(if resumed {
                    format!(
                        "Resumed run checkpoint at manifest row {} of {} for '{}'.",
                        manifest_processed_rows,
                        manifest_total_rows.unwrap_or(0),
                        job.collection_name
                    )
                } else {
                    format!(
                        "Manifest snapshot prepared with {} rows. Starting import for '{}'.",
                        manifest_total_rows.unwrap_or(0),
                        job.collection_name
                    )
                }),
                inter_batch_delay_ms: current_batch_delay_ms,
                next_batch_at: None,
                batches_completed,
            },
        )
        .await
        .map_err(|err| err.to_string())?;

        let batch_size = usize::try_from(job.batch_size.max(1)).unwrap_or(1);
        let total_batches = manifest_total_rows.map(|rows| {
            if rows == 0 {
                0usize
            } else {
                rows.div_ceil(batch_size.max(1))
            }
        });

        loop {
            if cancel_token.is_cancelled()
                || run_is_cancelled(&logging, run_id)
                    .await
                    .map_err(|err| err.to_string())?
            {
                return Err(TYPESENSE_CANCELLED_ERROR.to_string());
            }
            if !renew_sync_job_lock(&logging, job_id, lock_token)
                .await
                .map_err(|err| err.to_string())?
            {
                return Err(format!(
                    "Sync lock lost while importing job {}. Aborting to avoid duplicate mirror execution.",
                    job_id
                ));
            }

            let manifest_batch = load_manifest_batch(&logging, run_id, next_manifest_seq, batch_size)
                .await?;
            if manifest_batch.is_empty() {
                break;
            }

            let last_seq = manifest_batch.last().map(|row| row.manifest_seq).unwrap_or(0);
            let mut documents = Vec::with_capacity(manifest_batch.len());
            for row in manifest_batch {
                documents.push(row.row_data);
            }

            let current_batch_len = documents.len();
            manifest_processed_rows = manifest_processed_rows.saturating_add(current_batch_len);
            next_manifest_seq = usize::try_from(last_seq.saturating_add(1)).unwrap_or(usize::MAX);
            source_row_count = manifest_total_rows.unwrap_or(manifest_processed_rows);

            let normalized_documents = normalize_rows_for_import(
                documents,
                &job.document_id_column,
                &prepared_schema.field_types,
            )?;
            if state.typesense_sync_saga_backup_enabled() {
                apply_saga_backups_for_batch(
                    state,
                    &connection,
                    &job.collection_name,
                    &normalized_documents,
                    &mut backup_state,
                )
                .await
                .map_err(|err| err.message.clone())?;
            }

            let ndjson = documents_to_ndjson(&normalized_documents)?;
            let batch_summary = match import_documents_with_retry(
                state,
                &connection,
                &job.collection_name,
                ndjson,
            )
            .await
            {
                Ok((summary, retries_used)) => {
                    if retries_used > 0 {
                        retry_attempt_count += retries_used;
                        let _ = update_sync_run_retry_attempts(
                            &logging,
                            run_id,
                            retry_attempt_count,
                        )
                        .await;
                    }
                    summary
                }
                Err(err) => {
                    if state.typesense_sync_saga_backup_enabled()
                        && !backup_state.original_documents.is_empty()
                    {
                        rollback_attempted = true;
                        let _ = update_sync_run_rollback_state(
                            &logging,
                            run_id,
                            true,
                            "running",
                            0,
                            0,
                            None,
                        )
                        .await;
                        let _ = update_sync_run_progress(
                            &logging,
                            run_id,
                            &TypesenseSyncProgressUpdate {
                                source_row_count,
                                estimated_total_rows,
                                progress_total_kind: Some("exact".to_string()),
                                manifest_total_rows,
                                manifest_processed_rows: Some(manifest_processed_rows),
                                next_manifest_seq: Some(next_manifest_seq),
                                resume_attempt_count: Some(resume_attempt_count),
                                processed_row_count: manifest_processed_rows,
                                imported_count,
                                failed_count,
                                progress_pct: calculate_import_progress_pct(
                                    manifest_processed_rows,
                                    manifest_total_rows.or(estimated_total_rows),
                                ),
                                progress_stage: "rolling_back_batch".to_string(),
                                progress_message: Some(
                                    "Import retries exhausted. Rolling back this run to the pre-import state."
                                        .to_string(),
                                ),
                                inter_batch_delay_ms: current_batch_delay_ms,
                                next_batch_at: None,
                                batches_completed,
                            },
                        )
                        .await;

                        rollback_result = compensate_failed_import(
                            state,
                            &connection,
                            &job.collection_name,
                            &backup_state,
                        )
                        .await;
                        rollback_status = if rollback_result.error_message.is_some() {
                            "failed".to_string()
                        } else {
                            "succeeded".to_string()
                        };
                        let _ = update_sync_run_rollback_state(
                            &logging,
                            run_id,
                            true,
                            rollback_status.as_str(),
                            rollback_result.restored_count,
                            rollback_result.deleted_count,
                            rollback_result.error_message.as_deref(),
                        )
                        .await;
                    }

                    let status_label = err
                        .status
                        .map(|status| format!("HTTP {}", status.as_u16()))
                        .unwrap_or_else(|| "network failure".to_string());
                    let rollback_note = if rollback_attempted {
                        match &rollback_result.error_message {
                            Some(rollback_error) => format!(
                                " Rollback attempted but did not fully succeed: {}",
                                rollback_error
                            ),
                            None => format!(
                                " Rollback succeeded: restored {} previous documents and removed {} new documents.",
                                rollback_result.restored_count, rollback_result.deleted_count
                            ),
                        }
                    } else {
                        String::new()
                    };
                    return Err(format!(
                        "Typesense batch import failed ({status_label}): {}.{}",
                        err.message, rollback_note
                    ));
                }
            };

            imported_count += batch_summary.imported_count;
            failed_count += batch_summary.failed_count;
            batches_completed += 1;

            if preview.len() < TYPESENSE_SAGA_PREVIEW_LIMIT {
                let remaining = TYPESENSE_SAGA_PREVIEW_LIMIT - 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,
                    progress_total_kind: Some("exact".to_string()),
                    manifest_total_rows,
                    manifest_processed_rows: Some(manifest_processed_rows),
                    next_manifest_seq: Some(next_manifest_seq),
                    resume_attempt_count: Some(resume_attempt_count),
                    processed_row_count: manifest_processed_rows,
                    imported_count,
                    failed_count,
                    progress_pct: calculate_import_progress_pct(
                        manifest_processed_rows,
                        manifest_total_rows.or(estimated_total_rows),
                    ),
                    progress_stage: "importing_documents".to_string(),
                    progress_message: Some(match total_batches {
                        Some(total) if total > 0 => format!(
                            "Imported batch {} of {} snapshot batches into '{}'. Retries so far: {}.",
                            batches_completed, total, job.collection_name, retry_attempt_count
                        ),
                        _ => format!(
                            "Imported {} manifest rows into '{}'. Retries so far: {}.",
                            manifest_processed_rows, job.collection_name, retry_attempt_count
                        ),
                    }),
                    inter_batch_delay_ms: current_batch_delay_ms,
                    next_batch_at: None,
                    batches_completed,
                },
            )
            .await
            .map_err(|err| err.to_string())?;

            current_batch_delay_ms =
                usize::try_from(load_current_batch_delay_ms(&logging, job.id).await.map_err(
                    |err| format!("Failed to load current batch delay for job {}: {err}", job.id),
                )?)
                .unwrap_or(0);
            if current_batch_delay_ms > 0 {
                let delay_ms_i64 = i64::try_from(current_batch_delay_ms).unwrap_or(i64::MAX);
                let next_batch_at = Utc::now() + chrono::Duration::milliseconds(delay_ms_i64);
                let _ = update_sync_run_progress(
                    &logging,
                    run_id,
                    &TypesenseSyncProgressUpdate {
                        source_row_count,
                        estimated_total_rows,
                        progress_total_kind: Some("exact".to_string()),
                        manifest_total_rows,
                        manifest_processed_rows: Some(manifest_processed_rows),
                        next_manifest_seq: Some(next_manifest_seq),
                        resume_attempt_count: Some(resume_attempt_count),
                        processed_row_count: manifest_processed_rows,
                        imported_count,
                        failed_count,
                        progress_pct: calculate_import_progress_pct(
                            manifest_processed_rows,
                            manifest_total_rows.or(estimated_total_rows),
                        ),
                        progress_stage: "waiting_next_batch".to_string(),
                        progress_message: Some(format!(
                            "Waiting {}ms before the next batch. You can change this delay from the sync job settings.",
                            current_batch_delay_ms
                        )),
                        inter_batch_delay_ms: current_batch_delay_ms,
                        next_batch_at: Some(next_batch_at),
                        batches_completed,
                    },
                )
                .await;

                let mut remaining_ms = u64::try_from(current_batch_delay_ms).unwrap_or(0);
                while remaining_ms > 0 {
                    if cancel_token.is_cancelled()
                        || run_is_cancelled(&logging, run_id)
                            .await
                            .map_err(|err| err.to_string())?
                    {
                        return Err(TYPESENSE_CANCELLED_ERROR.to_string());
                    }
                    let step_ms = remaining_ms.min(1_000);
                    tokio::time::sleep(Duration::from_millis(step_ms)).await;
                    remaining_ms = remaining_ms.saturating_sub(step_ms);
                }
            }
        }

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

        if cancel_token.is_cancelled()
            || run_is_cancelled(&logging, run_id)
                .await
                .map_err(|err| err.to_string())?
        {
            return Err(TYPESENSE_CANCELLED_ERROR.to_string());
        }

        let run_updated = update_sync_run_success(
            &logging,
            run_id,
            source_row_count,
            estimated_total_rows,
            "exact",
            manifest_total_rows,
            manifest_processed_rows,
            next_manifest_seq,
            resume_attempt_count,
            imported_count,
            failed_count,
            batches_completed,
            duration_ms,
            retry_attempt_count,
            rollback_attempted,
            rollback_status.as_str(),
            rollback_result.restored_count,
            rollback_result.deleted_count,
            rollback_result.error_message.as_deref(),
            &result_preview,
        )
        .await
        .map_err(|err| err.to_string())?;
        if !run_updated {
            return Err(TYPESENSE_CANCELLED_ERROR.to_string());
        }
        update_job_success(
            &logging,
            &job,
            source_row_count,
            imported_count,
            failed_count,
            lock_token,
        )
        .await
        .map_err(|err| err.to_string())
        .and_then(|updated| {
            if updated {
                Ok(())
            } else {
                Err(format!(
                    "Sync lock ownership changed before finalizing job {}. Job state was not overwritten.",
                    job_id
                ))
            }
        })?;

        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,
            retry_attempt_count,
            rollback_attempted,
            rollback_status,
            rollback_restored_count: rollback_result.restored_count,
            rollback_deleted_count: rollback_result.deleted_count,
            result_preview,
            progress_total_kind: "exact".to_string(),
            manifest_total_rows,
            manifest_processed_rows,
            resumed,
        })
    }
    .await;

    if let Err(err) = &execute_result {
        if err == TYPESENSE_CANCELLED_ERROR {
            let _ = update_sync_run_cancelled(&logging, run_id, "Cancelled by operator.").await;
            let _ = update_job_cancelled(&logging, job_id, "Cancelled by operator.").await;
        } else {
            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, lock_token).await;
        }
    }

    execute_result
}

#[cfg(test)]
mod tests {
    use super::{
        calculate_import_progress_pct, is_retryable_typesense_status,
        is_typesense_worker_busy_error, normalize_catalog_row_estimate,
        normalize_progress_total_kind, typesense_retry_delay,
    };
    use reqwest::StatusCode as ReqwestStatusCode;

    #[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);
        assert_eq!(calculate_import_progress_pct(733_500, Some(708_995)), 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));
    }

    #[test]
    fn retryable_status_classification_matches_expected() {
        assert!(is_retryable_typesense_status(
            ReqwestStatusCode::TOO_MANY_REQUESTS
        ));
        assert!(is_retryable_typesense_status(
            ReqwestStatusCode::SERVICE_UNAVAILABLE
        ));
        assert!(is_retryable_typesense_status(
            ReqwestStatusCode::INTERNAL_SERVER_ERROR
        ));
        assert!(!is_retryable_typesense_status(
            ReqwestStatusCode::BAD_REQUEST
        ));
        assert!(!is_retryable_typesense_status(
            ReqwestStatusCode::UNAUTHORIZED
        ));
    }

    #[test]
    fn retry_delay_is_bounded_and_non_decreasing() {
        let first = typesense_retry_delay(400, 1);
        let second = typesense_retry_delay(400, 2);
        let third = typesense_retry_delay(400, 3);
        let fourth = typesense_retry_delay(400, 4);
        assert!(first <= second);
        assert!(second <= third);
        assert!(third <= fourth);
        assert!(fourth <= std::time::Duration::from_millis(5_000));
    }

    #[test]
    fn worker_busy_error_classification_matches_expected_message() {
        assert!(is_typesense_worker_busy_error(
            "Typesense worker is busy. At most 1 run can execute at a time."
        ));
    }

    #[test]
    fn worker_busy_error_classification_ignores_other_failures() {
        assert!(!is_typesense_worker_busy_error(
            "Typesense sync job failed: failed to load source rows"
        ));
    }

    #[test]
    fn progress_total_kind_normalization_is_strict() {
        assert_eq!(normalize_progress_total_kind("exact"), "exact");
        assert_eq!(normalize_progress_total_kind("estimated"), "estimated");
        assert_eq!(normalize_progress_total_kind("unknown"), "unknown");
        assert_eq!(normalize_progress_total_kind("EXACT"), "unknown");
        assert_eq!(normalize_progress_total_kind("something_else"), "unknown");
    }
}

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.batch_delay_ms,
            j.sync_interval_minutes,
            j.stale_after_minutes,
            j.execution_daemon_id,
            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
}

/// Validates one direct Typesense backend connection and returns health details.
pub async fn test_typesense_backend<R: TypesenseRuntime>(
    body: &TypesenseConnectionRequest,
    app_state: &R,
) -> HttpResponse {
    let connection = match normalize_typesense_connection(body, app_state.typesense_allow_http()) {
        Ok(connection) => connection,
        Err(resp) => return resp,
    };

    match typesense_healthcheck(app_state, &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),
    }
}

/// Lists collections exposed by one direct Typesense backend connection.
pub async fn list_typesense_collections<R: TypesenseRuntime>(
    body: &TypesenseConnectionRequest,
    app_state: &R,
) -> HttpResponse {
    let connection = match normalize_typesense_connection(body, app_state.typesense_allow_http()) {
        Ok(connection) => connection,
        Err(resp) => return resp,
    };

    let started = Instant::now();
    let (status, payload) =
        match typesense_get_json(app_state, &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,
        }),
    )
}

/// Executes one configured Typesense sync job immediately.
pub async fn run_typesense_sync_job<R: TypesenseRuntime>(
    app_state: &R,
    job_id: i64,
) -> HttpResponse {
    let Some(logging) = logging_pool_from_state(app_state) else {
        return service_unavailable(
            "Logging store unavailable",
            "Configure gateway.logging_client and ensure it is connected before running Typesense sync jobs.",
        );
    };
    if let Err(err) = ensure_typesense_sync_runs_schema_compat_once(&logging).await {
        return service_unavailable("Typesense schema compatibility unavailable", err);
    }

    match load_typesense_job(&logging, job_id).await {
        Ok(Some(_)) => {}
        Ok(None) => {
            return not_found(
                "Typesense sync job not found",
                format!("No Typesense sync job exists for id {}", job_id),
            );
        }
        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, job_id, "manual").await {
        Ok(result) => api_success("Typesense sync completed", json!(result)),
        Err(err) if err.contains("already running") || err.contains("worker is busy") => {
            conflict("Typesense sync job is already running", err)
        }
        Err(err)
            if err.contains("not currently connected")
                || err.contains(TYPESENSE_DAEMON_AFFINITY_ERROR_PREFIX) =>
        {
            service_unavailable("Typesense sync deferred", err)
        }
        Err(err) => internal_error("Typesense sync failed", err),
    }
}

/// Cancels the active run for a configured Typesense sync job.
pub async fn cancel_typesense_sync_job<R: TypesenseRuntime>(
    app_state: &R,
    job_id: i64,
) -> HttpResponse {
    let Some(pool) = logging_pool_from_state(app_state) else {
        return service_unavailable(
            "Logging store unavailable",
            "Configure gateway.logging_client and ensure it is connected before cancelling Typesense sync jobs.",
        );
    };

    let run_id = match load_running_run_for_job(&pool, job_id).await {
        Ok(Some(run_id)) => run_id,
        Ok(None) => {
            return bad_request(
                "Typesense job not active",
                "Only running sync jobs can be cancelled.",
            );
        }
        Err(err) if sqlx_relation_missing(&err) => return typesense_not_provisioned_response(),
        Err(err) => {
            return internal_error("Failed to cancel Typesense sync job", err.to_string());
        }
    };

    match update_sync_run_cancelled(&pool, run_id, "Cancelled by operator.").await {
        Ok(Some(_)) => {}
        Ok(None) => {
            return bad_request(
                "Typesense job not active",
                "Only running sync jobs can be cancelled.",
            );
        }
        Err(err) if sqlx_relation_missing(&err) => return typesense_not_provisioned_response(),
        Err(err) => {
            return internal_error("Failed to cancel Typesense sync job", err.to_string());
        }
    }

    if let Err(err) = update_job_cancelled(&pool, job_id, "Cancelled by operator.").await {
        return internal_error("Failed to cancel Typesense sync job", err.to_string());
    }
    trigger_typesense_run_cancel_token(run_id);

    api_success(
        "Typesense sync job cancelled",
        json!({
            "job_id": job_id,
            "run_id": run_id,
            "status": "cancelled",
        }),
    )
}

/// Force-kills a running Typesense sync run.
pub async fn kill_typesense_sync_run<R: TypesenseRuntime>(
    app_state: &R,
    run_id: i64,
) -> HttpResponse {
    let Some(pool) = logging_pool_from_state(app_state) else {
        return service_unavailable(
            "Logging store unavailable",
            "Configure gateway.logging_client and ensure it is connected before killing Typesense sync runs.",
        );
    };

    let job_id = match update_sync_run_cancelled(&pool, run_id, "Force killed by operator.").await {
        Ok(Some(job_id)) => job_id,
        Ok(None) => {
            return bad_request(
                "Typesense run not active",
                "Only running sync runs can be killed.",
            );
        }
        Err(err) if sqlx_relation_missing(&err) => return typesense_not_provisioned_response(),
        Err(err) => {
            return internal_error("Failed to kill Typesense sync run", err.to_string());
        }
    };

    if let Err(err) = update_job_cancelled(&pool, job_id, "Force killed by operator.").await {
        return internal_error("Failed to kill Typesense sync run", err.to_string());
    }
    trigger_typesense_run_cancel_token(run_id);

    api_success(
        "Typesense sync run killed",
        json!({
            "job_id": job_id,
            "run_id": run_id,
            "status": "cancelled",
        }),
    )
}

/// Searches a Typesense collection through one configured Athena sync binding.
pub async fn search_typesense<R: TypesenseRuntime>(
    body: &TypesenseSearchRequest,
    app_state: &R,
) -> HttpResponse {
    if body.q.trim().is_empty() {
        return bad_request("Invalid request", "q is required");
    }

    let Some(pool) = logging_pool_from_state(app_state) else {
        return service_unavailable(
            "Logging store unavailable",
            "Configure gateway.logging_client and ensure it is connected before running Typesense search.",
        );
    };
    if let Err(err) = ensure_typesense_sync_runs_schema_compat_once(&pool).await {
        return service_unavailable("Typesense schema compatibility unavailable", err);
    }
    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,
        &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,
        }),
    )
}

/// Runs one scheduled-sync worker tick against all due jobs for the current daemon.
pub async fn run_due_sync_jobs_once<R: TypesenseRuntime>(state: &R) -> Result<bool, String> {
    let Some(logging) = logging_pool_from_state(state) else {
        return Ok(false);
    };

    ensure_typesense_sync_runs_schema_compat_once(&logging).await?;
    let daemon_id = current_daemon_id();
    let due_jobs = due_jobs_for_daemon(&logging, &daemon_id)
        .await
        .map_err(|err| err.to_string())?;
    if due_jobs.is_empty() {
        return Ok(false);
    }

    let mut made_non_busy_progress = false;
    let mut saw_worker_busy = false;
    let mut attempted_job = false;

    for due_job in due_jobs {
        let job_id = due_job.id;
        if !daemon_affinity_matches(due_job.execution_daemon_id.as_deref(), &daemon_id) {
            debug!(
                job_id,
                daemon_id = %daemon_id,
                target = ?normalize_daemon_id(due_job.execution_daemon_id.as_deref()),
                "Typesense scheduled sync job ignored due to daemon affinity mismatch"
            );
            continue;
        }
        if state
            .postgres_pool_for_client(&due_job.source_client_name)
            .is_none()
        {
            debug!(
                job_id,
                daemon_id = %daemon_id,
                source_client = %due_job.source_client_name,
                "Typesense scheduled sync job deferred because source client is not connected on this daemon"
            );
            continue;
        }

        attempted_job = true;
        match execute_typesense_sync_job(state, job_id, "scheduled").await {
            Ok(_) => {
                made_non_busy_progress = true;
            }
            Err(err) if is_typesense_worker_busy_error(&err) => {
                saw_worker_busy = true;
                debug!(job_id, error = %err, "Typesense scheduled sync job deferred");
            }
            Err(err) if is_typesense_progress_total_kind_missing_error(&err) => {
                warn!(
                    job_id,
                    error = %err,
                    "Typesense scheduled sync hit schema drift; reconciling and retrying once"
                );
                match ensure_typesense_sync_runs_schema_compat_once(&logging).await {
                    Ok(()) => match execute_typesense_sync_job(state, job_id, "scheduled").await {
                        Ok(_) => {
                            made_non_busy_progress = true;
                            debug!(
                                job_id,
                                "Typesense scheduled sync recovered after schema reconciliation"
                            );
                        }
                        Err(retry_err) => {
                            made_non_busy_progress = true;
                            warn!(
                                job_id,
                                error = %retry_err,
                                "Typesense scheduled sync retry failed after schema reconciliation"
                            );
                        }
                    },
                    Err(reconcile_err) => {
                        made_non_busy_progress = true;
                        warn!(
                            job_id,
                            error = %reconcile_err,
                            "Typesense schema reconciliation failed before scheduled retry"
                        );
                    }
                }
            }
            Err(err) => {
                made_non_busy_progress = true;
                warn!(job_id, error = %err, "Typesense scheduled sync job failed");
            }
        }
    }

    if !attempted_job {
        return Ok(false);
    }
    if saw_worker_busy && !made_non_busy_progress {
        return Ok(false);
    }

    Ok(true)
}

/// Warms the Typesense sync-run schema compatibility cache for one logging pool.
pub async fn ensure_typesense_sync_runs_schema_compat_once_for_pool(
    pool: &PgPool,
) -> Result<(), String> {
    ensure_typesense_sync_runs_schema_compat_once(pool).await
}
