feat(webhook): Implement webhook functionality for bookmark events

apps/workers/crawlerWorker.ts

@@ -54,6 +54,7 @@ import {
   OpenAIQueue,
   triggerSearchReindex,
   triggerVideoWorker,
+  triggerWebhookWorker,
   zCrawlLinkRequestSchema,
 } from "@hoarder/shared/queues";
 import { BookmarkTypes } from "@hoarder/shared/types/bookmarks";
@@ -746,6 +747,9 @@ async function runCrawler(job: DequeuedJob<ZCrawlLinkRequest>) {
     // Trigger a potential download of a video from the URL
     await triggerVideoWorker(bookmarkId, url);
 
+    // Trigger a webhook
+    await triggerWebhookWorker(bookmarkId, url, "create");
+
     // Do the archival as a separate last step as it has the potential for failure
     await archivalLogic();
   }

apps/workers/index.ts

@@ -13,12 +13,13 @@ import { shutdownPromise } from "./exit";
 import { OpenAiWorker } from "./openaiWorker";
 import { SearchIndexingWorker } from "./searchWorker";
 import { VideoWorker } from "./videoWorker";
+import { WebhookWorker } from "./webhookWorker";
 
 async function main() {
   logger.info(`Workers version: ${serverConfig.serverVersion ?? "not set"}`);
   runQueueDBMigrations();
 
-  const [crawler, openai, search, tidyAssets, video, feed, assetPreprocessing] =
+  const [crawler, openai, search, tidyAssets, video, feed, assetPreprocessing, webhook] =
     [
       await CrawlerWorker.build(),
       OpenAiWorker.build(),
@@ -27,6 +28,7 @@ async function main() {
       VideoWorker.build(),
       FeedWorker.build(),
       AssetPreprocessingWorker.build(),
+      WebhookWorker.build(),
     ];
   FeedRefreshingWorker.start();
 
@@ -39,11 +41,12 @@ async function main() {
       video.run(),
       feed.run(),
       assetPreprocessing.run(),
+      webhook.run(),
     ]),
     shutdownPromise,
   ]);
   logger.info(
-    "Shutting down crawler, openai, tidyAssets, video, feed, assetPreprocessing and search workers ...",
+    "Shutting down crawler, openai, tidyAssets, video, feed, assetPreprocessing, webhook and search workers ...",
   );
 
   FeedRefreshingWorker.stop();
@@ -54,6 +57,7 @@ async function main() {
   video.stop();
   feed.stop();
   assetPreprocessing.stop();
+  webhook.stop();
 }
 
 main();

apps/workers/webhookWorker.ts

@@ -0,0 +1,109 @@
+import { DequeuedJob, Runner } from "liteque";
+import { db } from "@hoarder/db";
+import { bookmarks } from "@hoarder/db/schema";
+import serverConfig from "@hoarder/shared/config";
+import logger from "@hoarder/shared/logger";
+import { WebhookQueue, zWebhookRequestSchema, ZWebhookRequest } from "@hoarder/shared/queues";
+import { eq } from "drizzle-orm";
+import fetch from 'node-fetch';
+import { Response } from 'node-fetch';
+
+export class WebhookWorker {
+  static build() {
+    logger.info("Starting webhook worker ...");
+    const worker = new Runner<ZWebhookRequest>(
+      WebhookQueue,
+      {
+        run: runWebhook,
+        onComplete: async (job) => {
+          const jobId = job.id;
+          logger.info(`[webhook][${jobId}] Completed successfully`);
+        },
+        onError: async (job) => {
+          const jobId = job.id;
+          logger.error(
+            `[webhook][${jobId}] webhook job failed: ${job.error}\n${job.error.stack}`,
+          )
+        },
+      },
+      {
+        concurrency: 1,
+        pollIntervalMs: 1000,
+        timeoutSecs: serverConfig.inference.jobTimeoutSec,
+      },
+    );
+
+    return worker;
+  }
+}
+
+
+async function fetchBookmark(linkId: string) {
+  return await db.query.bookmarks.findFirst({
+    where: eq(bookmarks.id, linkId),
+    with: {
+      link: true,
+      text: true,
+      asset: true,
+    },
+  });
+}
+
+async function runWebhook(job: DequeuedJob<ZWebhookRequest>) {
+  const jobId = job.id;
+  const webhookUrls = serverConfig.webhook.urls;
+  const webhookToken = serverConfig.webhook.token;
+  const webhookTimeout = serverConfig.webhook.timeout;
+
+  const request = zWebhookRequestSchema.safeParse(job.data);
+  if (!request.success) {
+    throw new Error(
+      `[webhook][${jobId}] Got malformed job request: ${request.error.toString()}`,
+    );
+  }
+
+  const { bookmarkId } = request.data;
+  const bookmark = await fetchBookmark(bookmarkId);
+  if (!bookmark) {
+    throw new Error(
+      `[webhook][${jobId}] bookmark with id ${bookmarkId} was not found`,
+    );
+  }
+
+  logger.info(
+    `[webhook][${jobId}] Starting a webhook job for bookmark with id "${bookmark.id}"`,
+  );
+
+  for (const url of webhookUrls) {
+    try {
+      const response = await Promise.race([
+        fetch(url, {
+          method: "POST",
+          headers: {
+            "Content-Type": "application/json",
+            "Authorization": `Bearer ${webhookToken}`,
+          },
+          body: JSON.stringify({
+            jobId, bookmarkId, userId: bookmark.userId, url: bookmark.link.url, operation: job.data.operation
+          })
+        }),
+        new Promise((_, reject) =>
+          setTimeout(() => reject(new Error(`Request timed out in ${webhookTimeout}ms`)), webhookTimeout)
+        )
+      ]);
+
+      if (!(response instanceof Response)) {
+        throw new Error(`Webhook call to ${url} failed: response is not a Response object`);
+      }
+
+      if (!response.ok) {
+        logger.error(`Webhook call to ${url} failed with status: ${response.status}`);
+      }
+
+      logger.info(`[webhook][${jobId}] Webhook to ${url} call succeeded`);
+    } catch (error) {
+      logger.error(`[webhook][${jobId}] Webhook to ${url} call failed: ${error}`);
+      throw error;
+    }
+  }
+}

