"""Read-side projection (M6, plan §4.3): every payload passes through rep() HERE, at the API boundary, before it leaves the server. One module so no feature ever grows its own visibility logic — board, rally, comments, notifications, and SSE all call into the same PerspectiveLens. Conventions shared by every projector: - The lens is built ONCE per request (topology = 3 small queries) and memoizes rep(); bulk payloads (the board's hundreds of balls) project in O(rows) with no extra queries. - Projectors build PLAIN DICTS / new lists — they never mutate ORM rows. A projected value written onto a session-attached row would risk being flushed back to the DB as truth; history stays raw, projection stays read-time (§4.3). - assignment arrays (action_to / prev_action_to): each entry passes through rep(); results are de-duplicated preserving order. An entry whose rep bottoms out at a side label is DROPPED from the id list; when that leaves action_to empty while the raw list was not, the ball gets a holder_label (e.g. "Bowden Works") for the chip to render. A mixed list (some visible, some beyond the viewer's world) keeps the visible reps and silently drops the rest — the chip renders the players the viewer's game contains. """ from __future__ import annotations import uuid as uuidlib from sqlalchemy import select from sqlalchemy.orm import Session from .. import constants from ..models import Ball, Player from .perspective import MatchTopology, PerspectiveLens def build_lens(db: Session, match_id, viewer) -> PerspectiveLens: """Topology + lens for one request. `viewer` is a Player or None (the outside-observer default-deny path, see services.viewer).""" topology = MatchTopology(db, match_id) return PerspectiveLens(topology, viewer.id if viewer is not None else None) def project_assignment(lens: PerspectiveLens, player_ids) -> tuple[list, str | None]: """(projected id list, holder_label). See the module docstring for the drop/dedupe/label rules.""" projected: list = [] dropped_to_side = False for pid in player_ids or []: rep = lens.rep(pid) if rep.kind == "player": if rep.player_id not in projected: projected.append(rep.player_id) else: dropped_to_side = True holder_label = None if dropped_to_side and not projected: # Fully collapsed: the holder is beyond the viewer's deepest net. holder_label = lens.bottom_label return projected, holder_label def project_author(lens: PerspectiveLens, author_id): """author_id through rep(): the real id when visible, the boundary lead it collapses into otherwise, None past the viewer's world. Keeps every id in a payload resolvable against the viewer-scoped players list.""" if author_id is None: return None rep = lens.rep(author_id) return rep.player_id if rep.kind == "player" else None def can_spike(lens: PerspectiveLens, court_config_key: str, author_id) -> bool: """Plan §4.3b: the ONLY restricted automation is the cross-net spike. "Ready for Review" auto-spikes to the ball's AUTHOR; a viewer who cannot see the author must not be offered (or allowed — see services.board.patch_ball) that status. True everywhere else: courts without the spike automation, authorless balls (spike just clears hands), and visible authors.""" cfg = constants.court_config(court_config_key) or {} if not (cfg.get("automations") or {}).get("auto_spike_on_ready"): return True if author_id is None: return True return lens.player_visible(author_id) def project_ball_dict(lens: PerspectiveLens, ball: Ball, config_key: str) -> dict: """One ball for the wire (BallOut shape), assignment arrays and author rep()'d, can_spike computed for the viewer. A plain dict — never a mutated ORM row (module docstring).""" action_to, holder_label = project_assignment(lens, ball.action_to) prev_action_to, _ = project_assignment(lens, ball.prev_action_to) return { "id": ball.id, "court_id": ball.court_id, "title": ball.title, "body": ball.body, "status": ball.status, "review_state": ball.review_state, "category": ball.category, "tags": ball.tags or [], "url_staging": ball.url_staging, "url_live": ball.url_live, "action_to": action_to, "prev_action_to": prev_action_to, "holder_label": holder_label, "author_id": project_author(lens, ball.author_id), "resolved_at": ball.resolved_at, "position": ball.position, "created_at": ball.created_at, "last_touch_at": getattr(ball, "last_touch_at", None), "can_spike": can_spike(lens, config_key, ball.author_id), } def strip_bands(lens: PerspectiveLens) -> list[dict]: """The viewer's court strip (plan §4.5), band per court from courts_of() (observer court for outsiders): near ids left of the net, far ids (or just the label) across it. Band order is courts_of()'s order — member court first, then boundary courts (Adi: big court on top, Tingang beneath).""" bands = [] for court in lens.courts: bands.append( { "node_id": court.node_id, "name": court.name, "is_root": court.is_root, "near": sorted(court.near, key=str), "far": sorted(court.far, key=str), "far_label": court.far_label, } ) return bands def visible_sides(lens: PerspectiveLens) -> set: """Which side ROWS the viewer's payload may name. A viewer on a root court knows both sides; a sub-team-only viewer's world is their own root side ONLY (§4.2: naming the opposing side to Zeina would leak structure her game doesn't contain).""" topo = lens.topology side_ids: set = set() for court in lens.courts: own_side = topo.side_of(court.node_id) side_ids.add(own_side) if court.is_root: opp = topo.opposing_side(own_side) if opp is not None: side_ids.add(opp) return side_ids def visible_players_payload(db: Session, lens: PerspectiveLens) -> list[Player]: """Player rows for exactly the viewer's visible set — the board roster, the mention autocomplete, and the chip renderer all draw from this (§4.3: Erin cannot mention Zeina; she doesn't know Zeina exists). Every id any projected payload references is in this list by construction (rep() only ever returns visible players).""" if not lens.visible: return [] return ( db.execute( select(Player).where(Player.id.in_(lens.visible)).order_by(Player.username) ) .scalars() .all() ) # ------------------------------------------------------------------ rally def project_rally_entries(lens: PerspectiveLens, touches) -> list[dict]: """§4.3: touches whose ACTOR is invisible to the viewer collapse; consecutive invisible touches become ONE {kind:"collapsed", count, label} entry (label from the stretch's first acting_node, viewer- aware). Visible touches keep their real actor; to_players entries are rep()'d (deduped, side bottom-outs dropped). The auto flag survives projection — honesty about automation does not stop at the net.""" entries: list[dict] = [] collapsed_count = 0 collapsed_node = None for t in touches: if lens.touch_visible(t.from_player): if collapsed_count: entries.append( { "kind": "collapsed", "count": collapsed_count, "label": lens.collapsed_label(collapsed_node), } ) collapsed_count, collapsed_node = 0, None to_players, _ = project_assignment(lens, t.to_players) entries.append( { "kind": t.kind, "seq": t.seq, "from_player": t.from_player, "to_players": to_players, "auto": t.via_rule_id is not None, "created_at": t.created_at.isoformat(), } ) else: if collapsed_count == 0: collapsed_node = t.acting_node_id collapsed_count += 1 if collapsed_count: entries.append( { "kind": "collapsed", "count": collapsed_count, "label": lens.collapsed_label(collapsed_node), } ) return entries # --------------------------------------------------------------- comments def project_comments(lens: PerspectiveLens, comments: list[dict]) -> list[dict]: """§4.3 comment scoping (rule documented in services.perspective): a comment renders iff its acting_node_id is visible to the viewer (or NULL). Invisible stretches become {kind:"collapsed_comments", count} entries — the timeline admits the conversation exists, shows no bodies, no authors. Visible comments additionally get their author rep()'d (author_label when the author is beyond the viewer's world — the Tracy-to-Rob case in perspective.py's docstring).""" entries: list[dict] = [] collapsed_count = 0 for c in comments: if lens.node_visible(c.get("acting_node_id")): if collapsed_count: entries.append({"kind": "collapsed_comments", "count": collapsed_count}) collapsed_count = 0 projected = dict(c) author_id = c.get("author_id") if author_id is not None: rep = lens.rep(author_id) if rep.kind == "player": projected["author_id"] = rep.player_id else: projected["author_id"] = None projected["author_label"] = rep.label entries.append(projected) else: collapsed_count += 1 if collapsed_count: entries.append({"kind": "collapsed_comments", "count": collapsed_count}) return entries # ---------------------------------------------------------- notifications def project_notification_actor(lens: PerspectiveLens, item: dict) -> dict: """§4.3: notification actors render through rep() for the RECIPIENT. Visible actor: unchanged. Collapses to a boundary lead: actor_id becomes the lead. Beyond the recipient's world: actor_id None + actor_label side-label.""" actor_id = item.get("actor_id") if actor_id is None: return item rep = lens.rep(actor_id) out = dict(item) if rep.kind == "player": out["actor_id"] = rep.player_id else: out["actor_id"] = None out["actor_label"] = rep.label return out # ------------------------------------------------------------------- SSE def project_event(evt: dict, topology: MatchTopology | None, viewer_player_id) -> dict | None: """One SSE event, projected for one subscriber — a pure function over (event, topology, viewer). Returns the event to deliver, or None to drop it entirely. - notification.created: RECIPIENT-ONLY (this is the fix for the M5 broadcast leak — everyone's queue used to receive everyone's notifications). The embedded summary's actor also passes through rep() so the bell never learns an invisible actor's id. - ball.created / ball.updated: action_to is rep()'d (+ holder_label); a touch by an invisible actor keeps its KIND (the board may show a status flip) but sheds its actor (from_player None + from_label). A PURE MOVEMENT event (touch kind pass/auto) by an invisible actor whose projected movement stays inside one collapsed rep — rep(from) == every rep(to) — is DROPPED: the viewer's board shows the same holder before and after, so there is no visible change to announce (a Tingang inner pass never even reaches Rian's stream, §6.3). - comment.* events pass through untouched: they carry only ids, and the collapsed timeline deliberately RENDERS invisible stretches as live counts, so the count updating in realtime leaks nothing the projection doesn't already show. - Events without match context (hello/ping/unknown) pass through. """ etype = evt.get("type", "") if etype == "notification.created": if viewer_player_id is None or str(viewer_player_id) != evt.get("recipient_id"): return None if topology is None: return evt lens = PerspectiveLens(topology, viewer_player_id) out = dict(evt) # The SSE summary carries STRING ids (json-ready); rep() works on # UUIDs — convert in, project, stringify back out. summary = dict(evt.get("notification") or {}) raw_actor = summary.get("actor_id") if raw_actor is not None: summary["actor_id"] = uuidlib.UUID(str(raw_actor)) summary = project_notification_actor(lens, summary) if summary.get("actor_id") is not None: summary["actor_id"] = str(summary["actor_id"]) out["notification"] = summary return out if etype not in ("ball.created", "ball.updated") or topology is None: return evt lens = PerspectiveLens(topology, viewer_player_id) out = dict(evt) raw_action_to = evt.get("action_to") or [] action_ids = [uuidlib.UUID(p) for p in raw_action_to] projected, holder_label = project_assignment(lens, action_ids) out["action_to"] = [str(p) for p in projected] if holder_label: out["holder_label"] = holder_label touch = evt.get("touch") if touch: from_raw = touch.get("from_player") from_id = uuidlib.UUID(from_raw) if from_raw else None to_ids = [uuidlib.UUID(p) for p in (touch.get("to_players") or [])] if lens.touch_visible(from_id): to_projected, _ = project_assignment(lens, to_ids) out["touch"] = dict(touch, to_players=[str(p) for p in to_projected]) else: from_rep = lens.rep(from_id) to_projected, _ = project_assignment(lens, to_ids) if ( etype == "ball.updated" and touch.get("kind") in ("pass", "auto") and from_rep.kind == "player" and to_projected and all(p == from_rep.player_id for p in to_projected) ): # No visible change: the ball moved entirely inside one # collapsed blob (or over an inner net into its own lead). return None out["touch"] = dict( touch, from_player=None, from_label=lens.collapsed_label(None), to_players=[str(p) for p in to_projected], ) return out def topology_for_event(db: Session, evt: dict) -> MatchTopology | None: """Resolve the event's match topology for per-subscriber projection. Events stamped with match_id (all ball.* and notification.created envelopes) load directly; anything else projects topology-free.""" match_id = evt.get("match_id") if not match_id: return None return MatchTopology(db, uuidlib.UUID(match_id)) # ----------------------------------------------------------- write guards def ensure_targets_visible(lens: PerspectiveLens, to_players, error_cls) -> None: """§4.3b backstop: every pass/serve target must be VISIBLE to the acting player (the UI only offers visible ones; this stops a crafted request). `lens` is the ACTOR's lens — callers already hold one for the same request (never build a second topology just to validate). Raises error_cls(422, TARGET_NOT_VISIBLE) — error_cls is the calling service's structured error type. Passing to a boundary player (a rep target) is just a normal pass to that player, so it sails through here by construction.""" if not to_players: return invisible = [str(p) for p in to_players if not lens.player_visible(p)] if invisible: raise error_cls( 422, "TARGET_NOT_VISIBLE", "One or more recipients are not visible from your position.", "You can only pass or serve to players on your courts. " f"Not visible: {', '.join(invisible)}.", )