haematite 0.6.1

use std::fmt;
use std::io;
use std::path::{Path, PathBuf};

/// Errors surfaced by the top-level database handle.
#[derive(Debug)]
pub enum DatabaseError {
    DirectoryCreate(io::Error),
    ConfigWrite(io::Error),
    ConfigRead(io::Error),
    ConfigParse(String),
    InvalidShardCount,
    ShardSpawn(String),
    SweepSpawn(String),
    ShardError(String),
    SweepError(String),
    SyncSchedulerSpawn(String),
    SyncSchedulerError(String),
    IoError(io::Error),
    /// A retired TTL cadence field was present in stored configuration with
    /// the historically invalid zero value.
    InvalidSweepInterval,
    MissingSyncTopology,
    InvalidSyncInterval,
    /// The data directory's `config.json` carries an on-disk `format_version`
    /// stamp NEWER than this binary's [`crate::db::ON_DISK_FORMAT_VERSION`]
    /// (A5 downgrade refusal): the directory was written by a newer haematite
    /// whose layout this binary cannot know, so it refuses loudly instead of
    /// misreading data. Carries both versions so the operator can see exactly
    /// which side is behind. Distinct from [`Self::ConfigParse`]: the file is
    /// well-formed, it is the binary that is too old.
    FormatVersionTooNew {
        found: u32,
        supported: u32,
    },
    SequenceConflict {
        expected: u64,
        actual: u64,
    },
    CasMismatch {
        expected: Option<u64>,
        actual: Option<u64>,
    },
    ConsistencyError(String),
    /// A live distribution-endpoint operation failed (no endpoint attached, a
    /// transport send/connect failure, or a disconnected inbound drain).
    Distribution(String),
    /// A replicated write reached peer-quorum but the proposer could not durably
    /// apply its OWN committed value locally (see [`crate::db::Database::replicate_write`]).
    ///
    /// This is reported, never swallowed: a committed write that is absent on its
    /// own writer is a correctness hazard (it reopens the heal-mid-write
    /// split-brain hole). Under single-owner-per-key (the step-3 epoch fence) the
    /// local CAS can never mismatch, so this only ever surfaces a genuine local
    /// storage/IO fault.
    LocalCommitFailed(String),
    /// An [`crate::db::Database::acquire_shard`] election lost: a strictly higher
    /// ballot was promised elsewhere on every attempt. The candidate is NOT the
    /// owner and recorded no `owner_epoch`. Carries the highest competing counter
    /// seen so a caller could retry above it later. This is a clean, safe loss —
    /// the unique-ballot / majority invariants were never relaxed.
    ElectionLost {
        highest_seen: u64,
    },
    /// An [`crate::db::Database::acquire_shard`] election could not collect a
    /// majority of promises within the timeout on any attempt (e.g. a minority of
    /// nodes was reachable). The candidate is NOT the owner — never a false win.
    ElectionTimeout {
        attempts: u32,
    },
    /// A replicated CAS write was deterministically out-voted by the cluster: a
    /// stale/deposed owner's proposal collected enough rejects that a quorum of
    /// accepts is no longer reachable, so the writer is fenced and NOTHING was
    /// applied (the typed twin of [`crate::ConsistencyError::Fenced`]). Surfaced
    /// as its own variant — distinct from a generic [`Self::ConsistencyError`]
    /// string — so a consumer (e.g. an aion shard owner) can match the fence
    /// directly and re-resolve ownership rather than parsing a Display message.
    Fenced {
        required: usize,
        possible_accepts: usize,
    },
    /// A replicated CAS write was deterministically out-voted by *value-CAS
    /// mismatches alone* — the writer is still the live owner, but enough replicas
    /// refused the precondition that a quorum of accepts became unreachable (the
    /// typed twin of [`crate::ConsistencyError::CasConflict`]). Distinct from
    /// [`Self::Fenced`] (a higher-ballot owner deposed us, requiring ownership
    /// re-resolution): a `CasConflict` caller may simply re-read and re-CAS.
    CasConflict {
        required: usize,
        possible_accepts: usize,
    },
    /// The durable `cluster/members` record (CSOT-1, task #146) could not be
    /// encoded or a stored record could not be decoded/validated. Carries the
    /// underlying [`crate::sync::ClusterMembersError`] as a string so this variant
    /// stays dependency-light and matches the existing stringified-cause style.
    ClusterMembers(String),
    /// Another live writer holds the exclusive data-dir lock (A4): a second
    /// writer process (or a second `Database` handle in this process) already
    /// owns `<data_dir>/writer.lock`, and running two writers over the same
    /// shard WALs would corrupt them. The open fails immediately — it never
    /// blocks and never touches shard state. Advisory locks self-release on
    /// process death, so this is always a LIVE writer, never a stale lock.
    /// For observation alongside a live writer use
    /// [`crate::db::ReadOnlyDatabase`], which takes no lock.
    DataDirLocked {
        lock_path: PathBuf,
    },
    /// The data-dir writer lockfile could not be opened/created or the lock
    /// syscall failed for an I/O reason distinct from contention (A4).
    LockFileIo {
        lock_path: PathBuf,
        error: io::Error,
    },
    /// `Database::create` was pointed at a directory that already holds a
    /// database (its `config.json` exists) with no live writer attached (A5):
    /// create refuses rather than clobbering the existing config. A directory
    /// written by a NEWER binary refuses as [`Self::FormatVersionTooNew`]
    /// before this check is reached. Use [`crate::db::Database::open`] on the
    /// existing directory, or point create at a fresh path.
    DataDirAlreadyInitialised {
        config_path: PathBuf,
    },
}

