context_harness/

ingest.rs

//! Ingestion pipeline orchestration.
//!
//! Coordinates the full sync flow: connector → normalization → chunking →
//! embedding → storage. Supports incremental sync via checkpoints and
//! inline embedding (non-fatal on failure).
//!
//! # Sync Pipeline
//!
//! The [`run_sync`] function implements the following pipeline:
//!
//! 1. **Resolve targets** — expands the connector argument into one or more
//!    `(source_label, connector_type, instance_name)` tuples. Supports:
//!    - `"git:platform"` — a single named instance
//!    - `"git"` — all instances of a type
//!    - `"all"` — every configured connector
//! 2. **Scan (parallel)** — dispatches to the appropriate connector for each
//!    target, running scans concurrently via [`tokio::task::JoinSet`].
//! 3. **Filter** — applies checkpoint, `--since`, `--until`, and `--limit`
//!    filters to each connector's items.
//! 4. **Upsert documents** — inserts or updates each item in the `documents`
//!    table, computing a SHA-256 deduplication hash.
//! 5. **Replace chunks** — deletes old chunks (and their embeddings/FTS entries)
//!    for the document, then inserts fresh chunks.
//! 6. **Inline embed** — if embeddings are enabled, embeds new chunks
//!    immediately (non-fatal: failures are logged but do not abort the sync).
//! 7. **Update checkpoint** — persists the latest `updated_at` timestamp
//!    so the next incremental sync can skip unchanged items.
//!
//! # Deduplication
//!
//! Each document is identified by `(source, source_id)`. If a document with
//! the same composite key already exists, it is updated via an `ON CONFLICT`
//! upsert. The `dedup_hash` field is a SHA-256 digest of source + source_id +
//! updated_at + body, enabling downstream consumers to detect changes.
//!
//! # Checkpointing
//!
//! Checkpoints are stored in the `checkpoints` table as `(source, cursor)`
//! pairs. The cursor is the maximum `updated_at` timestamp seen during the
//! sync. On subsequent runs, only items newer than the checkpoint are processed.
//!
//! # Multi-Instance Connectors
//!
//! All connector types support named instances. Documents are tagged with
//! `source = "type:name"` (e.g. `"git:platform"`, `"filesystem:docs"`).
//! When syncing a type (e.g. `ctx sync git`), all instances of that type
//! are scanned in parallel.

use anyhow::{bail, Result};
use chrono::NaiveDate;
use sha2::{Digest, Sha256};
use sqlx::SqlitePool;
use uuid::Uuid;

use crate::chunk::chunk_text;
use crate::config::Config;
use crate::db;
use crate::embed_cmd;
use crate::extract;
use crate::models::SourceItem;
use crate::progress::{SyncProgressEvent, SyncProgressReporter};
use crate::traits::{Connector, ConnectorRegistry};

/// Default max extract size when connector is not filesystem or name not found (spec §4.1).
const DEFAULT_MAX_EXTRACT_BYTES: u64 = 50_000_000;

/// Resolve max_extract_bytes for a source from config. Parses "filesystem:name" and looks up
/// the connector config; non-filesystem or unknown name uses DEFAULT_MAX_EXTRACT_BYTES.
fn max_extract_bytes_for_source(config: &Config, source_label: &str) -> u64 {
    if let Some(name) = source_label.strip_prefix("filesystem:") {
        config
            .connectors
            .filesystem
            .get(name)
            .map(|c| c.max_extract_bytes)
            .unwrap_or(DEFAULT_MAX_EXTRACT_BYTES)
    } else {
        DEFAULT_MAX_EXTRACT_BYTES
    }
}

