use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use chrono::{DateTime, Utc};
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::broadcast;

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ContextEventKindV1 {
    ToolCallRecorded,
    SessionMutated,
    KnowledgeRemembered,
    ArtifactStored,
    GraphBuilt,
    ProofAdded,
}

impl ContextEventKindV1 {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::ToolCallRecorded => "tool_call_recorded",
            Self::SessionMutated => "session_mutated",
            Self::KnowledgeRemembered => "knowledge_remembered",
            Self::ArtifactStored => "artifact_stored",
            Self::GraphBuilt => "graph_built",
            Self::ProofAdded => "proof_added",
        }
    }

    pub fn parse(s: &str) -> Self {
        match s.trim().to_lowercase().as_str() {
            "session_mutated" => Self::SessionMutated,
            "knowledge_remembered" => Self::KnowledgeRemembered,
            "artifact_stored" => Self::ArtifactStored,
            "graph_built" => Self::GraphBuilt,
            "proof_added" => Self::ProofAdded,
            _ => Self::ToolCallRecorded,
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContextEventV1 {
    pub id: i64,
    pub workspace_id: String,
    pub channel_id: String,
    pub kind: String,
    pub actor: Option<String>,
    pub timestamp: DateTime<Utc>,
    pub payload: Value,
}

#[derive(Clone)]
pub struct ContextBus {
    inner: Arc<Inner>,
}

struct Inner {
    conn: Mutex<Connection>,
    tx: broadcast::Sender<ContextEventV1>,
}

impl Default for ContextBus {
    fn default() -> Self {
        Self::new()
    }
}

impl ContextBus {
    pub fn new() -> Self {
        let path = default_db_path();
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let conn = Connection::open(path).expect("open context-os db");
        conn.execute_batch(
            "PRAGMA journal_mode=WAL;
             CREATE TABLE IF NOT EXISTS context_events (
               id INTEGER PRIMARY KEY AUTOINCREMENT,
               workspace_id TEXT NOT NULL,
               channel_id TEXT NOT NULL,
               kind TEXT NOT NULL,
               actor TEXT,
               timestamp TEXT NOT NULL,
               payload_json TEXT NOT NULL
             );
             CREATE INDEX IF NOT EXISTS idx_context_events_stream
               ON context_events(workspace_id, channel_id, id);",
        )
        .expect("init context-os db");

        let (tx, _) = broadcast::channel(1024);
        Self {
            inner: Arc::new(Inner {
                conn: Mutex::new(conn),
                tx,
            }),
        }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<ContextEventV1> {
        self.inner.tx.subscribe()
    }

    pub fn append(
        &self,
        workspace_id: &str,
        channel_id: &str,
        kind: &ContextEventKindV1,
        actor: Option<&str>,
        payload: Value,
    ) -> Option<ContextEventV1> {
        let ts = Utc::now();
        let payload_json = payload.to_string();

        let id = {
            let Ok(conn) = self.inner.conn.lock() else {
                return None;
            };
            let _ = conn.execute(
                "INSERT INTO context_events (workspace_id, channel_id, kind, actor, timestamp, payload_json)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    workspace_id,
                    channel_id,
                    kind.as_str(),
                    actor.map(str::to_string),
                    ts.to_rfc3339(),
                    payload_json
                ],
            );
            conn.last_insert_rowid()
        };

        let ev = ContextEventV1 {
            id,
            workspace_id: workspace_id.to_string(),
            channel_id: channel_id.to_string(),
            kind: kind.as_str().to_string(),
            actor: actor.map(str::to_string),
            timestamp: ts,
            payload,
        };
        let _ = self.inner.tx.send(ev.clone());
        Some(ev)
    }

    pub fn read(
        &self,
        workspace_id: &str,
        channel_id: &str,
        since: i64,
        limit: usize,
    ) -> Vec<ContextEventV1> {
        let limit = limit.clamp(1, 1000) as i64;
        let Ok(conn) = self.inner.conn.lock() else {
            return Vec::new();
        };
        let Ok(mut stmt) = conn.prepare(
            "SELECT id, workspace_id, channel_id, kind, actor, timestamp, payload_json
             FROM context_events
             WHERE workspace_id = ?1 AND channel_id = ?2 AND id > ?3
             ORDER BY id ASC
             LIMIT ?4",
        ) else {
            return Vec::new();
        };
        let rows = stmt
            .query_map(params![workspace_id, channel_id, since, limit], |row| {
                let ts_str: String = row.get(5)?;
                let ts = DateTime::parse_from_rfc3339(&ts_str)
                    .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc));
                let payload_str: String = row.get(6)?;
                let payload: Value = serde_json::from_str(&payload_str).unwrap_or(Value::Null);
                Ok(ContextEventV1 {
                    id: row.get(0)?,
                    workspace_id: row.get(1)?,
                    channel_id: row.get(2)?,
                    kind: row.get(3)?,
                    actor: row.get::<_, Option<String>>(4)?,
                    timestamp: ts,
                    payload,
                })
            })
            .ok();
        let Some(rows) = rows else {
            return Vec::new();
        };
        rows.flatten().collect()
    }
}

