/// GraphQL surface over the ontology graph, impact traversal, project state,
/// backlog, ADRs, cross-project config, and notifications — with Apollo
/// Federation v2 entity resolution and WebSocket subscriptions.
use std::net::SocketAddr;
use std::sync::Arc;

use async_graphql::{
    futures_util::{self, StreamExt as _},
    http::ALL_WEBSOCKET_PROTOCOLS,
    Context, Enum, Object, Schema, SimpleObject, Subscription,
};
use async_graphql_axum::{GraphQLProtocol, GraphQLRequest, GraphQLResponse, GraphQLWebSocket};
use axum::{
    extract::{ConnectInfo, Query, State, WebSocketUpgrade},
    http::{header::AUTHORIZATION, HeaderMap, HeaderValue},
    response::{Html, IntoResponse},
    routing::get,
    Extension, Router,
};
use tokio_stream::wrappers::BroadcastStream;

use crate::{
    api::{extract_bearer, AppState},
    cache::NclCache,
    federation::FederatedQuery,
    notifications::{NotificationEvent, NotificationView},
};

pub mod memory_types;
pub use crate::memory_types::{ChronicleEntry, Criterion, PanelSpec, RoadmapItem};

pub type OntoSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;

// ── Output types ─────────────────────────────────────────────────────────────

#[derive(SimpleObject, Clone)]
pub struct OntologyNode {
    pub slug: String,
    pub id: String,
    pub name: String,
    pub kind: String,
    pub description: Option<String>,
}

#[derive(SimpleObject)]
pub struct ImpactEntry {
    pub slug: String,
    pub node_id: String,
    pub node_name: String,
    pub depth: i32,
    pub direction: String,
    pub via: String,
}

#[derive(SimpleObject)]
pub struct ProjectIdentity {
    pub slug: String,
    pub name: String,
    pub kind: String,
    pub description: Option<String>,
}

#[derive(SimpleObject)]
pub struct BacklogItem {
    pub id: String,
    pub title: String,
    pub kind: String,
    pub priority: String,
    pub status: String,
}

#[derive(SimpleObject)]
pub struct AdrSummary {
    pub id: String,
    pub slug: String,
    pub title: String,
    pub status: String,
    pub date: Option<String>,
}

#[derive(SimpleObject)]
pub struct AdrRecord {
    pub id: String,
    pub slug: String,
    pub title: String,
    pub status: String,
    pub date: Option<String>,
    pub context: Option<String>,
    pub decision: Option<String>,
    pub consequences: Option<String>,
}

#[derive(SimpleObject)]
pub struct CrossProjectSummary {
    pub slug: String,
    pub kind: String,
    pub export_ok: bool,
    pub section_count: i32,
}

#[derive(SimpleObject)]
pub struct CrossProjectConflict {
    pub section: String,
    pub field: String,
    pub project_a: String,
    pub value_a: String,
    pub project_b: String,
    pub value_b: String,
}

#[derive(Enum, Clone, Copy, PartialEq, Eq)]
pub enum NotificationEventGql {
    OntologyChanged,
    AdrChanged,
    ReflectionChanged,
    Custom,
}

impl From<NotificationEvent> for NotificationEventGql {
    fn from(e: NotificationEvent) -> Self {
        match e {
            NotificationEvent::OntologyChanged => Self::OntologyChanged,
            NotificationEvent::AdrChanged => Self::AdrChanged,
            NotificationEvent::ReflectionChanged => Self::ReflectionChanged,
            NotificationEvent::Custom => Self::Custom,
        }
    }
}

#[derive(SimpleObject, Clone)]
pub struct NotificationGql {
    pub id: i64,
    pub project: String,
    pub event: NotificationEventGql,
    pub files: Vec<String>,
    pub timestamp: i64,
    pub source_actor: Option<String>,
    pub custom_kind: Option<String>,
    pub custom_title: Option<String>,
}

impl From<NotificationView> for NotificationGql {
    fn from(n: NotificationView) -> Self {
        Self {
            id: n.id as i64,
            project: n.project,
            event: n.event.into(),
            files: n.files,
            timestamp: n.timestamp as i64,
            source_actor: n.source_actor,
            custom_kind: n.custom_kind,
            custom_title: n.custom_title,
        }
    }
}

// ── Registry topology types ──────────────────────────────────────────────────

#[derive(SimpleObject)]
pub struct RegistryNamespacesGql {
    pub own: Vec<String>,
    pub replicate_to: Vec<String>,
    pub mirror_from: Option<String>,
    pub prefixes: Vec<String>,
}

#[derive(SimpleObject)]
pub struct RegistryEntryGql {
    pub id: String,
    pub endpoint: String,
    pub role: String,
    pub tls: bool,
    pub namespaces: RegistryNamespacesGql,
}

#[derive(SimpleObject)]
pub struct RegistryTopologyGql {
    pub participant: String,
    pub default_id: Option<String>,
    pub entries: Vec<RegistryEntryGql>,
    /// Derived participant namespaces: ["domains/<participant>/",
    /// "modes/<participant>/"]
    pub namespaces: Vec<String>,
}

// ── Vault status types (ADR-017 + ADR-019) ──────────────────────────────────

