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 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 pub current_gen: u32,
55 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 pub mask_path: Option<PathBuf>,
68 pub mask_salt: Option<String>,
71 pub base: Option<String>,
73 pub branch: Option<String>,
75 pub base_gen: Option<u32>,
77 pub base_branch: Option<String>,
80 pub repo_rel_cwd: Option<PathBuf>,
82 pub repo_rel_migrations: Option<PathBuf>,
84 pub last_used: Option<i64>,
86}
87
88impl TemplateRow {
89 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 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#[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
182pub 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
263const 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
278const 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
289const 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
308const 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
335fn 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
434macro_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
461macro_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
494fn 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
509fn 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 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 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 conn.query_row("PRAGMA journal_mode=WAL", [], |_| Ok(()))?;
560 migrate(&mut conn)?;
561 Ok(Store { conn })
562 }
563
564 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 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 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 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 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 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 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 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 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 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 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 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 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 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 pub fn list_events(&self, f: &EventFilter) -> Result<Vec<EventRow>> {
945 self.query_events(0, f)
946 }
947
948 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}