fn default_db_path() -> PathBuf {
    let data = crate::core::data_dir::lean_ctx_data_dir().unwrap_or_else(|_| PathBuf::from("."));
    data.join("context-os").join("context-os.db")
}

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

    #[test]
    fn append_and_read_roundtrip() {
        let bus = ContextBus::new();
        let ev = bus
            .append(
                "ws",
                "ch",
                &ContextEventKindV1::ToolCallRecorded,
                Some("agent"),
                serde_json::json!({"tool":"ctx_read"}),
            )
            .expect("append");
        let got = bus.read("ws", "ch", ev.id - 1, 10);
        assert!(got.iter().any(|e| e.id == ev.id));
    }

    #[test]
    fn multi_client_concurrent_appends_have_deterministic_ordering() {
        let bus = Arc::new(ContextBus::new());
        let n_clients = 5;
        let n_events_per_client = 20;
        let ws = format!("ws-concurrent-{}", std::process::id());
        let ch = format!("ch-concurrent-{}", std::process::id());

        let mut handles = vec![];
        for client_idx in 0..n_clients {
            let bus = Arc::clone(&bus);
            let ws = ws.clone();
            let ch = ch.clone();
            handles.push(std::thread::spawn(move || {
                let agent = format!("agent-{client_idx}");
                for event_idx in 0..n_events_per_client {
                    bus.append(
                        &ws,
                        &ch,
                        &ContextEventKindV1::ToolCallRecorded,
                        Some(&agent),
                        serde_json::json!({"client": client_idx, "seq": event_idx}),
                    );
                }
            }));
        }

        for h in handles {
            h.join().unwrap();
        }

        let all = bus.read(&ws, &ch, 0, 1000);
        assert_eq!(
            all.len(),
            n_clients * n_events_per_client,
            "all events should be persisted"
        );

        let ids: Vec<i64> = all.iter().map(|e| e.id).collect();
        let mut sorted = ids.clone();
        sorted.sort_unstable();
        assert_eq!(ids, sorted, "events must be in strictly ascending ID order");

        for win in ids.windows(2) {
            assert!(
                win[1] > win[0],
                "IDs must be strictly monotonic (no gaps from concurrent access)"
            );
        }
    }

    #[test]
    fn workspace_channel_isolation() {
        let bus = ContextBus::new();
        let pid = std::process::id();
        let ws_a = format!("ws-iso-a-{pid}");
        let ws_b = format!("ws-iso-b-{pid}");
        let ws_c = format!("ws-iso-c-{pid}");
        let ch1 = format!("ch-iso-1-{pid}");
        let ch2 = format!("ch-iso-2-{pid}");

        bus.append(
            &ws_a,
            &ch1,
            &ContextEventKindV1::SessionMutated,
            Some("agent-a"),
            serde_json::json!({"ws":"a","ch":"1"}),
        );
        bus.append(
            &ws_a,
            &ch2,
            &ContextEventKindV1::KnowledgeRemembered,
            Some("agent-a"),
            serde_json::json!({"ws":"a","ch":"2"}),
        );
        bus.append(
            &ws_b,
            &ch1,
            &ContextEventKindV1::ArtifactStored,
            Some("agent-b"),
            serde_json::json!({"ws":"b","ch":"1"}),
        );

        let ws_a_ch_1 = bus.read(&ws_a, &ch1, 0, 100);
        assert_eq!(ws_a_ch_1.len(), 1);
        assert_eq!(ws_a_ch_1[0].kind, "session_mutated");

        let ws_a_ch_2 = bus.read(&ws_a, &ch2, 0, 100);
        assert_eq!(ws_a_ch_2.len(), 1);
        assert_eq!(ws_a_ch_2[0].kind, "knowledge_remembered");

        let ws_b_ch_1 = bus.read(&ws_b, &ch1, 0, 100);
        assert_eq!(ws_b_ch_1.len(), 1);
        assert_eq!(ws_b_ch_1[0].kind, "artifact_stored");

        let ws_c_ch_1 = bus.read(&ws_c, &ch1, 0, 100);
        assert!(ws_c_ch_1.is_empty(), "non-existent workspace returns empty");
    }

    #[test]
    fn replay_from_cursor_returns_only_newer_events() {
        let bus = ContextBus::new();
        let pid = std::process::id();
        let ws = &format!("ws-replay-{pid}");
        let ch = &format!("ch-replay-{pid}");

        let ev1 = bus
            .append(
                ws,
                ch,
                &ContextEventKindV1::ToolCallRecorded,
                None,
                serde_json::json!({"seq":1}),
            )
            .unwrap();
        let ev2 = bus
            .append(
                ws,
                ch,
                &ContextEventKindV1::SessionMutated,
                None,
                serde_json::json!({"seq":2}),
            )
            .unwrap();
        let _ev3 = bus
            .append(
                ws,
                ch,
                &ContextEventKindV1::GraphBuilt,
                None,
                serde_json::json!({"seq":3}),
            )
            .unwrap();

        let from_cursor = bus.read(ws, ch, ev2.id, 100);
        assert_eq!(from_cursor.len(), 1, "only events after cursor");
        assert_eq!(from_cursor[0].kind, "graph_built");

        let from_first = bus.read(ws, ch, ev1.id, 100);
        assert_eq!(from_first.len(), 2, "events after first");

        let from_zero = bus.read(ws, ch, 0, 100);
        assert_eq!(from_zero.len(), 3, "all events from zero");
    }

    #[test]
    fn broadcast_subscriber_receives_events() {
        let bus = ContextBus::new();
        let mut rx = bus.subscribe();

        let ev = bus
            .append(
                "ws",
                "ch",
                &ContextEventKindV1::ProofAdded,
                Some("verifier"),
                serde_json::json!({"proof":"hash"}),
            )
            .unwrap();

        let received = rx.try_recv().expect("subscriber should receive event");
        assert_eq!(received.id, ev.id);
        assert_eq!(received.kind, "proof_added");
        assert_eq!(received.actor.as_deref(), Some("verifier"));
    }
}
