//! # openfut-identity
//!
//! A **generic, game-scoped external-identity store**: a durable, reversible
//! mapping between an opaque OpenFUT core id and a game's external wire id,
//! keyed by `(game_id, entity_kind, core_id)`.
//!
//! It is game-independent infrastructure. It knows nothing about FIFA — a game
//! adapter supplies the numeric **policy** (e.g. FIFA 17 owned-item ids start at
//! `100_000_000`) by passing a `base_floor`; the store only guarantees the
//! structural properties every adapter needs:
//!
//! * **stable** — the same `(game, kind, core_id)` always resolves to the same
//!   external id, including after restart;
//! * **unique** — two distinct core ids in a scope never share an external id;
//! * **reversible** — an external id resolves back to its exact core id;
//! * **game-scoped** — `fifa17` and a future `fifa23` allocate in isolation;
//! * **persistent** — not process memory;
//! * **atomic** — allocation is serialized (one host process owns the file) and
//!   persisted with a temp+rename so a crash cannot leave a torn mapping;
//! * **explicit** — an unknown core id or external id returns `None`, never a
//!   fabricated value.
//!
//! Core never learns the external (FIFA) integer: only the host/adapter that
//! owns a game boundary uses this store. Persistence is a small JSON file (the
//! smallest durable mechanism; the host uses no database today). The
//! [`ExternalIdentityStore`] trait keeps that a swappable detail — a SQLite
//! implementation can drop in later without touching callers.

use std::collections::HashMap;
use std::path::{Path, PathBuf};

use parking_lot::Mutex;
use serde::{Deserialize, Serialize};

/// Store errors.
#[derive(Debug)]
pub enum IdError {
    Io(std::io::Error),
    Corrupt(String),
    /// A caller-supplied import mapping conflicts with an existing one.
    Conflict(String),
}

impl std::fmt::Display for IdError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            IdError::Io(e) => write!(f, "identity store io: {e}"),
            IdError::Corrupt(e) => write!(f, "identity store corrupt: {e}"),
            IdError::Conflict(e) => write!(f, "identity store conflict: {e}"),
        }
    }
}
impl std::error::Error for IdError {}

/// A generic, durable, reversible external-identity mapping.
pub trait ExternalIdentityStore: Send + Sync {
    /// Return the external id for `(game, kind, core_id)`, allocating a new one
    /// if absent. New ids are monotonic within the `(game, kind)` scope, never
    /// below `base_floor` (the adapter's namespace policy). Idempotent: an
    /// existing mapping is returned unchanged (base_floor ignored then).
    fn resolve_or_allocate(
        &self,
        game: &str,
        kind: &str,
        core_id: &str,
        base_floor: i64,
    ) -> Result<i64, IdError>;

    /// Existing external id for a core id, or `None` (never allocates).
    fn external_for(&self, game: &str, kind: &str, core_id: &str) -> Result<Option<i64>, IdError>;

    /// Reverse lookup: the core id that owns `external_id`, or `None`.
    fn core_for(&self, game: &str, kind: &str, external_id: i64)
        -> Result<Option<String>, IdError>;
}

/// One persisted mapping row.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Row {
    game_id: String,
    entity_kind: String,
    core_id: String,
    external_id: i64,
}

/// A persisted per-scope allocator watermark: the next external id to issue,
/// imported from a source system so historical allocation (including ids burned
/// in gaps) is preserved, not just currently-live mappings.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Watermark {
    game_id: String,
    entity_kind: String,
    next_id: i64,
}

/// On-disk store file. Backward-compatible: a legacy bare `[Row, ...]` array is
/// still accepted on load; new writes always use this object form so watermarks
/// persist alongside the rows.
#[derive(Debug, Default, Serialize, Deserialize)]
struct StoreFile {
    #[serde(default)]
    rows: Vec<Row>,
    #[serde(default)]
    watermarks: Vec<Watermark>,
}

#[derive(Default)]
struct Index {
    rows: Vec<Row>,
    fwd: HashMap<(String, String, String), i64>,
    rev: HashMap<(String, String, i64), String>,
    max: HashMap<(String, String), i64>,
    /// Per-scope imported allocation watermark (next id to issue): a floor that
    /// preserves source allocation history beyond the live `max`.
    watermarks: HashMap<(String, String), i64>,
}

