feat(scraper): implement efficient version refresh using ETags

This commit introduces a version refresh feature that enables efficient re-indexing of previously scraped library versions. By leveraging HTTP ETags, the new mechanism avoids re-processing unchanged pages, significantly reducing processing time, bandwidth usage, and embedding costs.

### Problem

Previously, re-indexing a library version was a wasteful process that required deleting all existing data and re-scraping every page from scratch, even if most of the content was unchanged.

### Solution

This implementation introduces a refresh mechanism that re-visits all previously scraped pages with their stored ETags. This allows the scraper to:
- **Skip** unchanged pages (HTTP 304 Not Modified).
- **Re-process** only pages that have changed (HTTP 200 OK).
- **Delete** documents for pages that are no longer available (HTTP 404 Not Found).

### Key Architectural Changes

- **Reused Existing Scraper Infrastructure**: The refresh operation is a standard scrape job with a pre-populated `initialQueue`. This leverages existing logic for progress tracking, error handling, and state management.
- **Database Schema Update**: A `depth` column has been added to the `pages` table to ensure that refresh operations respect the original `maxDepth` constraints. A database migration (`010-add-depth-to-pages.sql`) is included to apply this change.
- **Conditional Fetching**: The scraper's `processItem` logic has been updated to handle conditional requests. It now correctly processes 304, 404, and 200 HTTP responses to either skip, delete, or update documents.
- **Pipeline Manager Integration**: A new `enqueueRefreshJob` method was added to the `PipelineManager` to orchestrate the refresh process by fetching pages from the database and populating the `initialQueue`.

Author: arabold

db/migrations/010-add-depth-to-pages.sql

@@ -0,0 +1,10 @@
+-- Migration 010: Add depth column to pages table for refresh functionality
+-- This enables tracking the original crawl depth of each page, which is essential
+-- for maintaining consistent depth constraints during refresh operations.
+
+-- Add depth column to pages table
+ALTER TABLE pages ADD COLUMN depth INTEGER;
+
+-- Backfill existing pages with depth 0 (conservative default)
+-- This ensures all existing pages have a valid depth value
+UPDATE pages SET depth = 0 WHERE depth IS NULL;

src/cli/utils.ts

