//! The single writer task.
//!
//! Everything that writes to SQLite sends a [`WriteOp`] down an `mpsc` channel.
//! One task drains it, batching everything that arrives within 250 ms (or 512
//! operations, whichever comes first) into one transaction.
//!
//! Three properties follow, and all three matter:
//!
//! * **No `SQLITE_BUSY`, ever.** There is exactly one writer, so there is nothing
//!   to contend with.
//! * **fsyncs are amortised.** 100 devices reporting once a minute is ~4
//!   transactions per second, not 100.
//! * **The UDP loop never blocks on the database.** A full channel is answered
//!   with `THROTTLE` and the datagram is dropped; the client will retry. Applying
//!   backpressure to the receive loop instead would turn a storage stall into
//!   packet loss for *every* device, including the ones the server could still
//!   serve.

use std::time::Duration;

use anyhow::{Context, Result};
use otproto::Point;
use sqlx::{Sqlite, SqlitePool, Transaction};
use tokio::sync::mpsc;
use tracing::{debug, error, warn};

/// Channel depth. Deep enough to absorb a burst of queue flushes from every
/// device at once, shallow enough that a stalled disk is noticed in seconds
/// rather than after the process has eaten a gigabyte of positions.
const CHANNEL_DEPTH: usize = 8192;

/// Longest a write waits to be committed.
const BATCH_WINDOW: Duration = Duration::from_millis(250);

/// Most operations in one transaction.
const BATCH_MAX: usize = 512;

#[derive(Debug)]
pub enum WriteOp {
    /// Points from one authenticated `LOC`, all belonging to one account.
    Points {
        user_id: i64,
        /// Provenance only.
        src_token_id: i64,
        points: Vec<Point>,
        recv_at: i64,
    },
    /// Per-datagram telemetry for a token. Written on every authenticated
    /// packet, which is why it must be batched rather than done inline.
    TokenSeen {
        token_id: i64,
        at: i64,
        src_ip: String,
        src_port: u16,
        transport: &'static str,
    },
    /// Recorded from `HELLO`.
    TokenHello {
        token_id: i64,
        app_version: i64,
        os_api_level: i64,
    },
    Audit {
        user_id: Option<i64>,
        at: i64,
        action: String,
        detail: String,
        src_ip: Option<String>,
    },
}

/// Send handle. Cheap to clone; hand one to every task that needs to write.
#[derive(Clone)]
pub struct WriteHandle {
    tx: mpsc::Sender<WriteOp>,
}

/// Whether a write was accepted.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Accepted {
    Yes,
    /// The channel is full. The caller should signal `THROTTLE` and drop.
    Saturated,
    /// The writer task is gone — we are shutting down.
    Closed,
}

impl WriteHandle {
    /// Never awaits, never blocks. This is what the UDP loop calls.
    pub fn try_send(&self, op: WriteOp) -> Accepted {
        match self.tx.try_send(op) {
            Ok(()) => Accepted::Yes,
            Err(mpsc::error::TrySendError::Full(_)) => Accepted::Saturated,
            Err(mpsc::error::TrySendError::Closed(_)) => Accepted::Closed,
        }
    }

    /// For HTTP handlers, which can afford to wait for a slot.
    pub async fn send(&self, op: WriteOp) -> Result<()> {
        self.tx.send(op).await.context("writer task has stopped")
    }

    /// Approximate free capacity, for `/metrics` and for deciding when to warn.
    pub fn capacity(&self) -> usize {
        self.tx.capacity()
    }
}

/// Start the writer task. Returns the handle and a join handle for shutdown.
pub fn spawn(pool: SqlitePool) -> (WriteHandle, tokio::task::JoinHandle<()>) {
    let (tx, rx) = mpsc::channel(CHANNEL_DEPTH);
    let task = tokio::spawn(run(pool, rx));
    (WriteHandle { tx }, task)
}