impl Index {
    fn insert(&mut self, row: Row) {
        let scope = (row.game_id.clone(), row.entity_kind.clone());
        self.fwd.insert(
            (
                row.game_id.clone(),
                row.entity_kind.clone(),
                row.core_id.clone(),
            ),
            row.external_id,
        );
        self.rev.insert(
            (
                row.game_id.clone(),
                row.entity_kind.clone(),
                row.external_id,
            ),
            row.core_id.clone(),
        );
        let m = self.max.entry(scope).or_insert(i64::MIN);
        if row.external_id > *m {
            *m = row.external_id;
        }
        self.rows.push(row);
    }
}

/// The next external id to issue in a scope: the maximum of the namespace floor,
/// one past the highest live mapping, and any imported watermark. Taking the max
/// of all three preserves source allocation history (burned-id gaps) while never
/// reissuing a live id nor dropping below the adapter's policy floor.
fn next_external(ix: &Index, game: &str, kind: &str, base_floor: i64) -> i64 {
    let scope = (game.to_string(), kind.to_string());
    let from_max = ix.max.get(&scope).map(|m| m + 1);
    let from_wm = ix.watermarks.get(&scope).copied();
    [Some(base_floor), from_max, from_wm]
        .into_iter()
        .flatten()
        .max()
        .unwrap_or(base_floor)
}

/// JSON-file backed [`ExternalIdentityStore`]. Single-writer (one host process).
pub struct JsonIdentityStore {
    path: PathBuf,
    inner: Mutex<Index>,
}

