systemprompt_analytics/projection/
mod.rs

1//! Durable analytics projections over versioned source reporting contracts.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use sqlx::PgConnection;
9
10use crate::{AnalyticsError, Result};
11
12mod snapshot;
13mod sources;
14mod state;
15pub use snapshot::{SnapshotPage, lock_sources, write_snapshot_page};
16pub use sources::SOURCE_DEFINITIONS;
17pub use state::{
18    ProjectionStatus, RebuildState, heartbeat_rebuild, is_initialized, lock_projector,
19    lock_user_deletion, next_cutoff_revision, rebuild_state, status,
20};
21
22pub const REPORTING_CONSUMER: &str = "analytics_reporting";
23pub const REPORTING_KIND: &str = "reporting.row";
24pub const REPORTING_VERSION: u32 = 1;
25pub const REPORTING_STATE_SEED: &str =
26    "INSERT INTO analytics_projection_state(singleton) VALUES (TRUE) ON CONFLICT DO NOTHING";
27
28#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
29#[serde(rename_all = "snake_case")]
30pub enum ReportingSource {
31    Users,
32    UserSessions,
33    AgentTasks,
34    TaskMessages,
35    UserContexts,
36    AiRequests,
37    McpToolExecutions,
38    MarkdownContent,
39    AnalyticsEvents,
40}
41
42#[derive(Debug, Clone, Deserialize, Serialize)]
43#[serde(deny_unknown_fields)]
44pub struct ReportingRow {
45    pub source: ReportingSource,
46    pub key: String,
47    pub revision: i64,
48    pub deleted: bool,
49    pub row: Value,
50}
51
52#[derive(Debug, Clone, Copy)]
53pub struct SourceDefinition {
54    pub source: ReportingSource,
55    pub table: &'static str,
56    pub view: &'static str,
57    pub target: &'static str,
58    pub key: &'static str,
59    pub key_type: &'static str,
60    pub columns: &'static [&'static str],
61}
62
63impl ReportingSource {
64    pub fn definition(self) -> &'static SourceDefinition {
65        &SOURCE_DEFINITIONS[match self {
66            Self::Users => 0,
67            Self::UserSessions => 1,
68            Self::AgentTasks => 2,
69            Self::TaskMessages => 3,
70            Self::UserContexts => 4,
71            Self::AiRequests => 5,
72            Self::McpToolExecutions => 6,
73            Self::MarkdownContent => 7,
74            Self::AnalyticsEvents => 8,
75        }]
76    }
77}
78
79impl ReportingRow {
80    fn validate(&self) -> Result<()> {
81        if self.key.is_empty() || self.revision < 0 {
82            return Err(AnalyticsError::invalid_argument(
83                "invalid reporting key or revision",
84            ));
85        }
86        if self.deleted {
87            if !self.row.is_null() {
88                return Err(AnalyticsError::invalid_argument(
89                    "deleted reporting fact must have a null row",
90                ));
91            }
92            return Ok(());
93        }
94        let definition = self.source.definition();
95        let object = self
96            .row
97            .as_object()
98            .ok_or_else(|| AnalyticsError::invalid_argument("reporting row must be an object"))?;
99        if object.len() != definition.columns.len()
100            || definition
101                .columns
102                .iter()
103                .any(|column| !object.contains_key(*column))
104        {
105            return Err(AnalyticsError::invalid_argument(
106                "reporting row does not match its versioned column contract",
107            ));
108        }
109        let key = &object[definition.key];
110        let matches = key.as_str().is_some_and(|key| key == self.key)
111            || key.as_i64().is_some_and(|key| key.to_string() == self.key);
112        if !matches {
113            return Err(AnalyticsError::invalid_argument(
114                "reporting row key does not match envelope",
115            ));
116        }
117        Ok(())
118    }
119}
120
121/// Applies reporting facts inside the caller's transaction and projector lock.
122///
123/// `begin_rebuild` opens a new generation with its cutoff and the in-progress
124/// marker; callers hold the source locks, so every fact minted before the
125/// cutoff has committed and every later one is above it and will be applied.
126/// `clear_targets` empties every report table and the revision guard for that
127/// generation, in a separate transaction so no source lock is held while the
128/// targets are truncated. `finish_rebuild` marks the baseline complete,
129/// leaving the cutoff as recorded by `begin_rebuild` or raised since by a
130/// privacy compaction — lowering it would replay facts that compaction has
131/// already delivered.
132#[derive(Debug, Clone, Copy)]
133pub struct ReportingProjector;
134
135impl ReportingProjector {
136    pub async fn begin_rebuild(connection: &mut PgConnection, cutoff: i64) -> Result<i64> {
137        Ok(sqlx::query_scalar!(
138            r#"UPDATE analytics_projection_state
139               SET generation = generation + 1, initialized = FALSE, cutoff_revision = $1,
140                   rebuild_started_at = NOW(), rebuild_heartbeat_at = NOW(),
141                   rebuild_source = NULL, rebuild_rows = 0
142               WHERE singleton RETURNING generation AS "generation!""#,
143            cutoff
144        )
145        .fetch_one(&mut *connection)
146        .await?)
147    }
148
149    pub async fn clear_targets(connection: &mut PgConnection, generation: i64) -> Result<()> {
150        Self::verify_generation(connection, generation).await?;
151        let targets = SOURCE_DEFINITIONS
152            .iter()
153            .map(|definition| definition.target)
154            .collect::<Vec<_>>()
155            .join(", ");
156        sqlx::query(sqlx::AssertSqlSafe(format!("TRUNCATE TABLE {targets}")))
157            .execute(&mut *connection)
158            .await?;
159        sqlx::query!("DELETE FROM analytics_projection_revisions")
160            .execute(&mut *connection)
161            .await?;
162        Ok(())
163    }
164
165    pub async fn finish_rebuild(connection: &mut PgConnection, generation: i64) -> Result<()> {
166        let result = sqlx::query!(
167            "UPDATE analytics_projection_state
168             SET initialized = TRUE, rebuilt_at = NOW(), rebuild_started_at = NULL,
169                 rebuild_heartbeat_at = NULL, rebuild_source = NULL
170             WHERE singleton AND generation = $1 AND NOT initialized",
171            generation
172        )
173        .execute(&mut *connection)
174        .await?;
175        if result.rows_affected() != 1 {
176            return Err(AnalyticsError::rebuild_superseded());
177        }
178        Ok(())
179    }
180
181    async fn verify_generation(connection: &mut PgConnection, generation: i64) -> Result<()> {
182        let state = rebuild_state(connection).await?;
183        if state.generation != generation || state.initialized || state.rebuild_started_at.is_none()
184        {
185            return Err(AnalyticsError::rebuild_superseded());
186        }
187        Ok(())
188    }
189
190    pub async fn apply_fact(connection: &mut PgConnection, fact: &ReportingRow) -> Result<bool> {
191        fact.validate()?;
192        let state = sqlx::query!(
193            "SELECT initialized, cutoff_revision FROM analytics_projection_state WHERE singleton FOR UPDATE"
194        )
195        .fetch_one(&mut *connection)
196        .await?;
197        let cutoff = state.cutoff_revision;
198        if !state.initialized {
199            return Err(AnalyticsError::invalid_argument(
200                "analytics projection requires a baseline snapshot",
201            ));
202        }
203        if fact.revision <= cutoff {
204            return Ok(false);
205        }
206        let accepted = sqlx::query_scalar!(
207            "INSERT INTO analytics_projection_revisions(source, entity_key, revision)
208             VALUES ($1, $2, $3)
209             ON CONFLICT(source, entity_key) DO UPDATE SET revision = EXCLUDED.revision
210             WHERE analytics_projection_revisions.revision < EXCLUDED.revision
211             RETURNING revision",
212            fact.source.definition().table,
213            &fact.key,
214            fact.revision
215        )
216        .fetch_optional(&mut *connection)
217        .await?;
218        if accepted.is_none() {
219            return Ok(false);
220        }
221        Self::write_row(connection, fact).await?;
222        Ok(true)
223    }
224
225    async fn retained(connection: &mut PgConnection, fact: &ReportingRow) -> Result<bool> {
226        Ok(sqlx::query_scalar!(
227            r#"SELECT reporting_row_retained($1, $2) AS "retained!""#,
228            fact.source.definition().table,
229            &fact.row
230        )
231        .fetch_one(connection)
232        .await?)
233    }
234
235    async fn write_row(connection: &mut PgConnection, fact: &ReportingRow) -> Result<()> {
236        let definition = fact.source.definition();
237        if fact.deleted || !Self::retained(connection, fact).await? {
238            sqlx::query(sqlx::AssertSqlSafe(format!(
239                "DELETE FROM {} WHERE {} = CAST($1 AS {})",
240                definition.target, definition.key, definition.key_type,
241            )))
242            .bind(&fact.key)
243            .execute(connection)
244            .await?;
245        } else {
246            let assignments = definition
247                .columns
248                .iter()
249                .filter(|column| **column != definition.key)
250                .map(|column| format!("{column} = EXCLUDED.{column}"))
251                .collect::<Vec<_>>()
252                .join(", ");
253            sqlx::query(sqlx::AssertSqlSafe(format!(
254                "INSERT INTO {} SELECT * FROM jsonb_populate_record(NULL::{}, $1)
255                 ON CONFLICT ({}) DO UPDATE SET {}",
256                definition.target, definition.target, definition.key, assignments,
257            )))
258            .bind(&fact.row)
259            .execute(connection)
260            .await?;
261        }
262        Ok(())
263    }
264}
systemprompt_analytics/projection/mod.rs

systemprompt_analytics/projection/
mod.rs
