use std::collections::{BTreeMap, HashSet};
use std::fs;
use std::io::{Read, Write};
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};

use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tar::{Archive, EntryType};
use walkdir::WalkDir;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactManifest {
    pub packages: Vec<PackageRecord>,
    pub files: Vec<FileRecord>,
    pub initrd: Option<InitrdRecord>,
    pub services: Vec<ServiceRecord>,
    pub archive_members: Vec<ArchiveMemberRecord>,
}

impl ArtifactManifest {
    pub fn new(
        mut packages: Vec<PackageRecord>,
        mut files: Vec<FileRecord>,
        initrd: Option<InitrdRecord>,
        mut services: Vec<ServiceRecord>,
        mut archive_members: Vec<ArchiveMemberRecord>,
    ) -> Result<Self> {
        packages.sort_by(|left, right| left.name.cmp(&right.name));
        files.sort_by(|left, right| left.path.cmp(&right.path));
        services.sort_by(|left, right| left.name.cmp(&right.name));
        archive_members.sort_by(|left, right| left.path.cmp(&right.path));
        validate_file_records(&files)?;
        validate_initrd_record(initrd.as_ref())?;
        validate_manifest_paths(&files, initrd.as_ref(), &archive_members)?;
        ensure_unique("package", packages.iter().map(|record| record.name.as_str()))?;
        ensure_unique("file", files.iter().map(|record| record.path.as_str()))?;
        ensure_unique("service", services.iter().map(|record| record.name.as_str()))?;
        ensure_unique("archive member", archive_members.iter().map(|record| record.path.as_str()))?;
        Ok(Self { packages, files, initrd, services, archive_members })
    }

    pub fn load(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();
        let text = fs::read_to_string(path).with_context(|| format!("read artifact manifest {}", path.display()))?;
        let manifest: Self = toml::from_str(&text).with_context(|| format!("parse artifact manifest {}", path.display()))?;
        Self::new(manifest.packages, manifest.files, manifest.initrd, manifest.services, manifest.archive_members)
    }