async fn run(pool: SqlitePool, mut rx: mpsc::Receiver<WriteOp>) {
    let mut batch: Vec<WriteOp> = Vec::with_capacity(BATCH_MAX);

    loop {
        // Block until there is something to do — no idle polling.
        let Some(first) = rx.recv().await else {
            break; // all senders dropped: shutdown
        };
        batch.push(first);

        // Then take whatever else shows up inside the window.
        let deadline = tokio::time::Instant::now() + BATCH_WINDOW;
        while batch.len() < BATCH_MAX {
            match tokio::time::timeout_at(deadline, rx.recv()).await {
                Ok(Some(op)) => batch.push(op),
                Ok(None) => break, // channel closed; commit what we have
                Err(_) => break,   // window elapsed
            }
        }

        if let Err(e) = commit(&pool, &mut batch).await {
            // Losing a batch of positions is bad but survivable; the clients
            // still hold them unacked and will retry. Dying here is not
            // survivable, so log loudly and carry on.
            error!(error = %e, "write batch failed; clients will retry");
            batch.clear();
        }
    }

    // Drain whatever is left so a graceful SIGTERM does not lose points.
    while let Ok(op) = rx.try_recv() {
        batch.push(op);
        if batch.len() >= BATCH_MAX
            && let Err(e) = commit(&pool, &mut batch).await
        {
            error!(error = %e, "final write batch failed");
            batch.clear();
        }
    }
    if !batch.is_empty()
        && let Err(e) = commit(&pool, &mut batch).await
    {
        error!(error = %e, "final write batch failed");
    }
    debug!("writer task stopped");
}

async fn commit(pool: &SqlitePool, batch: &mut Vec<WriteOp>) -> Result<()> {
    let count = batch.len();
    let mut tx = pool.begin().await.context("begin transaction")?;
    for op in batch.drain(..) {
        apply(&mut tx, op).await?;
    }
    tx.commit().await.context("commit transaction")?;
    if count > BATCH_MAX / 2 {
        warn!(
            count,
            "large write batch — the writer may be falling behind"
        );
    } else {
        debug!(count, "committed write batch");
    }
    Ok(())
}

async fn apply(tx: &mut Transaction<'_, Sqlite>, op: WriteOp) -> Result<()> {
    match op {
        WriteOp::Points {
            user_id,
            src_token_id,
            points,
            recv_at,
        } => {
            for p in points {
                insert_point(tx, user_id, src_token_id, &p, recv_at).await?;
            }
        }

        WriteOp::TokenSeen {
            token_id,
            at,
            src_ip,
            src_port,
            transport,
        } => {
            // The source address is recorded but never used for authentication,
            // which is exactly why a Wi-Fi to LTE handoff needs no protocol work.
            sqlx::query(
                "UPDATE tokens SET last_seen_at = ?, last_src_ip = ?, last_src_port = ?, \
                 last_transport = ? WHERE token_id = ?",
            )
            .bind(at)
            .bind(src_ip)
            .bind(i64::from(src_port))
            .bind(transport)
            .bind(token_id)
            .execute(&mut **tx)
            .await
            .context("updating token telemetry")?;
        }

        WriteOp::TokenHello {
            token_id,
            app_version,
            os_api_level,
        } => {
            sqlx::query("UPDATE tokens SET app_version = ?, os_api_level = ? WHERE token_id = ?")
                .bind(app_version)
                .bind(os_api_level)
                .bind(token_id)
                .execute(&mut **tx)
                .await
                .context("recording HELLO")?;
        }

        WriteOp::Audit {
            user_id,
            at,
            action,
            detail,
            src_ip,
        } => {
            sqlx::query(
                "INSERT INTO audit_log (at, user_id, action, detail, src_ip) VALUES (?, ?, ?, ?, ?)",
            )
            .bind(at)
            .bind(user_id)
            .bind(action)
            .bind(detail)
            .bind(src_ip)
            .execute(&mut **tx)
            .await
            .context("writing audit log")?;
        }
    }
    Ok(())
}