/// Resolve a connector argument into a filtered list of connectors to scan.
///
/// The `registry` contains all connectors (built-in + custom). This function
/// filters them based on the user's connector argument.
///
/// # Supported Formats
///
/// | Input | Meaning |
/// |-------|---------|
/// | `"all"` | Every registered connector |
/// | `"git"` | All connectors of type `"git"` |
/// | `"filesystem"` | All connectors of type `"filesystem"` |
/// | `"s3"` | All connectors of type `"s3"` |
/// | `"script"` | All connectors of type `"script"` |
/// | `"custom"` | All connectors of type `"custom"` |
/// | `"git:platform"` | Specific named instance |
/// | `"custom:myconn"` | Specific named instance |
fn resolve_connectors<'a>(
    registry: &'a ConnectorRegistry,
    connector_arg: &str,
) -> Result<Vec<&'a dyn Connector>> {
    match connector_arg {
        "all" => {
            let all = registry.connectors();
            if all.is_empty() {
                bail!("No connectors configured. Add connector sections to your config.");
            }
            Ok(all.iter().map(|c| c.as_ref()).collect())
        }
        // Type-level filter: "git", "filesystem", "s3", "script", "custom"
        conn_type if matches!(conn_type, "filesystem" | "git" | "s3" | "script" | "custom") => {
            let matched = registry.connectors_by_type(conn_type);
            if matched.is_empty() {
                bail!("No {} connectors configured.", conn_type);
            }
            Ok(matched)
        }
        // Instance-level filter: "git:platform", "custom:mything"
        other => {
            if let Some((conn_type, name)) = other.split_once(':') {
                let conn = registry.find(conn_type, name).ok_or_else(|| {
                    let available: Vec<String> = registry
                        .connectors_by_type(conn_type)
                        .iter()
                        .map(|c| c.name().to_string())
                        .collect();
                    anyhow::anyhow!(
                        "No {} connector '{}'. Available: {}",
                        conn_type,
                        name,
                        if available.is_empty() {
                            "(none)".to_string()
                        } else {
                            available.join(", ")
                        }
                    )
                })?;
                Ok(vec![conn])
            } else {
                bail!(
                    "Unknown connector: '{}'. Use: all, filesystem, git, s3, script, custom, or type:name",
                    other
                );
            }
        }
    }
}

/// Runs the full ingestion pipeline for the specified connector(s).
///
/// This is the entry point for `ctx sync <connector>`. It builds a
/// [`ConnectorRegistry`] from the config and runs the pipeline.
///
/// # Arguments
///
/// - `config` — application configuration.
/// - `connector` — the connector specifier: `"all"`, a type name (`"git"`),
///   or a specific instance (`"git:platform"`).
/// - `full` — if `true`, ignores existing checkpoints and reprocesses all items.
/// - `dry_run` — if `true`, scans and counts items without writing to the database.
/// - `since` — optional `YYYY-MM-DD` date; only items updated on or after this date are processed.
/// - `until` — optional `YYYY-MM-DD` date; only items updated on or before this date are processed.
/// - `limit` — optional maximum number of items to process (per connector instance).
/// - `progress` — optional progress reporter; when provided, progress is emitted on stderr.
///
/// # Errors
///
/// Returns an error if:
/// - The specified connector is unknown or not configured.
/// - A connector fails to scan (e.g., network error).
/// - A database operation fails.
#[allow(clippy::too_many_arguments)]
pub async fn run_sync(
    config: &Config,
    connector: &str,
    full: bool,
    dry_run: bool,
    since: Option<String>,
    until: Option<String>,
    limit: Option<usize>,
    progress: Option<&dyn SyncProgressReporter>,
) -> Result<()> {
    let registry = ConnectorRegistry::from_config(config);
    run_sync_with_registry(
        config, connector, full, dry_run, since, until, limit, &registry, progress,
    )
    .await
}

/// Runs the ingestion pipeline with additional custom connectors.
///
/// Like [`run_sync`], but accepts extra connectors to merge with
/// the built-in ones from config. This is the entry point for custom
/// binaries that register `Connector` trait implementations.
///
/// The `extra_connectors` are combined with the built-in connectors
/// resolved from the config file.
#[allow(clippy::too_many_arguments, dead_code)]
pub async fn run_sync_with_extensions(
    config: &Config,
    connector: &str,
    full: bool,
    dry_run: bool,
    since: Option<String>,
    until: Option<String>,
    limit: Option<usize>,
    extra_connectors: &ConnectorRegistry,
) -> Result<()> {
    // Build combined connector list from config + extras
    let built_in = ConnectorRegistry::from_config(config);

    // Resolve from built-in registry (ignore error if extras might match)
    let mut resolved: Vec<&dyn Connector> = match resolve_connectors(&built_in, connector) {
        Ok(r) => r,
        Err(_) if !extra_connectors.is_empty() => Vec::new(),
        Err(e) => return Err(e),
    };

    // Also resolve from extras
    if let Ok(extras) = resolve_connectors(extra_connectors, connector) {
        resolved.extend(extras);
    }

    if resolved.is_empty() {
        bail!("No connectors matched '{}'.", connector);
    }

    run_connectors(config, &resolved, full, dry_run, since, until, limit, None).await
}

