//! The owner bridge — grouping nodes that belong to the same *owner* even
//! though the graph shows no Link between them.
//!
//! This is the subtlest part of the design (v3.4 §6.1.2). Measured fact it
//! exists to handle: a `module-loopback` forwarder's input leg and output
//! leg have **no Link between them**, so walking Links alone from the
//! leaking output leg finds no inbound links at all — a dead end that reads
//! as "clean". The legs are related only by shared properties.
//!
//! ## The rule
//!
//! A union of keys, strongest first:
//!
//! | # | key | scope |
//! | --- | --- | --- |
//! | 1 | `node.link-group` | per module/filter instance |
//! | 2 | `pulse.module.id` | per pactl module |
//! | 3 | `client.id` | per **connection** |
//! | 4 | `application.process.id` | per process |
//!
//! ⚠️ **"Resolves" means the two legs carry the key AND the values are
//! EQUAL — not "the first key present".** A first-present implementation
//! reproduces the exact measured leak: for `gst-launch pulsesrc ! pulsesink`
//! both legs carry `client.id` (209 and 210) but the values *differ*, so
//! first-present stops at key 3, sees a mismatch, and concludes "different
//! owners". The legs are in fact one process (`application.process.id`
//! 20172 on both). So: try each key in order, and a key resolves only if
//! both legs carry it and the values are equal; otherwise fall through.
//!
//! ## Two exceptions, both guarding against mass over-exclusion
//!
//! 1. **Never bridge on key 4 when the value is pipewire-pulse's own PID**
//!    (v3.4 §6.1.2). Module-created streams all carry the daemon's PID, so
//!    bridging on it fuses every Pulse module into one owner and a single
//!    tainted module input would exclude every module-created stream on the
//!    box. Keys 1 and 2 already cover those cases precisely.
//!
//! 2. **Coarse keys (3 and 4) may not bridge nodes exported from a real
//!    `Device`** — i.e. nodes carrying `device.id`. ⚠️ This rule is *not*
//!    in design v3.4; it was found while implementing, and it is the exact
//!    analogue of exception 1 for the session manager.
//!    ✅ **MEASURED on the live graph 2026-07-21:**
//!
//!    | node | `client.id` | `device.id` | `factory.name` |
//!    | --- | --- | --- | --- |
//!    | 5 × `alsa_{output,input}.*` | **42** (`WirePlumber [export]`) | 43/45/46 | `api.alsa.pcm.{sink,source}` |
//!    | 3 × `sink-sunshine-*` | 83 / 86 / 92 (each its own) | **absent** | `support.null-audio-sink` |
//!
//!    So one shared coarse key genuinely does relate every hardware device
//!    on the box, and `device.id` cleanly separates that set from virtual
//!    sinks. Without the rule, the hardware sink carrying peerspeak's
//!    playback (tainted by design, every single recompute) would bridge to
//!    *every other device node including the microphone source*, whose
//!    readers would then taint their owners' playback legs — reproducing
//!    precisely the §6.1.1 catastrophe ("excludes any app using a
//!    microphone") through a different door.
//!
//!    ⚠️ **Keyed on `device.id`, NOT on `media.class` being `Audio/Sink`.**
//!    The first cut suppressed coarse keys for every device-*role* node,
//!    and Codex refuted it: a **native virtual sink** — an app that creates
//!    an `Audio/Sink` plus a re-emitting stream on one client, with no
//!    `link-group` and no `pulse.module.id` — would then have had its only
//!    correlation stripped, and it would have leaked the whole call. Such a
//!    sink has no `device.id`, so it now bridges on `client.id` as it
//!    should.
//!
//! Grouping is **transitive** (union-find). That is the fail-closed
//! direction: bigger owner components mean more taint, never less.

use std::collections::{BTreeMap, BTreeSet};

use super::snapshot::{GlobalId, GraphSnapshot, NodeSnapshot, Serial};

/// Everything owner-key derivation needs from outside a single node.
///
/// Introduced by round 10 (R10-3). Before it, `keys_of` read only node
/// properties, and key 4 was therefore available **only** to nodes carrying
/// `application.process.id` — which native PipeWire clients do not. mpv on its
/// default ao, and peerspeak's own playback stream, expose nothing but
/// `client.id`, so both were *unbounded*, and the moment any tainted reader
/// existed anywhere, `propagate_unresolved_owner` excluded every one of them.
/// Measured: an untagged mpv went from eligible (alone) to `unresolved-owner`
/// the instant peerspeak played audio. That is "native-PipeWire apps are never
/// shareable", which is not a feature.
///
/// The missing pid is not missing at all — it is one hop away, on the node's
/// **Client**, as `pipewire.sec.pid`, and already in the snapshot.
pub struct OwnerCtx {
    pub pipewire_pulse_pid: Option<u32>,
    /// `client.id` → that Client's `pipewire.sec.pid`.
    ///
    /// Clients whose global id is **ambiguous** (two live objects claiming it,
    /// i.e. the observer missed a removal) are deliberately absent: resolving
    /// an ambiguous id to a pid would attribute a node to whichever Client won
    /// a coin toss, and inventing an owner key is the one direction that can
    /// *reduce* taint. Absent ⇒ unbounded ⇒ fails closed, as before.
    client_pids: BTreeMap<GlobalId, u32>,
}

