//! The medium loop (docs/04): seconds behind the write path, never blocking it.
//! Drains the durable gloss queue: interpret each new completion with the full
//! ensemble, canonicalize, publish gloss records, rebuild the user's term
//! graph, update their posterior.

use std::sync::Arc;
use std::time::Duration;

use smol_str::SmolStr;

use crate::atproto::records::{Gloss, Interpreter, Reading};
use crate::graph;
use crate::infer;
use crate::state::AppState;
use jacquard::api::com_atproto::repo::strong_ref::StrongRef;
use jacquard::types::string::{AtUri, Cid, Datetime};

use super::canon;
use super::{GlossInput, PROMPT_VERSION};

pub async fn run(state: Arc<AppState>) {
    loop {
        match drain(&state).await {
            Ok(0) => {
                // Idle: wake on notify or poll (Jetstream rows arrive without notify).
                tokio::select! {
                    _ = state.gloss_notify.notified() => {},
                    _ = tokio::time::sleep(Duration::from_secs(30)) => {},
                }
            }
            Ok(_) => {} // keep draining
            Err(e) => {
                tracing::error!("gloss worker error: {e:#}");
                tokio::time::sleep(Duration::from_secs(5)).await;
            }
        }
    }
}

async fn drain(state: &AppState) -> anyhow::Result<usize> {
    let batch: Vec<(i64, String, i64)> = sqlx::query_as(
        "SELECT id, completion_uri, attempts FROM gloss_queue
         WHERE attempts < 5 ORDER BY id LIMIT 8",
    )
    .fetch_all(&state.pool)
    .await?;

    let mut processed = 0;
    for (qid, completion_uri, _attempts) in batch {
        match gloss_one(state, &completion_uri).await {
            Ok(()) => {
                sqlx::query("DELETE FROM gloss_queue WHERE id = ?")
                    .bind(qid)
                    .execute(&state.pool)
                    .await?;
                processed += 1;
            }
            Err(e) => {
                tracing::warn!(%completion_uri, "gloss failed: {e:#}");
                sqlx::query("UPDATE gloss_queue SET attempts = attempts + 1 WHERE id = ?")
                    .bind(qid)
                    .execute(&state.pool)
                    .await?;
            }
        }
    }
    Ok(processed)
}

async fn gloss_one(state: &AppState, completion_uri: &str) -> anyhow::Result<()> {
    let row: Option<(String, Option<String>, String, String, String, Option<String>)> = sqlx::query_as(
        "SELECT c.did, c.text, c.mode, c.slot, c.square_uri, c.cid
         FROM completions c WHERE c.uri = ?",
    )
    .bind(completion_uri)
    .fetch_optional(&state.pool)
    .await?;
    let Some((did, text, mode, _slot, square_uri, completion_cid)) = row else {
        // Completion deleted before glossing: nothing to interpret.
        return Ok(());
    };

    let square: Option<(Option<String>, Option<String>, Option<String>, Option<String>, String)> =
        sqlx::query_as("SELECT a, b, c, d, open_slot FROM squares WHERE uri = ?")
            .bind(&square_uri)
            .fetch_optional(&state.pool)
            .await?;
    let Some((a, b, c, d, open_slot)) = square else {
        anyhow::bail!("square {square_uri} not indexed yet");
    };

    let input = GlossInput {
        a,
        b,
        c,
        d,
        open_slot,
        text: text.clone(),
        mode: mode.clone(),
    };

    // Skip if this ensemble already glossed this completion (idempotent re-runs).
    let (existing,): (i64,) =
        sqlx::query_as("SELECT COUNT(*) FROM glosses WHERE completion_uri = ?")
            .bind(completion_uri)
            .fetch_one(&state.pool)
            .await?;
    if existing >= state.ensemble.len() as i64 {
        return Ok(());
    }

    let completion_ref = StrongRef {
        uri: AtUri::new(SmolStr::new(completion_uri)).map_err(|e| anyhow::anyhow!("{e}"))?,
        cid: Cid::new_owned(
            completion_cid
                .as_deref()
                .unwrap_or("bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
                .as_bytes(),
        )
        .map_err(|e| anyhow::anyhow!("{e}"))?,
        extra_data: None,
    };

    for member in &state.ensemble {
        let member_id = member.id();
        let (already,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM glosses WHERE completion_uri = ? AND ensemble_member = ?",
        )
        .bind(completion_uri)
        .bind(&member_id)
        .fetch_one(&state.pool)
        .await?;
        if already > 0 {
            continue;
        }

        let output = member.gloss(&input).await?;

        // Canonicalize each reading into the growing relation space.
        let mut readings: Vec<Reading> = Vec::new();
        let mut readings_json: Vec<serde_json::Value> = Vec::new();
        for r in &output.readings {
            let canonical = canon::canonicalize(&state.pool, &state.embedder, &r.relation).await?;
            readings.push(Reading {
                relation: r.relation.clone(),
                canonical_relation: Some(canonical),
                weight: SmolStr::new(format!("{:.3}", r.weight)),
                valence: Some(SmolStr::new(format!("{:.3}", r.valence))),
            });
            readings_json.push(serde_json::json!({
                "relation": r.relation,
                "canonicalRelation": canonical,
                "weight": r.weight,
                "valence": r.valence,
            }));
        }

        let record = Gloss {
            completion: completion_ref.clone(),
            readings,
            interpreter: Interpreter {
                model: SmolStr::new(&output.model),
                prompt_version: SmolStr::new(PROMPT_VERSION),
                ensemble_member: SmolStr::new(&member_id),
                temperature: output.temperature.map(|t| SmolStr::new(format!("{t}"))),
            },
            created_at: Datetime::now(),
        };
        let published = state.publisher.publish_gloss(&record).await?;

        sqlx::query(
            "INSERT INTO glosses (uri, completion_uri, did, readings, interpreter, prompt_version, ensemble_member, created_at)
             VALUES (?,?,?,?,?,?,?,?)
             ON CONFLICT (uri) DO NOTHING",
        )
        .bind(&published.uri)
        .bind(completion_uri)
        .bind(&did)
        .bind(serde_json::to_string(&readings_json)?)
        .bind(&output.model)
        .bind(PROMPT_VERSION)
        .bind(&member_id)
        .bind(record.created_at.as_str())
        .execute(&state.pool)
        .await?;
    }

    // Medium-loop conclusions: term graph and posterior for this user.
    graph::rebuild_user_graph(&state.pool, &did).await?;
    infer::update_posterior(&state.pool, &did).await?;

    Ok(())
}

pub async fn enqueue(pool: &sqlx::SqlitePool, completion_uri: &str) -> anyhow::Result<()> {
    sqlx::query(
        "INSERT INTO gloss_queue (completion_uri, enqueued_at) VALUES (?, ?)
         ON CONFLICT (completion_uri) DO NOTHING",
    )
    .bind(completion_uri)
    .bind(crate::db::now())
    .execute(pool)
    .await?;
    Ok(())
}