fix: Propagate group ids in queue calls (#2177)

* fix: Propagate group ids

* fix tests

6 files changed, 85 insertions, 33 deletions

diff --git a/packages/trpc/lib/__tests__/ruleEngine.test.ts b/packages/trpc/lib/__tests__/ruleEngine.test.ts
index a108ede7..ede22ec6 100644
--- a/packages/trpc/lib/__tests__/ruleEngine.test.ts
+++ b/packages/trpc/lib/__tests__/ruleEngine.test.ts
@@ -528,11 +528,16 @@ describe("RuleEngine", () => {
       const action: RuleEngineAction = { type: "downloadFullPageArchive" };
       const result = await engine.executeAction(action);
       expect(result).toBe(`Enqueued full page archive`);
-      expect(LinkCrawlerQueue.enqueue).toHaveBeenCalledWith({
-        bookmarkId: bookmarkId,
-        archiveFullPage: true,
-        runInference: false,
-      });
+      expect(LinkCrawlerQueue.enqueue).toHaveBeenCalledWith(
+        {
+          bookmarkId: bookmarkId,
+          archiveFullPage: true,
+          runInference: false,
+        },
+        {
+          groupId: userId,
+        },
+      );
     });
 
     it("should execute favouriteBookmark action", async () => {
diff --git a/packages/trpc/lib/ruleEngine.ts b/packages/trpc/lib/ruleEngine.ts
index 2d5deae6..c191619b 100644
--- a/packages/trpc/lib/ruleEngine.ts
+++ b/packages/trpc/lib/ruleEngine.ts
@@ -189,11 +189,16 @@ export class RuleEngine {
         return `Removed from list ${action.listId}`;
       }
       case "downloadFullPageArchive": {
-        await LinkCrawlerQueue.enqueue({
-          bookmarkId: this.bookmark.id,
-          archiveFullPage: true,
-          runInference: false,
-        });
+        await LinkCrawlerQueue.enqueue(
+          {
+            bookmarkId: this.bookmark.id,
+            archiveFullPage: true,
+            runInference: false,
+          },
+          {
+            groupId: this.bookmark.userId,
+          },
+        );
         return `Enqueued full page archive`;
       }
       case "favouriteBookmark": {
diff --git a/packages/trpc/models/bookmarks.ts b/packages/trpc/models/bookmarks.ts
index bd696ee8..07fa8693 100644
--- a/packages/trpc/models/bookmarks.ts
+++ b/packages/trpc/models/bookmarks.ts
@@ -779,12 +779,19 @@ export class Bookmark extends BareBookmark {
         ),
       );
 
-    await SearchIndexingQueue.enqueue({
-      bookmarkId: this.bookmark.id,
-      type: "delete",
-    });
+    await SearchIndexingQueue.enqueue(
+      {
+        bookmarkId: this.bookmark.id,
+        type: "delete",
+      },
+      {
+        groupId: this.ctx.user.id,
+      },
+    );
 
-    await triggerWebhook(this.bookmark.id, "deleted", this.ctx.user.id);
+    await triggerWebhook(this.bookmark.id, "deleted", this.ctx.user.id, {
+      groupId: this.ctx.user.id,
+    });
     if (deleted.changes > 0) {
       await this.cleanupAssets();
     }
diff --git a/packages/trpc/models/tags.ts b/packages/trpc/models/tags.ts
index b230b6b4..55532077 100644
--- a/packages/trpc/models/tags.ts
+++ b/packages/trpc/models/tags.ts
@@ -280,7 +280,11 @@ export class Tag {
 
     try {
       await Promise.all(
-        affectedBookmarks.map((id) => triggerSearchReindex(id)),
+        affectedBookmarks.map((id) =>
+          triggerSearchReindex(id, {
+            groupId: ctx.user.id,
+          }),
+        ),
       );
     } catch (e) {
       console.error("Failed to reindex affected bookmarks", e);
@@ -315,7 +319,9 @@ export class Tag {
 
     await Promise.all(
       affectedBookmarks.map(({ bookmarkId }) =>
-        triggerSearchReindex(bookmarkId),
+        triggerSearchReindex(bookmarkId, {
+          groupId: this.ctx.user.id,
+        }),
       ),
     );
   }
@@ -352,7 +358,9 @@ export class Tag {
         await Promise.all(
           affectedBookmarks
             .map((b) => b.bookmarkId)
-            .map((id) => triggerSearchReindex(id)),
+            .map((id) =>
+              triggerSearchReindex(id, { groupId: this.ctx.user.id }),
+            ),
         );
       } catch (e) {
         console.error("Failed to reindex affected bookmarks", e);
diff --git a/packages/trpc/routers/bookmarks.ts b/packages/trpc/routers/bookmarks.ts
index 389f026c..05ffa240 100644
--- a/packages/trpc/routers/bookmarks.ts
+++ b/packages/trpc/routers/bookmarks.ts
@@ -281,6 +281,7 @@ export const bookmarksAppRouter = router({
       const enqueueOpts: EnqueueOptions = {
         // The lower the priority number, the sooner the job will be processed
         priority: input.crawlPriority === "low" ? 50 : 0,
+        groupId: ctx.user.id,
       };
 
       switch (bookmark.content.type) {
@@ -487,8 +488,12 @@ export const bookmarksAppRouter = router({
         );
       }
       // Trigger re-indexing and webhooks
-      await triggerSearchReindex(input.bookmarkId);
-      await triggerWebhook(input.bookmarkId, "edited");
+      await triggerSearchReindex(input.bookmarkId, {
+        groupId: ctx.user.id,
+      });
+      await triggerWebhook(input.bookmarkId, "edited", ctx.user.id, {
+        groupId: ctx.user.id,
+      });
 
       return updatedBookmark;
     }),
@@ -527,8 +532,12 @@ export const bookmarksAppRouter = router({
             ),
           );
       });
-      await triggerSearchReindex(input.bookmarkId);
-      await triggerWebhook(input.bookmarkId, "edited");
+      await triggerSearchReindex(input.bookmarkId, {
+        groupId: ctx.user.id,
+      });
+      await triggerWebhook(input.bookmarkId, "edited", ctx.user.id, {
+        groupId: ctx.user.id,
+      });
     }),
 
   deleteBookmark: authedProcedure
@@ -561,10 +570,15 @@ export const bookmarksAppRouter = router({
           crawlStatusCode: null,
         })
         .where(eq(bookmarkLinks.id, input.bookmarkId));
-      await LinkCrawlerQueue.enqueue({
-        bookmarkId: input.bookmarkId,
-        archiveFullPage: input.archiveFullPage,
-      });
+      await LinkCrawlerQueue.enqueue(
+        {
+          bookmarkId: input.bookmarkId,
+          archiveFullPage: input.archiveFullPage,
+        },
+        {
+          groupId: ctx.user.id,
+        },
+      );
     }),
   getBookmark: authedProcedure
     .input(
@@ -818,8 +832,12 @@ export const bookmarksAppRouter = router({
             tagId: t,
           })),
         ]);
