use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;

use anyhow::{Context, Result};
use flatbuffers::FlatBufferBuilder;

use crate::generated::{
    iam_wal_v1_generated::nexus::wal as fb_iam,
    migration_wal_v1_generated::nexus::wal as fb_migration,
    multipart_wal_v1_generated::nexus::wal as fb_multipart,
    object_wal_v1_generated::nexus::wal as fb_object,
    webhook_event_wal_v1_generated::nexus::wal as fb_webhook_event,
    webhook_wal_v1_generated::nexus::wal as fb_webhook,
};

const FRAME_HEADER_BYTES: usize = 9;
const RECORD_KIND_OBJECT: u8 = 1;
const RECORD_KIND_MULTIPART: u8 = 2;
const RECORD_KIND_IAM: u8 = 3;
const RECORD_KIND_WEBHOOK: u8 = 4;
const RECORD_KIND_WEBHOOK_EVENT: u8 = 5;
const RECORD_KIND_MIGRATION: u8 = 6;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DevicePointer {
    pub device_id: u16,
    pub offset: u64,
    pub len: u32,
    pub checksum_crc32c: u32,
    pub nonce_id: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WalEntryV1 {
    pub seq: u64,
    pub op: String,
    pub bucket: String,
    pub key: String,
    pub etag: Option<String>,
    pub pointer_or_manifest: Option<DevicePointer>,
    pub ts_unix_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultipartPart {
    pub part_number: u32,
    pub etag: String,
    pub pointer: DevicePointer,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultipartManifest {
    pub upload_id: String,
    pub bucket: String,
    pub key: String,
    pub parts: Vec<MultipartPart>,
    pub final_etag: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultipartWalEntryV1 {
    pub seq: u64,
    pub op: String,
    pub upload_id: String,
    pub bucket: String,
    pub key: String,
    pub final_etag: Option<String>,
    pub parts: Vec<MultipartPart>,
    pub ts_unix_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IamWalEntryV1 {
    pub seq: u64,
    pub op: String,
    pub target: String,
    pub payload: String,
    pub ts_unix_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebhookWalEntryV1 {
    pub seq: u64,
    pub op: String,
    pub bucket: String,
    pub rule_id: String,
    pub rule_json: String,
    pub ts_unix_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebhookEventWalEntryV1 {
    pub seq: u64,
    pub op: String,
    pub event_id: String,
    pub rule_id: String,
    pub bucket: String,
    pub key: String,
    pub payload_json: String,
    pub ts_unix_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MigrationWalEntryV1 {
    pub seq: u64,
    pub op: String,
    pub bucket: String,
    pub key: String,
    pub payload: String,
    pub ts_unix_ms: u64,
}

#[derive(Debug, Clone)]
pub struct WalReplayConfig {
    pub wal_path: PathBuf,
    pub dry_run: bool,
}

#[derive(Debug, Clone)]
pub struct WalReplayStats {
    pub wal_path: PathBuf,
    pub dry_run: bool,
    pub object_entries: u64,
    pub multipart_entries: u64,
    pub object_puts: u64,
    pub object_deletes: u64,
    pub multipart_creates: u64,
    pub multipart_parts: u64,
    pub multipart_completes: u64,
    pub multipart_aborts: u64,
    pub iam_entries: u64,
    pub webhook_entries: u64,
    pub webhook_event_entries: u64,
    pub migration_entries: u64,
    pub last_seq: u64,
}

impl WalReplayStats {
    pub fn to_json(&self) -> String {
        format!(
            "{{\"module\":\"WAL_REPLAY\",\"wal_path\":\"{}\",\"dry_run\":{},\"object_entries\":{},\"multipart_entries\":{},\"object_puts\":{},\"object_deletes\":{},\"multipart_creates\":{},\"multipart_parts\":{},\"multipart_completes\":{},\"multipart_aborts\":{},\"iam_entries\":{},\"webhook_entries\":{},\"webhook_event_entries\":{},\"migration_entries\":{},\"last_seq\":{}}}",
            self.wal_path.display(),
            self.dry_run,
            self.object_entries,
            self.multipart_entries,
            self.object_puts,
            self.object_deletes,
            self.multipart_creates,
            self.multipart_parts,
            self.multipart_completes,
            self.multipart_aborts,
            self.iam_entries,
            self.webhook_entries,
            self.webhook_event_entries,
            self.migration_entries,
            self.last_seq,
        )
    }
}

#[derive(Debug, Clone)]
pub struct WalReplayState {
    pub stats: WalReplayStats,
    pub object_index: BTreeMap<String, WalEntryV1>,
    pub multipart_uploads: BTreeMap<String, MultipartWalEntryV1>,
    pub iam_log: Vec<IamWalEntryV1>,
    pub webhook_log: Vec<WebhookWalEntryV1>,
    pub webhook_event_log: Vec<WebhookEventWalEntryV1>,
    pub migration_log: Vec<MigrationWalEntryV1>,
}

#[derive(Debug)]
pub struct WalWriter {
    wal_path: PathBuf,
    file: Mutex<File>,
    next_seq: AtomicU64,
}

impl WalWriter {
    pub fn open<P: AsRef<Path>>(wal_path: P) -> Result<Self> {
        let wal_path = wal_path.as_ref().to_path_buf();
        if let Some(parent) = wal_path.parent() {
            if !parent.as_os_str().is_empty() {
                std::fs::create_dir_all(parent)
                    .with_context(|| format!("failed creating WAL parent {}", parent.display()))?;
            }
        }

        let file = OpenOptions::new()
            .create(true)
            .append(true)
            .read(true)
            .open(&wal_path)
            .with_context(|| format!("failed opening WAL {}", wal_path.display()))?;

        let seed_seq = detect_next_seq(&wal_path).unwrap_or(1);

        Ok(Self {
            wal_path,
            file: Mutex::new(file),
            next_seq: AtomicU64::new(seed_seq),
        })
    }

    pub fn wal_path(&self) -> &Path {
        &self.wal_path
    }

    pub fn next_seq(&self) -> u64 {
        self.next_seq.fetch_add(1, Ordering::AcqRel)
    }

    pub fn append_object_entry(&self, entry: &WalEntryV1) -> Result<()> {
        let mut fbb = FlatBufferBuilder::with_capacity(512);

        let op = fbb.create_string(entry.op.as_str());
        let bucket = fbb.create_string(entry.bucket.as_str());
        let key = fbb.create_string(entry.key.as_str());
        let etag = entry.etag.as_ref().map(|v| fbb.create_string(v.as_str()));
        let pointer = entry.pointer_or_manifest.as_ref().map(|p| {
            fb_object::DevicePointer::create(
                &mut fbb,
                &fb_object::DevicePointerArgs {
                    device_id: p.device_id,
                    offset: p.offset,
                    len: p.len,
                    checksum_crc32c: p.checksum_crc32c,
                    nonce_id: p.nonce_id,
                },
            )
        });

        let record = fb_object::ObjectRecordV1::create(
            &mut fbb,
            &fb_object::ObjectRecordV1Args {
                seq: entry.seq,
                op: Some(op),
                bucket: Some(bucket),
                key: Some(key),
                etag,
                ts_unix_ms: entry.ts_unix_ms,
                pointer,
            },
        );
        fb_object::finish_size_prefixed_object_record_v1_buffer(&mut fbb, record);

        self.append_frame(RECORD_KIND_OBJECT, fbb.finished_data())
    }

    pub fn append_multipart_entry(&self, entry: &MultipartWalEntryV1) -> Result<()> {
        let mut fbb = FlatBufferBuilder::with_capacity(1024);

        let op = fbb.create_string(entry.op.as_str());
        let bucket = fbb.create_string(entry.bucket.as_str());
        let key = fbb.create_string(entry.key.as_str());
        let upload_id = fbb.create_string(entry.upload_id.as_str());
        let final_etag = entry
            .final_etag
            .as_ref()
            .map(|v| fbb.create_string(v.as_str()));

        let mut part_offsets = Vec::with_capacity(entry.parts.len());
        for part in &entry.parts {
            let etag = fbb.create_string(part.etag.as_str());
            let pointer = fb_multipart::DevicePointer::create(
                &mut fbb,
                &fb_multipart::DevicePointerArgs {
                    device_id: part.pointer.device_id,
                    offset: part.pointer.offset,
                    len: part.pointer.len,
                    checksum_crc32c: part.pointer.checksum_crc32c,
                    nonce_id: part.pointer.nonce_id,
                },
            );
            let part_entry = fb_multipart::MultipartPartPointer::create(
                &mut fbb,
                &fb_multipart::MultipartPartPointerArgs {
                    part_number: part.part_number,
                    etag: Some(etag),
                    pointer: Some(pointer),
                },
            );
            part_offsets.push(part_entry);
        }
        let parts = if part_offsets.is_empty() {
            None
        } else {
            Some(fbb.create_vector(&part_offsets))
        };

        let record = fb_multipart::MultipartRecordV1::create(
            &mut fbb,
            &fb_multipart::MultipartRecordV1Args {
                seq: entry.seq,
                op: Some(op),
                bucket: Some(bucket),
                key: Some(key),
                upload_id: Some(upload_id),
                final_etag,
                ts_unix_ms: entry.ts_unix_ms,
                parts,
            },
        );
        fb_multipart::finish_size_prefixed_multipart_record_v1_buffer(&mut fbb, record);

        self.append_frame(RECORD_KIND_MULTIPART, fbb.finished_data())
    }

    pub fn append_iam_entry(&self, entry: &IamWalEntryV1) -> Result<()> {
        let mut fbb = FlatBufferBuilder::with_capacity(512);
        let op = fbb.create_string(entry.op.as_str());
        let target = fbb.create_string(entry.target.as_str());
        let payload = fbb.create_string(entry.payload.as_str());

        let record = fb_iam::IamRecordV1::create(
            &mut fbb,
            &fb_iam::IamRecordV1Args {
                seq: entry.seq,
                op: Some(op),
                target: Some(target),
                payload: Some(payload),
                ts_unix_ms: entry.ts_unix_ms,
            },
        );
        fb_iam::finish_size_prefixed_iam_record_v1_buffer(&mut fbb, record);

        self.append_frame(RECORD_KIND_IAM, fbb.finished_data())
    }

    pub fn append_webhook_entry(&self, entry: &WebhookWalEntryV1) -> Result<()> {
        let mut fbb = FlatBufferBuilder::with_capacity(512);
        let op = fbb.create_string(entry.op.as_str());
        let bucket = fbb.create_string(entry.bucket.as_str());
        let rule_id = fbb.create_string(entry.rule_id.as_str());
        let rule_json = fbb.create_string(entry.rule_json.as_str());

        let record = fb_webhook::WebhookRecordV1::create(
            &mut fbb,
            &fb_webhook::WebhookRecordV1Args {
                seq: entry.seq,
                op: Some(op),
                bucket: Some(bucket),
                rule_id: Some(rule_id),
                rule_json: Some(rule_json),
                ts_unix_ms: entry.ts_unix_ms,
            },
        );
        fb_webhook::finish_size_prefixed_webhook_record_v1_buffer(&mut fbb, record);

        self.append_frame(RECORD_KIND_WEBHOOK, fbb.finished_data())
    }

    pub fn append_webhook_event_entry(&self, entry: &WebhookEventWalEntryV1) -> Result<()> {
        let mut fbb = FlatBufferBuilder::with_capacity(512);
        let op = fbb.create_string(entry.op.as_str());
        let event_id = fbb.create_string(entry.event_id.as_str());
        let rule_id = fbb.create_string(entry.rule_id.as_str());
        let bucket = fbb.create_string(entry.bucket.as_str());
        let key = fbb.create_string(entry.key.as_str());
        let payload_json = fbb.create_string(entry.payload_json.as_str());

        let record = fb_webhook_event::WebhookEventRecordV1::create(
            &mut fbb,
            &fb_webhook_event::WebhookEventRecordV1Args {
                seq: entry.seq,
                op: Some(op),
                event_id: Some(event_id),
                rule_id: Some(rule_id),
                bucket: Some(bucket),
                key: Some(key),
                payload_json: Some(payload_json),
                ts_unix_ms: entry.ts_unix_ms,
            },
        );
        fb_webhook_event::finish_size_prefixed_webhook_event_record_v1_buffer(&mut fbb, record);
        self.append_frame(RECORD_KIND_WEBHOOK_EVENT, fbb.finished_data())
    }

    pub fn append_migration_entry(&self, entry: &MigrationWalEntryV1) -> Result<()> {
        let mut fbb = FlatBufferBuilder::with_capacity(512);
        let op = fbb.create_string(entry.op.as_str());
        let bucket = fbb.create_string(entry.bucket.as_str());
        let key = fbb.create_string(entry.key.as_str());
        let payload = fbb.create_string(entry.payload.as_str());

        let record = fb_migration::MigrationRecordV1::create(
            &mut fbb,
            &fb_migration::MigrationRecordV1Args {
                seq: entry.seq,
                op: Some(op),
                bucket: Some(bucket),
                key: Some(key),
                payload: Some(payload),
                ts_unix_ms: entry.ts_unix_ms,
            },
        );
        fb_migration::finish_size_prefixed_migration_record_v1_buffer(&mut fbb, record);
        self.append_frame(RECORD_KIND_MIGRATION, fbb.finished_data())
    }

    fn append_frame(&self, kind: u8, payload: &[u8]) -> Result<()> {
        let payload_len: u32 = payload
            .len()
            .try_into()
            .context("WAL payload too large to frame")?;
        let payload_crc = crc32c::crc32c(payload);

        let mut header = [0_u8; FRAME_HEADER_BYTES];
        header[0] = kind;
        header[1..5].copy_from_slice(&payload_len.to_le_bytes());
        header[5..9].copy_from_slice(&payload_crc.to_le_bytes());

        let mut file = self
            .file
            .lock()
            .map_err(|_| anyhow::anyhow!("WAL file mutex poisoned"))?;
        file.write_all(&header)
            .context("failed writing WAL frame header")?;
        file.write_all(payload)
            .context("failed writing WAL frame payload")?;
        file.sync_data().context("failed syncing WAL append")?;
        Ok(())
    }
}

pub fn replay_wal(config: WalReplayConfig) -> Result<WalReplayStats> {
    let state = replay_wal_state(config)?;
    Ok(state.stats)
}

pub fn replay_wal_state(config: WalReplayConfig) -> Result<WalReplayState> {
    if !config.wal_path.exists() {
        return Ok(WalReplayState {
            stats: WalReplayStats {
                wal_path: config.wal_path,
                dry_run: config.dry_run,
                object_entries: 0,
                multipart_entries: 0,
                object_puts: 0,
                object_deletes: 0,
                multipart_creates: 0,
                multipart_parts: 0,
                multipart_completes: 0,
                multipart_aborts: 0,
                iam_entries: 0,
                webhook_entries: 0,
                webhook_event_entries: 0,
                migration_entries: 0,
                last_seq: 0,
            },
            object_index: BTreeMap::new(),
            multipart_uploads: BTreeMap::new(),
            iam_log: Vec::new(),
            webhook_log: Vec::new(),
            webhook_event_log: Vec::new(),
            migration_log: Vec::new(),
        });
    }

    let mut file = File::open(&config.wal_path)
        .with_context(|| format!("failed opening WAL {}", config.wal_path.display()))?;

    let mut object_index = BTreeMap::<String, WalEntryV1>::new();
    let mut multipart_uploads = BTreeMap::<String, MultipartWalEntryV1>::new();
    let mut stats = WalReplayStats {
        wal_path: config.wal_path.clone(),
        dry_run: config.dry_run,
        object_entries: 0,
        multipart_entries: 0,
        object_puts: 0,
        object_deletes: 0,
        multipart_creates: 0,
        multipart_parts: 0,
        multipart_completes: 0,
        multipart_aborts: 0,
        iam_entries: 0,
        webhook_entries: 0,
        webhook_event_entries: 0,
        migration_entries: 0,
        last_seq: 0,
    };
    let mut iam_log = Vec::<IamWalEntryV1>::new();
    let mut webhook_log = Vec::<WebhookWalEntryV1>::new();
    let mut webhook_event_log = Vec::<WebhookEventWalEntryV1>::new();
    let mut migration_log = Vec::<MigrationWalEntryV1>::new();

    loop {
        let mut header = [0_u8; FRAME_HEADER_BYTES];
        match file.read_exact(&mut header) {
            Ok(()) => {}
            Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => break,
            Err(err) => return Err(err).context("failed reading WAL frame header"),
        }

        let kind = header[0];
        let payload_len = u32::from_le_bytes(header[1..5].try_into().expect("header len"));
        let payload_crc = u32::from_le_bytes(header[5..9].try_into().expect("header crc"));

        // Raw block devices may expose a zero-initialized tail. Treat a zero header
        // as end-of-log rather than corruption.
        if kind == 0 && payload_len == 0 && payload_crc == 0 {
            break;
        }

        let mut payload = vec![0_u8; payload_len as usize];
        file.read_exact(&mut payload)
            .context("failed reading WAL frame payload")?;

        let actual_crc = crc32c::crc32c(&payload);
        if actual_crc != payload_crc {
            anyhow::bail!(
                "WAL CRC mismatch at seq {}: expected {}, got {}",
                stats.last_seq,
                payload_crc,
                actual_crc
            );
        }

        match kind {
            RECORD_KIND_OBJECT => {
                stats.object_entries += 1;
                let record = fb_object::size_prefixed_root_as_object_record_v1(&payload)
                    .map_err(|err| anyhow::anyhow!("invalid object WAL frame: {err}"))?;
                let entry = decode_object_record(record)?;
                enforce_monotonic_seq(stats.last_seq, entry.seq)?;
                stats.last_seq = entry.seq;

                match entry.op.as_str() {
                    "put" => {
                        stats.object_puts += 1;
                        if !config.dry_run {
                            object_index.insert(format!("{}/{}", entry.bucket, entry.key), entry);
                        }
                    }
                    "delete" => {
                        stats.object_deletes += 1;
                        if !config.dry_run {
                            object_index.remove(&format!("{}/{}", entry.bucket, entry.key));
                        }
                    }
                    other => anyhow::bail!("unsupported object WAL op: {other}"),
                }
            }
            RECORD_KIND_MULTIPART => {
                stats.multipart_entries += 1;
                let record = fb_multipart::size_prefixed_root_as_multipart_record_v1(&payload)
                    .map_err(|err| anyhow::anyhow!("invalid multipart WAL frame: {err}"))?;
                let entry = decode_multipart_record(record)?;
                enforce_monotonic_seq(stats.last_seq, entry.seq)?;
                stats.last_seq = entry.seq;

                match entry.op.as_str() {
                    "create" => {
                        stats.multipart_creates += 1;
                        if !config.dry_run {
                            multipart_uploads.insert(entry.upload_id.clone(), entry);
                        }
                    }
                    "upload_part" => {
                        stats.multipart_parts += 1;
                        if !config.dry_run {
                            let state = multipart_uploads
                                .entry(entry.upload_id.clone())
                                .or_insert_with(|| MultipartWalEntryV1 {
                                    seq: entry.seq,
                                    op: "create".to_string(),
                                    upload_id: entry.upload_id.clone(),
                                    bucket: entry.bucket.clone(),
                                    key: entry.key.clone(),
                                    final_etag: None,
                                    parts: Vec::new(),
                                    ts_unix_ms: entry.ts_unix_ms,
                                });
                            for incoming in entry.parts {
                                if let Some(existing) = state
                                    .parts
                                    .iter_mut()
                                    .find(|part| part.part_number == incoming.part_number)
                                {
                                    *existing = incoming;
                                } else {
                                    state.parts.push(incoming);
                                }
                            }
                            state.parts.sort_by_key(|part| part.part_number);
                            state.seq = entry.seq;
                            state.ts_unix_ms = entry.ts_unix_ms;
                        }
                    }
                    "complete" => {
                        stats.multipart_completes += 1;
                        if !config.dry_run {
                            multipart_uploads.insert(entry.upload_id.clone(), entry);
                        }
                    }
                    "abort" => {
                        stats.multipart_aborts += 1;
                        if !config.dry_run {
                            multipart_uploads.remove(&entry.upload_id);
                        }
                    }
                    other => anyhow::bail!("unsupported multipart WAL op: {other}"),
                }
            }
            RECORD_KIND_IAM => {
                stats.iam_entries += 1;
                let record = fb_iam::size_prefixed_root_as_iam_record_v1(&payload)
                    .map_err(|err| anyhow::anyhow!("invalid IAM WAL frame: {err}"))?;
                let entry = decode_iam_record(record)?;
                enforce_monotonic_seq(stats.last_seq, entry.seq)?;
                stats.last_seq = entry.seq;
                if !config.dry_run {
                    iam_log.push(entry);
                }
            }
            RECORD_KIND_WEBHOOK => {
                stats.webhook_entries += 1;
                let record = fb_webhook::size_prefixed_root_as_webhook_record_v1(&payload)
                    .map_err(|err| anyhow::anyhow!("invalid webhook WAL frame: {err}"))?;
                let entry = decode_webhook_record(record)?;
                enforce_monotonic_seq(stats.last_seq, entry.seq)?;
                stats.last_seq = entry.seq;
                if !config.dry_run {
                    webhook_log.push(entry);
                }
            }
            RECORD_KIND_WEBHOOK_EVENT => {
                stats.webhook_event_entries += 1;
                let record =
                    fb_webhook_event::size_prefixed_root_as_webhook_event_record_v1(&payload)
                        .map_err(|err| {
                            anyhow::anyhow!("invalid webhook event WAL frame: {err}")
                        })?;
                let entry = decode_webhook_event_record(record)?;
                enforce_monotonic_seq(stats.last_seq, entry.seq)?;
                stats.last_seq = entry.seq;
                if !config.dry_run {
                    webhook_event_log.push(entry);
                }
            }
            RECORD_KIND_MIGRATION => {
                stats.migration_entries += 1;
                let record = fb_migration::size_prefixed_root_as_migration_record_v1(&payload)
                    .map_err(|err| anyhow::anyhow!("invalid migration WAL frame: {err}"))?;
                let entry = decode_migration_record(record)?;
                enforce_monotonic_seq(stats.last_seq, entry.seq)?;
                stats.last_seq = entry.seq;
                if !config.dry_run {
                    migration_log.push(entry);
                }
            }
            other => anyhow::bail!("unknown WAL record kind: {other}"),
        }
    }

    Ok(WalReplayState {
        stats,
        object_index,
        multipart_uploads,
        iam_log,
        webhook_log,
        webhook_event_log,
        migration_log,
    })
}

fn decode_object_record(record: fb_object::ObjectRecordV1<'_>) -> Result<WalEntryV1> {
    let op = record
        .op()
        .context("object WAL record missing op")?
        .to_string();
    let bucket = record
        .bucket()
        .context("object WAL record missing bucket")?
        .to_string();
    let key = record
        .key()
        .context("object WAL record missing key")?
        .to_string();

    let pointer_or_manifest = record.pointer().map(|ptr| DevicePointer {
        device_id: ptr.device_id(),
        offset: ptr.offset(),
        len: ptr.len(),
        checksum_crc32c: ptr.checksum_crc32c(),
        nonce_id: ptr.nonce_id(),
    });

    Ok(WalEntryV1 {
        seq: record.seq(),
        op,
        bucket,
        key,
        etag: record.etag().map(ToString::to_string),
        pointer_or_manifest,
        ts_unix_ms: record.ts_unix_ms(),
    })
}

fn decode_multipart_record(
    record: fb_multipart::MultipartRecordV1<'_>,
) -> Result<MultipartWalEntryV1> {
    let op = record
        .op()
        .context("multipart WAL record missing op")?
        .to_string();
    let bucket = record
        .bucket()
        .context("multipart WAL record missing bucket")?
        .to_string();
    let key = record
        .key()
        .context("multipart WAL record missing key")?
        .to_string();
    let upload_id = record
        .upload_id()
        .context("multipart WAL record missing upload_id")?
        .to_string();

    let mut parts = Vec::new();
    if let Some(list) = record.parts() {
        for part in list {
            let pointer = part
                .pointer()
                .context("multipart WAL part missing pointer")?;
            parts.push(MultipartPart {
                part_number: part.part_number(),
                etag: part.etag().unwrap_or_default().to_string(),
                pointer: DevicePointer {
                    device_id: pointer.device_id(),
                    offset: pointer.offset(),
                    len: pointer.len(),
                    checksum_crc32c: pointer.checksum_crc32c(),
                    nonce_id: pointer.nonce_id(),
                },
            });
        }
    }

    Ok(MultipartWalEntryV1 {
        seq: record.seq(),
        op,
        upload_id,
        bucket,
        key,
        final_etag: record.final_etag().map(ToString::to_string),
        parts,
        ts_unix_ms: record.ts_unix_ms(),
    })
}

fn decode_iam_record(record: fb_iam::IamRecordV1<'_>) -> Result<IamWalEntryV1> {
    Ok(IamWalEntryV1 {
        seq: record.seq(),
        op: record
            .op()
            .context("iam WAL record missing op")?
            .to_string(),
        target: record
            .target()
            .context("iam WAL record missing target")?
            .to_string(),
        payload: record
            .payload()
            .context("iam WAL record missing payload")?
            .to_string(),
        ts_unix_ms: record.ts_unix_ms(),
    })
}

fn decode_webhook_record(record: fb_webhook::WebhookRecordV1<'_>) -> Result<WebhookWalEntryV1> {
    Ok(WebhookWalEntryV1 {
        seq: record.seq(),
        op: record
            .op()
            .context("webhook WAL record missing op")?
            .to_string(),
        bucket: record
            .bucket()
            .context("webhook WAL record missing bucket")?
            .to_string(),
        rule_id: record
            .rule_id()
            .context("webhook WAL record missing rule_id")?
            .to_string(),
        rule_json: record
            .rule_json()
            .context("webhook WAL record missing rule_json")?
            .to_string(),
        ts_unix_ms: record.ts_unix_ms(),
    })
}

fn decode_webhook_event_record(
    record: fb_webhook_event::WebhookEventRecordV1<'_>,
) -> Result<WebhookEventWalEntryV1> {
    Ok(WebhookEventWalEntryV1 {
        seq: record.seq(),
        op: record
            .op()
            .context("webhook event WAL record missing op")?
            .to_string(),
        event_id: record
            .event_id()
            .context("webhook event WAL record missing event_id")?
            .to_string(),
        rule_id: record
            .rule_id()
            .context("webhook event WAL record missing rule_id")?
            .to_string(),
        bucket: record
            .bucket()
            .context("webhook event WAL record missing bucket")?
            .to_string(),
        key: record
            .key()
            .context("webhook event WAL record missing key")?
            .to_string(),
        payload_json: record
            .payload_json()
            .context("webhook event WAL record missing payload_json")?
            .to_string(),
        ts_unix_ms: record.ts_unix_ms(),
    })
}

fn decode_migration_record(record: fb_migration::MigrationRecordV1<'_>) -> Result<MigrationWalEntryV1> {
    Ok(MigrationWalEntryV1 {
        seq: record.seq(),
        op: record
            .op()
            .context("migration WAL record missing op")?
            .to_string(),
        bucket: record
            .bucket()
            .context("migration WAL record missing bucket")?
            .to_string(),
        key: record
            .key()
            .context("migration WAL record missing key")?
            .to_string(),
        payload: record
            .payload()
            .context("migration WAL record missing payload")?
            .to_string(),
        ts_unix_ms: record.ts_unix_ms(),
    })
}

fn detect_next_seq(path: &Path) -> Option<u64> {
    let state = replay_wal_state(WalReplayConfig {
        wal_path: path.to_path_buf(),
        dry_run: true,
    })
    .ok()?;
    Some(state.stats.last_seq.saturating_add(1).max(1))
}

fn enforce_monotonic_seq(previous: u64, next: u64) -> Result<()> {
    if next <= previous {
        anyhow::bail!(
            "WAL sequence monotonicity violation: previous={}, next={}",
            previous,
            next
        );
    }
    Ok(())
}

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

    #[test]
    fn wal_round_trip_object_and_multipart() {
        let wal_path = unique_temp_path("wal-roundtrip");
        let writer = WalWriter::open(&wal_path).expect("open WAL writer");

        let seq1 = writer.next_seq();
        writer
            .append_object_entry(&WalEntryV1 {
                seq: seq1,
                op: "put".to_string(),
                bucket: "b".to_string(),
                key: "k".to_string(),
                etag: Some("etag-1".to_string()),
                pointer_or_manifest: Some(DevicePointer {
                    device_id: 1,
                    offset: 8192,
                    len: 4096,
                    checksum_crc32c: 123,
                    nonce_id: 77,
                }),
                ts_unix_ms: 1,
            })
            .expect("append object entry");

        let seq2 = writer.next_seq();
        writer
            .append_multipart_entry(&MultipartWalEntryV1 {
                seq: seq2,
                op: "create".to_string(),
                upload_id: "u1".to_string(),
                bucket: "b".to_string(),
                key: "k2".to_string(),
                final_etag: None,
                parts: Vec::new(),
                ts_unix_ms: 2,
            })
            .expect("append multipart entry");

        let replay = replay_wal_state(WalReplayConfig {
            wal_path: wal_path.clone(),
            dry_run: false,
        })
        .expect("replay should succeed");

        assert_eq!(replay.stats.object_entries, 1);
        assert_eq!(replay.stats.multipart_entries, 1);
        assert_eq!(replay.stats.last_seq, seq2);
        assert!(replay.object_index.contains_key("b/k"));
        assert!(replay.multipart_uploads.contains_key("u1"));

        let _ = std::fs::remove_file(wal_path);
    }

    #[test]
    fn wal_replay_rejects_non_monotonic_sequence() {
        let wal_path = unique_temp_path("wal-bad-seq");
        let writer = WalWriter::open(&wal_path).expect("open WAL writer");

        writer
            .append_object_entry(&WalEntryV1 {
                seq: 3,
                op: "put".to_string(),
                bucket: "b".to_string(),
                key: "k1".to_string(),
                etag: None,
                pointer_or_manifest: None,
                ts_unix_ms: 1,
            })
            .expect("append seq3");
        writer
            .append_object_entry(&WalEntryV1 {
                seq: 2,
                op: "put".to_string(),
                bucket: "b".to_string(),
                key: "k2".to_string(),
                etag: None,
                pointer_or_manifest: None,
                ts_unix_ms: 2,
            })
            .expect("append seq2");

        let err = replay_wal(WalReplayConfig {
            wal_path: wal_path.clone(),
            dry_run: false,
        })
        .expect_err("expected monotonicity violation");
        assert!(err.to_string().contains("monotonicity"));

        let _ = std::fs::remove_file(wal_path);
    }

    fn unique_temp_path(prefix: &str) -> PathBuf {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("valid system time")
            .as_nanos();
        std::env::temp_dir().join(format!("{prefix}-{}-{now}.wal", std::process::id()))
    }
}
