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::{Config, ManagerConfig, Pool, RecyclingMethod, Runtime};
12use tokio_postgres::NoTls;
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 `postgres://` URL or the pool cannot be
40    /// created. (Connection failures surface lazily on the first query.)
41    pub fn connect(db_url: &str) -> Result<Self> {
42        if !db_url.starts_with("postgres") {
43            anyhow::bail!(
44                "source cursor reads require a PostgreSQL connection URL (postgres://…); got: \
45                 {db_url}"
46            );
47        }
48        let mut cfg = Config::new();
49        cfg.url = Some(db_url.to_string());
50        cfg.manager = Some(ManagerConfig {
51            recycling_method: RecyclingMethod::Fast,
52        });
53        cfg.pool = Some(deadpool_postgres::PoolConfig::new(2));
54        let pool = cfg
55            .create_pool(Some(Runtime::Tokio1), NoTls)
56            .context("failed to create PostgreSQL connection pool for source cursor reads")?;
57        Ok(Self { pool })
58    }
59
60    /// Load every source cursor row.
61    ///
62    /// The age is computed in SQL against the database clock. `cursor_value` (JSONB)
63    /// decodes directly to a [`serde_json::Value`] (the `with-serde_json-1` codec).
64    /// Missing table ⇒ an empty result is *not* silently returned — the error
65    /// surfaces so the operator learns the source subsystem was never initialized.
66    ///
67    /// # Errors
68    ///
69    /// Returns an error if the connection or the query fails.
70    pub async fn load_cursors(&self) -> Result<Vec<CursorRow>> {
71        let client = self.pool.get().await.context("failed to acquire DB connection")?;
72        let rows = client
73            .query(
74                "SELECT source_name, cursor_value, version, updated_at::text AS updated_at_text, \
75                 EXTRACT(EPOCH FROM (now() - updated_at))::float8 AS age_seconds \
76                 FROM _fraiseql_source_cursor",
77                &[],
78            )
79            .await
80            .context(
81                "failed to read _fraiseql_source_cursor (is the source subsystem initialized?)",
82            )?;
83        Ok(rows
84            .iter()
85            .map(|row| CursorRow {
86                source_name: row.get("source_name"),
87                value:       row.get("cursor_value"),
88                version:     row.get("version"),
89                updated_at:  row.get("updated_at_text"),
90                age_seconds: row.get("age_seconds"),
91            })
92            .collect())
93    }
94}
fraiseql_cli/commands/sources/reader.rs

fraiseql_cli/commands/sources/
reader.rs