    pub fn write_beside(&self, artifact: impl AsRef<Path>) -> Result<PathBuf> {
        let path = Self::path_beside(artifact)?;
        let text = toml::to_string_pretty(self).context("serialize artifact manifest")?;
        fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&path)
            .with_context(|| format!("create artifact manifest {}", path.display()))?
            .write_all(text.as_bytes())
            .with_context(|| format!("write artifact manifest {}", path.display()))?;
        Ok(path)
    }

    /// Return the unambiguous companion-manifest path for one artifact.
    /// Keeping the artifact filename prevents legacy and native manifests from
    /// overwriting one another when both are written to the comparison output
    /// directory.
    pub fn path_beside(artifact: impl AsRef<Path>) -> Result<PathBuf> {
        let artifact = artifact.as_ref();
        let name = artifact
            .file_name()
            .ok_or_else(|| anyhow::anyhow!("artifact path has no filename: {}", artifact.display()))?;
        let mut manifest_name = name.to_os_string();
        manifest_name.push(".manifest.toml");
        Ok(artifact.with_file_name(manifest_name))
    }

    /// Collect native filesystem and archive facts from a completed build.
    /// Package records come from the typed package-installation boundary because
    /// the RPM database is not a portable filesystem format.
    pub fn collect(
        rootfs: impl AsRef<Path>,
        packages: Vec<PackageRecord>,
        initrd: Option<InitrdRecord>,
        archive: impl AsRef<Path>,
    ) -> Result<Self> {
        let rootfs = rootfs.as_ref();
        if !rootfs.is_dir() {
            bail!("rootfs is not a directory: {}", rootfs.display());
        }
        Self::new(
            packages,
            collect_files(rootfs, initrd.as_ref())?,
            initrd,
            collect_services(rootfs)?,
            collect_archive_members(archive.as_ref())?,
        )
    }

    /// Collect portable semantic facts from an already packaged rootfs archive.
    /// This permits comparison with legacy artifacts when their build root is
    /// unavailable. RPM package metadata remains unavailable in a plain tar
    /// archive and is consequently represented by an empty package set.
    pub fn collect_archive(archive_path: impl AsRef<Path>) -> Result<Self> {
        let archive_path = archive_path.as_ref();
        let file = fs::File::open(archive_path)
            .with_context(|| format!("open archive {}", archive_path.display()))?;
        let mut archive = Archive::new(file);
        let mut files = Vec::new();
        let mut services = Vec::new();
        let mut archive_members = Vec::new();
        let mut initrd = None;

        for entry in archive
            .entries()
            .with_context(|| format!("read archive {}", archive_path.display()))?
        {
            let mut entry = entry
                .with_context(|| format!("read archive member from {}", archive_path.display()))?;
            let path = portable_path(&entry.path()?)?;
            if is_archive_root_path(&path) || is_internal_metadata_path(&path) {
                continue;
            }
            let entry_type = entry.header().entry_type();
            let kind = match entry_type {
                EntryType::Regular => "file",
                EntryType::Directory => "directory",
                EntryType::Symlink => "symlink",
                EntryType::Link => "hardlink",
                other => bail!("unsupported archive member type {other:?} in {}", archive_path.display()),
            };
            archive_members.push(ArchiveMemberRecord::new(path.clone(), kind));

            match entry_type {
                EntryType::Regular => {
                    let digest = sha256_reader(&mut entry, archive_path)?;
                    if path.starts_with("boot/initrd-") && path.ends_with(".img") {
                        let record = InitrdRecord::new(path.clone(), digest.clone());
                        if initrd.replace(record).is_some() {
                            bail!("multiple initrd images found in archive {}", archive_path.display());
                        }
                    }
                    if !path.starts_with("boot/initrd-") || !path.ends_with(".img") {
                        files.push(FileRecord::file(path, digest));
                    }
                }
                EntryType::Symlink => {
                    let target = entry
                        .link_name()
                        .with_context(|| format!("read archive symlink target for {path}"))?
                        .ok_or_else(|| anyhow::anyhow!("archive symlink has no target: {path}"))?;
                    files.push(FileRecord::symlink(path.clone(), link_target(&target)?));
                    if is_enabled_service(&path) {
                        services.push(ServiceRecord::new(service_name(&path)?, true));
                    }
                }
                EntryType::Directory => {}
                EntryType::Link => {
                    let target = entry
                        .link_name()
                        .with_context(|| format!("read archive hardlink target for {path}"))?
                        .ok_or_else(|| anyhow::anyhow!("archive hardlink has no target: {path}"))?;
                    files.push(FileRecord::hardlink(path, portable_path(&target)?));
                }
                _ => unreachable!("entry type was checked above"),
            }
        }

        Self::new(
            Vec::new(),
            files,
            initrd,
            coalesce_enabled_services(services),
            archive_members,
        )
    }
}

fn collect_files(rootfs: &Path, initrd: Option<&InitrdRecord>) -> Result<Vec<FileRecord>> {
    let mut files = Vec::new();
    let mut entries = WalkDir::new(rootfs)
        .follow_links(false)
        .min_depth(1)
        .into_iter()
        .collect::<std::result::Result<Vec<_>, _>>()
        .with_context(|| format!("walk rootfs {}", rootfs.display()))?;
    entries.sort_by_key(|entry| entry.path().to_path_buf());
    let mut hardlink_targets: BTreeMap<(u64, u64), String> = BTreeMap::new();
    for entry in entries {
        let relative = entry.path().strip_prefix(rootfs).expect("walk entry is below rootfs");
        if is_internal_metadata_path(&portable_path(relative)?) {
            continue;
        }
        let path = portable_path(relative)?;
        if initrd.is_some_and(|record| record.path == path) {
            continue;
        }
        let file_type = entry.file_type();
        if file_type.is_file() {
            let metadata = entry.metadata().with_context(|| format!("inspect rootfs file {}", entry.path().display()))?;
            let key = (metadata.dev(), metadata.ino());
            if let Some(target) = hardlink_targets.get(&key) {
                files.push(FileRecord::hardlink(path, target.clone()));
            } else {
                hardlink_targets.insert(key, path.clone());
                files.push(FileRecord::file(path, sha256_file(entry.path())?));
            }
        } else if file_type.is_symlink() {
            let target = fs::read_link(entry.path())
                .with_context(|| format!("read rootfs symlink {}", entry.path().display()))?;
            files.push(FileRecord::symlink(path, link_target(&target)?));
        } else if !file_type.is_dir() {
            bail!("unsupported rootfs entry type: {}", entry.path().display());
        }
    }
    Ok(files)
}

