uhhm/portal
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 89 Wiki Activity

A person stays in dialogue with their case

Browse Source 
Five things a building authority needed and portal could not say,
found by building one and replaying its cases through the engine.

The one that mattered: a submitter could only ever act on a record
nobody had touched. `access.rs` pinned a self_transition's from-state
to `open`, though the runtime had always passed the real state and
asked. So the applicant written to for a missing drawing could not
send it, and nobody could withdraw once anything had happened.
`self_transition` now takes `from: [state, ...]` and compiles one rule
per state; omitted it still means the initial state alone, which is
what every unsubscribe link already meant.

Then the four beside it. A record can belong to another record
(`type: record`, `of:`), with `where: { field, equals }` on a listing
as the other half, so an application's page lists the objections that
name it. A public view can be narrowed to the states worth showing,
so a register that is public by law is public about what was decided
and not about what was withdrawn. A state can grant a group, so an
approval that means membership needs no second step. And a state can
be moved by a clock (`after: { days, to }`, src/deadlines.rs) - held
to the same declared edge as a desk button, firing the same mail,
publishing the same event.

Also: the README is a README again, and everything that was crowding
it is in docs/ - architecture, content reference, state machines,
access, operations.

95 tests, up from 82.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bendik Aagaard Lynghaug
2026-09-25 21:35:25 +02:00
co-authored by Claude Opus 5
parent 26c1762dfa
commit f18d1a5181
19 changed files with 1417 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/access.rs
+51 -10
View File
@@ -221,16 +221,20 @@ impl Policy {
}
}
if let Some(st) = &alt.self_transition {
let from = crate::answers::OPEN_STATE;
emit(
format!("self:{}:{from}>{}", st.bucket, st.to), "the submitter, by their link and email".to_string(), "self_transition",
format!("bucket {}", st.bucket), format!("{from} -> {}", st.to),
format!(
"permit(principal is Submitter, action == Action::\"self_transition\", resource == Bucket::{}) when {{ context.from == {} && context.to == {} && principal.email == context.stored_email }};",
quote(&st.bucket), quote(from), quote(&st.to)
),
&mut text, &mut rules,
);
// One rule per state the move may be made from: a
// person stays in dialogue with their case for as
// long as it runs, and `from` says how long that is.
for from in &st.from {
emit(
format!("self:{}:{from}>{}", st.bucket, st.to), "the submitter, by their link and email".to_string(), "self_transition",
format!("bucket {}", st.bucket), format!("{from} -> {}", st.to),
format!(
"permit(principal is Submitter, action == Action::\"self_transition\", resource == Bucket::{}) when {{ context.from == {} && context.to == {} && principal.email == context.stored_email }};",
quote(&st.bucket), quote(from), quote(&st.to)
),
&mut text, &mut rules,
);
}
}
if let (Some(invite), Some(g)) = (&alt.invite, &q.qualifies) {
emit(
@@ -536,6 +540,43 @@ mod tests {
assert!(!p.decide(&user(&["owners"]), Action::SelfTransition, &bucket, ctx("a@b.c")).allow);
}
// The applicant written to for a missing drawing sends it from
// `incomplete`, and may withdraw at any point up to the signature.
// Before `from`, every one of those moves was compiled as if it
// could only be made on the day the case arrived.
#[test]
fn a_submitter_stays_in_dialogue_for_as_long_as_the_case_runs() {
let files = vec![(
"index.yaml".to_string(),
"name: Home\nalternatives:\n - name: Apply\n record_as: cases\n - name: Send the papers\n self_transition: { bucket: cases, to: complete, from: [open, incomplete], label: Send }\n - name: Withdraw\n self_transition: { bucket: cases, to: withdrawn, from: [open, incomplete, complete], label: Withdraw }\n".to_string(),
)];
let questions = build_questions(&files).unwrap();
let aggregates = parse_aggregates_yaml(
"aggregates:\n - bucket: cases\n initial: open\n states:\n open: { event: a }\n incomplete: { event: b }\n complete: { event: c }\n withdrawn: { event: d }\n transitions:\n open: [incomplete, complete, withdrawn]\n incomplete: [complete, withdrawn]\n complete: [withdrawn]\n",
)
.unwrap();
let p = Policy::from_content(&questions, &aggregates);
assert_eq!(p.validate(), Ok(()), "{}", p.text);
let bucket = Resource::Bucket { name: "cases".into() };
let me = Principal::Submitter { record: "r1".into(), email: "a@b.c".into() };
let ctx = |from: &str, to: &str| json!({ "from": from, "to": to, "stored_email": "a@b.c" });
assert!(p.decide(&me, Action::SelfTransition, &bucket, ctx("incomplete", "complete")).allow);
assert!(p.decide(&me, Action::SelfTransition, &bucket, ctx("complete", "withdrawn")).allow);
// And only where it was said: the paperwork cannot be sent
// after the case is already withdrawn.
assert!(!p.decide(&me, Action::SelfTransition, &bucket, ctx("withdrawn", "complete")).allow);
assert!(!p.decide(&me, Action::SelfTransition, &bucket, ctx("complete", "complete")).allow);
}
// Omitted, `from` is the initial state alone - what every
// unsubscribe link written before this existed already meant.
#[test]
fn a_move_that_does_not_say_where_from_is_a_one_shot_on_arrival() {
let p = site();
assert!(p.rules.iter().any(|r| r.id == "self:subscribers:open>unsubscribed"));
assert_eq!(p.rules.iter().filter(|r| r.id.starts_with("self:subscribers:")).count(), 1);
}
#[test]
fn invites_and_automation() {
let p = site();
src/aggregates/mod.rs
+59
View File
@@ -41,6 +41,16 @@ pub struct AggregateSchema {
/// that enters it, so content decides when a person hears
/// something and the mailer never has to know a site.
pub mail_for_state: HashMap<String, MailSpec>,
/// What a state grants the person whose record entered it: a group
/// they become a member of. A firm approved to file is a member of
/// the approved firms from that moment, and the decision that made
/// it so is the same one the desk already pressed - nobody should
/// have to remember to invite them afterwards.
pub grants_for_state: HashMap<String, String>,
/// The move a clock makes rather than a person. A hearing that gets
/// no answer in six weeks has to go on without one, and nothing in
/// a building is going to press that button.
pub after_for_state: HashMap<String, AfterSpec>,
}
/// A mail to send when a record enters a state. `to` is `submitter`
@@ -64,6 +74,19 @@ impl MailSpec {
}
}
/// A move made by a clock: `after: { days: 42, to: no_answer }` on the
/// state a record is waiting in. The wait is measured from when the
/// record entered that state, and `to` must be an edge the same schema
/// already declares - a deadline may only do what a person could have
/// done. Days, because every deadline a public body keeps is in days
/// and a shorter unit would invite this to be used as a job queue.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AfterSpec {
pub days: u32,
pub to: String,
}
impl AggregateSchema {
pub fn has_state(&self, state: &str) -> bool {
self.event_for_state.contains_key(state)
@@ -73,6 +96,14 @@ impl AggregateSchema {
self.mail_for_state.get(state)
}
pub fn grants_for(&self, state: &str) -> Option<&str> {
self.grants_for_state.get(state).map(String::as_str)
}
pub fn after_for(&self, state: &str) -> Option<&AfterSpec> {
self.after_for_state.get(state)
}
pub fn allowed(&self, state: &str) -> &[String] {
self.transitions
.get(state)
@@ -86,6 +117,10 @@ struct RawState {
event: String,
#[serde(default)]
mail: Option<MailSpec>,
#[serde(default)]
grants: Option<String>,
#[serde(default)]
after: Option<AfterSpec>,
}
#[derive(Debug, serde::Deserialize)]
@@ -156,7 +191,29 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, Aggreg
}
let mut mail_for_state = HashMap::new();
let mut grants_for_state = HashMap::new();
let mut after_for_state = HashMap::new();
for (state, raw_state) in raw_schema.states {
if let Some(group) = raw_state.grants {
if group.trim().is_empty() {
anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: grants needs a group");
}
grants_for_state.insert(state.clone(), group);
}
if let Some(after) = raw_state.after {
if after.days == 0 {
anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: after.days must be at least one");
}
// A clock may only make a move a person could have
// made, so it is held to the same graph.
if !raw_schema.transitions.get(&state).is_some_and(|tos| tos.contains(&after.to)) {
anyhow::bail!(
"aggregates.yaml: bucket {bucket:?}: state {state:?}: after.to {:?} is not a declared transition from it",
after.to
);
}
after_for_state.insert(state.clone(), after);
}
if let Some(mail) = raw_state.mail {
if mail.subject.trim().is_empty() || mail.body.trim().is_empty() {
anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: mail needs a subject and a body");
@@ -181,6 +238,8 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, Aggreg
transitions: raw_schema.transitions,
attended_by: raw_schema.attended_by,
mail_for_state,
grants_for_state,
after_for_state,
},
);
}
src/answers.rs
+9 -1
View File
@@ -290,6 +290,7 @@ async fn apply_transition(
.await
.map_err(|e| ServerFnError::new(e.to_string()))?;
crate::mail::on_state(state, bucket, &item.item_id, &item.to, &answer.responses).await;
crate::people::on_state(state, bucket, &item.item_id, &item.to, &answer.responses).await;
// The decision extends the DAG as a child of the answer's own
// submission hash, published like any other answer.
@@ -370,7 +371,13 @@ pub async fn self_transition_answer(
serde_json::json!({ "from": answer.state, "to": self_transition.to, "stored_email": stored_email }),
);
if !decision.allow {
return Err(ServerFnError::new(if answer.state == OPEN_STATE { "not found" } else { "already processed" }));
// Told apart by which half of the rule failed. If the record is
// somewhere this button may be fired from, the email was wrong,
// and the answer must not confirm that the record exists. If it
// is not, the case has moved past this move and saying so is no
// longer a way to learn anything about it.
let in_reach = self_transition.from.iter().any(|s| s == &answer.state);
return Err(ServerFnError::new(if in_reach { "not found" } else { "already processed" }));
}
let decided_ms = chrono::Utc::now().timestamp_millis();
@@ -401,6 +408,7 @@ pub async fn self_transition_answer(
.await
.map_err(|e| ServerFnError::new(e.to_string()))?;
crate::mail::on_state(&state, &self_transition.bucket, &item_id, &self_transition.to, &answer.responses).await;
crate::people::on_state(&state, &self_transition.bucket, &item_id, &self_transition.to, &answer.responses).await;
let parent_hashes = vec![item_id.clone()];
let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload_for_event, decided_ms);
src/app.rs
+31 -1
View File
@@ -1451,7 +1451,12 @@ fn AlternativeCard(
.into_any();
}
if req.kind == "hidden" {
// `record` carries the case this one belongs to.
// Nobody types a chain hash, so it rides in on the
// link and is carried, not asked for — the label
// row would only ask a person to confirm a string
// that means nothing to them.
if req.kind == "hidden" || req.kind == "record" {
// A carrier value only — no label row, or a
// preset key reads as a second visible field.
return view! {
@@ -2402,6 +2407,30 @@ pub async fn submit_answer(
let mut responses: serde_json::Value = serde_json::from_str(&responses_json)
.map_err(|e| ServerFnError::new(format!("invalid responses: {e}")))?;
// A record that belongs to another record: check the other one is
// there before this one is written, so a reference in a stored
// record is one that resolved at least once. An optional field
// left empty points at nothing and is nothing to check.
for requirement in alt.features.iter().flat_map(|f| &f.requirements) {
let (Some(of), true) = (&requirement.of, requirement.kind == "record") else {
continue;
};
let given = responses.get(&requirement.name).and_then(|v| v.as_str()).map(str::trim).unwrap_or_default();
if given.is_empty() {
if requirement.optional {
continue;
}
return Err(ServerFnError::new(format!("this answer is missing the {} it is about", requirement.name)));
}
let found = match state.jetstream.get_key_value(of).await {
Ok(store) => store.get(given).await.ok().flatten().is_some(),
Err(_) => false,
};
if !found {
return Err(ServerFnError::new(format!("the {} this is about is not on file", requirement.name)));
}
}
// An invite: the policy says whether this group may extend one,
// Kanidm does the onboarding, and what came back (the account, the
// one-time link) is recorded with the answer for the inviter's desk
@@ -2494,6 +2523,7 @@ pub async fn submit_answer(
// state sends, it sends now.
if let Some(initial) = state.aggregates.load().get(bucket).map(|s| s.initial.clone()) {
crate::mail::on_state(&state, bucket, &chain_hash, &initial, &responses).await;
crate::people::on_state(&state, bucket, &chain_hash, &initial, &responses).await;
}
}
src/content/mod.rs
+138
View File
@@ -344,6 +344,19 @@ pub struct SelfTransition {
pub bucket: String,
pub to: String,
pub label: String,
/// The states this move may be made from. A person is in dialogue
/// with a case for as long as it runs, not only on the day they
/// sent it: an applicant written to for a missing drawing answers
/// from `incomplete`, and may withdraw from anywhere along the
/// way. Defaults to the initial state alone, which is what every
/// one-shot "unsubscribe" link wants and what this was before it
/// could be said.
#[serde(default = "initial_state_only")]
pub from: Vec<String>,
}
fn initial_state_only() -> Vec<String> {
vec![crate::answers::OPEN_STATE.to_string()]
}
#[derive(Clone, Debug, Serialize, Deserialize)]
@@ -391,6 +404,21 @@ pub struct ResourceSpec {
/// `requires_group` regardless of this flag.
#[serde(default)]
pub public: bool,
/// The record states this resource shows, for a `Kv` source; empty
/// shows every one. A register that is public by law is public
/// about what was decided, not about what was withdrawn or turned
/// down, and the desk that works the same bucket is a different
/// spec and still sees all of it.
#[serde(default)]
pub states: Vec<String>,
/// Show only the records whose `where.field` holds `where.equals`,
/// for a `Kv` source. With a `type: record` field on the other
/// side, this is the half that reads: an application's page lists
/// the objections that name it, because each objection stored the
/// application's id in a field and this asks for exactly those.
/// `equals` takes the page's own `{key}` like a resource key does.
#[serde(default, rename = "where")]
pub filter: Option<FieldIs>,
/// 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
@@ -505,6 +533,18 @@ pub struct Requirement {
/// page's key into the answer.
#[serde(default)]
pub value: Option<String>,
/// `type: record` only - the bucket the referenced record lives in.
/// This is how one case belongs to another: an objection names the
/// application it objects to, an appeal the decision it appeals.
/// The stored value is the parent's record id, and a submission
/// naming a record that is not there is refused, so a reference in
/// a record is one that resolved at least once.
///
/// A person never types one of these: the id is a chain hash. It
/// arrives the way it was given out - `value: "{key}"` on a dynamic
/// page, so the link in the letter carries the case it is about.
#[serde(default)]
pub of: Option<String>,
/// `type: gesture` / `type: voice` - ws(s):// URL of a redoal-relay instance
/// the drawing widget connects to for live echoes of similar
/// strokes. Absent means the widget works offline: the drawn path
@@ -518,6 +558,16 @@ pub struct Requirement {
/// value. When the sibling changes, `resource` is fetched with the
/// value as a parameter and the result replaces this field's value
/// (an empty sibling never fetches, and never clears an edit).
/// "Show the ones whose `field` is `equals`." The whole of what a
/// listing may ask about a record's own contents: one field, one
/// value, compared as strings. Anything more is a query language, and
/// a jq filter is already there for shaping what comes back.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct FieldIs {
pub field: String,
pub equals: String,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Bind {
/// The sibling requirement (same alternative) to watch.
@@ -1075,6 +1125,94 @@ alternatives:
std::collections::HashMap::from([(question.id.clone(), question)])
}
fn question_with_record(field_yaml: &str) -> std::collections::HashMap<String, Question> {
let question: Question = serde_yaml::from_str(&format!(
r#"
id: /object
name: Object to it
alternatives:
- name: The register
features:
- name: All
resource:
public: true
source: {{ kind: kv, bucket: things }}
- name: Object
record_as: things
features:
- name: Yours
requirements:
- name: what
{field_yaml}
"#
))
.unwrap();
std::collections::HashMap::from([(question.id.clone(), question)])
}
// An objection belongs to the application it objects to. The
// bucket it points at has to be one this repo declares, or the
// reference could never resolve.
#[test]
fn a_record_can_belong_to_another_record() {
let good = question_with_record(" - { name: case, type: record, of: things }");
validate_questions(&good, &schema_fixture()).unwrap();
let nowhere = question_with_record(" - { name: case, type: record, of: nothing_declared }");
let err = validate_questions(&nowhere, &schema_fixture()).unwrap_err();
assert!(err.to_string().contains("has no aggregate declared"), "{err}");
let unsaid = question_with_record(" - { name: case, type: record }");
let err = validate_questions(&unsaid, &schema_fixture()).unwrap_err();
assert!(err.to_string().contains("must say `of:"), "{err}");
// `of` means one thing, so it belongs to one type.
let stray = question_with_record(" - { name: case, of: things }");
let err = validate_questions(&stray, &schema_fixture()).unwrap_err();
assert!(err.to_string().contains("only a type: record field has"), "{err}");
// The id arrives on the link; it is not looked up.
let bound = question_with_record(" - name: case\n type: record\n of: things\n bind:\n field: what\n resource:\n source: { kind: url, url: \"https://x.example/{p}\" }");
let err = validate_questions(&bound, &schema_fixture()).unwrap_err();
assert!(err.to_string().contains("the link it came in on carries the id"), "{err}");
}
// A state that grants a group makes an account for someone, so the
// form that fills the bucket has to have asked who they are.
#[test]
fn a_state_that_grants_a_group_needs_a_name_and_an_address_on_file() {
let granting = crate::aggregates::parse_aggregates_yaml(
"aggregates:\n - bucket: things\n initial: open\n states:\n open: { event: opened }\n middle: { event: advanced }\n done: { event: finished, grants: approved_firms }\n transitions:\n open: [middle]\n middle: [done]\n",
)
.unwrap();
let asks = question_with_record(" - { name: name }\n - { name: email, type: email }");
validate_questions(&asks, &granting).unwrap();
let no_name = question_with_record(" - { name: email, type: email }");
let err = validate_questions(&no_name, &granting).unwrap_err();
assert!(err.to_string().contains("grant a group"), "{err}");
// The same form is fine against a schema that grants nothing.
validate_questions(&no_name, &schema_fixture()).unwrap();
}
// A deadline may only make a move a person could have made.
#[test]
fn a_states_deadline_is_held_to_the_same_graph_as_a_button() {
let ok = crate::aggregates::parse_aggregates_yaml(
"aggregates:\n - bucket: hearings\n initial: open\n states:\n open: { event: opened, after: { days: 42, to: no_answer } }\n no_answer: { event: gave_up }\n transitions:\n open: [no_answer]\n",
)
.unwrap();
assert_eq!(ok["hearings"].after_for("open").unwrap().days, 42);
let undeclared = crate::aggregates::parse_aggregates_yaml(
"aggregates:\n - bucket: hearings\n initial: open\n states:\n open: { event: opened, after: { days: 42, to: no_answer } }\n no_answer: { event: gave_up }\n transitions: {}\n",
);
assert!(undeclared.unwrap_err().to_string().contains("not a declared transition"));
}
#[test]
fn declared_edge_passes() {
let questions = question_with_transitions(
src/content/validate.rs
+93
View File
@@ -115,6 +115,38 @@ pub fn validate_questions(
}
}
}
// A state that grants a group makes an account for the person the
// record is about, so the record has to know who they are. Same
// shape as an invite's own rule, one step further back: the form
// that fills the bucket is where the name and the address have to
// be asked for, because by the time a desk presses approve there is
// nobody left to ask.
for (bucket, schema) in aggregates {
let mut granting: Vec<&String> = schema.grants_for_state.keys().collect();
if granting.is_empty() {
continue;
}
granting.sort();
for question in questions.values() {
for alternative in question.alternatives.iter().filter(|a| a.record_as.as_deref() == Some(bucket.as_str())) {
let required = |name: &str, kind: &str| {
alternative
.features
.iter()
.flat_map(|f| &f.requirements)
.any(|r| r.name == name && r.kind == kind && !r.optional)
};
if !required("email", "email") || !required("name", "text") {
anyhow::bail!(
"question {:?} alternative {:?}: records into {bucket:?}, whose state(s) {:?} grant a group, so it must collect a required `name` and a required `email` field of type email",
question.id, alternative.name, granting
);
}
}
}
}
for question in questions.values() {
if let Some(event) = &question.event {
if question.is_dynamic() {
@@ -210,6 +242,30 @@ pub fn validate_questions(
question.id, alternative.name, st.to, st.bucket
);
}
if st.from.is_empty() {
anyhow::bail!(
"question {:?} alternative {:?}: self_transition.from is empty, so the move can never be made",
question.id, alternative.name
);
}
for from in &st.from {
if !schema.has_state(from) {
anyhow::bail!(
"question {:?} alternative {:?}: self_transition.from {:?} is not a declared state for bucket {:?}",
question.id, alternative.name, from, st.bucket
);
}
// The policy would otherwise permit a move the
// engine refuses: the reader of the content
// should hear about that here, not from a
// person who clicked the button.
if !schema.allowed(from).iter().any(|s| s == &st.to) {
anyhow::bail!(
"question {:?} alternative {:?}: self_transition {:?} -> {:?} is not a declared edge for bucket {:?}",
question.id, alternative.name, from, st.to, st.bucket
);
}
}
}
}
// Bind targets resolve across the whole alternative, same
@@ -275,6 +331,35 @@ pub fn validate_questions(
}
}
}
// One case belonging to another. The bucket has to
// be one this repo declares, or the reference could
// never resolve; and it is the only thing `of`
// means, so it belongs to no other type.
if requirement.kind == "record" {
let Some(of) = &requirement.of else {
anyhow::bail!(
"question {:?} alternative {:?} feature {:?}: requirement {:?} is type: record - it must say `of: <bucket>`, the bucket the record it points at lives in",
question.id, alternative.name, feature.name, requirement.name
);
};
if !aggregates.contains_key(of) {
anyhow::bail!(
"question {:?} alternative {:?} feature {:?}: requirement {:?} points at bucket {of:?}, which has no aggregate declared",
question.id, alternative.name, feature.name, requirement.name
);
}
if requirement.resource.is_some() || requirement.bind.is_some() {
anyhow::bail!(
"question {:?} alternative {:?} feature {:?}: requirement {:?} is type: record - the link it came in on carries the id, it can't also load one from a resource or bind",
question.id, alternative.name, feature.name, requirement.name
);
}
} else if requirement.of.is_some() {
anyhow::bail!(
"question {:?} alternative {:?} feature {:?}: requirement {:?} has `of`, which only a type: record field has",
question.id, alternative.name, feature.name, requirement.name
);
}
if let Some(bind) = &requirement.bind {
if bind.field == requirement.name
|| !sibling_names.contains(bind.field.as_str())
@@ -295,6 +380,14 @@ pub fn validate_questions(
let Some(schema) = aggregates.get(bucket) else {
continue;
};
for state in &resource.states {
if !schema.has_state(state) {
anyhow::bail!(
"question {:?} alternative {:?} feature {:?}: states {:?} is not a declared state for bucket {:?}",
question.id, alternative.name, feature.name, state, bucket
);
}
}
for transition in &resource.transitions {
if !schema.has_state(&transition.from) {
anyhow::bail!(
src/deadlines.rs
+196
View File
@@ -0,0 +1,196 @@
//! The moves a clock makes rather than a person.
//!
//! A state in `aggregates.yaml` may carry `after: { days: 42, to:
//! no_answer }`: when a record has waited there that long, portal makes
//! the move itself. Every deadline a public body keeps is of this
//! shape - a hearing goes out and the authority has six weeks to
//! answer, and when it does not the case has to go on without one, with
//! nobody in the building in a position to press that button.
//!
//! A deadline may only do what a person could have done: `after.to` is
//! held to the same declared edge as a desk button (see
//! `parse_aggregates_yaml`), the move goes through the same
//! `aggregates::transition`, the same mail fires, and it lands on
//! `portal.answers.submitted` like any other decision. What it is not
//! is a job queue: the unit is days, the tick is an hour, and a record
//! that has waited long enough is moved once, because after the move it
//! is no longer in a state that has a deadline.
use crate::answers::Answer;
use crate::events::{emit_answer_submitted, AnswerSubmitted};
use crate::server::AppState;
/// Stamped on the record and on the transition event in place of a
/// username, so a desk and an n8n workflow can both tell a decision
/// nobody made from one somebody did.
pub const BY: &str = "deadline";
const DAY_MS: i64 = 24 * 60 * 60 * 1000;
pub async fn run(state: AppState) {
let mut tick = tokio::time::interval(std::time::Duration::from_secs(60 * 60));
loop {
tick.tick().await;
if let Err(e) = sweep(&state).await {
tracing::warn!(error = %e, "deadline sweep failed");
}
}
}
/// When a record entered the state it is in now. A record nobody has
/// moved has been waiting since it was sent.
fn waiting_since(answer: &Answer) -> i64 {
answer.decided_ms.unwrap_or(answer.submitted_ms)
}
/// Whether a record in `state` has waited out its deadline by `now_ms`.
pub fn due(schema: &crate::aggregates::AggregateSchema, answer: &Answer, now_ms: i64) -> Option<String> {
let after = schema.after_for(&answer.state)?;
let deadline = waiting_since(answer) + i64::from(after.days) * DAY_MS;
(now_ms >= deadline).then(|| after.to.clone())
}
/// One pass over every bucket that has a deadline on any of its states.
pub async fn sweep(state: &AppState) -> anyhow::Result<()> {
let now_ms = chrono::Utc::now().timestamp_millis();
let aggregates = state.aggregates.load();
for schema in aggregates.values() {
if schema.after_for_state.is_empty() {
continue;
}
// A bucket is created on first write, so one that has never
// been written to is an ordinary empty bucket, not an error.
let Ok(store) = state.jetstream.get_key_value(&schema.bucket).await else {
continue;
};
let keys: Vec<String> = {
use futures::TryStreamExt;
match store.keys().await {
Ok(keys) => keys.try_collect().await.unwrap_or_default(),
Err(_) => continue,
}
};
for id in keys {
let Ok(Some(bytes)) = store.get(&id).await else { continue };
let Ok(mut answer) = serde_json::from_slice::<Answer>(&bytes) else { continue };
let Some(target) = due(schema, &answer, now_ms) else { continue };
let payload = serde_json::json!({ "to": target, "item": id, "by": BY });
if let Err(e) = crate::aggregates::transition(
&state.jetstream, schema, &id, &target, payload.clone(), now_ms,
)
.await
{
// A record whose events are gone, or one another writer
// moved between the read and the write: neither is this
// sweep's to fix, and the next tick sees it again.
tracing::warn!(bucket = %schema.bucket, id, target, "a deadline could not be made: {e}");
continue;
}
answer.state = target.clone();
answer.decided_ms = Some(now_ms);
answer.decided_by = Some(BY.to_string());
if let Ok(data) = serde_json::to_vec(&answer) {
let _ = store.put(&id, data.into()).await;
}
crate::mail::on_state(state, &schema.bucket, &id, &target, &answer.responses).await;
crate::people::on_state(state, &schema.bucket, &id, &target, &answer.responses).await;
let parent_hashes = vec![id.clone()];
let chain_hash = crate::chain::hash_node(&answer.question_id, &parent_hashes, &payload, now_ms);
if let Err(e) = emit_answer_submitted(
&state.nats,
&AnswerSubmitted {
chain_hash,
parent_hashes,
question_id: answer.question_id.clone(),
alternative: target.clone(),
responses: payload,
timestamp_ms: now_ms,
},
)
.await
{
tracing::warn!(bucket = %schema.bucket, id, "the deadline was made but not published: {e}");
}
tracing::info!(bucket = %schema.bucket, id, target, "a deadline passed and the case went on");
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn schema(yaml: &str) -> crate::aggregates::AggregateSchema {
crate::aggregates::parse_aggregates_yaml(yaml).unwrap().into_values().next().unwrap()
}
const HEARINGS: &str = r#"
aggregates:
- bucket: hearings
initial: open
states:
open: { event: received, after: { days: 42, to: no_answer } }
answered: { event: answered }
no_answer: { event: no_answer }
transitions:
open: [answered, no_answer]
"#;
fn waiting(state: &str, since_ms: i64) -> Answer {
Answer {
id: "r".into(), question_id: "/q".into(), alternative: "a".into(),
responses: serde_json::json!({}), state: state.into(),
submitted_ms: since_ms, decided_ms: None, decided_by: None, notes: vec![],
}
}
#[test]
fn a_deadline_comes_due_the_day_it_says_and_not_before() {
let s = schema(HEARINGS);
let sent = 1_000_000_000_000;
let answer = waiting("open", sent);
assert_eq!(due(&s, &answer, sent), None);
assert_eq!(due(&s, &answer, sent + 41 * DAY_MS), None);
assert_eq!(due(&s, &answer, sent + 42 * DAY_MS), Some("no_answer".into()));
}
// The six weeks run from the last move, not from the day the case
// was opened: a record someone touched has been waiting since then.
#[test]
fn the_wait_is_measured_from_the_last_move() {
let s = schema(HEARINGS);
let sent = 1_000_000_000_000;
let mut answer = waiting("open", sent);
answer.decided_ms = Some(sent + 40 * DAY_MS);
assert_eq!(due(&s, &answer, sent + 42 * DAY_MS), None);
assert_eq!(due(&s, &answer, sent + 82 * DAY_MS), Some("no_answer".into()));
}
#[test]
fn a_state_with_no_deadline_waits_forever() {
let s = schema(HEARINGS);
let answer = waiting("answered", 0);
assert_eq!(due(&s, &answer, i64::MAX / 2), None);
}
// A clock may only make a move a person could have made, so a
// deadline onto an undeclared edge is refused when content loads,
// not when the hour comes round.
#[test]
fn a_deadline_onto_an_undeclared_edge_is_refused_on_load() {
let bad = HEARINGS.replace("to: no_answer }", "to: answered }").replace("open: [answered, no_answer]", "open: [no_answer]");
let err = crate::aggregates::parse_aggregates_yaml(&bad).unwrap_err().to_string();
assert!(err.contains("not a declared transition"), "{err}");
}
#[test]
fn a_deadline_of_no_days_is_refused() {
let bad = HEARINGS.replace("days: 42", "days: 0");
let err = crate::aggregates::parse_aggregates_yaml(&bad).unwrap_err().to_string();
assert!(err.contains("at least one"), "{err}");
}
}
src/lib.rs
+3
View File
@@ -13,6 +13,9 @@ pub mod aggregates;
#[cfg(feature = "ssr")]
pub mod announce;
#[cfg(feature = "ssr")]
pub mod deadlines;
#[cfg(feature = "ssr")]
pub mod http;
src/main.rs
+1
View File
@@ -107,6 +107,7 @@ async fn main() -> anyhow::Result<()> {
// Announced pages: keep their desk records current, close ended
// windows (see announce.rs).
tokio::spawn(portal::announce::run(state.clone()));
tokio::spawn(portal::deadlines::run(state.clone()));
// Replies to case mail, matched to their record by whatever reads
// the mailbox (uhhm/gdo over JMAP) and turned into notes here. A
src/people.rs
+62
View File
@@ -189,6 +189,68 @@ pub fn username_for(email: &str) -> String {
}
}
/// The person a record is about, for a state that grants a group.
/// Both are required by the lint of any form that fills such a bucket
/// (see `validate_questions`), so a record missing either is one that
/// predates the rule, and is reported rather than guessed at.
pub fn who_for(responses: &serde_json::Value) -> Option<(String, String)> {
let field = |name: &str| {
responses
.get(name)
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(String::from)
};
let email = field("email").filter(|e| e.contains('@'))?;
Some((field("name")?, email))
}
/// A record entered `state`: if that state grants a group, put the
/// person the record is about in it. The decision the desk already
/// pressed is the whole ceremony - an approval that means membership
/// should not also need somebody to remember to invite them.
///
/// Best-effort, exactly like the mail beside it: the transition is
/// committed either way, and an instance with no Kanidm configured
/// simply grants nothing and says so once per occurrence.
pub async fn on_state(
state: &crate::server::AppState,
bucket: &str,
id: &str,
to_state: &str,
responses: &serde_json::Value,
) {
let Some(group) = state
.aggregates
.load()
.get(bucket)
.and_then(|s| s.grants_for(to_state).map(str::to_string))
else {
return;
};
let Some(kanidm) = Kanidm::from_env() else {
tracing::warn!(bucket, id, to_state, group, "this state grants a group, but invites are not configured on this instance (KANIDM_API_TOKEN) - nobody was added");
return;
};
let Some((name, email)) = who_for(responses) else {
tracing::warn!(bucket, id, to_state, group, "this state grants a group, but the record carries no name and email - nobody was added");
return;
};
let invited = match kanidm.invite(&name, &email, &group).await {
Ok(invited) => invited,
Err(e) => {
tracing::error!(bucket, id, to_state, group, "granting the group failed: {e}");
return;
}
};
if let Ok(bytes) = serde_json::to_vec(&invited) {
let _ = state.nats.publish(INVITED_SUBJECT, bytes.into()).await;
}
crate::mail::on_invite(state, &invited).await;
tracing::info!(bucket, id, to_state, group, user = %invited.username, created = invited.created, "the decision granted a group");
}
#[cfg(test)]
mod tests {
use super::*;
src/resource.rs
+107 -2
View File
@@ -188,6 +188,50 @@ async fn authorize_resource(
}
}
/// Whether a stored record passes the resource's `states` filter. No
/// filter shows everything, which is what every resource written
/// before this existed means. A record with no `state` at all is not
/// an aggregate and is left alone.
#[cfg(feature = "ssr")]
fn shows_state(resource: &crate::content::ResourceSpec, value: &serde_json::Value) -> bool {
if resource.states.is_empty() {
return true;
}
match value.get("state").and_then(|s| s.as_str()) {
Some(state) => resource.states.iter().any(|s| s == state),
None => true,
}
}
/// Whether a stored record passes the resource's `where` filter. The
/// field is looked for in the record's `responses`, which is where a
/// form's answers live, and then on the record itself, so a listing
/// can be narrowed by what was submitted or by what the runtime
/// stamped on it. A missing field never matches: "everything that did
/// not say" is not something a page should be able to ask for by
/// accident.
#[cfg(feature = "ssr")]
fn matches_filter(
resource: &crate::content::ResourceSpec,
value: &serde_json::Value,
params: &std::collections::HashMap<String, String>,
) -> bool {
let Some(filter) = &resource.filter else {
return true;
};
let wanted = crate::content::template_action(&filter.equals, &serde_json::json!(params))
.unwrap_or_else(|| filter.equals.clone());
let found = value
.get("responses")
.and_then(|r| r.get(&filter.field))
.or_else(|| value.get(&filter.field));
match found {
Some(serde_json::Value::String(s)) => s == &wanted,
Some(other) => other.to_string() == wanted,
None => false,
}
}
#[cfg(feature = "ssr")]
async fn fetch_resource_value(
state: &crate::server::AppState,
@@ -216,7 +260,14 @@ async fn fetch_resource_value(
.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()))?
let value: serde_json::Value =
serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string()))?;
// A state filter holds for one record as well as a
// listing, or naming the key would walk around it.
if !shows_state(resource, &value) || !matches_filter(resource, &value, params) {
return Err(ServerFnError::new("resource key not found"));
}
value
}
(None, Some(store)) => {
use futures::TryStreamExt;
@@ -231,7 +282,9 @@ async fn fetch_resource_value(
for key in keys {
if let Ok(Some(bytes)) = store.get(&key).await {
if let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes) {
items.push(value);
if shows_state(resource, &value) && matches_filter(resource, &value, params) {
items.push(value);
}
}
}
}
@@ -470,6 +523,58 @@ pub fn apply_jq(
mod tests {
use super::*;
fn spec(yaml: &str) -> crate::content::ResourceSpec {
serde_yaml::from_str(yaml).unwrap()
}
fn record(state: &str, responses: serde_json::Value) -> serde_json::Value {
serde_json::json!({ "id": "r", "state": state, "responses": responses })
}
// A register that is public by law is public about what was
// decided, not about what was withdrawn.
#[test]
fn a_listing_can_be_narrowed_to_the_states_worth_showing() {
let all = spec("source: { kind: kv, bucket: cases }\npublic: true\n");
let decided = spec("source: { kind: kv, bucket: cases }\npublic: true\nstates: [approved, refused]\n");
for state in ["approved", "refused", "withdrawn", "open"] {
assert!(shows_state(&all, &record(state, serde_json::json!({}))));
}
assert!(shows_state(&decided, &record("approved", serde_json::json!({}))));
assert!(shows_state(&decided, &record("refused", serde_json::json!({}))));
assert!(!shows_state(&decided, &record("withdrawn", serde_json::json!({}))));
// Something that is not an aggregate has no state to filter on
// and is left alone.
assert!(shows_state(&decided, &serde_json::json!({ "name": "a file" })));
}
// The other half of a record belonging to a record: the
// application's page asking for the objections that name it.
#[test]
fn a_listing_can_be_narrowed_to_the_children_of_one_record() {
let children = spec("source: { kind: kv, bucket: objections }\nrequires_group: case_officers\nwhere: { field: case, equals: \"{key}\" }\n");
let params = std::collections::HashMap::from([("key".to_string(), "abc123".to_string())]);
let mine = record("open", serde_json::json!({ "case": "abc123", "your_objection": "the sun" }));
let theirs = record("open", serde_json::json!({ "case": "def456" }));
let neither = record("open", serde_json::json!({ "your_objection": "no case named" }));
assert!(matches_filter(&children, &mine, &params));
assert!(!matches_filter(&children, &theirs, &params));
// A record that did not say is not everyone's child.
assert!(!matches_filter(&children, &neither, &params));
// No filter, no narrowing: every resource written before this
// existed still lists its whole bucket.
let plain = spec("source: { kind: kv, bucket: objections }\npublic: true\n");
assert!(matches_filter(&plain, &theirs, &params));
}
#[test]
fn a_filter_can_name_a_value_outright() {
let fixed = spec("source: { kind: kv, bucket: cases }\npublic: true\nwhere: { field: which_authority, equals: The county }\n");
let params = std::collections::HashMap::new();
assert!(matches_filter(&fixed, &record("open", serde_json::json!({ "which_authority": "The county" })), &params));
assert!(!matches_filter(&fixed, &record("open", serde_json::json!({ "which_authority": "Fire service" })), &params));
}
/// A canned, Gitea-API-shaped fixture - the same fields the real
/// `/users/{username}/starred` endpoint returns (Gitea 1.27:
/// `stars_count`, not GitHub's `stargazers_count`; `website` is ""