armdb 0.8.0

use std::io;

use armour_core::persist::PersistError;

/// `" by pid 1234"` when the lock file carried a readable pid, `""` otherwise.
fn pid_hint(pid: &Option<u32>) -> String {
    match pid {
        Some(p) => format!(" by pid {p}"),
        None => String::new(),
    }
}

/// P2-4: `#[non_exhaustive]` so future variants are additive — downstream crates
/// must include a wildcard arm, and adding a variant (like `DuplicateCollection`)
/// is no longer a source-breaking change to their exhaustive matches.
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum DbError {
    /// 400
    #[error("client error: {0}")]
    Client(&'static str),
    /// 404
    #[error("key not found")]
    KeyNotFound,
    /// 501
    #[error("not implemented")]
    NotImplemented,
    #[error("storage error: {0}")]
    Internal(&'static str),
    #[error(transparent)]
    Io(#[from] io::Error),
    #[error("CRC32 mismatch: expected {expected:#x}, got {actual:#x}")]
    CrcMismatch { expected: u32, actual: u32 },
    #[error("corrupted entry at offset {offset}")]
    CorruptedEntry { offset: u64 },
    #[error("invalid configuration: {0}")]
    Config(&'static str),
    #[error("CAS value mismatch")]
    CasMismatch,
    #[error("key already exists")]
    KeyExists,
    /// A copied DiskLoc points to a file that has just been removed from
    /// inner.immutable by compaction. The data is logically still live;
    /// the caller should retry with a fresh DiskLoc from the index.
    #[error("stale DiskLoc - file removed by compaction")]
    StaleDiskLoc,
    #[error("all slots are occupied")]
    SlotsFull,
    #[error("key routes to a different shard")]
    ShardMismatch,
    /// Reserved for a future multi-collection transaction API (T5).
    ///
    /// Historically produced by `Db::atomic2/3/4` when the same collection
    /// appeared in two argument positions (re-locking its shards from the same
    /// thread would self-deadlock). Those methods were removed in 0.8.0; the
    /// variant and its default 500 mapping are kept so clients and T5 stay
    /// stable.
    /// armdb does not return this variant today.
    #[error("the same collection appears twice in a cross-collection transaction")]
    DuplicateCollectionInTx,
    /// The transaction log could not be repaired after a failed append; the
    /// database must be reopened before it can accept transactions again.
    #[error("transaction log is poisoned: {0}")]
    TxLogPoisoned(String),
    /// A committed transaction operation could not be applied to this shard, so
    /// the shard owes a write that only replay can complete. It refuses further
    /// writes until the database is reopened.
    ///
    /// Bitcask replay asks whether a reserved GSN is above the shard's recovered
    /// watermark. Letting a later write land would raise that watermark past the
    /// unapplied reservation, and the next open would read the committed
    /// operation as already on disk and drop it. Reads are unaffected, other
    /// shards are unaffected, and reopening the database replays the operation
    /// and clears the fence.
    #[error(
        "shard {shard} owes a committed transaction write (gsn {gsn}); reopen the database to replay it"
    )]
    ShardFenced { shard: usize, gsn: u64 },
    /// A collection with this name is already open on this `Db`, or the name is
    /// reserved for internal state (`__seq`, `db.info`, `db.user`). Opening it
    /// again would alias two in-memory indexes onto the same shard files and
    /// silently drop each other's updates.
    #[error("collection name already in use or reserved: {0}")]
    DuplicateCollection(&'static str),
    /// The database directory is already open — by another process, or by
    /// another live handle in this one. armdb requires exclusive access: two
    /// engines over the same shard files would run independent in-memory
    /// indexes and silently drop each other's writes.
    ///
    /// `pid` is a best-effort diagnostic read from the lock file; `None` means
    /// it was unreadable, not that no process holds the lock.
    #[error("database directory is locked{}: {}", pid_hint(pid), path.display())]
    Locked {
        path: std::path::PathBuf,
        pid: Option<u32>,
    },
    /// Optimistic-concurrency retry signal: a value read in phase 1 changed before
    /// the narrow transaction in phase 2 could commit. The caller should retry.
    /// 409. armdb does not return this variant today; reserved for T5 and callers.
    #[error("transaction conflict — retry")]
    TxConflict,
    /// `db.tx().commit()` was called from inside a collection's `atomic()` closure.
    /// Shard locks are not reentrant; commit while `atomic()` holds them would hang.
    #[error("cannot commit a transaction from inside atomic()")]
    ReentrantTx,
    /// A pending transaction record's schema fingerprint does not match the
    /// collection's stored type hash at open.
    #[error(
        "pending transaction schema mismatch on '{collection}' (tx_id={tx_id}): stored={stored:#x} logged={logged:#x}"
    )]
    TxSchemaMismatch {
        collection: &'static str,
        tx_id: u64,
        stored: u64,
        logged: u64,
    },
    /// A pending FixedStore operation cannot be ordered against the slot's
    /// current version: the two are more than 2^29 apart in the 30-bit ring, so
    /// neither "already applied" nor "never applied" can be established.
    /// Recovering requires human judgement, so the open fails rather than guess.
    #[error(
        "collection {collection}: pending transaction targets slot {slot} at meta \
         {target:#010x}, which cannot be ordered against the slot's current meta {current:#010x}"
    )]
    TxFixedSlot {
        collection: &'static str,
        slot: u32,
        current: u32,
        target: u32,
    },
    #[error("disk format mismatch: {0}")]
    FormatMismatch(String),
    #[cfg(feature = "fjall")]
    #[error("fjall backend error: {0}")]
    Fjall(String),
    #[cfg(feature = "encryption")]
    #[error("encryption error: {0}")]
    EncryptionError(String),
    #[cfg(feature = "replication")]
    #[error("replication error: {0}")]
    Replication(String),
    #[cfg(feature = "replication")]
    #[error("shard count mismatch: leader has {leader}, follower has {follower}")]
    ShardCountMismatch { leader: usize, follower: usize },
    #[error("catch-up run limit exceeded: observed {actual}, configured {limit}")]
    CatchupResourceLimit { actual: usize, limit: usize },
    /// Schema/version mismatch detected while opening a collection.
    /// See `SchemaMismatchKind` for the specific violation.
    #[error("schema mismatch on '{name}': {kind}")]
    SchemaMismatch {
        name: String,
        kind: SchemaMismatchKind,
    },
    #[error("recovery thread panicked")]
    RecoveryPanic,
}

