pgmt 0.6.4

use crate::config::types::TrackingTable;
use crate::migration::section_parser::MigrationSection;
use crate::migration_tracking::{calculate_checksum, version_from_db, version_to_db};
use anyhow::{Context, Result};
use sqlx::PgPool;
use std::collections::BTreeMap;
use std::str::FromStr;

/// pgmt version stamped onto section rows at registration.
const PGMT_VERSION: &str = env!("CARGO_PKG_VERSION");

/// Format the sections tracking table name with proper schema qualification
pub(crate) fn format_sections_table_name(tracking_table: &TrackingTable) -> String {
    format!(
        r#""{}"."{}_sections""#,
        tracking_table.schema, tracking_table.name
    )
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SectionStatus {
    Pending,
    Running,
    Completed,
    /// A remap section covered by an established source at provision/adoption
    /// time: the objects are already present under the
    /// source's name, so nothing ran here — but the section is accounted for.
    /// `completed` keeps its invariant ("this DDL committed here") and absence
    /// keeps its ("crashed or never requested"); the incomplete-baseline guard
    /// treats completed|satisfied alike as covered.
    Satisfied,
    Failed,
}

impl SectionStatus {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Running => "running",
            Self::Completed => "completed",
            Self::Satisfied => "satisfied",
            Self::Failed => "failed",
        }
    }

    /// A terminal, covered state: the section's objects are present (whether it
    /// executed here or was covered by an established source). The
    /// incomplete-baseline guard and every "is this section done" check treat
    /// completed and satisfied identically.
    pub fn is_covered(&self) -> bool {
        matches!(self, Self::Completed | Self::Satisfied)
    }
}

impl FromStr for SectionStatus {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            "pending" => Ok(Self::Pending),
            "running" => Ok(Self::Running),
            "completed" => Ok(Self::Completed),
            "satisfied" => Ok(Self::Satisfied),
            "failed" => Ok(Self::Failed),
            _ => Err(anyhow::anyhow!("Unknown section status: {}", s)),
        }
    }
}

/// Ensure the section tracking table exists (and migrate older tables to the
/// current shape).
///
/// Sections are keyed by `(migration_version, is_baseline, section_name)`:
/// baselines execute through the same section machinery as migrations, and a
/// version can host both (a baseline generated alongside a migration), so
/// their section rows must not collide.
pub async fn ensure_section_tracking_table(
    pool: &PgPool,
    tracking_table: &TrackingTable,
) -> Result<()> {
    let sections_table = format_sections_table_name(tracking_table);

    // Detect whether this call INTRODUCES section tracking: probe for the table
    // before creating it. The advisory lock every production caller holds guards
    // concurrency (no crossing or evolve interleaves), but NOT crash durability
    // — so the introducing path creates the table and backfills the synthetic
    // legacy rows in ONE transaction. A crash can then never leave the table
    // present with the backfill lost, which would brick every legacy version at
    // the apply-time "tracking row but no section rows" bail. Only the
    // introducing path backfills; once the table exists, a zero-section main row
    // is a legitimate ledger state (a version-level event that processed no
    // local work) and must never be backfilled.
    let table_existed: bool = sqlx::query_scalar("SELECT to_regclass($1) IS NOT NULL")
        .bind(&sections_table)
        .fetch_one(pool)
        .await
        .context("Failed to probe for the section tracking table")?;

    if !table_existed {
        // Introducing path: CREATE TABLE + backfill, atomic. CREATE TABLE is
        // transactional in Postgres, so either both land or neither does.
        let mut tx = pool.begin().await?;
        sqlx::query(sqlx::AssertSqlSafe(format!(
            r#"
            CREATE TABLE IF NOT EXISTS {} (
                migration_version BIGINT NOT NULL,
                is_baseline BOOLEAN NOT NULL,
                section_name TEXT NOT NULL,
                section_order INT NOT NULL,
                status TEXT NOT NULL,
                started_at TIMESTAMP WITH TIME ZONE,
                completed_at TIMESTAMP WITH TIME ZONE,
                attempts INT DEFAULT 0,
                last_error TEXT,
                rows_affected BIGINT,
                duration_ms BIGINT,
                checksum TEXT,
                mode TEXT,
                module TEXT,
                applied_by TEXT,
                pgmt_version TEXT,
                PRIMARY KEY (migration_version, is_baseline, section_name)
            )
            "#,
            sections_table
        )))
        .execute(&mut *tx)
        .await
        .context("Failed to create section tracking table")?;

        backfill_synthetic_legacy_sections(&mut tx, tracking_table, &sections_table).await?;

        tx.commit()
            .await
            .context("Failed to introduce section tracking table")?;
    } else {
        // Already-exists path: bring a pre-existing (possibly legacy) table to
        // the current shape. Never backfills — a zero-section main row here is a
        // legitimate state, not a legacy row awaiting materialization.
        migrate_section_table_schema(pool, &sections_table).await?;
    }

    // The stored module subscription is part of the same evolve step:
    // the tables arrive here so every apply/provision path that ensures the
    // section table also has the subscription tables available. (The read-only
    // `migrate status` path never calls this — it probes instead.)
    crate::migration_tracking::TrackingStore::new(pool, tracking_table)?
        .ensure_subscription_tables()
        .await?;

    // Create index for querying by status
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "CREATE INDEX IF NOT EXISTS idx_{}_status ON {}(migration_version, status)",
        sections_table.replace("\"", "").replace(".", "_"),
        sections_table
    )))
    .execute(pool)
    .await
    .context("Failed to create section tracking table index")?;

    Ok(())
}

