use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context, Result};

use second_brain_core::embedding::Embedder;
use second_brain_core::kuzu_store::KuzuStore;
use second_brain_core::query::{QueryEngine, QueryFilters, QueryRequest};
use second_brain_core::store::Store;

fn main() {
    tracing_subscriber::fmt()
        .with_writer(std::io::stderr)
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive("second_brain=info".parse().unwrap()),
        )
        .init();

    let args: Vec<String> = std::env::args().collect();
    let cmd = args.get(1).map(|s| s.as_str()).unwrap_or("help");

    if let Err(e) = run(cmd, &args) {
        eprintln!("error: {e:#}");
        std::process::exit(1);
    }
}

fn run(cmd: &str, args: &[String]) -> Result<()> {
    match cmd {
        "ingest" => {
            let source = args.get(2).map(|s| s.as_str()).unwrap_or("claude-code");
            cmd_ingest(source)
        }
        "recall" => {
            let query = args.get(2).map(|s| s.as_str()).unwrap_or("");
            cmd_recall(query)
        }
        "remember" => {
            let content = args[2..].join(" ");
            cmd_remember(&content)
        }
        "status" => cmd_status(),
        "setup" => cmd_setup(),
        "watch" => {
            eprintln!("watch is deprecated — the daemon handles file watching automatically");
            Ok(())
        }
        "sync" => {
            let peer = args
                .get(2)
                .ok_or_else(|| anyhow::anyhow!("usage: sb sync <peer-host>"))?;
            cmd_sync(peer, args)
        }
        "sync-export" => {
            let after_seq = parse_after_seq(args);
            cmd_sync_export(after_seq)
        }
        "sync-import" => cmd_sync_import(),
        "sync-status" => cmd_sync_status(),
        "sync-backfill" => cmd_sync_backfill(),
        "reingest" => {
            let clean = args.iter().any(|a| a == "--clean");
            cmd_reingest(clean)
        }
        "consolidate" => {
            let dry_run = args.iter().any(|a| a == "--dry-run");
            let limit = args
                .windows(2)
                .find(|w| w[0] == "--limit")
                .and_then(|w| w[1].parse().ok())
                .unwrap_or(50);
            let model = args
                .windows(2)
                .find(|w| w[0] == "--model")
                .map(|w| w[1].clone())
                .unwrap_or_else(|| "gemma-4-E4B-it-Q4_K_M".to_string());
            let llm_url = args
                .windows(2)
                .find(|w| w[0] == "--llm-url")
                .map(|w| w[1].clone())
                .unwrap_or_else(|| "http://localhost:8081".to_string());
            let concurrency: usize = args
                .windows(2)
                .find(|w| w[0] == "--concurrency")
                .and_then(|w| w[1].parse().ok())
                .unwrap_or(3);
            cmd_consolidate(dry_run, limit, &model, &llm_url, concurrency)
        }
        "daily-pipeline" => {
            let llm_url = args
                .windows(2)
                .find(|w| w[0] == "--llm-url")
                .map(|w| w[1].clone())
                .unwrap_or_else(|| "http://localhost:8081".to_string());
            let model = args
                .windows(2)
                .find(|w| w[0] == "--model")
                .map(|w| w[1].clone())
                .unwrap_or_else(|| "gemma-4-E4B-it-Q4_K_M".to_string());
            let consolidate_limit: usize = args
                .windows(2)
                .find(|w| w[0] == "--consolidate-limit")
                .and_then(|w| w[1].parse().ok())
                .unwrap_or(200);
            let enhance = args.iter().any(|a| a == "--enhance");

            cmd_daily_pipeline(&llm_url, &model, consolidate_limit, enhance)
        }
        _ => {
            println!("second-brain CLI");
            println!();
            println!("Usage: sb <command> [args]");
            println!();
            println!("Commands:");
            println!(
                "  ingest [source]     Ingest conversation history (claude-code, cursor, all)"
            );
            println!("  recall <query>      Search the brain");
            println!("  remember <text>     Store new knowledge");
            println!(
                "  setup               One-command install: systemd service, MCP config, first ingest"
            );
            println!("  status              Show brain stats");
            println!("  sync <peer>         Bidirectional sync with a peer");
            println!("  sync-export         Export changes as JSONL to stdout");
            println!("  sync-import         Import JSONL changes from stdin");
            println!("  sync-status         Show sync state for all peers");
            println!("  sync-backfill       Backfill sync log for existing data");
            println!(
                "  reingest [--clean]  Re-ingest all conversations (--clean removes old memories first)"
            );
            println!("  consolidate         Distill raw memories via local LLM");
            println!("  daily-pipeline      End-of-day: consolidate → daily → export → enhance");
            Ok(())
        }
    }
}

