haematite 0.6.1

//! Database startup and lazy shard-router construction.
//!
//! LAZY SHARD MATERIALISATION: startup no longer pre-creates all `shard_count`
//! directories or spawn one actor per shard. It builds a [`ShardRouter`] over
//! the fixed `shard_count` modulus base and lets the router spawn each shard on
//! first touch. Boot cost is
//! therefore O(shards actually used), not `O(shard_count)` — a very high
//! `shard_count` is ~free until its shards are exercised.

use std::fs;
use std::sync::Arc;
use std::time::Duration;

use beamr::module::ModuleRegistry;
use beamr::scheduler::{Scheduler, SchedulerConfig};

use crate::shard::router::{ShardMode, ShardRouter};
use crate::sync::scheduler::{
    NoopSyncPullTrigger, SyncSchedulerConfig, SyncSchedulerError, SyncSchedulerHandle,
};

use super::config::{ensure_uninitialised, validate_database_config, write_config};
use super::lock::DataDirLock;
use super::{Database, DatabaseConfig, DatabaseError};

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);

/// Everything a started database owns EXCEPT the writer lock. Splitting the
/// start from the assembly lets the create path keep the lock alive across
/// its failure cleanup: removing a partial directory is only safe while the
/// exclusive lock still excludes every other writer.
struct DatabaseParts {
    config: DatabaseConfig,
    scheduler: Arc<Scheduler>,
    router: ShardRouter,
    sync_schedulers: Vec<SyncSchedulerHandle>,
}

fn assemble(parts: DatabaseParts, lock: DataDirLock) -> Database {
    Database {
        config: parts.config,
        scheduler: parts.scheduler,
        router: parts.router,
        sync_schedulers: parts.sync_schedulers,
        distribution: None,
        owner_stamps: super::owner_stamp::OwnerStamps::default(),
        lock,
        timeout: DEFAULT_TIMEOUT,
    }
}

#[derive(Clone, Copy, Debug)]
pub(super) enum StartupMode {
    Create,
    Open,
}

impl StartupMode {
    const fn shard_mode(self) -> ShardMode {
        match self {
            Self::Create => ShardMode::Create,
            Self::Open => ShardMode::Open,
        }
    }
}

pub(super) fn initialise_database(config: DatabaseConfig) -> Result<Database, DatabaseError> {
    // Only the database root directory is created up front; per-shard directories
    // are created lazily by the router on each shard's first touch. This is what
    // makes create() O(1) in `shard_count` instead of O(shard_count).
    let created_dir = !config.data_dir.exists();
    fs::create_dir_all(&config.data_dir).map_err(DatabaseError::DirectoryCreate)?;
    if created_dir {
        sync_new_dir_entry(&config.data_dir)?;
    }
    // A4: the exclusive writer lock is taken BEFORE `config.json` is written, so
    // a create that loses the lock race fails loudly without clobbering the live
    // winner's config. A lock-acquire failure deliberately skips the cleanup
    // below even when this call created the directory: losing the race means a
    // concurrent winner already owns the dir it raced us into.
    let lock = DataDirLock::acquire(&config.data_dir)?;
    let data_dir = config.data_dir.clone();
    match create_locked(config) {
        Ok(parts) => Ok(assemble(parts, lock)),
        Err(error) => {
            // Cleanup runs while the exclusive lock is STILL HELD, so a
            // concurrent create/open cannot acquire the directory between this
            // failure and its removal — and only a directory this very call
            // created is removed, never a pre-existing one. The lock is
            // released only after the removal completes.
            if created_dir {
                drop(fs::remove_dir_all(&data_dir));
            }
            drop(lock);
            Err(error)
        }
    }
}

/// The fallible portion of create that must run under the writer lock: refuse
/// an already-initialised directory (A5), install the stamped config, start
/// the shards.
fn create_locked(config: DatabaseConfig) -> Result<DatabaseParts, DatabaseError> {
    ensure_uninitialised(&config.data_dir)?;
    write_config(&config)?;
    start_database(config, StartupMode::Create)
}

