portal/tests/jetstream.rs

//! 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,
        0,
    )
    .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 state_before_the_note = record.state.clone();
    let reply = portal::mail::Incoming {
        v: portal::mail::CONTRACT,
        id: "mail-1".into(),
        bucket: bucket.clone(),
        record: "rec-1".into(),
        from: "A <A@B.example>".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, state_before_the_note,
        "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");
}

/// A session outlives the process that made it. The `MemoryStore` this
/// replaced could not: every deploy signed out every desk, and the
/// only proof that this one does not is a store opened twice against
/// the same JetStream, as two runs of the binary would.
#[tokio::test]
async fn a_session_survives_the_process_that_made_it() {
    use tower_sessions::session::{Id, Record};
    use tower_sessions::SessionStore;

    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"));
    let store = portal::sessions::NatsStore::open(&js).await.expect("open");

    let mut record = Record {
        id: Id::default(),
        data: Default::default(),
        expiry_date: tower_sessions::cookie::time::OffsetDateTime::now_utc()
            + tower_sessions::cookie::time::Duration::hours(1),
    };
    record.data.insert("who".into(), json!("ingrid"));
    store.create(&mut record).await.expect("create");

    // A second store over the same JetStream is what the next boot of
    // the binary has: nothing carried over in memory.
    let next_boot = portal::sessions::NatsStore::open(&js)
        .await
        .expect("reopen");
    let found = next_boot
        .load(&record.id)
        .await
        .expect("load")
        .expect("still there");
    assert_eq!(found.data.get("who"), Some(&json!("ingrid")));

    // `create` never writes over somebody else's session: asked to
    // create one whose id is taken, it takes a new id instead.
    let mut collide = Record {
        id: record.id,
        data: Default::default(),
        expiry_date: record.expiry_date,
    };
    next_boot
        .create(&mut collide)
        .await
        .expect("create beside it");
    assert_ne!(
        collide.id, record.id,
        "an id already in use must not be reused"
    );
    assert_eq!(
        next_boot
            .load(&record.id)
            .await
            .expect("load")
            .expect("original intact")
            .data
            .get("who"),
        Some(&json!("ingrid")),
        "the original session must be untouched"
    );

    // An expired record is gone to anyone asking, whatever the bucket
    // still holds.
    let stale = Record {
        id: Id::default(),
        data: Default::default(),
        expiry_date: tower_sessions::cookie::time::OffsetDateTime::now_utc()
            - tower_sessions::cookie::time::Duration::minutes(1),
    };
    next_boot.save(&stale).await.expect("save stale");
    assert!(next_boot
        .load(&stale.id)
        .await
        .expect("load stale")
        .is_none());

    // Signing out leaves nothing to read back.
    next_boot.delete(&record.id).await.expect("delete");
    assert!(next_boot
        .load(&record.id)
        .await
        .expect("load after delete")
        .is_none());
    next_boot.delete(&collide.id).await.ok();
}

