systemprompt_analytics/projection/
state.rs

1//! Projection bookkeeping: the singleton state row, the projector lock, the
2//! rebuild-in-progress marker and the durable-queue lag behind it.
3//!
4//! `heartbeat_rebuild` records a page of the running rebuild, and its
5//! predicate is the fence: a forced rebuild or a privacy compaction that
6//! moved the generation, or a finished baseline, makes the write affect
7//! nothing and the caller stops.
8//!
9//! Copyright (c) systemprompt.io — Business Source License 1.1.
10//! See <https://systemprompt.io> for licensing details.
11
12use chrono::{DateTime, Utc};
13use serde::Serialize;
14use sqlx::{PgConnection, PgPool};
15
16use crate::{AnalyticsError, Result};
17
18const PROJECTOR_LOCK: i64 = 0x5350_414e_414c_5954;
19
20/// Baseline state plus the pending durable facts still owed to the projection.
21#[derive(Debug, Clone, Serialize)]
22pub struct ProjectionStatus {
23    pub initialized: bool,
24    pub generation: i64,
25    pub rebuilt_at: Option<DateTime<Utc>>,
26    pub rebuild_started_at: Option<DateTime<Utc>>,
27    pub rebuild_heartbeat_at: Option<DateTime<Utc>>,
28    pub rebuild_source: Option<String>,
29    pub rebuild_rows: i64,
30    pub pending_count: i64,
31    pub applied_last_minute: i64,
32    pub oldest_pending_at: Option<DateTime<Utc>>,
33    pub last_processed_at: Option<DateTime<Utc>>,
34}
35
36/// The state row as a rebuild sees it under the projector lock.
37#[derive(Debug, Clone, Copy)]
38pub struct RebuildState {
39    pub generation: i64,
40    pub initialized: bool,
41    pub rebuild_started_at: Option<DateTime<Utc>>,
42    pub rebuild_heartbeat_at: Option<DateTime<Utc>>,
43}
44
45pub async fn lock_projector(connection: &mut PgConnection) -> Result<()> {
46    sqlx::query!("SELECT pg_advisory_xact_lock($1)", PROJECTOR_LOCK)
47        .fetch_one(connection)
48        .await?;
49    Ok(())
50}
51
52pub async fn lock_user_deletion(connection: &mut PgConnection) -> Result<()> {
53    sqlx::query_scalar!(r#"SELECT public.lock_user_deletion_for_retention() AS "locked!""#)
54        .fetch_one(connection)
55        .await?;
56    Ok(())
57}
58
59pub async fn is_initialized(connection: &mut PgConnection) -> Result<bool> {
60    Ok(sqlx::query_scalar!(
61        r#"SELECT initialized AS "initialized!" FROM analytics_projection_state WHERE singleton"#
62    )
63    .fetch_one(connection)
64    .await?)
65}
66
67pub async fn rebuild_state(connection: &mut PgConnection) -> Result<RebuildState> {
68    Ok(sqlx::query_as!(
69        RebuildState,
70        "SELECT generation, initialized, rebuild_started_at, rebuild_heartbeat_at
71         FROM analytics_projection_state WHERE singleton"
72    )
73    .fetch_one(connection)
74    .await?)
75}
76
77pub async fn heartbeat_rebuild(
78    connection: &mut PgConnection,
79    generation: i64,
80    source: &str,
81    rows: i64,
82) -> Result<()> {
83    let result = sqlx::query!(
84        "UPDATE analytics_projection_state
85         SET rebuild_heartbeat_at = NOW(), rebuild_source = $2, rebuild_rows = rebuild_rows + $3
86         WHERE singleton AND generation = $1 AND NOT initialized AND rebuild_started_at IS NOT NULL",
87        generation,
88        source,
89        rows
90    )
91    .execute(connection)
92    .await?;
93    if result.rows_affected() != 1 {
94        return Err(AnalyticsError::rebuild_superseded());
95    }
96    Ok(())
97}
98
99pub async fn next_cutoff_revision(connection: &mut PgConnection) -> Result<i64> {
100    Ok(
101        sqlx::query_scalar!(r#"SELECT nextval('event_outbox_reporting_revision') AS "cutoff!""#)
102            .fetch_one(connection)
103            .await?,
104    )
105}
106
107pub async fn status(pool: &PgPool, consumer: &str) -> Result<ProjectionStatus> {
108    Ok(sqlx::query_as!(
109        ProjectionStatus,
110        r#"SELECT initialized, generation, rebuilt_at,
111            rebuild_started_at, rebuild_heartbeat_at, rebuild_source, rebuild_rows,
112            (SELECT COUNT(*) FROM event_outbox WHERE consumer = $1 AND processed_at IS NULL) AS "pending_count!",
113            (SELECT COUNT(*) FROM event_outbox WHERE consumer = $1 AND processed_at >= NOW() - INTERVAL '1 minute') AS "applied_last_minute!",
114            (SELECT MIN(created_at) FROM event_outbox WHERE consumer = $1 AND processed_at IS NULL) AS oldest_pending_at,
115            (SELECT MAX(processed_at) FROM event_outbox WHERE consumer = $1) AS last_processed_at
116         FROM analytics_projection_state WHERE singleton"#,
117        consumer
118    )
119    .fetch_one(pool)
120    .await?)
121}
systemprompt_analytics/projection/state.rs

systemprompt_analytics/projection/
state.rs