/// A4: open path — acquire the exclusive writer lock, then start the shards.
pub(super) fn open_database(config: DatabaseConfig) -> Result<Database, DatabaseError> {
    let lock = DataDirLock::acquire(&config.data_dir)?;
    start_database(config, StartupMode::Open).map(|parts| assemble(parts, lock))
}

fn start_database(
    config: DatabaseConfig,
    mode: StartupMode,
) -> Result<DatabaseParts, DatabaseError> {
    validate_database_config(&config)?;
    let scheduler = create_scheduler()?;
    let router = build_router(&scheduler, &config, mode)?;
    let sync_schedulers = match spawn_sync_schedulers(&scheduler, &config, &router) {
        Ok(handles) => handles,
        Err(error) => {
            router.shutdown_all(DEFAULT_TIMEOUT);
            return Err(error);
        }
    };
    Ok(DatabaseParts {
        config,
        scheduler,
        router,
        sync_schedulers,
    })
}

/// Flush the parent-directory entry for a data dir this create just made:
/// config.json and its content are fsynced, but their dirents live inside the
/// new directory, whose OWN entry in its parent is what makes the database
/// reachable after power loss. `create_dir_all` may create deeper ancestors;
/// only the immediate parent is synced here — pre-existing ancestors are
/// assumed durable. Windows: `std` cannot open a directory for fsync, so the
/// sync is skipped there (the same platform gap as `branch/persist.rs`).
#[cfg(unix)]
fn sync_new_dir_entry(data_dir: &std::path::Path) -> Result<(), DatabaseError> {
    let Some(parent) = data_dir.parent() else {
        return Ok(());
    };
    let parent = if parent.as_os_str().is_empty() {
        std::path::Path::new(".")
    } else {
        parent
    };
    fs::File::open(parent)
        .and_then(|dir| dir.sync_all())
        .map_err(DatabaseError::DirectoryCreate)
}

#[cfg(not(unix))]
fn sync_new_dir_entry(_data_dir: &std::path::Path) -> Result<(), DatabaseError> {
    Ok(())
}

fn create_scheduler() -> Result<Arc<Scheduler>, DatabaseError> {
    Scheduler::new(SchedulerConfig::default(), Arc::new(ModuleRegistry::new()))
        .map(Arc::new)
        .map_err(DatabaseError::ShardSpawn)
}

fn build_router(
    scheduler: &Arc<Scheduler>,
    config: &DatabaseConfig,
    mode: StartupMode,
) -> Result<ShardRouter, DatabaseError> {
    ShardRouter::new(
        Arc::clone(scheduler),
        &config.data_dir,
        config.shard_count,
        mode.shard_mode(),
    )
    .ok_or(DatabaseError::InvalidShardCount)
}

fn spawn_sync_schedulers(
    scheduler: &Arc<Scheduler>,
    config: &DatabaseConfig,
    router: &ShardRouter,
) -> Result<Vec<SyncSchedulerHandle>, DatabaseError> {
    let Some(distributed) = &config.distributed else {
        return Ok(Vec::new());
    };
    let topology = distributed
        .topology
        .clone()
        .ok_or(DatabaseError::MissingSyncTopology)?;
    let scheduler_config = SyncSchedulerConfig::new(
        distributed.local_node.clone(),
        distributed.nodes.clone(),
        topology,
        config.shard_count,
        Duration::from_millis(distributed.sync_interval),
    );
    // LAZY: the scheduler syncs ONLY materialised shards (via the router's
    // membership view) — an un-materialised shard holds no data, so pulling it
    // would be pure waste.
    SyncSchedulerHandle::spawn_with_shard_source(
        Arc::clone(scheduler),
        scheduler_config,
        Arc::new(NoopSyncPullTrigger),
        Arc::new(router.membership()),
        DEFAULT_TIMEOUT,
    )
    .map(|handle| vec![handle])
    .map_err(map_sync_scheduler_error)
}

fn map_sync_scheduler_error(error: SyncSchedulerError) -> DatabaseError {
    match error {
        SyncSchedulerError::Spawn(message) => DatabaseError::SyncSchedulerSpawn(message),
        other => DatabaseError::SyncSchedulerError(other.to_string()),
    }
}