From 55708390ee795b32f6ca82643846f3cf34cd15c6 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 15:25:23 -0700 Subject: [PATCH] feat: add member and guild tombstones --- CHANGELOG.md | 1 + README.md | 4 +- internal/cli/cloud_commands.go | 4 +- internal/discord/client.go | 68 +++++++++++ internal/share/merge.go | 50 +++++++- internal/share/merge_test.go | 143 +++++++++++++++++++++- internal/share/share.go | 114 +++++++++++++++++- internal/share/share_test.go | 16 ++- internal/store/attachments.go | 4 +- internal/store/coverage.go | 2 +- internal/store/members_profile.go | 5 +- internal/store/members_profile_test.go | 2 +- internal/store/mentions.go | 2 +- internal/store/messages.go | 10 +- internal/store/sqlc/queries.sql | 67 +++++++---- internal/store/sqlc/schema.sql | 8 +- internal/store/store.go | 45 ++++++- internal/store/store_test.go | 114 +++++++++++++++++- internal/store/storedb/models.go | 42 ++++--- internal/store/storedb/queries.sql.go | 160 ++++++++++++++----------- internal/store/write.go | 88 +++++++------- internal/syncer/syncer.go | 6 +- internal/syncer/syncer_tail_test.go | 18 +++ internal/syncer/tail.go | 21 +++- 24 files changed, 801 insertions(+), 193 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 74b0a73f..d6aefde8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixes +- Preserve guild and member history with source-attributed tombstones, explicit restore semantics, omission-safe member refreshes, and revision-aware Git-share merges. - Keep attachment fetches compatible with three allowed Discord CDN redirects and injected HTTP transports while preserving final-response host validation. - Reject malformed tombstone timestamps before routine incremental snapshot imports mutate the archive. diff --git a/README.md b/README.md index abd5dcac..5a8ac8d8 100644 --- a/README.md +++ b/README.md @@ -293,7 +293,7 @@ Each channel crawl also has a bounded runtime budget, so a pathological channel Retryable failures and unavailable-channel markers are tracked per channel; stale unavailable markers are cleared after a later successful crawl, and marker cleanup is best-effort so one missing local sync-state row cannot crash the run. Full sync member refresh is best-effort and currently gives up after five minutes without a caller-supplied deadline, so message sync completion is not held hostage by a slow guild member crawl. When the archive is already complete, `sync --full` now reuses the stored backlog markers and limits steady-state refresh to live top-level channels plus active threads instead of revisiting every stored archived thread. -If a guild already has a local member snapshot, routine syncs reuse it and skip another full member crawl until that snapshot ages out. +If a guild already has a local member snapshot, routine syncs reuse it and skip another full member crawl until that snapshot ages out. A completed refresh merges the members it observed; an omitted member is retained unless Discord sends an explicit member-remove event. ### `tail` @@ -640,7 +640,7 @@ discrawl subscribe --stale-after 15m https://github.com/example/discord-archive. discrawl subscribe --no-auto-update https://github.com/example/discord-archive.git ``` -Once `share.remote` is configured, read commands auto-fetch and import when the last share check is older than `share.stale_after` (default `15m`). Imports are planned from crawlkit shard fingerprints, with a Git-object fallback for older manifests, so routine updates normally read only changed canonical shards, preserve destination-only cache rows, and avoid rebuilding message FTS. `discrawl update` runs the same safe merge manually. Removed shards or incompatible table changes keep the database untouched and require `discrawl update --force`; historical restores require `update --force --ref `. `discrawl sync` does not auto-import the share unless `--update=auto` or `--update=force` is provided. +Once `share.remote` is configured, read commands auto-fetch and import when the last share check is older than `share.stale_after` (default `15m`). Imports are planned from crawlkit shard fingerprints, with a Git-object fallback for older manifests, so routine updates normally read only changed canonical shards, preserve destination-only cache rows, and avoid rebuilding message FTS. Guild and member rows carry explicit deletion metadata; their merges keep newer revisions, prefer tombstones at equal revisions, and allow later live revisions to restore them. `discrawl update` runs the same safe merge manually. Removed shards or incompatible table changes keep the database untouched and require `discrawl update --force`; tombstone-compatible guild/member schema changes remain merge-safe, while historical restores require `update --force --ref `. `discrawl sync` does not auto-import the share unless `--update=auto` or `--update=force` is provided. Hybrid mode is supported too: keep normal Discord credentials configured and set `share.remote`. `discrawl sync --update=auto` and `discrawl messages --sync` safely merge the Git snapshot first, usually as a changed-shard delta, then use live Discord for latest-message deltas. `discrawl sync --update=force` intentionally replaces public snapshot tables before live sync. Use `sync --all-channels` or `sync --full` when you want a broader live repair/backfill pass. diff --git a/internal/cli/cloud_commands.go b/internal/cli/cloud_commands.go index 93c580df..cc1e2d9b 100644 --- a/internal/cli/cloud_commands.go +++ b/internal/cli/cloud_commands.go @@ -460,7 +460,7 @@ var discrawlGuildColumns = []string{"guild_id", "name", "updated_at"} const discrawlGuildExportSQL = ` select id as guild_id, name, updated_at from guilds -where id != '@me' +where id != '@me' and deleted_at is null order by id` var discrawlChannelColumns = []string{"channel_id", "guild_id", "name", "type", "parent_id", "updated_at"} @@ -476,7 +476,7 @@ var discrawlMemberColumns = []string{"guild_id", "user_id", "username", "display const discrawlMemberExportSQL = ` select guild_id, user_id, username, coalesce(nullif(display_name, ''), nullif(nick, ''), username) as display_name, updated_at from members -where guild_id != '@me' +where guild_id != '@me' and deleted_at is null order by guild_id, user_id` var discrawlMessageColumns = []string{"message_id", "channel_id", "guild_id", "author_id", "author_username", "content", "created_at", "edited_at"} diff --git a/internal/discord/client.go b/internal/discord/client.go index 91e4d817..3488b870 100644 --- a/internal/discord/client.go +++ b/internal/discord/client.go @@ -22,6 +22,11 @@ type EventHandler interface { OnMemberDelete(context.Context, string, string) error } +type guildEventHandler interface { + OnGuildUpsert(context.Context, *discordgo.Guild) error + OnGuildDelete(context.Context, *discordgo.Guild) error +} + type tailGuildFilter interface { TailAllowsGuild(string) bool } @@ -603,6 +608,51 @@ func (c *Client) Tail(ctx context.Context, handler EventHandler) error { before, )) }) + addHandler(func(_ *discordgo.Session, evt *discordgo.GuildCreate) { + guildHandler, ok := handler.(guildEventHandler) + if !ok { + return + } + var guild *discordgo.Guild + if evt != nil { + guild = evt.Guild + } + c.enqueueTailTask(tailCtx, orderedWorkCh, fatal, newGuildTailTask( + "GUILD_CREATE", + func(taskCtx context.Context) error { return guildHandler.OnGuildUpsert(taskCtx, guild) }, + guild, + )) + }) + addHandler(func(_ *discordgo.Session, evt *discordgo.GuildUpdate) { + guildHandler, ok := handler.(guildEventHandler) + if !ok { + return + } + var guild *discordgo.Guild + if evt != nil { + guild = evt.Guild + } + c.enqueueTailTask(tailCtx, orderedWorkCh, fatal, newGuildTailTask( + "GUILD_UPDATE", + func(taskCtx context.Context) error { return guildHandler.OnGuildUpsert(taskCtx, guild) }, + guild, + )) + }) + addHandler(func(_ *discordgo.Session, evt *discordgo.GuildDelete) { + guildHandler, ok := handler.(guildEventHandler) + if !ok { + return + } + var guild *discordgo.Guild + if evt != nil { + guild = evt.Guild + } + c.enqueueTailTask(tailCtx, orderedWorkCh, fatal, newGuildTailTask( + "GUILD_DELETE", + func(taskCtx context.Context) error { return guildHandler.OnGuildDelete(taskCtx, guild) }, + guild, + )) + }) addHandler(func(_ *discordgo.Session, evt *discordgo.GuildMemberAdd) { var member *discordgo.Member if evt != nil { @@ -1297,6 +1347,24 @@ func newChannelTailTask( return task } +func newGuildTailTask( + eventType string, + run func(context.Context) error, + guilds ...*discordgo.Guild, +) tailTask { + task := tailTask{ + eventType: eventType, + failureClass: tailFailureClassOrdered, + run: run, + } + for _, guild := range guilds { + if guild != nil { + setTailTaskID(&task.guildID, guild.ID) + } + } + return task +} + func newMemberTailTask( eventType string, run func(context.Context) error, diff --git a/internal/share/merge.go b/internal/share/merge.go index 12347d90..85c46b70 100644 --- a/internal/share/merge.go +++ b/internal/share/merge.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strings" "time" @@ -62,7 +63,7 @@ func MergeIfChanged(ctx context.Context, s *store.Store, opts Options) (Manifest } } plan := snapshot.PlanMergeImport(snapshotManifest(previous), snapshotManifest(manifest)) - plan, err = shareMergePlan(plan, allowEventMerge) + plan, err = shareMergePlan(plan, snapshotManifest(previous), allowEventMerge) if err != nil { if markErr := MarkReplacementPending(ctx, s, manifest, err.Error()); markErr != nil { return Manifest{}, false, errors.Join(err, markErr) @@ -72,7 +73,7 @@ func MergeIfChanged(ctx context.Context, s *store.Store, opts Options) (Manifest return importMergePlan(ctx, s, opts, previous, manifest, plan) } -func shareMergePlan(plan snapshot.ImportPlan, allowEventMerge bool) (snapshot.ImportPlan, error) { +func shareMergePlan(plan snapshot.ImportPlan, previous snapshot.Manifest, allowEventMerge bool) (snapshot.ImportPlan, error) { if plan.Full { return snapshot.ImportPlan{}, &ReplacementRequiredError{Reason: plan.Reason} } @@ -104,6 +105,14 @@ func shareMergePlan(plan snapshot.ImportPlan, allowEventMerge bool) (snapshot.Im continue } if tablePlan.Mode == snapshot.TableImportReplace { + if (tablePlan.Table.Name == "guilds" || tablePlan.Table.Name == "members") && + tablePlan.Reason == "columns changed" && + isTombstoneColumnAddition(manifestTable(previous, tablePlan.Table.Name).Columns, tablePlan.Table.Columns) { + tablePlan.Mode = snapshot.TableImportFiles + tablePlan.Reason = "merge tombstone-aware entity rows" + out.Tables = append(out.Tables, tablePlan) + continue + } replacements = append(replacements, tablePlan.Table.Name) continue } @@ -116,6 +125,43 @@ func shareMergePlan(plan snapshot.ImportPlan, allowEventMerge bool) (snapshot.Im return out, nil } +func manifestTable(manifest snapshot.Manifest, name string) snapshot.TableManifest { + for _, table := range manifest.Tables { + if table.Name == name { + return table + } + } + return snapshot.TableManifest{} +} + +func isTombstoneColumnAddition(previous, current []string) bool { + if len(current) != len(previous)+3 { + return false + } + base := make([]string, 0, len(previous)) + tombstones := map[string]bool{ + "deleted_at": false, + "deletion_source": false, + "deletion_reason": false, + } + for _, column := range current { + if _, ok := tombstones[column]; ok { + if tombstones[column] { + return false + } + tombstones[column] = true + continue + } + base = append(base, column) + } + for _, found := range tombstones { + if !found { + return false + } + } + return slices.Equal(previous, base) +} + func eventTablesEmpty(ctx context.Context, s *store.Store) (bool, error) { var rows int if err := s.DB().QueryRowContext(ctx, `select (select count(*) from message_events) + (select count(*) from mention_events)`).Scan(&rows); err != nil { diff --git a/internal/share/merge_test.go b/internal/share/merge_test.go index 46593ca7..98e3d9ac 100644 --- a/internal/share/merge_test.go +++ b/internal/share/merge_test.go @@ -132,6 +132,117 @@ func TestMergeIfChangedPreservesLocalRowsUntilForcedReplacement(t *testing.T) { require.Equal(t, "0", rows[0][0], "force must reconcile even when the manifest is unchanged") } +func TestMemberAndGuildTombstonesMergeByRevisionAndRestore(t *testing.T) { + ctx := context.Background() + src := seedStore(t, filepath.Join(t.TempDir(), "src.db")) + defer func() { _ = src.Close() }() + repo := filepath.Join(t.TempDir(), "share") + _, err := Export(ctx, src, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + + dst := seedStore(t, filepath.Join(t.TempDir(), "dst.db")) + defer func() { _ = dst.Close() }() + _, _, err = MergeIfChanged(ctx, dst, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + + offsetRevision := time.Now().UTC().Add(time.Hour).Truncate(time.Second) + offsetText := offsetRevision.In(time.FixedZone("proof-offset", -2*60*60)).Format(time.RFC3339Nano) + _, err = src.DB().ExecContext(ctx, ` + update guilds set name = 'offset revision guild', updated_at = ? where id = 'g1'; + update members set display_name = 'Offset Revision Member', updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, offsetText, offsetText) + require.NoError(t, err) + _, err = dst.DB().ExecContext(ctx, ` + update guilds set updated_at = ? where id = 'g1'; + update members set updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, offsetRevision.Add(-time.Minute).Format(time.RFC3339Nano), offsetRevision.Add(-time.Minute).Format(time.RFC3339Nano)) + require.NoError(t, err) + _, err = Export(ctx, src, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + _, _, err = MergeIfChanged(ctx, dst, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + var guildName, memberName, guildRevision, memberRevision string + require.NoError(t, dst.DB().QueryRowContext(ctx, `select name, updated_at from guilds where id = 'g1'`).Scan(&guildName, &guildRevision)) + require.NoError(t, dst.DB().QueryRowContext(ctx, `select display_name, updated_at from members where guild_id = 'g1' and user_id = 'u1'`).Scan(&memberName, &memberRevision)) + require.Equal(t, "offset revision guild", guildName) + require.Equal(t, "Offset Revision Member", memberName) + require.Equal(t, offsetRevision.Format(time.RFC3339Nano), guildRevision) + require.Equal(t, offsetRevision.Format(time.RFC3339Nano), memberRevision) + + equalRevision := offsetRevision.Add(time.Second) + _, err = src.DB().ExecContext(ctx, ` + update guilds set name = 'stale live guild', updated_at = ? where id = 'g1'; + update members set display_name = 'Stale Live Member', updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, equalRevision.Format(time.RFC3339Nano), equalRevision.Format(time.RFC3339Nano)) + require.NoError(t, err) + _, err = dst.DB().ExecContext(ctx, ` + update guilds set deleted_at = ?, deletion_source = 'local-test', deletion_reason = 'explicit-local-delete', updated_at = ? where id = 'g1'; + update members set deleted_at = ?, deletion_source = 'local-test', deletion_reason = 'explicit-local-delete', updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, equalRevision.Format(time.RFC3339Nano), equalRevision.Format(time.RFC3339Nano), equalRevision.Format(time.RFC3339Nano), equalRevision.Format(time.RFC3339Nano)) + require.NoError(t, err) + _, err = Export(ctx, src, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + _, changed, err := MergeIfChanged(ctx, dst, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + require.True(t, changed) + assertEntityTombstones(t, ctx, dst, true, "local-test") + + newerRevision := equalRevision.Add(time.Second) + _, err = src.DB().ExecContext(ctx, ` + update guilds set name = 'restored guild', deleted_at = null, deletion_source = null, deletion_reason = null, updated_at = ? where id = 'g1'; + update members set display_name = 'Restored Member', deleted_at = null, deletion_source = null, deletion_reason = null, updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, newerRevision.Format(time.RFC3339Nano), newerRevision.Format(time.RFC3339Nano)) + require.NoError(t, err) + _, err = Export(ctx, src, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + _, _, err = MergeIfChanged(ctx, dst, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + assertEntityTombstones(t, ctx, dst, false, "") + members, err := dst.Members(ctx, "g1", "Restored", 10) + require.NoError(t, err) + require.Len(t, members, 1) + + tombstoneRevision := newerRevision.Add(time.Second) + _, err = src.DB().ExecContext(ctx, ` + update guilds set deleted_at = ?, deletion_source = 'remote-share', deletion_reason = 'guild-delete-event', updated_at = ? where id = 'g1'; + update members set deleted_at = ?, deletion_source = 'remote-share', deletion_reason = 'member-remove-event', updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, tombstoneRevision.Format(time.RFC3339Nano), tombstoneRevision.Format(time.RFC3339Nano), tombstoneRevision.Format(time.RFC3339Nano), tombstoneRevision.Format(time.RFC3339Nano)) + require.NoError(t, err) + _, err = Export(ctx, src, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + _, _, err = MergeIfChanged(ctx, dst, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + assertEntityTombstones(t, ctx, dst, true, "remote-share") + + restoreRevision := tombstoneRevision.Add(time.Second) + _, err = src.DB().ExecContext(ctx, ` + update guilds set deleted_at = null, deletion_source = null, deletion_reason = null, updated_at = ? where id = 'g1'; + update members set deleted_at = null, deletion_source = null, deletion_reason = null, updated_at = ? where guild_id = 'g1' and user_id = 'u1'; + `, restoreRevision.Format(time.RFC3339Nano), restoreRevision.Format(time.RFC3339Nano)) + require.NoError(t, err) + _, err = Export(ctx, src, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + require.NoError(t, dst.UpsertGuild(ctx, store.GuildRecord{ID: "local-only", Name: "Local only", RawJSON: `{}`})) + _, _, err = Replace(ctx, dst, Options{RepoPath: repo, Branch: "main"}) + require.NoError(t, err) + assertEntityTombstones(t, ctx, dst, false, "") + var localOnly int + require.NoError(t, dst.DB().QueryRowContext(ctx, `select count(*) from guilds where id = 'local-only'`).Scan(&localOnly)) + require.Zero(t, localOnly) +} + +func assertEntityTombstones(t *testing.T, ctx context.Context, s *store.Store, deleted bool, source string) { + t.Helper() + var guildDeleted, memberDeleted bool + var guildSource, memberSource string + require.NoError(t, s.DB().QueryRowContext(ctx, `select deleted_at is not null, coalesce(deletion_source, '') from guilds where id = 'g1'`).Scan(&guildDeleted, &guildSource)) + require.NoError(t, s.DB().QueryRowContext(ctx, `select deleted_at is not null, coalesce(deletion_source, '') from members where guild_id = 'g1' and user_id = 'u1'`).Scan(&memberDeleted, &memberSource)) + require.Equal(t, deleted, guildDeleted) + require.Equal(t, deleted, memberDeleted) + require.Equal(t, source, guildSource) + require.Equal(t, source, memberSource) +} + func TestMergeIfChangedMarksReplacementPendingWithoutChangingRows(t *testing.T) { ctx := context.Background() src := seedStore(t, filepath.Join(t.TempDir(), "src.db")) @@ -164,7 +275,7 @@ func TestShareMergePlanKeepsGeneratedHistoryForceOnly(t *testing.T) { {Table: snapshot.TableManifest{Name: "message_events"}, Mode: snapshot.TableImportFiles}, {Table: snapshot.TableManifest{Name: "mention_events"}, Mode: snapshot.TableImportFiles}, {Table: snapshot.TableManifest{Name: "sync_state"}, Mode: snapshot.TableImportFiles}, - }}, false) + }}, snapshot.Manifest{}, false) require.NoError(t, err) require.Equal(t, snapshot.TableImportFiles, importPlanTable(t, plan, "messages").Mode) require.Equal(t, snapshot.TableImportSkip, importPlanTable(t, plan, "message_events").Mode) @@ -174,7 +285,7 @@ func TestShareMergePlanKeepsGeneratedHistoryForceOnly(t *testing.T) { bootstrap, err := shareMergePlan(snapshot.ImportPlan{Tables: []snapshot.TableImportPlan{ {Table: snapshot.TableManifest{Name: "message_events"}, Mode: snapshot.TableImportFiles}, {Table: snapshot.TableManifest{Name: "sync_state"}, Mode: snapshot.TableImportFiles}, - }}, true) + }}, snapshot.Manifest{}, true) require.NoError(t, err) require.Equal(t, snapshot.TableImportFiles, importPlanTable(t, bootstrap, "message_events").Mode) require.Equal(t, snapshot.TableImportSkip, importPlanTable(t, bootstrap, "sync_state").Mode) @@ -182,17 +293,39 @@ func TestShareMergePlanKeepsGeneratedHistoryForceOnly(t *testing.T) { _, err = shareMergePlan(snapshot.ImportPlan{Tables: []snapshot.TableImportPlan{{ Table: snapshot.TableManifest{Name: "messages"}, Mode: snapshot.TableImportReplace, - }}}, false) + }}}, snapshot.Manifest{}, false) var replacement *ReplacementRequiredError require.ErrorAs(t, err, &replacement) require.Equal(t, []string{"messages"}, replacement.Tables) + legacyGuildColumns := []string{"id", "name", "icon", "raw_json", "updated_at"} + legacyMemberColumns := []string{"guild_id", "user_id", "username", "updated_at"} + entityPlan, err := shareMergePlan(snapshot.ImportPlan{Tables: []snapshot.TableImportPlan{ + {Table: snapshot.TableManifest{Name: "guilds", Columns: append(append([]string{}, legacyGuildColumns...), "deleted_at", "deletion_source", "deletion_reason")}, Mode: snapshot.TableImportReplace, Reason: "columns changed"}, + {Table: snapshot.TableManifest{Name: "members", Columns: append(append([]string{}, legacyMemberColumns...), "deleted_at", "deletion_source", "deletion_reason")}, Mode: snapshot.TableImportReplace, Reason: "columns changed"}, + }}, snapshot.Manifest{Tables: []snapshot.TableManifest{ + {Name: "guilds", Columns: legacyGuildColumns}, + {Name: "members", Columns: legacyMemberColumns}, + }}, false) + require.NoError(t, err) + require.Equal(t, snapshot.TableImportFiles, importPlanTable(t, entityPlan, "guilds").Mode) + require.Equal(t, snapshot.TableImportFiles, importPlanTable(t, entityPlan, "members").Mode) + + _, err = shareMergePlan(snapshot.ImportPlan{Tables: []snapshot.TableImportPlan{{ + Table: snapshot.TableManifest{Name: "guilds", Columns: []string{"id", "name", "updated_at", "deleted_at", "deletion_source", "deletion_reason"}}, + Mode: snapshot.TableImportReplace, + Reason: "columns changed", + }}}, snapshot.Manifest{Tables: []snapshot.TableManifest{{ + Name: "guilds", Columns: []string{"id", "renamed_name", "updated_at"}, + }}}, false) + require.ErrorAs(t, err, &replacement) + require.Equal(t, []string{"guilds"}, replacement.Tables) - _, err = shareMergePlan(snapshot.ImportPlan{Full: true, Reason: "manifest version changed"}, false) + _, err = shareMergePlan(snapshot.ImportPlan{Full: true, Reason: "manifest version changed"}, snapshot.Manifest{}, false) require.ErrorContains(t, err, "manifest version changed") _, err = shareMergePlan(snapshot.ImportPlan{Tables: []snapshot.TableImportPlan{{ Table: snapshot.TableManifest{Name: "unknown"}, Mode: snapshot.TableImportFiles, - }}}, false) + }}}, snapshot.Manifest{}, false) require.ErrorContains(t, err, "unknown") } diff --git a/internal/share/share.go b/internal/share/share.go index 5638e078..5e6e88e0 100644 --- a/internal/share/share.go +++ b/internal/share/share.go @@ -1687,11 +1687,24 @@ func importTableFile(ctx context.Context, stmt *sql.Stmt, repoPath string, table } func validateSnapshotRow(table string, row map[string]any) error { - if table != "messages" { + if table != "messages" && table != "guilds" && table != "members" { return nil } + if table == "guilds" || table == "members" { + for _, column := range []string{"deleted_at", "deletion_source", "deletion_reason"} { + if _, ok := row[column]; !ok { + row[column] = nil + } + } + if err := validateSnapshotRevision(table, row); err != nil { + return err + } + } raw, ok := row["deleted_at"] if !ok || raw == nil { + if table == "guilds" || table == "members" { + return validateLiveSnapshotTombstoneMetadata(table, row) + } return nil } value, ok := raw.(string) @@ -1701,13 +1714,54 @@ func validateSnapshotRow(table string, row map[string]any) error { value = strings.TrimSpace(value) if value == "" { row["deleted_at"] = nil + if table == "guilds" || table == "members" { + return validateLiveSnapshotTombstoneMetadata(table, row) + } return nil } parsed, err := time.Parse(time.RFC3339Nano, value) if err != nil { - return fmt.Errorf("messages.deleted_at must be RFC3339: %w", err) + return fmt.Errorf("%s.deleted_at must be RFC3339: %w", table, err) } row["deleted_at"] = parsed.UTC().Format(time.RFC3339Nano) + if table == "guilds" || table == "members" { + for _, column := range []string{"deletion_source", "deletion_reason"} { + value, ok := row[column].(string) + if !ok || strings.TrimSpace(value) == "" { + return fmt.Errorf("%s.%s must be a non-empty string for a tombstone", table, column) + } + row[column] = strings.TrimSpace(value) + } + } + return nil +} + +func validateSnapshotRevision(table string, row map[string]any) error { + raw := row["updated_at"] + value, ok := raw.(string) + if !ok || strings.TrimSpace(value) == "" { + return fmt.Errorf("%s.updated_at must be an RFC3339 string", table) + } + parsed, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(value)) + if err != nil { + return fmt.Errorf("%s.updated_at must be RFC3339: %w", table, err) + } + row["updated_at"] = parsed.UTC().Format(time.RFC3339Nano) + return nil +} + +func validateLiveSnapshotTombstoneMetadata(table string, row map[string]any) error { + for _, column := range []string{"deletion_source", "deletion_reason"} { + raw := row[column] + if raw == nil { + continue + } + value, ok := raw.(string) + if !ok || strings.TrimSpace(value) != "" { + return fmt.Errorf("%s.%s must be null for a live row", table, column) + } + row[column] = nil + } return nil } @@ -2509,10 +2563,64 @@ func upsertMergeSnapshotRow(ctx context.Context, tx *sql.Tx, table string, row m return false, err } } - protectNewer := slices.Contains([]string{"guilds", "channels", "members", "messages"}, table) + if table == "guilds" || table == "members" { + apply, err := shouldMergeTombstoneEntityRow(ctx, tx, table, row) + if err != nil || !apply { + return false, err + } + } + // Guild/member revisions were compared chronologically above. Reapplying the + // generic lexical SQL guard would reject valid RFC3339 offset timestamps. + protectNewer := slices.Contains([]string{"channels", "messages"}, table) return upsertSnapshotRow(ctx, tx, table, row, protectNewer) } +func shouldMergeTombstoneEntityRow(ctx context.Context, tx *sql.Tx, table string, row map[string]any) (bool, error) { + var query string + var args []any + switch table { + case "guilds": + query = `select updated_at, deleted_at is not null from guilds where id = ?` + args = []any{importValue(row["id"])} + case "members": + query = `select updated_at, deleted_at is not null from members where guild_id = ? and user_id = ?` + args = []any{importValue(row["guild_id"]), importValue(row["user_id"])} + default: + return true, nil + } + var localUpdated string + var localTombstone bool + err := tx.QueryRowContext(ctx, query, args...).Scan(&localUpdated, &localTombstone) + if errors.Is(err, sql.ErrNoRows) { + return true, nil + } + if err != nil { + return false, fmt.Errorf("read existing %s merge revision: %w", table, err) + } + comparison := compareSnapshotRevisions(stringValue(row["updated_at"]), localUpdated) + if comparison != 0 { + return comparison > 0, nil + } + incomingTombstone := row["deleted_at"] != nil && strings.TrimSpace(stringValue(row["deleted_at"])) != "" + return !localTombstone || incomingTombstone, nil +} + +func compareSnapshotRevisions(left, right string) int { + leftTime, leftErr := time.Parse(time.RFC3339Nano, strings.TrimSpace(left)) + rightTime, rightErr := time.Parse(time.RFC3339Nano, strings.TrimSpace(right)) + if leftErr == nil && rightErr == nil { + switch { + case leftTime.Before(rightTime): + return -1 + case leftTime.After(rightTime): + return 1 + default: + return 0 + } + } + return strings.Compare(strings.TrimSpace(left), strings.TrimSpace(right)) +} + func preserveIncrementalAttachmentState(ctx context.Context, tx *sql.Tx, row map[string]any) error { attachmentID := stringValue(row["attachment_id"]) if attachmentID == "" { diff --git a/internal/share/share_test.go b/internal/share/share_test.go index c93df912..6e1c4128 100644 --- a/internal/share/share_test.go +++ b/internal/share/share_test.go @@ -1043,7 +1043,7 @@ func TestMergeIfChangedUsesMixedIncrementalPlanForMetadataChanges(t *testing.T) previous, ok := PreviousMergedManifest(ctx, dst, Options{RepoPath: repo, Branch: "main"}) require.True(t, ok) - planned, err := shareMergePlan(snapshot.PlanMergeImport(snapshotManifest(previous), snapshotManifest(updated)), false) + planned, err := shareMergePlan(snapshot.PlanMergeImport(snapshotManifest(previous), snapshotManifest(updated)), snapshotManifest(previous), false) require.NoError(t, err) require.Equal(t, snapshot.TableImportFiles, importPlanTable(t, planned, "channels").Mode) require.Equal(t, snapshot.TableImportFiles, importPlanTable(t, planned, "members").Mode) @@ -1134,7 +1134,7 @@ func TestMergeIfChangedInfersLegacyManifestFilesFromGit(t *testing.T) { previous, ok := PreviousMergedManifest(ctx, dst, Options{RepoPath: repo, Branch: "main"}) require.True(t, ok) - planned, err := shareMergePlan(snapshot.PlanMergeImport(snapshotManifest(previous), snapshotManifest(enrichManifestFromGit(ctx, repo, "HEAD", stripFileManifests(updated)))), false) + planned, err := shareMergePlan(snapshot.PlanMergeImport(snapshotManifest(previous), snapshotManifest(enrichManifestFromGit(ctx, repo, "HEAD", stripFileManifests(updated)))), snapshotManifest(previous), false) require.NoError(t, err) require.True(t, planned.Changed(), "%+v", planned) @@ -2323,7 +2323,17 @@ func TestValidateSnapshotRowRejectsMalformedDeletedAtBeforeImport(t *testing.T) require.Equal(t, "2026-07-14T12:00:00Z", padded["deleted_at"]) require.ErrorContains(t, validateSnapshotRow("messages", map[string]any{"deleted_at": "not-a-timestamp"}), "must be RFC3339") require.ErrorContains(t, validateSnapshotRow("messages", map[string]any{"deleted_at": json.Number("123")}), "must be a string or null") - require.NoError(t, validateSnapshotRow("guilds", map[string]any{"deleted_at": "not-a-timestamp"})) + require.ErrorContains(t, validateSnapshotRow("guilds", map[string]any{ + "updated_at": "2026-07-14T12:00:00Z", "deleted_at": "not-a-timestamp", + }), "must be RFC3339") + require.ErrorContains(t, validateSnapshotRow("members", map[string]any{ + "updated_at": "2026-07-14T12:00:00Z", "deleted_at": "2026-07-14T12:00:01Z", + }), "deletion_source") + offsetRevision := map[string]any{ + "updated_at": "2026-07-18T11:30:00-02:00", "deleted_at": nil, + } + require.NoError(t, validateSnapshotRow("guilds", offsetRevision)) + require.Equal(t, "2026-07-18T13:30:00Z", offsetRevision["updated_at"]) ctx := context.Background() s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) diff --git a/internal/store/attachments.go b/internal/store/attachments.go index f4b5347f..f37bae78 100644 --- a/internal/store/attachments.go +++ b/internal/store/attachments.go @@ -142,9 +142,9 @@ func (s *Store) ListAttachments(ctx context.Context, opts AttachmentListOptions) m.created_at from message_attachments a left join messages m on m.id = a.message_id - left join guilds g on g.id = a.guild_id + left join guilds g on g.id = a.guild_id and g.deleted_at is null left join channels c on c.id = a.channel_id - left join members mem on mem.guild_id = a.guild_id and mem.user_id = a.author_id + left join members mem on mem.guild_id = a.guild_id and mem.user_id = a.author_id and mem.deleted_at is null where ` + strings.Join(clauses, " and ") + ` order by m.created_at asc, a.attachment_id asc ` diff --git a/internal/store/coverage.go b/internal/store/coverage.go index e77bc8d7..62b04311 100644 --- a/internal/store/coverage.go +++ b/internal/store/coverage.go @@ -108,7 +108,7 @@ func (s *Store) Coverage(ctx context.Context, guildID string, generatedAt time.T rows, err := s.db.QueryContext(queryCtx, ` select id, name from guilds - where ? = '' or id = ? + where deleted_at is null and (? = '' or id = ?) order by lower(name), id `, guildID, guildID) if err != nil { diff --git a/internal/store/members_profile.go b/internal/store/members_profile.go index 59b7131a..07097da1 100644 --- a/internal/store/members_profile.go +++ b/internal/store/members_profile.go @@ -82,6 +82,7 @@ func (s *Store) rebuildMemberFTS(ctx context.Context) error { coalesce(nullif(display_name, ''), nullif(nick, ''), nullif(global_name, ''), username, ''), raw_json from members + where deleted_at is null order by guild_id, user_id `) if err != nil { @@ -163,7 +164,7 @@ func (s *Store) searchMembers(ctx context.Context, guildID, query string, limit coalesce(m.joined_at, ''), m.raw_json from member_fts - join members m on m.guild_id = member_fts.guild_id and m.user_id = member_fts.user_id + join members m on m.guild_id = member_fts.guild_id and m.user_id = member_fts.user_id and m.deleted_at is null where `+strings.Join(clauses, " and ")+` order by bm25(member_fts), coalesce(nullif(m.display_name, ''), nullif(m.nick, ''), nullif(m.global_name, ''), m.username), m.username limit ? @@ -177,7 +178,7 @@ func (s *Store) searchMembers(ctx context.Context, guildID, query string, limit func (s *Store) searchMembersFallback(ctx context.Context, guildID, query string, limit int) ([]MemberRow, error) { args := []any{} - clauses := []string{"1=1"} + clauses := []string{"deleted_at is null"} if guildID != "" { clauses = append(clauses, "guild_id = ?") args = append(args, guildID) diff --git a/internal/store/members_profile_test.go b/internal/store/members_profile_test.go index dd2275ce..65e54c5f 100644 --- a/internal/store/members_profile_test.go +++ b/internal/store/members_profile_test.go @@ -17,7 +17,7 @@ func TestMembersSearchesArchivedProfileText(t *testing.T) { require.NoError(t, err) defer func() { _ = s.Close() }() - require.NoError(t, s.ReplaceMembers(ctx, "g1", []MemberRecord{ + require.NoError(t, s.MergeMembers(ctx, "g1", []MemberRecord{ { GuildID: "g1", UserID: "u1", diff --git a/internal/store/mentions.go b/internal/store/mentions.go index 3a2e342b..d119fc1c 100644 --- a/internal/store/mentions.go +++ b/internal/store/mentions.go @@ -69,7 +69,7 @@ func (s *Store) ListMentions(ctx context.Context, opts MentionListOptions) ([]Me from mention_events me left join messages m on m.id = me.message_id left join channels c on c.id = me.channel_id - left join members mem on mem.guild_id = me.guild_id and mem.user_id = me.author_id + left join members mem on mem.guild_id = me.guild_id and mem.user_id = me.author_id and mem.deleted_at is null where `+strings.Join(clauses, " and ")+` order by me.event_at desc, me.event_id desc limit ? diff --git a/internal/store/messages.go b/internal/store/messages.go index d5f43ef9..719f84e2 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -111,9 +111,9 @@ func (s *Store) ListMessages(ctx context.Context, opts MessageListOptions) ([]Me coalesce((select group_concat(a.text_content, char(10)) from message_attachments a where a.message_id = m.id and trim(a.text_content) <> ''), ''), m.pinned from messages m - left join guilds g on g.id = m.guild_id + left join guilds g on g.id = m.guild_id and g.deleted_at is null left join channels c on c.id = m.channel_id - left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id + left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id and mem.deleted_at is null where ` + strings.Join(clauses, " and ") + ` ` @@ -249,9 +249,9 @@ func (s *Store) hydrateMessageThreadContext(ctx context.Context, rows []MessageR coalesce((select group_concat(a.text_content, char(10)) from message_attachments a where a.message_id = m.id and trim(a.text_content) <> ''), ''), m.pinned from messages m - left join guilds g on g.id = m.guild_id + left join guilds g on g.id = m.guild_id and g.deleted_at is null left join channels c on c.id = m.channel_id - left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id + left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id and mem.deleted_at is null where m.id in (` + placeholders(len(rootIDs)) + `) order by m.created_at asc, m.id asc` contextRows, err := s.db.QueryContext(ctx, query, rootIDs...) @@ -464,7 +464,7 @@ func (s *Store) discordMemberDisplayNames(ctx context.Context, ids map[string]st '' ) from members - where user_id in (` + placeholders(len(args)) + `) + where deleted_at is null and user_id in (` + placeholders(len(args)) + `) ` rows, err := s.db.QueryContext(ctx, query, args...) if err != nil { diff --git a/internal/store/sqlc/queries.sql b/internal/store/sqlc/queries.sql index 1d4c8450..72e069ef 100644 --- a/internal/store/sqlc/queries.sql +++ b/internal/store/sqlc/queries.sql @@ -56,7 +56,7 @@ select exists( ) as present; -- name: CountGuilds :one -select count(*) as count from guilds; +select count(*) as count from guilds where deleted_at is null; -- name: CountChannels :one select count(*) as count from channels; @@ -65,7 +65,7 @@ select count(*) as count from channels; select count(*) as count from messages; -- name: CountMembers :one -select count(*) as count from members; +select count(*) as count from members where deleted_at is null; -- name: CountThreads :one select count(*) as count @@ -80,11 +80,12 @@ where scope = ?; -- name: GetGuildName :one select name from guilds -where id = ?; +where id = ? and deleted_at is null; -- name: ListGuildIDs :many select id from guilds +where deleted_at is null order by id; -- name: CountChannelsByGuild :one @@ -95,7 +96,7 @@ where guild_id = ?; -- name: CountMembersByGuild :one select count(*) as count from members -where guild_id = ?; +where guild_id = ? and deleted_at is null; -- name: ListMembers :many select guild_id, user_id, username, coalesce(global_name, '') as global_name, @@ -103,6 +104,7 @@ select guild_id, user_id, username, coalesce(global_name, '') as global_name, coalesce(discriminator, '') as discriminator, coalesce(avatar, '') as avatar, role_ids_json, bot, coalesce(joined_at, '') as joined_at, raw_json from members +where deleted_at is null order by coalesce(nullif(display_name, ''), nullif(nick, ''), nullif(global_name, ''), username), username limit ?; @@ -112,7 +114,7 @@ select guild_id, user_id, username, coalesce(global_name, '') as global_name, coalesce(discriminator, '') as discriminator, coalesce(avatar, '') as avatar, role_ids_json, bot, coalesce(joined_at, '') as joined_at, raw_json from members -where guild_id = ? +where guild_id = ? and deleted_at is null order by coalesce(nullif(display_name, ''), nullif(nick, ''), nullif(global_name, ''), username), username limit ?; @@ -122,7 +124,7 @@ select guild_id, user_id, username, coalesce(global_name, '') as global_name, coalesce(discriminator, '') as discriminator, coalesce(avatar, '') as avatar, role_ids_json, bot, coalesce(joined_at, '') as joined_at, raw_json from members -where user_id = ? +where user_id = ? and deleted_at is null order by guild_id, username; -- name: MemberMessageStats :one @@ -158,7 +160,7 @@ select m.pinned from messages m left join channels c on c.id = m.channel_id -left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id +left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id and mem.deleted_at is null where m.guild_id = ? and m.author_id = ? order by m.created_at desc, m.id desc limit ?; @@ -214,12 +216,27 @@ where c.kind in ('text', 'news', 'announcement', 'thread_public', 'thread_privat order by c.id; -- name: UpsertGuild :exec -insert into guilds(id, name, icon, raw_json, updated_at) -values(?, ?, ?, ?, ?) +insert into guilds(id, name, icon, raw_json, updated_at, deleted_at, deletion_source, deletion_reason) +values(?, ?, ?, ?, ?, null, null, null) on conflict(id) do update set name = excluded.name, icon = excluded.icon, raw_json = excluded.raw_json, + updated_at = excluded.updated_at, + deleted_at = null, + deletion_source = null, + deletion_reason = null; + +-- name: MarkGuildDeleted :exec +insert into guilds(id, name, icon, raw_json, updated_at, deleted_at, deletion_source, deletion_reason) +values( + sqlc.arg(id), sqlc.arg(id), null, '{}', sqlc.arg(updated_at), + sqlc.arg(deleted_at), sqlc.arg(deletion_source), sqlc.arg(deletion_reason) +) +on conflict(id) do update set + deleted_at = excluded.deleted_at, + deletion_source = excluded.deletion_source, + deletion_reason = excluded.deletion_reason, updated_at = excluded.updated_at; -- name: UpsertChannel :exec @@ -248,17 +265,12 @@ on conflict(id) do update set delete from members where guild_id = ?; --- name: InsertMember :exec -insert into members( - guild_id, user_id, username, global_name, display_name, nick, discriminator, - avatar, bot, joined_at, role_ids_json, raw_json, updated_at -) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); - -- name: UpsertMember :exec insert into members( guild_id, user_id, username, global_name, display_name, nick, discriminator, - avatar, bot, joined_at, role_ids_json, raw_json, updated_at -) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + avatar, bot, joined_at, role_ids_json, raw_json, updated_at, + deleted_at, deletion_source, deletion_reason +) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, null, null, null) on conflict(guild_id, user_id) do update set username = excluded.username, global_name = excluded.global_name, @@ -270,11 +282,24 @@ on conflict(guild_id, user_id) do update set joined_at = excluded.joined_at, role_ids_json = excluded.role_ids_json, raw_json = excluded.raw_json, - updated_at = excluded.updated_at; + updated_at = excluded.updated_at, + deleted_at = null, + deletion_source = null, + deletion_reason = null; --- name: DeleteMember :exec -delete from members -where guild_id = ? and user_id = ?; +-- name: MarkMemberDeleted :exec +insert into members( + guild_id, user_id, username, role_ids_json, raw_json, updated_at, + deleted_at, deletion_source, deletion_reason +) values( + sqlc.arg(guild_id), sqlc.arg(user_id), sqlc.arg(user_id), '[]', '{}', sqlc.arg(updated_at), + sqlc.arg(deleted_at), sqlc.arg(deletion_source), sqlc.arg(deletion_reason) +) +on conflict(guild_id, user_id) do update set + deleted_at = excluded.deleted_at, + deletion_source = excluded.deletion_source, + deletion_reason = excluded.deletion_reason, + updated_at = excluded.updated_at; -- name: DeleteOrphanChannels :exec delete from channels diff --git a/internal/store/sqlc/schema.sql b/internal/store/sqlc/schema.sql index f0fe0914..1e4c620e 100644 --- a/internal/store/sqlc/schema.sql +++ b/internal/store/sqlc/schema.sql @@ -3,7 +3,10 @@ create table guilds ( name text not null, icon text, raw_json text not null, - updated_at text not null + updated_at text not null, + deleted_at text, + deletion_source text, + deletion_reason text ); create table channels ( @@ -38,6 +41,9 @@ create table members ( role_ids_json text not null, raw_json text not null, updated_at text not null, + deleted_at text, + deletion_source text, + deletion_reason text, primary key (guild_id, user_id) ); diff --git a/internal/store/store.go b/internal/store/store.go index e8d03468..df0cd818 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -17,7 +17,7 @@ const ( timeLayout = "2006-01-02T15:04:05.000000000Z07:00" messageFTSVersion = "2" memberFTSVersion = "1" - storeSchemaVersion = 4 + storeSchemaVersion = 5 ) var ErrSchemaVersionMismatch = errors.New("database schema version mismatch") @@ -199,6 +199,15 @@ func (s *Store) migrate(ctx context.Context) error { if err := s.setSchemaVersion(ctx, 4); err != nil { return err } + currentVersion = 4 + } + if currentVersion < 5 { + if err := s.applyEntityTombstoneMigration(ctx); err != nil { + return err + } + if err := s.setSchemaVersion(ctx, 5); err != nil { + return err + } } if version, err := s.schemaVersion(ctx); err != nil { return err @@ -214,6 +223,9 @@ func (s *Store) migrate(ctx context.Context) error { if err := s.applyFailureLedgerMigration(ctx); err != nil { return err } + if err := s.applyEntityTombstoneMigration(ctx); err != nil { + return err + } if err := s.ensureFTSRowIDs(ctx); err != nil { return err } @@ -314,7 +326,10 @@ func (s *Store) applyBaselineSchema(ctx context.Context) error { name text not null, icon text, raw_json text not null, - updated_at text not null + updated_at text not null, + deleted_at text, + deletion_source text, + deletion_reason text );`, `create table if not exists channels ( id text primary key, @@ -347,6 +362,9 @@ func (s *Store) applyBaselineSchema(ctx context.Context) error { role_ids_json text not null, raw_json text not null, updated_at text not null, + deleted_at text, + deletion_source text, + deletion_reason text, primary key (guild_id, user_id) );`, `create table if not exists messages ( @@ -563,6 +581,29 @@ func (s *Store) applyFailureLedgerMigration(ctx context.Context) error { return tx.Commit() } +func (s *Store) applyEntityTombstoneMigration(ctx context.Context) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer rollback(tx) + for _, table := range []string{"guilds", "members"} { + for _, column := range []string{"deleted_at", "deletion_source", "deletion_reason"} { + exists, err := columnExists(ctx, tx, table, column) + if err != nil { + return err + } + if exists { + continue + } + if _, err := tx.ExecContext(ctx, "alter table "+table+" add column "+column+" text"); err != nil { + return fmt.Errorf("migrate %s.%s: %w", table, column, err) + } + } + } + return tx.Commit() +} + func (s *Store) applyQueryIndexMigration(ctx context.Context) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { diff --git a/internal/store/store_test.go b/internal/store/store_test.go index deba64dc..cd9e359c 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -226,9 +226,9 @@ func TestClosedStoreOperationsReturnErrors(t *testing.T) { require.Error(t, s.UpsertGuild(ctx, GuildRecord{ID: "g1"})) require.Error(t, s.UpsertChannel(ctx, ChannelRecord{ID: "c1"})) - require.Error(t, s.ReplaceMembers(ctx, "g1", nil)) + require.Error(t, s.MergeMembers(ctx, "g1", nil)) require.Error(t, s.UpsertMember(ctx, MemberRecord{GuildID: "g1", UserID: "u1"})) - require.Error(t, s.DeleteMember(ctx, "g1", "u1")) + require.Error(t, s.MarkMemberDeleted(ctx, "g1", "u1", "test", "closed-store")) require.Error(t, s.UpsertMessageWithOptions(ctx, MessageRecord{ID: "m1"}, WriteOptions{})) require.Error(t, s.UpsertMessages(ctx, []MessageMutation{{Record: MessageRecord{ID: "m1"}}})) require.NoError(t, s.UpsertMessages(ctx, nil)) @@ -304,7 +304,7 @@ func TestStoreReadWriteAndSearch(t *testing.T) { require.NoError(t, s.UpsertGuild(ctx, GuildRecord{ID: "g1", Name: "Guild", RawJSON: `{}`})) require.NoError(t, s.UpsertChannel(ctx, ChannelRecord{ID: "c1", GuildID: "g1", Kind: "text", Name: "general", RawJSON: `{}`})) require.NoError(t, s.UpsertChannel(ctx, ChannelRecord{ID: "t1", GuildID: "g1", Kind: "thread_public", Name: "thread", RawJSON: `{}`})) - require.NoError(t, s.ReplaceMembers(ctx, "g1", []MemberRecord{{ + require.NoError(t, s.MergeMembers(ctx, "g1", []MemberRecord{{ GuildID: "g1", UserID: "u1", Username: "peter", @@ -1401,6 +1401,43 @@ func TestOpenMigratesUnversionedV1SchemaToV2(t *testing.T) { require.Equal(t, "0", rows[0][0]) } +func TestOpenMigratesV4MemberAndGuildRowsLosslessly(t *testing.T) { + t.Parallel() + + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "discrawl.db") + require.NoError(t, createV1Schema(ctx, dbPath)) + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + _, err = db.ExecContext(ctx, ` + insert into guilds(id, name, raw_json, updated_at) values('g1', 'Legacy Guild', '{"legacy":true}', '2026-07-17T00:00:00Z'); + insert into members(guild_id, user_id, username, role_ids_json, raw_json, updated_at) + values('g1', 'u1', 'legacy-user', '[]', '{"legacy":true}', '2026-07-17T00:00:01Z'); + pragma user_version = 4; + `) + require.NoError(t, err) + require.NoError(t, db.Close()) + + s, err := Open(ctx, dbPath) + require.NoError(t, err) + defer func() { _ = s.Close() }() + var version int + require.NoError(t, s.DB().QueryRowContext(ctx, `pragma user_version`).Scan(&version)) + require.Equal(t, storeSchemaVersion, version) + var guildName, username string + var tombstoneFields int + require.NoError(t, s.DB().QueryRowContext(ctx, ` + select g.name, m.username, + (g.deleted_at is not null) + (g.deletion_source is not null) + (g.deletion_reason is not null) + + (m.deleted_at is not null) + (m.deletion_source is not null) + (m.deletion_reason is not null) + from guilds g join members m on m.guild_id = g.id + where g.id = 'g1' and m.user_id = 'u1' + `).Scan(&guildName, &username, &tombstoneFields)) + require.Equal(t, "Legacy Guild", guildName) + require.Equal(t, "legacy-user", username) + require.Zero(t, tombstoneFields) +} + func TestOpenHealsVersionTwoMissingEmbeddingStorage(t *testing.T) { t.Parallel() @@ -1607,7 +1644,7 @@ func TestQueryAndExec(t *testing.T) { require.Error(t, err) } -func TestUpsertAndDeleteMember(t *testing.T) { +func TestUpsertAndTombstoneMember(t *testing.T) { t.Parallel() ctx := context.Background() @@ -1630,12 +1667,12 @@ func TestUpsertAndDeleteMember(t *testing.T) { require.Equal(t, "steipete", rows[0].GitHubLogin) require.Equal(t, "https://steipete.me", rows[0].Website) - require.NoError(t, s.DeleteMember(ctx, "g1", "u1")) + require.NoError(t, s.MarkMemberDeleted(ctx, "g1", "u1", "test", "explicit-delete")) rows, err = s.MemberByID(ctx, "u1") require.NoError(t, err) require.Empty(t, rows) - require.NoError(t, s.ReplaceMembers(ctx, "g1", []MemberRecord{{ + require.NoError(t, s.MergeMembers(ctx, "g1", []MemberRecord{{ GuildID: "g1", UserID: "u2", Username: "other", @@ -1649,6 +1686,71 @@ func TestUpsertAndDeleteMember(t *testing.T) { require.Equal(t, "Other bio", rows[0].Bio) } +func TestMemberAndGuildTombstonesRestoreAndMissingRowsStayLive(t *testing.T) { + t.Parallel() + + ctx := context.Background() + s, err := Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + defer func() { _ = s.Close() }() + require.NoError(t, s.UpsertGuild(ctx, GuildRecord{ID: "g1", Name: "Guild", RawJSON: `{}`})) + require.NoError(t, s.MarkGuildDeleted(ctx, "unseen-guild", "discord-gateway", "guild-delete-event")) + require.NoError(t, s.MarkMemberDeleted(ctx, "unseen-guild", "unseen-user", "discord-gateway", "member-remove-event")) + var unseenTombstones int + require.NoError(t, s.DB().QueryRowContext(ctx, ` + select (select count(*) from guilds where id = 'unseen-guild' and deleted_at is not null) + + (select count(*) from members where guild_id = 'unseen-guild' and user_id = 'unseen-user' and deleted_at is not null) + `).Scan(&unseenTombstones)) + require.Equal(t, 2, unseenTombstones, "explicit removals must survive even when the live row was never observed") + for _, userID := range []string{"u1", "u2"} { + require.NoError(t, s.UpsertMember(ctx, MemberRecord{ + GuildID: "g1", UserID: userID, Username: userID, DisplayName: userID, + RoleIDsJSON: `[]`, RawJSON: `{}`, + })) + } + require.NoError(t, s.MergeMembers(ctx, "g1", []MemberRecord{{ + GuildID: "g1", UserID: "u1", Username: "u1-new", DisplayName: "U1 New", + RoleIDsJSON: `[]`, RawJSON: `{}`, + }})) + members, err := s.Members(ctx, "g1", "", 10) + require.NoError(t, err) + require.Len(t, members, 2, "an omitted member is not an explicit deletion") + + require.NoError(t, s.MarkMemberDeleted(ctx, "g1", "u1", "discord-gateway", "member-remove-event")) + require.NoError(t, s.MarkGuildDeleted(ctx, "g1", "discord-gateway", "guild-delete-event")) + status, err := s.Status(ctx, "db", "") + require.NoError(t, err) + require.Zero(t, status.GuildCount) + require.Equal(t, 1, status.MemberCount) + members, err = s.Members(ctx, "g1", "u1", 10) + require.NoError(t, err) + require.Empty(t, members) + var memberSource, memberReason, guildSource, guildReason string + require.NoError(t, s.DB().QueryRowContext(ctx, `select deletion_source, deletion_reason from members where guild_id = 'g1' and user_id = 'u1'`).Scan(&memberSource, &memberReason)) + require.NoError(t, s.DB().QueryRowContext(ctx, `select deletion_source, deletion_reason from guilds where id = 'g1'`).Scan(&guildSource, &guildReason)) + require.Equal(t, "discord-gateway", memberSource) + require.Equal(t, "member-remove-event", memberReason) + require.Equal(t, "discord-gateway", guildSource) + require.Equal(t, "guild-delete-event", guildReason) + + require.NoError(t, s.UpsertGuild(ctx, GuildRecord{ID: "g1", Name: "Restored Guild", RawJSON: `{}`})) + require.NoError(t, s.UpsertMember(ctx, MemberRecord{ + GuildID: "g1", UserID: "u1", Username: "restored", DisplayName: "Restored", + RoleIDsJSON: `[]`, RawJSON: `{}`, + })) + var tombstoneFields int + require.NoError(t, s.DB().QueryRowContext(ctx, ` + select (g.deleted_at is not null) + (g.deletion_source is not null) + (g.deletion_reason is not null) + + (m.deleted_at is not null) + (m.deletion_source is not null) + (m.deletion_reason is not null) + from guilds g join members m on m.guild_id = g.id + where g.id = 'g1' and m.user_id = 'u1' + `).Scan(&tombstoneFields)) + require.Zero(t, tombstoneFields) + members, err = s.Members(ctx, "g1", "Restored", 10) + require.NoError(t, err) + require.Len(t, members, 1) +} + func TestOpenTightensDBFilePerms(t *testing.T) { t.Parallel() diff --git a/internal/store/storedb/models.go b/internal/store/storedb/models.go index a28d3bec..514a3f3d 100644 --- a/internal/store/storedb/models.go +++ b/internal/store/storedb/models.go @@ -56,27 +56,33 @@ type FailureLedger struct { } type Guild struct { - ID string - Name string - Icon sql.NullString - RawJson string - UpdatedAt string + ID string + Name string + Icon sql.NullString + RawJson string + UpdatedAt string + DeletedAt sql.NullString + DeletionSource sql.NullString + DeletionReason sql.NullString } type Member struct { - GuildID string - UserID string - Username string - GlobalName sql.NullString - DisplayName sql.NullString - Nick sql.NullString - Discriminator sql.NullString - Avatar sql.NullString - Bot int64 - JoinedAt sql.NullString - RoleIdsJson string - RawJson string - UpdatedAt string + GuildID string + UserID string + Username string + GlobalName sql.NullString + DisplayName sql.NullString + Nick sql.NullString + Discriminator sql.NullString + Avatar sql.NullString + Bot int64 + JoinedAt sql.NullString + RoleIdsJson string + RawJson string + UpdatedAt string + DeletedAt sql.NullString + DeletionSource sql.NullString + DeletionReason sql.NullString } type MemberFt struct { diff --git a/internal/store/storedb/queries.sql.go b/internal/store/storedb/queries.sql.go index 9b2eb1be..e54a5557 100644 --- a/internal/store/storedb/queries.sql.go +++ b/internal/store/storedb/queries.sql.go @@ -82,7 +82,7 @@ func (q *Queries) CountEmbeddingJobsByMessage(ctx context.Context, messageID str } const countGuilds = `-- name: CountGuilds :one -select count(*) as count from guilds +select count(*) as count from guilds where deleted_at is null ` func (q *Queries) CountGuilds(ctx context.Context) (int64, error) { @@ -93,7 +93,7 @@ func (q *Queries) CountGuilds(ctx context.Context) (int64, error) { } const countMembers = `-- name: CountMembers :one -select count(*) as count from members +select count(*) as count from members where deleted_at is null ` func (q *Queries) CountMembers(ctx context.Context) (int64, error) { @@ -106,7 +106,7 @@ func (q *Queries) CountMembers(ctx context.Context) (int64, error) { const countMembersByGuild = `-- name: CountMembersByGuild :one select count(*) as count from members -where guild_id = ? +where guild_id = ? and deleted_at is null ` func (q *Queries) CountMembersByGuild(ctx context.Context, guildID string) (int64, error) { @@ -190,21 +190,6 @@ func (q *Queries) DeleteGuild(ctx context.Context, id string) error { return err } -const deleteMember = `-- name: DeleteMember :exec -delete from members -where guild_id = ? and user_id = ? -` - -type DeleteMemberParams struct { - GuildID string - UserID string -} - -func (q *Queries) DeleteMember(ctx context.Context, arg DeleteMemberParams) error { - _, err := q.db.ExecContext(ctx, deleteMember, arg.GuildID, arg.UserID) - return err -} - const deleteMembersByGuild = `-- name: DeleteMembersByGuild :exec delete from members where guild_id = ? @@ -303,7 +288,7 @@ func (q *Queries) DeleteSyncState(ctx context.Context, scope string) error { const getGuildName = `-- name: GetGuildName :one select name from guilds -where id = ? +where id = ? and deleted_at is null ` func (q *Queries) GetGuildName(ctx context.Context, id string) (string, error) { @@ -375,48 +360,6 @@ func (q *Queries) HasMessageEmbeddings(ctx context.Context, arg HasMessageEmbedd return present, err } -const insertMember = `-- name: InsertMember :exec -insert into members( - guild_id, user_id, username, global_name, display_name, nick, discriminator, - avatar, bot, joined_at, role_ids_json, raw_json, updated_at -) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) -` - -type InsertMemberParams struct { - GuildID string - UserID string - Username string - GlobalName sql.NullString - DisplayName sql.NullString - Nick sql.NullString - Discriminator sql.NullString - Avatar sql.NullString - Bot int64 - JoinedAt sql.NullString - RoleIdsJson string - RawJson string - UpdatedAt string -} - -func (q *Queries) InsertMember(ctx context.Context, arg InsertMemberParams) error { - _, err := q.db.ExecContext(ctx, insertMember, - arg.GuildID, - arg.UserID, - arg.Username, - arg.GlobalName, - arg.DisplayName, - arg.Nick, - arg.Discriminator, - arg.Avatar, - arg.Bot, - arg.JoinedAt, - arg.RoleIdsJson, - arg.RawJson, - arg.UpdatedAt, - ) - return err -} - const insertMentionEvent = `-- name: InsertMentionEvent :exec insert into mention_events( message_id, guild_id, channel_id, author_id, target_type, target_id, target_name, event_at @@ -758,6 +701,7 @@ func (q *Queries) ListExistingAttachmentMedia(ctx context.Context, messageID str const listGuildIDs = `-- name: ListGuildIDs :many select id from guilds +where deleted_at is null order by id ` @@ -871,6 +815,7 @@ select guild_id, user_id, username, coalesce(global_name, '') as global_name, coalesce(discriminator, '') as discriminator, coalesce(avatar, '') as avatar, role_ids_json, bot, coalesce(joined_at, '') as joined_at, raw_json from members +where deleted_at is null order by coalesce(nullif(display_name, ''), nullif(nick, ''), nullif(global_name, ''), username), username limit ? ` @@ -932,7 +877,7 @@ select guild_id, user_id, username, coalesce(global_name, '') as global_name, coalesce(discriminator, '') as discriminator, coalesce(avatar, '') as avatar, role_ids_json, bot, coalesce(joined_at, '') as joined_at, raw_json from members -where guild_id = ? +where guild_id = ? and deleted_at is null order by coalesce(nullif(display_name, ''), nullif(nick, ''), nullif(global_name, ''), username), username limit ? ` @@ -999,7 +944,7 @@ select guild_id, user_id, username, coalesce(global_name, '') as global_name, coalesce(discriminator, '') as discriminator, coalesce(avatar, '') as avatar, role_ids_json, bot, coalesce(joined_at, '') as joined_at, raw_json from members -where user_id = ? +where user_id = ? and deleted_at is null order by guild_id, username ` @@ -1141,7 +1086,7 @@ select m.pinned from messages m left join channels c on c.id = m.channel_id -left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id +left join members mem on mem.guild_id = m.guild_id and mem.user_id = m.author_id and mem.deleted_at is null where m.guild_id = ? and m.author_id = ? order by m.created_at desc, m.id desc limit ? @@ -1364,6 +1309,74 @@ func (q *Queries) MarkEmptyEmbeddingJobDone(ctx context.Context, arg MarkEmptyEm return err } +const markGuildDeleted = `-- name: MarkGuildDeleted :exec +insert into guilds(id, name, icon, raw_json, updated_at, deleted_at, deletion_source, deletion_reason) +values( + ?1, ?1, null, '{}', ?2, + ?3, ?4, ?5 +) +on conflict(id) do update set + deleted_at = excluded.deleted_at, + deletion_source = excluded.deletion_source, + deletion_reason = excluded.deletion_reason, + updated_at = excluded.updated_at +` + +type MarkGuildDeletedParams struct { + ID string + UpdatedAt string + DeletedAt sql.NullString + DeletionSource sql.NullString + DeletionReason sql.NullString +} + +func (q *Queries) MarkGuildDeleted(ctx context.Context, arg MarkGuildDeletedParams) error { + _, err := q.db.ExecContext(ctx, markGuildDeleted, + arg.ID, + arg.UpdatedAt, + arg.DeletedAt, + arg.DeletionSource, + arg.DeletionReason, + ) + return err +} + +const markMemberDeleted = `-- name: MarkMemberDeleted :exec +insert into members( + guild_id, user_id, username, role_ids_json, raw_json, updated_at, + deleted_at, deletion_source, deletion_reason +) values( + ?1, ?2, ?2, '[]', '{}', ?3, + ?4, ?5, ?6 +) +on conflict(guild_id, user_id) do update set + deleted_at = excluded.deleted_at, + deletion_source = excluded.deletion_source, + deletion_reason = excluded.deletion_reason, + updated_at = excluded.updated_at +` + +type MarkMemberDeletedParams struct { + GuildID string + UserID string + UpdatedAt string + DeletedAt sql.NullString + DeletionSource sql.NullString + DeletionReason sql.NullString +} + +func (q *Queries) MarkMemberDeleted(ctx context.Context, arg MarkMemberDeletedParams) error { + _, err := q.db.ExecContext(ctx, markMemberDeleted, + arg.GuildID, + arg.UserID, + arg.UpdatedAt, + arg.DeletedAt, + arg.DeletionSource, + arg.DeletionReason, + ) + return err +} + const markMessageDeleted = `-- name: MarkMessageDeleted :exec update messages set deleted_at = ?, updated_at = ? @@ -1669,13 +1682,16 @@ func (q *Queries) UpsertEmbeddingJobPending(ctx context.Context, arg UpsertEmbed } const upsertGuild = `-- name: UpsertGuild :exec -insert into guilds(id, name, icon, raw_json, updated_at) -values(?, ?, ?, ?, ?) +insert into guilds(id, name, icon, raw_json, updated_at, deleted_at, deletion_source, deletion_reason) +values(?, ?, ?, ?, ?, null, null, null) on conflict(id) do update set name = excluded.name, icon = excluded.icon, raw_json = excluded.raw_json, - updated_at = excluded.updated_at + updated_at = excluded.updated_at, + deleted_at = null, + deletion_source = null, + deletion_reason = null ` type UpsertGuildParams struct { @@ -1700,8 +1716,9 @@ func (q *Queries) UpsertGuild(ctx context.Context, arg UpsertGuildParams) error const upsertMember = `-- name: UpsertMember :exec insert into members( guild_id, user_id, username, global_name, display_name, nick, discriminator, - avatar, bot, joined_at, role_ids_json, raw_json, updated_at -) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + avatar, bot, joined_at, role_ids_json, raw_json, updated_at, + deleted_at, deletion_source, deletion_reason +) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, null, null, null) on conflict(guild_id, user_id) do update set username = excluded.username, global_name = excluded.global_name, @@ -1713,7 +1730,10 @@ on conflict(guild_id, user_id) do update set joined_at = excluded.joined_at, role_ids_json = excluded.role_ids_json, raw_json = excluded.raw_json, - updated_at = excluded.updated_at + updated_at = excluded.updated_at, + deleted_at = null, + deletion_source = null, + deletion_reason = null ` type UpsertMemberParams struct { diff --git a/internal/store/write.go b/internal/store/write.go index 1021f58e..70174fec 100644 --- a/internal/store/write.go +++ b/internal/store/write.go @@ -127,26 +127,38 @@ func (s *Store) UpsertGuild(ctx context.Context, guild GuildRecord) error { }) } +func (s *Store) MarkGuildDeleted(ctx context.Context, guildID, source, reason string) error { + if strings.TrimSpace(guildID) == "" || strings.TrimSpace(source) == "" || strings.TrimSpace(reason) == "" { + return errors.New("guild tombstone requires guild id, source, and reason") + } + now := time.Now().UTC().Format(timeLayout) + return s.q.MarkGuildDeleted(ctx, storedb.MarkGuildDeletedParams{ + DeletedAt: nullString(now), + DeletionSource: nullString(source), + DeletionReason: nullString(reason), + UpdatedAt: now, + ID: guildID, + }) +} + func (s *Store) UpsertChannel(ctx context.Context, channel ChannelRecord) error { return s.q.UpsertChannel(ctx, upsertChannelParams(channel, time.Now().UTC().Format(timeLayout))) } -func (s *Store) ReplaceMembers(ctx context.Context, guildID string, members []MemberRecord) error { +// MergeMembers refreshes observed members without treating absence as deletion. +func (s *Store) MergeMembers(ctx context.Context, guildID string, members []MemberRecord) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer rollback(tx) qtx := s.q.WithTx(tx) - if err := qtx.DeleteMembersByGuild(ctx, guildID); err != nil { - return err - } - if _, err := tx.ExecContext(ctx, `delete from member_fts where guild_id = ?`, guildID); err != nil { - return err - } now := time.Now().UTC().Format(timeLayout) for _, member := range members { - if err := qtx.InsertMember(ctx, insertMemberParams(member, now)); err != nil { + if member.GuildID != guildID { + return fmt.Errorf("member guild %q does not match refresh guild %q", member.GuildID, guildID) + } + if err := qtx.UpsertMember(ctx, upsertMemberParams(member, now)); err != nil { return err } if err := upsertMemberFTSTx(ctx, tx, member); err != nil { @@ -156,6 +168,32 @@ func (s *Store) ReplaceMembers(ctx context.Context, guildID string, members []Me return tx.Commit() } +func (s *Store) MarkMemberDeleted(ctx context.Context, guildID, userID, source, reason string) error { + if strings.TrimSpace(guildID) == "" || strings.TrimSpace(userID) == "" || strings.TrimSpace(source) == "" || strings.TrimSpace(reason) == "" { + return errors.New("member tombstone requires guild id, user id, source, and reason") + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer rollback(tx) + now := time.Now().UTC().Format(timeLayout) + if err := s.q.WithTx(tx).MarkMemberDeleted(ctx, storedb.MarkMemberDeletedParams{ + DeletedAt: nullString(now), + DeletionSource: nullString(source), + DeletionReason: nullString(reason), + UpdatedAt: now, + GuildID: guildID, + UserID: userID, + }); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `delete from member_fts where rowid = ?`, memberFTSRowID(guildID, userID)); err != nil { + return err + } + return tx.Commit() +} + func (s *Store) UpsertMember(ctx context.Context, member MemberRecord) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { @@ -220,22 +258,6 @@ func (s *Store) DeleteOrphanChannels(ctx context.Context, guildID string) error return s.q.DeleteOrphanChannels(ctx, guildID) } -func (s *Store) DeleteMember(ctx context.Context, guildID, userID string) error { - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return err - } - defer rollback(tx) - qtx := s.q.WithTx(tx) - if err := qtx.DeleteMember(ctx, storedb.DeleteMemberParams{GuildID: guildID, UserID: userID}); err != nil { - return err - } - if _, err := tx.ExecContext(ctx, `delete from member_fts where rowid = ?`, memberFTSRowID(guildID, userID)); err != nil { - return err - } - return tx.Commit() -} - func (s *Store) UpsertMessage(ctx context.Context, message MessageRecord) error { return s.UpsertMessageWithOptions(ctx, message, WriteOptions{}) } @@ -682,24 +704,6 @@ func upsertChannelParams(channel ChannelRecord, now string) storedb.UpsertChanne } } -func insertMemberParams(member MemberRecord, now string) storedb.InsertMemberParams { - return storedb.InsertMemberParams{ - GuildID: member.GuildID, - UserID: member.UserID, - Username: member.Username, - GlobalName: nullString(member.GlobalName), - DisplayName: nullString(member.DisplayName), - Nick: nullString(member.Nick), - Discriminator: nullString(member.Discriminator), - Avatar: nullString(member.Avatar), - Bot: int64(boolInt(member.Bot)), - JoinedAt: nullString(member.JoinedAt), - RoleIdsJson: member.RoleIDsJSON, - RawJson: member.RawJSON, - UpdatedAt: now, - } -} - func upsertMemberParams(member MemberRecord, now string) storedb.UpsertMemberParams { return storedb.UpsertMemberParams{ GuildID: member.GuildID, diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index 2564ae23..f64cff84 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -305,9 +305,9 @@ func (s *Syncer) refreshGuildMembers(ctx context.Context, guildID string, force for _, member := range members { converted = append(converted, toMemberRecord(guildID, member)) } - if err := s.store.ReplaceMembers(ctx, guildID, converted); err != nil { - s.logger.Warn("member replace failed", "guild_id", guildID, "err", err) - return 0, fmt.Errorf("replace guild members: %w", err) + if err := s.store.MergeMembers(ctx, guildID, converted); err != nil { + s.logger.Warn("member merge failed", "guild_id", guildID, "err", err) + return 0, fmt.Errorf("merge guild members: %w", err) } if s.store != nil { if err := s.store.SetSyncState(ctx, guildMemberSyncSuccessScope(guildID), time.Now().UTC().Format(time.RFC3339Nano)); err != nil { diff --git a/internal/syncer/syncer_tail_test.go b/internal/syncer/syncer_tail_test.go index 44c4e55e..d8a2bf2d 100644 --- a/internal/syncer/syncer_tail_test.go +++ b/internal/syncer/syncer_tail_test.go @@ -82,6 +82,18 @@ func TestTailHandlerWritesEvents(t *testing.T) { defer func() { _ = s.Close() }() handler := &tailHandler{store: s} + require.NoError(t, handler.OnGuildUpsert(ctx, &discordgo.Guild{ID: "g1", Name: "Guild"})) + require.NoError(t, handler.OnGuildDelete(ctx, &discordgo.Guild{ID: "g1", Unavailable: true})) + var guildDeleted bool + require.NoError(t, s.DB().QueryRowContext(ctx, `select deleted_at is not null from guilds where id = 'g1'`).Scan(&guildDeleted)) + require.False(t, guildDeleted, "temporary Discord unavailability is not a deletion") + require.NoError(t, handler.OnGuildDelete(ctx, &discordgo.Guild{ID: "g1"})) + var guildSource, guildReason string + require.NoError(t, s.DB().QueryRowContext(ctx, `select deleted_at is not null, deletion_source, deletion_reason from guilds where id = 'g1'`).Scan(&guildDeleted, &guildSource, &guildReason)) + require.True(t, guildDeleted) + require.Equal(t, "discord-gateway", guildSource) + require.Equal(t, "guild-delete-event", guildReason) + require.NoError(t, handler.OnGuildUpsert(ctx, &discordgo.Guild{ID: "g1", Name: "Restored Guild"})) msg := &discordgo.Message{ ID: "9", GuildID: "g1", @@ -109,6 +121,12 @@ func TestTailHandlerWritesEvents(t *testing.T) { User: &discordgo.User{ID: "u1", Username: "peter"}, })) require.NoError(t, handler.OnMemberDelete(ctx, "g1", "u1")) + var memberDeleted bool + var memberSource, memberReason string + require.NoError(t, s.DB().QueryRowContext(ctx, `select deleted_at is not null, deletion_source, deletion_reason from members where guild_id = 'g1' and user_id = 'u1'`).Scan(&memberDeleted, &memberSource, &memberReason)) + require.True(t, memberDeleted) + require.Equal(t, "discord-gateway", memberSource) + require.Equal(t, "member-remove-event", memberReason) status, err := s.Status(context.Background(), "db", "") require.NoError(t, err) diff --git a/internal/syncer/tail.go b/internal/syncer/tail.go index 85e57e55..17862aa7 100644 --- a/internal/syncer/tail.go +++ b/internal/syncer/tail.go @@ -382,6 +382,25 @@ func (t *tailHandler) OnChannelUpsert(ctx context.Context, channel *discordgo.Ch return t.store.UpsertChannel(ctx, toChannelRecord(channel, marshalJSONString(channel, "{}"))) } +func (t *tailHandler) OnGuildUpsert(ctx context.Context, guild *discordgo.Guild) error { + if guild == nil || guild.Unavailable || !t.allowGuild(guild.ID) { + return nil + } + return t.store.UpsertGuild(ctx, store.GuildRecord{ + ID: guild.ID, + Name: guild.Name, + Icon: guild.Icon, + RawJSON: marshalJSONString(guild, "{}"), + }) +} + +func (t *tailHandler) OnGuildDelete(ctx context.Context, guild *discordgo.Guild) error { + if guild == nil || guild.Unavailable || !t.allowGuild(guild.ID) { + return nil + } + return t.store.MarkGuildDeleted(ctx, guild.ID, "discord-gateway", "guild-delete-event") +} + func (t *tailHandler) OnMemberUpsert(ctx context.Context, guildID string, member *discordgo.Member) error { if !t.allowGuild(guildID) || member == nil || member.User == nil { return nil @@ -393,7 +412,7 @@ func (t *tailHandler) OnMemberDelete(ctx context.Context, guildID, userID string if !t.allowGuild(guildID) { return nil } - return t.store.DeleteMember(ctx, guildID, userID) + return t.store.MarkMemberDeleted(ctx, guildID, userID, "discord-gateway", "member-remove-event") } func (t *tailHandler) TailAllowsGuild(guildID string) bool {