#[derive(SimpleObject)]
pub struct RecipientGroupGql {
    pub name: String,
    pub keys: Vec<String>,
}

#[derive(SimpleObject)]
pub struct RecipientRuleGql {
    pub path: String,
    pub groups: Vec<String>,
}

#[derive(SimpleObject)]
pub struct ActorBindingGql {
    pub actor: String,
    pub role: String,
}

#[derive(SimpleObject)]
pub struct VaultLastOpGql {
    pub ts: String,
    pub actor: String,
    pub op: String,
    pub registry_id: Option<String>,
    pub detail: Option<String>,
}

#[derive(SimpleObject)]
pub struct VaultStatusGql {
    pub vault_id: String,
    pub vault_dir: String,
    pub backend: String,
    pub registry_endpoint: Option<String>,
    pub master_key: Option<String>,
    pub access_sops_present: bool,
    pub recipient_count: i32,
    pub recipient_groups: Vec<RecipientGroupGql>,
    pub recipient_rules: Vec<RecipientRuleGql>,
    pub actor_key_bindings: Vec<ActorBindingGql>,
    pub last_operation: Option<VaultLastOpGql>,
}

// ── Domain types ─────────────────────────────────────────────────────────────

#[derive(SimpleObject, Clone)]
pub struct DomainDeclGql {
    pub id: String,
    pub kind: String,
    pub path: String,
}

#[derive(SimpleObject, Clone)]
pub struct DomainConsumerGql {
    pub slug: String,
    pub root: String,
    pub kind: String,
}

#[derive(SimpleObject, Clone)]
pub struct DomainMetaGql {
    pub id: String,
    pub name: String,
    pub description: String,
    pub schema_cmd: String,
    pub consumers: Vec<DomainConsumerGql>,
}

// ── Tier types (ADR-029) ─────────────────────────────────────────────────────

#[derive(SimpleObject, Clone)]
pub struct TierCapabilityGql {
    pub id: String,
    pub label: String,
    pub min_tier: String,
    pub description: String,
}

/// Current tier + capability split for a project.
#[derive(SimpleObject, Clone)]
pub struct TierStatusGql {
    pub slug: String,
    pub tier: String,
    pub phase: String,
    pub available: Vec<TierCapabilityGql>,
    pub locked: Vec<TierCapabilityGql>,
}

/// Forward projection: what the next tier adds and why.
#[derive(SimpleObject, Clone)]
pub struct TierProjectionGql {
    pub from: String,
    pub to: Option<String>,
    pub adds: Vec<TierCapabilityGql>,
    pub narrative: String,
}

/// Runtime config + project beacon combined.
#[derive(SimpleObject, Clone)]
pub struct ProjectConfigGql {
    pub slug: String,
    /// Full config.ncl export as JSON string (null when absent).
    pub config_json: Option<String>,
    /// Full card.ncl export as JSON string (null when absent).
    pub card_json: Option<String>,
}

// ── Query root ───────────────────────────────────────────────────────────────

pub struct QueryRoot;

fn collect_conflicts(
    index: &std::collections::BTreeMap<(String, String), Vec<(String, String)>>,
) -> Vec<CrossProjectConflict> {
    let mut conflicts = Vec::new();
    for ((section, field), entries) in index {
        for i in 0..entries.len() {
            for j in (i + 1)..entries.len() {
                if entries[i].1 != entries[j].1 {
                    conflicts.push(CrossProjectConflict {
                        section: section.clone(),
                        field: field.clone(),
                        project_a: entries[i].0.clone(),
                        value_a: entries[i].1.clone(),
                        project_b: entries[j].0.clone(),
                        value_b: entries[j].1.clone(),
                    });
                }
            }
        }
    }
    conflicts
}

#[Object]
impl QueryRoot {
    /// Resolve a single ontology node by project slug and node id.
    async fn ontology_node(
        &self,
        ctx: &Context<'_>,
        slug: String,
        id: String,
    ) -> async_graphql::Result<Option<OntologyNode>> {
        let state = ctx.data::<AppState>()?;
        let fed = FederatedQuery::new(state.registry.clone());
        Ok(fed.resolve(&slug, &id).await.map(|f| {
            let name = str_field(&f.node, "name").unwrap_or_else(|| f.node_id.clone());
            let kind = str_field(&f.node, "kind").unwrap_or_else(|| "unknown".to_owned());
            let description = str_field(&f.node, "description");
            OntologyNode {
                slug: f.slug,
                id: f.node_id,
                name,
                kind,
                description,
            }
        }))
    }

    /// BFS impact graph from a node, with optional cross-project traversal.
    async fn impact_graph(
        &self,
        ctx: &Context<'_>,
        slug: String,
        node_id: String,
        max_depth: Option<i32>,
        include_external: Option<bool>,
    ) -> async_graphql::Result<Vec<ImpactEntry>> {
        let state = ctx.data::<AppState>()?;
        let fed = FederatedQuery::new(state.registry.clone());
        let depth = max_depth.unwrap_or(3).max(0) as u32;
        let entries = fed
            .impact_graph(&slug, &node_id, depth, include_external.unwrap_or(false))
            .await;
        Ok(entries
            .into_iter()
            .map(|e| ImpactEntry {
                slug: e.slug,
                node_id: e.node_id,
                node_name: e.node_name,
                depth: e.depth as i32,
                direction: e.direction,
                via: e.via,
            })
            .collect())
    }

