fix: auto-batch full sync backlog

steipete · steipete · commit ecf00b2820ce · 2026-03-14T05:20:14.000Z
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,7 @@ All notable changes to `discrawl` will be documented in this file.
 
 ## 0.2.0 - Unreleased
 
+- `sync --full` now auto-batches incomplete message channels from the local archive instead of stalling on large forum/thread catalogs
 - offline member-profile search via `member_fts`
 - `members search` now matches archived profile fields in addition to names
 - `members show` now accepts ids or queries and shows recent messages plus message stats when uniquely resolved
diff --git a/internal/store/query.go b/internal/store/query.go
@@ -241,6 +241,46 @@ func (s *Store) Channels(ctx context.Context, guildID string) ([]ChannelRow, err
 	return out, rows.Err()
 }
 
+func (s *Store) IncompleteMessageChannelIDs(ctx context.Context, guildID string) ([]string, error) {
+	args := []any{}
+	query := `
+		select c.id
+		from channels c
+		where c.kind in ('text', 'news', 'announcement', 'thread_public', 'thread_private', 'thread_news', 'thread_announcement')
+	`
+	if guildID != "" {
+		query += ` and c.guild_id = ?`
+		args = append(args, guildID)
+	}
+	query += `
+		and not exists (
+			select 1
+			from sync_state s
+			where s.scope = 'channel:' || c.id || ':history_complete'
+		)
+		and not exists (
+			select 1
+			from sync_state s
+			where s.scope = 'channel:' || c.id || ':unavailable'
+		)
+		order by c.id
+	`
+	rows, err := s.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	var out []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		out = append(out, id)
+	}
+	return out, rows.Err()
+}
+
 func (s *Store) Status(ctx context.Context, dbPath, defaultGuildID string) (Status, error) {
 	status := Status{DBPath: dbPath, DefaultGuildID: defaultGuildID}
 	queries := map[string]*int{
diff --git a/internal/syncer/channel_catalog.go b/internal/syncer/channel_catalog.go
@@ -70,7 +70,21 @@ func (s *Syncer) liveChannelList(ctx context.Context, guildID string) ([]*discor
 	for _, channel := range channels {
 		allChannels[channel.ID] = channel
 	}
-	if err := s.appendThreadCatalog(ctx, allChannels, threadParentIDs(channels)); err != nil {
+	var storedRows []store.ChannelRow
+	if s.store != nil {
+		rows, err := s.store.Channels(ctx, guildID)
+		if err != nil {
+			return nil, err
+		}
+		storedRows = rows
+		mergeStoredThreadChannels(allChannels, rows)
+	}
+	parentIDs := fullSyncThreadParentIDs(channels, storedRows)
+	if len(storedThreadParentIDs(storedRows)) == 0 {
+		if err := s.appendThreadCatalog(ctx, allChannels, parentIDs); err != nil {
+			return nil, err
+		}
+	} else if err := s.appendActiveThreadCatalog(ctx, allChannels, parentIDs); err != nil {
 		return nil, err
 	}
 	return mapsToSlice(allChannels), nil
@@ -82,11 +96,8 @@ func (s *Syncer) appendThreadCatalog(ctx context.Context, allChannels map[string
 		if !isThreadParent(channel) {
 			continue
 		}
-		active, err := s.client.ThreadsActive(ctx, channel.ID)
-		if err == nil {
-			for _, thread := range active {
-				allChannels[thread.ID] = thread
-			}
+		if err := s.appendActiveThreads(ctx, allChannels, channel.ID); err != nil {
+			return err
 		}
 		for _, private := range []bool{false, true} {
 			archived, err := s.client.ThreadsArchived(ctx, channel.ID, private)
@@ -102,6 +113,30 @@ func (s *Syncer) appendThreadCatalog(ctx context.Context, allChannels map[string
 	return nil
 }
 
+func (s *Syncer) appendActiveThreadCatalog(ctx context.Context, allChannels map[string]*discordgo.Channel, parents []string) error {
+	for _, parentID := range uniqueIDs(parents) {
+		channel := allChannels[parentID]
+		if !isThreadParent(channel) {
+			continue
+		}
+		if err := s.appendActiveThreads(ctx, allChannels, channel.ID); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *Syncer) appendActiveThreads(ctx context.Context, allChannels map[string]*discordgo.Channel, channelID string) error {
+	active, err := s.client.ThreadsActive(ctx, channelID)
+	if err != nil {
+		return err
+	}
+	for _, thread := range active {
+		allChannels[thread.ID] = thread
+	}
+	return nil
+}
+
 func mapsToSlice(in map[string]*discordgo.Channel) []*discordgo.Channel {
 	out := make([]*discordgo.Channel, 0, len(in))
 	for _, channel := range in {
@@ -111,6 +146,18 @@ func mapsToSlice(in map[string]*discordgo.Channel) []*discordgo.Channel {
 	return out
 }
 
+func mergeStoredThreadChannels(allChannels map[string]*discordgo.Channel, rows []store.ChannelRow) {
+	for _, row := range rows {
+		if !strings.HasPrefix(row.Kind, "thread_") {
+			continue
+		}
+		if _, ok := allChannels[row.ID]; ok {
+			continue
+		}
+		allChannels[row.ID] = channelFromRow(row)
+	}
+}
+
 func selectStoredChannels(rows []store.ChannelRow, requested map[string]struct{}) map[string]*discordgo.Channel {
 	if len(rows) == 0 || len(requested) == 0 {
 		return nil
@@ -120,30 +167,34 @@ func selectStoredChannels(rows []store.ChannelRow, requested map[string]struct{}
 		if _, ok := requested[row.ID]; !ok {
 			continue
 		}
-		channelType := channelTypeFromKind(row.Kind)
-		var threadMeta *discordgo.ThreadMetadata
-		if strings.HasPrefix(row.Kind, "thread_") {
-			threadMeta = &discordgo.ThreadMetadata{
-				Archived:         row.IsArchived,
-				Locked:           row.IsLocked,
-				ArchiveTimestamp: row.ArchiveTimestamp,
-			}
-		}
-		out[row.ID] = &discordgo.Channel{
-			ID:             row.ID,
-			GuildID:        row.GuildID,
-			ParentID:       row.ParentID,
-			Name:           row.Name,
-			Topic:          row.Topic,
-			Position:       row.Position,
-			NSFW:           row.IsNSFW,
-			Type:           channelType,
-			ThreadMetadata: threadMeta,
-		}
+		out[row.ID] = channelFromRow(row)
 	}
 	return out
 }
 
+func channelFromRow(row store.ChannelRow) *discordgo.Channel {
+	channelType := channelTypeFromKind(row.Kind)
+	var threadMeta *discordgo.ThreadMetadata
+	if strings.HasPrefix(row.Kind, "thread_") {
+		threadMeta = &discordgo.ThreadMetadata{
+			Archived:         row.IsArchived,
+			Locked:           row.IsLocked,
+			ArchiveTimestamp: row.ArchiveTimestamp,
+		}
+	}
+	return &discordgo.Channel{
+		ID:             row.ID,
+		GuildID:        row.GuildID,
+		ParentID:       row.ParentID,
+		Name:           row.Name,
+		Topic:          row.Topic,
+		Position:       row.Position,
+		NSFW:           row.IsNSFW,
+		Type:           channelType,
+		ThreadMetadata: threadMeta,
+	}
+}
+
 func canUseStoredTargets(storedByID map[string]*discordgo.Channel, requested map[string]struct{}) bool {
 	if len(requested) == 0 || len(storedByID) != len(requested) {
 		return false
@@ -232,6 +283,41 @@ func threadParentIDs(channels []*discordgo.Channel) []string {
 	return parents
 }
 
+func fullSyncThreadParentIDs(topLevel []*discordgo.Channel, storedRows []store.ChannelRow) []string {
+	storedParents := storedThreadParentIDs(storedRows)
+	if len(storedParents) == 0 {
+		return threadParentIDs(topLevel)
+	}
+	parents := make([]string, 0, len(topLevel))
+	for _, channel := range topLevel {
+		if channel == nil {
+			continue
+		}
+		if channel.Type == discordgo.ChannelTypeGuildForum {
+			parents = append(parents, channel.ID)
+			continue
+		}
+		if _, ok := storedParents[channel.ID]; ok && isThreadParent(channel) {
+			parents = append(parents, channel.ID)
+		}
+	}
+	return uniqueIDs(parents)
+}
+
+func storedThreadParentIDs(rows []store.ChannelRow) map[string]struct{} {
+	if len(rows) == 0 {
+		return nil
+	}
+	parents := make(map[string]struct{})
+	for _, row := range rows {
+		if !strings.HasPrefix(row.Kind, "thread_") || row.ParentID == "" {
+			continue
+		}
+		parents[row.ParentID] = struct{}{}
+	}
+	return parents
+}
+
 func uniqueIDs(ids []string) []string {
 	set := make(map[string]struct{}, len(ids))
 	out := make([]string, 0, len(ids))
diff --git a/internal/syncer/channel_catalog_test.go b/internal/syncer/channel_catalog_test.go
@@ -186,3 +186,89 @@ func TestSyncSkipsUnchangedThreadsWhenHistoryComplete(t *testing.T) {
 	require.Zero(t, stats.Messages)
 	require.Zero(t, client.messageCalls["t1"])
 }
+
+func TestFullSyncReusesStoredThreadParents(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db"))
+	require.NoError(t, err)
+	defer func() { _ = s.Close() }()
+
+	require.NoError(t, s.UpsertGuild(ctx, store.GuildRecord{ID: "g1", Name: "Guild", RawJSON: `{}`}))
+	require.NoError(t, s.UpsertChannel(ctx, store.ChannelRecord{
+		ID:      "c1",
+		GuildID: "g1",
+		Kind:    "text",
+		Name:    "general",
+		RawJSON: `{"id":"c1"}`,
+	}))
+	require.NoError(t, s.UpsertChannel(ctx, store.ChannelRecord{
+		ID:       "t1",
+		GuildID:  "g1",
+		ParentID: "c1",
+		Kind:     "thread_public",
+		Name:     "bug-report",
+		RawJSON:  `{"id":"t1"}`,
+	}))
+
+	client := &fakeClient{
+		guilds: []*discordgo.UserGuild{{ID: "g1", Name: "Guild"}},
+		guildByID: map[string]*discordgo.Guild{
+			"g1": {ID: "g1", Name: "Guild"},
+		},
+		channels: map[string][]*discordgo.Channel{
+			"g1": {
+				{ID: "c1", GuildID: "g1", Name: "general", Type: discordgo.ChannelTypeGuildText},
+				{ID: "c2", GuildID: "g1", Name: "random", Type: discordgo.ChannelTypeGuildText},
+			},
+		},
+		messages: map[string][]*discordgo.Message{
+			"t1": {{
+				ID:        "10",
+				GuildID:   "g1",
+				ChannelID: "t1",
+				Content:   "stored thread body",
+				Timestamp: time.Now().UTC(),
+				Author:    &discordgo.User{ID: "u1", Username: "user"},
+			}},
+		},
+	}
+
+	svc := New(client, s, nil)
+	stats, err := svc.Sync(ctx, SyncOptions{Full: true, GuildIDs: []string{"g1"}})
+	require.NoError(t, err)
+	require.Equal(t, 1, stats.Threads)
+	require.Equal(t, 1, stats.Messages)
+	require.Zero(t, client.threadCalls)
+	require.Equal(t, 1, client.messageCalls["t1"])
+}
+
+func TestFullSyncFallsBackToBroadThreadDiscoveryWithoutStoredThreads(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db"))
+	require.NoError(t, err)
+	defer func() { _ = s.Close() }()
+
+	require.NoError(t, s.UpsertGuild(ctx, store.GuildRecord{ID: "g1", Name: "Guild", RawJSON: `{}`}))
+
+	client := &fakeClient{
+		guilds: []*discordgo.UserGuild{{ID: "g1", Name: "Guild"}},
+		guildByID: map[string]*discordgo.Guild{
+			"g1": {ID: "g1", Name: "Guild"},
+		},
+		channels: map[string][]*discordgo.Channel{
+			"g1": {
+				{ID: "c1", GuildID: "g1", Name: "general", Type: discordgo.ChannelTypeGuildText},
+				{ID: "c2", GuildID: "g1", Name: "random", Type: discordgo.ChannelTypeGuildText},
+			},
+		},
+	}
+
+	svc := New(client, s, nil)
+	_, err = svc.Sync(ctx, SyncOptions{Full: true, GuildIDs: []string{"g1"}})
+	require.NoError(t, err)
+	require.Equal(t, 6, client.threadCalls)
+}
diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go
diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go
