djogi 0.1.0-alpha.4

//! Migration ledger — `djogi_schema_migrations` DDL bootstrap, row
//! CRUD helpers, and the `V1:<sha256-hex>` checksum format.
//!
//! # Schema (Phase 7 v3 §6)
//!
//! ```sql
//! CREATE TABLE IF NOT EXISTS djogi_schema_migrations (
//!     id                    BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
//!     version               TEXT        NOT NULL UNIQUE,
//!     description           TEXT        NOT NULL DEFAULT '',
//!     checksum_up           VARCHAR(68) NOT NULL,
//!     checksum_down         VARCHAR(68),
//!     execution_mode        TEXT        NOT NULL DEFAULT 'transactional'
//!                              CHECK (execution_mode IN ('transactional',
//!                                                          'non_transactional')),
//!     status                TEXT        NOT NULL DEFAULT 'pending'
//!                              CHECK (status IN ('pending', 'applied',
//!                                                  'baseline', 'faked',
//!                                                  'rolled_back', 'failed')),
//!     applied_at            TIMESTAMPTZ NOT NULL DEFAULT now(),
//!     applied_by            TEXT        NOT NULL DEFAULT current_user,
//!     execution_time_ms     BIGINT      NOT NULL DEFAULT 0,
//!     out_of_order_flag     BOOLEAN     NOT NULL DEFAULT FALSE,
//!     applied_steps_count   INTEGER     NOT NULL DEFAULT 0,
//!     total_steps           INTEGER,
//!     partial_apply_note    TEXT,
//!     run_id                BIGINT      NOT NULL,
//!     snapshot_version      TEXT        NOT NULL,
//!     app_label             TEXT        NOT NULL DEFAULT '',
//!     leaf_identity         TEXT
//! );
//! ```
//!
//! `out_of_order_flag` is a separate Boolean from `status` — the
//! flag describes ordering, the status describes lifecycle. Re-adding
//! an `out_of_order` status enum value would conflate the two; the
//! schema CHECK enforces the separation.
//!
//! # Checksum format
//!
//! Per OQ-05: every row carries a `V1:<sha256-hex>` checksum on
//! `checksum_up` (always) and `checksum_down` (when the operation has
//! a non-comment rollback). Format: literal `V1:` (3 bytes) + 64
//! lowercase hex chars (the hex SHA-256 of the per-segment `up`/`down`
//! SQL concatenated with `\n`). Total = 67 bytes; the column type is
//! `VARCHAR(68)` for one byte of tolerance against future versioning.
//!
//! # API surface
//!
//! - [`bootstrap`] — idempotently create the ledger table.
//! - [`compute_checksum`] — hash a sequence of SQL fragments into the
//!   `V1:<hex>` form.
//! - [`LedgerRow`] / [`LedgerStatus`] / [`ExecutionMode`] — typed row
//!   shape used by the runner T4.
//! - [`insert_pending`] / [`mark_applied`] / [`mark_failed`] /
//!   [`mark_partial`] — row mutation helpers. Each opens its own
//!   pool connection or runs inside the runner-supplied transaction
//!   per the dispatch rules in T4.
//!
//! All CRUD helpers take `&mut DjogiContext` so the runner can route
//! through either an open transaction (for the per-migration
//! transactional segment) or the pool (for the standalone INSERT
//! that records the run before the transaction opens).

use sha2::{Digest, Sha256};

use crate::__bypass::RawAccessExt as _;
use crate::context::DjogiContext;
use crate::error::{DbError, DjogiError};

// ── LedgerRow row conversion ──────────────────────────────────────────────

/// Constant SQL for the 15-column ledger SELECT. Shared between
/// [`load_full_row_by_version`] and callers that append their own
/// WHERE / ORDER BY clauses, keeping the column list in one place.
pub(crate) const LEDGER_SELECT_COLS: &str = "SELECT version, description, checksum_up, checksum_down, execution_mode, \
     status, execution_time_ms, out_of_order_flag, applied_steps_count, \
     total_steps, partial_apply_note, run_id, snapshot_version, app_label, leaf_identity \
     FROM djogi_schema_migrations";

/// Convert a raw `tokio_postgres::Row` (returned by any of the 15-column
/// ledger queries) into a [`LedgerRow`]. Column order matches
/// [`LEDGER_SELECT_COLS`] exactly; callers that append a different
/// SELECT list must NOT use this impl.
///
/// `version` is read from column 0, so both the by-version query
/// (which supplies `$1` in the WHERE) and the all-rows query (which
/// SELECTs `version` as column 0) produce a compatible row.
impl TryFrom<&tokio_postgres::Row> for LedgerRow {
    type Error = tokio_postgres::Error;

    fn try_from(row: &tokio_postgres::Row) -> Result<Self, Self::Error> {
        let version: String = row.try_get(0)?;
        let description: String = row.try_get(1)?;
        let checksum_up: String = row.try_get(2)?;
        let checksum_down: Option<String> = row.try_get(3)?;
        let execution_mode_s: String = row.try_get(4)?;
        let status_s: String = row.try_get(5)?;
        let execution_time_ms: i64 = row.try_get(6)?;
        let out_of_order_flag: bool = row.try_get(7)?;
        let applied_steps_count: i32 = row.try_get(8)?;
        let total_steps: Option<i32> = row.try_get(9)?;
        let partial_apply_note: Option<String> = row.try_get(10)?;
        let run_id: i64 = row.try_get(11)?;
        let snapshot_version: String = row.try_get(12)?;
        let app_label: String = row.try_get(13)?;
        let leaf_identity: Option<String> = row.try_get(14)?;

        let execution_mode = match execution_mode_s.as_str() {
            "transactional" => ExecutionMode::Transactional,
            _ => ExecutionMode::NonTransactional,
        };
        let status = LedgerStatus::from_db_str(&status_s).unwrap_or(LedgerStatus::Failed);

        Ok(LedgerRow {
            version,
            description,
            checksum_up,
            checksum_down,
            execution_mode,
            status,
            execution_time_ms,
            out_of_order_flag,
            applied_steps_count,
            total_steps,
            partial_apply_note,
            run_id,
            snapshot_version,
            app_label,
            leaf_identity,
        })
    }
}