impl JsonIdentityStore {
    /// Open (or create) the store at `path`, loading any existing mappings.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, IdError> {
        let path = path.as_ref().to_path_buf();
        let mut index = Index::default();
        if path.exists() {
            let raw = std::fs::read_to_string(&path).map_err(IdError::Io)?;
            if !raw.trim().is_empty() {
                // Accept the modern object form; fall back to the legacy bare
                // `[Row, ...]` array so pre-watermark stores keep loading.
                let file: StoreFile = match serde_json::from_str::<StoreFile>(&raw) {
                    Ok(f) => f,
                    Err(_) => {
                        let rows: Vec<Row> = serde_json::from_str(&raw)
                            .map_err(|e| IdError::Corrupt(e.to_string()))?;
                        StoreFile {
                            rows,
                            watermarks: Vec::new(),
                        }
                    }
                };
                for row in file.rows {
                    // Reject a torn file that violates reverse-uniqueness rather
                    // than silently serving an ambiguous reverse lookup.
                    let rkey = (
                        row.game_id.clone(),
                        row.entity_kind.clone(),
                        row.external_id,
                    );
                    if index.rev.contains_key(&rkey) {
                        return Err(IdError::Corrupt(format!(
                            "duplicate external id {} in scope ({}, {})",
                            row.external_id, row.game_id, row.entity_kind
                        )));
                    }
                    index.insert(row);
                }
                for wm in file.watermarks {
                    index
                        .watermarks
                        .insert((wm.game_id, wm.entity_kind), wm.next_id);
                }
            }
        }
        Ok(JsonIdentityStore {
            path,
            inner: Mutex::new(index),
        })
    }

    /// Atomically persist the full row set (temp + rename).
    fn persist(&self, index: &Index) -> Result<(), IdError> {
        let file = StoreFile {
            rows: index.rows.clone(),
            watermarks: index
                .watermarks
                .iter()
                .map(|((g, k), &n)| Watermark {
                    game_id: g.clone(),
                    entity_kind: k.clone(),
                    next_id: n,
                })
                .collect(),
        };
        let json = serde_json::to_vec_pretty(&file).map_err(|e| IdError::Corrupt(e.to_string()))?;
        let tmp = self.path.with_extension("tmp");
        std::fs::write(&tmp, &json).map_err(IdError::Io)?;
        std::fs::rename(&tmp, &self.path).map_err(IdError::Io)?;
        Ok(())
    }

    /// Register an EXISTING external id for `(game, kind, core_id)` instead of
    /// allocating one — the import path that preserves a source system's wire
    /// ids. Idempotent for an identical mapping; rejects a conflicting forward
    /// (core already mapped elsewhere) or reverse (external already owned)
    /// mapping with [`IdError::Conflict`]; persists both directions atomically.
    pub fn insert_existing_mapping(
        &self,
        game: &str,
        kind: &str,
        core_id: &str,
        external_id: i64,
    ) -> Result<(), IdError> {
        let mut ix = self.inner.lock();
        let fkey = (game.to_string(), kind.to_string(), core_id.to_string());
        let rkey = (game.to_string(), kind.to_string(), external_id);
        let fwd_existing = ix.fwd.get(&fkey).copied();
        let rev_existing = ix.rev.get(&rkey).cloned();
        if let Some(e) = fwd_existing {
            if e == external_id && rev_existing.as_deref() == Some(core_id) {
                return Ok(()); // identical mapping — idempotent
            }
            return Err(IdError::Conflict(format!(
                "core '{core_id}' already maps to {e} in ({game}, {kind}), not {external_id}"
            )));
        }
        if let Some(c) = rev_existing {
            return Err(IdError::Conflict(format!(
                "external {external_id} already owned by core '{c}' in ({game}, {kind}), not '{core_id}'"
            )));
        }
        ix.insert(Row {
            game_id: game.to_string(),
            entity_kind: kind.to_string(),
            core_id: core_id.to_string(),
            external_id,
        });
        if let Err(e) = self.persist(&ix) {
            // roll the in-memory insert back so memory and disk never diverge
            ix.rows.pop();
            ix.fwd.remove(&fkey);
            ix.rev.remove(&rkey);
            let scope = (game.to_string(), kind.to_string());
            let new_max = ix
                .rows
                .iter()
                .filter(|r| r.game_id == game && r.entity_kind == kind)
                .map(|r| r.external_id)
                .max();
            match new_max {
                Some(m) => {
                    ix.max.insert(scope, m);
                }
                None => {
                    ix.max.remove(&scope);
                }
            }
            return Err(e);
        }
        Ok(())
    }

    /// Set the per-scope allocator watermark (the next external id to issue),
    /// imported from the source system so future allocations continue past its
    /// historical high-water even where those ids left no live mapping (gaps).
    /// A floor: it never lowers the effective next id below the live max.
    pub fn set_watermark(&self, game: &str, kind: &str, next_id: i64) -> Result<(), IdError> {
        let mut ix = self.inner.lock();
        let scope = (game.to_string(), kind.to_string());
        let prev = ix.watermarks.insert(scope.clone(), next_id);
        if let Err(e) = self.persist(&ix) {
            match prev {
                Some(p) => {
                    ix.watermarks.insert(scope, p);
                }
                None => {
                    ix.watermarks.remove(&scope);
                }
            }
            return Err(e);
        }
        Ok(())
    }

    /// The imported watermark for a scope, if any.
    pub fn watermark_for(&self, game: &str, kind: &str) -> Option<i64> {
        self.inner
            .lock()
            .watermarks
            .get(&(game.to_string(), kind.to_string()))
            .copied()
    }

    /// The external id the next allocation in this scope WOULD receive, without
    /// allocating. For dry-run reporting.
    pub fn peek_next_external(&self, game: &str, kind: &str, base_floor: i64) -> i64 {
        let ix = self.inner.lock();
        next_external(&ix, game, kind, base_floor)
    }
}

