feat: Import workflow v3 (#2378)

* feat: import workflow v3

* batch stage

* revert migration

* cleanups

* pr comments

* move to models

* add allowed workers

* e2e tests

* import list ids

* add missing indicies

* merge test

* more fixes

* add resume/pause to UI

* fix ui states

* fix tests

* simplify progress tracking

* remove backpressure

* fix list imports

* fix race on claiming bookmarks

* remove the codex file

1 files changed, 200 insertions, 33 deletions

diff --git a/packages/trpc/routers/importSessions.test.ts b/packages/trpc/routers/importSessions.test.ts
index 9ef0de6f..f257ad3b 100644
--- a/packages/trpc/routers/importSessions.test.ts
+++ b/packages/trpc/routers/importSessions.test.ts
@@ -1,12 +1,13 @@
-import { eq } from "drizzle-orm";
 import { beforeEach, describe, expect, test } from "vitest";
 import { z } from "zod";
 
-import { bookmarks } from "@karakeep/db/schema";
 import {
-  BookmarkTypes,
-  zNewBookmarkRequestSchema,
-} from "@karakeep/shared/types/bookmarks";
+  bookmarkLinks,
+  bookmarks,
+  bookmarkTexts,
+  importStagingBookmarks,
+} from "@karakeep/db/schema";
+import { BookmarkTypes } from "@karakeep/shared/types/bookmarks";
 import {
   zCreateImportSessionRequestSchema,
   zDeleteImportSessionRequestSchema,
@@ -20,17 +21,6 @@ import { defaultBeforeEach } from "../testUtils";
 beforeEach<CustomTestContext>(defaultBeforeEach(true));
 
 describe("ImportSessions Routes", () => {
-  async function createTestBookmark(api: APICallerType, sessionId: string) {
-    const newBookmarkInput: z.infer<typeof zNewBookmarkRequestSchema> = {
-      type: BookmarkTypes.TEXT,
-      text: "Test bookmark text",
-      importSessionId: sessionId,
-    };
-    const createdBookmark =
-      await api.bookmarks.createBookmark(newBookmarkInput);
-    return createdBookmark.id;
-  }
-
   async function createTestList(api: APICallerType) {
     const newListInput: z.infer<typeof zNewBookmarkListSchema> = {
       name: "Test Import List",
@@ -98,8 +88,15 @@ describe("ImportSessions Routes", () => {
     const session = await api.importSessions.createImportSession({
       name: "Test Import Session",
     });
-    await createTestBookmark(api, session.id);
-    await createTestBookmark(api, session.id);
+
+    // Stage bookmarks using the staging flow
+    await api.importSessions.stageImportedBookmarks({
+      importSessionId: session.id,
+      bookmarks: [
+        { type: "text", content: "Test bookmark 1", tags: [], listIds: [] },
+        { type: "text", content: "Test bookmark 2", tags: [], listIds: [] },
+      ],
+    });
 
     const statsInput: z.infer<typeof zGetImportSessionStatsRequestSchema> = {
       importSessionId: session.id,
@@ -110,7 +107,7 @@ describe("ImportSessions Routes", () => {
     expect(stats).toMatchObject({
       id: session.id,
       name: "Test Import Session",
-      status: "in_progress",
+      status: "staging",
       totalBookmarks: 2,
       pendingBookmarks: 2,
       completedBookmarks: 0,
@@ -119,31 +116,191 @@ describe("ImportSessions Routes", () => {
     });
   });
 
-  test<CustomTestContext>("marks text-only imports as completed when tagging succeeds", async ({
+  test<CustomTestContext>("stats reflect crawl and tagging status for completed staging bookmarks", async ({
     apiCallers,
     db,
   }) => {
     const api = apiCallers[0];
+
     const session = await api.importSessions.createImportSession({
-      name: "Text Import Session",
+      name: "Test Import Session",
+    });
+
+    // Create bookmarks with different crawl/tag statuses
+    const user = (await db.query.users.findFirst())!;
+
+    // 1. Link bookmark: crawl success, tag success -> completed
+    const [completedLinkBookmark] = await db
+      .insert(bookmarks)
+      .values({
+        userId: user.id,
+        type: BookmarkTypes.LINK,
+        taggingStatus: "success",
+      })
+      .returning();
+    await db.insert(bookmarkLinks).values({
+      id: completedLinkBookmark.id,
+      url: "https://example.com/1",
+      crawlStatus: "success",
+    });
+
+    // 2. Link bookmark: crawl pending, tag success -> processing
+    const [crawlPendingBookmark] = await db
+      .insert(bookmarks)
+      .values({
+        userId: user.id,
+        type: BookmarkTypes.LINK,
+        taggingStatus: "success",
+      })
+      .returning();
+    await db.insert(bookmarkLinks).values({
+      id: crawlPendingBookmark.id,
+      url: "https://example.com/2",
+      crawlStatus: "pending",
+    });
+
+    // 3. Text bookmark: tag pending -> processing
+    const [tagPendingBookmark] = await db
+      .insert(bookmarks)
+      .values({
+        userId: user.id,
+        type: BookmarkTypes.TEXT,
+        taggingStatus: "pending",
+      })
+      .returning();
+    await db.insert(bookmarkTexts).values({
+      id: tagPendingBookmark.id,
+      text: "Test text",
+    });
+
+    // 4. Link bookmark: crawl failure -> failed
+    const [crawlFailedBookmark] = await db
+      .insert(bookmarks)
+      .values({
+        userId: user.id,
+        type: BookmarkTypes.LINK,
+        taggingStatus: "success",
+      })
+      .returning();
+    await db.insert(bookmarkLinks).values({
+      id: crawlFailedBookmark.id,
+      url: "https://example.com/3",
+      crawlStatus: "failure",
     });
-    const bookmarkId = await createTestBookmark(api, session.id);
 
-    await db
-      .update(bookmarks)
-      .set({ taggingStatus: "success" })
-      .where(eq(bookmarks.id, bookmarkId));
+    // 5. Text bookmark: tag failure -> failed
+    const [tagFailedBookmark] = await db
+      .insert(bookmarks)
+      .values({
+        userId: user.id,
+        type: BookmarkTypes.TEXT,
+        taggingStatus: "failure",
+      })
+      .returning();
+    await db.insert(bookmarkTexts).values({
+      id: tagFailedBookmark.id,
+      text: "Test text 2",
+    });
+
+    // 6. Text bookmark: tag success (no crawl needed) -> completed
+    const [completedTextBookmark] = await db
+      .insert(bookmarks)
+      .values({
+        userId: user.id,
+        type: BookmarkTypes.TEXT,
+        taggingStatus: "success",
+      })
+      .returning();
+    await db.insert(bookmarkTexts).values({
+      id: completedTextBookmark.id,
+      text: "Test text 3",
+    });
+
+    // Create staging bookmarks in different states
+    // Note: With the new import worker design, items stay in "processing" until
+    // crawl/tag is done. Only then do they move to "completed".
+    await db.insert(importStagingBookmarks).values([
+      // Staging pending -> pendingBookmarks
+      {
+        importSessionId: session.id,
+        type: "text",
+        content: "pending staging",
+        status: "pending",
+      },
+      // Staging processing (no bookmark yet) -> processingBookmarks
+      {
+        importSessionId: session.id,
+        type: "text",
+        content: "processing staging",
+        status: "processing",
+      },
+      // Staging failed -> failedBookmarks
+      {
+        importSessionId: session.id,
+        type: "text",
+        content: "failed staging",
+        status: "failed",
+      },
+      // Staging completed + crawl/tag success -> completedBookmarks
+      {
+        importSessionId: session.id,
+        type: "link",
+        url: "https://example.com/1",
+        status: "completed",
+        resultBookmarkId: completedLinkBookmark.id,
+      },
+      // Staging processing + crawl pending -> processingBookmarks (waiting for crawl)
+      {
+        importSessionId: session.id,
+        type: "link",
+        url: "https://example.com/2",
+        status: "processing",
+        resultBookmarkId: crawlPendingBookmark.id,
+      },
+      // Staging processing + tag pending -> processingBookmarks (waiting for tag)
+      {
+        importSessionId: session.id,
+        type: "text",
+        content: "tag pending",
+        status: "processing",
+        resultBookmarkId: tagPendingBookmark.id,
+      },
+      // Staging completed + crawl failure -> completedBookmarks (failure is terminal)
+      {
+        importSessionId: session.id,
+        type: "link",
+        url: "https://example.com/3",
+        status: "completed",
+        resultBookmarkId: crawlFailedBookmark.id,
+      },
+      // Staging completed + tag failure -> completedBookmarks (failure is terminal)
+      {
+        importSessionId: session.id,
+        type: "text",
+        content: "tag failed",
+        status: "completed",
+        resultBookmarkId: tagFailedBookmark.id,
+      },
+      // Staging completed + tag success (text, no crawl) -> completedBookmarks
+      {
+        importSessionId: session.id,
+        type: "text",
+        content: "completed text",
+        status: "completed",
+        resultBookmarkId: completedTextBookmark.id,
+      },
+    ]);
 
     const stats = await api.importSessions.getImportSessionStats({
       importSessionId: session.id,
     });
 
     expect(stats).toMatchObject({
-      completedBookmarks: 1,
-      pendingBookmarks: 0,
-      failedBookmarks: 0,
-      totalBookmarks: 1,
-      status: "completed",
+      totalBookmarks: 9,
+      pendingBookmarks: 1, // staging pending
+      processingBookmarks: 3, // staging processing (no bookmark) + crawl pending + tag pending
+      completedBookmarks: 4, // link success + text success + crawl failure + tag failure
+      failedBookmarks: 1, // staging failed
     });
   });
 
@@ -215,7 +372,7 @@ describe("ImportSessions Routes", () => {
     ).rejects.toThrow("Import session not found");
   });
 
-  test<CustomTestContext>("cannot attach other user's bookmark", async ({
+  test<CustomTestContext>("cannot stage other user's session", async ({
     apiCallers,
   }) => {
     const api1 = apiCallers[0];
@@ -228,7 +385,17 @@ describe("ImportSessions Routes", () => {
 
     // User 1 tries to attach User 2's bookmark
     await expect(
-      createTestBookmark(api2, session.id), // User 2's bookmark
+      api2.importSessions.stageImportedBookmarks({
+        importSessionId: session.id,
+        bookmarks: [
+          {
+            type: "text",
+            content: "Test bookmark",
+            tags: [],
+            listIds: [],
+          },
+        ],
+      }),
     ).rejects.toThrow("Import session not found");
   });
 });