/// Load the full ledger row for a given `version`. Returns `None` when
/// the row is absent (not a hard error — callers that distinguish
/// "missing" from "DB error" handle both arms). Surfaced as a
/// `pub(crate)` helper so runner, repair, and verify can share the
/// 15-column SELECT without duplicating the column list or the
/// try_get cascade.
pub(crate) async fn load_full_row_by_version(
    ctx: &mut DjogiContext,
    version: &str,
) -> Result<Option<LedgerRow>, DjogiError> {
    let sql = format!("{LEDGER_SELECT_COLS} WHERE version = $1");
    let row_opt = ctx.query_opt(&sql, &[&version]).await?;
    let Some(row) = row_opt else {
        return Ok(None);
    };
    let ledger_row = LedgerRow::try_from(&row)?;
    Ok(Some(ledger_row))
}

/// Constant emitting the `djogi_schema_migrations` DDL. Public so
/// integration tests and the T6 `init` command can replay it
/// without reaching into the runner.
pub const LEDGER_TABLE_DDL: &str = r#"
CREATE TABLE IF NOT EXISTS djogi_schema_migrations (
    id                    BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    version               TEXT        NOT NULL UNIQUE,
    description           TEXT        NOT NULL DEFAULT '',
    checksum_up           VARCHAR(68) NOT NULL,
    checksum_down         VARCHAR(68),
    execution_mode        TEXT        NOT NULL DEFAULT 'transactional'
                              CHECK (execution_mode IN ('transactional', 'non_transactional')),
    status                TEXT        NOT NULL DEFAULT 'pending'
                              CHECK (status IN (
                                  'pending', 'applied', 'baseline',
                                  'faked', 'rolled_back', 'failed'
                              )),
    applied_at            TIMESTAMPTZ NOT NULL DEFAULT now(),
    applied_by            TEXT        NOT NULL DEFAULT current_user,
    execution_time_ms     BIGINT      NOT NULL DEFAULT 0,
    out_of_order_flag     BOOLEAN     NOT NULL DEFAULT FALSE,
    applied_steps_count   INTEGER     NOT NULL DEFAULT 0,
    total_steps           INTEGER,
    partial_apply_note    TEXT,
    run_id                BIGINT      NOT NULL,
    snapshot_version      TEXT        NOT NULL,
    app_label             TEXT        NOT NULL DEFAULT '',
    leaf_identity         TEXT
);
"#;

/// SHA-256 hex digest length (64 lowercase hex chars).
pub const SHA256_HEX_LEN: usize = 64;

/// Versioned-checksum prefix. Bumping past `V1:` would require the
/// runner to dual-verify both forms during the transition window.
pub const CHECKSUM_PREFIX: &str = "V1:";

/// Total byte length of a `V1:<sha256-hex>` checksum string. Equals
/// `CHECKSUM_PREFIX.len() + SHA256_HEX_LEN`.
pub const CHECKSUM_LEN: usize = 3 + SHA256_HEX_LEN;

/// Prefix for a structured `partial_apply_note` that records an
/// in-flight non-transactional step claim. The claim exists so a
/// crash or post-DDL ledger failure cannot silently re-run the next
/// step: repair resume detects the prefix and refuses automatic
/// replay until an operator reconciles the ambiguous boundary.
pub(crate) const NON_TX_PROGRESS_CLAIM_PREFIX: &str = "non-tx progress claim:";

/// Lifecycle status enum — mirrors the database CHECK constraint.
///
/// Ordering of variants matches the database CHECK list; do not
/// reshuffle without a migration. The `Display` impl writes the
/// string used by the database (lowercase, underscore-separated),
/// and [`LedgerStatus::from_db_str`] is the inverse.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LedgerStatus {
    /// Row inserted before SQL ran; no segment has executed yet.
    Pending,
    /// Every segment ran successfully; snapshot was persisted.
    Applied,
    /// Migration represents a baseline of an existing database. Used
    /// by `djogi migrations baseline` (T8).
    Baseline,
    /// Migration was marked applied without running its SQL — used
    /// when an out-of-band tool already applied the change.
    Faked,
    /// Migration was rolled back via `djogi migrations down` (T8).
    RolledBack,
    /// Apply failed; no further work attempted. The runner records
    /// `partial_apply_note` for split applies that crashed mid-stream.
    Failed,
}

impl LedgerStatus {
    pub const fn as_db_str(self) -> &'static str {
        match self {
            LedgerStatus::Pending => "pending",
            LedgerStatus::Applied => "applied",
            LedgerStatus::Baseline => "baseline",
            LedgerStatus::Faked => "faked",
            LedgerStatus::RolledBack => "rolled_back",
            LedgerStatus::Failed => "failed",
        }
    }

    /// Inverse of [`LedgerStatus::as_db_str`]. Returns `None` for
    /// strings that do not match any known status — the caller
    /// surfaces this as a database corruption indicator.
    pub fn from_db_str(s: &str) -> Option<Self> {
        Some(match s {
            "pending" => LedgerStatus::Pending,
            "applied" => LedgerStatus::Applied,
            "baseline" => LedgerStatus::Baseline,
            "faked" => LedgerStatus::Faked,
            "rolled_back" => LedgerStatus::RolledBack,
            "failed" => LedgerStatus::Failed,
            _ => return None,
        })
    }
}