impl ExternalIdentityStore for JsonIdentityStore {
    fn resolve_or_allocate(
        &self,
        game: &str,
        kind: &str,
        core_id: &str,
        base_floor: i64,
    ) -> Result<i64, IdError> {
        let mut ix = self.inner.lock();
        let fkey = (game.to_string(), kind.to_string(), core_id.to_string());
        if let Some(&ext) = ix.fwd.get(&fkey) {
            return Ok(ext);
        }
        let scope = (game.to_string(), kind.to_string());
        let next = next_external(&ix, game, kind, base_floor);
        let row = Row {
            game_id: game.to_string(),
            entity_kind: kind.to_string(),
            core_id: core_id.to_string(),
            external_id: next,
        };
        ix.insert(row);
        // Persist before returning; on failure roll the in-memory insert back so
        // memory and disk never diverge.
        if let Err(e) = self.persist(&ix) {
            ix.rows.pop();
            ix.fwd.remove(&fkey);
            ix.rev.remove(&(game.to_string(), kind.to_string(), next));
            // `max` may now be stale-high; recompute for the scope.
            let new_max = ix
                .rows
                .iter()
                .filter(|r| r.game_id == game && r.entity_kind == kind)
                .map(|r| r.external_id)
                .max();
            match new_max {
                Some(m) => {
                    ix.max.insert(scope, m);
                }
                None => {
                    ix.max.remove(&scope);
                }
            }
            return Err(e);
        }
        Ok(next)
    }

    fn external_for(&self, game: &str, kind: &str, core_id: &str) -> Result<Option<i64>, IdError> {
        let ix = self.inner.lock();
        Ok(ix
            .fwd
            .get(&(game.to_string(), kind.to_string(), core_id.to_string()))
            .copied())
    }