/// Runs the ingestion pipeline with a pre-built [`ConnectorRegistry`].
///
/// This is the unified entry point. All connectors in the registry are
/// trait objects — built-in, Lua, and custom all go through the same path.
#[allow(clippy::too_many_arguments)]
pub async fn run_sync_with_registry(
    config: &Config,
    connector: &str,
    full: bool,
    dry_run: bool,
    since: Option<String>,
    until: Option<String>,
    limit: Option<usize>,
    registry: &ConnectorRegistry,
    progress: Option<&dyn SyncProgressReporter>,
) -> Result<()> {
    let connectors = resolve_connectors(registry, connector)?;
    run_connectors(
        config,
        &connectors,
        full,
        dry_run,
        since,
        until,
        limit,
        progress,
    )
    .await
}

/// Report progress every N items during ingest (avoids flooding stderr).
const INGEST_PROGRESS_INTERVAL: u64 = 10;

/// Core sync engine: scans connectors sequentially, ingests items.
///
/// Scans are run sequentially since connectors hold references. Each
/// connector's items flow through the standard checkpoint → filter → upsert →
/// chunk → embed pipeline.
#[allow(clippy::too_many_arguments)]
async fn run_connectors(
    config: &Config,
    connectors: &[&dyn Connector],
    full: bool,
    dry_run: bool,
    since: Option<String>,
    until: Option<String>,
    limit: Option<usize>,
    progress: Option<&dyn SyncProgressReporter>,
) -> Result<()> {
    if connectors.len() > 1 {
        println!("Syncing {} connector instances...", connectors.len());
    }

    // Scan all connectors and collect results
    let mut scan_results: Vec<(String, Vec<SourceItem>)> = Vec::new();
    let mut scan_errors: Vec<String> = Vec::new();

    for conn in connectors {
        let label = conn.source_label();
        if let Some(p) = progress {
            p.report(SyncProgressEvent::Discovering {
                connector: label.to_string(),
            });
        }
        match conn.scan().await {
            Ok(items) => {
                scan_results.push((label, items));
            }
            Err(e) => {
                scan_errors.push(format!("{}: {:#}", label, e));
            }
        }
    }

    // Report scan errors but continue with successful scans
    for err in &scan_errors {
        eprintln!("Warning: scan failed: {}", err);
    }

    if scan_results.is_empty() && !scan_errors.is_empty() {
        bail!("All connector scans failed:\n{}", scan_errors.join("\n"));
    }

    // Sort for deterministic output ordering
    scan_results.sort_by(|a, b| a.0.cmp(&b.0));

    // Ingest each target's items (sequential — SQLite writes are serialized)
    let pool = db::connect(config).await?;

    for (source_label, mut items) in scan_results {
        // Load checkpoint
        let checkpoint: Option<i64> = if full {
            None
        } else {
            get_checkpoint(&pool, &source_label).await?
        };

        // Filter by checkpoint (skip files not modified since checkpoint)
        if let Some(cp) = checkpoint {
            items.retain(|item| item.updated_at.timestamp() > cp);
        }

        // Apply --since filter
        if let Some(ref since_str) = since {
            let since_date = NaiveDate::parse_from_str(since_str, "%Y-%m-%d")?;
            let since_ts = since_date
                .and_hms_opt(0, 0, 0)
                .unwrap()
                .and_utc()
                .timestamp();
            items.retain(|item| item.updated_at.timestamp() >= since_ts);
        }

        // Apply --until filter
        if let Some(ref until_str) = until {
            let until_date = NaiveDate::parse_from_str(until_str, "%Y-%m-%d")?;
            let until_ts = until_date
                .and_hms_opt(23, 59, 59)
                .unwrap()
                .and_utc()
                .timestamp();
            items.retain(|item| item.updated_at.timestamp() <= until_ts);
        }

        // Apply --limit (per connector instance)
        if let Some(lim) = limit {
            items.truncate(lim);
        }

        if dry_run {
            println!("sync {} (dry-run)", source_label);
            println!("  items found: {}", items.len());
            let total_chunks: usize = items
                .iter()
                .map(|item| chunk_text("tmp", &item.body, config.chunking.max_tokens).len())
                .sum();
            println!("  estimated chunks: {}", total_chunks);
            continue;
        }

        let mut docs_upserted = 0u64;
        let mut chunks_written = 0u64;
        let mut embeddings_written = 0u64;
        let mut embeddings_pending = 0u64;
        let mut extraction_skipped = 0u64;
        let mut max_updated: i64 = checkpoint.unwrap_or(0);
        let max_extract_bytes = max_extract_bytes_for_source(config, &source_label);
        let total_items = items.len() as u64;

        if let Some(p) = progress {
            if total_items > 0 {
                p.report(SyncProgressEvent::Ingesting {
                    connector: source_label.clone(),
                    n: 0,
                    total: total_items,
                });
            }
        }

        for item in items.iter_mut() {
            if let Some(ref bytes) = item.raw_bytes {
                if bytes.len() as u64 > max_extract_bytes {
                    extraction_skipped += 1;
                    eprintln!(
                        "Warning: skipping {} (size {} > max_extract_bytes {})",
                        item.source_id,
                        bytes.len(),
                        max_extract_bytes
                    );
                    continue;
                }
                match extract::extract_text(bytes, &item.content_type) {
                    Ok(text) => {
                        item.body = text;
                        item.raw_bytes = None;
                    }
                    Err(e) => {
                        extraction_skipped += 1;
                        eprintln!("Warning: extraction failed for {}: {}", item.source_id, e);
                        continue;
                    }
                }
            }

            let doc_id = upsert_document(&pool, item).await?;
            let chunks = chunk_text(&doc_id, &item.body, config.chunking.max_tokens);
            let chunk_count = chunks.len() as u64;
            replace_chunks(&pool, &doc_id, &chunks).await?;

            // Inline embedding (non-fatal)
            let (emb_ok, emb_pending) =
                embed_cmd::embed_chunks_inline(config, &pool, &chunks).await;
            embeddings_written += emb_ok;
            embeddings_pending += emb_pending;

            docs_upserted += 1;
            chunks_written += chunk_count;

            if let Some(p) = progress {
                let n = docs_upserted;
                if n.is_multiple_of(INGEST_PROGRESS_INTERVAL) || n == total_items {
                    p.report(SyncProgressEvent::Ingesting {
                        connector: source_label.clone(),
                        n,
                        total: total_items,
                    });
                }
            }

            let ts = item.updated_at.timestamp();
            if ts > max_updated {
                max_updated = ts;
            }
        }

        // Update checkpoint
        set_checkpoint(&pool, &source_label, max_updated).await?;

        println!("sync {}", source_label);
        println!("  fetched: {} items", items.len());
        println!("  upserted documents: {}", docs_upserted);
        println!("  chunks written: {}", chunks_written);
        println!("  extraction skipped: {}", extraction_skipped);
        if config.embedding.is_enabled() {
            println!("  embeddings written: {}", embeddings_written);
            println!("  embeddings pending: {}", embeddings_pending);
        }
        println!("  checkpoint: {}", max_updated);
        println!("ok");
    }

    pool.close().await;
    Ok(())
}

