Skip to content

Commit 8bc7ca5

Browse files
morlutomorluto
committed
fix(actors): preserve identity and snapshot integrity
1 parent b1a082e commit 8bc7ca5

8 files changed

Lines changed: 450 additions & 23 deletions

File tree

‎internal/app/mcp_actor_facets.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -426,6 +426,9 @@ func (r *MCPReader) SearchContributions(ctx context.Context, in mcpcontract.Sear
426426
}
427427
out.Coverage = append(out.Coverage, mcpcontract.ActorContributionCoverage{ActorID: actor.Key, Facet: coverage})
428428
}
429+
if err := finishCorpusRead(ctx, c, revision); err != nil {
430+
return mcpcontract.SearchContributionsOutput{}, err
431+
}
429432
return out, nil
430433
}
431434

‎internal/app/mcp_actors.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,9 @@ func (r *MCPReader) SearchActors(ctx context.Context, in mcpcontract.SearchActor
233233
for _, actor := range page.Actors {
234234
out.Items = append(out.Items, actorOutput(actor))
235235
}
236+
if err := finishCorpusRead(ctx, c, revision); err != nil {
237+
return mcpcontract.SearchActorsOutput{}, err
238+
}
236239
return out, nil
237240
}
238241

@@ -267,6 +270,9 @@ func (r *MCPReader) GetActors(ctx context.Context, in mcpcontract.GetActorsInput
267270
}
268271
out.Items[index] = item
269272
}
273+
if err := finishCorpusRead(ctx, c, revision); err != nil {
274+
return mcpcontract.GetActorsOutput{}, err
275+
}
270276
return out, nil
271277
}
272278

@@ -340,6 +346,9 @@ func (r *MCPReader) GetActorFacets(ctx context.Context, in mcpcontract.GetActorF
340346
item.Value = &value
341347
out.Items[index] = item
342348
}
349+
if err := finishCorpusRead(ctx, c, revision); err != nil {
350+
return mcpcontract.GetActorFacetsOutput{}, err
351+
}
343352
return out, nil
344353
}
345354

