//! `db reset` orchestrator — Phase 7 v3 §8 / T8.
//!
//! `db reset` is the destructive triple-gated path: it drops the
//! application database, recreates it, and replays every committed
//! migration found under `migrations/<database>/<app>/`. The triple
//! gate per the v3 §8 brief:
//!
//! 1. `DATABASE_URL` MUST resolve to localhost (reused
//! [`super::policy::is_localhost_connection`]).
//! 2. `Djogi.toml::profile` MUST NOT equal `"production"`.
//! 3. The caller MUST supply explicit confirmation (a `--yes` flag in
//! the CLI; programmatic callers pass [`ResetRequest::confirmed`]
//! `= true`).
//!
//! All three gates are enforced before any I/O. A refusal returns a
//! typed [`ResetError::Refused`] so the operator-facing message is
//! actionable.
//!
//! # Logging-DB isolation
//!
//! Per CLAUDE.md, the CRUD-log and event-log databases survive every
//! `db reset` invocation. Today the runner is single-context (Phase 4)
//! so this module only operates on the application DB; the seam is
//! documented at [`reset_app_database`] for the day the three-database
//! `DjogiContext::pool_for(database)` API lands.
//!
//! # `DROP DATABASE` connection plumbing
//!
//! Postgres refuses to drop the database the current session is
//! connected to. We follow the standard libpq idiom: connect to the
//! `postgres` maintenance database with the same credentials, issue
//! `DROP DATABASE … WITH (FORCE)` (Postgres 13+; we target 18+ per
//! `docs/spec/decisions.md`), then `CREATE DATABASE …`. The forced
//! variant terminates other sessions to avoid the classic "another
//! session is connected" bounce.
//!
//! After recreation the runner re-points at the fresh database via
//! [`crate::pg::pool::DjogiPool::connect`] and replays each migration
//! file pair in HISTORICAL apply order (Codex umbrella U-4 per
//! `docs/spec/configuration.md`). T7's out-of-order policy allows a
//! hotfix migration to apply AFTER a later one, so lexical version
//! sort is NOT a faithful replay of what the live DB experienced.
//! `db reset` pre-flight reads `djogi_schema_migrations.applied_at`
//! BEFORE the drop, then uses that order during replay; versions
//! absent from the historical order (typically disk files added after
//! the last apply) sort lexically afterwards. Fresh DBs with no
//! ledger fall back to lexical sort safely.
//!
//! # Historical-order capture error policy (Codex umbrella round-2 U-6)
//!
//! The pre-flight capture step has TWO qualitatively different
//! failure modes that pre-U-6 collapsed to the same outcome:
//!
//! - **Ledger genuinely missing** (the `pg_class` probe returns
//! `false`): legitimate fresh-DB fallback. Reset proceeds with an
//! empty historical map, and `build_replay_plan` falls back to
//! lexical version sort.
//! - **Anything else** — connection failure, query failure, decode
//! failure, permission denied: opaque. Reset propagates as
//! [`ResetError::HistoricalOrderCaptureFailed`] and refuses to
//! drop / recreate.
//!
//! Pre-U-6 every failure mode swallowed itself via `unwrap_or_default()`
//! at the call site, so a flaky ledger read on a populated DB still
//! triggered the destructive operation. The fix is the
//! [`HistoricalCaptureError`] split: `LedgerMissing` is the only
//! legitimate fall-back signal; `Transient(DjogiError)` propagates.
//!
//! # No regex
//!
//! URL parsing reuses the byte-level extractor in
//! [`super::policy::extract_host`] for the localhost gate, plus a
//! minimal forward-scan helper to split out the `<host>/<dbname>` parts.
//! No regex engine, no regex notation.
// `ResetError` carries an embedded `RunnerError`, which itself embeds
// boxed and string-rich variants; the resulting `Result` payload
// exceeds clippy's default 128-byte threshold for `result_large_err`.
// Boxing the whole error type would force every caller to indirect
// through a heap allocation just to discriminate among the variants —
// the signal-vs-cost tradeoff favours allowing the lint at file
// scope here, mirroring the same `#[allow(clippy::result_large_err)]`
// pattern that `crate::config` and `crate::migrate::projection` use.
#![allow(clippy::result_large_err)]
use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};
use tokio_postgres::NoTls;
use crate::config::MigrateConfig;
use crate::context::DjogiContext;
use crate::error::{DbError, DjogiError};
use crate::pg::pool::DjogiPool;
use super::compose::{
DATE_ARRAY_HELPER_PRELUDE, NUMERIC_ARRAY_HELPER_PRELUDE, TSTZ_ARRAY_HELPER_PRELUDE,
date_array_helper_operation, numeric_array_helper_operation, tstz_array_helper_operation,
};
use super::ledger::compute_checksum;
use super::naming::{down_filename, up_filename};
use super::policy::{OutOfOrderPolicy, is_localhost_connection};
use super::projection::BucketKey;
use super::replay_plan::{
ReplayPlanLoadStatus, find_non_transactional_statement_shape, load_committed_replay_plan,
};
use super::runner::{RunnerCtx, apply_plan};
use super::segment::{MigrationPlan, Segment, SegmentKind};
use super::sql::OperationSql;
use super::target::{app_dirname, bucket_dir, migrations_root};
// ── Public types ──────────────────────────────────────────────────────────
/// Configuration handed to [`reset_app_database`].
///
/// Constructed by the CLI glue from the resolved workspace, the
/// loaded [`crate::config::DjogiConfig`], and the operator's `--yes`
/// flag.
pub struct ResetRequest<'a> {
/// Workspace root — `migrations/<database>/<app>/` lives below it.
pub workspace_root: &'a Path,
/// The operator's full `DATABASE_URL` — used for the localhost
/// gate and for deriving the maintenance + reconnection URLs.
pub database_url: &'a str,
/// `Djogi.toml::profile` — production refuses unconditionally.
pub profile: &'a str,
/// `true` when the operator passed `--yes` (or the programmatic
/// caller has otherwise confirmed). The default is `false`; the
/// runner refuses without it.
pub confirmed: bool,
/// `true` when the operator explicitly accepts replaying
/// drifted on-disk SQL after the parity preflight reports that
/// the current files no longer match the live ledger. Default:
/// `false` — drift refuses before `DROP DATABASE`.
pub allow_checksum_drift_reset: bool,
/// Maintenance database name. Defaults to `"postgres"` when
/// the caller has nothing more specific (the conventional
/// administrative DB present on every cluster).
pub maintenance_database: &'a str,
/// Migration-engine config the runner consults during the replay
/// phase. Operators rarely override this; the CLI default is the
/// loaded `Djogi.toml::migrate` block.
pub migrate_config: MigrateConfig,
/// Optional pool pointing at the **audit DB** (`crud_log_url` in
/// `Djogi.toml`, by default `crud_log` derived from
/// `database.url`). When `Some`, every replayed migration writes
/// one `djogi_ddl_audit` row per executed (non-metadata) segment
/// so the audit trail captures the post-reset apply just as a
/// regular `apply` would. When `None` the audit write is silently
/// skipped — appropriate for adopters who have not yet provisioned
/// the second DB OR for tests that only care about the app-side
/// replay.
///
/// **Why a raw `deadpool_postgres::Pool`:** mirrors
/// [`super::runner::RunnerCtx::audit_pool`] so the replay
/// orchestrator can pass the pool through without re-wrapping.
/// See the doc on `RunnerCtx::audit_pool` for the rationale (the
/// audit pool is internal substrate; `DjogiPool`'s wider invariants
/// such as post-connect callbacks and status reporting are not
/// needed for the audit-side context the runner builds).
///
/// **Construction.** Production callers build this via
/// [`super::resolve_audit_url`] + [`super::build_audit_pool`]. The
/// CLI's `db reset` glue degrades to `None` (with a warn log) if
/// audit URL resolution or pool construction fails — losing the
/// audit row is preferable to refusing the destructive operation
/// over a sibling-DB outage. Tests typically pass `None` unless
/// they explicitly want to assert the per-segment audit-row
/// behaviour.
pub audit_pool: Option<deadpool_postgres::Pool>,
}
/// Successful-reset report. Names every replayed migration so the
/// operator can confirm the post-reset state matches expectation.
#[derive(Debug, Clone)]
pub struct ResetReport {
/// The application database name that was dropped + recreated.
pub database: String,
/// One entry per replayed migration version in apply order.
pub replayed_versions: Vec<ReplayedMigration>,
}
/// Per-replay record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplayedMigration {
/// Bucket — `(database, app)` identity.
pub bucket: BucketKey,
/// Version id — `V<ts>__<slug>`.
pub version: String,
}
/// Which file side a checksum-parity issue concerns.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ResetSqlSide {
Up,
Down,
}
impl ResetSqlSide {
const fn as_str(self) -> &'static str {
match self {
ResetSqlSide::Up => "up",
ResetSqlSide::Down => "down",
}
}
}
/// Why checksum parity could not be satisfied for one historical row.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ResetChecksumParityProblem {
Drift,
MissingFile,
UnsupportedBaseline,
}
/// One checksum-parity issue found before the destructive reset step.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResetChecksumParityIssue {
/// Bucket whose current on-disk SQL no longer matches the live ledger.
pub bucket: BucketKey,
/// Migration version carrying the mismatch.
pub version: String,
/// Whether the issue concerns `up.sdjql` or `down.sdjql`.
pub sql_side: ResetSqlSide,
/// Checksum recorded on the live ledger before reset.
pub ledger_checksum: String,
/// Checksum computed from the current on-disk file, or `None`
/// when the expected file is missing.
pub on_disk_checksum: Option<String>,
/// Why the parity preflight refused this historical row.
pub problem: ResetChecksumParityProblem,
}
/// Why reset cannot prove faithful replay semantics from the committed
/// migration artifacts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ResetReplaySemanticsProblem {
MissingReplayPlan,
InvalidReplayPlan,
}
/// One replay-semantics issue found before the destructive reset step.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResetReplaySemanticsIssue {
/// Bucket carrying the affected migration.
pub bucket: BucketKey,
/// Migration version that cannot be replayed safely from disk.
pub version: String,
/// Statement class that requires non-transactional replay semantics.
pub statement_shape: String,
/// Why reset could not recover a committed replay plan.
pub problem: ResetReplaySemanticsProblem,
}
/// Errors surfaced by [`reset_app_database`].
#[derive(Debug)]
pub enum ResetError {
/// One of the three triple-gates rejected the request. The
/// embedded variant names which gate refused so the operator
/// message is precise.
Refused(ResetRefusal),
/// Connecting to the maintenance database failed (admin URL
/// unreachable, credentials wrong, ssl handshake failed, …).
MaintenanceConnectFailed { source: DjogiError },
/// `DROP DATABASE` or `CREATE DATABASE` returned an error from the
/// server — typically permission denied (the connecting role lacks
/// CREATEDB) or another session is still connected.
MaintenanceSqlFailed { sql: String, source: DjogiError },
/// Connecting to the freshly-created database failed.
AppConnectFailed { source: DjogiError },
/// Walking `migrations/<database>/` failed (I/O error reading the
/// committed migration tree).
MigrationScanFailed {
path: PathBuf,
source: std::io::Error,
},
/// Reading one of the on-disk SQL files failed.
SqlReadFailed {
path: PathBuf,
source: std::io::Error,
},
/// Replay of one migration via [`apply_plan`] failed. Carries the
/// version id and the underlying runner error so the operator can
/// see which migration broke the replay.
ReplayFailed {
version: String,
source: super::runner::RunnerError,
},
/// The supplied `DATABASE_URL` is missing the database-name
/// component that `db reset` would otherwise drop. Surfaces a
/// typed error rather than a panic deep inside `replace_db_in_url`.
DatabaseUrlMalformed { database_url: String },
/// The decoded database name is not a valid Postgres identifier.
/// Defence-in-depth against URL-injection — we percent-decode the
/// path component first, then check the resulting bytes match the
/// strict grammar (ASCII letter or underscore, followed by ASCII
/// alphanumerics / underscores, up to 63 bytes). Anything else
/// refuses BEFORE we splice the value into `DROP DATABASE` /
/// `CREATE DATABASE` DDL. Surfaced separately from
/// `DatabaseUrlMalformed` so the operator can tell "no database
/// component" from "the component decodes to something we won't
/// quote into DDL".
InvalidDatabaseName { name: String },
/// Workspace lock acquisition failed before the replay could run.
WorkspaceLockFailed { source: super::guard::GuardError },
/// Codex umbrella round-2 U-6: capturing the live ledger's
/// historical apply order failed for a reason that is NOT
/// "ledger table is missing on a fresh DB". Pre-fix every
/// failure mode of the capture step (connection error, decode
/// error, generic SQL error, …) collapsed to an empty map via
/// `unwrap_or_default()`, which silently fell through to the
/// drop / recreate path on a transient error. That re-opens the
/// U-4 hazard: a flaky ledger read that swallows itself, then
/// the destructive operation runs anyway against a database
/// whose true state we never confirmed.
///
/// Post-fix: the ONLY legitimate fall-back-to-lexical signal is
/// the `pg_class` probe returning `false` (genuinely fresh DB
/// or freshly-recreated DB without bootstrap yet). Every other
/// failure mode propagates through this variant, refusing the
/// destructive operation. The operator-facing message names the
/// underlying `DjogiError` so the failure point is unambiguous.
HistoricalOrderCaptureFailed { source: DjogiError },
}
/// Specific refusal kind for [`ResetError::Refused`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ResetRefusal {
/// `DATABASE_URL` does not resolve to localhost.
NotLocalhost { database_url: String },
/// `Djogi.toml::profile = "production"`.
ProductionProfile { profile: String },
/// The caller did not supply explicit confirmation.
NotConfirmed,
/// The live ledger's checksums do not match the current on-disk
/// migration files. Reset refuses before `DROP DATABASE` unless
/// the operator explicitly overrides the drift gate.
ChecksumParity {
issues: Vec<ResetChecksumParityIssue>,
},
/// Reset cannot prove faithful replay semantics for at least one
/// committed migration before the destructive drop/recreate.
ReplaySemantics {
issues: Vec<ResetReplaySemanticsIssue>,
},
}
impl std::fmt::Display for ResetError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ResetError::Refused(r) => write!(f, "db reset refused: {r}"),
ResetError::MaintenanceConnectFailed { source } => write!(
f,
"db reset: connect to maintenance database failed: {source}"
),
ResetError::MaintenanceSqlFailed { sql, source } => {
write!(f, "db reset: maintenance SQL `{sql}` failed: {source}")
}
ResetError::AppConnectFailed { source } => {
write!(
f,
"db reset: connect to fresh app database failed: {source}"
)
}
ResetError::MigrationScanFailed { path, source } => write!(
f,
"db reset: scanning {} for migration files failed: {source}",
path.display(),
),
ResetError::SqlReadFailed { path, source } => write!(
f,
"db reset: reading migration SQL at {} failed: {source}",
path.display(),
),
ResetError::ReplayFailed { version, source } => write!(
f,
"db reset: replay of `{version}` failed: {source}; the database \
has been recreated but is now in a partial state — fix the \
underlying issue and re-run db reset",
),
ResetError::DatabaseUrlMalformed { database_url } => write!(
f,
"db reset: DATABASE_URL `{database_url}` does not contain a \
database-name component (no `/` after the host); db reset \
cannot derive the database name to drop"
),
ResetError::InvalidDatabaseName { name } => write!(
f,
"db reset: decoded database name `{name}` is not a valid \
Postgres identifier (expected: ASCII letter or underscore, \
followed by ASCII alphanumerics or underscores, up to 63 \
bytes); db reset refuses to splice arbitrary bytes into \
DROP DATABASE / CREATE DATABASE DDL"
),
ResetError::WorkspaceLockFailed { source } => {
write!(f, "db reset: workspace lock acquisition failed: {source}")
}
ResetError::HistoricalOrderCaptureFailed { source } => write!(
f,
"db reset: capturing the live ledger's historical apply order \
failed: {source}; refusing to proceed with the destructive \
drop / recreate because we cannot confirm the live state — \
fix the underlying connection / query failure and re-run, or \
(if the database genuinely does not exist yet) create it \
first then re-run db reset"
),
}
}
}
impl std::fmt::Display for ResetRefusal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ResetRefusal::NotLocalhost { database_url } => write!(
f,
"DATABASE_URL is not localhost (got `{database_url}`); db reset is \
a destructive operation and must not be invoked against a remote \
database"
),
ResetRefusal::ProductionProfile { profile } => write!(
f,
"Djogi.toml::profile = `{profile}`; db reset refuses to run on a \
production profile"
),
ResetRefusal::NotConfirmed => f.write_str(
"db reset requires explicit confirmation — pass `--yes` (or set \
ResetRequest::confirmed = true) to acknowledge that the entire \
application database will be dropped",
),
ResetRefusal::ChecksumParity { issues } => {
let rendered = issues
.iter()
.map(render_checksum_parity_issue)
.collect::<Vec<_>>()
.join("; ");
write!(
f,
"db reset checksum parity preflight found drift against the live ledger: \
{rendered}; refusing destructive drop / recreate unless you pass \
`--allow-checksum-drift-reset` (or set \
`ResetRequest::allow_checksum_drift_reset = true`)"
)
}
ResetRefusal::ReplaySemantics { issues } => {
let rendered = issues
.iter()
.map(render_replay_semantics_issue)
.collect::<Vec<_>>()
.join("; ");
write!(
f,
"db reset cannot prove faithful replay semantics for at least one committed \
migration: {rendered}; refusing destructive drop / recreate until the migration \
has a committed replay manifest or is replay-safe as a single transactional plan"
)
}
}
}
}
impl std::error::Error for ResetError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
ResetError::MaintenanceConnectFailed { source } => Some(source),
ResetError::MaintenanceSqlFailed { source, .. } => Some(source),
ResetError::AppConnectFailed { source } => Some(source),
ResetError::MigrationScanFailed { source, .. } => Some(source),
ResetError::SqlReadFailed { source, .. } => Some(source),
ResetError::ReplayFailed { source, .. } => Some(source),
ResetError::WorkspaceLockFailed { source } => Some(source),
ResetError::HistoricalOrderCaptureFailed { source } => Some(source),
_ => None,
}
}
}
// ── Public entry point ────────────────────────────────────────────────────
/// Drop, recreate, and replay every committed migration against the
/// application database in `req.database_url`.
///
/// Triple-gated per the module docs. Returns a [`ResetReport`] on
/// success or a [`ResetError`] on any failure mode — including a
/// gate refusal, which is surfaced as `ResetError::Refused` rather
/// than as a successful no-op.
pub async fn reset_app_database(req: ResetRequest<'_>) -> Result<ResetReport, ResetError> {
// 1. Triple gate — every gate runs BEFORE any I/O so a refusal
// leaves zero side effects on the workspace OR the database.
if !is_localhost_connection(req.database_url) {
return Err(ResetError::Refused(ResetRefusal::NotLocalhost {
database_url: req.database_url.to_string(),
}));
}
if req.profile == "production" {
return Err(ResetError::Refused(ResetRefusal::ProductionProfile {
profile: req.profile.to_string(),
}));
}
if !req.confirmed {
return Err(ResetError::Refused(ResetRefusal::NotConfirmed));
}
// 2. Derive the app database name and the maintenance URL.
//
// Two-step parse: first extract+percent-decode the path
// component, then validate the decoded bytes against the strict
// Postgres-identifier grammar. We refuse weird-looking names
// BEFORE splicing them into `DROP DATABASE` / `CREATE DATABASE`
// DDL — defence-in-depth against URL-injection where a crafted
// URL like `postgres://localhost/'; DROP TABLE foo; --` could
// otherwise reach the DDL builder. The maintenance database
// name flows from operator config (`--maintenance-database`,
// default `postgres`) so it's validated separately.
let database =
extract_database_from_url(req.database_url).ok_or(ResetError::DatabaseUrlMalformed {
database_url: req.database_url.to_string(),
})?;
if !is_valid_pg_identifier(&database) {
return Err(ResetError::InvalidDatabaseName { name: database });
}
if !is_valid_pg_identifier(req.maintenance_database) {
return Err(ResetError::InvalidDatabaseName {
name: req.maintenance_database.to_string(),
});
}
let maintenance_url = replace_db_in_url(req.database_url, req.maintenance_database).ok_or(
ResetError::DatabaseUrlMalformed {
database_url: req.database_url.to_string(),
},
)?;
// 3. Acquire workspace lock — replay mutates ledger state on the
// fresh DB; concurrent compose / apply / repair operations
// against the same workspace must not interleave with reset.
let lock_path = req.workspace_root.join(super::guard::LOCK_FILE_NAME);
let _guard = super::guard::acquire(&lock_path, super::guard::DEFAULT_TIMEOUT)
.map_err(|e| ResetError::WorkspaceLockFailed { source: e })?;
// 4. Codex umbrella U-4: capture the HISTORICAL apply order from
// the live ledger BEFORE the drop. T7's out-of-order policy
// allows a hotfix migration to apply AFTER a later one, e.g.
// `applied_at` of `0001 < 0003 < 0002`. Lexical version-string
// sort would replay them as `0001, 0002, 0003`, which is NOT
// the sequence the live database actually experienced. If
// `0002` only succeeded historically because `0003` was
// already in place, lexical replay would re-apply it
// out-of-order on a fresh DB — different state from what we
// just dropped.
//
// Strategy: pre-flight a read-only connection to the live DB,
// query `djogi_schema_migrations` ordered by `applied_at`, and
// capture `(bucket, version) -> rank`. We then use that rank
// as the replay sort key. Versions absent from the historical
// order (e.g. files added on disk after the last apply) sort
// AFTER any historical entry, lexically among themselves.
//
// Codex umbrella round-2 U-6 — error-policy split:
// `HistoricalCaptureError::LedgerMissing` is the ONLY legitimate
// fall-back-to-lexical signal (`pg_class` probe returned false:
// genuinely fresh DB). Every OTHER failure mode (connection
// failure, decode failure, generic SQL error, permission
// denied) surfaces as `Transient(..)` and propagates through
// `ResetError::HistoricalOrderCaptureFailed`. Pre-U-6 every
// error collapsed to `()` and the reset proceeded with an
// empty map — which re-opened the U-4 hazard for transient
// failures (the empty map masquerades as "fresh DB with no
// history" and the destructive drop / recreate runs anyway).
let historical_entries = match capture_historical_replay_entries(req.database_url).await {
Ok(entries) => entries,
Err(HistoricalCaptureError::LedgerMissing) => Vec::new(),
Err(HistoricalCaptureError::Transient(e)) => {
return Err(ResetError::HistoricalOrderCaptureFailed { source: e });
}
};
preflight_reset_checksum_parity(
req.workspace_root,
&database,
&historical_entries,
req.allow_checksum_drift_reset,
)?;
preflight_reset_replay_semantics(req.workspace_root, &database)?;
let historical_order = build_historical_order(&historical_entries);
// 5. Drop + recreate the application database via the maintenance
// connection. A fresh tokio_postgres client is opened just for
// the two DDLs — the maintenance pool is intentionally NOT
// cached because db reset is interactive / one-shot.
drop_and_create_database(&maintenance_url, &database).await?;
// 6. Connect to the freshly-created application DB and replay
// every committed migration.
let pool = DjogiPool::connect(req.database_url)
.await
.map_err(|e| ResetError::AppConnectFailed { source: e })?;
let mut ctx = DjogiContext::from_pool(pool);
let buckets = scan_committed_migrations(req.workspace_root, &database)?;
// Codex umbrella U-4: replay order = historical apply order
// (`applied_at` ascending) for versions that have a historical
// entry; lexical-after-historical for versions that do not.
let replay_plan = build_replay_plan(&buckets, &historical_order);
let mut replayed: Vec<ReplayedMigration> = Vec::new();
for (bucket, version) in replay_plan {
replay_one_migration(
&mut ctx,
req.workspace_root,
&bucket,
&version,
&req.migrate_config,
&_guard,
req.audit_pool.as_ref(),
)
.await?;
replayed.push(ReplayedMigration {
bucket: bucket.clone(),
version,
});
}
Ok(ResetReport {
database,
replayed_versions: replayed,
})
}
/// Internal error classifier for [`capture_historical_apply_order`]
/// per Codex umbrella round-2 U-6.
///
/// The capture step has two qualitatively different failure modes:
///
/// - **`LedgerMissing`** — the `pg_class` probe came back `false`.
/// The connection succeeded, the catalog query succeeded, and the
/// answer was "no `djogi_schema_migrations` table here". This is
/// the legitimate fresh-DB / freshly-recreated-DB signal. The
/// caller falls back to lexical sort and the destructive drop /
/// recreate proceeds.
/// - **`Transient(DjogiError)`** — anything else: tokio_postgres
/// connect failure (DB unreachable, auth fail, network drop, DB
/// does not exist), `current_database()` query failure, probe
/// query failure, decode error, generic SELECT failure. None of
/// these prove the DB is fresh; they prove we cannot CONFIRM the
/// live state. The caller propagates as
/// `ResetError::HistoricalOrderCaptureFailed` and refuses to
/// drop / recreate.
///
/// Pre-U-6 the helper returned `Result<_, ()>` and `unwrap_or_default()`
/// at the call site collapsed every failure mode to "empty map →
/// proceed with lexical fallback". That re-opened the U-4 hazard
/// under a transient connection / query failure: the destructive
/// path runs against a database whose history we never read.
#[derive(Debug)]
enum HistoricalCaptureError {
/// `pg_class` probe returned `false` — ledger genuinely absent.
LedgerMissing,
/// Connection / query / decode failure — treat as opaque, do
/// NOT proceed with the destructive operation.
Transient(DjogiError),
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct HistoricalReplayEntry {
bucket: BucketKey,
version: String,
status: String,
checksum_up: String,
checksum_down: Option<String>,
}
/// Codex umbrella U-4 + round-2 U-6 — capture the historical apply
/// order from the live ledger before the drop.
///
/// Connects to the application DB at `database_url`, probes for the
/// presence of `djogi_schema_migrations`, and (when present) queries
/// it ordered by `applied_at ASC, id ASC`. Returns a
/// `(bucket, version) -> rank` map where lower ranks applied first
/// historically.
///
/// Per U-6, the error classification is intentional and load-bearing:
///
/// - Probe says ledger absent → `Err(HistoricalCaptureError::LedgerMissing)`
/// (caller falls back to lexical).
/// - Anything else → `Err(HistoricalCaptureError::Transient(..))`
/// (caller propagates and refuses the destructive drop).
///
/// Only `Applied` / `Faked` / `Baseline` rows participate — `Pending`,
/// `Failed`, `RolledBack` do not represent migrations whose effect
/// the live DB carries forward.
#[cfg(test)]
#[allow(clippy::disallowed_methods)]
async fn capture_historical_apply_order(
database_url: &str,
) -> Result<BTreeMap<(BucketKey, String), u64>, HistoricalCaptureError> {
let entries = capture_historical_replay_entries(database_url).await?;
Ok(build_historical_order(&entries))
}
fn build_historical_order(entries: &[HistoricalReplayEntry]) -> BTreeMap<(BucketKey, String), u64> {
entries
.iter()
.enumerate()
.map(|(rank, entry)| ((entry.bucket.clone(), entry.version.clone()), rank as u64))
.collect()
}
#[allow(clippy::disallowed_methods)]
async fn capture_historical_replay_entries(
database_url: &str,
) -> Result<Vec<HistoricalReplayEntry>, HistoricalCaptureError> {
let (client, conn) = tokio_postgres::connect(database_url, NoTls)
.await
.map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"tokio-postgres connect failed during historical-order capture: {e}"
))))
})?;
let driver = tokio::spawn(async move {
if let Err(e) = conn.await {
tracing::debug!("[db reset] historical-order driver: {e}");
}
});
// Resolve the active database name so the captured map's bucket
// identity matches what `scan_committed_migrations` produces.
let db_row = client
.query_one("SELECT current_database()::text", &[])
.await
.map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"current_database() failed during historical-order capture: {e}"
))))
})?;
let database: String = db_row.try_get(0).map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding current_database() result failed: {e}"
))))
})?;
// Probe the ledger table. THIS is the canonical fresh-DB decision
// point: the connection succeeded AND the catalog query returned
// a typed answer. A `false` here means the ledger has not been
// bootstrapped yet — legitimate fresh-DB fallback. A failure of
// the probe itself is opaque and propagates as Transient.
let probe = client
.query_one(
"SELECT EXISTS(SELECT 1 FROM pg_class \
WHERE relname = 'djogi_schema_migrations' AND relkind = 'r')",
&[],
)
.await
.map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"pg_class probe for djogi_schema_migrations failed: {e}"
))))
})?;
let exists: bool = probe.try_get(0).map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding pg_class probe result failed: {e}"
))))
})?;
if !exists {
drop(client);
let _ = driver.await;
return Err(HistoricalCaptureError::LedgerMissing);
}
let rows = client
.query(
"SELECT version, app_label, status, checksum_up, checksum_down \
FROM djogi_schema_migrations \
WHERE status IN ('applied', 'faked', 'baseline') \
ORDER BY applied_at ASC, id ASC",
&[],
)
.await
.map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"SELECT djogi_schema_migrations failed during historical-order capture: {e}"
))))
})?;
let mut out: Vec<HistoricalReplayEntry> = Vec::with_capacity(rows.len());
for (rank, row) in rows.iter().enumerate() {
let version: String = row.try_get("version").map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding ledger version column failed at rank {rank}: {e}"
))))
})?;
let app_label: String = row.try_get("app_label").map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding ledger app_label column failed at rank {rank}: {e}"
))))
})?;
let status: String = row.try_get("status").map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding ledger status column failed at rank {rank}: {e}"
))))
})?;
let checksum_up: String = row.try_get("checksum_up").map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding ledger checksum_up column failed at rank {rank}: {e}"
))))
})?;
let checksum_down: Option<String> = row.try_get("checksum_down").map_err(|e| {
HistoricalCaptureError::Transient(DjogiError::Db(DbError::other(format!(
"decoding ledger checksum_down column failed at rank {rank}: {e}"
))))
})?;
out.push(HistoricalReplayEntry {
bucket: BucketKey {
database: database.clone(),
app: app_label,
},
version,
status,
checksum_up,
checksum_down,
});
}
drop(client);
let _ = driver.await;
Ok(out)
}
fn preflight_reset_checksum_parity(
workspace_root: &Path,
database: &str,
historical_entries: &[HistoricalReplayEntry],
allow_checksum_drift_reset: bool,
) -> Result<(), ResetError> {
let issues = collect_checksum_parity_issues(workspace_root, database, historical_entries)?;
if issues.is_empty() || allow_checksum_drift_reset {
return Ok(());
}
Err(ResetError::Refused(ResetRefusal::ChecksumParity { issues }))
}
fn collect_checksum_parity_issues(
workspace_root: &Path,
database: &str,
historical_entries: &[HistoricalReplayEntry],
) -> Result<Vec<ResetChecksumParityIssue>, ResetError> {
let on_disk = super::target::scan_filesystem_with_files(workspace_root, Some(database))
.map_err(|err| ResetError::MigrationScanFailed {
path: migrations_root(workspace_root).join(database),
source: err,
})?;
let mut issues = Vec::new();
for entry in historical_entries {
if entry.status == "baseline" {
issues.push(ResetChecksumParityIssue {
bucket: entry.bucket.clone(),
version: entry.version.clone(),
sql_side: ResetSqlSide::Up,
ledger_checksum: entry.checksum_up.clone(),
on_disk_checksum: None,
problem: ResetChecksumParityProblem::UnsupportedBaseline,
});
continue;
}
let up_path = on_disk
.get(&entry.bucket)
.and_then(|versions| versions.get(&entry.version))
.cloned()
.unwrap_or_else(|| {
bucket_dir(workspace_root, &entry.bucket).join(up_filename(&entry.version))
});
push_checksum_issue_if_needed(
&mut issues,
entry,
ResetSqlSide::Up,
&entry.checksum_up,
&up_path,
)?;
if let Some(ledger_checksum_down) = entry.checksum_down.as_deref() {
let down_path =
bucket_dir(workspace_root, &entry.bucket).join(down_filename(&entry.version));
push_checksum_issue_if_needed(
&mut issues,
entry,
ResetSqlSide::Down,
ledger_checksum_down,
&down_path,
)?;
}
}
Ok(issues)
}
fn push_checksum_issue_if_needed(
issues: &mut Vec<ResetChecksumParityIssue>,
entry: &HistoricalReplayEntry,
sql_side: ResetSqlSide,
ledger_checksum: &str,
path: &Path,
) -> Result<(), ResetError> {
let on_disk_sql = match fs::read_to_string(path) {
Ok(sql) => sql,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
issues.push(ResetChecksumParityIssue {
bucket: entry.bucket.clone(),
version: entry.version.clone(),
sql_side,
ledger_checksum: ledger_checksum.to_string(),
on_disk_checksum: None,
problem: ResetChecksumParityProblem::MissingFile,
});
return Ok(());
}
Err(e) => {
return Err(ResetError::SqlReadFailed {
path: path.to_path_buf(),
source: e,
});
}
};
let on_disk_checksum = compute_committed_sql_checksum(&on_disk_sql, sql_side);
if on_disk_checksum != ledger_checksum {
issues.push(ResetChecksumParityIssue {
bucket: entry.bucket.clone(),
version: entry.version.clone(),
sql_side,
ledger_checksum: ledger_checksum.to_string(),
on_disk_checksum: Some(on_disk_checksum),
problem: ResetChecksumParityProblem::Drift,
});
}
Ok(())
}
fn render_checksum_parity_issue(issue: &ResetChecksumParityIssue) -> String {
let version_path = format!(
"{}/{}/{}",
issue.bucket.database,
app_dirname(&issue.bucket.app),
issue.version
);
match issue.problem {
ResetChecksumParityProblem::Drift => format!(
"{version_path} {} checksum drift: ledger `{}` vs on-disk `{}`",
issue.sql_side.as_str(),
issue.ledger_checksum,
issue.on_disk_checksum.as_deref().unwrap_or("<missing>")
),
ResetChecksumParityProblem::MissingFile => format!(
"{version_path} {} file missing: ledger checksum `{}` has no on-disk peer",
issue.sql_side.as_str(),
issue.ledger_checksum
),
ResetChecksumParityProblem::UnsupportedBaseline => format!(
"{version_path} baseline checksum `{}` cannot be compared to migration file bytes; \
db reset cannot establish safe parity for a baseline row",
issue.ledger_checksum
),
}
}
fn preflight_reset_replay_semantics(
workspace_root: &Path,
database: &str,
) -> Result<(), ResetError> {
let buckets = scan_committed_migrations(workspace_root, database)?;
let mut issues = Vec::new();
for (bucket, versions) in &buckets {
for version in versions {
let replay_sql = read_replay_sql_files(workspace_root, bucket, version)?;
let plan_status =
replay_plan_status_for_reset(workspace_root, bucket, version, &replay_sql);
if let Some(issue) =
replay_semantics_issue_for_plan_status(bucket, version, &replay_sql, &plan_status)
{
issues.push(issue);
}
}
}
if issues.is_empty() {
Ok(())
} else {
Err(ResetError::Refused(ResetRefusal::ReplaySemantics {
issues,
}))
}
}
fn replay_plan_status_for_reset(
workspace_root: &Path,
bucket: &BucketKey,
version: &str,
replay_sql: &ReplaySqlFiles,
) -> ReplayPlanLoadStatus {
load_committed_replay_plan(
workspace_root,
bucket,
version,
&replay_sql.checksum_up,
replay_sql.checksum_down.as_deref(),
)
}
fn load_reset_replay_plan(
workspace_root: &Path,
bucket: &BucketKey,
version: &str,
replay_sql: &ReplaySqlFiles,
) -> Result<Option<MigrationPlan>, ResetError> {
let plan_status = replay_plan_status_for_reset(workspace_root, bucket, version, replay_sql);
if let Some(issue) =
replay_semantics_issue_for_plan_status(bucket, version, replay_sql, &plan_status)
{
return Err(ResetError::Refused(ResetRefusal::ReplaySemantics {
issues: vec![issue],
}));
}
match plan_status {
ReplayPlanLoadStatus::Loaded(plan) => Ok(Some(plan)),
ReplayPlanLoadStatus::Missing | ReplayPlanLoadStatus::Invalid(_) => Ok(None),
}
}
fn replay_semantics_issue_for_plan_status(
bucket: &BucketKey,
version: &str,
replay_sql: &ReplaySqlFiles,
plan_status: &ReplayPlanLoadStatus,
) -> Option<ResetReplaySemanticsIssue> {
let statement_shape = find_non_transactional_statement_shape(&replay_sql.up_sql)?;
let problem = match plan_status {
ReplayPlanLoadStatus::Loaded(_) => return None,
ReplayPlanLoadStatus::Missing => ResetReplaySemanticsProblem::MissingReplayPlan,
ReplayPlanLoadStatus::Invalid(_) => ResetReplaySemanticsProblem::InvalidReplayPlan,
};
Some(ResetReplaySemanticsIssue {
bucket: bucket.clone(),
version: version.to_string(),
statement_shape: statement_shape.to_string(),
problem,
})
}
fn render_replay_semantics_issue(issue: &ResetReplaySemanticsIssue) -> String {
let version_path = format!(
"{}/{}/{}",
issue.bucket.database,
app_dirname(&issue.bucket.app),
issue.version
);
match issue.problem {
ResetReplaySemanticsProblem::MissingReplayPlan => format!(
"{version_path} contains `{}` but has no committed replay manifest",
issue.statement_shape
),
ResetReplaySemanticsProblem::InvalidReplayPlan => format!(
"{version_path} contains `{}` but its committed replay manifest is missing or stale",
issue.statement_shape
),
}
}
/// Codex umbrella U-4 — given the on-disk bucket map and the captured
/// historical apply order, produce the deterministic replay plan as a
/// flat `Vec<(BucketKey, String)>` in the order migrations should be
/// re-applied.
///
/// **Sort key** (lower wins): `(historical_rank.unwrap_or(u64::MAX),
/// bucket.database, bucket.app, version)`. Versions WITH a historical
/// rank apply first (in apply-order); versions WITHOUT (typically
/// disk files added after the last historical apply) apply last,
/// sorted lexically among themselves so re-running the reset
/// produces byte-identical output.
///
/// Pulled out as a free function so unit tests can pin every edge
/// case without standing up a live connection.
fn build_replay_plan(
buckets: &BTreeMap<BucketKey, Vec<String>>,
historical_order: &BTreeMap<(BucketKey, String), u64>,
) -> Vec<(BucketKey, String)> {
let mut flat: Vec<(BucketKey, String)> = Vec::new();
for (bucket, versions) in buckets {
for v in versions {
flat.push((bucket.clone(), v.clone()));
}
}
flat.sort_by(|a, b| {
let ra = historical_order
.get(&(a.0.clone(), a.1.clone()))
.copied()
.unwrap_or(u64::MAX);
let rb = historical_order
.get(&(b.0.clone(), b.1.clone()))
.copied()
.unwrap_or(u64::MAX);
ra.cmp(&rb)
.then_with(|| a.0.database.cmp(&b.0.database))
.then_with(|| a.0.app.cmp(&b.0.app))
.then_with(|| a.1.cmp(&b.1))
});
flat
}
// ── Internals ─────────────────────────────────────────────────────────────
/// Tokio-postgres-based DROP + CREATE helper. Connects to the
/// maintenance database, issues both statements via `batch_execute`
/// (the simple-query protocol — Postgres refuses to prepare DROP /
/// CREATE DATABASE), and returns once both succeed.
#[allow(clippy::disallowed_methods)]
async fn drop_and_create_database(maintenance_url: &str, database: &str) -> Result<(), ResetError> {
let (client, conn) = tokio_postgres::connect(maintenance_url, NoTls)
.await
.map_err(|e| ResetError::MaintenanceConnectFailed {
source: DjogiError::Db(DbError::other(format!(
"tokio-postgres connect to maintenance DB failed: {e}"
))),
})?;
// The connection task must run for the lifetime of the client.
let driver = tokio::spawn(async move {
if let Err(e) = conn.await {
tracing::error!("[db reset] maintenance connection error: {e}");
}
});
// Quote the database name for safety. Postgres identifier rules
// allow `"` to appear inside a quoted identifier only as an
// escaped `""` pair. We replace each `"` byte with `""` so an
// operator who somehow has a quote in the database name still
// gets a syntactically-valid identifier; in practice the database
// grammar (set by the connection URL parser upstream) precludes
// that, but the defensive escape is free.
let quoted_db = quote_identifier(database);
let drop_sql = format!("DROP DATABASE IF EXISTS {quoted_db} WITH (FORCE)");
client
.batch_execute(&drop_sql)
.await
.map_err(|e| ResetError::MaintenanceSqlFailed {
sql: drop_sql.clone(),
source: DjogiError::Db(DbError::other(format!("{e}"))),
})?;
let create_sql = format!("CREATE DATABASE {quoted_db}");
client
.batch_execute(&create_sql)
.await
.map_err(|e| ResetError::MaintenanceSqlFailed {
sql: create_sql.clone(),
source: DjogiError::Db(DbError::other(format!("{e}"))),
})?;
// Drop the client so the connection task finishes; await the task
// so the connection close lands deterministically.
drop(client);
let _ = driver.await;
Ok(())
}
/// Walk `migrations/<database>/` and collect every committed
/// `V<ts>__<slug>.sql` migration grouped by `(database, app)` bucket.
///
/// Returns a `BTreeMap` so iteration order is deterministic across
/// runs — key order is `(database, app)` ASCII-sorted; per-bucket
/// migration lists are version-sorted (lexical = chronological per
/// the [`super::naming`] convention).
///
/// Files matching the down-side suffix (`.down.sdjql`) are skipped —
/// the up-side filename serves as the canonical version identifier.
fn scan_committed_migrations(
workspace_root: &Path,
database: &str,
) -> Result<BTreeMap<BucketKey, Vec<String>>, ResetError> {
let with_paths = super::target::scan_filesystem_with_files(workspace_root, Some(database))
.map_err(|err| ResetError::MigrationScanFailed {
path: migrations_root(workspace_root).join(database),
source: err,
})?;
Ok(with_paths
.into_iter()
.map(|(bucket, vers)| (bucket, vers.into_keys().collect()))
.collect())
}
struct ReplaySqlFiles {
up_sql: String,
down_sql: String,
checksum_up: String,
checksum_down: Option<String>,
}
/// Compute the canonical checksum of a committed migration SQL file's
/// contents, in the same domain compose uses when it records the ledger
/// `checksum_up` / `checksum_down` values.
///
/// # Why this exists
///
/// Compose computes checksums over the [`super::OperationSql`] fragments
/// (`label` + `up` / `down` SQL), NOT over the rendered file that those
/// fragments are written into. A composed migration file carries a
/// `-- Djogi composed migration — {up,down}` header, a
/// `-- DO NOT EDIT …` banner, and per-statement `-- <label>` comment
/// lines that are absent from the fragment domain. A naive
/// [`compute_checksum`] over the whole file therefore yields a different
/// digest than the ledger stores. This helper strips the file framing
/// (parsing the composed file back into its canonical fragments) so a
/// recomputed checksum matches what compose persisted — load-bearing for
/// `djogi migrations repair checksum-drift`, which recomputes from disk
/// when the operator omits `--checksum-up` / `--checksum-down`.
///
/// # Behavior
///
/// When `sql` is a recognizable composed file (correct header + banner),
/// the digest is computed over its canonical fragments for `side`.
/// Otherwise — a hand-authored or legacy file with no composed framing —
/// it falls back to the whole-file digest, matching how such files are
/// checksummed elsewhere.
pub fn compute_committed_sql_checksum(sql: &str, side: ResetSqlSide) -> String {
canonical_composed_sql_fragments(sql, side)
.map(|fragments| compute_checksum(fragments.iter().map(String::as_str)))
.unwrap_or_else(|| compute_checksum([sql]))
}
/// Compute the canonical checksum of a committed *down* SQL file, or
/// `None` when the down side carries no real statements.
///
/// Shares the fragment-level domain documented on
/// [`compute_committed_sql_checksum`]. Returns `None` (matching compose's
/// `NULL` `checksum_down` sentinel) when the down file is comment-only —
/// either every composed fragment is a comment, or, for a non-composed
/// file, every non-blank line is a `--` comment. A down side with at
/// least one real statement returns `Some(digest)`.
///
/// Note this takes file *contents* already read from disk; a missing
/// down file (which also maps to a `None` / `NULL` down checksum) is the
/// caller's concern, not handled here.
pub fn compute_committed_down_sql_checksum(sql: &str) -> Option<String> {
if let Some(fragments) = canonical_composed_sql_fragments(sql, ResetSqlSide::Down) {
if fragments.iter().all(|fragment| fragment.starts_with("--")) {
None
} else {
Some(compute_checksum(fragments.iter().map(String::as_str)))
}
} else if is_comment_only_sql(sql) {
None
} else {
Some(compute_checksum([sql]))
}
}
fn is_comment_only_sql(sql: &str) -> bool {
sql.lines()
.map(str::trim_start)
.filter(|line| !line.trim().is_empty())
.all(|line| line.starts_with("--"))
}
fn canonical_composed_sql_fragments(sql: &str, side: ResetSqlSide) -> Option<Vec<String>> {
let expected_header = match side {
ResetSqlSide::Up => "-- Djogi composed migration — up\n",
ResetSqlSide::Down => "-- Djogi composed migration — down\n",
};
if !sql.starts_with(expected_header) {
return None;
}
let (_, mut body) =
sql.split_once("-- DO NOT EDIT — regenerate via `djogi migrations compose`.\n\n")?;
let mut fragments = Vec::new();
let helper_pairs: [(&str, String); 3] = match side {
ResetSqlSide::Up => [
(
NUMERIC_ARRAY_HELPER_PRELUDE,
NUMERIC_ARRAY_HELPER_PRELUDE.to_string(),
),
(
DATE_ARRAY_HELPER_PRELUDE,
DATE_ARRAY_HELPER_PRELUDE.to_string(),
),
(
TSTZ_ARRAY_HELPER_PRELUDE,
TSTZ_ARRAY_HELPER_PRELUDE.to_string(),
),
],
ResetSqlSide::Down => [
(
NUMERIC_ARRAY_HELPER_PRELUDE,
numeric_array_helper_operation().down,
),
(
DATE_ARRAY_HELPER_PRELUDE,
date_array_helper_operation().down,
),
(
TSTZ_ARRAY_HELPER_PRELUDE,
tstz_array_helper_operation().down,
),
],
};
for (rendered_prelude, checksum_fragment) in helper_pairs {
if let Some(rest) = body.strip_prefix(rendered_prelude) {
fragments.push(checksum_fragment);
body = rest.strip_prefix('\n').unwrap_or(rest);
}
}
let mut operation_fragments = parse_composed_operation_fragments(body, side)?;
if side == ResetSqlSide::Down {
operation_fragments.reverse();
}
fragments.extend(operation_fragments);
Some(fragments)
}
fn parse_composed_operation_fragments(body: &str, side: ResetSqlSide) -> Option<Vec<String>> {
let mut rest = body.trim_end_matches('\n');
let mut fragments = Vec::new();
if rest.trim().is_empty() {
return Some(fragments);
}
loop {
let after_label = rest.strip_prefix("-- ")?;
let (_, after_label) = after_label.split_once('\n')?;
let (fragment, next) = match after_label.find("\n\n-- ") {
Some(next_label) => (
&after_label[..next_label],
Some(&after_label[next_label + 2..]),
),
None => (after_label, None),
};
let fragment = if side == ResetSqlSide::Down {
fragment
.lines()
.filter(|line| !line.starts_with("-- LOSSY:"))
.collect::<Vec<_>>()
.join("\n")
} else {
fragment.to_string()
};
fragments.push(fragment);
match next {
Some(next_rest) => rest = next_rest,
None => break,
}
}
Some(fragments)
}
fn read_replay_sql_files(
workspace_root: &Path,
bucket: &BucketKey,
version: &str,
) -> Result<ReplaySqlFiles, ResetError> {
let bucket_dir = super::target::bucket_dir(workspace_root, bucket);
let up_path = bucket_dir.join(up_filename(version));
let down_path = bucket_dir.join(down_filename(version));
let up_sql = fs::read_to_string(&up_path).map_err(|e| ResetError::SqlReadFailed {
path: up_path.clone(),
source: e,
})?;
let checksum_up = compute_committed_sql_checksum(&up_sql, ResetSqlSide::Up);
let (down_sql, checksum_down) = match fs::read_to_string(&down_path) {
Ok(sql) => {
let checksum_down = compute_committed_down_sql_checksum(&sql);
(sql, checksum_down)
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => (String::new(), None),
Err(e) => {
return Err(ResetError::SqlReadFailed {
path: down_path,
source: e,
});
}
};
Ok(ReplaySqlFiles {
up_sql,
down_sql,
checksum_up,
checksum_down,
})
}
/// Read one migration's committed SQL and apply it through the runner.
///
/// Manifest-backed migrations replay the committed segment plan so
/// non-transactional statements keep their original execution shape.
/// Legacy migrations without a manifest fall back to a single
/// transactional segment. In both paths, the runner receives the same
/// canonical operation-fragment checksum domain that compose records
/// in the ledger.
async fn replay_one_migration(
ctx: &mut DjogiContext,
workspace_root: &Path,
bucket: &BucketKey,
version: &str,
migrate_config: &MigrateConfig,
guard: &super::guard::WorkspaceGuard,
audit_pool: Option<&deadpool_postgres::Pool>,
) -> Result<(), ResetError> {
let replay_sql = read_replay_sql_files(workspace_root, bucket, version)?;
let plan = load_reset_replay_plan(workspace_root, bucket, version, &replay_sql)?
.unwrap_or_else(|| MigrationPlan {
bucket: bucket.clone(),
classification: super::diff::Classification::Additive,
segments: vec![Segment {
kind: SegmentKind::Transactional,
statements: vec![OperationSql {
label: format!("replay {version}"),
up: replay_sql.up_sql.clone(),
down: replay_sql.down_sql.clone(),
lossy: None,
}],
}],
});
let runner_ctx = RunnerCtx {
bucket: bucket.clone(),
version: version.to_string(),
description: format!("db reset replay of {version}"),
checksum_up: replay_sql.checksum_up,
checksum_down: replay_sql.checksum_down,
snapshot: None,
snapshot_path: None,
// `MigrateConfig` does not derive `Clone` (the type carries
// a small fixed-size payload but the wider Phase 7 stance is
// to construct it explicitly per call so future changes
// surface at every callsite). We mirror the operator's
// settings into a fresh instance.
config: MigrateConfig {
concurrent_warn_relpages: migrate_config.concurrent_warn_relpages,
strict_concurrent_warnings: migrate_config.strict_concurrent_warnings,
pk_flip_long_tx_threshold_secs: migrate_config.pk_flip_long_tx_threshold_secs,
pk_flip_join_table_option: migrate_config.pk_flip_join_table_option,
},
// Replay applies in lexical order, so the runner's
// out-of-order detection should never trip — but supplying
// `AllowWithDiagnostic` matches the usual dev default and
// means a bug here surfaces as a warning rather than a hard
// failure during the time-sensitive reset window.
out_of_order_policy: OutOfOrderPolicy::AllowWithDiagnostic,
// Phase 8.5 Cluster 2 issue #118 — production wire-up. When the
// caller supplied an audit pool on `ResetRequest::audit_pool`
// we plumb it through to `RunnerCtx` so each replayed
// migration writes one `djogi_ddl_audit` row per executed
// segment, exactly as a regular `apply` would. `cloned()`
// bumps the underlying `Arc` (deadpool pools are Arc-shaped)
// so the runner's per-segment context can take ownership of
// its own handle without disturbing the orchestrator's. When
// the caller passed `None` the runner's audit-write loop
// gracefully skips — matching the runner's own best-effort
// stance documented on `record_ddl_audit_for_plan`.
audit_pool: audit_pool.cloned(),
};
apply_plan(ctx, &plan, &runner_ctx, guard)
.await
.map_err(|e| ResetError::ReplayFailed {
version: version.to_string(),
source: e,
})?;
Ok(())
}
// ── URL helpers ───────────────────────────────────────────────────────────
/// Extract the database-name component from a Postgres URL.
///
/// Returns `None` when the URL has no path-component database name
/// (e.g. `postgres://localhost`) — `db reset` cannot derive a database
/// to drop in that case.
///
/// **Percent-decoding.** The path-component bytes are percent-decoded
/// before being returned: a path of `my%2Fdb` decodes to `my/db`.
/// Without this step a URL like `postgres://localhost/my%2Fdb` would
/// make the runner drop the literal identifier `my%2Fdb` (with a
/// `%2F` byte sequence in it) while the post-recreate reconnection
/// would target the correctly-decoded `my/db` — different databases.
/// Decoding produces the SAME byte sequence libpq itself sees when it
/// connects, so the maintenance-DB DROP target matches what the runner
/// re-connects to. Returns `None` on malformed escapes (a `%` not
/// followed by two hex digits) — refusing rather than guessing keeps
/// the destructive path defensive.
///
/// Validation against the Postgres identifier grammar (ASCII letter
/// or underscore followed by ASCII alphanumerics or underscores, up
/// to 63 bytes) is layered on top by [`is_valid_pg_identifier`] —
/// extraction returns the raw decoded string so error messages can
/// surface what the operator actually supplied.
///
/// **No regex.** Walks the URL bytes from the rightmost `/` once and
/// then walks the path bytes once more during the percent-decode.
fn extract_database_from_url(url: &str) -> Option<String> {
// Confirm the URL has a recognised scheme. We accept both
// `postgres://` and `postgresql://`. Anything else is treated as
// libpq parameter form, which db reset doesn't support today
// (the operator would need the URL form for the libpq path).
let body = url
.strip_prefix("postgres://")
.or_else(|| url.strip_prefix("postgresql://"))?;
// Skip past the authority — the database name follows the FIRST
// `/` after the scheme. Authority byte indexing within `body`
// walks until that slash or end-of-string.
let mut idx = 0usize;
let body_bytes = body.as_bytes();
while idx < body_bytes.len() && body_bytes[idx] != b'/' {
idx += 1;
}
if idx >= body_bytes.len() {
return None; // no path component
}
// Path starts after the slash; database name runs until the next
// `?` (query parameters) or end-of-string.
let path_start = idx + 1;
let mut path_end = path_start;
while path_end < body_bytes.len() && body_bytes[path_end] != b'?' {
path_end += 1;
}
if path_end == path_start {
return None; // empty database name
}
percent_decode_strict(&body_bytes[path_start..path_end])
}
/// Percent-decode a byte slice strictly. A `%` must be followed by
/// exactly two hex digits (case-insensitive ASCII); any other shape
/// returns `None`. Output is treated as UTF-8 — non-UTF-8 byte
/// sequences also return `None`.
///
/// Kept private to this module so the destructive `db reset` path is
/// the only consumer; if a future caller needs the same primitive we
/// can promote it without churning the public surface.
fn percent_decode_strict(bytes: &[u8]) -> Option<String> {
let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
let mut i = 0usize;
while i < bytes.len() {
let b = bytes[i];
if b == b'%' {
// Need exactly two hex digits after the `%`.
if i + 2 >= bytes.len() {
return None;
}
let hi = hex_digit_value(bytes[i + 1])?;
let lo = hex_digit_value(bytes[i + 2])?;
out.push((hi << 4) | lo);
i += 3;
} else {
out.push(b);
i += 1;
}
}
String::from_utf8(out).ok()
}
/// Map an ASCII hex digit byte to its 0..=15 value. Returns `None`
/// for any non-hex byte — used by [`percent_decode_strict`] to refuse
/// malformed escapes outright.
fn hex_digit_value(b: u8) -> Option<u8> {
match b {
b'0'..=b'9' => Some(b - b'0'),
b'a'..=b'f' => Some(10 + (b - b'a')),
b'A'..=b'F' => Some(10 + (b - b'A')),
_ => None,
}
}
/// Validate a string against the strict Postgres-identifier grammar:
///
/// > ASCII letter or underscore, followed by zero-or-more ASCII
/// > alphanumerics or underscores, up to 63 bytes total.
///
/// **No regex.** Byte-level checks per `docs/spec/decisions.md` —
/// `u8::is_ascii_alphabetic`, `u8::is_ascii_alphanumeric`, and
/// explicit byte equality against `b'_'`.
///
/// Postgres' own grammar is technically more permissive (it accepts
/// any byte sequence inside double-quoted identifiers), but the
/// grammar above is the one every Djogi-emitted identifier obeys.
/// Refusing anything wider keeps the `DROP DATABASE` /
/// `CREATE DATABASE` paths free of operator-supplied bytes that the
/// double-quote escape elsewhere in the codebase wouldn't otherwise
/// surface.
fn is_valid_pg_identifier(name: &str) -> bool {
let bytes = name.as_bytes();
// Length: 1..=63 bytes (the standard Postgres `NAMEDATALEN - 1`).
if bytes.is_empty() || bytes.len() > 63 {
return false;
}
// Leading byte: ASCII letter or underscore.
let first = bytes[0];
if !first.is_ascii_alphabetic() && first != b'_' {
return false;
}
// Trailing bytes: ASCII alphanumerics or underscore.
for &b in &bytes[1..] {
if !b.is_ascii_alphanumeric() && b != b'_' {
return false;
}
}
true
}
/// Replace the database-name component in a Postgres URL with a new
/// value. Preserves the scheme, authority, and any trailing query
/// string.
///
/// Returns `None` when the URL has no recognisable database component.
///
/// Visible to the rest of the crate so the seed runner (Codex B-1)
/// can reuse the same splice — `db seed --database <name>` derives
/// the per-database connection URL from the application URL by
/// replacing the path component in place.
pub fn replace_db_in_url(url: &str, new_db: &str) -> Option<String> {
let body = url
.strip_prefix("postgres://")
.or_else(|| url.strip_prefix("postgresql://"))?;
let scheme = if url.starts_with("postgres://") {
"postgres://"
} else {
"postgresql://"
};
// Find the path slash.
let mut idx = 0usize;
let body_bytes = body.as_bytes();
while idx < body_bytes.len() && body_bytes[idx] != b'/' {
idx += 1;
}
if idx >= body_bytes.len() {
return None;
}
let authority = &body[..idx];
// Capture any trailing `?query` from the original path.
let path_start = idx + 1;
let mut path_end = path_start;
while path_end < body_bytes.len() && body_bytes[path_end] != b'?' {
path_end += 1;
}
let trailing = &body[path_end..]; // includes leading `?` if present, else empty.
Some(format!("{scheme}{authority}/{new_db}{trailing}"))
}
/// Quote a Postgres identifier for embedding in DDL. Doubles each
/// internal `"` byte and wraps the result in `"`. The byte-level
/// approach matches the rest of the migrate substrate.
fn quote_identifier(name: &str) -> String {
let mut out = String::with_capacity(name.len() + 2);
out.push('"');
for b in name.bytes() {
if b == b'"' {
out.push('"');
out.push('"');
} else {
out.push(b as char);
}
}
out.push('"');
out
}
// ── Tests ─────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
fn temp_root(tag: &str) -> PathBuf {
static COUNTER: AtomicUsize = AtomicUsize::new(0);
let n = COUNTER.fetch_add(1, Ordering::SeqCst);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let p = std::env::temp_dir().join(format!("djogi-reset-{tag}-{nanos}-{n}"));
fs::create_dir_all(&p).unwrap();
p
}
fn req<'a>(
workspace: &'a Path,
url: &'a str,
profile: &'a str,
confirmed: bool,
) -> ResetRequest<'a> {
ResetRequest {
workspace_root: workspace,
database_url: url,
profile,
confirmed,
allow_checksum_drift_reset: false,
maintenance_database: "postgres",
migrate_config: MigrateConfig::default(),
// Gate / URL / replay tests do not assert audit-row
// behaviour — the focused audit-pool wire-up coverage
// lives in `tests/internal/sources/phase8_5_c2_118_*`.
audit_pool: None,
}
}
/// Gate 1 — non-localhost URLs must refuse before any I/O.
#[tokio::test]
async fn refuses_when_url_is_not_localhost() {
let work = temp_root("not_localhost");
let res = reset_app_database(req(
&work,
"postgres://prod.example.com:5432/main",
"development",
true,
))
.await;
match res {
Err(ResetError::Refused(ResetRefusal::NotLocalhost { database_url })) => {
assert_eq!(database_url, "postgres://prod.example.com:5432/main");
}
other => panic!("expected NotLocalhost refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
/// Gate 2 — production profile refuses even against localhost
/// (production-looking infra running locally is still production
/// from a policy perspective).
#[tokio::test]
async fn refuses_when_profile_is_production() {
let work = temp_root("production");
let res =
reset_app_database(req(&work, "postgres://localhost/main", "production", true)).await;
match res {
Err(ResetError::Refused(ResetRefusal::ProductionProfile { profile })) => {
assert_eq!(profile, "production");
}
other => panic!("expected ProductionProfile refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
/// Gate 3 — unconfirmed invocation refuses. The default for
/// `confirmed` (in CLI usage: when `--yes` is absent) is `false`.
#[tokio::test]
async fn refuses_when_not_confirmed() {
let work = temp_root("not_confirmed");
let res = reset_app_database(req(
&work,
"postgres://localhost/main",
"development",
false,
))
.await;
match res {
Err(ResetError::Refused(ResetRefusal::NotConfirmed)) => {}
other => panic!("expected NotConfirmed refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
/// Gate ordering — a request that fails multiple gates should
/// surface the FIRST gate's refusal (localhost > production >
/// confirmation). Operators get a single, deterministic refusal
/// reason rather than a moving target.
#[tokio::test]
async fn gates_evaluate_in_documented_order() {
let work = temp_root("gate_order");
// Non-localhost + production + unconfirmed — every gate
// refuses. The localhost gate should fire first.
let res = reset_app_database(req(
&work,
"postgres://prod.example.com/main",
"production",
false,
))
.await;
match res {
Err(ResetError::Refused(ResetRefusal::NotLocalhost { .. })) => {}
other => panic!("expected NotLocalhost first, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn extract_database_from_url_basic() {
assert_eq!(
extract_database_from_url("postgres://localhost/main"),
Some("main".to_string())
);
assert_eq!(
extract_database_from_url("postgres://user:pass@localhost:5432/main"),
Some("main".to_string())
);
assert_eq!(
extract_database_from_url("postgresql://localhost/main?sslmode=disable"),
Some("main".to_string())
);
}
#[test]
fn extract_database_from_url_missing_path() {
// Authority-only URL (no path component) → None.
assert_eq!(extract_database_from_url("postgres://localhost"), None);
// Trailing slash but empty database → None.
assert_eq!(extract_database_from_url("postgres://localhost/"), None);
// Non-postgres scheme → None.
assert_eq!(extract_database_from_url("mysql://localhost/main"), None);
}
/// Codex round-1 B-2 — percent-decoding has to happen BEFORE the
/// runner sees the database name, otherwise the maintenance-DB
/// drop and the post-recreate reconnect target two different
/// strings (one literal-percent, one decoded). The extractor MUST
/// surface the decoded string so the validator can refuse it.
#[test]
fn extract_database_from_url_percent_decodes_path_bytes() {
// `my%2Fdb` decodes to `my/db` — this is exactly what libpq
// sees when it parses the same URL.
assert_eq!(
extract_database_from_url("postgres://localhost/my%2Fdb"),
Some("my/db".to_string())
);
// Mixed case hex digits round-trip identically.
assert_eq!(
extract_database_from_url("postgres://localhost/foo%2fbar"),
Some("foo/bar".to_string())
);
// A multi-byte UTF-8 sequence (`é` = `c3 a9`) decodes back to
// the same sequence the operator would have typed unencoded.
assert_eq!(
extract_database_from_url("postgres://localhost/caf%C3%A9"),
Some("café".to_string())
);
}
/// Codex round-1 B-2 — malformed `%XX` escapes must refuse rather
/// than silently fall through to a literal `%`. A `%Z9` is not a
/// valid escape; we don't pretend it is.
#[test]
fn extract_database_from_url_rejects_malformed_percent_escapes() {
// Trailing `%` with no following bytes.
assert_eq!(
extract_database_from_url("postgres://localhost/main%"),
None
);
// `%` followed by a single hex digit, then end-of-string.
assert_eq!(
extract_database_from_url("postgres://localhost/main%2"),
None
);
// Non-hex bytes after the `%`.
assert_eq!(
extract_database_from_url("postgres://localhost/main%ZZ"),
None
);
}
/// Codex round-1 B-2 — the strict-identifier grammar covers the
/// happy path (typical names) and refuses anything we won't
/// emit into DDL.
#[test]
fn is_valid_pg_identifier_accepts_typical_names() {
assert!(is_valid_pg_identifier("main"));
assert!(is_valid_pg_identifier("crud_log"));
assert!(is_valid_pg_identifier("event_log"));
assert!(is_valid_pg_identifier("_underscore_lead"));
assert!(is_valid_pg_identifier("a"));
assert!(is_valid_pg_identifier("MyDatabase42"));
// Boundary — exactly 63 bytes is accepted (the Postgres
// NAMEDATALEN-1 limit).
let sixty_three: String = std::iter::repeat_n('a', 63).collect();
assert!(is_valid_pg_identifier(&sixty_three));
}
#[test]
fn is_valid_pg_identifier_refuses_invalid_inputs() {
// Empty.
assert!(!is_valid_pg_identifier(""));
// 64 bytes — one over the limit.
let sixty_four: String = std::iter::repeat_n('a', 64).collect();
assert!(!is_valid_pg_identifier(&sixty_four));
// Leading digit.
assert!(!is_valid_pg_identifier("1main"));
// Internal slash (this is what `my%2Fdb` percent-decodes to).
assert!(!is_valid_pg_identifier("my/db"));
// SQL-injection shape — a single quote is not an identifier
// byte, so the validator rejects the whole string.
assert!(!is_valid_pg_identifier("'; DROP TABLE foo; --"));
// Spaces.
assert!(!is_valid_pg_identifier("my db"));
// Hyphen.
assert!(!is_valid_pg_identifier("my-db"));
// Multi-byte UTF-8 (`café`).
assert!(!is_valid_pg_identifier("café"));
}
/// Codex round-1 B-2 — `reset_app_database` must surface
/// `InvalidDatabaseName` rather than splicing decoded bytes into
/// DDL. We exercise three failure shapes plus the maintenance-DB
/// override path through the public entry.
#[tokio::test]
async fn reset_refuses_when_decoded_database_name_is_not_an_identifier() {
// `my%2Fdb` decodes to `my/db` — gate-passing localhost URL
// but the decoded name fails the identifier grammar.
let work = temp_root("invalid_decoded");
let res = reset_app_database(req(
&work,
"postgres://localhost/my%2Fdb",
"development",
true,
))
.await;
match res {
Err(ResetError::InvalidDatabaseName { name }) => assert_eq!(name, "my/db"),
other => panic!("expected InvalidDatabaseName for `my/db`, got {other:?}"),
}
// The `--maintenance-database` operator-supplied value flows
// through the same validator; a crafted value must refuse.
let bogus_maint = ResetRequest {
workspace_root: &work,
database_url: "postgres://localhost/main",
profile: "development",
confirmed: true,
allow_checksum_drift_reset: false,
maintenance_database: "'; DROP DATABASE main; --",
migrate_config: MigrateConfig::default(),
audit_pool: None,
};
match reset_app_database(bogus_maint).await {
Err(ResetError::InvalidDatabaseName { name }) => {
assert_eq!(name, "'; DROP DATABASE main; --");
}
other => panic!("expected InvalidDatabaseName for maintenance, got {other:?}"),
}
// Boundary — a 64-character all-`a` database refuses (over
// the 63-byte NAMEDATALEN-1 limit).
let too_long: String = std::iter::repeat_n('a', 64).collect();
let url = format!("postgres://localhost/{too_long}");
match reset_app_database(req(&work, &url, "development", true)).await {
Err(ResetError::InvalidDatabaseName { name }) => assert_eq!(name, too_long),
other => panic!("expected InvalidDatabaseName, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn replace_db_in_url_round_trips() {
assert_eq!(
replace_db_in_url("postgres://localhost/main", "postgres"),
Some("postgres://localhost/postgres".to_string())
);
assert_eq!(
replace_db_in_url("postgres://user:pass@localhost:5432/main", "postgres"),
Some("postgres://user:pass@localhost:5432/postgres".to_string())
);
// Query string preserved.
assert_eq!(
replace_db_in_url("postgresql://localhost/main?sslmode=disable", "postgres"),
Some("postgresql://localhost/postgres?sslmode=disable".to_string())
);
}
#[test]
fn quote_identifier_doubles_internal_quotes() {
assert_eq!(quote_identifier("main"), "\"main\"");
// Defensive — Postgres URL parsers don't emit quote bytes in
// database names, but the escape is still correct.
assert_eq!(quote_identifier("a\"b"), "\"a\"\"b\"");
}
#[test]
fn scan_committed_migrations_returns_versions_in_lexical_order() {
use super::super::target::{GLOBAL_BUCKET_DIRNAME, MIGRATIONS_DIR};
let work = temp_root("scan");
// Lay down two buckets with two up files each.
let main_global = work.join(format!("{MIGRATIONS_DIR}/main/{GLOBAL_BUCKET_DIRNAME}"));
let main_billing = work.join(format!("{MIGRATIONS_DIR}/main/billing"));
fs::create_dir_all(&main_global).unwrap();
fs::create_dir_all(&main_billing).unwrap();
fs::write(
main_global.join("V20260301000000__init.sdjql"),
"-- up\nCREATE TABLE foo (id BIGINT PRIMARY KEY);",
)
.unwrap();
fs::write(
main_global.join("V20260301000000__init.down.sdjql"),
"-- down\nDROP TABLE foo;",
)
.unwrap();
fs::write(
main_global.join("V20260201000000__earlier.sdjql"),
"-- up\nCREATE TABLE bar (id BIGINT PRIMARY KEY);",
)
.unwrap();
fs::write(
main_billing.join("V20260401000000__widgets.sdjql"),
"-- up\nCREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
// Hand-written `seed.sql` (no `V` prefix) should be skipped.
fs::write(main_global.join("seed.sql"), "-- not a migration").unwrap();
// The schema_snapshot.json should be skipped (no `.sdjql`
// suffix).
fs::write(main_global.join("schema_snapshot.json"), "{}").unwrap();
let scanned = scan_committed_migrations(&work, "main").unwrap();
// Two buckets, in BTreeMap order.
let mut buckets: Vec<&BucketKey> = scanned.keys().collect();
buckets.sort();
assert_eq!(buckets.len(), 2);
// Global bucket — versions sorted ascending.
let global_bucket = BucketKey {
database: "main".to_string(),
app: String::new(),
};
let billing_bucket = BucketKey {
database: "main".to_string(),
app: "billing".to_string(),
};
assert_eq!(
scanned[&global_bucket],
vec![
"V20260201000000__earlier".to_string(),
"V20260301000000__init".to_string(),
]
);
assert_eq!(
scanned[&billing_bucket],
vec!["V20260401000000__widgets".to_string()]
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn scan_committed_migrations_handles_missing_database_dir() {
let work = temp_root("missing_db");
// No migrations/ tree at all → empty map, no error.
let scanned = scan_committed_migrations(&work, "main").unwrap();
assert!(scanned.is_empty());
let _ = fs::remove_dir_all(&work);
}
fn historical_entry(
database: &str,
app: &str,
version: &str,
checksum_up: &str,
checksum_down: Option<&str>,
) -> HistoricalReplayEntry {
HistoricalReplayEntry {
bucket: bk(database, app),
version: version.to_string(),
status: "applied".to_string(),
checksum_up: checksum_up.to_string(),
checksum_down: checksum_down.map(str::to_string),
}
}
#[test]
fn u275_preflight_checksum_parity_refuses_when_up_sql_drifted() {
let work = temp_root("u275_up_drift");
let bucket = bk("main", "");
let version = "V20260301000000__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
let err = preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
version,
&compute_checksum(["CREATE TABLE widgets (id BIGINT);"]),
None,
)],
false,
)
.expect_err("edited up SQL must refuse before destructive reset");
match err {
ResetError::Refused(ResetRefusal::ChecksumParity { issues }) => {
assert_eq!(issues.len(), 1, "expected one drift issue");
assert_eq!(issues[0].bucket, bucket);
assert_eq!(issues[0].version, version);
assert_eq!(issues[0].sql_side, ResetSqlSide::Up);
assert_eq!(issues[0].problem, ResetChecksumParityProblem::Drift);
assert_eq!(
issues[0].ledger_checksum,
compute_checksum(["CREATE TABLE widgets (id BIGINT);"])
);
assert_eq!(
issues[0].on_disk_checksum.as_deref(),
Some(
compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"])
.as_str()
)
);
}
other => panic!("expected checksum-parity refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_preflight_checksum_parity_refuses_when_down_sql_drifted() {
let work = temp_root("u275_down_drift");
let bucket = bk("main", "");
let version = "V20260301000001__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
"DROP TABLE widgets;",
)
.unwrap();
let err = preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
version,
&compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"]),
Some(&compute_checksum(["DROP TABLE widgets CASCADE;"])),
)],
false,
)
.expect_err("edited down SQL must refuse before destructive reset");
match err {
ResetError::Refused(ResetRefusal::ChecksumParity { issues }) => {
assert_eq!(issues.len(), 1, "expected one drift issue");
assert_eq!(issues[0].bucket, bucket);
assert_eq!(issues[0].version, version);
assert_eq!(issues[0].sql_side, ResetSqlSide::Down);
assert_eq!(issues[0].problem, ResetChecksumParityProblem::Drift);
assert_eq!(
issues[0].ledger_checksum,
compute_checksum(["DROP TABLE widgets CASCADE;"])
);
assert_eq!(
issues[0].on_disk_checksum.as_deref(),
Some(compute_checksum(["DROP TABLE widgets;"]).as_str())
);
}
other => panic!("expected checksum-parity refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_preflight_checksum_parity_refuses_when_historical_file_is_missing() {
let work = temp_root("u275_missing_file");
let err = preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
"V20260301000002__widgets",
&compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"]),
None,
)],
false,
)
.expect_err("missing historical files must refuse before destructive reset");
match err {
ResetError::Refused(ResetRefusal::ChecksumParity { issues }) => {
assert_eq!(issues.len(), 1, "expected one missing-file issue");
assert_eq!(issues[0].version, "V20260301000002__widgets");
assert_eq!(issues[0].sql_side, ResetSqlSide::Up);
assert_eq!(issues[0].problem, ResetChecksumParityProblem::MissingFile);
assert!(
issues[0].on_disk_checksum.is_none(),
"missing file should not claim an on-disk checksum"
);
}
other => panic!("expected checksum-parity refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_preflight_checksum_parity_override_allows_drift() {
let work = temp_root("u275_override");
let bucket = bk("main", "");
let version = "V20260301000003__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
version,
&compute_checksum(["CREATE TABLE widgets (id BIGINT);"]),
None,
)],
true,
)
.expect("explicit override should bypass checksum-parity refusal");
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_preflight_checksum_parity_refuses_when_down_file_is_missing() {
let work = temp_root("u275_missing_down");
let bucket = bk("main", "");
let version = "V20260301000003__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
let err = preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
version,
&compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"]),
Some(&compute_checksum(["DROP TABLE widgets;"])),
)],
false,
)
.expect_err("missing down SQL must refuse before destructive reset");
match err {
ResetError::Refused(ResetRefusal::ChecksumParity { issues }) => {
assert_eq!(issues.len(), 1, "expected one missing-file issue");
assert_eq!(issues[0].version, version);
assert_eq!(issues[0].sql_side, ResetSqlSide::Down);
assert_eq!(issues[0].problem, ResetChecksumParityProblem::MissingFile);
assert!(issues[0].on_disk_checksum.is_none());
}
other => panic!("expected checksum-parity refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
fn composed_up_sql(version: &str, body: &str) -> String {
format!(
"-- Djogi composed migration — up\n\
-- Version: {version}\n\
-- Bucket: main/_global_\n\
-- Classification: Additive\n\
-- DO NOT EDIT — regenerate via `djogi migrations compose`.\n\n\
-- AddTable widgets\n\
{body}\n\n"
)
}
fn composed_down_sql(version: &str, body: &str) -> String {
format!(
"-- Djogi composed migration — down\n\
-- Version: {version}\n\
-- Bucket: main/_global_\n\
-- DO NOT EDIT — regenerate via `djogi migrations compose`.\n\n\
-- DropTable widgets\n\
{body}\n\n"
)
}
#[test]
fn u275_preflight_checksum_parity_accepts_composed_sql_headers() {
let work = temp_root("u275_composed_headers");
let bucket = bk("main", "");
let version = "V20260301000012__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
composed_up_sql(version, "CREATE TABLE widgets (id BIGINT PRIMARY KEY);"),
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
composed_down_sql(version, "DROP TABLE widgets;"),
)
.unwrap();
preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
version,
&compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"]),
Some(&compute_checksum(["DROP TABLE widgets;"])),
)],
false,
)
.expect("composed comments and labels must not count as checksum drift");
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_preflight_checksum_parity_refuses_edited_composed_operation_sql() {
let work = temp_root("u275_composed_operation_drift");
let bucket = bk("main", "");
let version = "V20260301000013__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
composed_up_sql(version, "CREATE TABLE widgets (id BIGINT PRIMARY KEY);"),
)
.unwrap();
let err = preflight_reset_checksum_parity(
&work,
"main",
&[historical_entry(
"main",
"",
version,
&compute_checksum(["CREATE TABLE widgets (id BIGINT);"]),
None,
)],
false,
)
.expect_err("operation SQL drift inside composed file must refuse");
match err {
ResetError::Refused(ResetRefusal::ChecksumParity { issues }) => {
assert_eq!(issues.len(), 1);
assert_eq!(issues[0].problem, ResetChecksumParityProblem::Drift);
assert_eq!(
issues[0].on_disk_checksum.as_deref(),
Some(
compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"])
.as_str()
)
);
}
other => panic!("expected checksum-parity refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_preflight_checksum_parity_refuses_when_baseline_row_cannot_be_compared() {
let work = temp_root("u275_baseline");
let bucket = bk("main", "");
let version = "V20260301000004__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
let err = preflight_reset_checksum_parity(
&work,
"main",
&[HistoricalReplayEntry {
bucket,
version: version.to_string(),
status: "baseline".to_string(),
checksum_up: "V1:baseline-projection".to_string(),
checksum_down: None,
}],
false,
)
.expect_err("baseline rows must refuse when reset cannot establish file parity");
match err {
ResetError::Refused(ResetRefusal::ChecksumParity { issues }) => {
assert_eq!(issues.len(), 1, "expected one baseline issue");
assert_eq!(issues[0].version, version);
assert_eq!(
issues[0].problem,
ResetChecksumParityProblem::UnsupportedBaseline
);
assert!(issues[0].on_disk_checksum.is_none());
}
other => panic!("expected checksum-parity refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_replay_sql_checksums_include_down_when_down_file_exists() {
let work = temp_root("u275_replay_checksums");
let bucket = bk("main", "");
let version = "V20260301000005__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
"DROP TABLE widgets;",
)
.unwrap();
let replay_sql =
read_replay_sql_files(&work, &bucket, version).expect("load replay SQL files");
assert_eq!(
replay_sql.checksum_up,
compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"])
);
assert_eq!(
replay_sql.checksum_down.as_deref(),
Some(compute_checksum(["DROP TABLE widgets;"]).as_str()),
"reset replay must preserve checksum_down so later resets still enforce down-side parity"
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_replay_sql_checksums_treat_comment_only_down_as_none() {
let work = temp_root("u275_comment_only_down");
let bucket = bk("main", "");
let version = "V20260301000016__phase_zero_like";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE SCHEMA IF NOT EXISTS heeranjid;",
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
"-- no meaningful rollback\n-- framework bootstrap is dependency-only\n",
)
.unwrap();
let replay_sql =
read_replay_sql_files(&work, &bucket, version).expect("load replay SQL files");
assert!(
replay_sql.checksum_down.is_none(),
"comment-only down files must preserve the no-real-rollback null sentinel"
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_preflight_replay_semantics_refuses_non_transactional_sql_without_manifest() {
let work = temp_root("u276_missing_manifest");
let bucket = bk("main", "");
let version = "V20260301000006__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"-- AddTable widgets\n\
CREATE TABLE widgets (id BIGINT PRIMARY KEY);\n\n\
-- AddIndex widgets_id_idx\n\
CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);\n",
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
"DROP INDEX CONCURRENTLY widgets_id_idx;\nDROP TABLE widgets;\n",
)
.unwrap();
let err = preflight_reset_replay_semantics(&work, "main")
.expect_err("legacy file-only concurrent index replay must refuse before drop");
match err {
ResetError::Refused(ResetRefusal::ReplaySemantics { issues }) => {
assert_eq!(issues.len(), 1, "expected one replay-semantics issue");
assert_eq!(issues[0].bucket, bucket);
assert_eq!(issues[0].version, version);
assert_eq!(
issues[0].problem,
ResetReplaySemanticsProblem::MissingReplayPlan
);
assert_eq!(issues[0].statement_shape, "CREATE INDEX CONCURRENTLY");
}
other => panic!("expected replay-semantics refusal, got {other:?}"),
}
let _ = fs::remove_dir_all(&work);
}
fn assert_single_replay_semantics_issue(
err: ResetError,
expected_bucket: &BucketKey,
expected_version: &str,
expected_problem: ResetReplaySemanticsProblem,
expected_shape: &str,
) {
match err {
ResetError::Refused(ResetRefusal::ReplaySemantics { issues }) => {
assert_eq!(issues.len(), 1, "expected one replay-semantics issue");
assert_eq!(issues[0].bucket, *expected_bucket);
assert_eq!(issues[0].version, expected_version);
assert_eq!(issues[0].problem, expected_problem);
assert_eq!(issues[0].statement_shape, expected_shape);
}
other => panic!("expected replay-semantics refusal, got {other:?}"),
}
}
#[test]
fn u276_preflight_replay_semantics_refuses_call_backfill_without_manifest() {
let work = temp_root("u276_missing_manifest_call");
let bucket = bk("main", "");
let version = "V20260301000008__pk_flip_call";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CALL heeranjid_bulk_backfill('widgets', 'id', 'id_desc', 'heer', 10000);\n",
)
.unwrap();
fs::write(bucket_dir.join(down_filename(version)), "SELECT 1;\n").unwrap();
let err = preflight_reset_replay_semantics(&work, "main")
.expect_err("CALL backfill replay without manifest must refuse before drop");
assert_single_replay_semantics_issue(
err,
&bucket,
version,
ResetReplaySemanticsProblem::MissingReplayPlan,
"CALL heeranjid_bulk_backfill",
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_preflight_replay_semantics_refuses_do_backfill_without_manifest() {
let work = temp_root("u276_missing_manifest_do");
let bucket = bk("main", "");
let version = "V20260301000009__pk_flip_do";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"DO $$\n\
BEGIN\n\
COMMIT;\n\
END\n\
$$;\n",
)
.unwrap();
fs::write(bucket_dir.join(down_filename(version)), "SELECT 1;\n").unwrap();
let err = preflight_reset_replay_semantics(&work, "main")
.expect_err("DO backfill replay without manifest must refuse before drop");
assert_single_replay_semantics_issue(
err,
&bucket,
version,
ResetReplaySemanticsProblem::MissingReplayPlan,
"DO block with COMMIT",
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_preflight_replay_semantics_refuses_partitioned_placeholder_without_manifest() {
let work = temp_root("u276_missing_manifest_partitioned");
let bucket = bk("main", "");
let version = "V20260301000010__pk_flip_partitioned";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
"CREATE UNIQUE INDEX events_partition_key_id_desc_idx\n \
ON ONLY events (partition_key, id_desc);\n\
-- Per leaf: CREATE UNIQUE INDEX CONCURRENTLY <leaf>_partition_key_id_desc_idx\n\
-- ON <leaf> (partition_key, id_desc);\n\
-- Then ALTER INDEX events_partition_key_id_desc_idx ATTACH PARTITION\n\
-- <leaf>_partition_key_id_desc_idx;\n",
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
"DROP INDEX IF EXISTS events_partition_key_id_desc_idx;\n",
)
.unwrap();
let err = preflight_reset_replay_semantics(&work, "main")
.expect_err("partitioned placeholder replay without manifest must refuse before drop");
assert_single_replay_semantics_issue(
err,
&bucket,
version,
ResetReplaySemanticsProblem::MissingReplayPlan,
"PARTITIONED CONCURRENTLY placeholder",
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_load_reset_replay_plan_refuses_invalid_call_manifest() {
let work = temp_root("u276_invalid_manifest_call");
let bucket = bk("main", "");
let version = "V20260301000011__pk_flip_call_invalid";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
let up_sql = "CALL heeranjid_bulk_backfill('widgets', 'id', 'id_desc', 'heer', 10000);\n";
fs::write(bucket_dir.join(up_filename(version)), up_sql).unwrap();
fs::write(bucket_dir.join(down_filename(version)), "SELECT 1;\n").unwrap();
fs::write(
bucket_dir.join(format!("{version}.plan.json")),
"{\n \"format_version\": \"1\",\n \"checksum_up\": \"V1:stale\",\n \"checksum_down\": null,\n \"classification\": { \"kind\": \"additive\" },\n \"segments\": []\n}\n",
)
.unwrap();
let replay_sql =
read_replay_sql_files(&work, &bucket, version).expect("load replay SQL files");
let err = load_reset_replay_plan(&work, &bucket, version, &replay_sql)
.expect_err("invalid non-transactional replay manifest must refuse");
assert_single_replay_semantics_issue(
err,
&bucket,
version,
ResetReplaySemanticsProblem::InvalidReplayPlan,
"CALL heeranjid_bulk_backfill",
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_load_reset_replay_plan_preserves_committed_segment_kinds() {
let work = temp_root("u276_manifest_roundtrip");
let bucket = bk("main", "");
let version = "V20260301000007__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
let up_sql = "-- AddTable widgets\n\
CREATE TABLE widgets (id BIGINT PRIMARY KEY);\n\n\
-- AddIndex widgets_id_idx\n\
CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);\n";
let down_sql = "DROP INDEX CONCURRENTLY widgets_id_idx;\nDROP TABLE widgets;\n";
fs::write(bucket_dir.join(up_filename(version)), up_sql).unwrap();
fs::write(bucket_dir.join(down_filename(version)), down_sql).unwrap();
fs::write(
bucket_dir.join(format!("{version}.plan.json")),
format!(
"{{\n \"format_version\": \"1\",\n \"checksum_up\": \"{}\",\n \"checksum_down\": \"{}\",\n \"classification\": {{ \"kind\": \"additive\" }},\n \"segments\": [\n {{\n \"kind\": \"transactional\",\n \"statements\": [\n {{ \"label\": \"AddTable widgets\", \"up\": \"CREATE TABLE widgets (id BIGINT PRIMARY KEY);\" }}\n ]\n }},\n {{\n \"kind\": \"non_transactional\",\n \"statements\": [\n {{ \"label\": \"AddIndex widgets_id_idx\", \"up\": \"CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);\" }}\n ]\n }}\n ]\n}}\n",
compute_checksum([up_sql]),
compute_checksum([down_sql]),
),
)
.unwrap();
let replay_sql =
read_replay_sql_files(&work, &bucket, version).expect("load replay SQL files");
let plan = load_reset_replay_plan(&work, &bucket, version, &replay_sql)
.expect("manifest should load")
.expect("manifest should produce a replay plan");
assert_eq!(plan.bucket, bucket);
assert_eq!(
plan.classification,
crate::migrate::Classification::Additive
);
assert_eq!(plan.segments.len(), 2);
assert_eq!(plan.segments[0].kind, SegmentKind::Transactional);
assert_eq!(plan.segments[0].statements[0].label, "AddTable widgets");
assert_eq!(plan.segments[1].kind, SegmentKind::NonTransactional);
assert_eq!(
plan.segments[1].statements[0].label,
"AddIndex widgets_id_idx"
);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_load_reset_replay_plan_accepts_composed_sql_operation_checksums() {
let work = temp_root("u276_manifest_composed_checksum");
let bucket = bk("main", "");
let version = "V20260301000014__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
composed_up_sql(
version,
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);\n\n\
-- AddIndex widgets_id_idx\n\
CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);",
),
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
composed_down_sql(
version,
"DROP INDEX CONCURRENTLY widgets_id_idx;\nDROP TABLE widgets;",
),
)
.unwrap();
fs::write(
bucket_dir.join(format!("{version}.plan.json")),
format!(
"{{\n \"format_version\": \"1\",\n \"checksum_up\": \"{}\",\n \"checksum_down\": \"{}\",\n \"classification\": {{ \"kind\": \"additive\" }},\n \"segments\": [\n {{\n \"kind\": \"transactional\",\n \"statements\": [\n {{ \"label\": \"AddTable widgets\", \"up\": \"CREATE TABLE widgets (id BIGINT PRIMARY KEY);\" }}\n ]\n }},\n {{\n \"kind\": \"non_transactional\",\n \"statements\": [\n {{ \"label\": \"AddIndex widgets_id_idx\", \"up\": \"CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);\" }}\n ]\n }}\n ]\n}}\n",
compute_checksum([
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
"CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);"
]),
compute_checksum(["DROP INDEX CONCURRENTLY widgets_id_idx;\nDROP TABLE widgets;"]),
),
)
.unwrap();
let replay_sql =
read_replay_sql_files(&work, &bucket, version).expect("load replay SQL files");
assert_eq!(
replay_sql.checksum_up,
compute_checksum([
"CREATE TABLE widgets (id BIGINT PRIMARY KEY);",
"CREATE INDEX CONCURRENTLY widgets_id_idx ON widgets (id);"
])
);
let plan = load_reset_replay_plan(&work, &bucket, version, &replay_sql)
.expect("manifest should load")
.expect("manifest should produce a replay plan");
assert_eq!(plan.segments.len(), 2);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u276_load_reset_replay_plan_accepts_composed_comment_only_down_manifest() {
let work = temp_root("u276_manifest_comment_only_down");
let bucket = bk("main", "");
let version = "V20260301000015__widgets";
let bucket_dir = super::super::target::bucket_dir(&work, &bucket);
fs::create_dir_all(&bucket_dir).unwrap();
fs::write(
bucket_dir.join(up_filename(version)),
composed_up_sql(version, "CREATE TABLE widgets (id BIGINT PRIMARY KEY);"),
)
.unwrap();
fs::write(
bucket_dir.join(down_filename(version)),
composed_down_sql(
version,
"-- no-op rollback placeholder: lossy operation requires manual rollback",
),
)
.unwrap();
fs::write(
bucket_dir.join(format!("{version}.plan.json")),
format!(
"{{\n \"format_version\": \"1\",\n \"checksum_up\": \"{}\",\n \"checksum_down\": null,\n \"classification\": {{ \"kind\": \"lossy\" }},\n \"segments\": [\n {{\n \"kind\": \"transactional\",\n \"statements\": [\n {{ \"label\": \"AddTable widgets\", \"up\": \"CREATE TABLE widgets (id BIGINT PRIMARY KEY);\" }}\n ]\n }}\n ]\n}}\n",
compute_checksum(["CREATE TABLE widgets (id BIGINT PRIMARY KEY);"]),
),
)
.unwrap();
let replay_sql =
read_replay_sql_files(&work, &bucket, version).expect("load replay SQL files");
assert!(
replay_sql.checksum_down.is_none(),
"composed comment-only down files must preserve compose's checksum_down = null"
);
let plan = load_reset_replay_plan(&work, &bucket, version, &replay_sql)
.expect("manifest should load")
.expect("manifest should produce a replay plan");
assert_eq!(plan.segments.len(), 1);
let _ = fs::remove_dir_all(&work);
}
#[test]
fn u275_checksum_parity_refusal_display_names_bucket_version_and_checksums() {
let refusal = ResetRefusal::ChecksumParity {
issues: vec![
ResetChecksumParityIssue {
bucket: bk("main", ""),
version: "V20260301000004__widgets".to_string(),
sql_side: ResetSqlSide::Up,
ledger_checksum: "V1:ledger".to_string(),
on_disk_checksum: Some("V1:disk".to_string()),
problem: ResetChecksumParityProblem::Drift,
},
ResetChecksumParityIssue {
bucket: bk("main", "billing"),
version: "V20260301000005__billing".to_string(),
sql_side: ResetSqlSide::Down,
ledger_checksum: "V1:down-ledger".to_string(),
on_disk_checksum: None,
problem: ResetChecksumParityProblem::MissingFile,
},
],
};
let rendered = refusal.to_string();
assert!(
rendered.contains("main/_global_/V20260301000004__widgets"),
"{rendered}"
);
assert!(
rendered.contains("main/billing/V20260301000005__billing"),
"{rendered}"
);
assert!(rendered.contains("V1:ledger"), "{rendered}");
assert!(rendered.contains("V1:disk"), "{rendered}");
assert!(rendered.contains("V1:down-ledger"), "{rendered}");
assert!(
rendered.contains("--allow-checksum-drift-reset")
|| rendered.contains("allow_checksum_drift_reset"),
"{rendered}"
);
}
// ── Codex umbrella U-4: historical-order replay plan ────────────────
fn bk(database: &str, app: &str) -> BucketKey {
BucketKey {
database: database.to_string(),
app: app.to_string(),
}
}
/// `build_replay_plan` honours the historical apply order: when
/// `0001 → applied_at_rank 0`, `0003 → rank 1`, `0002 → rank 2`,
/// the replay plan is `[0001, 0003, 0002]` — NOT lexical
/// `[0001, 0002, 0003]`. This is the load-bearing umbrella U-4
/// invariant.
#[test]
fn u4_replay_plan_honours_historical_apply_order_over_lexical() {
let bucket = bk("main", "");
let mut buckets = BTreeMap::new();
buckets.insert(
bucket.clone(),
vec![
"V20260101000000__a".to_string(),
"V20260201000000__b".to_string(),
"V20260301000000__c".to_string(),
],
);
// Out-of-order historical apply: a (0), c (1), b (2). Lexical
// would be a, b, c — different order!
let mut historical = BTreeMap::new();
historical.insert((bucket.clone(), "V20260101000000__a".to_string()), 0u64);
historical.insert((bucket.clone(), "V20260301000000__c".to_string()), 1u64);
historical.insert((bucket.clone(), "V20260201000000__b".to_string()), 2u64);
let plan = build_replay_plan(&buckets, &historical);
let versions: Vec<&str> = plan.iter().map(|(_, v)| v.as_str()).collect();
assert_eq!(
versions,
vec![
"V20260101000000__a",
"V20260301000000__c",
"V20260201000000__b",
],
"historical apply order MUST win over lexical version sort"
);
}
/// When NO historical order exists (fresh DB, ledger missing),
/// the plan falls back to lexical version-string sort. This is
/// the safe-by-default behaviour that pre-umbrella reset always
/// used.
#[test]
fn u4_replay_plan_falls_back_to_lexical_when_historical_empty() {
let bucket = bk("main", "");
let mut buckets = BTreeMap::new();
buckets.insert(
bucket.clone(),
vec![
"V20260201000000__b".to_string(),
"V20260101000000__a".to_string(),
"V20260301000000__c".to_string(),
],
);
let historical: BTreeMap<(BucketKey, String), u64> = BTreeMap::new();
let plan = build_replay_plan(&buckets, &historical);
let versions: Vec<&str> = plan.iter().map(|(_, v)| v.as_str()).collect();
assert_eq!(
versions,
vec![
"V20260101000000__a",
"V20260201000000__b",
"V20260301000000__c",
],
"empty historical map → lexical sort"
);
}
/// Mixed shape: some versions have historical entries, some
/// don't (typical when files were added on disk after the last
/// apply). The historical ones come first in apply-order; the
/// rest sort lexically afterwards.
#[test]
fn u4_replay_plan_mixes_historical_and_new_disk_files() {
let bucket = bk("main", "");
let mut buckets = BTreeMap::new();
buckets.insert(
bucket.clone(),
vec![
"V20260101000000__a".to_string(), // historical, rank 0
"V20260201000000__b".to_string(), // not historical
"V20260301000000__c".to_string(), // historical, rank 1
"V20260401000000__d".to_string(), // not historical
],
);
let mut historical = BTreeMap::new();
historical.insert((bucket.clone(), "V20260101000000__a".to_string()), 0u64);
historical.insert((bucket.clone(), "V20260301000000__c".to_string()), 1u64);
let plan = build_replay_plan(&buckets, &historical);
let versions: Vec<&str> = plan.iter().map(|(_, v)| v.as_str()).collect();
assert_eq!(
versions,
vec![
"V20260101000000__a", // rank 0
"V20260301000000__c", // rank 1
"V20260201000000__b", // no rank, lexical first among non-historical
"V20260401000000__d", // no rank, lexical second
],
);
}
// ── Codex umbrella round-2 U-6: error-policy classification ─────────
/// Connecting to a syntactically valid but unreachable URL must
/// classify as `Transient`, NOT `LedgerMissing`. Pre-U-6 the
/// connect-failure path collapsed to an empty map and the
/// destructive operation would proceed; post-U-6 the call surfaces
/// the failure so the caller refuses to drop / recreate.
///
/// We point at a port nobody listens on (TCP `:1` is the standard
/// "discard" pseudo-port — kernels reject the connect immediately).
#[tokio::test]
async fn u6_capture_failure_unreachable_url_classifies_as_transient() {
let url = "postgres://djogi:djogi@127.0.0.1:1/nonexistent_db";
let res = capture_historical_apply_order(url).await;
match res {
Err(HistoricalCaptureError::Transient(e)) => {
let msg = format!("{e}");
assert!(
msg.contains("connect failed")
|| msg.contains("connect")
|| msg.contains("Connection")
|| msg.contains("connection")
|| msg.contains("refused"),
"Transient message must surface the connect failure: {msg}"
);
}
Err(HistoricalCaptureError::LedgerMissing) => {
panic!(
"U-6: connect failure must classify as Transient, NOT LedgerMissing — \
pre-fix the unwrap_or_default() collapsed both into the same fallback"
);
}
Ok(_) => panic!("U-6: connect to :1 must fail"),
}
}
/// `ResetError::HistoricalOrderCaptureFailed` is plumbed through
/// `reset_app_database` end-to-end. We construct a request that
/// passes every gate (localhost URL with valid identifier name,
/// development profile, confirmed) but points at an unreachable
/// port — the historical-order capture's connect step fails and
/// the variant must propagate.
///
/// CRITICAL invariant: pre-fix this same scenario would have
/// `unwrap_or_default()` ed the failure and proceeded into the
/// destructive `drop_and_create_database` call. Post-fix the
/// request returns `HistoricalOrderCaptureFailed` BEFORE any
/// destructive operation runs.
#[tokio::test]
async fn u6_reset_propagates_capture_failure_before_destructive_op() {
let work = temp_root("u6_capture_propagate");
// Localhost with a deliberately-wrong port so the gate passes
// but the connect fails. The is_localhost_connection helper
// accepts host=127.0.0.1 / host=localhost regardless of port.
let url = "postgres://djogi:djogi@127.0.0.1:1/main";
let res = reset_app_database(req(&work, url, "development", true)).await;
match res {
Err(ResetError::HistoricalOrderCaptureFailed { source }) => {
let msg = format!("{source}");
assert!(
msg.contains("connect")
|| msg.contains("connection")
|| msg.contains("refused"),
"source message must surface the underlying connect failure: {msg}"
);
}
other => panic!(
"U-6: expected HistoricalOrderCaptureFailed; got {other:?} \
(pre-fix this would have proceeded into the destructive drop)"
),
}
let _ = fs::remove_dir_all(&work);
}
/// The Display impl for `HistoricalOrderCaptureFailed` carries
/// operator-actionable language: it names the underlying source,
/// explains why we refused, and tells the operator what to do
/// next. This is the message a CI script or human will see when
/// the gate fires.
#[test]
fn u6_historical_order_capture_failed_display_is_actionable() {
let e = ResetError::HistoricalOrderCaptureFailed {
source: DjogiError::Db(DbError::other("connection refused".to_string())),
};
let s = format!("{e}");
assert!(s.contains("connection refused"), "must echo source: {s}");
assert!(
s.contains("refusing to proceed") || s.contains("refusing"),
"must explain we refused: {s}"
);
assert!(
s.contains("re-run") || s.contains("rerun"),
"must guide remediation: {s}"
);
}
/// `ResetError::source()` returns the underlying `DjogiError` for
/// the U-6 variant — the `?` operator and `tracing` style error
/// chains depend on that. The pre-existing variants in the impl
/// already do this; the test guards against forgetting the new
/// arm if someone touches the match block later.
#[test]
fn u6_historical_order_capture_failed_carries_source() {
use std::error::Error;
let e = ResetError::HistoricalOrderCaptureFailed {
source: DjogiError::Db(DbError::other("decode failed".to_string())),
};
let src = e.source().expect("must have source");
assert!(
src.to_string().contains("decode failed"),
"source must carry inner message: {}",
src
);
}
/// Cross-bucket: historical apply ranks impose order across
/// different buckets too. Without that property, two buckets
/// applied historically as `bucketA/v1, bucketB/v1, bucketA/v2`
/// would lose the interleaved ordering.
#[test]
fn u4_replay_plan_orders_across_buckets_by_historical_rank() {
let a = bk("main", "users");
let b = bk("main", "billing");
let mut buckets = BTreeMap::new();
buckets.insert(
a.clone(),
vec!["V0001__a1".to_string(), "V0003__a2".to_string()],
);
buckets.insert(b.clone(), vec!["V0002__b1".to_string()]);
// Historical apply: a/V0001 (0), b/V0002 (1), a/V0003 (2).
let mut historical = BTreeMap::new();
historical.insert((a.clone(), "V0001__a1".to_string()), 0u64);
historical.insert((b.clone(), "V0002__b1".to_string()), 1u64);
historical.insert((a.clone(), "V0003__a2".to_string()), 2u64);
let plan = build_replay_plan(&buckets, &historical);
let render: Vec<String> = plan
.iter()
.map(|(bucket, v)| format!("{}/{}/{}", bucket.database, bucket.app, v))
.collect();
assert_eq!(
render,
vec![
"main/users/V0001__a1".to_string(),
"main/billing/V0002__b1".to_string(),
"main/users/V0003__a2".to_string(),
],
"interleaved cross-bucket apply order must be preserved"
);
}
#[test]
fn reset_discovers_sdjql_migration_files() {
let root = temp_root("sdjql-discovery");
let bucket = super::super::target::bucket_dir(&root, &bk("main", "myapp"));
fs::create_dir_all(&bucket).unwrap();
fs::write(
bucket.join("V20260501000000__new.sdjql"),
"-- Djogi composed migration — up\n-- Version: V20260501000000__new\n\
-- Bucket: main/myapp\n-- Classification: Additive\n--\n\
-- Apply via `djogi migrations apply`, not psql...\n-- DO NOT EDIT...\n\nCREATE TABLE items (id bigint PRIMARY KEY);\n"
).unwrap();
fs::write(
bucket.join("V20260501000000__new.down.sdjql"),
"-- Djogi composed migration — down\n-- Version: V20260501000000__new\n\
-- Bucket: main/myapp\n-- DO NOT EDIT...\n\nDROP TABLE items;\n",
)
.unwrap();
let scanned =
super::super::target::scan_filesystem_with_files(&root, Some("main")).unwrap();
let bk_main = bk("main", "myapp");
assert!(
scanned.contains_key(&bk_main),
"scanner must discover .sdjql files"
);
assert_eq!(
scanned[&bk_main].len(),
1,
"should find exactly one up-side file"
);
assert!(scanned[&bk_main].contains_key("V20260501000000__new"));
let _ = fs::remove_dir_all(&root);
}
#[test]
fn reset_rejects_legacy_sql_migration_files() {
let root = temp_root("legacy-sql-reject");
let bucket = super::super::target::bucket_dir(&root, &bk("main", "myapp"));
fs::create_dir_all(&bucket).unwrap();
fs::write(
bucket.join("V20260301000000__legacy.sql"),
"-- Djogi composed migration — up\n-- Version: V20260301000000__legacy\n\
CREATE TABLE users (id bigint PRIMARY KEY);\n",
)
.unwrap();
let result = super::super::target::scan_filesystem_with_files(&root, Some("main"));
assert!(
result.is_err(),
"reset must reject legacy .sql schema migration files"
);
let err = result.unwrap_err().to_string();
assert!(
err.contains("V20260301000000__legacy.sql") || err.contains("legacy"),
"error must mention the legacy file: {err}"
);
let _ = fs::remove_dir_all(&root);
}
}