systemprompt_analytics/projection/
snapshot.rs

1//! Rebuild snapshot: a share lock over every source table and a server-side
2//! cursor over each owner's reporting view.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use serde_json::Value;
8use sqlx::PgConnection;
9
10use super::{SOURCE_DEFINITIONS, SourceDefinition};
11use crate::Result;
12
13/// One owner-published reporting row read from a rebuild snapshot cursor.
14#[derive(Debug, sqlx::FromRow)]
15pub struct SnapshotRow {
16    pub entity_key: String,
17    // JSON: owner-published reporting views provide the versioned row payload.
18    pub row: Value,
19}
20
21/// Server-side cursor over one source's reporting view, held open for the
22/// rebuild transaction so the snapshot is read in bounded batches.
23#[derive(Debug, Clone, Copy)]
24pub struct SnapshotCursor {
25    definition: &'static SourceDefinition,
26}
27
28impl SnapshotCursor {
29    pub async fn lock_sources(connection: &mut PgConnection) -> Result<()> {
30        let tables = SOURCE_DEFINITIONS
31            .iter()
32            .map(|definition| definition.table)
33            .collect::<Vec<_>>()
34            .join(", ");
35        sqlx::query(sqlx::AssertSqlSafe(format!(
36            "LOCK TABLE {tables} IN SHARE MODE"
37        )))
38        .execute(connection)
39        .await?;
40        Ok(())
41    }
42
43    pub async fn open(
44        connection: &mut PgConnection,
45        definition: &'static SourceDefinition,
46    ) -> Result<Self> {
47        sqlx::query(sqlx::AssertSqlSafe(format!(
48            "DECLARE reporting_snapshot NO SCROLL CURSOR FOR SELECT entity_key, row FROM {}",
49            definition.view,
50        )))
51        .execute(connection)
52        .await?;
53        Ok(Self { definition })
54    }
55
56    pub const fn definition(&self) -> &'static SourceDefinition {
57        self.definition
58    }
59
60    pub async fn fetch(&self, connection: &mut PgConnection) -> Result<Vec<SnapshotRow>> {
61        Ok(sqlx::query_as("FETCH FORWARD 1000 FROM reporting_snapshot")
62            .fetch_all(connection)
63            .await?)
64    }
65
66    pub async fn close(self, connection: &mut PgConnection) -> Result<()> {
67        sqlx::query("CLOSE reporting_snapshot")
68            .execute(connection)
69            .await?;
70        Ok(())
71    }
72}
systemprompt_analytics/projection/snapshot.rs

systemprompt_analytics/projection/
snapshot.rs
