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)
}
}