feat: Add full text search over chat messages (#27126)

Closes CODAGT-721 Closes CODAGT-722 Closes CODAGT-723 Closes CODAGT-724 Closes CODAGT-725 This PR adds the database and API pieces necessary to support full-text chat message search. - Adds required chat schema for full-text search - Adds dbpurge job to populate search_tsv in the background - Adds `search` parameter to GetChats query - Adds `search` filter to `searchquery.Chats` - Wires chat search filter into chats API > Implemented by Coder Agents, reviewed and tested by a human.
2026-09-24 15:04:27 +08:00 · 2026-07-16 15:21:57 +01:00
parent 9862f10484
commit f7481c5d08
26 changed files with 1756 additions and 35 deletions
@@ -51,6 +51,12 @@ const (
 	// Chat debug run deletions can cascade into steps with large JSONB
 	// payloads, so they use the same conservative batch size.
 	chatDebugRunsBatchSize = 1000
+	// Chat search tsvector backfill is capped at 5 batches of 10k
+	// rows per tick. Benchmarks on a dogfood-class machine (EPYC 9454P)
+	// with containerized Postgres were measured to take ~800ms per batch.
+	// This is considered acceptable but may need dialing in later.
+	chatSearchBackfillBatchSize  = 10000
+	chatSearchBackfillMaxBatches = 5
 )

 type Option func(*instance)
@@ -61,6 +67,14 @@ func WithClock(clk quartz.Clock) Option {
 	return func(i *instance) { i.clk = clk }
 }

+// WithChatSearchBackfillLimits overrides backfill batch size and cap. For tests.
+func WithChatSearchBackfillLimits(batchSize int32, maxBatches int) Option {
+	return func(i *instance) {
+		i.chatSearchBackfillBatchSize = batchSize
+		i.chatSearchBackfillMaxBatches = maxBatches
+	}
+}
+
 // New creates a new periodically purging database instance.
 // Callers must Close the returned instance.
 func New(ctx context.Context, logger slog.Logger, db database.Store, vals *codersdk.DeploymentValues, reg prometheus.Registerer, opts ...Option) io.Closer {
@@ -87,14 +101,25 @@ func New(ctx context.Context, logger slog.Logger, db database.Store, vals *coder
 	}, []string{"record_type"})
 	reg.MustRegister(recordsPurged)

+	chatSearchRowsBackfilled := prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: "coderd",
+		Subsystem: "dbpurge",
+		Name:      "chat_search_rows_backfilled_total",
+		Help:      "Total number of chat message rows whose search_tsv was backfilled.",
+	})
+	reg.MustRegister(chatSearchRowsBackfilled)
+
 	inst := &instance{
-		cancel:            cancelFunc,
-		closed:            closed,
-		logger:            logger,
-		vals:              vals,
-		clk:               quartz.NewReal(),
-		iterationDuration: iterationDuration,
-		recordsPurged:     recordsPurged,
+		cancel:                       cancelFunc,
+		closed:                       closed,
+		logger:                       logger,
+		vals:                         vals,
+		clk:                          quartz.NewReal(),
+		iterationDuration:            iterationDuration,
+		recordsPurged:                recordsPurged,
+		chatSearchRowsBackfilled:     chatSearchRowsBackfilled,
+		chatSearchBackfillBatchSize:  chatSearchBackfillBatchSize,
+		chatSearchBackfillMaxBatches: chatSearchBackfillMaxBatches,
 	}
 	for _, opt := range opts {
 		opt(inst)
@@ -310,6 +335,25 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
 			}
 		}

+		// Backfill search_tsv tsvector on chat_messages in batches. Doing this here because it's
+		// potentially too much for a regular migration, especially on larger deployments:
+		// - Each row with search_tsv = NULL is present in idx_chat_messages_search_tsv_pending.
+		// - Content of chat_messages is not changed after insert.
+		// - Rows that are soft-deleted are no longer part of the index.
+		// NOTE: This should not remain in dbpurge and should be adjusted when the "DBOps" gets
+		//   implemented.
+		var backfilledChatSearchRows int64
+		for range i.chatSearchBackfillMaxBatches {
+			n, err := tx.BackfillChatMessagesSearchTsv(ctx, i.chatSearchBackfillBatchSize)
+			if err != nil {
+				return xerrors.Errorf("backfill chat_messages.search_tsv: %w", err)
+			}
+			backfilledChatSearchRows += n
+			if n < int64(i.chatSearchBackfillBatchSize) {
+				break
+			}
+		}
+
 		i.logger.Debug(ctx, "purged old database entries",
 			slog.F("workspace_agent_logs", purgedWorkspaceAgentLogs),
 			slog.F("expired_api_keys", expiredAPIKeys),
@@ -322,6 +366,7 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
 			slog.F("chats", purgedChats),
 			slog.F("chat_files", purgedChatFiles),
 			slog.F("chat_debug_runs", purgedChatDebugRuns),
+			slog.F("chat_search_rows_backfilled", backfilledChatSearchRows),
 			slog.F("duration", i.clk.Since(start)),
 		)

