//! A dedicated SQLite writer batches entries and serves ordered queries without propagating disk failures into the host.
use rusqlite::{Connection, OpenFlags, params};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::io::Write;
use std::{
    collections::HashSet,
    path::PathBuf,
    sync::{Mutex, mpsc},
};
use tokio::sync::oneshot;
#[derive(Debug, Serialize, Deserialize)]
struct Entry {
    time: String,
    id: String,
    pid: String,
    plugin: String,
    stream: String,
    line: String,
}
enum Command {
    Write(Entry),
    Query(Value, oneshot::Sender<Value>),
}
pub(crate) struct Journal {
    sender: mpsc::Sender<Command>,
    warned: Mutex<HashSet<String>>,
    spool: PathBuf,
}
impl Journal {
    pub(crate) async fn memory() -> anyhow::Result<Self> {
        Self::open(PathBuf::from(":memory:")).await
    }
    pub(crate) async fn open(path: PathBuf) -> anyhow::Result<Self> {
        Self::open_with_legacy(path, None).await
    }
    pub(crate) async fn open_with_legacy(
        path: PathBuf,
        legacy: Option<PathBuf>,
    ) -> anyhow::Result<Self> {
        let spool = path
            .parent()
            .unwrap_or(std::path::Path::new("."))
            .join("failed-batches");
        let worker_spool = spool.clone();
        let (sender, receiver) = mpsc::channel();
        let (ready, initialized) = oneshot::channel();
        std::thread::spawn(move || {
            match (|| -> anyhow::Result<_> {
                let db = open_database(path)?;
                // Never rename, checkpoint, or write the old SQLite/WAL files.
                let old = legacy
                    .filter(|p| p.exists())
                    .map(|p| Connection::open_with_flags(p, OpenFlags::SQLITE_OPEN_READ_ONLY))
                    .transpose()?;
                Ok((db, old))
            })() {
                Ok((mut db, old)) => {
                    let _ = ready.send(Ok(()));
                    let mut pending = Vec::new();
                    while let Ok(first) = receiver.recv() {
                        let mut batch = vec![first];
                        batch.extend(receiver.try_iter().take(255));
                        pending = write_batch(&mut db, old.as_ref(), &worker_spool, pending, batch);
                    }
                    if !pending.is_empty() {
                        // Both storage locations failed. Last resort: retain full text in host stderr.
                        eprintln!(
                            "journal shutdown with unpersisted entries: {}",
                            serde_json::to_string(&pending).unwrap()
                        );
                    }
                }
                Err(error) => {
                    eprintln!("journal open failed: {error}");
                    let _ = ready.send(Err(error));
                }
            }
        });
        initialized.await??;
        Ok(Self {
            sender,
            warned: Mutex::new(HashSet::new()),
            spool,
        })
    }
    pub(crate) fn write(&self, pid: &str, plugin: &str, stream: &str, line: &str) {
        let entry = Entry {
            time: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
            id: uuid::Uuid::now_v7().to_string(),
            pid: pid.into(),
            plugin: plugin.into(),
            stream: stream.into(),
            line: line.into(),
        };
        if let Err(mpsc::SendError(Command::Write(entry))) = self.sender.send(Command::Write(entry))
        {
            eprintln!("journal writer stopped; saving entry outside SQLite");
            if let Err(error) = spool_batch(&self.spool, std::slice::from_ref(&entry)) {
                eprintln!(
                    "journal spool failed: {error}; unpersisted entry: {}",
                    serde_json::to_string(&entry).unwrap()
                );
            }
        }
    }
    pub(crate) fn warning(&self, pid: &str, plugin: &str) {
        if self.warned.lock().unwrap().insert(pid.into()) {
            self.write(pid, plugin, "core", "warning: discarded stdout");
        }
    }
    pub(crate) fn forget(&self, pid: &str) {
        self.warned.lock().unwrap().remove(pid);
    }
    pub(crate) fn query_later(&self, filters: Value) -> oneshot::Receiver<Value> {
        let (tx, rx) = oneshot::channel();
        if self.sender.send(Command::Query(filters, tx)).is_err() {
            eprintln!("journal writer stopped");
        }
        rx
    }
    pub(crate) async fn query(&self, filters: Value) -> Value {
        self.query_later(filters)
            .await
            .unwrap_or_else(|_| json!([]))
    }
}
fn open_database(path: PathBuf) -> anyhow::Result<Connection> {
    if path != std::path::Path::new(":memory:") {
        std::fs::create_dir_all(path.parent().unwrap())?;
    }
    let db = Connection::open(path)?;
    db.execute_batch("PRAGMA journal_mode=WAL;
        PRAGMA synchronous=NORMAL;
        CREATE TABLE IF NOT EXISTS journal (id TEXT PRIMARY KEY, time TEXT, pid TEXT, plugin TEXT, stream TEXT, line TEXT);
        CREATE INDEX IF NOT EXISTS journal_pid ON journal(pid);
        CREATE INDEX IF NOT EXISTS journal_plugin ON journal(plugin);")?;
    Ok(db)
}
/// Exclusive, synced JSON files preserve failed transactions without a busy retry loop.
/// Recovery is deliberate: files are never automatically removed or replayed.
fn spool_batch(dir: &std::path::Path, entries: &[Entry]) -> anyhow::Result<PathBuf> {
    std::fs::create_dir_all(dir)?;
    let path = dir.join(format!("{}.json", uuid::Uuid::now_v7()));
    // A failed/partial write must not look like a complete recovery batch.
    let mut file = tempfile::NamedTempFile::new_in(dir)?;
    serde_json::to_writer(&mut file, entries)?;
    file.write_all(b"\n")?;
    file.as_file().sync_all()?;
    file.persist_noclobber(&path)?;
    std::fs::File::open(dir)?.sync_all()?;
    eprintln!(
        "journal failed batch preserved: {} ({} entries)",
        path.display(),
        entries.len()
    );
    Ok(path)
}

fn write_batch(
    db: &mut Connection,
    old: Option<&Connection>,
    spool: &std::path::Path,
    mut pending: Vec<Entry>,
    batch: Vec<Command>,
) -> Vec<Entry> {
    let mut queries = Vec::new();
    for command in batch {
        match command {
            Command::Write(entry) => pending.push(entry),
            Command::Query(filters, response) => queries.push((filters, response)),
        }
    }
    if !pending.is_empty() {
        let write = (|| -> rusqlite::Result<()> {
            let transaction = db.transaction()?;
            for e in &pending {
                transaction.execute("INSERT OR IGNORE INTO journal (id,time,pid,plugin,stream,line) VALUES (?1,?2,?3,?4,?5,?6)",params![e.id,e.time,e.pid,e.plugin,e.stream,e.line])?;
            }
            transaction.commit()
        })();
        match write {
            Ok(()) => pending.clear(),
            Err(error) => {
                eprintln!("journal write failed: {error}");
                match spool_batch(spool, &pending) {
                    Ok(_) => pending.clear(),
                    Err(error) => eprintln!(
                        "journal spool failed: {error}; retaining {} entries in memory until next command",
                        pending.len()
                    ),
                }
            }
        }
    }
    for (filters, response) in queries {
        let value = query_all(db, old, &filters).unwrap_or_else(|error| {
            eprintln!("journal query failed: {error}");
            json!([])
        });
        let _ = response.send(value);
    }
    pending
}

fn query_all(
    db: &Connection,
    old: Option<&Connection>,
    filters: &Value,
) -> rusqlite::Result<Value> {
    let mut entries = query_database(db, filters)?.as_array().unwrap().clone();
    if let Some(old) = old {
        entries.extend(
            query_database(old, filters)?
                .as_array()
                .unwrap()
                .iter()
                .cloned(),
        );
        entries.sort_by(|a, b| a["id"].as_str().cmp(&b["id"].as_str()));
        entries.dedup_by(|a, b| a["id"] == b["id"]);
        let limit = filters["limit"].as_i64().unwrap_or(100);
        if limit >= 0 {
            entries.drain(..entries.len().saturating_sub(limit as usize));
        }
    }
    Ok(json!(entries))
}
fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result<Value> {
    let mut statement = db.prepare(
        "SELECT id,time,pid,plugin,stream,line FROM journal
        WHERE (?1 IS NULL OR plugin=?1) AND (?2 IS NULL OR pid=?2)
        AND (?3 IS NULL OR id>=?3) AND (?4 IS NULL OR id<=?4)
        AND (?5 IS NULL OR instr(line,?5)>0) ORDER BY id DESC LIMIT ?6",
    )?;
    let rows=statement.query_map(params![filters["plugin"].as_str(),filters["pid"].as_str(),filters["since"].as_str(),filters["until"].as_str(),filters["grep"].as_str(),filters["limit"].as_i64().unwrap_or(100)],|row| Ok(json!({
        "id":row.get::<_,String>(0)?,"time":row.get::<_,String>(1)?,"pid":row.get::<_,String>(2)?,"plugin":row.get::<_,String>(3)?,"stream":row.get::<_,String>(4)?,"line":row.get::<_,String>(5)?,
    })))?;
    let mut entries: Vec<_> = rows.collect::<rusqlite::Result<_>>()?;
    entries.reverse();
    Ok(json!(entries))
}

#[cfg(test)]
mod tests {
    use super::*;
    fn entry(line: &str) -> Entry {
        Entry {
            id: uuid::Uuid::now_v7().to_string(),
            time: "2026-10-26T12:00:00.000Z".into(),
            pid: "pid".into(),
            plugin: "test".into(),
            stream: "core".into(),
            line: line.into(),
        }
    }

    #[test]
    fn failed_transaction_is_spooled_in_full_and_queries_finish() {
        let dir = tempfile::tempdir().unwrap();
        let mut db = open_database(PathBuf::from(":memory:")).unwrap();
        db.execute_batch("PRAGMA query_only=ON").unwrap();
        let original = "😀 full failure\n".repeat(6000);
        let (tx, mut rx) = oneshot::channel();
        let pending = write_batch(
            &mut db,
            None,
            dir.path(),
            vec![],
            vec![
                Command::Write(entry(&original)),
                Command::Query(json!({}), tx),
            ],
        );
        assert!(pending.is_empty());
        assert_eq!(rx.try_recv().unwrap(), json!([]));
        let path = std::fs::read_dir(dir.path())
            .unwrap()
            .next()
            .unwrap()
            .unwrap()
            .path();
        let saved: Vec<Entry> = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
        assert_eq!(saved.len(), 1);
        assert_eq!(saved[0].line, original);
        assert_eq!(saved[0].time, "2026-10-26T12:00:00.000Z");
    }

    #[test]
    fn failed_spool_retries_on_next_command_without_duplicates() {
        let dir = tempfile::tempdir().unwrap();
        let blocked = dir.path().join("blocked");
        std::fs::write(&blocked, "not a directory").unwrap();
        let mut db = open_database(PathBuf::from(":memory:")).unwrap();
        db.execute_batch("PRAGMA query_only=ON").unwrap();
        let pending = write_batch(
            &mut db,
            None,
            &blocked,
            vec![],
            vec![Command::Write(entry("retained"))],
        );
        assert_eq!(pending.len(), 1);
        db.execute_batch("PRAGMA query_only=OFF").unwrap();
        let (tx, mut rx) = oneshot::channel();
        let pending = write_batch(
            &mut db,
            None,
            &blocked,
            pending,
            vec![Command::Query(json!({}), tx)],
        );
        assert!(pending.is_empty());
        assert_eq!(rx.try_recv().unwrap()[0]["line"], "retained");
        assert_eq!(
            query_all(&db, None, &json!({}))
                .unwrap()
                .as_array()
                .unwrap()
                .len(),
            1
        );
    }

    #[tokio::test]
    async fn legacy_wal_and_new_journal_share_filters_order_and_limit() {
        let dir = tempfile::tempdir().unwrap();
        let legacy = dir.path().join("state/core/journal.sqlite");
        let mut old = open_database(legacy.clone()).unwrap();
        write_batch(
            &mut old,
            None,
            dir.path(),
            vec![],
            vec![Command::Write(entry("old match"))],
        );
        // Leave the WAL connection open: opening the new journal must not move its files.
        let wal = legacy.with_extension("sqlite-wal");
        let before = std::fs::read(&wal).unwrap();
        let new = dir.path().join("logs/core/journal.sqlite");
        let journal = Journal::open_with_legacy(new.clone(), Some(legacy.clone()))
            .await
            .unwrap();
        journal.write("pid", "test", "core", "new match");
        let entries = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            journal.query(json!({"pid":"pid","grep":"match","limit":-1})),
        )
        .await
        .unwrap();
        assert_eq!(entries.as_array().unwrap().len(), 2);
        assert_eq!(entries[0]["line"], "old match");
        assert_eq!(entries[1]["line"], "new match");
        assert_eq!(
            journal.query(json!({"limit":1})).await[0]["line"],
            "new match"
        );
        assert_eq!(
            journal.query(json!({"until":entries[0]["id"]})).await[0]["line"],
            "old match"
        );
        assert_eq!(
            journal
                .query(json!({"since":entries[1]["id"],"plugin":"test"}))
                .await[0]["line"],
            "new match"
        );
        assert_eq!(journal.query(json!({"grep":"absent"})).await, json!([]));
        assert_eq!(std::fs::read(wal).unwrap(), before);
        assert!(legacy.exists() && new.exists());
        assert_eq!(
            query_database(&old, &json!({}))
                .unwrap()
                .as_array()
                .unwrap()
                .len(),
            1
        );
    }

    /// PROTOCOL 7: completed processes do not leave warning identities in memory.
    #[tokio::test]
    async fn completion_forgets_warning_identity() {
        let folder = std::env::temp_dir().join(format!("sausage-warning-{}", uuid::Uuid::now_v7()));
        crate::world::World::create(&folder).unwrap();
        let host = crate::Host::start_with_cache(&folder, &folder.join("cache"))
            .await
            .unwrap();
        host.register_native(
            "warn",
            std::sync::Arc::new(|ctx, _, _| {
                Box::pin(async move {
                    ctx.host.output(&ctx.pid, "not JSON".into());
                    Ok(json!(ctx.pid))
                })
            }),
        );
        let pid = host.call("warn.once", json!({})).await.unwrap();
        host.call("core.stop", json!({"pid":pid})).await.unwrap();
        let retained = host
            .0
            .journal
            .warned
            .lock()
            .unwrap()
            .contains(pid.as_str().unwrap());
        host.stop().await;
        std::fs::remove_dir_all(folder).unwrap();
        assert!(!retained);
    }
}