    /// Project identity from core.ncl (slug defaults to primary project).
    async fn project(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<Option<ProjectIdentity>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let path = ontoref_ontology::layout::resolve_section(&root, "ontology/core.ncl");
        if !path.exists() {
            return Ok(None);
        }
        let (json, _) = cache
            .export(&path, import_path.as_deref())
            .await
            .map_err(|e| async_graphql::Error::new(e.to_string()))?;

        let identity = json.get("identity").unwrap_or(&json);
        let name = str_field(identity, "name")
            .or_else(|| str_field(&json, "name"))
            .unwrap_or_else(|| "unknown".to_owned());
        let kind = str_field(identity, "kind")
            .or_else(|| str_field(&json, "kind"))
            .unwrap_or_else(|| "project".to_owned());
        let description =
            str_field(identity, "description").or_else(|| str_field(&json, "description"));
        let resolved_slug = slug.unwrap_or_else(|| state.registry.primary_slug().to_owned());
        Ok(Some(ProjectIdentity {
            slug: resolved_slug,
            name,
            kind,
            description,
        }))
    }

    /// Backlog items from reflection/backlog.ncl (slug defaults to primary).
    async fn backlog(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<Vec<BacklogItem>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let path = ontoref_ontology::layout::resolve_section(&root, "reflection/backlog.ncl");
        if !path.exists() {
            return Ok(vec![]);
        }
        let (json, _) = cache
            .export(&path, import_path.as_deref())
            .await
            .map_err(|e| async_graphql::Error::new(e.to_string()))?;
        let items = json
            .get("items")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        Ok(items
            .iter()
            .map(|item| BacklogItem {
                id: str_field(item, "id").unwrap_or_default(),
                title: str_field(item, "title").unwrap_or_default(),
                kind: str_field(item, "kind").unwrap_or_default(),
                priority: str_field(item, "priority").unwrap_or_default(),
                status: str_field(item, "status").unwrap_or_default(),
            })
            .collect())
    }