/// "Migrate the migrator" for the sections table: older tables lack
/// `is_baseline` (every existing row is a migration section — backfill FALSE,
/// no column DEFAULT) and key sections by `(migration_version, section_name)`.
/// Later revisions added per-section `checksum`/`mode`/`module`/`applied_by`/
/// `pgmt_version` (all nullable; NULL = legacy/pre-upgrade row).
async fn migrate_section_table_schema(pool: &PgPool, sections_table: &str) -> Result<()> {
    let existing_columns: Vec<String> = sqlx::query_scalar(
        "SELECT attname::text FROM pg_attribute
         WHERE attrelid = $1::regclass AND attnum > 0 AND NOT attisdropped",
    )
    .bind(sections_table)
    .fetch_all(pool)
    .await
    .with_context(|| format!("Failed to inspect columns of {}", sections_table))?;
    let has = |name: &str| existing_columns.iter().any(|c| c == name);

    // `is_baseline`: add nullable, backfill existing rows to FALSE (every
    // existing row is a migration), then enforce NOT NULL — in one transaction
    // so the column is never observed half-migrated.
    if !has("is_baseline") {
        let mut tx = pool.begin().await?;
        sqlx::query(sqlx::AssertSqlSafe(format!(
            "ALTER TABLE {} ADD COLUMN is_baseline BOOLEAN",
            sections_table
        )))
        .execute(&mut *tx)
        .await?;
        sqlx::query(sqlx::AssertSqlSafe(format!(
            "UPDATE {} SET is_baseline = FALSE",
            sections_table
        )))
        .execute(&mut *tx)
        .await?;
        sqlx::query(sqlx::AssertSqlSafe(format!(
            "ALTER TABLE {} ALTER COLUMN is_baseline SET NOT NULL",
            sections_table
        )))
        .execute(&mut *tx)
        .await?;
        tx.commit()
            .await
            .with_context(|| format!("Failed to add is_baseline to {}", sections_table))?;
    }

    // Per-section immutability + self-describing attribution columns. All
    // nullable: a NULL checksum marks a legacy/pre-upgrade row (passes
    // validation), and `module` NULL means the base (unmoduled) section.
    for column in ["checksum", "mode", "module", "applied_by", "pgmt_version"] {
        if !has(column) {
            sqlx::query(sqlx::AssertSqlSafe(format!(
                "ALTER TABLE {} ADD COLUMN {} TEXT",
                sections_table, column
            )))
            .execute(pool)
            .await
            .with_context(|| format!("Failed to add {} to {}", column, sections_table))?;
        }
    }

    crate::migration_tracking::ensure_primary_key(
        pool,
        sections_table,
        &["migration_version", "is_baseline", "section_name"],
    )
    .await?;

    Ok(())
}

