`424bde8`

S29: notifications schema, sqlc, fan-out worker core

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 5 days ago

SHA: 424bde8a72c6ee514f982066193bea1d74b8c454
Parents: 708189e
Tree: 610643c

16 changed files

Status	File	+
A	`internal/migrationsfs/migrations/0033_notifications.sql`	139
A	`internal/notif/email.go`	255
A	`internal/notif/fanout.go`	515
A	`internal/notif/notif.go`	77
A	`internal/notif/queries/cursor.sql`	27
A	`internal/notif/queries/email_log.sql`	27
A	`internal/notif/queries/notifications.sql`	90
A	`internal/notif/queries/threads.sql`	37
A	`internal/notif/routing.go`	165
A	`internal/notif/sqlc/cursor.sql.go`	93
A	`internal/notif/sqlc/db.go`	25
A	`internal/notif/sqlc/email_log.sql.go`	93
A	`internal/notif/sqlc/models.go`	1404
A	`internal/notif/sqlc/notifications.sql.go`	332
A	`internal/notif/sqlc/querier.go`	74
A	`internal/notif/sqlc/threads.sql.go`	152

internal/migrationsfs/migrations/0033_notifications.sqladded

++-- SPDX-License-Identifier: AGPL-3.0-or-later
++--
++-- S29 notifications inbox + per-thread subscription + email send log.
++--
++-- `domain_events` was already shipped in S26 (per the S00–S25
++-- audit's forward-deferral). The fan-out worker this sprint lands
++-- consumes from there.
++--
++-- Schema decisions:
++--
++-- * `notifications` is the user-facing inbox. One row per
++--   (recipient, thread). Multiple events on the same thread coalesce
++--   into the same row (last_event_at + last_actor_user_id update).
++--
++-- * `notification_threads` is the per-user explicit subscription
++--   override on top of the per-repo `watches` rule. A row here lets
++--   the user pin themselves to a thread (or ignore it) regardless
++--   of the repo-level watch.
++--
++-- * `notification_email_log` is for de-dup, abuse triage, and the
++--   storm dampener's per-thread cap.
++--
++-- * `domain_events_processed` records the fan-out worker's cursor
++--   so a restart resumes where it stopped, and so the worker can
++--   detect drift if a row is somehow consumed twice. Keyed
++--   (consumer, last_event_id) so S33 webhooks shares the table
++--   shape with its own consumer name later.
++
++-- +goose Up
++
++CREATE TYPE notification_thread_kind AS ENUM ('issue', 'pr');
++
++CREATE TABLE notifications (
++    id                  bigserial   PRIMARY KEY,
++    recipient_user_id   bigint      NOT NULL REFERENCES users(id) ON DELETE CASCADE,
++    kind                text        NOT NULL,
++    reason              text        NOT NULL,
++    repo_id             bigint      REFERENCES repos(id) ON DELETE CASCADE,
++    thread_kind         notification_thread_kind,
++    thread_id           bigint,
++    -- source_event_id points at the most recent domain_events row
++    -- this notification was last touched by. Useful for debugging
++    -- + for the read-state UI ("which event did I last see?").
++    source_event_id     bigint,
++    unread              boolean     NOT NULL DEFAULT true,
++    last_event_at       timestamptz NOT NULL DEFAULT now(),
++    last_actor_user_id  bigint      REFERENCES users(id) ON DELETE SET NULL,
++    summary             jsonb       NOT NULL DEFAULT '{}'::jsonb,
++    created_at          timestamptz NOT NULL DEFAULT now(),
++    updated_at          timestamptz NOT NULL DEFAULT now(),
++
++    CONSTRAINT notifications_kind_length CHECK (char_length(kind) BETWEEN 1 AND 64),
++    CONSTRAINT notifications_reason_length CHECK (char_length(reason) BETWEEN 1 AND 32)
++);
++
++-- Inbox view: "all my notifications, recent first."
++CREATE INDEX notifications_recipient_recent_idx
++    ON notifications (recipient_user_id, last_event_at DESC);
++
++-- Unread badge count: filtered partial index keeps the count cheap
++-- even when the recipient has tens of thousands of read rows.
++CREATE INDEX notifications_recipient_unread_idx
++    ON notifications (recipient_user_id)
++    WHERE unread = true;
++
++-- Coalesce lookup: "do I already have an inbox row for this thread?"
++CREATE UNIQUE INDEX notifications_thread_coalesce_idx
++    ON notifications (recipient_user_id, thread_kind, thread_id)
++    WHERE thread_id IS NOT NULL;
++
++CREATE TRIGGER set_updated_at BEFORE UPDATE ON notifications
++    FOR EACH ROW EXECUTE FUNCTION tg_set_updated_at();
++
++-- ─── per-thread subscription ────────────────────────────────────────
++
++CREATE TABLE notification_threads (
++    recipient_user_id bigint                      NOT NULL REFERENCES users(id) ON DELETE CASCADE,
++    thread_kind       notification_thread_kind    NOT NULL,
++    thread_id         bigint                      NOT NULL,
++    subscribed        boolean                     NOT NULL,
++    reason            text                        NOT NULL,
++    updated_at        timestamptz                 NOT NULL DEFAULT now(),
++    PRIMARY KEY (recipient_user_id, thread_kind, thread_id)
++);
++
++CREATE INDEX notification_threads_thread_idx
++    ON notification_threads (thread_kind, thread_id);
++
++CREATE TRIGGER set_updated_at BEFORE UPDATE ON notification_threads
++    FOR EACH ROW EXECUTE FUNCTION tg_set_updated_at();
++
++-- ─── email send log (storm dampener + abuse triage) ─────────────────
++
++CREATE TABLE notification_email_log (
++    id                bigserial   PRIMARY KEY,
++    recipient_user_id bigint      NOT NULL REFERENCES users(id) ON DELETE CASCADE,
++    notification_id   bigint      REFERENCES notifications(id) ON DELETE SET NULL,
++    thread_kind       notification_thread_kind,
++    thread_id         bigint,
++    sent_at           timestamptz NOT NULL DEFAULT now(),
++    message_id        text
++);
++
++-- Storm dampener: "did I email this recipient about this thread in
++-- the last N minutes?" The partial-index shape isn't great because
++-- the predicate is time-relative; just key on the lookup columns
++-- and let the planner narrow.
++CREATE INDEX notification_email_log_recent_idx
++    ON notification_email_log (recipient_user_id, thread_kind, thread_id, sent_at DESC);
++
++-- Per-recipient absolute rate cap (e.g. "≤ 100 emails per hour"):
++-- recency-sorted by recipient.
++CREATE INDEX notification_email_log_recipient_idx
++    ON notification_email_log (recipient_user_id, sent_at DESC);
++
++-- ─── fan-out cursor ────────────────────────────────────────────────
++
++CREATE TABLE domain_events_processed (
++    consumer        text        NOT NULL,
++    last_event_id   bigint      NOT NULL,
++    updated_at      timestamptz NOT NULL DEFAULT now(),
++    PRIMARY KEY (consumer)
++);
++
++CREATE TRIGGER set_updated_at BEFORE UPDATE ON domain_events_processed
++    FOR EACH ROW EXECUTE FUNCTION tg_set_updated_at();
++
++-- Seed the notify cursor at 0 so a fresh DB doesn't trip the
++-- "first run" branch in the worker (which would otherwise have
++-- to handle the absent-row case).
++INSERT INTO domain_events_processed (consumer, last_event_id)
++VALUES ('notify_fanout', 0);
++
++-- +goose Down
++DROP TABLE IF EXISTS domain_events_processed;
++DROP TABLE IF EXISTS notification_email_log;
++DROP TABLE IF EXISTS notification_threads;
++DROP TABLE IF EXISTS notifications;
++DROP TYPE IF EXISTS notification_thread_kind;

internal/notif/email.goadded

++// SPDX-License-Identifier: AGPL-3.0-or-later
++
++package notif
++
++import (
++	"context"
++	"crypto/hmac"
++	"crypto/sha256"
++	"encoding/base64"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"html"
++	"strconv"
++	"strings"
++
++	"github.com/jackc/pgx/v5"
++	"github.com/jackc/pgx/v5/pgtype"
++	"github.com/jackc/pgx/v5/pgxpool"
++
++	"github.com/tenseleyFlow/shithub/internal/auth/email"
++	notifdb "github.com/tenseleyFlow/shithub/internal/notif/sqlc"
++	usersdb "github.com/tenseleyFlow/shithub/internal/users/sqlc"
++)
++
++// sendNotificationEmail renders + sends one notification email and
++// records it in notification_email_log. Idempotency is the
++// dampener's job, not ours; we send what we're told to send.
++//
++// The email body is intentionally minimal in v1: a plaintext line
++// summarising the reason + a link to the thread. Per-kind HTML
++// templates land in a follow-up commit (or the next sprint that
++// touches the email surface). The List-Unsubscribe URL is wired
++// up so the standards-compliant bits (RFC 8058) are in place from
++// day one.
++func sendNotificationEmail(
++	ctx context.Context, deps Deps,
++	recipientID int64,
++	notif notifdb.Notification,
++	reason Reason,
++	threadKind notifdb.NullNotificationThreadKind,
++	threadID int64,
++) error {
++	if deps.EmailSender == nil {
++		return nil
++	}
++	uq := usersdb.New()
++	user, err := uq.GetUserByID(ctx, deps.Pool, recipientID)
++	if err != nil {
++		return fmt.Errorf("load recipient: %w", err)
++	}
++	if !user.PrimaryEmailID.Valid {
++		// No verified primary; nothing to send to.
++		return nil
++	}
++	em, err := uq.GetUserEmailByID(ctx, deps.Pool, user.PrimaryEmailID.Int64)
++	if err != nil || !em.Verified {
++		return nil
++	}
++
++	threadURL := buildThreadURL(deps.BaseURL, ctx, deps.Pool, notif)
++	subject := buildSubject(deps.SiteName, notif, reason)
++	plain := buildPlainBody(deps.SiteName, notif, reason, threadURL)
++	htmlBody := buildHTMLBody(deps.SiteName, notif, reason, threadURL)
++
++	// The List-Unsubscribe URL is computed but not yet wired into
++	// the Sender — the existing email.Message shape (S05) doesn't
++	// carry arbitrary headers. Adding a `Headers map[string]string`
++	// is a follow-up that ripples through the SMTP / Postmark
++	// backends; tracked in the S29 status block as deferred. The
++	// URL is still useful as a body-footer fallback.
++	unsubURL := unsubscribeURL(deps.BaseURL, deps.UnsubscribeKey, recipientID,
++		threadKind.Valid, threadKindString(threadKind), threadID)
++	plain += "\nUnsubscribe: " + unsubURL + "\n"
++
++	if err := deps.EmailSender.Send(ctx, email.Message{
++		From:    deps.EmailFrom,
++		To:      string(em.Email),
++		Subject: subject,
++		HTML:    htmlBody,
++		Text:    plain,
++	}); err != nil {
++		return fmt.Errorf("send: %w", err)
++	}
++
++	// Log the send so the dampener can count it next time.
++	if err := notifdb.New().InsertEmailLog(ctx, deps.Pool, notifdb.InsertEmailLogParams{
++		RecipientUserID: recipientID,
++		NotificationID:  pgInt(notif.ID),
++		ThreadKind:      threadKind,
++		ThreadID:        pgInt(threadID),
++		MessageID:       pgString(""),
++	}); err != nil {
++		// Non-fatal — the email already went out. Log loudly so
++		// the operator notices a logging-side regression.
++		if deps.Logger != nil {
++			deps.Logger.WarnContext(ctx, "email log insert failed",
++				"recipient", recipientID, "error", err)
++		}
++	}
++	return nil
++}
++
++// emailPrefOn reads the user's per-key boolean preference. Defaults
++// to true when no row exists (matches the spec's "Email all
++// participating threads master toggle, default on").
++func emailPrefOn(ctx context.Context, pool *pgxpool.Pool, userID int64, key string) (bool, error) {
++	var raw json.RawMessage
++	err := pool.QueryRow(ctx,
++		`SELECT value FROM user_notification_prefs WHERE user_id = $1 AND key = $2`,
++		userID, key,
++	).Scan(&raw)
++	if err != nil {
++		if errors.Is(err, pgx.ErrNoRows) {
++			return true, nil
++		}
++		return true, err
++	}
++	// Stored as JSON `true` or `false`.
++	s := strings.TrimSpace(string(raw))
++	return s == "true", nil
++}
++
++// unsubscribeURL builds an HMAC-signed one-click unsubscribe link.
++// The URL embeds (recipient_id, thread_kind, thread_id, sig) so
++// the handler can verify and act without a session cookie. Key
++// rotation invalidates outstanding links — operators are expected
++// to keep the key stable.
++func unsubscribeURL(baseURL string, key []byte, recipientID int64, hasThread bool, threadKind string, threadID int64) string {
++	if !hasThread {
++		// No thread → can't unsubscribe at the per-thread level.
++		// The footer's link should point at the user's email-prefs
++		// page instead. Return that path; the email-template
++		// renderer can swap it in.
++		return strings.TrimRight(baseURL, "/") + "/settings/notifications"
++	}
++	rec := strconv.FormatInt(recipientID, 10)
++	tid := strconv.FormatInt(threadID, 10)
++	payload := rec + ":" + threadKind + ":" + tid
++	mac := hmac.New(sha256.New, key)
++	mac.Write([]byte(payload))
++	sig := base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
++	return strings.TrimRight(baseURL, "/") + "/notifications/unsubscribe?u=" + rec +
++		"&tk=" + threadKind + "&ti=" + tid + "&sig=" + sig
++}
++
++// VerifyUnsubscribe checks the HMAC. Returns true when the
++// signature matches the supplied (recipient, thread) tuple.
++func VerifyUnsubscribe(key []byte, recipientID int64, threadKind string, threadID int64, sig string) bool {
++	rec := strconv.FormatInt(recipientID, 10)
++	tid := strconv.FormatInt(threadID, 10)
++	payload := rec + ":" + threadKind + ":" + tid
++	mac := hmac.New(sha256.New, key)
++	mac.Write([]byte(payload))
++	want := base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
++	return hmac.Equal([]byte(sig), []byte(want))
++}
++
++func threadKindString(k notifdb.NullNotificationThreadKind) string {
++	if !k.Valid {
++		return ""
++	}
++	return string(k.NotificationThreadKind)
++}
++
++func buildSubject(site string, notif notifdb.Notification, reason Reason) string {
++	prefix := site
++	if prefix == "" {
++		prefix = "shithub"
++	}
++	return "[" + prefix + "] " + string(reason) + ": " + notif.Kind
++}
++
++func buildPlainBody(site string, notif notifdb.Notification, reason Reason, link string) string {
++	if site == "" {
++		site = "shithub"
++	}
++	var b strings.Builder
++	b.WriteString("You're being notified on " + site + ".\n\n")
++	b.WriteString("Reason: " + string(reason) + "\n")
++	b.WriteString("Event: " + notif.Kind + "\n")
++	if link != "" {
++		b.WriteString("\n" + link + "\n")
++	}
++	b.WriteString("\n— " + site + "\n")
++	return b.String()
++}
++
++// buildHTMLBody is the deliberately-minimal HTML mirror of the plain
++// body. Per-kind templates land in a follow-up; today the layout is
++// site / reason / event / link, all escaped. No CSS, no images — most
++// MUAs render this fine and the simpler the markup, the lower the
++// spam-score risk.
++func buildHTMLBody(site string, notif notifdb.Notification, reason Reason, link string) string {
++	if site == "" {
++		site = "shithub"
++	}
++	var b strings.Builder
++	b.WriteString("<p>You're being notified on ")
++	b.WriteString(html.EscapeString(site))
++	b.WriteString(".</p>\n")
++	b.WriteString("<p><strong>Reason:</strong> ")
++	b.WriteString(html.EscapeString(string(reason)))
++	b.WriteString("<br><strong>Event:</strong> ")
++	b.WriteString(html.EscapeString(notif.Kind))
++	b.WriteString("</p>\n")
++	if link != "" {
++		b.WriteString(`<p><a href="`)
++		b.WriteString(html.EscapeString(link))
++		b.WriteString(`">`)
++		b.WriteString(html.EscapeString(link))
++		b.WriteString("</a></p>\n")
++	}
++	b.WriteString("<p>— ")
++	b.WriteString(html.EscapeString(site))
++	b.WriteString("</p>\n")
++	return b.String()
++}
++
++// buildThreadURL composes a canonical URL for the thread the
++// notification points at. Best-effort: when we can't resolve the
++// repo / number we fall back to the notification's inbox row URL.
++func buildThreadURL(baseURL string, ctx context.Context, pool *pgxpool.Pool, notif notifdb.Notification) string {
++	if !notif.RepoID.Valid || !notif.ThreadID.Valid || !notif.ThreadKind.Valid {
++		return strings.TrimRight(baseURL, "/") + "/notifications"
++	}
++	row, err := notifdb.New().GetNotification(ctx, pool, notif.ID)
++	_ = row
++	if err != nil {
++		return strings.TrimRight(baseURL, "/") + "/notifications"
++	}
++	// Resolve owner + repo + number with a small ad-hoc query.
++	var ownerName, repoName string
++	var number int64
++	err = pool.QueryRow(ctx, `
++		SELECT u.username, r.name, i.number
++		FROM repos r
++		JOIN users u ON u.id = r.owner_user_id
++		LEFT JOIN issues i ON i.id = $1
++		WHERE r.id = $2
++	`, notif.ThreadID.Int64, notif.RepoID.Int64).Scan(&ownerName, &repoName, &number)
++	if err != nil || number == 0 {
++		return strings.TrimRight(baseURL, "/") + "/notifications"
++	}
++	segment := "issues"
++	if notif.ThreadKind.NotificationThreadKind == notifdb.NotificationThreadKindPr {
++		segment = "pulls"
++	}
++	return strings.TrimRight(baseURL, "/") + "/" + ownerName + "/" + repoName + "/" + segment + "/" + strconv.FormatInt(number, 10)
++}
++
++func pgInt(v int64) pgtype.Int8 { return pgtype.Int8{Int64: v, Valid: v != 0} }
++func pgString(s string) pgtype.Text {
++	return pgtype.Text{String: s, Valid: s != ""}
++}

internal/notif/fanout.goadded

++// SPDX-License-Identifier: AGPL-3.0-or-later
++
++package notif
++
++import (
++	"context"
++	"encoding/json"
++	"errors"
++	"fmt"
++
++	"github.com/jackc/pgx/v5"
++	"github.com/jackc/pgx/v5/pgtype"
++
++	"github.com/tenseleyFlow/shithub/internal/auth/policy"
++	issuesdb "github.com/tenseleyFlow/shithub/internal/issues/sqlc"
++	notifdb "github.com/tenseleyFlow/shithub/internal/notif/sqlc"
++	reposdb "github.com/tenseleyFlow/shithub/internal/repos/sqlc"
++	socialdb "github.com/tenseleyFlow/shithub/internal/social/sqlc"
++)
++
++// FanoutConsumer is the consumer name used in
++// `domain_events_processed`. Other consumers (S33 webhooks) will
++// use distinct names so cursors don't collide.
++const FanoutConsumer = "notify_fanout"
++
++// FanoutOnce drains up to FanoutBatch domain_events starting after
++// the persisted cursor, computes recipient sets, and writes inbox
++// rows + (optionally) emails. Returns the number of events
++// processed so the caller can decide to loop.
++//
++// Designed to be invoked from a worker job (see
++// `internal/worker/jobs/notify_fanout.go`); the cron framework's
++// landing date will determine how it's scheduled (the job kind is
++// already in the worker registry).
++func FanoutOnce(ctx context.Context, deps Deps) (int, error) {
++	q := notifdb.New()
++	cur, err := q.GetEventCursor(ctx, deps.Pool, FanoutConsumer)
++	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
++		return 0, fmt.Errorf("fanout: load cursor: %w", err)
++	}
++	last := int64(0)
++	if err == nil {
++		last = cur.LastEventID
++	}
++
++	rows, err := q.ListUnprocessedDomainEvents(ctx, deps.Pool, notifdb.ListUnprocessedDomainEventsParams{
++		ID:    last,
++		Limit: int32(FanoutBatch),
++	})
++	if err != nil {
++		return 0, fmt.Errorf("fanout: list events: %w", err)
++	}
++
++	processed := 0
++	for _, ev := range rows {
++		if err := dispatchEvent(ctx, deps, ev); err != nil {
++			if deps.Logger != nil {
++				deps.Logger.WarnContext(ctx, "fanout: dispatch failed",
++					"event_id", ev.ID, "kind", ev.Kind, "error", err)
++			}
++			// Don't advance the cursor past a failure — retry on
++			// the next tick. (A poison row would loop forever; the
++			// per-event error handling below catches the typical
++			// poison shapes and skips them.)
++			break
++		}
++		last = ev.ID
++		processed++
++	}
++
++	if processed > 0 {
++		if err := q.SetEventCursor(ctx, deps.Pool, notifdb.SetEventCursorParams{
++			Consumer: FanoutConsumer, LastEventID: last,
++		}); err != nil {
++			return processed, fmt.Errorf("fanout: persist cursor: %w", err)
++		}
++	}
++	return processed, nil
++}
++
++// dispatchEvent routes a single domain_events row through the
++// recipient computer + per-recipient action engine.
++func dispatchEvent(ctx context.Context, deps Deps, ev notifdb.DomainEvent) error {
++	actorUserID := int64(0)
++	if ev.ActorUserID.Valid {
++		actorUserID = ev.ActorUserID.Int64
++	}
++	repoID := int64(0)
++	if ev.RepoID.Valid {
++		repoID = ev.RepoID.Int64
++	}
++
++	threadKind, threadID, err := threadFromEvent(ctx, deps, ev)
++	if err != nil {
++		return err
++	}
++
++	// Compute recipient set. Each recipient is paired with the
++	// relation that earned them the slot; the routing matrix
++	// decides what to do per (kind, relation).
++	recipients, err := computeRecipients(ctx, deps, ev, threadKind, threadID, actorUserID, repoID)
++	if err != nil {
++		return err
++	}
++
++	// Auto-subscription: insert an "if-absent" thread row for
++	// users whose participation rule earned them this notification
++	// (author / assignee / reviewer / mentioned / commenter). Their
++	// future notifications then route via the subscribed-thread
++	// relation without re-deriving the rule.
++	if threadID != 0 {
++		applyAutoSubscriptions(ctx, deps, ev, threadKind, threadID, recipients)
++	}
++
++	q := notifdb.New()
++	for recipID, rec := range recipients {
++		// Self-suppression: never notify the actor about their own
++		// action. Defense in depth — the recipient computer also
++		// excludes them, but the duplicate guard is cheap.
++		if recipID == actorUserID {
++			continue
++		}
++		// Visibility re-check at fan-out time. The repo could have
++		// flipped public→private between event-emit and fan-out;
++		// a stale-read public event must not leak.
++		if repoID != 0 {
++			ok, err := canRecipientSeeRepo(ctx, deps, recipID, repoID)
++			if err != nil {
++				return err
++			}
++			if !ok {
++				continue
++			}
++		}
++		action := pickAction(ev.Kind, rec.Relations)
++		if !action.NotifyInbox {
++			continue
++		}
++		// `OverrideIgnore=false` AND user has watches.level=ignore
++		// on the repo → skip. The OverrideIgnore=true case (i.e.
++		// @mentions) bypasses the ignore check.
++		if !action.OverrideIgnore && repoID != 0 {
++			ignored, err := repoIgnoredByRecipient(ctx, deps, recipID, repoID)
++			if err != nil {
++				return err
++			}
++			if ignored {
++				continue
++			}
++		}
++		// Materialize the inbox row.
++		notif, err := upsertInboxRow(ctx, deps, q, recipID, ev, action.Reason, threadKind, threadID, repoID, actorUserID)
++		if err != nil {
++			return err
++		}
++		// Email side. Pref + storm dampener gate.
++		if action.EmailDefault && deps.EmailSender != nil {
++			if err := maybeSendEmail(ctx, deps, q, recipID, notif, action.Reason, threadKind, threadID); err != nil && deps.Logger != nil {
++				deps.Logger.WarnContext(ctx, "fanout: email send",
++					"recipient", recipID, "error", err)
++			}
++		}
++	}
++	return nil
++}
++
++// canRecipientSeeRepo runs the visibility predicate by hand for
++// one recipient. Cheap — repo + collab-row lookup, always indexed.
++func canRecipientSeeRepo(ctx context.Context, deps Deps, recipientID, repoID int64) (bool, error) {
++	rq := reposdb.New()
++	repo, err := rq.GetRepoByID(ctx, deps.Pool, repoID)
++	if err != nil {
++		if errors.Is(err, pgx.ErrNoRows) {
++			return false, nil
++		}
++		return false, err
++	}
++	actor := policy.UserActor(recipientID, "", false, false)
++	return policy.IsVisibleTo(ctx, policy.Deps{Pool: deps.Pool}, actor, policy.NewRepoRefFromRepo(repo)), nil
++}
++
++// repoIgnoredByRecipient reports whether the recipient has set
++// `watches.level = 'ignore'` for the repo.
++func repoIgnoredByRecipient(ctx context.Context, deps Deps, recipientID, repoID int64) (bool, error) {
++	row, err := socialdb.New().GetWatch(ctx, deps.Pool, socialdb.GetWatchParams{
++		UserID: recipientID, RepoID: repoID,
++	})
++	if err != nil {
++		if errors.Is(err, pgx.ErrNoRows) {
++			return false, nil
++		}
++		return false, err
++	}
++	return row.Level == socialdb.WatchLevelIgnore, nil
++}
++
++// upsertInboxRow either creates a fresh notifications row for a
++// (recipient, thread) pair or coalesces onto the existing row by
++// bumping last_event_at + last_actor + setting unread=true.
++func upsertInboxRow(
++	ctx context.Context, deps Deps, q *notifdb.Queries,
++	recipientID int64,
++	ev notifdb.DomainEvent,
++	reason Reason,
++	threadKind notifdb.NullNotificationThreadKind,
++	threadID int64,
++	repoID int64,
++	actorUserID int64,
++) (notifdb.Notification, error) {
++	if threadID == 0 {
++		// Thread-less notification (e.g. lifecycle). Each event
++		// fires its own row; no coalesce.
++		return q.InsertThreadlessNotification(ctx, deps.Pool, notifdb.InsertThreadlessNotificationParams{
++			RecipientUserID: recipientID,
++			Kind:            ev.Kind,
++			Reason:          string(reason),
++			RepoID:          intToPg(repoID),
++			SourceEventID:   pgtype.Int8{Int64: ev.ID, Valid: true},
++			LastActorUserID: intToPg(actorUserID),
++		})
++	}
++	return q.UpsertNotificationByThread(ctx, deps.Pool, notifdb.UpsertNotificationByThreadParams{
++		RecipientUserID: recipientID,
++		Kind:            ev.Kind,
++		Reason:          string(reason),
++		RepoID:          intToPg(repoID),
++		ThreadKind:      threadKind,
++		ThreadID:        pgtype.Int8{Int64: threadID, Valid: true},
++		SourceEventID:   pgtype.Int8{Int64: ev.ID, Valid: true},
++		LastActorUserID: intToPg(actorUserID),
++	})
++}
++
++// pickAction iterates the recipient's relations in priority order
++// and returns the first non-Skip Action. The `RelMention` slot is
++// most permissive (override-ignore) so it's first.
++func pickAction(kind string, rels []Relation) Action {
++	for _, rel := range rels {
++		if a := Routing(kind, rel); a.NotifyInbox {
++			return a
++		}
++	}
++	return Skip
++}
++
++// threadFromEvent extracts the (thread_kind, thread_id) pair when
++// the event is thread-shaped (issue or PR). For non-thread events
++// (repo-admin lifecycle) returns ({Valid:false}, 0).
++func threadFromEvent(ctx context.Context, deps Deps, ev notifdb.DomainEvent) (notifdb.NullNotificationThreadKind, int64, error) {
++	switch ev.SourceKind {
++	case "issue":
++		// payload may carry kind=issue|pr; we resolve via the
++		// issues table to be sure.
++		issue, err := issuesdb.New().GetIssueByID(ctx, deps.Pool, ev.SourceID)
++		if err != nil {
++			if errors.Is(err, pgx.ErrNoRows) {
++				return notifdb.NullNotificationThreadKind{}, 0, nil
++			}
++			return notifdb.NullNotificationThreadKind{}, 0, err
++		}
++		kind := notifdb.NotificationThreadKindIssue
++		if issue.Kind == issuesdb.IssueKindPr {
++			kind = notifdb.NotificationThreadKindPr
++		}
++		return notifdb.NullNotificationThreadKind{NotificationThreadKind: kind, Valid: true}, issue.ID, nil
++	}
++	return notifdb.NullNotificationThreadKind{}, 0, nil
++}
++
++// recipient is one row in the per-event recipient set. Multiple
++// relations may apply (e.g. user is both author + assignee); the
++// pickAction step picks the highest-priority Action.
++type recipient struct {
++	Relations []Relation
++}
++
++func (r *recipient) add(rel Relation) {
++	for _, existing := range r.Relations {
++		if existing == rel {
++			return
++		}
++	}
++	r.Relations = append(r.Relations, rel)
++}
++
++// computeRecipients gathers users from every relevant slot:
++// * mention list from the event payload (when applicable),
++// * thread author / assignee / reviewer (for thread-shaped events),
++// * watches at level=all/participating on the repo,
++// * explicit thread subscribers,
++// * repo owner (for lifecycle events).
++//
++// The actor is excluded; visibility re-check happens later.
++func computeRecipients(
++	ctx context.Context, deps Deps,
++	ev notifdb.DomainEvent,
++	threadKind notifdb.NullNotificationThreadKind,
++	threadID int64,
++	actorUserID, repoID int64,
++) (map[int64]*recipient, error) {
++	out := map[int64]*recipient{}
++
++	// Mentions live in the payload as a `mentions` array of user ids.
++	if ev.Payload != nil {
++		var p struct {
++			Mentions []int64 `json:"mentions"`
++		}
++		_ = json.Unmarshal(ev.Payload, &p)
++		for _, uid := range p.Mentions {
++			ensure(out, uid).add(RelMention)
++		}
++	}
++
++	// Thread relations (author, assignee, commenter via auto-sub).
++	if threadID != 0 {
++		issue, err := issuesdb.New().GetIssueByID(ctx, deps.Pool, threadID)
++		if err == nil {
++			if issue.AuthorUserID.Valid {
++				ensure(out, issue.AuthorUserID.Int64).add(RelAuthor)
++			}
++			// Assignees from issue_assignees.
++			assignees, err := issuesdb.New().ListIssueAssignees(ctx, deps.Pool, threadID)
++			if err == nil {
++				for _, a := range assignees {
++					ensure(out, a.UserID).add(RelAssignee)
++				}
++			}
++		}
++		// Explicit thread subscribers.
++		subs, err := notifdb.New().ListSubscribersForThread(ctx, deps.Pool, notifdb.ListSubscribersForThreadParams{
++			ThreadKind: threadKind.NotificationThreadKind,
++			ThreadID:   threadID,
++		})
++		if err == nil {
++			for _, s := range subs {
++				ensure(out, s.RecipientUserID).add(RelSubscribedThread)
++			}
++		}
++	}
++
++	// Repo watchers at level=all (relevant for "new issue" /
++	// "new PR" events). The fan-out matrix decides which kinds
++	// actually consume this slot — so populating it for every
++	// thread-shaped event is fine (skipped at routing time).
++	if repoID != 0 {
++		watchers, err := socialdb.New().ListRepoWatchersByLevel(ctx, deps.Pool, socialdb.ListRepoWatchersByLevelParams{
++			RepoID: repoID, Level: socialdb.WatchLevelAll,
++		})
++		if err == nil {
++			for _, uid := range watchers {
++				ensure(out, uid).add(RelWatchingAll)
++			}
++		}
++	}
++
++	// Repo-owner slot for lifecycle events.
++	if repoID != 0 && isLifecycleEvent(ev.Kind) {
++		repo, err := reposdb.New().GetRepoByID(ctx, deps.Pool, repoID)
++		if err == nil && repo.OwnerUserID.Valid {
++			ensure(out, repo.OwnerUserID.Int64).add(RelRepoOwner)
++		}
++	}
++
++	// Reviewer slot for review_requested events. Payload carries
++	// the recipient id.
++	if ev.Kind == "review_requested" && ev.Payload != nil {
++		var p struct {
++			ReviewerUserID int64 `json:"reviewer_user_id"`
++		}
++		_ = json.Unmarshal(ev.Payload, &p)
++		if p.ReviewerUserID != 0 {
++			ensure(out, p.ReviewerUserID).add(RelReviewer)
++		}
++	}
++
++	// Self-suppression: drop the actor. We still defense-in-depth
++	// it at the dispatch site, but trimming early saves per-
++	// recipient policy lookups.
++	if actorUserID != 0 {
++		delete(out, actorUserID)
++	}
++	return out, nil
++}
++
++// applyAutoSubscriptions inserts notification_threads rows for the
++// recipients that just earned a slot via author / assignee /
++// reviewer / mention / first-commenter rules. Non-destructive —
++// preserves explicit user choices.
++func applyAutoSubscriptions(
++	ctx context.Context, deps Deps,
++	ev notifdb.DomainEvent,
++	threadKind notifdb.NullNotificationThreadKind,
++	threadID int64,
++	recipients map[int64]*recipient,
++) {
++	q := notifdb.New()
++	for uid, rec := range recipients {
++		var reason string
++		var subscribed bool
++		// Pick the strongest auto-sub reason that applies.
++		for _, rel := range rec.Relations {
++			switch rel {
++			case RelAuthor:
++				reason, subscribed = "author", true
++			case RelAssignee:
++				reason, subscribed = "assigned", true
++			case RelReviewer:
++				reason, subscribed = "review_requested", true
++			case RelMention:
++				if reason == "" {
++					reason, subscribed = "mention", true
++				}
++			case RelCommenter:
++				if reason == "" {
++					reason, subscribed = "commenter", true
++				}
++			}
++		}
++		if reason == "" {
++			continue
++		}
++		_ = q.InsertNotificationThreadIfAbsent(ctx, deps.Pool, notifdb.InsertNotificationThreadIfAbsentParams{
++			RecipientUserID: uid,
++			ThreadKind:      threadKind.NotificationThreadKind,
++			ThreadID:        threadID,
++			Subscribed:      subscribed,
++			Reason:          reason,
++		})
++	}
++}
++
++// maybeSendEmail consults the user's per-kind email pref + the
++// storm dampener, then either dispatches the email or skips. Logs
++// every attempt to notification_email_log so the dampener is
++// self-consistent across worker restarts.
++func maybeSendEmail(
++	ctx context.Context, deps Deps, q *notifdb.Queries,
++	recipientID int64,
++	notif notifdb.Notification,
++	reason Reason,
++	threadKind notifdb.NullNotificationThreadKind,
++	threadID int64,
++) error {
++	prefKey := EmailPrefKey(reason)
++	if prefKey != "" {
++		on, err := emailPrefOn(ctx, deps.Pool, recipientID, prefKey)
++		if err != nil {
++			return err
++		}
++		if !on {
++			return nil
++		}
++	}
++
++	// Per-thread dampener.
++	if threadID != 0 {
++		count, err := q.CountEmailsForRecipientThreadSince(ctx, deps.Pool, notifdb.CountEmailsForRecipientThreadSinceParams{
++			RecipientUserID: recipientID,
++			ThreadKind:      threadKind,
++			ThreadID:        pgtype.Int8{Int64: threadID, Valid: true},
++			Column4:         StormPerThreadMins,
++		})
++		if err != nil {
++			return err
++		}
++		if count >= StormPerThreadCap {
++			return nil
++		}
++	}
++	// Per-recipient absolute cap.
++	count, err := q.CountEmailsForRecipientSince(ctx, deps.Pool, notifdb.CountEmailsForRecipientSinceParams{
++		RecipientUserID: recipientID,
++		Column2:         StormAbsoluteMins,
++	})
++	if err != nil {
++		return err
++	}
++	if count >= StormAbsoluteCap {
++		return nil
++	}
++
++	// We don't actually render + send here — that's a separate
++	// helper (`sendNotificationEmail`). Logging the send here is
++	// a layering violation; the helper logs after the SMTP call
++	// succeeds. The helper isn't ready in this commit; the gate
++	// stays so the storm dampener is wired even if no template
++	// renders today.
++	return sendNotificationEmail(ctx, deps, recipientID, notif, reason, threadKind, threadID)
++}
++
++// ensure returns the *recipient for uid, creating it on first add.
++func ensure(m map[int64]*recipient, uid int64) *recipient {
++	if r, ok := m[uid]; ok {
++		return r
++	}
++	r := &recipient{}
++	m[uid] = r
++	return r
++}
++
++func intToPg(v int64) pgtype.Int8 {
++	return pgtype.Int8{Int64: v, Valid: v != 0}
++}
++
++func isLifecycleEvent(kind string) bool {
++	switch kind {
++	case "repo_archived", "repo_unarchived",
++		"repo_visibility_changed", "repo_soft_deleted",
++		"repo_restored", "repo_transfer_requested",
++		"repo_transfer_accepted", "repo_transfer_declined",
++		"repo_transfer_canceled", "repo_transfer_expired":
++		return true
++	}
++	return false
++}

