Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436package deliberi
import ( "context" "encoding/json" "io" "log/slog" "path/filepath" "testing"
comatprototypes "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" jmodels "github.com/bluesky-social/jetstream/pkg/models" "tangled.org/core/api/tangled" deldb "tangled.org/core/deliberi/db" models "tangled.org/core/deliberi/models")
type fakeResolver struct { dids []string // bySubject, when set, answers per subject instead of unconditionally. bySubject map[string][]string err error
// repo owner lookups, keyed by repo did. owners map[string]string repoNames map[string]string ownerErr error // ownerCalls counts RepoOwner calls, to prove cache hits skip the network. ownerCalls *int}
func (f fakeResolver) ListRecipients(ctx context.Context, uri string, collection string) ([]string, error) { if f.bySubject != nil { return f.bySubject[uri], f.err } return f.dids, f.err}
func (f fakeResolver) 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 starEvent(t *testing.T, actorDid, repoDid, rkey string) *jmodels.Event { t.Helper() raw, err := json.Marshal(tangled.FeedStar{ CreatedAt: "2026-01-01T00:00:00Z", Subject: &tangled.FeedStar_Subject{ FeedStar_Repo: &tangled.FeedStar_Repo{Did: repoDid}, }, }) if err != nil { t.Fatalf("marshal star: %v", err) } return &jmodels.Event{ Did: actorDid, Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ Operation: jmodels.CommitOperationCreate, Collection: tangled.FeedStarNSID, RKey: rkey, Record: raw, }, }}
func newTestIngester(t *testing.T, r recipientResolver) *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() }) return &Ingester{ db: database, recipients: r, logger: slog.New(slog.NewTextHandler(io.Discard, nil)), }}
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 TestNotifyEntitySubscriberGetsRow(t *testing.T) { i := newTestIngester(t, fakeResolver{dids: []string{"did:sub"}}) i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil, "") if got := countFor(t, i, "did:sub"); got != 1 { t.Fatalf("subscriber rows = %d, want 1", got) }}
func TestActorNeverNotified(t *testing.T) { i := newTestIngester(t, fakeResolver{dids: []string{"did:actor"}}) i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", []string{"did:actor"}, "") if got := countFor(t, i, "did:actor"); got != 0 { t.Fatalf("actor rows = %d, want 0", got) }}
func TestMentionDeliveredOnResolverError(t *testing.T) { i := newTestIngester(t, fakeResolver{err: io.ErrUnexpectedEOF}) i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", []string{"did:mention"}, "") if got := countFor(t, i, "did:mention"); got != 1 { t.Fatalf("mention rows = %d, want 1", got) }}
func TestMentionDeliveredAsUserMentioned(t *testing.T) { i := newTestIngester(t, fakeResolver{}) i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", []string{"did:mention"}, "")
var gotType string err := i.db.QueryRow(`select type from notifications where recipient_did = ?`, "did:mention").Scan(&gotType) if err != nil { t.Fatalf("scan notification: %v", err) } if gotType != string(models.NotificationTypeUserMentioned) { t.Errorf("type = %q, want %q", gotType, models.NotificationTypeUserMentioned) }}
func TestMentionTakesPriorityOverSubscription(t *testing.T) { // subscribed AND mentioned: one row, tagged as the mention. i := newTestIngester(t, fakeResolver{dids: []string{"did:both"}}) i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", []string{"did:both"}, "")
if got := countFor(t, i, "did:both"); got != 1 { t.Fatalf("rows = %d, want 1", got) } var gotType string err := i.db.QueryRow(`select type from notifications where recipient_did = ?`, "did:both").Scan(&gotType) if err != nil { t.Fatalf("scan notification: %v", err) } if gotType != string(models.NotificationTypeUserMentioned) { t.Errorf("type = %q, want %q", gotType, models.NotificationTypeUserMentioned) }}
func TestCreateNotificationDedupe(t *testing.T) { i := newTestIngester(t, fakeResolver{dids: []string{"did:sub"}}) i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil, "") i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil, "") if got := countFor(t, i, "did:sub"); got != 1 { t.Fatalf("deduped rows = %d, want 1", got) }}
func TestStarNotifiesRepoOwner(t *testing.T) { calls := 0 i := newTestIngester(t, fakeResolver{ownerCalls: &calls}) const repoDid = "did:plc:therepo" const ownerDid = "did:plc:bob" defer func() { if calls != 0 { t.Errorf("RepoOwner calls = %d, want 0 (cache hit must not hit bobbin)", calls) } }()
// Seed the repo DID → owner DID mapping (what the repo handler now does). if err := deldb.PutRepoName(i.db, repoDid, ownerDid, "my-repo"); err != nil { t.Fatalf("PutRepoName: %v", err) }
// Construct a firehose event for Alice starring Bob's repo. raw, err := json.Marshal(tangled.FeedStar{ CreatedAt: "2026-01-01T00:00:00Z", Subject: &tangled.FeedStar_Subject{ FeedStar_Repo: &tangled.FeedStar_Repo{Did: repoDid}, }, }) if err != nil { t.Fatalf("marshal star: %v", err) } ev := &jmodels.Event{ Did: "did:plc:alice", Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ Operation: jmodels.CommitOperationCreate, Collection: tangled.FeedStarNSID, RKey: "star1", Record: raw, }, }
if err := i.process(context.Background(), ev); err != nil { t.Fatalf("process: %v", err) }
if got := countFor(t, i, ownerDid); got != 1 { t.Fatalf("owner rows = %d, want 1", got) } if got := countFor(t, i, repoDid); got != 0 { t.Fatalf("repo DID rows = %d, want 0 (notification must go to owner, not repo)", got) } if got := countFor(t, i, "did:plc:alice"); got != 0 { t.Fatalf("actor rows = %d, want 0 (star author must not self-notify)", got) }}
func TestStarResolvesOwnerFromBobbinOnCacheMiss(t *testing.T) { const repoDid = "did:plc:therepo" const ownerDid = "did:plc:bob"
// nothing cached: the owner must come from bobbin. i := newTestIngester(t, fakeResolver{ owners: map[string]string{repoDid: ownerDid}, repoNames: map[string]string{repoDid: "my-repo"}, })
if err := i.process(context.Background(), starEvent(t, "did:plc:alice", repoDid, "star1")); err != nil { t.Fatalf("process: %v", err) }
if got := countFor(t, i, ownerDid); got != 1 { t.Fatalf("owner rows = %d, want 1", got) } // the lookup must be cached, so the next star on this repo is a local hit. if got := deldb.GetRepoOwner(i.db, repoDid); got != ownerDid { t.Errorf("cached owner = %q, want %q", got, ownerDid) } if got := deldb.GetRepoName(i.db, repoDid); got != "my-repo" { t.Errorf("cached name = %q, want %q", got, "my-repo") }}
func TestStarKeepsCachedNameWhenRecordIsNameless(t *testing.T) { const repoDid = "did:plc:therepo" const ownerDid = "did:plc:bob"
// a row with a name but no owner, as migrated rows can have. i := newTestIngester(t, fakeResolver{owners: map[string]string{repoDid: ownerDid}}) if err := deldb.PutRepoName(i.db, repoDid, "", "my-repo"); err != nil { t.Fatalf("PutRepoName: %v", err) }
if err := i.process(context.Background(), starEvent(t, "did:plc:alice", repoDid, "star1")); err != nil { t.Fatalf("process: %v", err) }
if got := deldb.GetRepoName(i.db, repoDid); got != "my-repo" { t.Errorf("cached name = %q, want %q (nameless record must not blank it)", got, "my-repo") }}
func TestStarSkipsWhenOwnerUnresolvable(t *testing.T) { // nothing cached and bobbin is unreachable, so the star handler should skip. i := newTestIngester(t, fakeResolver{ownerErr: io.ErrUnexpectedEOF}) const repoDid = "did:plc:therepo"
if err := i.process(context.Background(), starEvent(t, "did:plc:alice", repoDid, "star1")); err != nil { t.Fatalf("process: %v", err) }
if got := countFor(t, i, "did:plc:bob"); got != 0 { t.Fatalf("bob rows = %d, want 0 (unresolvable owner => skip)", got) }}
func TestRepoHandlerSeedsOwner(t *testing.T) { // Verify that the ingester's repo processing stores the correct mapping: // repo_did → owner_did and name. // This test exercises the same code path as the repo NSID case in process(). db, err := deldb.Make(context.Background(), filepath.Join(t.TempDir(), "x.db")) if err != nil { t.Fatalf("make db: %v", err) } t.Cleanup(func() { db.Close() })
const ownerDid = "did:plc:owner" const repoDid = "did:plc:therepo" const repoName = "my-repo"
if err := deldb.PutRepoName(db, repoDid, ownerDid, repoName); err != nil { t.Fatalf("PutRepoName: %v", err) }
if got := deldb.GetRepoOwner(db, repoDid); got != ownerDid { t.Fatalf("GetRepoOwner = %q, want %q", got, ownerDid) } if got := deldb.GetRepoName(db, repoDid); got != repoName { t.Fatalf("GetRepoName = %q, want %q", got, repoName) } // Owner DID alone should NOT resolve as a repo name (it's not the key). if got := deldb.GetRepoName(db, ownerDid); got != "" { t.Fatalf("GetRepoName(ownerDid) = %q, want empty (owner is not the cache key)", got) }}
func TestCommentNotifiesRepoSubscriber(t *testing.T) { const ( repoDid = "did:plc:therepo" repoName = "my-repo" issueUri = "at://did:plc:bob/sh.tangled.repo.issue/issue1" )
// subscribed to the repo, not the issue, so a row proves the repo lookup ran. i := newTestIngester(t, fakeResolver{bySubject: map[string][]string{ repoDid: {"did:sub"}, }})
if err := deldb.PutRepoName(i.db, repoDid, "did:plc:bob", repoName); err != nil { t.Fatalf("PutRepoName: %v", err) } if err := deldb.PutEntityTitle(i.db, issueUri, "the issue", repoDid); err != nil { t.Fatalf("PutEntityTitle: %v", err) }
raw, err := json.Marshal(tangled.FeedComment{ CreatedAt: "2026-01-01T00:00:00Z", Subject: &comatprototypes.RepoStrongRef{Uri: issueUri, Cid: "bafyfake"}, }) if err != nil { t.Fatalf("marshal comment: %v", err) } ev := &jmodels.Event{ Did: "did:plc:alice", Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ Operation: jmodels.CommitOperationCreate, Collection: tangled.FeedCommentNSID, RKey: "comment1", Record: raw, }, } if err := i.process(context.Background(), ev); err != nil { t.Fatalf("process: %v", err) }
if got := countFor(t, i, "did:sub"); got != 1 { t.Fatalf("repo subscriber rows = %d, want 1", got) }
var gotRepoDid, gotTitle string err = i.db.QueryRow( `select repo_did, entity_title from notifications where recipient_did = ?`, "did:sub", ).Scan(&gotRepoDid, &gotTitle) if err != nil { t.Fatalf("scan notification: %v", err) } if gotRepoDid != repoDid { t.Errorf("repo_did = %q, want %q", gotRepoDid, repoDid) } if gotTitle != "the issue" { t.Errorf("entity_title = %q, want %q", gotTitle, "the issue") }}
func TestCommentNotifiesEntitySubscriber(t *testing.T) { const ( repoDid = "did:plc:therepo" issueUri = "at://did:plc:bob/sh.tangled.repo.issue/issue1" )
i := newTestIngester(t, fakeResolver{bySubject: map[string][]string{ issueUri: {"did:entitysub"}, }}) if err := deldb.PutEntityTitle(i.db, issueUri, "the issue", repoDid); err != nil { t.Fatalf("PutEntityTitle: %v", err) }
raw, err := json.Marshal(tangled.FeedComment{ CreatedAt: "2026-01-01T00:00:00Z", Subject: &comatprototypes.RepoStrongRef{Uri: issueUri, Cid: "bafyfake"}, }) if err != nil { t.Fatalf("marshal comment: %v", err) } ev := &jmodels.Event{ Did: "did:plc:alice", Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ Operation: jmodels.CommitOperationCreate, Collection: tangled.FeedCommentNSID, RKey: "comment1", Record: raw, }, } if err := i.process(context.Background(), ev); err != nil { t.Fatalf("process: %v", err) }
if got := countFor(t, i, "did:entitysub"); got != 1 { t.Fatalf("entity subscriber rows = %d, want 1", got) }}
func TestPutEntityTitleKeepsRepoDid(t *testing.T) { 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() })
const uri = "at://did:plc:bob/sh.tangled.repo.issue/issue1" if err := deldb.PutEntityTitle(database, uri, "t1", "did:plc:therepo"); err != nil { t.Fatalf("first put: %v", err) } // a later write that does not know the repo must not erase it. if err := deldb.PutEntityTitle(database, uri, "t2", ""); err != nil { t.Fatalf("second put: %v", err) } if got := deldb.GetEntityRepo(database, uri); got != "did:plc:therepo" { t.Errorf("repo did = %q, want it preserved", got) } if got := deldb.GetEntityTitle(database, uri); got != "t2" { t.Errorf("title = %q, want %q", got, "t2") }}
func TestDisabledPrefSuppressesRow(t *testing.T) { i := newTestIngester(t, fakeResolver{dids: []string{"did:sub"}}) prefs := models.DefaultNotificationPreferences(syntax.DID("did:sub")) prefs.IssueCreated = false if err := deldb.UpsertNotificationPreferences(i.db, prefs); err != nil { t.Fatalf("upsert prefs: %v", err) } i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil, "") if got := countFor(t, i, "did:sub"); got != 0 { t.Fatalf("disabled-pref rows = %d, want 0", got) }}