/// Give every main-table row present when section tracking was introduced one
/// synthetic `default` completed section row.
///
/// A ONE-TIME upgrade action, run only by the evolve that *creates* the section
/// table — never a standing per-row rule. Before per-section tracking, "a main
/// row with no section rows" *meant* "fully applied by an older pgmt" — an
/// ABSENCE carrying meaning. Materializing the implicit `default` section at the
/// moment section tracking arrives removes that heuristic: from then on,
/// applied-ness is a derived function of present section rows. At that instant
/// every zero-section main row is legacy by definition, so all of them are
/// materialized.
///
/// After introduction this never runs again, and it must not: every writer now
/// registers its main row and section rows in ONE transaction, so a main row
/// with zero section rows is a LEGITIMATE state — a version-level event that
/// processed no local work (e.g. a crossing that consumed a re-anchor whose
/// remaps relabeled nothing this target holds). Backfilling a `default`
/// completed row onto such a row would forge content the target never applied.
///
/// Runs on the caller's transaction (the introducing evolve's), so the CREATE
/// TABLE and this backfill commit together — a crash can't leave the table
/// present with the backfill lost. Skipped when the main table is absent (the
/// section table can be created before it); the `NOT EXISTS` guard keeps the
/// insert harmless if invoked against a partially-populated table.
async fn backfill_synthetic_legacy_sections(
    conn: &mut sqlx::PgConnection,
    tracking_table: &TrackingTable,
    sections_table: &str,
) -> Result<()> {
    let main_table = crate::migration_tracking::format_tracking_table_name(tracking_table)?;

    let main_exists: Option<String> = sqlx::query_scalar("SELECT to_regclass($1)::text")
        .bind(&main_table)
        .fetch_one(&mut *conn)
        .await?;
    if main_exists.is_none() {
        return Ok(());
    }

    sqlx::query(sqlx::AssertSqlSafe(format!(
        "INSERT INTO {sections}
             (migration_version, is_baseline, section_name, section_order,
              status, completed_at, attempts)
         SELECT m.version, m.is_baseline, 'default', 0, 'completed', m.applied_at, 0
         FROM {main} m
         WHERE NOT EXISTS (
             SELECT 1 FROM {sections} s
             WHERE s.migration_version = m.version AND s.is_baseline = m.is_baseline)",
        sections = sections_table,
        main = main_table,
    )))
    .execute(&mut *conn)
    .await
    .with_context(|| {
        format!(
            "Failed to backfill synthetic legacy sections into {}",
            sections_table
        )
    })?;

    Ok(())
}

/// Insert a single Pending section row carrying its checksum/mode/module and
/// the recording pgmt version. `ON CONFLICT DO NOTHING` keeps registration
/// idempotent across resume / module-adoption re-registration. The column
/// list lives here so every registration path (migration, baseline,
/// initialize_sections) stores identical metadata.
pub(crate) async fn insert_pending_section<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    sections_table: &str,
    version: u64,
    is_baseline: bool,
    order: i32,
    section: &MigrationSection,
) -> Result<()> {
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "INSERT INTO {} (migration_version, is_baseline, section_name, section_order,
                         status, attempts, checksum, mode, module, pgmt_version)
         VALUES ($1, $2, $3, $4, $5, 0, $6, $7, $8, $9)
         ON CONFLICT (migration_version, is_baseline, section_name) DO NOTHING",
        sections_table
    )))
    .bind(version_to_db(version)?)
    .bind(is_baseline)
    .bind(&section.name)
    .bind(order)
    .bind(SectionStatus::Pending.as_str())
    .bind(calculate_checksum(&section.checksum_content()))
    .bind(section.mode.as_str())
    .bind(section.module.as_deref())
    .bind(PGMT_VERSION)
    .execute(executor)
    .await?;
    Ok(())
}