impl fmt::Display for DatabaseError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DirectoryCreate(error) => {
                write!(formatter, "failed to create database directory: {error}")
            }
            Self::ConfigWrite(error) => {
                write!(formatter, "failed to write database config: {error}")
            }
            Self::ConfigRead(error) => write!(formatter, "failed to read database config: {error}"),
            Self::ConfigParse(message) => {
                write!(formatter, "failed to parse database config: {message}")
            }
            Self::InvalidShardCount => write!(formatter, "database shard_count must be at least 1"),
            Self::ShardSpawn(message) => {
                write!(formatter, "failed to spawn shard actor: {message}")
            }
            Self::SweepSpawn(message) => {
                write!(formatter, "failed to spawn sweep actor: {message}")
            }
            Self::ShardError(message) => write!(formatter, "shard operation failed: {message}"),
            Self::SweepError(message) => write!(formatter, "sweep operation failed: {message}"),
            Self::SyncSchedulerSpawn(message) => {
                write!(formatter, "failed to spawn sync scheduler: {message}")
            }
            Self::SyncSchedulerError(message) => {
                write!(formatter, "sync scheduler failed: {message}")
            }
            Self::IoError(error) => write!(formatter, "database I/O error: {error}"),
            Self::InvalidSweepInterval => {
                write!(formatter, "legacy TTL cadence must be greater than zero")
            }
            Self::MissingSyncTopology => {
                write!(formatter, "distributed database requires sync topology")
            }
            Self::InvalidSyncInterval => {
                write!(formatter, "sync_interval must be greater than zero")
            }
            Self::FormatVersionTooNew { found, supported } => {
                fmt_format_version_too_new(formatter, *found, *supported)
            }
            Self::SequenceConflict { expected, actual } => write!(
                formatter,
                "sequence conflict on append: expected {expected}, actual {actual}"
            ),
            Self::CasMismatch { expected, actual } => write!(
                formatter,
                "cas mismatch: expected {expected:?}, actual {actual:?}"
            ),
            Self::ConsistencyError(message) => {
                write!(formatter, "consistency requirement failed: {message}")
            }
            Self::Distribution(message) => {
                write!(formatter, "distribution endpoint error: {message}")
            }
            Self::LocalCommitFailed(message) => write!(
                formatter,
                "replicated write reached quorum but local durable commit failed: {message}"
            ),
            Self::ElectionLost { highest_seen } => write!(
                formatter,
                "shard election lost: a higher ballot (counter {highest_seen}) was promised elsewhere"
            ),
            Self::ElectionTimeout { attempts } => write!(
                formatter,
                "shard election timed out without a majority after {attempts} attempts"
            ),
            Self::Fenced {
                required,
                possible_accepts,
            } => write!(
                formatter,
                "fenced by CAS rejects: required {required} accepts, only {possible_accepts} still possible"
            ),
            Self::CasConflict {
                required,
                possible_accepts,
            } => write!(
                formatter,
                "lost CAS by value mismatch: required {required} accepts, only {possible_accepts} still possible"
            ),
            Self::ClusterMembers(message) => {
                write!(formatter, "cluster/members record error: {message}")
            }
            Self::DataDirLocked { lock_path } => fmt_data_dir_locked(formatter, lock_path),
            Self::LockFileIo { lock_path, error } => write!(
                formatter,
                "failed to acquire the data-dir writer lock at {}: {error}",
                lock_path.display()
            ),
            Self::DataDirAlreadyInitialised { config_path } => {
                fmt_data_dir_already_initialised(formatter, config_path)
            }
        }
    }
}

/// Long-form message bodies for the A4/A5 data-dir refusals, split out of
/// `Display::fmt` to keep that exhaustive match within the function-length
/// lint as the error surface grows.
fn fmt_format_version_too_new(
    formatter: &mut fmt::Formatter<'_>,
    found: u32,
    supported: u32,
) -> fmt::Result {
    write!(
        formatter,
        "database on-disk format version {found} is newer than the newest format version \
         this binary supports ({supported}); refusing to open — use a haematite build \
         that understands format version {found}"
    )
}

