fix: backfill old sessions and do queue backpressure (#2449)

* fix: backfill old sessions and do queue backpressure

* fix typo

2 files changed, 64 insertions, 1 deletions

diff --git a/packages/db/drizzle/0077_import_listpaths_to_listids.sql b/packages/db/drizzle/0077_import_listpaths_to_listids.sql
index f1c4f883..0a28ea20 100644
--- a/packages/db/drizzle/0077_import_listpaths_to_listids.sql
+++ b/packages/db/drizzle/0077_import_listpaths_to_listids.sql
@@ -23,4 +23,55 @@ CREATE TABLE `importStagingBookmarks` (
 CREATE INDEX `importStaging_session_status_idx` ON `importStagingBookmarks` (`importSessionId`,`status`);--> statement-breakpoint
 CREATE INDEX `importStaging_completedAt_idx` ON `importStagingBookmarks` (`completedAt`);--> statement-breakpoint
 ALTER TABLE `importSessions` ADD `status` text DEFAULT 'staging' NOT NULL;--> statement-breakpoint
-ALTER TABLE `importSessions` ADD `lastProcessedAt` integer;
\ No newline at end of file
+ALTER TABLE `importSessions` ADD `lastProcessedAt` integer;--> statement-breakpoint
+-- Migrate legacy importSessionBookmarks into importStagingBookmarks.
+-- Reuses the same ID from the old table.
+-- Calculates status based on actual downstream crawl/tagging state.
+INSERT INTO importStagingBookmarks (
+  id, importSessionId, type, url,
+  status, processingStartedAt, result, resultBookmarkId, createdAt, completedAt
+)
+SELECT
+  isb.id,
+  isb.importSessionId,
+  b.type,
+  bl.url,
+  CASE
+    WHEN (bl.crawlStatus IS NULL OR bl.crawlStatus IN ('success', 'failure'))
+     AND (b.taggingStatus IS NULL OR b.taggingStatus IN ('success', 'failure'))
+    THEN 'completed'
+    ELSE 'processing'
+  END,
+  isb.createdAt,
+  'accepted',
+  isb.bookmarkId,
+  isb.createdAt,
+  CASE
+    WHEN (bl.crawlStatus IS NULL OR bl.crawlStatus IN ('success', 'failure'))
+     AND (b.taggingStatus IS NULL OR b.taggingStatus IN ('success', 'failure'))
+    THEN isb.createdAt
+    ELSE NULL
+  END
+FROM importSessionBookmarks isb
+JOIN bookmarks b ON b.id = isb.bookmarkId
+LEFT JOIN bookmarkLinks bl ON bl.id = isb.bookmarkId
+WHERE NOT EXISTS (
+  SELECT 1 FROM importStagingBookmarks stg
+  WHERE stg.importSessionId = isb.importSessionId
+);
+--> statement-breakpoint
+-- Move legacy sessions out of staging:
+-- - Running if any items are still processing downstream
+-- - Completed otherwise (including sessions with no remaining items)
+UPDATE importSessions
+SET status = CASE
+  WHEN EXISTS (
+    SELECT 1
+    FROM importStagingBookmarks stg
+    WHERE stg.importSessionId = importSessions.id
+      AND stg.status = 'processing'
+  )
+  THEN 'running'
+  ELSE 'completed'
+END
+WHERE status = 'staging';
diff --git a/packages/shared/logger.ts b/packages/shared/logger.ts
index efe78ff3..f3c5d45d 100644
--- a/packages/shared/logger.ts
+++ b/packages/shared/logger.ts
@@ -14,4 +14,16 @@ const logger = winston.createLogger({
   transports: [new winston.transports.Console()],
 });
 
+export function throttledLogger(periodMs: number) {
+  let lastLogTime = 0;
+
+  return (level: string, message: string) => {
+    const now = Date.now();
+    if (now - lastLogTime >= periodMs) {
+      lastLogTime = now;
+      logger.log(level, message);
+    }
+  };
+}
+
 export default logger;