use std::fmt;
use std::io::{self, Read, Seek, SeekFrom, Write};

use crate::{crc32c, ondisk, FormatPlan};

const MAX_MATERIALIZED_ALLOCATION_GROUPS: u32 = 4096;
const MAX_MATERIALIZED_BYTES: u64 = 64 << 20;
const MAX_MANIFEST_EXTENTS: usize = 100_000;
const MAX_MANIFEST_EXTENT_LEN: u64 = 4 << 30;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MetadataExtent {
    pub offset: u64,
    pub bytes: Vec<u8>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ZeroExtent {
    pub offset: u64,
    pub len: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MetadataImage {
    extents: Vec<MetadataExtent>,
    zero_extents: Vec<ZeroExtent>,
}

#[derive(Debug)]
pub enum ImageError {
    TooManyAllocationGroups { count: u32, max: u32 },
    TooManyMaterializedBytes { bytes: u64, max: u64 },
    InvalidPlan(&'static str),
    Io(io::Error),
}

impl fmt::Display for ImageError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooManyAllocationGroups { count, max } => {
                write!(
                    f,
                    "materialized metadata image would contain {count} allocation groups, maximum is {max}"
                )
            }
            Self::TooManyMaterializedBytes { bytes, max } => {
                write!(
                    f,
                    "materialized metadata image would contain {bytes} bytes, maximum is {max}"
                )
            }
            Self::InvalidPlan(reason) => write!(f, "invalid format plan: {reason}"),
            Self::Io(err) => write!(f, "{err}"),
        }
    }
}

impl std::error::Error for ImageError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(err) => Some(err),
            _ => None,
        }
    }
}

impl From<io::Error> for ImageError {
    fn from(err: io::Error) -> Self {
        Self::Io(err)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MetadataExtentSummary {
    pub offset: u64,
    pub len: u64,
    pub crc32c: u32,
    pub magic: [u8; 4],
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ManifestParseError {
    MissingHeader,
    InvalidHeader(String),
    InvalidFieldCount {
        line: usize,
        count: usize,
    },
    InvalidInteger {
        line: usize,
        field: &'static str,
        value: String,
    },
    InvalidMagic {
        line: usize,
        value: String,
    },
    TooManyExtents {
        max: usize,
    },
    ExtentTooLarge {
        line: usize,
        len: u64,
        max: u64,
    },
}

impl fmt::Display for ManifestParseError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::MissingHeader => write!(f, "manifest is missing a header"),
            Self::InvalidHeader(header) => write!(f, "invalid manifest header: {header}"),
            Self::InvalidFieldCount { line, count } => {
                write!(f, "line {line} has {count} fields, expected 4")
            }
            Self::InvalidInteger { line, field, value } => {
                write!(f, "line {line} has invalid {field}: {value}")
            }
            Self::InvalidMagic { line, value } => {
                write!(f, "line {line} has invalid magic: {value}")
            }
            Self::TooManyExtents { max } => write!(f, "manifest has more than {max} extents"),
            Self::ExtentTooLarge { line, len, max } => {
                write!(f, "line {line} extent length {len} exceeds maximum {max}")
            }
        }
    }
}

impl std::error::Error for ManifestParseError {}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ManifestDiff {
    Missing {
        expected: MetadataExtentSummary,
    },
    Extra {
        actual: MetadataExtentSummary,
    },
    Mismatch {
        expected: MetadataExtentSummary,
        actual: MetadataExtentSummary,
    },
}

impl MetadataImage {
    /// Build a bounded, materialized metadata image for a validated format
    /// plan.
    ///
    /// The returned image is sparse: initialized metadata is represented as
    /// byte extents, while large zeroed regions are represented separately as
    /// [`ZeroExtent`] values. For very large filesystems, this returns
    /// [`ImageError::TooManyAllocationGroups`] or
    /// [`ImageError::TooManyMaterializedBytes`] instead of materializing
    /// excessive metadata. Use
    /// [`write_plan_to`] to stream metadata directly to a target without this
    /// allocation cap.
    ///
    /// ```rust
    /// use mkfs_xfs::image::MetadataImage;
    /// use mkfs_xfs::{FormatOptions, FormatPlan};
    ///
    /// let uuid = [0_u8; 16];
    /// let plan = FormatPlan::new(FormatOptions::new(1 << 30, uuid))?;
    /// let image = MetadataImage::from_plan(&plan)?;
    ///
    /// assert!(!image.extents().is_empty());
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn from_plan(plan: &FormatPlan) -> Result<Self, ImageError> {
        if plan.ag_count > MAX_MATERIALIZED_ALLOCATION_GROUPS {
            return Err(ImageError::TooManyAllocationGroups {
                count: plan.ag_count,
                max: MAX_MATERIALIZED_ALLOCATION_GROUPS,
            });
        }
        let estimated_bytes = estimate_materialized_bytes(plan)?;
        if estimated_bytes > MAX_MATERIALIZED_BYTES {
            return Err(ImageError::TooManyMaterializedBytes {
                bytes: estimated_bytes,
                max: MAX_MATERIALIZED_BYTES,
            });
        }

        let mut builder = MetadataImageBuilder::new();
        MetadataLayoutEmitter::new(plan).emit_initial_metadata(&mut builder)?;
        Ok(builder.finish())
    }

    /// Initialized metadata byte extents, sorted by device offset.
    pub fn extents(&self) -> &[MetadataExtent] {
        &self.extents
    }