fn fmt_data_dir_locked(formatter: &mut fmt::Formatter<'_>, lock_path: &Path) -> fmt::Result {
    write!(
        formatter,
        "data dir is locked by another live writer (writer lock held at {}); \
         a second writer would corrupt the shard WALs — use ReadOnlyDatabase to observe",
        lock_path.display()
    )
}

fn fmt_data_dir_already_initialised(
    formatter: &mut fmt::Formatter<'_>,
    config_path: &Path,
) -> fmt::Result {
    write!(
        formatter,
        "data dir already holds a database ({} exists); refusing to clobber it — \
         open the existing database instead of creating over it",
        config_path.display()
    )
}

impl std::error::Error for DatabaseError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::DirectoryCreate(error)
            | Self::ConfigWrite(error)
            | Self::ConfigRead(error)
            | Self::IoError(error)
            | Self::LockFileIo { error, .. } => Some(error),
            Self::ConfigParse(_)
            | Self::InvalidShardCount
            | Self::ShardSpawn(_)
            | Self::SweepSpawn(_)
            | Self::ShardError(_)
            | Self::SweepError(_)
            | Self::SyncSchedulerSpawn(_)
            | Self::SyncSchedulerError(_)
            | Self::InvalidSweepInterval
            | Self::MissingSyncTopology
            | Self::InvalidSyncInterval
            | Self::FormatVersionTooNew { .. }
            | Self::SequenceConflict { .. }
            | Self::CasMismatch { .. }
            | Self::ConsistencyError(_)
            | Self::Distribution(_)
            | Self::LocalCommitFailed(_)
            | Self::ElectionLost { .. }
            | Self::ElectionTimeout { .. }
            | Self::Fenced { .. }
            | Self::CasConflict { .. }
            | Self::ClusterMembers(_)
            | Self::DataDirLocked { .. }
            | Self::DataDirAlreadyInitialised { .. } => None,
        }
    }
}

impl From<crate::sync::ClusterMembersError> for DatabaseError {
    fn from(error: crate::sync::ClusterMembersError) -> Self {
        Self::ClusterMembers(error.to_string())
    }
}

impl From<io::Error> for DatabaseError {
    fn from(error: io::Error) -> Self {
        Self::IoError(error)
    }
}

impl From<crate::sync::ConsistencyError> for DatabaseError {
    /// Preserve the deterministic CAS fence as the typed [`Self::Fenced`] so
    /// consumers can match it; every other consistency failure keeps its existing
    /// stringified [`Self::ConsistencyError`] form (behaviour unchanged).
    fn from(error: crate::sync::ConsistencyError) -> Self {
        match error {
            crate::sync::ConsistencyError::Fenced {
                required,
                possible_accepts,
            } => Self::Fenced {
                required,
                possible_accepts,
            },
            crate::sync::ConsistencyError::CasConflict {
                required,
                possible_accepts,
            } => Self::CasConflict {
                required,
                possible_accepts,
            },
            other => Self::ConsistencyError(other.to_string()),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::DatabaseError;
    use crate::sync::ConsistencyError;

    #[test]
    fn consistency_fence_maps_to_typed_fenced() {
        let mapped = DatabaseError::from(ConsistencyError::Fenced {
            required: 3,
            possible_accepts: 1,
        });
        assert!(
            matches!(
                mapped,
                DatabaseError::Fenced {
                    required: 3,
                    possible_accepts: 1
                }
            ),
            "the deterministic CAS fence must survive as the typed DatabaseError::Fenced"
        );
    }

    #[test]
    fn consistency_cas_conflict_maps_to_typed_cas_conflict() {
        // The value-CAS loss must survive as the typed DatabaseError::CasConflict
        // (preserving required/possible_accepts), distinct from the typed Fenced and
        // from the stringified fallback.
        let mapped = DatabaseError::from(ConsistencyError::CasConflict {
            required: 3,
            possible_accepts: 1,
        });
        assert!(
            matches!(
                mapped,
                DatabaseError::CasConflict {
                    required: 3,
                    possible_accepts: 1
                }
            ),
            "the value-CAS loss must survive as the typed DatabaseError::CasConflict"
        );
    }

    #[test]
    fn other_consistency_failures_stay_stringified() {
        // A non-fence consistency failure must NOT be misclassified as a fence; it
        // keeps its existing stringified ConsistencyError form (behaviour unchanged).
        for error in [
            ConsistencyError::QuorumUnavailable {
                required: 2,
                possible: 1,
            },
            ConsistencyError::TransportUnavailable,
            ConsistencyError::AckFailed,
        ] {
            let display = error.to_string();
            let mapped = DatabaseError::from(error);
            assert!(
                matches!(mapped, DatabaseError::ConsistencyError(ref message) if *message == display),
                "non-fence consistency failures must remain DatabaseError::ConsistencyError"
            );
        }
    }
}