fn open_store() -> Result<Arc<KuzuStore>> {
    let db_path = std::env::var("SECOND_BRAIN_DB")
        .map(PathBuf::from)
        .unwrap_or_else(|_| {
            let home = std::env::var("HOME").unwrap_or_else(|_| "/tmp".to_string());
            PathBuf::from(home).join(".second-brain").join("graph.kuzu")
        });

    if let Some(parent) = db_path.parent() {
        std::fs::create_dir_all(parent).ok();
    }

    let store = KuzuStore::open(&db_path, machine_id())
        .with_context(|| format!("opening database at {}", db_path.display()))?;

    eprintln!("db: {}", db_path.display());
    Ok(Arc::new(store))
}

fn load_embedder() -> Option<Embedder> {
    eprintln!("loading embedding model...");
    match Embedder::new() {
        Ok(e) => {
            eprintln!("embedder ready (BGE-small-en-v1.5, 384-dim)");
            Some(e)
        }
        Err(e) => {
            eprintln!("embedder failed, proceeding without embeddings: {e}");
            None
        }
    }
}

fn machine_id() -> String {
    hostname::get()
        .map(|h| h.to_string_lossy().to_string())
        .unwrap_or_else(|_| "unknown".to_string())
}

fn cmd_ingest(_source: &str) -> Result<()> {
    let api = api_base_url();
    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()?;

    eprintln!("ingesting via daemon...");

    let resp = client
        .post(format!("{api}/v1/ingest"))
        .json(&serde_json::json!({}))
        .send()?;

    if !resp.status().is_success() {
        anyhow::bail!(
            "ingest failed: {} {}",
            resp.status(),
            resp.text().unwrap_or_default()
        );
    }

    let body: serde_json::Value = resp.json()?;
    eprintln!(
        "ingested: {} memories, {} entities, {} conversations ({} dupes, {} unchanged, {} errors)",
        body["new_memories"],
        body["new_entities"],
        body["new_conversations"],
        body["skipped_dupes"],
        body["unchanged"],
        body["errors"],
    );

    Ok(())
}
fn cmd_reingest(clean: bool) -> Result<()> {
    let api = api_base_url();
    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()?;

    eprintln!("reingest via daemon (clean={clean})...");

    let resp = client
        .post(format!("{api}/v1/reingest"))
        .json(&serde_json::json!({"clean": clean}))
        .send()?;

    if !resp.status().is_success() {
        anyhow::bail!(
            "reingest failed: {} {}",
            resp.status(),
            resp.text().unwrap_or_default()
        );
    }

    let body: serde_json::Value = resp.json()?;
    if clean {
        eprintln!(
            "cleaned {} claude-code + {} cursor memories",
            body["cleaned_claude"], body["cleaned_cursor"]
        );
    }
    eprintln!("cleared {} ingest log entries", body["cleared_log"]);
    eprintln!(
        "ingested: {} memories, {} entities, {} conversations ({} dupes, {} unchanged, {} errors)",
        body["new_memories"],
        body["new_entities"],
        body["new_conversations"],
        body["skipped_dupes"],
        body["unchanged"],
        body["errors"],
    );

    Ok(())
}
fn cmd_recall(query: &str) -> Result<()> {
    if query.is_empty() {
        anyhow::bail!("usage: sb recall <query>");
    }

    let store = open_store()?;
    let embedder = load_embedder();

    let results = if let Some(ref emb) = embedder {
        if let Ok(embedding) = emb.embed(query) {
            let engine = QueryEngine::new(store.as_ref());
            let request = QueryRequest {
                text: query.to_string(),
                embedding,
                limit: 10,
                filters: QueryFilters::default(),
            };
            match engine.recall(&request) {
                Ok(results) if !results.is_empty() => results,
                other => {
                    if let Err(e) = &other {
                        eprintln!("warning: vector recall failed: {e}");
                    }
                    let text_results = store.text_search(query, 10)?;
                    if text_results.is_empty() {
                        println!("no memories found for: \"{query}\"");
                        return Ok(());
                    }
                    println!("(text search fallback — no vector matches)\n");
                    for (i, memory) in text_results.iter().enumerate() {
                        println!(
                            "{}. [{:?}] (source: {})",
                            i + 1,
                            memory.memory_type,
                            memory.source,
                        );
                        let preview: String = memory.content.chars().take(200).collect();
                        println!("   {preview}\n");
                    }
                    return Ok(());
                }
            }
        } else {
            eprintln!("warning: embedding failed, falling back to text search");
            let text_results = store.text_search(query, 10)?;
            if text_results.is_empty() {
                println!("no memories found for: \"{query}\"");
                return Ok(());
            }
            for (i, memory) in text_results.iter().enumerate() {
                println!(
                    "{}. [{:?}] (source: {})",
                    i + 1,
                    memory.memory_type,
                    memory.source,
                );
                let preview: String = memory.content.chars().take(200).collect();
                println!("   {preview}\n");
            }
            return Ok(());
        }
    } else {
        eprintln!("warning: no embedder available, falling back to text search");
        let text_results = store.text_search(query, 10)?;
        if text_results.is_empty() {
            println!("no memories found for: \"{query}\"");
            return Ok(());
        }
        for (i, memory) in text_results.iter().enumerate() {
            println!(
                "{}. [{:?}] (source: {})",
                i + 1,
                memory.memory_type,
                memory.source,
            );
            let preview: String = memory.content.chars().take(200).collect();
            println!("   {preview}\n");
        }
        return Ok(());
    };

    for (i, result) in results.iter().enumerate() {
        println!(
            "{}. [{:?}] (score: {:.4}, source: {})",
            i + 1,
            result.memory.memory_type,
            result.score,
            result.memory.source,
        );
        let preview: String = result.memory.content.chars().take(200).collect();
        println!("   {preview}\n");
    }

    Ok(())
}

