//! 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); // A reply to the mail this record was sent becomes a note on it, // through the same store the desks read - and moves nothing: the // record is still where the desk left it. let mut record: Answer = serde_json::from_slice(&store.get("rec-1").await.unwrap().expect("record")).unwrap(); let reply = portal::mail::Incoming { v: portal::mail::CONTRACT, id: "mail-1".into(), bucket: bucket.clone(), record: "rec-1".into(), from: "A ".into(), subject: "Re: your enquiry".into(), text: "Thursday suits us.".into(), received_ms: now + 10, }; let note = portal::mail::note_for(&reply, &record).expect("the address matches the record"); record.notes.push(note); store.put("rec-1", serde_json::to_vec(&record).unwrap().into()).await.expect("note stored"); let read_back: Answer = serde_json::from_slice(&store.get("rec-1").await.unwrap().expect("record")).unwrap(); assert_eq!(read_back.notes.len(), 1); assert_eq!(read_back.notes[0].text, "Thursday suits us."); assert_eq!(read_back.notes[0].from, "a@b.example"); assert_eq!(read_back.state, "quoted", "a reply is a note, never a decision"); // The same reply read twice adds nothing. assert!(portal::mail::note_for(&reply, &read_back).is_err()); // And the event log is untouched by it. let after = load_events(&js, &bucket, "rec-1").await.expect("events"); assert_eq!(after.len(), 3, "a note writes no event"); }