use heed::{
    Database, EnvFlags, EnvOpenOptions,
    byteorder::BigEndian,
    types::{Bytes, U64},
};

use super::journal_scan::{
    ArchivedEnvelopeOf, JournalEvent, JournalScanSource, ScanStartPolicy, ScanSummary,
    ValidatedScan, decode_journal_event, replay_wal, validate_recovery_window,
    validate_segment_sequence,
};
use crate::disruptor::{
    Envelope, RingSlot,
    traits::{RkyvError, RkyvToBytes},
};
use crate::error::WalError;
use crate::{book::BookEventEnvelope, config::JournalConfig};
use core_affinity::CoreId;
use disrupt_rs::{EventPoller, GatedSequence, Polling, Sequence, wait_strategies::WaitStrategy};
use std::{
    cell::{Cell, RefCell},
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    thread,
    time::Instant,
};
use tracing::{error, info, trace, warn};

type BEU64 = U64<BigEndian>;
type LogDb = Database<BEU64, Bytes>;
type MetaDb = Database<Bytes, BEU64>;

const META_SCHEMA_VERSION_KEY: &[u8] = b"schema_version";
const WAL_SCHEMA_VERSION: u64 = 1;
const TRASH_DIR_NAME: &str = "trash";

#[inline]
fn journaler_error_shutdown_ordering() -> Ordering {
    Ordering::Release
}

thread_local! {
    static SCRATCH: RefCell<rkyv::util::AlignedVec> =
        RefCell::new(rkyv::util::AlignedVec::with_capacity(1024));
}

pub(crate) const ENGINE_WAL_DB_NAME: &str = "walw";

// ---------------------------------------------------------------------------
// Segment directory helpers
// ---------------------------------------------------------------------------

fn segment_dir_name(start_seq: u64) -> String {
    // u64 max has 20 decimal digits; fixed width keeps names stable and predictable.
    format!("seg_{start_seq:020}")
}

pub(crate) fn parse_segment_dir(name: &str) -> Option<u64> {
    name.strip_prefix("seg_")
        .and_then(|s| s.parse::<u64>().ok())
}

pub(crate) fn list_segment_dirs(root: &Path) -> anyhow::Result<Vec<String>> {
    let mut dirs = Vec::new();
    if !root.exists() {
        return Ok(dirs);
    }
    for entry in std::fs::read_dir(root)? {
        let entry = entry?;
        if entry.file_type()?.is_dir() {
            let name = entry.file_name().to_string_lossy().into_owned();
            if parse_segment_dir(&name).is_some() {
                dirs.push(name);
            }
        }
    }
    // Sort by numeric start sequence so ordering stays correct regardless of digit width.
    dirs.sort_unstable_by_key(|name| parse_segment_dir(name).unwrap_or(0));
    Ok(dirs)
}

fn trash_dir(root: &Path) -> PathBuf {
    root.join(TRASH_DIR_NAME)
}

fn clear_trash_dir(root: &Path) -> anyhow::Result<usize> {
    let trash = trash_dir(root);
    if !trash.exists() {
        return Ok(0);
    }

    let mut removed = 0usize;
    for entry in std::fs::read_dir(&trash)? {
        let entry = entry?;
        let path = entry.path();
        if entry.file_type()?.is_dir() {
            std::fs::remove_dir_all(&path)?;
        } else {
            std::fs::remove_file(&path)?;
        }
        removed += 1;
    }

    if std::fs::read_dir(&trash)?.next().is_none() {
        let _ = std::fs::remove_dir(&trash);
    }

    Ok(removed)
}

/// Binary-search for the segment whose start_seq <= target.
/// Returns the index into `dirs`. Assumes `dirs` is sorted.
pub(crate) fn find_segment_for_seq(dirs: &[String], target: u64) -> usize {
    match dirs.binary_search_by_key(&target, |d| parse_segment_dir(d).unwrap_or(0)) {
        Ok(i) => i,
        Err(i) => i.saturating_sub(1),
    }
}

