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::{SnapshotCursor, SnapshotRow};
16pub use sources::SOURCE_DEFINITIONS;
17pub use state::{
18    ProjectionStatus, is_initialized, lock_projector, lock_user_deletion, next_cutoff_revision,
19    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    AiRequestMessages,
38    McpToolExecutions,
39    MarkdownContent,
40    Logs,
41    AnalyticsEvents,
42}
43
44#[derive(Debug, Clone, Deserialize, Serialize)]
45#[serde(deny_unknown_fields)]
46pub struct ReportingRow {
47    pub source: ReportingSource,
48    pub key: String,
49    pub revision: i64,
50    pub deleted: bool,
51    pub row: Value,
52}
53
54#[derive(Debug, Clone, Copy)]
55pub struct SourceDefinition {
56    pub source: ReportingSource,
57    pub table: &'static str,
58    pub view: &'static str,
59    pub target: &'static str,
60    pub key: &'static str,
61    pub key_type: &'static str,
62    pub columns: &'static [&'static str],
63}
64
65impl ReportingSource {
66    pub fn definition(self) -> &'static SourceDefinition {
67        &SOURCE_DEFINITIONS[match self {
68            Self::Users => 0,
69            Self::UserSessions => 1,
70            Self::AgentTasks => 2,
71            Self::TaskMessages => 3,
72            Self::UserContexts => 4,
73            Self::AiRequests => 5,
74            Self::AiRequestMessages => 6,
75            Self::McpToolExecutions => 7,
76            Self::MarkdownContent => 8,
77            Self::Logs => 9,
78            Self::AnalyticsEvents => 10,
79        }]
80    }
81}
82
83impl ReportingRow {
84    fn validate(&self) -> Result<()> {
85        if self.key.is_empty() || self.revision < 0 {
86            return Err(AnalyticsError::invalid_argument(
87                "invalid reporting key or revision",
88            ));
89        }
90        if self.deleted {
91            if !self.row.is_null() {
92                return Err(AnalyticsError::invalid_argument(
93                    "deleted reporting fact must have a null row",
94                ));
95            }
96            return Ok(());
97        }
98        let definition = self.source.definition();
99        let object = self
100            .row
101            .as_object()
102            .ok_or_else(|| AnalyticsError::invalid_argument("reporting row must be an object"))?;
103        if object.len() != definition.columns.len()
104            || definition
105                .columns
106                .iter()
107                .any(|column| !object.contains_key(*column))
108        {
109            return Err(AnalyticsError::invalid_argument(
110                "reporting row does not match its versioned column contract",
111            ));
112        }
113        let key = &object[definition.key];
114        let matches = key.as_str().is_some_and(|key| key == self.key)
115            || key.as_i64().is_some_and(|key| key.to_string() == self.key);
116        if !matches {
117            return Err(AnalyticsError::invalid_argument(
118                "reporting row key does not match envelope",
119            ));
120        }
121        Ok(())
122    }
123}
124
125/// Applies reporting facts inside the caller's transaction and projector lock.
126#[derive(Debug, Clone, Copy)]
127pub struct ReportingProjector;
128
129impl ReportingProjector {
130    pub async fn begin_rebuild(connection: &mut PgConnection) -> Result<i64> {
131        let generation = sqlx::query_scalar!(
132            r#"SELECT generation + 1 AS "generation!" FROM analytics_projection_state WHERE singleton FOR UPDATE"#
133        )
134        .fetch_one(&mut *connection)
135        .await?;
136        for definition in SOURCE_DEFINITIONS {
137            sqlx::query(sqlx::AssertSqlSafe(format!(
138                "DELETE FROM {}",
139                definition.target
140            )))
141            .execute(&mut *connection)
142            .await?;
143        }
144        sqlx::query!("DELETE FROM analytics_projection_revisions")
145            .execute(&mut *connection)
146            .await?;
147        Ok(generation)
148    }
149
150    pub async fn apply_snapshot(connection: &mut PgConnection, fact: &ReportingRow) -> Result<()> {
151        fact.validate()?;
152        if fact.deleted {
153            return Err(AnalyticsError::invalid_argument(
154                "a snapshot cannot contain deleted rows",
155            ));
156        }
157        if Self::retained(connection, fact).await? {
158            Self::write_row(connection, fact).await?;
159        }
160        Ok(())
161    }
162
163    pub async fn finish_rebuild(
164        connection: &mut PgConnection,
165        generation: i64,
166        cutoff_revision: i64,
167    ) -> Result<()> {
168        if generation < 1 || cutoff_revision < 0 {
169            return Err(AnalyticsError::invalid_argument(
170                "invalid projection generation or cutoff",
171            ));
172        }
173        let result = sqlx::query!(
174            "UPDATE analytics_projection_state SET generation = $1::BIGINT, cutoff_revision = $2,
175             initialized = TRUE, rebuilt_at = NOW() WHERE singleton AND generation = $1::BIGINT - 1",
176            generation,
177            cutoff_revision
178        )
179        .execute(&mut *connection)
180        .await?;
181        if result.rows_affected() != 1 {
182            return Err(AnalyticsError::invalid_argument(
183                "projection generation changed during rebuild",
184            ));
185        }
186        Ok(())
187    }
188
189    pub async fn apply_fact(connection: &mut PgConnection, fact: &ReportingRow) -> Result<bool> {
190        fact.validate()?;
191        let state = sqlx::query!(
192            "SELECT initialized, cutoff_revision FROM analytics_projection_state WHERE singleton FOR UPDATE"
193        )
194        .fetch_one(&mut *connection)
195        .await?;
196        let cutoff = state.cutoff_revision;
197        if !state.initialized {
198            return Err(AnalyticsError::invalid_argument(
199                "analytics projection requires a baseline snapshot",
200            ));
201        }
202        if fact.revision <= cutoff {
203            return Ok(false);
204        }
205        let accepted = sqlx::query_scalar!(
206            "INSERT INTO analytics_projection_revisions(source, entity_key, revision)
207             VALUES ($1, $2, $3)
208             ON CONFLICT(source, entity_key) DO UPDATE SET revision = EXCLUDED.revision
209             WHERE analytics_projection_revisions.revision < EXCLUDED.revision
210             RETURNING revision",
211            fact.source.definition().table,
212            &fact.key,
213            fact.revision
214        )
215        .fetch_optional(&mut *connection)
216        .await?;
217        if accepted.is_none() {
218            return Ok(false);
219        }
220        Self::write_row(connection, fact).await?;
221        Ok(true)
222    }
223
224    async fn retained(connection: &mut PgConnection, fact: &ReportingRow) -> Result<bool> {
225        Ok(sqlx::query_scalar!(
226            r#"SELECT reporting_row_retained($1, $2) AS "retained!""#,
227            fact.source.definition().table,
228            &fact.row
229        )
230        .fetch_one(connection)
231        .await?)
232    }
233
234    async fn write_row(connection: &mut PgConnection, fact: &ReportingRow) -> Result<()> {
235        let definition = fact.source.definition();
236        if fact.deleted || !Self::retained(connection, fact).await? {
237            sqlx::query(sqlx::AssertSqlSafe(format!(
238                "DELETE FROM {} WHERE {} = CAST($1 AS {})",
239                definition.target, definition.key, definition.key_type,
240            )))
241            .bind(&fact.key)
242            .execute(connection)
243            .await?;
244        } else {
245            let assignments = definition
246                .columns
247                .iter()
248                .filter(|column| **column != definition.key)
249                .map(|column| format!("{column} = EXCLUDED.{column}"))
250                .collect::<Vec<_>>()
251                .join(", ");
252            sqlx::query(sqlx::AssertSqlSafe(format!(
253                "INSERT INTO {} SELECT * FROM jsonb_populate_record(NULL::{}, $1)
254                 ON CONFLICT ({}) DO UPDATE SET {}",
255                definition.target, definition.target, definition.key, assignments,
256            )))
257            .bind(&fact.row)
258            .execute(connection)
259            .await?;
260        }
261        Ok(())
262    }
263}
systemprompt_analytics/projection/mod.rs

systemprompt_analytics/projection/
mod.rs