/// Execution-mode enum — mirrors the database CHECK constraint.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExecutionMode {
    /// All SQL ran inside a single Postgres transaction.
    Transactional,
    /// At least one segment ran outside any transaction
    /// (e.g. `CREATE INDEX CONCURRENTLY`). The runner records the
    /// row as `non_transactional` even when only one segment was
    /// non-transactional — the lifecycle is "this row may be in a
    /// partial-apply state if the process crashed mid-segment".
    NonTransactional,
}

impl ExecutionMode {
    pub const fn as_db_str(self) -> &'static str {
        match self {
            ExecutionMode::Transactional => "transactional",
            ExecutionMode::NonTransactional => "non_transactional",
        }
    }
}

/// Owned shape of a `djogi_schema_migrations` row. Used by the runner
/// to assemble a row before insertion and by the row-CRUD helpers
/// when reading back. Public so T5 (`repair`) and T8 (`status`) can
/// share the type.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LedgerRow {
    /// Unique migration version label (e.g. `V20260425010203__add_users`).
    pub version: String,
    /// Operator-facing one-line description.
    pub description: String,
    /// `V1:<sha256-hex>` of the concatenated `up` SQL.
    pub checksum_up: String,
    /// `V1:<sha256-hex>` of the concatenated `down` SQL, or `None`
    /// when the operation was wholly lossy and `down` is just a
    /// SQL-comment placeholder.
    pub checksum_down: Option<String>,
    /// Whether this migration ran in / required a transaction.
    pub execution_mode: ExecutionMode,
    /// Lifecycle status — see [`LedgerStatus`].
    pub status: LedgerStatus,
    /// Wall-clock time spent applying, in milliseconds. `0` for
    /// `pending` / `failed` rows that never completed.
    pub execution_time_ms: i64,
    /// `true` when the migration applied later than its sort order
    /// would suggest (e.g. a new dev branch picks up a peer's
    /// migration after the dev's own newer one). T7 sets this; T4
    /// always writes `false`.
    pub out_of_order_flag: bool,
    /// For split-apply migrations: how many non-transactional steps
    /// have completed. `0` for transactional-only migrations.
    pub applied_steps_count: i32,
    /// Total non-transactional step count. `None` for transactional-
    /// only migrations.
    pub total_steps: Option<i32>,
    /// Human-readable note describing a partial-apply state — set
    /// when the runner's split-segment dispatch crashed mid-stream.
    pub partial_apply_note: Option<String>,
    /// Per-runner-invocation ID. Generated via
    /// [`HeerId::generate_id`](crate::types::HeerId)-equivalent in
    /// the runner (the actual generation lives in
    /// [`crate::migrate::runner`]). Stored as `BIGINT`.
    pub run_id: i64,
    /// Value of [`crate::migrate::schema::SNAPSHOT_FORMAT_VERSION`]
    /// at apply time — anchors the migration to the exact snapshot
    /// shape it consumed.
    pub snapshot_version: String,
    /// App label this migration belongs to. Empty string for the
    /// synthetic global bucket.
    pub app_label: String,
    /// Partition leaf identity snapshot. Stored as newline-delimited
    /// `parent:leaf1,leaf2` entries. `None` means no partition-expanded
    /// segments were materialized (non-partitioned migration, or a row
    /// seeded via fake-apply / baseline / attune). A partitioned migration
    /// that applied when the parent had zero attached leaves stores
    /// `Some("parent:")` — the parent key is present but the leaf list
    /// is empty.
    pub leaf_identity: Option<String>,
}

/// Build the machine-detectable note written immediately before a
/// non-transactional step runs. `step_ordinal` is 1-based across the
/// whole migration (not just within the segment).
pub(crate) fn format_non_tx_progress_claim(
    prior_note: Option<&str>,
    step_ordinal: i32,
    total_steps: Option<i32>,
    segment_index: usize,
    statement_label: &str,
) -> String {
    let mut note = match total_steps {
        Some(total) => format!(
            "{NON_TX_PROGRESS_CLAIM_PREFIX} step {step_ordinal} of {total} \
             `{statement_label}` in segment {} is/was in flight; automatic resume \
             is blocked until an operator reconciles whether that step committed",
            segment_index + 1,
        ),
        None => format!(
            "{NON_TX_PROGRESS_CLAIM_PREFIX} step {step_ordinal} `{statement_label}` \
             in segment {} is/was in flight; automatic resume is blocked until an \
             operator reconciles whether that step committed",
            segment_index + 1,
        ),
    };
    if let Some(prior) = prior_note.filter(|note| !note.is_empty()) {
        note.push_str("\nprior note: ");
        note.push_str(prior);
    }
    note
}

/// True when `partial_apply_note` currently carries an outstanding
/// non-transactional progress claim.
pub(crate) fn note_has_non_tx_progress_claim(note: Option<&str>) -> bool {
    note.is_some_and(|value| value.starts_with(NON_TX_PROGRESS_CLAIM_PREFIX))
}