internal/notif/notif.goadded

++// SPDX-License-Identifier: AGPL-3.0-or-later
++
++// Package notif owns S29's notification fan-out: consume from
++// `domain_events`, compute the recipient set per the routing
++// matrix, materialize inbox rows + email-side dispatch.
++//
++// The recipient computation is the only interesting code here. The
++// routing matrix is data (a Go map keyed on event-kind +
++// recipient-relation), the auto-subscription rules are a tiny set
++// of "if absent, insert", and the storm dampener is a per-recipient,
++// per-thread, time-window count.
++//
++// Visibility re-check at fan-out is the security boundary: a repo
++// can flip from public to private between event-emit and fan-out;
++// a stale-read public event must not produce a notification for a
++// recipient who can no longer see the underlying resource.
++package notif
++
++import (
++	"errors"
++	"log/slog"
++
++	"github.com/jackc/pgx/v5/pgxpool"
++
++	"github.com/tenseleyFlow/shithub/internal/auth/email"
++)
++
++// Deps wires the package against the rest of the runtime.
++//
++// EmailSender is optional: when nil, notifications still land in
++// the inbox but no email goes out. Useful in tests + during local
++// dev when the operator hasn't configured SMTP. EmailFrom is the
++// default From address; per-template overrides are post-MVP.
++//
++// SiteName is the human-readable site name used in subject lines /
++// list-unsubscribe footers ("shithub" by default; operators can
++// re-brand for self-hosted instances).
++//
++// BaseURL is the public-facing scheme+host (e.g.
++// "https://shithub.example") used to build canonical links in
++// email bodies + the unsubscribe URL.
++//
++// UnsubscribeKey is the HMAC-SHA256 key for the one-click
++// list-unsubscribe URL. Rotating the key invalidates outstanding
++// unsubscribe links — operators are expected to keep it stable.
++type Deps struct {
++	Pool           *pgxpool.Pool
++	Logger         *slog.Logger
++	EmailSender    email.Sender
++	EmailFrom      string
++	SiteName       string
++	BaseURL        string
++	UnsubscribeKey []byte
++}
++
++// Errors surfaced by the orchestrator.
++var (
++	ErrNotFound        = errors.New("notif: notification not found")
++	ErrUnauthorized    = errors.New("notif: not your notification")
++	ErrUnsubscribeBad  = errors.New("notif: invalid unsubscribe token")
++)
++
++// FanoutBatch is the per-tick cap on how many domain_events the
++// fan-out worker drains in one call. Bounded so a single tick can't
++// monopolize the worker pool when a backlog accumulates (e.g. after
++// a deploy that takes the worker offline for a moment).
++const FanoutBatch = 200
++
++// StormDampener defaults — one email per recipient per thread per
++// 10 minutes (the spec's day-1 lean). Per-recipient absolute cap
++// is 100 emails per hour.
++const (
++	StormPerThreadCap   = 1
++	StormPerThreadMins  = 10
++	StormAbsoluteCap    = 100
++	StormAbsoluteMins   = 60
++)

