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

4 files changed, 143 insertions, 105 deletions

diff --git a/apps/web/components/settings/ImportSessionCard.tsx b/apps/web/components/settings/ImportSessionCard.tsx
index 690caaa5..f20710ca 100644
--- a/apps/web/components/settings/ImportSessionCard.tsx
+++ b/apps/web/components/settings/ImportSessionCard.tsx
@@ -9,6 +9,8 @@ import { Progress } from "@/components/ui/progress";
 import {
   useDeleteImportSession,
   useImportSessionStats,
+  usePauseImportSession,
+  useResumeImportSession,
 } from "@/lib/hooks/useImportSessions";
 import { useTranslation } from "@/lib/i18n/client";
 import { formatDistanceToNow } from "date-fns";
@@ -19,10 +21,17 @@ import {
   Clock,
   ExternalLink,
   Loader2,
+  Pause,
+  Play,
   Trash2,
+  Upload,
 } from "lucide-react";
 
-import type { ZImportSessionWithStats } from "@karakeep/shared/types/importSessions";
+import type {
+  ZImportSessionStatus,
+  ZImportSessionWithStats,
+} from "@karakeep/shared/types/importSessions";
+import { switchCase } from "@karakeep/shared/utils/switch";
 
 interface ImportSessionCardProps {
   session: ZImportSessionWithStats;
@@ -30,10 +39,14 @@ interface ImportSessionCardProps {
 
 function getStatusColor(status: string) {
   switch (status) {
+    case "staging":
+      return "bg-purple-500/10 text-purple-700 dark:text-purple-400";
     case "pending":
       return "bg-muted text-muted-foreground";
-    case "in_progress":
+    case "running":
       return "bg-blue-500/10 text-blue-700 dark:text-blue-400";
+    case "paused":
+      return "bg-yellow-500/10 text-yellow-700 dark:text-yellow-400";
     case "completed":
       return "bg-green-500/10 text-green-700 dark:text-green-400";
     case "failed":
@@ -45,10 +58,14 @@ function getStatusColor(status: string) {
 
 function getStatusIcon(status: string) {
   switch (status) {
+    case "staging":
+      return <Upload className="h-4 w-4" />;
     case "pending":
       return <Clock className="h-4 w-4" />;
-    case "in_progress":
+    case "running":
       return <Loader2 className="h-4 w-4 animate-spin" />;
+    case "paused":
+      return <Pause className="h-4 w-4" />;
     case "completed":
       return <CheckCircle2 className="h-4 w-4" />;
     case "failed":
@@ -62,13 +79,18 @@ export function ImportSessionCard({ session }: ImportSessionCardProps) {
   const { t } = useTranslation();
   const { data: liveStats } = useImportSessionStats(session.id);
   const deleteSession = useDeleteImportSession();
+  const pauseSession = usePauseImportSession();
+  const resumeSession = useResumeImportSession();
 
-  const statusLabels: Record<string, string> = {
-    pending: t("settings.import_sessions.status.pending"),
-    in_progress: t("settings.import_sessions.status.in_progress"),
-    completed: t("settings.import_sessions.status.completed"),
-    failed: t("settings.import_sessions.status.failed"),
-  };
+  const statusLabels = (s: ZImportSessionStatus) =>
+    switchCase(s, {
+      staging: t("settings.import_sessions.status.staging"),
+      pending: t("settings.import_sessions.status.pending"),
+      running: t("settings.import_sessions.status.running"),
+      paused: t("settings.import_sessions.status.paused"),
+      completed: t("settings.import_sessions.status.completed"),
+      failed: t("settings.import_sessions.status.failed"),
+    });
 
   // Use live stats if available, otherwise fallback to session stats
   const stats = liveStats || session;
@@ -79,7 +101,14 @@ export function ImportSessionCard({ session }: ImportSessionCardProps) {
         100
       : 0;
 
-  const canDelete = stats.status !== "in_progress";
+  const canDelete =
+    stats.status === "completed" ||
+    stats.status === "failed" ||
+    stats.status === "paused";
+
+  const canPause = stats.status === "pending" || stats.status === "running";
+
+  const canResume = stats.status === "paused";
 
   return (
     <Card className="transition-all hover:shadow-md">
@@ -101,7 +130,7 @@ export function ImportSessionCard({ session }: ImportSessionCardProps) {
             >
               {getStatusIcon(stats.status)}
               <span className="ml-1 capitalize">
-                {statusLabels[stats.status] ?? stats.status.replace("_", " ")}
+                {statusLabels(stats.status)}
               </span>
             </Badge>
           </div>
@@ -213,6 +242,32 @@ export function ImportSessionCard({ session }: ImportSessionCardProps) {
           {/* Actions */}
           <div className="flex items-center justify-end pt-2">
             <div className="flex items-center gap-2">
+              {canPause && (
+                <Button
+                  variant="outline"
+                  size="sm"
+                  onClick={() =>
+                    pauseSession.mutate({ importSessionId: session.id })
+                  }
+                  disabled={pauseSession.isPending}
+                >
+                  <Pause className="mr-1 h-4 w-4" />
+                  {t("settings.import_sessions.pause_session")}
+                </Button>
+              )}
+              {canResume && (
+                <Button
+                  variant="outline"
+                  size="sm"
+                  onClick={() =>
+                    resumeSession.mutate({ importSessionId: session.id })
+                  }
+                  disabled={resumeSession.isPending}
+                >
+                  <Play className="mr-1 h-4 w-4" />
+                  {t("settings.import_sessions.resume_session")}
+                </Button>
+              )}
               {canDelete && (
                 <ActionConfirmingDialog
                   title={t("settings.import_sessions.delete_dialog_title")}
diff --git a/apps/web/lib/hooks/useBookmarkImport.ts b/apps/web/lib/hooks/useBookmarkImport.ts
index c0681924..35c04c1b 100644
--- a/apps/web/lib/hooks/useBookmarkImport.ts
+++ b/apps/web/lib/hooks/useBookmarkImport.ts
@@ -5,25 +5,13 @@ import { toast } from "@/components/ui/sonner";
 import { useTranslation } from "@/lib/i18n/client";
 import { useMutation, useQueryClient } from "@tanstack/react-query";
 
-import {
-  useCreateBookmarkWithPostHook,
-  useUpdateBookmarkTags,
-} from "@karakeep/shared-react/hooks/bookmarks";
-import {
-  useAddBookmarkToList,
-  useCreateBookmarkList,
-} from "@karakeep/shared-react/hooks/lists";
+import { useCreateBookmarkList } from "@karakeep/shared-react/hooks/lists";
 import { useTRPC } from "@karakeep/shared-react/trpc";
 import {
   importBookmarksFromFile,
   ImportSource,
-  ParsedBookmark,
   parseImportFile,
 } from "@karakeep/shared/import-export";
-import {
-  BookmarkTypes,
-  MAX_BOOKMARK_TITLE_LENGTH,
-} from "@karakeep/shared/types/bookmarks";
 
 import { useCreateImportSession } from "./useImportSessions";
 
@@ -43,10 +31,13 @@ export function useBookmarkImport() {
 
   const queryClient = useQueryClient();
   const { mutateAsync: createImportSession } = useCreateImportSession();
-  const { mutateAsync: createBookmark } = useCreateBookmarkWithPostHook();
   const { mutateAsync: createList } = useCreateBookmarkList();
-  const { mutateAsync: addToList } = useAddBookmarkToList();
-  const { mutateAsync: updateTags } = useUpdateBookmarkTags();
+  const { mutateAsync: stageImportedBookmarks } = useMutation(
+    api.importSessions.stageImportedBookmarks.mutationOptions(),
+  );
+  const { mutateAsync: finalizeImportStaging } = useMutation(
+    api.importSessions.finalizeImportStaging.mutationOptions(),
+  );
 
   const uploadBookmarkFileMutation = useMutation({
     mutationFn: async ({
@@ -86,7 +77,6 @@ export function useBookmarkImport() {
       }
 
       // Proceed with import if quota check passes
-      // Use a custom parser to avoid re-parsing the file
       const result = await importBookmarksFromFile(
         {
           file,
@@ -95,65 +85,9 @@ export function useBookmarkImport() {
           deps: {
             createImportSession,
             createList,
-            createBookmark: async (
-              bookmark: ParsedBookmark,
-              sessionId: string,
-            ) => {
-              if (bookmark.content === undefined) {
-                throw new Error("Content is undefined");
-              }
-              const created = await createBookmark({
-                crawlPriority: "low",
-                title: bookmark.title.substring(0, MAX_BOOKMARK_TITLE_LENGTH),
-                createdAt: bookmark.addDate
-                  ? new Date(bookmark.addDate * 1000)
-                  : undefined,
-                note: bookmark.notes,
-                archived: bookmark.archived,
-                importSessionId: sessionId,
-                source: "import",
-                ...(bookmark.content.type === BookmarkTypes.LINK
-                  ? {
-                      type: BookmarkTypes.LINK,
-                      url: bookmark.content.url,
-                    }
-                  : {
-                      type: BookmarkTypes.TEXT,
-                      text: bookmark.content.text,
-                    }),
-              });
-              return created as { id: string; alreadyExists?: boolean };
-            },
-            addBookmarkToLists: async ({
-              bookmarkId,
-              listIds,
-            }: {
-              bookmarkId: string;
-              listIds: string[];
-            }) => {
-              await Promise.all(
-                listIds.map((listId) =>
-                  addToList({
-                    bookmarkId,
-                    listId,
-                  }),
-                ),
-              );
-            },
-            updateBookmarkTags: async ({
-              bookmarkId,
-              tags,
-            }: {
-              bookmarkId: string;
-              tags: string[];
-            }) => {
-              if (tags.length > 0) {
-                await updateTags({
-                  bookmarkId,
-                  attach: tags.map((t) => ({ tagName: t })),
-                  detach: [],
-                });
-              }
+            stageImportedBookmarks,
+            finalizeImportStaging: async (sessionId: string) => {
+              await finalizeImportStaging({ importSessionId: sessionId });
             },
           },
           onProgress: (done, total) => setImportProgress({ done, total }),
@@ -174,19 +108,11 @@ export function useBookmarkImport() {
         toast({ description: "No bookmarks found in the file." });
         return;
       }
-      const { successes, failures, alreadyExisted } = result.counts;
-      if (successes > 0 || alreadyExisted > 0) {
-        toast({
-          description: `Imported ${successes} bookmarks into import session. Background processing will start automatically.`,
-          variant: "default",
-        });
-      }
-      if (failures > 0) {
-        toast({
-          description: `Failed to import ${failures} bookmarks. Check console for details.`,
-          variant: "destructive",
-        });
-      }
+
+      toast({
+        description: `Staged ${result.counts.total} bookmarks for import. Background processing will start automatically.`,
+        variant: "default",
+      });
     },
     onError: (error) => {
       setImportProgress(null);
diff --git a/apps/web/lib/hooks/useImportSessions.ts b/apps/web/lib/hooks/useImportSessions.ts
index 133bb29b..4d095c0b 100644
--- a/apps/web/lib/hooks/useImportSessions.ts
+++ b/apps/web/lib/hooks/useImportSessions.ts
@@ -46,7 +46,11 @@ export function useImportSessionStats(importSessionId: string) {
         importSessionId,
       },
       {
-        refetchInterval: 5000, // Refetch every 5 seconds to show progress
+        refetchInterval: (q) =>
+          !q.state.data ||
+          !["completed", "failed"].includes(q.state.data.status)
+            ? 5000
+            : false, // Refetch every 5 seconds to show progress
         enabled: !!importSessionId,
       },
     ),
@@ -77,3 +81,53 @@ export function useDeleteImportSession() {
     }),
   );
 }
+
+export function usePauseImportSession() {
+  const api = useTRPC();
+  const queryClient = useQueryClient();
+
+  return useMutation(
+    api.importSessions.pauseImportSession.mutationOptions({
+      onSuccess: () => {
+        queryClient.invalidateQueries(
+          api.importSessions.listImportSessions.pathFilter(),
+        );
+        toast({
+          description: "Import session paused",
+          variant: "default",
+        });
+      },
+      onError: (error) => {
+        toast({
+          description: error.message || "Failed to pause import session",
+          variant: "destructive",
+        });
+      },
+    }),
+  );
+}
+
+export function useResumeImportSession() {
+  const api = useTRPC();
+  const queryClient = useQueryClient();
+
+  return useMutation(
+    api.importSessions.resumeImportSession.mutationOptions({
+      onSuccess: () => {
+        queryClient.invalidateQueries(
+          api.importSessions.listImportSessions.pathFilter(),
+        );
+        toast({
+          description: "Import session resumed",
+          variant: "default",
+        });
+      },
+      onError: (error) => {
+        toast({
+          description: error.message || "Failed to resume import session",
+          variant: "destructive",
+        });
+      },
+    }),
+  );
+}
diff --git a/apps/web/lib/i18n/locales/en/translation.json b/apps/web/lib/i18n/locales/en/translation.json
index ce607920..59d2098e 100644
--- a/apps/web/lib/i18n/locales/en/translation.json
+++ b/apps/web/lib/i18n/locales/en/translation.json
@@ -416,11 +416,12 @@
       "created_at": "Created {{time}}",
       "progress": "Progress",
       "status": {
+        "staging": "Staging",
         "pending": "Pending",
-        "in_progress": "In progress",
+        "running": "Running",
+        "paused": "Paused",
         "completed": "Completed",
-        "failed": "Failed",
-        "processing": "Processing"
+        "failed": "Failed"
       },
       "badges": {
         "pending": "{{count}} pending",
@@ -432,7 +433,9 @@
       "view_list": "View List",
       "delete_dialog_title": "Delete Import Session",
       "delete_dialog_description": "Are you sure you want to delete \"{{name}}\"? This action cannot be undone. The bookmarks themselves will not be deleted.",
-      "delete_session": "Delete Session"
+      "delete_session": "Delete Session",
+      "pause_session": "Pause",
+      "resume_session": "Resume"
     },
     "backups": {
       "backups": "Backups",