    /// Explicit zero ranges that must be written before initialized extents.
    pub fn zero_extents(&self) -> &[ZeroExtent] {
        &self.zero_extents
    }

    pub fn into_extents(self) -> Vec<MetadataExtent> {
        self.extents
    }

    pub fn extent_summaries(&self) -> Vec<MetadataExtentSummary> {
        let mut summaries: Vec<_> = self
            .extents
            .iter()
            .map(|extent| MetadataExtentSummary {
                offset: extent.offset,
                len: extent.bytes.len() as u64,
                crc32c: crc32c::checksum(&extent.bytes),
                magic: first_four(&extent.bytes),
            })
            .collect();
        summaries.extend(
            self.zero_extents
                .iter()
                .map(|extent| MetadataExtentSummary {
                    offset: extent.offset,
                    len: extent.len,
                    crc32c: checksum_zeroes(extent.len),
                    magic: [0; 4],
                }),
        );
        summaries.sort_by_key(|summary| summary.offset);
        summaries
    }

    pub fn manifest(&self) -> String {
        let mut out = String::from("offset,len,crc32c,magic\n");
        for summary in self.extent_summaries() {
            out.push_str(&format!(
                "{},{},{:08x},{}\n",
                summary.offset,
                summary.len,
                summary.crc32c,
                MagicDisplay(summary.magic)
            ));
        }
        out
    }

    pub fn parse_manifest(input: &str) -> Result<Vec<MetadataExtentSummary>, ManifestParseError> {
        parse_manifest(input)
    }

    pub fn diff_manifest(&self, actual: &[MetadataExtentSummary]) -> Vec<ManifestDiff> {
        diff_manifest(&self.extent_summaries(), actual)
    }

    pub fn summarize_reader_extents<R: Read + Seek>(
        &self,
        reader: &mut R,
    ) -> io::Result<Vec<MetadataExtentSummary>> {
        summarize_reader_extents(reader, &self.extent_summaries())
    }

    pub fn end_offset(&self) -> u64 {
        let data_end = self
            .extents
            .iter()
            .map(|extent| extent.offset + extent.bytes.len() as u64)
            .max()
            .unwrap_or(0);
        let zero_end = self
            .zero_extents
            .iter()
            .map(|extent| extent.offset + extent.len)
            .max()
            .unwrap_or(0);
        data_end.max(zero_end)
    }

    /// Write the sparse metadata image to a seekable target.
    ///
    /// Zero extents are written first, followed by initialized metadata
    /// extents.
    pub fn write_to<W: Seek + Write>(&self, writer: &mut W) -> io::Result<()> {
        for extent in &self.zero_extents {
            write_zeroes(writer, extent.offset, extent.len)?;
        }
        for extent in &self.extents {
            writer.seek(SeekFrom::Start(extent.offset))?;
            writer.write_all(&extent.bytes)?;
        }
        Ok(())
    }

    pub fn apply_to_slice(&self, image: &mut [u8]) -> bool {
        for extent in &self.zero_extents {
            let start = match usize::try_from(extent.offset) {
                Ok(start) => start,
                Err(_) => return false,
            };
            let len = match usize::try_from(extent.len) {
                Ok(len) => len,
                Err(_) => return false,
            };
            let Some(end) = start.checked_add(len) else {
                return false;
            };
            if end > image.len() {
                return false;
            }
            image[start..end].fill(0);
        }
        for extent in &self.extents {
            let start = match usize::try_from(extent.offset) {
                Ok(start) => start,
                Err(_) => return false,
            };
            let Some(end) = start.checked_add(extent.bytes.len()) else {
                return false;
            };
            if end > image.len() {
                return false;
            }
            image[start..end].copy_from_slice(&extent.bytes);
        }
        true
    }
}

/// Stream the metadata for `plan` directly to a seekable writer.
///
/// Unlike [`MetadataImage::from_plan`], this does not materialize per-AG
/// metadata in memory. Working memory is bounded to the largest encoded
/// metadata block or initial inode chunk.
pub fn write_plan_to<W: Seek + Write>(plan: &FormatPlan, writer: &mut W) -> Result<(), ImageError> {
    let mut sink = MetadataWriter { writer };
    MetadataLayoutEmitter::new(plan).emit_initial_metadata(&mut sink)
}

pub fn diff_manifest(
    expected: &[MetadataExtentSummary],
    actual: &[MetadataExtentSummary],
) -> Vec<ManifestDiff> {
    let mut diffs = Vec::new();
    let mut expected_idx = 0;
    let mut actual_idx = 0;

    while expected_idx < expected.len() || actual_idx < actual.len() {
        match (expected.get(expected_idx), actual.get(actual_idx)) {
            (Some(&e), Some(&a)) if e.offset == a.offset => {
                if e != a {
                    diffs.push(ManifestDiff::Mismatch {
                        expected: e,
                        actual: a,
                    });
                }
                expected_idx += 1;
                actual_idx += 1;
            }
            (Some(&e), Some(&a)) if e.offset < a.offset => {
                diffs.push(ManifestDiff::Missing { expected: e });
                expected_idx += 1;
            }
            (Some(_), Some(&a)) => {
                diffs.push(ManifestDiff::Extra { actual: a });
                actual_idx += 1;
            }
            (Some(&e), None) => {
                diffs.push(ManifestDiff::Missing { expected: e });
                expected_idx += 1;
            }
            (None, Some(&a)) => {
                diffs.push(ManifestDiff::Extra { actual: a });
                actual_idx += 1;
            }
            (None, None) => break,
        }
    }

    diffs
}