fn collect_services(rootfs: &Path) -> Result<Vec<ServiceRecord>> {
    let system = rootfs.join("etc/systemd/system");
    if !system.is_dir() {
        return Ok(Vec::new());
    }
    let mut services = Vec::new();
    for entry in WalkDir::new(&system).follow_links(false).min_depth(2) {
        let entry = entry.with_context(|| format!("walk service state directory {}", system.display()))?;
        let parent_is_wants_directory = entry
            .path()
            .parent()
            .and_then(Path::file_name)
            .is_some_and(|name| name.to_string_lossy().ends_with(".target.wants"));
        if !parent_is_wants_directory || !entry.file_type().is_symlink() {
            continue;
        }
        let name = entry
            .file_name()
            .to_str()
            .ok_or_else(|| anyhow::anyhow!("non-UTF-8 service name in {}", entry.path().display()))?;
        if name.ends_with(".service") {
            services.push(ServiceRecord::new(name, true));
        }
    }
    Ok(coalesce_enabled_services(services))
}

fn coalesce_enabled_services(mut services: Vec<ServiceRecord>) -> Vec<ServiceRecord> {
    services.sort_by(|left, right| left.name.cmp(&right.name));
    services.dedup_by(|left, right| left.name == right.name);
    services
}

fn collect_archive_members(archive_path: &Path) -> Result<Vec<ArchiveMemberRecord>> {
    let file = fs::File::open(archive_path).with_context(|| format!("open archive {}", archive_path.display()))?;
    let mut archive = Archive::new(file);
    let mut members = Vec::new();
    for entry in archive.entries().with_context(|| format!("read archive {}", archive_path.display()))? {
        let entry = entry.with_context(|| format!("read archive member from {}", archive_path.display()))?;
        let path = portable_path(&entry.path()?)?;
        if is_archive_root_path(&path) || is_internal_metadata_path(&path) {
            continue;
        }
        let kind = match entry.header().entry_type() {
            EntryType::Regular => "file",
            EntryType::Directory => "directory",
            EntryType::Symlink => "symlink",
            EntryType::Link => "hardlink",
            other => bail!("unsupported archive member type {other:?} in {}", archive_path.display()),
        };
        members.push(ArchiveMemberRecord::new(path, kind));
    }
    Ok(members)
}

fn sha256_file(path: &Path) -> Result<String> {
    let mut file = fs::File::open(path).with_context(|| format!("open rootfs file {}", path.display()))?;
    sha256_reader(&mut file, path)
}

fn sha256_reader(reader: &mut impl Read, source: &Path) -> Result<String> {
    let mut hasher = Sha256::new();
    let mut buffer = [0; 8192];
    loop {
        let read = reader.read(&mut buffer).with_context(|| format!("read {}", source.display()))?;
        if read == 0 {
            break;
        }
        hasher.update(&buffer[..read]);
    }
    Ok(format!("{:x}", hasher.finalize()))
}

fn portable_path(path: &Path) -> Result<String> {
    if path.is_absolute() {
        bail!("absolute path is not valid in an artifact manifest: {}", path.display());
    }
    let value = path.to_str().ok_or_else(|| anyhow::anyhow!("non-UTF-8 path is not valid in an artifact manifest: {}", path.display()))?;
    if value.is_empty() {
        bail!("empty path is not valid in an artifact manifest");
    }
    let mut normalized = PathBuf::new();
    for component in path.components() {
        match component {
            std::path::Component::Normal(part) => normalized.push(part),
            std::path::Component::CurDir => {}
            std::path::Component::ParentDir => {
                bail!("parent path is not valid in an artifact manifest: {}", path.display());
            }
            std::path::Component::RootDir | std::path::Component::Prefix(_) => {
                bail!("absolute path is not valid in an artifact manifest: {}", path.display());
            }
        }
    }
    if normalized.as_os_str().is_empty() {
        return Ok(".".into());
    }
    let normalized = normalized
        .to_str()
        .ok_or_else(|| anyhow::anyhow!("non-UTF-8 path is not valid in an artifact manifest: {}", path.display()))?;
    Ok(normalized.to_owned())
}

fn is_enabled_service(path: &str) -> bool {
    let mut components = path.split('/');
    matches!(
        (
            components.next(),
            components.next(),
            components.next(),
            components.next(),
            components.next(),
            components.next(),
        ),
        (
            Some("etc"),
            Some("systemd"),
            Some("system"),
            Some(wants),
            Some(service),
            None,
        ) if wants.ends_with(".target.wants") && service.ends_with(".service")
    )
}