@@ -338,6 +383,9 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
 			i.recordsPurged.WithLabelValues("chat_debug_runs").Add(float64(purgedChatDebugRuns))
 			i.recordsPurged.WithLabelValues("chat_files").Add(float64(purgedChatFiles))
 		}
+		if i.chatSearchRowsBackfilled != nil {
+			i.chatSearchRowsBackfilled.Add(float64(backfilledChatSearchRows))
+		}

 		// chatConfigErr is returned after the tx, so do not record this
 		// iteration as successful when only the deferred config read failed.
@@ -362,13 +410,16 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
 }

 type instance struct {
-	cancel            context.CancelFunc
-	closed            chan struct{}
-	logger            slog.Logger
-	vals              *codersdk.DeploymentValues
-	clk               quartz.Clock
-	iterationDuration *prometheus.HistogramVec
-	recordsPurged     *prometheus.CounterVec
+	cancel                       context.CancelFunc
+	closed                       chan struct{}
+	logger                       slog.Logger
+	vals                         *codersdk.DeploymentValues
+	clk                          quartz.Clock
+	iterationDuration            *prometheus.HistogramVec
+	recordsPurged                *prometheus.CounterVec
+	chatSearchRowsBackfilled     prometheus.Counter
+	chatSearchBackfillBatchSize  int32
+	chatSearchBackfillMaxBatches int
 }

 func (i *instance) Close() error {
@@ -14,6 +14,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/lib/pq"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sqlc-dev/pqtype"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
@@ -254,6 +255,7 @@ func TestMetrics(t *testing.T) {
 		mDB.EXPECT().DeleteOldTelemetryLocks(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		mDB.EXPECT().DeleteOldWorkspaceBuildOrchestrations(gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes()
 		mDB.EXPECT().DeleteOldAuditLogConnectionEvents(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+		mDB.EXPECT().BackfillChatMessagesSearchTsv(gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes()
 		mDB.EXPECT().DeleteOldChatDebugRuns(gomock.Any(), gomock.AssignableToTypeOf(database.DeleteOldChatDebugRunsParams{})).Return(int64(0), nil).MinTimes(1)
 		mDB.EXPECT().InTx(gomock.Any(), database.DefaultTXOptions().WithID("db_purge")).
 			DoAndReturn(func(f func(database.Store) error, _ *database.TxOptions) error {
@@ -305,6 +307,7 @@ func TestMetrics(t *testing.T) {
 		mDB.EXPECT().DeleteOldTelemetryLocks(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		mDB.EXPECT().DeleteOldWorkspaceBuildOrchestrations(gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes()
 		mDB.EXPECT().DeleteOldAuditLogConnectionEvents(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+		mDB.EXPECT().BackfillChatMessagesSearchTsv(gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes()
 		mDB.EXPECT().DeleteOldChats(gomock.Any(), gomock.AssignableToTypeOf(database.DeleteOldChatsParams{})).Return(int64(0), nil).MinTimes(1)
 		mDB.EXPECT().DeleteOldChatFiles(gomock.Any(), gomock.AssignableToTypeOf(database.DeleteOldChatFilesParams{})).Return(int64(0), nil).MinTimes(1)
 		mDB.EXPECT().InTx(gomock.Any(), database.DefaultTXOptions().WithID("db_purge")).
@@ -2861,3 +2864,375 @@ func TestDeleteOldChatFiles(t *testing.T) {
 		})
 	}
 }
+
+func awaitDoTicks(ctx context.Context, t *testing.T, clk *quartz.Mock, n int) func() {
+	t.Helper()
+	completed := make(chan struct{})
+	advance := make(chan struct{})
+	trapNow := clk.Trap().Now()
+	trapStop := clk.Trap().TickerStop()
+	trapReset := clk.Trap().TickerReset()
+	go func() {
+		defer close(completed)
+		defer trapReset.Close()
+		defer trapStop.Close()
+		defer trapNow.Close()
+		trapNow.MustWait(ctx).MustRelease(ctx)
+		trapReset.MustWait(ctx).MustRelease(ctx)
+		select {
+		case completed <- struct{}{}:
+		case <-ctx.Done():
+			return
+		}
+		for i := 1; i < n; i++ {
+			select {
+			case <-advance:
+			case <-ctx.Done():
+				return
+			}
+			d, w := clk.AdvanceNext()
+			if !assert.Equal(t, 10*time.Minute, d) {
+				return
+			}
+			w.MustWait(ctx)
+			trapStop.MustWait(ctx).MustRelease(ctx)
+			trapReset.MustWait(ctx).MustRelease(ctx)
+			select {
+			case completed <- struct{}{}:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	first := true
+	return func() {
+		t.Helper()
+		if !first {
+			testutil.RequireSend(ctx, t, advance, struct{}{})
+		}
+		first = false
+		testutil.TryReceive(ctx, t, completed)
+	}
+}
+
+//nolint:paralleltest // It uses LockIDDBPurge.
+func TestBackfillChatMessagesSearchTsv(t *testing.T) {
+	now := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC)
+
+	type chatSearchDeps struct {
+		user        database.User
+		modelConfig database.ChatModelConfig
+		chat        database.Chat
+	}
+	setupDeps := func(t *testing.T, db database.Store) chatSearchDeps {
+		t.Helper()
+		user := dbgen.User(t, db, database.User{})
+		org := dbgen.Organization(t, db, database.Organization{})
+		_ = dbgen.OrganizationMember(t, db, database.OrganizationMember{
+			UserID:         user.ID,
+			OrganizationID: org.ID,
+		})
+		_ = dbgen.ChatProvider(t, db, database.ChatProvider{
+			Provider:    "openai",
+			DisplayName: "OpenAI",
+		})
+		modelConfig := dbgen.ChatModelConfig(t, db, database.ChatModelConfig{
+			Model:        "test-model",
+			ContextLimit: 8192,
+		})
+		chat := dbgen.Chat(t, db, database.Chat{
+			OrganizationID:    org.ID,
+			OwnerID:           user.ID,
+			LastModelConfigID: modelConfig.ID,
+			Title:             "search-backfill-test-chat",
+		})
+		return chatSearchDeps{user: user, modelConfig: modelConfig, chat: chat}
+	}
+	textContent := func(text string) pqtype.NullRawMessage {
+		return pqtype.NullRawMessage{
+			RawMessage: json.RawMessage(fmt.Sprintf(`[{"type":"text","text":%q}]`, text)),
+			Valid:      true,
+		}
+	}
+	createMessage := func(t *testing.T, db database.Store, deps chatSearchDeps, role database.ChatMessageRole, visibility database.ChatMessageVisibility, content pqtype.NullRawMessage) database.ChatMessage {
+		t.Helper()
+		return dbgen.ChatMessage(t, db, database.ChatMessage{
+			ChatID:        deps.chat.ID,
+			CreatedBy:     uuid.NullUUID{UUID: deps.user.ID, Valid: true},
+			ModelConfigID: uuid.NullUUID{UUID: deps.modelConfig.ID, Valid: true},
+			Role:          role,
+			Visibility:    visibility,
+			Content:       content,
+		})
+	}
+	softDelete := func(ctx context.Context, t *testing.T, rawDB *sql.DB, id int64) {
+		t.Helper()
+		_, err := rawDB.ExecContext(ctx, "UPDATE chat_messages SET deleted = true WHERE id = $1", id)
+		require.NoError(t, err)
+	}
+	// The WHERE clause below must match the predicate of idx_chat_messages_search_tsv_pending.
+	countPending := func(ctx context.Context, t *testing.T, rawDB *sql.DB) int {
+		t.Helper()
+		var count int
+		err := rawDB.QueryRowContext(ctx, `
+			SELECT COUNT(*) FROM chat_messages
+			WHERE search_tsv IS NULL
+			  AND deleted = false
+			  AND visibility IN ('user', 'both')
+			  AND role IN ('user', 'assistant')`).Scan(&count)
+		require.NoError(t, err)
+		return count
+	}
+	searchTsv := func(ctx context.Context, t *testing.T, rawDB *sql.DB, id int64) (isNull bool, text string) {
+		t.Helper()
+		err := rawDB.QueryRowContext(ctx,
+			"SELECT search_tsv IS NULL, COALESCE(search_tsv::text, '') FROM chat_messages WHERE id = $1", id).
+			Scan(&isNull, &text)
+		require.NoError(t, err)
+		return isNull, text
+	}
+	requireBackfilled := func(ctx context.Context, t *testing.T, rawDB *sql.DB, id int64, msg string) {
+		t.Helper()
+		isNull, _ := searchTsv(ctx, t, rawDB, id)
+		require.False(t, isNull, msg)
+	}
+	// Asserts the row's tsvector matches expectedText, not just non-NULL.
+	requireTsvFor := func(ctx context.Context, t *testing.T, rawDB *sql.DB, id int64, expectedText string) {
+		t.Helper()
+		var matches bool
+		err := rawDB.QueryRowContext(ctx,
+			"SELECT search_tsv = to_tsvector('simple', $2::text) FROM chat_messages WHERE id = $1", id, expectedText).
+			Scan(&matches)
+		require.NoError(t, err)
+		require.True(t, matches, "search_tsv should contain the lexemes of %q", expectedText)
+	}
+	requireNotBackfilled := func(ctx context.Context, t *testing.T, rawDB *sql.DB, id int64, msg string) {
+		t.Helper()
+		isNull, _ := searchTsv(ctx, t, rawDB, id)
+		require.True(t, isNull, msg)
+	}
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("DrainConverges", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+
+		eligibleBoth := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent("hello world"))
+		eligibleUserVis := createMessage(t, db, deps, database.ChatMessageRoleAssistant, database.ChatMessageVisibilityUser, textContent("assistant reply"))
+		eligibleNoText := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, pqtype.NullRawMessage{RawMessage: json.RawMessage(`[]`), Valid: true})
+		toolMsg := createMessage(t, db, deps, database.ChatMessageRoleTool, database.ChatMessageVisibilityBoth, textContent("tool output"))
+		modelOnlyMsg := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityModel, textContent("model only"))
+		deletedMsg := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent("deleted message"))
+		softDelete(ctx, t, rawDB, deletedMsg.ID)
+
+		tick := awaitDoTicks(ctx, t, clk, 1)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
+		defer closer.Close()
+		tick()
+
+		require.Zero(t, countPending(ctx, t, rawDB), "queue should be drained")
+		requireTsvFor(ctx, t, rawDB, eligibleBoth.ID, "hello world")
+		requireTsvFor(ctx, t, rawDB, eligibleUserVis.ID, "assistant reply")
+		requireBackfilled(ctx, t, rawDB, eligibleNoText.ID, "eligible message with no text should be backfilled (sentinel)")
+		requireNotBackfilled(ctx, t, rawDB, toolMsg.ID, "tool message should never be backfilled")
+		requireNotBackfilled(ctx, t, rawDB, modelOnlyMsg.ID, "model-only message should never be backfilled")
+		requireNotBackfilled(ctx, t, rawDB, deletedMsg.ID, "deleted message should never be backfilled")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("BackfillsNewestFirst", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+
+		var ids []int64
+		for i := range 5 {
+			msg := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent(fmt.Sprintf("message %d", i)))
+			ids = append(ids, msg.ID)
+		}
+
+		tick := awaitDoTicks(ctx, t, clk, 1)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(),
+			dbpurge.WithClock(clk), dbpurge.WithChatSearchBackfillLimits(2, 1))
+		defer closer.Close()
+		tick()
+
+		slices.Sort(ids)
+		requireBackfilled(ctx, t, rawDB, ids[4], "newest message should be backfilled first")
+		requireBackfilled(ctx, t, rawDB, ids[3], "second-newest message should be backfilled first")
+		for _, id := range ids[:3] {
+			requireNotBackfilled(ctx, t, rawDB, id, "older messages should remain pending after one batch")
+		}
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("NoTextSentinel", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+
+		emptyArr := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, pqtype.NullRawMessage{RawMessage: json.RawMessage(`[]`), Valid: true})
+		noTextParts := createMessage(t, db, deps, database.ChatMessageRoleAssistant, database.ChatMessageVisibilityBoth, pqtype.NullRawMessage{RawMessage: json.RawMessage(`[{"type":"tool_call","id":"x"}]`), Valid: true})
+
+		tick := awaitDoTicks(ctx, t, clk, 1)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
+		defer closer.Close()
+		tick()
+
+		for _, id := range []int64{emptyArr.ID, noTextParts.ID} {
+			isNull, text := searchTsv(ctx, t, rawDB, id)
+			require.False(t, isNull, "no-text row should get the empty-tsvector sentinel, not stay NULL")
+			require.Empty(t, text, "no-text row should have an empty tsvector")
+		}
+		require.Zero(t, countPending(ctx, t, rawDB), "sentinel rows should not reappear as pending")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("PerTickBound", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+
+		for i := range 6 {
+			createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent(fmt.Sprintf("message %d", i)))
+		}
+
+		tick := awaitDoTicks(ctx, t, clk, 2)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(),
+			dbpurge.WithClock(clk), dbpurge.WithChatSearchBackfillLimits(2, 2))
+		defer closer.Close()
+
+		tick()
+		require.Equal(t, 2, countPending(ctx, t, rawDB), "one tick backfills at most maxBatches*batchSize rows")
+
+		tick()
+		require.Zero(t, countPending(ctx, t, rawDB), "next tick continues draining")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("SkipsDeletedRows", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+
+		msg := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent("soft deleted before backfill"))
+		softDelete(ctx, t, rawDB, msg.ID)
+		require.Zero(t, countPending(ctx, t, rawDB), "deleted rows should not appear as pending")
+
+		tick := awaitDoTicks(ctx, t, clk, 1)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
+		defer closer.Close()
+		tick()
+
+		requireNotBackfilled(ctx, t, rawDB, msg.ID, "deleted row should never be backfilled")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("BackfillsNewMessagesAfterDrain", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+
+		initial := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent("initial message"))
+
+		tick := awaitDoTicks(ctx, t, clk, 2)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
+		defer closer.Close()
+
+		tick()
+		requireBackfilled(ctx, t, rawDB, initial.ID, "initial message should be backfilled")
+		require.Zero(t, countPending(ctx, t, rawDB))
+
+		fresh := createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent("post drain message"))
+		tick()
+		requireBackfilled(ctx, t, rawDB, fresh.ID, "message inserted after drain should be backfilled on the next tick")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("SteadyStateNoop", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, rawDB := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		_ = setupDeps(t, db)
+		reg := prometheus.NewRegistry()
+
+		tick := awaitDoTicks(ctx, t, clk, 1)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
+		defer closer.Close()
+		tick()
+
+		require.Zero(t, countPending(ctx, t, rawDB))
+		backfilled := promhelp.CounterValue(t, reg, "coderd_dbpurge_chat_search_rows_backfilled_total", nil)
+		require.Zero(t, backfilled, "empty queue should backfill zero rows")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("MetricsCountsBackfilledRows", func(t *testing.T) {
+		ctx := testutil.Context(t, testutil.WaitLong)
+		clk := quartz.NewMock(t)
+		clk.Set(now).MustWait(ctx)
+		db, _, _ := dbtestutil.NewDBWithSQLDB(t, dbtestutil.WithDumpOnFailure())
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		deps := setupDeps(t, db)
+		reg := prometheus.NewRegistry()
+
+		for i := range 3 {
+			createMessage(t, db, deps, database.ChatMessageRoleUser, database.ChatMessageVisibilityBoth, textContent(fmt.Sprintf("message %d", i)))
+		}
+		createMessage(t, db, deps, database.ChatMessageRoleTool, database.ChatMessageVisibilityBoth, textContent("tool output"))
+
+		tick := awaitDoTicks(ctx, t, clk, 1)
+		closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
+		defer closer.Close()
+		tick()
+
+		backfilled := promhelp.CounterValue(t, reg, "coderd_dbpurge_chat_search_rows_backfilled_total", nil)
+		require.Equal(t, 3, backfilled, "counter should count exactly the eligible backfilled rows")
+	})
+
+	//nolint:paralleltest // It uses LockIDDBPurge.
+	t.Run("SkippedWhenLockHeld", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
+		defer cancel()
+
+		clk := quartz.NewMock(t)
+		ctrl := gomock.NewController(t)
+		mDB := dbmock.NewMockStore(ctrl)
+		mDB.EXPECT().GetChatRetentionDays(gomock.Any()).Return(int32(0), nil).AnyTimes()
+		mDB.EXPECT().GetChatDebugRetentionDays(gomock.Any(), codersdk.DefaultChatDebugRetentionDays).
+			Return(int32(0), nil).AnyTimes()
+		mDB.EXPECT().TryAcquireLock(gomock.Any(), int64(database.LockIDDBPurge)).Return(false, nil).AnyTimes()
+		mDB.EXPECT().BackfillChatMessagesSearchTsv(gomock.Any(), gomock.Any()).Times(0)
+		mDB.EXPECT().InTx(gomock.Any(), database.DefaultTXOptions().WithID("db_purge")).
+			DoAndReturn(func(f func(database.Store) error, _ *database.TxOptions) error {
+				return f(mDB)
+			}).MinTimes(1)
+
+		logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+		done := awaitDoTick(ctx, t, clk)
+		closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
+		defer closer.Close()
+		testutil.TryReceive(ctx, t, done)
+	})
+}