pub fn summarize_reader_extents<R: Read + Seek>(
    reader: &mut R,
    layout: &[MetadataExtentSummary],
) -> io::Result<Vec<MetadataExtentSummary>> {
    validate_extent_count(layout.len())?;
    let mut summaries = Vec::with_capacity(layout.len());

    for extent in layout {
        validate_extent_len(extent.len)?;
        reader.seek(SeekFrom::Start(extent.offset))?;
        let mut hasher = crc32c::Hasher::new();
        let magic = hash_reader_extent(reader, extent.len, &mut hasher)?;
        summaries.push(MetadataExtentSummary {
            offset: extent.offset,
            len: extent.len,
            crc32c: hasher.finalize(),
            magic,
        });
    }

    Ok(summaries)
}

pub fn parse_manifest(input: &str) -> Result<Vec<MetadataExtentSummary>, ManifestParseError> {
    let mut lines = input.lines();
    let Some(header) = lines.next() else {
        return Err(ManifestParseError::MissingHeader);
    };
    if header != "offset,len,crc32c,magic" {
        return Err(ManifestParseError::InvalidHeader(header.to_string()));
    }

    let mut summaries = Vec::new();
    for (idx, line) in lines.enumerate() {
        let line_no = idx + 2;
        if line.is_empty() {
            continue;
        }
        if summaries.len() == MAX_MANIFEST_EXTENTS {
            return Err(ManifestParseError::TooManyExtents {
                max: MAX_MANIFEST_EXTENTS,
            });
        }

        let mut fields = line.split(',');
        let Some(offset_field) = fields.next() else {
            return Err(ManifestParseError::InvalidFieldCount {
                line: line_no,
                count: 0,
            });
        };
        let Some(len_field) = fields.next() else {
            return Err(ManifestParseError::InvalidFieldCount {
                line: line_no,
                count: 1,
            });
        };
        let Some(crc32c_field) = fields.next() else {
            return Err(ManifestParseError::InvalidFieldCount {
                line: line_no,
                count: 2,
            });
        };
        let Some(magic_field) = fields.next() else {
            return Err(ManifestParseError::InvalidFieldCount {
                line: line_no,
                count: 3,
            });
        };
        if fields.next().is_some() {
            return Err(ManifestParseError::InvalidFieldCount {
                line: line_no,
                count: line.split(',').count(),
            });
        }

        let offset = parse_u64_field(offset_field, line_no, "offset")?;
        let len = parse_u64_field(len_field, line_no, "len")?;
        if len > MAX_MANIFEST_EXTENT_LEN {
            return Err(ManifestParseError::ExtentTooLarge {
                line: line_no,
                len,
                max: MAX_MANIFEST_EXTENT_LEN,
            });
        }
        let crc32c = u32::from_str_radix(crc32c_field, 16).map_err(|_| {
            ManifestParseError::InvalidInteger {
                line: line_no,
                field: "crc32c",
                value: crc32c_field.to_string(),
            }
        })?;
        let magic = parse_magic(magic_field).ok_or_else(|| ManifestParseError::InvalidMagic {
            line: line_no,
            value: magic_field.to_string(),
        })?;

        summaries.push(MetadataExtentSummary {
            offset,
            len,
            crc32c,
            magic,
        });
    }

    Ok(summaries)
}

fn validate_extent_count(count: usize) -> io::Result<()> {
    if count <= MAX_MANIFEST_EXTENTS {
        Ok(())
    } else {
        Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("manifest has more than {MAX_MANIFEST_EXTENTS} extents"),
        ))
    }
}

fn validate_extent_len(len: u64) -> io::Result<()> {
    if len <= MAX_MANIFEST_EXTENT_LEN {
        Ok(())
    } else {
        Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("manifest extent length {len} exceeds maximum {MAX_MANIFEST_EXTENT_LEN}"),
        ))
    }
}

fn parse_u64_field(
    value: &str,
    line: usize,
    field: &'static str,
) -> Result<u64, ManifestParseError> {
    value
        .parse()
        .map_err(|_| ManifestParseError::InvalidInteger {
            line,
            field,
            value: value.to_string(),
        })
}

fn parse_magic(value: &str) -> Option<[u8; 4]> {
    let bytes = value.as_bytes();
    let mut out = [0_u8; 4];
    let mut out_idx = 0;
    let mut idx = 0;

    while idx < bytes.len() {
        if out_idx == out.len() {
            return None;
        }

        if bytes[idx] == b'\\' {
            if bytes.get(idx + 1) != Some(&b'x') {
                return None;
            }
            let hi = hex_nibble(*bytes.get(idx + 2)?)?;
            let lo = hex_nibble(*bytes.get(idx + 3)?)?;
            out[out_idx] = (hi << 4) | lo;
            idx += 4;
        } else {
            out[out_idx] = bytes[idx];
            idx += 1;
        }
        out_idx += 1;
    }

    (out_idx == out.len()).then_some(out)
}

fn hex_nibble(byte: u8) -> Option<u8> {
    match byte {
        b'0'..=b'9' => Some(byte - b'0'),
        b'a'..=b'f' => Some(byte - b'a' + 10),
        b'A'..=b'F' => Some(byte - b'A' + 10),
        _ => None,
    }
}