fn first_log_seq(env: &heed::Env, log: &LogDb) -> anyhow::Result<Option<u64>> {
    let rtxn = env.read_txn()?;
    Ok(log.first(&rtxn)?.map(|(k, _)| k))
}

fn last_log_seq(env: &heed::Env, log: &LogDb) -> anyhow::Result<Option<u64>> {
    let rtxn = env.read_txn()?;
    Ok(log.last(&rtxn)?.map(|(k, _)| k))
}

fn visit_log_entries<F>(
    env: &heed::Env,
    log: &LogDb,
    start_seq: u64,
    up_to_seq: Option<u64>,
    mut on_entry: F,
) -> anyhow::Result<()>
where
    F: FnMut(u64, &[u8]) -> anyhow::Result<()>,
{
    let rtxn = env.read_txn()?;

    match up_to_seq {
        Some(up_to_seq) => {
            let mut iter = log.range(&rtxn, &(start_seq..=up_to_seq))?;
            while let Some(entry) = iter.next().transpose()? {
                let (seq, val_bytes) = entry;
                on_entry(seq, val_bytes)?;
            }
        }
        None => {
            let mut iter = log.range(&rtxn, &(start_seq..))?;
            while let Some(entry) = iter.next().transpose()? {
                let (seq, val_bytes) = entry;
                on_entry(seq, val_bytes)?;
            }
        }
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// Segment — a single LMDB environment
// ---------------------------------------------------------------------------

struct Segment {
    env: heed::Env,
    log: LogDb,
    event_count: Cell<u64>,
}

impl Segment {
    fn open(dir_path: &Path, db_name: &str, map_size: usize) -> anyhow::Result<Self> {
        std::fs::create_dir_all(dir_path)?;

        let env = unsafe {
            EnvOpenOptions::new()
                .map_size(map_size)
                .max_dbs(8)
                .open(dir_path)?
        };

        let log = {
            let mut wtxn = env.write_txn()?;
            let db: LogDb = env.create_database(&mut wtxn, Some(db_name))?;
            let meta: MetaDb = env.create_database(&mut wtxn, Some("meta"))?;

            match meta.get(&wtxn, META_SCHEMA_VERSION_KEY)? {
                Some(v) if v == WAL_SCHEMA_VERSION => {}
                Some(v) => {
                    return Err(WalError::SchemaMismatch {
                        expected: WAL_SCHEMA_VERSION,
                        found: v,
                    }
                    .into());
                }
                None => {
                    meta.put(&mut wtxn, META_SCHEMA_VERSION_KEY, &WAL_SCHEMA_VERSION)?;
                }
            }
            wtxn.commit()?;
            db
        };

        let event_count = {
            let rtxn = env.read_txn()?;
            log.len(&rtxn)?
        };

        Ok(Self {
            env,
            log,
            event_count: Cell::new(event_count),
        })
    }

    fn write_txn(&self) -> anyhow::Result<heed::RwTxn<'_>> {
        Ok(self.env.write_txn()?)
    }

    fn put_envelope<T>(
        &self,
        wtxn: &mut heed::RwTxn<'_>,
        envelope: &Envelope<T>,
    ) -> anyhow::Result<()>
    where
        T: RkyvToBytes,
    {
        SCRATCH.with(|scratch| -> anyhow::Result<()> {
            let mut bytes = scratch.borrow_mut();
            let mut writer = std::mem::take(&mut *bytes);
            writer.clear();
            let writer = rkyv::api::high::to_bytes_in::<_, RkyvError>(envelope, writer)?;
            self.log.put(wtxn, &envelope.seq, writer.as_slice())?;
            *bytes = writer;
            Ok(())
        })?;
        Ok(())
    }
}

pub(crate) struct ReadSegment {
    env: heed::Env,
    log: LogDb,
    event_count: u64,
}

impl ReadSegment {
    pub(crate) fn open(dir_path: &Path, db_name: &str, map_size: usize) -> anyhow::Result<Self> {
        let env = unsafe {
            let mut options = EnvOpenOptions::new();
            options.map_size(map_size).max_dbs(8);
            options.flags(EnvFlags::READ_ONLY);
            options.open(dir_path)?
        };

        let (log, event_count) = {
            let rtxn = env.read_txn()?;
            let log: LogDb = env
                .open_database(&rtxn, Some(db_name))?
                .ok_or_else(|| anyhow::anyhow!("journal database `{db_name}` not found"))?;
            let meta: MetaDb =
                env.open_database(&rtxn, Some("meta"))?
                    .ok_or(WalError::SchemaMismatch {
                        expected: WAL_SCHEMA_VERSION,
                        found: 0,
                    })?;
            match meta.get(&rtxn, META_SCHEMA_VERSION_KEY)? {
                Some(v) if v == WAL_SCHEMA_VERSION => {}
                Some(v) => {
                    return Err(WalError::SchemaMismatch {
                        expected: WAL_SCHEMA_VERSION,
                        found: v,
                    }
                    .into());
                }
                None => {
                    return Err(WalError::SchemaMismatch {
                        expected: WAL_SCHEMA_VERSION,
                        found: 0,
                    }
                    .into());
                }
            }
            let event_count = log.len(&rtxn)?;
            rtxn.commit()?;
            (log, event_count)
        };

        Ok(Self {
            env,
            log,
            event_count,
        })
    }

    pub(crate) fn visit_entries<F>(
        &self,
        start_seq: u64,
        up_to_seq: Option<u64>,
        on_entry: F,
    ) -> anyhow::Result<()>
    where
        F: FnMut(u64, &[u8]) -> anyhow::Result<()>,
    {
        visit_log_entries(&self.env, &self.log, start_seq, up_to_seq, on_entry)
    }
}

// ---------------------------------------------------------------------------
// SegmentInfo — public metadata about a segment
// ---------------------------------------------------------------------------

pub struct SegmentInfo {
    pub dir_name: String,
    pub path: PathBuf,
    pub start_seq: Option<u64>,
    pub end_seq: Option<u64>,
    pub event_count: u64,
    pub file_size_bytes: u64,
}

fn build_segment_info(
    root_dir: &Path,
    dir_name: &str,
    start_seq: Option<u64>,
    end_seq: Option<u64>,
    event_count: u64,
) -> anyhow::Result<SegmentInfo> {
    let path = root_dir.join(dir_name);
    let file_size_bytes = dir_size(&path)?;

    Ok(SegmentInfo {
        dir_name: dir_name.to_string(),
        path,
        start_seq,
        end_seq,
        event_count,
        file_size_bytes,
    })
}

fn segment_info_from_open(
    root_dir: &Path,
    dir_name: &str,
    seg: &Segment,
) -> anyhow::Result<SegmentInfo> {
    build_segment_info(
        root_dir,
        dir_name,
        first_log_seq(&seg.env, &seg.log)?,
        last_log_seq(&seg.env, &seg.log)?,
        seg.event_count.get(),
    )
}

fn segment_info_static(
    root_dir: &Path,
    dir_name: &str,
    db_name: &str,
    map_size: usize,
) -> anyhow::Result<SegmentInfo> {
    let path = root_dir.join(dir_name);
    let seg = Segment::open(&path, db_name, map_size)?;
    segment_info_from_open(root_dir, dir_name, &seg)
}

fn segment_info_from_read(
    root_dir: &Path,
    dir_name: &str,
    seg: &ReadSegment,
) -> anyhow::Result<SegmentInfo> {
    build_segment_info(
        root_dir,
        dir_name,
        first_log_seq(&seg.env, &seg.log)?,
        last_log_seq(&seg.env, &seg.log)?,
        seg.event_count,
    )
}

pub(crate) fn segment_info_readonly(
    root_dir: &Path,
    dir_name: &str,
    db_name: &str,
    map_size: usize,
) -> anyhow::Result<SegmentInfo> {
    let path = root_dir.join(dir_name);
    let seg = ReadSegment::open(&path, db_name, map_size)?;
    segment_info_from_read(root_dir, dir_name, &seg)
}

fn dir_size(path: &Path) -> anyhow::Result<u64> {
    let mut total = 0u64;
    if path.is_dir() {
        for entry in std::fs::read_dir(path)? {
            let entry = entry?;
            let meta = entry.metadata()?;
            if meta.is_file() {
                total += meta.len();
            }
        }
    }
    Ok(total)
}

// ---------------------------------------------------------------------------
// JournalHandler — segmented WAL
// ---------------------------------------------------------------------------

/// Higher-level durability + recovery handler backed by a segmented LMDB journal.
///
/// Each segment is its own LMDB environment in a subdirectory. Old segments can
/// be deleted wholesale (`rm -rf`) for true disk reclamation.
pub struct JournalHandler {
    root_dir: PathBuf,
    db_name: String,
    config: JournalConfig,
    active: Segment,
    active_dir_name: String,
    rotation_requested: bool,
}

impl JournalScanSource for JournalHandler {
    fn root_dir(&self) -> &Path {
        &self.root_dir
    }

    fn visit_segment_entries<F>(
        &self,
        dir_name: &str,
        start_seq: u64,
        up_to_seq: Option<u64>,
        on_entry: F,
    ) -> anyhow::Result<()>
    where
        F: FnMut(u64, &[u8]) -> anyhow::Result<()>,
    {
        let owned_seg = self.open_segment_if_not_active(dir_name)?;
        let seg = owned_seg.as_ref().unwrap_or(&self.active);
        visit_log_entries(&seg.env, &seg.log, start_seq, up_to_seq, on_entry)
    }
}

impl JournalHandler {
    pub fn open_engine(root_dir: impl AsRef<Path>, config: JournalConfig) -> anyhow::Result<Self> {
        Self::open_raw(root_dir, ENGINE_WAL_DB_NAME, config)
    }

    pub fn open(
        root_dir: impl AsRef<Path>,
        db_name: &str,
        config: JournalConfig,
    ) -> anyhow::Result<Self> {
        let mut handler = Self::open_raw(root_dir, db_name, config)?;
        handler.apply_startup_maintenance()?;
        if handler.config.validate_on_startup {
            handler.validate_segments()?;
        }
        Ok(handler)
    }

    fn open_raw(
        root_dir: impl AsRef<Path>,
        db_name: &str,
        config: JournalConfig,
    ) -> anyhow::Result<Self> {
        let root_dir = root_dir.as_ref().to_path_buf();
        std::fs::create_dir_all(&root_dir)?;

        let dirs = list_segment_dirs(&root_dir)?;

        let (active, active_dir_name) = if dirs.is_empty() {
            let dir_name = segment_dir_name(0);
            let seg_path = root_dir.join(&dir_name);
            info!(segment = %dir_name, "creating initial segment");
            let seg = Segment::open(&seg_path, db_name, config.segment_map_size_bytes)?;
            (seg, dir_name)
        } else {
            // Open the last segment as active
            let last = dirs.last().unwrap().clone();
            let seg_path = root_dir.join(&last);
            info!(segment = %last, "opening last segment as active");
            let seg = Segment::open(&seg_path, db_name, config.segment_map_size_bytes)?;
            (seg, last)
        };

        let handler = Self {
            root_dir,
            db_name: db_name.to_string(),
            config,
            active,
            active_dir_name,
            rotation_requested: false,
        };

        Ok(handler)
    }

    pub fn apply_startup_maintenance(&mut self) -> anyhow::Result<()> {
        if let Some(confirmed_seq) = self.config.prune_before_seq_on_startup {
            let moved = self.prune_segments_before(confirmed_seq)?;
            if !moved.is_empty() {
                info!(
                    moved_segments = moved.len(),
                    confirmed_seq, "pruned sealed segments after startup recovery"
                );
            }
        }

        if self.config.clear_trash_on_startup {
            match clear_trash_dir(&self.root_dir) {
                Ok(0) => {}
                Ok(n) => info!(
                    removed_entries = n,
                    "deleted trashed journal segments after startup recovery"
                ),
                Err(e) => warn!(
                    error = %e,
                    "failed to clear trash directory after startup recovery"
                ),
            }
        }

        Ok(())
    }

    /// Get segment info, using the already-open active segment when the dir matches.
    fn segment_info(&self, dir_name: &str) -> anyhow::Result<SegmentInfo> {
        if dir_name == self.active_dir_name {
            segment_info_from_open(&self.root_dir, dir_name, &self.active)
        } else {
            segment_info_static(
                &self.root_dir,
                dir_name,
                &self.db_name,
                self.config.segment_map_size_bytes,
            )
        }
    }

    /// Open a segment, using the already-open active segment when the dir matches.
    /// Returns None if dir_name is the active segment (caller should use self.active).
    fn open_segment_if_not_active(&self, dir_name: &str) -> anyhow::Result<Option<Segment>> {
        if dir_name == self.active_dir_name {
            Ok(None)
        } else {
            let path = self.root_dir.join(dir_name);
            Ok(Some(Segment::open(
                &path,
                &self.db_name,
                self.config.segment_map_size_bytes,
            )?))
        }
    }

    pub fn write_txn(&self) -> anyhow::Result<heed::RwTxn<'_>> {
        self.active.write_txn()
    }

    pub fn put_envelope<T>(
        &self,
        wtxn: &mut heed::RwTxn<'_>,
        envelope: &Envelope<T>,
    ) -> anyhow::Result<()>
    where
        T: RkyvToBytes,
    {
        self.active.put_envelope(wtxn, envelope)?;
        self.active
            .event_count
            .set(self.active.event_count.get() + 1);
        Ok(())
    }

    pub fn request_rotation(&mut self) {
        self.rotation_requested = true;
    }

    pub fn should_rotate(&self) -> bool {
        if self.rotation_requested {
            return true;
        }
        self.active.event_count.get() >= self.config.max_segment_events
    }

    pub fn rotate(&mut self, expected_start_seq: u64) -> anyhow::Result<()> {
        let dir_name = segment_dir_name(expected_start_seq);
        let seg_path = self.root_dir.join(&dir_name);
        info!(segment = %dir_name, expected_start_seq, "rotating to new segment");
        let new_seg = Segment::open(&seg_path, &self.db_name, self.config.segment_map_size_bytes)?;
        self.active = new_seg;
        self.active_dir_name = dir_name;
        self.rotation_requested = false;
        Ok(())
    }

    /// Recover events from the journal, walking across segment boundaries.
    ///
    /// The tx framing logic is identical to the old single-env approach, but each
    /// segment's read transaction is scoped to that segment's lifetime. Framing
    /// state (current_tx, pending_last) is carried as plain values across segments.
    pub fn recover_from<T, F>(&self, after_seq: Option<u64>, mut on_event: F) -> anyhow::Result<()>
    where
        T: rkyv::Archive,
        ArchivedEnvelopeOf<T>:
            for<'a> rkyv::bytecheck::CheckBytes<rkyv::api::high::HighValidator<'a, RkyvError>>,
        F: for<'a> FnMut(&'a ArchivedEnvelopeOf<T>, bool) -> anyhow::Result<()>,
    {
        let validation = validate_recovery_window::<T, _>(
            self,
            after_seq,
            None,
            ScanStartPolicy::ExactContinuation,
            None,
        )?;
        let Some(validated_up_to_seq) = validation.summary.last_emitted_seq else {
            return Ok(());
        };
        replay_wal::<T, _, _>(
            self,
            after_seq,
            Some(validated_up_to_seq),
            ScanStartPolicy::ExactContinuation,
            |bytes, end_of_batch| {
                let archived = rkyv::api::high::access::<ArchivedEnvelopeOf<T>, RkyvError>(bytes)?;
                on_event(archived, end_of_batch)
            },
        )?;
        Ok(())
    }

    pub(crate) fn validate_events_from_required(
        &self,
        after_seq: Option<u64>,
        required_through_seq: Option<u64>,
    ) -> anyhow::Result<ValidatedScan> {
        validate_recovery_window::<BookEventEnvelope, _>(
            self,
            after_seq,
            None,
            ScanStartPolicy::ExactContinuation,
            required_through_seq,
        )
    }

    pub(crate) fn replay_events_from<F>(
        &self,
        after_seq: Option<u64>,
        up_to_seq: Option<u64>,
        mut on_event: F,
    ) -> anyhow::Result<ScanSummary>
    where
        F: FnMut(&JournalEvent, bool) -> anyhow::Result<()>,
    {
        replay_wal::<BookEventEnvelope, _, _>(
            self,
            after_seq,
            up_to_seq,
            ScanStartPolicy::ExactContinuation,
            |bytes, end_of_batch| {
                let event = decode_journal_event(bytes)?;
                on_event(&event, end_of_batch)
            },
        )
    }

    /// Validate contiguity across all segments.
    pub fn validate_segments(&self) -> anyhow::Result<()> {
        validate_segment_sequence(&self.root_dir, |dir_name| self.segment_info(dir_name))
    }

    pub fn list_segments(&self) -> anyhow::Result<Vec<SegmentInfo>> {
        let dirs = list_segment_dirs(&self.root_dir)?;
        let mut infos = Vec::with_capacity(dirs.len());
        for dir_name in &dirs {
            infos.push(self.segment_info(dir_name)?);
        }
        Ok(infos)
    }

    /// Move sealed segments where `end_seq <= confirmed_seq` into `trash/`.
    /// Never moves the last (active) segment.
    /// Returns the list of moved segment dir names.
    pub fn prune_segments_before(&mut self, confirmed_seq: u64) -> anyhow::Result<Vec<String>> {
        let dirs = list_segment_dirs(&self.root_dir)?;
        if dirs.len() <= 1 {
            return Ok(Vec::new());
        }

        let trash_root = trash_dir(&self.root_dir);
        let mut deleted = Vec::new();
        // Never delete the last directory (the active segment)
        for dir_name in &dirs[..dirs.len() - 1] {
            let info = self.segment_info(dir_name)?;
            if let Some(end_seq) = info.end_seq
                && end_seq <= confirmed_seq
            {
                std::fs::create_dir_all(&trash_root)?;
                let target = trash_root.join(dir_name);
                if target.exists() {
                    warn!(segment = %dir_name, target = %target.display(), "trash target already exists, replacing");
                    if target.is_dir() {
                        std::fs::remove_dir_all(&target)?;
                    } else {
                        std::fs::remove_file(&target)?;
                    }
                }
                info!(segment = %dir_name, end_seq, confirmed_seq, "moving confirmed segment to trash");
                std::fs::rename(&info.path, &target)?;
                deleted.push(dir_name.clone());
            }
        }

        Ok(deleted)
    }

    pub fn into_poller<T, W, B>(
        self,
        poller: EventPoller<RingSlot<T>, B, W::Notifier>,
        wait_strategy: W,
        gate: GatedSequence<W::Notifier>,
        shutdown: Arc<AtomicBool>,
    ) -> JournalPoller<T, W, B>
    where
        W: WaitStrategy,
    {
        JournalPoller::<T, W, B>::new(self, poller, wait_strategy, gate, shutdown)
    }
}