    /// List ADR summaries for a project (slug defaults to primary).
    async fn adrs(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<Vec<AdrSummary>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let adrs_dir = ontoref_ontology::layout::resolve_section(&root, "adrs");
        let resolved_slug = slug.unwrap_or_else(|| state.registry.primary_slug().to_owned());

        let entries = match std::fs::read_dir(&adrs_dir) {
            Ok(e) => e,
            Err(_) => return Ok(vec![]),
        };

        let mut summaries = Vec::new();
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("ncl") {
                continue;
            }
            let file_id = path
                .file_stem()
                .and_then(|s| s.to_str())
                .unwrap_or("")
                .to_owned();
            let Ok((json, _)) = cache.export(&path, import_path.as_deref()).await else {
                continue;
            };
            summaries.push(AdrSummary {
                id: str_field(&json, "id").unwrap_or_else(|| file_id.clone()),
                slug: resolved_slug.clone(),
                title: str_field(&json, "title").unwrap_or_else(|| file_id.clone()),
                status: str_field(&json, "status").unwrap_or_else(|| "unknown".to_owned()),
                date: str_field(&json, "date"),
            });
        }
        summaries.sort_by(|a, b| a.id.cmp(&b.id));
        Ok(summaries)
    }

    /// Full ADR record by id fragment (slug defaults to primary).
    async fn adr(
        &self,
        ctx: &Context<'_>,
        id: String,
        slug: Option<String>,
    ) -> async_graphql::Result<Option<AdrRecord>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let resolved_slug = slug.unwrap_or_else(|| state.registry.primary_slug().to_owned());
        let adrs_dir = ontoref_ontology::layout::resolve_section(&root, "adrs");

        let entries = match std::fs::read_dir(&adrs_dir) {
            Ok(e) => e,
            Err(_) => return Ok(None),
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("ncl") {
                continue;
            }
            let stem = path
                .file_stem()
                .and_then(|s| s.to_str())
                .unwrap_or("")
                .to_string();
            if !stem.contains(id.as_str()) {
                continue;
            }
            let (json, _) = cache
                .export(&path, import_path.as_deref())
                .await
                .map_err(|e| async_graphql::Error::new(e.to_string()))?;
            return Ok(Some(AdrRecord {
                id: str_field(&json, "id").unwrap_or_else(|| stem.clone()),
                slug: resolved_slug,
                title: str_field(&json, "title").unwrap_or_else(|| stem.clone()),
                status: str_field(&json, "status").unwrap_or_default(),
                date: str_field(&json, "date"),
                context: str_field(&json, "context"),
                decision: str_field(&json, "decision"),
                consequences: str_field(&json, "consequences"),
            }));
        }
        Ok(None)
    }

    /// Project summaries for all registered projects with config surfaces.
    async fn cross_project_summaries(
        &self,
        ctx: &Context<'_>,
    ) -> async_graphql::Result<Vec<CrossProjectSummary>> {
        let state = ctx.data::<AppState>()?;
        let candidates: Vec<_> = state.registry.all();

        let mut out = Vec::new();
        for proj in candidates {
            let Some(surface) = proj.config_surface.as_ref() else {
                continue;
            };
            let entry = surface.entry_point_path(&proj.root);
            let export_ok = proj
                .cache
                .export(&entry, proj.import_path.as_deref())
                .await
                .is_ok();
            out.push(CrossProjectSummary {
                slug: proj.slug.clone(),
                kind: format!("{:?}", surface.kind),
                export_ok,
                section_count: surface.sections.len() as i32,
            });
        }
        Ok(out)
    }

    /// Cross-project config conflicts — all pairs of projects with differing
    /// values for the same (section, field).
    async fn cross_project_conflicts(
        &self,
        ctx: &Context<'_>,
    ) -> async_graphql::Result<Vec<CrossProjectConflict>> {
        let state = ctx.data::<AppState>()?;
        let candidates: Vec<_> = state.registry.all();

        // (section, field) → Vec<(slug, value_string)>
        let mut index: std::collections::BTreeMap<(String, String), Vec<(String, String)>> =
            std::collections::BTreeMap::new();

        for proj in &candidates {
            let Some(surface) = proj.config_surface.as_ref() else {
                continue;
            };
            let entry = surface.entry_point_path(&proj.root);
            let Ok((json, _)) = proj.cache.export(&entry, proj.import_path.as_deref()).await else {
                continue;
            };
            for section in &surface.sections {
                let Some(sec_val) = json.get(&section.id).and_then(|v| v.as_object()) else {
                    continue;
                };
                for (field, val) in sec_val {
                    index
                        .entry((section.id.clone(), field.clone()))
                        .or_default()
                        .push((proj.slug.clone(), val.to_string()));
                }
            }
        }

        let conflicts = collect_conflicts(&index);
        Ok(conflicts)
    }

    /// Federation entity resolver — resolves OntologyNode by (slug, id).
    #[graphql(entity)]
    async fn find_ontology_node_by_slug_and_id(
        &self,
        ctx: &Context<'_>,
        slug: String,
        id: String,
    ) -> async_graphql::Result<Option<OntologyNode>> {
        self.ontology_node(ctx, slug, id).await
    }

    /// Registry topology declared in a project's manifest.ncl registry_provides
    /// field.
    async fn registry_topology(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<Option<RegistryTopologyGql>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let path = ontoref_ontology::layout::resolve_section(&root, "ontology/manifest.ncl");
        if !path.exists() {
            return Ok(None);
        }
        let (json, _) = cache
            .export(&path, import_path.as_deref())
            .await
            .map_err(|e| async_graphql::Error::new(e.to_string()))?;

        let Some(provides) = json.get("registry_provides") else {
            return Ok(None);
        };
        let participant = str_field(provides, "participant").unwrap_or_default();
        if participant.is_empty() {
            return Ok(None);
        }

        let regs = provides
            .get("registries")
            .and_then(|r| r.get("registries"))
            .and_then(|r| r.as_array())
            .cloned()
            .unwrap_or_default();
        let default_id = provides
            .get("registries")
            .and_then(|r| r.get("default"))
            .and_then(|v| v.as_str())
            .map(|s| s.to_owned());

        let entries = regs
            .iter()
            .map(|e| {
                let ns = e.get("namespaces").unwrap_or(&serde_json::Value::Null);
                RegistryEntryGql {
                    id: str_field(e, "id").unwrap_or_default(),
                    endpoint: str_field(e, "endpoint").unwrap_or_default(),
                    role: str_field(e, "role").unwrap_or_else(|| "dev".to_owned()),
                    tls: e.get("tls").and_then(|v| v.as_bool()).unwrap_or(true),
                    namespaces: RegistryNamespacesGql {
                        own: string_array(ns, "own"),
                        replicate_to: string_array(ns, "replicate_to"),
                        mirror_from: str_field(ns, "mirror_from"),
                        prefixes: string_array(ns, "prefixes"),
                    },
                }
            })
            .collect();

        let namespaces = vec![
            format!("domains/{participant}/"),
            format!("modes/{participant}/"),
        ];

        Ok(Some(RegistryTopologyGql {
            participant,
            default_id,
            entries,
            namespaces,
        }))
    }

    /// Canonical participant namespaces for a project id.
    /// Returns ["domains/<project_id>/", "modes/<project_id>/"].
    async fn participant_namespace(
        &self,
        _ctx: &Context<'_>,
        project_id: String,
    ) -> async_graphql::Result<Vec<String>> {
        Ok(vec![
            format!("domains/{project_id}/"),
            format!("modes/{project_id}/"),
        ])
    }

    /// Credential vault state declared in a project's
    /// .ontoref/project.ncl::sops block (ADR-017 + ADR-019). Read-only —
    /// never decrypts; only inspects declarative state and the local vault
    /// dir.
    async fn vault_status(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<Option<VaultStatusGql>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let project_ncl = root.join(".ontoref").join("project.ncl");
        if !project_ncl.exists() {
            return Ok(None);
        }
        let (project, _) = cache
            .export(&project_ncl, import_path.as_deref())
            .await
            .map_err(|e| async_graphql::Error::new(e.to_string()))?;

        let Some(sops) = project.get("sops").filter(|v| !v.is_null()) else {
            return Ok(None);
        };
        if !sops
            .get("enabled")
            .and_then(|v| v.as_bool())
            .unwrap_or(false)
        {
            return Ok(None);
        }

        let project_slug = str_field(&project, "slug").unwrap_or_default();
        let vault_id = sops
            .get("vault_id")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .unwrap_or(&project_slug)
            .to_string();
        let backend = sops
            .get("vault_backend")
            .and_then(|v| v.as_str())
            .unwrap_or("restic")
            .to_string();
        let registry_endpoint = sops
            .get("registry_endpoint")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(String::from);

        // master_key: project override → global config.
        let master_key = sops
            .get("master_key_path")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(String::from)
            .or_else(|| {
                let home = std::env::var_os("HOME").map(std::path::PathBuf::from)?;
                let global = home.join(".config").join("ontoref").join("config.ncl");
                if !global.exists() {
                    return None;
                }
                let out = std::process::Command::new("nickel")
                    .arg("export")
                    .arg(&global)
                    .output()
                    .ok()?;
                if !out.status.success() {
                    return None;
                }
                let v: serde_json::Value = serde_json::from_slice(&out.stdout).ok()?;
                v.get("vault")
                    .and_then(|x| x.get("master_key_path"))
                    .and_then(|s| s.as_str())
                    .filter(|s| !s.is_empty())
                    .map(String::from)
            });

        let home = std::env::var_os("HOME")
            .map(std::path::PathBuf::from)
            .ok_or_else(|| async_graphql::Error::new("HOME not resolvable"))?;
        let vault_dir = home
            .join(".config")
            .join("ontoref")
            .join("vaults")
            .join(&vault_id);
        let access_sops = vault_dir.join("access.sops.yaml");
        let logs_jsonl = vault_dir.join("logs").join("access.jsonl");

        let access_sops_present = access_sops.exists();
        let recipient_count: i32 = if access_sops_present {
            std::fs::read_to_string(&access_sops)
                .map(|s| {
                    s.lines()
                        .filter(|l| l.trim_start().starts_with("- recipient: age1"))
                        .count() as i32
                })
                .unwrap_or(0)
        } else {
            0
        };

        let recipient_groups: Vec<RecipientGroupGql> = sops
            .get("recipient_groups")
            .and_then(|v| v.as_object())
            .map(|map| {
                map.iter()
                    .map(|(k, v)| RecipientGroupGql {
                        name: k.clone(),
                        keys: v
                            .as_array()
                            .map(|a| {
                                a.iter()
                                    .filter_map(|x| x.as_str().map(String::from))
                                    .collect()
                            })
                            .unwrap_or_default(),
                    })
                    .collect()
            })
            .unwrap_or_default();

        let recipient_rules: Vec<RecipientRuleGql> = sops
            .get("recipient_rules")
            .and_then(|v| v.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|r| {
                        Some(RecipientRuleGql {
                            path: r.get("path")?.as_str()?.to_string(),
                            groups: r
                                .get("groups")?
                                .as_array()?
                                .iter()
                                .filter_map(|g| g.as_str().map(String::from))
                                .collect(),
                        })
                    })
                    .collect()
            })
            .unwrap_or_default();

        let actor_key_bindings: Vec<ActorBindingGql> = sops
            .get("actor_key_bindings")
            .and_then(|v| v.as_object())
            .map(|m| {
                m.iter()
                    .filter_map(|(k, v)| {
                        Some(ActorBindingGql {
                            actor: k.clone(),
                            role: v.as_str()?.to_string(),
                        })
                    })
                    .collect()
            })
            .unwrap_or_default();

        let last_operation = if logs_jsonl.exists() {
            std::fs::read_to_string(&logs_jsonl)
                .ok()
                .and_then(|s| s.lines().last().map(String::from))
                .and_then(|line| serde_json::from_str::<serde_json::Value>(&line).ok())
                .and_then(|v| {
                    Some(VaultLastOpGql {
                        ts: v.get("ts")?.as_str()?.to_string(),
                        actor: v.get("actor")?.as_str()?.to_string(),
                        op: v.get("op")?.as_str()?.to_string(),
                        registry_id: v
                            .get("registry_id")
                            .and_then(|x| x.as_str())
                            .map(String::from),
                        detail: v.get("detail").and_then(|x| x.as_str()).map(String::from),
                    })
                })
        } else {
            None
        };

        Ok(Some(VaultStatusGql {
            vault_id,
            vault_dir: vault_dir.display().to_string(),
            backend,
            registry_endpoint,
            master_key,
            access_sops_present,
            recipient_count,
            recipient_groups,
            recipient_rules,
            actor_key_bindings,
            last_operation,
        }))
    }

    /// Domain implementations for a project — what domains it consumes.
    async fn domain_list(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<Vec<DomainDeclGql>> {
        let state = ctx.data::<AppState>()?;
        let pctx = slug
            .as_deref()
            .and_then(|s| state.registry.get(s))
            .unwrap_or_else(|| state.registry.primary());
        let decls = crate::domain::project_domains(&pctx.root);
        Ok(decls
            .into_iter()
            .map(|d| DomainDeclGql {
                id: d.id,
                kind: format!("{:?}", d.kind),
                path: d.path,
            })
            .collect())
    }

    /// All registered projects implementing a given domain.
    async fn domain_consumers(
        &self,
        ctx: &Context<'_>,
        domain_id: String,
    ) -> async_graphql::Result<Vec<DomainConsumerGql>> {
        let state = ctx.data::<AppState>()?;
        let consumers = crate::domain::consumers_of(&state.registry, &domain_id);
        Ok(consumers
            .into_iter()
            .map(|c| DomainConsumerGql {
                slug: c.slug,
                root: c.root,
                kind: format!("{:?}", c.kind),
            })
            .collect())
    }

    /// Domain metadata + all its consumers.
    async fn domain_show(
        &self,
        ctx: &Context<'_>,
        domain_id: String,
    ) -> async_graphql::Result<Option<DomainMetaGql>> {
        let state = ctx.data::<AppState>()?;
        let ontoref_root = std::env::var("ONTOREF_ROOT")
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|_| state.project_root.clone());
        let Some(meta) = crate::domain::domain_meta(&ontoref_root, &domain_id) else {
            return Ok(None);
        };
        let consumers = crate::domain::consumers_of(&state.registry, &domain_id);
        Ok(Some(DomainMetaGql {
            id: meta.id,
            name: meta.name,
            description: meta.description,
            schema_cmd: meta.schema_cmd,
            consumers: consumers
                .into_iter()
                .map(|c| DomainConsumerGql {
                    slug: c.slug,
                    root: c.root,
                    kind: format!("{:?}", c.kind),
                })
                .collect(),
        }))
    }

    /// Current tier declaration and capability matrix for a project.
    async fn tier_status(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<TierStatusGql> {
        let state = ctx.data::<AppState>()?;
        let effective = slug
            .as_deref()
            .unwrap_or_else(|| state.registry.primary_slug());
        let pctx = state
            .registry
            .get(effective)
            .unwrap_or_else(|| state.registry.primary());
        let ops = pctx.ops();
        let view = ontoref_ops::capabilities::capability_view(ops.tier);

        let cap_gql = |c: ontoref_ops::capabilities::CapabilityRef| TierCapabilityGql {
            id: c.id.to_string(),
            label: c.label.to_string(),
            min_tier: format!("{:?}", c.min_tier),
            description: c.description.to_string(),
        };

        Ok(TierStatusGql {
            slug: pctx.slug.clone(),
            tier: format!("{:?}", ops.tier),
            phase: format!("{:?}", ops.phase),
            available: view.available.into_iter().map(cap_gql).collect(),
            locked: view.locked.into_iter().map(cap_gql).collect(),
        })
    }

    /// Forward projection — what the next tier brings and what it requires.
    async fn tier_next(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<TierProjectionGql> {
        let state = ctx.data::<AppState>()?;
        let effective = slug
            .as_deref()
            .unwrap_or_else(|| state.registry.primary_slug());
        let pctx = state
            .registry
            .get(effective)
            .unwrap_or_else(|| state.registry.primary());
        let ops = pctx.ops();
        let proj = ontoref_ops::capabilities::tier_projection(ops.tier);

        let cap_gql = |c: ontoref_ops::capabilities::CapabilityRef| TierCapabilityGql {
            id: c.id.to_string(),
            label: c.label.to_string(),
            min_tier: format!("{:?}", c.min_tier),
            description: c.description.to_string(),
        };

        Ok(TierProjectionGql {
            from: format!("{:?}", proj.from),
            to: proj.to.map(|t| format!("{:?}", t)),
            adds: proj.adds.into_iter().map(cap_gql).collect(),
            narrative: proj.narrative.to_string(),
        })
    }

    /// Chronicle timeline entries woven from Accepted ADRs and protocol
    /// migrations.
    async fn chronicle(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
        since: Option<String>,
        kind: Option<String>,
        domain: Option<String>,
    ) -> async_graphql::Result<Vec<ChronicleEntry>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        Ok(crate::memory::chronicle_entries(
            &root,
            &cache,
            import_path.as_deref(),
            since.as_deref(),
            kind.as_deref(),
            domain.as_deref(),
        )
        .await)
    }

    /// Roadmap items derived from the project backlog.
    async fn roadmap(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
        status: Option<String>,
        horizon: Option<String>,
    ) -> async_graphql::Result<Vec<RoadmapItem>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        Ok(crate::memory::roadmap_items(
            &root,
            &cache,
            import_path.as_deref(),
            status.as_deref(),
            horizon.as_deref(),
        )
        .await)
    }

    /// Criteria promoted to catalog validators (ADR-049/050).
    async fn criteria(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
        kind: Option<String>,
    ) -> async_graphql::Result<Vec<Criterion>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        Ok(
            crate::memory::criteria_list(&root, &cache, import_path.as_deref(), kind.as_deref())
                .await,
        )
    }

    /// Resolved panel specs from the presentation directory (ADR-053).
    async fn panels(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
        kind: Option<String>,
    ) -> async_graphql::Result<Vec<PanelSpec>> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        Ok(
            crate::memory::panels_list(&root, &cache, import_path.as_deref(), kind.as_deref())
                .await,
        )
    }

    /// Combined config.ncl + card.ncl for a project.
    async fn project_config(
        &self,
        ctx: &Context<'_>,
        slug: Option<String>,
    ) -> async_graphql::Result<ProjectConfigGql> {
        let state = ctx.data::<AppState>()?;
        let (root, cache, import_path) = project_ctx(state, slug.as_deref());
        let pctx = slug
            .as_deref()
            .and_then(|s| state.registry.get(s))
            .unwrap_or_else(|| state.registry.primary());

        let export_or_null = |path: std::path::PathBuf| {
            let cache = Arc::clone(&cache);
            let ip = import_path.clone();
            async move {
                if !path.exists() {
                    return None;
                }
                cache
                    .export(&path, ip.as_deref())
                    .await
                    .ok()
                    .map(|(v, _)| v.to_string())
            }
        };

        let config_json = export_or_null(root.join(".ontoref").join("config.ncl")).await;
        let card_json = export_or_null(root.join(".ontoref").join("card.ncl")).await;

        Ok(ProjectConfigGql {
            slug: pctx.slug.clone(),
            config_json,
            card_json,
        })
    }
}