/// Compute a `V1:<sha256-hex>` checksum over a sequence of SQL
/// fragments concatenated with `\n` separators.
///
/// Order matters — passing the same fragments in a different order
/// produces a different hash. The runner concatenates each segment's
/// `up` (or `down`) SQL in segment order before hashing.
///
/// Returns a 67-byte string consisting of the literal `V1:` followed
/// by 64 lowercase hex characters.
pub fn compute_checksum<I, S>(fragments: I) -> String
where
    I: IntoIterator<Item = S>,
    S: AsRef<str>,
{
    let mut hasher = Sha256::new();
    let mut first = true;
    for frag in fragments {
        if !first {
            hasher.update(b"\n");
        }
        first = false;
        hasher.update(frag.as_ref().as_bytes());
    }
    let digest = hasher.finalize();
    let mut out = String::with_capacity(CHECKSUM_LEN);
    out.push_str(CHECKSUM_PREFIX);
    for b in digest {
        // Manually format as 2-digit lowercase hex. No regex, no
        // dependency on the `hex` crate; this is a tight inner loop.
        out.push(hex_digit(b >> 4));
        out.push(hex_digit(b & 0x0f));
    }
    debug_assert_eq!(out.len(), CHECKSUM_LEN);
    out
}

/// Map a 4-bit nibble to its lowercase hex character.
fn hex_digit(n: u8) -> char {
    match n {
        0..=9 => (b'0' + n) as char,
        10..=15 => (b'a' + (n - 10)) as char,
        // Caller passes a nibble (`b >> 4` or `b & 0x0f`); this is
        // unreachable at runtime but the explicit panic gives any
        // future refactor a clean failure mode rather than silently
        // emitting `\0`.
        _ => unreachable!("hex_digit takes a 4-bit nibble"),
    }
}

/// Reason a checksum string failed [`validate_checksum_format`].
///
/// Surfaces the precise rule violated so the operator can correct the
/// stored value (typically by re-running `compose` against the
/// canonical SQL fragments).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChecksumFormatErrorKind {
    /// String did not start with the literal [`CHECKSUM_PREFIX`]
    /// (`V1:`). Bumping the prefix to `V2:` would be a breaking
    /// change; until then any other prefix is malformed.
    WrongPrefix,
    /// Total length differed from [`CHECKSUM_LEN`] (67 bytes). A
    /// well-formed checksum is `V1:` followed by exactly 64 hex
    /// characters.
    WrongLength { observed: usize },
    /// One of the 64 hex bytes was not a lowercase ASCII hex
    /// character (`0`..=`9` or `a`..=`f`). Uppercase hex is also
    /// rejected — the canonical encoding is lowercase to match
    /// `compute_checksum`'s output exactly so byte-comparisons are
    /// total.
    NonLowercaseHex { offset: usize, byte: u8 },
}

impl std::fmt::Display for ChecksumFormatErrorKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::WrongPrefix => write!(
                f,
                "checksum string does not start with the `{CHECKSUM_PREFIX}` prefix",
            ),
            Self::WrongLength { observed } => write!(
                f,
                "checksum string length is {observed} bytes; expected {CHECKSUM_LEN}",
            ),
            Self::NonLowercaseHex { offset, byte } => write!(
                f,
                "checksum hex byte at offset {offset} is 0x{byte:02x}; expected lowercase ASCII hex",
            ),
        }
    }
}

/// A checksum string failed [`validate_checksum_format`]. The
/// `side` field indicates which argument to [`verify_checksum`] was
/// malformed so operators can pinpoint the bad value (a hand-edited
/// ledger row vs. a runner that emitted a wrong-shape string).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChecksumFormatError {
    /// Which side of the verification carried the malformed string.
    pub side: ChecksumSide,
    /// The offending checksum string.
    pub value: String,
    /// The exact rule violated.
    pub kind: ChecksumFormatErrorKind,
}

/// Identifies which checksum argument failed format validation —
/// the stored `expected` value (typically from the ledger row) or
/// the freshly-computed `actual` value (from the runner).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChecksumSide {
    /// The persisted / supplied `expected` checksum (typically
    /// loaded from `djogi_schema_migrations.checksum_up`).
    Expected,
    /// The freshly-computed `actual` checksum produced by the
    /// runner from the in-memory plan.
    Actual,
}

impl std::fmt::Display for ChecksumSide {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Expected => f.write_str("expected"),
            Self::Actual => f.write_str("actual"),
        }
    }
}

impl std::fmt::Display for ChecksumFormatError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{side} checksum `{value}` is malformed: {kind}",
            side = self.side,
            value = self.value,
            kind = self.kind,
        )
    }
}

impl std::error::Error for ChecksumFormatError {}