impl OwnerCtx {
    pub fn new(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option<u32>) -> Self {
        let mut client_pids: BTreeMap<GlobalId, u32> = BTreeMap::new();
        let mut ambiguous: BTreeSet<GlobalId> = BTreeSet::new();
        for client in snapshot.clients() {
            if client_pids.contains_key(&client.id) || ambiguous.contains(&client.id) {
                // Two Clients claiming one id: drop it entirely rather than
                // pick. See the field docs.
                client_pids.remove(&client.id);
                ambiguous.insert(client.id);
                continue;
            }
            if let Some(pid) = client.sec_pid {
                client_pids.insert(client.id, pid);
            }
        }
        Self {
            pipewire_pulse_pid,
            client_pids,
        }
    }

    /// The `pipewire.sec.pid` of this node's Client, if it has one and that
    /// Client's id is unambiguous.
    fn client_pid(&self, node: &NodeSnapshot) -> Option<u32> {
        self.client_pids.get(&node.props.client_id?).copied()
    }
}

/// Which key bridged two legs. Ordered strongest first; the `Ord` derive is
/// load-bearing for "report the strongest shared key".
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
pub enum OwnerKey {
    LinkGroup,
    PulseModuleId,
    ClientId,
    ProcessId,
}

impl OwnerKey {
    /// Stable, machine-readable — this ends up in the phase 5 audit output
    /// and the phase 6 status event.
    pub fn code(self) -> &'static str {
        match self {
            Self::LinkGroup => "node.link-group",
            Self::PulseModuleId => "pulse.module.id",
            Self::ClientId => "client.id",
            Self::ProcessId => "application.process.id",
        }
    }
}

/// The value a node presents for a given key, if it presents one at all.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
enum KeyValue {
    Text(String),
    Num(u64),
}

/// Owner keys usable on this node, strongest first.
///
/// A key that is present but unusable (the pipewire-pulse PID; a coarse key
/// on a device node) is **absent** here — that is the whole mechanism of the
/// two exceptions.
fn keys_of(node: &NodeSnapshot, ctx: &OwnerCtx) -> Vec<(OwnerKey, KeyValue)> {
    let mut out = Vec::new();
    if let Some(group) = &node.props.link_group {
        out.push((OwnerKey::LinkGroup, KeyValue::Text(group.clone())));
    }
    if let Some(module) = node.props.pulse_module_id {
        out.push((OwnerKey::PulseModuleId, KeyValue::Num(module)));
    }
    // Exception 2: coarse keys never bridge passive session-manager device
    // nodes — they all share the session manager's client.
    if node.props.session_device {
        return out;
    }
    if let Some(client) = node.props.client_id {
        out.push((OwnerKey::ClientId, KeyValue::Num(u64::from(client.0))));
    }
    // Key 4, from the node if it has one, otherwise from its Client
    // (round 10, R10-3). The node's own property wins when both exist: it is
    // the direct statement, and the Client's is a one-hop inference.
    //
    // ⚠️ **Exception 1 must apply to the fallback too, and this is the whole
    // risk of the fallback.** Measured on this host: 15 unrelated Clients
    // share `sec_pid` 2528, which is pipewire-pulse's own — every
    // Pulse-emulated app has one. Adding key 4 unguarded would fuse all
    // fifteen into a single owner and bridge taint between completely
    // unrelated applications, which is enormous over-exclusion. Guarded, the
    // fallback strictly *adds* correct bounding: it fires only for native
    // clients, which are exactly the ones that carry a real per-app pid here.
    if let Some(pid) = node.props.process_id.or_else(|| ctx.client_pid(node)) {
        // Note the fail-closed asymmetry when the daemon PID is unknown
        // (`None`): the exception does *not* fire, key 4 applies to
        // everything, and Pulse modules fuse into one owner. That is broad
        // over-exclusion — annoying and safe — which is the direction v3.4
        // §6.1.2's failure-mode paragraph asks for.
        if Some(pid) != ctx.pipewire_pulse_pid {
            out.push((OwnerKey::ProcessId, KeyValue::Num(u64::from(pid))));
        }
    }
    out
}