fn cmd_remember(content: &str) -> Result<()> {
    if content.is_empty() {
        anyhow::bail!("usage: sb remember <text>");
    }

    let store = open_store()?;
    let embedder = load_embedder();

    let mut memory = second_brain_core::schema::Memory::new(
        content.to_string(),
        second_brain_core::schema::MemoryType::Semantic,
        "cli".to_string(),
        String::new(),
    );

    if let Some(ref emb) = embedder
        && let Ok(embedding) = emb.embed(content)
    {
        memory.embedding = embedding;
    }

    store.store_memory(&memory)?;
    println!("remembered [{}]: {}", memory.id, content);
    Ok(())
}

fn cmd_status() -> Result<()> {
    let store = open_store()?;
    let memories = store.memory_count()?;
    let files = store.ingested_file_count().unwrap_or(0);
    println!("memories: {memories}");
    println!("ingested files: {files}");
    println!();
    match store.diagnostic() {
        Ok(diag) => println!("{diag}"),
        Err(e) => eprintln!("diagnostic error: {e}"),
    }
    Ok(())
}

fn cmd_setup() -> Result<()> {
    let home = std::env::var("HOME").context("HOME not set")?;
    let home = PathBuf::from(&home);

    eprintln!("=== second-brain setup ===\n");

    // 1. Create data directory
    let db_dir = home.join(".second-brain");
    std::fs::create_dir_all(&db_dir)?;
    eprintln!("[1/5] data dir: {}", db_dir.display());

    // 2. Locate binaries
    let (api_bin, mcp_bin, cli_bin) = locate_binaries()?;
    eprintln!("[2/5] binaries:");
    eprintln!("  api: {}", api_bin.display());
    eprintln!("  mcp: {}", mcp_bin.display());
    eprintln!("  cli: {}", cli_bin.display());

    // 3. Install systemd user service
    let systemd_dir = home.join(".config/systemd/user");
    std::fs::create_dir_all(&systemd_dir)?;
    let service_path = systemd_dir.join("second-brain.service");
    let db_path = db_dir.join("graph.kuzu");

    let service_content = format!(
        "[Unit]\n\
         Description=Second Brain daemon (KuzuDB, embedder, file watcher)\n\
         After=network.target\n\
         \n\
         [Service]\n\
         Type=simple\n\
         ExecStart={api}\n\
         Environment=SECOND_BRAIN_DB={db}\n\
         Environment=SECOND_BRAIN_BIND=127.0.0.1:7200\n\
         Restart=on-failure\n\
         RestartSec=5\n\
         TimeoutStartSec=60\n\
         \n\
         [Install]\n\
         WantedBy=default.target\n",
        api = api_bin.display(),
        db = db_path.display(),
    );

    std::fs::write(&service_path, &service_content)?;
    eprintln!("[3/5] systemd service: {}", service_path.display());

    let reload = std::process::Command::new("systemctl")
        .args(["--user", "daemon-reload"])
        .status();
    if reload.is_ok_and(|s| s.success()) {
        let enable = std::process::Command::new("systemctl")
            .args(["--user", "enable", "--now", "second-brain.service"])
            .status();
        match enable {
            Ok(s) if s.success() => eprintln!("  enabled and started"),
            _ => eprintln!("  warning: could not enable service (systemd not available?)"),
        }
    } else {
        eprintln!(
            "  warning: systemctl not available, start daemon manually: {}",
            api_bin.display()
        );
    }

    // 4. Configure Claude Code MCP
    let claude_dir = home.join(".claude");
    std::fs::create_dir_all(&claude_dir)?;
    let settings_path = claude_dir.join("settings.json");

    let mut settings: serde_json::Value = if settings_path.exists() {
        let content = std::fs::read_to_string(&settings_path)?;
        serde_json::from_str(&content).unwrap_or_else(|_| serde_json::json!({}))
    } else {
        serde_json::json!({})
    };

    let mcp_servers = settings
        .as_object_mut()
        .unwrap()
        .entry("mcpServers")
        .or_insert_with(|| serde_json::json!({}));

    mcp_servers["second-brain"] = serde_json::json!({
        "command": mcp_bin.to_string_lossy(),
        "env": {
            "SECOND_BRAIN_API": "http://127.0.0.1:7200",
            "SECOND_BRAIN_DAEMON_BIN": api_bin.to_string_lossy(),
        }
    });

    std::fs::write(&settings_path, serde_json::to_string_pretty(&settings)?)?;
    eprintln!("[4/5] claude code MCP: {}", settings_path.display());

    // 5. Wait for daemon, run first ingest
    eprintln!("[5/5] running first ingest...");
    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()?;

    let api = api_base_url();
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    let mut ready = false;
    while std::time::Instant::now() < deadline {
        if client
            .get(format!("{api}/health"))
            .timeout(std::time::Duration::from_secs(2))
            .send()
            .is_ok_and(|r| r.status().is_success())
        {
            ready = true;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(500));
    }

    if !ready {
        eprintln!("  daemon not ready after 30s — run 'second-brain-cli ingest' later");
        return Ok(());
    }

    match client
        .post(format!("{api}/v1/ingest"))
        .json(&serde_json::json!({}))
        .send()
    {
        Ok(resp) if resp.status().is_success() => {
            let body: serde_json::Value = resp.json().unwrap_or_default();
            eprintln!(
                "  ingested: {} memories, {} conversations",
                body["new_memories"], body["new_conversations"]
            );
        }
        Ok(resp) => eprintln!("  ingest warning: {}", resp.status()),
        Err(e) => eprintln!("  ingest warning: {e}"),
    }

    eprintln!("\n=== setup complete ===");
    eprintln!("daemon:  systemctl --user status second-brain");
    eprintln!("recall:  sb recall \"your query\"");
    eprintln!("status:  sb status");
    eprintln!("\nRun your bootstrapper (harness) for wiki vault + daily timer setup.");

    Ok(())
}

