fraiseql_cli/commands/sources/
reader.rs

1//! The thin PostgreSQL projection behind `fraiseql sources` (#573).
2//!
3//! Reads the durable `_fraiseql_source_cursor` watermark table — the same table the
4//! runtime poller advances — so an operator can see, per source, where its cursor
5//! stands and how stale it is. Deliberately dependency-light (deadpool +
6//! tokio-postgres, mirroring the `perf` reader): all merge/format logic lives in
7//! pure functions in the parent module, so correctness is covered without a live
8//! database.
9
10use anyhow::{Context, Result};
11use deadpool_postgres::Pool;
12use fraiseql_db::postgres::PostgresTlsConfig;
13
14/// One `_fraiseql_source_cursor` row: a source's durable watermark and its
15/// staleness, computed against the database clock so there is no app↔DB skew.
16pub struct CursorRow {
17    /// The cursor key — the source name the runtime poller advances under.
18    pub source_name: String,
19    /// The opaque cursor value, or `None` if the row exists but holds SQL NULL.
20    pub value:       Option<serde_json::Value>,
21    /// The compare-and-swap generation counter.
22    pub version:     i64,
23    /// Last-advance timestamp, rendered as text (`updated_at::text`).
24    pub updated_at:  String,
25    /// Seconds since the last advance, from the database clock (the staleness/lag).
26    pub age_seconds: f64,
27}
28
29/// A live PostgreSQL connection pool for reading source cursors.
30pub struct SourceCursorReader {
31    pool: Pool,
32}
33
34impl SourceCursorReader {
35    /// Connect to `db_url` (PostgreSQL only) for cursor reads.
36    ///
37    /// # Errors
38    ///
39    /// Returns an error if `db_url` is not a PostgreSQL connection string (see
40    /// [`crate::connection::require_postgres`]), the server cannot be reached, or
41    /// it is older than PostgreSQL 18.
42    pub async fn connect(db_url: &str, tls: &PostgresTlsConfig) -> Result<Self> {
43        let pool = crate::connection::postgres_pool(db_url, "source cursor reads", tls).await?;
44        Ok(Self { pool })
45    }
46
47    /// Load every source cursor row.
48    ///
49    /// The age is computed in SQL against the database clock. `cursor_value` (JSONB)
50    /// decodes directly to a [`serde_json::Value`] (the `with-serde_json-1` codec).
51    /// Missing table ⇒ an empty result is *not* silently returned — the error
52    /// surfaces so the operator learns the source subsystem was never initialized.
53    ///
54    /// # Errors
55    ///
56    /// Returns an error if the connection or the query fails.
57    pub async fn load_cursors(&self) -> Result<Vec<CursorRow>> {
58        let client = self.pool.get().await.context("failed to acquire DB connection")?;
59        let rows = client
60            .query(
61                "SELECT source_name, cursor_value, version, updated_at::text AS updated_at_text, \
62                 EXTRACT(EPOCH FROM (now() - updated_at))::float8 AS age_seconds \
63                 FROM _fraiseql_source_cursor",
64                &[],
65            )
66            .await
67            .context(
68                "failed to read _fraiseql_source_cursor (is the source subsystem initialized?)",
69            )?;
70        Ok(rows
71            .iter()
72            .map(|row| CursorRow {
73                source_name: row.get("source_name"),
74                value:       row.get("cursor_value"),
75                version:     row.get("version"),
76                updated_at:  row.get("updated_at_text"),
77                age_seconds: row.get("age_seconds"),
78            })
79            .collect())
80    }
81}
fraiseql_cli/commands/sources/reader.rs

fraiseql_cli/commands/sources/
reader.rs
