Drops porting-history narratives (dodrenett), superseded-behavior explanations, and restatements of what the next line does. Constraint notes (fail-closed policies, CAS semantics, cascade behavior, id uniqueness) stay, just shorter. No code changes. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
20b24f08c2
commit
6932816c42
@@ -174,10 +174,8 @@ 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).
|
||||
/// A concurrent writer advanced this aggregate past the sequence
|
||||
/// this call read (see `events::store::append_event`'s CAS).
|
||||
Conflict,
|
||||
Store(String),
|
||||
}
|
||||
|
||||
+26
-52
@@ -1,16 +1,8 @@
|
||||
//! Read-model storage for submitted answers - the KV-backed projection
|
||||
//! of the durable event log now underneath it (`events::store`,
|
||||
//! `aggregates`) for every bucket that has a declared state graph in
|
||||
//! `AppState.aggregates` (loaded from `aggregates.yaml` - see
|
||||
//! `content::load_aggregates_from_gitea`). 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 the real, content-declared
|
||||
//! transition table, for the buckets that have one. A bucket with no
|
||||
//! entry in `aggregates` 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 declare a state graph
|
||||
//! before it needs one.
|
||||
//! Read-model storage for submitted answers: the KV projection of the
|
||||
//! event log (`events::store`, `aggregates`) for every bucket with a
|
||||
//! state graph in `aggregates.yaml`. A bucket with no graph still
|
||||
//! works - direct KV mutate, no event log, no CAS - so content isn't
|
||||
//! forced to declare one before it needs it.
|
||||
|
||||
use leptos::prelude::*;
|
||||
use serde::{Deserialize, Serialize};
|
||||
@@ -95,11 +87,9 @@ pub async fn store_answer(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// One selected row's pending transition - `feature_name` is carried
|
||||
/// per-item (not once per call) so a batch can span more than one
|
||||
/// resource feature on the same alternative, even though today's
|
||||
/// content never actually declares more than one transitionable
|
||||
/// resource per alternative.
|
||||
/// One selected row's pending transition. `feature_name` is per-item
|
||||
/// so a batch can span more than one resource feature on the same
|
||||
/// alternative.
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct TransitionItem {
|
||||
pub feature_name: String,
|
||||
@@ -107,15 +97,10 @@ pub struct TransitionItem {
|
||||
pub to: String,
|
||||
}
|
||||
|
||||
/// Moves every selected row to its chosen target state in one call -
|
||||
/// the batched counterpart to what used to be a `transition_answer`
|
||||
/// fired separately per row, one "Confirm" button each. Now there's
|
||||
/// one shared button per alternative (see `AlternativeCard`), and a
|
||||
/// click here applies whatever was selected across every row at once.
|
||||
/// Each item is independent - there's no cross-item transaction to
|
||||
/// have, only per-item CAS (see `aggregates::transition`) - so one
|
||||
/// item's failure doesn't roll back or block the others. Failures are
|
||||
/// collected and reported together; whatever succeeded stays applied.
|
||||
/// Moves every selected row to its chosen target state in one call
|
||||
/// (the shared per-alternative Confirm button). Items are independent
|
||||
/// - per-item CAS, no cross-item transaction - so one failure doesn't
|
||||
/// roll back the others; failures are collected and reported together.
|
||||
#[server]
|
||||
pub async fn transition_answers(
|
||||
question_id: String,
|
||||
@@ -233,12 +218,9 @@ async fn apply_transition(
|
||||
.ok_or_else(|| ServerFnError::new("unknown answer"))?;
|
||||
let mut answer: Answer =
|
||||
serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string()))?;
|
||||
// Matched on (from, to), not to alone - two declared transitions
|
||||
// may share a target (open -> declined and in_dialogue -> declined),
|
||||
// and only the one whose `from` is the row's actual current state
|
||||
// is valid to fire. This also subsumes the old blanket
|
||||
// "already decided" check: a row in a state no declared transition
|
||||
// starts from simply has no legal move here.
|
||||
// Matched on (from, to), not to alone: two declared transitions may
|
||||
// share a target (open -> declined, in_dialogue -> declined), and
|
||||
// only the one starting from the row's actual state may fire.
|
||||
let transition = resource
|
||||
.transitions
|
||||
.iter()
|
||||
@@ -255,13 +237,9 @@ async fn apply_transition(
|
||||
let decision_payload_for_event =
|
||||
serde_json::json!({ "to": item.to, "item": item.item_id, "by": user.username });
|
||||
|
||||
// For any bucket with a declared state graph, 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 `aggregates` falls back to
|
||||
// the direct KV mutate this always did - not every resource has to
|
||||
// be event-sourced to keep working.
|
||||
// CAS on the aggregate's replayed sequence guards two concurrent
|
||||
// decisions on the same item; a schemaless bucket falls back to the
|
||||
// plain KV mutate below.
|
||||
if let Some(schema) = state.aggregates.load().get(bucket) {
|
||||
transition_or_reseed(
|
||||
&state.jetstream,
|
||||
@@ -286,10 +264,8 @@ async fn apply_transition(
|
||||
.await
|
||||
.map_err(|e| ServerFnError::new(e.to_string()))?;
|
||||
|
||||
// Extend the DAG: the decision is a child node of the answer's own
|
||||
// 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.
|
||||
// The decision extends the DAG as a child of the answer's own
|
||||
// submission hash, published like any other answer.
|
||||
let parent_hashes = vec![item.item_id.clone()];
|
||||
let chain_hash = hash_node(question_id, &parent_hashes, &decision_payload_for_event, decided_ms);
|
||||
let event = AnswerSubmitted {
|
||||
@@ -307,14 +283,12 @@ async fn apply_transition(
|
||||
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
|
||||
/// `email`, both carried in the link itself (see
|
||||
/// `content::SelfTransition`). Deliberately returns the same generic
|
||||
/// error for "no such item" and "email doesn't match" - a real
|
||||
/// unsubscribe link should never let someone probe which chain hashes
|
||||
/// or emails exist.
|
||||
/// The self-service counterpart to `transition_answers`: no session,
|
||||
/// no group check - authorized by holding `item_id` (an unguessable
|
||||
/// chain hash) plus a matching `email`, both carried in the link
|
||||
/// itself. Deliberately returns the same generic error for "no such
|
||||
/// item" and "email doesn't match", so an unsubscribe link can't be
|
||||
/// used to probe which hashes or emails exist.
|
||||
#[server]
|
||||
pub async fn self_transition_answer(
|
||||
question_id: String,
|
||||
|
||||
+6
-12
@@ -479,14 +479,9 @@ fn AlternativeCard(
|
||||
}
|
||||
});
|
||||
|
||||
// File fields don't fit a live-typed RwSignal<String> - they get
|
||||
// their own map of element refs, read (and uploaded) only at submit
|
||||
// time, and are excluded from `field_map` below. `select` fields
|
||||
// don't fit it either (a multi-select's value is a *set* of ids,
|
||||
// not one string) - they get their own `Vec<String>`-signal map,
|
||||
// plus a plain (non-reactive, content-derived) record of which
|
||||
// select fields are `multiple`, read back at submit time to decide
|
||||
// whether to emit a JSON array or a single string.
|
||||
// Text fields live in field_map; file fields (element refs, read at
|
||||
// submit) and select fields (a set of ids, plus which are multiple)
|
||||
// each need their own shape.
|
||||
let mut field_map: std::collections::HashMap<String, RwSignal<String>> =
|
||||
std::collections::HashMap::new();
|
||||
let mut file_refs: std::collections::HashMap<String, NodeRef<leptos::html::Input>> =
|
||||
@@ -511,10 +506,9 @@ fn AlternativeCard(
|
||||
}
|
||||
}
|
||||
|
||||
// Prefixes field DOM ids with the alternative too, not just the
|
||||
// question - two alternatives on the same page (e.g. both asking for
|
||||
// "email") would otherwise mint the same id twice, which is invalid
|
||||
// HTML and makes <label for=…> pick the wrong input.
|
||||
// DOM ids are prefixed with question AND alternative - two
|
||||
// alternatives on one page can both ask for "email", and duplicate
|
||||
// ids would make <label for=...> pick the wrong input.
|
||||
let field_prefix = format!(
|
||||
"{question_id}-{}",
|
||||
alternative.name.to_lowercase().replace(' ', "-")
|
||||
|
||||
+3
-8
@@ -1,11 +1,6 @@
|
||||
//! SHA-256 hashed DAG chain, ported from dodrenett's lib/challenge.ts.
|
||||
//! Kept deliberately plain: every submission hashes in its parent(s), but
|
||||
//! nothing here produces more than one parent yet - that only starts
|
||||
//! mattering once something (an LLM, or a human) posts follow-up
|
||||
//! questions back into a visitor's chain instead of the fixed content in
|
||||
//! ../portal-content. The DAG-shaped `parent_hashes` field exists so that
|
||||
//! can slot in later without a schema change; the hashing itself doesn't
|
||||
//! need to.
|
||||
//! SHA-256 hashed answer chain. `parent_hashes` is DAG-shaped, but
|
||||
//! nothing produces more than one parent yet - the shape is reserved
|
||||
//! for follow-up questions posted back into a visitor's chain.
|
||||
#![cfg(feature = "ssr")]
|
||||
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
+46
-95
@@ -71,25 +71,15 @@ pub struct Alternative {
|
||||
pub images: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub features: Vec<Feature>,
|
||||
/// Names a NATS KV bucket to also durably store this submission
|
||||
/// into, beyond the ephemeral NATS event every submission already
|
||||
/// gets published as - just a bucket name, not a keyword the
|
||||
/// runtime special-cases (see `src/answers.rs`). Content-driven
|
||||
/// rather than matching on the alternative's name/copy, which would
|
||||
/// silently break if the wording changes later.
|
||||
/// NATS KV bucket to durably store this submission into - just a
|
||||
/// bucket name, never a keyword the runtime special-cases.
|
||||
#[serde(default)]
|
||||
pub record_as: Option<String>,
|
||||
/// A transition anyone holding the *right item's own reference* can
|
||||
/// fire - not gated by `requires_group` like `ResourceSpec.transitions`,
|
||||
/// which is for signed-in owners browsing a whole bucket. This is
|
||||
/// the opposite shape: an anonymous visitor who already holds one
|
||||
/// specific chain hash (from a `?chain=` link - see `chain.rs`,
|
||||
/// the same reference `/subscribed?chain=...` already carries) may
|
||||
/// transition *that one item*, and only that one, without ever
|
||||
/// being able to enumerate or touch anyone else's. `email` is a
|
||||
/// second factor checked against the stored item's own `email`
|
||||
/// response field - not the lookup key, just cheap defense in depth
|
||||
/// against a leaked/guessed chain hash alone being sufficient.
|
||||
/// A transition fireable by anyone holding one specific item's own
|
||||
/// reference (`?chain=` link) plus a matching `email` - the
|
||||
/// anonymous, single-item counterpart to `ResourceSpec.transitions`'
|
||||
/// group-gated bucket browsing. The email is a second factor
|
||||
/// checked against the stored item, not the lookup key.
|
||||
#[serde(default)]
|
||||
pub self_transition: Option<SelfTransition>,
|
||||
}
|
||||
@@ -106,21 +96,14 @@ pub struct Feature {
|
||||
pub name: String,
|
||||
#[serde(default)]
|
||||
pub description: String,
|
||||
/// Any valid CSS color (hex, named, rgb()/oklch()/...) - set as this
|
||||
/// feature's own `--feature-accent` custom property rather than
|
||||
/// interpolated into a stylesheet, so a bad value just fails to
|
||||
/// apply instead of being live CSS content could inject arbitrary
|
||||
/// rules into. Unset means no accent border at all (see
|
||||
/// style/main.css's `.feature`), not a silent fallback to the
|
||||
/// global `--accent` - a feature that never asked for a color
|
||||
/// shouldn't suddenly gain a visible border.
|
||||
/// Any valid CSS color - set as the feature's `--feature-accent`
|
||||
/// custom property (never interpolated into a stylesheet, so a bad
|
||||
/// value fails to apply instead of injecting CSS). Unset means no
|
||||
/// accent border at all.
|
||||
#[serde(default)]
|
||||
pub color: Option<String>,
|
||||
/// An Iconify icon name (`{prefix}:{name}`, e.g. `lucide:star`),
|
||||
/// rendered via Iconify's public SVG API
|
||||
/// (`https://api.iconify.design/{icon}.svg`) - no icon library
|
||||
/// bundled here, matching this app's zero-JS-dependency content
|
||||
/// otherwise keeps to.
|
||||
/// rendered via Iconify's public SVG API - no icon library bundled.
|
||||
#[serde(default)]
|
||||
pub icon: Option<String>,
|
||||
#[serde(default)]
|
||||
@@ -132,15 +115,11 @@ pub struct Feature {
|
||||
pub resource: Option<ResourceSpec>,
|
||||
}
|
||||
|
||||
/// Declares a NATS KV-backed live-data read. The bucket/key here are
|
||||
/// only ever read server-side, from content the server itself loaded at
|
||||
/// startup - `get_resource` (`src/resource.rs`) takes a question id +
|
||||
/// feature name from the client, never a bucket/key directly, so a
|
||||
/// visitor can never probe an arbitrary bucket by naming it. There is
|
||||
/// no render-mode/"kind" tag here on purpose - what a resource displays
|
||||
/// as follows from its data's own shape (a plain value, or a list of
|
||||
/// `answers::Answer`s), not from a Rust-side special case per content
|
||||
/// item.
|
||||
/// A live-data read declared in content. The bucket/key are only ever
|
||||
/// resolved server-side from trusted content - a client names a
|
||||
/// question + feature, never a bucket directly, so a visitor can't
|
||||
/// probe arbitrary buckets. No render-mode tag: what a resource
|
||||
/// displays as follows from its data's shape.
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct ResourceSpec {
|
||||
pub source: ResourceSource,
|
||||
@@ -152,60 +131,43 @@ pub struct ResourceSpec {
|
||||
#[serde(default)]
|
||||
pub requires_group: Option<String>,
|
||||
/// Must be explicitly set for an anonymous-readable resource - a
|
||||
/// spec with neither this nor `requires_group` is unreachable by
|
||||
/// design (fail closed, not fail open). Reads only - mutations
|
||||
/// (`transitions` below) always require `requires_group` regardless
|
||||
/// of this flag.
|
||||
/// spec with neither this nor `requires_group` is unreachable
|
||||
/// (fail closed). Reads only: mutations always require
|
||||
/// `requires_group` regardless of this flag.
|
||||
#[serde(default)]
|
||||
pub public: bool,
|
||||
/// 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. Only
|
||||
/// meaningful for a `Kv` source - a live external pull is always
|
||||
/// read-only.
|
||||
/// The moves a listed answer may make, one button each - empty
|
||||
/// means read-only. Server calls are checked against this
|
||||
/// allow-list, so a client can never fire a transition content
|
||||
/// didn't declare. `Kv` sources only.
|
||||
#[serde(default)]
|
||||
pub transitions: Vec<Transition>,
|
||||
/// 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.
|
||||
/// A jq filter (evaluated via `jaq`, no shell-out) reshaping the
|
||||
/// fetched value before it reaches the frontend - e.g.
|
||||
/// `.[] | {name, url: .html_url}`. `None` returns it as-is.
|
||||
#[serde(default)]
|
||||
pub jq: Option<String>,
|
||||
}
|
||||
|
||||
/// 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.
|
||||
/// Where a resource's live data comes from. `Kv` (a NATS KV bucket
|
||||
/// this server owns) is the only mutable source; the rest are
|
||||
/// read-only live pulls.
|
||||
#[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.
|
||||
/// Any other HTTPS JSON endpoint. Scheme-restricted and checked
|
||||
/// against loopback/private/link-local addresses at fetch time
|
||||
/// (`resource::fetch_url_resource`) - a server-side fetch of a
|
||||
/// content-supplied URL is SSRF surface, so it fails closed.
|
||||
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.
|
||||
/// The KV bucket this resource reads/writes - `None` for a live
|
||||
/// external pull.
|
||||
pub fn bucket(&self) -> Option<&str> {
|
||||
match &self.source {
|
||||
ResourceSource::Kv { bucket } => Some(bucket),
|
||||
@@ -218,10 +180,8 @@ impl ResourceSpec {
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Transition {
|
||||
/// The state a row must currently be in for this transition's
|
||||
/// button to render and its server call to be accepted - defaults
|
||||
/// to `"open"`, which is what every transition meant before graphs
|
||||
/// grew deeper than one decision (invited -> active, etc.).
|
||||
/// The state a row must be in for this transition's button to
|
||||
/// render and its server call to be accepted. Defaults to "open".
|
||||
#[serde(default = "default_transition_from")]
|
||||
pub from: String,
|
||||
pub to: String,
|
||||
@@ -253,17 +213,12 @@ pub struct Requirement {
|
||||
/// boundary - the upload handler re-checks content-type itself).
|
||||
#[serde(default)]
|
||||
pub accept: Option<String>,
|
||||
/// `type: select` only - where the selectable options come from.
|
||||
/// Reuses the exact same `ResourceSpec`/`ResourceSource`/`jq`
|
||||
/// mechanism a `Feature.resource` uses (`resource::get_requirement_options`)
|
||||
/// - a resource is a resource regardless of whether it's displayed
|
||||
/// read-only or offered as choices to pick from.
|
||||
/// `type: select` only - where the options come from; the same
|
||||
/// `ResourceSpec` mechanism a `Feature.resource` uses.
|
||||
#[serde(default)]
|
||||
pub resource: Option<ResourceSpec>,
|
||||
/// `type: select` only - which field in each resource item is that
|
||||
/// option's stable identifier, submitted as the requirement's value
|
||||
/// (or one entry of it, if `multiple`). Defaults to trying `_id`
|
||||
/// then `id` if unset.
|
||||
/// `type: select` only - which field in each item is the option's
|
||||
/// stable id. Defaults to trying `_id` then `id`.
|
||||
#[serde(default)]
|
||||
pub id_field: Option<String>,
|
||||
}
|
||||
@@ -361,14 +316,10 @@ async fn fetch_gitea_file(
|
||||
.map_err(|e| anyhow::anyhow!("reading {path}: {e}"))
|
||||
}
|
||||
|
||||
/// Fetches and parses `aggregates.yaml` from a Gitea repo's root -
|
||||
/// sibling to the pages `subdir`, not inside it, so
|
||||
/// `load_questions_from_gitea`'s "every `*.yaml` under `subdir` is a
|
||||
/// page" directory-listing convention needs no special-case exclusion.
|
||||
/// Validation (well-formed states/transitions, no duplicate buckets or
|
||||
/// event-type strings) happens inside `aggregates::parse_aggregates_yaml`
|
||||
/// itself - a malformed file fails here, before ever reaching
|
||||
/// `AppState`.
|
||||
/// Fetches and parses `aggregates.yaml` from the repo root (sibling to
|
||||
/// the pages `subdir`, so the "every yaml under subdir is a page"
|
||||
/// convention needs no exclusion). A malformed file fails here, before
|
||||
/// ever reaching `AppState`.
|
||||
#[cfg(feature = "ssr")]
|
||||
pub async fn load_aggregates_from_gitea(
|
||||
repo_url: &str,
|
||||
|
||||
+3
-6
@@ -1,9 +1,6 @@
|
||||
//! Plain NATS event publishing - ported from dodrenett's lib/nats.ts
|
||||
//! (AnswerSubmitted). Deliberately just a publish, no stream/consumer
|
||||
//! bookkeeping here: whoever reads `portal.answers.submitted` today is a
|
||||
//! person watching the subject; a future LLM-driven follow-up poster
|
||||
//! reads the exact same subject and shape, so this doesn't need to
|
||||
//! change shape when that shows up, only gain a second subscriber.
|
||||
//! Plain NATS publish of every submission/decision - fire and forget,
|
||||
//! no stream bookkeeping. The n8n automations subscribe to
|
||||
//! `portal.answers.submitted` and gate on `question_id`/`alternative`.
|
||||
#![cfg(feature = "ssr")]
|
||||
|
||||
pub mod store;
|
||||
|
||||
+16
-40
@@ -1,21 +1,11 @@
|
||||
//! 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.
|
||||
//! Durable, replayable event log behind `crate::aggregates` - the
|
||||
//! source of truth the `answers::Answer` KV records only project.
|
||||
//!
|
||||
//! Subject scheme is `events.<aggregate_type>.<aggregate_id>`,
|
||||
//! deliberately *not* `events.<type>.<id>.<eventType>` like dodrenett -
|
||||
//! dropping the event type from the subject means
|
||||
//! `expected_last_subject_sequence` gives real optimistic concurrency
|
||||
//! across an aggregate's *entire* history, not just same-typed events.
|
||||
//! The cost: no cheap subject-wildcard query for "every `invited` event
|
||||
//! across all applicants" - nobody needs that today, and it's a filter
|
||||
//! over `load_events`'s output if it's ever needed.
|
||||
//! Subject scheme is `events.<aggregate_type>.<aggregate_id>` with no
|
||||
//! event type in the subject: that's what lets
|
||||
//! `expected_last_subject_sequence` give optimistic concurrency across
|
||||
//! an aggregate's entire history, at the cost of subject-wildcard
|
||||
//! queries by event type (filter `load_events` output instead).
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
@@ -40,10 +30,8 @@ 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.
|
||||
/// Idempotent - safe to call on every boot. `max_age` deliberately
|
||||
/// unlimited: this is the durable source of truth.
|
||||
pub async fn ensure_stream(
|
||||
js: &async_nats::jetstream::Context,
|
||||
) -> anyhow::Result<async_nats::jetstream::stream::Stream> {
|
||||
@@ -58,17 +46,10 @@ pub async fn ensure_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.
|
||||
/// sequence makes this a compare-and-swap: a racing writer with a stale
|
||||
/// sequence gets `PublishError` instead of silently winning. Returns
|
||||
/// the raw `PublishError` so callers can distinguish a CAS conflict
|
||||
/// (`WrongLastSequence`) from other failures.
|
||||
pub async fn append_event(
|
||||
js: &async_nats::jetstream::Context,
|
||||
aggregate_type: &str,
|
||||
@@ -88,9 +69,6 @@ pub async fn append_event(
|
||||
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()
|
||||
@@ -105,11 +83,9 @@ pub async fn append_event(
|
||||
.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.
|
||||
/// Loads one aggregate's events in JetStream append order, via an
|
||||
/// ephemeral pull consumer scoped to its subject - appropriate at this
|
||||
/// scale (single-digit events per aggregate).
|
||||
pub async fn load_events(
|
||||
js: &async_nats::jetstream::Context,
|
||||
aggregate_type: &str,
|
||||
|
||||
+3
-5
@@ -24,11 +24,9 @@ pub struct Garage {
|
||||
}
|
||||
|
||||
impl Garage {
|
||||
/// Builds a client from `GARAGE_S3_ENDPOINT`/`GARAGE_ACCESS_KEY`/
|
||||
/// `GARAGE_SECRET_KEY`/`GARAGE_UPLOADS_BUCKET` - same secret names
|
||||
/// dodrenett's own Woodpecker pipeline already uses. `None` (not an
|
||||
/// error) when unset, since uploads are optional: everything else
|
||||
/// works without Garage configured.
|
||||
/// Builds a client from the `GARAGE_*` env vars. `None` (not an
|
||||
/// error) when unset - uploads are optional, everything else works
|
||||
/// without Garage configured.
|
||||
pub fn from_env() -> Option<Self> {
|
||||
let endpoint = std::env::var("GARAGE_S3_ENDPOINT")
|
||||
.unwrap_or_else(|_| "http://garage:3900".to_string());
|
||||
|
||||
Reference in New Issue
Block a user