fn first_four(bytes: &[u8]) -> [u8; 4] {
    let mut magic = [0_u8; 4];
    let len = bytes.len().min(magic.len());
    magic[..len].copy_from_slice(&bytes[..len]);
    magic
}

fn hash_reader_extent<R: Read>(
    reader: &mut R,
    len: u64,
    hasher: &mut crc32c::Hasher,
) -> io::Result<[u8; 4]> {
    let mut magic = [0_u8; 4];
    let magic_len = len.min(4) as usize;
    if magic_len > 0 {
        reader.read_exact(&mut magic[..magic_len])?;
        hasher.update(&magic[..magic_len]);
    }

    let mut remaining = len - magic_len as u64;
    let mut buffer = [0_u8; 64 * 1024];
    while remaining > 0 {
        let read_len = remaining.min(buffer.len() as u64) as usize;
        reader.read_exact(&mut buffer[..read_len])?;
        hasher.update(&buffer[..read_len]);
        remaining -= read_len as u64;
    }

    Ok(magic)
}

fn checksum_zeroes(len: u64) -> u32 {
    let zeroes = [0_u8; 64 * 1024];
    let mut remaining = len;
    let mut hasher = crc32c::Hasher::new();
    while remaining > 0 {
        let chunk = remaining.min(zeroes.len() as u64) as usize;
        hasher.update(&zeroes[..chunk]);
        remaining -= chunk as u64;
    }
    hasher.finalize()
}

fn write_zeroes<W: Seek + Write>(writer: &mut W, offset: u64, len: u64) -> io::Result<()> {
    let zeroes = [0_u8; 64 * 1024];
    let mut remaining = len;
    writer.seek(SeekFrom::Start(offset))?;
    while remaining > 0 {
        let chunk = remaining.min(zeroes.len() as u64) as usize;
        writer.write_all(&zeroes[..chunk])?;
        remaining -= chunk as u64;
    }
    Ok(())
}

fn estimate_materialized_bytes(plan: &FormatPlan) -> Result<u64, ImageError> {
    let mut total = 0_u64;

    let inode_chunk_bytes =
        u64::from(plan.inode_size).checked_mul(u64::from(ondisk::XFS_INODES_PER_CHUNK));
    add_materialized_bytes(
        &mut total,
        inode_chunk_bytes.ok_or(ImageError::InvalidPlan(
            "initial inode chunk size overflows u64",
        ))?,
    )?;

    let log_record_size = ondisk::initial_log_record_size(plan.sector_size).ok_or(
        ImageError::InvalidPlan("initial log record size is invalid"),
    )?;
    add_materialized_bytes(&mut total, log_record_size as u64)?;

    add_materialized_bytes(&mut total, u64::from(plan.sector_size))?;

    let features = plan.features();
    let sector_size = u64::from(plan.sector_size);
    let block_size = u64::from(plan.block_size);
    let mut per_ag = 3_u64
        .checked_mul(sector_size)
        .ok_or(ImageError::InvalidPlan("AG header estimate overflows u64"))?;
    per_ag = per_ag
        .checked_add(
            3_u64
                .checked_mul(block_size)
                .ok_or(ImageError::InvalidPlan("btree root estimate overflows u64"))?,
        )
        .ok_or(ImageError::InvalidPlan(
            "per-allocation-group estimate overflows u64",
        ))?;
    if features.finobt() {
        per_ag = per_ag
            .checked_add(block_size)
            .ok_or(ImageError::InvalidPlan("FINOBT estimate overflows u64"))?;
    }
    if features.rmapbt() {
        per_ag = per_ag
            .checked_add(block_size)
            .ok_or(ImageError::InvalidPlan("RMAPBT estimate overflows u64"))?;
    }
    if features.reflink() {
        per_ag = per_ag
            .checked_add(block_size)
            .ok_or(ImageError::InvalidPlan("REFCOUNTBT estimate overflows u64"))?;
    }

    let ag_bytes = per_ag
        .checked_mul(u64::from(plan.ag_count))
        .ok_or(ImageError::InvalidPlan(
            "AG metadata estimate overflows u64",
        ))?;
    add_materialized_bytes(&mut total, ag_bytes)?;

    let secondary_superblocks = plan.ag_count.saturating_sub(1);
    let secondary_superblock_bytes = u64::from(secondary_superblocks)
        .checked_mul(sector_size)
        .ok_or(ImageError::InvalidPlan(
            "secondary superblock estimate overflows u64",
        ))?;
    add_materialized_bytes(&mut total, secondary_superblock_bytes)?;

    Ok(total)
}

fn add_materialized_bytes(total: &mut u64, bytes: u64) -> Result<(), ImageError> {
    *total = total.checked_add(bytes).ok_or(ImageError::InvalidPlan(
        "materialized byte estimate overflows u64",
    ))?;
    Ok(())
}

struct MagicDisplay([u8; 4]);

impl std::fmt::Display for MagicDisplay {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        for &byte in &self.0 {
            if byte.is_ascii_graphic() || byte == b' ' {
                write!(f, "{}", char::from(byte))?;
            } else {
                write!(f, "\\x{byte:02x}")?;
            }
        }
        Ok(())
    }
}

trait MetadataSink {
    fn metadata(&mut self, offset: u64, bytes: Vec<u8>) -> Result<(), ImageError>;
    fn zero(&mut self, offset: u64, len: u64) -> Result<(), ImageError>;
}

