//! Aggregator-side session + lease store (Phase 9 S1-S3).
//!
//! A **session** is a TTL-bounded container that owns one or more
//! **tag leases**. Callers create a session, attach pins to it, and
//! either `commit` (make the tags permanent, drop tracking) or let
//! the session `expire` / `DELETE` it (unpin every tag on every peer).
//!
//! This solves the wizard-cancel problem: the client picks a repo,
//! we mint a session + pin under it, and if the browser closes we
//! reap automatically. No orphaned fleet-wide state.
//!
//! ## Persistence
//!
//! One JSON file at `path`. Rewritten on every mutation (small store,
//! aggregator only, correctness > throughput). Survives aggregator
//! restart so mid-flight leases don't leak.
//!
//! ## Concurrency
//!
//! One `RwLock<HashMap<SessionId, Session>>`. Handlers take the write
//! lock briefly (mutation + `save()`), the sweeper takes it for the
//! per-tick scan + reap. Peer fan-out RPCs happen **outside** the
//! lock so a slow peer never blocks other API calls.
//!
//! ## Auth
//!
//! Every session carries the caller's `namespace` at creation. The
//! aggregator enforces "same-namespace only" access on every session
//! endpoint. Admin callers see all sessions.

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;
use uuid::Uuid;

/// Opaque session id — UUID v4 hex, stable in the URL. Public so
/// HTTP handlers can parse from path params.
pub type SessionId = String;

/// A single tag that a session owns. On reap, the aggregator issues
/// a fleet-wide `DELETE /api/v2/tags/:name` for each of these.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LeasedTag {
    pub tag: String,
    /// 64-char lowercase hex — the blob the tag was pinned to. Kept
    /// so listing endpoints can show what the session is holding.
    pub blob_id_hex: String,
    /// When the pin actually landed on the fleet, unix seconds. For
    /// operator inspection; not consulted for reap decisions.
    pub pinned_at_unix: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Session {
    pub id: SessionId,
    /// Namespace the session belongs to. `None` means an admin-owned
    /// session (created via the root `api_token`). Enforced by the
    /// HTTP layer, not this module.
    pub namespace: Option<String>,
    pub created_at_unix: u64,
    /// When the sweeper will reap this session (unix seconds).
    /// Extended by `renew`; frozen by `commit`.
    pub expires_at_unix: u64,
    pub leases: Vec<LeasedTag>,
    /// Optional operator note, e.g. `"research-wizard draft"`.
    #[serde(default)]
    pub note: Option<String>,
    /// After `commit`, the session sticks around for a short grace
    /// window so the UI can still show "committed" state, but the
    /// sweeper stops treating expiry as a reap trigger. Reapable
    /// only via explicit `DELETE`.
    #[serde(default)]
    pub committed: bool,
}

/// Handle to the on-disk + in-memory session store.
#[derive(Clone)]
pub struct SessionStore {
    inner: Arc<RwLock<HashMap<SessionId, Session>>>,
    path: PathBuf,
}

impl SessionStore {
    /// Load from disk (empty map if the file doesn't exist). Called
    /// once at aggregator boot; concurrent handlers share the returned
    /// clone via `Arc`.
    pub fn load(path: PathBuf) -> Result<Self> {
        let map: HashMap<SessionId, Session> = if path.exists() {
            let bytes = std::fs::read(&path)
                .with_context(|| format!("read sessions store at {}", path.display()))?;
            serde_json::from_slice(&bytes).context("parse sessions store")?
        } else {
            HashMap::new()
        };
        Ok(Self {
            inner: Arc::new(RwLock::new(map)),
            path,
        })
    }

    /// Rewrite the whole store to disk. Called under the write lock.
    /// Small store, correctness > throughput.
    fn save_locked(&self, map: &HashMap<SessionId, Session>) -> Result<()> {
        if let Some(parent) = self.path.parent() {
            std::fs::create_dir_all(parent).ok();
        }
        let tmp = self.path.with_extension("json.tmp");
        std::fs::write(&tmp, serde_json::to_vec_pretty(map)?)?;
        std::fs::rename(&tmp, &self.path)?;
        Ok(())
    }

    pub async fn create(
        &self,
        namespace: Option<String>,
        ttl_secs: u64,
        note: Option<String>,
    ) -> Result<Session> {
        let now = now_unix();
        let sess = Session {
            id: Uuid::new_v4().simple().to_string(),
            namespace,
            created_at_unix: now,
            expires_at_unix: now.saturating_add(ttl_secs),
            leases: Vec::new(),
            note,
            committed: false,
        };
        let mut w = self.inner.write().await;
        w.insert(sess.id.clone(), sess.clone());
        self.save_locked(&w)?;
        Ok(sess)
    }

