sequel_mcp/backup/
mod.rs

1//! Backup capture and restore planning over the audit database.
2
3pub mod extractor;
4pub mod journal;
5pub mod restore;
6
7use crate::audit::AuditDb;
8use crate::policy::model::Policy;
9use extractor::{BackupSpec, BackupTable};
10use rusqlite::params;
11use std::sync::Arc;
12use thiserror::Error;
13
14#[derive(Debug, Error)]
15pub enum BackupError {
16    #[error("backup row cap exceeded ({observed} > {cap})")]
17    RowOverflow { observed: u64, cap: u64 },
18    #[error("backup byte cap exceeded ({observed} > {cap})")]
19    ByteOverflow { observed: u64, cap: u64 },
20    #[error("database error: {0}")]
21    Db(#[from] rusqlite::Error),
22    #[error("backup query cannot be safely bounded: {0}")]
23    Unbounded(String),
24}
25
26#[derive(Debug, Clone)]
27pub struct CapturedBackup {
28    pub backup_id: i64,
29    pub total_rows: u64,
30    pub truncated: bool,
31    pub total_bytes: u64,
32}
33
34fn iso_now() -> String {
35    time::OffsetDateTime::now_utc()
36        .format(&time::format_description::well_known::Rfc3339)
37        .unwrap_or_else(|_| "1970-01-01T00:00:00Z".into())
38}
39
40#[allow(clippy::too_many_arguments)]
41fn insert_backup_row(
42    audit: &Option<Arc<AuditDb>>,
43    ts: &str,
44    connection: &str,
45    database: Option<&str>,
46    table: &str,
47    kind: &str,
48    rows_json: Option<&str>,
49    schema_sql: Option<&str>,
50    primary_key: Option<&str>,
51    row_count: u64,
52    truncated: bool,
53    bytes: u64,
54) -> rusqlite::Result<i64> {
55    let db = match audit {
56        Some(a) => a.clone(),
57        None => AuditDb::shared(),
58    };
59    db.with_tx(|c| {
60        c.execute(
61            "INSERT INTO backup
62               (ts, connection, database, table_name, backup_kind, rows_json,
63                schema_sql, primary_key, row_count, truncated, size_bytes)
64             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
65            params![
66                ts,
67                connection,
68                database,
69                table,
70                kind,
71                rows_json,
72                schema_sql,
73                primary_key,
74                row_count as i64,
75                truncated as i64,
76                bytes as i64,
77            ],
78        )?;
79        Ok(c.last_insert_rowid())
80    })
81}
82
83fn show_create_table_sqlite(
84    db: &rusqlite::Connection,
85    database: Option<&str>,
86    table: &str,
87) -> Option<String> {
88    let schema_table = match database {
89        Some(d) => format!("{}.sqlite_schema", extractor::quote_ident(d)),
90        None => "sqlite_schema".to_string(),
91    };
92    db.query_row(
93        &format!(
94            "SELECT sql FROM {schema_table}
95              WHERE name = ?1 AND type IN ('table', 'view')
96              ORDER BY CASE type WHEN 'table' THEN 0 ELSE 1 END LIMIT 1"
97        ),
98        params![table],
99        |r| r.get::<_, Option<String>>(0),
100    )
101    .ok()
102    .flatten()
103}
104
105pub mod capture {
106    //! Re-exports used by the executors.
107    pub use super::{
108        BackupError, CapturedBackup, capture_backup_sqlite, capture_insert_hint,
109        insert_rows_backup_row, insert_schema_backup_row,
110    };
111}
112
113/// Insert a schema-only backup row (used by the MySQL and SQLite paths).
114pub fn insert_schema_backup_row(
115    audit: &Option<Arc<AuditDb>>,
116    ts: &str,
117    connection_name: &str,
118    database: Option<&str>,
119    table: &str,
120    schema_sql: Option<&str>,
121) -> rusqlite::Result<i64> {
122    insert_backup_row(
123        audit,
124        ts,
125        connection_name,
126        database,
127        table,
128        "schema",
129        None,
130        schema_sql,
131        None,
132        0,
133        false,
134        0,
135    )
136}
137
138/// Insert a rows/combined backup row (used by the MySQL path).
139#[allow(clippy::too_many_arguments)]
140pub fn insert_rows_backup_row(
141    audit: &Option<Arc<AuditDb>>,
142    ts: &str,
143    connection_name: &str,
144    database: Option<&str>,
145    table: &str,
146    kind: &str,
147    rows_json: Option<&str>,
148    schema_sql: Option<&str>,
149    row_count: u64,
150    truncated: bool,
151    bytes: u64,
152) -> rusqlite::Result<i64> {
153    insert_backup_row(
154        audit,
155        ts,
156        connection_name,
157        database,
158        table,
159        kind,
160        rows_json,
161        schema_sql,
162        None,
163        row_count,
164        truncated,
165        bytes,
166    )
167}
168
169/// Capture a pre-mutation backup from a SQLite handle, enforcing the row
170/// and byte caps with the policy's overflow behaviour.
171pub fn capture_backup_sqlite(
172    db: &rusqlite::Connection,
173    spec: &BackupSpec,
174    connection_name: &str,
175    database: Option<&str>,
176    policy: &Policy,
177    audit: Option<&Arc<AuditDb>>,
178) -> Result<Option<CapturedBackup>, BackupError> {
179    let ts = iso_now();
180    let row_cap = policy.max_backup_rows as u64;
181    let byte_cap = policy.max_backup_bytes;
182
183    let tables = match spec {
184        BackupSpec::None { .. } | BackupSpec::InsertHint { .. } => return Ok(None),
185        BackupSpec::Rows { tables } | BackupSpec::Combined { tables } => tables,
186        BackupSpec::Schema { tables } => {
187            let mut first_id = None;
188            let mut total = 0u64;
189            for t in tables {
190                let schema_sql = show_create_table_sqlite(db, database, &t.table);
191                let id = insert_backup_row(
192                    &audit.map(Arc::clone),
193                    &ts,
194                    connection_name,
195                    database,
196                    &t.table,
197                    "schema",
198                    None,
199                    schema_sql.as_deref(),
200                    None,
201                    0,
202                    false,
203                    0,
204                )?;
205                if first_id.is_none() {
206                    first_id = Some(id);
207                }
208                total += 1;
209            }
210            return Ok(first_id.map(|id| CapturedBackup {
211                backup_id: id,
212                total_rows: 0,
213                truncated: false,
214                total_bytes: total,
215            }));
216        }
217    };
218
219    let mut first_id: Option<i64> = None;
220    let mut total_rows = 0u64;
221    let mut total_bytes = 0u64;
222    let mut truncated_any = false;
223
224    for t in tables {
225        let (rows_json, schema_sql, row_count, truncated, bytes) =
226            fetch_rows_sqlite(db, t, row_cap, database)?;
227        if truncated
228            && matches!(
229                policy.on_backup_overflow,
230                crate::policy::model::BackupOverflow::Abort
231            )
232        {
233            return Err(BackupError::RowOverflow {
234                observed: row_cap + 1,
235                cap: row_cap,
236            });
237        }
238        if bytes > byte_cap
239            && matches!(
240                policy.on_backup_overflow,
241                crate::policy::model::BackupOverflow::Abort
242            )
243        {
244            return Err(BackupError::ByteOverflow {
245                observed: bytes,
246                cap: byte_cap,
247            });
248        }
249        let kind = match spec {
250            BackupSpec::Combined { .. } => "combined",
251            _ => "rows",
252        };
253        let schema_sql = if matches!(spec, BackupSpec::Combined { .. }) {
254            schema_sql.or_else(|| show_create_table_sqlite(db, database, &t.table))
255        } else {
256            schema_sql
257        };
258        let id = insert_backup_row(
259            &audit.map(Arc::clone),
260            &ts,
261            connection_name,
262            database,
263            &t.table,
264            kind,
265            rows_json.as_deref(),
266            schema_sql.as_deref(),
267            None,
268            row_count,
269            truncated,
270            bytes,
271        )?;
272        if first_id.is_none() {
273            first_id = Some(id);
274        }
275        total_rows += row_count;
276        total_bytes += bytes;
277        truncated_any |= truncated;
278    }
279
280    Ok(first_id.map(|id| CapturedBackup {
281        backup_id: id,
282        total_rows,
283        truncated: truncated_any,
284        total_bytes,
285    }))
286}
287
288/// Row payload fetched for a backup.
289type RowsFetch = (Option<String>, Option<String>, u64, bool, u64);
290
291fn fetch_rows_sqlite(
292    db: &rusqlite::Connection,
293    t: &BackupTable,
294    row_cap: u64,
295    _database: Option<&str>,
296) -> Result<RowsFetch, BackupError> {
297    // Unrewritable backup SELECT (existing LIMIT, set op, CTE, comment,
298    // unmatched lock clause, ...): deny the mutation rather than run an
299    // unbounded pre-image query.
300    let capped =
301        crate::backup::extractor::with_limit(&t.select_sql, row_cap + 1).ok_or_else(|| {
302            BackupError::Unbounded(format!(
303                "backup query not safely limitable: {}",
304                t.select_sql.chars().take(80).collect::<String>()
305            ))
306        })?;
307    let mut stmt = db.prepare(&capped)?;
308    let names: Vec<String> = stmt.column_names().iter().map(|s| s.to_string()).collect();
309    let mut rows = stmt.query([])?;
310    let mut out: Vec<serde_json::Value> = Vec::new();
311    let mut truncated = false;
312    while let Some(row) = rows.next()? {
313        if out.len() as u64 >= row_cap {
314            truncated = true;
315            break;
316        }
317        let mut obj = serde_json::Map::with_capacity(names.len());
318        for (i, name) in names.iter().enumerate() {
319            let v = match row.get_ref(i)? {
320                rusqlite::types::ValueRef::Null => serde_json::Value::Null,
321                rusqlite::types::ValueRef::Integer(n) => serde_json::json!(n),
322                rusqlite::types::ValueRef::Real(f) => serde_json::json!(f),
323                rusqlite::types::ValueRef::Text(s) => {
324                    serde_json::json!(String::from_utf8_lossy(s))
325                }
326                rusqlite::types::ValueRef::Blob(b) => {
327                    use base64::Engine;
328                    serde_json::json!(base64::engine::general_purpose::STANDARD.encode(b))
329                }
330            };
331            obj.insert(name.clone(), v);
332        }
333        out.push(serde_json::Value::Object(obj));
334    }
335    let bytes = serde_json::to_string(&out)
336        .map(|s| s.len() as u64)
337        .unwrap_or(0);
338    let count = out.len() as u64;
339    let json = if count > 0 {
340        Some(serde_json::to_string(&out).unwrap_or_default())
341    } else {
342        None
343    };
344    Ok((json, None, count, truncated, bytes))
345}
346
347/// Record an insert rollback hint (explicit PK values or autoincrement
348/// range) after a successful INSERT.
349pub fn capture_insert_hint(
350    spec: &BackupSpec,
351    connection_name: &str,
352    database: Option<&str>,
353    insert_id: Option<i64>,
354    affected_rows: u64,
355    audit: Option<&Arc<AuditDb>>,
356) -> Option<i64> {
357    let BackupSpec::InsertHint {
358        table,
359        explicit_pk_values,
360        ..
361    } = spec
362    else {
363        return None;
364    };
365    let (primary_key, rows_json, row_count) = if let Some(values) = explicit_pk_values {
366        if values.is_empty() {
367            return None;
368        }
369        let pk = serde_json::json!({
370            "kind": "explicit",
371            "columns": ["id"],
372            "values": values,
373        })
374        .to_string();
375        let rows = serde_json::to_string(values).ok()?;
376        (pk, Some(rows), values.len() as u64)
377    } else if let (Some(start), true) = (insert_id, affected_rows > 0) {
378        let end = start + affected_rows as i64 - 1;
379        let pk = serde_json::json!({
380            "kind": "range",
381            "column": "id",
382            "start": start,
383            "end": end,
384        })
385        .to_string();
386        (pk, None, affected_rows)
387    } else {
388        return None;
389    };
390    let bytes = rows_json.as_ref().map(|r| r.len() as u64).unwrap_or(0);
391    insert_backup_row(
392        &audit.map(Arc::clone),
393        &iso_now(),
394        connection_name,
395        database.or(table.db.as_deref()),
396        &table.table,
397        "insert-hint",
398        rows_json.as_deref(),
399        None,
400        Some(&primary_key),
401        row_count,
402        false,
403        bytes,
404    )
405    .ok()
406}
407
408/// Backup listing row (legacy shape).
409#[derive(Debug, Clone, serde::Serialize)]
410pub struct BackupRow {
411    pub id: i64,
412    pub ts: String,
413    pub connection: String,
414    pub database: Option<String>,
415    pub table_name: String,
416    pub backup_kind: String,
417    pub row_count: u64,
418    pub truncated: bool,
419    pub size_bytes: u64,
420}
421
422pub fn list_backups(
423    audit: &Arc<AuditDb>,
424    connection: Option<&str>,
425    limit: u32,
426) -> rusqlite::Result<Vec<BackupRow>> {
427    let limit = limit.min(1000);
428    audit.with(|c| {
429        let mut stmt = c.prepare(
430            "SELECT id, ts, connection, database, table_name, backup_kind, row_count, truncated, size_bytes
431               FROM backup
432              WHERE (?1 IS NULL OR connection = ?1)
433              ORDER BY id DESC LIMIT ?2",
434        )?;
435        let rows = stmt.query_map(params![connection, limit], |r| {
436            Ok(BackupRow {
437                id: r.get(0)?,
438                ts: r.get(1)?,
439                connection: r.get(2)?,
440                database: r.get(3)?,
441                table_name: r.get(4)?,
442                backup_kind: r.get(5)?,
443                row_count: r.get::<_, i64>(6)?.max(0) as u64,
444                truncated: r.get::<_, i64>(7)? != 0,
445                size_bytes: r.get::<_, i64>(8)?.max(0) as u64,
446            })
447        })?;
448        rows.collect()
449    })
450}
451
452#[derive(Debug, Clone, serde::Serialize)]
453pub struct BackupDetail {
454    pub id: i64,
455    pub ts: String,
456    pub connection: String,
457    pub database: Option<String>,
458    pub table_name: String,
459    pub backup_kind: String,
460    pub row_count: u64,
461    pub truncated: bool,
462    pub size_bytes: u64,
463    pub rows: Option<serde_json::Value>,
464    pub schema_sql: Option<String>,
465    pub primary_key: Option<String>,
466}
467
468pub fn get_backup(audit: &Arc<AuditDb>, id: i64) -> rusqlite::Result<Option<BackupDetail>> {
469    audit.with(|c| {
470        c.query_row(
471            "SELECT id, ts, connection, database, table_name, backup_kind, rows_json,
472                    schema_sql, primary_key, row_count, truncated, size_bytes
473               FROM backup WHERE id = ?1",
474            params![id],
475            |r| {
476                let rows_json: Option<String> = r.get(6)?;
477                Ok(BackupDetail {
478                    id: r.get(0)?,
479                    ts: r.get(1)?,
480                    connection: r.get(2)?,
481                    database: r.get(3)?,
482                    table_name: r.get(4)?,
483                    backup_kind: r.get(5)?,
484                    row_count: r.get::<_, i64>(9)?.max(0) as u64,
485                    truncated: r.get::<_, i64>(10)? != 0,
486                    size_bytes: r.get::<_, i64>(11)?.max(0) as u64,
487                    rows: rows_json.and_then(|j| serde_json::from_str(&j).ok()),
488                    schema_sql: r.get(7)?,
489                    primary_key: r.get(8)?,
490                })
491            },
492        )
493        .map(Some)
494        .or_else(|e| match e {
495            rusqlite::Error::QueryReturnedNoRows => Ok(None),
496            other => Err(other),
497        })
498    })
499}
sequel_mcp/backup/mod.rs

sequel_mcp/backup/
mod.rs