struct MetadataImageBuilder {
    extents: Vec<MetadataExtent>,
    zero_extents: Vec<ZeroExtent>,
}

impl MetadataImageBuilder {
    fn new() -> Self {
        Self {
            extents: Vec::new(),
            zero_extents: Vec::new(),
        }
    }

    fn finish(mut self) -> MetadataImage {
        self.extents.sort_by_key(|extent| extent.offset);
        self.zero_extents.sort_by_key(|extent| extent.offset);
        let mut merged: Vec<MetadataExtent> = Vec::with_capacity(self.extents.len());
        for extent in self.extents.drain(..) {
            if let Some(last) = merged.last_mut() {
                if last.offset == extent.offset {
                    last.bytes = extent.bytes;
                    continue;
                }
            }
            merged.push(extent);
        }
        self.extents = merged;
        MetadataImage {
            extents: self.extents,
            zero_extents: self.zero_extents,
        }
    }
}

impl MetadataSink for MetadataImageBuilder {
    fn metadata(&mut self, offset: u64, bytes: Vec<u8>) -> Result<(), ImageError> {
        self.extents.push(MetadataExtent { offset, bytes });
        Ok(())
    }

    fn zero(&mut self, offset: u64, len: u64) -> Result<(), ImageError> {
        self.zero_extents.push(ZeroExtent { offset, len });
        Ok(())
    }
}

struct MetadataWriter<'a, W> {
    writer: &'a mut W,
}

impl<W: Seek + Write> MetadataSink for MetadataWriter<'_, W> {
    fn metadata(&mut self, offset: u64, bytes: Vec<u8>) -> Result<(), ImageError> {
        self.writer.seek(SeekFrom::Start(offset))?;
        self.writer.write_all(&bytes)?;
        Ok(())
    }

    fn zero(&mut self, offset: u64, len: u64) -> Result<(), ImageError> {
        write_zeroes(self.writer, offset, len)?;
        Ok(())
    }
}

struct MetadataLayoutEmitter<'a> {
    plan: &'a FormatPlan,
}

impl<'a> MetadataLayoutEmitter<'a> {
    fn new(plan: &'a FormatPlan) -> Self {
        Self { plan }
    }

    fn emit_initial_metadata<S: MetadataSink>(&self, sink: &mut S) -> Result<(), ImageError> {
        self.emit_initial_inode_chunk(self.plan.root_inode_location(), sink)?;

        let log = self.plan.internal_log();
        let Some(log_offset) = log.start_block.checked_mul(u64::from(self.plan.block_size)) else {
            return Err(ImageError::InvalidPlan("log offset overflows u64"));
        };
        let Some(log_len) = u64::from(log.blocks).checked_mul(u64::from(self.plan.block_size))
        else {
            return Err(ImageError::InvalidPlan("log length overflows u64"));
        };
        let Some(log_record) = ondisk::initial_log_record(self.plan.uuid, self.plan.sector_size)
        else {
            return Err(ImageError::InvalidPlan("initial log record is invalid"));
        };
        let log_record_size = log_record.len() as u64;
        sink.metadata(log_offset, log_record)?;

        let Some(zero_offset) = log_offset.checked_add(log_record_size) else {
            return Err(ImageError::InvalidPlan("log zero offset overflows u64"));
        };
        let Some(zero_len) = log_len.checked_sub(log_record_size) else {
            return Err(ImageError::InvalidPlan(
                "initial log record exceeds log length",
            ));
        };
        sink.zero(zero_offset, zero_len)?;

        let Some(primary_sb) = self
            .plan
            .initial_superblock()
            .encode_sector_with_crc(self.plan.sector_size as usize)
        else {
            return Err(ImageError::InvalidPlan("superblock encoding failed"));
        };
        sink.metadata(0, primary_sb)?;

        for agno in 0..self.plan.ag_count {
            let Some(headers) = self.plan.initial_ag_headers(agno) else {
                return Err(ImageError::InvalidPlan(
                    "allocation group headers are invalid",
                ));
            };
            let Some(roots) = self.plan.initial_btree_roots(agno) else {
                return Err(ImageError::InvalidPlan("btree roots are invalid"));
            };

            if agno > 0 {
                let mut sb = self.plan.initial_superblock();
                sb.inprogress = 1;
                sb.inode_count = 0;
                sb.free_inode_count = 0;
                sb.realtime_bitmap_inode = ondisk::NULLFSINO;
                sb.realtime_summary_inode = ondisk::NULLFSINO;
                if self.plan.internal_log().agno == agno {
                    sb.root_inode = ondisk::NULLFSINO;
                }
                let Some(free_data_blocks) = sb
                    .free_data_blocks
                    .checked_add(u64::from(self.plan.initial_inode_chunk_blocks()))
                else {
                    return Err(ImageError::InvalidPlan(
                        "free data block count overflows u64",
                    ));
                };
                sb.free_data_blocks = free_data_blocks;
                let Some(bytes) = sb.encode_sector_with_crc(self.plan.sector_size as usize) else {
                    return Err(ImageError::InvalidPlan(
                        "secondary superblock encoding failed",
                    ));
                };
                sink.metadata(self.ag_sector_offset(agno, 0), bytes)?;
            }

            let Some(bytes) = headers
                .agf
                .encode_sector_with_crc(self.plan.sector_size as usize)
            else {
                return Err(ImageError::InvalidPlan("AGF encoding failed"));
            };
            sink.metadata(self.ag_sector_offset(agno, 1), bytes)?;

            let Some(bytes) = headers
                .agi
                .encode_sector_with_crc(self.plan.sector_size as usize)
            else {
                return Err(ImageError::InvalidPlan("AGI encoding failed"));
            };
            sink.metadata(self.ag_sector_offset(agno, 2), bytes)?;

            let Some(bytes) = headers
                .agfl
                .encode_sector_with_crc(self.plan.sector_size as usize)
            else {
                return Err(ImageError::InvalidPlan("AGFL encoding failed"));
            };
            sink.metadata(self.ag_sector_offset(agno, 3), bytes)?;

            let Some(bytes) = roots
                .bno
                .encode_block_with_crc(self.plan.block_size as usize)
            else {
                return Err(ImageError::InvalidPlan("BNO btree root encoding failed"));
            };
            sink.metadata(self.ag_block_offset(agno, headers.agf.bno_root), bytes)?;

            let Some(bytes) = roots
                .cnt
                .encode_block_with_crc(self.plan.block_size as usize)
            else {
                return Err(ImageError::InvalidPlan("CNT btree root encoding failed"));
            };
            sink.metadata(self.ag_block_offset(agno, headers.agf.cnt_root), bytes)?;

            let Some(bytes) = roots
                .inobt
                .encode_block_with_crc(self.plan.block_size as usize)
            else {
                return Err(ImageError::InvalidPlan("INOBT root encoding failed"));
            };
            sink.metadata(self.ag_block_offset(agno, headers.agi.root), bytes)?;

            if let Some(finobt) = roots.finobt {
                let Some(bytes) = finobt.encode_block_with_crc(self.plan.block_size as usize)
                else {
                    return Err(ImageError::InvalidPlan("FINOBT root encoding failed"));
                };
                sink.metadata(self.ag_block_offset(agno, headers.agi.free_root), bytes)?;
            }
            if let Some(rmap) = roots.rmap {
                let Some(bytes) = rmap.encode_block_with_crc(self.plan.block_size as usize) else {
                    return Err(ImageError::InvalidPlan("RMAPBT root encoding failed"));
                };
                sink.metadata(self.ag_block_offset(agno, headers.agf.rmap_root), bytes)?;
            }
            if let Some(refcount) = roots.refcount {
                let Some(bytes) = refcount.encode_block_with_crc(self.plan.block_size as usize)
                else {
                    return Err(ImageError::InvalidPlan("REFCOUNTBT root encoding failed"));
                };
                sink.metadata(self.ag_block_offset(agno, headers.agf.refcount_root), bytes)?;
            }
        }

        Ok(())
    }

