portal/src/deadlines.rs

//! The moves a clock makes rather than a person.
//!
//! A state in `aggregates.yaml` may carry `after: { days: 42, to:
//! no_answer }`: when a record has waited there that long, portal makes
//! the move itself. Every deadline a public body keeps is of this
//! shape - a hearing goes out and the authority has six weeks to
//! answer, and when it does not the case has to go on without one, with
//! nobody in the building in a position to press that button.
//!
//! A deadline may only do what a person could have done: `after.to` is
//! held to the same declared edge as a desk button (see
//! `parse_aggregates_yaml`), the move goes through the same
//! `aggregates::transition`, the same mail fires, and it lands on
//! `portal.answers.submitted` like any other decision. What it is not
//! is a job queue: the unit is days, the tick is an hour, and a record
//! that has waited long enough is moved once, because after the move it
//! is no longer in a state that has a deadline.

use crate::answers::Answer;
use crate::events::{emit_answer_submitted, AnswerSubmitted};
use crate::server::AppState;

/// Stamped on the record and on the transition event in place of a
/// username, so a desk and an n8n workflow can both tell a decision
/// nobody made from one somebody did.
pub const BY: &str = "deadline";

const DAY_MS: i64 = 24 * 60 * 60 * 1000;

pub async fn run(state: AppState) {
    let mut tick = tokio::time::interval(std::time::Duration::from_secs(60 * 60));
    loop {
        tick.tick().await;
        if let Err(e) = sweep(&state).await {
            tracing::warn!(error = %e, "deadline sweep failed");
        }
    }
}

/// Records leaving their list: once a record has been in a state with
/// `leaves_after` for that long, it is taken out of the bucket. Not a
/// move - nothing is decided, no mail goes - and the event log keeps
/// the record. Every minute, over the buckets that use it and no
/// others.
pub async fn run_leaving(state: AppState) {
    let mut tick = tokio::time::interval(std::time::Duration::from_secs(60));
    loop {
        tick.tick().await;
        if let Err(e) = sweep_leaving(&state).await {
            tracing::warn!(error = %e, "leaving sweep failed");
        }
    }
}

/// Whether a record has stayed out its time in the state it is in.
pub fn has_left(schema: &crate::aggregates::AggregateSchema, answer: &Answer, now_ms: i64) -> bool {
    schema
        .leaves_for_state
        .get(&answer.state)
        .is_some_and(|leaves| now_ms >= waiting_since(answer) + leaves.millis())
}

pub async fn sweep_leaving(state: &AppState) -> anyhow::Result<()> {
    let now_ms = chrono::Utc::now().timestamp_millis();
    let aggregates = state.aggregates.load();
    for schema in aggregates.values().filter(|s| !s.leaves_for_state.is_empty()) {
        let Ok(store) = state.jetstream.get_key_value(&schema.bucket).await else { continue };
        let keys: Vec<String> = {
            use futures::TryStreamExt;
            match store.keys().await {
                Ok(keys) => keys.try_collect().await.unwrap_or_default(),
                Err(_) => continue,
            }
        };
        for id in keys {
            let Ok(Some(bytes)) = store.get(&id).await else { continue };
            let Ok(answer) = serde_json::from_slice::<Answer>(&bytes) else { continue };
            if has_left(schema, &answer, now_ms) {
                match store.delete(&id).await {
                    Ok(()) => tracing::info!(bucket = %schema.bucket, id, state = %answer.state, "a record left its list"),
                    Err(e) => tracing::warn!(bucket = %schema.bucket, id, "a record could not leave its list: {e}"),
                }
            }
        }
    }
    Ok(())
}

/// When a record entered the state it is in now. A record nobody has
/// moved has been waiting since it was sent.
fn waiting_since(answer: &Answer) -> i64 {
    answer.decided_ms.unwrap_or(answer.submitted_ms)
}

/// Whether a record in `state` has waited out its deadline by `now_ms`.
pub fn due(schema: &crate::aggregates::AggregateSchema, answer: &Answer, now_ms: i64) -> Option<String> {
    let after = schema.after_for(&answer.state)?;
    let deadline = waiting_since(answer) + i64::from(after.days) * DAY_MS;
    (now_ms >= deadline).then(|| after.to.clone())
}

