package deliberi

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"net/url"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/gorilla/websocket"
	"tangled.org/core/deliberi/analytics"
	"tangled.org/core/deliberi/config"
	deldb "tangled.org/core/deliberi/db"
	models "tangled.org/core/deliberi/models"
	"tangled.org/core/jetstream"
)

type fakeBobbin struct {
	owners    map[string]string
	repoNames map[string]string
	ownerErr  error
	// proves cache hits skip bobbin
	ownerCalls *int

	authors map[string]string
}

func (f fakeBobbin) RepoOwner(ctx context.Context, repoDid string) (string, string, error) {
	if f.ownerCalls != nil {
		*f.ownerCalls++
	}
	if f.ownerErr != nil {
		return "", "", f.ownerErr
	}
	return f.owners[repoDid], f.repoNames[repoDid], nil
}

func (f fakeBobbin) TicketAuthor(ctx context.Context, ticketUri string) (string, error) {
	if author, ok := f.authors[ticketUri]; ok {
		return author, nil
	}
	return "", io.ErrUnexpectedEOF
}

const (
	ownerDid  = "did:plc:bob"
	editorDid = "did:plc:alice"
	collabDid = "did:plc:carol"
	otherDid  = "did:plc:dave"
	ticketUri = "at://" + repoDid + "/org.tangled.track.ticket/1"
)

// an hour ahead of the db, so new frames count as live
var frameClock = time.Now().Add(time.Hour).UnixMicro()

func frame(did, op, collection, rkey, rev, record string) *jetstream.Event {
	frameClock++
	live := true
	return &jetstream.Event{
		Did:    did,
		TimeUS: frameClock,
		Kind:   jetstream.EventKindCommit,
		Commit: &jetstream.Commit{
			Rev:        rev,
			Operation:  op,
			Collection: collection,
			RKey:       rkey,
			Record:     []byte(record),
			CID:        "cid-" + rev,
			Live:       &live,
		},
	}
}

func backfill(e *jetstream.Event) *jetstream.Event {
	live := false
	e.Commit.Live = &live
	return e
}

func newTestIngester(t *testing.T, b bobbinLookup) *Ingester {
	t.Helper()
	database, err := deldb.Make(context.Background(), filepath.Join(t.TempDir(), "x.db"))
	if err != nil {
		t.Fatalf("make db: %v", err)
	}
	t.Cleanup(func() { database.Close() })
	i := NewIngester(database, b, nil, "ws://hydrant/subscribe", slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err := deldb.PutRepoName(i.db, repoDid, ownerDid, "my-repo"); err != nil {
		t.Fatalf("seed repo: %v", err)
	}
	return i
}

func mustProcess(t *testing.T, i *Ingester, events ...*jetstream.Event) {
	t.Helper()
	for _, e := range events {
		if err := i.process(context.Background(), e); err != nil {
			t.Fatalf("process: %v", err)
		}
	}
}

func countFor(t *testing.T, i *Ingester, did string) int64 {
	t.Helper()
	n, err := deldb.CountNotifications(i.db, did)
	if err != nil {
		t.Fatalf("count: %v", err)
	}
	return n
}

func notifsFor(t *testing.T, i *Ingester, did string) []*models.Notification {
	t.Helper()
	ns, err := deldb.GetNotifications(i.db, did, 0)
	if err != nil {
		t.Fatalf("get notifications: %v", err)
	}
	return ns
}

func onlyFor(t *testing.T, i *Ingester, did string) *models.Notification {
	t.Helper()
	ns := notifsFor(t, i, did)
	if len(ns) != 1 {
		t.Fatalf("%s has %d notifications, want 1", did, len(ns))
	}
	return ns[0]
}

func wantTypes(t *testing.T, i *Ingester, want map[string]models.NotificationType) {
	t.Helper()
	for _, did := range []string{ownerDid, editorDid, collabDid, otherDid, "did:plc:sub", "did:plc:ignorer"} {
		ns := notifsFor(t, i, did)
		wantType, ok := want[did]
		switch {
		case !ok && len(ns) != 0:
			t.Errorf("%s has %d notifications (first %q), want none", did, len(ns), ns[0].Type)
		case ok && (len(ns) != 1 || ns[0].Type != wantType):
			t.Errorf("%s notifications = %d, want exactly one %q", did, len(ns), wantType)
		}
	}
}

// rows can share a created second, so order is no help
func byType(t *testing.T, i *Ingester, did string, want models.NotificationType) *models.Notification {
	t.Helper()
	for _, n := range notifsFor(t, i, did) {
		if n.Type == want {
			return n
		}
	}
	t.Fatalf("%s has no %q notification", did, want)
	return nil
}

func ticketJSON(state, editor, extra string) string {
	return `{"$type":"org.tangled.track.ticket","createdAt":"2026-01-01T00:00:00Z","state":"` + state + `",` +
		`"title":{"$type":"org.tangled.markup.markdown#inline","text":"the ticket"},` + extra +
		`"x-tngl-editor":"` + editor + `"}`
}

const pullExtra = `"targetBranch":"main",`

func ticketFrame(op, rev, state, editor, extra string) *jetstream.Event {
	return frame(repoDid, op, ticketNSID, "1", rev, ticketJSON(state, editor, extra))
}

func subscriptionJSON(subject string, ignore bool) string {
	return `{"$type":"org.tangled.feed.subscription","createdAt":"2026-01-01T00:00:00Z","ignore":` + strconv.FormatBool(ignore) + `,"subject":` + subject + `}`
}

func recordSubject(uri string) string {
	return `{"$type":"org.tangled.feed.subscription#record","uri":"` + uri + `"}`
}

func repoSubject(did string) string {
	return `{"$type":"org.tangled.feed.subscription#repo","did":"` + did + `"}`
}

func subscribe(user, rkey, subject string, ignore bool) *jetstream.Event {
	return frame(user, jetstream.CommitOperationCreate, subscriptionNSID, rkey, "rev-"+rkey, subscriptionJSON(subject, ignore))
}

func collaboratorFrame(op, rev, did string) *jetstream.Event {
	return frame(repoDid, op, collaboratorNSID, did, rev, `{"$type":"org.tangled.graph.collaborator","createdAt":"2026-01-01T00:00:00Z"}`)
}

func commentJSON(subjectUri, replyTo, editor string) string {
	reply := ""
	if replyTo != "" {
		reply = `"replyTo":{"uri":"` + replyTo + `","cid":"bafyfake"},`
	}
	return `{"$type":"org.tangled.feed.comment","createdAt":"2026-01-01T00:00:00Z",` +
		`"body":{"$type":"org.tangled.markup.markdown","text":"looks good"},` + reply +
		`"subject":{"uri":"` + subjectUri + `","cid":"bafyfake"},"x-tngl-editor":"` + editor + `"}`
}

func commentUri(rkey string) string {
	return "at://" + repoDid + "/org.tangled.feed.comment/" + rkey
}

func commentFrame(rkey, rev, subjectUri, replyTo, editor string) *jetstream.Event {
	return frame(repoDid, jetstream.CommitOperationCreate, commentNSID, rkey, rev, commentJSON(subjectUri, replyTo, editor))
}

func TestLiveCreateNotifiesOwnerCollaboratorsAndSubscribers(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		collaboratorFrame(jetstream.CommitOperationCreate, "r1", collabDid),
		subscribe("did:plc:sub", "s1", recordSubject(ticketUri), false),
		subscribe(editorDid, "s2", repoSubject(repoDid), false),
		subscribe("did:plc:ignorer", "s3", repoSubject(repoDid), true),
		subscribe(ownerDid, "s4", recordSubject(ticketUri), true),
		ticketFrame(jetstream.CommitOperationCreate, "r2", "open", editorDid, ""),
	)

	wantTypes(t, i, map[string]models.NotificationType{
		collabDid:     models.NotificationTypeIssueCreated,
		"did:plc:sub": models.NotificationTypeIssueCreated,
	})

	if got := countFor(t, i, ownerDid); got != 0 {
		t.Errorf("owner rows = %d, want 0 (ignored the ticket)", got)
	}
	n := onlyFor(t, i, collabDid)
	if n.ActorDid != editorDid || n.EntityAt != ticketUri || n.RepoDid != repoDid || n.EntityTitle != "the ticket" || n.EntityKind != models.EntityKindIssue {
		t.Errorf("notification = %+v", n)
	}
	if n.AtUri != ticketUri {
		t.Errorf("key = %q, want the ticket uri", n.AtUri)
	}
}

