From ebf4bf91b3d55a24ea0697ce139a4150bbc1984d Mon Sep 17 00:00:00 2001 From: Bendik Aagaard Lynghaug Date: Thu, 6 Aug 2026 08:52:29 +0200 Subject: [PATCH] Event-sourced applicant/subscriber/project aggregates, generalized resources Replaces the free-string, direct-KV-mutate state model in answers.rs with a proper event log (events/store.rs, JetStream-backed, CAS via expected_last_subject_sequence) and three pure state machines (aggregates/{applicant,subscriber,project}.rs). Closes a real lost-update race in the old transition_answer (concurrent decisions on the same item could both win, publishing contradictory events). KV buckets become best-effort read-model projections, not the source of truth. Content-declared transition targets are now validated at load/reload time against the real compiled transition tables, not accepted as arbitrary strings. Buckets renamed to describe their content, not their relation to the app (portal_applicants -> applicants, etc); "inquiry" folded into a richer "project" concept. ResourceSpec generalized beyond a single KV bucket: Kv | GiteaStarred | GiteaOrgRepos | Url sources, with an optional jq filter (via the jaq crate) to shape live data for the frontend. Url source is SSRF-guarded (https-only, rejects loopback/private/link-local, real DNS resolve). New headless question_lint binary (validates content against compiled transition tables with no NATS/OIDC/server involved) and a one-time backfill_events binary (dry-run by default) for migrating existing KV data onto the new event log. Questions get an optional `responsible` contact plus a lightweight "report this question" action. Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 326 ++++++++++++++++++++++++++++++- Cargo.toml | 23 ++- src/aggregates/applicant.rs | 40 ++++ src/aggregates/mod.rs | 294 ++++++++++++++++++++++++++++ src/aggregates/project.rs | 43 ++++ src/aggregates/subscriber.rs | 38 ++++ src/answers.rs | 177 +++++++++++++++-- src/app.rs | 93 ++++++++- src/bin/backfill_events.rs | 148 ++++++++++++++ src/bin/question_lint.rs | 76 +++++++ src/content.rs | 154 ++++++++++++++- src/{events.rs => events/mod.rs} | 2 + src/events/store.rs | 150 ++++++++++++++ src/lib.rs | 3 + src/main.rs | 2 + src/resource.rs | 309 +++++++++++++++++++++++++---- 16 files changed, 1811 insertions(+), 67 deletions(-) create mode 100644 src/aggregates/applicant.rs create mode 100644 src/aggregates/mod.rs create mode 100644 src/aggregates/project.rs create mode 100644 src/aggregates/subscriber.rs create mode 100644 src/bin/backfill_events.rs create mode 100644 src/bin/question_lint.rs rename src/{events.rs => events/mod.rs} (98%) create mode 100644 src/events/store.rs diff --git a/Cargo.lock b/Cargo.lock index 486c64a..c483cea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -616,6 +616,27 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bit-set" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3" +dependencies = [ + "bit-vec", +] + +[[package]] +name = "bit-vec" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" + +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.13.1" @@ -649,6 +670,17 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "bstr" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f7dc094d718f2e1c1559ad110e27eeaae14a5465d3d56dd6dbd793079fbd530" +dependencies = [ + "memchr", + "regex-automata", + "serde_core", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -1037,6 +1069,37 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" +[[package]] +name = "defmt" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1" +dependencies = [ + "bitflags 1.3.2", + "defmt-macros", +] + +[[package]] +name = "defmt-macros" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8" +dependencies = [ + "defmt-parser", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "defmt-parser" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" +dependencies = [ + "thiserror 2.0.19", +] + [[package]] name = "der" version = "0.7.10" @@ -1288,6 +1351,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "foldhash" version = "0.2.0" @@ -1550,7 +1619,7 @@ checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" dependencies = [ "allocator-api2", "equivalent", - "foldhash", + "foldhash 0.2.0", ] [[package]] @@ -1565,6 +1634,12 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hifijson" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "242402749acf71e6f32f5857598b7002c4058a4e3c3b22b4c7d51cab9aea754e" + [[package]] name = "hkdf" version = "0.12.4" @@ -1997,6 +2072,106 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jaq-core" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7561783b20275a6c9cb576e39208b0c635f34ef14357f1f05a2927a774f3adec" +dependencies = [ + "dyn-clone", + "once_cell", + "typed-arena", +] + +[[package]] +name = "jaq-json" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48d801b0b57f10064c4e9f5a4f6c97d0ccf62649b179ff8ac23cd494a3120ee9" +dependencies = [ + "bstr", + "bytes", + "foldhash 0.1.5", + "hifijson", + "indexmap 2.14.0", + "jaq-core", + "jaq-std", + "num-bigint", + "num-traits", + "ryu", + "self_cell", +] + +[[package]] +name = "jaq-std" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bdc5a74b0feeb5e6a1dc2dd08c34280a61e37668d10a6a3b27ad69d0fb9ce2e" +dependencies = [ + "aho-corasick", + "base64 0.22.1", + "bstr", + "jaq-core", + "jiff", + "libm", + "log", + "regex-bites", + "urlencoding", +] + +[[package]] +name = "jiff" +version = "0.2.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "668b7183bd07af9a4885f5c35b0cc5c83c4607a913c16b7e17291832910d2dcc" +dependencies = [ + "defmt", + "jiff-core", + "jiff-static", + "jiff-tzdb-platform", + "log", + "portable-atomic", + "portable-atomic-util", + "serde_core", + "windows-link", +] + +[[package]] +name = "jiff-core" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7feca88439efe53da3754500c1851dedf3cb36c524dd5cf8225cc0794de95d09" +dependencies = [ + "defmt", +] + +[[package]] +name = "jiff-static" +version = "0.2.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a69dcb3a21cfb32ce1cd056169337ca284af0766dd766e7878819b251a49204" +dependencies = [ + "jiff-core", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "jiff-tzdb" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "142bd39932ad231f10513df9ab62661fead8719872150b7ad02a2df79f4e141e" + +[[package]] +name = "jiff-tzdb-platform" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8" +dependencies = [ + "jiff-tzdb", +] + [[package]] name = "jobserver" version = "0.1.35" @@ -2276,6 +2451,12 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + [[package]] name = "litemap" version = "0.8.2" @@ -2450,6 +2631,16 @@ dependencies = [ "rand 0.8.7", ] +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -2746,6 +2937,15 @@ version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d20d5497ef88037a52ff98267d066e7f11fcc5e99bbfbd58a42336193aacec3" +[[package]] +name = "portable-atomic-util" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618" +dependencies = [ + "portable-atomic", +] + [[package]] name = "portal" version = "0.1.0" @@ -2760,12 +2960,16 @@ dependencies = [ "console_error_panic_hook", "dotenvy", "futures", + "jaq-core", + "jaq-json", + "jaq-std", "js-sys", "leptos", "leptos_axum", "leptos_meta", "leptos_router", "openidconnect", + "proptest", "serde", "serde_json", "serde_yaml", @@ -2881,6 +3085,31 @@ dependencies = [ "yansi", ] +[[package]] +name = "proptest" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744" +dependencies = [ + "bit-set", + "bit-vec", + "bitflags 2.13.1", + "num-traits", + "rand 0.9.5", + "rand_chacha 0.9.0", + "rand_xorshift", + "regex-syntax", + "rusty-fork", + "tempfile", + "unarray", +] + +[[package]] +name = "quick-error" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" + [[package]] name = "quinn" version = "0.11.11" @@ -3065,6 +3294,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_xorshift" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a" +dependencies = [ + "rand_core 0.9.5", +] + [[package]] name = "reactive_graph" version = "0.2.14" @@ -3125,7 +3363,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags", + "bitflags 2.13.1", ] [[package]] @@ -3171,6 +3409,12 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "regex-bites" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6a15a2fa0bfda9361941c45550896ae87b15cc6c8c939ea350079670332e211" + [[package]] name = "regex-lite" version = "0.1.9" @@ -3295,6 +3539,19 @@ dependencies = [ "semver", ] +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags 2.13.1", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + [[package]] name = "rustls" version = "0.21.12" @@ -3405,6 +3662,18 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "rusty-fork" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc6bf79ff24e648f6da1f8d1f011e9cac26491b619e6b9280f2b47f1774e6ee2" +dependencies = [ + "fnv", + "quick-error", + "tempfile", + "wait-timeout", +] + [[package]] name = "ryu" version = "1.0.23" @@ -3489,7 +3758,7 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" dependencies = [ - "bitflags", + "bitflags 2.13.1", "core-foundation 0.9.4", "core-foundation-sys", "libc", @@ -3502,7 +3771,7 @@ version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags", + "bitflags 2.13.1", "core-foundation 0.10.1", "core-foundation-sys", "libc", @@ -3519,6 +3788,12 @@ dependencies = [ "libc", ] +[[package]] +name = "self_cell" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ab42ca02749e120097e328d91d415325bdf43b1c72c4c8badf37375fe40a813" + [[package]] name = "semver" version = "1.0.28" @@ -4027,6 +4302,19 @@ dependencies = [ "web-sys", ] +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.4.3", + "once_cell", + "rustix", + "windows-sys 0.61.2", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -4303,7 +4591,7 @@ version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ - "bitflags", + "bitflags 2.13.1", "bytes", "futures-core", "futures-util", @@ -4481,6 +4769,12 @@ dependencies = [ "thiserror 2.0.19", ] +[[package]] +name = "typed-arena" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6af6ae20167a9ece4bcb41af5b80f8a1f1df981f6391189ce00fd257af04126a" + [[package]] name = "typed-builder" version = "0.23.2" @@ -4507,6 +4801,12 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "unarray" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" + [[package]] name = "unicase" version = "2.9.0" @@ -4556,6 +4856,12 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "urlencoding" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -4570,6 +4876,7 @@ checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" dependencies = [ "getrandom 0.4.3", "js-sys", + "serde_core", "wasm-bindgen", ] @@ -4591,6 +4898,15 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" +[[package]] +name = "wait-timeout" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ac3b126d3914f9849036f826e054cbabdc8519970b8998ddaf3b5bd3c65f11" +dependencies = [ + "libc", +] + [[package]] name = "walkdir" version = "2.5.0" diff --git a/Cargo.toml b/Cargo.toml index 9838fa6..e146c6a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,12 +31,17 @@ openidconnect = { version = "4", optional = true } # token's raw JWT payload - see server/oidc.rs::extract_groups_claim. base64 = { version = "0.22", optional = true } chrono = { version = "0.4", features = ["serde"], optional = true } -uuid = { version = "1", features = ["v4"], optional = true } +uuid = { version = "1", features = ["v4", "serde"], optional = true } dotenvy = { version = "0.15", optional = true } anyhow = { version = "1", optional = true } futures = { version = "0.3", optional = true } tracing = { version = "0.1", optional = true } tracing-subscriber = { version = "0.3", features = ["env-filter"], optional = true } +# jq-filter evaluation for ResourceSource's optional `jq` field +# (src/resource.rs) - pure Rust, no shell-out to a `jq` binary. +jaq-core = { version = "3", optional = true } +jaq-std = { version = "3", optional = true } +jaq-json = { version = "2", features = ["sync"], optional = true } # --- browser only --- wasm-bindgen = { version = "0.2", optional = true } @@ -90,8 +95,24 @@ ssr = [ "dep:futures", "dep:tracing", "dep:tracing-subscriber", + "dep:jaq-core", + "dep:jaq-std", + "dep:jaq-json", ] +[dev-dependencies] +proptest = "1" + +[[bin]] +name = "question_lint" +path = "src/bin/question_lint.rs" +required-features = ["ssr"] + +[[bin]] +name = "backfill_events" +path = "src/bin/backfill_events.rs" +required-features = ["ssr"] + [profile.wasm-release] inherits = "release" opt-level = 'z' diff --git a/src/aggregates/applicant.rs b/src/aggregates/applicant.rs new file mode 100644 index 0000000..cadc4cf --- /dev/null +++ b/src/aggregates/applicant.rs @@ -0,0 +1,40 @@ +//! Collaborator applicants (`/review`'s "Applicants" alternative, +//! `record_as: applicants`). `Open -> Invited | Declined`, straight +//! from the alternatives already declared in `questions/review.yaml`. +use super::AggregateKind; + +#[derive(Copy, Clone, PartialEq, Eq, Debug)] +pub enum State { + Open, + Invited, + Declined, +} + +impl AggregateKind for State { + const AGGREGATE_TYPE: &'static str = "applicant"; + const INITIAL_STATE: Self = State::Open; + + fn from_event_type(event_type: &str) -> Option { + match event_type { + "applied" => Some(State::Open), + "invited" => Some(State::Invited), + "declined" => Some(State::Declined), + _ => None, + } + } + + fn event_type(self) -> &'static str { + match self { + State::Open => "applied", + State::Invited => "invited", + State::Declined => "declined", + } + } + + fn allowed(self) -> &'static [Self] { + match self { + State::Open => &[State::Invited, State::Declined], + State::Invited | State::Declined => &[], + } + } +} diff --git a/src/aggregates/mod.rs b/src/aggregates/mod.rs new file mode 100644 index 0000000..3bb94c0 --- /dev/null +++ b/src/aggregates/mod.rs @@ -0,0 +1,294 @@ +//! Generic pure-replay + CAS-write machinery shared by every aggregate +//! type (`applicant`, `subscriber`, `project`) - one small `AggregateKind` +//! impl per type supplies its own states/transition table/event-type +//! names, this module supplies the mechanics once. Mirrors dodrenett's +//! `lib/machines/*.ts` shape (a `Record` table + a pure +//! `replay()`), but generalized across types instead of copy-pasted per +//! type, and with real optimistic concurrency on write (dodrenett had +//! none - see `crate::events::store`). +#![cfg(feature = "ssr")] + +pub mod applicant; +pub mod project; +pub mod subscriber; + +use crate::events::store::{append_event, load_events, Event}; + +/// A state machine's shape: its states, its transition table, and how +/// states map to/from the event-log's `event_type` strings. Kept as a +/// trait (not a generic enum param alone) so each aggregate module can +/// own its own `State` enum and transition table as plain, readable +/// Rust, not a shared data structure everyone has to thread through. +pub trait AggregateKind: Copy + Clone + PartialEq + Eq + std::fmt::Debug + 'static { + /// The bucket-name -> aggregate-type key used both for `events.rs`'s + /// subject scheme and for `content.rs`'s load-time transition + /// validation (see `content::aggregate_type_for_bucket`). + const AGGREGATE_TYPE: &'static str; + /// The state a brand-new aggregate starts in - also the state a + /// creation event (see `create`) asserts. + const INITIAL_STATE: Self; + + /// Maps an event-log `event_type` string to the state it represents + /// - `None` for anything this aggregate type doesn't recognize + /// (replay skips it, never errors). + fn from_event_type(event_type: &str) -> Option; + /// The event-log `event_type` string that represents transitioning + /// *into* this state. + fn event_type(self) -> &'static str; + /// States reachable directly from this one. Checked at write time + /// (`transition`); replay is deliberately more lenient (see below). + fn allowed(self) -> &'static [Self]; +} + +/// One aggregate's current view, replayed from its event history. +#[derive(Clone, Debug)] +pub struct Aggregate { + pub id: String, + pub state: K, + pub payload: serde_json::Value, + pub sequence: u64, +} + +/// Pure, no IO, never panics. An event that doesn't fit the transition +/// table from wherever replay has gotten to so far is silently skipped +/// - not an error, not even the *first* recognized event has anything +/// to be "from" yet, so it's always accepted. This lenience (not +/// enforcing the transition table during replay, only at write time - +/// see `transition` below) is deliberate and matches dodrenett's proven +/// "self-healing against malformed history" policy: a corrupted or +/// out-of-order event never crashes a read, it's just ignored. +pub fn replay(id: &str, events: &[Event]) -> Option> { + let mut current: Option> = None; + for event in events { + let Some(next) = K::from_event_type(&event.event_type) else { + continue; + }; + let accept = match ¤t { + None => true, + Some(agg) => next == agg.state || agg.state.allowed().contains(&next), + }; + if !accept { + continue; + } + current = Some(Aggregate { + id: id.to_string(), + state: next, + payload: event.payload.clone(), + sequence: event.sequence, + }); + } + current +} + +#[derive(Debug, Clone)] +pub enum TransitionError { + /// No creation event found for this id - nothing to transition. + UnknownAggregate, + DisallowedTransition { from: String, to: String }, + /// A concurrent writer already advanced this aggregate past the + /// sequence this call read - the fix for the lost-update race + /// `answers::transition_answer` used to have (see + /// `crate::events::store::append_event`'s CAS). + Conflict, + Store(String), +} + +impl std::fmt::Display for TransitionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::UnknownAggregate => write!(f, "unknown item"), + Self::DisallowedTransition { from, to } => { + write!(f, "cannot go from {from} to {to}") + } + Self::Conflict => write!(f, "item was just changed by someone else, try again"), + Self::Store(e) => write!(f, "storage error: {e}"), + } + } +} + +fn classify_publish_err(e: async_nats::jetstream::context::PublishError) -> TransitionError { + use async_nats::jetstream::context::PublishErrorKind; + match e.kind() { + PublishErrorKind::WrongLastSequence | PublishErrorKind::WrongLastMessageId => { + TransitionError::Conflict + } + _ => TransitionError::Store(e.to_string()), + } +} + +/// Appends the creation event for a brand-new aggregate. `expected_seq: +/// Some(0)` (no message yet on this id's subject) doubles as a +/// create-only-once guard - a double-submit race on the same id fails +/// the same way a disallowed transition does, rather than silently +/// overwriting. +pub async fn create( + js: &async_nats::jetstream::Context, + id: &str, + payload: serde_json::Value, + occurred_at_ms: i64, +) -> Result, TransitionError> { + let ack = append_event( + js, + K::AGGREGATE_TYPE, + id, + K::INITIAL_STATE.event_type(), + payload.clone(), + occurred_at_ms, + Some(0), + ) + .await + .map_err(classify_publish_err)?; + Ok(Aggregate { + id: id.to_string(), + state: K::INITIAL_STATE, + payload, + sequence: ack.sequence, + }) +} + +/// Loads and replays `id`'s current state, then - if `target` is +/// actually reachable from it - appends the corresponding event with +/// the aggregate's just-replayed sequence as the CAS guard. +pub async fn transition( + js: &async_nats::jetstream::Context, + id: &str, + target: K, + payload: serde_json::Value, + occurred_at_ms: i64, +) -> Result, TransitionError> { + let events = load_events(js, K::AGGREGATE_TYPE, id) + .await + .map_err(|e| TransitionError::Store(e.to_string()))?; + let current = replay::(id, &events).ok_or(TransitionError::UnknownAggregate)?; + if !current.state.allowed().contains(&target) { + return Err(TransitionError::DisallowedTransition { + from: format!("{:?}", current.state), + to: format!("{target:?}"), + }); + } + let ack = append_event( + js, + K::AGGREGATE_TYPE, + id, + target.event_type(), + payload.clone(), + occurred_at_ms, + Some(current.sequence), + ) + .await + .map_err(classify_publish_err)?; + Ok(Aggregate { + id: id.to_string(), + state: target, + payload, + sequence: ack.sequence, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::events::store::Event; + + #[derive(Copy, Clone, PartialEq, Eq, Debug)] + enum TestState { + Open, + A, + B, + } + + impl AggregateKind for TestState { + const AGGREGATE_TYPE: &'static str = "test"; + const INITIAL_STATE: Self = TestState::Open; + + fn from_event_type(event_type: &str) -> Option { + match event_type { + "opened" => Some(TestState::Open), + "a" => Some(TestState::A), + "b" => Some(TestState::B), + _ => None, + } + } + fn event_type(self) -> &'static str { + match self { + TestState::Open => "opened", + TestState::A => "a", + TestState::B => "b", + } + } + fn allowed(self) -> &'static [Self] { + match self { + TestState::Open => &[TestState::A, TestState::B], + TestState::A | TestState::B => &[], + } + } + } + + fn event(seq: u64, event_type: &str) -> Event { + Event { + id: uuid::Uuid::new_v4(), + aggregate_type: "test".to_string(), + aggregate_id: "x".to_string(), + event_type: event_type.to_string(), + payload: serde_json::json!({}), + occurred_at_ms: 0, + sequence: seq, + } + } + + #[test] + fn empty_history_is_no_aggregate() { + assert!(replay::("x", &[]).is_none()); + } + + #[test] + fn valid_path_replays() { + let events = vec![event(1, "opened"), event(2, "a")]; + let agg = replay::("x", &events).unwrap(); + assert_eq!(agg.state, TestState::A); + assert_eq!(agg.sequence, 2); + } + + #[test] + fn disallowed_edge_is_skipped_not_fatal() { + // "a" then "b" - B isn't reachable from A, so the second event + // is silently ignored, not an error. + let events = vec![event(1, "opened"), event(2, "a"), event(3, "b")]; + let agg = replay::("x", &events).unwrap(); + assert_eq!(agg.state, TestState::A); + } + + #[test] + fn unrecognized_event_type_is_skipped() { + let events = vec![event(1, "opened"), event(2, "bogus"), event(3, "a")]; + let agg = replay::("x", &events).unwrap(); + assert_eq!(agg.state, TestState::A); + } + + proptest::proptest! { + #[test] + fn replay_never_panics_and_state_always_valid( + types in proptest::collection::vec( + proptest::sample::select(vec!["opened", "a", "b", "bogus"]), + 0..30, + ) + ) { + let events: Vec = types + .iter() + .enumerate() + .map(|(i, t)| event(i as u64, t)) + .collect(); + let result = replay::("x", &events); + if let Some(agg) = result { + // Whatever state we landed in must itself be a real + // state this machine declares - trivially true here + // since TestState::allowed() only ever returns TestState + // values, but this is the shape the real aggregates' + // proptests repeat: replaying arbitrary event-type noise + // must never produce anything but a legitimate state. + let _: TestState = agg.state; + assert!(agg.sequence <= events.len() as u64); + } + } + } +} diff --git a/src/aggregates/project.rs b/src/aggregates/project.rs new file mode 100644 index 0000000..6262882 --- /dev/null +++ b/src/aggregates/project.rs @@ -0,0 +1,43 @@ +//! Project submissions (`record_as: projects` - absorbs what used to be +//! called "inquiries"). A visitor submits a project idea plus who it's +//! for (org/contact details, captured as plain payload fields - no +//! separate Organization aggregate yet, see the event-sourcing plan). +//! `Open -> Accepted | Declined` - a real b2b-pipeline pair of terminal +//! states rather than the old generic `Handled`. +use super::AggregateKind; + +#[derive(Copy, Clone, PartialEq, Eq, Debug)] +pub enum State { + Open, + Accepted, + Declined, +} + +impl AggregateKind for State { + const AGGREGATE_TYPE: &'static str = "project"; + const INITIAL_STATE: Self = State::Open; + + fn from_event_type(event_type: &str) -> Option { + match event_type { + "submitted" => Some(State::Open), + "accepted" => Some(State::Accepted), + "declined" => Some(State::Declined), + _ => None, + } + } + + fn event_type(self) -> &'static str { + match self { + State::Open => "submitted", + State::Accepted => "accepted", + State::Declined => "declined", + } + } + + fn allowed(self) -> &'static [Self] { + match self { + State::Open => &[State::Accepted, State::Declined], + State::Accepted | State::Declined => &[], + } + } +} diff --git a/src/aggregates/subscriber.rs b/src/aggregates/subscriber.rs new file mode 100644 index 0000000..83a689a --- /dev/null +++ b/src/aggregates/subscriber.rs @@ -0,0 +1,38 @@ +//! Newsletter subscribers (`record_as: subscribers`). `Open -> +//! Unsubscribed`, fired via `self_transition` (see `answers.rs`) rather +//! than an owner-gated `transition`, so a subscriber never needs a +//! signed-in session to leave. +use super::AggregateKind; + +#[derive(Copy, Clone, PartialEq, Eq, Debug)] +pub enum State { + Open, + Unsubscribed, +} + +impl AggregateKind for State { + const AGGREGATE_TYPE: &'static str = "subscriber"; + const INITIAL_STATE: Self = State::Open; + + fn from_event_type(event_type: &str) -> Option { + match event_type { + "subscribed" => Some(State::Open), + "unsubscribed" => Some(State::Unsubscribed), + _ => None, + } + } + + fn event_type(self) -> &'static str { + match self { + State::Open => "subscribed", + State::Unsubscribed => "unsubscribed", + } + } + + fn allowed(self) -> &'static [Self] { + match self { + State::Open => &[State::Unsubscribed], + State::Unsubscribed => &[], + } + } +} diff --git a/src/answers.rs b/src/answers.rs index 06350fa..af3fe62 100644 --- a/src/answers.rs +++ b/src/answers.rs @@ -1,15 +1,15 @@ -//! Durable, queryable storage for submitted answers - the KV-backed -//! counterpart to the ephemeral `events::AnswerSubmitted` every -//! submission already publishes. The runtime here has no notion of -//! "applicant" or any other specific workflow: an `Alternative.record_as` -//! names a NATS KV bucket (nothing more); on submit, the response gets -//! wrapped in the same envelope (`Answer`) regardless of what content -//! produced it. `resource::get_resource` reads it back generically, and -//! `transition_answer` below moves an answer between states named in -//! content (`ResourceSpec.transitions`), never states hardcoded here. -//! Whatever workflow this ends up powering - applicant review, order -//! triage, comment moderation - is entirely a property of some -//! `portal-content` YAML, not of this file. +//! Read-model storage for submitted answers - the KV-backed projection +//! of the durable event log now underneath it (`events::store`, +//! `aggregates`) for the three buckets migrated onto that machinery +//! this pass (`content::aggregate_type_for_bucket`). An +//! `Alternative.record_as`/`ResourceSpec`'s bucket is still just a +//! bucket name as far as `resource::get_resource` is concerned - this +//! file is where that name additionally gets checked against a real, +//! Rust-defined transition table, for the buckets that have one. +//! A bucket outside that three-entry map still works exactly as +//! before (direct KV mutate-in-place, no event log, no CAS) - this is +//! deliberately not a hard cutover, so content isn't forced to migrate +//! before it needs to. use leptos::prelude::*; use serde::{Deserialize, Serialize}; @@ -71,7 +71,30 @@ pub async fn store_answer( decided_by: None, }; let data = serde_json::to_vec(&answer)?; - store.put(id, data.into()).await?; + store.put(id.clone(), data.into()).await?; + + // Also seed the event log for the three aggregate types migrated + // onto it this pass - best-effort, same "a hiccup here shouldn't + // fail a submission the NATS notification event has already + // recorded" policy as the KV write above. + if let Some(agg_type) = crate::content::aggregate_type_for_bucket(bucket) { + use crate::aggregates::{applicant, create, project, subscriber}; + let seed = match agg_type { + "applicant" => create::(js, &id, responses.clone(), submitted_ms) + .await + .map(|_| ()), + "subscriber" => create::(js, &id, responses.clone(), submitted_ms) + .await + .map(|_| ()), + "project" => create::(js, &id, responses.clone(), submitted_ms) + .await + .map(|_| ()), + _ => Ok(()), + }; + if let Err(e) = seed { + tracing::error!(bucket, id, error = %e, "failed to seed event log for submission"); + } + } Ok(()) } @@ -138,9 +161,12 @@ pub async fn transition_answer( return Err(ServerFnError::new("not authorized")); } + let bucket = resource + .bucket() + .ok_or_else(|| ServerFnError::new("resource is not kv-backed, has no transitions"))?; let store = state .jetstream - .get_key_value(&resource.bucket) + .get_key_value(bucket) .await .map_err(|e| ServerFnError::new(format!("resource bucket unavailable: {e}")))?; let bytes = store @@ -155,6 +181,22 @@ pub async fn transition_answer( } let decided_ms = chrono::Utc::now().timestamp_millis(); + let decision_payload_for_event = + serde_json::json!({ "to": to, "item": item_id, "by": user.username }); + + // For the three buckets migrated onto the event log this pass, the + // real fix for the lost-update race two concurrent decisions on the + // same item used to hit: append with CAS on the aggregate's + // just-replayed sequence, so a second racing caller's write is + // rejected instead of silently overwriting the first. A bucket + // outside that map falls back to the direct KV mutate this always + // did - not every resource has to be migrated to keep working. + if let Some(agg_type) = crate::content::aggregate_type_for_bucket(bucket) { + transition_by_aggregate(&state.jetstream, agg_type, &item_id, &to, decision_payload_for_event.clone(), decided_ms) + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + } + answer.state = to.clone(); answer.decided_ms = Some(decided_ms); answer.decided_by = Some(user.username.clone()); @@ -169,15 +211,14 @@ pub async fn transition_answer( // submission hash, published the same way any other answer is - // question_id/alternative come from this call's own arguments and // the content-declared transition label, never a hardcoded value. - let decision_payload = serde_json::json!({ "to": to, "item": item_id, "by": user.username }); let parent_hashes = vec![item_id.clone()]; - let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload, decided_ms); + let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload_for_event, decided_ms); let event = AnswerSubmitted { chain_hash, parent_hashes, question_id, alternative: transition.label, - responses: decision_payload, + responses: decision_payload_for_event, timestamp_ms: decided_ms, }; emit_answer_submitted(&state.nats, &event) @@ -187,6 +228,43 @@ pub async fn transition_answer( Ok(()) } +/// Dispatches to the right aggregate type's `transition()` by name - +/// the one place `answers.rs` has to know the fixed, three-entry set of +/// migrated aggregate types (`content::aggregate_type_for_bucket`'s +/// counterpart), since `crate::aggregates` itself stays generic over +/// any `AggregateKind` and has no list of its own to dispatch through. +#[cfg(feature = "ssr")] +async fn transition_by_aggregate( + js: &async_nats::jetstream::Context, + aggregate_type: &str, + item_id: &str, + to: &str, + payload: serde_json::Value, + occurred_at_ms: i64, +) -> Result<(), crate::aggregates::TransitionError> { + use crate::aggregates::{applicant, project, subscriber, transition, AggregateKind, TransitionError}; + + macro_rules! dispatch { + ($state:ty) => {{ + let target = <$state>::from_event_type(to).ok_or_else(|| { + TransitionError::DisallowedTransition { + from: "?".to_string(), + to: to.to_string(), + } + })?; + transition::<$state>(js, item_id, target, payload, occurred_at_ms).await?; + }}; + } + + match aggregate_type { + "applicant" => dispatch!(applicant::State), + "subscriber" => dispatch!(subscriber::State), + "project" => dispatch!(project::State), + _ => return Err(TransitionError::UnknownAggregate), + } + Ok(()) +} + /// The self-service counterpart to `transition_answer`: no signed-in /// session, no group check - authorized instead by already holding /// `item_id` (a chain hash, opaque and unguessable) plus a matching @@ -244,6 +322,22 @@ pub async fn self_transition_answer( } let decided_ms = chrono::Utc::now().timestamp_millis(); + let decision_payload_for_event = + serde_json::json!({ "to": self_transition.to, "item": item_id, "by": email }); + + if let Some(agg_type) = crate::content::aggregate_type_for_bucket(&self_transition.bucket) { + transition_by_aggregate( + &state.jetstream, + agg_type, + &item_id, + &self_transition.to, + decision_payload_for_event.clone(), + decided_ms, + ) + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + } + answer.state = self_transition.to.clone(); answer.decided_ms = Some(decided_ms); answer.decided_by = Some(email.clone()); @@ -254,15 +348,14 @@ pub async fn self_transition_answer( .await .map_err(|e| ServerFnError::new(e.to_string()))?; - let decision_payload = serde_json::json!({ "to": self_transition.to, "item": item_id, "by": email }); let parent_hashes = vec![item_id.clone()]; - let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload, decided_ms); + let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload_for_event, decided_ms); let event = AnswerSubmitted { chain_hash, parent_hashes, question_id, alternative: self_transition.label, - responses: decision_payload, + responses: decision_payload_for_event, timestamp_ms: decided_ms, }; emit_answer_submitted(&state.nats, &event) @@ -271,3 +364,47 @@ pub async fn self_transition_answer( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// `app.rs`'s list-rendering path only shows the nice + /// button-per-transition UI if a resource's raw JSON value + /// deserializes as `Vec` - otherwise it silently falls + /// back to a plain read-only dump with no error. Guards that shape + /// contract directly, since nothing else would catch a drift here. + #[test] + fn answer_list_round_trips_through_the_shape_app_rs_expects() { + let answers = vec![ + Answer { + id: "abc123".to_string(), + question_id: "/review".to_string(), + alternative: "Applicants".to_string(), + responses: serde_json::json!({"name": "Ada"}), + state: OPEN_STATE.to_string(), + submitted_ms: 1_700_000_000_000, + decided_ms: None, + decided_by: None, + }, + Answer { + id: "def456".to_string(), + question_id: "/review".to_string(), + alternative: "Applicants".to_string(), + responses: serde_json::json!({"name": "Grace"}), + state: "invited".to_string(), + submitted_ms: 1_700_000_001_000, + decided_ms: Some(1_700_000_002_000), + decided_by: Some("bl".to_string()), + }, + ]; + let value = serde_json::Value::Array( + answers.iter().map(|a| serde_json::to_value(a).unwrap()).collect(), + ); + let round_tripped: Vec = + serde_json::from_value(value).expect("Answer list round-trips"); + assert_eq!(round_tripped.len(), 2); + assert_eq!(round_tripped[1].state, "invited"); + assert_eq!(round_tripped[1].decided_by.as_deref(), Some("bl")); + } +} diff --git a/src/app.rs b/src/app.rs index d9652cd..3b80b04 100644 --- a/src/app.rs +++ b/src/app.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use crate::answers::{Answer, SelfTransitionAnswer, TransitionAnswer}; use crate::auth::{current_user, User}; -use crate::content::{is_qualified, Alternative, Question, Transition}; +use crate::content::{is_qualified, Alternative, Question, Responsible, Transition}; use crate::resource::get_resource; /// The visible site name/wordmark - "portal" is just this codebase's @@ -157,6 +157,7 @@ fn QuestionView( .into_any(); } + let question_id_for_note = question_id.clone(); view! { + {question + .responsible + .clone() + .map(|r| { + view! { } + })} } .into_any() } +/// "Asked by X — contact them if you get stuck", plus a one-click +/// "report this question" action (see `report_question`) - mainly +/// meant for a question that reads as unhelpfully LLM-generated, so +/// whoever's responsible for it hears about it. +#[component] +fn ResponsibleNote(question_id: String, responsible: Responsible) -> impl IntoView { + let report = ServerAction::::new(); + let reported = RwSignal::new(false); + Effect::new(move |_| { + if report.value().get().is_some_and(|r| r.is_ok()) { + reported.set(true); + } + }); + + view! { +

+ "Asked by " {responsible.name.clone()} " — " + "contact them" + " if you get stuck." + {move || { + if reported.get() { + view! { " Reported — thanks." } + .into_any() + } else { + let question_id = question_id.clone(); + view! { + + } + .into_any() + } + }} +

+ } +} + // Binds the actual "YES - Rasterized Lines" piece live at uhhm.no // (ported to `public/yes.js`, an ES module now instead of a page-owning // script) as a typed JS class - the idiomatic wasm-bindgen way to drive @@ -691,7 +746,11 @@ fn ResourceFeature( let feature_name = feature_name.clone(); move || (question_id.clone(), alternative.clone(), feature_name.clone()) }, - |(q, a, f)| get_resource(q, a, f), + // Empty params for now - forwarding sibling requirement values + // as query params (for a resource parameterized by other form + // fields) is deferred, see the resource-backed multi-select + // requirement UI follow-up. + |(q, a, f)| get_resource(q, a, f, std::collections::HashMap::new()), ); view! { @@ -996,6 +1055,36 @@ pub async fn submit_answer( Ok(SubmitResult { next, chain_hash }) } +/// Fire-and-forget "this question wasn't helpful" report - a plain log +/// entry in the event store (`aggregate_type: "question_report"`, no +/// state machine, nothing to transition), reviewable via +/// `events::store::load_events(js, "question_report", question_id)` the +/// same way any other aggregate's history is. No auth, same as reading +/// a public question - a report is cheap, low-stakes signal, not +/// something worth gating behind a session. +#[server] +pub async fn report_question( + question_id: String, + reason: Option, +) -> Result<(), ServerFnError> { + use crate::server::AppState; + + let state = expect_context::(); + let occurred_at_ms = chrono::Utc::now().timestamp_millis(); + crate::events::store::append_event( + &state.jetstream, + "question_report", + &question_id, + "reported", + serde_json::json!({ "reason": reason }), + occurred_at_ms, + None, + ) + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + Ok(()) +} + /// Uploads one file to `POST /upload` and returns its stored object key /// - the value a `type: file` requirement contributes to /// `responses_json`, same as any other field. A plain `fetch`, not a diff --git a/src/bin/backfill_events.rs b/src/bin/backfill_events.rs new file mode 100644 index 0000000..f64e0a2 --- /dev/null +++ b/src/bin/backfill_events.rs @@ -0,0 +1,148 @@ +//! One-time migration: for each of the three renamed buckets +//! (`portal_applicants` -> `applicants`, `portal_subscribers` -> +//! `subscribers`, `portal_project_inquiries` -> `projects`), seeds the +//! new event log with a creation event (plus a decision event for any +//! record already decided) and re-puts the record into the new-named +//! KV bucket - then immediately replays what it just wrote and checks +//! it reproduces the original record's state exactly, per the +//! event-sourcing plan's verification step. +//! +//! Defaults to a dry run (lists what it *would* do, writes nothing) - +//! pass `--apply` to actually write. The old-named buckets are never +//! touched or deleted by this tool; remove them by hand once every +//! record's replay has been confirmed to match. +#![cfg(feature = "ssr")] + +use portal::aggregates::{self, applicant, project, subscriber, AggregateKind}; +use portal::answers::{Answer, OPEN_STATE}; +use portal::events::store as event_store; + +const MIGRATIONS: &[(&str, &str, &str)] = &[ + ("portal_applicants", "applicants", "applicant"), + ("portal_subscribers", "subscribers", "subscriber"), + ("portal_project_inquiries", "projects", "project"), +]; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let apply = std::env::args().any(|a| a == "--apply"); + + let nats_url = + std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); + let parsed = url::Url::parse(&nats_url)?; + let mut opts = async_nats::ConnectOptions::new(); + if !parsed.username().is_empty() { + opts = opts.user_and_password( + parsed.username().to_string(), + parsed.password().unwrap_or_default().to_string(), + ); + } + let nats = opts.connect(&nats_url).await?; + let js = async_nats::jetstream::new(nats); + event_store::ensure_stream(&js).await?; + + println!("mode: {}\n", if apply { "APPLY (writing)" } else { "DRY RUN (no writes)" }); + + for (old_bucket, new_bucket, agg_type) in MIGRATIONS { + println!("=== {old_bucket} -> {new_bucket} ({agg_type}) ==="); + let store = match js.get_key_value(*old_bucket).await { + Ok(s) => s, + Err(e) => { + println!(" skip: bucket unavailable: {e}"); + continue; + } + }; + use futures::TryStreamExt; + let keys: Vec = store.keys().await?.try_collect().await?; + if keys.is_empty() { + println!(" (empty)"); + } + for key in keys { + let Some(bytes) = store.get(&key).await? else { continue }; + let answer: Answer = serde_json::from_slice(&bytes)?; + println!(" {key}: state={:?} submitted_ms={}", answer.state, answer.submitted_ms); + if apply { + if let Err(e) = seed_and_verify(&js, agg_type, &key, &answer, new_bucket).await { + println!(" ! {e}"); + } + } + } + } + + if !apply { + println!("\nDry run only - no writes made. Re-run with --apply to actually backfill."); + } + Ok(()) +} + +async fn seed_and_verify( + js: &async_nats::jetstream::Context, + agg_type: &str, + id: &str, + answer: &Answer, + new_bucket: &str, +) -> anyhow::Result<()> { + match agg_type { + "applicant" => seed::(js, id, answer).await?, + "subscriber" => seed::(js, id, answer).await?, + "project" => seed::(js, id, answer).await?, + other => anyhow::bail!("unknown aggregate type {other}"), + } + + let new_store = match js.get_key_value(new_bucket).await { + Ok(s) => s, + Err(_) => { + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: new_bucket.to_string(), + ..Default::default() + }) + .await? + } + }; + new_store.put(id, serde_json::to_vec(answer)?.into()).await?; + Ok(()) +} + +/// Appends the initial event, then - if the original record was +/// already decided - a second event carrying the real decision, before +/// replaying and checking the result matches. Two events for a decided +/// record, not one: a single "imported" event would show a decided +/// record as forever open on replay. +async fn seed( + js: &async_nats::jetstream::Context, + id: &str, + answer: &Answer, +) -> anyhow::Result<()> { + aggregates::create::(js, id, answer.responses.clone(), answer.submitted_ms) + .await + .map_err(|e| anyhow::anyhow!("seeding initial event: {e}"))?; + + if answer.state != OPEN_STATE { + match K::from_event_type(&answer.state) { + Some(target) => { + let payload = serde_json::json!({ + "to": answer.state, + "item": id, + "by": answer.decided_by, + }); + let occurred = answer.decided_ms.unwrap_or(answer.submitted_ms); + aggregates::transition::(js, id, target, payload, occurred) + .await + .map_err(|e| anyhow::anyhow!("seeding decision event: {e}"))?; + } + None => println!(" ! unrecognized decided state {:?}, only seeded initial event", answer.state), + } + } + + let events = event_store::load_events(js, K::AGGREGATE_TYPE, id).await?; + match aggregates::replay::(id, &events) { + Some(agg) if agg.state.event_type() == answer.state => { + println!(" OK: replay matches original state {:?}", answer.state); + } + Some(agg) => { + println!(" MISMATCH: replayed {:?}, original was {:?}", agg.state, answer.state); + } + None => println!(" MISMATCH: no aggregate replayed at all"), + } + Ok(()) +} diff --git a/src/bin/question_lint.rs b/src/bin/question_lint.rs new file mode 100644 index 0000000..5980232 --- /dev/null +++ b/src/bin/question_lint.rs @@ -0,0 +1,76 @@ +//! Headless schema-check binary - loads content (from a Gitea repo URL +//! or a local directory) and validates it exactly the way +//! `content::watch_for_reload`/`main.rs`'s boot path do, with no NATS, +//! OIDC, web server, or JetStream connection involved. Built once by +//! portal's own deploy workflow and downloaded as a release artifact by +//! `questions`' own CI, rather than compiled there - keeps that repo's +//! CI coupling to "run a static binary," not "build a Rust workspace." +#![cfg(feature = "ssr")] + +use portal::content; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let mut args = std::env::args().skip(1); + let mut repo: Option = None; + let mut branch = "main".to_string(); + let mut subdir = "questions".to_string(); + let mut path: Option = None; + + while let Some(arg) = args.next() { + match arg.as_str() { + "--repo" => repo = args.next(), + "--branch" => branch = args.next().unwrap_or(branch), + "--subdir" => subdir = args.next().unwrap_or(subdir), + "--path" => path = args.next(), + other => { + eprintln!("unknown argument: {other}"); + std::process::exit(2); + } + } + } + + let questions = if let Some(dir) = path { + load_from_dir(&dir)? + } else if let Some(repo_url) = repo { + content::load_questions_from_gitea(&repo_url, &branch, &subdir).await? + } else { + eprintln!( + "usage: question-lint --repo [--branch main] [--subdir questions] | --path " + ); + std::process::exit(2); + }; + + match content::validate_questions(&questions) { + Ok(()) => { + println!("OK: {} question(s) valid", questions.len()); + Ok(()) + } + Err(e) => { + eprintln!("FAIL: {e}"); + std::process::exit(1); + } + } +} + +/// The offline counterpart to `content::load_questions_from_gitea` - +/// same "every `*.yaml` file becomes a `Question` keyed by its own +/// `id`" shape, just reading a local checkout instead of Gitea's API, +/// for linting a branch that hasn't been pushed yet. +fn load_from_dir( + dir: &str, +) -> anyhow::Result> { + let mut out = std::collections::HashMap::new(); + for entry in std::fs::read_dir(dir)? { + let entry = entry?; + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("yaml") { + continue; + } + let raw = std::fs::read_to_string(&path)?; + let question: content::Question = serde_yaml::from_str(&raw) + .map_err(|e| anyhow::anyhow!("parsing {}: {e}", path.display()))?; + out.insert(question.id.clone(), question); + } + Ok(out) +} diff --git a/src/content.rs b/src/content.rs index d7c76fc..2e53a46 100644 --- a/src/content.rs +++ b/src/content.rs @@ -21,6 +21,20 @@ pub struct Question { pub qualifies: Option, #[serde(default)] pub alternatives: Vec, + /// Who to contact if a visitor gets stuck or finds this question + /// unhelpful - rendered as a small line on the page, and the person + /// a "report this question" action names in its own event payload + /// (see `events::store`, `question_report` aggregate-less events). + #[serde(default)] + pub responsible: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Responsible { + pub name: String, + /// A mailto address or plain email - deliberately just a string, + /// same as everywhere else content keeps contact info simple. + pub contact: String, } /// Whether `user` may view/submit `question`. `true` when the question @@ -105,8 +119,9 @@ pub struct Feature { /// item. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ResourceSpec { - pub bucket: String, - /// A single item; omit to list the whole bucket. + pub source: ResourceSource, + /// A single item; omit to list the whole bucket. Only meaningful + /// for a `Kv` source. #[serde(default)] pub key: Option, /// Kanidm group required to read this resource. @@ -122,9 +137,59 @@ pub struct ResourceSpec { /// States a listed answer may move to, and the button label for /// each - empty means the resource is read-only. The allow-list a /// `transition_answer` call is checked against, so a client can - /// never move an answer to a state content didn't declare. + /// never move an answer to a state content didn't declare. Only + /// meaningful for a `Kv` source - a live external pull is always + /// read-only. #[serde(default)] pub transitions: Vec, + /// A jq filter reshaping the fetched value before it reaches the + /// frontend - e.g. `.[] | {name, url: .html_url}` to pick just the + /// fields a showcase card needs out of a raw Gitea API response. + /// Evaluated via the `jaq` crate (`resource::apply_jq`), no shell + /// -out. `None` (or a `Kv` source) returns the fetched value as-is. + #[serde(default)] + pub jq: Option, +} + +/// Where a resource's live data actually comes from. `Kv` is the +/// original (and still only mutable) behavior - a NATS KV bucket this +/// server itself owns. The `Gitea*` variants are live pulls from the +/// same Gitea instance content is loaded from (`AppState.gitea_base`), +/// for surfacing things like a portfolio of starred repos without +/// hand-curating a KV bucket for it. Not mutually exclusive with +/// owner-curated resources - a hand-picked feature list is just a `Kv` +/// resource an owner writes into; a `GiteaStarred`/`GiteaOrgRepos` pull +/// is the automatic catalog to choose from. Both can appear as +/// different features on the same page. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ResourceSource { + Kv { bucket: String }, + GiteaStarred { username: String }, + GiteaOrgRepos { org: String }, + /// Any other HTTPS JSON endpoint - the general escape hatch for a + /// live resource that isn't this Gitea instance. Deliberately + /// scheme-restricted and checked against loopback/private/link + /// -local addresses at fetch time (`resource::fetch_url_resource`) + /// - content is trusted today (same org owns both repos), but a + /// server-side fetch of a content-supplied URL is real SSRF surface + /// the moment that stops being true, so it fails closed rather than + /// trusting every host by default. + Url { url: String }, +} + +impl ResourceSpec { + /// The KV bucket this resource reads/writes, if it's `Kv`-sourced - + /// `None` for a live external pull, which has no bucket and (see + /// `transitions` above) can't be mutated through this mechanism. + pub fn bucket(&self) -> Option<&str> { + match &self.source { + ResourceSource::Kv { bucket } => Some(bucket), + ResourceSource::GiteaStarred { .. } + | ResourceSource::GiteaOrgRepos { .. } + | ResourceSource::Url { .. } => None, + } + } } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -251,6 +316,85 @@ pub async fn load_questions_from_gitea( Ok(out) } +/// Maps a content-declared KV bucket name to the aggregate type +/// (`crate::aggregates::*::State::AGGREGATE_TYPE`) it's event-sourced +/// under - the fixed, three-entry table for this pass (see the +/// event-sourcing plan). A bucket name not listed here is left alone +/// entirely (no validation applied) - deliberate: only buckets actually +/// migrated onto the generic aggregate machinery get their transitions +/// checked. +#[cfg(feature = "ssr")] +pub fn aggregate_type_for_bucket(bucket: &str) -> Option<&'static str> { + match bucket { + "applicants" => Some("applicant"), + "subscribers" => Some("subscriber"), + "projects" => Some("project"), + _ => None, + } +} + +#[cfg(feature = "ssr")] +fn is_valid_transition_target(aggregate_type: &str, to: &str) -> bool { + use crate::aggregates::AggregateKind; + fn check(to: &str) -> bool { + K::INITIAL_STATE.allowed().iter().any(|s| s.event_type() == to) + } + match aggregate_type { + "applicant" => check::(to), + "subscriber" => check::(to), + "project" => check::(to), + _ => false, + } +} + +/// Validates every declared transition target (`SelfTransition.to`, +/// `ResourceSpec.transitions[].to`) against the real compiled +/// transition table for its bucket's aggregate type, for every bucket +/// this pass actually migrated (see `aggregate_type_for_bucket`). +/// Called on every content load/reload (`watch_for_reload`, `main.rs`'s +/// boot path) and by the standalone `question-lint` binary, so a YAML +/// typo becomes a caught, logged rejection instead of a silently +/// -accepted, later-broken string. +#[cfg(feature = "ssr")] +pub fn validate_questions( + questions: &std::collections::HashMap, +) -> anyhow::Result<()> { + for question in questions.values() { + for alternative in &question.alternatives { + if let Some(st) = &alternative.self_transition { + if let Some(agg) = aggregate_type_for_bucket(&st.bucket) { + if !is_valid_transition_target(agg, &st.to) { + anyhow::bail!( + "question {:?} alternative {:?}: self_transition.to {:?} is not a valid transition for bucket {:?} ({agg})", + question.id, alternative.name, st.to, st.bucket + ); + } + } + } + for feature in &alternative.features { + let Some(resource) = &feature.resource else { + continue; + }; + let Some(bucket) = resource.bucket() else { + continue; + }; + let Some(agg) = aggregate_type_for_bucket(bucket) else { + continue; + }; + for transition in &resource.transitions { + if !is_valid_transition_target(agg, &transition.to) { + anyhow::bail!( + "question {:?} alternative {:?} feature {:?}: transition.to {:?} is not a valid transition for bucket {:?} ({agg})", + question.id, alternative.name, feature.name, transition.to, bucket + ); + } + } + } + } + } + Ok(()) +} + /// Published by the content repo's own CI (after it lints a push) to /// tell every running instance to pick up the change - a plain fire /// and forget NATS publish, no payload, matching `events.rs`'s @@ -283,6 +427,10 @@ pub async fn watch_for_reload( while sub.next().await.is_some() { match load_questions_from_gitea(&repo_url, &branch, &subdir).await { Ok(loaded) => { + if let Err(e) = validate_questions(&loaded) { + tracing::error!(error = %e, "content reload failed validation, keeping last-good content"); + continue; + } let count = loaded.len(); questions.store(std::sync::Arc::new(loaded)); tracing::info!(count, "reloaded content"); diff --git a/src/events.rs b/src/events/mod.rs similarity index 98% rename from src/events.rs rename to src/events/mod.rs index 364d2c3..b61e734 100644 --- a/src/events.rs +++ b/src/events/mod.rs @@ -6,6 +6,8 @@ //! change shape when that shows up, only gain a second subscriber. #![cfg(feature = "ssr")] +pub mod store; + use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, Serialize, Deserialize)] diff --git a/src/events/store.rs b/src/events/store.rs new file mode 100644 index 0000000..0d9cff1 --- /dev/null +++ b/src/events/store.rs @@ -0,0 +1,150 @@ +//! 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..`, +//! deliberately *not* `events...` 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 { + 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, +) -> Result +{ + 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> { + 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) +} diff --git a/src/lib.rs b/src/lib.rs index d3e214f..37998d9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,6 +6,9 @@ pub mod content; pub mod events; pub mod resource; +#[cfg(feature = "ssr")] +pub mod aggregates; + #[cfg(feature = "ssr")] pub mod server; diff --git a/src/main.rs b/src/main.rs index 60cea3b..b2e60a0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -30,6 +30,7 @@ async fn main() -> anyhow::Result<()> { let content_branch = std::env::var("CONTENT_BRANCH").unwrap_or_else(|_| "main".to_string()); let gitea_base = content::gitea_api_base(&content_repo)?; let questions = content::load_questions_from_gitea(&content_repo, &content_branch, "questions").await?; + content::validate_questions(&questions)?; tracing::info!(count = questions.len(), repo = %content_repo, branch = %content_branch, "loaded content"); let questions = Arc::new(arc_swap::ArcSwap::from_pointee(questions)); @@ -50,6 +51,7 @@ async fn main() -> anyhow::Result<()> { // names whatever bucket an alternative's `record_as` should write // into, and `answers::store_answer` creates it on first use. let jetstream = async_nats::jetstream::new(nats.clone()); + portal::events::store::ensure_stream(&jetstream).await?; let oidc_state = Arc::new(oidc::Oidc::from_env().await?); diff --git a/src/resource.rs b/src/resource.rs index 71df30d..b347517 100644 --- a/src/resource.rs +++ b/src/resource.rs @@ -1,24 +1,30 @@ -//! Generic, authorized reads of NATS KV data declared in content +//! Generic, authorized reads of live data declared in content //! (`content::ResourceSpec`, on a `Feature`). The one property that -//! matters: the bucket/key a resource reads is never a client-supplied -//! parameter, only something the server resolves from its own -//! trusted, YAML-loaded content - the client only ever names a -//! question id + feature name, so it can't probe an arbitrary bucket -//! by just naming it. +//! matters: what a resource reads is never a client-supplied parameter, +//! only something the server resolves from its own trusted, YAML-loaded +//! content - the client only ever names a question id + feature name +//! (plus, now, whatever form-field values it already has, forwarded as +//! query params), never a bucket/URL directly. use leptos::prelude::*; /// Fetches the live data for `question_id`'s `alternative`'s /// `feature_name` feature. Fails closed: a resource with neither /// `public: true` nor `requires_group` set is unreachable, not "open" -/// by omission. +/// by omission. `params` are the requesting form's current field +/// values (if any) - forwarded as query parameters to a live external +/// fetch (`GiteaStarred`/`GiteaOrgRepos`/`Url`), the mechanism behind +/// "a resource parameterized by other form fields"; a `Kv` resource +/// ignores them entirely, same as today. #[server] pub async fn get_resource( question_id: String, alternative: String, feature_name: String, + params: std::collections::HashMap, ) -> Result { use crate::auth::{User, SESSION_USER_KEY}; + use crate::content::ResourceSource; use crate::server::AppState; let state = expect_context::(); @@ -66,39 +72,270 @@ pub async fn get_resource( } } - let store = state - .jetstream - .get_key_value(&resource.bucket) - .await - .map_err(|e| ServerFnError::new(format!("resource bucket unavailable: {e}")))?; - - match &resource.key { - Some(key) => { - let bytes = store - .get(key) + let value = match &resource.source { + ResourceSource::Kv { bucket } => { + let store = state + .jetstream + .get_key_value(bucket) .await - .map_err(|e| ServerFnError::new(e.to_string()))? - .ok_or_else(|| ServerFnError::new("resource key not found"))?; - serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string())) - } - None => { - use futures::TryStreamExt; - let keys: Vec = store - .keys() - .await - .map_err(|e| ServerFnError::new(e.to_string()))? - .try_collect() - .await - .map_err(|e| ServerFnError::new(e.to_string()))?; - let mut items = Vec::new(); - for key in keys { - if let Ok(Some(bytes)) = store.get(&key).await { - if let Ok(value) = serde_json::from_slice::(&bytes) { - items.push(value); + .map_err(|e| ServerFnError::new(format!("resource bucket unavailable: {e}")))?; + match &resource.key { + Some(key) => { + let bytes = store + .get(key) + .await + .map_err(|e| ServerFnError::new(e.to_string()))? + .ok_or_else(|| ServerFnError::new("resource key not found"))?; + serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string()))? + } + None => { + use futures::TryStreamExt; + let keys: Vec = store + .keys() + .await + .map_err(|e| ServerFnError::new(e.to_string()))? + .try_collect() + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + let mut items = Vec::new(); + for key in keys { + if let Ok(Some(bytes)) = store.get(&key).await { + if let Ok(value) = serde_json::from_slice::(&bytes) { + items.push(value); + } + } } + serde_json::Value::Array(items) } } - Ok(serde_json::Value::Array(items)) + } + ResourceSource::GiteaStarred { username } => { + fetch_gitea_json(&state, &format!("/api/v1/users/{username}/starred"), ¶ms).await? + } + ResourceSource::GiteaOrgRepos { org } => { + fetch_gitea_json(&state, &format!("/api/v1/orgs/{org}/repos"), ¶ms).await? + } + ResourceSource::Url { url } => fetch_url_json(url, ¶ms).await?, + }; + + match &resource.jq { + Some(filter) => apply_jq(filter, &value), + None => Ok(value), + } +} + +/// Fetches JSON from a path on the same Gitea instance content is +/// loaded from (`AppState.gitea_base`), attaching `GITEA_API_TOKEN` if +/// set - confirmed live against `project.uhhm.no` that at least the +/// starred-repos endpoint 401s unauthenticated, unlike the repo +/// -contents/repo-info endpoints `content.rs` already calls anonymously. +#[cfg(feature = "ssr")] +async fn fetch_gitea_json( + state: &crate::server::AppState, + path: &str, + params: &std::collections::HashMap, +) -> Result { + let mut url = url::Url::parse(&format!("{}{}", state.gitea_base, path)) + .map_err(|e| ServerFnError::new(e.to_string()))?; + for (k, v) in params { + url.query_pairs_mut().append_pair(k, v); + } + let client = openidconnect::reqwest::Client::new(); + let mut req = client.get(url.clone()); + if let Ok(token) = std::env::var("GITEA_API_TOKEN") { + if !token.is_empty() { + req = req.header("Authorization", format!("token {token}")); + } + } + let text = req + .send() + .await + .and_then(|r| r.error_for_status()) + .map_err(|e| ServerFnError::new(format!("fetching {url}: {e}")))? + .text() + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + serde_json::from_str(&text).map_err(|e| ServerFnError::new(e.to_string())) +} + +/// Fetches JSON from an arbitrary content-declared URL +/// (`content::ResourceSource::Url`). Content is trusted today (same +/// org owns both this app and its content repo), but a server-side +/// fetch of a content-supplied URL is real SSRF surface the moment +/// that stops being true - `assert_public_https_url` fails closed on +/// anything that isn't `https://` and actually resolves to a public +/// address, not just a syntactically-plausible one. +#[cfg(feature = "ssr")] +async fn fetch_url_json( + url: &str, + params: &std::collections::HashMap, +) -> Result { + let mut parsed = assert_public_https_url(url) + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + for (k, v) in params { + parsed.query_pairs_mut().append_pair(k, v); + } + let client = openidconnect::reqwest::Client::new(); + let text = client + .get(parsed.clone()) + .send() + .await + .and_then(|r| r.error_for_status()) + .map_err(|e| ServerFnError::new(format!("fetching {parsed}: {e}")))? + .text() + .await + .map_err(|e| ServerFnError::new(e.to_string()))?; + serde_json::from_str(&text).map_err(|e| ServerFnError::new(e.to_string())) +} + +/// `https://` only, and only if every address the host resolves to is +/// public - rejects loopback (127.0.0.1, ::1), RFC1918/link-local +/// private ranges, and unspecified/multicast addresses. Resolves via +/// real DNS (not just a syntax check) so a hostname that *points at* an +/// internal address is caught too, not just a literal internal IP +/// spelled out directly. +#[cfg(feature = "ssr")] +async fn assert_public_https_url(url: &str) -> anyhow::Result { + let parsed = url::Url::parse(url)?; + if parsed.scheme() != "https" { + anyhow::bail!("only https:// resource urls are allowed"); + } + let host = parsed + .host_str() + .ok_or_else(|| anyhow::anyhow!("resource url has no host"))?; + let port = parsed.port_or_known_default().unwrap_or(443); + let mut resolved = false; + for addr in tokio::net::lookup_host((host, port)) + .await + .map_err(|e| anyhow::anyhow!("resolving resource host {host}: {e}"))? + { + resolved = true; + if !is_public_ip(&addr.ip()) { + anyhow::bail!("resource host {host} resolves to a non-public address"); + } + } + if !resolved { + anyhow::bail!("resource host {host} did not resolve to any address"); + } + Ok(parsed) +} + +#[cfg(feature = "ssr")] +fn is_public_ip(ip: &std::net::IpAddr) -> bool { + match ip { + std::net::IpAddr::V4(v4) => { + !(v4.is_private() + || v4.is_loopback() + || v4.is_link_local() + || v4.is_multicast() + || v4.is_broadcast() + || v4.is_unspecified() + || v4.is_documentation()) + } + std::net::IpAddr::V6(v6) => { + let is_unique_local = (v6.segments()[0] & 0xfe00) == 0xfc00; + let is_link_local = (v6.segments()[0] & 0xffc0) == 0xfe80; + !(v6.is_loopback() + || v6.is_unspecified() + || v6.is_multicast() + || is_unique_local + || is_link_local) } } } + +/// Runs `filter_src` (a jq filter) over `input`, collecting every value +/// it yields into a JSON array - e.g. `.[] | {name, url: .html_url}` +/// over a Gitea repo list yields one shaped object per repo. Pure Rust +/// via the `jaq` crate family, no shell-out to a `jq` binary. +#[cfg(feature = "ssr")] +pub fn apply_jq( + filter_src: &str, + input: &serde_json::Value, +) -> Result { + use jaq_core::load::{Arena, File, Loader}; + use jaq_core::{data, unwrap_valr, Compiler, Ctx, Vars}; + use jaq_json::{read, Val}; + + let input_text = input.to_string(); + let input_val = read::parse_single(input_text.as_bytes()) + .map_err(|e| ServerFnError::new(format!("internal: re-parsing json for jq: {e:?}")))?; + + let program = File { code: filter_src, path: () }; + let defs = jaq_core::defs().chain(jaq_std::defs()).chain(jaq_json::defs()); + let funs = jaq_core::funs().chain(jaq_std::funs()).chain(jaq_json::funs()); + let loader = Loader::new(defs); + let arena = Arena::default(); + let modules = loader + .load(&arena, program) + .map_err(|e| ServerFnError::new(format!("jq filter parse error: {e:?}")))?; + let filter = Compiler::default() + .with_funs(funs) + .compile(modules) + .map_err(|e| ServerFnError::new(format!("jq filter compile error: {e:?}")))?; + let ctx = Ctx::>::new(&filter.lut, Vars::new([])); + + let mut outputs = Vec::new(); + for result in filter.id.run((ctx, input_val)).map(unwrap_valr) { + let val = result.map_err(|e| ServerFnError::new(format!("jq filter runtime error: {e:?}")))?; + let text = val.to_string(); + let json: serde_json::Value = serde_json::from_str(&text) + .map_err(|e| ServerFnError::new(format!("internal: parsing jq output: {e}")))?; + outputs.push(json); + } + Ok(serde_json::Value::Array(outputs)) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A canned, Gitea-API-shaped fixture - the same fields the real + /// `/users/{username}/starred` endpoint returns - run through the + /// filter `index.yaml`'s "What we've built" resource actually + /// declares, confirming the `jaq` integration produces the shape + /// the frontend showcase card expects. + #[test] + fn jq_shapes_gitea_repo_list_for_the_showcase() { + let input = serde_json::json!([ + { + "name": "cnats", + "description": "A NATS-backed chat client", + "html_url": "https://project.uhhm.no/bl/cnats", + "stargazers_count": 3, + "private": false + }, + { + "name": "portal", + "description": "This app", + "html_url": "https://project.uhhm.no/uhhm/portal", + "stargazers_count": 1, + "private": false + } + ]); + let filter = ".[] | {name: .name, description: .description, url: .html_url, stars: .stargazers_count}"; + + let shaped = apply_jq(filter, &input).expect("filter runs"); + let items = shaped.as_array().expect("array output"); + assert_eq!(items.len(), 2); + assert_eq!(items[0]["name"], "cnats"); + assert_eq!(items[0]["url"], "https://project.uhhm.no/bl/cnats"); + assert_eq!(items[0]["stars"], 3); + // The filter never mentions `private` - confirms shaping + // actually drops fields, not just passes the object through. + assert!(items[0].get("private").is_none()); + } + + #[test] + fn is_public_ip_rejects_loopback_and_private() { + let loopback: std::net::IpAddr = "127.0.0.1".parse().unwrap(); + let private: std::net::IpAddr = "10.0.0.5".parse().unwrap(); + let link_local: std::net::IpAddr = "169.254.1.1".parse().unwrap(); + let public: std::net::IpAddr = "1.1.1.1".parse().unwrap(); + assert!(!is_public_ip(&loopback)); + assert!(!is_public_ip(&private)); + assert!(!is_public_ip(&link_local)); + assert!(is_public_ip(&public)); + } +}