/// Inserts or updates a document in the `documents` table.
///
/// Computes a SHA-256 deduplication hash from the item's source, source_id,
/// updated_at timestamp, and body. If a document with the same `(source, source_id)`
/// already exists, it is updated with the new data; otherwise a new UUID is assigned.
///
/// # Returns
///
/// The document's UUID (existing or newly generated).
async fn upsert_document(pool: &SqlitePool, item: &SourceItem) -> Result<String> {
    // Compute dedup hash
    let mut hasher = Sha256::new();
    hasher.update(item.source.as_bytes());
    hasher.update(item.source_id.as_bytes());
    hasher.update(item.updated_at.timestamp().to_le_bytes());
    hasher.update(item.body.as_bytes());
    let dedup_hash = format!("{:x}", hasher.finalize());

    // Check if document exists
    let existing_id: Option<String> =
        sqlx::query_scalar("SELECT id FROM documents WHERE source = ? AND source_id = ?")
            .bind(&item.source)
            .bind(&item.source_id)
            .fetch_optional(pool)
            .await?;

    let doc_id = existing_id.unwrap_or_else(|| Uuid::new_v4().to_string());

    sqlx::query(
        r#"
        INSERT INTO documents (id, source, source_id, source_url, title, author, created_at, updated_at, content_type, body, metadata_json, raw_json, dedup_hash)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(source, source_id) DO UPDATE SET
            source_url = excluded.source_url,
            title = excluded.title,
            author = excluded.author,
            updated_at = excluded.updated_at,
            content_type = excluded.content_type,
            body = excluded.body,
            metadata_json = excluded.metadata_json,
            raw_json = excluded.raw_json,
            dedup_hash = excluded.dedup_hash
        "#,
    )
    .bind(&doc_id)
    .bind(&item.source)
    .bind(&item.source_id)
    .bind(&item.source_url)
    .bind(&item.title)
    .bind(&item.author)
    .bind(item.created_at.timestamp())
    .bind(item.updated_at.timestamp())
    .bind(&item.content_type)
    .bind(&item.body)
    .bind(&item.metadata_json)
    .bind(&item.raw_json)
    .bind(&dedup_hash)
    .execute(pool)
    .await?;

    Ok(doc_id)
}