func TestLiveCreateOfPullNotifiesOwner(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, pullExtra))

	n := onlyFor(t, i, ownerDid)
	if n.Type != models.NotificationTypePullCreated || n.EntityKind != models.EntityKindPull {
		t.Errorf("type/kind = %q/%q, want pull_created/pull", n.Type, n.EntityKind)
	}
}

func TestBackfillStoresStateWithoutNotifying(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{authors: map[string]string{ticketUri: editorDid}})
	mustProcess(t, i,
		backfill(ticketFrame(jetstream.CommitOperationCreate, "r1", "open", otherDid, "")),
		backfill(ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", otherDid, "")),
	)

	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("owner rows = %d, want 0 for a backfill", got)
	}
	got := deldb.GetEntity(i.db, ticketUri)
	if got.State != "closed" || got.Cid != "cid-r2" || got.Title != "the ticket" {
		t.Errorf("stored ticket = %+v", got)
	}
	if got.Author != "" {
		t.Errorf("author = %q, want none yet: a snapshot's editor isn't the author and a replay doesn't ask bobbin", got.Author)
	}

	// bobbin supplies the opener on a live update, not the frame's editor
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r3", "open", otherDid, ""))
	if got := deldb.GetEntity(i.db, ticketUri).Author; got != editorDid {
		t.Errorf("author = %q, want bobbin's original author once a live update notifies", got)
	}
	if n := onlyFor(t, i, editorDid); n.Type != models.NotificationTypeIssueReopen {
		t.Errorf("author's notification = %q, want issue_reopen", n.Type)
	}
}

func TestBackfillSetsBaselineForTheNextLiveUpdate(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		backfill(ticketFrame(jetstream.CommitOperationCreate, "r1", "closed", editorDid, "")),
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "open", otherDid, ""),
	)

	if n := onlyFor(t, i, ownerDid); n.Type != models.NotificationTypeIssueReopen {
		t.Errorf("type = %q, want issue_reopen", n.Type)
	}
}

func TestLiveUpdateWithoutRowOnlyAdoptsTheTicket(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{authors: map[string]string{ticketUri: editorDid}})
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", otherDid, ""))

	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("owner rows = %d, want 0 (no baseline to diff)", got)
	}
	if got := deldb.GetEntity(i.db, ticketUri); got.Author != editorDid || got.State != "closed" {
		t.Errorf("adopted ticket = %+v", got)
	}
}