/// Can this node's owner be positively bounded — i.e. can we enumerate its
/// sibling legs and be right?
///
/// ⚠️ Not the same as "has any usable key", and the difference is a leak.
/// `client.id` alone does **not** bound an owner: that is the measured
/// GStreamer refutation, where one process presented two different
/// `client.id`s for its two legs. So an owner is bounded only by a strong
/// key (link-group / pulse.module.id) or by a *usable* process id — usable
/// meaning key 4 was not suppressed as pipewire-pulse's own PID.
///
/// The case this exists for is v3.4 §12's "module forwarder with neither
/// `link-group` nor `pulse.module.id`": its process id is the daemon's and
/// therefore suppressed, its two legs may carry different `client.id`s, and
/// nothing else relates them. Its sibling output leg cannot be found, so
/// the engine must fail closed rather than declare it clean
/// (v3.4 §6.1.1, final paragraph).
pub fn owner_is_bounded(node: &NodeSnapshot, ctx: &OwnerCtx) -> bool {
    keys_of(node, ctx)
        .iter()
        .any(|(key, _)| *key != OwnerKey::ClientId)
}

/// Owner keys computed once per snapshot.
///
/// `keys_of` allocates a `Vec` and clones the `link-group` string, and the
/// bridge asks for keys once per (tainted member × component member) pair —
/// so recomputing was the hot spot in an otherwise linear pass.
#[derive(Debug, Default)]
pub struct OwnerKeyIndex {
    keys: BTreeMap<Serial, Vec<(OwnerKey, KeyValue)>>,
}

impl OwnerKeyIndex {
    pub fn build(snapshot: &GraphSnapshot, ctx: &OwnerCtx) -> Self {
        Self {
            keys: snapshot
                .nodes()
                .map(|node| (node.serial, keys_of(node, ctx)))
                .collect(),
        }
    }

    /// The strongest key these two nodes share directly, if any.
    pub fn strongest_shared(&self, a: Serial, b: Serial) -> Option<OwnerKey> {
        let (Some(a_keys), Some(b_keys)) = (self.keys.get(&a), self.keys.get(&b)) else {
            return None;
        };
        // Stored strongest-first, so the first match is the strongest.
        a_keys.iter().find_map(|(key, value)| {
            b_keys
                .iter()
                .any(|(other_key, other_value)| other_key == key && other_value == value)
                .then_some(*key)
        })
    }

    /// Is `client.id` a usable owner key for this node?
    ///
    /// ⚠️ Load-bearing for sticky state. A device node's `client.id` is
    /// suppressed by exception 2, so recording the session manager's Client
    /// as a *member* of a tainted device's sticky owner would smuggle the
    /// suppressed key back in: the next recompute would expand that Client
    /// to every hardware node on the box — the microphone included — and
    /// the §6.1.1 catastrophe would arrive one epoch late instead of never.
    /// (Codex round 2, finding 1.)
    pub fn uses_client_key(&self, serial: Serial) -> bool {
        self.keys
            .get(&serial)
            .is_some_and(|keys| keys.iter().any(|(key, _)| *key == OwnerKey::ClientId))
    }

    /// The owner keys that are safe to remember *across* connections, for
    /// sticky taint: the strong keys plus a usable process id.
    ///
    /// `client.id` is deliberately excluded — it identifies a *connection*,
    /// and the whole point of a fingerprint is to survive one process
    /// closing a connection and opening another. A live Client member is
    /// what covers the same-connection case, precisely.
    ///
    /// These are recyclable strings and numbers, so they are only ever
    /// applied while some **serial** member of the owner is still live
    /// (v3.4 §6.1.3): while the process is alive, its PID cannot have been
    /// handed to anyone else.
    pub fn fingerprints(&self, serial: Serial) -> Vec<Fingerprint> {
        self.keys
            .get(&serial)
            .map(|keys| {
                keys.iter()
                    .filter(|(key, _)| *key != OwnerKey::ClientId)
                    .map(|(key, value)| Fingerprint(*key, value.clone()))
                    .collect()
            })
            .unwrap_or_default()
    }

    /// Does this node currently present `fingerprint`?
    pub fn has_fingerprint(&self, serial: Serial, fingerprint: &Fingerprint) -> bool {
        self.keys.get(&serial).is_some_and(|keys| {
            keys.iter()
                .any(|(key, value)| *key == fingerprint.0 && *value == fingerprint.1)
        })
    }

    /// See [`owner_is_bounded`].
    pub fn is_bounded(&self, serial: Serial) -> bool {
        self.keys
            .get(&serial)
            .is_some_and(|keys| keys.iter().any(|(key, _)| *key != OwnerKey::ClientId))
    }
}