// ── Mutation root ────────────────────────────────────────────────────────────

pub struct MutationRoot;

#[Object]
impl MutationRoot {
    /// Emit a custom notification on a project. Returns the new notification
    /// id.
    async fn emit_notification(
        &self,
        ctx: &Context<'_>,
        kind: String,
        title: String,
        project: Option<String>,
        payload_json: Option<String>,
    ) -> async_graphql::Result<i64> {
        let state = ctx.data::<AppState>()?;
        let slug = project
            .as_deref()
            .unwrap_or_else(|| state.registry.primary_slug());
        let payload = payload_json
            .as_deref()
            .map(serde_json::from_str)
            .transpose()
            .map_err(|e| async_graphql::Error::new(format!("payload_json: {e}")))?;
        let id = state
            .notifications
            .push_custom(slug, kind, title, payload, None, None);
        Ok(id as i64)
    }

    /// Propose a status change for a backlog item (creates a backlog_review
    /// notification pending admin approval).
    async fn propose_backlog_status(
        &self,
        ctx: &Context<'_>,
        id: String,
        proposed_status: String,
        slug: Option<String>,
    ) -> async_graphql::Result<bool> {
        let state = ctx.data::<AppState>()?;
        let project = slug
            .as_deref()
            .unwrap_or_else(|| state.registry.primary_slug());
        state.notifications.push_custom(
            project,
            "backlog_review",
            format!("Status proposal: {id} → {proposed_status}"),
            Some(serde_json::json!({ "id": id, "proposed_status": proposed_status })),
            None,
            None,
        );
        Ok(true)
    }

