Skip to main content

mitosis/
state.rs

1use crate::error::{ErrorCode, MitosisError, Result};
2use crate::protocol::{
3    EventFilter, EventRow, GenKind, MigrateCmd, OnStale, PendingRestore, RefreshMode,
4    TemplateSource,
5};
6use rusqlite::{params, Connection, OptionalExtension, Row};
7use serde::de::DeserializeOwned;
8use serde::Serialize;
9use std::collections::HashSet;
10use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
11use std::path::{Path, PathBuf};
12
13#[derive(Clone, Copy, Debug, PartialEq, Eq)]
14pub enum CloneStatus {
15    Creating,
16    Starting,
17    Ready,
18    Stopping,
19    /// Postgres stopped cleanly; folder, port, password and `.env` records kept until a wake.
20    Sleeping,
21}
22
23impl CloneStatus {
24    pub fn as_str(self) -> &'static str {
25        match self {
26            CloneStatus::Creating => "creating",
27            CloneStatus::Starting => "starting",
28            CloneStatus::Ready => "ready",
29            CloneStatus::Stopping => "stopping",
30            CloneStatus::Sleeping => "sleeping",
31        }
32    }
33
34    fn parse(s: &str) -> Option<CloneStatus> {
35        Some(match s {
36            "creating" => CloneStatus::Creating,
37            "starting" => CloneStatus::Starting,
38            "ready" => CloneStatus::Ready,
39            "stopping" => CloneStatus::Stopping,
40            "sleeping" => CloneStatus::Sleeping,
41            _ => return None,
42        })
43    }
44}
45
46#[derive(Clone, Debug, PartialEq)]
47pub struct TemplateRow {
48    pub name: String,
49    pub database: String,
50    pub source: TemplateSource,
51    pub migrations: Option<PathBuf>,
52    pub max_age_secs: u64,
53    /// 0 until the first generation is committed.
54    pub current_gen: u32,
55    /// The number `alloc_gen` hands out next; never goes down.
56    pub next_gen: u32,
57    pub keep: u32,
58    pub every_secs: Option<u64>,
59    pub every_mode: RefreshMode,
60    pub migrate: Option<MigrateCmd>,
61    pub auto_migrate: bool,
62    pub on_stale: OnStale,
63    pub paused: bool,
64    pub last_error: Option<String>,
65    pub last_attempt_at: Option<i64>,
66    /// Masking file (absolute path); full builds then dump and restore through a staging cluster.
67    pub mask_path: Option<PathBuf>,
68    /// Hex of the 32 random bytes that key masked values; made at the first masked build.
69    /// Written only by `ensure_mask_salt`, never by `update_template_settings`.
70    pub mask_salt: Option<String>,
71    /// A branch template (`<base>~<branch>`): its base template.
72    pub base: Option<String>,
73    /// A branch template: the git branch it was built for, as git names it.
74    pub branch: Option<String>,
75    /// A branch template: the base generation its current generation was layered on.
76    pub base_gen: Option<u32>,
77    /// A base template: the branch its recorded checkout follows; layered builds of the base skip
78    /// while that checkout is on another branch.
79    pub base_branch: Option<String>,
80    /// The migrate command's folder, relative to the root of the git work tree it is in.
81    pub repo_rel_cwd: Option<PathBuf>,
82    /// The migrations folder, relative to the root of the git work tree it is in.
83    pub repo_rel_migrations: Option<PathBuf>,
84    /// A branch template: when a clone last asked for it (gc drops long-unused ones).
85    pub last_used: Option<i64>,
86}
87
88impl TemplateRow {
89    /// A template with no generation yet and default settings.
90    pub fn new(
91        name: &str,
92        database: &str,
93        source: TemplateSource,
94        migrations: Option<PathBuf>,
95        max_age_secs: u64,
96    ) -> TemplateRow {
97        TemplateRow {
98            name: name.into(),
99            database: database.into(),
100            source,
101            migrations,
102            max_age_secs,
103            current_gen: 0,
104            next_gen: 1,
105            keep: DEFAULT_KEEP,
106            every_secs: None,
107            every_mode: RefreshMode::default(),
108            migrate: None,
109            auto_migrate: false,
110            on_stale: OnStale::default(),
111            paused: false,
112            last_error: None,
113            last_attempt_at: None,
114            mask_path: None,
115            mask_salt: None,
116            base: None,
117            branch: None,
118            base_gen: None,
119            base_branch: None,
120            repo_rel_cwd: None,
121            repo_rel_migrations: None,
122            last_used: None,
123        }
124    }
125}
126
127#[derive(Clone, Debug, PartialEq)]
128pub struct GenerationRow {
129    pub template: String,
130    pub gen: u32,
131    pub kind: GenKind,
132    pub dir: PathBuf,
133    pub pg_major: u32,
134    pub bin_dir: PathBuf,
135    pub extensions: Vec<String>,
136    pub migrations_hash: Option<String>,
137    pub built_at: i64,
138    pub size_bytes: u64,
139    pub from_clone: Option<String>,
140    pub from_gen: Option<u32>,
141    /// Hash of the masking file the generation was masked with.
142    pub mask_hash: Option<String>,
143}
144
145#[derive(Clone, Debug, PartialEq)]
146pub struct CloneRow {
147    pub id: String,
148    pub name: Option<String>,
149    pub template: String,
150    pub dir: PathBuf,
151    pub port: u16,
152    pub socket_dir: PathBuf,
153    pub password: String,
154    pub database: String,
155    pub pg_pid: Option<i32>,
156    pub pg_start_time: Option<u64>,
157    pub parent_pid: Option<i32>,
158    pub parent_start_time: Option<u64>,
159    pub created_at: i64,
160    pub expires_at: i64,
161    pub durable: bool,
162    pub status: CloneStatus,
163    pub generation: u32,
164    pub protected: bool,
165    pub suspect: bool,
166}
167
168/// A clone name's sticky port and password; kept after the clone is destroyed.
169#[derive(Clone, Debug, PartialEq)]
170pub struct ReservationRow {
171    pub name: String,
172    pub port: u16,
173    pub password: String,
174    pub template: String,
175    pub last_used: i64,
176}
177
178pub struct Store {
179    conn: Connection,
180}
181
182/// SQLite `user_version` of the current schema; v0.1 stores are version 0.
183pub const SCHEMA_VERSION: i32 = 3;
184pub const DEFAULT_KEEP: u32 = 3;
185
186const SCHEMA: &str = "
187CREATE TABLE IF NOT EXISTS templates (
188    name TEXT PRIMARY KEY,
189    database TEXT NOT NULL,
190    source TEXT NOT NULL,
191    migrations TEXT,
192    max_age_secs INTEGER NOT NULL,
193    current_gen INTEGER NOT NULL,
194    next_gen INTEGER NOT NULL,
195    keep INTEGER NOT NULL DEFAULT 3,
196    every_secs INTEGER,
197    every_mode TEXT NOT NULL DEFAULT 'full',
198    migrate TEXT,
199    auto_migrate INTEGER NOT NULL DEFAULT 0,
200    on_stale TEXT NOT NULL DEFAULT 'warn',
201    paused INTEGER NOT NULL DEFAULT 0,
202    last_error TEXT,
203    last_attempt_at INTEGER,
204    mask_salt TEXT,
205    mask_path TEXT,
206    base TEXT,
207    branch TEXT,
208    base_gen INTEGER,
209    base_branch TEXT,
210    repo_rel_cwd TEXT,
211    repo_rel_migrations TEXT,
212    last_used INTEGER
213);
214CREATE TABLE IF NOT EXISTS generations (
215    template TEXT NOT NULL,
216    gen INTEGER NOT NULL,
217    kind TEXT NOT NULL,
218    dir TEXT NOT NULL,
219    pg_major INTEGER NOT NULL,
220    bin_dir TEXT NOT NULL,
221    extensions TEXT NOT NULL,
222    migrations_hash TEXT,
223    built_at INTEGER NOT NULL,
224    size_bytes INTEGER NOT NULL,
225    from_clone TEXT,
226    from_gen INTEGER,
227    mask_hash TEXT,
228    PRIMARY KEY (template, gen)
229);
230CREATE TABLE IF NOT EXISTS clones (
231    id TEXT PRIMARY KEY,
232    name TEXT,
233    template TEXT NOT NULL,
234    dir TEXT NOT NULL,
235    port INTEGER NOT NULL,
236    socket_dir TEXT NOT NULL,
237    password TEXT NOT NULL,
238    database TEXT NOT NULL,
239    pg_pid INTEGER,
240    pg_start_time INTEGER,
241    parent_pid INTEGER,
242    parent_start_time INTEGER,
243    created_at INTEGER NOT NULL,
244    expires_at INTEGER NOT NULL,
245    durable INTEGER NOT NULL,
246    status TEXT NOT NULL,
247    generation INTEGER NOT NULL DEFAULT 1,
248    protected INTEGER NOT NULL DEFAULT 0,
249    suspect INTEGER NOT NULL DEFAULT 0
250);
251CREATE UNIQUE INDEX IF NOT EXISTS clones_name_uniq ON clones(name) WHERE name IS NOT NULL;
252CREATE TABLE IF NOT EXISTS events (
253    id INTEGER PRIMARY KEY,
254    ts INTEGER NOT NULL,
255    kind TEXT NOT NULL,
256    clone TEXT,
257    template TEXT,
258    detail TEXT NOT NULL
259);
260CREATE INDEX IF NOT EXISTS events_ts ON events(ts);
261";
262
263/// v0.1 -> v2: each template row becomes a template plus generation 1 in its old folder.
264const MIGRATE_V1: &str = "
265ALTER TABLE templates RENAME TO templates_v1;
266ALTER TABLE clones ADD COLUMN generation INTEGER NOT NULL DEFAULT 1;
267ALTER TABLE clones ADD COLUMN protected INTEGER NOT NULL DEFAULT 0;
268ALTER TABLE clones ADD COLUMN suspect INTEGER NOT NULL DEFAULT 0;
269";
270const MIGRATE_V1_ROWS: &str = "
271INSERT INTO templates (name, database, source, migrations, max_age_secs, current_gen, next_gen)
272    SELECT name, database, source, migrations, max_age_secs, 1, 2 FROM templates_v1;
273INSERT INTO generations (template, gen, kind, dir, pg_major, bin_dir, extensions, migrations_hash, built_at, size_bytes)
274    SELECT name, 1, 'full', dir, pg_major, bin_dir, extensions, migrations_hash, built_at, size_bytes FROM templates_v1;
275DROP TABLE templates_v1;
276";
277
278/// Runs on every open: a v2 store made before reservations existed gains the table too.
279const RESERVATIONS: &str = "
280CREATE TABLE IF NOT EXISTS reservations (
281    name TEXT PRIMARY KEY,
282    port INTEGER NOT NULL,
283    password TEXT NOT NULL,
284    template TEXT NOT NULL,
285    last_used INTEGER NOT NULL
286);
287";
288
289/// Runs on every open, like `RESERVATIONS`.
290const ENV_TABLES: &str = "
291CREATE TABLE IF NOT EXISTS env_records (
292    clone_id TEXT NOT NULL,
293    file TEXT NOT NULL,
294    var TEXT NOT NULL,
295    previous TEXT,
296    url TEXT NOT NULL,
297    PRIMARY KEY (clone_id, file, var)
298);
299CREATE TABLE IF NOT EXISTS pending_restores (
300    id INTEGER PRIMARY KEY AUTOINCREMENT,
301    file TEXT NOT NULL,
302    var TEXT NOT NULL,
303    previous TEXT,
304    url TEXT NOT NULL
305);
306";
307
308/// Columns added after their table first shipped: added on open when missing, so v0.2 stores
309/// and v3 stores made before them gain them too.
310const ADDED_COLUMNS: &[(&str, &str, &str)] = &[
311    ("templates", "mask_salt", "TEXT"),
312    ("templates", "mask_path", "TEXT"),
313    ("generations", "mask_hash", "TEXT"),
314    ("templates", "base", "TEXT"),
315    ("templates", "branch", "TEXT"),
316    ("templates", "base_gen", "INTEGER"),
317    ("templates", "base_branch", "TEXT"),
318    ("templates", "repo_rel_cwd", "TEXT"),
319    ("templates", "repo_rel_migrations", "TEXT"),
320    ("templates", "last_used", "INTEGER"),
321];
322
323const TEMPLATE_COLS: &str = "name, database, source, migrations, max_age_secs, current_gen, next_gen, keep, every_secs, every_mode, migrate, auto_migrate, on_stale, paused, last_error, last_attempt_at, mask_salt, mask_path, base, branch, base_gen, base_branch, repo_rel_cwd, repo_rel_migrations, last_used";
324const GEN_COLS: &str = "template, gen, kind, dir, pg_major, bin_dir, extensions, migrations_hash, built_at, size_bytes, from_clone, from_gen, mask_hash";
325const CLONE_COLS: &str = "id, name, template, dir, port, socket_dir, password, database, pg_pid, pg_start_time, parent_pid, parent_start_time, created_at, expires_at, durable, status, generation, protected, suspect";
326
327fn conv_err(idx: usize, msg: String) -> rusqlite::Error {
328    rusqlite::Error::FromSqlConversionFailure(
329        idx,
330        rusqlite::types::Type::Text,
331        Box::new(std::io::Error::new(std::io::ErrorKind::InvalidData, msg)),
332    )
333}
334
335/// A unit enum as its serde name (`"full"`), for TEXT columns.
336fn enum_str<T: Serialize>(v: &T) -> String {
337    serde_json::to_value(v)
338        .ok()
339        .and_then(|v| v.as_str().map(String::from))
340        .unwrap_or_default()
341}
342
343fn enum_col<T: DeserializeOwned>(r: &Row, idx: usize) -> rusqlite::Result<T> {
344    let s: String = r.get(idx)?;
345    serde_json::from_value(serde_json::Value::String(s)).map_err(|e| conv_err(idx, e.to_string()))
346}
347
348fn json_col<T: DeserializeOwned>(r: &Row, idx: usize) -> rusqlite::Result<T> {
349    let s: String = r.get(idx)?;
350    serde_json::from_str(&s).map_err(|e| conv_err(idx, e.to_string()))
351}
352
353fn template_from_row(r: &Row) -> rusqlite::Result<TemplateRow> {
354    Ok(TemplateRow {
355        name: r.get(0)?,
356        database: r.get(1)?,
357        source: json_col(r, 2)?,
358        migrations: r.get::<_, Option<String>>(3)?.map(PathBuf::from),
359        max_age_secs: r.get::<_, i64>(4)? as u64,
360        current_gen: r.get::<_, i64>(5)? as u32,
361        next_gen: r.get::<_, i64>(6)? as u32,
362        keep: r.get::<_, i64>(7)? as u32,
363        every_secs: r.get::<_, Option<i64>>(8)?.map(|v| v as u64),
364        every_mode: enum_col(r, 9)?,
365        migrate: match r.get::<_, Option<String>>(10)? {
366            Some(s) => Some(serde_json::from_str(&s).map_err(|e| conv_err(10, e.to_string()))?),
367            None => None,
368        },
369        auto_migrate: r.get::<_, i64>(11)? != 0,
370        on_stale: enum_col(r, 12)?,
371        paused: r.get::<_, i64>(13)? != 0,
372        last_error: r.get(14)?,
373        last_attempt_at: r.get(15)?,
374        mask_salt: r.get(16)?,
375        mask_path: r.get::<_, Option<String>>(17)?.map(PathBuf::from),
376        base: r.get(18)?,
377        branch: r.get(19)?,
378        base_gen: r.get::<_, Option<i64>>(20)?.map(|v| v as u32),
379        base_branch: r.get(21)?,
380        repo_rel_cwd: r.get::<_, Option<String>>(22)?.map(PathBuf::from),
381        repo_rel_migrations: r.get::<_, Option<String>>(23)?.map(PathBuf::from),
382        last_used: r.get(24)?,
383    })
384}
385
386fn gen_from_row(r: &Row) -> rusqlite::Result<GenerationRow> {
387    Ok(GenerationRow {
388        template: r.get(0)?,
389        gen: r.get::<_, i64>(1)? as u32,
390        kind: enum_col(r, 2)?,
391        dir: PathBuf::from(r.get::<_, String>(3)?),
392        pg_major: r.get::<_, i64>(4)? as u32,
393        bin_dir: PathBuf::from(r.get::<_, String>(5)?),
394        extensions: json_col(r, 6)?,
395        migrations_hash: r.get(7)?,
396        built_at: r.get(8)?,
397        size_bytes: r.get::<_, i64>(9)? as u64,
398        from_clone: r.get(10)?,
399        from_gen: r.get::<_, Option<i64>>(11)?.map(|v| v as u32),
400        mask_hash: r.get(12)?,
401    })
402}
403
404fn clone_from_row(r: &Row) -> rusqlite::Result<CloneRow> {
405    let status: String = r.get(15)?;
406    Ok(CloneRow {
407        id: r.get(0)?,
408        name: r.get(1)?,
409        template: r.get(2)?,
410        dir: PathBuf::from(r.get::<_, String>(3)?),
411        port: r.get::<_, i64>(4)? as u16,
412        socket_dir: PathBuf::from(r.get::<_, String>(5)?),
413        password: r.get(6)?,
414        database: r.get(7)?,
415        pg_pid: r.get::<_, Option<i64>>(8)?.map(|v| v as i32),
416        pg_start_time: r.get::<_, Option<i64>>(9)?.map(|v| v as u64),
417        parent_pid: r.get::<_, Option<i64>>(10)?.map(|v| v as i32),
418        parent_start_time: r.get::<_, Option<i64>>(11)?.map(|v| v as u64),
419        created_at: r.get(12)?,
420        expires_at: r.get(13)?,
421        durable: r.get::<_, i64>(14)? != 0,
422        status: CloneStatus::parse(&status)
423            .ok_or_else(|| conv_err(15, format!("bad status {status}")))?,
424        generation: r.get::<_, i64>(16)? as u32,
425        protected: r.get::<_, i64>(17)? != 0,
426        suspect: r.get::<_, i64>(18)? != 0,
427    })
428}
429
430fn p(path: &Path) -> String {
431    path.to_string_lossy().into_owned()
432}
433
434// Shared by insert/update; column order matches CLONE_COLS.
435macro_rules! clone_params {
436    ($c:expr) => {
437        params![
438            $c.id,
439            $c.name,
440            $c.template,
441            p(&$c.dir),
442            $c.port as i64,
443            p(&$c.socket_dir),
444            $c.password,
445            $c.database,
446            $c.pg_pid.map(|v| v as i64),
447            $c.pg_start_time.map(|v| v as i64),
448            $c.parent_pid.map(|v| v as i64),
449            $c.parent_start_time.map(|v| v as i64),
450            $c.created_at,
451            $c.expires_at,
452            $c.durable as i64,
453            $c.status.as_str(),
454            $c.generation as i64,
455            $c.protected as i64,
456            $c.suspect as i64,
457        ]
458    };
459}
460
461// Shared by insert/update; column order matches TEMPLATE_COLS.
462macro_rules! template_params {
463    ($t:expr) => {
464        params![
465            $t.name,
466            $t.database,
467            serde_json::to_string(&$t.source)?,
468            $t.migrations.as_deref().map(p),
469            $t.max_age_secs as i64,
470            $t.current_gen as i64,
471            $t.next_gen as i64,
472            $t.keep as i64,
473            $t.every_secs.map(|v| v as i64),
474            enum_str(&$t.every_mode),
475            $t.migrate.as_ref().map(serde_json::to_string).transpose()?,
476            $t.auto_migrate as i64,
477            enum_str(&$t.on_stale),
478            $t.paused as i64,
479            $t.last_error,
480            $t.last_attempt_at,
481            $t.mask_salt,
482            $t.mask_path.as_deref().map(p),
483            $t.base,
484            $t.branch,
485            $t.base_gen.map(|v| v as i64),
486            $t.base_branch,
487            $t.repo_rel_cwd.as_deref().map(p),
488            $t.repo_rel_migrations.as_deref().map(p),
489            $t.last_used,
490        ]
491    };
492}
493
494/// Adds the `ADDED_COLUMNS` a table lacks.
495fn add_missing_columns(conn: &Connection) -> Result<()> {
496    for (table, col, decl) in ADDED_COLUMNS {
497        let has = conn.query_row(
498            "SELECT count(*) FROM pragma_table_info(?1) WHERE name = ?2",
499            params![table, col],
500            |r| r.get::<_, i64>(0),
501        )? > 0;
502        if !has {
503            conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {col} {decl}"))?;
504        }
505    }
506    Ok(())
507}
508
509/// Brings a store of any older version to `SCHEMA_VERSION` in one transaction.
510fn migrate(conn: &mut Connection) -> Result<()> {
511    let version: i32 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
512    if version > SCHEMA_VERSION {
513        return Err(MitosisError::new(
514            ErrorCode::Internal,
515            format!(
516                "state.db has schema version {version}; this mitosis knows up to {SCHEMA_VERSION}"
517            ),
518        )
519        .with_hint("run a newer mitosis"));
520    }
521    conn.execute_batch(RESERVATIONS)?;
522    conn.execute_batch(ENV_TABLES)?;
523    if version == SCHEMA_VERSION {
524        return add_missing_columns(conn);
525    }
526    let tx = conn.transaction()?;
527    // A v0.1 store has a templates table but no version; a new file has no tables. A v2 store
528    // has a templates table too, but already in the new shape.
529    let v01 = version < 2
530        && tx.query_row(
531            "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = 'templates'",
532            [],
533            |r| r.get::<_, i64>(0),
534        )? > 0;
535    if v01 {
536        tx.execute_batch(MIGRATE_V1)?;
537    }
538    tx.execute_batch(SCHEMA)?;
539    add_missing_columns(&tx)?;
540    if v01 {
541        tx.execute_batch(MIGRATE_V1_ROWS)?;
542    }
543    tx.execute_batch(&format!("PRAGMA user_version = {SCHEMA_VERSION}"))?;
544    tx.commit()?;
545    Ok(())
546}
547
548impl Store {
549    pub fn open(path: &Path) -> Result<Store> {
550        // Create the file 0600 first so the -wal/-shm files SQLite derives from it inherit that mode.
551        std::fs::OpenOptions::new()
552            .create(true)
553            .append(true)
554            .mode(0o600)
555            .open(path)?;
556        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
557        let mut conn = Connection::open(path)?;
558        // journal_mode returns a row, so read it instead of executing.
559        conn.query_row("PRAGMA journal_mode=WAL", [], |_| Ok(()))?;
560        migrate(&mut conn)?;
561        Ok(Store { conn })
562    }
563
564    /// Fails when the name is taken.
565    pub fn insert_template(&self, t: &TemplateRow) -> Result<()> {
566        self.conn.execute(
567            &format!("INSERT INTO templates ({TEMPLATE_COLS}) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,?23,?24,?25)"),
568            template_params!(t),
569        )?;
570        Ok(())
571    }
572
573    /// Writes every column except the generation pointers (`current_gen`, `next_gen`).
574    pub fn update_template_settings(&self, t: &TemplateRow) -> Result<()> {
575        let n = self.conn.execute(
576            "UPDATE templates SET database=?2, source=?3, migrations=?4, max_age_secs=?5, keep=?8, \
577             every_secs=?9, every_mode=?10, migrate=?11, auto_migrate=?12, on_stale=?13, paused=?14, \
578             last_error=?15, last_attempt_at=?16, mask_path=?18, base=?19, branch=?20, \
579             base_gen=?21, base_branch=?22, repo_rel_cwd=?23, repo_rel_migrations=?24, \
580             last_used=?25 WHERE name=?1",
581            template_params!(t),
582        )?;
583        if n == 0 {
584            return Err(MitosisError::new(
585                ErrorCode::TemplateNotFound,
586                format!("template {} not found", t.name),
587            ));
588        }
589        Ok(())
590    }
591
592    /// Stores `salt` as the masking salt of `name` unless it already has one; returns the salt in
593    /// effect, so every build of a template masks with the same key.
594    pub fn ensure_mask_salt(&self, name: &str, salt: &str) -> Result<String> {
595        self.conn.execute(
596            "UPDATE templates SET mask_salt = ?2 WHERE name = ?1 AND mask_salt IS NULL",
597            params![name, salt],
598        )?;
599        self.conn
600            .query_row(
601                "SELECT mask_salt FROM templates WHERE name = ?1",
602                [name],
603                |r| r.get::<_, Option<String>>(0),
604            )
605            .optional()?
606            .flatten()
607            .ok_or_else(|| {
608                MitosisError::new(
609                    ErrorCode::TemplateNotFound,
610                    format!("template {name} not found"),
611                )
612            })
613    }
614
615    /// Removes the template and all its generation rows (not their folders).
616    /// Removes the template and its generation rows together, so a failure never leaves one half.
617    /// A base template takes its branch templates (`base = name`) with it.
618    pub fn delete_template(&self, name: &str) -> Result<()> {
619        let tx = self.conn.unchecked_transaction()?;
620        tx.execute(
621            "DELETE FROM generations WHERE template = ?1 \
622             OR template IN (SELECT name FROM templates WHERE base = ?1)",
623            [name],
624        )?;
625        tx.execute("DELETE FROM templates WHERE name = ?1 OR base = ?1", [name])?;
626        tx.commit()?;
627        Ok(())
628    }
629
630    pub fn get_template(&self, name: &str) -> Result<Option<TemplateRow>> {
631        Ok(self
632            .conn
633            .query_row(
634                &format!("SELECT {TEMPLATE_COLS} FROM templates WHERE name = ?1"),
635                [name],
636                template_from_row,
637            )
638            .optional()?)
639    }
640
641    pub fn list_templates(&self) -> Result<Vec<TemplateRow>> {
642        let mut stmt = self.conn.prepare(&format!(
643            "SELECT {TEMPLATE_COLS} FROM templates ORDER BY name"
644        ))?;
645        let rows = stmt
646            .query_map([], template_from_row)?
647            .collect::<rusqlite::Result<Vec<_>>>()?;
648        Ok(rows)
649    }
650
651    /// Returns the template's next generation number and moves `next_gen` past it.
652    pub fn alloc_gen(&self, name: &str) -> Result<u32> {
653        let gen = self
654            .conn
655            .query_row(
656                "UPDATE templates SET next_gen = next_gen + 1 WHERE name = ?1 RETURNING next_gen - 1",
657                [name],
658                |r| r.get::<_, i64>(0),
659            )
660            .optional()?;
661        gen.map(|g| g as u32).ok_or_else(|| {
662            MitosisError::new(
663                ErrorCode::TemplateNotFound,
664                format!("template {name} not found"),
665            )
666        })
667    }
668
669    pub fn insert_generation(&self, g: &GenerationRow) -> Result<()> {
670        self.conn.execute(
671            &format!("INSERT INTO generations ({GEN_COLS}) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13)"),
672            params![
673                g.template,
674                g.gen as i64,
675                enum_str(&g.kind),
676                p(&g.dir),
677                g.pg_major as i64,
678                p(&g.bin_dir),
679                serde_json::to_string(&g.extensions)?,
680                g.migrations_hash,
681                g.built_at,
682                g.size_bytes as i64,
683                g.from_clone,
684                g.from_gen.map(|v| v as i64),
685                g.mask_hash,
686            ],
687        )?;
688        Ok(())
689    }
690
691    pub fn get_generation(&self, name: &str, gen: u32) -> Result<Option<GenerationRow>> {
692        Ok(self
693            .conn
694            .query_row(
695                &format!("SELECT {GEN_COLS} FROM generations WHERE template = ?1 AND gen = ?2"),
696                params![name, gen as i64],
697                gen_from_row,
698            )
699            .optional()?)
700    }
701
702    /// Oldest first.
703    pub fn list_generations(&self, name: &str) -> Result<Vec<GenerationRow>> {
704        let mut stmt = self.conn.prepare(&format!(
705            "SELECT {GEN_COLS} FROM generations WHERE template = ?1 ORDER BY gen"
706        ))?;
707        let rows = stmt
708            .query_map([name], gen_from_row)?
709            .collect::<rusqlite::Result<Vec<_>>>()?;
710        Ok(rows)
711    }
712
713    /// The template and its current generation; `None` while it has none.
714    pub fn current(&self, name: &str) -> Result<Option<(TemplateRow, GenerationRow)>> {
715        let Some(t) = self.get_template(name)? else {
716            return Ok(None);
717        };
718        Ok(self.get_generation(name, t.current_gen)?.map(|g| (t, g)))
719    }
720
721    /// Compare-and-set of the current pointer: changes it only while it still equals `expected`
722    /// (`None`: always) and generation `gen` exists. True when it was changed.
723    pub fn set_current(&self, name: &str, gen: u32, expected: Option<u32>) -> Result<bool> {
724        Ok(self.conn.execute(
725            "UPDATE templates SET current_gen = ?2 WHERE name = ?1 AND (?3 IS NULL OR current_gen = ?3) \
726             AND EXISTS (SELECT 1 FROM generations WHERE template = ?1 AND gen = ?2)",
727            params![name, gen as i64, expected.map(|e| e as i64)],
728        )? > 0)
729    }
730
731    pub fn delete_generation(&self, name: &str, gen: u32) -> Result<()> {
732        self.conn.execute(
733            "DELETE FROM generations WHERE template = ?1 AND gen = ?2",
734            params![name, gen as i64],
735        )?;
736        Ok(())
737    }
738
739    pub fn insert_clone(&self, c: &CloneRow) -> Result<()> {
740        self.conn
741            .execute(
742                &format!("INSERT INTO clones ({CLONE_COLS}) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19)"),
743                clone_params!(c),
744            )
745            .map_err(|e| match &e {
746                rusqlite::Error::SqliteFailure(f, _) if f.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE => {
747                    MitosisError::new(
748                        ErrorCode::InvalidRequest,
749                        format!("a clone named '{}' already exists", c.name.as_deref().unwrap_or(&c.id)),
750                    )
751                }
752                _ => e.into(),
753            })?;
754        Ok(())
755    }
756
757    /// `CloneNotFound` when the row is gone (for example destroyed meanwhile).
758    pub fn update_clone(&self, c: &CloneRow) -> Result<()> {
759        let n = self.conn.execute(
760            "UPDATE clones SET name=?2, template=?3, dir=?4, port=?5, socket_dir=?6, password=?7, database=?8, \
761             pg_pid=?9, pg_start_time=?10, parent_pid=?11, parent_start_time=?12, created_at=?13, expires_at=?14, \
762             durable=?15, status=?16, generation=?17, protected=?18, suspect=?19 WHERE id=?1",
763            clone_params!(c),
764        )?;
765        if n == 0 {
766            return Err(MitosisError::new(
767                ErrorCode::CloneNotFound,
768                format!("clone {} no longer exists", c.id),
769            ));
770        }
771        Ok(())
772    }
773
774    /// Changes only the status column; false when no row has this id.
775    pub fn set_clone_status(&self, id: &str, status: CloneStatus) -> Result<bool> {
776        Ok(self.conn.execute(
777            "UPDATE clones SET status = ?2 WHERE id = ?1",
778            params![id, status.as_str()],
779        )? > 0)
780    }
781
782    /// Matches id first, then name.
783    pub fn get_clone(&self, id_or_name: &str) -> Result<Option<CloneRow>> {
784        Ok(self
785            .conn
786            .query_row(
787                &format!("SELECT {CLONE_COLS} FROM clones WHERE id = ?1 OR name = ?1 ORDER BY id = ?1 DESC LIMIT 1"),
788                [id_or_name],
789                clone_from_row,
790            )
791            .optional()?)
792    }
793
794    pub fn list_clones(&self) -> Result<Vec<CloneRow>> {
795        let mut stmt = self.conn.prepare(&format!(
796            "SELECT {CLONE_COLS} FROM clones ORDER BY created_at, id"
797        ))?;
798        let rows = stmt
799            .query_map([], clone_from_row)?
800            .collect::<rusqlite::Result<Vec<_>>>()?;
801        Ok(rows)
802    }
803
804    pub fn delete_clone(&self, id: &str) -> Result<()> {
805        self.conn
806            .execute("DELETE FROM clones WHERE id = ?1", [id])?;
807        Ok(())
808    }
809
810    /// Remembers a `.env` entry that holds the clone's URL. A second record for the same file and
811    /// variable keeps the first `previous` (the user's own value) and only updates the URL.
812    pub fn add_env_record(&self, clone_id: &str, rec: &PendingRestore) -> Result<()> {
813        self.conn.execute(
814            "INSERT INTO env_records (clone_id, file, var, previous, url) VALUES (?1,?2,?3,?4,?5) \
815             ON CONFLICT(clone_id, file, var) DO UPDATE SET url = excluded.url",
816            params![
817                clone_id,
818                rec.file.to_string_lossy(),
819                rec.var,
820                rec.previous,
821                rec.url
822            ],
823        )?;
824        Ok(())
825    }
826
827    /// The clone's `.env` records as restores (`id` is 0).
828    pub fn env_records(&self, clone_id: &str) -> Result<Vec<PendingRestore>> {
829        let mut stmt = self.conn.prepare(
830            "SELECT file, var, previous, url FROM env_records WHERE clone_id = ?1 ORDER BY rowid",
831        )?;
832        let rows = stmt
833            .query_map([clone_id], |r| {
834                Ok(PendingRestore {
835                    id: 0,
836                    file: PathBuf::from(r.get::<_, String>(0)?),
837                    var: r.get(1)?,
838                    previous: r.get(2)?,
839                    url: r.get(3)?,
840                })
841            })?
842            .collect::<rusqlite::Result<Vec<_>>>()?;
843        Ok(rows)
844    }
845
846    pub fn delete_env_records(&self, clone_id: &str) -> Result<()> {
847        self.conn
848            .execute("DELETE FROM env_records WHERE clone_id = ?1", [clone_id])?;
849        Ok(())
850    }
851
852    pub fn add_pending_restore(&self, rec: &PendingRestore) -> Result<()> {
853        self.conn.execute(
854            "INSERT INTO pending_restores (file, var, previous, url) VALUES (?1,?2,?3,?4)",
855            params![rec.file.to_string_lossy(), rec.var, rec.previous, rec.url],
856        )?;
857        Ok(())
858    }
859
860    pub fn list_pending_restores(&self) -> Result<Vec<PendingRestore>> {
861        let mut stmt = self
862            .conn
863            .prepare("SELECT id, file, var, previous, url FROM pending_restores ORDER BY id")?;
864        let rows = stmt
865            .query_map([], |r| {
866                Ok(PendingRestore {
867                    id: r.get(0)?,
868                    file: PathBuf::from(r.get::<_, String>(1)?),
869                    var: r.get(2)?,
870                    previous: r.get(3)?,
871                    url: r.get(4)?,
872                })
873            })?
874            .collect::<rusqlite::Result<Vec<_>>>()?;
875        Ok(rows)
876    }
877
878    pub fn delete_pending_restore(&self, id: i64) -> Result<()> {
879        self.conn
880            .execute("DELETE FROM pending_restores WHERE id = ?1", [id])?;
881        Ok(())
882    }
883
884    pub fn get_reservation(&self, name: &str) -> Result<Option<ReservationRow>> {
885        Ok(self
886            .conn
887            .query_row(
888                "SELECT name, port, password, template, last_used FROM reservations WHERE name = ?1",
889                [name],
890                |r| {
891                    Ok(ReservationRow {
892                        name: r.get(0)?,
893                        port: r.get::<_, i64>(1)? as u16,
894                        password: r.get(2)?,
895                        template: r.get(3)?,
896                        last_used: r.get(4)?,
897                    })
898                },
899            )
900            .optional()?)
901    }
902
903    pub fn upsert_reservation(&self, r: &ReservationRow) -> Result<()> {
904        self.conn.execute(
905            "INSERT OR REPLACE INTO reservations (name, port, password, template, last_used) VALUES (?1,?2,?3,?4,?5)",
906            params![r.name, i64::from(r.port), r.password, r.template, r.last_used],
907        )?;
908        Ok(())
909    }
910
911    pub fn delete_reservation(&self, name: &str) -> Result<()> {
912        self.conn
913            .execute("DELETE FROM reservations WHERE name = ?1", [name])?;
914        Ok(())
915    }
916
917    /// Appends one entry to the event log (the time is now).
918    pub fn add_event(
919        &self,
920        kind: &str,
921        clone: Option<&str>,
922        template: Option<&str>,
923        detail: &serde_json::Value,
924    ) -> Result<EventRow> {
925        let ts = std::time::SystemTime::now()
926            .duration_since(std::time::UNIX_EPOCH)
927            .map_or(0, |d| d.as_secs() as i64);
928        let id = self.conn.query_row(
929            "INSERT INTO events (ts, kind, clone, template, detail) VALUES (?1,?2,?3,?4,?5) RETURNING id",
930            params![ts, kind, clone, template, serde_json::to_string(detail)?],
931            |r| r.get::<_, i64>(0),
932        )?;
933        Ok(EventRow {
934            id,
935            ts,
936            kind: kind.to_string(),
937            clone: clone.map(String::from),
938            template: template.map(String::from),
939            detail: detail.clone(),
940        })
941    }
942
943    /// The events matching `f`, oldest first; with a limit, the newest `limit` of them.
944    pub fn list_events(&self, f: &EventFilter) -> Result<Vec<EventRow>> {
945        self.query_events(0, f)
946    }
947
948    /// Every event with `id > after` that matches `f` (its `limit` is ignored), oldest first.
949    /// A follower reads the log with this, so it can neither lose nor reorder an event.
950    pub fn list_events_after(&self, after: i64, f: &EventFilter) -> Result<Vec<EventRow>> {
951        let all = EventFilter {
952            limit: None,
953            ..f.clone()
954        };
955        self.query_events(after, &all)
956    }
957
958    fn query_events(&self, after: i64, f: &EventFilter) -> Result<Vec<EventRow>> {
959        // A clone filter matches the name or id column, or the id kept in a named clone's detail.
960        let mut stmt = self.conn.prepare(
961            "SELECT id, ts, kind, clone, template, detail FROM (
962                 SELECT id, ts, kind, clone, template, detail FROM events
963                 WHERE id > ?6
964                   AND (?1 IS NULL OR clone = ?1 OR json_extract(detail, '$.id') = ?1)
965                   AND (?2 IS NULL OR template = ?2)
966                   AND (?3 IS NULL OR kind = ?3 OR substr(kind, 1, length(?3) + 1) = ?3 || '.')
967                   AND (?4 IS NULL OR ts >= ?4)
968                 ORDER BY id DESC LIMIT ?5
969             ) ORDER BY id",
970        )?;
971        let rows = stmt
972            .query_map(
973                params![
974                    f.clone,
975                    f.template,
976                    f.kind,
977                    f.since,
978                    f.limit.map_or(-1, i64::from),
979                    after
980                ],
981                |r| {
982                    Ok(EventRow {
983                        id: r.get(0)?,
984                        ts: r.get(1)?,
985                        kind: r.get(2)?,
986                        clone: r.get(3)?,
987                        template: r.get(4)?,
988                        detail: json_col(r, 5)?,
989                    })
990                },
991            )?
992            .collect::<rusqlite::Result<Vec<_>>>()?;
993        Ok(rows)
994    }
995
996    /// Deletes events older than `before_ts` and all but the newest `max_rows`; returns how many.
997    pub fn prune_events(&self, before_ts: i64, max_rows: u32) -> Result<usize> {
998        Ok(self.conn.execute(
999            "DELETE FROM events WHERE ts < ?1 \
1000             OR id <= (SELECT id FROM events ORDER BY id DESC LIMIT 1 OFFSET ?2)",
1001            params![before_ts, i64::from(max_rows)],
1002        )?)
1003    }
1004
1005    /// Every reserved port, whether or not a clone of that name runs.
1006    pub fn reserved_ports(&self) -> Result<HashSet<u16>> {
1007        let mut stmt = self.conn.prepare("SELECT port FROM reservations")?;
1008        let ports = stmt
1009            .query_map([], |r| Ok(r.get::<_, i64>(0)? as u16))?
1010            .collect::<rusqlite::Result<HashSet<_>>>()?;
1011        Ok(ports)
1012    }
1013
1014    /// Deletes reservations last used before `before` whose name has no clone; returns their names.
1015    pub fn drop_stale_reservations(&self, before: i64) -> Result<Vec<String>> {
1016        let mut stmt = self.conn.prepare(
1017            "DELETE FROM reservations WHERE last_used < ?1 \
1018                 AND name NOT IN (SELECT name FROM clones WHERE name IS NOT NULL) RETURNING name",
1019        )?;
1020        let names = stmt
1021            .query_map([before], |r| r.get::<_, String>(0))?
1022            .collect::<rusqlite::Result<Vec<_>>>()?;
1023        Ok(names)
1024    }
1025
1026    /// The clone row is the port lease.
1027    pub fn leased_ports(&self) -> Result<HashSet<u16>> {
1028        let mut stmt = self.conn.prepare("SELECT port FROM clones")?;
1029        let ports = stmt
1030            .query_map([], |r| Ok(r.get::<_, i64>(0)? as u16))?
1031            .collect::<rusqlite::Result<HashSet<_>>>()?;
1032        Ok(ports)
1033    }
1034}
1035
1036#[cfg(test)]
1037mod tests {
1038    use super::*;
1039
1040    fn tpl(name: &str) -> TemplateRow {
1041        TemplateRow::new(
1042            name,
1043            "app",
1044            TemplateSource::Url {
1045                url: "postgres://x".into(),
1046            },
1047            Some(PathBuf::from("/tmp/m")),
1048            604_800,
1049        )
1050    }
1051
1052    fn gen(name: &str, gen: u32) -> GenerationRow {
1053        GenerationRow {
1054            template: name.into(),
1055            gen,
1056            kind: GenKind::Full,
1057            dir: PathBuf::from(format!("/tmp/t/{name}@{gen}")),
1058            pg_major: 16,
1059            bin_dir: PathBuf::from("/opt/pg16/bin"),
1060            extensions: vec!["pgcrypto".into()],
1061            migrations_hash: Some("abc".into()),
1062            built_at: 1_700_000_000,
1063            size_bytes: 1,
1064            from_clone: None,
1065            from_gen: None,
1066            mask_hash: Some("m1".into()),
1067        }
1068    }
1069
1070    fn clone_row(id: &str, name: Option<&str>, port: u16) -> CloneRow {
1071        CloneRow {
1072            id: id.into(),
1073            name: name.map(String::from),
1074            template: "main".into(),
1075            dir: PathBuf::from("/tmp/c"),
1076            port,
1077            socket_dir: PathBuf::from("/tmp/s"),
1078            password: "pw".into(),
1079            database: "app".into(),
1080            pg_pid: Some(123),
1081            pg_start_time: Some(456),
1082            parent_pid: None,
1083            parent_start_time: None,
1084            created_at: 1_700_000_000,
1085            expires_at: 1_700_007_200,
1086            durable: false,
1087            status: CloneStatus::Ready,
1088            generation: 1,
1089            protected: false,
1090            suspect: false,
1091        }
1092    }
1093
1094    #[test]
1095    fn template_roundtrip_and_settings() {
1096        let dir = tempfile::tempdir().unwrap();
1097        let store = Store::open(&dir.path().join("state.db")).unwrap();
1098        assert!(store.get_template("main").unwrap().is_none());
1099        store.insert_template(&tpl("main")).unwrap();
1100        assert!(store.insert_template(&tpl("main")).is_err());
1101        assert_eq!(store.get_template("main").unwrap().unwrap(), tpl("main"));
1102        let changed = TemplateRow {
1103            keep: 5,
1104            every_secs: Some(3600),
1105            every_mode: RefreshMode::Layered,
1106            migrate: Some(MigrateCmd {
1107                command: "make migrate".into(),
1108                cwd: "/w".into(),
1109                path: Some("/bin".into()),
1110            }),
1111            auto_migrate: true,
1112            on_stale: OnStale::Refresh,
1113            paused: true,
1114            last_error: Some("boom".into()),
1115            last_attempt_at: Some(7),
1116            // pointers are not settings: these must be ignored
1117            current_gen: 9,
1118            next_gen: 9,
1119            ..tpl("main")
1120        };
1121        store.update_template_settings(&changed).unwrap();
1122        assert_eq!(
1123            store.get_template("main").unwrap().unwrap(),
1124            TemplateRow {
1125                current_gen: 0,
1126                next_gen: 1,
1127                ..changed
1128            }
1129        );
1130        assert_eq!(
1131            store
1132                .update_template_settings(&tpl("nope"))
1133                .unwrap_err()
1134                .code,
1135            ErrorCode::TemplateNotFound
1136        );
1137        store.insert_template(&tpl("other")).unwrap();
1138        assert_eq!(store.list_templates().unwrap().len(), 2);
1139        store.delete_template("other").unwrap();
1140        assert_eq!(store.list_templates().unwrap().len(), 1);
1141    }
1142
1143    #[test]
1144    fn delete_template_is_all_or_nothing() {
1145        let dir = tempfile::tempdir().unwrap();
1146        let store = Store::open(&dir.path().join("state.db")).unwrap();
1147        store.insert_template(&tpl("main")).unwrap();
1148        store.insert_generation(&gen("main", 1)).unwrap();
1149        // A failing second delete must take the first one back with it.
1150        store
1151            .conn
1152            .execute_batch(
1153                "CREATE TRIGGER no_tpl_delete BEFORE DELETE ON templates
1154                 BEGIN SELECT RAISE(ABORT, 'no'); END;",
1155            )
1156            .unwrap();
1157        assert!(store.delete_template("main").is_err());
1158        assert_eq!(store.list_generations("main").unwrap().len(), 1);
1159        store
1160            .conn
1161            .execute_batch("DROP TRIGGER no_tpl_delete")
1162            .unwrap();
1163        store.delete_template("main").unwrap();
1164        assert!(store.list_generations("main").unwrap().is_empty());
1165        assert!(store.get_template("main").unwrap().is_none());
1166    }
1167
1168    #[test]
1169    fn generations_alloc_cas_and_delete() {
1170        let dir = tempfile::tempdir().unwrap();
1171        let store = Store::open(&dir.path().join("state.db")).unwrap();
1172        assert_eq!(
1173            store.alloc_gen("main").unwrap_err().code,
1174            ErrorCode::TemplateNotFound
1175        );
1176        store.insert_template(&tpl("main")).unwrap();
1177        assert!(store.current("main").unwrap().is_none());
1178        assert_eq!(store.alloc_gen("main").unwrap(), 1);
1179        assert_eq!(store.alloc_gen("main").unwrap(), 2);
1180        // gen 2 never got a row; numbers still move on
1181        assert_eq!(store.alloc_gen("main").unwrap(), 3);
1182        store.insert_generation(&gen("main", 3)).unwrap();
1183        store.insert_generation(&gen("main", 1)).unwrap();
1184        assert!(store.insert_generation(&gen("main", 1)).is_err());
1185        assert_eq!(
1186            store
1187                .list_generations("main")
1188                .unwrap()
1189                .iter()
1190                .map(|g| g.gen)
1191                .collect::<Vec<_>>(),
1192            vec![1, 3]
1193        );
1194        assert_eq!(
1195            store.get_generation("main", 3).unwrap().unwrap(),
1196            gen("main", 3)
1197        );
1198
1199        // CAS: wrong expectation and missing generation both refuse
1200        assert!(!store.set_current("main", 3, Some(1)).unwrap());
1201        assert!(!store.set_current("main", 2, Some(0)).unwrap());
1202        assert!(store.set_current("main", 3, Some(0)).unwrap());
1203        let (t, g) = store.current("main").unwrap().unwrap();
1204        assert_eq!((t.current_gen, g.gen), (3, 3));
1205        assert!(store.set_current("main", 1, None).unwrap());
1206        assert_eq!(store.current("main").unwrap().unwrap().1.gen, 1);
1207
1208        store.delete_generation("main", 3).unwrap();
1209        assert!(store.get_generation("main", 3).unwrap().is_none());
1210        assert_eq!(store.alloc_gen("main").unwrap(), 4);
1211    }
1212
1213    /// The v0.1 schema, verbatim.
1214    const V01_SCHEMA: &str = "
1215CREATE TABLE IF NOT EXISTS templates (
1216    name TEXT PRIMARY KEY, dir TEXT NOT NULL, pg_major INTEGER NOT NULL, bin_dir TEXT NOT NULL,
1217    extensions TEXT NOT NULL, database TEXT NOT NULL, source TEXT NOT NULL, migrations TEXT,
1218    migrations_hash TEXT, max_age_secs INTEGER NOT NULL, built_at INTEGER NOT NULL,
1219    size_bytes INTEGER NOT NULL
1220);
1221CREATE TABLE IF NOT EXISTS clones (
1222    id TEXT PRIMARY KEY, name TEXT, template TEXT NOT NULL, dir TEXT NOT NULL, port INTEGER NOT NULL,
1223    socket_dir TEXT NOT NULL, password TEXT NOT NULL, database TEXT NOT NULL, pg_pid INTEGER,
1224    pg_start_time INTEGER, parent_pid INTEGER, parent_start_time INTEGER, created_at INTEGER NOT NULL,
1225    expires_at INTEGER NOT NULL, durable INTEGER NOT NULL, status TEXT NOT NULL
1226);
1227CREATE UNIQUE INDEX IF NOT EXISTS clones_name_uniq ON clones(name) WHERE name IS NOT NULL;
1228";
1229
1230    #[test]
1231    fn v01_store_is_migrated() {
1232        let dir = tempfile::tempdir().unwrap();
1233        let path = dir.path().join("state.db");
1234        {
1235            let conn = Connection::open(&path).unwrap();
1236            conn.execute_batch(V01_SCHEMA).unwrap();
1237            conn.execute(
1238                "INSERT INTO templates VALUES ('main', '/h/templates/main', 16, '/opt/pg16/bin', \
1239                 '[\"plpgsql\"]', 'app', '{\"kind\":\"url\",\"url\":\"postgres://x\"}', NULL, 'h1', 604800, 5, 9)",
1240                [],
1241            )
1242            .unwrap();
1243            conn.execute(
1244                "INSERT INTO clones VALUES ('c1', 'x', 'main', '/h/clones/c1', 54320, '/h/s/c1', 'pw', \
1245                 'app', 1, 2, NULL, NULL, 3, 4, 0, 'ready')",
1246                [],
1247            )
1248            .unwrap();
1249        }
1250        let store = Store::open(&path).unwrap();
1251        let (t, g) = store.current("main").unwrap().unwrap();
1252        assert_eq!(
1253            (t.current_gen, t.next_gen, t.keep, t.paused),
1254            (1, 2, DEFAULT_KEEP, false)
1255        );
1256        assert_eq!(t.database, "app");
1257        assert_eq!(
1258            (g.gen, g.kind, g.dir.as_path(), g.pg_major),
1259            (1, GenKind::Full, Path::new("/h/templates/main"), 16)
1260        );
1261        assert_eq!(
1262            (
1263                g.extensions,
1264                g.migrations_hash.as_deref(),
1265                g.built_at,
1266                g.size_bytes
1267            ),
1268            (vec!["plpgsql".to_string()], Some("h1"), 5, 9)
1269        );
1270        let c = store.get_clone("c1").unwrap().unwrap();
1271        assert_eq!((c.generation, c.protected, c.suspect), (1, false, false));
1272        assert_eq!(store.alloc_gen("main").unwrap(), 2);
1273        drop(store);
1274        // Reopening is a no-op; a newer store is refused.
1275        let store = Store::open(&path).unwrap();
1276        assert_eq!(store.list_generations("main").unwrap().len(), 1);
1277        store
1278            .conn
1279            .execute_batch("PRAGMA user_version = 99")
1280            .unwrap();
1281        drop(store);
1282        assert!(Store::open(&path).is_err());
1283    }
1284
1285    #[test]
1286    fn clone_crud_names_and_ports() {
1287        let dir = tempfile::tempdir().unwrap();
1288        let store = Store::open(&dir.path().join("state.db")).unwrap();
1289        store
1290            .insert_clone(&clone_row("c1", Some("x"), 54320))
1291            .unwrap();
1292        assert_eq!(
1293            store
1294                .insert_clone(&clone_row("c2", Some("x"), 54321))
1295                .unwrap_err()
1296                .code,
1297            ErrorCode::InvalidRequest
1298        );
1299        // unnamed clones never collide on name
1300        store.insert_clone(&clone_row("c3", None, 54322)).unwrap();
1301        store.insert_clone(&clone_row("c4", None, 54323)).unwrap();
1302
1303        assert_eq!(store.get_clone("x").unwrap().unwrap().id, "c1");
1304        assert_eq!(store.get_clone("c1").unwrap().unwrap().port, 54320);
1305        assert_eq!(
1306            store.get_clone("c1").unwrap().unwrap(),
1307            clone_row("c1", Some("x"), 54320)
1308        );
1309        assert!(store.get_clone("nope").unwrap().is_none());
1310        assert!(store.leased_ports().unwrap().contains(&54320));
1311        assert_eq!(store.list_clones().unwrap().len(), 3);
1312
1313        let mut c = clone_row("c1", Some("x"), 54320);
1314        c.status = CloneStatus::Stopping;
1315        c.durable = true;
1316        c.parent_pid = Some(9);
1317        store.update_clone(&c).unwrap();
1318        assert_eq!(store.get_clone("c1").unwrap().unwrap(), c);
1319
1320        assert!(store.set_clone_status("c3", CloneStatus::Stopping).unwrap());
1321        let c3 = store.get_clone("c3").unwrap().unwrap();
1322        assert_eq!(
1323            c3,
1324            CloneRow {
1325                status: CloneStatus::Stopping,
1326                ..clone_row("c3", None, 54322)
1327            }
1328        );
1329
1330        store.delete_clone("c1").unwrap();
1331        assert!(store.get_clone("c1").unwrap().is_none());
1332        assert_eq!(
1333            store.update_clone(&c).unwrap_err().code,
1334            ErrorCode::CloneNotFound
1335        );
1336        assert!(!store.set_clone_status("c1", CloneStatus::Ready).unwrap());
1337        assert!(!store.leased_ports().unwrap().contains(&54320));
1338    }
1339
1340    #[test]
1341    fn reopening_keeps_rows_and_file_is_private() {
1342        let dir = tempfile::tempdir().unwrap();
1343        let path = dir.path().join("state.db");
1344        {
1345            let store = Store::open(&path).unwrap();
1346            store.insert_template(&tpl("main")).unwrap();
1347            store
1348                .insert_clone(&clone_row("c1", Some("x"), 54320))
1349                .unwrap();
1350            // while open, the WAL side files exist and must be private too
1351            for suffix in ["-wal", "-shm"] {
1352                let side = dir.path().join(format!("state.db{suffix}"));
1353                let mode = std::fs::metadata(&side).unwrap().permissions().mode() & 0o777;
1354                assert_eq!(mode, 0o600, "{suffix}");
1355            }
1356        }
1357        let store = Store::open(&path).unwrap();
1358        assert!(store.get_template("main").unwrap().is_some());
1359        assert_eq!(store.get_clone("x").unwrap().unwrap().id, "c1");
1360        assert_eq!(
1361            std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
1362            0o600
1363        );
1364    }
1365
1366    /// The v0.2 schema (version 2), verbatim: no `events` table.
1367    const V02_SCHEMA: &str = "
1368CREATE TABLE templates (
1369    name TEXT PRIMARY KEY, database TEXT NOT NULL, source TEXT NOT NULL, migrations TEXT,
1370    max_age_secs INTEGER NOT NULL, current_gen INTEGER NOT NULL, next_gen INTEGER NOT NULL,
1371    keep INTEGER NOT NULL DEFAULT 3, every_secs INTEGER, every_mode TEXT NOT NULL DEFAULT 'full',
1372    migrate TEXT, auto_migrate INTEGER NOT NULL DEFAULT 0, on_stale TEXT NOT NULL DEFAULT 'warn',
1373    paused INTEGER NOT NULL DEFAULT 0, last_error TEXT, last_attempt_at INTEGER
1374);
1375CREATE TABLE generations (
1376    template TEXT NOT NULL, gen INTEGER NOT NULL, kind TEXT NOT NULL, dir TEXT NOT NULL,
1377    pg_major INTEGER NOT NULL, bin_dir TEXT NOT NULL, extensions TEXT NOT NULL, migrations_hash TEXT,
1378    built_at INTEGER NOT NULL, size_bytes INTEGER NOT NULL, from_clone TEXT, from_gen INTEGER,
1379    PRIMARY KEY (template, gen)
1380);
1381CREATE TABLE clones (
1382    id TEXT PRIMARY KEY, name TEXT, template TEXT NOT NULL, dir TEXT NOT NULL, port INTEGER NOT NULL,
1383    socket_dir TEXT NOT NULL, password TEXT NOT NULL, database TEXT NOT NULL, pg_pid INTEGER,
1384    pg_start_time INTEGER, parent_pid INTEGER, parent_start_time INTEGER, created_at INTEGER NOT NULL,
1385    expires_at INTEGER NOT NULL, durable INTEGER NOT NULL, status TEXT NOT NULL,
1386    generation INTEGER NOT NULL DEFAULT 1, protected INTEGER NOT NULL DEFAULT 0,
1387    suspect INTEGER NOT NULL DEFAULT 0
1388);
1389CREATE UNIQUE INDEX clones_name_uniq ON clones(name) WHERE name IS NOT NULL;
1390CREATE TABLE reservations (
1391    name TEXT PRIMARY KEY, port INTEGER NOT NULL, password TEXT NOT NULL, template TEXT NOT NULL,
1392    last_used INTEGER NOT NULL
1393);
1394CREATE TABLE env_records (
1395    clone_id TEXT NOT NULL, file TEXT NOT NULL, var TEXT NOT NULL, previous TEXT, url TEXT NOT NULL,
1396    PRIMARY KEY (clone_id, file, var)
1397);
1398CREATE TABLE pending_restores (
1399    id INTEGER PRIMARY KEY AUTOINCREMENT, file TEXT NOT NULL, var TEXT NOT NULL, previous TEXT,
1400    url TEXT NOT NULL
1401);
1402PRAGMA user_version = 2;
1403";
1404
1405    #[test]
1406    fn v02_store_migrates_to_v3() {
1407        let dir = tempfile::tempdir().unwrap();
1408        let path = dir.path().join("state.db");
1409        {
1410            let conn = Connection::open(&path).unwrap();
1411            conn.execute_batch(V02_SCHEMA).unwrap();
1412            conn.execute_batch(
1413                "INSERT INTO templates (name, database, source, max_age_secs, current_gen, next_gen, keep)
1414                     VALUES ('main', 'app', '{\"kind\":\"seed\",\"command\":\"make seed\"}', 604800, 2, 3, 5);
1415                 INSERT INTO generations VALUES ('main', 1, 'full', '/h/templates/main@1', 16, '/b', '[]', 'h1', 5, 9, NULL, NULL);
1416                 INSERT INTO generations VALUES ('main', 2, 'layered', '/h/templates/main@2', 16, '/b', '[\"pgcrypto\"]', 'h2', 6, 9, NULL, 1);
1417                 INSERT INTO clones VALUES ('c1', 'x', 'main', '/h/clones/c1', 54320, '/h/s/c1', 'pw', 'app',
1418                     1, 2, NULL, NULL, 3, 4, 0, 'ready', 2, 1, 0);
1419                 INSERT INTO reservations VALUES ('x', 54320, 'pw', 'main', 77);
1420                 INSERT INTO env_records VALUES ('c1', '/w/.env', 'DATABASE_URL', 'old', 'postgres://u:***@h/db');
1421                 INSERT INTO pending_restores (file, var, previous, url) VALUES ('/w/.env', 'URL', NULL, 'u');",
1422            )
1423            .unwrap();
1424        }
1425        let store = Store::open(&path).unwrap();
1426        let version: i32 = store
1427            .conn
1428            .query_row("PRAGMA user_version", [], |r| r.get(0))
1429            .unwrap();
1430        assert_eq!(version, SCHEMA_VERSION);
1431        assert_eq!(SCHEMA_VERSION, 3);
1432
1433        let (t, g) = store.current("main").unwrap().unwrap();
1434        assert_eq!((t.current_gen, t.next_gen, t.keep), (2, 3, 5));
1435        assert_eq!((g.gen, g.kind, g.from_gen), (2, GenKind::Layered, Some(1)));
1436        assert_eq!(store.list_generations("main").unwrap().len(), 2);
1437        let c = store.get_clone("x").unwrap().unwrap();
1438        assert_eq!((c.id.as_str(), c.generation, c.protected), ("c1", 2, true));
1439        assert_eq!(store.get_reservation("x").unwrap().unwrap().port, 54320);
1440        let recs = store.env_records("c1").unwrap();
1441        assert_eq!(recs.len(), 1);
1442        assert_eq!(recs[0].previous.as_deref(), Some("old"));
1443        assert_eq!(store.list_pending_restores().unwrap().len(), 1);
1444
1445        // The new table works, and a reopen changes nothing.
1446        let ev = store
1447            .add_event("gc", None, None, &serde_json::json!({ "n": 1 }))
1448            .unwrap();
1449        drop(store);
1450        let store = Store::open(&path).unwrap();
1451        assert_eq!(
1452            store.list_events(&EventFilter::default()).unwrap(),
1453            vec![ev]
1454        );
1455        assert_eq!(store.list_generations("main").unwrap().len(), 2);
1456    }
1457
1458    #[test]
1459    fn mask_columns_reach_v02_and_early_v3_stores() {
1460        for version in [2, 3] {
1461            let dir = tempfile::tempdir().unwrap();
1462            let path = dir.path().join("state.db");
1463            {
1464                let conn = Connection::open(&path).unwrap();
1465                conn.execute_batch(V02_SCHEMA).unwrap();
1466                conn.execute_batch(&format!(
1467                    "INSERT INTO templates (name, database, source, max_age_secs, current_gen, next_gen)
1468                         VALUES ('main', 'app', '{{\"kind\":\"seed\",\"command\":\"x\"}}', 1, 1, 2);
1469                     INSERT INTO generations VALUES ('main', 1, 'full', '/d', 16, '/b', '[]', NULL, 5, 9, NULL, NULL);
1470                     PRAGMA user_version = {version};"
1471                ))
1472                .unwrap();
1473            }
1474            let store = Store::open(&path).unwrap();
1475            let (t, g) = store.current("main").unwrap().unwrap();
1476            assert_eq!((t.mask_path, t.mask_salt, g.mask_hash), (None, None, None));
1477            assert_eq!(store.ensure_mask_salt("main", "aa").unwrap(), "aa");
1478            // The first salt sticks, and a settings write leaves it alone.
1479            assert_eq!(store.ensure_mask_salt("main", "bb").unwrap(), "aa");
1480            let mut t = store.get_template("main").unwrap().unwrap();
1481            t.mask_salt = None;
1482            t.mask_path = Some(PathBuf::from("/w/mask.toml"));
1483            store.update_template_settings(&t).unwrap();
1484            let t = store.get_template("main").unwrap().unwrap();
1485            assert_eq!(t.mask_salt.as_deref(), Some("aa"));
1486            assert_eq!(t.mask_path, Some(PathBuf::from("/w/mask.toml")));
1487            store.insert_generation(&gen("main", 2)).unwrap();
1488            assert_eq!(
1489                store.get_generation("main", 2).unwrap().unwrap(),
1490                gen("main", 2)
1491            );
1492        }
1493    }
1494
1495    #[test]
1496    fn prune_keeps_limits() {
1497        let dir = tempfile::tempdir().unwrap();
1498        let store = Store::open(&dir.path().join("state.db")).unwrap();
1499        for i in 0..10 {
1500            store
1501                .add_event("gc", None, None, &serde_json::json!({ "i": i }))
1502                .unwrap();
1503        }
1504        // Nothing is old, and the row limit is above the count: nothing goes.
1505        assert_eq!(store.prune_events(0, 10).unwrap(), 0);
1506        assert_eq!(store.prune_events(0, 3).unwrap(), 7);
1507        let left = store.list_events(&EventFilter::default()).unwrap();
1508        let kept: Vec<i64> = left
1509            .iter()
1510            .map(|e| e.detail["i"].as_i64().unwrap())
1511            .collect();
1512        assert_eq!(kept, vec![7, 8, 9]);
1513        // Everything is older than a time in the future.
1514        assert_eq!(store.prune_events(i64::MAX, 100).unwrap(), 3);
1515    }
1516
1517    #[test]
1518    fn list_events_filters_and_limits() {
1519        let dir = tempfile::tempdir().unwrap();
1520        let store = Store::open(&dir.path().join("state.db")).unwrap();
1521        let add = |kind, clone: Option<&str>, tpl: Option<&str>, detail| {
1522            store.add_event(kind, clone, tpl, &detail).unwrap()
1523        };
1524        let a = add(
1525            "clone.created",
1526            Some("x"),
1527            Some("main"),
1528            serde_json::json!({ "id": "c1" }),
1529        );
1530        let b = add(
1531            "clone.destroyed",
1532            Some("x"),
1533            Some("main"),
1534            serde_json::json!({ "id": "c1" }),
1535        );
1536        let c = add("template.built", None, Some("main"), serde_json::json!({}));
1537        let d = add(
1538            "clone.created",
1539            Some("c2"),
1540            Some("other"),
1541            serde_json::json!({ "id": "c2" }),
1542        );
1543        let list = |f: EventFilter| -> Vec<i64> {
1544            store
1545                .list_events(&f)
1546                .unwrap()
1547                .iter()
1548                .map(|e| e.id)
1549                .collect()
1550        };
1551        assert_eq!(list(EventFilter::default()), vec![a.id, b.id, c.id, d.id]);
1552        // A clone filter takes the name or the id.
1553        let by = |c: &str| EventFilter {
1554            clone: Some(c.into()),
1555            ..Default::default()
1556        };
1557        assert_eq!(list(by("x")), vec![a.id, b.id]);
1558        assert_eq!(list(by("c1")), vec![a.id, b.id]);
1559        let tpl = EventFilter {
1560            template: Some("main".into()),
1561            ..Default::default()
1562        };
1563        assert_eq!(list(tpl), vec![a.id, b.id, c.id]);
1564        // A kind is exact or a group, never a loose prefix.
1565        let kind = |k: &str| EventFilter {
1566            kind: Some(k.into()),
1567            ..Default::default()
1568        };
1569        assert_eq!(list(kind("clone.created")), vec![a.id, d.id]);
1570        assert_eq!(list(kind("clone")), vec![a.id, b.id, d.id]);
1571        assert_eq!(list(kind("clon")), Vec::<i64>::new());
1572        // The limit keeps the newest matches, still oldest first.
1573        let last2 = EventFilter {
1574            limit: Some(2),
1575            ..Default::default()
1576        };
1577        assert_eq!(list(last2), vec![c.id, d.id]);
1578        let future = EventFilter {
1579            since: Some(i64::MAX),
1580            ..Default::default()
1581        };
1582        assert!(list(future).is_empty());
1583    }
1584}