‎internal/corpus/actor_facets.go‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -163,12 +163,13 @@ func (c *Corpus) ApplyActorContributionPeriod(ctx context.Context, input ActorCo
163163
if _, err := tx.ExecContext(ctx, `INSERT INTO actor_observations(actor_id,facet,source_updated_at,observation_sequence,observed_at,complete,authorization_scope,payload) VALUES(?, 'contributions', ?, ?, ?, ?, ?, ?)`, input.ActorID, encodeTime(input.SourceUpdatedAt), sequence, encodeTime(input.ObservedAt), boolToInt(input.Complete), input.AuthorizationScope, string(payload)); err != nil {
164164
return err
165165
}
166+
var existingComplete bool
166167
var existingSource, existingSequence int64
167-
err = tx.QueryRowContext(ctx, `SELECT source_updated_at,observation_sequence FROM actor_contribution_periods WHERE actor_id=? AND period_start=? AND period_end=? AND organization_node_id=? AND authorization_scope=?`, input.ActorID, encodeTime(input.From), encodeTime(input.To), input.OrganizationNodeID, input.AuthorizationScope).Scan(&existingSource, &existingSequence)
168+
err = tx.QueryRowContext(ctx, `SELECT complete,source_updated_at,observation_sequence FROM actor_contribution_periods WHERE actor_id=? AND period_start=? AND period_end=? AND organization_node_id=? AND authorization_scope=?`, input.ActorID, encodeTime(input.From), encodeTime(input.To), input.OrganizationNodeID, input.AuthorizationScope).Scan(&existingComplete, &existingSource, &existingSequence)
168169
if err != nil && !errors.Is(err, sql.ErrNoRows) {
169170
return err
170171
}
171-
if err == nil && !orderingNewer(encodeTime(input.SourceUpdatedAt), sequence, existingSource, existingSequence) {
172+
if err == nil && (!orderingNewer(encodeTime(input.SourceUpdatedAt), sequence, existingSource, existingSequence) || (!input.Complete && existingComplete)) {
172173
return tx.Commit()
173174
}
174175
var periodID int64
@@ -303,7 +304,7 @@ func (c *Corpus) SearchActorContributions(ctx context.Context, opts Contribution
303304
for i := range opts.ActorRefs {
304305
placeholders[i] = "?"
305306
}
306-
where += ` AND (a.actor_key IN (` + strings.Join(placeholders, ",") + `) OR a.node_id IN (` + strings.Join(placeholders, ",") + `) OR a.id IN (SELECT actor_id FROM actor_aliases WHERE normalized_login IN (` + strings.Join(placeholders, ",") + `)))`
307+
where += ` AND (a.actor_key IN (` + strings.Join(placeholders, ",") + `) OR a.node_id IN (` + strings.Join(placeholders, ",") + `) OR a.id IN (SELECT actor_id FROM actor_aliases WHERE active=1 AND normalized_login IN (` + strings.Join(placeholders, ",") + `)))`
307308
for _, ref := range opts.ActorRefs {
308309
args = append(args, strings.TrimSpace(ref))
309310
}

‎internal/corpus/actors.go‎

Lines changed: 46 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -203,11 +203,7 @@ func (c *Corpus) ApplyActorIdentityObservation(ctx context.Context, provider, lo
203203
`, actorKey(provider, nodeID, login), nodeID, databaseID, kind, login, sequence, encodeTime(observedAt), actorID); err != nil {
204204
return Actor{}, fmt.Errorf("advance actor identity: %w", err)
205205
}
206-
if _, err := tx.ExecContext(ctx, `
207-
INSERT INTO actor_aliases (actor_id, login, normalized_login, active, first_observed_at, last_observed_at)
208-
VALUES (?, ?, ?, 1, ?, ?)
209-
ON CONFLICT(actor_id, normalized_login) DO UPDATE SET login=excluded.login, active=1, last_observed_at=excluded.last_observed_at
210-
`, actorID, login, normalizeLogin(login), encodeTime(observedAt), encodeTime(observedAt)); err != nil {
206+
if err := activateActorAlias(ctx, tx, actorID, provider, login, nodeID, encodeTime(observedAt)); err != nil {
211207
return Actor{}, fmt.Errorf("upsert actor identity alias: %w", err)
212208
}
213209
if err := refreshActorFTS(ctx, tx, actorID); err != nil {
@@ -313,12 +309,7 @@ func (c *Corpus) ApplyActorProfileObservation(ctx context.Context, input ActorPr
313309
encodeTime(profile.ObservedAt), profile.AuthorizationScope); err != nil {
314310
return Actor{}, fmt.Errorf("advance actor profile: %w", err)
315311
}
316-
if _, err := tx.ExecContext(ctx, `
317-
INSERT INTO actor_aliases (actor_id, login, normalized_login, active, first_observed_at, last_observed_at)
318-
VALUES (?, ?, ?, 1, ?, ?)
319-
ON CONFLICT(actor_id, normalized_login) DO UPDATE SET
320-
login=excluded.login, active=1, last_observed_at=excluded.last_observed_at
321-
`, actorID, input.Login, normalizeLogin(input.Login), encodeTime(input.ObservedAt), encodeTime(input.ObservedAt)); err != nil {
312+
if err := activateActorAlias(ctx, tx, actorID, input.Provider, input.Login, input.NodeID, encodeTime(input.ObservedAt)); err != nil {
322313
return Actor{}, fmt.Errorf("upsert actor alias: %w", err)
323314
}
324315
if err := refreshActorFTS(ctx, tx, actorID); err != nil {
@@ -348,11 +339,14 @@ func resolveActorID(ctx context.Context, tx *sql.Tx, input ActorProfileObservati
348339
return 0, fmt.Errorf("resolve actor node id: %w", err)
349340
}
350341
}
351-
err := tx.QueryRowContext(ctx, `
342+
aliasQuery := `
352343
SELECT a.id FROM actor_aliases aa JOIN actors a ON a.id=aa.actor_id
353-
WHERE a.provider=? AND aa.normalized_login=? AND aa.active=1
354-
ORDER BY aa.last_observed_at DESC, a.id DESC LIMIT 1
355-
`, input.Provider, normalizeLogin(input.Login)).Scan(&id)
344+
WHERE a.provider=? AND aa.normalized_login=? AND aa.active=1`
345+
if input.NodeID != "" {
346+
aliasQuery += ` AND (a.node_id IS NULL OR a.node_id='')`
347+
}
348+
aliasQuery += ` ORDER BY aa.last_observed_at DESC, a.id DESC LIMIT 1`
349+
err := tx.QueryRowContext(ctx, aliasQuery, input.Provider, normalizeLogin(input.Login)).Scan(&id)
356350
if err == nil {
357351
return id, nil
358352
}
@@ -375,6 +369,43 @@ func resolveActorID(ctx context.Context, tx *sql.Tx, input ActorProfileObservati
375369
return id, nil
376370
}
377371

372+
// activateActorAlias keeps a reused login historical on the old node-backed
373+
// identity while making the newly observed node-backed identity current.
374+
func activateActorAlias(ctx context.Context, tx *sql.Tx, actorID int64, provider, login, nodeID string, observedAt int64) error {
375+
normalized := normalizeLogin(login)
376+
active := true
377+
if nodeID != "" {
378+
var currentObservedAt int64
379+
err := tx.QueryRowContext(ctx, `
380+
SELECT aa.last_observed_at
381+
FROM actor_aliases aa JOIN actors a ON a.id=aa.actor_id
382+
WHERE aa.normalized_login=? AND aa.actor_id<>? AND aa.active=1 AND a.provider=?
383+
ORDER BY aa.last_observed_at DESC, aa.actor_id DESC LIMIT 1
384+
`, normalized, actorID, provider).Scan(&currentObservedAt)
385+
if err != nil && !errors.Is(err, sql.ErrNoRows) {
386+
return err
387+
}
388+
active = errors.Is(err, sql.ErrNoRows) || observedAt >= currentObservedAt
389+
if active {
390+
if _, err := tx.ExecContext(ctx, `
391+
UPDATE actor_aliases SET active=0
392+
WHERE normalized_login=? AND actor_id<>?
393+
AND actor_id IN (SELECT id FROM actors WHERE provider=?)
394+
`, normalized, actorID, provider); err != nil {
395+
return err
396+
}
397+
}
398+
}
399+
_, err := tx.ExecContext(ctx, `
400+
INSERT INTO actor_aliases (actor_id, login, normalized_login, active, first_observed_at, last_observed_at)
401+
VALUES (?, ?, ?, ?, ?, ?)
402+
ON CONFLICT(actor_id, normalized_login) DO UPDATE SET
403+
login=excluded.login, active=excluded.active,
404+
last_observed_at=MAX(actor_aliases.last_observed_at, excluded.last_observed_at)
405+
`, actorID, login, normalized, boolToInt(active), observedAt, observedAt)
406+
return err
407+
}
408+
378409
func actorKey(provider, nodeID, login string) string {
379410
if nodeID != "" {
380411
return provider + ":node:" + nodeID

‎internal/corpus/actors_test.go‎

Lines changed: 83 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,79 @@ func TestActorProfileObservationReconcilesLoginToNodeIDAndPreservesNewerProjecti
5858
}
5959
}
6060

61+
func TestActorObservationDoesNotMergeReusedLoginAcrossNodeIDs(t *testing.T) {
62+
t.Parallel()
63+
ctx := context.Background()
64+
c, _ := openTestCorpus(t)
65+
otherProvider, err := c.ApplyActorIdentityObservation(ctx, "gitlab", "mona", "GL_1", nil, "user", "public", time.Unix(1, 0).UTC(), nil)
66+
if err != nil {
67+
t.Fatal(err)
68+
}
69+
first, err := c.ApplyActorIdentityObservation(ctx, "github", "mona", "U_1", nil, "user", "public", time.Unix(1, 0).UTC(), nil)
70+
if err != nil {
71+
t.Fatal(err)
72+
}
73+
second, err := c.ApplyActorIdentityObservation(ctx, "github", "mona", "U_2", nil, "user", "public", time.Unix(2, 0).UTC(), nil)
74+
if err != nil {
75+
t.Fatal(err)
76+
}
77+
if first.ID == second.ID || first.NodeID != "U_1" || second.NodeID != "U_2" {
78+
t.Fatalf("reused login identities merged: first=%+v second=%+v", first, second)
79+
}
80+
byFirstNode, err := c.GetActor(ctx, "U_1")
81+
if err != nil {
82+
t.Fatal(err)
83+
}
84+
byLogin, err := c.GetActor(ctx, "mona")
85+
if err != nil {
86+
t.Fatal(err)
87+
}
88+
if byFirstNode == nil || byFirstNode.ID != first.ID || byLogin == nil || byLogin.ID != second.ID {
89+
t.Fatalf("reused login lookup: first node=%+v current login=%+v", byFirstNode, byLogin)
90+
}
91+
if _, err := c.ApplyActorIdentityObservation(ctx, "github", "mona", "U_1", nil, "user", "public", time.Unix(1, 0).UTC(), nil); err != nil {
92+
t.Fatal(err)
93+
}
94+
byLogin, err = c.GetActor(ctx, "mona")
95+
if err != nil {
96+
t.Fatal(err)
97+
}
98+
if byLogin == nil || byLogin.ID != second.ID {
99+
t.Fatalf("delayed older observation reclaimed reused login: %+v", byLogin)
100+
}
101+
var otherProviderAliasActive bool
102+
if err := c.db.QueryRowContext(ctx, `SELECT active FROM actor_aliases WHERE actor_id=? AND normalized_login='mona'`, otherProvider.ID).Scan(&otherProviderAliasActive); err != nil {
103+
t.Fatal(err)
104+
}
105+
if !otherProviderAliasActive {
106+
t.Fatal("reusing a GitHub login deactivated the same alias for another provider")
107+
}
108+
}
109+
110+
func TestIncompleteContributionPeriodMaterializesUntilCompleteSnapshotExists(t *testing.T) {
111+
t.Parallel()
112+
ctx := context.Background()
113+
c, _ := openTestCorpus(t)
114+
actor, err := c.ApplyActorIdentityObservation(ctx, "github", "alice", "U_alice", nil, "user", "public", time.Unix(1, 0).UTC(), nil)
115+
if err != nil {
116+
t.Fatal(err)
117+
}
118+
from := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
119+
if err := c.ApplyActorContributionPeriod(ctx, ActorContributionPeriodInput{
120+
ActorID: actor.ID, From: from, To: from.Add(24 * time.Hour), Complete: false,
121+
ObservedAt: from.Add(25 * time.Hour), SourceUpdatedAt: from.Add(24 * time.Hour),
122+
}); err != nil {
123+
t.Fatal(err)
124+
}
125+
var complete bool
126+
if err := c.db.QueryRowContext(ctx, `SELECT complete FROM actor_contribution_periods WHERE actor_id=?`, actor.ID).Scan(&complete); err != nil {
127+
t.Fatal(err)
128+
}
129+
if complete {
130+
t.Fatal("incomplete contribution period was materialized as complete")
131+
}
132+
}
133+
61134
func TestSearchActorsReturnsNullableProfilesAndBoundedCursor(t *testing.T) {
62135
t.Parallel()
63136
ctx := context.Background()
@@ -155,12 +228,19 @@ func TestActorContributionSearchBindsCursorToFilters(t *testing.T) {
155228
if err := c.ApplyActorContributionPeriod(ctx, ActorContributionPeriodInput{ActorID: actor.ID, From: from, To: from.Add(24 * time.Hour), Complete: false, ObservedAt: from.Add(27 * time.Hour), SourceUpdatedAt: from.Add(24 * time.Hour)}); err != nil {
156229
t.Fatal(err)
157230
}
158-
partial, err := c.GetActorContributionCoverage(ctx, actor.ID, "", from.Add(time.Hour), from.Add(12*time.Hour))
231+
retained, err := c.GetActorContributionCoverage(ctx, actor.ID, "", from.Add(time.Hour), from.Add(12*time.Hour))
232+
if err != nil {
233+
t.Fatal(err)
234+
}
235+
if retained == nil || !retained.Complete {
236+
t.Fatalf("partial refresh replaced complete coverage: %+v", retained)
237+
}
238+
pageAfterPartial, err := c.SearchActorContributions(ctx, ContributionSearchOptions{ActorRefs: []string{"alice"}, Sort: "occurred_at", Limit: 10})
159239
if err != nil {
160240
t.Fatal(err)
161241
}
162-
if partial != nil {
163-
t.Fatalf("partial refresh retained complete coverage: %+v", partial)
242+
if pageAfterPartial.Total != 2 {
243+
t.Fatalf("partial refresh replaced complete contribution items: %+v", pageAfterPartial)
164244
}
165245
organizationCovered, err := c.GetActorContributionCoverage(ctx, actor.ID, "O_acme", from.Add(time.Hour), from.Add(12*time.Hour))
166246
if err != nil {

‎internal/corpus/migration_test.go‎

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,19 @@ func TestBaselineMigrationCreatesCurrentSchema(t *testing.T) {
3737
t.Fatalf("table %s missing after baseline migration", table)
3838
}
3939
}
40+
for _, table := range []string{
41+
"actors", "actor_aliases", "actor_observations", "actor_profiles", "actor_social_accounts",
42+
"actor_organization_memberships", "actor_pinned_items", "actor_repository_affiliations",
43+
"actor_contribution_periods", "actor_contribution_days", "actor_contribution_items",
44+
"actor_repository_contribution_totals",
45+
} {
46+
for _, suffix := range []string{"ai", "au", "ad"} {
47+
trigger := "corpus_revision_" + table + "_" + suffix
48+
if !migrationTriggerExists(ctx, t, c.db, trigger) {
49+
t.Fatalf("trigger %s missing after baseline migration", trigger)
50+
}
51+
}
52+
}
4053

4154
for _, col := range []string{"merged_known", "author_association", "assignees", "draft", "locked", "state_reason", "milestone"} {
4255
if !migrationColumnExists(ctx, t, c.db, "threads", col) {
@@ -60,8 +73,10 @@ func TestActorMigrationDeduplicatesExistingLoginsCaseInsensitively(t *testing.T)
6073
if err != nil {
6174
t.Fatal(err)
6275
}
63-
if _, err := provider.Down(ctx); err != nil {
64-
t.Fatal(err)
76+
for range 2 {
77+
if _, err := provider.Down(ctx); err != nil {
78+
t.Fatal(err)
79+
}
6580
}
6681
if err := logger.Err(); err != nil {
6782
t.Fatal(err)
@@ -103,3 +118,12 @@ func migrationColumnExists(ctx context.Context, t *testing.T, db *sql.DB, table,
103118
}
104119
return found == 1
105120
}
121+
122+
func migrationTriggerExists(ctx context.Context, t *testing.T, db *sql.DB, trigger string) bool {
123+
t.Helper()
124+
var found int
125+
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM sqlite_master WHERE type='trigger' AND name=?`, trigger).Scan(&found); err != nil {
126+
t.Fatalf("query migration trigger %s: %v", trigger, err)
127+
}
128+
return found == 1
129+
}

0 commit comments

Comments
 (0)