portal/src/announce.rs

//! Announced pages (`Question.event`): keeps one `portal_events` record
//! per event question and, once a window closes, moves it to
//! `awaiting_summary` - the "post what happened" task - publishing the
//! transition on `portal.answers.submitted` like any desk decision, so
//! the desktop notifier and n8n hear it. Runs on a one-minute tick;
//! every step is idempotent, so a restart or a content reload in the
//! middle changes nothing.

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

/// Alternative name stamped on the auto-created record and on the
/// transition event - what an n8n workflow gates on.
pub const EVENT_ALTERNATIVE: &str = "Announced";
pub const SUMMARY_DUE_LABEL: &str = "Summary due";

/// Record id for an event question: its id with `/` folded to `-`
/// (`/events/opening` -> `events-opening`), so one page is one record
/// however many times the sweeper runs.
pub fn record_id(question_id: &str) -> String {
    question_id.trim_matches('/').replace('/', "-")
}

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

/// One pass: create missing records, close ended windows.
pub async fn sweep(state: &AppState) -> anyhow::Result<()> {
    let now_ms = chrono::Utc::now().timestamp_millis();
    let questions = state.questions.load();
    let events: Vec<_> = questions
        .values()
        .filter(|q| q.event.is_some())
        .cloned()
        .collect();
    if events.is_empty() {
        return Ok(());
    }
    let aggregates = state.aggregates.load();
    let schema = aggregates
        .get(EVENTS_BUCKET)
        .ok_or_else(|| anyhow::anyhow!("no {EVENTS_BUCKET} schema"))?;
    let store = match state.jetstream.get_key_value(EVENTS_BUCKET).await {
        Ok(store) => Some(store),
        Err(_) => None,
    };

    for q in events {
        let event = q.event.as_ref().expect("filtered");
        let (starts, ends) = match event.window_ms() {
            Ok(w) => w,
            Err(_) => continue, // validated on load; belt and braces
        };
        let id = record_id(&q.id);
        let existing: Option<Answer> = match &store {
            Some(store) => store
                .get(&id)
                .await?
                .and_then(|bytes| serde_json::from_slice(&bytes).ok()),
            None => None,
        };
        let responses = serde_json::json!({
            "name": q.name,
            "starts": event.starts,
            "starts_ms": starts,
            "ends_ms": ends,
            "place": event.place,
            "page": q.id,
        });

        let answer = match existing {
            None => {
                store_answer(
                    &state.jetstream,
                    &aggregates,
                    EVENTS_BUCKET,
                    id.clone(),
                    &q.id,
                    EVENT_ALTERNATIVE,
                    &responses,
                    now_ms,
                    0,
                )
                .await?;
                tracing::info!(question = %q.id, "event record created");
                continue;
            }
            Some(a) => a,
        };

        // Content is the source of truth while the window is still
        // open: a corrected date or place flows into the record, so the
        // followup fires on the schedule the page actually announces.
        if answer.state == schema.initial && answer.responses != responses {
            let mut refreshed = answer.clone();
            refreshed.responses = responses.clone();
            if let Some(store) = &store {
                store
                    .put(&id, serde_json::to_vec(&refreshed)?.into())
                    .await?;
                tracing::info!(question = %q.id, "event record refreshed from content");
            }
        }

        if answer.state == schema.initial && now_ms >= ends {
            let target = "awaiting_summary";
            let payload = serde_json::json!({ "to": target, "item": id, "by": "portal" });
            match crate::aggregates::transition(
                &state.jetstream,
                schema,
                &id,
                target,
                payload.clone(),
                now_ms,
            )
            .await
            {
                Ok(_) => {}
                Err(crate::aggregates::TransitionError::UnknownAggregate) => {
                    crate::aggregates::reseed(&state.jetstream, schema, &id, &answer.state, now_ms)
                        .await
                        .map_err(|e| anyhow::anyhow!("{e}"))?;
                    crate::aggregates::transition(
                        &state.jetstream,
                        schema,
                        &id,
                        target,
                        payload.clone(),
                        now_ms,
                    )
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))?;
                }
                Err(e) => return Err(anyhow::anyhow!("{e}")),
            }
            let mut answer = answer;
            answer.state = target.to_string();
            answer.decided_ms = Some(now_ms);
            answer.decided_by = Some("portal".to_string());
            let store = state.jetstream.get_key_value(EVENTS_BUCKET).await?;
            store.put(&id, serde_json::to_vec(&answer)?.into()).await?;

            let parent_hashes = vec![id.clone()];
            let chain_hash = crate::chain::hash_node(&q.id, &parent_hashes, &payload, now_ms);
            emit_answer_submitted(
                &state.nats,
                &AnswerSubmitted {
                    chain_hash,
                    parent_hashes,
                    question_id: q.id.clone(),
                    alternative: SUMMARY_DUE_LABEL.to_string(),
                    responses: payload,
                    timestamp_ms: now_ms,
                    storyline: 0,
                },
            )
            .await?;
            tracing::info!(question = %q.id, "event ended - summary due");
        }
    }
    Ok(())
}

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

    #[test]
    fn record_id_is_stable_and_flat() {
        assert_eq!(record_id("/events/opening"), "events-opening");
        assert_eq!(record_id("/"), "");
        assert_eq!(record_id("/opening"), "opening");
    }
}