/// The strongest key two nodes share, or `None` if they share none. Used to
/// *name* the key in a bridge decision; membership itself is transitive and
/// comes from [`OwnerComponents`].
pub fn strongest_shared_key(
    a: &NodeSnapshot,
    b: &NodeSnapshot,
    ctx: &OwnerCtx,
) -> Option<OwnerKey> {
    let a_keys = keys_of(a, ctx);
    let b_keys = keys_of(b, ctx);
    // `keys_of` yields strongest-first, so the first match is the strongest.
    a_keys.iter().find_map(|(key, value)| {
        b_keys
            .iter()
            .any(|(other_key, other_value)| other_key == key && other_value == value)
            .then_some(*key)
    })
}

/// A remembered owner key — see [`OwnerKeyIndex::fingerprints`].
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct Fingerprint(OwnerKey, KeyValue);

/// Nodes partitioned into owner components.
#[derive(Clone, Debug, Default)]
pub struct OwnerComponents {
    /// node serial → component index.
    of_node: BTreeMap<Serial, usize>,
    /// component index → member node serials, ascending.
    members: Vec<Vec<Serial>>,
}

impl OwnerComponents {
    pub fn build(snapshot: &GraphSnapshot, ctx: &OwnerCtx) -> Self {
        let serials: Vec<Serial> = snapshot.nodes().map(|n| n.serial).collect();
        let index: BTreeMap<Serial, usize> =
            serials.iter().enumerate().map(|(i, s)| (*s, i)).collect();
        let mut uf = UnionFind::new(serials.len());

        // Group by (key, value) and union within each group. Equivalent to
        // the pairwise "some key resolves" rule, and O(n log n).
        let mut buckets: BTreeMap<(OwnerKey, KeyValue), Vec<usize>> = BTreeMap::new();
        for node in snapshot.nodes() {
            let slot = index[&node.serial];
            for (key, value) in keys_of(node, ctx) {
                buckets.entry((key, value)).or_default().push(slot);
            }
        }
        for group in buckets.values() {
            for pair in group.windows(2) {
                uf.union(pair[0], pair[1]);
            }
        }

        // Compact roots into dense component indices, deterministically.
        let mut root_to_component: BTreeMap<usize, usize> = BTreeMap::new();
        let mut members: Vec<Vec<Serial>> = Vec::new();
        let mut of_node = BTreeMap::new();
        for (slot, serial) in serials.iter().enumerate() {
            let root = uf.find(slot);
            let component = *root_to_component.entry(root).or_insert_with(|| {
                members.push(Vec::new());
                members.len() - 1
            });
            members[component].push(*serial);
            of_node.insert(*serial, component);
        }
        Self { of_node, members }
    }

    pub fn component_of(&self, serial: Serial) -> Option<usize> {
        self.of_node.get(&serial).copied()
    }

    /// Member serials of the component containing `serial`, including it.
    /// Empty if the node is not in this snapshot.
    pub fn members_with(&self, serial: Serial) -> &[Serial] {
        match self.component_of(serial) {
            Some(component) => &self.members[component],
            None => &[],
        }
    }

    pub fn components(&self) -> impl Iterator<Item = &[Serial]> {
        self.members.iter().map(Vec::as_slice)
    }
}

struct UnionFind {
    parent: Vec<usize>,
}

impl UnionFind {
    fn new(len: usize) -> Self {
        Self {
            parent: (0..len).collect(),
        }
    }

    fn find(&mut self, mut node: usize) -> usize {
        while self.parent[node] != node {
            self.parent[node] = self.parent[self.parent[node]];
            node = self.parent[node];
        }
        node
    }

    fn union(&mut self, a: usize, b: usize) {
        let (a, b) = (self.find(a), self.find(b));
        if a != b {
            // Lowest root wins, so components are deterministic.
            let (low, high) = if a < b { (a, b) } else { (b, a) };
            self.parent[high] = low;
        }
    }
}

/// Client objects belonging to an owner component, so sticky taint can be
/// keyed on every object that constitutes the owner (v3.4 §6.1.3: clear the
/// entry only once **all** member objects are gone).
pub fn client_serials_of(
    snapshot: &GraphSnapshot,
    keys: &OwnerKeyIndex,
    nodes: &[Serial],
) -> Vec<Serial> {
    let mut out: Vec<Serial> = nodes
        .iter()
        // Only nodes for which `client.id` is a *usable* owner key. See
        // `uses_client_key`: recording a device node's shared session-manager
        // Client here would defeat exception 2 on the next recompute.
        .filter(|serial| keys.uses_client_key(**serial))
        .filter_map(|serial| snapshot.node(*serial))
        .filter_map(|node| node.props.client_id)
        // An ambiguous client id means two Clients claim it and we cannot
        // say which one is ours, so remember both: an entry that recorded
        // neither could be retired while its owner was still live.
        .flat_map(|id: GlobalId| snapshot.clients_with_id(id).map(|client| client.serial))
        .collect();
    out.sort_unstable();
    out.dedup();
    out
}