@@ -173,7 +173,7 @@ export async function createPipelineWithCallbacks(
 ): Promise<IPipeline> {
   logger.debug(`Initializing pipeline with options: ${JSON.stringify(options)}`);
   const { serverUrl, ...rest } = options;
-  const pipeline = serverUrl
+  const pipeline: IPipeline = serverUrl
     ? await PipelineFactory.createPipeline(undefined, { serverUrl, ...rest })
     : await (async () => {
         if (!docService) {
@@ -194,7 +194,7 @@ export async function createPipelineWithCallbacks(
     },
     onJobError: async (job, error, document) => {
       logger.warn(
-        `⚠️ Job ${job.id} error ${document ? `on document ${document.metadata.url}` : ""}: ${error.message}`,
+        `⚠️ Job ${job.id} error ${document ? `on document ${document.url}` : ""}: ${error.message}`,
       );
     },
   });

src/pipeline/PipelineClient.test.ts

@@ -6,7 +6,8 @@ vi.mock("../utils/logger");
 // Mock tRPC client factory
 const mockClient: any = {
   ping: { query: vi.fn() },
-  enqueueJob: { mutate: vi.fn() },
+  enqueueScrapeJob: { mutate: vi.fn() },
+  enqueueRefreshJob: { mutate: vi.fn() },
   getJob: { query: vi.fn() },
   getJobs: { query: vi.fn() },
   cancelJob: { mutate: vi.fn() },
@@ -28,7 +29,8 @@ describe("PipelineClient", () => {
     vi.resetAllMocks();
     // Reset default mock behaviors
     mockClient.ping.query.mockResolvedValue({ status: "ok" });
-    mockClient.enqueueJob.mutate.mockResolvedValue({ jobId: "job-123" });
+    mockClient.enqueueScrapeJob.mutate.mockResolvedValue({ jobId: "job-123" });
+    mockClient.enqueueRefreshJob.mutate.mockResolvedValue({ jobId: "job-456" });
     mockClient.getJob.query.mockResolvedValue(undefined);
     mockClient.getJobs.query.mockResolvedValue({ jobs: [] });
     mockClient.cancelJob.mutate.mockResolvedValue({ success: true });
@@ -50,18 +52,18 @@ describe("PipelineClient", () => {
     });
   });
 
-  describe("enqueueJob", () => {
+  describe("enqueueScrapeJob", () => {
     it("should delegate job creation to external API", async () => {
       const mockJobId = "job-123";
-      mockClient.enqueueJob.mutate.mockResolvedValueOnce({ jobId: mockJobId });
-      const jobId = await client.enqueueJob("react", "18.0.0", {
+      mockClient.enqueueScrapeJob.mutate.mockResolvedValueOnce({ jobId: mockJobId });
+      const jobId = await client.enqueueScrapeJob("react", "18.0.0", {
         url: "https://react.dev",
         library: "react",
         version: "18.0.0",
       });
 
       expect(jobId).toBe(mockJobId);
-      expect(mockClient.enqueueJob.mutate).toHaveBeenCalledWith({
+      expect(mockClient.enqueueScrapeJob.mutate).toHaveBeenCalledWith({
         library: "react",
         version: "18.0.0",
         options: {
@@ -73,9 +75,9 @@ describe("PipelineClient", () => {
     });
 
     it("should handle API errors gracefully", async () => {
-      mockClient.enqueueJob.mutate.mockRejectedValueOnce(new Error("Bad request"));
+      mockClient.enqueueScrapeJob.mutate.mockRejectedValueOnce(new Error("Bad request"));
 
-      await expect(client.enqueueJob("invalid", null, {} as any)).rejects.toThrow(
+      await expect(client.enqueueScrapeJob("invalid", null, {} as any)).rejects.toThrow(
         "Failed to enqueue job: Bad request",
       );
     });

src/pipeline/PipelineClient.ts

@@ -68,7 +68,7 @@ export class PipelineClient implements IPipeline {
     logger.debug("PipelineClient stopped");
   }
 
-  async enqueueJob(
+  async enqueueScrapeJob(
     library: string,
     version: string | undefined | null,
     options: ScraperOptions,
@@ -78,7 +78,7 @@ export class PipelineClient implements IPipeline {
         typeof version === "string" && version.trim().length === 0
           ? null
           : (version ?? null);
-      const result = await this.client.enqueueJob.mutate({
+      const result = await this.client.enqueueScrapeJob.mutate({
         library,
         version: normalizedVersion,
         options,
@@ -92,6 +92,28 @@ export class PipelineClient implements IPipeline {
     }
   }
 
+  async enqueueRefreshJob(
+    library: string,
+    version: string | undefined | null,
+  ): Promise<string> {
+    try {
+      const normalizedVersion =
+        typeof version === "string" && version.trim().length === 0
+          ? null
+          : (version ?? null);
+      const result = await this.client.enqueueRefreshJob.mutate({
+        library,
+        version: normalizedVersion,
+      });
+      logger.debug(`Refresh job ${result.jobId} enqueued successfully`);
+      return result.jobId;
+    } catch (error) {
+      throw new Error(
+        `Failed to enqueue refresh job: ${error instanceof Error ? error.message : String(error)}`,
+      );
+    }
+  }
+
   async getJob(jobId: string): Promise<PipelineJob | undefined> {
     try {
       const serializedJob = await this.client.getJob.query({ id: jobId });

src/pipeline/PipelineFactory.ts

@@ -24,8 +24,7 @@ export namespace PipelineFactory {
   // Overload: Remote pipeline client (out-of-process worker)
   export async function createPipeline(
     docService: undefined,
-    options: Required<Pick<PipelineOptions, "serverUrl">> &
-      Omit<PipelineOptions, "serverUrl">,
+    options: PipelineOptions & { serverUrl: string },
   ): Promise<PipelineClient>;
   // Implementation
   export async function createPipeline(