remem/db/query/
legacy_surfaces.rs

1use anyhow::Result;
2use rusqlite::{Connection, OptionalExtension};
3
4#[derive(Debug, Clone, PartialEq, Eq)]
5pub struct LegacySurfaceStats {
6    pub surface: String,
7    pub disposition: String,
8    pub row_count: i64,
9    pub last_write_epoch: Option<i64>,
10    pub frozen_write_violations: i64,
11}
12
13pub(super) fn query_legacy_surface_stats(conn: &Connection) -> Result<Vec<LegacySurfaceStats>> {
14    let observations = legacy_table_surface(conn, "observations", "reclassify-current", &[])?;
15    let observations_fts =
16        legacy_table_surface(conn, "observations_fts", "reclassify-current", &[])?;
17    let session_summaries = legacy_table_surface(conn, "session_summaries", "keep", &[])?;
18    let pending_observations = legacy_pending_observation_surface(conn)?;
19    let summary_jobs = legacy_summary_job_surface(conn)?;
20
21    Ok(vec![
22        observations,
23        observations_fts,
24        session_summaries,
25        pending_observations,
26        summary_jobs,
27    ])
28}
29
30fn legacy_table_surface(
31    conn: &Connection,
32    table: &str,
33    disposition: &str,
34    violation_epoch_columns: &[&str],
35) -> Result<LegacySurfaceStats> {
36    let row_count = table_count_or_zero(conn, table)?;
37    let last_write_epoch = max_write_epoch(conn, table)?;
38    let mut has_violation_epoch = false;
39    for column in violation_epoch_columns {
40        has_violation_epoch |= column_exists(conn, table, column)?;
41    }
42    let frozen_write_violations = if row_count > 0 && has_violation_epoch {
43        row_count
44    } else {
45        0
46    };
47    Ok(LegacySurfaceStats {
48        surface: table.to_string(),
49        disposition: disposition.to_string(),
50        row_count,
51        last_write_epoch,
52        frozen_write_violations,
53    })
54}
55
56fn legacy_pending_observation_surface(conn: &Connection) -> Result<LegacySurfaceStats> {
57    let table = "pending_observations";
58    let row_count = table_count_or_zero(conn, table)?;
59    let last_write_epoch = max_write_epoch(conn, table)?;
60    let frozen_write_violations = if !table_exists(conn, table)? {
61        0
62    } else if column_exists(conn, table, "status")? {
63        let archived_filter = if column_exists(conn, table, "archived_at_epoch")? {
64            "AND archived_at_epoch IS NULL"
65        } else {
66            ""
67        };
68        conn.query_row(
69            &format!(
70                "SELECT COUNT(*) FROM pending_observations
71                 WHERE status <> 'migrated'
72                 {archived_filter}"
73            ),
74            [],
75            |row| row.get(0),
76        )?
77    } else {
78        row_count
79    };
80
81    Ok(LegacySurfaceStats {
82        surface: table.to_string(),
83        disposition: "retire".to_string(),
84        row_count,
85        last_write_epoch,
86        frozen_write_violations,
87    })
88}
89
90fn legacy_summary_job_surface(conn: &Connection) -> Result<LegacySurfaceStats> {
91    let surface = "summary_jobs".to_string();
92    let disposition = "retire-summary-only".to_string();
93    if !table_exists(conn, "jobs")? || !column_exists(conn, "jobs", "job_type")? {
94        return Ok(LegacySurfaceStats {
95            surface,
96            disposition,
97            row_count: 0,
98            last_write_epoch: None,
99            frozen_write_violations: 0,
100        });
101    }
102
103    let row_count = conn.query_row(
104        "SELECT COUNT(*) FROM jobs WHERE job_type = 'summary'",
105        [],
106        |row| row.get(0),
107    )?;
108    let frozen_write_violations = if column_exists(conn, "jobs", "state")? {
109        let archived_filter = if column_exists(conn, "jobs", "archived_at_epoch")? {
110            "AND archived_at_epoch IS NULL"
111        } else {
112            ""
113        };
114        let rejection_filter = if column_exists(conn, "jobs", "failure_class")?
115            && column_exists(conn, "jobs", "last_error")?
116        {
117            "AND NOT (
118                 state = 'failed'
119                 AND failure_class = 'permanent'
120                 AND last_error = 'legacy summary job rejected during GH684 summary retirement upgrade; SessionRollup owns session summary output'
121               )"
122        } else {
123            ""
124        };
125        conn.query_row(
126            &format!(
127                "SELECT COUNT(*) FROM jobs
128                 WHERE job_type = 'summary'
129                   AND state <> 'done'
130                   {rejection_filter}
131                   {archived_filter}"
132            ),
133            [],
134            |row| row.get(0),
135        )?
136    } else {
137        row_count
138    };
139    let last_write_epoch = max_write_epoch_where(conn, "jobs", "job_type = 'summary'")?;
140    Ok(LegacySurfaceStats {
141        surface,
142        disposition,
143        row_count,
144        last_write_epoch,
145        frozen_write_violations,
146    })
147}
148
149fn table_exists(conn: &Connection, table: &str) -> Result<bool> {
150    Ok(conn
151        .query_row(
152            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
153            [table],
154            |_| Ok(()),
155        )
156        .optional()?
157        .is_some())
158}
159
160fn table_count_or_zero(conn: &Connection, table: &str) -> Result<i64> {
161    if !table_exists(conn, table)? {
162        return Ok(0);
163    }
164    Ok(conn.query_row(
165        &format!("SELECT COUNT(*) FROM {}", quote_identifier(table)),
166        [],
167        |row| row.get(0),
168    )?)
169}
170
171fn max_write_epoch(conn: &Connection, table: &str) -> Result<Option<i64>> {
172    max_write_epoch_where(conn, table, "1 = 1")
173}
174
175fn max_write_epoch_where(
176    conn: &Connection,
177    table: &str,
178    where_clause: &str,
179) -> Result<Option<i64>> {
180    if !table_exists(conn, table)? {
181        return Ok(None);
182    }
183    let mut columns = Vec::new();
184    if column_exists(conn, table, "updated_at_epoch")? {
185        columns.push("updated_at_epoch");
186    }
187    if column_exists(conn, table, "created_at_epoch")? {
188        columns.push("created_at_epoch");
189    }
190    if columns.is_empty() {
191        return Ok(None);
192    }
193
194    let expression = match columns.as_slice() {
195        ["updated_at_epoch", "created_at_epoch"] => {
196            "CASE
197                WHEN NULLIF(updated_at_epoch, 0) IS NULL THEN NULLIF(created_at_epoch, 0)
198                WHEN NULLIF(created_at_epoch, 0) IS NULL THEN NULLIF(updated_at_epoch, 0)
199                WHEN updated_at_epoch >= created_at_epoch THEN updated_at_epoch
200                ELSE created_at_epoch
201             END"
202        }
203        [single] => match *single {
204            "updated_at_epoch" => "NULLIF(updated_at_epoch, 0)",
205            "created_at_epoch" => "NULLIF(created_at_epoch, 0)",
206            _ => unreachable!("legacy write epoch columns are fixed"),
207        },
208        _ => unreachable!("legacy write epoch columns are fixed"),
209    };
210    let sql = format!(
211        "SELECT MAX({expression}) FROM {} WHERE {where_clause}",
212        quote_identifier(table)
213    );
214    Ok(conn.query_row(&sql, [], |row| row.get(0))?)
215}
216
217fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
218    if !table_exists(conn, table)? {
219        return Ok(false);
220    }
221    let mut stmt = conn.prepare(&format!("PRAGMA table_info({})", quote_identifier(table)))?;
222    let mut rows = stmt.query([])?;
223    while let Some(row) = rows.next()? {
224        let name: String = row.get(1)?;
225        if name == column {
226            return Ok(true);
227        }
228    }
229    Ok(false)
230}
231
232fn quote_identifier(identifier: &str) -> String {
233    format!("\"{}\"", identifier.replace('"', "\"\""))
234}
remem/db/query/legacy_surfaces.rs

remem/db/query/
legacy_surfaces.rs