func TestCloseByMaintainerNotifiesEveryone(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		collaboratorFrame(jetstream.CommitOperationCreate, "r1", collabDid),
		subscribe("did:plc:sub", "s1", repoSubject(repoDid), false),
		subscribe("did:plc:ignorer", "s2", recordSubject(ticketUri), true),
		ticketFrame(jetstream.CommitOperationCreate, "r2", "open", editorDid, ""),
		commentFrame("c1", "r3", ticketUri, "", otherDid),
	)
	for _, did := range []string{ownerDid, editorDid, collabDid, otherDid, "did:plc:sub"} {
		if err := deldb.DeleteNotification(i.db, did, ticketUri); err != nil {
			t.Fatal(err)
		}
		if err := deldb.DeleteNotification(i.db, did, commentUri("c1")); err != nil {
			t.Fatal(err)
		}
	}

	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r4", "closed", collabDid, ""))

	wantTypes(t, i, map[string]models.NotificationType{
		ownerDid:      models.NotificationTypeIssueClosed,
		editorDid:     models.NotificationTypeIssueClosed,
		otherDid:      models.NotificationTypeIssueClosed,
		"did:plc:sub": models.NotificationTypeIssueClosed,
	})
	n := onlyFor(t, i, ownerDid)
	if n.ActorDid != collabDid || n.AtUri != ticketUri+"#r4" {
		t.Errorf("actor/key = %q/%q", n.ActorDid, n.AtUri)
	}
}

func TestIgnoreBeatsTheOwnerOnStateChange(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		subscribe(ownerDid, "s1", recordSubject(ticketUri), true),
	)
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", editorDid, ""))

	if got := notifsFor(t, i, ownerDid); len(got) != 1 || got[0].Type != models.NotificationTypeIssueCreated {
		t.Errorf("owner notifications = %+v, want only the earlier created", got)
	}
}

func TestPullStateChanges(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, pullExtra))

	for n, step := range []struct {
		state string
		want  models.NotificationType
	}{
		{"closed", models.NotificationTypePullClosed},
		{"open", models.NotificationTypePullReopen},
		{"completed", models.NotificationTypePullMerged},
	} {
		mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r"+strconv.Itoa(n+2), step.state, otherDid, pullExtra))
		n := byType(t, i, ownerDid, step.want)
		if n.EntityKind != models.EntityKindPull || n.ActorDid != otherDid {
			t.Errorf("after %s: %q is %q by %q", step.state, step.want, n.EntityKind, n.ActorDid)
		}
	}
	if got := countFor(t, i, otherDid); got != 0 {
		t.Errorf("actor rows = %d, want 0", got)
	}
}

type capturedEvent struct {
	Uuid       string         `json:"uuid"`
	Event      string         `json:"event"`
	DistinctId string         `json:"distinct_id"`
	Properties map[string]any `json:"properties"`
}