/// Specific kind of schema mismatch (see `DbError::SchemaMismatch`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SchemaMismatchKind {
    /// Stored and expected `typ_hash` differ at the same `version`.
    /// Indicates the schema changed without bumping `T::VERSION`.
    TypHash { stored: u64, expected: u64 },
    /// Stored `version` is newer than the requested `meta.version`
    /// (rollback to an older binary).
    Downgrade { stored: u16, requested: u16 },
    /// No migration step is registered for the source version `from`.
    MissingStep { from: u16 },
}

impl core::fmt::Display for SchemaMismatchKind {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::TypHash { stored, expected } => write!(
                f,
                "type hash drift: stored={stored:#x} expected={expected:#x}"
            ),
            Self::Downgrade { stored, requested } => write!(
                f,
                "version downgrade: stored={stored} requested={requested}"
            ),
            Self::MissingStep { from } => write!(f, "missing migration from version {from}"),
        }
    }
}

impl DbError {
    pub fn status_code(&self) -> u16 {
        match self {
            DbError::Client(_) => 400,
            DbError::KeyNotFound => 404,
            DbError::NotImplemented => 501,
            DbError::SchemaMismatch { .. } => 422,
            DbError::KeyExists => 409,
            DbError::CasMismatch => 412,
            DbError::TxConflict => 409,
            DbError::TxSchemaMismatch { .. } => 422,
            _ => 500,
        }
    }
}

pub type DbResult<T> = Result<T, DbError>;

impl From<PersistError> for DbError {
    fn from(err: PersistError) -> Self {
        match err {
            PersistError::Io(e) => Self::Io(e),
            PersistError::Json(e) => Self::FormatMismatch(e.to_string()),
        }
    }
}

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

    #[test]
    fn schema_mismatch_typ_hash_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "users".into(),
            kind: SchemaMismatchKind::TypHash {
                stored: 0xDEAD,
                expected: 0xBEEF,
            },
        };
        assert_eq!(e.status_code(), 422);
    }

    #[test]
    fn schema_mismatch_downgrade_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "users".into(),
            kind: SchemaMismatchKind::Downgrade {
                stored: 3,
                requested: 2,
            },
        };
        assert_eq!(e.status_code(), 422);
    }

    #[test]
    fn schema_mismatch_missing_step_maps_to_422() {
        let e = DbError::SchemaMismatch {
            name: "users".into(),
            kind: SchemaMismatchKind::MissingStep { from: 1 },
        };
        assert_eq!(e.status_code(), 422);
    }

    #[test]
    fn key_exists_maps_to_409() {
        assert_eq!(DbError::KeyExists.status_code(), 409);
    }

    #[test]
    fn cas_mismatch_maps_to_412() {
        assert_eq!(DbError::CasMismatch.status_code(), 412);
    }
}