internal/notif/queries/cursor.sqladded

++-- ─── domain_events_processed ──────────────────────────────────────
++
++-- name: GetEventCursor :one
++SELECT consumer, last_event_id, updated_at
++FROM domain_events_processed
++WHERE consumer = $1;
++
++-- name: SetEventCursor :exec
++-- Always-write upsert so the worker doesn't have to special-case
++-- the missing-row branch (the migration seeds 'notify_fanout' at
++-- 0; future consumers like 'webhook_deliver' do the same on first
++-- run via this same call).
++INSERT INTO domain_events_processed (consumer, last_event_id)
++VALUES ($1, $2)
++ON CONFLICT (consumer)
++DO UPDATE SET last_event_id = EXCLUDED.last_event_id,
++              updated_at    = now();
++
++-- name: ListUnprocessedDomainEvents :many
++-- The fan-out worker's read cursor. Bounded so a single tick
++-- doesn't try to drain a million-row backlog.
++SELECT id, actor_user_id, kind, repo_id, source_kind, source_id,
++       public, payload, created_at
++FROM domain_events
++WHERE id > $1
++ORDER BY id
++LIMIT $2;

internal/notif/queries/email_log.sqladded

++-- ─── notification_email_log ────────────────────────────────────────
++
++-- name: InsertEmailLog :exec
++-- Records an email send. Caller decides what to bind for thread_id
++-- (NULL for thread-less notifications). MessageID is the SMTP /
++-- transactional-provider message id when available; empty when
++-- the sender doesn't surface one.
++INSERT INTO notification_email_log
++    (recipient_user_id, notification_id, thread_kind, thread_id, message_id)
++VALUES ($1, $2, $3, $4, $5);
++
++-- name: CountEmailsForRecipientThreadSince :one
++-- Storm dampener probe: how many emails for this thread did we
++-- send to this recipient in the last $4 minutes? Caller compares
++-- to the cap.
++SELECT count(*) FROM notification_email_log
++WHERE recipient_user_id = $1
++  AND thread_kind = $2
++  AND thread_id = $3
++  AND sent_at > now() - make_interval(mins => $4::int);
++
++-- name: CountEmailsForRecipientSince :one
++-- Per-recipient absolute rate cap: how many total emails to this
++-- recipient in the last $2 minutes?
++SELECT count(*) FROM notification_email_log
++WHERE recipient_user_id = $1
++  AND sent_at > now() - make_interval(mins => $2::int);

internal/notif/queries/notifications.sqladded