    fn core_for(
        &self,
        game: &str,
        kind: &str,
        external_id: i64,
    ) -> Result<Option<String>, IdError> {
        let ix = self.inner.lock();
        Ok(ix
            .rev
            .get(&(game.to_string(), kind.to_string(), external_id))
            .cloned())
    }
}

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

    fn store() -> (JsonIdentityStore, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let s = JsonIdentityStore::open(dir.path().join("ids.json")).unwrap();
        (s, dir)
    }

    const G: &str = "fifa17";
    const K: &str = "owned-item";
    const BASE: i64 = 100_000_001;

    #[test]
    fn allocation_is_stable_and_reversible() {
        let (s, _d) = store();
        let a = s.resolve_or_allocate(G, K, "oc-A", BASE).unwrap();
        // idempotent
        assert_eq!(s.resolve_or_allocate(G, K, "oc-A", BASE).unwrap(), a);
        assert_eq!(s.external_for(G, K, "oc-A").unwrap(), Some(a));
        // reversible
        assert_eq!(s.core_for(G, K, a).unwrap().as_deref(), Some("oc-A"));
    }

    #[test]
    fn first_allocation_respects_base_floor_then_monotonic() {
        let (s, _d) = store();
        let a = s.resolve_or_allocate(G, K, "oc-A", BASE).unwrap();
        let b = s.resolve_or_allocate(G, K, "oc-B", BASE).unwrap();
        assert_eq!(a, BASE);
        assert_eq!(b, BASE + 1);
        assert_ne!(a, b, "two owned items get distinct wire ids");
    }

    #[test]
    fn unknown_ids_are_explicit_none() {
        let (s, _d) = store();
        assert_eq!(s.external_for(G, K, "nope").unwrap(), None);
        assert_eq!(s.core_for(G, K, 999_999).unwrap(), None);
    }

    #[test]
    fn game_and_kind_scopes_are_isolated() {
        let (s, _d) = store();
        let f = s.resolve_or_allocate("fifa17", K, "oc-A", BASE).unwrap();
        let g = s.resolve_or_allocate("fifa23", K, "oc-A", BASE).unwrap();
        // same core id, different games -> independent allocations, both from base
        assert_eq!(f, BASE);
        assert_eq!(g, BASE);
        // reverse lookup respects the game scope
        assert_eq!(s.core_for("fifa17", K, f).unwrap().as_deref(), Some("oc-A"));
        assert_eq!(s.core_for("fifa23", K, f).unwrap().as_deref(), Some("oc-A"));
        // a different entity kind is its own scope
        let k2 = s
            .resolve_or_allocate("fifa17", "card-def", "oc-A", 1)
            .unwrap();
        assert_eq!(k2, 1);
    }

    #[test]
    fn mappings_persist_across_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("ids.json");
        let (a, b);
        {
            let s = JsonIdentityStore::open(&path).unwrap();
            a = s.resolve_or_allocate(G, K, "oc-A", BASE).unwrap();
            b = s.resolve_or_allocate(G, K, "oc-B", BASE).unwrap();
        }
        // reopen: same ids, reverse intact, next allocation continues monotonic
        let s = JsonIdentityStore::open(&path).unwrap();
        assert_eq!(s.external_for(G, K, "oc-A").unwrap(), Some(a));
        assert_eq!(s.external_for(G, K, "oc-B").unwrap(), Some(b));
        assert_eq!(s.core_for(G, K, a).unwrap().as_deref(), Some("oc-A"));
        let c = s.resolve_or_allocate(G, K, "oc-C", BASE).unwrap();
        assert_eq!(c, b + 1, "counter resumes after restart, no reuse");
    }

    #[test]
    fn corrupt_reverse_duplicate_is_rejected_on_open() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("ids.json");
        // hand-write a file with two core ids sharing one external id in a scope
        std::fs::write(
            &path,
            r#"[{"game_id":"fifa17","entity_kind":"owned-item","core_id":"a","external_id":5},
                {"game_id":"fifa17","entity_kind":"owned-item","core_id":"b","external_id":5}]"#,
        )
        .unwrap();
        assert!(
            JsonIdentityStore::open(&path).is_err(),
            "ambiguous reverse must not load"
        );
    }

    #[test]
    fn insert_existing_mapping_preserves_and_is_idempotent() {
        let (s, _d) = store();
        s.insert_existing_mapping(G, K, "oc-A", 100_004_600)
            .unwrap();
        assert_eq!(s.external_for(G, K, "oc-A").unwrap(), Some(100_004_600));
        assert_eq!(
            s.core_for(G, K, 100_004_600).unwrap().as_deref(),
            Some("oc-A")
        );
        // identical re-insert is idempotent
        s.insert_existing_mapping(G, K, "oc-A", 100_004_600)
            .unwrap();
    }

    #[test]
    fn insert_existing_mapping_rejects_conflicts() {
        let (s, _d) = store();
        s.insert_existing_mapping(G, K, "oc-A", 100_000_010)
            .unwrap();
        // same core -> different external: reject
        assert!(matches!(
            s.insert_existing_mapping(G, K, "oc-A", 100_000_011),
            Err(IdError::Conflict(_))
        ));
        // same external -> different core: reject
        assert!(matches!(
            s.insert_existing_mapping(G, K, "oc-B", 100_000_010),
            Err(IdError::Conflict(_))
        ));
    }

    #[test]
    fn watermark_preserves_allocation_history_past_gaps() {
        let (s, _d) = store();
        // a live mapping (highest surviving id) plus a source watermark that sits
        // ABOVE it because ids in between were issued then removed (burned gap).
        s.insert_existing_mapping(G, K, "oc-live", 100_004_605)
            .unwrap();
        s.set_watermark(G, K, 100_004_617).unwrap();
        assert_eq!(s.peek_next_external(G, K, BASE), 100_004_617);
        // first NEW allocation is the source next-to-issue, NOT live_max+1
        assert_eq!(
            s.resolve_or_allocate(G, K, "oc-new", BASE).unwrap(),
            100_004_617
        );
        // then continues monotonically
        assert_eq!(
            s.resolve_or_allocate(G, K, "oc-new2", BASE).unwrap(),
            100_004_618
        );
    }

    #[test]
    fn watermark_and_imported_ids_survive_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("ids.json");
        {
            let s = JsonIdentityStore::open(&path).unwrap();
            s.insert_existing_mapping(G, K, "oc-A", 100_000_001)
                .unwrap();
            s.insert_existing_mapping(G, K, "oc-B", 100_004_605)
                .unwrap();
            s.set_watermark(G, K, 100_004_617).unwrap();
        }
        let s = JsonIdentityStore::open(&path).unwrap();
        assert_eq!(s.external_for(G, K, "oc-A").unwrap(), Some(100_000_001));
        assert_eq!(
            s.core_for(G, K, 100_004_605).unwrap().as_deref(),
            Some("oc-B")
        );
        assert_eq!(s.watermark_for(G, K), Some(100_004_617));
        // sparse import: next allocation respects the persisted watermark
        assert_eq!(
            s.resolve_or_allocate(G, K, "oc-C", BASE).unwrap(),
            100_004_617
        );
    }
}