portal/src/events/store.rs

//! Durable, replayable event log for the aggregate state machines in
//! `crate::aggregates` - the source of truth `answers::Answer` KV
//! records now only *project*. Ported from dodrenett's `lib/eventStore.ts`
//! with two gaps closed rather than copied: dodrenett's `sequence`
//! field was always hardcoded `0` (ordering relied entirely on
//! JetStream's own append order anyway - here the real per-message
//! sequence is used directly, nothing fake); dodrenett never used
//! `Nats-Msg-Id`/CAS at all. This module uses both, via `async-nats`
//! 0.38's native support - no new dependency.
//!
//! Subject scheme is `events.<aggregate_type>.<aggregate_id>`,
//! deliberately *not* `events.<type>.<id>.<eventType>` like dodrenett -
//! dropping the event type from the subject means
//! `expected_last_subject_sequence` gives real optimistic concurrency
//! across an aggregate's *entire* history, not just same-typed events.
//! The cost: no cheap subject-wildcard query for "every `invited` event
//! across all applicants" - nobody needs that today, and it's a filter
//! over `load_events`'s output if it's ever needed.

use serde::{Deserialize, Serialize};

pub const STREAM_NAME: &str = "EVENTS";

/// One durably-stored fact about an aggregate. `sequence` is filled in
/// from JetStream's own per-message sequence number on read
/// (`load_events`) - never trust a value baked into the payload itself.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Event {
    pub id: uuid::Uuid,
    pub aggregate_type: String,
    pub aggregate_id: String,
    pub event_type: String,
    pub payload: serde_json::Value,
    pub occurred_at_ms: i64,
    #[serde(default)]
    pub sequence: u64,
}

fn subject(aggregate_type: &str, aggregate_id: &str) -> String {
    format!("events.{aggregate_type}.{aggregate_id}")
}

/// Idempotent - safe to call on every boot. `max_age` deliberately left
/// at the default (unlimited) - dodrenett's 7-day retention was an
/// explicit dev default, not something worth inheriting for what's
/// meant to be the durable source of truth.
pub async fn ensure_stream(
    js: &async_nats::jetstream::Context,
) -> anyhow::Result<async_nats::jetstream::stream::Stream> {
    let stream = js
        .get_or_create_stream(async_nats::jetstream::stream::Config {
            name: STREAM_NAME.to_string(),
            subjects: vec!["events.>".to_string()],
            ..Default::default()
        })
        .await?;
    Ok(stream)
}

/// Appends one event. `expected_seq` set to the aggregate's last-known
/// sequence turns this into a compare-and-swap: a concurrent caller
/// racing against the same aggregate with a stale `expected_seq` gets
/// `PublishError` back instead of silently winning a lost-update race
/// (see `crate::aggregates`, `transition_answer`'s old direct
/// KV-mutate-in-place bug this replaces). `event.id` is set as the
/// JetStream `Nats-Msg-Id` for the stream's dedup window - only
/// catches a genuine retry with the *same* id, not a fresh submission;
/// callers that want real retry-dedup must pass a stable id themselves.
/// Returns the raw `PublishError` (not wrapped in `anyhow`) so callers
/// like `crate::aggregates::classify_publish_err` can distinguish a CAS
/// conflict (`WrongLastSequence`) from every other failure mode.
pub async fn append_event(
    js: &async_nats::jetstream::Context,
    aggregate_type: &str,
    aggregate_id: &str,
    event_type: &str,
    payload: serde_json::Value,
    occurred_at_ms: i64,
    expected_seq: Option<u64>,
) -> Result<async_nats::jetstream::publish::PublishAck, async_nats::jetstream::context::PublishError>
{
    let event = Event {
        id: uuid::Uuid::new_v4(),
        aggregate_type: aggregate_type.to_string(),
        aggregate_id: aggregate_id.to_string(),
        event_type: event_type.to_string(),
        payload,
        occurred_at_ms,
        sequence: 0, // ignored on write - the server assigns the real one
    };
    // Every field here is a plain, always-serializable type (strings,
    // an i64, a Uuid, a serde_json::Value) - matches `chain.rs::hash_node`'s
    // own "json values always serialize" precedent.
    let data = serde_json::to_vec(&event).expect("Event always serializes");

    let mut publish = async_nats::jetstream::context::Publish::build()
        .payload(data.into())
        .message_id(event.id.to_string());
    if let Some(seq) = expected_seq {
        publish = publish.expected_last_subject_sequence(seq);
    }

    js.send_publish(subject(aggregate_type, aggregate_id), publish)
        .await?
        .await
}

/// Loads every event for one aggregate, in real JetStream append order.
/// Uses an ephemeral pull consumer scoped to this one aggregate's
/// subject, fetched once and torn down - same idiom dodrenett's
/// `getEvents` used, appropriate at this scale (single-digit events per
/// aggregate), not a durable/long-lived consumer.
pub async fn load_events(
    js: &async_nats::jetstream::Context,
    aggregate_type: &str,
    aggregate_id: &str,
) -> anyhow::Result<Vec<Event>> {
    use futures::TryStreamExt;

    let stream = js.get_stream(STREAM_NAME).await?;
    let subj = subject(aggregate_type, aggregate_id);
    let consumer = stream
        .create_consumer(async_nats::jetstream::consumer::pull::Config {
            filter_subject: subj,
            ..Default::default()
        })
        .await?;

    let mut events = Vec::new();
    let mut messages = consumer.fetch().max_messages(10_000).messages().await?;
    while let Some(message) = messages
        .try_next()
        .await
        .map_err(|e| anyhow::anyhow!("fetching messages: {e}"))?
    {
        let info = message.info().map_err(|e| anyhow::anyhow!("reading message info: {e}"))?;
        let mut event: Event = serde_json::from_slice(&message.payload)?;
        event.sequence = info.stream_sequence;
        events.push(event);
        message.ack().await.map_err(|e| anyhow::anyhow!("acking: {e}"))?;
    }
    events.sort_by_key(|e| e.sequence);

    // Best-effort teardown - an ephemeral consumer left behind expires
    // on its own (default inactive_threshold), so a delete failure here
    // isn't worth failing the read over.
    let _ = stream.delete_consumer(&consumer.cached_info().name).await;

    Ok(events)
}