// a posthog stand-in collecting what the client batches to /batch/
func fakePosthog(t *testing.T) (*analytics.Client, func() []capturedEvent) {
	t.Helper()
	var events []capturedEvent
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var body struct {
			Batch []capturedEvent `json:"batch"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			t.Errorf("decoding posthog batch: %v", err)
		}
		events = append(events, body.Batch...)
	}))
	t.Cleanup(srv.Close)
	client := analytics.New(config.PosthogConfig{ApiKey: "key", Endpoint: srv.URL, Env: "test"},
		slog.New(slog.NewTextHandler(io.Discard, nil)))
	return client, func() []capturedEvent {
		client.Close()
		return events
	}
}

func TestStateChangesReachPosthogUnderTheAppviewNames(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	client, flushed := fakePosthog(t)
	i.analytics = client
	frames := []*jetstream.Event{
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, pullExtra),
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", otherDid, pullExtra),
		ticketFrame(jetstream.CommitOperationUpdate, "r3", "open", otherDid, pullExtra),
		ticketFrame(jetstream.CommitOperationUpdate, "r4", "completed", otherDid, pullExtra),
	}
	mustProcess(t, i, frames...)
	// a replay from another hydrant must not count twice
	mustProcess(t, i, frames...)

	events := flushed()
	want := []string{"pull_closed", "pull_reopen", "pull_merged"}
	if len(events) != len(want) {
		t.Fatalf("got %d events %v, want %v", len(events), events, want)
	}
	for n, e := range events {
		if e.Event != want[n] {
			t.Errorf("event %d = %q, want %q", n, e.Event, want[n])
		}
		// the author is the subject, the maintainer who acted is a property
		if e.DistinctId != editorDid || e.Properties["actor"] != otherDid {
			t.Errorf("%s: distinct id %q, actor %v, want %q and %q", e.Event, e.DistinctId, e.Properties["actor"], editorDid, otherDid)
		}
		if e.Properties["repo_did"] != repoDid || e.Properties["uri"] != ticketUri || e.Properties["environment"] != "test" {
			t.Errorf("%s: properties %v", e.Event, e.Properties)
		}
	}
}

func TestIssueReopen(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", editorDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r3", "open", editorDid, ""),
	)
	if got := countFor(t, i, ownerDid); got != 3 {
		t.Fatalf("owner rows = %d, want created, closed and reopen", got)
	}
	byType(t, i, ownerDid, models.NotificationTypeIssueReopen)
}

func TestEditsThatKeepTheStateDoNotNotify(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "open", otherDid, `"tangledLabels":[{"key":"bug"}],`),
	)
	if got := countFor(t, i, ownerDid); got != 1 {
		t.Fatalf("owner rows = %d, want only the created", got)
	}
	if got := deldb.GetEntity(i.db, ticketUri); got.Cid != "cid-r2" {
		t.Errorf("cid = %q, want the edit stored", got.Cid)
	}
}

func TestReplayDoesNotNotifyTwiceOrReverseTheState(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	frames := []*jetstream.Event{
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", editorDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r3", "open", editorDid, ""),
	}
	mustProcess(t, i, frames...)
	mustProcess(t, i, frames...)
	// a stale r2 close can show up late from another hydrant
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", editorDid, ""))

	if got := countFor(t, i, ownerDid); got != 3 {
		t.Errorf("owner rows = %d, want 3", got)
	}
	if got := deldb.GetEntity(i.db, ticketUri); got.State != "open" || got.Cid != "cid-r3" {
		t.Errorf("stored ticket = %+v, want the newest version untouched", got)
	}
}

func TestDeleteTombstoneBlocksAnOlderCreate(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		frame(repoDid, jetstream.CommitOperationDelete, ticketNSID, "1", "r2", ""),
	)
	if got := deldb.GetEntity(i.db, ticketUri); got.Kind != "" {
		t.Fatalf("ticket after delete = %+v, want gone", got)
	}

	mustProcess(t, i, ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""))

	if got := deldb.GetEntity(i.db, ticketUri); got.Kind != "" {
		t.Errorf("an older create revived the ticket: %+v", got)
	}
	// a replayed create must not bring back a notice the erase removed
	if got := countFor(t, i, ownerDid); got != 0 {
		t.Errorf("owner rows = %d, want 0", got)
	}
}

func TestCommentRecipientsFollowLegacyRules(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		commentFrame("top", "r2", ticketUri, "", otherDid),
		commentFrame("reply", "r3", ticketUri, commentUri("top"), collabDid),
		commentFrame("lonely", "r4", ticketUri, "", "did:plc:sub"),
	)
	for _, did := range []string{ownerDid, editorDid, collabDid, otherDid, "did:plc:sub"} {
		for _, key := range []string{ticketUri, commentUri("top"), commentUri("reply"), commentUri("lonely")} {
			if err := deldb.DeleteNotification(i.db, did, key); err != nil {
				t.Fatal(err)
			}
		}
	}

	mustProcess(t, i, commentFrame("c1", "r5", ticketUri, "", "did:plc:ignorer"))
	wantTypes(t, i, map[string]models.NotificationType{
		ownerDid:  models.NotificationTypeIssueCommented,
		editorDid: models.NotificationTypeIssueCommented,
	})
	for _, did := range []string{ownerDid, editorDid} {
		if err := deldb.DeleteNotification(i.db, did, commentUri("c1")); err != nil {
			t.Fatal(err)
		}
	}

	mustProcess(t, i, commentFrame("c2", "r6", ticketUri, commentUri("top"), editorDid))
	wantTypes(t, i, map[string]models.NotificationType{
		ownerDid:  models.NotificationTypeIssueCommented,
		collabDid: models.NotificationTypeIssueCommented,
		otherDid:  models.NotificationTypeIssueCommented,
	})
}

func TestPullCommentTellsEveryParticipant(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, pullExtra),
		commentFrame("c1", "r2", ticketUri, "", otherDid),
		commentFrame("c2", "r3", ticketUri, "", "did:plc:sub"),
	)

	got := notifsFor(t, i, otherDid)
	if len(got) != 1 || got[0].AtUri != commentUri("c2") || got[0].Type != models.NotificationTypePullCommented || got[0].EntityKind != models.EntityKindPull {
		t.Errorf("earlier commenter notifications = %+v, want the later pull comment", got)
	}
	if got := countFor(t, i, editorDid); got != 2 {
		t.Errorf("pull author rows = %d, want both comments", got)
	}
	if got := countFor(t, i, collabDid); got != 0 {
		t.Errorf("stranger rows = %d, want 0", got)
	}
}

func TestCommentOnPatchResolvesToTheTicket(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	patchUri := "at://" + repoDid + "/org.tangled.review.patch/bafypatch"
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, pullExtra),
		frame(repoDid, jetstream.CommitOperationCreate, patchNSID, "bafypatch", "r2", `{"ticket":"1","base":"a","head":"b","createdAt":"2026-01-01T00:00:00Z"}`),
		commentFrame("c1", "r3", patchUri, "", otherDid),
	)

	n := onlyFor(t, i, editorDid)
	if n.Type != models.NotificationTypePullCommented || n.EntityAt != ticketUri {
		t.Errorf("notification = %+v, want a pull comment on the ticket", n)
	}
}

// hydrant backfills a repo in key order, so feed.comment, then review.patch, then track.ticket
func TestBackfillKeepsCommentsSentBeforeTheirTicket(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	patchUri := "at://" + repoDid + "/org.tangled.review.patch/bafypatch"
	mustProcess(t, i,
		backfill(commentFrame("c1", "r1", ticketUri, "", otherDid)),
		backfill(commentFrame("c2", "r2", patchUri, "", "did:plc:sub")),
		backfill(frame(repoDid, jetstream.CommitOperationCreate, patchNSID, "bafypatch", "r3", `{"ticket":"1","base":"a","head":"b","createdAt":"2026-01-01T00:00:00Z"}`)),
		backfill(ticketFrame(jetstream.CommitOperationCreate, "r4", "open", editorDid, pullExtra)),
		ticketFrame(jetstream.CommitOperationUpdate, "r5", "closed", ownerDid, pullExtra),
	)

	for _, did := range []string{otherDid, "did:plc:sub"} {
		if n := onlyFor(t, i, did); n.Type != models.NotificationTypePullClosed {
			t.Errorf("%s got %q, want the close", did, n.Type)
		}
	}
}

// like a comment on a deleted ticket. nothing to name in a notice, but the commenter still counts
func TestLiveCommentOnAnUnseenTicketOnlyKeepsTheCommenter(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		commentFrame("c1", "r1", ticketUri, "", otherDid),
		// the first version deliberi sees is only a baseline
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", ownerDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r3", "open", ownerDid, ""),
	)
	if n := onlyFor(t, i, otherDid); n.Type != models.NotificationTypeIssueReopen {
		t.Errorf("commenter got %q, want only the reopen", n.Type)
	}
}

func TestCommentOnAnotherReposTicketIgnored(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	const foreign = "at://did:plc:elsewhere/org.tangled.track.ticket/1"
	if err := deldb.PutEntity(i.db, foreign, deldb.Entity{Title: "t", RepoDid: "did:plc:elsewhere", Kind: models.EntityKindIssue, Author: editorDid}); err != nil {
		t.Fatalf("PutEntity: %v", err)
	}

	mustProcess(t, i, commentFrame("c1", "r1", foreign, "", otherDid))

	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("rows = %d, want 0 (a repo can't comment into another repo's ticket)", got)
	}
}

func TestCommentEditAndBackfillDoNotNotify(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		backfill(ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, "")),
		backfill(commentFrame("c1", "r2", ticketUri, "", otherDid)),
		frame(repoDid, jetstream.CommitOperationUpdate, commentNSID, "c1", "r3", commentJSON(ticketUri, "", otherDid)),
	)
	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("rows = %d, want 0", got)
	}
	// a backfilled comment still makes its author a participant
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r4", "closed", editorDid, ""))
	if got := countFor(t, i, otherDid); got != 1 {
		t.Errorf("backfilled commenter rows = %d, want the close", got)
	}
}

func TestEditorlessCommentAttributedToRepo(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		frame(repoDid, jetstream.CommitOperationCreate, commentNSID, "c1", "r2",
			`{"$type":"org.tangled.feed.comment","createdAt":"2026-01-01T00:00:00Z",`+
				`"body":{"$type":"org.tangled.markup.markdown","text":"hi"},"subject":{"uri":"`+ticketUri+`","cid":"bafyfake"}}`),
	)
	if n := byType(t, i, editorDid, models.NotificationTypeIssueCommented); n.ActorDid != repoDid {
		t.Errorf("actor = %q, want the repo did", n.ActorDid)
	}
}

func TestCommentOnUnknownTicketTakesAuthorFromBobbin(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{authors: map[string]string{ticketUri: editorDid}})
	if err := deldb.PutEntity(i.db, ticketUri, deldb.Entity{Title: "old", RepoDid: repoDid, Kind: models.EntityKindIssue}); err != nil {
		t.Fatal(err)
	}

	mustProcess(t, i, commentFrame("c1", "r1", ticketUri, "", otherDid))

	if n := onlyFor(t, i, editorDid); n.Type != models.NotificationTypeIssueCommented {
		t.Errorf("type = %q", n.Type)
	}
	if got := deldb.GetEntity(i.db, ticketUri).Author; got != editorDid {
		t.Errorf("cached author = %q", got)
	}
}

func TestSubscriptionResolution(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	const sub = "did:plc:sub"
	resolve := func() (subs, ignored []string) {
		subs, ignored, err := deldb.ListSubscribers(i.db, ticketUri, repoDid)
		if err != nil {
			t.Fatal(err)
		}
		return subs, ignored
	}

	mustProcess(t, i, subscribe(sub, "3a", recordSubject(ticketUri), false))
	if subs, ign := resolve(); !slices.Equal(subs, []string{sub}) || len(ign) != 0 {
		t.Fatalf("subscribed: subs=%v ignored=%v", subs, ign)
	}

	mustProcess(t, i, subscribe(sub, "3b", recordSubject(ticketUri), true))
	if subs, ign := resolve(); len(subs) != 0 || !slices.Equal(ign, []string{sub}) {
		t.Fatalf("ignored: subs=%v ignored=%v", subs, ign)
	}

	// 3a2 sorts before 3b, so arriving late it can't undo the ignore
	mustProcess(t, i, subscribe(sub, "3a2", recordSubject(ticketUri), false))
	if subs, ign := resolve(); len(subs) != 0 || !slices.Equal(ign, []string{sub}) {
		t.Fatalf("stale duplicate: subs=%v ignored=%v", subs, ign)
	}

	mustProcess(t, i, frame(sub, jetstream.CommitOperationDelete, subscriptionNSID, "3b", "rev-3b-del", ""))
	if subs, ign := resolve(); !slices.Equal(subs, []string{sub}) || len(ign) != 0 {
		t.Fatalf("after delete: subs=%v ignored=%v", subs, ign)
	}

	mustProcess(t, i,
		frame(sub, jetstream.CommitOperationDelete, subscriptionNSID, "3a", "rev-3a-del", ""),
		frame(sub, jetstream.CommitOperationDelete, subscriptionNSID, "3a2", "rev-3a2-del", ""),
	)
	if subs, ign := resolve(); len(subs) != 0 || len(ign) != 0 {
		t.Fatalf("all deleted: subs=%v ignored=%v", subs, ign)
	}

	mustProcess(t, i, subscribe(sub, "3c", repoSubject(repoDid), false))
	if subs, _ := resolve(); !slices.Equal(subs, []string{sub}) {
		t.Fatalf("repo level: subs=%v", subs)
	}
}

func TestCollaboratorCreateAndDelete(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i, collaboratorFrame(jetstream.CommitOperationCreate, "r1", collabDid))
	got, err := deldb.ListCollaborators(i.db, repoDid)
	if err != nil || !slices.Equal(got, []string{collabDid}) {
		t.Fatalf("collaborators = %v, %v", got, err)
	}

	mustProcess(t, i, collaboratorFrame(jetstream.CommitOperationDelete, "r2", collabDid), collaboratorFrame(jetstream.CommitOperationCreate, "r1", collabDid))
	if got, _ := deldb.ListCollaborators(i.db, repoDid); len(got) != 0 {
		t.Fatalf("collaborators after revoke = %v, want none even with the old create replayed", got)
	}

	mustProcess(t, i, collaboratorFrame(jetstream.CommitOperationCreate, "r3", collabDid))
	if got, _ := deldb.ListCollaborators(i.db, repoDid); !slices.Equal(got, []string{collabDid}) {
		t.Fatalf("collaborators after re-add = %v", got)
	}
}

func TestBootstrapHistoryStaysSilent(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	bootstrap := deldb.GetBootstrapUs(i.db)
	if bootstrap == 0 {
		t.Fatal("a fresh db must record when it was created")
	}

	old := ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, "")
	old.TimeUS = bootstrap - 1_000_000
	mustProcess(t, i, old)
	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("owner rows = %d, want 0 for history from before the db existed", got)
	}

	// history built the state silently, so this close still notifies
	mustProcess(t, i, ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", editorDid, ""))
	if n := onlyFor(t, i, ownerDid); n.Type != models.NotificationTypeIssueClosed {
		t.Errorf("type = %q, want issue_closed", n.Type)
	}
}

func TestUpgradedDatabaseIsNotSilenced(t *testing.T) {
	path := filepath.Join(t.TempDir(), "x.db")
	d, err := deldb.Make(context.Background(), path)
	if err != nil {
		t.Fatal(err)
	}
	for _, q := range []string{
		`delete from ingest_bootstrap`,
		`delete from migrations where name = 'ingest-bootstrap'`,
		`insert into jetstream_cursor (id, last_time_us) values (0, 5)`,
	} {
		if _, err := d.Exec(q); err != nil {
			t.Fatal(err)
		}
	}
	d.Close()

	d, err = deldb.Make(context.Background(), path)
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	if got := deldb.GetBootstrapUs(d); got != 0 {
		t.Errorf("bootstrap = %d, want 0 for a db that already ingested", got)
	}
}

func TestEventAndCursorCommitTogether(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	e := ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, "")
	mustProcess(t, i, e)
	if us, ok, _ := deldb.GetCursor(i.db); !ok || us != e.TimeUS {
		t.Fatalf("cursor = %d/%v, want %d", us, ok, e.TimeUS)
	}

	// a failed write must not move the cursor or the rev, or a retry gets skipped
	if _, err := i.db.Exec(`drop table repo_collaborators`); err != nil {
		t.Fatal(err)
	}
	failing := collaboratorFrame(jetstream.CommitOperationCreate, "r2", collabDid)
	if err := i.process(context.Background(), failing); err == nil {
		t.Fatal("process succeeded without its table")
	}
	if us, _, _ := deldb.GetCursor(i.db); us != e.TimeUS {
		t.Errorf("cursor = %d, want it left at %d", us, e.TimeUS)
	}
	if !deldb.RevIsNew(i.db, "at://"+repoDid+"/"+collaboratorNSID+"/"+collabDid, "r2") {
		t.Error("the failed event's rev was kept, a retry would be skipped")
	}
}

func TestCursorNeverMovesBack(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	if _, ok, err := deldb.GetCursor(i.db); err != nil || ok {
		t.Fatalf("fresh cursor ok = %v, %v, want none", ok, err)
	}

	now := time.Now().UnixMicro()
	for _, us := range []int64{now, now - 1_000_000} {
		if err := deldb.SaveCursor(i.db, us); err != nil {
			t.Fatal(err)
		}
	}
	if got, _, err := deldb.GetCursor(i.db); err != nil || got != now {
		t.Fatalf("cursor = %d, %v, want %d", got, err, now)
	}
}

func TestRunIngestsFromTheStreamAndMovesTheCursor(t *testing.T) {
	e := ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, "")
	frameJSON, err := json.Marshal(e)
	if err != nil {
		t.Fatal(err)
	}
	var query url.Values
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		query = r.URL.Query()
		con, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
		if err != nil {
			return
		}
		defer con.Close()
		con.WriteMessage(websocket.TextMessage, frameJSON)
		con.ReadMessage()
	}))
	defer srv.Close()

	i := newTestIngester(t, fakeBobbin{})
	i.endpoint = "ws" + strings.TrimPrefix(srv.URL, "http") + "/subscribe"
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- i.Run(ctx) }()

	deadline := time.Now().Add(5 * time.Second)
	for deldb.GetEntity(i.db, ticketUri).Kind == "" {
		if time.Now().After(deadline) {
			t.Fatal("the streamed ticket never landed")
		}
		time.Sleep(10 * time.Millisecond)
	}
	if us, _, _ := deldb.GetCursor(i.db); us != e.TimeUS {
		t.Errorf("cursor = %d, want %d", us, e.TimeUS)
	}
	if query.Get("cursor") != "0" || !slices.Equal(query["wantedEventTypes"], []string{"live", "historical"}) {
		t.Errorf("query = %v, want a replay from 0 for live and historical", query)
	}
	cancel()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

func TestDisabledPrefsSuppressStateNotifications(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	prefs := models.DefaultNotificationPreferences(syntax.DID(ownerDid))
	prefs.IssueClosed = false
	prefs.PullMerged = false
	if err := deldb.UpsertNotificationPreferences(i.db, prefs); err != nil {
		t.Fatal(err)
	}
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		ticketFrame(jetstream.CommitOperationUpdate, "r2", "closed", editorDid, ""),
		frame(repoDid, jetstream.CommitOperationCreate, ticketNSID, "2", "r3", ticketJSON("open", editorDid, pullExtra)),
		frame(repoDid, jetstream.CommitOperationUpdate, ticketNSID, "2", "r4", ticketJSON("completed", editorDid, pullExtra)),
	)
	for _, n := range notifsFor(t, i, ownerDid) {
		if n.Type == models.NotificationTypeIssueClosed || n.Type == models.NotificationTypePullMerged {
			t.Errorf("owner got %q with that pref off", n.Type)
		}
	}
	if got := countFor(t, i, ownerDid); got != 2 {
		t.Errorf("owner rows = %d, want the two created", got)
	}
}

func TestFanOutMentions(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	n := models.Notification{AtUri: "at://src", Type: models.NotificationTypeIssueCreated, ActorDid: editorDid, RepoDid: repoDid, EntityAt: ticketUri, EntityKind: models.EntityKindIssue}

	if err := i.fanOut(i.db, n, []string{ownerDid, otherDid}, []string{otherDid}); err != nil {
		t.Fatal(err)
	}
	if got := onlyFor(t, i, otherDid); got.Type != models.NotificationTypeUserMentioned || got.EntityKind != models.EntityKindIssue {
		t.Errorf("mentioned user got %+v, want the mention in place of the generic one", got)
	}
	if got := onlyFor(t, i, ownerDid); got.Type != models.NotificationTypeIssueCreated {
		t.Errorf("owner got %q", got.Type)
	}
}

func TestMoreThanEightMentionsIsSpam(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	n := models.Notification{AtUri: "at://src", Type: models.NotificationTypeIssueCreated, ActorDid: editorDid, RepoDid: repoDid, EntityAt: ticketUri, EntityKind: models.EntityKindIssue}
	var mentions []string
	for k := range 9 {
		mentions = append(mentions, "did:plc:m"+strconv.Itoa(k))
	}

	if err := i.fanOut(i.db, n, []string{ownerDid, mentions[0]}, mentions); err != nil {
		t.Fatal(err)
	}
	for _, did := range mentions {
		if k := countFor(t, i, did); did != mentions[0] && k != 0 {
			t.Errorf("%s was told about a spammy mention", did)
		}
	}
	// mentions[0] is a plain recipient too, so it keeps the generic notice
	if n := onlyFor(t, i, mentions[0]); n.Type != models.NotificationTypeIssueCreated {
		t.Errorf("type = %q, want the generic notification", n.Type)
	}
}

func TestMentionStillReachesAMutedThread(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i, subscribe(otherDid, "s1", recordSubject(ticketUri), true))
	n := models.Notification{AtUri: "at://src", Type: models.NotificationTypeIssueCreated, ActorDid: editorDid, RepoDid: repoDid, EntityAt: ticketUri, EntityKind: models.EntityKindIssue}

	if err := i.fanOut(i.db, n, []string{otherDid}, []string{otherDid}); err != nil {
		t.Fatal(err)
	}
	if got := onlyFor(t, i, otherDid); got.Type != models.NotificationTypeUserMentioned {
		t.Errorf("type = %q", got.Type)
	}
}

func TestManifestSeedsOwnerAndName(t *testing.T) {
	calls := 0
	i := newTestIngester(t, fakeBobbin{ownerCalls: &calls})
	const other = "did:plc:other"

	mustProcess(t, i,
		frame(other, jetstream.CommitOperationCreate, repoManifestNSID, "self", "r1",
			`{"$type":"org.tangled.repo.manifest","createdAt":"2026-01-01T00:00:00Z","declaration":{"owner":"`+ownerDid+`","name":"my-repo"}}`),
		starFrame(editorDid, other, "star1"),
	)

	if got := deldb.GetRepoName(i.db, other); got != "my-repo" {
		t.Errorf("cached name = %q, want my-repo", got)
	}
	if got := countFor(t, i, ownerDid); got != 1 {
		t.Fatalf("owner rows = %d, want 1", got)
	}
	if calls != 0 {
		t.Errorf("RepoOwner calls = %d, want 0 (manifest cache hit must not hit bobbin)", calls)
	}
}

func starFrame(actorDid, repoDid, rkey string) *jetstream.Event {
	return frame(actorDid, jetstream.CommitOperationCreate, starNSID, rkey, "rev-"+rkey,
		`{"$type":"org.tangled.feed.star","createdAt":"2026-01-01T00:00:00Z","subject":{"$type":"org.tangled.feed.star#repo","identity":"`+repoDid+`"}}`)
}

func TestStarResolvesOwnerFromBobbinOnCacheMiss(t *testing.T) {
	const other = "did:plc:other"
	i := newTestIngester(t, fakeBobbin{owners: map[string]string{other: ownerDid}, repoNames: map[string]string{other: "my-repo"}})

	mustProcess(t, i, starFrame(editorDid, other, "star1"))

	n := onlyFor(t, i, ownerDid)
	if n.Type != models.NotificationTypeRepoStarred || n.ActorDid != editorDid || n.RepoDid != other {
		t.Errorf("notification = %+v", n)
	}
	if got := deldb.GetRepoOwner(i.db, other); got != ownerDid {
		t.Errorf("cached owner = %q, want %q", got, ownerDid)
	}
}

func TestStarKeepsCachedNameWhenNameMissing(t *testing.T) {
	const other = "did:plc:other"
	i := newTestIngester(t, fakeBobbin{owners: map[string]string{other: ownerDid}})
	if err := deldb.PutRepoName(i.db, other, "", "my-repo"); err != nil {
		t.Fatalf("PutRepoName: %v", err)
	}

	mustProcess(t, i, starFrame(editorDid, other, "star1"))

	if got := deldb.GetRepoName(i.db, other); got != "my-repo" {
		t.Errorf("cached name = %q, want my-repo", got)
	}
}

func TestStarSkipsWhenOwnerUnresolvable(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{ownerErr: io.ErrUnexpectedEOF})

	mustProcess(t, i, starFrame(editorDid, "did:plc:other", "star1"))

	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("owner rows = %d, want 0", got)
	}
}

func TestBackfilledStarAndFollowAreSilent(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		backfill(starFrame(editorDid, repoDid, "star1")),
		backfill(frame(editorDid, jetstream.CommitOperationCreate, followNSID, ownerDid, "r1", `{"$type":"org.tangled.graph.follow","createdAt":"2026-01-01T00:00:00Z"}`)),
	)
	if got := countFor(t, i, ownerDid); got != 0 {
		t.Fatalf("owner rows = %d, want 0", got)
	}
}

func TestFollowNotifiesRkeySubject(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})

	mustProcess(t, i, frame(editorDid, jetstream.CommitOperationCreate, followNSID, ownerDid, "r1", `{"$type":"org.tangled.graph.follow","createdAt":"2026-01-01T00:00:00Z"}`))

	if n := onlyFor(t, i, ownerDid); n.Type != models.NotificationTypeFollowed || n.ActorDid != editorDid {
		t.Errorf("notification = %+v", n)
	}
}

func TestUndoingAStarOrFollowTakesTheNoticeBack(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	const follow = `{"$type":"org.tangled.graph.follow","createdAt":"2026-01-01T00:00:00Z"}`
	mustProcess(t, i,
		starFrame(editorDid, repoDid, "star1"),
		frame(editorDid, jetstream.CommitOperationCreate, followNSID, ownerDid, "r1", follow),
	)
	if got := countFor(t, i, ownerDid); got != 2 {
		t.Fatalf("owner rows = %d, want the star and the follow", got)
	}

	mustProcess(t, i,
		frame(editorDid, jetstream.CommitOperationDelete, starNSID, "star1", "r2", ""),
		frame(editorDid, jetstream.CommitOperationDelete, followNSID, ownerDid, "r3", ""),
	)
	if got := countFor(t, i, ownerDid); got != 0 {
		t.Errorf("owner rows = %d, want both taken back", got)
	}
}

func TestPutEntityKeepsKnownFields(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	first := deldb.Entity{Title: "t1", RepoDid: repoDid, Kind: models.EntityKindPull, State: "open", Author: editorDid, Cid: "c1"}
	if err := deldb.PutEntity(i.db, ticketUri, first); err != nil {
		t.Fatalf("first put: %v", err)
	}
	if err := deldb.PutEntity(i.db, ticketUri, deldb.Entity{Title: "t2"}); err != nil {
		t.Fatalf("second put: %v", err)
	}
	want := first
	want.Title = "t2"
	if got := deldb.GetEntity(i.db, ticketUri); got != want {
		t.Errorf("entity = %+v, want %+v", got, want)
	}
}

func TestCreatedNotificationUsesEventTime(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	e := ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, "")
	mustProcess(t, i, e)
	if got := onlyFor(t, i, ownerDid).Created; got.UnixMicro()/1_000_000 != e.TimeUS/1_000_000 {
		t.Errorf("created = %v, want the event time", got)
	}
}

func TestIngestedRevIsCommittedWithItsNotices(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		commentFrame("c1", "r2", ticketUri, "", otherDid),
	)

	rev, _, err := deldb.GetIngested(i.db, commentUri("c1"))
	if err != nil || rev != "r2" {
		t.Fatalf("comment ingested = %q/%v, want rev r2", rev, err)
	}
	if n := byType(t, i, ownerDid, models.NotificationTypeIssueCommented); n.AtUri != commentUri("c1") {
		t.Errorf("notice key = %q, want the comment uri the rev was stored under", n.AtUri)
	}
	if rev, cid, err := deldb.GetIngested(i.db, ticketUri); err != nil || rev != "r1" || cid != "cid-r1" {
		t.Errorf("ticket ingested = %q/%q/%v, want r1 and cid-r1", rev, cid, err)
	}
	if _, _, err := deldb.GetIngested(i.db, commentUri("never")); !errors.Is(err, sql.ErrNoRows) {
		t.Errorf("a record that was never processed answered %v, want sql.ErrNoRows", err)
	}
}

func TestErasedTicketTakesItsNoticesWithIt(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		commentFrame("c1", "r2", ticketUri, "", otherDid),
		ticketFrame(jetstream.CommitOperationUpdate, "r3", "closed", collabDid, ""),
	)
	if got := countFor(t, i, ownerDid); got != 3 {
		t.Fatalf("owner rows before the erase = %d, want created, commented and closed", got)
	}

	mustProcess(t, i, frame(repoDid, jetstream.CommitOperationDelete, ticketNSID, "1", "r4", ""))

	for _, did := range []string{ownerDid, editorDid} {
		if got := countFor(t, i, did); got != 0 {
			t.Errorf("%s still has %d notices about the erased ticket", did, got)
		}
	}
}

func TestErasedCommentTakesOnlyItsOwnNoticesWithIt(t *testing.T) {
	i := newTestIngester(t, fakeBobbin{})
	mustProcess(t, i,
		ticketFrame(jetstream.CommitOperationCreate, "r1", "open", editorDid, ""),
		commentFrame("c1", "r2", ticketUri, "", otherDid),
		commentFrame("c2", "r3", ticketUri, "", otherDid),
	)

	mustProcess(t, i, frame(repoDid, jetstream.CommitOperationDelete, commentNSID, "c1", "r4", ""))

	got := map[models.NotificationType]int{}
	for _, n := range notifsFor(t, i, ownerDid) {
		got[n.Type]++
		if n.AtUri == commentUri("c1") {
			t.Errorf("the erased comment's notice is still there")
		}
	}
	if got[models.NotificationTypeIssueCommented] != 1 || got[models.NotificationTypeIssueCreated] != 1 {
		t.Errorf("owner notices = %v, want the created one and the other comment's", got)
	}
}