fn locate_binaries() -> Result<(PathBuf, PathBuf, PathBuf)> {
    let cli_exe = std::env::current_exe().context("cannot determine own binary path")?;
    let bin_dir = cli_exe.parent().unwrap();

    let api_candidate = bin_dir.join("second-brain-api");
    let mcp_candidate = bin_dir.join("second-brain-mcp");

    let api_bin = if api_candidate.exists() {
        api_candidate
    } else {
        which_bin("second-brain-api")?
    };

    let mcp_bin = if mcp_candidate.exists() {
        mcp_candidate
    } else {
        which_bin("second-brain-mcp")?
    };

    Ok((api_bin, mcp_bin, cli_exe))
}

fn which_bin(name: &str) -> Result<PathBuf> {
    let home = std::env::var("HOME").unwrap_or_default();
    let cargo_bin = PathBuf::from(&home).join(".cargo/bin").join(name);
    if cargo_bin.exists() {
        return Ok(cargo_bin);
    }

    let output = std::process::Command::new("which")
        .arg(name)
        .output()
        .context(format!("{name} not found"))?;

    if output.status.success() {
        let path = String::from_utf8_lossy(&output.stdout).trim().to_string();
        Ok(PathBuf::from(path))
    } else {
        anyhow::bail!(
            "{name} not found in PATH or ~/.cargo/bin — install with: cargo install {name}"
        )
    }
}

fn parse_after_seq(args: &[String]) -> u64 {
    for (i, arg) in args.iter().enumerate() {
        if arg == "--after-seq"
            && let Some(val) = args.get(i + 1)
        {
            return val.parse().unwrap_or(0);
        }
    }
    0
}