// ---------------------------------------------------------------------------
// JournalPoller
// ---------------------------------------------------------------------------

pub struct JournalPoller<T, W: WaitStrategy, B> {
    journal: JournalHandler,
    poller: EventPoller<RingSlot<T>, B, W::Notifier>,
    wait_strategy: W,
    gate: GatedSequence<W::Notifier>,
    shutdown: Arc<AtomicBool>,
}

impl<T, W: WaitStrategy, B> JournalPoller<T, W, B> {
    pub fn new(
        journal: JournalHandler,
        poller: EventPoller<RingSlot<T>, B, W::Notifier>,
        wait_strategy: W,
        gate: GatedSequence<W::Notifier>,
        shutdown: Arc<AtomicBool>,
    ) -> Self {
        Self {
            journal,
            poller,
            wait_strategy,
            gate,
            shutdown,
        }
    }
}

impl<T, W, B> JournalPoller<T, W, B>
where
    W: WaitStrategy + Send + 'static,
    W::Notifier: Send + 'static,
    T: RkyvToBytes + Send + Sync + 'static,
    B: disrupt_rs::Barrier + 'static,
{
    pub fn poll(self) -> thread::JoinHandle<anyhow::Result<()>> {
        self.spawn(None)
    }

    pub fn poll_pinned(self, core_id: Option<usize>) -> thread::JoinHandle<anyhow::Result<()>> {
        self.spawn(core_id)
    }

    fn spawn(self, core_id: Option<usize>) -> thread::JoinHandle<anyhow::Result<()>> {
        let mut this = self;
        thread::spawn(move || {
            trace!(core_id = ?core_id, "starting wal poller");
            if let Some(id) = core_id
                && !core_affinity::set_for_current(CoreId { id })
            {
                warn!(
                    core_id = id,
                    "failed to pin wal poller thread to requested core"
                );
            }

            crate::metrics_stage!("wal_poller");
            let mut waiter = this.wait_strategy.new_waiter();
            let res = (|| {
                let mut wtxn = this.journal.write_txn()?;
                let mut last_completed_tx_slot: Sequence = 0;

                'poll: loop {
                    match this.poller.poll_wait(&mut waiter) {
                        Ok(mut events) => {
                            let t0 = Instant::now();
                            let events_len = events.len();
                            crate::metric!({ inflight: 1, in_total: events_len as u64 });

                            let mut batch_end_app_seq: Option<u64> = None;
                            for (s, e) in &mut events {
                                let envelope: &Envelope<T> = e;
                                this.journal.put_envelope(&mut wtxn, envelope)?;
                                if envelope.is_tx_end() {
                                    last_completed_tx_slot = s;
                                    batch_end_app_seq = Some(envelope.seq);
                                } else {
                                    batch_end_app_seq = None;
                                }
                            }

                            wtxn.commit()?;
                            this.gate.set(last_completed_tx_slot);

                            // Only rotate when the batch ends on a complete tx boundary.
                            // This guarantees no transaction ever spans two segments.
                            if this.journal.should_rotate()
                                && let Some(last_app_seq) = batch_end_app_seq
                            {
                                let next_seq = last_app_seq.saturating_add(1);
                                this.journal.rotate(next_seq)?;
                            }

                            wtxn = this.journal.write_txn()?;

                            crate::metric!({
                                out_total: 1, // 1 commit per batch
                                inflight: 0,
                                duration_ns: t0.elapsed().as_nanos() as u64
                            });
                        }
                        Err(Polling::NoEvents) => {
                            // The disruptor poller doesn't consult our shutdown flag, so ensure we
                            // can exit cleanly even if no new events are arriving.
                            if this.shutdown.load(Ordering::Acquire) {
                                break 'poll;
                            }
                            continue;
                        }
                        Err(Polling::Shutdown) => break 'poll,
                    }
                }
                Ok::<_, anyhow::Error>(())
            })();

            if let Err(e) = res {
                error!(error = ?e, "journal poller error");
                crate::metric!({ err_total: 1 });
                this.shutdown
                    .store(true, journaler_error_shutdown_ordering());
                return Err(e);
            }

            Ok(())
        })
    }
}

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

    #[test]
    fn journaler_error_shutdown_store_uses_release_ordering() {
        assert!(matches!(
            journaler_error_shutdown_ordering(),
            Ordering::Release
        ));
    }
}