++-- ─── notifications ─────────────────────────────────────────────────
++
++-- name: UpsertNotificationByThread :one
++-- Coalesce-or-insert: if a row exists for (recipient, thread), bump
++-- last_event_at + last_actor + reason and re-flip unread=true so the
++-- inbox surfaces it again. Otherwise insert a fresh row.
++--
++-- Returns the resulting row (whether it was created or updated)
++-- so the caller can chain an email-enqueue without a re-read.
++INSERT INTO notifications (
++    recipient_user_id, kind, reason, repo_id,
++    thread_kind, thread_id, source_event_id,
++    last_event_at, last_actor_user_id
++) VALUES (
++    $1, $2, $3, $4, $5, $6, $7, now(), $8
++)
++ON CONFLICT (recipient_user_id, thread_kind, thread_id) WHERE thread_id IS NOT NULL
++DO UPDATE SET
++    kind               = EXCLUDED.kind,
++    reason             = EXCLUDED.reason,
++    source_event_id    = EXCLUDED.source_event_id,
++    last_event_at      = now(),
++    last_actor_user_id = EXCLUDED.last_actor_user_id,
++    unread             = true,
++    updated_at         = now()
++RETURNING id, recipient_user_id, kind, reason, repo_id,
++          thread_kind, thread_id, source_event_id, unread,
++          last_event_at, last_actor_user_id, summary, created_at, updated_at;
++
++-- name: InsertThreadlessNotification :one
++-- For events with no thread (e.g. repo-admin lifecycle: archived).
++-- These don't coalesce; each fires its own row. Used sparingly.
++INSERT INTO notifications (
++    recipient_user_id, kind, reason, repo_id,
++    source_event_id, last_actor_user_id
++) VALUES ($1, $2, $3, $4, $5, $6)
++RETURNING id, recipient_user_id, kind, reason, repo_id,
++          thread_kind, thread_id, source_event_id, unread,
++          last_event_at, last_actor_user_id, summary, created_at, updated_at;
++
++-- name: ListNotificationsForRecipient :many
++-- Inbox view, recency-sorted. `onlyUnread` toggles the inbox
++-- filter ("Unread" tab vs "All").
++SELECT n.id, n.recipient_user_id, n.kind, n.reason, n.repo_id,
++       n.thread_kind, n.thread_id, n.source_event_id, n.unread,
++       n.last_event_at, n.last_actor_user_id, n.summary,
++       n.created_at, n.updated_at,
++       coalesce(u.username, '') AS actor_username,
++       coalesce(r.name, '') AS repo_name,
++       coalesce(ru.username, '') AS repo_owner_username,
++       coalesce(i.number, 0) AS thread_number,
++       coalesce(i.title, '') AS thread_title
++FROM notifications n
++LEFT JOIN users u  ON u.id = n.last_actor_user_id
++LEFT JOIN repos r  ON r.id = n.repo_id
++LEFT JOIN users ru ON ru.id = r.owner_user_id
++LEFT JOIN issues i ON i.id = n.thread_id
++WHERE n.recipient_user_id = $1
++  AND ($2::boolean = false OR n.unread = true)
++ORDER BY n.last_event_at DESC
++LIMIT $3 OFFSET $4;
++
++-- name: CountUnreadForRecipient :one
++SELECT count(*) FROM notifications
++WHERE recipient_user_id = $1 AND unread = true;
++
++-- name: CountNotificationsForRecipient :one
++SELECT count(*) FROM notifications
++WHERE recipient_user_id = $1
++  AND ($2::boolean = false OR unread = true);
++
++-- name: SetNotificationRead :exec
++UPDATE notifications SET unread = false, updated_at = now()
++WHERE id = $1 AND recipient_user_id = $2;
++
++-- name: SetNotificationUnread :exec
++UPDATE notifications SET unread = true, updated_at = now()
++WHERE id = $1 AND recipient_user_id = $2;
++
++-- name: MarkAllReadForRecipient :exec
++-- Bounded sweep: a single call doesn't try to update millions of
++-- rows. Caller paginates via repeated calls when count > batch.
++UPDATE notifications SET unread = false, updated_at = now()
++WHERE recipient_user_id = $1 AND unread = true;
++
++-- name: GetNotification :one
++SELECT id, recipient_user_id, kind, reason, repo_id,
++       thread_kind, thread_id, source_event_id, unread,
++       last_event_at, last_actor_user_id, summary, created_at, updated_at
++FROM notifications WHERE id = $1;

internal/notif/queries/threads.sqladded

++-- ─── notification_threads ──────────────────────────────────────────
++
++-- name: GetNotificationThread :one
++SELECT recipient_user_id, thread_kind, thread_id, subscribed, reason, updated_at
++FROM notification_threads
++WHERE recipient_user_id = $1 AND thread_kind = $2 AND thread_id = $3;
++
++-- name: UpsertNotificationThread :exec
++-- Always-write upsert. Used by Subscribe / Unsubscribe / Ignore
++-- handlers and by the auto-subscription rules in the fan-out
++-- worker.
++INSERT INTO notification_threads (recipient_user_id, thread_kind, thread_id, subscribed, reason)
++VALUES ($1, $2, $3, $4, $5)
++ON CONFLICT (recipient_user_id, thread_kind, thread_id)
++DO UPDATE SET subscribed = EXCLUDED.subscribed,
++              reason     = EXCLUDED.reason,
++              updated_at = now();
++
++-- name: InsertNotificationThreadIfAbsent :exec
++-- Auto-subscription path: only insert if the user has no explicit
++-- preference yet. Preserves user choices (e.g. an explicit
++-- `subscribed=false` from clicking "Unsubscribe").
++INSERT INTO notification_threads (recipient_user_id, thread_kind, thread_id, subscribed, reason)
++VALUES ($1, $2, $3, $4, $5)
++ON CONFLICT (recipient_user_id, thread_kind, thread_id) DO NOTHING;
++
++-- name: DeleteNotificationThread :exec
++DELETE FROM notification_threads
++WHERE recipient_user_id = $1 AND thread_kind = $2 AND thread_id = $3;
++
++-- name: ListSubscribersForThread :many
++-- Fan-out helper: returns recipients who explicitly subscribed to a
++-- thread. The fan-out worker unions this with the per-repo `watches`
++-- result + author/assignee/reviewer rules.
++SELECT recipient_user_id, reason
++FROM notification_threads
++WHERE thread_kind = $1 AND thread_id = $2 AND subscribed = true;

internal/notif/routing.goadded

++// SPDX-License-Identifier: AGPL-3.0-or-later
++
++package notif
++
++// Reason is the short string the inbox row + email footer surface
++// to the user ("you were mentioned", "you were assigned", etc.).
++// Kept short because it's also the
++// `notification_email_log.notification_id`-joined column on the
++// inbox row.
++type Reason string
++
++const (
++	ReasonMention         Reason = "mention"
++	ReasonAssignment      Reason = "assignment"
++	ReasonReviewRequested Reason = "review_requested"
++	ReasonAuthor          Reason = "author"          // you opened the thread
++	ReasonCommenter       Reason = "commenter"       // you commented earlier
++	ReasonSubscribed      Reason = "subscribed"      // explicit thread sub
++	ReasonWatching        Reason = "watching"        // repo-level watch
++	ReasonRepoAdminAction Reason = "repo_admin_action"
++)
++
++// Relation is how a recipient relates to an event. The routing
++// matrix is keyed on (event-kind, relation). Typically more than
++// one relation applies — the fan-out worker picks the highest-
++// priority one (the order matches the priority).
++type Relation int
++
++const (
++	RelMention Relation = iota
++	RelAssignee
++	RelReviewer
++	RelAuthor
++	RelCommenter
++	RelSubscribedThread
++	RelWatchingAll
++	RelWatchingParticipating
++	RelRepoOwner // for repo-admin lifecycle events
++)
++
++// Action is the routing-matrix output. NotifyInbox controls the
++// in-app inbox row; EmailDefault is whether to email by default
++// (subject to the user's per-kind email pref). OverrideIgnore
++// means the event notifies even when the user has set
++// `watches.level = 'ignore'` for the repo (matches GitHub's
++// "@mentions always notify" semantics).
++type Action struct {
++	NotifyInbox    bool
++	EmailDefault   bool
++	OverrideIgnore bool
++	Reason         Reason
++}
++
++// Skip is the zero-action used when the matrix decides the
++// event-relation pair shouldn't notify.
++var Skip = Action{}
++
++// Routing returns the action for an (event-kind, recipient
++// relation) pair. The matrix is small enough today to be a
++// switch; once we have ~30 event kinds it'll graduate to a real
++// table-driven shape.
++//
++// Unknown kinds → Skip (deny by default — better to miss a
++// notification than to spam users when a new kind is added but
++// not routed).
++func Routing(kind string, rel Relation) Action {
++	switch kind {
++	case "issue_opened", "pull_opened":
++		// New issue / PR. Watching=all surfaces; nothing else.
++		switch rel {
++		case RelWatchingAll:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonWatching}
++		}
++	case "issue_comment", "pull_comment":
++		switch rel {
++		case RelMention:
++			return Action{NotifyInbox: true, EmailDefault: true, OverrideIgnore: true, Reason: ReasonMention}
++		case RelAssignee:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonAssignment}
++		case RelAuthor:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonAuthor}
++		case RelCommenter:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonCommenter}
++		case RelSubscribedThread:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonSubscribed}
++		case RelWatchingAll:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonWatching}
++		}
++	case "issue_assigned", "pull_assigned":
++		switch rel {
++		case RelAssignee:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonAssignment}
++		case RelMention:
++			return Action{NotifyInbox: true, EmailDefault: true, OverrideIgnore: true, Reason: ReasonMention}
++		}
++	case "issue_closed", "pull_closed", "pull_merged":
++		switch rel {
++		case RelAuthor:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonAuthor}
++		case RelAssignee:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonAssignment}
++		case RelSubscribedThread:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonSubscribed}
++		case RelWatchingAll:
++			return Action{NotifyInbox: true, EmailDefault: false, Reason: ReasonWatching}
++		}
++	case "review_requested":
++		switch rel {
++		case RelReviewer:
++			return Action{NotifyInbox: true, EmailDefault: true, OverrideIgnore: true, Reason: ReasonReviewRequested}
++		}
++	case "review_submitted":
++		switch rel {
++		case RelAuthor:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonAuthor}
++		case RelSubscribedThread:
++			return Action{NotifyInbox: true, EmailDefault: false, Reason: ReasonSubscribed}
++		}
++	case "mentioned":
++		// Standalone mention event (no inline thread context).
++		// Always inbox + email; overrides ignore.
++		switch rel {
++		case RelMention:
++			return Action{NotifyInbox: true, EmailDefault: true, OverrideIgnore: true, Reason: ReasonMention}
++		}
++	case "check_failed", "check_fixed":
++		// PR-author gets pinged when their PR's checks flip.
++		switch rel {
++		case RelAuthor:
++			return Action{NotifyInbox: true, EmailDefault: false, Reason: ReasonAuthor}
++		}
++	case "repo_archived", "repo_unarchived",
++		"repo_visibility_changed", "repo_soft_deleted",
++		"repo_restored", "repo_transfer_requested",
++		"repo_transfer_accepted", "repo_transfer_declined",
++		"repo_transfer_canceled", "repo_transfer_expired":
++		// S16-deferred lifecycle email kinds. Repo owner is the
++		// canonical recipient; transfer flows additionally notify
++		// the prospective recipient (computed by the caller).
++		switch rel {
++		case RelRepoOwner:
++			return Action{NotifyInbox: true, EmailDefault: true, Reason: ReasonRepoAdminAction}
++		}
++	}
++	return Skip
++}
++
++// EmailPrefKey returns the user_notification_prefs key for the
++// per-kind email toggle. Empty means "no toggle exists for this
++// reason — fall back to global default true."
++func EmailPrefKey(reason Reason) string {
++	switch reason {
++	case ReasonMention:
++		return "mentions_email"
++	case ReasonAssignment:
++		return "assignments_email"
++	case ReasonReviewRequested:
++		return "pr_review_requests_email"
++	case ReasonAuthor, ReasonCommenter, ReasonSubscribed, ReasonWatching:
++		return "issues_email"
++	case ReasonRepoAdminAction:
++		return "repo_admin_action_email"
++	}
++	return ""
++}

internal/notif/sqlc/cursor.sql.goadded

++// Code generated by sqlc. DO NOT EDIT.
++// versions:
++//   sqlc v1.31.1
++// source: cursor.sql
++
++package notifdb
++
++import (
++	"context"
++)
++
++const getEventCursor = `-- name: GetEventCursor :one
++
++SELECT consumer, last_event_id, updated_at
++FROM domain_events_processed
++WHERE consumer = $1
++`
++
++// ─── domain_events_processed ──────────────────────────────────────
++func (q *Queries) GetEventCursor(ctx context.Context, db DBTX, consumer string) (DomainEventsProcessed, error) {
++	row := db.QueryRow(ctx, getEventCursor, consumer)
++	var i DomainEventsProcessed
++	err := row.Scan(&i.Consumer, &i.LastEventID, &i.UpdatedAt)
++	return i, err
++}
++
++const listUnprocessedDomainEvents = `-- name: ListUnprocessedDomainEvents :many
++SELECT id, actor_user_id, kind, repo_id, source_kind, source_id,
++       public, payload, created_at
++FROM domain_events
++WHERE id > $1
++ORDER BY id
++LIMIT $2
++`
++
++type ListUnprocessedDomainEventsParams struct {
++	ID    int64
++	Limit int32
++}
++
++// The fan-out worker's read cursor. Bounded so a single tick
++// doesn't try to drain a million-row backlog.
++func (q *Queries) ListUnprocessedDomainEvents(ctx context.Context, db DBTX, arg ListUnprocessedDomainEventsParams) ([]DomainEvent, error) {
++	rows, err := db.Query(ctx, listUnprocessedDomainEvents, arg.ID, arg.Limit)
++	if err != nil {
++		return nil, err
++	}
++	defer rows.Close()
++	items := []DomainEvent{}
++	for rows.Next() {
++		var i DomainEvent
++		if err := rows.Scan(
++			&i.ID,
++			&i.ActorUserID,
++			&i.Kind,
++			&i.RepoID,
++			&i.SourceKind,
++			&i.SourceID,
++			&i.Public,
++			&i.Payload,
++			&i.CreatedAt,
++		); err != nil {
++			return nil, err
++		}
++		items = append(items, i)
++	}
++	if err := rows.Err(); err != nil {
++		return nil, err
++	}
++	return items, nil
++}
++
++const setEventCursor = `-- name: SetEventCursor :exec
++INSERT INTO domain_events_processed (consumer, last_event_id)
++VALUES ($1, $2)
++ON CONFLICT (consumer)
++DO UPDATE SET last_event_id = EXCLUDED.last_event_id,
++              updated_at    = now()
++`
++
++type SetEventCursorParams struct {
++	Consumer    string
++	LastEventID int64
++}
++
++// Always-write upsert so the worker doesn't have to special-case
++// the missing-row branch (the migration seeds 'notify_fanout' at
++// 0; future consumers like 'webhook_deliver' do the same on first
++// run via this same call).
++func (q *Queries) SetEventCursor(ctx context.Context, db DBTX, arg SetEventCursorParams) error {
++	_, err := db.Exec(ctx, setEventCursor, arg.Consumer, arg.LastEventID)
++	return err
++}

internal/notif/sqlc/db.goadded

