Merge pull request #694 from rusq/review-679

[codex] fix skip complete threads resume

Author: rusq

cmd/slackdump/internal/resume/resume.go

@@ -74,9 +74,6 @@ type ResumeParams struct {
 	// SkipCompleteThreads skips threads where the database already holds all
 	// replies (DB count == API reply_count + 1).  Faster, but won't detect
 	// edited or deleted messages.  Use only when threads are append-only.
-	// Note: threads with parent messages older than the lookback window will
-	// not be checked for updates, as only channel messages within the lookback
-	// window are scanned to discover threads.
 	SkipCompleteThreads bool
 	// Dedupe runs duplicate entity cleanup after a successful resume.
 	Dedupe bool
@@ -92,7 +89,7 @@ func init() {
 	CmdResume.Flag.BoolVar(&resumeFlags.IncludeThreads, "threads", false, "include threads (slow, and flaky business)")
 	CmdResume.Flag.BoolVar(&resumeFlags.RecordOnlyNewUsers, "only-new-users", true, "record only new or updated users")
 	CmdResume.Flag.Var(resumeFlags.Lookback, "lookback", "lookback window `duration`")
-	CmdResume.Flag.BoolVar(&resumeFlags.SkipCompleteThreads, "skip-complete-threads", false, "skip threads where DB already has all replies (faster, but won't detect edits/deletes or new replies to messages older than lookback window)")
+	CmdResume.Flag.BoolVar(&resumeFlags.SkipCompleteThreads, "skip-complete-threads", false, "skip threads where DB already has all replies (faster, but won't detect edits/deletes)")
 	CmdResume.Flag.BoolVar(&resumeFlags.Dedupe, "dedupe", false, "run dedupe cleanup after successful resume finish")
 }
 
@@ -136,9 +133,6 @@ func runResume(ctx context.Context, cmd *base.Command, args []string) error {
 		return fmt.Errorf("source type %q does not support resume, use 'slackdump convert -f database' to convert it", src.Type())
 	}
 
-	if resumeFlags.IncludeThreads && resumeFlags.SkipCompleteThreads {
-		slog.WarnContext(ctx, "threads whose parent message is older than the lookback window will not be checked for new replies")
-	}
 	latest, err := latest(ctx, src, resumeFlags.IncludeThreads, resumeFlags.SkipCompleteThreads, time.Duration((*duration.Duration)(resumeFlags.Lookback).ToTimeDuration()), list)
 	if err != nil {
 		base.SetExitStatus(base.SApplicationError)
@@ -267,7 +261,7 @@ func latest(ctx context.Context, src source.Resumer, includeThreads bool, skipCo
 
 	ei := make([]structures.EntityItem, 0, len(latest))
 	for sl, ts := range latest {
-		if sl.IsThread() && (!includeThreads || skipCompleteThreads) {
+		if sl.IsThread() && !includeThreads {
 			continue
 		}
 		item := structures.EntityItem{

cmd/slackdump/internal/resume/resume_test.go

@@ -30,8 +30,8 @@ import (
 	"go.uber.org/mock/gomock"
 
 	"github.com/rusq/slackdump/v4"
-	dedupecmd "github.com/rusq/slackdump/v4/cmd/slackdump/internal/diag/dedupe"
 	"github.com/rusq/slackdump/v4/cmd/slackdump/internal/cfg"
+	dedupecmd "github.com/rusq/slackdump/v4/cmd/slackdump/internal/diag/dedupe"
 	"github.com/rusq/slackdump/v4/internal/fixtures"
 	"github.com/rusq/slackdump/v4/internal/structures"
 	"github.com/rusq/slackdump/v4/source"
@@ -476,6 +476,7 @@ func Test_latest(t *testing.T) {
 			},
 			want: structures.NewEntityListFromItems(
 				structures.EntityItem{Id: "C123", Oldest: time.Date(2021, 1, 1, 0, 0, 0, 0, time.UTC), Latest: time.Time(cfg.Latest), Include: true},
+				structures.EntityItem{Id: "C456:123.456", Oldest: time.Date(2021, 1, 2, 0, 0, 0, 0, time.UTC), Latest: time.Time(cfg.Latest), Include: true},
 			),
 			wantErr: false,
 		},

stream/conversation.go

@@ -244,6 +244,7 @@ func (cs *Stream) thread(ctx context.Context, req request, callback func(mm []sl
 	lg.DebugContext(ctx, "- getting thread")
 
 	var cursor string
+	firstPage := true
 	for {
 		var (
 			msgs    []slack.Message
@@ -281,6 +282,15 @@ func (cs *Stream) thread(ctx context.Context, req request, callback func(mm []sl
 			return err
 		}
 
+		if firstPage && req.threadOnly && cs.skipThread != nil {
+			replyCount := msgs[0].ReplyCount
+			if cs.skipThread(ctx, req.sl.Channel, req.sl.ThreadTS, replyCount) {
+				lg.DebugContext(ctx, "skipping complete thread", "channel_id", req.sl.Channel, "thread_ts", req.sl.ThreadTS, "reply_count", replyCount)
+				return nil
+			}
+		}
+		firstPage = false
+
 		r := trace.StartRegion(ctx, "thread_callback")
 		err := callback(msgs, !hasmore)
 		r.End()

stream/stream.go

@@ -180,9 +180,10 @@ func OptFailOnNonCritError(b bool) Option {
 }
 
 // OptSkipThreadFunc sets a predicate that is called for each thread head message
-// during channel processing.  If it returns true, the thread is skipped (no API
-// call is made).  Returns false on any error (safe default: re-fetch the thread).
-// Intended for resume operations to skip threads already complete in the database.
+// during channel processing and after the first replies page for direct thread
+// requests.  If it returns true, the thread is skipped.  Returns false on any
+// error (safe default: re-fetch the thread).  Intended for resume operations to
+// skip threads already complete in the database.
 func OptSkipThreadFunc(fn func(ctx context.Context, channelID, threadTS string, replyCount int) bool) Option {
 	return func(cs *Stream) {
 		cs.skipThread = fn

stream/stream_test.go

@@ -203,6 +203,85 @@ func TestStream_thread(t *testing.T) {
 			assert.Equal(t, 1, calls)
 		})
 	}
+
+	t.Run("skips complete direct thread after first page", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := mock_client.NewMockSlack(ctrl)
+		cs := New(ms, network.NoLimits, OptSkipThreadFunc(func(ctx context.Context, channelID, threadTS string, replyCount int) bool {
+			assert.Equal(t, "CTM1", channelID)
+			assert.Equal(t, "1610000000.000000", threadTS)
+			assert.Equal(t, 2, replyCount)
+			return true
+		}))
+		msgs := []slack.Message{
+			{Msg: slack.Msg{
+				Channel:         "CTM1",
+				Timestamp:       "1610000000.000000",
+				ThreadTimestamp: "1610000000.000000",
+				ReplyCount:      2,
+			}},
+		}
+		ms.EXPECT().
+			GetConversationRepliesContext(gomock.Any(), gomock.Any()).
+			Return(msgs, true, "next-cursor", nil).
+			Times(1)
+
+		var calls int
+		err := cs.thread(t.Context(), request{
+			sl:         &structures.SlackLink{Channel: "CTM1", ThreadTS: "1610000000.000000"},
+			threadOnly: true,
+		}, func(mm []slack.Message, isLast bool) error {
+			calls++
+			return nil
+		})
+		if err != nil {
+			t.Fatalf("thread() error = %v", err)
+		}
+		assert.Equal(t, 0, calls)
+	})
+
+	t.Run("processes incomplete direct thread after first page", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := mock_client.NewMockSlack(ctrl)
+		cs := New(ms, network.NoLimits, OptSkipThreadFunc(func(ctx context.Context, channelID, threadTS string, replyCount int) bool {
+			assert.Equal(t, "CTM1", channelID)
+			assert.Equal(t, "1610000000.000000", threadTS)
+			assert.Equal(t, 3, replyCount)
+			return false
+		}))
+		msgs := []slack.Message{
+			{Msg: slack.Msg{
+				Channel:         "CTM1",
+				Timestamp:       "1610000000.000000",
+				ThreadTimestamp: "1610000000.000000",
+				ReplyCount:      3,
+			}},
+			{Msg: slack.Msg{
+				Channel:         "CTM1",
+				Timestamp:       "1610000001.000000",
+				ThreadTimestamp: "1610000000.000000",
+			}},
+		}
+		ms.EXPECT().
+			GetConversationRepliesContext(gomock.Any(), gomock.Any()).
+			Return(msgs, false, "", nil).
+			Times(1)
+
+		var calls int
+		err := cs.thread(t.Context(), request{
+			sl:         &structures.SlackLink{Channel: "CTM1", ThreadTS: "1610000000.000000"},
+			threadOnly: true,
+		}, func(mm []slack.Message, isLast bool) error {
+			calls++
+			assert.True(t, isLast)
+			assert.Equal(t, msgs, mm)
+			return nil
+		})
+		if err != nil {
+			t.Fatalf("thread() error = %v", err)
+		}
+		assert.Equal(t, 1, calls)
+	})
 }
 
 func Test_processLink(t *testing.T) {