//! 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();
        // ⚠️ Tracked separately from `client_pids`, and that is the point: a
        // Client with no `sec_pid` still *claims* its id. Detecting duplicates
        // by looking in the pid map would let a pid-less first claimant leave
        // no trace, so the next Client claiming the same id would look unique
        // and its pid would be used — resolving an ambiguous id, which is the
        // one guess this guard exists to refuse. Pid-less Clients are ordinary
        // (the session manager's is one).
        let mut seen: BTreeSet<GlobalId> = BTreeSet::new();
        for client in snapshot.clients() {
            if !seen.insert(client.id) {
                // Two Clients claiming one id: drop it entirely rather than
                // pick. See the field docs.
                client_pids.remove(&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.
    ///
    /// ⚠️ **Known imprecision, deliberately not fixed here.** `ProcessId` now
    /// covers two sources — the node's `application.process.id` and its
    /// Client's `pipewire.sec.pid` (see [`keys_of`]) — so a bridge reported as
    /// `application.process.id` may in fact have resolved on the Client's
    /// protected pid. Pre-existing since R10-3 made the Client a fallback, and
    /// widened by the review's finding 1 making it a union. Splitting it would
    /// add a code to a set that is explicitly a stable contract for the audit
    /// output and the "why isn't this app being shared?" answer, so it wants
    /// its own decision rather than a drive-by.
    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 **and** from its Client (round 10, R10-3; made a
    // union rather than a fallback by the round-10 review, finding 1).
    //
    // ⚠️ **A union, not `node.or_else(client)`, and the difference is a leak.**
    // The node's `application.process.id` is client-controlled and optional;
    // the Client's `pipewire.sec.pid` is `pipewire.*`, protected, and the only
    // one that can carry a soundness argument (the same reason
    // `propagate_unresolved_owner` sweeps everything for an unbounded reader).
    // Letting the node's value *replace* the Client's meant one process using
    // two Clients could escape the bridge entirely: its tainted reader reports
    // a bogus node pid, its output leg omits the node pid and falls back to
    // the Client's real one, the two legs are bounded by different values, so
    // they neither bridge nor trip the unbounded sweep — and the output stays
    // eligible while re-emitting the call. Carrying both values costs nothing
    // and closes it: a leg that presents *either* value bridges.
    //
    // ⚠️ **Exception 1 applies to each value independently, and that is the
    // whole risk here.** Measured on this host: 15 unrelated Clients share
    // `sec_pid` 2528, which is pipewire-pulse's own — every Pulse-emulated app
    // has one. Suppressing it per value is what keeps the union from fusing
    // all fifteen into a single owner while still keeping each app's real
    // per-app pid. For the common Pulse shape (node pid = the app's, Client
    // `sec_pid` = the daemon's) the union therefore reduces to exactly the
    // node's pid, as before.
    //
    // 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.
    for pid in [node.props.process_id, ctx.client_pid(node)]
        .into_iter()
        .flatten()
    {
        if Some(pid) == ctx.pipewire_pulse_pid {
            continue;
        }
        let key = (OwnerKey::ProcessId, KeyValue::Num(u64::from(pid)));
        // The two agree far more often than not; a duplicate entry would be
        // harmless but would make the audit's key list read oddly.
        if !out.contains(&key) {
            out.push(key);
        }
    }
    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).
///
/// # 🔴 OPEN, phase-6 blocking — the key union can *reduce* taint here
///
/// **Round 11 review, finding 1. Verified correct; deliberately not fixed in
/// that round.** Round 10 made key 4 a union of the node's
/// `application.process.id` and its Client's `pipewire.sec.pid`, and the claim
/// that this was "strictly additive" was too strong: the same key list also
/// feeds *this* predicate, so adding a value can move a node from unbounded to
/// bounded, and `propagate_unresolved_owner`'s global sweep is triggered by an
/// **un**bounded tainted reader. Concretely:
///
/// 1. A tainted reader's node claims the pipewire-pulse PID while its Client
///    holds a real protected PID `A`. Under `or_else` the node's value won and
///    exception 1 suppressed it, leaving the reader unbounded; under the union
///    it is bounded by `A`.
/// 2. Its process's output leg uses a second Client whose id is **ambiguous**
///    (the observer missed a removal), so no protected PID is available — but
///    the leg claims a bogus `application.process.id` `B`, which bounds it.
/// 3. Neither the bridge nor the sweep fires, and the output stays eligible
///    while re-emitting the call.
///
/// It cannot leak today: `evaluate()` is reached only by the dry-run audit,
/// which creates no links. It becomes live when phase 6 consumes eligibility.
///
/// **Why it is not fixed yet.** The principled repair is provenance: a
/// self-claimed `application.process.id` is not a *sound* bound, only the
/// protected keys are. But applying that bluntly makes every Pulse-emulated
/// app unbounded — their Client's `sec_pid` is the daemon's and suppressed, so
/// the node's own claim is their only per-app identity — which re-triggers the
/// §6.1.1 mass over-exclusion the whole design is built to avoid, and would
/// make the eligible half of the §5.1 matrix empty.
///
/// The targeted rule that closes the path above without that cost: **a node
/// whose Client cannot be resolved at all must not be bounded by its own
/// self-claimed PID.** An ambiguous Client already means "we do not know who
/// owns this", and a self-claim must not paper over it; a Pulse app's Client
/// *is* resolved (to the daemon's PID, then suppressed), so it keeps its
/// bound. Implementing it needs `OwnerCtx` to distinguish "resolved" from
/// "absent", and `OwnerKeyIndex` to carry boundedness separately from the key
/// set, since bridging must keep using the full union.
///
/// ⚠️ Do this **with the §5.1 matrix data in hand**, not before: the whole
/// question is how much over-exclusion the rule actually causes on a real
/// graph, and that is measurable rather than arguable.
///
/// ## Round 12 — the deferral holds, and "resolved" has a trap in it
///
/// Codex re-examined this and agreed the deferral is defensible while
/// `evaluate()` is audit-only, and that the rule above closes the recorded path
/// without unbounding normal Pulse-emulated apps — **but only under one
/// reading of "resolves"**, and the wrong reading reintroduces the hole:
///
/// - ✅ "Resolved" must mean **an unambiguous Client that yields
///   `Some(pipewire.sec.pid)`**, taken *before* the pipewire-pulse suppression
///   step. A Pulse app then still has the daemon's protected PID as
///   provenance, even though that value is omitted from the bridge keys, so it
///   stays bounded and the eligible half survives.
/// - ❌ **Do not** implement it as "a unique Client object exists". A unique
///   Client with `sec_pid = None` would satisfy that test while providing no
///   protected identity at all, leaving exactly the self-claimed-PID hole this
///   rule is meant to close.
///
/// So the matrix needs five Client cases, not two: **absent**, **ambiguous**,
/// **unique but pid-less**, **resolved-native**, and
/// **resolved-to-pipewire-pulse**. The third is the one that distinguishes the
/// two readings, and it is the row a two-case matrix would silently skip.
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
}