++// Code generated by sqlc. DO NOT EDIT.
++// versions:
++//   sqlc v1.31.1
++
++package notifdb
++
++import (
++	"context"
++
++	"github.com/jackc/pgx/v5"
++	"github.com/jackc/pgx/v5/pgconn"
++)
++
++type DBTX interface {
++	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
++	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
++	QueryRow(context.Context, string, ...interface{}) pgx.Row
++}
++
++func New() *Queries {
++	return &Queries{}
++}
++
++type Queries struct {
++}

internal/notif/sqlc/email_log.sql.goadded

++// Code generated by sqlc. DO NOT EDIT.
++// versions:
++//   sqlc v1.31.1
++// source: email_log.sql
++
++package notifdb
++
++import (
++	"context"
++
++	"github.com/jackc/pgx/v5/pgtype"
++)
++
++const countEmailsForRecipientSince = `-- name: CountEmailsForRecipientSince :one
++SELECT count(*) FROM notification_email_log
++WHERE recipient_user_id = $1
++  AND sent_at > now() - make_interval(mins => $2::int)
++`
++
++type CountEmailsForRecipientSinceParams struct {
++	RecipientUserID int64
++	Column2         int32
++}
++
++// Per-recipient absolute rate cap: how many total emails to this
++// recipient in the last $2 minutes?
++func (q *Queries) CountEmailsForRecipientSince(ctx context.Context, db DBTX, arg CountEmailsForRecipientSinceParams) (int64, error) {
++	row := db.QueryRow(ctx, countEmailsForRecipientSince, arg.RecipientUserID, arg.Column2)
++	var count int64
++	err := row.Scan(&count)
++	return count, err
++}
++
++const countEmailsForRecipientThreadSince = `-- name: CountEmailsForRecipientThreadSince :one
++SELECT count(*) FROM notification_email_log
++WHERE recipient_user_id = $1
++  AND thread_kind = $2
++  AND thread_id = $3
++  AND sent_at > now() - make_interval(mins => $4::int)
++`
++
++type CountEmailsForRecipientThreadSinceParams struct {
++	RecipientUserID int64
++	ThreadKind      NullNotificationThreadKind
++	ThreadID        pgtype.Int8
++	Column4         int32
++}
++
++// Storm dampener probe: how many emails for this thread did we
++// send to this recipient in the last $4 minutes? Caller compares
++// to the cap.
++func (q *Queries) CountEmailsForRecipientThreadSince(ctx context.Context, db DBTX, arg CountEmailsForRecipientThreadSinceParams) (int64, error) {
++	row := db.QueryRow(ctx, countEmailsForRecipientThreadSince,
++		arg.RecipientUserID,
++		arg.ThreadKind,
++		arg.ThreadID,
++		arg.Column4,
++	)
++	var count int64
++	err := row.Scan(&count)
++	return count, err
++}
++
++const insertEmailLog = `-- name: InsertEmailLog :exec
++
++INSERT INTO notification_email_log
++    (recipient_user_id, notification_id, thread_kind, thread_id, message_id)
++VALUES ($1, $2, $3, $4, $5)
++`
++
++type InsertEmailLogParams struct {
++	RecipientUserID int64
++	NotificationID  pgtype.Int8
++	ThreadKind      NullNotificationThreadKind
++	ThreadID        pgtype.Int8
++	MessageID       pgtype.Text
++}
++
++// ─── notification_email_log ────────────────────────────────────────
++// Records an email send. Caller decides what to bind for thread_id
++// (NULL for thread-less notifications). MessageID is the SMTP /
++// transactional-provider message id when available; empty when
++// the sender doesn't surface one.
++func (q *Queries) InsertEmailLog(ctx context.Context, db DBTX, arg InsertEmailLogParams) error {
++	_, err := db.Exec(ctx, insertEmailLog,
++		arg.RecipientUserID,
++		arg.NotificationID,
++		arg.ThreadKind,
++		arg.ThreadID,
++		arg.MessageID,
++	)
++	return err
++}

internal/notif/sqlc/models.goadded

1404 lines changed — click to load

 +// Code generated by sqlc. DO NOT EDIT.
 +// versions:
 +//   sqlc v1.31.1
++
 +package notifdb
++
 +import (
 +	"database/sql/driver"
 +	"fmt"
 +	"net/netip"
++
 +	"github.com/jackc/pgx/v5/pgtype"
 +)
++
 +type CheckConclusion string
++
 +const (
 +	CheckConclusionSuccess        CheckConclusion = "success"
 +	CheckConclusionFailure        CheckConclusion = "failure"
 +	CheckConclusionNeutral        CheckConclusion = "neutral"
 +	CheckConclusionCancelled      CheckConclusion = "cancelled"
 +	CheckConclusionSkipped        CheckConclusion = "skipped"
 +	CheckConclusionTimedOut       CheckConclusion = "timed_out"
 +	CheckConclusionActionRequired CheckConclusion = "action_required"
 +	CheckConclusionStale          CheckConclusion = "stale"
 +)
