Ben Stull
949 lines
34 KiB
Python
949 lines
34 KiB
Python
"""§15: the notifications fan-out and the SSE broadcaster.
|
||
|
||
This module is the single chokepoint for the producer side of §15.
|
||
Every write that names an `rfc_slug` flows through `fan_out_from_action`
|
||
(called from `bot._log`) or `fan_out_chat_message` (called from the chat
|
||
write paths in `api_branches` / `api_prs`, since chat messages don't go
|
||
through the bot wrapper). The chokepoint:
|
||
|
||
- upserts a `watches` row for the actor per §15.6's auto-watch rule
|
||
(a substantive gesture sets `watching`, never downgrades, and bumps
|
||
`last_participation_at` for the 90-day decay);
|
||
- applies the §15.1 routing rules to determine which other users
|
||
receive a notification for the event;
|
||
- filters out muted recipients per §15.8 (per-user mute list);
|
||
- filters out `watches.state='muted'` per §15.6 (per-RFC mute);
|
||
- inserts one `notifications` row per recipient with the underlying
|
||
actor's user id per §15.9 (the bot never appears as actor);
|
||
- dispatches the row to the §15.4 email path if the recipient's
|
||
category toggle is on (held during quiet hours per §15.8, deferred
|
||
to the digest otherwise);
|
||
- publishes each new row to the per-user SSE subscriber queue so the
|
||
inbox refresh and header-badge count back the same stream per §15.3.
|
||
|
||
If you find yourself wanting to insert a row into `notifications`
|
||
directly from an endpoint, the spec is right and you are wrong: the
|
||
chokepoint is the invariant. Read paths can query the table from
|
||
anywhere; the producer side flows through here.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Iterable
|
||
|
||
from . import db
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# §15.1 routing — action_kind → (event_kind, category, recipient_set)
|
||
# ---------------------------------------------------------------------------
|
||
#
|
||
# Three recipient sets are used by the routing rules below:
|
||
#
|
||
# "watchers" — every user with a watches row on this slug whose state
|
||
# is 'watching' (full stream) or 'following' (structural
|
||
# only), minus the actor and minus 'muted' rows. The
|
||
# structural/full split is applied per-event below.
|
||
# "owners" — every user named in the entry's owners_json frontmatter
|
||
# on cached_rfcs (resolved to user_ids via gitea_login).
|
||
# "proposer" — the user whose action_kind=propose_rfc opened the slug,
|
||
# resolved via the most recent matching audit row.
|
||
#
|
||
# The category of each event maps to one of the §15.4 buckets:
|
||
#
|
||
# "personal-direct" — recipient is named subject of the action.
|
||
# "structural" — structural beats on watched RFCs.
|
||
# "churn" — per-commit / per-message activity. Never emails
|
||
# per §15.4; aggregated by the §15.5 digest.
|
||
|
||
CATEGORY_PERSONAL = "personal-direct"
|
||
CATEGORY_STRUCTURAL = "structural"
|
||
CATEGORY_CHURN = "churn"
|
||
|
||
# Action kinds whose actor's first interaction with a slug triggers
|
||
# auto-watch per §15.6. The substantive-gesture list in the spec is
|
||
# the source; we map onto our action_kind taxonomy. Reads, view advances,
|
||
# and own-action observations are intentionally excluded.
|
||
_AUTO_WATCH_ACTIONS = {
|
||
"propose_rfc",
|
||
"merge_proposal",
|
||
"decline_proposal",
|
||
"create_branch",
|
||
"accept_change",
|
||
"decline_change",
|
||
"manual_flush",
|
||
"open_branch_pr",
|
||
"merge_branch_pr",
|
||
"withdraw_branch_pr",
|
||
"supersede_branch_pr",
|
||
"open_metadata_pr",
|
||
"open_claim_pr",
|
||
"merge_claim_pr",
|
||
"create_resolution_branch",
|
||
"replay_change",
|
||
"graduate_start",
|
||
"graduate_complete",
|
||
"open_review_thread",
|
||
"resolve_thread",
|
||
"drop_flag",
|
||
"resolve_flag",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SSE broadcaster — per-user subscriber registry
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class _Subscriber:
|
||
user_id: int
|
||
queue: asyncio.Queue = field(default_factory=asyncio.Queue)
|
||
|
||
|
||
_subscribers: dict[int, list[_Subscriber]] = {}
|
||
_subscribers_lock: asyncio.Lock | None = None
|
||
|
||
|
||
def _get_lock() -> asyncio.Lock:
|
||
global _subscribers_lock
|
||
if _subscribers_lock is None:
|
||
_subscribers_lock = asyncio.Lock()
|
||
return _subscribers_lock
|
||
|
||
|
||
async def subscribe(user_id: int) -> _Subscriber:
|
||
"""Register a new SSE subscriber for a user. Each browser tab gets its
|
||
own subscriber; the per-user fan-out pushes the same event to every
|
||
subscriber for that user per §15 (multiple tabs see the same updates)."""
|
||
sub = _Subscriber(user_id=user_id)
|
||
async with _get_lock():
|
||
_subscribers.setdefault(user_id, []).append(sub)
|
||
return sub
|
||
|
||
|
||
async def unsubscribe(sub: _Subscriber) -> None:
|
||
async with _get_lock():
|
||
bucket = _subscribers.get(sub.user_id, [])
|
||
if sub in bucket:
|
||
bucket.remove(sub)
|
||
if not bucket:
|
||
_subscribers.pop(sub.user_id, None)
|
||
|
||
|
||
async def _broadcast(user_id: int, event: str, payload: Any) -> None:
|
||
"""Push an event to every subscriber for one user.
|
||
|
||
Best-effort; if the queue is unbounded a slow consumer just sees the
|
||
backlog when it next drains. We deliberately do not block writers on
|
||
consumer presence — the durable inbox row is the source of truth."""
|
||
async with _get_lock():
|
||
subs = list(_subscribers.get(user_id, []))
|
||
for sub in subs:
|
||
try:
|
||
sub.queue.put_nowait({"event": event, "payload": payload})
|
||
except Exception:
|
||
log.exception("notify broadcast: drop event for user %s", user_id)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Producer-side entry points
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def fan_out_from_action(
|
||
*,
|
||
actor_user_id: int | None,
|
||
action_kind: str,
|
||
rfc_slug: str | None,
|
||
branch_name: str | None,
|
||
pr_number: int | None,
|
||
details: dict | None,
|
||
) -> None:
|
||
"""Called from `bot._log` after every successful Git write.
|
||
|
||
Synchronous on purpose — runs inside the same request as the action,
|
||
so a failure here surfaces with the action rather than as a silent
|
||
drop. The SSE push is asyncio-scheduled if a loop is running; if
|
||
no loop is available (test harness, sync test client) the dispatch
|
||
falls back to writing the row and relying on the next subscriber
|
||
poll to surface it.
|
||
"""
|
||
if rfc_slug is None or action_kind not in _AUTO_WATCH_ACTIONS:
|
||
# Actions without an rfc_slug (e.g. permission events) or that
|
||
# aren't a §15.6 substantive gesture don't drive the chokepoint.
|
||
return
|
||
|
||
if actor_user_id is not None:
|
||
_bump_auto_watch(actor_user_id, rfc_slug)
|
||
|
||
rules = _ROUTING.get(action_kind)
|
||
if rules is None:
|
||
return
|
||
|
||
for rule in rules:
|
||
recipients = _resolve_recipients(
|
||
rule=rule,
|
||
actor_user_id=actor_user_id,
|
||
rfc_slug=rfc_slug,
|
||
details=details,
|
||
)
|
||
if not recipients:
|
||
continue
|
||
for recipient_id in recipients:
|
||
_emit_one(
|
||
recipient_user_id=recipient_id,
|
||
event_kind=rule["event_kind"],
|
||
category=rule["category"],
|
||
actor_user_id=actor_user_id,
|
||
rfc_slug=rfc_slug,
|
||
branch_name=branch_name,
|
||
pr_number=pr_number,
|
||
details=details or {},
|
||
)
|
||
|
||
|
||
def fan_out_chat_message(
|
||
*,
|
||
actor_user_id: int,
|
||
rfc_slug: str,
|
||
branch_name: str,
|
||
thread_id: int,
|
||
message_id: int,
|
||
is_review_thread: bool = False,
|
||
pr_number: int | None = None,
|
||
) -> None:
|
||
"""Called from chat write paths after a user message is persisted.
|
||
|
||
Chat messages don't flow through bot.py (no Git write); the chokepoint
|
||
here is the equivalent for §15's chat-message events. The routing
|
||
rule is: prior authors in the thread (besides the message author) get
|
||
a personal-direct `chat_reply_to_my_message`; users watching the RFC
|
||
(state='watching', i.e. full stream) get a churn-class
|
||
`chat_message_in_participated_thread`. The two are union'd so a user
|
||
who is both gets only the personal-direct row.
|
||
"""
|
||
_bump_auto_watch(actor_user_id, rfc_slug)
|
||
|
||
prior_authors = {
|
||
row["author_user_id"]
|
||
for row in db.conn().execute(
|
||
"""
|
||
SELECT DISTINCT author_user_id
|
||
FROM thread_messages
|
||
WHERE thread_id = ?
|
||
AND author_user_id IS NOT NULL
|
||
AND author_user_id != ?
|
||
AND id < ?
|
||
""",
|
||
(thread_id, actor_user_id, message_id),
|
||
)
|
||
}
|
||
|
||
# §15.6 / §15.8 filters apply here too — a muted recipient never sees
|
||
# a row regardless of which producer path generated it.
|
||
muted_rfc = _muted_user_ids_for_rfc(rfc_slug)
|
||
muters = _muters_of(actor_user_id)
|
||
prior_authors = (prior_authors - muted_rfc) - muters
|
||
|
||
# Personal-direct: prior thread authors.
|
||
for recipient_id in prior_authors:
|
||
event_kind = (
|
||
"pr_review_thread_reply" if is_review_thread else "chat_reply_to_my_message"
|
||
)
|
||
_emit_one(
|
||
recipient_user_id=recipient_id,
|
||
event_kind=event_kind,
|
||
category=CATEGORY_PERSONAL,
|
||
actor_user_id=actor_user_id,
|
||
rfc_slug=rfc_slug,
|
||
branch_name=branch_name,
|
||
pr_number=pr_number,
|
||
thread_id=thread_id,
|
||
details={"message_id": message_id},
|
||
)
|
||
|
||
# Churn: watchers of the RFC who weren't already covered.
|
||
watcher_ids = _watcher_user_ids(rfc_slug, scope="watching") - prior_authors
|
||
watcher_ids.discard(actor_user_id)
|
||
watcher_ids = (watcher_ids - muted_rfc) - muters
|
||
for recipient_id in watcher_ids:
|
||
event_kind = (
|
||
"pr_review_thread_new" if is_review_thread
|
||
else "chat_message_in_participated_thread"
|
||
)
|
||
_emit_one(
|
||
recipient_user_id=recipient_id,
|
||
event_kind=event_kind,
|
||
category=CATEGORY_CHURN,
|
||
actor_user_id=actor_user_id,
|
||
rfc_slug=rfc_slug,
|
||
branch_name=branch_name,
|
||
pr_number=pr_number,
|
||
thread_id=thread_id,
|
||
details={"message_id": message_id},
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Routing rules — per action_kind, a list of recipient×event×category
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
_ROUTING: dict[str, list[dict]] = {
|
||
# §9.1: propose surfaces the new slug. The proposer is the actor;
|
||
# the personal-direct beats fire on merge/decline, not on propose
|
||
# itself. Admins and owners get a structural ping so the
|
||
# pending-ideas queue isn't invisible.
|
||
"propose_rfc": [
|
||
{"event_kind": "proposal_opened_on_watched_topic",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "admins"},
|
||
],
|
||
# §9.3: merge fires personal-direct to the proposer per §15.4's
|
||
# named-subject rule. Watchers also see the structural beat.
|
||
"merge_proposal": [
|
||
{"event_kind": "proposal_merged",
|
||
"category": CATEGORY_PERSONAL, "recipients": "proposer"},
|
||
{"event_kind": "proposal_merged",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"decline_proposal": [
|
||
{"event_kind": "proposal_declined",
|
||
"category": CATEGORY_PERSONAL, "recipients": "proposer"},
|
||
],
|
||
# §10.1: PR opened. Watchers see it; the opener auto-watches but
|
||
# doesn't get an inbox row for their own gesture.
|
||
"open_branch_pr": [
|
||
{"event_kind": "pr_opened",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"open_metadata_pr": [
|
||
{"event_kind": "pr_opened",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"open_claim_pr": [
|
||
{"event_kind": "claim_opened",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
# §10.5: merge fires personal-direct to the PR opener if someone
|
||
# else merged; watchers see the structural beat regardless.
|
||
"merge_branch_pr": [
|
||
{"event_kind": "pr_merged",
|
||
"category": CATEGORY_PERSONAL, "recipients": "pr_opener_if_other"},
|
||
{"event_kind": "pr_merged",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"withdraw_branch_pr": [
|
||
{"event_kind": "pr_withdrawn",
|
||
"category": CATEGORY_PERSONAL, "recipients": "pr_opener_if_other"},
|
||
{"event_kind": "pr_withdrawn",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"supersede_branch_pr": [
|
||
{"event_kind": "pr_withdrawn",
|
||
"category": CATEGORY_PERSONAL, "recipients": "pr_opener_if_other"},
|
||
],
|
||
# §8.6: accept_change is per-commit churn. Watchers (full stream)
|
||
# get the row; following-only watchers don't.
|
||
"accept_change": [
|
||
{"event_kind": "pr_commit_added",
|
||
"category": CATEGORY_CHURN, "recipients": "watchers_full"},
|
||
],
|
||
"manual_flush": [
|
||
{"event_kind": "pr_commit_added",
|
||
"category": CATEGORY_CHURN, "recipients": "watchers_full"},
|
||
],
|
||
"replay_change": [
|
||
{"event_kind": "pr_commit_added",
|
||
"category": CATEGORY_CHURN, "recipients": "watchers_full"},
|
||
],
|
||
# §13.3: graduation_complete is personal-direct to owners (the
|
||
# super-draft's owners list) and structural to watchers.
|
||
"graduate_complete": [
|
||
{"event_kind": "graduation_complete",
|
||
"category": CATEGORY_PERSONAL, "recipients": "owners"},
|
||
{"event_kind": "graduation_complete",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"graduate_start": [
|
||
{"event_kind": "super_draft_graduation_ready",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"create_resolution_branch": [
|
||
{"event_kind": "pr_conflict_with_main",
|
||
"category": CATEGORY_STRUCTURAL, "recipients": "watchers_structural"},
|
||
],
|
||
"create_branch": [
|
||
# No notification — branch creation is a private gesture until
|
||
# work lands on it. Auto-watch is the visible effect.
|
||
],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Recipient resolvers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _resolve_recipients(
|
||
*,
|
||
rule: dict,
|
||
actor_user_id: int | None,
|
||
rfc_slug: str,
|
||
details: dict | None,
|
||
) -> set[int]:
|
||
kind = rule["recipients"]
|
||
if kind == "watchers_structural":
|
||
# following (structural only) + watching (full stream) — both
|
||
# see structural events.
|
||
ids = _watcher_user_ids(rfc_slug, scope="any")
|
||
elif kind == "watchers_full":
|
||
ids = _watcher_user_ids(rfc_slug, scope="watching")
|
||
elif kind == "owners":
|
||
ids = _entry_owner_user_ids(rfc_slug)
|
||
elif kind == "admins":
|
||
ids = _admin_user_ids()
|
||
elif kind == "proposer":
|
||
ids = _proposer_user_id(rfc_slug)
|
||
elif kind == "pr_opener_if_other":
|
||
opener = _pr_opener_user_id(rfc_slug, details)
|
||
ids = {opener} if opener is not None and opener != actor_user_id else set()
|
||
else:
|
||
log.warning("notify: unknown recipient kind %s", kind)
|
||
return set()
|
||
|
||
if actor_user_id is not None:
|
||
ids.discard(actor_user_id)
|
||
|
||
# §15.6: per-RFC mute suppresses every signal for the (user, slug).
|
||
muted = _muted_user_ids_for_rfc(rfc_slug)
|
||
ids -= muted
|
||
|
||
# §15.8: per-user mute suppresses notifications produced by the actor
|
||
# for each muter. (Inbox-volume only; content visibility unchanged.)
|
||
if actor_user_id is not None:
|
||
muters = _muters_of(actor_user_id)
|
||
ids -= muters
|
||
|
||
return ids
|
||
|
||
|
||
def _watcher_user_ids(rfc_slug: str, *, scope: str) -> set[int]:
|
||
"""scope: 'watching' = full stream only; 'any' = watching+following."""
|
||
if scope == "watching":
|
||
rows = db.conn().execute(
|
||
"SELECT user_id FROM watches WHERE rfc_slug = ? AND state = 'watching'",
|
||
(rfc_slug,),
|
||
)
|
||
else:
|
||
rows = db.conn().execute(
|
||
"SELECT user_id FROM watches WHERE rfc_slug = ? AND state IN ('watching', 'following')",
|
||
(rfc_slug,),
|
||
)
|
||
return {r["user_id"] for r in rows}
|
||
|
||
|
||
def _entry_owner_user_ids(rfc_slug: str) -> set[int]:
|
||
row = db.conn().execute(
|
||
"SELECT owners_json FROM cached_rfcs WHERE slug = ?", (rfc_slug,),
|
||
).fetchone()
|
||
if row is None:
|
||
return set()
|
||
try:
|
||
owners = json.loads(row["owners_json"] or "[]")
|
||
except json.JSONDecodeError:
|
||
return set()
|
||
if not owners:
|
||
return set()
|
||
placeholders = ",".join("?" * len(owners))
|
||
user_rows = db.conn().execute(
|
||
f"SELECT id FROM users WHERE gitea_login IN ({placeholders})",
|
||
tuple(owners),
|
||
)
|
||
return {r["id"] for r in user_rows}
|
||
|
||
|
||
def _admin_user_ids() -> set[int]:
|
||
return {
|
||
r["id"]
|
||
for r in db.conn().execute("SELECT id FROM users WHERE role IN ('owner', 'admin')")
|
||
}
|
||
|
||
|
||
def _proposer_user_id(rfc_slug: str) -> set[int]:
|
||
row = db.conn().execute(
|
||
"""
|
||
SELECT actor_user_id FROM actions
|
||
WHERE action_kind = 'propose_rfc' AND rfc_slug = ?
|
||
ORDER BY id LIMIT 1
|
||
""",
|
||
(rfc_slug,),
|
||
).fetchone()
|
||
if row is None or row["actor_user_id"] is None:
|
||
return set()
|
||
return {row["actor_user_id"]}
|
||
|
||
|
||
def _pr_opener_user_id(rfc_slug: str, details: dict | None) -> int | None:
|
||
"""For pr_opener_if_other recipients, find the user who opened the PR.
|
||
|
||
Uses cached_prs.opened_by (gitea_login) → users.id, falling back to
|
||
the audit log if the cache row isn't there yet.
|
||
"""
|
||
pr_number = (details or {}).get("pr_number")
|
||
# When called from merge/withdraw routing, details is the original
|
||
# action's details — pr_number is on the audit row, not here. Pull
|
||
# it from the most-recent relevant action.
|
||
row = db.conn().execute(
|
||
"""
|
||
SELECT actor_user_id FROM actions
|
||
WHERE action_kind = 'open_branch_pr' AND rfc_slug = ?
|
||
ORDER BY id DESC LIMIT 1
|
||
""",
|
||
(rfc_slug,),
|
||
).fetchone()
|
||
if row and row["actor_user_id"] is not None:
|
||
return row["actor_user_id"]
|
||
return pr_number # last-ditch: keep typing consistent (rarely matters)
|
||
|
||
|
||
def _muted_user_ids_for_rfc(rfc_slug: str) -> set[int]:
|
||
return {
|
||
r["user_id"]
|
||
for r in db.conn().execute(
|
||
"SELECT user_id FROM watches WHERE rfc_slug = ? AND state = 'muted'",
|
||
(rfc_slug,),
|
||
)
|
||
}
|
||
|
||
|
||
def _muters_of(actor_user_id: int) -> set[int]:
|
||
return {
|
||
r["muter_user_id"]
|
||
for r in db.conn().execute(
|
||
"SELECT muter_user_id FROM notification_user_mutes WHERE muted_user_id = ?",
|
||
(actor_user_id,),
|
||
)
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Auto-watch upsert
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _bump_auto_watch(user_id: int, rfc_slug: str) -> None:
|
||
"""Per §15.6 substantive-gesture rule: upsert a `watching` row for
|
||
the actor, never downgrade. Bumps `last_participation_at` for the
|
||
90-day decay timer regardless of current state.
|
||
|
||
The role-implicit watching for owners/arbiters is computed at fan-out
|
||
time (recipient resolution), not stored on the row, so this upsert
|
||
doesn't need to special-case them.
|
||
"""
|
||
existing = db.conn().execute(
|
||
"SELECT state, set_by FROM watches WHERE user_id = ? AND rfc_slug = ?",
|
||
(user_id, rfc_slug),
|
||
).fetchone()
|
||
now = "datetime('now')"
|
||
if existing is None:
|
||
db.conn().execute(
|
||
f"""
|
||
INSERT INTO watches
|
||
(user_id, rfc_slug, state, set_by, set_at, last_participation_at)
|
||
VALUES (?, ?, 'watching', 'auto', {now}, {now})
|
||
""",
|
||
(user_id, rfc_slug),
|
||
)
|
||
return
|
||
# Don't downgrade: a 'muted' explicit setting stays muted; a 'watching'
|
||
# row stays watching. The only auto-upgrade is following → watching.
|
||
if existing["state"] == "following" and existing["set_by"] != "explicit":
|
||
db.conn().execute(
|
||
f"UPDATE watches SET state = 'watching', last_participation_at = {now} WHERE user_id = ? AND rfc_slug = ?",
|
||
(user_id, rfc_slug),
|
||
)
|
||
else:
|
||
db.conn().execute(
|
||
f"UPDATE watches SET last_participation_at = {now} WHERE user_id = ? AND rfc_slug = ?",
|
||
(user_id, rfc_slug),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Inserting a notification row + dispatch
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _emit_one(
|
||
*,
|
||
recipient_user_id: int,
|
||
event_kind: str,
|
||
category: str,
|
||
actor_user_id: int | None,
|
||
rfc_slug: str | None,
|
||
branch_name: str | None,
|
||
pr_number: int | None,
|
||
details: dict,
|
||
thread_id: int | None = None,
|
||
) -> int:
|
||
payload = {"category": category, **details}
|
||
cur = db.conn().execute(
|
||
"""
|
||
INSERT INTO notifications
|
||
(recipient_user_id, event_kind, rfc_slug, branch_name, pr_number,
|
||
thread_id, actor_user_id, payload)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
recipient_user_id,
|
||
event_kind,
|
||
rfc_slug,
|
||
branch_name,
|
||
pr_number,
|
||
thread_id,
|
||
actor_user_id,
|
||
json.dumps(payload),
|
||
),
|
||
)
|
||
notif_id = cur.lastrowid
|
||
|
||
row_payload = _row_payload(notif_id)
|
||
_schedule_broadcast(recipient_user_id, "notification", row_payload)
|
||
|
||
# Email dispatch: §15.4. Held during quiet hours per §15.8; held
|
||
# during a global opt-out (bounce) per §15.4. Defers to the digest
|
||
# otherwise per §15.5's exclusion rules.
|
||
_schedule_email(notif_id, recipient_user_id, event_kind, category, row_payload)
|
||
|
||
return notif_id
|
||
|
||
|
||
def _row_payload(notif_id: int) -> dict:
|
||
row = db.conn().execute(
|
||
"""
|
||
SELECT n.id, n.event_kind, n.rfc_slug, n.branch_name, n.pr_number,
|
||
n.thread_id, n.actor_user_id, n.payload, n.created_at, n.read_at,
|
||
u.display_name AS actor_display, u.gitea_login AS actor_login,
|
||
r.title AS rfc_title
|
||
FROM notifications n
|
||
LEFT JOIN users u ON u.id = n.actor_user_id
|
||
LEFT JOIN cached_rfcs r ON r.slug = n.rfc_slug
|
||
WHERE n.id = ?
|
||
""",
|
||
(notif_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
return {}
|
||
try:
|
||
extras = json.loads(row["payload"] or "{}")
|
||
except json.JSONDecodeError:
|
||
extras = {}
|
||
return {
|
||
"id": row["id"],
|
||
"event_kind": row["event_kind"],
|
||
"rfc_slug": row["rfc_slug"],
|
||
"rfc_title": row["rfc_title"],
|
||
"branch_name": row["branch_name"],
|
||
"pr_number": row["pr_number"],
|
||
"thread_id": row["thread_id"],
|
||
"actor_user_id": row["actor_user_id"],
|
||
"actor_login": row["actor_login"],
|
||
"actor_display": row["actor_display"],
|
||
"created_at": row["created_at"],
|
||
"read_at": row["read_at"],
|
||
"category": extras.get("category"),
|
||
"summary": render_summary(row["event_kind"], row["actor_display"], row["rfc_title"], extras),
|
||
"extras": extras,
|
||
}
|
||
|
||
|
||
def render_summary(event_kind: str, actor_display: str | None, rfc_title: str | None, extras: dict) -> str:
|
||
"""One short sentence shared by the inbox row and the email body
|
||
per §15.4. Actor is the underlying user per §15.9; null actor renders
|
||
as "the app" so a system-generated event still reads as a sentence."""
|
||
actor = actor_display or "the app"
|
||
title = rfc_title or extras.get("slug") or "this RFC"
|
||
if event_kind == "proposal_merged":
|
||
return f"{actor} merged your proposal — {title} is now a super-draft."
|
||
if event_kind == "proposal_declined":
|
||
return f"{actor} declined your proposal for {title}."
|
||
if event_kind == "proposal_opened_on_watched_topic":
|
||
return f"{actor} proposed a new RFC: {title}."
|
||
if event_kind == "pr_opened":
|
||
return f"{actor} opened a PR on {title}."
|
||
if event_kind == "pr_merged":
|
||
return f"{actor} merged a PR on {title}."
|
||
if event_kind == "pr_withdrawn":
|
||
return f"{actor} withdrew a PR on {title}."
|
||
if event_kind == "pr_commit_added":
|
||
return f"{actor} added a commit on {title}."
|
||
if event_kind == "pr_review_thread_new":
|
||
return f"{actor} opened a review thread on {title}."
|
||
if event_kind == "pr_review_thread_reply":
|
||
return f"{actor} replied to your review thread on {title}."
|
||
if event_kind == "chat_message_in_participated_thread":
|
||
return f"{actor} posted a chat message on {title}."
|
||
if event_kind == "chat_reply_to_my_message":
|
||
return f"{actor} replied to your message on {title}."
|
||
if event_kind == "claim_opened":
|
||
return f"{actor} opened a claim PR on {title}."
|
||
if event_kind == "graduation_complete":
|
||
return f"{title} graduated to an active RFC."
|
||
if event_kind == "super_draft_graduation_ready":
|
||
return f"{actor} began graduating {title}."
|
||
if event_kind == "pr_conflict_with_main":
|
||
return f"{actor} started a resolution branch on {title}."
|
||
return f"{event_kind} on {title}"
|
||
|
||
|
||
def _schedule_broadcast(user_id: int, event: str, payload: Any) -> None:
|
||
"""Best-effort SSE push; survives the absence of a running loop so
|
||
sync test clients still write the row."""
|
||
try:
|
||
loop = asyncio.get_running_loop()
|
||
except RuntimeError:
|
||
return
|
||
loop.create_task(_broadcast(user_id, event, payload))
|
||
|
||
|
||
def _schedule_email(
|
||
notif_id: int,
|
||
recipient_user_id: int,
|
||
event_kind: str,
|
||
category: str,
|
||
payload: dict,
|
||
) -> None:
|
||
"""§15.4 dispatch — synchronous so the email_sent_at timestamp lands
|
||
in the same tick as the notification row, which is what the §15.5
|
||
digest's exclusion rule 1 keys on. The actual SMTP send happens in
|
||
`email.py`; this just consults the recipient's preferences and quiet
|
||
hours, then calls into the adapter."""
|
||
from . import email as email_mod # avoid import cycle
|
||
|
||
email_mod.maybe_send(
|
||
notif_id=notif_id,
|
||
recipient_user_id=recipient_user_id,
|
||
event_kind=event_kind,
|
||
category=category,
|
||
payload=payload,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Per-user mute and watch helpers used by api_notifications
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def list_inbox(
|
||
*,
|
||
user_id: int,
|
||
unread: bool | None = None,
|
||
rfc_slug: str | None = None,
|
||
category: str | None = None,
|
||
actor_user_id: int | None = None,
|
||
bundled: bool = False,
|
||
limit: int = 200,
|
||
) -> dict:
|
||
"""§15.2: inbox query. Filter chips are AND-combined.
|
||
|
||
When `bundled=True` is set, rows are grouped server-side by
|
||
(rfc_slug, event_kind) per the §15.2 bundle toggle, with the most
|
||
recent timestamp on each bundle.
|
||
"""
|
||
where = ["recipient_user_id = ?"]
|
||
args: list[Any] = [user_id]
|
||
if unread:
|
||
where.append("read_at IS NULL")
|
||
if rfc_slug:
|
||
where.append("rfc_slug = ?")
|
||
args.append(rfc_slug)
|
||
if actor_user_id is not None:
|
||
where.append("actor_user_id = ?")
|
||
args.append(actor_user_id)
|
||
rows = db.conn().execute(
|
||
f"""
|
||
SELECT n.id, n.event_kind, n.rfc_slug, n.branch_name, n.pr_number,
|
||
n.thread_id, n.actor_user_id, n.payload, n.created_at, n.read_at,
|
||
u.display_name AS actor_display, u.gitea_login AS actor_login,
|
||
r.title AS rfc_title
|
||
FROM notifications n
|
||
LEFT JOIN users u ON u.id = n.actor_user_id
|
||
LEFT JOIN cached_rfcs r ON r.slug = n.rfc_slug
|
||
WHERE {' AND '.join(where)}
|
||
ORDER BY n.id DESC
|
||
LIMIT ?
|
||
""",
|
||
(*args, limit),
|
||
).fetchall()
|
||
items = []
|
||
for row in rows:
|
||
try:
|
||
extras = json.loads(row["payload"] or "{}")
|
||
except json.JSONDecodeError:
|
||
extras = {}
|
||
if category and extras.get("category") != category:
|
||
continue
|
||
items.append({
|
||
"id": row["id"],
|
||
"event_kind": row["event_kind"],
|
||
"rfc_slug": row["rfc_slug"],
|
||
"rfc_title": row["rfc_title"],
|
||
"branch_name": row["branch_name"],
|
||
"pr_number": row["pr_number"],
|
||
"thread_id": row["thread_id"],
|
||
"actor_user_id": row["actor_user_id"],
|
||
"actor_login": row["actor_login"],
|
||
"actor_display": row["actor_display"],
|
||
"created_at": row["created_at"],
|
||
"read_at": row["read_at"],
|
||
"category": extras.get("category"),
|
||
"summary": render_summary(row["event_kind"], row["actor_display"], row["rfc_title"], extras),
|
||
})
|
||
|
||
if bundled:
|
||
items = _bundle(items)
|
||
|
||
unread_count = db.conn().execute(
|
||
"SELECT COUNT(*) AS c FROM notifications WHERE recipient_user_id = ? AND read_at IS NULL",
|
||
(user_id,),
|
||
).fetchone()["c"]
|
||
|
||
return {"items": items, "unread_count": unread_count}
|
||
|
||
|
||
def _bundle(items: list[dict]) -> list[dict]:
|
||
"""§15.2: collapse per-row notifications into per-(RFC, event_kind)
|
||
bundles; the bundle's representative row is the most-recent
|
||
constituent, with a `bundled_ids` array carrying the rest."""
|
||
buckets: dict[tuple, list[dict]] = {}
|
||
for it in items:
|
||
key = (it["rfc_slug"], it["event_kind"])
|
||
buckets.setdefault(key, []).append(it)
|
||
bundled = []
|
||
for key, rows in buckets.items():
|
||
rep = dict(rows[0]) # most recent (sorted DESC above)
|
||
rep["bundled_ids"] = [r["id"] for r in rows]
|
||
rep["bundled_count"] = len(rows)
|
||
bundled.append(rep)
|
||
bundled.sort(key=lambda r: r["created_at"], reverse=True)
|
||
return bundled
|
||
|
||
|
||
def mark_read_by_filter(
|
||
*,
|
||
user_id: int,
|
||
unread: bool | None = None,
|
||
rfc_slug: str | None = None,
|
||
category: str | None = None,
|
||
actor_user_id: int | None = None,
|
||
ids: Iterable[int] | None = None,
|
||
) -> int:
|
||
"""`Mark all read` per §15.2 — respects the active filter so the
|
||
user can mark all churn read without touching personal-direct."""
|
||
where = ["recipient_user_id = ?", "read_at IS NULL"]
|
||
args: list[Any] = [user_id]
|
||
if rfc_slug:
|
||
where.append("rfc_slug = ?")
|
||
args.append(rfc_slug)
|
||
if actor_user_id is not None:
|
||
where.append("actor_user_id = ?")
|
||
args.append(actor_user_id)
|
||
if ids:
|
||
ids = list(ids)
|
||
if not ids:
|
||
return 0
|
||
placeholders = ",".join("?" * len(ids))
|
||
where.append(f"id IN ({placeholders})")
|
||
args.extend(ids)
|
||
# Category lives inside payload JSON; we filter post-fetch to keep
|
||
# the SQL portable.
|
||
rows = db.conn().execute(
|
||
f"SELECT id, payload FROM notifications WHERE {' AND '.join(where)}",
|
||
args,
|
||
).fetchall()
|
||
selected_ids: list[int] = []
|
||
for r in rows:
|
||
if category:
|
||
try:
|
||
extras = json.loads(r["payload"] or "{}")
|
||
except json.JSONDecodeError:
|
||
extras = {}
|
||
if extras.get("category") != category:
|
||
continue
|
||
selected_ids.append(r["id"])
|
||
if not selected_ids:
|
||
return 0
|
||
placeholders = ",".join("?" * len(selected_ids))
|
||
db.conn().execute(
|
||
f"UPDATE notifications SET read_at = datetime('now') WHERE id IN ({placeholders})",
|
||
selected_ids,
|
||
)
|
||
for nid in selected_ids:
|
||
_schedule_broadcast(user_id, "read", {"id": nid})
|
||
return len(selected_ids)
|
||
|
||
|
||
def reconcile_seen_advance(
|
||
*,
|
||
user_id: int,
|
||
rfc_slug: str,
|
||
branch_name: str | None = None,
|
||
pr_number: int | None = None,
|
||
) -> int:
|
||
"""§15.7 reconciler — when a scope cursor advances on visit, mark
|
||
matching unread notifications read. Called from the chat-seen and
|
||
pr-seen advance endpoints.
|
||
"""
|
||
where = ["recipient_user_id = ?", "read_at IS NULL", "rfc_slug = ?"]
|
||
args: list[Any] = [user_id, rfc_slug]
|
||
if pr_number is not None:
|
||
where.append("pr_number = ?")
|
||
args.append(pr_number)
|
||
elif branch_name is not None:
|
||
where.append("branch_name = ?")
|
||
args.append(branch_name)
|
||
rows = db.conn().execute(
|
||
f"SELECT id FROM notifications WHERE {' AND '.join(where)}",
|
||
args,
|
||
).fetchall()
|
||
ids = [r["id"] for r in rows]
|
||
if not ids:
|
||
return 0
|
||
placeholders = ",".join("?" * len(ids))
|
||
db.conn().execute(
|
||
f"UPDATE notifications SET read_at = datetime('now') WHERE id IN ({placeholders})",
|
||
ids,
|
||
)
|
||
for nid in ids:
|
||
_schedule_broadcast(user_id, "read", {"id": nid})
|
||
return len(ids)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 90-day decay sweep (§15.6) — called by the digest job's nightly loop
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def decay_watches() -> int:
|
||
"""Downgrade `watching` rows whose last_participation_at is older
|
||
than 90 days to `following`. Explicit and role-implicit rows are
|
||
exempt; role-implicit isn't stored on the row, so it's a no-op for
|
||
those. Returns the number of rows downgraded."""
|
||
cur = db.conn().execute(
|
||
"""
|
||
UPDATE watches
|
||
SET state = 'following', last_participation_at = datetime('now')
|
||
WHERE state = 'watching'
|
||
AND set_by = 'auto'
|
||
AND (last_participation_at IS NULL
|
||
OR datetime(last_participation_at, '+90 days') < datetime('now'))
|
||
"""
|
||
)
|
||
return cur.rowcount or 0
|