/// Structurally validate a checksum string against the
/// `V1:<sha256-hex>` shape.
///
/// Rules:
///
/// 1. Starts with the literal [`CHECKSUM_PREFIX`] (`V1:`). No other
///    prefix is accepted; uppercase / lowercase variants are rejected.
/// 2. Total byte length equals [`CHECKSUM_LEN`] (67).
/// 3. The 64 bytes after the prefix are all lowercase ASCII hex —
///    a digit `0`..=`9` or a lowercase letter `a`..=`f`. Uppercase
///    hex is rejected because [`compute_checksum`] always emits
///    lowercase, so an uppercase value cannot be a runner-produced
///    checksum.
///
/// Used by [`verify_checksum`] on BOTH operands before string
/// comparison so a malformed persisted value cannot accidentally
/// pass a `==` check against another malformed value.
///
/// Implementation note: byte-level rules per the Djogi-wide
/// no-regex policy. No allocation; constant-stack only.
pub fn validate_checksum_format(s: &str) -> Result<(), ChecksumFormatErrorKind> {
    let bytes = s.as_bytes();
    if !s.starts_with(CHECKSUM_PREFIX) {
        return Err(ChecksumFormatErrorKind::WrongPrefix);
    }
    if bytes.len() != CHECKSUM_LEN {
        return Err(ChecksumFormatErrorKind::WrongLength {
            observed: bytes.len(),
        });
    }
    // Walk the 64-byte hex tail. Lowercase hex only — `0..=9`
    // (0x30..=0x39) or `a..=f` (0x61..=0x66).
    let prefix_len = CHECKSUM_PREFIX.len();
    for (i, &b) in bytes[prefix_len..].iter().enumerate() {
        let is_lower_hex = b.is_ascii_digit() || (b'a'..=b'f').contains(&b);
        if !is_lower_hex {
            return Err(ChecksumFormatErrorKind::NonLowercaseHex {
                offset: prefix_len + i,
                byte: b,
            });
        }
    }
    Ok(())
}

/// Verify that a stored checksum matches a freshly-computed one.
/// Returns `Ok(())` on match; on mismatch returns the structured
/// error variant the runner surfaces verbatim.
///
/// **Format validation runs first.** Both operands are passed
/// through [`validate_checksum_format`] BEFORE the string compare so
/// a malformed-but-equal pair (e.g. two uppercase-hex strings) cannot
/// accidentally verify clean. The format-failure path returns
/// [`VerifyError::Format`] identifying which side was bad; only when
/// both sides are well-formed does the function fall through to the
/// hash-comparison path which yields [`VerifyError::Mismatch`] on
/// disagreement.
///
/// Distinct from a plain `==` so the call site reads as a verification
/// step rather than a generic equality check.
pub fn verify_checksum(version: &str, expected: &str, actual: &str) -> Result<(), VerifyError> {
    if let Err(kind) = validate_checksum_format(expected) {
        return Err(VerifyError::Format(ChecksumFormatError {
            side: ChecksumSide::Expected,
            value: expected.to_string(),
            kind,
        }));
    }
    if let Err(kind) = validate_checksum_format(actual) {
        return Err(VerifyError::Format(ChecksumFormatError {
            side: ChecksumSide::Actual,
            value: actual.to_string(),
            kind,
        }));
    }
    if expected == actual {
        Ok(())
    } else {
        Err(VerifyError::Mismatch(ChecksumMismatch {
            version: version.to_string(),
            expected: expected.to_string(),
            actual: actual.to_string(),
        }))
    }
}

/// Failure modes of [`verify_checksum`]. Either a malformed input
/// (rejected before comparison) or a content mismatch between two
/// well-formed checksums.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum VerifyError {
    /// One of the two checksum strings did not pass
    /// [`validate_checksum_format`]. The wrapped error carries the
    /// offending side and the precise rule violated.
    Format(ChecksumFormatError),
    /// Both inputs were well-formed but the values differed. The
    /// runner surfaces this verbatim as
    /// `RunnerError::ChecksumMismatch`.
    Mismatch(ChecksumMismatch),
}

impl std::fmt::Display for VerifyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            VerifyError::Format(e) => write!(f, "{e}"),
            VerifyError::Mismatch(e) => write!(f, "{e}"),
        }
    }
}

impl std::error::Error for VerifyError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            VerifyError::Format(e) => Some(e),
            VerifyError::Mismatch(e) => Some(e),
        }
    }
}

/// Structured payload for a checksum-verification failure. The runner
/// wraps this in `RunnerError::ChecksumMismatch`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChecksumMismatch {
    pub version: String,
    pub expected: String,
    pub actual: String,
}

impl std::fmt::Display for ChecksumMismatch {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "checksum mismatch on version `{ver}`: expected {exp}, computed {act}",
            ver = self.version,
            exp = self.expected,
            act = self.actual,
        )
    }
}

impl std::error::Error for ChecksumMismatch {}

/// Idempotently create the `djogi_schema_migrations` table. Safe to
/// call on every runner invocation — the `IF NOT EXISTS` clause makes
/// repeated calls a no-op.
///
/// Routes through `DjogiContext::raw_ddl` which uses Postgres's simple
/// query protocol (DDL statements that the prepare-cache cannot
/// handle).
pub async fn bootstrap(ctx: &mut DjogiContext) -> Result<(), DjogiError> {
    ctx.raw_ddl(LEDGER_TABLE_DDL).await?;
    ctx.raw_ddl("ALTER TABLE djogi_schema_migrations ADD COLUMN IF NOT EXISTS leaf_identity TEXT")
        .await?;
    Ok(())
}

/// Insert a `pending` ledger row before SQL runs. Returns the row's
/// `id` (BIGINT IDENTITY).
///
/// The row is intentionally written through whatever context the
/// runner provides — the runner inserts the pending row OUTSIDE the
/// migration's transaction so a rollback of the migration's tx does
/// not also wipe the pending-row record. (For non-transactional
/// migrations the row update path is even more important — the
/// pending row anchors crash recovery.)
pub async fn insert_pending(ctx: &mut DjogiContext, row: &LedgerRow) -> Result<i64, DjogiError> {
    let sql = "INSERT INTO djogi_schema_migrations \
               (version, description, checksum_up, checksum_down, execution_mode, \
                status, execution_time_ms, out_of_order_flag, applied_steps_count, \
                total_steps, partial_apply_note, run_id, snapshot_version, app_label, leaf_identity) \
               VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15) \
               RETURNING id";
    let exec_mode = row.execution_mode.as_db_str();
    let status = row.status.as_db_str();
    let row_db = ctx
        .query_one(
            sql,
            &[
                &row.version,
                &row.description,
                &row.checksum_up,
                &row.checksum_down,
                &exec_mode,
                &status,
                &row.execution_time_ms,
                &row.out_of_order_flag,
                &row.applied_steps_count,
                &row.total_steps,
                &row.partial_apply_note,
                &row.run_id,
                &row.snapshot_version,
                &row.app_label,
                &row.leaf_identity,
            ],
        )
        .await?;
    Ok(row_db.try_get::<_, i64>("id")?)
}

