//! Todos the swarm notification router queues for this agent.
//!
//! The router publishes each change as a `swarm_wire::router_todo::RouterTodoOp`
//! on this agent's subject of the `ROUTER_TODOS` work-queue stream. This
//! task creates the agent's own durable consumer on it (`todo-<agent>`,
//! `swarm_queue_client::router_todo::open_own`), pulls, applies each op to the
//! todo store as subsystem `router`, and acks it, which removes it from the
//! stream. Ops queued while the agent was down wait there, so nothing is lost
//! to downtime.
//!
//! **Redelivery is harmless.** The store dedups on `(subsystem, key)`, so an
//! op applied twice changes nothing the second time, except that an upsert
//! reopens a todo the agent marked done in between (`reopen_if_acked`).
//!
//! **Own credential only.** The queue grants this agent's credential creating
//! its own consumer (the extended create subject, whose filter the server
//! checks), pulling from it and acking it, and nothing else of the stream.
//!
//! Until the router has created the stream, the consumer create fails; this
//! task retries every [`RETRY`].

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

use futures_util::StreamExt as _;
use swarm_wire::router_todo::{Op, RouterTodoOp, TODO_SUBSYSTEM};
use tokio::sync::Notify;

use crate::todos::Todos;

/// How long one pull waits for ops.
const PULL_WAIT: Duration = Duration::from_secs(30);

/// How many ops one pull asks for.
const BATCH: usize = 32;

/// How long to wait after a failure before creating the consumer again.
const RETRY: Duration = Duration::from_mins(1);

/// What applying one op came to.
#[derive(Debug, PartialEq, Eq)]
enum Applied {
    /// The list changed in a way the agent should be woken for.
    Wake,
    /// Nothing to wake for: unchanged, or a clear.
    Quiet,
}

/// Apply one op to the store.
fn apply(todos: &Todos, op: &RouterTodoOp) -> anyhow::Result<Applied> {
    match op.op {
        Op::Upsert => {
            let (_, changed) = todos.upsert(
                TODO_SUBSYSTEM,
                Some(&op.key),
                &op.summary,
                Some(&op.source),
                true,
            )?;
            Ok(if changed {
                Applied::Wake
            } else {
                Applied::Quiet
            })
        }
        Op::Clear if op.prefix => {
            todos.clear_prefix(TODO_SUBSYSTEM, &op.key)?;
            Ok(Applied::Quiet)
        }
        Op::Clear => {
            todos.clear(TODO_SUBSYSTEM, Some(&op.key))?;
            Ok(Applied::Quiet)
        }
    }
}

/// Drain this agent's consumer for the harness's lifetime, if it has a label
/// to name it with.
pub async fn run(todos: Arc<Todos>, wake: Arc<Notify>) {
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; not taking router todos");
        return;
    }
    let client = crate::swarm_queue::client().await;
    loop {
        if let Err(e) = drain(&client, &agent, &todos, &wake).await {
            tracing::info!(
                error = format!("{e:#}"),
                "router todos: not draining; retrying"
            );
        }
        tokio::time::sleep(RETRY).await;
    }
}

/// Create the consumer and pull from it until something fails.
async fn drain(
    client: &async_nats::Client,
    agent: &str,
    todos: &Todos,
    wake: &Notify,
) -> anyhow::Result<()> {
    swarm_queue_client::ensure_connected(client)?;
    let consumer = swarm_queue_client::router_todo::open_own(client, agent)
        .await
        .map_err(|e| anyhow::anyhow!("creating this agent's consumer: {e}"))?;
    tracing::info!(%agent, "router todos: draining this agent's queue");
    loop {
        let mut batch = consumer
            .batch()
            .max_messages(BATCH)
            .expires(PULL_WAIT)
            .messages()
            .await
            .map_err(|e| anyhow::anyhow!("requesting a batch: {e}"))?;
        let mut refused = false;
        while let Some(message) = batch.next().await {
            let message = message.map_err(|e| anyhow::anyhow!("pulling: {e}"))?;
            let ack = settle_op(&message.payload, &mut refused, |op| apply(todos, op), wake);
            message
                .ack_with(ack)
                .await
                .map_err(|e| anyhow::anyhow!("acking: {e}"))?;
        }
        if refused {
            anyhow::bail!(
                "the store refused an op; it and the rest of its batch come back in order"
            );
        }
    }
}