fn cmd_sync(peer_host: &str, args: &[String]) -> Result<()> {
    let api_base = api_base_url();

    let mut sb_binary = "sb".to_string();
    for (i, arg) in args.iter().enumerate() {
        if arg == "--sb-path"
            && let Some(val) = args.get(i + 1)
        {
            sb_binary = val.clone();
        }
    }

    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()?;

    eprintln!("pulling from {peer_host}...");
    let pull_stats = sync_pull_via_daemon(&client, &api_base, peer_host, &sb_binary)?;
    eprintln!(
        "  {} created, {} updated, {} deleted, {} skipped",
        pull_stats.created, pull_stats.updated, pull_stats.deleted, pull_stats.skipped
    );

    eprintln!("pushing to {peer_host}...");
    let max_seq = sync_push_via_daemon(&client, &api_base, peer_host, &sb_binary)?;
    eprintln!("  pushed through seq {max_seq}");

    eprintln!("sync complete.");
    Ok(())
}

fn api_base_url() -> String {
    std::env::var("SECOND_BRAIN_API").unwrap_or_else(|_| "http://127.0.0.1:7200".to_string())
}

fn sync_pull_via_daemon(
    client: &reqwest::blocking::Client,
    api_base: &str,
    peer_host: &str,
    sb_binary: &str,
) -> Result<SyncImportStats> {
    #[derive(serde::Deserialize)]
    struct PeerInfo {
        peers: Vec<PeerState>,
    }
    #[derive(serde::Deserialize)]
    struct PeerState {
        peer_id: String,
        last_seq: u64,
    }

    let status: PeerInfo = client
        .get(format!("{api_base}/v1/sync/status"))
        .send()
        .context("getting sync status")?
        .json()?;

    let after_seq = status
        .peers
        .iter()
        .find(|p| p.peer_id == peer_host)
        .map(|p| p.last_seq)
        .unwrap_or(0);

    let mut child = std::process::Command::new("ssh")
        .arg("-o")
        .arg("StrictHostKeyChecking=no")
        .arg("-o")
        .arg("ServerAliveInterval=30")
        .arg(peer_host)
        .arg(sb_binary)
        .arg("sync-export")
        .arg("--after-seq")
        .arg(after_seq.to_string())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::inherit())
        .spawn()
        .context("spawning ssh for pull")?;

    let stdout = child.stdout.take().unwrap();
    let reader = std::io::BufReader::new(stdout);

    let mut records = Vec::new();
    for line in std::io::BufRead::lines(reader) {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }
        let val: serde_json::Value = serde_json::from_str(&line)?;
        records.push(val);
    }

    let status = child.wait()?;
    if !status.success() {
        anyhow::bail!("ssh pull exited with {}", status);
    }

    #[derive(serde::Deserialize)]
    struct ImportResp {
        created: u64,
        updated: u64,
        deleted: u64,
        skipped: u64,
        #[allow(dead_code)]
        errors: u64,
        max_seq: u64,
    }

    let resp: ImportResp = client
        .post(format!("{api_base}/v1/sync/import"))
        .json(&serde_json::json!({"records": records}))
        .send()
        .context("posting import to daemon")?
        .error_for_status()
        .context("import returned error")?
        .json()?;

    if resp.max_seq > after_seq {
        client
            .post(format!("{api_base}/v1/sync/state"))
            .json(&serde_json::json!({"peer_id": peer_host, "last_seq": resp.max_seq}))
            .send()
            .context("updating sync state")?;
    }

    Ok(SyncImportStats {
        created: resp.created,
        updated: resp.updated,
        deleted: resp.deleted,
        skipped: resp.skipped,
    })
}

struct SyncImportStats {
    created: u64,
    updated: u64,
    deleted: u64,
    skipped: u64,
}