/// One pass over every bucket that has a deadline on any of its states.
pub async fn sweep(state: &AppState) -> anyhow::Result<()> {
    let now_ms = chrono::Utc::now().timestamp_millis();
    let aggregates = state.aggregates.load();
    for schema in aggregates.values() {
        if schema.after_for_state.is_empty() {
            continue;
        }
        // A bucket is created on first write, so one that has never
        // been written to is an ordinary empty bucket, not an error.
        let Ok(store) = state.jetstream.get_key_value(&schema.bucket).await else {
            continue;
        };
        let keys: Vec<String> = {
            use futures::TryStreamExt;
            match store.keys().await {
                Ok(keys) => keys.try_collect().await.unwrap_or_default(),
                Err(_) => continue,
            }
        };
        for id in keys {
            let Ok(Some(bytes)) = store.get(&id).await else { continue };
            let Ok(mut answer) = serde_json::from_slice::<Answer>(&bytes) else { continue };
            let Some(target) = due(schema, &answer, now_ms) else { continue };

            let payload = serde_json::json!({ "to": target, "item": id, "by": BY });
            if let Err(e) = crate::aggregates::transition(
                &state.jetstream, schema, &id, &target, payload.clone(), now_ms,
            )
            .await
            {
                // A record whose events are gone, or one another writer
                // moved between the read and the write: neither is this
                // sweep's to fix, and the next tick sees it again.
                tracing::warn!(bucket = %schema.bucket, id, target, "a deadline could not be made: {e}");
                continue;
            }

            answer.state = target.clone();
            answer.decided_ms = Some(now_ms);
            answer.decided_by = Some(BY.to_string());
            if let Ok(data) = serde_json::to_vec(&answer) {
                let _ = store.put(&id, data.into()).await;
            }
            crate::mail::on_state(state, &schema.bucket, &id, &target, &answer.responses, answer.storyline).await;
            crate::people::on_state(state, &schema.bucket, &id, &target, &answer.responses).await;

            let parent_hashes = vec![id.clone()];
            let chain_hash = crate::chain::hash_node(&answer.question_id, &parent_hashes, &payload, now_ms);
            if let Err(e) = emit_answer_submitted(
                &state.nats,
                &AnswerSubmitted {
                    chain_hash,
                    parent_hashes,
                    question_id: answer.question_id.clone(),
                    alternative: target.clone(),
                    responses: payload,
                    timestamp_ms: now_ms,
                    storyline: answer.storyline,
                },
            )
            .await
            {
                tracing::warn!(bucket = %schema.bucket, id, "the deadline was made but not published: {e}");
            }
            tracing::info!(bucket = %schema.bucket, id, target, "a deadline passed and the case went on");
        }
    }
    Ok(())
}

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

    fn schema(yaml: &str) -> crate::aggregates::AggregateSchema {
        crate::aggregates::parse_aggregates_yaml(yaml).unwrap().into_values().next().unwrap()
    }

    const HEARINGS: &str = r#"
aggregates:
  - bucket: hearings
    initial: open
    states:
      open: { event: received, after: { days: 42, to: no_answer } }
      answered: { event: answered }
      no_answer: { event: no_answer }
    transitions:
      open: [answered, no_answer]
"#;

    fn waiting(state: &str, since_ms: i64) -> Answer {
        Answer {
            id: "r".into(), question_id: "/q".into(), alternative: "a".into(),
            responses: serde_json::json!({}), state: state.into(),
            submitted_ms: since_ms, decided_ms: None, decided_by: None, notes: vec![],
            storyline: 0,
        }
    }

    #[test]
    fn a_record_leaves_its_list_when_its_time_in_the_state_is_up() {
        let s = schema(
            "aggregates:\n  - bucket: people\n    initial: open\n    states:\n      open: { event: shown }\n      gone: { event: gone, leaves_after: { minutes: 10 } }\n    transitions:\n      open: [gone]\n",
        );
        let t = 1_000_000_000_000;
        let mut answer = waiting("gone", t - 60 * 60 * 1000);
        answer.decided_ms = Some(t);
        assert!(!has_left(&s, &answer, t + 9 * 60 * 1000), "measured from entering the state, not from sending");
        assert!(has_left(&s, &answer, t + 10 * 60 * 1000));
        // A state without it keeps its records.
        assert!(!has_left(&s, &waiting("open", 0), t));
        assert!(crate::aggregates::parse_aggregates_yaml("aggregates:\n  - bucket: p\n    initial: open\n    states:\n      open: { event: o, leaves_after: {} }\n    transitions: {}\n").is_err());
    }

    #[test]
    fn a_deadline_comes_due_the_day_it_says_and_not_before() {
        let s = schema(HEARINGS);
        let sent = 1_000_000_000_000;
        let answer = waiting("open", sent);
        assert_eq!(due(&s, &answer, sent), None);
        assert_eq!(due(&s, &answer, sent + 41 * DAY_MS), None);
        assert_eq!(due(&s, &answer, sent + 42 * DAY_MS), Some("no_answer".into()));
    }

    // The six weeks run from the last move, not from the day the case
    // was opened: a record someone touched has been waiting since then.
    #[test]
    fn the_wait_is_measured_from_the_last_move() {
        let s = schema(HEARINGS);
        let sent = 1_000_000_000_000;
        let mut answer = waiting("open", sent);
        answer.decided_ms = Some(sent + 40 * DAY_MS);
        assert_eq!(due(&s, &answer, sent + 42 * DAY_MS), None);
        assert_eq!(due(&s, &answer, sent + 82 * DAY_MS), Some("no_answer".into()));
    }

    #[test]
    fn a_state_with_no_deadline_waits_forever() {
        let s = schema(HEARINGS);
        let answer = waiting("answered", 0);
        assert_eq!(due(&s, &answer, i64::MAX / 2), None);
    }

    // A clock may only make a move a person could have made, so a
    // deadline onto an undeclared edge is refused when content loads,
    // not when the hour comes round.
    #[test]
    fn a_deadline_onto_an_undeclared_edge_is_refused_on_load() {
        let bad = HEARINGS.replace("to: no_answer }", "to: answered }").replace("open: [answered, no_answer]", "open: [no_answer]");
        let err = crate::aggregates::parse_aggregates_yaml(&bad).unwrap_err().to_string();
        assert!(err.contains("not a declared transition"), "{err}");
    }

    #[test]
    fn a_deadline_of_no_days_is_refused() {
        let bad = HEARINGS.replace("days: 42", "days: 0");
        let err = crate::aggregates::parse_aggregates_yaml(&bad).unwrap_err().to_string();
        assert!(err.contains("at least one"), "{err}");
    }
}