/// Mark a previously-inserted ledger row as `applied`. Updates
/// `status`, `execution_time_ms`, `applied_steps_count`,
/// `applied_at`, and clears any prior `partial_apply_note`.
pub async fn mark_applied(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    execution_time_ms: i64,
    applied_steps_count: i32,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET status = 'applied', \
                   execution_time_ms = $2, \
                   applied_steps_count = $3, \
                   applied_at = now(), \
                   partial_apply_note = NULL \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &execution_time_ms, &applied_steps_count])
        .await?;
    Ok(())
}

/// Mark a ledger row as `applied` while PRESERVING the
/// `partial_apply_note`. T7's out-of-order detection sets the note
/// at insert time (the conflicting peer + optional override reason);
/// preserving it on success keeps the audit trail honest about why
/// the row carries `out_of_order_flag = TRUE` even after it reached
/// the final `applied` state.
///
/// Call this in place of [`mark_applied`] whenever the runner's apply
/// path inserted a non-empty initial note — the note describes a
/// historical condition (out-of-order arrival) that does not stop
/// being true once the migration applies cleanly.
pub async fn mark_applied_keep_note(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    execution_time_ms: i64,
    applied_steps_count: i32,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET status = 'applied', \
                   execution_time_ms = $2, \
                   applied_steps_count = $3, \
                   applied_at = now() \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &execution_time_ms, &applied_steps_count])
        .await?;
    Ok(())
}

/// Mark a previously-inserted ledger row as `failed` and attach a
/// human-readable note. Used when an entirely transactional apply
/// failed before any non-tx work began.
pub async fn mark_failed(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    note: &str,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET status = 'failed', \
                   partial_apply_note = $2 \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &note]).await?;
    Ok(())
}

/// Record the next non-transactional step as claimed but not yet
/// durably acknowledged. The existing `applied_steps_count` remains
/// untouched until the post-DDL ack succeeds.
pub async fn claim_non_tx_progress(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    note: &str,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET partial_apply_note = $2 \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &note]).await?;
    Ok(())
}

/// Update a ledger row mid-apply with the latest non-transactional
/// step count. Used after each non-tx step completes — the runner
/// can then resume from `applied_steps_count` if the process crashes
/// before the next step.
pub async fn update_progress(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    applied_steps_count: i32,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET applied_steps_count = $2 \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &applied_steps_count])
        .await?;
    Ok(())
}

/// Acknowledge a previously claimed non-transactional step by
/// advancing `applied_steps_count` and restoring the durable
/// background note (or clearing the claim entirely).
pub async fn ack_non_tx_progress(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    applied_steps_count: i32,
    restored_note: Option<&str>,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET applied_steps_count = $2, \
                   partial_apply_note = $3 \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &applied_steps_count, &restored_note])
        .await?;
    Ok(())
}

/// Mark a ledger row as `failed` with partial-apply context — used
/// when a non-transactional step crashed mid-stream. The note string
/// captures the failing step ordinal and the underlying error.
pub async fn mark_partial(
    ctx: &mut DjogiContext,
    ledger_id: i64,
    applied_steps_count: i32,
    note: &str,
) -> Result<(), DjogiError> {
    let sql = "UPDATE djogi_schema_migrations \
               SET status = 'failed', \
                   applied_steps_count = $2, \
                   partial_apply_note = $3 \
               WHERE id = $1";
    ctx.execute(sql, &[&ledger_id, &applied_steps_count, &note])
        .await?;
    Ok(())
}

/// Read-only projection of a ledger row used by `migrations status`.
///
/// Carries enough fields for the operator-facing list output without
/// over-fetching (the full [`LedgerRow`] is heavier and includes
/// fields `status` does not surface). T6 owns this projection because
/// it is shaped specifically for the status command's grouped /
/// time-sorted output.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LedgerSummaryRow {
    /// Ledger row primary key.
    pub id: i64,
    /// `V<ts>__<slug>` version key.
    pub version: String,
    /// Operator-facing description.
    pub description: String,
    /// Lifecycle status — see [`LedgerStatus`].
    pub status: LedgerStatus,
    /// Apply wall-clock time in milliseconds.
    pub execution_time_ms: i64,
    /// `applied_at` rendered as RFC 3339 UTC second-precision so the
    /// command output is deterministic across locales / time zones.
    pub applied_at_rfc3339: String,
    /// Operator (`applied_by`) recorded by the runner.
    pub applied_by: String,
    /// Per-runner-invocation ID — surfaced truncated by the status
    /// command, full value here so consumers can join with logs.
    pub run_id: i64,
    /// `partial_apply_note` if the migration ended in a partial
    /// state.
    pub partial_apply_note: Option<String>,
    /// `app_label` recorded on the row. Empty string for the
    /// synthetic global bucket.
    pub app_label: String,
    /// `true` when this row applied out-of-order — its `version`
    /// sorts before some peer that was already applied at the time.
    /// Surfaced by the status command via the `[ooo]` line marker.
    pub out_of_order_flag: bool,
}