fn is_internal_metadata_path(path: &str) -> bool {
    path.split('/').any(|component| matches!(component, ".host" | ".fakedata"))
}

fn is_archive_root_path(path: &str) -> bool {
    path == "." || path == "./"
}

fn service_name(path: &str) -> Result<&str> {
    path.rsplit('/')
        .next()
        .filter(|name| !name.is_empty())
        .ok_or_else(|| anyhow::anyhow!("invalid service path in archive: {path}"))
}

fn link_target(path: &Path) -> Result<String> {
    let value = path.to_str().ok_or_else(|| anyhow::anyhow!("non-UTF-8 symlink target is not valid in an artifact manifest: {}", path.display()))?;
    if value.is_empty() {
        bail!("empty symlink target is not valid in an artifact manifest");
    }
    Ok(value.to_owned())
}

fn validate_manifest_paths(
    files: &[FileRecord],
    initrd: Option<&InitrdRecord>,
    archive_members: &[ArchiveMemberRecord],
) -> Result<()> {
    for path in files.iter().map(|record| record.path.as_str()).chain(archive_members.iter().map(|record| record.path.as_str())) {
        portable_path(Path::new(path))?;
    }
    if let Some(initrd) = initrd {
        portable_path(Path::new(&initrd.path))?;
    }
    Ok(())
}

fn validate_initrd_record(record: Option<&InitrdRecord>) -> Result<()> {
    if let Some(record) = record {
        if record.sha256.is_empty() {
            bail!("initrd record requires a sha256 digest: {}", record.path);
        }
    }
    Ok(())
}

fn validate_file_records(records: &[FileRecord]) -> Result<()> {
    for record in records {
        match record.kind.as_str() {
            "file" => {
                if record.digest.as_deref().is_none_or(str::is_empty) {
                    bail!("regular file record requires a digest: {}", record.path);
                }
                if record.target.is_some() {
                    bail!("regular file record cannot carry a target: {}", record.path);
                }
            }
            "symlink" | "hardlink" => {
                if record.digest.is_some() {
                    bail!("{} record cannot carry a digest: {}", record.kind, record.path);
                }
                if record.target.as_deref().is_none_or(str::is_empty) {
                    bail!("{} record requires a target: {}", record.kind, record.path);
                }
            }
            _ => bail!("unsupported file record kind {}: {}", record.kind, record.path),
        }
    }
    Ok(())
}

fn ensure_unique<'a>(kind: &str, keys: impl IntoIterator<Item = &'a str>) -> Result<()> {
    let mut seen = HashSet::new();
    for key in keys {
        if key.is_empty() || !seen.insert(key) {
            bail!("duplicate {kind} record: {key}");
        }
    }
    Ok(())
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PackageRecord {
    pub name: String,
    pub version: String,
}

impl PackageRecord {
    pub fn new(name: impl Into<String>, version: impl Into<String>) -> Self {
        Self { name: name.into(), version: version.into() }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FileRecord {
    pub path: String,
    pub kind: String,
    pub digest: Option<String>,
    pub target: Option<String>,
}

impl FileRecord {
    pub fn file(path: impl Into<String>, digest: impl Into<String>) -> Self {
        Self { path: path.into(), kind: "file".into(), digest: Some(digest.into()), target: None }
    }

    pub fn symlink(path: impl Into<String>, target: impl Into<String>) -> Self {
        Self { path: path.into(), kind: "symlink".into(), digest: None, target: Some(target.into()) }
    }

    pub fn hardlink(path: impl Into<String>, target: impl Into<String>) -> Self {
        Self { path: path.into(), kind: "hardlink".into(), digest: None, target: Some(target.into()) }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct InitrdRecord {
    pub path: String,
    pub sha256: String,
}

impl InitrdRecord {
    pub fn new(path: impl Into<String>, sha256: impl Into<String>) -> Self {
        Self { path: path.into(), sha256: sha256.into() }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ServiceRecord {
    pub name: String,
    pub enabled: bool,
}

impl ServiceRecord {
    pub fn new(name: impl Into<String>, enabled: bool) -> Self {
        Self { name: name.into(), enabled }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArchiveMemberRecord {
    pub path: String,
    pub kind: String,
}

impl ArchiveMemberRecord {
    pub fn new(path: impl Into<String>, kind: impl Into<String>) -> Self {
        Self { path: path.into(), kind: kind.into() }
    }
}