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 depths<'e, E>(executor: E) -> Result<Vec<(String, i64)>, RepositoryError>
112where
113 E: PgExecutor<'e>,
114{
115 let rows: Vec<(String, i64)> = sqlx::query_as(
116 "SELECT projection, COUNT(*)::BIGINT FROM projection_backlog GROUP BY projection",
117 )
118 .fetch_all(executor)
119 .await?;
120 Ok(rows)
121}
122
123pub async fn drain<F>(
124 pool: &PgPool,
125 projection: &str,
126 batch: i32,
127 budget: Duration,
128 refresh: F,
129) -> Result<DrainReport, RepositoryError>
130where
131 F: for<'c> Fn(&'c mut PgConnection, Vec<String>) -> RefreshFuture<'c> + Send + Sync,
132{
133 let started = Instant::now();
134 let mut report = DrainReport::default();
135 while started.elapsed() < budget {
136 let mut tx = pool.begin().await?;
137 let keys: Vec<String> = sqlx::query_scalar("SELECT take_projection_backlog($1, $2)")
138 .bind(projection)
139 .bind(batch)
140 .fetch_one(&mut *tx)
141 .await?;
142 if keys.is_empty() {
143 tx.rollback().await?;
144 break;
145 }
146 let taken = u64::try_from(keys.len()).unwrap_or(u64::MAX);
147 match refresh(&mut tx, keys.clone()).await {
148 Ok(_) => {
149 tx.commit().await?;
150 report.keys = report.keys.saturating_add(taken);
151 report.batches = report.batches.saturating_add(1);
152 },
153 Err(error) => {
154 tx.rollback().await?;
155 let requeued: i64 = sqlx::query_scalar("SELECT requeue_projection_backlog($1, $2)")
156 .bind(projection)
157 .bind(&keys)
158 .fetch_one(pool)
159 .await?;
160 report.requeued = u64::try_from(requeued).unwrap_or(0);
161 warn!(
162 projection,
163 keys = keys.len(),
164 error = %error,
165 "projection refresh failed; batch re-queued one attempt dearer"
166 );
167 break;
168 },
169 }
170 }
171 report.remaining = depth(pool, projection).await?;
172 report.elapsed = started.elapsed();
173 Ok(report)
174}
175
176pub async fn drain_with_sql(
177 pool: &PgPool,
178 projection: &str,
179 batch: i32,
180 budget: Duration,
181 projector: &str,
182) -> Result<DrainReport, RepositoryError> {
183 let statement = projector_statement(projector)?;
184 let statement = statement.as_str();
185 drain(pool, projection, batch, budget, move |conn, keys| {
186 let statement = statement.to_owned();
187 Box::pin(async move {
188 // Why: Postgres — the only dynamic part is the projector's name,
189 // checked as a bare identifier by `projector_statement`; the keys
190 // are a bind parameter.
191 let written: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(statement.as_str()))
192 .bind(&keys)
193 .fetch_one(conn)
194 .await?;
195 Ok(u64::try_from(written).unwrap_or(0))
196 })
197 })
198 .await
199}
200
201fn projector_statement(projector: &str) -> Result<String, RepositoryError> {
202 let mut chars = projector.chars();
203 let leads = chars
204 .next()
205 .is_some_and(|c| c.is_ascii_lowercase() || c == '_');
206 let rest = chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_');
207 if !(leads && rest) {
208 return Err(RepositoryError::InvalidArgument {
209 field: "projector",
210 reason: format!("{projector:?} is not a bare lower-case identifier"),
211 });
212 }
213 Ok(format!("SELECT {projector}($1)"))
214}