fraiseql_cli/commands/perf/
reader.rs

1//! Thin PostgreSQL reader over the change-log contract view
2//! `core.v_entity_change_log`.
3//!
4//! Deliberately minimal: it pulls a trailing window of rows into plain
5//! [`ChangeLogSample`] values and hands them to the pure analysis functions. All
6//! columns are cast to dependency-light scalar types in SQL (`text`, `int4`,
7//! `int8`, `float8`) so the reader needs no `chrono`/`time`/`uuid` decode
8//! features — `created_at` arrives as an epoch `f64` and `object_id` as text.
9//!
10//! PostgreSQL-only, mirroring the `--against-db` checks
11//! ([`crate::schema::pg_catalog`]): the change-log contract view is a PostgreSQL
12//! object and the marker lives in a `jsonb` column.
13
14use anyhow::{Context, Result};
15use deadpool_postgres::Pool;
16use fraiseql_db::postgres::PostgresTlsConfig;
17
18/// The framework-owned read-path view shipped by the change-log contract
19/// migration (`08_create_entity_change_log_contract.sql`). Trusted identifier —
20/// never interpolated from user input.
21pub const CONTRACT_VIEW: &str = "core.v_entity_change_log";
22
23/// One change-log row, projected to the columns `perf` analysis needs.
24///
25/// `duration_ms` is `NULL` for cooperative external producers (no FraiseQL
26/// request clock); `duration_calc_version` is the data-quality marker
27/// (`extra_metadata->>'duration_calc_version'`) used to refuse mixing pre-fix
28/// rows with the wall-clock-correct ones — see
29/// [`fraiseql_db::changelog::DURATION_CALC_VERSION`].
30#[derive(Debug, Clone, PartialEq)]
31pub struct ChangeLogSample {
32    /// Entity type the mutation touched (`object_type`).
33    pub object_type:           String,
34    /// `INSERT` / `UPDATE` / `DELETE` / `CUSTOM`.
35    pub modification_type:     String,
36    /// Wall-clock duration in milliseconds; `None` for non-executor producers.
37    pub duration_ms:           Option<i32>,
38    /// `extra_metadata->>'duration_calc_version'`; `None` when unmarked (legacy).
39    pub duration_calc_version: Option<i64>,
40    /// `EXTRACT(EPOCH FROM created_at)` — seconds since the Unix epoch.
41    pub created_at_epoch:      f64,
42    /// The changed entity's id (`object_id::text`); `None` when absent.
43    pub object_id:             Option<String>,
44    /// W3C trace id, when populated.
45    pub trace_id:              Option<String>,
46}
47
48/// A live PostgreSQL connection pool for reading the change-log contract view.
49pub struct PerfReader {
50    pool: Pool,
51}
52
53impl PerfReader {
54    /// Connect to `db_url` (PostgreSQL only) for change-log reads.
55    ///
56    /// # Errors
57    ///
58    /// Returns an error if `db_url` is not a PostgreSQL connection string (see
59    /// [`crate::connection::require_postgres`]), the server cannot be reached, or
60    /// it is older than PostgreSQL 18.
61    pub async fn connect(db_url: &str, tls: &PostgresTlsConfig) -> Result<Self> {
62        let pool = crate::connection::postgres_pool(db_url, "perf reads", tls).await?;
63        Ok(Self { pool })
64    }
65
66    /// The database clock as a Unix epoch in seconds — the reference the analysis
67    /// windows back from, so there is no app↔DB skew at the window boundary.
68    ///
69    /// # Errors
70    ///
71    /// Returns an error if the connection or the query fails.
72    pub async fn db_now_epoch(&self) -> Result<f64> {
73        let client = self.pool.get().await.context("failed to acquire DB connection")?;
74        let row = client
75            .query_one("SELECT EXTRACT(EPOCH FROM now())::float8 AS now_epoch", &[])
76            .await
77            .context("failed to read database clock")?;
78        Ok(row.get("now_epoch"))
79    }
80
81    /// Load every change-log sample within the trailing `window_days` window,
82    /// optionally restricted to a single `object_type`.
83    ///
84    /// The window is applied in SQL (`created_at >= now() - make_interval(days =>
85    /// $1)`) so only the rows the analysis needs cross the wire.
86    ///
87    /// # Errors
88    ///
89    /// Returns an error if the connection or the query fails.
90    pub async fn load_samples(
91        &self,
92        window_days: i32,
93        object_type: Option<&str>,
94    ) -> Result<Vec<ChangeLogSample>> {
95        let client = self.pool.get().await.context("failed to acquire DB connection")?;
96
97        // All columns cast to dependency-light scalars; the marker is pulled out
98        // of the view's `data` JSONB (it is not a top-level view column).
99        let base = format!(
100            "SELECT object_type, modification_type, duration_ms, \
101             EXTRACT(EPOCH FROM created_at)::float8 AS created_at_epoch, \
102             (data->'extra_metadata'->>'duration_calc_version')::bigint AS duration_calc_version, \
103             object_id::text AS object_id_text, trace_id \
104             FROM {CONTRACT_VIEW} \
105             WHERE created_at >= now() - make_interval(days => $1)"
106        );
107
108        let rows = if let Some(object_type) = object_type {
109            client
110                .query(&format!("{base} AND object_type = $2"), &[&window_days, &object_type])
111                .await
112        } else {
113            client.query(&base, &[&window_days]).await
114        }
115        .with_context(|| format!("failed to read {CONTRACT_VIEW}"))?;
116
117        Ok(rows.iter().map(row_to_sample).collect())
118    }
119}
120
121/// Decode one `core.v_entity_change_log` row into a [`ChangeLogSample`].
122fn row_to_sample(row: &tokio_postgres::Row) -> ChangeLogSample {
123    ChangeLogSample {
124        object_type:           row.get("object_type"),
125        modification_type:     row.get("modification_type"),
126        duration_ms:           row.get("duration_ms"),
127        duration_calc_version: row.get("duration_calc_version"),
128        created_at_epoch:      row.get("created_at_epoch"),
129        object_id:             row.get("object_id_text"),
130        trace_id:              row.get("trace_id"),
131    }
132}
fraiseql_cli/commands/perf/reader.rs

fraiseql_cli/commands/perf/
reader.rs
