refactor(webapp,clickhouse,database): make logs search v2-only

Read v2 directly, use the projector enable setting as the operational switch, and store initialization in append-only checkpoints.

Author: carderne

apps/webapp/app/env.server.ts

@@ -2094,9 +2094,6 @@ const EnvironmentSchema = z
       .nonnegative()
       .optional(),
 
-    // Keep reads on v1 until the scheduled v2 projector has enough history.
-    LOGS_SEARCH_TABLE_VERSION: z.enum(["v1", "v2"]).default("v1"),
-
     // Scheduled logs-search projection. Disabled by default. LOGS_CLICKHOUSE_URL, or the
     // CLICKHOUSE_URL fallback, must reach both source and destination tables and allow writes.
     LOGS_SEARCH_PROJECTOR_ENABLED: BoolEnv.default(false),

apps/webapp/app/presenters/v3/LogsListPresenter.server.ts

@@ -227,10 +227,7 @@ export class LogsListPresenter extends BasePresenter {
     }
 
     const effectivePageSize = Math.min(pageSize, env.LOGS_LIST_MAX_PAGE_SIZE);
-    const usesV2Search = env.LOGS_SEARCH_TABLE_VERSION === "v2";
-    const queryLimit = usesV2Search
-      ? (effectivePageSize + 1) * LOGS_SEARCH_RETRY_OVERFETCH_FACTOR
-      : effectivePageSize + 1;
+    const queryLimit = (effectivePageSize + 1) * LOGS_SEARCH_RETRY_OVERFETCH_FACTOR;
 
     // Only honor a cursor scoped to this org+env; one copied from another scope would shift the
     // pagination anchor instead of resetting to the first page.
@@ -247,9 +244,7 @@ export class LogsListPresenter extends BasePresenter {
     const clampedTo = effectiveTo !== undefined ? (effectiveTo > now ? now : effectiveTo) : now;
 
     const rawSearchTerm = search?.trim() ?? "";
-    const normalizedSearchTerm = usesV2Search
-      ? normalizeLogsSearchTerm(rawSearchTerm)
-      : rawSearchTerm.toLowerCase();
+    const normalizedSearchTerm = normalizeLogsSearchTerm(rawSearchTerm);
     if (rawSearchTerm !== "" && !hasMinimumLogsSearchLength(normalizedSearchTerm)) {
       throw new ServiceValidationError(
         `Log searches must be at least ${MIN_LOGS_SEARCH_LENGTH} characters.`
@@ -261,11 +256,9 @@ export class LogsListPresenter extends BasePresenter {
     // Run exactly one bounded query. Broadening a search window is an explicit user action;
     // silently rescanning the same recent rows makes absence queries needlessly expensive.
     const runQuery = () => {
-      const queryBuilder = this.clickhouse.taskEventsSearch.logsListQueryBuilder(
-        env.LOGS_SEARCH_TABLE_VERSION
-      );
+      const queryBuilder = this.clickhouse.taskEventsSearch.logsListQueryBuilder();
 
-      // The materialized view excludes events without a trace_id; this guards the legacy tail.
+      // The projector excludes events without a trace_id.
       queryBuilder.where("trace_id != ''");
       queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
       queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
@@ -294,18 +287,9 @@ export class LogsListPresenter extends BasePresenter {
       }
 
       if (searchTerm !== undefined) {
-        if (usesV2Search) {
-          // One predicate lets the text index answer substring searches without an OR across
-          // independently indexed columns.
-          queryBuilder.where("search_text LIKE {searchPattern: String}", {
-            searchPattern: `%${searchTerm}%`,
-          });
-        } else {
-          queryBuilder.where(
-            "(lower(message) LIKE {searchPattern: String} OR lower(attributes_text) LIKE {searchPattern: String})",
-            { searchPattern: `%${searchTerm}%` }
-          );
-        }
+        queryBuilder.where("search_text LIKE {searchPattern: String}", {
+          searchPattern: `%${searchTerm}%`,
+        });
       }
 
       if (levels && levels.length > 0) {
@@ -343,12 +327,12 @@ export class LogsListPresenter extends BasePresenter {
           cursorTriggeredTimestamp: decodedCursor.triggeredTimestamp,
           cursorTraceId: decodedCursor.traceId,
           cursorSpanId: decodedCursor.spanId,
-          ...(usesV2Search && decodedCursor.projectionFingerprint
+          ...(decodedCursor.projectionFingerprint
             ? { cursorProjectionFingerprint: decodedCursor.projectionFingerprint }
             : {}),
         };
         queryBuilder.where(
-          usesV2Search && decodedCursor.projectionFingerprint
+          decodedCursor.projectionFingerprint
             ? `(triggered_timestamp < {cursorTriggeredTimestamp: String}
               OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String})
               OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String})
@@ -361,9 +345,7 @@ export class LogsListPresenter extends BasePresenter {
       }
 
       queryBuilder.orderBy(
-        usesV2Search
-          ? "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC"
-          : "triggered_timestamp DESC, trace_id DESC, span_id DESC"
+        "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC"
       );
       queryBuilder.limit(queryLimit);
 
@@ -379,12 +361,7 @@ export class LogsListPresenter extends BasePresenter {
     // marker. Keep the default throw behavior so the product never presents truncated results as
     // complete.
     const results = queryResult ?? [];
-    const page = usesV2Search
-      ? prepareLogsSearchPage(results, effectivePageSize, queryLimit)
-      : {
-          rows: results.slice(0, effectivePageSize),
-          hasMore: results.length > effectivePageSize,
-        };
+    const page = prepareLogsSearchPage(results, effectivePageSize, queryLimit);
     const hasMore = page.hasMore;
     const logs = page.rows;
 

apps/webapp/app/routes/admin.api.v1.logs-search-projector.ts

@@ -1,44 +1,8 @@
-import { type ActionFunctionArgs, type LoaderFunctionArgs, json } from "@remix-run/server-runtime";
-import { z } from "zod";
-import { LogsSearchProjectorConflictError } from "~/services/logsSearchProjector.server";
+import { type LoaderFunctionArgs, json } from "@remix-run/server-runtime";
 import { getLogsSearchProjector } from "~/services/logsSearchProjectorInstance.server";
-import { logger } from "~/services/logger.server";
 import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
 
-const Body = z.discriminatedUnion("action", [
-  z.object({ action: z.literal("pause") }),
-  z.object({ action: z.literal("resume") }),
-]);
-
 export async function loader({ request }: LoaderFunctionArgs) {
   await requireAdminApiRequest(request);
   return json(await getLogsSearchProjector().status());
 }
-
-export async function action({ request }: ActionFunctionArgs) {
-  const user = await requireAdminApiRequest(request);
-
-  try {
-    const body = Body.parse(await request.json());
-    const logsSearchProjector = getLogsSearchProjector();
-    logger.info("Updating logs search projector", { userId: user.id, action: body.action });
-
-    switch (body.action) {
-      case "pause":
-        return json(await logsSearchProjector.pause());
-      case "resume":
-        return json(await logsSearchProjector.resume());
-    }
-  } catch (error) {
-    if (error instanceof LogsSearchProjectorConflictError) {
-      return json({ error: error.message }, { status: 409 });
-    }
-    if (error instanceof z.ZodError || error instanceof SyntaxError) {
-      return json(
-        { error: error instanceof Error ? error.message : String(error) },
-        { status: 400 }
-      );
-    }
-    throw error;
-  }
-}