//! Bitemporal EAV triple store backed by redb.

use std::collections::HashMap;
use std::path::Path;

use ontoref_types::{
    canonical,
    operation::{OpPayload, Operation},
    AttrId, EntityId, Hlc, OpId, Value,
};
use redb::{Database, TableDefinition};
use serde::{Deserialize, Serialize};

use crate::error::{Error, Result};
use crate::key;

const EAV: TableDefinition<&[u8], &[u8]> = TableDefinition::new("eav");
const VAE: TableDefinition<&[u8], ()> = TableDefinition::new("vae");

/// Stored body of an EAV row.
///
/// Field declaration order is lexicographic so the canonical encoding agrees
/// with sorted CBOR maps.
#[derive(Clone, Debug, Serialize, Deserialize)]
struct EavRow {
    op_id: OpId,
    /// `true` for a retraction (ADR-041): the attribute is removed as of
    /// `tx_time`. Defaults to `false` so rows written before retract
    /// support decode unchanged. `value` is `Null` for retractions.
    #[serde(default)]
    retracted: bool,
    valid_from: Hlc,
    value: Value,
}

/// EAV bitemporal triple store.
pub struct TripleStore {
    db: Database,
}

impl TripleStore {
    /// Open or create the store at `path`.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let db = Database::create(path).map_err(Error::from)?;
        let txn = db.begin_write()?;
        {
            let _ = txn.open_table(EAV)?;
            let _ = txn.open_table(VAE)?;
        }
        txn.commit()?;
        Ok(Self { db })
    }

    /// Apply an operation. `Noop` operations contribute no triples; an
    /// `EntityAssert` materialises one EAV row (and one VAE index row) per
    /// (attr, value) pair.
    pub fn apply(&self, op: &Operation) -> Result<()> {
        match &op.body.payload {
            // Noop, Witness, TierDescent, and KeySuccession contribute no
            // triples — they attest/mark provenance rather than assert any
            // (attr, value).
            OpPayload::Noop { .. }
            | OpPayload::Witness { .. }
            | OpPayload::TierDescent { .. }
            | OpPayload::KeySuccession { .. } => Ok(()),
            OpPayload::EntityAssert {
                attrs,
                entity,
                valid_from,
            } => self.apply_entity_assert(
                entity,
                attrs,
                op.body.timestamp,
                valid_from.unwrap_or(op.body.timestamp),
                op.id(),
            ),
            OpPayload::Retract { attrs, entity } => {
                self.apply_retract(entity, attrs, op.body.timestamp, op.id())
            }
        }
    }

    fn apply_entity_assert(
        &self,
        entity: &EntityId,
        attrs: &[(AttrId, Value)],
        tx_time: Hlc,
        valid_from: Hlc,
        op_id: OpId,
    ) -> Result<()> {
        let txn = self.db.begin_write()?;
        {
            let mut eav = txn.open_table(EAV)?;
            let mut vae = txn.open_table(VAE)?;
            for (attr, value) in attrs {
                let eav_k = key::eav_key(entity.as_str(), attr.as_str(), tx_time)?;
                let row = EavRow {
                    op_id,
                    retracted: false,
                    valid_from,
                    value: value.clone(),
                };
                let eav_v = canonical::encode(&row)?;
                eav.insert(eav_k.as_slice(), eav_v.as_slice())?;

                let value_h = key::value_hash(value);
                let vae_k = key::vae_key(attr.as_str(), value_h, entity.as_str(), tx_time)?;
                vae.insert(vae_k.as_slice(), ())?;
            }
        }
        txn.commit()?;
        Ok(())
    }

    /// Record a retraction (ADR-041) for each attribute of `entity` at
    /// `tx_time`: an `EavRow` with `retracted = true` and a `Null` value, so
    /// the bitemporal query (`entity_attrs`) stops returning the attribute
    /// as of that transaction time. Append-only — prior assertions remain in
    /// history, consistent with the bitemporal model.
    fn apply_retract(
        &self,
        entity: &EntityId,
        attrs: &[AttrId],
        tx_time: Hlc,
        op_id: OpId,
    ) -> Result<()> {
        let txn = self.db.begin_write()?;
        {
            let mut eav = txn.open_table(EAV)?;
            for attr in attrs {
                let eav_k = key::eav_key(entity.as_str(), attr.as_str(), tx_time)?;
                let row = EavRow {
                    op_id,
                    retracted: true,
                    valid_from: tx_time,
                    value: Value::Null,
                };
                let eav_v = canonical::encode(&row)?;
                eav.insert(eav_k.as_slice(), eav_v.as_slice())?;
            }
        }
        txn.commit()?;
        Ok(())
    }

    /// All `(attr, value)` pairs for `entity` in effect at `valid_at`,
    /// recorded as of `as_of`. For each attribute the assertion with the
    /// greatest `valid_from <= valid_at` and `tx_time <= as_of` wins
    /// (ties broken by larger `tx_time`).
    pub fn entity_attrs(
        &self,
        entity: &EntityId,
        as_of: Hlc,
        valid_at: Hlc,
    ) -> Result<Vec<(AttrId, Value)>> {
        let txn = self.db.begin_read().map_err(Error::from)?;
        let eav = txn.open_table(EAV)?;
        let prefix = key::eav_prefix_entity(entity.as_str())?;

        let mut winners: HashMap<String, (Hlc, Hlc, Value, bool)> = HashMap::new();

        for row in eav.range::<&[u8]>(prefix.as_slice()..)? {
            let (key_guard, value_guard) = row?;
            let key_bytes = key_guard.value();
            if !key_bytes.starts_with(&prefix) {
                break;
            }
            let (attr, tx_time) = key::parse_eav_key_attr_tx(key_bytes, entity.as_str())?;
            if tx_time > as_of {
                continue;
            }
            let eav_row: EavRow = canonical::decode(value_guard.value())?;
            if eav_row.valid_from > valid_at {
                continue;
            }

            let candidate = (
                eav_row.valid_from,
                tx_time,
                eav_row.value,
                eav_row.retracted,
            );
            let better = match winners.get(&attr) {
                None => true,
                Some((current_vf, current_tx, _, _)) => {
                    candidate.0 > *current_vf
                        || (candidate.0 == *current_vf && candidate.1 > *current_tx)
                }
            };
            if better {
                winners.insert(attr, candidate);
            }
        }

        // A winning retraction means the attribute is absent as of the
        // viewpoint — drop it from the result (ADR-041).
        let mut out: Vec<(AttrId, Value)> = winners
            .into_iter()
            .filter(|(_, (_, _, _, retracted))| !retracted)
            .map(|(attr, (_, _, value, _))| (AttrId::new(attr), value))
            .collect();
        out.sort_by(|a, b| a.0.cmp(&b.0));
        Ok(out)
    }

    /// Single attribute lookup. Returns `None` if no assertion is visible
    /// from the `(as_of, valid_at)` viewpoint.
    pub fn entity_attr(
        &self,
        entity: &EntityId,
        attr: &AttrId,
        as_of: Hlc,
        valid_at: Hlc,
    ) -> Result<Option<Value>> {
        let txn = self.db.begin_read().map_err(Error::from)?;
        let eav = txn.open_table(EAV)?;
        let prefix = key::eav_prefix_entity_attr(entity.as_str(), attr.as_str())?;
        let prefix_len = prefix.len();

        let mut winner: Option<(Hlc, Hlc, Value)> = None;

        for row in eav.range::<&[u8]>(prefix.as_slice()..)? {
            let (key_guard, value_guard) = row?;
            let key_bytes = key_guard.value();
            if !key_bytes.starts_with(&prefix) {
                break;
            }
            let tx_time = key::parse_eav_key_tx_only(key_bytes, prefix_len)?;
            if tx_time > as_of {
                continue;
            }
            let eav_row: EavRow = canonical::decode(value_guard.value())?;
            if eav_row.valid_from > valid_at {
                continue;
            }

            let candidate = (eav_row.valid_from, tx_time, eav_row.value);
            let better = match &winner {
                None => true,
                Some((cvf, ctx, _)) => {
                    candidate.0 > *cvf || (candidate.0 == *cvf && candidate.1 > *ctx)
                }
            };
            if better {
                winner = Some(candidate);
            }
        }

        Ok(winner.map(|(_, _, value)| value))
    }

    /// Entities whose `attr` equals `value` from the `(as_of, valid_at)`
    /// viewpoint. The reverse index narrows the candidate set; each
    /// candidate is then checked through the forward query to confirm the
    /// current visible value still matches (excluding entities whose later
    /// assertion overwrote the value).
    pub fn entities_by_value(
        &self,
        attr: &AttrId,
        value: &Value,
        as_of: Hlc,
        valid_at: Hlc,
    ) -> Result<Vec<EntityId>> {
        let value_h = key::value_hash(value);
        let prefix = key::vae_prefix_attr_value(attr.as_str(), value_h)?;
        let prefix_len = prefix.len();

        let mut candidates: HashMap<String, Hlc> = HashMap::new();
        {
            let txn = self.db.begin_read().map_err(Error::from)?;
            let vae = txn.open_table(VAE)?;
            for row in vae.range::<&[u8]>(prefix.as_slice()..)? {
                let (key_guard, _) = row?;
                let key_bytes = key_guard.value();
                if !key_bytes.starts_with(&prefix) {
                    break;
                }
                let (entity, tx_time) = key::parse_vae_key_entity_tx(key_bytes, prefix_len)?;
                if tx_time > as_of {
                    continue;
                }
                candidates
                    .entry(entity)
                    .and_modify(|current| *current = (*current).max(tx_time))
                    .or_insert(tx_time);
            }
        }

        let mut out = Vec::new();
        for entity in candidates.into_keys() {
            let entity_id = EntityId::new(entity);
            let current = self.entity_attr(&entity_id, attr, as_of, valid_at)?;
            if current.as_ref() == Some(value) {
                out.push(entity_id);
            }
        }
        out.sort();
        Ok(out)
    }
}