    /// Acknowledge specific notifications by id list. Returns count acked.
    async fn ack_notifications(
        &self,
        ctx: &Context<'_>,
        ids: Vec<i64>,
        actor_token: String,
        project: Option<String>,
    ) -> async_graphql::Result<i32> {
        let state = ctx.data::<AppState>()?;
        let slug = project
            .as_deref()
            .unwrap_or_else(|| state.registry.primary_slug());
        let count = ids
            .iter()
            .filter(|&&i| state.notifications.ack_one(slug, &actor_token, i as u64))
            .count();
        Ok(count as i32)
    }

    /// Acknowledge all pending notifications for an actor. Returns count acked.
    async fn ack_all_notifications(
        &self,
        ctx: &Context<'_>,
        actor_token: String,
        project: Option<String>,
    ) -> async_graphql::Result<i32> {
        let state = ctx.data::<AppState>()?;
        let slug = project
            .as_deref()
            .unwrap_or_else(|| state.registry.primary_slug());
        Ok(state.notifications.ack_all(slug, &actor_token) as i32)
    }
}

// ── Subscription root ────────────────────────────────────────────────────────

pub struct SubscriptionRoot;

#[Subscription]
impl SubscriptionRoot {
    /// Real-time notification stream. Optionally filter by project slug.
    async fn notifications(
        &self,
        ctx: &Context<'_>,
        project: Option<String>,
    ) -> async_graphql::Result<impl futures_util::Stream<Item = NotificationGql>> {
        let state = ctx.data::<AppState>()?;
        let rx = state.notifications.subscribe();
        Ok(BroadcastStream::new(rx)
            .filter_map(|r| std::future::ready(r.ok()))
            .filter(move |notif| {
                let keep = project
                    .as_deref()
                    .is_none_or(|p| p == notif.project.as_str());
                std::future::ready(keep)
            })
            .map(NotificationGql::from))
    }
}

