//! Local SQLite write-staging store (the CQRS command side).
//!
//! Writes are appended to an `_outbox` table and acknowledged immediately; the
//! sync worker later drains pending rows into Postgres. `rusqlite` is blocking,
//! so the connection lives behind an `Arc<Mutex<_>>` and every access runs on a
//! blocking thread via `tokio::task::spawn_blocking`.

use rusqlite::Connection;
use std::sync::{Arc, Mutex};

pub type SharedSqlite = Arc<Mutex<Connection>>;

/// One staged write awaiting migration to Postgres.
#[derive(Debug, Clone)]
pub struct OutboxRow {
    pub id: i64,
    pub entity: String,
    pub table_name: String,
    pub row_id: String,
    /// JSON object of accepted column -> value changes.
    pub payload_json: String,
}

/// Open (or create) the staging database and ensure the outbox table exists.
pub fn open(path: &str) -> rusqlite::Result<SharedSqlite> {
    let conn = Connection::open(path)?;
    conn.execute_batch(
        "CREATE TABLE IF NOT EXISTS _outbox (
            id           INTEGER PRIMARY KEY AUTOINCREMENT,
            entity       TEXT NOT NULL,
            table_name   TEXT NOT NULL,
            row_id       TEXT NOT NULL,
            payload_json TEXT NOT NULL,
            status       TEXT NOT NULL DEFAULT 'pending',
            created_at   TEXT NOT NULL DEFAULT (datetime('now'))
        );",
    )?;
    Ok(Arc::new(Mutex::new(conn)))
}

/// Append a staged write. Returns the new outbox id.
pub async fn enqueue(
    db: &SharedSqlite,
    entity: String,
    table: String,
    row_id: String,
    payload_json: String,
) -> rusqlite::Result<i64> {
    let db = db.clone();
    tokio::task::spawn_blocking(move || {
        let conn = db.lock().unwrap();
        conn.execute(
            "INSERT INTO _outbox (entity, table_name, row_id, payload_json) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params![entity, table, row_id, payload_json],
        )?;
        Ok(conn.last_insert_rowid())
    })
    .await
    .expect("sqlite enqueue task panicked")
}

/// Fetch up to `limit` pending outbox rows in insertion order.
pub async fn fetch_pending(db: &SharedSqlite, limit: usize) -> rusqlite::Result<Vec<OutboxRow>> {
    let db = db.clone();
    tokio::task::spawn_blocking(move || {
        let conn = db.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, entity, table_name, row_id, payload_json
             FROM _outbox WHERE status = 'pending' ORDER BY id LIMIT ?1",
        )?;
        let rows = stmt
            .query_map([limit as i64], |r| {
                Ok(OutboxRow {
                    id: r.get(0)?,
                    entity: r.get(1)?,
                    table_name: r.get(2)?,
                    row_id: r.get(3)?,
                    payload_json: r.get(4)?,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        Ok(rows)
    })
    .await
    .expect("sqlite fetch task panicked")
}

/// Update the status of an outbox row (`done` / `error`).
pub async fn set_status(db: &SharedSqlite, id: i64, status: &str) -> rusqlite::Result<()> {
    let db = db.clone();
    let status = status.to_string();
    tokio::task::spawn_blocking(move || {
        let conn = db.lock().unwrap();
        conn.execute(
            "UPDATE _outbox SET status = ?1 WHERE id = ?2",
            rusqlite::params![status, id],
        )?;
        Ok(())
    })
    .await
    .expect("sqlite status task panicked")
}
