systemprompt_database/projection/mod.rs
1//! Store, don't compute: the data paradigm and the queue that carries it.
2//!
3//! Every analytics-shaped read in the system follows one rule. The answer a
4//! page, report or policy needs was known once — when the row it summarises
5//! was written — and is stored then, so no reader ever re-derives it from
6//! the raw history. The cost of a read is then the size of the answer, never
7//! the size of the ledger behind it.
8//!
9//! The vocabulary, from raw to read:
10//!
11//! 1. **Ledger** — the append-only raw tables the gateway, the hook plane, the
12//! tool plane and the governance spine write (`ai_requests`,
13//! `ai_request_payloads`, `ai_request_messages`, `mcp_tool_executions`,
14//! `plugin_usage_events`, `governance_decisions`, `logs`, …). Read by
15//! exactly two things: projectors, and single-entity detail queries (`WHERE
16//! pk = $1`, no aggregate).
17//! 2. **Stamped fact** — a pure function of the row and what is in hand when it
18//! is written, stored as a column in the same transaction by the table's
19//! owner (`has_offered_tools`, `is_builtin`, `skill_ref`).
20//! 3. **Counter** — an additive aggregate the writer keeps current through a
21//! `FOR EACH STATEMENT … REFERENCING NEW TABLE` trigger (never a row
22//! trigger: a row trigger's cost is the fan-out that turned a 3,644-row
23//! update into 27 minutes).
24//! 4. **Projection** — one denormalised row per entity (`<entity>_facts`),
25//! maintained by `refresh_<projection>(keys TEXT[])` and drained from the
26//! generic queue in this module.
27//! 5. **Rollup** — `<subject>_rollup_hourly(bucket, dims…)` of additive
28//! measures, upserted at write time. Hourly, because analytics windows are
29//! arbitrary timestamps; percentiles, distinct counts and top-N substrings
30//! are not additive and keep reading raw rows for their window.
31//!
32//! **Every fact has four parts, all in the base schema**: the structure, the
33//! projector or stamp, the statement trigger (stamp directly when the
34//! transition table alone determines the value, else queue the keys) and the
35//! drainer job. An upgrade migration is the same DDL text plus one
36//! `INSERT INTO projection_backlog SELECT '<projection>', <key> FROM …`.
37//! Nothing else: a fresh install stamps migrations without running them, so
38//! anything a migration computes exists only on upgraded databases.
39//!
40//! **NULL means pending.** A reader never falls back to the ledger expression
41//! (`COALESCE(fact, <ledger expr>)`); it shows the fact as pending and
42//! [`depth`] says how many are. Two static gates keep this true: the
43//! ledger-reads gate over repository SQL, and `no_backfill_after_floor` over
44//! migrations, built on [`crate::backfill_statements`].
45//!
46//! ## The queue
47//!
48//! `projection_backlog(projection, key, queued_at, attempts)` is one table
49//! for every projection. Writers queue keys ([`queue`] or the SQL
50//! `queue_projection_backlog`); a drainer job calls [`drain`] with the
51//! projection's refresh, which takes a batch with `FOR UPDATE SKIP LOCKED`,
52//! refreshes it and deletes the rows in one transaction. A failed batch is
53//! rolled back and re-queued one attempt dearer; a key past its retry budget
54//! stays queued for an operator and never blocks the rest. A statement that
55//! keeps a `refresh_*(keys TEXT[])` SQL function can use [`drain_with_sql`]
56//! and name it.
57//!
58//! Copyright (c) systemprompt.io — Business Source License 1.1.
59//! See <https://systemprompt.io> for licensing details.
60
61use std::time::{Duration, Instant};
62
63use sqlx::{PgConnection, PgExecutor, PgPool};
64use systemprompt_traits::RepositoryError;
65use tracing::warn;
66
67use crate::services::BoxFuture;
68
69/// How a drainer tick went. `remaining` is the queue depth when it stopped,
70/// so a job can report "partly drained" without another query.
71#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
72pub struct DrainReport {
73 pub keys: u64,
74 pub batches: u32,
75 pub remaining: i64,
76 pub elapsed: Duration,
77 pub requeued: u64,
78}
79
80/// A projection's refresh: given a connection inside the drain transaction
81/// and the keys taken, rebuild them and return how many rows were written.
82pub type RefreshFuture<'c> = BoxFuture<'c, Result<u64, RepositoryError>>;
83
84pub async fn queue<'e, E>(
85 executor: E,
86 projection: &str,
87 keys: &[String],
88) -> Result<i64, RepositoryError>
89where
90 E: PgExecutor<'e>,
91{
92 let queued: i64 = sqlx::query_scalar("SELECT queue_projection_backlog($1, $2)")
93 .bind(projection)
94 .bind(keys)
95 .fetch_one(executor)
96 .await?;
97 Ok(queued)
98}
99
100pub async fn depth<'e, E>(executor: E, projection: &str) -> Result<i64, RepositoryError>
101where
102 E: PgExecutor<'e>,
103{
104 let depth: i64 = sqlx::query_scalar("SELECT projection_backlog_depth($1)")
105 .bind(projection)
106 .fetch_one(executor)
107 .await?;
108 Ok(depth)
109}
110
111pub async fn drain<F>(
112 pool: &PgPool,
113 projection: &str,
114 batch: i32,
115 budget: Duration,
116 refresh: F,
117) -> Result<DrainReport, RepositoryError>
118where
119 F: for<'c> Fn(&'c mut PgConnection, Vec<String>) -> RefreshFuture<'c> + Send + Sync,
120{
121 let started = Instant::now();
122 let mut report = DrainReport::default();
123 while started.elapsed() < budget {
124 let mut tx = pool.begin().await?;
125 let keys: Vec<String> = sqlx::query_scalar("SELECT take_projection_backlog($1, $2)")
126 .bind(projection)
127 .bind(batch)
128 .fetch_one(&mut *tx)
129 .await?;
130 if keys.is_empty() {
131 tx.rollback().await?;
132 break;
133 }
134 let taken = u64::try_from(keys.len()).unwrap_or(u64::MAX);
135 match refresh(&mut tx, keys.clone()).await {
136 Ok(_) => {
137 tx.commit().await?;
138 report.keys = report.keys.saturating_add(taken);
139 report.batches = report.batches.saturating_add(1);
140 },
141 Err(error) => {
142 tx.rollback().await?;
143 let requeued: i64 = sqlx::query_scalar("SELECT requeue_projection_backlog($1, $2)")
144 .bind(projection)
145 .bind(&keys)
146 .fetch_one(pool)
147 .await?;
148 report.requeued = u64::try_from(requeued).unwrap_or(0);
149 warn!(
150 projection,
151 keys = keys.len(),
152 error = %error,
153 "projection refresh failed; batch re-queued one attempt dearer"
154 );
155 break;
156 },
157 }
158 }
159 report.remaining = depth(pool, projection).await?;
160 report.elapsed = started.elapsed();
161 Ok(report)
162}
163
164pub async fn drain_with_sql(
165 pool: &PgPool,
166 projection: &str,
167 batch: i32,
168 budget: Duration,
169 projector: &str,
170) -> Result<DrainReport, RepositoryError> {
171 let statement = projector_statement(projector)?;
172 let statement = statement.as_str();
173 drain(pool, projection, batch, budget, move |conn, keys| {
174 let statement = statement.to_owned();
175 Box::pin(async move {
176 // Why: Postgres — the only dynamic part is the projector's name,
177 // checked as a bare identifier by `projector_statement`; the keys
178 // are a bind parameter.
179 let written: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(statement.as_str()))
180 .bind(&keys)
181 .fetch_one(conn)
182 .await?;
183 Ok(u64::try_from(written).unwrap_or(0))
184 })
185 })
186 .await
187}
188
189fn projector_statement(projector: &str) -> Result<String, RepositoryError> {
190 let mut chars = projector.chars();
191 let leads = chars
192 .next()
193 .is_some_and(|c| c.is_ascii_lowercase() || c == '_');
194 let rest = chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_');
195 if !(leads && rest) {
196 return Err(RepositoryError::InvalidArgument {
197 field: "projector",
198 reason: format!("{projector:?} is not a bare lower-case identifier"),
199 });
200 }
201 Ok(format!("SELECT {projector}($1)"))
202}