/// Register one remap section as `satisfied` on the caller's connection:
/// insert its Pending row (idempotent), then flip it to `satisfied` — but only
/// if still `pending`, so a row that already completed/satisfied here is never
/// demoted. The two statements share the caller's connection so the
/// determination and the row land together (registration folds this into its
/// own transaction; the crossing recorder runs it inside the crossing
/// transaction). THE single definition of the insert-then-mark-satisfied
/// statement pair — every satisfied-recording path routes through here.
pub(crate) async fn insert_satisfied_section(
    conn: &mut sqlx::PgConnection,
    sections_table: &str,
    version: u64,
    is_baseline: bool,
    order: i32,
    section: &MigrationSection,
) -> Result<()> {
    insert_pending_section(
        &mut *conn,
        sections_table,
        version,
        is_baseline,
        order,
        section,
    )
    .await?;
    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE {} SET status = $1, completed_at = NOW(), applied_by = CURRENT_USER
         WHERE migration_version = $2 AND is_baseline = $3 AND section_name = $4
           AND status = 'pending'",
        sections_table
    )))
    .bind(SectionStatus::Satisfied.as_str())
    .bind(version_to_db(version)?)
    .bind(is_baseline)
    .bind(&section.name)
    .execute(&mut *conn)
    .await?;
    Ok(())
}

/// One recorded section row, as stored in `{tracking}_sections`.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct StoredSection {
    pub section_name: String,
    pub status: String,
    pub checksum: Option<String>,
    pub mode: Option<String>,
    pub section_order: i32,
    pub module: Option<String>,
}

/// Must name EVERY field of [`StoredSection`] — `FromRow` resolves fields by
/// column name, so an omitted one is a decode error at runtime, not a compile
/// error.
const STORED_SECTION_COLUMNS: &str = "section_name, status, checksum, mode, section_order, module";

/// The recorded section rows for one `(version, is_baseline)` coordinate.
///
/// For a caller that walks every migration version, use
/// [`RecordedSections::load_migrations`] instead — this issues one query per
/// call.
pub async fn fetch_section_rows(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    version: u64,
    is_baseline: bool,
) -> Result<Vec<StoredSection>> {
    let rows = sqlx::query_as(sqlx::AssertSqlSafe(format!(
        "SELECT {STORED_SECTION_COLUMNS}
             FROM {} WHERE migration_version = $1 AND is_baseline = $2",
        format_sections_table_name(tracking_table)
    )))
    .bind(version_to_db(version)?)
    .bind(is_baseline)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Every recorded MIGRATION section row on the target (`is_baseline = FALSE`),
/// grouped by version and read in a SINGLE query.
///
/// This is a snapshot: it is sound only for a caller that reads a version's
/// rows before writing any of them, and never re-reads a version it has since
/// written.
pub struct RecordedSections(BTreeMap<u64, Vec<StoredSection>>);

impl RecordedSections {
    pub async fn load_migrations(pool: &PgPool, tracking_table: &TrackingTable) -> Result<Self> {
        #[derive(sqlx::FromRow)]
        struct VersionedSection {
            migration_version: i64,
            #[sqlx(flatten)]
            section: StoredSection,
        }

        let rows: Vec<VersionedSection> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
            "SELECT migration_version, {STORED_SECTION_COLUMNS}
                 FROM {} WHERE NOT is_baseline",
            format_sections_table_name(tracking_table)
        )))
        .fetch_all(pool)
        .await
        .context("Failed to read the recorded migration sections")?;

        let mut by_version: BTreeMap<u64, Vec<StoredSection>> = BTreeMap::new();
        for row in rows {
            by_version
                .entry(version_from_db(row.migration_version))
                .or_default()
                .push(row.section);
        }
        Ok(Self(by_version))
    }

    /// The rows recorded for `version` — empty when the version has none.
    pub fn for_version(&self, version: u64) -> &[StoredSection] {
        self.0.get(&version).map_or(&[], Vec::as_slice)
    }
}

