+8

internal/firehose/index.go

reviewed

··· 1171 1171 return profile, nil 1172 1172 } 1173 1173 1174 1174 + // StoreProfile writes a profile to both in-memory and persistent caches and 1175 1175 + // maintains the did_by_handle index. Use this when you've already fetched a 1176 1176 + // profile (backfill workers, tests, externally-provided data) and want to seed 1177 1177 + // the cache without going through the public API. 1178 1178 + func (idx *FeedIndex) StoreProfile(ctx context.Context, did string, profile *atproto.Profile) { 1179 1179 + idx.storeProfile(ctx, did, profile) 1180 1180 + } 1181 1181 + 1174 1182 // storeProfile writes a profile to both in-memory and persistent caches, and 1175 1183 // maintains the did_by_handle index so handle lookups stay accurate across 1176 1184 // handle changes and handle reassignment between DIDs.

+221

tests/integration/handle_resolution_test.go

reviewed

··· 1 1 + package integration 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "testing" 6 6 + "time" 7 7 + 8 8 + "tangled.org/arabica.social/arabica/internal/atproto" 9 9 + "tangled.org/arabica.social/arabica/internal/firehose" 10 10 + 11 11 + "github.com/stretchr/testify/assert" 12 12 + "github.com/stretchr/testify/require" 13 13 + ) 14 14 + 15 15 + // These tests cover the bug where two accounts share a handle (because the 16 16 + // first account's PDS went away and the user re-registered with a new DID) 17 17 + // and the witness cache could resolve the handle to the orphan DID. The fix 18 18 + // keys handle resolution off a `did_by_handle` table with last-writer-wins 19 19 + // semantics, plus identity-event-driven invalidation of stale prior owners. 20 20 + 21 21 + const sharedHandle = "shared.test" 22 22 + 23 23 + // seedProfile writes a profile cache entry for did/handle without going 24 24 + // through the public API. Mirrors what the profile watcher would do after 25 25 + // observing a profile commit on the firehose. 26 26 + func seedProfile(t *testing.T, h *Harness, did, handle string) { 27 27 + t.Helper() 28 28 + h.FeedIndex.StoreProfile(context.Background(), did, &atproto.Profile{ 29 29 + DID: did, 30 30 + Handle: handle, 31 31 + }) 32 32 + } 33 33 + 34 34 + // TestHandleReassignment_NewOwnerWinsLookup is the direct repro of the bug. 35 35 + // Two DIDs claim the same handle in sequence; the lookup must return the 36 36 + // most-recently-written owner. 37 37 + func TestHandleReassignment_NewOwnerWinsLookup(t *testing.T) { 38 38 + h := StartHarness(t, nil) 39 39 + ctx := context.Background() 40 40 + 41 41 + const orphanDID = "did:plc:orphanaccount" 42 42 + const newDID = "did:plc:freshaccount" 43 43 + 44 44 + seedProfile(t, h, orphanDID, sharedHandle) 45 45 + 46 46 + got, ok := h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 47 47 + require.True(t, ok, "lookup should find the seeded profile") 48 48 + assert.Equal(t, orphanDID, got) 49 49 + 50 50 + // The new account claims the same handle. After its profile is observed, 51 51 + // the handle must resolve to the new DID, not the orphan. 52 52 + seedProfile(t, h, newDID, sharedHandle) 53 53 + 54 54 + got, ok = h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 55 55 + require.True(t, ok) 56 56 + assert.Equal(t, newDID, got, "handle should resolve to the most recent owner, not the orphan") 57 57 + } 58 58 + 59 59 + // TestHandleChange_OldHandleStopsResolving covers a single DID changing its 60 60 + // handle (no reassignment between DIDs). The stale handle row must be cleared 61 61 + // so the old handle no longer resolves. 62 62 + func TestHandleChange_OldHandleStopsResolving(t *testing.T) { 63 63 + h := StartHarness(t, nil) 64 64 + ctx := context.Background() 65 65 + 66 66 + const did = "did:plc:movinghandles" 67 67 + 68 68 + seedProfile(t, h, did, "old.test") 69 69 + 70 70 + got, ok := h.FeedIndex.GetDIDByHandle(ctx, "old.test") 71 71 + require.True(t, ok) 72 72 + assert.Equal(t, did, got) 73 73 + 74 74 + seedProfile(t, h, did, "new.test") 75 75 + 76 76 + _, ok = h.FeedIndex.GetDIDByHandle(ctx, "old.test") 77 77 + assert.False(t, ok, "old handle should no longer resolve after the DID changes its handle") 78 78 + 79 79 + got, ok = h.FeedIndex.GetDIDByHandle(ctx, "new.test") 80 80 + require.True(t, ok) 81 81 + assert.Equal(t, did, got) 82 82 + } 83 83 + 84 84 + // TestPurgeDID_FreesHandleForReuse reproduces the cleanup path: an orphan DID 85 85 + // is purged, then a new account claiming the same handle resolves cleanly. 86 86 + // The admin /_mod/purge endpoint isn't reachable from the harness (no 87 87 + // moderation service), so we exercise the same DeleteAllByDID code path it 88 88 + // invokes and assert the same downstream invariants. 89 89 + func TestPurgeDID_FreesHandleForReuse(t *testing.T) { 90 90 + h := StartHarness(t, nil) 91 91 + ctx := context.Background() 92 92 + 93 93 + const orphanDID = "did:plc:orphandead" 94 94 + const newDID = "did:plc:newowner" 95 95 + 96 96 + seedProfile(t, h, orphanDID, sharedHandle) 97 97 + 98 98 + // The orphan also has indexed records — purge must remove them. 99 99 + const collection = atproto.NSIDRoaster 100 100 + const rkey = "orphanrkey" 101 101 + now := time.Now().Unix() 102 102 + require.NoError(t, h.FeedIndex.UpsertRecord( 103 103 + ctx, orphanDID, collection, rkey, "ciddead", 104 104 + []byte(`{"$type":"social.arabica.alpha.roaster","name":"Orphan Roaster","createdAt":"2025-01-01T00:00:00Z"}`), 105 105 + now, 106 106 + )) 107 107 + 108 108 + uri := atproto.BuildATURI(orphanDID, collection, rkey) 109 109 + rec, err := h.FeedIndex.GetRecord(ctx, uri) 110 110 + require.NoError(t, err) 111 111 + require.NotNil(t, rec, "orphan record should be indexed before purge") 112 112 + 113 113 + require.NoError(t, h.FeedIndex.DeleteAllByDID(ctx, orphanDID)) 114 114 + h.FeedIndex.InvalidatePublicCachesForDID(orphanDID) 115 115 + 116 116 + rec, err = h.FeedIndex.GetRecord(ctx, uri) 117 117 + assert.NoError(t, err) 118 118 + assert.Nil(t, rec, "orphan record should be gone after purge") 119 119 + 120 120 + _, ok := h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 121 121 + assert.False(t, ok, "purge should remove the handle index entry") 122 122 + 123 123 + // New account claims the freed handle. 124 124 + seedProfile(t, h, newDID, sharedHandle) 125 125 + got, ok := h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 126 126 + require.True(t, ok) 127 127 + assert.Equal(t, newDID, got, "freed handle should resolve to the new owner") 128 128 + } 129 129 + 130 130 + // TestIdentityEvent_EvictsPriorOwnerOfHandle drives the OnIdentityEvent 131 131 + // reconciliation path through the profile watcher. When a Jetstream identity 132 132 + // event reports that a new DID has claimed an existing handle, the prior 133 133 + // owner's profile cache and handle mapping must be invalidated immediately — 134 134 + // before the eventual API refresh happens — so a concurrent lookup of the 135 135 + // shared handle never returns the stale owner. 136 136 + func TestIdentityEvent_EvictsPriorOwnerOfHandle(t *testing.T) { 137 137 + h := StartHarness(t, &HarnessOptions{EnableFirehose: true}) 138 138 + ctx := context.Background() 139 139 + 140 140 + const orphanDID = "did:plc:identityorphan" 141 141 + const newDID = "did:plc:identitynew" 142 142 + 143 143 + seedProfile(t, h, orphanDID, sharedHandle) 144 144 + got, ok := h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 145 145 + require.True(t, ok) 146 146 + require.Equal(t, orphanDID, got, "precondition: orphan owns the handle") 147 147 + 148 148 + // Synthesize an identity event for the new DID claiming the shared handle. 149 149 + // OnIdentityEvent runs prior-owner eviction synchronously; the subsequent 150 150 + // RefreshProfile call hits the public bsky API and fails for our synthetic 151 151 + // DID, which is fine — we're verifying the eviction half of the path. 152 152 + require.NotNil(t, h.ProfileWatcher) 153 153 + h.ProfileWatcher.ProcessEvent(firehose.JetstreamEvent{ 154 154 + DID: newDID, 155 155 + TimeUS: time.Now().UnixMicro(), 156 156 + Kind: "identity", 157 157 + Identity: &firehose.JetstreamIdentity{ 158 158 + DID: newDID, 159 159 + Handle: sharedHandle, 160 160 + Time: time.Now().Format(time.RFC3339), 161 161 + }, 162 162 + }) 163 163 + 164 164 + // The orphan must no longer be the owner of the shared handle. Either the 165 165 + // lookup misses entirely (because RefreshProfile failed and InvalidateProfile 166 166 + // cleared everything) or it resolves to newDID — both are acceptable; what 167 167 + // must not happen is the stale orphan winning. 168 168 + got, ok = h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 169 169 + if ok { 170 170 + assert.NotEqual(t, orphanDID, got, "shared handle must not resolve to the evicted prior owner") 171 171 + } 172 172 + 173 173 + // Now seed the new DID's profile (as the watcher would, on a working API). 174 174 + // The handle must resolve to newDID, never orphanDID. 175 175 + seedProfile(t, h, newDID, sharedHandle) 176 176 + got, ok = h.FeedIndex.GetDIDByHandle(ctx, sharedHandle) 177 177 + require.True(t, ok) 178 178 + assert.Equal(t, newDID, got, "after reconciliation, shared handle resolves to the new owner") 179 179 + } 180 180 + 181 181 + // TestHandleResolution_BackfillFromExistingProfiles verifies that a FeedIndex 182 182 + // opened against a database with pre-existing profile rows but no 183 183 + // did_by_handle table populates the index on startup, so handle lookups work 184 184 + // for users observed before this fix shipped. 185 185 + func TestHandleResolution_BackfillFromExistingProfiles(t *testing.T) { 186 186 + tmpDir := t.TempDir() 187 187 + dbPath := tmpDir + "/feed-backfill.db" 188 188 + 189 189 + // First open: seed two profiles, then close. 190 190 + idx, err := firehose.NewFeedIndex(dbPath, time.Hour) 191 191 + require.NoError(t, err) 192 192 + idx.StoreProfile(context.Background(), "did:plc:backfilla", &atproto.Profile{ 193 193 + DID: "did:plc:backfilla", Handle: "alice.bf", 194 194 + }) 195 195 + idx.StoreProfile(context.Background(), "did:plc:backfillb", &atproto.Profile{ 196 196 + DID: "did:plc:backfillb", Handle: "bob.bf", 197 197 + }) 198 198 + require.NoError(t, idx.Close()) 199 199 + 200 200 + // Drop did_by_handle to simulate a database that predates the index. 201 201 + idx2, err := firehose.NewFeedIndex(dbPath, time.Hour) 202 202 + require.NoError(t, err) 203 203 + defer idx2.Close() 204 204 + 205 205 + _, err = idx2.DB().Exec(`DELETE FROM did_by_handle`) 206 206 + require.NoError(t, err) 207 207 + require.NoError(t, idx2.Close()) 208 208 + 209 209 + // Reopen — backfill should re-populate the table from the profile rows. 210 210 + idx3, err := firehose.NewFeedIndex(dbPath, time.Hour) 211 211 + require.NoError(t, err) 212 212 + defer idx3.Close() 213 213 + 214 214 + got, ok := idx3.GetDIDByHandle(context.Background(), "alice.bf") 215 215 + require.True(t, ok, "backfill should restore alice's handle mapping") 216 216 + assert.Equal(t, "did:plc:backfilla", got) 217 217 + 218 218 + got, ok = idx3.GetDIDByHandle(context.Background(), "bob.bf") 219 219 + require.True(t, ok, "backfill should restore bob's handle mapping") 220 220 + assert.Equal(t, "did:plc:backfillb", got) 221 221 + }