docs/docs/03-configuration.md

@@ -95,3 +95,33 @@ Hoarder uses [tesseract.js](https://github.com/naptha/tesseract.js) to extract t
 | OCR_CACHE_DIR            | No       | $TEMP_DIR | The dir where tesseract will download its models. By default, those models are not persisted and stored in the OS' temp dir.                                                                                                              |
 | OCR_LANGS                | No       | eng       | Comma separated list of the language codes that you want tesseract to support. You can find the language codes [here](https://tesseract-ocr.github.io/tessdoc/Data-Files-in-different-versions.html). Set to empty string to disable OCR. |
 | OCR_CONFIDENCE_THRESHOLD | No       | 50        | A number between 0 and 100 indicating the minimum acceptable confidence from tessaract. If tessaract's confidence is lower than this value, extracted text won't be stored.                                                               |
+
+## Webhook Configs
+
+You can use webhooks to trigger actions when bookmarks are changed ( only support *create* now ).
+
+| Name            | Required | Default | Description                                                                                    |
+| --------------- | -------- | ------- | ---------------------------------------------------------------------------------------------- |
+| WEBHOOK_URLS    | No       |         | The urls of the webhooks to trigger, separated by commas.                                      |
+| WEBHOOK_TOKEN   | No       |         | The token to use for authentication. Will appears in the Authorization header as Bearer token. |
+| WEBHOOK_TIMEOUT | No       | 5000    | The timeout for the webhook request in milliseconds.                                           |
+
+:::info
+
+- If a url is add to hoarder , after it is crawled, the webhook will be triggered.
+- The WEBHOOK_TOKEN is used for authentication. It will appear in the Authorization header as Bearer token.
+  ```
+  Authorization: Bearer <WEBHOOK_TOKEN>
+  ```
+- The webhook will be triggered with the job id (used for idempotent), bookmark id, the user id, the url and the operation in JSON format in the body.
+
+  ```json
+  {
+    "jobId": 123,
+    "bookmarkId": "exampleBookmarkId",
+    "userId": "exampleUserId",
+    "url": "https://example.com",
+    "operation": "create"
+  }
+  ```
+:::

packages/shared/config.ts

@@ -56,6 +56,12 @@ const allEnv = z.object({
   DATA_DIR: z.string().default(""),
   MAX_ASSET_SIZE_MB: z.coerce.number().default(4),
   INFERENCE_LANG: z.string().default("english"),
+  WEBHOOK_URLS: z.preprocess(
+    (val) => typeof val === "string" ? val.split(",") : val,
+    z.array(z.string().url())
+  ).default([]),
+  WEBHOOK_TOKEN: z.string().default(""),
+  WEBHOOK_TIMEOUT: z.coerce.number().default(5000),
   // Build only flag
   SERVER_VERSION: z.string().optional(),
   DISABLE_NEW_RELEASE_CHECK: stringBool("false"),
@@ -118,22 +124,27 @@ const serverConfigSchema = allEnv.transform((val) => {
     },
     meilisearch: val.MEILI_ADDR
       ? {
-          address: val.MEILI_ADDR,
-          key: val.MEILI_MASTER_KEY,
-        }
+        address: val.MEILI_ADDR,
+        key: val.MEILI_MASTER_KEY,
+      }
       : undefined,
     logLevel: val.LOG_LEVEL,
     demoMode: val.DEMO_MODE
       ? {
-          email: val.DEMO_MODE_EMAIL,
-          password: val.DEMO_MODE_PASSWORD,
-        }
+        email: val.DEMO_MODE_EMAIL,
+        password: val.DEMO_MODE_PASSWORD,
+      }
       : undefined,
     dataDir: val.DATA_DIR,
     maxAssetSizeMb: val.MAX_ASSET_SIZE_MB,
     serverVersion: val.SERVER_VERSION,
     disableNewReleaseCheck: val.DISABLE_NEW_RELEASE_CHECK,
     usingLegacySeparateContainers: val.USING_LEGACY_SEPARATE_CONTAINERS,
+    webhook: {
+      urls: val.WEBHOOK_URLS,
+      token: val.WEBHOOK_TOKEN,
+      timeout: val.WEBHOOK_TIMEOUT,
+    },
   };
 });
 

packages/shared/queues.ts

@@ -158,3 +158,30 @@ export const AssetPreprocessingQueue =
       keepFailedJobs: false,
     },
   );
+
+
+//Webhook worker
+export const zWebhookRequestSchema = z.object({
+  bookmarkId: z.string(),
+  url: z.string(),
+  operation: z.enum(["create"]),
+});
+export type ZWebhookRequest = z.infer<typeof zWebhookRequestSchema>;
+export const WebhookQueue = new SqliteQueue<ZWebhookRequest>(
+  "webhook_queue",
+  queueDB,
+  {
+    defaultJobArgs: {
+      numRetries: 1,
+    },
+    keepFailedJobs: false,
+  },
+);
+
+export async function triggerWebhookWorker(bookmarkId: string, url: string, operation: "create") {
+  await WebhookQueue.enqueue({
+    bookmarkId,
+    url,
+    operation,
+  });
+}