//! Worker pool: claims jobs from `mem_tree_jobs`, dispatches them through
//! [`handlers::handle_job`], and settles the row.
//!
//! Concurrency control for LLM-bound work is delegated to
//! [`crate::openhuman::scheduler_gate`] — its global single-slot
//! semaphore (`LlmPermit`) is the one source of truth across this
//! worker, voice cleanup, autocomplete, triage, and reflection. The
//! worker itself just calls `wait_for_capacity()`; non-LLM jobs
//! (`AppendBuffer`, `FlushStale`) run without acquiring a permit.

use std::sync::{Arc, OnceLock};
use std::time::Duration;

use anyhow::Result;
use tokio::sync::Notify;

use crate::openhuman::config::Config;
use crate::openhuman::memory::tree::jobs::handlers;
use crate::openhuman::memory::tree::jobs::redact::scrub_for_log;
use crate::openhuman::memory::tree::jobs::store::{
    claim_next, mark_deferred, mark_done, mark_failed, recover_stale_locks,
    DEFAULT_LOCK_DURATION_MS,
};
use crate::openhuman::memory::tree::jobs::types::JobOutcome;

/// Number of concurrent job-worker tasks. Each worker claims one job
/// at a time via `claim_next` (atomic UPDATE under SQLite WAL with
/// `locked_until_ms` + status='running'), so multiple workers
/// parallelize independent jobs without double-claim risk.
///
/// On cloud backends, LLM-bound jobs drop the global LLM permit
/// after claim (see `run_once`) so all 4 workers can run cloud
/// extract/summarise calls in parallel.
///
/// On local backends, the single global LLM slot still serialises
/// Ollama calls for laptop-RAM safety. Note that `wait_for_capacity`
/// is acquired **before** `claim_next`, so non-LLM jobs (AppendBuffer,
/// FlushStale, TopicRoute) also block on the gate when an LLM job
/// holds the permit — they only run in parallel with each other while
/// no LLM job is in flight. Bumping `WORKER_COUNT` therefore helps
/// throughput most when local LLM calls are sparse.
const WORKER_COUNT: usize = 4;
const POLL_INTERVAL: Duration = Duration::from_secs(5);

static WORKER_NOTIFY: OnceLock<Arc<Notify>> = OnceLock::new();
static STARTED: std::sync::Once = std::sync::Once::new();

/// Notify any idle workers so they re-poll immediately instead of waiting
/// out [`POLL_INTERVAL`]. Cheap no-op before [`start`] has run.
pub fn wake_workers() {
    if let Some(notify) = WORKER_NOTIFY.get() {
        notify.notify_waiters();
    }
}

/// Start the worker pool + daily scheduler. Takes the full `Config` so
/// each spawned task sees the user's actual settings (LLM endpoints,
/// embedder model, timeouts) — not `Config::default()`. Without this,
/// workers fall back to inert/regex-only behavior regardless of what's
/// in `config.toml`, defeating the entire async pipeline.
///
/// Idempotent (`Once`-guarded) so repeat calls during bootstrap are
/// safe no-ops after the first.
pub fn start(config: Config) {
    STARTED.call_once(|| {
        let notify = WORKER_NOTIFY
            .get_or_init(|| Arc::new(Notify::new()))
            .clone();
        if let Err(err) = recover_stale_locks(&config) {
            log::warn!("[memory_tree::jobs] recover_stale_locks failed at startup: {err:#}");
        }

        for idx in 0..WORKER_COUNT {
            let notify = notify.clone();
            let cfg = config.clone();
            tokio::spawn(async move {
                loop {
                    match run_once(&cfg).await {
                        Ok(true) => continue,
                        Ok(false) => {
                            tokio::select! {
                                _ = notify.notified() => {}
                                _ = tokio::time::sleep(POLL_INTERVAL) => {}
                            }
                        }
                        Err(err) => {
                            crate::core::observability::report_error(
                                &err,
                                "memory",
                                "tree_jobs_worker",
                                &[("worker_idx", &idx.to_string())],
                            );
                            tokio::time::sleep(Duration::from_secs(1)).await;
                        }
                    }
                }
            });
        }

        super::scheduler::start(config);
    });
}

