use std::collections::BTreeMap;

use rusqlite::{OptionalExtension, params};
use serde::{Deserialize, Serialize};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};

use crate::domain::AuditEvent;
use crate::{
    domain::{Entry, FolderId, Manifest},
    storage::{StateStore, audit::append_audit_event, prune_manifest_history},
};

const HISTORY_CHECKPOINT_INTERVAL: u64 = 20;
const COMPRESSION_LEVEL: i32 = 3;
const CHECKPOINT_RECORD_KIND: &str = "checkpoint";
const DELTA_RECORD_KIND: &str = "delta";

#[derive(Clone, Debug)]
pub struct ManifestRevision {
    pub revision: i64,
    pub saved_at: OffsetDateTime,
    pub manifest: Manifest,
}

#[derive(Deserialize, Serialize)]
struct ManifestDelta {
    entries: BTreeMap<String, Option<Entry>>,
}

impl StateStore {
    pub fn load_manifest(&self, folder_id: FolderId) -> anyhow::Result<Manifest> {
        let manifest = self
            .connection
            .query_row(
                "SELECT manifest FROM manifests WHERE folder_id = ?1",
                params![folder_id.to_string()],
                |row| row.get::<_, String>(0),
            )
            .optional()?;
        manifest
            .map(|value| serde_json::from_str::<Manifest>(&value).map_err(anyhow::Error::from))
            .transpose()?
            .ok_or_else(|| anyhow::anyhow!("missing manifest for folder {folder_id}"))
    }

    pub fn save_manifest(&self, manifest: &Manifest) -> anyhow::Result<()> {
        self.save_manifest_with_optional_audit(manifest, None)
    }

    pub fn save_manifest_with_audit(
        &self,
        manifest: &Manifest,
        audit_event: &AuditEvent,
    ) -> anyhow::Result<()> {
        self.save_manifest_with_optional_audit(manifest, Some(audit_event))
    }

    fn save_manifest_with_optional_audit(
        &self,
        manifest: &Manifest,
        audit_event: Option<&AuditEvent>,
    ) -> anyhow::Result<()> {
        let serialized = serde_json::to_string(manifest)?;
        let existing = self.load_existing_manifest(manifest.folder_id)?;
        if existing.as_deref() == Some(&serialized) {
            return Ok(());
        }
        let transaction = self.connection.unchecked_transaction()?;
        transaction.execute("INSERT INTO manifests (folder_id, manifest) VALUES (?1, ?2) ON CONFLICT(folder_id) DO UPDATE SET manifest = excluded.manifest", params![manifest.folder_id.to_string(), serialized])?;
        let revision_count = history_count(&transaction, manifest.folder_id)?;
        let payload = history_payload(existing.as_deref(), manifest, revision_count)?;
        transaction.execute("INSERT INTO manifest_history_compact (folder_id, saved_at, kind, payload) VALUES (?1, ?2, ?3, ?4)", params![manifest.folder_id.to_string(), OffsetDateTime::now_utc().format(&Rfc3339)?, payload.kind, payload.bytes])?;
        checkpoint_retention_boundary(&transaction, manifest.folder_id, self.history_limit)?;
        prune_manifest_history(&transaction, manifest.folder_id, self.history_limit)?;
        if let Some(audit_event) = audit_event {
            append_audit_event(&transaction, audit_event)?;
        }
        transaction.commit()?;
        Ok(())
    }

    pub fn manifest_history(&self, folder_id: FolderId) -> anyhow::Result<Vec<ManifestRevision>> {
        let revisions = self.history_records(folder_id)?;
        reconstruct_history(folder_id, revisions)
    }

    pub fn manifest_history_count(&self, folder_id: FolderId) -> anyhow::Result<u64> {
        let count: i64 = self.connection.query_row(
            "SELECT COUNT(*) FROM manifest_history_compact WHERE folder_id = ?1",
            params![folder_id.to_string()],
            |row| row.get(0),
        )?;
        Ok(u64::try_from(count)?)
    }

    pub fn manifest_revision(
        &self,
        folder_id: FolderId,
        revision: i64,
    ) -> anyhow::Result<Manifest> {
        self.manifest_history(folder_id)?
            .into_iter()
            .find(|item| item.revision == revision)
            .map(|item| item.manifest)
            .ok_or_else(|| anyhow::anyhow!("unknown manifest revision {revision}"))
    }

    fn load_existing_manifest(&self, folder_id: FolderId) -> anyhow::Result<Option<String>> {
        self.connection
            .query_row(
                "SELECT manifest FROM manifests WHERE folder_id = ?1",
                params![folder_id.to_string()],
                |row| row.get(0),
            )
            .optional()
            .map_err(Into::into)
    }

    fn history_records(&self, folder_id: FolderId) -> anyhow::Result<Vec<HistoryRecord>> {
        history_records(&self.connection, folder_id)
    }
}

struct HistoryRecord {
    revision: i64,
    saved_at: String,
    kind: String,
    payload: Vec<u8>,
}

struct HistoryPayload {
    kind: &'static str,
    bytes: Vec<u8>,
}