/// Read every ledger row from the active database, ordered by
/// `app_label` ASC then `applied_at` ASC.
///
/// Read-only — does not bootstrap the table. Callers expecting an
/// uninitialised database should bootstrap first via [`bootstrap`].
/// The status command intentionally skips bootstrap so it can run
/// against a database that has never seen a Djogi migration and emit
/// an empty list rather than create infrastructure.
pub async fn select_all(ctx: &mut DjogiContext) -> Result<Vec<LedgerSummaryRow>, DjogiError> {
    // The `to_char` formatter renders `applied_at` deterministically
    // in UTC second-precision. Postgres `to_char` does not understand
    // timezone-zulu shorthand; we format the components explicitly.
    let sql = "SELECT \
               id, version, description, status, execution_time_ms, \
               to_char(applied_at AT TIME ZONE 'UTC', \
                       'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS applied_at_rfc3339, \
               applied_by, run_id, partial_apply_note, app_label, \
               out_of_order_flag \
               FROM djogi_schema_migrations \
               ORDER BY app_label ASC, applied_at ASC, id ASC";
    let rows = ctx.query_all(sql, &[]).await?;
    let mut out = Vec::with_capacity(rows.len());
    for row in rows {
        let status_str: String = row.try_get("status")?;
        let status = LedgerStatus::from_db_str(&status_str).ok_or_else(|| {
            DjogiError::Db(DbError::other(format!(
                "unknown LedgerStatus value in djogi_schema_migrations: {status_str:?}"
            )))
        })?;
        out.push(LedgerSummaryRow {
            id: row.try_get("id")?,
            version: row.try_get("version")?,
            description: row.try_get("description")?,
            status,
            execution_time_ms: row.try_get("execution_time_ms")?,
            applied_at_rfc3339: row.try_get("applied_at_rfc3339")?,
            applied_by: row.try_get("applied_by")?,
            run_id: row.try_get("run_id")?,
            partial_apply_note: row.try_get("partial_apply_note")?,
            app_label: row.try_get("app_label")?,
            out_of_order_flag: row.try_get("out_of_order_flag")?,
        });
    }
    Ok(out)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn checksum_has_v1_prefix_and_64_hex() {
        let csum = compute_checksum(["CREATE TABLE foo ();"]);
        assert!(csum.starts_with(CHECKSUM_PREFIX));
        assert_eq!(csum.len(), CHECKSUM_LEN);
        let hex = &csum[CHECKSUM_PREFIX.len()..];
        assert_eq!(hex.len(), SHA256_HEX_LEN);
        assert!(hex.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')));
    }

    #[test]
    fn empty_input_still_hashes_deterministically() {
        let csum = compute_checksum(std::iter::empty::<&str>());
        // SHA-256 of empty input is the well-known
        // e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
        assert_eq!(
            csum,
            "V1:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
    }

    #[test]
    fn checksum_is_deterministic_across_calls() {
        let a = compute_checksum(["a", "b", "c"]);
        let b = compute_checksum(["a", "b", "c"]);
        assert_eq!(a, b);
    }

    #[test]
    fn checksum_changes_on_fragment_reorder() {
        let a = compute_checksum(["a", "b"]);
        let b = compute_checksum(["b", "a"]);
        assert_ne!(a, b, "order must be load-bearing");
    }

    #[test]
    fn checksum_separates_fragments_with_newline() {
        // Direct hashing of "a\nb" must equal compute_checksum(["a", "b"]).
        let direct = {
            let mut h = Sha256::new();
            h.update(b"a\nb");
            let d = h.finalize();
            let mut s = String::from(CHECKSUM_PREFIX);
            for b in d {
                s.push(hex_digit(b >> 4));
                s.push(hex_digit(b & 0x0f));
            }
            s
        };
        assert_eq!(compute_checksum(["a", "b"]), direct);
    }

    #[test]
    fn verify_checksum_ok_on_match() {
        let csum = compute_checksum(["x"]);
        assert!(verify_checksum("V_test", &csum, &csum).is_ok());
    }

    #[test]
    fn verify_checksum_err_on_mismatch_carries_payload() {
        let a = compute_checksum(["x"]);
        let b = compute_checksum(["y"]);
        let err = verify_checksum("V_test", &a, &b).unwrap_err();
        match err {
            VerifyError::Mismatch(m) => {
                assert_eq!(m.version, "V_test");
                assert_eq!(m.expected, a);
                assert_eq!(m.actual, b);
                let msg = m.to_string();
                assert!(msg.contains("V_test"));
                assert!(msg.contains("checksum mismatch"));
            }
            other => panic!("expected Mismatch, got {other:?}"),
        }
    }

    // ── validate_checksum_format ─────────────────────────────────────────

    #[test]
    fn validate_format_accepts_canonical_compute_checksum_output() {
        let csum = compute_checksum(["foo"]);
        validate_checksum_format(&csum).expect("canonical output is valid");
    }

    #[test]
    fn validate_format_rejects_uppercase_v_prefix() {
        // The canonical prefix is exactly `V1:` (capital V, digit 1,
        // colon). An uppercase-V is fine; `v1:` lowercase must reject.
        let bad = "v1:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        assert_eq!(bad.len(), CHECKSUM_LEN);
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(err, ChecksumFormatErrorKind::WrongPrefix));
    }

    #[test]
    fn validate_format_rejects_v2_prefix() {
        let bad = "V2:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        assert_eq!(bad.len(), CHECKSUM_LEN);
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(err, ChecksumFormatErrorKind::WrongPrefix));
    }

    #[test]
    fn validate_format_rejects_no_prefix() {
        let bad = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(err, ChecksumFormatErrorKind::WrongPrefix));
    }

    #[test]
    fn validate_format_rejects_short_length() {
        let bad = "V1:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b85"; // 66 bytes
        assert_eq!(bad.len(), CHECKSUM_LEN - 1);
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(
            err,
            ChecksumFormatErrorKind::WrongLength { observed: 66 }
        ));
    }

    #[test]
    fn validate_format_rejects_long_length() {
        let bad = "V1:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b8550"; // 68 bytes
        assert_eq!(bad.len(), CHECKSUM_LEN + 1);
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(
            err,
            ChecksumFormatErrorKind::WrongLength { observed: 68 }
        ));
    }

    #[test]
    fn validate_format_rejects_uppercase_hex() {
        // Same length, valid prefix, but uppercase A in the hex tail.
        let bad = "V1:ABC0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        assert_eq!(bad.len(), CHECKSUM_LEN);
        let err = validate_checksum_format(bad).unwrap_err();
        match err {
            ChecksumFormatErrorKind::NonLowercaseHex { offset, byte } => {
                assert_eq!(offset, 3); // first byte after `V1:` prefix
                assert_eq!(byte, b'A');
            }
            other => panic!("expected NonLowercaseHex, got {other:?}"),
        }
    }

    #[test]
    fn validate_format_rejects_non_hex_letters() {
        // `z` is past `f` — must reject.
        let bad = "V1:zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz";
        assert_eq!(bad.len(), CHECKSUM_LEN);
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(
            err,
            ChecksumFormatErrorKind::NonLowercaseHex {
                offset: 3,
                byte: b'z'
            }
        ));
    }

    #[test]
    fn validate_format_rejects_uppercase_g() {
        // `G` is also past `f` and uppercase — must reject.
        let bad = "V1:GGG0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        assert_eq!(bad.len(), CHECKSUM_LEN);
        let err = validate_checksum_format(bad).unwrap_err();
        assert!(matches!(
            err,
            ChecksumFormatErrorKind::NonLowercaseHex {
                offset: 3,
                byte: b'G'
            }
        ));
    }

    #[test]
    fn verify_checksum_rejects_malformed_expected() {
        let actual = compute_checksum(["x"]);
        let err = verify_checksum("V_test", "V1:NOT_HEX", &actual).unwrap_err();
        match err {
            VerifyError::Format(f) => {
                assert_eq!(f.side, ChecksumSide::Expected);
            }
            other => panic!("expected Format(Expected), got {other:?}"),
        }
    }

    #[test]
    fn verify_checksum_rejects_malformed_actual() {
        // Expected is well-formed, actual is malformed (uppercase hex).
        let expected = compute_checksum(["x"]);
        let bad_actual = "V1:ABC0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        let err = verify_checksum("V_test", &expected, bad_actual).unwrap_err();
        match err {
            VerifyError::Format(f) => {
                assert_eq!(f.side, ChecksumSide::Actual);
            }
            other => panic!("expected Format(Actual), got {other:?}"),
        }
    }

    #[test]
    fn verify_checksum_runs_format_check_before_equality() {
        // Two malformed-but-equal strings would pass a naive `==` check.
        // The format gate must reject them first.
        let bad = "V1:ABC0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
        let err = verify_checksum("V_test", bad, bad).unwrap_err();
        assert!(matches!(err, VerifyError::Format(_)));
    }

    #[test]
    fn ledger_status_round_trips_through_db_str() {
        for s in [
            LedgerStatus::Pending,
            LedgerStatus::Applied,
            LedgerStatus::Baseline,
            LedgerStatus::Faked,
            LedgerStatus::RolledBack,
            LedgerStatus::Failed,
        ] {
            assert_eq!(LedgerStatus::from_db_str(s.as_db_str()), Some(s));
        }
    }

    #[test]
    fn ledger_status_unknown_returns_none() {
        assert_eq!(LedgerStatus::from_db_str("out_of_order"), None);
        assert_eq!(LedgerStatus::from_db_str(""), None);
    }

    #[test]
    fn execution_mode_db_strings_are_stable() {
        assert_eq!(ExecutionMode::Transactional.as_db_str(), "transactional");
        assert_eq!(
            ExecutionMode::NonTransactional.as_db_str(),
            "non_transactional"
        );
    }

    #[test]
    fn checksum_constants_are_consistent() {
        assert_eq!(CHECKSUM_LEN, CHECKSUM_PREFIX.len() + SHA256_HEX_LEN);
        assert_eq!(SHA256_HEX_LEN, 64);
    }

    #[test]
    fn non_tx_progress_claim_note_round_trips_through_prefix_detector() {
        let note = format_non_tx_progress_claim(
            Some("out-of-order apply: peer V20260524__older"),
            2,
            Some(3),
            1,
            "AddIndex widgets_name_idx",
        );
        assert!(note_has_non_tx_progress_claim(Some(&note)));
        assert!(note.contains("step 2 of 3"));
        assert!(note.contains("AddIndex widgets_name_idx"));
        assert!(note.contains("prior note: out-of-order apply"));
    }

    #[test]
    fn non_tx_progress_claim_detector_ignores_regular_partial_notes() {
        assert!(!note_has_non_tx_progress_claim(None));
        assert!(!note_has_non_tx_progress_claim(Some(
            "non-tx step 2 of segment 1 failed: AddIndex widgets_name_idx"
        )));
    }
}