fn sync_push_via_daemon(
    client: &reqwest::blocking::Client,
    api_base: &str,
    peer_host: &str,
    sb_binary: &str,
) -> Result<u64> {
    let push_peer = format!("push:{}", peer_host);
    const BATCH_SIZE: usize = 2000;

    #[derive(serde::Deserialize)]
    struct PeerInfo {
        peers: Vec<PeerState>,
    }
    #[derive(serde::Deserialize)]
    struct PeerState {
        peer_id: String,
        last_seq: u64,
    }

    let status: PeerInfo = client
        .get(format!("{api_base}/v1/sync/status"))
        .send()
        .context("getting sync status")?
        .json()?;

    let mut cursor = status
        .peers
        .iter()
        .find(|p| p.peer_id == push_peer)
        .map(|p| p.last_seq)
        .unwrap_or(0);

    #[derive(serde::Deserialize)]
    struct ExportResp {
        records: Vec<serde_json::Value>,
        max_seq: u64,
    }

    let mut child = std::process::Command::new("ssh")
        .arg("-o")
        .arg("StrictHostKeyChecking=no")
        .arg("-o")
        .arg("ServerAliveInterval=60")
        .arg("-o")
        .arg("ServerAliveCountMax=60")
        .arg(peer_host)
        .arg(sb_binary)
        .arg("sync-import")
        .stdin(std::process::Stdio::piped())
        .stderr(std::process::Stdio::inherit())
        .spawn()
        .context("spawning ssh for push")?;

    let mut total_pushed = 0u64;

    {
        let stdin = child.stdin.take().unwrap();
        let mut writer = std::io::BufWriter::new(stdin);

        loop {
            let export: ExportResp = client
                .post(format!("{api_base}/v1/sync/export"))
                .json(&serde_json::json!({"after_seq": cursor, "limit": BATCH_SIZE}))
                .send()
                .context("getting export from daemon")?
                .error_for_status()
                .context("export returned error")?
                .json()?;

            if export.records.is_empty() {
                break;
            }

            let batch_count = export.records.len();

            for record in &export.records {
                use std::io::Write;
                serde_json::to_writer(&mut writer, record)?;
                writer.write_all(b"\n")?;
            }
            use std::io::Write;
            writer.flush()?;

            total_pushed += batch_count as u64;
            cursor = export.max_seq;

            eprintln!("  batch: {} records (through seq {})", batch_count, cursor);

            if batch_count < BATCH_SIZE {
                break;
            }
        }
    }

    let exit = child.wait()?;
    if !exit.success() {
        anyhow::bail!("ssh push exited with {}", exit);
    }

    if total_pushed > 0 {
        client
            .post(format!("{api_base}/v1/sync/state"))
            .json(&serde_json::json!({"peer_id": push_peer, "last_seq": cursor}))
            .send()
            .context("updating push sync state")?;
    }

    eprintln!("  total: {} records pushed", total_pushed);
    Ok(cursor)
}

fn cmd_sync_export(after_seq: u64) -> Result<()> {
    let api_base = api_base_url();
    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(120))
        .build()?;

    #[derive(serde::Deserialize)]
    struct ExportResp {
        records: Vec<serde_json::Value>,
        max_seq: u64,
    }

    let resp: ExportResp = client
        .post(format!("{api_base}/v1/sync/export"))
        .json(&serde_json::json!({"after_seq": after_seq}))
        .send()
        .context("calling sync/export on daemon")?
        .error_for_status()
        .context("sync/export returned error")?
        .json()?;

    let stdout = std::io::stdout();
    let mut writer = std::io::BufWriter::new(stdout.lock());
    for record in &resp.records {
        use std::io::Write;
        serde_json::to_writer(&mut writer, record)?;
        writer.write_all(b"\n")?;
    }

    eprintln!("exported through seq {}", resp.max_seq);
    Ok(())
}

fn cmd_sync_import() -> Result<()> {
    let api_base = api_base_url();
    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()?;
    const BATCH_SIZE: usize = 500;

    let stdin = std::io::stdin();
    let reader = std::io::BufReader::new(stdin.lock());

    #[derive(serde::Deserialize)]
    struct ImportResp {
        created: u64,
        updated: u64,
        deleted: u64,
        skipped: u64,
        errors: u64,
        max_seq: u64,
    }

    let mut totals = (0u64, 0u64, 0u64, 0u64, 0u64, 0u64);
    let mut batch: Vec<serde_json::Value> = Vec::with_capacity(BATCH_SIZE);

    for line in std::io::BufRead::lines(reader) {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }
        batch.push(serde_json::from_str(&line)?);

        if batch.len() >= BATCH_SIZE {
            let resp: ImportResp = client
                .post(format!("{api_base}/v1/sync/import"))
                .json(&serde_json::json!({"records": batch}))
                .send()
                .context("calling sync/import on daemon")?
                .error_for_status()
                .context("sync/import returned error")?
                .json()?;
            totals.0 += resp.created;
            totals.1 += resp.updated;
            totals.2 += resp.deleted;
            totals.3 += resp.skipped;
            totals.4 += resp.errors;
            totals.5 = resp.max_seq;
            batch = Vec::with_capacity(BATCH_SIZE);
        }
    }

    if !batch.is_empty() {
        let resp: ImportResp = client
            .post(format!("{api_base}/v1/sync/import"))
            .json(&serde_json::json!({"records": batch}))
            .send()
            .context("calling sync/import on daemon")?
            .error_for_status()
            .context("sync/import returned error")?
            .json()?;
        totals.0 += resp.created;
        totals.1 += resp.updated;
        totals.2 += resp.deleted;
        totals.3 += resp.skipped;
        totals.4 += resp.errors;
        totals.5 = resp.max_seq;
    }

    eprintln!(
        "imported: {} created, {} updated, {} deleted, {} skipped, {} errors (through seq {})",
        totals.0, totals.1, totals.2, totals.3, totals.4, totals.5
    );
    Ok(())
}