/// A completed section is immutable at section granularity; an unapplied
/// (pending/failed/running) one may be fixed in the repo and re-run. Validate
/// the CURRENT file's sections against the rows registered for this version and
/// bring unapplied rows into sync.
///
/// Rules (matching "unit of resume = unit of immutability"):
/// - Rows with a NULL stored checksum are legacy/pre-upgrade — they always
///   pass and are ignored here (the caller keeps a file-level fallback guard).
/// - A COMPLETED checksummed row must match the file exactly: same checksum,
///   same transaction mode, same section_order, same module tag, and its name
///   must still exist in the file. Any divergence is a hard error naming the
///   section.
/// - A non-completed checksummed row whose checksum drifted is UPDATEd to the
///   current content (the fix-in-repo recovery path) with a printed notice.
///
/// Returns whether any checksummed section row was found — when none were, the
/// caller falls back to the legacy file-level checksum bail.
///
/// `rows` are the recorded rows for this `(version, is_baseline)` coordinate,
/// supplied by the caller ([`fetch_section_rows`] for a single coordinate,
/// [`RecordedSections`] when walking many).
pub async fn validate_and_sync_section_checksums(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    version: u64,
    is_baseline: bool,
    file_sections: &[MigrationSection],
    rows: &[StoredSection],
) -> Result<bool> {
    let sections_table = format_sections_table_name(tracking_table);
    let kind = if is_baseline { "baseline" } else { "migration" };
    // Recovery hints must carry --baseline when the drifted row IS a baseline
    // section — the bare coordinate targets the migration row space and the
    // verb would find no row.
    let baseline_flag = if is_baseline { " --baseline" } else { "" };

    // Current file: name -> (order, checksum, mode, module).
    let mut file_map: BTreeMap<&str, (i32, String, &'static str, Option<&str>)> = BTreeMap::new();
    for (idx, section) in file_sections.iter().enumerate() {
        file_map.insert(
            section.name.as_str(),
            (
                idx as i32,
                calculate_checksum(&section.checksum_content()),
                section.mode.as_str(),
                section.module.as_deref(),
            ),
        );
    }

    let mut has_checksummed = false;
    for row in rows {
        let StoredSection {
            section_name: name,
            status,
            checksum: stored_checksum,
            mode: stored_mode,
            section_order: stored_order,
            module: stored_module,
        } = row;
        // Legacy row: no stored checksum → always passes.
        let Some(stored_checksum) = stored_checksum else {
            continue;
        };
        has_checksummed = true;
        // Satisfied rows are terminal and covered, like completed: their
        // section is immutable (source-covered adoption record).
        let completed = matches!(
            status.parse::<SectionStatus>(),
            Ok(s) if s.is_covered()
        );

        let Some((order, checksum, mode, module)) = file_map.get(name.as_str()) else {
            // The section is gone from the file.
            if completed {
                anyhow::bail!(
                    "section '{name}' of {kind} {version} was applied but no longer exists in \
                     the file. Applied sections are immutable. Create a new migration for \
                     further changes."
                );
            }
            continue;
        };

        if completed {
            if stored_checksum != checksum {
                anyhow::bail!(
                    "section '{name}' of {kind} {version} was modified after it was applied.\n\
                     Expected checksum: {stored_checksum}\n\
                     Actual checksum:   {checksum}\n\n\
                     Applied sections are immutable. Create a new migration for further changes. \
                     If the edit was conscious and its effects already match the database, \
                     re-stamp the stored checksum: pgmt migrate resolve --restamp {version}/{name}{baseline_flag}."
                );
            }
            if *stored_order != *order {
                anyhow::bail!(
                    "section '{name}' of {kind} {version} was reordered after it was applied \
                     (was position {stored_order}, now {order}). Applied sections are immutable. \
                     Create a new migration for further changes. If the reorder is intentional, \
                     accept it with `pgmt migrate resolve --restamp {version}{baseline_flag}`."
                );
            }
            if let Some(sm) = stored_mode
                && sm != mode
            {
                anyhow::bail!(
                    "section '{name}' of {kind} {version} changed transaction mode after it was \
                     applied (was '{sm}', now '{mode}'). Applied sections are immutable. Create a \
                     new migration for further changes."
                );
            }
            if stored_module.as_deref() != *module {
                anyhow::bail!(
                    "section '{name}' of {kind} {version} was re-tagged to a different module \
                     after it was applied. Applied sections are immutable. Create a new migration \
                     for further changes."
                );
            }
        } else if stored_checksum != checksum {
            // Fix-in-repo recovery: an unapplied section may be edited and
            // re-run. Bring the row up to date before it executes.
            println!("section '{name}' changed since registration; updating");
            sqlx::query(sqlx::AssertSqlSafe(format!(
                "UPDATE {} SET checksum = $1, mode = $2, module = $3
                 WHERE migration_version = $4 AND is_baseline = $5 AND section_name = $6",
                sections_table
            )))
            .bind(checksum)
            .bind(*mode)
            .bind(*module)
            .bind(version_to_db(version)?)
            .bind(is_baseline)
            .bind(name)
            .execute(pool)
            .await?;
        }
    }

    Ok(has_checksummed)
}

/// Initialize sections for a migration or baseline (idempotent: rows already
/// present — e.g. on a resumed run — are left untouched).
///
/// Each entry pairs a section with its `section_order`: the section's index in
/// the FULL parsed migration/baseline file, supplied by the caller. The order
/// MUST be stable per migration version rather than derived from a local
/// `enumerate()` — under module subset deploy the sections of one version are
/// registered across SEPARATE calls (base first, a module later), so a
/// per-call index would restart at 0 and collide distinct sections on the same
/// `section_order`.
pub async fn initialize_sections(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    migration_version: u64,
    is_baseline: bool,
    sections: &[(i32, MigrationSection)],
) -> Result<()> {
    let sections_table = format_sections_table_name(tracking_table);

    for (order, section) in sections {
        insert_pending_section(
            pool,
            &sections_table,
            migration_version,
            is_baseline,
            *order,
            section,
        )
        .await?;
    }

    Ok(())
}

/// Record the given sections as `satisfied`: a remap
/// section whose source the target already holds, so nothing runs — the
/// objects are present under the source's name and a later crossing relabels
/// them. Insert-then-mark inside ONE transaction so the determination and the
/// row are atomic (the caller holds the advisory lock). Idempotent: an existing
/// row keeps its status (a resumed adoption never demotes a completed section).
pub async fn record_sections_satisfied(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    version: u64,
    is_baseline: bool,
    sections: &[(i32, MigrationSection)],
) -> Result<()> {
    if sections.is_empty() {
        return Ok(());
    }
    let sections_table = format_sections_table_name(tracking_table);
    let mut tx = pool.begin().await?;
    for (order, section) in sections {
        insert_satisfied_section(
            &mut tx,
            &sections_table,
            version,
            is_baseline,
            *order,
            section,
        )
        .await?;
    }
    tx.commit()
        .await
        .context("Failed to record satisfied sections")?;
    Ok(())
}

/// Status of every recorded section for a version, derived from rows the caller
/// already holds. Empty for legacy rows (recorded on completion by older pgmt,
/// before per-section registration).
pub fn section_statuses(rows: &[StoredSection]) -> Result<BTreeMap<String, SectionStatus>> {
    rows.iter()
        .map(|row| {
            Ok((
                row.section_name.clone(),
                SectionStatus::from_str(&row.status)?,
            ))
        })
        .collect()
}

/// Get status of a specific section
pub async fn get_section_status(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    migration_version: u64,
    is_baseline: bool,
    section_name: &str,
) -> Result<Option<SectionStatus>> {
    let sections_table = format_sections_table_name(tracking_table);

    let row: Option<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
        "SELECT status FROM {} WHERE migration_version = $1 AND is_baseline = $2 AND section_name = $3",
        sections_table
    )))
    .bind(version_to_db(migration_version)?)
    .bind(is_baseline)
    .bind(section_name)
    .fetch_optional(pool)
    .await?;

    row.map(|(status,)| SectionStatus::from_str(&status))
        .transpose()
}

