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::{Config, ManagerConfig, Pool, RecyclingMethod, Runtime};
16use tokio_postgres::NoTls;
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 `postgres://` URL or the pool cannot
59 /// be created. (Connection failures surface lazily on first query.)
60 pub fn connect(db_url: &str) -> Result<Self> {
61 if !db_url.starts_with("postgres") {
62 anyhow::bail!(
63 "perf reads require a PostgreSQL connection URL (postgres://…); got: {db_url}"
64 );
65 }
66 let mut cfg = Config::new();
67 cfg.url = Some(db_url.to_string());
68 cfg.manager = Some(ManagerConfig {
69 recycling_method: RecyclingMethod::Fast,
70 });
71 cfg.pool = Some(deadpool_postgres::PoolConfig::new(2));
72 let pool = cfg
73 .create_pool(Some(Runtime::Tokio1), NoTls)
74 .context("failed to create PostgreSQL connection pool for perf reads")?;
75 Ok(Self { pool })
76 }
77
78 /// The database clock as a Unix epoch in seconds — the reference the analysis
79 /// windows back from, so there is no app↔DB skew at the window boundary.
80 ///
81 /// # Errors
82 ///
83 /// Returns an error if the connection or the query fails.
84 pub async fn db_now_epoch(&self) -> Result<f64> {
85 let client = self.pool.get().await.context("failed to acquire DB connection")?;
86 let row = client
87 .query_one("SELECT EXTRACT(EPOCH FROM now())::float8 AS now_epoch", &[])
88 .await
89 .context("failed to read database clock")?;
90 Ok(row.get("now_epoch"))
91 }
92
93 /// Load every change-log sample within the trailing `window_days` window,
94 /// optionally restricted to a single `object_type`.
95 ///
96 /// The window is applied in SQL (`created_at >= now() - make_interval(days =>
97 /// $1)`) so only the rows the analysis needs cross the wire.
98 ///
99 /// # Errors
100 ///
101 /// Returns an error if the connection or the query fails.
102 pub async fn load_samples(
103 &self,
104 window_days: i32,
105 object_type: Option<&str>,
106 ) -> Result<Vec<ChangeLogSample>> {
107 let client = self.pool.get().await.context("failed to acquire DB connection")?;
108
109 // All columns cast to dependency-light scalars; the marker is pulled out
110 // of the view's `data` JSONB (it is not a top-level view column).
111 let base = format!(
112 "SELECT object_type, modification_type, duration_ms, \
113 EXTRACT(EPOCH FROM created_at)::float8 AS created_at_epoch, \
114 (data->'extra_metadata'->>'duration_calc_version')::bigint AS duration_calc_version, \
115 object_id::text AS object_id_text, trace_id \
116 FROM {CONTRACT_VIEW} \
117 WHERE created_at >= now() - make_interval(days => $1)"
118 );
119
120 let rows = if let Some(object_type) = object_type {
121 client
122 .query(&format!("{base} AND object_type = $2"), &[&window_days, &object_type])
123 .await
124 } else {
125 client.query(&base, &[&window_days]).await
126 }
127 .with_context(|| format!("failed to read {CONTRACT_VIEW}"))?;
128
129 Ok(rows.iter().map(row_to_sample).collect())
130 }
131}
132
133/// Decode one `core.v_entity_change_log` row into a [`ChangeLogSample`].
134fn row_to_sample(row: &tokio_postgres::Row) -> ChangeLogSample {
135 ChangeLogSample {
136 object_type: row.get("object_type"),
137 modification_type: row.get("modification_type"),
138 duration_ms: row.get("duration_ms"),
139 duration_calc_version: row.get("duration_calc_version"),
140 created_at_epoch: row.get("created_at_epoch"),
141 object_id: row.get("object_id_text"),
142 trace_id: row.get("trace_id"),
143 }
144}