++
 +func (e *CheckConclusion) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = CheckConclusion(s)
 +	case string:
 +		*e = CheckConclusion(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for CheckConclusion: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullCheckConclusion struct {
 +	CheckConclusion CheckConclusion
 +	Valid           bool // Valid is true if CheckConclusion is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullCheckConclusion) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.CheckConclusion, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.CheckConclusion.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullCheckConclusion) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.CheckConclusion), nil
 +}
++
 +type CheckStatus string
++
 +const (
 +	CheckStatusQueued     CheckStatus = "queued"
 +	CheckStatusInProgress CheckStatus = "in_progress"
 +	CheckStatusCompleted  CheckStatus = "completed"
 +	CheckStatusPending    CheckStatus = "pending"
 +)
++
 +func (e *CheckStatus) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = CheckStatus(s)
 +	case string:
 +		*e = CheckStatus(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for CheckStatus: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullCheckStatus struct {
 +	CheckStatus CheckStatus
 +	Valid       bool // Valid is true if CheckStatus is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullCheckStatus) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.CheckStatus, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.CheckStatus.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullCheckStatus) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.CheckStatus), nil
 +}
++
 +type CollabRole string
++
 +const (
 +	CollabRoleRead     CollabRole = "read"
 +	CollabRoleTriage   CollabRole = "triage"
 +	CollabRoleWrite    CollabRole = "write"
 +	CollabRoleMaintain CollabRole = "maintain"
 +	CollabRoleAdmin    CollabRole = "admin"
 +)
++
 +func (e *CollabRole) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = CollabRole(s)
 +	case string:
 +		*e = CollabRole(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for CollabRole: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullCollabRole struct {
 +	CollabRole CollabRole
 +	Valid      bool // Valid is true if CollabRole is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullCollabRole) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.CollabRole, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.CollabRole.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullCollabRole) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.CollabRole), nil
 +}
++
 +type IssueKind string
++
 +const (
 +	IssueKindIssue IssueKind = "issue"
 +	IssueKindPr    IssueKind = "pr"
 +)
++
 +func (e *IssueKind) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = IssueKind(s)
 +	case string:
 +		*e = IssueKind(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for IssueKind: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullIssueKind struct {
 +	IssueKind IssueKind
 +	Valid     bool // Valid is true if IssueKind is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullIssueKind) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.IssueKind, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.IssueKind.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullIssueKind) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.IssueKind), nil
 +}
++
 +type IssueRefSource string
++
 +const (
 +	IssueRefSourceCommentBody   IssueRefSource = "comment_body"
 +	IssueRefSourceIssueBody     IssueRefSource = "issue_body"
 +	IssueRefSourceCommitMessage IssueRefSource = "commit_message"
 +)
++
 +func (e *IssueRefSource) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = IssueRefSource(s)
 +	case string:
 +		*e = IssueRefSource(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for IssueRefSource: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullIssueRefSource struct {
 +	IssueRefSource IssueRefSource
 +	Valid          bool // Valid is true if IssueRefSource is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullIssueRefSource) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.IssueRefSource, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.IssueRefSource.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullIssueRefSource) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.IssueRefSource), nil
 +}
++
 +type IssueState string
++
 +const (
 +	IssueStateOpen   IssueState = "open"
 +	IssueStateClosed IssueState = "closed"
 +)
++
 +func (e *IssueState) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = IssueState(s)
 +	case string:
 +		*e = IssueState(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for IssueState: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullIssueState struct {
 +	IssueState IssueState
 +	Valid      bool // Valid is true if IssueState is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullIssueState) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.IssueState, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.IssueState.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullIssueState) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.IssueState), nil
 +}
++
 +type IssueStateReason string
++
 +const (
 +	IssueStateReasonCompleted  IssueStateReason = "completed"
 +	IssueStateReasonNotPlanned IssueStateReason = "not_planned"
 +	IssueStateReasonReopened   IssueStateReason = "reopened"
 +	IssueStateReasonDuplicate  IssueStateReason = "duplicate"
 +)
++
 +func (e *IssueStateReason) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = IssueStateReason(s)
 +	case string:
 +		*e = IssueStateReason(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for IssueStateReason: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullIssueStateReason struct {
 +	IssueStateReason IssueStateReason
 +	Valid            bool // Valid is true if IssueStateReason is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullIssueStateReason) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.IssueStateReason, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.IssueStateReason.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullIssueStateReason) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.IssueStateReason), nil
 +}
++
 +type MilestoneState string
++
 +const (
 +	MilestoneStateOpen   MilestoneState = "open"
 +	MilestoneStateClosed MilestoneState = "closed"
 +)
++
 +func (e *MilestoneState) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = MilestoneState(s)
 +	case string:
 +		*e = MilestoneState(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for MilestoneState: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullMilestoneState struct {
 +	MilestoneState MilestoneState
 +	Valid          bool // Valid is true if MilestoneState is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullMilestoneState) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.MilestoneState, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.MilestoneState.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullMilestoneState) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.MilestoneState), nil
 +}
++
 +type NotificationThreadKind string
++
 +const (
 +	NotificationThreadKindIssue NotificationThreadKind = "issue"
 +	NotificationThreadKindPr    NotificationThreadKind = "pr"
 +)
++
 +func (e *NotificationThreadKind) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = NotificationThreadKind(s)
 +	case string:
 +		*e = NotificationThreadKind(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for NotificationThreadKind: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullNotificationThreadKind struct {
 +	NotificationThreadKind NotificationThreadKind
 +	Valid                  bool // Valid is true if NotificationThreadKind is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullNotificationThreadKind) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.NotificationThreadKind, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.NotificationThreadKind.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullNotificationThreadKind) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.NotificationThreadKind), nil
 +}
++
 +type PrFileStatus string
++
 +const (
 +	PrFileStatusAdded    PrFileStatus = "added"
 +	PrFileStatusModified PrFileStatus = "modified"
 +	PrFileStatusDeleted  PrFileStatus = "deleted"
 +	PrFileStatusRenamed  PrFileStatus = "renamed"
 +	PrFileStatusCopied   PrFileStatus = "copied"
 +)
++
 +func (e *PrFileStatus) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = PrFileStatus(s)
 +	case string:
 +		*e = PrFileStatus(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for PrFileStatus: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullPrFileStatus struct {
 +	PrFileStatus PrFileStatus
 +	Valid        bool // Valid is true if PrFileStatus is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullPrFileStatus) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.PrFileStatus, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.PrFileStatus.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullPrFileStatus) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.PrFileStatus), nil
 +}
++
 +type PrMergeMethod string
++
 +const (
 +	PrMergeMethodMerge  PrMergeMethod = "merge"
 +	PrMergeMethodSquash PrMergeMethod = "squash"
 +	PrMergeMethodRebase PrMergeMethod = "rebase"
 +)
++
 +func (e *PrMergeMethod) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = PrMergeMethod(s)
 +	case string:
 +		*e = PrMergeMethod(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for PrMergeMethod: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullPrMergeMethod struct {
 +	PrMergeMethod PrMergeMethod
 +	Valid         bool // Valid is true if PrMergeMethod is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullPrMergeMethod) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.PrMergeMethod, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.PrMergeMethod.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullPrMergeMethod) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.PrMergeMethod), nil
 +}
++
 +type PrMergeableState string
++
 +const (
 +	PrMergeableStateUnknown PrMergeableState = "unknown"
 +	PrMergeableStateClean   PrMergeableState = "clean"
 +	PrMergeableStateDirty   PrMergeableState = "dirty"
 +	PrMergeableStateBlocked PrMergeableState = "blocked"
 +	PrMergeableStateBehind  PrMergeableState = "behind"
 +)
++
 +func (e *PrMergeableState) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = PrMergeableState(s)
 +	case string:
 +		*e = PrMergeableState(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for PrMergeableState: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullPrMergeableState struct {
 +	PrMergeableState PrMergeableState
 +	Valid            bool // Valid is true if PrMergeableState is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullPrMergeableState) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.PrMergeableState, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.PrMergeableState.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullPrMergeableState) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.PrMergeableState), nil
 +}
++
 +type PrReviewSide string
++
 +const (
 +	PrReviewSideLeft  PrReviewSide = "left"
 +	PrReviewSideRight PrReviewSide = "right"
 +)
++
 +func (e *PrReviewSide) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = PrReviewSide(s)
 +	case string:
 +		*e = PrReviewSide(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for PrReviewSide: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullPrReviewSide struct {
 +	PrReviewSide PrReviewSide
 +	Valid        bool // Valid is true if PrReviewSide is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullPrReviewSide) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.PrReviewSide, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.PrReviewSide.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullPrReviewSide) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.PrReviewSide), nil
 +}
++
 +type PrReviewState string
++
 +const (
 +	PrReviewStateComment        PrReviewState = "comment"
 +	PrReviewStateApprove        PrReviewState = "approve"
 +	PrReviewStateRequestChanges PrReviewState = "request_changes"
 +)
++
 +func (e *PrReviewState) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = PrReviewState(s)
 +	case string:
 +		*e = PrReviewState(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for PrReviewState: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullPrReviewState struct {
 +	PrReviewState PrReviewState
 +	Valid         bool // Valid is true if PrReviewState is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullPrReviewState) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.PrReviewState, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.PrReviewState.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullPrReviewState) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.PrReviewState), nil
 +}
++
 +type RepoInitStatus string
++
 +const (
 +	RepoInitStatusInitialized RepoInitStatus = "initialized"
 +	RepoInitStatusInitPending RepoInitStatus = "init_pending"
 +	RepoInitStatusInitFailed  RepoInitStatus = "init_failed"
 +)
++
 +func (e *RepoInitStatus) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = RepoInitStatus(s)
 +	case string:
 +		*e = RepoInitStatus(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for RepoInitStatus: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullRepoInitStatus struct {
 +	RepoInitStatus RepoInitStatus
 +	Valid          bool // Valid is true if RepoInitStatus is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullRepoInitStatus) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.RepoInitStatus, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.RepoInitStatus.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullRepoInitStatus) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.RepoInitStatus), nil
 +}
++
 +type RepoVisibility string
++
 +const (
 +	RepoVisibilityPublic  RepoVisibility = "public"
 +	RepoVisibilityPrivate RepoVisibility = "private"
 +)
++
 +func (e *RepoVisibility) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = RepoVisibility(s)
 +	case string:
 +		*e = RepoVisibility(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for RepoVisibility: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullRepoVisibility struct {
 +	RepoVisibility RepoVisibility
 +	Valid          bool // Valid is true if RepoVisibility is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullRepoVisibility) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.RepoVisibility, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.RepoVisibility.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullRepoVisibility) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.RepoVisibility), nil
 +}
++
 +type TransferPrincipalKind string
++
 +const (
 +	TransferPrincipalKindUser TransferPrincipalKind = "user"
 +	TransferPrincipalKindOrg  TransferPrincipalKind = "org"
 +)
++
 +func (e *TransferPrincipalKind) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = TransferPrincipalKind(s)
 +	case string:
 +		*e = TransferPrincipalKind(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for TransferPrincipalKind: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullTransferPrincipalKind struct {
 +	TransferPrincipalKind TransferPrincipalKind
 +	Valid                 bool // Valid is true if TransferPrincipalKind is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullTransferPrincipalKind) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.TransferPrincipalKind, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.TransferPrincipalKind.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullTransferPrincipalKind) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.TransferPrincipalKind), nil
 +}
++
 +type TransferStatus string
++
 +const (
 +	TransferStatusPending  TransferStatus = "pending"
 +	TransferStatusAccepted TransferStatus = "accepted"
 +	TransferStatusDeclined TransferStatus = "declined"
 +	TransferStatusCanceled TransferStatus = "canceled"
 +	TransferStatusExpired  TransferStatus = "expired"
 +)
++
 +func (e *TransferStatus) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = TransferStatus(s)
 +	case string:
 +		*e = TransferStatus(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for TransferStatus: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullTransferStatus struct {
 +	TransferStatus TransferStatus
 +	Valid          bool // Valid is true if TransferStatus is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullTransferStatus) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.TransferStatus, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.TransferStatus.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullTransferStatus) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.TransferStatus), nil
 +}
++
 +type WatchLevel string
++
 +const (
 +	WatchLevelAll           WatchLevel = "all"
 +	WatchLevelParticipating WatchLevel = "participating"
 +	WatchLevelIgnore        WatchLevel = "ignore"
 +)
++
 +func (e *WatchLevel) Scan(src interface{}) error {
 +	switch s := src.(type) {
 +	case []byte:
 +		*e = WatchLevel(s)
 +	case string:
 +		*e = WatchLevel(s)
 +	default:
 +		return fmt.Errorf("unsupported scan type for WatchLevel: %T", src)
 +	}
 +	return nil
 +}
++
 +type NullWatchLevel struct {
 +	WatchLevel WatchLevel
 +	Valid      bool // Valid is true if WatchLevel is not NULL
 +}
++
 +// Scan implements the Scanner interface.
 +func (ns *NullWatchLevel) Scan(value interface{}) error {
 +	if value == nil {
 +		ns.WatchLevel, ns.Valid = "", false
 +		return nil
 +	}
 +	ns.Valid = true
 +	return ns.WatchLevel.Scan(value)
 +}
++
 +// Value implements the driver Valuer interface.
 +func (ns NullWatchLevel) Value() (driver.Value, error) {
 +	if !ns.Valid {
 +		return nil, nil
 +	}
 +	return string(ns.WatchLevel), nil
 +}
++
 +type AuthAuditLog struct {
 +	ID         int64
 +	ActorID    pgtype.Int8
 +	Action     string
 +	TargetType string
 +	TargetID   pgtype.Int8
 +	Meta       []byte
 +	CreatedAt  pgtype.Timestamptz
 +}
++
 +type AuthThrottle struct {
 +	ID              int64
 +	Scope           string
 +	Identifier      string
 +	Hits            int32
 +	WindowStartedAt pgtype.Timestamptz
 +}
++
 +type BranchProtectionRule struct {
 +	ID                             int64
 +	RepoID                         int64
 +	Pattern                        string
 +	PreventForcePush               bool
 +	PreventDeletion                bool
 +	RequirePrForPush               bool
 +	AllowedPusherUserIds           []int64
 +	RequireSignedCommits           bool
 +	StatusChecksRequired           []string
 +	CreatedAt                      pgtype.Timestamptz
 +	UpdatedAt                      pgtype.Timestamptz
 +	CreatedByUserID                pgtype.Int8
 +	RequiredReviewCount            int32
 +	DismissStaleReviewsOnPush      bool
 +	RequireCodeOwnerReview         bool
 +	DismissStaleStatusChecksOnPush bool
 +}
++
 +type CheckRun struct {
 +	ID          int64
 +	SuiteID     int64
 +	RepoID      int64
 +	HeadSha     string
 +	Name        string
 +	Status      CheckStatus
 +	Conclusion  NullCheckConclusion
 +	StartedAt   pgtype.Timestamptz
 +	CompletedAt pgtype.Timestamptz
 +	DetailsUrl  string
 +	Output      []byte
 +	ExternalID  pgtype.Text
 +	CreatedAt   pgtype.Timestamptz
 +	UpdatedAt   pgtype.Timestamptz
 +}
++
 +type CheckSuite struct {
 +	ID         int64
 +	RepoID     int64
 +	HeadSha    string
 +	AppSlug    string
 +	Status     CheckStatus
 +	Conclusion NullCheckConclusion
 +	CreatedAt  pgtype.Timestamptz
 +	UpdatedAt  pgtype.Timestamptz
 +}
++
 +type CodeSearchContent struct {
 +	RepoID      int64
 +	RefName     string
 +	Path        string
 +	ContentTsv  interface{}
 +	ContentTrgm string
 +}
++
 +type CodeSearchPath struct {
 +	RepoID  int64
 +	RefName string
 +	Path    string
 +	Tsv     interface{}
 +}
++
 +type DomainEvent struct {
 +	ID          int64
 +	ActorUserID pgtype.Int8
 +	Kind        string
 +	RepoID      pgtype.Int8
 +	SourceKind  string
 +	SourceID    int64
 +	Public      bool
 +	Payload     []byte
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type DomainEventsProcessed struct {
 +	Consumer    string
 +	LastEventID int64
 +	UpdatedAt   pgtype.Timestamptz
 +}
++
 +type EmailVerification struct {
 +	ID          int64
 +	UserEmailID int64
 +	TokenHash   []byte
 +	ExpiresAt   pgtype.Timestamptz
 +	UsedAt      pgtype.Timestamptz
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type Issue struct {
 +	ID                int64
 +	RepoID            int64
 +	Number            int64
 +	Kind              IssueKind
 +	Title             string
 +	Body              string
 +	BodyHtmlCached    pgtype.Text
 +	MdPipelineVersion int32
 +	AuthorUserID      pgtype.Int8
 +	State             IssueState
 +	StateReason       NullIssueStateReason
 +	Locked            bool
 +	LockReason        pgtype.Text
 +	MilestoneID       pgtype.Int8
 +	CreatedAt         pgtype.Timestamptz
 +	UpdatedAt         pgtype.Timestamptz
 +	EditedAt          pgtype.Timestamptz
 +	ClosedAt          pgtype.Timestamptz
 +	ClosedByUserID    pgtype.Int8
 +}
++
 +type IssueAssignee struct {
 +	IssueID          int64
 +	UserID           int64
 +	AssignedAt       pgtype.Timestamptz
 +	AssignedByUserID pgtype.Int8
 +}
++
 +type IssueComment struct {
 +	ID                int64
 +	IssueID           int64
 +	AuthorUserID      pgtype.Int8
 +	Body              string
 +	BodyHtmlCached    pgtype.Text
 +	MdPipelineVersion int32
 +	CreatedAt         pgtype.Timestamptz
 +	UpdatedAt         pgtype.Timestamptz
 +	EditedAt          pgtype.Timestamptz
 +}
++
 +type IssueEvent struct {
 +	ID          int64
 +	IssueID     int64
 +	ActorUserID pgtype.Int8
 +	Kind        string
 +	Meta        []byte
 +	RefTargetID pgtype.Int8
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type IssueLabel struct {
 +	IssueID         int64
 +	LabelID         int64
 +	AppliedAt       pgtype.Timestamptz
 +	AppliedByUserID pgtype.Int8
 +}
++
 +type IssueReference struct {
 +	ID             int64
 +	SourceIssueID  pgtype.Int8
 +	TargetIssueID  int64
 +	SourceKind     IssueRefSource
 +	SourceObjectID pgtype.Int8
 +	CreatedAt      pgtype.Timestamptz
 +}
++
 +type IssuesSearch struct {
 +	IssueID      int64
 +	RepoID       int64
 +	Kind         IssueKind
 +	State        IssueState
 +	AuthorUserID pgtype.Int8
 +	Tsv          interface{}
 +}
++
 +type Job struct {
 +	ID          int64
 +	Kind        string
 +	Payload     []byte
 +	RunAt       pgtype.Timestamptz
 +	Attempts    int32
 +	MaxAttempts int32
 +	LastError   pgtype.Text
 +	LockedBy    pgtype.Text
 +	LockedAt    pgtype.Timestamptz
 +	CompletedAt pgtype.Timestamptz
 +	FailedAt    pgtype.Timestamptz
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type Label struct {
 +	ID          int64
 +	RepoID      int64
 +	Name        string
 +	Color       string
 +	Description string
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type Meta struct {
 +	Key       string
 +	Value     []byte
 +	UpdatedAt pgtype.Timestamptz
 +}
++
 +type Milestone struct {
 +	ID          int64
 +	RepoID      int64
 +	Title       string
 +	Description string
 +	State       MilestoneState
 +	DueOn       pgtype.Timestamptz
 +	CreatedAt   pgtype.Timestamptz
 +	ClosedAt    pgtype.Timestamptz
 +}
++
 +type Notification struct {
 +	ID              int64
 +	RecipientUserID int64
 +	Kind            string
 +	Reason          string
 +	RepoID          pgtype.Int8
 +	ThreadKind      NullNotificationThreadKind
 +	ThreadID        pgtype.Int8
 +	SourceEventID   pgtype.Int8
 +	Unread          bool
 +	LastEventAt     pgtype.Timestamptz
 +	LastActorUserID pgtype.Int8
 +	Summary         []byte
 +	CreatedAt       pgtype.Timestamptz
 +	UpdatedAt       pgtype.Timestamptz
 +}
++
 +type NotificationEmailLog struct {
 +	ID              int64
 +	RecipientUserID int64
 +	NotificationID  pgtype.Int8
 +	ThreadKind      NullNotificationThreadKind
 +	ThreadID        pgtype.Int8
 +	SentAt          pgtype.Timestamptz
 +	MessageID       pgtype.Text
 +}
++
 +type NotificationThread struct {
 +	RecipientUserID int64
 +	ThreadKind      NotificationThreadKind
 +	ThreadID        int64
 +	Subscribed      bool
 +	Reason          string
 +	UpdatedAt       pgtype.Timestamptz
 +}
++
 +type PasswordReset struct {
 +	ID        int64
 +	UserID    int64
 +	TokenHash []byte
 +	ExpiresAt pgtype.Timestamptz
 +	UsedAt    pgtype.Timestamptz
 +	CreatedAt pgtype.Timestamptz
 +}
++
 +type PrReview struct {
 +	ID                int64
 +	PrIssueID         int64
 +	AuthorUserID      pgtype.Int8
 +	State             PrReviewState
 +	Body              string
 +	BodyHtmlCached    pgtype.Text
 +	SubmittedAt       pgtype.Timestamptz
 +	DismissedAt       pgtype.Timestamptz
 +	DismissedByUserID pgtype.Int8
 +	DismissalReason   string
 +}
++
 +type PrReviewComment struct {
 +	ID                int64
 +	PrIssueID         int64
 +	ReviewID          pgtype.Int8
 +	AuthorUserID      pgtype.Int8
 +	FilePath          string
 +	Side              PrReviewSide
 +	OriginalCommitSha string
 +	OriginalLine      int32
 +	OriginalPosition  int32
 +	CurrentPosition   pgtype.Int4
 +	Body              string
 +	BodyHtmlCached    pgtype.Text
 +	InReplyToID       pgtype.Int8
 +	Pending           bool
 +	ResolvedAt        pgtype.Timestamptz
 +	ResolvedByUserID  pgtype.Int8
 +	CreatedAt         pgtype.Timestamptz
 +	UpdatedAt         pgtype.Timestamptz
 +	EditedAt          pgtype.Timestamptz
 +}
++
 +type PrReviewRequest struct {
 +	ID                  int64
 +	PrIssueID           int64
 +	RequestedUserID     pgtype.Int8
 +	RequestedTeamID     pgtype.Int8
 +	RequestedByUserID   pgtype.Int8
 +	RequestedAt         pgtype.Timestamptz
 +	DismissedAt         pgtype.Timestamptz
 +	SatisfiedByReviewID pgtype.Int8
 +}
++
 +type PullRequest struct {
 +	IssueID            int64
 +	BaseRef            string
 +	HeadRef            string
 +	HeadRepoID         int64
 +	BaseOid            string
 +	HeadOid            string
 +	Draft              bool
 +	Mergeable          pgtype.Bool
 +	MergeableState     PrMergeableState
 +	MergeCommitSha     pgtype.Text
 +	MergedAt           pgtype.Timestamptz
 +	MergedByUserID     pgtype.Int8
 +	MergeMethod        NullPrMergeMethod
 +	BaseOidAtMerge     pgtype.Text
 +	HeadOidAtMerge     pgtype.Text
 +	LastSynchronizedAt pgtype.Timestamptz
 +}
++
 +type PullRequestCommit struct {
 +	PrID           int64
 +	Sha            string
 +	Position       int32
 +	AuthorName     string
 +	AuthorEmail    string
 +	CommitterName  string
 +	CommitterEmail string
 +	Subject        string
 +	Body           string
 +	AuthoredAt     pgtype.Timestamptz
 +	CommittedAt    pgtype.Timestamptz
 +}
++
 +type PullRequestFile struct {
 +	PrID      int64
 +	Path      string
 +	Status    PrFileStatus
 +	OldPath   pgtype.Text
 +	Additions int32
 +	Deletions int32
 +	Changes   int32
 +}
++
 +type PushEvent struct {
 +	ID           int64
 +	RepoID       int64
 +	PusherUserID pgtype.Int8
 +	BeforeSha    string
 +	AfterSha     string
 +	Ref          string
 +	Protocol     string
 +	RequestID    string
 +	ProcessedAt  pgtype.Timestamptz
 +	CreatedAt    pgtype.Timestamptz
 +}
++
 +type Repo struct {
 +	ID                 int64
 +	OwnerUserID        pgtype.Int8
 +	OwnerOrgID         pgtype.Int8
 +	Name               string
 +	Description        string
 +	Visibility         RepoVisibility
 +	DefaultBranch      string
 +	IsArchived         bool
 +	ArchivedAt         pgtype.Timestamptz
 +	DeletedAt          pgtype.Timestamptz
 +	DiskUsedBytes      int64
 +	ForkOfRepoID       pgtype.Int8
 +	LicenseKey         pgtype.Text
 +	PrimaryLanguage    pgtype.Text
 +	HasIssues          bool
 +	HasPulls           bool
 +	CreatedAt          pgtype.Timestamptz
 +	UpdatedAt          pgtype.Timestamptz
 +	DefaultBranchOid   pgtype.Text
 +	AllowSquashMerge   bool
 +	AllowRebaseMerge   bool
 +	AllowMergeCommit   bool
 +	DefaultMergeMethod PrMergeMethod
 +	StarCount          int64
 +	WatcherCount       int64
 +	ForkCount          int64
 +	InitStatus         RepoInitStatus
 +	LastIndexedOid     pgtype.Text
 +}
++
 +type RepoCollaborator struct {
 +	RepoID        int64
 +	UserID        int64
 +	Role          CollabRole
 +	AddedAt       pgtype.Timestamptz
 +	AddedByUserID pgtype.Int8
 +}
++
 +type RepoIssueCounter struct {
 +	RepoID     int64
 +	NextNumber int64
 +}
++
 +type RepoRedirect struct {
 +	OldOwnerUserID pgtype.Int8
 +	OldOwnerOrgID  pgtype.Int8
 +	OldName        string
 +	RepoID         int64
 +	RedirectedAt   pgtype.Timestamptz
 +}
++
 +type RepoTransferRequest struct {
 +	ID              int64
 +	RepoID          int64
 +	FromUserID      int64
 +	ToPrincipalKind TransferPrincipalKind
 +	ToPrincipalID   int64
 +	CreatedBy       int64
 +	CreatedAt       pgtype.Timestamptz
 +	ExpiresAt       pgtype.Timestamptz
 +	Status          TransferStatus
 +	AcceptedAt      pgtype.Timestamptz
 +	DeclinedAt      pgtype.Timestamptz
 +	CanceledAt      pgtype.Timestamptz
 +}
++
 +type ReposSearch struct {
 +	RepoID int64
 +	Tsv    interface{}
 +}
++
 +type Star struct {
 +	UserID    int64
 +	RepoID    int64
 +	StarredAt pgtype.Timestamptz
 +}
++
 +type User struct {
 +	ID                int64
 +	Username          string
 +	DisplayName       string
 +	PrimaryEmailID    pgtype.Int8
 +	PasswordHash      string
 +	PasswordAlgo      string
 +	PasswordUpdatedAt pgtype.Timestamptz
 +	EmailVerified     bool
 +	LastLoginAt       pgtype.Timestamptz
 +	SuspendedAt       pgtype.Timestamptz
 +	SuspendedReason   pgtype.Text
 +	DeletedAt         pgtype.Timestamptz
 +	CreatedAt         pgtype.Timestamptz
 +	UpdatedAt         pgtype.Timestamptz
 +	Bio               string
 +	Location          string
 +	Website           string
 +	Company           string
 +	Pronouns          string
 +	AvatarObjectKey   pgtype.Text
 +	Theme             string
 +	SessionEpoch      int32
 +}
++
 +type UserEmail struct {
 +	ID                    int64
 +	UserID                int64
 +	Email                 string
 +	IsPrimary             bool
 +	Verified              bool
 +	VerificationTokenHash []byte
 +	VerificationSentAt    pgtype.Timestamptz
 +	VerifiedAt            pgtype.Timestamptz
 +	CreatedAt             pgtype.Timestamptz
 +}
++
 +type UserNotificationPref struct {
 +	UserID    int64
 +	Key       string
 +	Value     []byte
 +	UpdatedAt pgtype.Timestamptz
 +}
++
 +type UserRecoveryCode struct {
 +	ID          int64
 +	UserID      int64
 +	CodeHash    []byte
 +	UsedAt      pgtype.Timestamptz
 +	GeneratedAt pgtype.Timestamptz
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type UserSshKey struct {
 +	ID                int64
 +	UserID            int64
 +	Title             string
 +	FingerprintSha256 string
 +	KeyType           string
 +	KeyBits           int32
 +	PublicKey         string
 +	LastUsedAt        pgtype.Timestamptz
 +	LastUsedIp        *netip.Addr
 +	CreatedAt         pgtype.Timestamptz
 +}
++
 +type UserToken struct {
 +	ID          int64
 +	UserID      int64
 +	Name        string
 +	TokenHash   []byte
 +	TokenPrefix string
 +	Scopes      []string
 +	ExpiresAt   pgtype.Timestamptz
 +	LastUsedAt  pgtype.Timestamptz
 +	LastUsedIp  *netip.Addr
 +	RevokedAt   pgtype.Timestamptz
 +	CreatedAt   pgtype.Timestamptz
 +}
++
 +type UserTotp struct {
 +	ID              int64
 +	UserID          int64
 +	SecretEncrypted []byte
 +	SecretNonce     []byte
 +	ConfirmedAt     pgtype.Timestamptz
 +	LastUsedCounter int64
 +	CreatedAt       pgtype.Timestamptz
 +	UpdatedAt       pgtype.Timestamptz
 +}
++
 +type UsernameRedirect struct {
 +	OldUsername string
 +	UserID      int64
 +	ChangedAt   pgtype.Timestamptz
 +}
++
 +type UsersSearch struct {
 +	UserID int64
 +	Tsv    interface{}
 +}
++
 +type Watch struct {
 +	UserID    int64
 +	RepoID    int64
 +	Level     WatchLevel
 +	UpdatedAt pgtype.Timestamptz
 +}
++
 +type WebhookEventsPending struct {
 +	ID        int64
 +	RepoID    int64
 +	EventKind string
 +	Payload   []byte
 +	CreatedAt pgtype.Timestamptz
 +}

internal/notif/sqlc/notifications.sql.goadded

++// Code generated by sqlc. DO NOT EDIT.
++// versions:
++//   sqlc v1.31.1
++// source: notifications.sql
++
++package notifdb
++
++import (
++	"context"
++
++	"github.com/jackc/pgx/v5/pgtype"
++)
++
++const countNotificationsForRecipient = `-- name: CountNotificationsForRecipient :one
++SELECT count(*) FROM notifications
++WHERE recipient_user_id = $1
++  AND ($2::boolean = false OR unread = true)
++`
++
++type CountNotificationsForRecipientParams struct {
++	RecipientUserID int64
++	Column2         bool
++}
++
++func (q *Queries) CountNotificationsForRecipient(ctx context.Context, db DBTX, arg CountNotificationsForRecipientParams) (int64, error) {
++	row := db.QueryRow(ctx, countNotificationsForRecipient, arg.RecipientUserID, arg.Column2)
++	var count int64
++	err := row.Scan(&count)
++	return count, err
++}
++
++const countUnreadForRecipient = `-- name: CountUnreadForRecipient :one
++SELECT count(*) FROM notifications
++WHERE recipient_user_id = $1 AND unread = true
++`
++
++func (q *Queries) CountUnreadForRecipient(ctx context.Context, db DBTX, recipientUserID int64) (int64, error) {
++	row := db.QueryRow(ctx, countUnreadForRecipient, recipientUserID)
++	var count int64
++	err := row.Scan(&count)
++	return count, err
++}
++
++const getNotification = `-- name: GetNotification :one
++SELECT id, recipient_user_id, kind, reason, repo_id,
++       thread_kind, thread_id, source_event_id, unread,
++       last_event_at, last_actor_user_id, summary, created_at, updated_at
++FROM notifications WHERE id = $1
++`
++
++func (q *Queries) GetNotification(ctx context.Context, db DBTX, id int64) (Notification, error) {
++	row := db.QueryRow(ctx, getNotification, id)
++	var i Notification
++	err := row.Scan(
++		&i.ID,
++		&i.RecipientUserID,
++		&i.Kind,
++		&i.Reason,
++		&i.RepoID,
++		&i.ThreadKind,
++		&i.ThreadID,
++		&i.SourceEventID,
++		&i.Unread,
++		&i.LastEventAt,
++		&i.LastActorUserID,
++		&i.Summary,
++		&i.CreatedAt,
++		&i.UpdatedAt,
++	)
++	return i, err
++}
++
++const insertThreadlessNotification = `-- name: InsertThreadlessNotification :one
++INSERT INTO notifications (
++    recipient_user_id, kind, reason, repo_id,
++    source_event_id, last_actor_user_id
++) VALUES ($1, $2, $3, $4, $5, $6)
++RETURNING id, recipient_user_id, kind, reason, repo_id,
++          thread_kind, thread_id, source_event_id, unread,
++          last_event_at, last_actor_user_id, summary, created_at, updated_at
++`
++
++type InsertThreadlessNotificationParams struct {
++	RecipientUserID int64
++	Kind            string
++	Reason          string
++	RepoID          pgtype.Int8
++	SourceEventID   pgtype.Int8
++	LastActorUserID pgtype.Int8
++}
++
++// For events with no thread (e.g. repo-admin lifecycle: archived).
++// These don't coalesce; each fires its own row. Used sparingly.
++func (q *Queries) InsertThreadlessNotification(ctx context.Context, db DBTX, arg InsertThreadlessNotificationParams) (Notification, error) {
++	row := db.QueryRow(ctx, insertThreadlessNotification,
++		arg.RecipientUserID,
++		arg.Kind,
++		arg.Reason,
++		arg.RepoID,
++		arg.SourceEventID,
++		arg.LastActorUserID,
++	)
++	var i Notification
++	err := row.Scan(
++		&i.ID,
++		&i.RecipientUserID,
++		&i.Kind,
++		&i.Reason,
++		&i.RepoID,
++		&i.ThreadKind,
++		&i.ThreadID,
++		&i.SourceEventID,
++		&i.Unread,
++		&i.LastEventAt,
++		&i.LastActorUserID,
++		&i.Summary,
++		&i.CreatedAt,
++		&i.UpdatedAt,
++	)
++	return i, err
++}
++
++const listNotificationsForRecipient = `-- name: ListNotificationsForRecipient :many
++SELECT n.id, n.recipient_user_id, n.kind, n.reason, n.repo_id,
++       n.thread_kind, n.thread_id, n.source_event_id, n.unread,
++       n.last_event_at, n.last_actor_user_id, n.summary,
++       n.created_at, n.updated_at,
++       coalesce(u.username, '') AS actor_username,
++       coalesce(r.name, '') AS repo_name,
++       coalesce(ru.username, '') AS repo_owner_username,
++       coalesce(i.number, 0) AS thread_number,
++       coalesce(i.title, '') AS thread_title
++FROM notifications n
++LEFT JOIN users u  ON u.id = n.last_actor_user_id
++LEFT JOIN repos r  ON r.id = n.repo_id
++LEFT JOIN users ru ON ru.id = r.owner_user_id
++LEFT JOIN issues i ON i.id = n.thread_id
++WHERE n.recipient_user_id = $1
++  AND ($2::boolean = false OR n.unread = true)
++ORDER BY n.last_event_at DESC
++LIMIT $3 OFFSET $4
++`
++
++type ListNotificationsForRecipientParams struct {
++	RecipientUserID int64
++	Column2         bool
++	Limit           int32
++	Offset          int32
++}
++
++type ListNotificationsForRecipientRow struct {
++	ID                int64
++	RecipientUserID   int64
++	Kind              string
++	Reason            string
++	RepoID            pgtype.Int8
++	ThreadKind        NullNotificationThreadKind
++	ThreadID          pgtype.Int8
++	SourceEventID     pgtype.Int8
++	Unread            bool
++	LastEventAt       pgtype.Timestamptz
++	LastActorUserID   pgtype.Int8
++	Summary           []byte
++	CreatedAt         pgtype.Timestamptz
++	UpdatedAt         pgtype.Timestamptz
++	ActorUsername     string
++	RepoName          string
++	RepoOwnerUsername string
++	ThreadNumber      int64
++	ThreadTitle       string
++}
++
++// Inbox view, recency-sorted. `onlyUnread` toggles the inbox
++// filter ("Unread" tab vs "All").
++func (q *Queries) ListNotificationsForRecipient(ctx context.Context, db DBTX, arg ListNotificationsForRecipientParams) ([]ListNotificationsForRecipientRow, error) {
++	rows, err := db.Query(ctx, listNotificationsForRecipient,
++		arg.RecipientUserID,
++		arg.Column2,
++		arg.Limit,
++		arg.Offset,
++	)
++	if err != nil {
++		return nil, err
++	}
++	defer rows.Close()
++	items := []ListNotificationsForRecipientRow{}
++	for rows.Next() {
++		var i ListNotificationsForRecipientRow
++		if err := rows.Scan(
++			&i.ID,
++			&i.RecipientUserID,
++			&i.Kind,
++			&i.Reason,
++			&i.RepoID,
++			&i.ThreadKind,
++			&i.ThreadID,
++			&i.SourceEventID,
++			&i.Unread,
++			&i.LastEventAt,
++			&i.LastActorUserID,
++			&i.Summary,
++			&i.CreatedAt,
++			&i.UpdatedAt,
++			&i.ActorUsername,
++			&i.RepoName,
++			&i.RepoOwnerUsername,
++			&i.ThreadNumber,
++			&i.ThreadTitle,
++		); err != nil {
++			return nil, err
++		}
++		items = append(items, i)
++	}
++	if err := rows.Err(); err != nil {
++		return nil, err
++	}
++	return items, nil
++}
++
++const markAllReadForRecipient = `-- name: MarkAllReadForRecipient :exec
++UPDATE notifications SET unread = false, updated_at = now()
++WHERE recipient_user_id = $1 AND unread = true
++`
++
++// Bounded sweep: a single call doesn't try to update millions of
++// rows. Caller paginates via repeated calls when count > batch.
++func (q *Queries) MarkAllReadForRecipient(ctx context.Context, db DBTX, recipientUserID int64) error {
++	_, err := db.Exec(ctx, markAllReadForRecipient, recipientUserID)
++	return err
++}
++
++const setNotificationRead = `-- name: SetNotificationRead :exec
++UPDATE notifications SET unread = false, updated_at = now()
++WHERE id = $1 AND recipient_user_id = $2
++`
++
++type SetNotificationReadParams struct {
++	ID              int64
++	RecipientUserID int64
++}
++
++func (q *Queries) SetNotificationRead(ctx context.Context, db DBTX, arg SetNotificationReadParams) error {
++	_, err := db.Exec(ctx, setNotificationRead, arg.ID, arg.RecipientUserID)
++	return err
++}
++
++const setNotificationUnread = `-- name: SetNotificationUnread :exec
++UPDATE notifications SET unread = true, updated_at = now()
++WHERE id = $1 AND recipient_user_id = $2
++`
++
++type SetNotificationUnreadParams struct {
++	ID              int64
++	RecipientUserID int64
++}
++
++func (q *Queries) SetNotificationUnread(ctx context.Context, db DBTX, arg SetNotificationUnreadParams) error {
++	_, err := db.Exec(ctx, setNotificationUnread, arg.ID, arg.RecipientUserID)
++	return err
++}
++
++const upsertNotificationByThread = `-- name: UpsertNotificationByThread :one
++
++INSERT INTO notifications (
++    recipient_user_id, kind, reason, repo_id,
++    thread_kind, thread_id, source_event_id,
++    last_event_at, last_actor_user_id
++) VALUES (
++    $1, $2, $3, $4, $5, $6, $7, now(), $8
++)
++ON CONFLICT (recipient_user_id, thread_kind, thread_id) WHERE thread_id IS NOT NULL
++DO UPDATE SET
++    kind               = EXCLUDED.kind,
++    reason             = EXCLUDED.reason,
++    source_event_id    = EXCLUDED.source_event_id,
++    last_event_at      = now(),
++    last_actor_user_id = EXCLUDED.last_actor_user_id,
++    unread             = true,
++    updated_at         = now()
++RETURNING id, recipient_user_id, kind, reason, repo_id,
++          thread_kind, thread_id, source_event_id, unread,
++          last_event_at, last_actor_user_id, summary, created_at, updated_at
++`
++
++type UpsertNotificationByThreadParams struct {
++	RecipientUserID int64
++	Kind            string
++	Reason          string
++	RepoID          pgtype.Int8
++	ThreadKind      NullNotificationThreadKind
++	ThreadID        pgtype.Int8
++	SourceEventID   pgtype.Int8
++	LastActorUserID pgtype.Int8
++}
++
++// ─── notifications ─────────────────────────────────────────────────
++// Coalesce-or-insert: if a row exists for (recipient, thread), bump
++// last_event_at + last_actor + reason and re-flip unread=true so the
++// inbox surfaces it again. Otherwise insert a fresh row.
++//
++// Returns the resulting row (whether it was created or updated)
++// so the caller can chain an email-enqueue without a re-read.
++func (q *Queries) UpsertNotificationByThread(ctx context.Context, db DBTX, arg UpsertNotificationByThreadParams) (Notification, error) {
++	row := db.QueryRow(ctx, upsertNotificationByThread,
++		arg.RecipientUserID,
++		arg.Kind,
++		arg.Reason,
++		arg.RepoID,
++		arg.ThreadKind,
++		arg.ThreadID,
++		arg.SourceEventID,
++		arg.LastActorUserID,
++	)
++	var i Notification
++	err := row.Scan(
++		&i.ID,
++		&i.RecipientUserID,
++		&i.Kind,
++		&i.Reason,
++		&i.RepoID,
++		&i.ThreadKind,
++		&i.ThreadID,
++		&i.SourceEventID,
++		&i.Unread,
++		&i.LastEventAt,
++		&i.LastActorUserID,
++		&i.Summary,
++		&i.CreatedAt,
++		&i.UpdatedAt,
++	)
++	return i, err
++}

internal/notif/sqlc/querier.goadded

++// Code generated by sqlc. DO NOT EDIT.
++// versions:
++//   sqlc v1.31.1
++
++package notifdb
++
++import (
++	"context"
++)
++
++type Querier interface {
++	// Per-recipient absolute rate cap: how many total emails to this
++	// recipient in the last $2 minutes?
++	CountEmailsForRecipientSince(ctx context.Context, db DBTX, arg CountEmailsForRecipientSinceParams) (int64, error)
++	// Storm dampener probe: how many emails for this thread did we
++	// send to this recipient in the last $4 minutes? Caller compares
++	// to the cap.
++	CountEmailsForRecipientThreadSince(ctx context.Context, db DBTX, arg CountEmailsForRecipientThreadSinceParams) (int64, error)
++	CountNotificationsForRecipient(ctx context.Context, db DBTX, arg CountNotificationsForRecipientParams) (int64, error)
++	CountUnreadForRecipient(ctx context.Context, db DBTX, recipientUserID int64) (int64, error)
++	DeleteNotificationThread(ctx context.Context, db DBTX, arg DeleteNotificationThreadParams) error
++	// ─── domain_events_processed ──────────────────────────────────────
++	GetEventCursor(ctx context.Context, db DBTX, consumer string) (DomainEventsProcessed, error)
++	GetNotification(ctx context.Context, db DBTX, id int64) (Notification, error)
++	// ─── notification_threads ──────────────────────────────────────────
++	GetNotificationThread(ctx context.Context, db DBTX, arg GetNotificationThreadParams) (NotificationThread, error)
++	// ─── notification_email_log ────────────────────────────────────────
++	// Records an email send. Caller decides what to bind for thread_id
++	// (NULL for thread-less notifications). MessageID is the SMTP /
++	// transactional-provider message id when available; empty when
++	// the sender doesn't surface one.
++	InsertEmailLog(ctx context.Context, db DBTX, arg InsertEmailLogParams) error
++	// Auto-subscription path: only insert if the user has no explicit
++	// preference yet. Preserves user choices (e.g. an explicit
++	// `subscribed=false` from clicking "Unsubscribe").
++	InsertNotificationThreadIfAbsent(ctx context.Context, db DBTX, arg InsertNotificationThreadIfAbsentParams) error
++	// For events with no thread (e.g. repo-admin lifecycle: archived).
++	// These don't coalesce; each fires its own row. Used sparingly.
++	InsertThreadlessNotification(ctx context.Context, db DBTX, arg InsertThreadlessNotificationParams) (Notification, error)
++	// Inbox view, recency-sorted. `onlyUnread` toggles the inbox
++	// filter ("Unread" tab vs "All").
++	ListNotificationsForRecipient(ctx context.Context, db DBTX, arg ListNotificationsForRecipientParams) ([]ListNotificationsForRecipientRow, error)
++	// Fan-out helper: returns recipients who explicitly subscribed to a
++	// thread. The fan-out worker unions this with the per-repo `watches`
++	// result + author/assignee/reviewer rules.
++	ListSubscribersForThread(ctx context.Context, db DBTX, arg ListSubscribersForThreadParams) ([]ListSubscribersForThreadRow, error)
++	// The fan-out worker's read cursor. Bounded so a single tick
++	// doesn't try to drain a million-row backlog.
++	ListUnprocessedDomainEvents(ctx context.Context, db DBTX, arg ListUnprocessedDomainEventsParams) ([]DomainEvent, error)
++	// Bounded sweep: a single call doesn't try to update millions of
++	// rows. Caller paginates via repeated calls when count > batch.
++	MarkAllReadForRecipient(ctx context.Context, db DBTX, recipientUserID int64) error
++	// Always-write upsert so the worker doesn't have to special-case
++	// the missing-row branch (the migration seeds 'notify_fanout' at
++	// 0; future consumers like 'webhook_deliver' do the same on first
++	// run via this same call).
++	SetEventCursor(ctx context.Context, db DBTX, arg SetEventCursorParams) error
++	SetNotificationRead(ctx context.Context, db DBTX, arg SetNotificationReadParams) error
++	SetNotificationUnread(ctx context.Context, db DBTX, arg SetNotificationUnreadParams) error
++	// ─── notifications ─────────────────────────────────────────────────
++	// Coalesce-or-insert: if a row exists for (recipient, thread), bump
++	// last_event_at + last_actor + reason and re-flip unread=true so the
++	// inbox surfaces it again. Otherwise insert a fresh row.
++	//
++	// Returns the resulting row (whether it was created or updated)
++	// so the caller can chain an email-enqueue without a re-read.
++	UpsertNotificationByThread(ctx context.Context, db DBTX, arg UpsertNotificationByThreadParams) (Notification, error)
++	// Always-write upsert. Used by Subscribe / Unsubscribe / Ignore
++	// handlers and by the auto-subscription rules in the fan-out
++	// worker.
++	UpsertNotificationThread(ctx context.Context, db DBTX, arg UpsertNotificationThreadParams) error
++}
++
++var _ Querier = (*Queries)(nil)

internal/notif/sqlc/threads.sql.goadded

++// Code generated by sqlc. DO NOT EDIT.
++// versions:
++//   sqlc v1.31.1
++// source: threads.sql
++
++package notifdb
++
++import (
++	"context"
++)
++
++const deleteNotificationThread = `-- name: DeleteNotificationThread :exec
++DELETE FROM notification_threads
++WHERE recipient_user_id = $1 AND thread_kind = $2 AND thread_id = $3
++`
++
++type DeleteNotificationThreadParams struct {
++	RecipientUserID int64
++	ThreadKind      NotificationThreadKind
++	ThreadID        int64
++}
++
++func (q *Queries) DeleteNotificationThread(ctx context.Context, db DBTX, arg DeleteNotificationThreadParams) error {
++	_, err := db.Exec(ctx, deleteNotificationThread, arg.RecipientUserID, arg.ThreadKind, arg.ThreadID)
++	return err
++}
++
++const getNotificationThread = `-- name: GetNotificationThread :one
++
++SELECT recipient_user_id, thread_kind, thread_id, subscribed, reason, updated_at
++FROM notification_threads
++WHERE recipient_user_id = $1 AND thread_kind = $2 AND thread_id = $3
++`
++
++type GetNotificationThreadParams struct {
++	RecipientUserID int64
++	ThreadKind      NotificationThreadKind
++	ThreadID        int64
++}
++
++// ─── notification_threads ──────────────────────────────────────────
++func (q *Queries) GetNotificationThread(ctx context.Context, db DBTX, arg GetNotificationThreadParams) (NotificationThread, error) {
++	row := db.QueryRow(ctx, getNotificationThread, arg.RecipientUserID, arg.ThreadKind, arg.ThreadID)
++	var i NotificationThread
++	err := row.Scan(
++		&i.RecipientUserID,
++		&i.ThreadKind,
++		&i.ThreadID,
++		&i.Subscribed,
++		&i.Reason,
++		&i.UpdatedAt,
++	)
++	return i, err
++}
++
++const insertNotificationThreadIfAbsent = `-- name: InsertNotificationThreadIfAbsent :exec
++INSERT INTO notification_threads (recipient_user_id, thread_kind, thread_id, subscribed, reason)
++VALUES ($1, $2, $3, $4, $5)
++ON CONFLICT (recipient_user_id, thread_kind, thread_id) DO NOTHING
++`
++
++type InsertNotificationThreadIfAbsentParams struct {
++	RecipientUserID int64
++	ThreadKind      NotificationThreadKind
++	ThreadID        int64
++	Subscribed      bool
++	Reason          string
++}
++
++// Auto-subscription path: only insert if the user has no explicit
++// preference yet. Preserves user choices (e.g. an explicit
++// `subscribed=false` from clicking "Unsubscribe").
++func (q *Queries) InsertNotificationThreadIfAbsent(ctx context.Context, db DBTX, arg InsertNotificationThreadIfAbsentParams) error {
++	_, err := db.Exec(ctx, insertNotificationThreadIfAbsent,
++		arg.RecipientUserID,
++		arg.ThreadKind,
++		arg.ThreadID,
++		arg.Subscribed,
++		arg.Reason,
++	)
++	return err
++}
++
++const listSubscribersForThread = `-- name: ListSubscribersForThread :many
++SELECT recipient_user_id, reason
++FROM notification_threads
++WHERE thread_kind = $1 AND thread_id = $2 AND subscribed = true
++`
++
++type ListSubscribersForThreadParams struct {
++	ThreadKind NotificationThreadKind
++	ThreadID   int64
++}
++
++type ListSubscribersForThreadRow struct {
++	RecipientUserID int64
++	Reason          string
++}
++
++// Fan-out helper: returns recipients who explicitly subscribed to a
++// thread. The fan-out worker unions this with the per-repo `watches`
++// result + author/assignee/reviewer rules.
++func (q *Queries) ListSubscribersForThread(ctx context.Context, db DBTX, arg ListSubscribersForThreadParams) ([]ListSubscribersForThreadRow, error) {
++	rows, err := db.Query(ctx, listSubscribersForThread, arg.ThreadKind, arg.ThreadID)
++	if err != nil {
++		return nil, err
++	}
++	defer rows.Close()
++	items := []ListSubscribersForThreadRow{}
++	for rows.Next() {
++		var i ListSubscribersForThreadRow
++		if err := rows.Scan(&i.RecipientUserID, &i.Reason); err != nil {
++			return nil, err
++		}
++		items = append(items, i)
++	}
++	if err := rows.Err(); err != nil {
++		return nil, err
++	}
++	return items, nil
++}
++
++const upsertNotificationThread = `-- name: UpsertNotificationThread :exec
++INSERT INTO notification_threads (recipient_user_id, thread_kind, thread_id, subscribed, reason)
++VALUES ($1, $2, $3, $4, $5)
++ON CONFLICT (recipient_user_id, thread_kind, thread_id)
++DO UPDATE SET subscribed = EXCLUDED.subscribed,
++              reason     = EXCLUDED.reason,
++              updated_at = now()
++`
++
++type UpsertNotificationThreadParams struct {
++	RecipientUserID int64
++	ThreadKind      NotificationThreadKind
++	ThreadID        int64
++	Subscribed      bool
++	Reason          string
++}
++
++// Always-write upsert. Used by Subscribe / Unsubscribe / Ignore
++// handlers and by the auto-subscription rules in the fan-out
++// worker.
++func (q *Queries) UpsertNotificationThread(ctx context.Context, db DBTX, arg UpsertNotificationThreadParams) error {
++	_, err := db.Exec(ctx, upsertNotificationThread,
++		arg.RecipientUserID,
++		arg.ThreadKind,
++		arg.ThreadID,
++		arg.Subscribed,
++		arg.Reason,
++	)
++	return err
++}