-        await triggerSearchReindex(input.bookmarkId);
-        await triggerWebhook(input.bookmarkId, "edited");
+        await triggerSearchReindex(input.bookmarkId, {
+          groupId: ctx.user.id,
+        });
+        await triggerWebhook(input.bookmarkId, "edited", ctx.user.id, {
+          groupId: ctx.user.id,
+        });
         return {
           bookmarkId: input.bookmarkId,
           attached: allIds,
@@ -959,8 +977,12 @@ Author: ${bookmark.author ?? ""}
           summary: summary.response,
         })
         .where(eq(bookmarks.id, input.bookmarkId));
-      await triggerSearchReindex(input.bookmarkId);
-      await triggerWebhook(input.bookmarkId, "edited");
+      await triggerSearchReindex(input.bookmarkId, {
+        groupId: ctx.user.id,
+      });
+      await triggerWebhook(input.bookmarkId, "edited", ctx.user.id, {
+        groupId: ctx.user.id,
+      });
 
       return {
         bookmarkId: input.bookmarkId,
diff --git a/packages/trpc/routers/feeds.ts b/packages/trpc/routers/feeds.ts
index 57c88084..8591c98c 100644
--- a/packages/trpc/routers/feeds.ts
+++ b/packages/trpc/routers/feeds.ts
@@ -57,8 +57,13 @@ export const feedsAppRouter = router({
     .input(z.object({ feedId: z.string() }))
     .mutation(async ({ input, ctx }) => {
       await Feed.fromId(ctx, input.feedId);
-      await FeedQueue.enqueue({
-        feedId: input.feedId,
-      });
+      await FeedQueue.enqueue(
+        {
+          feedId: input.feedId,
+        },
+        {
+          groupId: ctx.user.id,
+        },
+      );
     }),
 });