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, 109 insertions, 0 deletions

diff --git a/packages/trpc/routers/importSessions.ts b/packages/trpc/routers/importSessions.ts
index 4bdc4f29..62263bdd 100644
--- a/packages/trpc/routers/importSessions.ts
+++ b/packages/trpc/routers/importSessions.ts
@@ -1,5 +1,8 @@
+import { experimental_trpcMiddleware } from "@trpc/server";
+import { and, eq, gt } from "drizzle-orm";
 import { z } from "zod";
 
+import { importStagingBookmarks } from "@karakeep/db/schema";
 import {
   zCreateImportSessionRequestSchema,
   zDeleteImportSessionRequestSchema,
@@ -9,9 +12,26 @@ import {
   zListImportSessionsResponseSchema,
 } from "@karakeep/shared/types/importSessions";
 
+import type { AuthedContext } from "../index";
 import { authedProcedure, router } from "../index";
 import { ImportSession } from "../models/importSessions";
 
+const ensureImportSessionAccess = experimental_trpcMiddleware<{
+  ctx: AuthedContext;
+  input: { importSessionId: string };
+}>().create(async (opts) => {
+  const importSession = await ImportSession.fromId(
+    opts.ctx,
+    opts.input.importSessionId,
+  );
+  return opts.next({
+    ctx: {
+      ...opts.ctx,
+      importSession,
+    },
+  });
+});
+
 export const importSessionsRouter = router({
   createImportSession: authedProcedure
     .input(zCreateImportSessionRequestSchema)
@@ -45,4 +65,93 @@ export const importSessionsRouter = router({
       await session.delete();
       return { success: true };
     }),
+
+  stageImportedBookmarks: authedProcedure
+    .input(
+      z.object({
+        importSessionId: z.string(),
+        bookmarks: z
+          .array(
+            z.object({
+              type: z.enum(["link", "text", "asset"]),
+              url: z.string().optional(),
+              title: z.string().optional(),
+              content: z.string().optional(),
+              note: z.string().optional(),
+              tags: z.array(z.string()).default([]),
+              listIds: z.array(z.string()).default([]),
+              sourceAddedAt: z.date().optional(),
+            }),
+          )
+          .max(50),
+      }),
+    )
+    .use(ensureImportSessionAccess)
+    .mutation(async ({ input, ctx }) => {
+      await ctx.importSession.stageBookmarks(input.bookmarks);
+    }),
+
+  finalizeImportStaging: authedProcedure
+    .input(z.object({ importSessionId: z.string() }))
+    .use(ensureImportSessionAccess)
+    .mutation(async ({ ctx }) => {
+      await ctx.importSession.finalize();
+    }),
+
+  pauseImportSession: authedProcedure
+    .input(z.object({ importSessionId: z.string() }))
+    .use(ensureImportSessionAccess)
+    .mutation(async ({ ctx }) => {
+      await ctx.importSession.pause();
+    }),
+
+  resumeImportSession: authedProcedure
+    .input(z.object({ importSessionId: z.string() }))
+    .use(ensureImportSessionAccess)
+    .mutation(async ({ ctx }) => {
+      await ctx.importSession.resume();
+    }),
+
+  getImportSessionResults: authedProcedure
+    .input(
+      z.object({
+        importSessionId: z.string(),
+        filter: z
+          .enum(["all", "accepted", "rejected", "skipped_duplicate", "pending"])
+          .optional(),
+        cursor: z.string().optional(),
+        limit: z.number().default(50),
+      }),
+    )
+    .use(ensureImportSessionAccess)
+    .query(async ({ ctx, input }) => {
+      const results = await ctx.db
+        .select()
+        .from(importStagingBookmarks)
+        .where(
+          and(
+            eq(
+              importStagingBookmarks.importSessionId,
+              ctx.importSession.session.id,
+            ),
+            input.filter && input.filter !== "all"
+              ? input.filter === "pending"
+                ? eq(importStagingBookmarks.status, "pending")
+                : eq(importStagingBookmarks.result, input.filter)
+              : undefined,
+            input.cursor
+              ? gt(importStagingBookmarks.id, input.cursor)
+              : undefined,
+          ),
+        )
+        .orderBy(importStagingBookmarks.id)
+        .limit(input.limit + 1);
+
+      // Return with pagination info
+      const hasMore = results.length > input.limit;
+      return {
+        items: results.slice(0, input.limit),
+        nextCursor: hasMore ? results[input.limit - 1].id : null,
+      };
+    }),
 });