/// A membership is a case portal keeps itself: invited by a desk or a
/// decision, active from the person's first sign-in, removed by a desk,
/// and invited again after that. What the person belongs to is read
/// back from the same records, matched on phone number or address.
#[tokio::test]
async fn a_membership_is_invited_becomes_active_at_sign_in_and_can_be_taken_back() {
    use portal::memberships::{self, Identity, ACTIVE, INVITED, REMOVED};

    let Ok(url) = std::env::var("PORTAL_TEST_NATS_URL") else {
        eprintln!("skipped: PORTAL_TEST_NATS_URL is not set");
        return;
    };
    let nats = async_nats::connect(&url).await.expect("connect");
    let js = async_nats::jetstream::new(nats.clone());
    ensure_stream(&js).await.expect("event stream");
    // A clean bucket per run.
    let _ = js.delete_key_value(memberships::BUCKET).await;
    let aggregates = portal::content::with_builtin_aggregates(Default::default());
    let questions = std::collections::HashMap::new();
    let access = portal::access::Policy::from_content(&questions, &aggregates);
    let state = portal::server::AppState {
        leptos_options: leptos::config::LeptosOptions::builder()
            .output_name("test")
            .build(),
        nats,
        jetstream: js,
        questions: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(
            portal::content::Storylines::one(questions),
        )),
        aggregates: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(
            portal::content::Storylines::one(aggregates),
        )),
        site: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(
            portal::content::Storylines::one(Default::default()),
        )),
        access: std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(access)),
        gitea_base: String::new(),
        content_raw_base: String::new(),
        content_branch: "main".into(),
        oidc: std::sync::Arc::new(portal::server::oidc::Oidc::disabled()),
        garage: None,
    };

    // Groups of this run's own: the event log outlives the bucket, and
    // a record id seen before would replay its old history.
    let run = chrono::Utc::now().timestamp_millis();
    let styret = format!("styret{run}");
    let medlemmer = format!("medlemmer{run}");
    let (styret, medlemmer) = (styret.as_str(), medlemmer.as_str());

    // A desk invites Kari by phone number; a decision grants her a
    // second group by address. Both are invitations until she is seen.
    let by_phone = Identity {
        sub: None,
        phone: Some("+4791234567".into()),
        email: None,
    };
    let by_mail = Identity {
        sub: None,
        phone: None,
        email: Some("kari@vel.no".into()),
    };
    let first = memberships::grant(&state, styret, "Kari", &by_phone, "desk")
        .await
        .expect("grant");
    assert!(first.created && first.reset_link.is_none());
    let again = memberships::grant(&state, styret, "Kari", &by_phone, "desk")
        .await
        .expect("grant again");
    assert!(
        !again.created,
        "a second invite into the same group finds the first"
    );
    memberships::grant(&state, medlemmer, "Kari", &by_mail, "husstander:abc")
        .await
        .expect("grant by mail");
    let records = memberships::all(&state).await;
    assert_eq!(records.len(), 2);
    assert!(records.iter().all(|r| r.state == INVITED));
    assert!(
        memberships::groups_of(&records, &by_phone).is_empty(),
        "an invitation is not yet a group"
    );

    // She signs in: the provider knows her subject, phone and address.
    let kari = portal::auth::User {
        sub: "vipps-abc".into(),
        username: "kari".into(),
        display_name: "Kari".into(),
        groups: vec![],
        phone: Some("912 34 567".into()),
        email: Some("Kari@Vel.no".into()),
    };
    let mut groups = memberships::on_signed_in(&state, &kari).await;
    groups.sort();
    assert_eq!(groups, vec![medlemmer, styret]);
    let records = memberships::all(&state).await;
    assert!(records.iter().all(|r| r.state == ACTIVE));
    // Both records now know her by everything the provider verified.
    for r in &records {
        let held = memberships::identity_of(r);
        assert_eq!(held.sub.as_deref(), Some("vipps-abc"));
        assert_eq!(held.phone.as_deref(), Some("+4791234567"));
        assert_eq!(held.email.as_deref(), Some("kari@vel.no"));
    }

    // A desk takes her out of styret; the record stays, as removed.
    assert!(memberships::revoke(&state, styret, &by_phone, "ola")
        .await
        .expect("revoke"));
    assert!(
        !memberships::revoke(&state, styret, &by_phone, "ola")
            .await
            .expect("revoke twice"),
        "already out"
    );
    let records = memberships::all(&state).await;
    let styret_record = records
        .iter()
        .find(|r| memberships::group_of(r) == styret)
        .unwrap();
    assert_eq!(styret_record.state, REMOVED);
    assert_eq!(
        memberships::groups_of(
            &records,
            &Identity {
                sub: Some("vipps-abc".into()),
                phone: None,
                email: None
            }
        ),
        vec![medlemmer]
    );

    // Invited again: the same record, invited, and hers at the next sign-in.
    let back = memberships::grant(&state, styret, "Kari", &by_mail, "desk")
        .await
        .expect("re-grant");
    assert!(back.created);
    let records = memberships::all(&state).await;
    assert_eq!(
        records
            .iter()
            .find(|r| memberships::group_of(r) == styret)
            .unwrap()
            .state,
        INVITED
    );
    let groups = memberships::on_signed_in(&state, &kari).await;
    assert!(groups.contains(&styret.to_string()));
}

#[tokio::test]
async fn a_card_of_ones_own_is_replaced_not_added_to_even_on_a_double_click() {
    use portal::answers::{keep_newest_own, own_records, OWNER_FIELD};
    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");
    let bucket = format!("own_{}", chrono::Utc::now().timestamp_millis());
    let schemas = std::collections::HashMap::new();
    let now = chrono::Utc::now().timestamp_millis();
    let card = |name: &str, by: &str| json!({"name": name, OWNER_FIELD: by});

    // Kari's first card, then two sends of her second pressed at once,
    // and Ola's card beside them.
    store_answer(
        &js,
        &schemas,
        &bucket,
        "k1".into(),
        "/kontor/ansatt",
        "a",
        &card("Kari", "kari"),
        now,
        0,
    )
    .await
    .unwrap();
    store_answer(
        &js,
        &schemas,
        &bucket,
        "o1".into(),
        "/kontor/ansatt",
        "a",
        &card("Ola", "ola"),
        now + 1,
        0,
    )
    .await
    .unwrap();
    store_answer(
        &js,
        &schemas,
        &bucket,
        "k2".into(),
        "/kontor/ansatt",
        "a",
        &card("Kari N.", "kari"),
        now + 2,
        0,
    )
    .await
    .unwrap();
    store_answer(
        &js,
        &schemas,
        &bucket,
        "k3".into(),
        "/kontor/ansatt",
        "a",
        &card("Kari N.", "kari"),
        now + 2,
        0,
    )
    .await
    .unwrap();
    // Both racing sends settle after storing; both must keep the same one.
    let (a, b) = tokio::join!(
        keep_newest_own(&js, &bucket, "open", "kari"),
        keep_newest_own(&js, &bucket, "open", "kari")
    );
    let _ = (a, b);

    let kari = own_records(&js, &bucket, "kari").await;
    assert_eq!(
        kari.iter().map(|a| a.id.as_str()).collect::<Vec<_>>(),
        vec!["k3"],
        "one card, the newest"
    );
    assert_eq!(
        own_records(&js, &bucket, "ola").await.len(),
        1,
        "nobody else's card is touched"
    );
}