/// Claim and run a single job. Returns `true` when work was processed,
/// `false` when no eligible row was available.
pub async fn run_once(config: &Config) -> Result<bool> {
    // Cooperative throttle BEFORE `claim_next()`. Holding the DB claim
    // across an awaited `wait_for_capacity()` would let `Paused` mode
    // sit on the row past `DEFAULT_LOCK_DURATION_MS`, after which
    // `recover_stale_locks()` would requeue it for another worker to
    // pick up — duplicating side effects. Throttling here means
    // non-LLM jobs (AppendBuffer/FlushStale) also experience the same
    // gate delay, but that's fine: in Throttled mode the host is
    // already overloaded and a 30s breather between any DB-write batch
    // is welcome; in Paused mode the user has explicitly asked us to
    // stand down. Returns immediately in Aggressive/Normal so plugged-in
    // desktops with headroom pay zero cost.
    //
    // For LLM-bound jobs the returned `LlmPermit` reserves the global
    // single slot for the lifetime of `handle_job`. Non-LLM jobs
    // (`AppendBuffer`, `FlushStale`) drop the permit before the
    // handler runs so they don't block the slot.
    let gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await;

    let Some(job) = claim_next(config, DEFAULT_LOCK_DURATION_MS)? else {
        return Ok(false);
    };

    let llm_permit = if job.kind.is_llm_bound() {
        // Local Ollama loads ~1.3 GB resident per concurrent call —
        // hold the gate to enforce process-wide single-slot RAM
        // safety. Cloud calls are bandwidth-bound, not RAM-bound:
        // drop the permit so multiple workers can run cloud
        // extract/summarise calls in parallel (the worker pool
        // itself, sized to `WORKER_COUNT`, is the upstream bound).
        match config.memory_tree.llm_backend {
            crate::openhuman::config::LlmBackend::Local => gate_permit,
            crate::openhuman::config::LlmBackend::Cloud => {
                drop(gate_permit);
                None
            }
        }
    } else {
        // Non-LLM jobs don't need the global slot; release it so an
        // LLM-bound caller waiting elsewhere in the process can run.
        drop(gate_permit);
        None
    };

    let result = handlers::handle_job(config, &job).await;
    drop(llm_permit);

    match result {
        Ok(JobOutcome::Done) => {
            log::debug!(
                "[memory_tree::jobs] done id={} kind={}",
                job.id,
                job.kind.as_str()
            );
            mark_done(config, &job)?;
        }
        Ok(JobOutcome::Defer { until_ms, reason }) => {
            // Defer is normal operation (transient blocker, e.g. rate
            // limit) — log at info, not warn — and do NOT count this
            // claim toward the failure-attempt budget. `mark_deferred`
            // reverts the bump applied by `claim_next` so the row's
            // attempts counter stays where it was before this claim.
            //
            // `reason` is handler-supplied free-form text and may
            // include upstream provider responses; scrub for log
            // emission while keeping the original in DB state.
            log::info!(
                "[memory_tree::jobs] deferred id={} kind={} until_ms={} reason={}",
                job.id,
                job.kind.as_str(),
                until_ms,
                scrub_for_log(&reason)
            );
            mark_deferred(config, &job, until_ms, &reason)?;
        }
        Err(err) => {
            // Preserve the full anyhow cause chain in the persisted
            // last_error so a reader of mem_tree_jobs can see the root
            // cause, not just the top-level message. The log line gets
            // the same chain after `scrub_for_log`, since anyhow chains
            // commonly embed upstream HTTP bodies / auth headers.
            let message = format!("{err:#}");
            log::warn!(
                "[memory_tree::jobs] job failed id={} kind={} err={}",
                job.id,
                job.kind.as_str(),
                scrub_for_log(&message)
            );
            mark_failed(config, &job, &message)?;
        }
    }

    Ok(true)
}