fn cmd_sync_status() -> Result<()> {
    let api_base = api_base_url();
    let client = reqwest::blocking::Client::new();

    #[derive(serde::Deserialize)]
    struct PeerInfo {
        peers: Vec<PeerState>,
    }
    #[derive(serde::Deserialize)]
    struct PeerState {
        peer_id: String,
        last_seq: u64,
        last_sync_at: String,
    }

    let resp: PeerInfo = client
        .get(format!("{api_base}/v1/sync/status"))
        .send()
        .context("getting sync status from daemon")?
        .error_for_status()?
        .json()?;

    if resp.peers.is_empty() {
        println!("no sync peers configured");
    } else {
        for state in &resp.peers {
            println!(
                "peer: {}, last_seq: {}, last_sync: {}",
                state.peer_id, state.last_seq, state.last_sync_at
            );
        }
    }
    Ok(())
}

fn cmd_sync_backfill() -> Result<()> {
    let api_base = api_base_url();
    let client = reqwest::blocking::Client::new();

    #[derive(serde::Deserialize)]
    struct BackfillResp {
        backfilled: usize,
    }

    let resp: BackfillResp = client
        .post(format!("{api_base}/v1/sync/backfill"))
        .send()
        .context("calling backfill on daemon")?
        .error_for_status()?
        .json()?;

    eprintln!("backfilled {} sync log entries", resp.backfilled);
    Ok(())
}

