A political conference and discussion platform, in Rust and Dioxus
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599//! Every live subscription the app opens, as TYPED operations.//!//! These used to be format! strings. Nothing checked them: not the compiler,//! which cannot read GraphQL, and not a test, which would only have asserted the//! same string back. The first thing that checked one was the server, in front of//! a user — twice in one evening. A filter that carried its own braces went out//! as `where: { {_or: …} }` and every reader of the feed silently stopped getting//! updates, and an aggregate over a table with no timestamp would have been//! rejected the same way.//!//! Built with cynic against `graphql/schema.graphql`, so a wrong field name, a//! wrong argument or a filter on a column that does not exist is a compile error//! here rather than a refused subscription there. The `where` clauses are the//! same typed input objects the queries use, so a filter can no longer be//! malformed at all.
use super::*;
// --- Streaming cursor ---
/// Which column a stream advances on, and from where.////// `updated_at` is the only usable cursor on `nodes`: it is trigger-maintained,/// so it moves on an edit as well as an insert. `created_at` would miss every/// update, and no other table this app subscribes to has either.#[derive(cynic::InputObject, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "nodes_stream_cursor_value_input")]pub struct NodesStreamCursorValue { #[cynic(rename = "updatedAt", skip_serializing_if = "Option::is_none")] pub updated_at: Option<Timestamptz>,}
#[derive(cynic::InputObject, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "nodes_stream_cursor_input")]pub struct NodesStreamCursor { #[cynic(rename = "initial_value")] pub initial_value: NodesStreamCursorValue, #[cynic(skip_serializing_if = "Option::is_none")] pub ordering: Option<CursorOrdering>,}
#[derive(cynic::Enum, Clone, Copy, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "cursor_ordering")]pub enum CursorOrdering { #[cynic(rename = "ASC")] Asc, #[cynic(rename = "DESC")] Desc,}
#[derive(cynic::QueryVariables, Debug)]pub struct StreamVariables { pub where_clause: NodesBoolExp, pub cursor: Vec<NodesStreamCursor>, pub batch: i32,}
// --- The rows a stream carries ---
/// Just the id: the caller fetches the rows properly (the feed).#[derive(cynic::QueryFragment, Debug, Clone)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "nodes")]pub struct StreamedId { pub id: Uuid,}
/// Whose child changed, so a watcher can tell whether it was its own.#[derive(cynic::QueryFragment, Debug, Clone)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "nodes")]pub struct StreamedParent { pub parent_id: Option<Uuid>,}
/// A node's live state (a poll opening or closing).#[derive(cynic::QueryFragment, Debug, Clone)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "nodes")]pub struct StreamedState { pub id: Uuid, pub mutable: bool,}
/// A painted cell.#[derive(cynic::QueryFragment, Debug, Clone)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "nodes")]pub struct StreamedCell { pub key: String, pub data: Option<Jsonb>, /// Who painted it and when, so a cell that arrives while you are looking can /// be attributed without re-reading the board. pub owner_id: Option<Uuid>, pub updated_at: Option<Timestamptz>,}
// --- The subscriptions ---
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "StreamVariables")]pub struct IdStream { #[cynic(rename = "nodes_stream")] #[arguments(batch_size: $batch, cursor: $cursor, where: $where_clause)] pub nodes_stream: Vec<StreamedId>,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "StreamVariables")]pub struct ParentStream { #[cynic(rename = "nodes_stream")] #[arguments(batch_size: $batch, cursor: $cursor, where: $where_clause)] pub nodes_stream: Vec<StreamedParent>,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "StreamVariables")]pub struct StateStream { #[cynic(rename = "nodes_stream")] #[arguments(batch_size: $batch, cursor: $cursor, where: $where_clause)] pub nodes_stream: Vec<StreamedState>,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "StreamVariables")]pub struct CellStream { #[cynic(rename = "nodes_stream")] #[arguments(batch_size: $batch, cursor: $cursor, where: $where_clause)] pub nodes_stream: Vec<StreamedCell>,}
/// A query and its variables, ready for the subscription hub.////// The hub speaks the wire protocol, not cynic, so an operation is flattened/// here — but it was BUILT from the schema, which is the whole point.pub struct Wire { pub query: String, pub variables: serde_json::Value,}
#[expect( clippy::needless_pass_by_value, reason = "consuming the operation matches how every caller hands it off; a reference would force callers to keep a dead value alive")]fn wire<Q, V: serde::Serialize>(op: cynic::StreamingOperation<Q, V>) -> Wire { // A StreamingOperation keeps its parts private and serialises to exactly the // wire shape, `{query, variables}`, which is what the hub sends. let body = serde_json::to_value(&op).unwrap_or(serde_json::Value::Null); Wire { query: body .get("query") .and_then(|q| q.as_str()) .unwrap_or_default() .to_string(), variables: body .get("variables") .cloned() .unwrap_or(serde_json::Value::Null), }}
/// The cursor every stream starts from: "everything after this moment".fn from(since: &str) -> Vec<NodesStreamCursor> { vec![NodesStreamCursor { initial_value: NodesStreamCursorValue { updated_at: Some(Timestamptz(since.to_string())), }, ordering: Some(CursorOrdering::Asc), }]}
/// Ids of nodes matching `where_clause`, as they change.pub fn id_stream(where_clause: NodesBoolExp, since: &str, batch: i32) -> Wire { use cynic::SubscriptionBuilder; wire(IdStream::build(StreamVariables { where_clause, cursor: from(since), batch, }))}
/// Which parent's children changed, for a watcher that only cares about its own.pub fn parent_stream(where_clause: NodesBoolExp, since: &str, batch: i32) -> Wire { use cynic::SubscriptionBuilder; wire(ParentStream::build(StreamVariables { where_clause, cursor: from(since), batch, }))}
/// One node's `mutable`, as it changes.pub fn state_stream(where_clause: NodesBoolExp, since: &str) -> Wire { use cynic::SubscriptionBuilder; wire(StateStream::build(StreamVariables { where_clause, cursor: from(since), batch: 10, }))}
/// Cells painted on a canvas.pub fn cell_stream(where_clause: NodesBoolExp, since: &str) -> Wire { use cynic::SubscriptionBuilder; wire(CellStream::build(StreamVariables { where_clause, cursor: from(since), batch: 200, }))}
#[cfg(test)]mod tests { use super::*;
/// The filter is a typed input object, so the double-wrapped `where` that /// broke the feed cannot be expressed: there is no string to wrap. #[test] fn a_stream_carries_its_filter_as_a_variable() { let w = id_stream(NodesBoolExp::default(), "2026-01-01T00:00:00Z", 100); assert!(w.query.contains("nodes_stream"), "{}", w.query); assert!( w.query.contains("$whereClause"), "the filter travels as a variable: {}", w.query ); assert!(!w.query.contains("{ {"), "{}", w.query); // ...and every variable it declares is actually sent. let vars = w.variables.as_object().expect("variables are an object"); for key in ["whereClause", "cursor", "batch"] { assert!(vars.contains_key(key), "missing {key} in {vars:?}"); assert!( w.query.contains(&format!("${key}")), "undeclared {key}: {}", w.query ); } }
/// The cursor is ascending on `updated_at`, which is what makes a stream /// deliver edits and not only insertions. #[test] fn the_cursor_is_ascending_on_updated_at() { let w = cell_stream(NodesBoolExp::default(), "2026-01-01T00:00:00Z"); let cursor = &w.variables["cursor"][0]; assert_eq!(cursor["initial_value"]["updatedAt"], "2026-01-01T00:00:00Z"); assert_eq!(cursor["ordering"], "ASC"); }
/// Hasura folds subscribers into one cohort only when query AND variables /// are byte-identical, and that folding is the signal's whole point. #[test] fn the_context_signal_is_identical_for_every_view() { let ctx = "c732d24a-36b6-41aa-8fdb-4db3a1537b14"; let a = context_changed(ctx); let b = context_changed(ctx); assert_eq!(a.query, b.query); assert_eq!(a.variables, b.variables); assert!(a.query.contains("contextTouch"), "{}", a.query); assert!(a.query.contains("seq"), "{}", a.query); assert!(a.query.contains("$whereClause"), "{}", a.query); assert_eq!(a.variables["whereClause"]["contextId"]["_eq"], ctx); }}
// --- Change tokens (for the tables a stream cannot cover) ---
/// The newest timestamp among the matching rows, beside their count.////// Together these move on an insert, an edit and a delete, which is what makes a/// two-field aggregate a complete change token for `nodes` — and 101 bytes rather/// than the rows themselves.#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "nodes_max_fields")]pub struct NodesMaxUpdated { pub updated_at: Option<Timestamptz>,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "nodes_aggregate_fields")]pub struct NodesChangeFields { pub count: i32, pub max: Option<NodesMaxUpdated>,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "nodes_aggregate")]pub struct NodesChangeAggregate { pub aggregate: Option<NodesChangeFields>,}
#[derive(cynic::QueryVariables, Debug)]pub struct NodesWhereOnly { pub where_clause: NodesBoolExp,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "NodesWhereOnly")]pub struct NodesChanged { #[cynic(rename = "nodesAggregate")] #[arguments(where: $where_clause)] pub nodes_aggregate: NodesChangeAggregate,}
/// Something under `where_clause` changed. Carries no rows.pub fn nodes_changed_typed(where_clause: NodesBoolExp) -> Wire { use cynic::SubscriptionBuilder; wire(NodesChanged::build(NodesWhereOnly { where_clause }))}
// --- The context change signal: one row, one probe (migration 0026) ---
/// The payload is a sequence that strictly increases on every write in the/// context, so the result always differs and the push always fires. `count`/// or a timestamp could repeat; `seq` cannot.#[derive(cynic::QueryFragment, Debug)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "context_touch")]pub struct ContextTouchRow { pub seq: i64,}
#[derive(cynic::InputObject, Debug, Default)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "context_touch_bool_exp")]pub struct ContextTouchBoolExp { #[cynic(rename = "contextId", skip_serializing_if = "Option::is_none")] pub context_id: Option<UuidComparisonExp>,}
#[derive(cynic::QueryVariables, Debug)]pub struct ContextTouchWhereOnly { pub where_clause: ContextTouchBoolExp,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "ContextTouchWhereOnly")]pub struct ContextChanged { #[cynic(rename = "contextTouch")] #[arguments(where: $where_clause)] pub context_touch: Vec<ContextTouchRow>,}
/// Something in this context changed. Carries no rows.////// Every view of a context must build this with the SAME context id and/// nothing else: identical query + identical variables is what lets Hasura/// fold all of a user's views (and tabs) into one cohort, whose poll reads one/// row and checks one permission instead of aggregating over the subtree per/// view. That folding is the entire point of the signal; do not add/// per-view parameters to it.pub fn context_changed(context_id: &str) -> Wire { use cynic::SubscriptionBuilder; wire(ContextChanged::build(ContextTouchWhereOnly { where_clause: ContextTouchBoolExp { context_id: Some(UuidComparisonExp { eq: Some(Uuid(context_id.to_string())), ..Default::default() }), }, }))}
/// The change signal for a node, preferring its context's one-row token and/// falling back to the node-scoped aggregate for the rare node outside any/// context (where no touch row exists and the signal would never fire).pub fn node_changed(context_id: Option<&str>, fallback: NodesBoolExp) -> Wire { match context_id { Some(ctx) => context_changed(ctx), None => nodes_changed_typed(fallback), }}
// --- relations and members: no cursor, so a live query and not a stream ---
#[derive(cynic::QueryFragment, Debug)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "relations")]pub struct WatchedRelation { pub node_id: Option<Uuid>, /// Selected as well as the id because a subscription fires when its RESULT /// changes: these rows are upserted, and the name is what tells two of them /// apart when one is replaced. pub name: String,}
#[derive(cynic::QueryVariables, Debug)]pub struct RelationsWhereOnly { pub where_clause: RelationsBoolExp,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "RelationsWhereOnly")]pub struct RelationsChanged { #[arguments(where: $where_clause)] pub relations: Vec<WatchedRelation>,}
/// What the chair has pointed the room at, as it changes.pub fn relations_changed(where_clause: RelationsBoolExp) -> Wire { use cynic::SubscriptionBuilder; wire(RelationsChanged::build(RelationsWhereOnly { where_clause }))}
#[derive(cynic::QueryFragment, Debug)]#[cynic(schema_path = "graphql/schema.graphql", graphql_type = "members")]pub struct WatchedMember { pub id: Uuid, /// `accepted` and `active` as well as the id: accepting an invitation is an /// UPDATE, and a selection of ids alone never noticed it. pub accepted: bool, pub active: bool,}
#[derive(cynic::QueryVariables, Debug)]pub struct MembersWhereOnly { pub where_clause: MembersBoolExp,}
#[derive(cynic::QueryFragment, Debug)]#[cynic( schema_path = "graphql/schema.graphql", graphql_type = "subscription_root", variables = "MembersWhereOnly")]pub struct MembersChanged { #[arguments(where: $where_clause)] pub members: Vec<WatchedMember>,}
/// This person's memberships and invitations, as they change.pub fn members_changed(where_clause: MembersBoolExp) -> Wire { use cynic::SubscriptionBuilder; wire(MembersChanged::build(MembersWhereOnly { where_clause }))}
// --- Filters, spelled once ---//// Each of these used to be a format! string at a call site. As values they are// checked by the compiler, cannot be composed wrongly, and read the same// wherever they appear.
/// Children of a node.pub fn children_of(parent_id: &str) -> NodesBoolExp { NodesBoolExp { parent_id: Some(UuidComparisonExp { eq: Some(Uuid(parent_id.to_string())), ..Default::default() }), ..Default::default() }}
/// Children of a node, of one kind.pub fn children_of_mime(parent_id: &str, mime: &str) -> NodesBoolExp { NodesBoolExp { and: Some(vec![children_of(parent_id), of_mime(mime)]), ..Default::default() }}
/// One kind of node.pub fn of_mime(mime: &str) -> NodesBoolExp { NodesBoolExp { mime_id: Some(StringComparisonExp { eq: Some(mime.to_string()), ..Default::default() }), ..Default::default() }}
/// One node, by id.pub fn node_is(id: &str) -> NodesBoolExp { NodesBoolExp { id: Some(UuidComparisonExp { eq: Some(Uuid(id.to_string())), ..Default::default() }), ..Default::default() }}
/// Everything of one kind in a context — or, lacking a context, under one node.////// The fallback matters: a comment on a node outside any context still has to/// wake its own thread.pub fn in_context_or_under(context_id: Option<&str>, node_id: &str, mime: &str) -> NodesBoolExp { let scope = match context_id { Some(ctx) => NodesBoolExp { context_id: Some(UuidComparisonExp { eq: Some(Uuid(ctx.to_string())), ..Default::default() }), ..Default::default() }, None => children_of(node_id), }; NodesBoolExp { and: Some(vec![scope, of_mime(mime)]), ..Default::default() }}
/// A named relation on a context (the projector's active node, and friends).pub fn relation_named(parent_id: &str, name: &str) -> RelationsBoolExp { RelationsBoolExp { parent_id: Some(UuidComparisonExp { eq: Some(Uuid(parent_id.to_string())), ..Default::default() }), name: Some(StringComparisonExp { eq: Some(name.to_string()), ..Default::default() }), }}
/// Several named relations on a context, watched together.pub fn relations_named(parent_id: &str, names: &[&str]) -> RelationsBoolExp { RelationsBoolExp { parent_id: Some(UuidComparisonExp { eq: Some(Uuid(parent_id.to_string())), ..Default::default() }), name: Some(StringComparisonExp { in_: Some(names.iter().map(|n| n.to_string()).collect()), ..Default::default() }), }}
/// Relations whose name matches a pattern (the projector's `focus:<anchor>`).pub fn relations_like(parent_id: &str, pattern: &str) -> RelationsBoolExp { RelationsBoolExp { parent_id: Some(UuidComparisonExp { eq: Some(Uuid(parent_id.to_string())), ..Default::default() }), name: Some(StringComparisonExp { ilike: Some(pattern.to_string()), ..Default::default() }), }}
/// This person's memberships, wherever they are.pub fn memberships_of(user_id: &str) -> MembersBoolExp { MembersBoolExp { node_id: Some(UuidComparisonExp { eq: Some(Uuid(user_id.to_string())), ..Default::default() }), ..Default::default() }}