/// Atomically replaces all chunks (and their embeddings/FTS entries) for a document.
///
/// This function runs inside a single SQLite transaction to ensure consistency:
/// 1. Deletes old `chunk_vectors` rows for the document's chunks.
/// 2. Deletes old `embeddings` metadata rows for the document's chunks.
/// 3. Deletes old `chunks_fts` (FTS5) entries for the document.
/// 4. Deletes old `chunks` rows for the document.
/// 5. Inserts new chunks and their corresponding FTS entries.
///
/// # Arguments
///
/// - `pool` — the database connection pool.
/// - `document_id` — the UUID of the parent document.
/// - `chunks` — the new set of chunks to insert.
async fn replace_chunks(
    pool: &SqlitePool,
    document_id: &str,
    chunks: &[crate::models::Chunk],
) -> Result<()> {
    let mut tx = pool.begin().await?;

    // Delete old embeddings for this document's chunks
    sqlx::query(
        "DELETE FROM chunk_vectors WHERE chunk_id IN (SELECT id FROM chunks WHERE document_id = ?)",
    )
    .bind(document_id)
    .execute(&mut *tx)
    .await?;
    sqlx::query(
        "DELETE FROM embeddings WHERE chunk_id IN (SELECT id FROM chunks WHERE document_id = ?)",
    )
    .bind(document_id)
    .execute(&mut *tx)
    .await?;

    // Delete old FTS entries for this document's chunks
    sqlx::query("DELETE FROM chunks_fts WHERE document_id = ?")
        .bind(document_id)
        .execute(&mut *tx)
        .await?;

    // Delete old chunks
    sqlx::query("DELETE FROM chunks WHERE document_id = ?")
        .bind(document_id)
        .execute(&mut *tx)
        .await?;

    // Insert new chunks + FTS entries
    for chunk in chunks {
        sqlx::query(
            "INSERT INTO chunks (id, document_id, chunk_index, text, hash) VALUES (?, ?, ?, ?, ?)",
        )
        .bind(&chunk.id)
        .bind(&chunk.document_id)
        .bind(chunk.chunk_index)
        .bind(&chunk.text)
        .bind(&chunk.hash)
        .execute(&mut *tx)
        .await?;

        sqlx::query("INSERT INTO chunks_fts (chunk_id, document_id, text) VALUES (?, ?, ?)")
            .bind(&chunk.id)
            .bind(&chunk.document_id)
            .bind(&chunk.text)
            .execute(&mut *tx)
            .await?;
    }

    tx.commit().await?;
    Ok(())
}

/// Retrieves the last sync checkpoint for a given connector.
///
/// Returns `Some(timestamp)` if a checkpoint exists, or `None` if this is
/// the first sync for the connector.
async fn get_checkpoint(pool: &SqlitePool, source: &str) -> Result<Option<i64>> {
    let result: Option<String> =
        sqlx::query_scalar("SELECT cursor FROM checkpoints WHERE source = ?")
            .bind(source)
            .fetch_optional(pool)
            .await?;

    Ok(result.and_then(|s| s.parse::<i64>().ok()))
}

/// Persists the sync checkpoint for a connector.
///
/// Uses an upsert to create or update the checkpoint row. The `cursor`
/// value is typically the maximum `updated_at` timestamp seen during the sync.
async fn set_checkpoint(pool: &SqlitePool, source: &str, cursor_val: i64) -> Result<()> {
    let now = chrono::Utc::now().timestamp();
    sqlx::query(
        r#"
        INSERT INTO checkpoints (source, cursor, updated_at) VALUES (?, ?, ?)
        ON CONFLICT(source) DO UPDATE SET cursor = excluded.cursor, updated_at = excluded.updated_at
        "#,
    )
    .bind(source)
    .bind(cursor_val.to_string())
    .bind(now)
    .execute(pool)
    .await?;

    Ok(())
}