    pub async fn get(&self, id: &str) -> Option<Session> {
        self.inner.read().await.get(id).cloned()
    }

    /// List all sessions visible to a caller. `None` namespace ==
    /// admin (sees everything); `Some(ns)` filters to sessions owned
    /// by that namespace.
    pub async fn list(&self, namespace_filter: Option<&str>) -> Vec<Session> {
        let r = self.inner.read().await;
        let mut out: Vec<Session> = r
            .values()
            .filter(|s| match namespace_filter {
                None => true,
                Some(ns) => s.namespace.as_deref() == Some(ns),
            })
            .cloned()
            .collect();
        // Newest first — mirrors the pattern in dashboards.
        out.sort_by(|a, b| b.created_at_unix.cmp(&a.created_at_unix));
        out
    }

    /// Attach a tag lease to an existing session. Caller must have
    /// already fanned the actual pin out to peers. This just records
    /// the tag so the sweeper can reap it on expiry.
    pub async fn attach_lease(&self, id: &str, lease: LeasedTag) -> Result<Session> {
        let mut w = self.inner.write().await;
        let s = w
            .get_mut(id)
            .ok_or_else(|| anyhow::anyhow!("session {id} not found"))?;
        // Idempotent: dedupe by tag name so retries don't double-record.
        if !s.leases.iter().any(|l| l.tag == lease.tag) {
            s.leases.push(lease);
        }
        let out = s.clone();
        self.save_locked(&w)?;
        Ok(out)
    }

    /// Extend expiry. `ttl_secs` is absolute-from-now, not additive,
    /// so heartbeats stay idempotent — sending "1 hour" repeatedly
    /// pins expiry to `now + 1h` no matter how many arrive.
    pub async fn renew(&self, id: &str, ttl_secs: u64) -> Result<Session> {
        let mut w = self.inner.write().await;
        let s = w
            .get_mut(id)
            .ok_or_else(|| anyhow::anyhow!("session {id} not found"))?;
        s.expires_at_unix = now_unix().saturating_add(ttl_secs);
        let out = s.clone();
        self.save_locked(&w)?;
        Ok(out)
    }

    /// Freeze the session's tags: mark committed, stop reaping on
    /// expiry, but keep the record for a while so the UI can show
    /// history. Explicit `DELETE` still works.
    pub async fn commit(&self, id: &str) -> Result<Session> {
        let mut w = self.inner.write().await;
        let s = w
            .get_mut(id)
            .ok_or_else(|| anyhow::anyhow!("session {id} not found"))?;
        s.committed = true;
        let out = s.clone();
        self.save_locked(&w)?;
        Ok(out)
    }

    /// Remove a session from the store and return the tags that
    /// need to be unpinned across the fleet. Callers must actually
    /// issue the unpin fan-out — this module doesn't dial peers.
    pub async fn remove(&self, id: &str) -> Result<Option<Session>> {
        let mut w = self.inner.write().await;
        let removed = w.remove(id);
        if removed.is_some() {
            self.save_locked(&w)?;
        }
        Ok(removed)
    }

    /// Snapshot the currently-expired sessions (used by the sweeper).
    /// Split into snapshot-then-reap so the peer RPCs happen outside
    /// the store's write lock.
    pub async fn snapshot_expired(&self) -> Vec<Session> {
        let now = now_unix();
        let r = self.inner.read().await;
        r.values()
            .filter(|s| !s.committed && s.expires_at_unix <= now)
            .cloned()
            .collect()
    }
}

fn now_unix() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Spawn a background task that scans for expired sessions every
/// `tick`, calls `reap` for each expired one (which is responsible
/// for the actual per-peer unpin), then removes them from the store.
/// The reap callback owns fan-out so this module stays free of
/// clawstor RPC + config types.
pub fn spawn_sweeper<F, Fut>(
    store: SessionStore,
    tick: Duration,
    reap: F,
) -> tokio::task::JoinHandle<()>
where
    F: Fn(Session) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = ()> + Send + 'static,
{
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(tick);
        // Skip the immediate first tick — nothing to reap at boot.
        interval.tick().await;
        loop {
            interval.tick().await;
            let expired = store.snapshot_expired().await;
            for sess in expired {
                let id = sess.id.clone();
                reap(sess).await;
                let _ = store.remove(&id).await;
            }
        }
    })
}