/// Record section start
pub async fn record_section_start(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    migration_version: u64,
    is_baseline: bool,
    section_name: &str,
) -> Result<()> {
    let sections_table = format_sections_table_name(tracking_table);

    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE {}
         SET status = $1, started_at = NOW(), attempts = attempts + 1,
             applied_by = CURRENT_USER
         WHERE migration_version = $2 AND is_baseline = $3 AND section_name = $4",
        sections_table
    )))
    .bind(SectionStatus::Running.as_str())
    .bind(version_to_db(migration_version)?)
    .bind(is_baseline)
    .bind(section_name)
    .execute(pool)
    .await?;

    Ok(())
}

/// Record section completion.
///
/// Accepts any executor so the caller can pass either a pool (autocommit /
/// non-transactional paths) or the section's own transaction (`&mut *tx`).
/// For transactional sections, recording completion inside the section's
/// transaction — before its commit — makes the `completed` tracking row and
/// the section's DDL atomic: the row exists if and only if the DDL committed.
pub async fn record_section_complete<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    tracking_table: &TrackingTable,
    migration_version: u64,
    is_baseline: bool,
    section_name: &str,
    rows_affected: Option<i64>,
    duration_ms: i64,
) -> Result<()> {
    let sections_table = format_sections_table_name(tracking_table);

    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE {}
         SET status = $1, completed_at = NOW(), rows_affected = $2, duration_ms = $3
         WHERE migration_version = $4 AND is_baseline = $5 AND section_name = $6",
        sections_table
    )))
    .bind(SectionStatus::Completed.as_str())
    .bind(rows_affected)
    .bind(duration_ms)
    .bind(version_to_db(migration_version)?)
    .bind(is_baseline)
    .bind(section_name)
    .execute(executor)
    .await?;

    Ok(())
}

