package spindle
import (
"context"
"encoding/json"
"testing"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/jetstream"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/tapc"
)
func recordEvent(did syntax.DID, collection, rkey string, action tapc.RecordAction, record any) tapc.Event {
raw, _ := json.Marshal(record)
return tapc.Event{
Type: tapc.EvtRecord,
Record: &tapc.RecordEventData{
Live: true,
Did: did,
Collection: syntax.NSID(collection),
Rkey: syntax.RecordKey(rkey),
Action: action,
Record: raw,
},
}
}
func declarationEvent(owner, repo syntax.DID, action tapc.RecordAction) tapc.Event {
return recordEvent(owner, org_tangled.RepoDeclarationNSID, "repo", action, org_tangled.RepoDeclaration{Repo: repo.String()})
}
func processOK(t *testing.T, env *ensureEnv, evt tapc.Event) {
t.Helper()
if err := env.s.processEvent(context.Background(), evt); err != nil {
t.Fatal(err)
}
}
func TestProcessEventIgnoresOtherCollections(t *testing.T) {
env := newEnsureEnv(t)
for _, collection := range []string{org_tangled.FeedStarNSID, org_tangled.FeedCommentNSID, org_tangled.FeedReactionNSID, org_tangled.FeedSubscriptionNSID} {
evt := recordEvent("did:plc:someone", collection, "3mrhpypucbsg4", tapc.RecordCreateAction, map[string]string{"title": "x"})
if err := env.s.processEvent(context.Background(), evt); err != nil {
t.Fatalf("processEvent(%s): %v", collection, err)
}
}
if env.knot.count() != 0 {
t.Fatalf("knot was asked %d times for records spindle no longer reads", env.knot.count())
}
}
func TestManifestEventForAnUnknownRepoNeedsToNameThisSpindle(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
other := org_tangled.RepoManifest{Spindle: ptr("did:web:other.example.com")}
evt := recordEvent(testRepoDid, org_tangled.RepoManifestNSID, "self", tapc.RecordUpdateAction, other)
processOK(t, env, evt)
if env.knot.count() != 0 || len(env.rows(t)) != 0 {
t.Fatalf("a manifest for another spindle cost %d knot requests, rows %+v", env.knot.count(), env.rows(t))
}
processOK(t, env, markEvent(testRepoDid, env.spindleDid(), testOwner))
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("rows = %+v", rows)
}
}
func TestManifestEventForAKnownRepoReadsTheKnot(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
// the event body is empty on purpose, the knot's manifest is what counts
env.manifest("repo", "did:web:other.example.com")
evt := recordEvent(testRepoDid, org_tangled.RepoManifestNSID, "self", tapc.RecordUpdateAction, org_tangled.RepoManifest{})
processOK(t, env, evt)
if rows := env.rows(t); len(rows) != 0 {
t.Fatalf("moved repo kept, rows = %+v", rows)
}
}
func TestManifestDeleteForAKnownRepoForgetsIt(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
env.knot.erase(testRepoDid, org_tangled.RepoManifestNSID, "self")
evt := recordEvent(testRepoDid, org_tangled.RepoManifestNSID, "self", tapc.RecordDeleteAction, nil)
processOK(t, env, evt)
if rows := env.rows(t); len(rows) != 0 {
t.Fatalf("rows = %+v", rows)
}
}
func TestCollaboratorEventsOnlyMatterForKnownRepos(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo", testBob)
evt := recordEvent(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String(), tapc.RecordCreateAction, map[string]string{})
processOK(t, env, evt)
if env.knot.count() != 0 {
t.Fatalf("collaborator of an unknown repo cost %d knot requests", env.knot.count())
}
env.ensure(t)
env.knot.erase(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String())
evt = recordEvent(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String(), tapc.RecordDeleteAction, nil)
processOK(t, env, evt)
if env.collaborator(t, testBob) {
t.Error("bob kept access after his collaborator record went away")
}
}
func TestDeclarationEventsLeadToTheRepo(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
evt := declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction)
processOK(t, env, evt)
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("a declaration backfill didn't register the repo, rows = %+v", rows)
}
env.knot.undeclare(testOwner, "repo")
evt = recordEvent(testOwner, org_tangled.RepoDeclarationNSID, "repo", tapc.RecordDeleteAction, nil)
processOK(t, env, evt)
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("deleting the declaration forgot the repo, rows = %+v", rows)
}
}
func TestDeclarationDeleteForARowWithoutARepoDidIsSkipped(t *testing.T) {
env := newEnsureEnv(t)
if err := env.s.db.AddRepo(db.Repo{Knot: env.knot.host(), Owner: testOwner, Rkey: "repo"}); err != nil {
t.Fatal(err)
}
evt := recordEvent(testOwner, org_tangled.RepoDeclarationNSID, "repo", tapc.RecordDeleteAction, nil)
processOK(t, env, evt)
if env.knot.count() != 0 {
t.Fatalf("a legacy row cost %d knot requests", env.knot.count())
}
}
func TestDeclarationFromAnOutsiderCostsNothing(t *testing.T) {
env := newEnsureEnv(t)
env.s.cfg.Server.Admission = config.AdmissionClosed
env.serve("repo")
if err := db.AddToAllowlist(env.s.db, "did:plc:someoneelse"); err != nil {
t.Fatal(err)
}
evt := declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction)
processOK(t, env, evt)
if env.knot.count() != 0 {
t.Fatalf("an owner outside the allow-list cost %d knot requests", env.knot.count())
}
}
func TestDeclarationFromAnAdminIsServedWhileClosed(t *testing.T) {
env := newEnsureEnv(t)
env.s.cfg.Server.Admission = config.AdmissionClosed
env.s.cfg.Server.Admins = []string{testOwner.String()}
env.serve("repo")
evt := declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction)
processOK(t, env, evt)
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("an admin's declaration wasn't served, rows = %+v", rows)
}
}
func TestJetstreamToTapEventMarksRecordsLive(t *testing.T) {
tests := []struct {
name string
operation string
action tapc.RecordAction
}{
{name: "create", operation: jetstream.CommitOperationCreate, action: tapc.RecordCreateAction},
{name: "update", operation: jetstream.CommitOperationUpdate, action: tapc.RecordUpdateAction},
{name: "delete", operation: jetstream.CommitOperationDelete, action: tapc.RecordDeleteAction},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
event, ok := jetstreamToTapEvent(&jetstream.Event{
Did: "did:plc:jge3zxi7lgrfnvhzcgrimeo7",
Kind: jetstream.EventKindCommit,
Commit: &jetstream.Commit{
Operation: tt.operation,
Collection: org_tangled.RepoManifestNSID,
RKey: "self",
Record: json.RawMessage(`{"declaration":{}}`),
},
})
if !ok {
t.Fatal("jetstreamToTapEvent() rejected a valid event")
}
if event.Record == nil || !event.Record.Live {
t.Fatalf("converted event = %+v, want a live record", event.Record)
}
if event.Record.Collection.String() != org_tangled.RepoManifestNSID {
t.Errorf("collection = %q", event.Record.Collection)
}
if event.Record.Action != tt.action {
t.Errorf("action = %q, want %q", event.Record.Action, tt.action)
}
})
}
}
func manifestCommit(env *ensureEnv) *jetstream.Event {
raw, _ := json.Marshal(org_tangled.RepoManifest{Spindle: ptr(env.spindleDid()), Declaration: &org_tangled.RepoManifest_Declaration{Owner: testOwner.String(), Name: "repo"}})
return &jetstream.Event{
Did: testRepoDid.String(),
Kind: jetstream.EventKindCommit,
Commit: &jetstream.Commit{
Operation: jetstream.CommitOperationCreate,
Collection: org_tangled.RepoManifestNSID,
RKey: "self",
Record: raw,
},
}
}
func TestIngestRoutesJetstreamCommitsIntoEnsureRepo(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo", testBob)
if err := env.s.ingest()(context.Background(), manifestCommit(env)); err != nil {
t.Fatal(err)
}
if rows := env.rows(t); len(rows) != 1 || !env.collaborator(t, testBob) {
t.Fatalf("jetstream commit didn't register the repo, rows = %+v", rows)
}
}
func TestDeclarationArrivingAfterTheManifestEnrollsTheRepo(t *testing.T) {
env := newEnsureEnv(t)
env.manifest("repo", env.spindleDid())
processOK(t, env, markEvent(testRepoDid, env.spindleDid(), testOwner))
if rows := env.rows(t); len(rows) != 0 {
t.Fatalf("repo served before its owner declared it, rows = %+v", rows)
}
env.knot.declare(testOwner, "repo", testRepoDid)
processOK(t, env, declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction))
if rows := env.rows(t); len(rows) != 1 {
t.Fatalf("the declaration didn't enrol the repo, rows = %+v", rows)
}
}
func TestIngestReturnsAnEventCutByShutdownSoTheCursorStaysBehindIt(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo", testBob)
event := manifestCommit(env)
ctx, cancel := context.WithCancel(context.Background())
cancel()
if err := env.s.ingest()(ctx, event); err == nil {
t.Fatal("an event cut by shutdown was reported handled")
}
if rows := env.rows(t); len(rows) != 0 {
t.Fatalf("a cancelled event still registered the repo: %+v", rows)
}
if err := env.s.ingest()(context.Background(), event); err != nil {
t.Fatal(err)
}
}
func TestDeclarationFromANonOwnerDoesNotRefreshAServedRepo(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
before := env.knot.count()
evt := recordEvent("did:plc:newowner", org_tangled.RepoDeclarationNSID, "whatever", tapc.RecordCreateAction, org_tangled.RepoDeclaration{Repo: testRepoDid.String()})
processOK(t, env, evt)
if got := env.knot.count(); got != before {
t.Fatalf("a stranger's declaration cost %d knot requests", got-before)
}
evt = declarationEvent(testOwner, testRepoDid, tapc.RecordUpdateAction)
processOK(t, env, evt)
if env.knot.count() == before {
t.Error("the owner's own declaration no longer refreshes the repo")
}
}
func TestAMarkThatFailsOnTheDbIsReportedInsteadOfSwallowed(t *testing.T) {
env := newEnsureEnv(t)
env.serve("repo")
env.ensure(t)
if _, err := env.s.db.Exec(`drop table record_retries`); err != nil {
t.Fatal(err)
}
event := &jetstream.Event{Did: testRepoDid.String(), Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
Operation: jetstream.CommitOperationCreate, Collection: org_tangled.TrackTicketNSID, RKey: "3kticket00001", Record: []byte("{}"),
}}
if err := env.s.ingestEvent(withMarkOnly(context.Background()), event); err == nil {
t.Fatal("an event whose mark couldn't be written was reported as handled")
}
}