/// Insert one point and, if it is newer than what we have, update the live marker.
///
/// The `ON CONFLICT` clause is the load-bearing part of the whole storage design:
///
/// * A **retry or replay** carries a `ts` that already exists, so it collapses
///   into the existing row. This is why the protocol needs no replay window.
/// * **Two phones on one account reporting in the same second** would otherwise
///   resolve to whichever packet landed last, which is arbitrary. The
///   `WHERE excluded.acc_dm < points.acc_dm` guard keeps the better fix.
async fn insert_point(
    tx: &mut Transaction<'_, Sqlite>,
    user_id: i64,
    src_token_id: i64,
    p: &Point,
    recv_at: i64,
) -> Result<()> {
    let ts = i64::from(p.ts);
    let acc = p.acc_dm.map(i64::from);

    sqlx::query(
        "INSERT INTO points \
           (user_id, ts, lat, lon, acc_dm, alt_m, spd_cms, brg_cdeg, bat_pct, flags, recv_at, src_token_id) \
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
         ON CONFLICT (user_id, ts) DO UPDATE SET \
           lat = excluded.lat, lon = excluded.lon, acc_dm = excluded.acc_dm, \
           alt_m = excluded.alt_m, spd_cms = excluded.spd_cms, brg_cdeg = excluded.brg_cdeg, \
           bat_pct = excluded.bat_pct, flags = excluded.flags, recv_at = excluded.recv_at, \
           src_token_id = excluded.src_token_id \
         WHERE excluded.acc_dm IS NOT NULL \
           AND (points.acc_dm IS NULL OR excluded.acc_dm < points.acc_dm)",
    )
    .bind(user_id)
    .bind(ts)
    .bind(p.lat_e7)
    .bind(p.lon_e7)
    .bind(acc)
    .bind(p.alt_m.map(i64::from))
    .bind(p.spd_cms.map(i64::from))
    .bind(p.brg_cdeg.map(i64::from))
    .bind(p.bat_pct.map(i64::from))
    .bind(i64::from(p.flags.0))
    .bind(recv_at)
    .bind(src_token_id)
    .execute(&mut **tx)
    .await
    .context("inserting point")?;

    // The live marker only moves forward in client time. A phone that has been
    // in a drawer and wakes up with an old queued fix must not drag the dot back.
    sqlx::query(
        "INSERT INTO user_latest \
           (user_id, ts, lat, lon, acc_dm, alt_m, spd_cms, brg_cdeg, bat_pct, flags, recv_at, src_token_id) \
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
         ON CONFLICT (user_id) DO UPDATE SET \
           ts = excluded.ts, lat = excluded.lat, lon = excluded.lon, acc_dm = excluded.acc_dm, \
           alt_m = excluded.alt_m, spd_cms = excluded.spd_cms, brg_cdeg = excluded.brg_cdeg, \
           bat_pct = excluded.bat_pct, flags = excluded.flags, recv_at = excluded.recv_at, \
           src_token_id = excluded.src_token_id \
         WHERE excluded.ts > user_latest.ts",
    )
    .bind(user_id)
    .bind(ts)
    .bind(p.lat_e7)
    .bind(p.lon_e7)
    .bind(acc)
    .bind(p.alt_m.map(i64::from))
    .bind(p.spd_cms.map(i64::from))
    .bind(p.brg_cdeg.map(i64::from))
    .bind(p.bat_pct.map(i64::from))
    .bind(i64::from(p.flags.0))
    .bind(recv_at)
    .bind(src_token_id)
    .execute(&mut **tx)
    .await
    .context("updating live position")?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use otproto::point::Flags;

    async fn seeded() -> (crate::db::Db, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("temp dir");
        let db = crate::db::Db::open(&dir.path().join("t.db"))
            .await
            .expect("open");
        sqlx::query(
            "INSERT INTO users (id, username, pw_hash, display_name, created_at, pw_changed_at) \
             VALUES (1, 'a', 'x', 'A', 0, 0)",
        )
        .execute(&db.write)
        .await
        .expect("user");
        (db, dir)
    }

    fn point(ts: u32, lat: i32, acc: Option<u16>) -> Point {
        Point {
            acc_dm: acc,
            flags: Flags::NONE,
            ..Point::new(ts, lat, 0)
        }
    }

    async fn drain(handle: &WriteHandle, task: tokio::task::JoinHandle<()>) {
        drop(handle.clone());
        // Dropping the last handle ends the task; the caller keeps one, so use an
        // explicit timeout instead of awaiting forever.
        let _ = tokio::time::timeout(Duration::from_millis(50), task).await;
    }

    #[tokio::test]
    async fn a_batch_of_points_lands_in_one_transaction() {
        let (db, _dir) = seeded().await;
        let (handle, task) = spawn(db.write.clone());

        handle
            .send(WriteOp::Points {
                user_id: 1,
                src_token_id: 7,
                points: (0..40)
                    .map(|i| point(1000 + i, i as i32, Some(50)))
                    .collect(),
                recv_at: 2000,
            })
            .await
            .expect("send");

        tokio::time::sleep(Duration::from_millis(400)).await;
        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM points")
            .fetch_one(&db.read)
            .await
            .expect("count");
        assert_eq!(count, 40);
        drain(&handle, task).await;
    }

    #[tokio::test]
    async fn the_live_marker_only_moves_forward_in_client_time() {
        let (db, _dir) = seeded().await;
        let (handle, task) = spawn(db.write.clone());

        // Newest first, then an older queued fix — the drawer-phone scenario.
        for (ts, lat) in [(2000u32, 100), (1000, 999)] {
            handle
                .send(WriteOp::Points {
                    user_id: 1,
                    src_token_id: 7,
                    points: vec![point(ts, lat, Some(50))],
                    recv_at: 3000,
                })
                .await
                .expect("send");
        }
        tokio::time::sleep(Duration::from_millis(400)).await;

        let (ts, lat): (i64, i64) =
            sqlx::query_as("SELECT ts, lat FROM user_latest WHERE user_id = 1")
                .fetch_one(&db.read)
                .await
                .expect("latest");
        assert_eq!(
            (ts, lat),
            (2000, 100),
            "an older fix must not drag the marker back"
        );

        // Both points are still in the history, though.
        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM points")
            .fetch_one(&db.read)
            .await
            .expect("count");
        assert_eq!(count, 2);
        drain(&handle, task).await;
    }

    #[tokio::test]
    async fn a_replayed_point_is_idempotent() {
        let (db, _dir) = seeded().await;
        let (handle, task) = spawn(db.write.clone());

        for _ in 0..3 {
            handle
                .send(WriteOp::Points {
                    user_id: 1,
                    src_token_id: 7,
                    points: vec![point(1000, 42, Some(80))],
                    recv_at: 2000,
                })
                .await
                .expect("send");
        }
        tokio::time::sleep(Duration::from_millis(400)).await;

        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM points")
            .fetch_one(&db.read)
            .await
            .expect("count");
        assert_eq!(count, 1, "replay must not create duplicate rows");
        drain(&handle, task).await;
    }

    #[tokio::test]
    async fn a_same_second_collision_keeps_the_better_accuracy() {
        let (db, _dir) = seeded().await;
        let (handle, task) = spawn(db.write.clone());

        for (acc, lat) in [(200u16, 1), (30, 2), (500, 3)] {
            handle
                .send(WriteOp::Points {
                    user_id: 1,
                    src_token_id: 7,
                    points: vec![point(1000, lat, Some(acc))],
                    recv_at: 2000,
                })
                .await
                .expect("send");
        }
        tokio::time::sleep(Duration::from_millis(400)).await;

        let (acc, lat): (i64, i64) =
            sqlx::query_as("SELECT acc_dm, lat FROM points WHERE user_id = 1 AND ts = 1000")
                .fetch_one(&db.read)
                .await
                .expect("point");
        assert_eq!(
            (acc, lat),
            (30, 2),
            "the better fix must win, whatever the arrival order"
        );
        drain(&handle, task).await;
    }
}