    fn emit_initial_inode_chunk<S: MetadataSink>(
        &self,
        root: crate::InodeLocation,
        sink: &mut S,
    ) -> Result<(), ImageError> {
        let inode_size = self.plan.inode_size as usize;
        let Some(chunk_size) = inode_size.checked_mul(ondisk::XFS_INODES_PER_CHUNK as usize) else {
            return Err(ImageError::InvalidPlan(
                "initial inode chunk size overflows usize",
            ));
        };
        let mut chunk = Vec::with_capacity(chunk_size);
        let generation = self.plan.inode_generation.generation()?;

        let Some(data_fork) = ondisk::root_shortform_directory(root.ino) else {
            return Err(ImageError::InvalidPlan(
                "root shortform directory is invalid",
            ));
        };
        let Some(root_inode) = ondisk::Dinode::root_directory(root.ino, self.plan.uuid)
            .encode_inode_with_data_fork_and_crc(inode_size, &data_fork)
        else {
            return Err(ImageError::InvalidPlan("root inode encoding failed"));
        };
        chunk.extend_from_slice(&root_inode);

        let mut rt_bitmap = ondisk::Dinode::realtime_bitmap(root.ino + 1, self.plan.uuid);
        rt_bitmap.generation = generation;
        let Some(rt_bitmap) = rt_bitmap.encode_inode_with_crc(inode_size) else {
            return Err(ImageError::InvalidPlan(
                "realtime bitmap inode encoding failed",
            ));
        };
        chunk.extend_from_slice(&rt_bitmap);

        let mut rt_summary = ondisk::Dinode::realtime_summary(root.ino + 2, self.plan.uuid);
        rt_summary.generation = generation;
        let Some(rt_summary) = rt_summary.encode_inode_with_crc(inode_size) else {
            return Err(ImageError::InvalidPlan(
                "realtime summary inode encoding failed",
            ));
        };
        chunk.extend_from_slice(&rt_summary);

        for ino in root.ino + 3..root.ino + u64::from(ondisk::XFS_INODES_PER_CHUNK) {
            let mut free = ondisk::Dinode::free(ino, self.plan.uuid);
            free.generation = generation;
            let Some(free) = free.encode_inode_with_crc(inode_size) else {
                return Err(ImageError::InvalidPlan("free inode encoding failed"));
            };
            chunk.extend_from_slice(&free);
        }

        debug_assert_eq!(chunk.len(), chunk_size);
        sink.metadata(root.byte_offset, chunk)?;
        Ok(())
    }

    fn ag_sector_offset(&self, agno: u32, sector: u32) -> u64 {
        self.ag_base_offset(agno) + u64::from(sector) * u64::from(self.plan.sector_size)
    }

