safe_migrate/
sync.rs

1// FILE: src/sync.rs
2
3use crate::ast::identifiers::ObjectId;
4use crate::db::cache::{DbCache, ForeignKeyCache, IndexCache};
5use crate::model::relation::{Persistence, RelationKind, RelationState};
6use anyhow::{Context, Result};
7use postgres::{Client, NoTls};
8use std::fs;
9use std::path::Path;
10
11pub fn sync_cache(out_path: &Path) -> Result<()> {
12    // Strict env-only credential enforcement
13    let db_url = std::env::var("DATABASE_URL")
14        .context("DATABASE_URL environment variable is required to sync database stats. Do not pass credentials via CLI flags or config files.")?;
15
16    // Destructive cache removal prevents corrupted reads on failures
17    if out_path.exists() {
18        fs::remove_file(out_path).context("Failed to remove old cache file before sync")?;
19    }
20
21    let mut client = Client::connect(&db_url, NoTls).context("Failed to connect to PostgreSQL")?;
22
23    let mut cache = DbCache::new();
24
25    // Query 1: Server Version
26    let version_row = client.query_one("SHOW server_version_num;", &[])?;
27    let version_str: String = version_row.get(0);
28    cache.pg_version_num = version_str.parse::<u32>().ok();
29
30    // Query 2: Relations + Staleness
31    let table_query = "
32        SELECT
33            n.nspname AS schema_name,
34            c.relname AS relation_name,
35            c.relkind AS relation_kind,
36            c.relpersistence AS persistence,
37            CASE WHEN c.reltuples < 0 THEN -1 ELSE c.reltuples::bigint END AS estimated_rows,
38            GREATEST(c.relpages::bigint, 0) AS relpages,
39            to_char(s.last_analyze, 'YYYY-MM-DD HH24:MI:SS') AS last_analyze,
40            to_char(s.last_autoanalyze, 'YYYY-MM-DD HH24:MI:SS') AS last_autoanalyze
41        FROM pg_class c
42        JOIN pg_namespace n ON n.oid = c.relnamespace
43        LEFT JOIN pg_stat_user_tables s ON s.relid = c.oid
44        WHERE c.relkind IN ('r', 'p', 'v', 'm')
45          AND n.nspname NOT IN ('pg_catalog', 'information_schema');
46    ";
47
48    for row in client.query(table_query, &[])? {
49        let schema_name: String = row.get("schema_name");
50        let relation_name: String = row.get("relation_name");
51        let relkind: i8 = row.get("relation_kind");
52        let persistence_char: i8 = row.get("persistence");
53        let raw_rows: i64 = row.get("estimated_rows");
54        let relpages: i64 = row.get("relpages");
55
56        let last_analyze: Option<String> = row.get("last_analyze");
57        let last_autoanalyze: Option<String> = row.get("last_autoanalyze");
58
59        let object_id = ObjectId::new(&schema_name, &relation_name);
60
61        let kind = match relkind as u8 {
62            b'v' => RelationKind::View,
63            b'm' => RelationKind::MaterializedView,
64            _ => RelationKind::Table,
65        };
66
67        let persistence = match persistence_char as u8 {
68            b't' => Persistence::Temporary,
69            b'u' => Persistence::Unlogged,
70            _ => Persistence::Permanent,
71        };
72
73        let estimated_rows = if raw_rows < 0 {
74            None
75        } else {
76            Some(raw_rows as u64)
77        };
78
79        let mut state =
80            RelationState::new(object_id.clone(), 0, estimated_rows, kind, persistence, 0);
81        state.relpages = Some(relpages as u64);
82        state.last_analyze = last_analyze;
83        state.last_autoanalyze = last_autoanalyze;
84
85        cache.insert_baseline(object_id, state);
86    }
87
88    // Query 3: Columns + Width
89    let col_query = "
90        SELECT
91            n.nspname AS schema_name,
92            c.relname AS relation_name,
93            a.attname AS column_name,
94            pg_catalog.format_type(a.atttypid, a.atttypmod) AS type_name,
95            a.attnotnull AS not_null,
96            s.avg_width AS avg_width
97        FROM pg_attribute a
98        JOIN pg_class c ON a.attrelid = c.oid
99        JOIN pg_namespace n ON n.oid = c.relnamespace
100        LEFT JOIN pg_stats s ON s.schemaname = n.nspname AND s.tablename = c.relname AND s.attname = a.attname
101        WHERE a.attnum > 0 AND NOT a.attisdropped
102          AND c.relkind IN ('r', 'p', 'v', 'm')
103          AND n.nspname NOT IN ('pg_catalog', 'information_schema');
104    ";
105
106    for row in client.query(col_query, &[])? {
107        let schema_name: String = row.get("schema_name");
108        let relation_name: String = row.get("relation_name");
109        let column_name: String = row.get("column_name");
110        let type_name: String = row.get("type_name");
111        let not_null: bool = row.get("not_null");
112        let avg_width: Option<i32> = row.get("avg_width");
113
114        let object_id = ObjectId::new(&schema_name, &relation_name);
115
116        if let Some(rel) = cache.relations.get_mut(&object_id) {
117            rel.columns.push(crate::model::column::Column {
118                name: column_name,
119                data_type: Some(type_name),
120                is_nullable: !not_null,
121                default: None,
122                avg_width,
123            });
124        }
125    }
126
127    // Query 4: Triggers & Policies
128    let tp_query = "
129        SELECT 
130            n.nspname AS schema_name,
131            c.relname AS relation_name,
132            COALESCE(array_agg(DISTINCT t.tgname) FILTER (WHERE t.tgname IS NOT NULL AND t.tgisinternal = false), '{}') as triggers,
133            COALESCE(array_agg(DISTINCT p.polname) FILTER (WHERE p.polname IS NOT NULL), '{}') as policies
134        FROM pg_class c
135        JOIN pg_namespace n ON n.oid = c.relnamespace
136        LEFT JOIN pg_trigger t ON t.tgrelid = c.oid
137        LEFT JOIN pg_policy p ON p.polrelid = c.oid
138        WHERE c.relkind IN ('r', 'p', 'v', 'm') AND n.nspname NOT IN ('pg_catalog', 'information_schema')
139        GROUP BY n.nspname, c.relname;
140    ";
141
142    for row in client.query(tp_query, &[])? {
143        let schema_name: String = row.get("schema_name");
144        let relation_name: String = row.get("relation_name");
145        let triggers: Vec<String> = row.get("triggers");
146        let policies: Vec<String> = row.get("policies");
147
148        let object_id = ObjectId::new(&schema_name, &relation_name);
149
150        if let Some(rel) = cache.relations.get_mut(&object_id) {
151            rel.triggers.extend(triggers);
152            rel.policies.extend(policies);
153        }
154    }
155
156    // Query 5: Foreign Keys
157    let fk_query = "
158        SELECT 
159            c.conname AS constraint_name,
160            n1.nspname AS from_schema, t1.relname AS from_table,
161            n2.nspname AS to_schema, t2.relname AS to_table
162        FROM pg_constraint c
163        JOIN pg_class t1 ON t1.oid = c.conrelid
164        JOIN pg_namespace n1 ON n1.oid = t1.relnamespace
165        JOIN pg_class t2 ON t2.oid = c.confrelid
166        JOIN pg_namespace n2 ON n2.oid = t2.relnamespace
167        WHERE c.contype = 'f';
168    ";
169
170    for row in client.query(fk_query, &[])? {
171        let constraint_name: String = row.get("constraint_name");
172        let from_schema: String = row.get("from_schema");
173        let from_table: String = row.get("from_table");
174        let to_schema: String = row.get("to_schema");
175        let to_table: String = row.get("to_table");
176
177        cache.foreign_keys.push(ForeignKeyCache {
178            constraint_name,
179            from_table: ObjectId::new(&from_schema, &from_table),
180            to_table: ObjectId::new(&to_schema, &to_table),
181        });
182    }
183
184    // Query 6: Indexes
185    let idx_query = "
186        SELECT 
187            n_i.nspname AS index_schema, i.relname AS index_name,
188            n_t.nspname AS table_schema, t.relname AS table_name
189        FROM pg_index x
190        JOIN pg_class i ON i.oid = x.indexrelid
191        JOIN pg_namespace n_i ON n_i.oid = i.relnamespace
192        JOIN pg_class t ON t.oid = x.indrelid
193        JOIN pg_namespace n_t ON n_t.oid = t.relnamespace
194        WHERE x.indisvalid = true;
195    ";
196
197    for row in client.query(idx_query, &[])? {
198        let index_schema: String = row.get("index_schema");
199        let index_name: String = row.get("index_name");
200        let table_schema: String = row.get("table_schema");
201        let table_name: String = row.get("table_name");
202
203        cache.indexes.push(IndexCache {
204            index_id: ObjectId::new(&index_schema, &index_name),
205            table_id: ObjectId::new(&table_schema, &table_name),
206        });
207    }
208
209    // Atomic write via temp file
210    let json = serde_json::to_string_pretty(&cache)?;
211    let tmp_path = out_path.with_extension("tmp");
212
213    fs::write(&tmp_path, json).context("Failed to write temporary cache file")?;
214    fs::rename(&tmp_path, out_path).context("Failed to atomically rename cache file")?;
215
216    Ok(())
217}
safe_migrate/sync.rs

safe_migrate/
sync.rs