fn history_records(
    connection: &rusqlite::Connection,
    folder_id: FolderId,
) -> anyhow::Result<Vec<HistoryRecord>> {
    let mut statement = connection.prepare("SELECT revision, saved_at, kind, payload FROM manifest_history_compact WHERE folder_id = ?1 ORDER BY revision")?;
    statement
        .query_map(params![folder_id.to_string()], |row| {
            Ok(HistoryRecord {
                revision: row.get(0)?,
                saved_at: row.get(1)?,
                kind: row.get(2)?,
                payload: row.get(3)?,
            })
        })?
        .collect::<Result<Vec<_>, _>>()
        .map_err(Into::into)
}

fn history_count(
    transaction: &rusqlite::Transaction<'_>,
    folder_id: FolderId,
) -> anyhow::Result<u64> {
    let count: i64 = transaction.query_row(
        "SELECT COUNT(*) FROM manifest_history_compact WHERE folder_id = ?1",
        params![folder_id.to_string()],
        |row| row.get(0),
    )?;
    Ok(u64::try_from(count)?)
}
fn history_payload(
    previous: Option<&str>,
    manifest: &Manifest,
    revision_count: u64,
) -> anyhow::Result<HistoryPayload> {
    // Deltas keep ordinary revisions compact. Periodic checkpoints bound the
    // amount of history needed to reconstruct any retained revision.
    if previous.is_none() || revision_count.is_multiple_of(HISTORY_CHECKPOINT_INTERVAL) {
        return Ok(HistoryPayload {
            kind: CHECKPOINT_RECORD_KIND,
            bytes: compress(manifest)?,
        });
    }
    let previous =
        previous.ok_or_else(|| anyhow::anyhow!("manifest history is missing its predecessor"))?;
    let previous = serde_json::from_str(previous)?;
    Ok(HistoryPayload {
        kind: DELTA_RECORD_KIND,
        bytes: compress(&manifest_delta(&previous, manifest))?,
    })
}

fn checkpoint_retention_boundary(
    transaction: &rusqlite::Transaction<'_>,
    folder_id: FolderId,
    history_limit: usize,
) -> anyhow::Result<()> {
    let records = history_records(transaction, folder_id)?;
    if records.len() <= history_limit {
        return Ok(());
    }

    let boundary_index = records.len() - history_limit;
    let boundary = &records[boundary_index];
    let manifest = reconstruct_manifest(folder_id, &records[..=boundary_index])?;
    transaction.execute(
        "UPDATE manifest_history_compact SET kind = ?1, payload = ?2 WHERE revision = ?3",
        params![
            CHECKPOINT_RECORD_KIND,
            compress(&manifest)?,
            boundary.revision
        ],
    )?;
    Ok(())
}
fn manifest_delta(previous: &Manifest, current: &Manifest) -> ManifestDelta {
    let paths = previous
        .entries
        .keys()
        .chain(current.entries.keys())
        .collect::<std::collections::BTreeSet<_>>();
    let entries = paths
        .into_iter()
        .filter_map(|path| {
            let before = previous.entries.get(path);
            let after = current.entries.get(path);
            (before != after).then(|| (path.clone(), after.cloned()))
        })
        .collect();
    ManifestDelta { entries }
}
fn reconstruct_history(
    folder_id: FolderId,
    records: Vec<HistoryRecord>,
) -> anyhow::Result<Vec<ManifestRevision>> {
    let mut manifest = Manifest::empty(folder_id);
    let mut revisions = Vec::new();
    for record in records {
        manifest = apply_history_record(manifest, &record)?;
        revisions.push(ManifestRevision {
            revision: record.revision,
            saved_at: OffsetDateTime::parse(&record.saved_at, &Rfc3339)?,
            manifest: manifest.clone(),
        });
    }
    revisions.reverse();
    Ok(revisions)
}

fn reconstruct_manifest(
    folder_id: FolderId,
    records: &[HistoryRecord],
) -> anyhow::Result<Manifest> {
    records
        .iter()
        .try_fold(Manifest::empty(folder_id), |manifest, record| {
            apply_history_record(manifest, record)
        })
}
fn apply_history_record(
    mut manifest: Manifest,
    record: &HistoryRecord,
) -> anyhow::Result<Manifest> {
    match record.kind.as_str() {
        CHECKPOINT_RECORD_KIND => Ok(decompress(&record.payload)?),
        DELTA_RECORD_KIND => {
            let delta: ManifestDelta = decompress(&record.payload)?;
            for (path, entry) in delta.entries {
                match entry {
                    Some(entry) => {
                        manifest.entries.insert(path, entry);
                    }
                    None => {
                        manifest.entries.remove(&path);
                    }
                }
            }
            Ok(manifest)
        }
        _ => anyhow::bail!("unknown manifest history record kind"),
    }
}
fn compress(value: &impl Serialize) -> anyhow::Result<Vec<u8>> {
    Ok(zstd::encode_all(
        serde_json::to_vec(value)?.as_slice(),
        COMPRESSION_LEVEL,
    )?)
}
fn decompress<T: for<'de> Deserialize<'de>>(payload: &[u8]) -> anyhow::Result<T> {
    Ok(serde_json::from_slice(&zstd::decode_all(payload)?)?)
}