/// How one pulled op is acked. Once the store has refused an op of the batch
/// (`refused`), no later op is applied: they are all nakked for immediate
/// redelivery, so two ops on one key are never applied out of order, and the
/// caller waits before it pulls again.
fn settle_op(
    payload: &[u8],
    refused: &mut bool,
    apply: impl FnOnce(&RouterTodoOp) -> anyhow::Result<Applied>,
    wake: &Notify,
) -> async_nats::jetstream::AckKind {
    use async_nats::jetstream::AckKind;

    if *refused {
        return AckKind::Nak(None);
    }
    let op = match serde_json::from_slice::<RouterTodoOp>(payload) {
        Ok(op) => op,
        Err(e) => {
            tracing::warn!(error = %e, "router todos: an undecodable op; dropped");
            return AckKind::Term;
        }
    };
    match apply(&op) {
        Ok(applied) => {
            tracing::info!(key = %op.key, op = ?op.op, ?applied, "router todos: applied");
            if applied == Applied::Wake {
                wake.notify_one();
            }
            AckKind::Ack
        }
        Err(e) => {
            tracing::warn!(key = %op.key, error = ?e, "router todos: the store refused an op");
            *refused = true;
            AckKind::Nak(None)
        }
    }
}

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

    fn store() -> (tempfile::TempDir, Todos) {
        let dir = tempfile::tempdir().unwrap();
        let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap();
        (dir, db)
    }

    fn upsert(key: &str, summary: &str) -> RouterTodoOp {
        RouterTodoOp {
            op: Op::Upsert,
            key: key.to_owned(),
            prefix: false,
            summary: summary.to_owned(),
            source: "router:assigned".to_owned(),
        }
    }

    fn clear(key: &str, prefix: bool) -> RouterTodoOp {
        RouterTodoOp {
            op: Op::Clear,
            key: key.to_owned(),
            prefix,
            summary: String::new(),
            source: "router:closed".to_owned(),
        }
    }

    fn keys(todos: &Todos) -> Vec<String> {
        let mut k: Vec<String> = todos
            .list(Some(TODO_SUBSYSTEM))
            .unwrap()
            .into_iter()
            .filter_map(|t| t.subsystem_key)
            .collect();
        k.sort();
        k
    }

    #[test]
    fn an_upsert_wakes_once_and_its_redelivery_is_quiet() {
        let (_dir, s) = store();
        let op = upsert("forge:o/r#1:assignment", "you are assigned");
        assert_eq!(apply(&s, &op).unwrap(), Applied::Wake);
        assert_eq!(apply(&s, &op).unwrap(), Applied::Quiet);
        let rows = s.list(Some(TODO_SUBSYSTEM)).unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].source.as_deref(), Some("router:assigned"));
        // A new text under the same key rewrites the one row and wakes.
        assert_eq!(
            apply(&s, &upsert("forge:o/r#1:assignment", "not assigned")).unwrap(),
            Applied::Wake
        );
        assert_eq!(s.list(Some(TODO_SUBSYSTEM)).unwrap().len(), 1);
    }

    #[test]
    fn a_clear_removes_one_key_and_a_prefix_clear_every_match() {
        // Another issue whose number starts with the first one's.
        const OTHER: &str = "forge:o/r#12:assignment"; // lint:allow: a test key, not a tracker reference
        let (_dir, s) = store();
        for key in ["forge:o/r#1:assignment", "forge:o/r#1:review", OTHER] {
            apply(&s, &upsert(key, "x")).unwrap();
        }
        assert_eq!(
            apply(&s, &clear("forge:o/r#1:review", false)).unwrap(),
            Applied::Quiet
        );
        assert_eq!(keys(&s), [OTHER, "forge:o/r#1:assignment"]);
        // Not a prefix clear: the key must match whole.
        apply(&s, &clear("forge:o/r#1:", false)).unwrap();
        assert_eq!(keys(&s).len(), 2);
        apply(&s, &clear("forge:o/r#1:", true)).unwrap();
        assert_eq!(keys(&s), [OTHER]);
    }

    /// Two ops on one key: the first is refused, so the second is not applied
    /// ahead of it, and both are nakked to come back in order.
    #[test]
    fn after_a_refused_op_no_later_op_is_applied() {
        use async_nats::jetstream::AckKind;

        let wake = Notify::new();
        let assigned = serde_json::to_vec(&upsert("forge:o/r#1:assignment", "assigned")).unwrap();
        let unassigned =
            serde_json::to_vec(&upsert("forge:o/r#1:assignment", "not assigned")).unwrap();
        let mut refused = false;
        let mut applied = Vec::new();
        let mut step = |payload: &[u8], refuse: bool| {
            settle_op(
                payload,
                &mut refused,
                |op| {
                    applied.push(op.summary.clone());
                    if refuse {
                        anyhow::bail!("the store is locked")
                    }
                    Ok(Applied::Wake)
                },
                &wake,
            )
        };
        let first = step(&assigned, true);
        let second = step(&unassigned, false);
        assert_eq!(format!("{first:?}"), format!("{:?}", AckKind::Nak(None)));
        assert_eq!(format!("{second:?}"), format!("{:?}", AckKind::Nak(None)));
        assert_eq!(applied, ["assigned"]);
        assert!(refused);

        // Control: with nothing refused, both apply and are acked.
        let mut refused = false;
        for payload in [&assigned, &unassigned] {
            let ack = settle_op(payload, &mut refused, |_| Ok(Applied::Quiet), &wake);
            assert_eq!(format!("{ack:?}"), format!("{:?}", AckKind::Ack));
        }
    }

    #[test]
    fn router_ops_touch_no_other_subsystem() {
        let (_dir, s) = store();
        s.upsert(
            "forge",
            Some("forge:o/r#1:assignment"),
            "notify",
            None,
            false,
        )
        .unwrap();
        apply(&s, &upsert("forge:o/r#1:assignment", "router")).unwrap();
        apply(&s, &clear("forge:o/r#1:", true)).unwrap();
        let left = s.list(None).unwrap();
        assert_eq!(left.len(), 1);
        assert_eq!(left[0].subsystem, "forge");
    }
}