// ── Helpers ──────────────────────────────────────────────────────────────────

fn string_array(v: &serde_json::Value, key: &str) -> Vec<String> {
    v.get(key)
        .and_then(|a| a.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|s| s.as_str().map(|s| s.to_owned()))
                .collect()
        })
        .unwrap_or_default()
}

fn str_field(v: &serde_json::Value, key: &str) -> Option<String> {
    v.get(key)?.as_str().map(|s| s.to_owned())
}

fn project_ctx(
    state: &AppState,
    slug: Option<&str>,
) -> (std::path::PathBuf, Arc<NclCache>, Option<String>) {
    if let Some(s) = slug {
        if let Some(ctx) = state.registry.get(s) {
            return (ctx.root.clone(), ctx.cache.clone(), ctx.import_path.clone());
        }
    }
    (
        state.project_root.clone(),
        state.cache.clone(),
        state.nickel_import_path.clone(),
    )
}

// ── Axum handlers ────────────────────────────────────────────────────────────

#[derive(serde::Deserialize)]
struct GraphiqlQuery {
    token: Option<String>,
}

/// GET /graphql — serves GraphiQL IDE.
///
/// When the primary project has auth enabled, a `?token=<password>` query
/// parameter is required. On success the token is pre-configured in GraphiQL
/// as the default `Authorization: Bearer` header so queries run without manual
/// setup. The same token is forwarded as a WebSocket `connectionParam` for
/// subscriptions.
async fn graphiql_handler(
    Extension(app_state): Extension<AppState>,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    Query(q): Query<GraphiqlQuery>,
) -> impl IntoResponse {
    // If a graphql read token is configured, require it as ?token=.
    if app_state.graphql_read_token_hash.is_some() {
        let provided = q.token.as_deref().unwrap_or("");
        if !app_state.verify_graphql_token(provided) {
            let mut headers = HeaderMap::new();
            if let Ok(val) = HeaderValue::from_str(&format!("Bearer {}", provided)) {
                headers.insert(AUTHORIZATION, val);
            }
            app_state.auth_rate_limiter.record_failure(addr.ip());
            tracing::warn!(peer_ip = %addr.ip(), endpoint = "GET /graphql (IDE)", "unauthorized GraphQL access attempt");
            return (
                axum::http::StatusCode::UNAUTHORIZED,
                axum::Json(serde_json::json!({"error": "invalid or missing GraphQL token — use ?token=<token>"})),
            ).into_response();
        }
    }

    let bearer_value = q.token.as_deref().map(|t| format!("Bearer {t}"));
    let mut builder = async_graphql::http::GraphiQLSource::build()
        .endpoint("/graphql")
        .subscription_endpoint("/graphql/ws");
    if let Some(ref bv) = bearer_value {
        builder = builder
            .header("Authorization", bv)
            .ws_connection_param("Authorization", bv);
    }
    Html(builder.finish()).into_response()
}

