tests/jetstream.rs: the store, event log and projection against a real JetStream

Until now nothing outside a live site exercised the NATS side: the
aggregate engine was tested by in-memory replay only. This test
submits an answer, moves it from a desk through the engine, checks
that an undeclared move and an unknown record are refused, and reads
the record back from its own event log. It runs when
PORTAL_TEST_NATS_URL is set; CI starts a throwaway nats:2.14.6-alpine
with JetStream for it and tears it down. Without the variable the
test passes as skipped, so `cargo test` stays green offline.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L4jrCgLiKKHAEFuZUJjckH

.gitea/workflows/test.yml

@@ -22,5 +22,22 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
+      # A throwaway JetStream for tests/jetstream.rs. Best effort: if the
+      # runner cannot use docker the variable stays unset and that one
+      # test reports itself skipped rather than failing the build.
+      - name: Start a throwaway JetStream
+        run: |
+          docker rm -f portal-test-nats >/dev/null 2>&1 || true
+          if docker run -d --rm --name portal-test-nats -p 127.0.0.1:14223:4222 nats:2.14.6-alpine -js >/dev/null 2>&1; then
+            sleep 2
+            echo "PORTAL_TEST_NATS_URL=nats://127.0.0.1:14223" >> "$GITHUB_ENV"
+          else
+            echo "::warning::no docker on this runner - the JetStream integration test will be skipped"
+          fi
+
       - name: Test
         run: cargo test --features ssr
+
+      - name: Stop the throwaway JetStream
+        if: always()
+        run: docker rm -f portal-test-nats >/dev/null 2>&1 || true

Cargo.toml

@@ -106,6 +106,10 @@ ssr = [
 [dev-dependencies]
 proptest = "1"
 
+[[test]]
+name = "jetstream"
+required-features = ["ssr"]
+
 [[bin]]
 name = "question_lint"
 path = "src/bin/question_lint.rs"

tests/jetstream.rs

@@ -0,0 +1,66 @@
+//! The one test that meets a real JetStream: a submission becomes an
+//! aggregate, a desk move is accepted, an undeclared move is refused,
+//! and the record reads back from its own event log. Everything else
+//! in the suite replays in memory; this is what proves the store, the
+//! event log and the projection agree with each other.
+//!
+//! Runs only when `PORTAL_TEST_NATS_URL` points at a JetStream (CI
+//! starts a throwaway one, see `.gitea/workflows/test.yml`); without
+//! it the test passes as skipped, so `cargo test` stays green offline.
+#![cfg(feature = "ssr")]
+
+use portal::aggregates::{self, parse_aggregates_yaml, TransitionError};
+use portal::answers::{store_answer, Answer};
+use portal::events::store::{ensure_stream, load_events};
+use serde_json::json;
+
+#[tokio::test]
+async fn a_submission_is_created_moved_refused_and_read_back_through_the_real_engine() {
+    let Ok(url) = std::env::var("PORTAL_TEST_NATS_URL") else {
+        eprintln!("skipped: PORTAL_TEST_NATS_URL is not set");
+        return;
+    };
+    let js = async_nats::jetstream::new(async_nats::connect(&url).await.expect("connect"));
+    ensure_stream(&js).await.expect("event stream");
+
+    // A bucket of its own per run, so reruns and parallel runs never
+    // read each other's records.
+    let bucket = format!("it_{}", chrono::Utc::now().timestamp_millis());
+    let schemas = parse_aggregates_yaml(&format!(
+        "aggregates:\n  - bucket: {bucket}\n    initial: open\n    states:\n      open: {{ event: received }}\n      quoted: {{ event: quoted }}\n      won: {{ event: won }}\n      lost: {{ event: lost }}\n    transitions:\n      open: [quoted, lost]\n      quoted: [won, lost]\n"
+    ))
+    .expect("schema");
+    let schema = &schemas[&bucket];
+    let now = chrono::Utc::now().timestamp_millis();
+
+    // The submission: a KV record in the initial state, plus the first event.
+    store_answer(&js, &schemas, &bucket, "rec-1".into(), "/", "Hire us", &json!({"email": "a@b.example"}), now)
+        .await
+        .expect("store");
+    let store = js.get_key_value(&bucket).await.expect("bucket exists");
+    let stored: Answer = serde_json::from_slice(&store.get("rec-1").await.unwrap().expect("record")).unwrap();
+    assert_eq!(stored.state, "open");
+    assert_eq!(stored.responses["email"], "a@b.example");
+
+    // A desk move the machine allows.
+    let moved = aggregates::transition(&js, schema, "rec-1", "quoted", json!({"by": "test"}), now + 1)
+        .await
+        .expect("open -> quoted is declared");
+    assert_eq!(moved.state, "quoted");
+
+    // Moves the machine does not allow are refused, whoever asks.
+    let back = aggregates::transition(&js, schema, "rec-1", "open", json!({}), now + 2).await;
+    assert!(matches!(back, Err(TransitionError::DisallowedTransition { .. })), "{back:?}");
+    let skip = aggregates::transition(&js, schema, "rec-1", "won", json!({}), now + 2).await;
+    assert!(skip.is_ok(), "quoted -> won is declared: {skip:?}");
+    let nobody = aggregates::transition(&js, schema, "rec-9", "quoted", json!({}), now + 3).await;
+    assert!(matches!(nobody, Err(TransitionError::UnknownAggregate)), "{nobody:?}");
+
+    // The record reads back from its own event log, in the state the
+    // accepted moves left it, with a sequence that moved forward.
+    let events = load_events(&js, &bucket, "rec-1").await.expect("events");
+    assert_eq!(events.len(), 3, "received, quoted, won");
+    let replayed = aggregates::replay(schema, "rec-1", &events).expect("replays");
+    assert_eq!(replayed.state, "won");
+    assert!(replayed.sequence > moved.sequence);
+}