/// Record section failure
pub async fn record_section_failed(
    pool: &PgPool,
    tracking_table: &TrackingTable,
    migration_version: u64,
    is_baseline: bool,
    section_name: &str,
    error: &str,
) -> Result<()> {
    let sections_table = format_sections_table_name(tracking_table);

    sqlx::query(sqlx::AssertSqlSafe(format!(
        "UPDATE {}
         SET status = $1, last_error = $2
         WHERE migration_version = $3 AND is_baseline = $4 AND section_name = $5",
        sections_table
    )))
    .bind(SectionStatus::Failed.as_str())
    .bind(error)
    .bind(version_to_db(migration_version)?)
    .bind(is_baseline)
    .bind(section_name)
    .execute(pool)
    .await?;

    Ok(())
}

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

    #[test]
    fn test_section_status_as_str() {
        assert_eq!(SectionStatus::Pending.as_str(), "pending");
        assert_eq!(SectionStatus::Running.as_str(), "running");
        assert_eq!(SectionStatus::Completed.as_str(), "completed");
        assert_eq!(SectionStatus::Satisfied.as_str(), "satisfied");
        assert_eq!(SectionStatus::Failed.as_str(), "failed");
    }

    #[test]
    fn test_section_status_covered() {
        assert!(SectionStatus::Completed.is_covered());
        assert!(SectionStatus::Satisfied.is_covered());
        assert!(!SectionStatus::Pending.is_covered());
        assert!(!SectionStatus::Running.is_covered());
        assert!(!SectionStatus::Failed.is_covered());
    }

    #[test]
    fn test_section_status_from_str() {
        assert_eq!(
            "pending".parse::<SectionStatus>().unwrap(),
            SectionStatus::Pending
        );
        assert_eq!(
            "running".parse::<SectionStatus>().unwrap(),
            SectionStatus::Running
        );
        assert_eq!(
            "completed".parse::<SectionStatus>().unwrap(),
            SectionStatus::Completed
        );
        assert_eq!(
            "satisfied".parse::<SectionStatus>().unwrap(),
            SectionStatus::Satisfied
        );
        assert_eq!(
            "failed".parse::<SectionStatus>().unwrap(),
            SectionStatus::Failed
        );
        assert!("invalid".parse::<SectionStatus>().is_err());
    }
}