async fn graphql_handler(
    State(schema): State<OntoSchema>,
    Extension(app_state): Extension<AppState>,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    req: GraphQLRequest,
) -> impl IntoResponse {
    if let Some(token) = extract_bearer(&headers) {
        if !app_state.verify_graphql_token(token) {
            app_state.auth_rate_limiter.record_failure(addr.ip());
            tracing::warn!(peer_ip = %addr.ip(), endpoint = "POST /graphql", "unauthorized GraphQL access attempt");
            return (
                axum::http::StatusCode::UNAUTHORIZED,
                axum::Json(serde_json::json!({"error": "invalid GraphQL token"})),
            )
                .into_response();
        }
    } else if app_state.graphql_read_token_hash.is_some() {
        tracing::warn!(peer_ip = %addr.ip(), endpoint = "POST /graphql", "unauthenticated GraphQL request — token required");
        return (
            axum::http::StatusCode::UNAUTHORIZED,
            axum::Json(serde_json::json!({"error": "Authorization: Bearer <token> required"})),
        )
            .into_response();
    }
    let gql: GraphQLResponse = schema.execute(req.into_inner()).await.into();
    gql.into_response()
}

async fn graphql_ws_handler(
    State(schema): State<OntoSchema>,
    Extension(app_state): Extension<AppState>,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    protocol: GraphQLProtocol,
    ws: WebSocketUpgrade,
) -> impl IntoResponse {
    // HTTP-level auth — verify graphql token from Bearer header.
    if let Some(token) = extract_bearer(&headers) {
        if !app_state.verify_graphql_token(token) {
            app_state.auth_rate_limiter.record_failure(addr.ip());
            tracing::warn!(peer_ip = %addr.ip(), endpoint = "GET /graphql/ws", "unauthorized GraphQL WS upgrade attempt");
            return (
                axum::http::StatusCode::UNAUTHORIZED,
                axum::Json(serde_json::json!({"error": "invalid GraphQL token"})),
            )
                .into_response();
        }
    } else if app_state.graphql_read_token_hash.is_some() {
        tracing::warn!(peer_ip = %addr.ip(), endpoint = "GET /graphql/ws", "unauthenticated GraphQL WS upgrade — token required");
        return (
            axum::http::StatusCode::UNAUTHORIZED,
            axum::Json(serde_json::json!({"error": "Authorization: Bearer <token> required"})),
        )
            .into_response();
    }
    // Token forwarded into WS context for subscription resolvers if needed.
    let bearer = extract_bearer(&headers).map(|s| s.to_owned());
    ws.protocols(ALL_WEBSOCKET_PROTOCOLS)
        .on_upgrade(move |stream| async move {
            let mut data = async_graphql::Data::default();
            if let Some(token) = bearer {
                data.insert(token);
            }
            GraphQLWebSocket::new(stream, schema, protocol)
                .with_data(data)
                .serve()
                .await;
        })
}

// ── Router ───────────────────────────────────────────────────────────────────

pub fn graphql_router(state: &AppState) -> Router {
    let schema = Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
        .data(state.clone())
        .enable_federation()
        .finish();
    Router::new()
        .route("/", get(graphiql_handler).post(graphql_handler))
        .route("/ws", get(graphql_ws_handler))
        .layer(Extension(state.clone()))
        .with_state(schema)
}