    fn ag_block_offset(&self, agno: u32, agbno: u32) -> u64 {
        self.ag_base_offset(agno) + u64::from(agbno) * u64::from(self.plan.block_size)
    }

    fn ag_base_offset(&self, agno: u32) -> u64 {
        u64::from(agno) * u64::from(self.plan.ag_blocks) * u64::from(self.plan.block_size)
    }
}

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

    const UUID: [u8; 16] = [
        0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee,
        0xff,
    ];

    #[test]
    fn image_contains_primary_metadata_for_default_plan() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let extents = image.extents();

        assert_eq!(extents.len(), 42);
        assert_eq!(image.zero_extents().len(), 1);
        assert_eq!(extents[0].offset, 0);
        assert_eq!(&extents[0].bytes[0..4], b"XFSB");
        assert_eq!(extents[1].offset, 512);
        assert_eq!(&extents[1].bytes[0..4], b"XAGF");
        assert_eq!(extents[2].offset, 1024);
        assert_eq!(&extents[2].bytes[0..4], b"XAGI");
        assert_eq!(extents[3].offset, 1536);
        assert_eq!(&extents[3].bytes[0..4], b"XAFL");
        assert_eq!(extents[4].offset, 4096);
        assert_eq!(&extents[4].bytes[0..4], b"AB3B");
        assert_eq!(extents[9].offset, 24_576);
        assert_eq!(&extents[9].bytes[0..4], b"R3FC");
    }

    #[test]
    fn image_offsets_are_sorted_and_unique() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();

        for pair in image.extents().windows(2) {
            assert!(pair[0].offset < pair[1].offset);
        }
    }

    #[test]
    fn image_extent_summaries_are_stable_and_compact() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let summaries = image.extent_summaries();

        assert_eq!(
            summaries.len(),
            image.extents().len() + image.zero_extents().len()
        );
        assert_eq!(
            summaries[0],
            MetadataExtentSummary {
                offset: 0,
                len: 512,
                crc32c: crc32c::checksum(&image.extents()[0].bytes),
                magic: *b"XFSB",
            }
        );
        assert_eq!(summaries[1].magic, *b"XAGF");
        assert_eq!(summaries[4].magic, *b"AB3B");
        assert_eq!(summaries[9].magic, *b"R3FC");
    }

    #[test]
    fn image_includes_initialized_internal_log_extent() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let log = plan.internal_log();
        let log_offset = log.start_block * u64::from(plan.block_size);
        let log_len = u64::from(log.blocks) * u64::from(plan.block_size);
        let log_record_size = crate::ondisk::initial_log_record_size(plan.sector_size).unwrap();

        let log_record = image
            .extents()
            .iter()
            .find(|extent| extent.offset == log_offset)
            .unwrap();
        assert_eq!(log_record.bytes.len(), log_record_size);
        assert_eq!(&log_record.bytes[0..4], &[0xfe, 0xed, 0xba, 0xbe]);
        assert_eq!(&log_record.bytes[512..516], &1_u32.to_be_bytes());
        assert_eq!(&log_record.bytes[516..520], &8_u32.to_be_bytes());
        assert_eq!(log_record.bytes[520], 0xaa);
        assert_eq!(log_record.bytes[521], 0x20);
        assert_eq!(&log_record.bytes[524..526], &0x556e_u16.to_le_bytes());
        assert!(image.zero_extents().contains(&ZeroExtent {
            offset: log_offset + log_record_size as u64,
            len: log_len - log_record_size as u64,
        }));
        assert!(image
            .extents()
            .iter()
            .all(|extent| extent.offset <= log_offset || extent.offset >= log_offset + log_len));
        assert!(image.extent_summaries().iter().any(|summary| {
            summary.offset == log_offset + log_record_size as u64
                && summary.len == log_len - log_record_size as u64
                && summary.crc32c == checksum_zeroes(log_len - log_record_size as u64)
                && summary.magic == [0; 4]
        }));
    }

    #[test]
    fn image_includes_initial_inode_chunk_extent() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let root = plan.root_inode_location();
        let inode_size = plan.inode_size as usize;

        let extent = image
            .extents()
            .iter()
            .find(|extent| extent.offset == root.byte_offset)
            .unwrap();
        assert_eq!(
            extent.bytes.len(),
            inode_size * crate::ondisk::XFS_INODES_PER_CHUNK as usize
        );
        assert_eq!(&extent.bytes[0..2], b"IN");
        assert_eq!(&extent.bytes[inode_size..inode_size + 2], b"IN");
        assert_eq!(&extent.bytes[inode_size * 2..inode_size * 2 + 2], b"IN");
        assert_eq!(
            &extent.bytes[crate::ondisk::XFS_DINODE_SIZE
                ..crate::ondisk::XFS_DINODE_SIZE + crate::ondisk::XFS_DIR2_SF_HDR_SIZE_SHORT],
            &[0, 0, 0, 0, 0, 128]
        );
        assert!(image.extent_summaries().iter().any(|summary| {
            summary.offset == root.byte_offset
                && summary.len
                    == u64::from(plan.inode_size) * u64::from(crate::ondisk::XFS_INODES_PER_CHUNK)
        }));
    }

    #[test]
    fn image_does_not_zero_initialized_inode_chunk_tail() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let root = plan.root_inode_location();
        let inode_size = u64::from(plan.inode_size);
        let zero_tail_offset = root.byte_offset + inode_size;
        let zero_tail_len = (u64::from(crate::ondisk::XFS_INODES_PER_CHUNK) - 1) * inode_size;

        assert!(!image.zero_extents().contains(&ZeroExtent {
            offset: zero_tail_offset,
            len: zero_tail_len,
        }));
    }

    #[test]
    fn image_manifest_is_deterministic_text() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let manifest = image.manifest();
        let lines: Vec<&str> = manifest.lines().collect();

        assert_eq!(lines[0], "offset,len,crc32c,magic");
        assert_eq!(
            lines.len(),
            image.extents().len() + image.zero_extents().len() + 1
        );
        assert!(lines[1].starts_with("0,512,"));
        assert!(lines[1].ends_with(",XFSB"));
        assert!(lines[2].ends_with(",XAGF"));
        assert!(lines[5].ends_with(",AB3B"));
    }

    #[test]
    fn image_contains_secondary_ag_metadata_offsets() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let base = u64::from(plan.ag_blocks) * u64::from(plan.block_size);

        assert!(image
            .extents()
            .iter()
            .any(|extent| extent.offset == base && &extent.bytes[0..4] == b"XFSB"));
        assert!(image
            .extents()
            .iter()
            .any(|extent| extent.offset == base + 4096 && &extent.bytes[0..4] == b"AB3B"));
    }

    #[test]
    fn image_can_apply_to_sparse_buffer() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let mut bytes = vec![0_u8; image.end_offset() as usize];

        assert!(image.apply_to_slice(&mut bytes));
        assert_eq!(&bytes[0..4], b"XFSB");
        assert_eq!(&bytes[512..516], b"XAGF");
        assert_eq!(&bytes[4096..4100], b"AB3B");
    }

    #[test]
    fn image_write_to_seekable_writer() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let mut cursor = std::io::Cursor::new(Vec::new());

        image.write_to(&mut cursor).unwrap();
        let bytes = cursor.into_inner();

        assert_eq!(bytes.len() as u64, image.end_offset());
        assert_eq!(&bytes[0..4], b"XFSB");
        assert_eq!(&bytes[1024..1028], b"XAGI");
        assert_eq!(&bytes[24_576..24_580], b"R3FC");
    }

    #[test]
    fn streaming_writer_matches_materialized_extents() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let mut writer = RecordingWriter::default();

        write_plan_to(&plan, &mut writer).unwrap();

        let mut written_extents: Vec<_> = writer
            .writes
            .iter()
            .filter(|(_, bytes)| bytes.iter().any(|&byte| byte != 0))
            .map(|(offset, bytes)| MetadataExtent {
                offset: *offset,
                bytes: bytes.clone(),
            })
            .collect();
        written_extents.sort_by_key(|extent| extent.offset);
        assert_eq!(written_extents, image.extents());

        let zero_bytes: u64 = writer
            .writes
            .iter()
            .filter(|(_, bytes)| bytes.iter().all(|&byte| byte == 0))
            .map(|(_, bytes)| bytes.len() as u64)
            .sum();
        assert_eq!(
            zero_bytes,
            image
                .zero_extents()
                .iter()
                .map(|extent| extent.len)
                .sum::<u64>()
        );
    }

    #[test]
    fn materialized_image_rejects_excessive_ag_count() {
        let oversized_bytes = (u64::from(MAX_MATERIALIZED_ALLOCATION_GROUPS) + 1) << 40;
        let plan = crate::FormatPlan::new(FormatOptions::new(oversized_bytes, UUID)).unwrap();

        assert!(plan.ag_count() > MAX_MATERIALIZED_ALLOCATION_GROUPS);
        assert!(matches!(
            MetadataImage::from_plan(&plan),
            Err(ImageError::TooManyAllocationGroups { .. })
        ));
    }

    #[test]
    fn materialized_image_rejects_excessive_byte_budget() {
        let mut options = FormatOptions::new(
            u64::from(MAX_MATERIALIZED_ALLOCATION_GROUPS - 1) << 40,
            UUID,
        );
        options.block_size = 65_536;
        options.sector_size = 32_768;
        options.inode_size = 32_768;
        let plan = crate::FormatPlan::new(options).unwrap();

        assert_eq!(plan.ag_count(), MAX_MATERIALIZED_ALLOCATION_GROUPS);
        assert!(matches!(
            MetadataImage::from_plan(&plan),
            Err(ImageError::TooManyMaterializedBytes { .. })
        ));
    }

    #[test]
    fn apply_to_slice_rejects_too_small_target() {
        let plan = crate::FormatPlan::new(FormatOptions::new(1 << 30, UUID)).unwrap();
        let image = MetadataImage::from_plan(&plan).unwrap();
        let mut bytes = vec![0_u8; 4096];

        assert!(!image.apply_to_slice(&mut bytes));
    }

    #[derive(Default)]
    struct RecordingWriter {
        pos: u64,
        writes: Vec<(u64, Vec<u8>)>,
    }

    impl Write for RecordingWriter {
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.writes.push((self.pos, buf.to_vec()));
            self.pos += buf.len() as u64;
            Ok(buf.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    impl Seek for RecordingWriter {
        fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
            self.pos = match pos {
                SeekFrom::Start(offset) => offset,
                SeekFrom::End(_) | SeekFrom::Current(_) => {
                    return Err(io::Error::new(
                        io::ErrorKind::Unsupported,
                        "relative seeks are not supported",
                    ));
                }
            };
            Ok(self.pos)
        }
    }
}