fn cmd_consolidate(
    dry_run: bool,
    limit: usize,
    model: &str,
    llm_url: &str,
    concurrency: usize,
) -> Result<()> {
    let api = api_base_url();
    let client = reqwest::blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(30))
        .build()?;

    let batch_resp = client
        .get(format!("{api}/v1/consolidate/batch?limit={limit}"))
        .send()?;
    if !batch_resp.status().is_success() {
        anyhow::bail!("batch fetch failed: {}", batch_resp.status());
    }
    let batch: serde_json::Value = batch_resp.json()?;

    let total = batch["total"].as_u64().unwrap_or(0);
    let consolidated = batch["consolidated"].as_u64().unwrap_or(0);
    let memories = batch["memories"].as_array().cloned().unwrap_or_default();

    eprintln!("consolidated: {consolidated}/{total} memories");
    if memories.is_empty() {
        eprintln!("nothing to consolidate");
        return Ok(());
    }
    eprintln!(
        "processing {} memories with {model} at {llm_url} (concurrency={concurrency})",
        memories.len()
    );

    let rt = tokio::runtime::Runtime::new()?;
    let chat_url = format!("{llm_url}/v1/chat/completions");

    let success = std::sync::atomic::AtomicUsize::new(0);
    let skipped = std::sync::atomic::AtomicUsize::new(0);
    let errors = std::sync::atomic::AtomicUsize::new(0);
    let processed = std::sync::atomic::AtomicUsize::new(0);

    let chunks: Vec<Vec<serde_json::Value>> =
        memories.chunks(concurrency).map(|c| c.to_vec()).collect();

    let total_count = memories.len();

    for chunk in &chunks {
        let futures: Vec<_> = chunk
            .iter()
            .map(|mem| {
                let id = mem["id"].as_str().unwrap_or("").to_string();
                let content = mem["content"].as_str().unwrap_or("").to_string();
                let chat_url = chat_url.clone();
                let model = model.to_string();
                async move {
                    let prompt = build_consolidation_prompt(&content);
                    let result = call_llm(&chat_url, &model, &prompt).await;
                    (id, content, result)
                }
            })
            .collect();

        let results = rt.block_on(async { futures::future::join_all(futures).await });

        for (id, content, result) in results {
            let i = processed.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;

            let response = match result {
                Ok(r) => r,
                Err(e) => {
                    eprintln!("  [{i}/{total_count}] error: {e}");
                    errors.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    continue;
                }
            };

            let distilled = response.trim().to_string();

            if distilled.is_empty() || distilled.len() < 20 || distilled == "SKIP" {
                skipped.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                client
                    .post(format!("{api}/v1/consolidate/store"))
                    .json(&serde_json::json!({
                        "raw_id": id,
                        "distilled_content": "",
                        "model": model,
                        "skip": true,
                    }))
                    .send()
                    .ok();
                continue;
            }

            if dry_run {
                let preview: String = content.chars().take(100).collect();
                println!("--- [{i}/{total_count}] raw ({} chars) ---", content.len());
                println!("  {preview}...");
                println!("--- distilled ({} chars) ---", distilled.len());
                println!("  {distilled}");
                println!();
                success.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                continue;
            }

            let store_resp = client
                .post(format!("{api}/v1/consolidate/store"))
                .json(&serde_json::json!({
                    "raw_id": id,
                    "distilled_content": distilled,
                    "model": model,
                    "skip": false,
                }))
                .send();

            match store_resp {
                Ok(r) if r.status().is_success() => {
                    success.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
                Ok(r) => {
                    eprintln!("  [{i}/{total_count}] store error: {}", r.status());
                    errors.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
                Err(e) => {
                    eprintln!("  [{i}/{total_count}] store error: {e}");
                    errors.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }

            if i.is_multiple_of(10) {
                let s = success.load(std::sync::atomic::Ordering::Relaxed);
                let sk = skipped.load(std::sync::atomic::Ordering::Relaxed);
                let e = errors.load(std::sync::atomic::Ordering::Relaxed);
                eprintln!("  [{i}/{total_count}] {s} distilled, {sk} skipped, {e} errors");
            }
        }
    }

    let s = success.load(std::sync::atomic::Ordering::Relaxed);
    let sk = skipped.load(std::sync::atomic::Ordering::Relaxed);
    let e = errors.load(std::sync::atomic::Ordering::Relaxed);
    eprintln!();
    eprintln!("consolidation complete: {s} distilled, {sk} skipped, {e} errors");
    if dry_run {
        eprintln!("(dry run — nothing written)");
        return Ok(());
    }

    if s > 0 || sk > 0 {
        eprintln!("cleaning up consolidated raw memories...");
        match client.post(format!("{api}/v1/consolidate/cleanup")).send() {
            Ok(resp) if resp.status().is_success() => {
                let body: serde_json::Value = resp.json().unwrap_or_default();
                let deleted = body["deleted"].as_u64().unwrap_or(0);
                eprintln!("deleted {deleted} raw memories that were consolidated");
            }
            Ok(resp) => {
                eprintln!("cleanup warning: {}", resp.status());
            }
            Err(e) => {
                eprintln!("cleanup warning: {e}");
            }
        }
    }

    Ok(())
}

fn build_consolidation_prompt(raw_content: &str) -> String {
    format!(
        r#"Extract the key knowledge facts from this conversation fragment. Return ONLY the distilled facts as a concise paragraph. No preamble, no explanation, no bullet points unless the content is a list of items.

Rules:
- Keep technical details exact (names, versions, commands, file paths, config values)
- Remove conversational noise ("Let me check...", "Perfect!", "Now let me...")
- Preserve decisions and their reasoning
- Preserve architecture choices and trade-offs
- If the fragment contains no extractable knowledge, respond with just "SKIP"
- Keep it under 200 words

Fragment:
{raw_content}"#
    )
}

async fn call_llm(url: &str, model: &str, prompt: &str) -> Result<String> {
    let client = reqwest::Client::new();
    let body = serde_json::json!({
        "model": model,
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.1,
        "max_tokens": 4096,
    });

    let resp = client
        .post(url)
        .json(&body)
        .timeout(std::time::Duration::from_secs(120))
        .send()
        .await
        .context("connecting to LLM")?;

    if !resp.status().is_success() {
        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        anyhow::bail!("LLM returned {status}: {text}");
    }

    let json: serde_json::Value = resp.json().await.context("parsing LLM response")?;
    let content = json["choices"][0]["message"]["content"]
        .as_str()
        .unwrap_or("");
    Ok(content.to_string())
}

fn cmd_daily_pipeline(
    llm_url: &str,
    model: &str,
    consolidate_limit: usize,
    _enhance: bool,
) -> Result<()> {
    println!("=== daily pipeline ===\n");

    println!(
        "[1/1] consolidating up to {} memories...",
        consolidate_limit
    );
    match cmd_consolidate(false, consolidate_limit, model, llm_url, 3) {
        Ok(()) => println!("      consolidation complete"),
        Err(e) => println!("      consolidation failed (non-fatal): {}", e),
    }

    println!("\n=== pipeline complete ===");
    Ok(())
}
