From 78580f65df626df8fec971107f42b709fcaff2d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gerg=C5=91=20M=C3=B3ricz?= Date: Thu, 5 Jun 2025 22:04:22 +0200 Subject: [PATCH] feat(webhook): refactor callWebhook and add logWebhook (FIR-2218) (#1629) * feat(webhook): refactor callWebhook and add logWebhook * feat(queue-worker): fix crawl pre-finishing logic (#1628) * feat(ci): verify typescript errors * fix(ci): * feat(api/tests): add webhook tests + refactor batch scrape lib (#1630) * feat(api/tests): add webhook tests + refactor batch scrape lib * fix(ci): * feat(webhook/log): insert queue --- .../v0/admin/index-queue-prometheus.ts | 3 + apps/api/src/controllers/v1/batch-scrape.ts | 16 +- .../api/src/services/indexing/index-worker.ts | 10 ++ apps/api/src/services/queue-worker.ts | 77 ++++----- apps/api/src/services/webhook.ts | 151 +++++++++++++++--- 5 files changed, 190 insertions(+), 67 deletions(-) diff --git a/apps/api/src/controllers/v0/admin/index-queue-prometheus.ts b/apps/api/src/controllers/v0/admin/index-queue-prometheus.ts index d1476869..a9ec6b03 100644 --- a/apps/api/src/controllers/v0/admin/index-queue-prometheus.ts +++ b/apps/api/src/controllers/v0/admin/index-queue-prometheus.ts @@ -1,12 +1,15 @@ import type { Request, Response } from "express"; import { getIndexInsertQueueLength } from "../../../services"; +import { getWebhookInsertQueueLength } from "../../../services/webhook"; export async function indexQueuePrometheus(req: Request, res: Response) { const queueLength = await getIndexInsertQueueLength(); + const webhookQueueLength = await getWebhookInsertQueueLength(); res.setHeader("Content-Type", "text/plain"); res.send(`\ # HELP firecrawl_index_queue_length The number of items in the index insert queue # TYPE firecrawl_index_queue_length gauge firecrawl_index_queue_length ${queueLength} +firecrawl_webhook_queue_length ${webhookQueueLength} `); } \ No newline at end of file diff --git a/apps/api/src/controllers/v1/batch-scrape.ts b/apps/api/src/controllers/v1/batch-scrape.ts index 5016e2ab..556551c0 100644 --- a/apps/api/src/controllers/v1/batch-scrape.ts +++ b/apps/api/src/controllers/v1/batch-scrape.ts @@ -171,14 +171,14 @@ export async function batchScrapeController( logger.debug("Calling webhook with batch_scrape.started...", { webhook: req.body.webhook, }); - await callWebhook( - req.auth.team_id, - id, - null, - req.body.webhook, - true, - "batch_scrape.started", - ); + await callWebhook({ + teamId: req.auth.team_id, + crawlId: id, + data: null, + webhook: req.body.webhook, + v1: true, + eventType: "batch_scrape.started", + }); } const protocol = process.env.ENV === "local" ? req.protocol : "https"; diff --git a/apps/api/src/services/indexing/index-worker.ts b/apps/api/src/services/indexing/index-worker.ts index 677d33fa..06ad21f5 100644 --- a/apps/api/src/services/indexing/index-worker.ts +++ b/apps/api/src/services/indexing/index-worker.ts @@ -15,6 +15,7 @@ import { processBillingBatch, queueBillingOperation, startBillingBatchProcessing import systemMonitor from "../system-monitor"; import { v4 as uuidv4 } from "uuid"; import { processIndexInsertJobs } from ".."; +import { processWebhookInsertJobs } from "../webhook"; const workerLockDuration = Number(process.env.WORKER_LOCK_DURATION) || 60000; const workerStalledCheckInterval = @@ -228,6 +229,7 @@ const workerFun = async (queue: Queue, jobProcessor: (token: string, job: Job) = }; const INDEX_INSERT_INTERVAL = 15000; +const WEBHOOK_INSERT_INTERVAL = 15000; // Start the workers (async () => { @@ -246,8 +248,16 @@ const INDEX_INSERT_INTERVAL = 15000; await processIndexInsertJobs(); }, INDEX_INSERT_INTERVAL); + const webhookInserterInterval = setInterval(async () => { + if (isShuttingDown) { + return; + } + await processWebhookInsertJobs(); + }, WEBHOOK_INSERT_INTERVAL); + // Wait for both workers to complete (which should only happen on shutdown) await Promise.all([indexWorkerPromise, billingWorkerPromise]); clearInterval(indexInserterInterval); + clearInterval(webhookInserterInterval); })(); diff --git a/apps/api/src/services/queue-worker.ts b/apps/api/src/services/queue-worker.ts index 4f178cbf..793fa85c 100644 --- a/apps/api/src/services/queue-worker.ts +++ b/apps/api/src/services/queue-worker.ts @@ -344,16 +344,16 @@ async function finishCrawlIfNeeded(job: Job & { id: string }, sc: StoredCrawl) { // v0 web hooks, call when done with all the data if (!job.data.v1) { - callWebhook( - job.data.team_id, - job.data.crawl_id, + callWebhook({ + teamId: job.data.team_id, + crawlId: job.data.crawl_id, data, - job.data.webhook, - job.data.v1, - job.data.crawlerOptions !== null + webhook: job.data.webhook, + v1: job.data.v1, + eventType: job.data.crawlerOptions !== null ? "crawl.completed" : "batch_scrape.completed", - ); + }); } } else { const num_docs = await getDoneJobsOrderedLength(job.data.crawl_id); @@ -384,16 +384,16 @@ async function finishCrawlIfNeeded(job: Job & { id: string }, sc: StoredCrawl) { // v1 web hooks, call when done with no data, but with event completed if (job.data.v1 && job.data.webhook) { - callWebhook( - job.data.team_id, - job.data.crawl_id, - [], - job.data.webhook, - job.data.v1, - job.data.crawlerOptions !== null + callWebhook({ + teamId: job.data.team_id, + crawlId: job.data.crawl_id, + data: [], + webhook: job.data.webhook, + v1: job.data.v1, + eventType: job.data.crawlerOptions !== null ? "crawl.completed" : "batch_scrape.completed", - ); + }); } } } @@ -988,14 +988,14 @@ async function processKickoffJob(job: Job & { id: string }, token: string) { logger.debug("Calling webhook with crawl.started...", { webhook: job.data.webhook, }); - await callWebhook( - job.data.team_id, - job.data.crawl_id, - null, - job.data.webhook, - true, - "crawl.started", - ); + callWebhook({ + teamId: job.data.team_id, + crawlId: job.data.crawl_id, + data: null, + webhook: job.data.webhook, + v1: job.data.v1, + eventType: "crawl.started", + }); } const sitemap = sc.crawlerOptions.ignoreSitemap @@ -1478,15 +1478,15 @@ async function processJob(job: Job & { id: string }, token: string) { logger.debug("Calling webhook with success...", { webhook: job.data.webhook, }); - await callWebhook( - job.data.team_id, - job.data.crawl_id, + callWebhook({ + teamId: job.data.team_id, + crawlId: job.data.crawl_id, + scrapeId: job.id, data, - job.data.webhook, - job.data.v1, - job.data.crawlerOptions !== null ? "crawl.page" : "batch_scrape.page", - true, - ); + webhook: job.data.webhook, + v1: job.data.v1, + eventType: job.data.crawlerOptions !== null ? "crawl.page" : "batch_scrape.page", + }); } logger.debug("Declaring job as done..."); @@ -1583,14 +1583,15 @@ async function processJob(job: Job & { id: string }, token: string) { }; if (!job.data.v1 && (job.data.mode === "crawl" || job.data.crawl_id)) { - callWebhook( - job.data.team_id, - job.data.crawl_id ?? (job.id as string), + callWebhook({ + teamId: job.data.team_id, + crawlId: job.data.crawl_id ?? (job.id as string), + scrapeId: job.id, data, - job.data.webhook, - job.data.v1, - job.data.crawlerOptions !== null ? "crawl.page" : "batch_scrape.page", - ); + webhook: job.data.webhook, + v1: job.data.v1, + eventType: job.data.crawlerOptions !== null ? "crawl.page" : "batch_scrape.page", + }); } const end = Date.now(); diff --git a/apps/api/src/services/webhook.ts b/apps/api/src/services/webhook.ts index 48d864a8..ae007107 100644 --- a/apps/api/src/services/webhook.ts +++ b/apps/api/src/services/webhook.ts @@ -1,38 +1,107 @@ -import axios from "axios"; -import { logger as _logger } from "../lib/logger"; +import axios, { AxiosError } from "axios"; +import { logger as _logger, logger } from "../lib/logger"; import { supabase_rr_service, supabase_service } from "./supabase"; import { WebhookEventType } from "../types"; import { configDotenv } from "dotenv"; import { z } from "zod"; import { webhookSchema } from "../controllers/v1/types"; +import { redisEvictConnection } from "./redis"; +import { index_supabase_service } from "."; configDotenv(); -export const callWebhook = async ( - teamId: string, - id: string, - data: any | null, - specified?: z.infer, - v1 = false, - eventType: WebhookEventType = "crawl.page", - awaitWebhook: boolean = false, -) => { +const WEBHOOK_INSERT_QUEUE_KEY = "webhook-insert-queue"; +const WEBHOOK_INSERT_BATCH_SIZE = 1000; + +async function addWebhookInsertJob(data: any) { + await redisEvictConnection.rpush(WEBHOOK_INSERT_QUEUE_KEY, JSON.stringify(data)); +} + +export async function getWebhookInsertQueueLength(): Promise { + return await redisEvictConnection.llen(WEBHOOK_INSERT_QUEUE_KEY) ?? 0; +} + +async function getWebhookInsertJobs(): Promise { + const jobs = (await redisEvictConnection.lpop(WEBHOOK_INSERT_QUEUE_KEY, WEBHOOK_INSERT_BATCH_SIZE)) ?? []; + return jobs.map(x => JSON.parse(x)); +} + +export async function processWebhookInsertJobs() { + const jobs = await getWebhookInsertJobs(); + if (jobs.length === 0) { + return; + } + logger.info(`Webhook inserter found jobs to insert`, { jobCount: jobs.length }); + try { + await supabase_service.from("webhook_logs").insert(jobs); + logger.info(`Webhook inserter inserted jobs`, { jobCount: jobs.length }); + } catch (error) { + logger.error(`Webhook inserter failed to insert jobs`, { error, jobCount: jobs.length }); + } +} + +async function logWebhook(data: { + success: boolean; + error?: string; + teamId: string; + crawlId: string; + scrapeId?: string; + url: string; + statusCode?: number; + event: WebhookEventType +}) { + try { + await addWebhookInsertJob({ + success: data.success, + error: data.error ?? null, + team_id: data.teamId, + crawl_id: data.crawlId, + scrape_id: data.scrapeId ?? null, + url: data.url, + status_code: data.statusCode ?? null, + event: data.event, + }); + } catch (error) { + _logger.error("Error logging webhook", { error, crawlId: data.crawlId, scrapeId: data.scrapeId, teamId: data.teamId, team_id: data.teamId, module: "webhook", method: "logWebhook" }); + } +} + +export const callWebhook = async ({ + teamId, + crawlId, + scrapeId, + data, + webhook, + v1, + eventType, + awaitWebhook = false, +}: { + teamId: string; + crawlId: string; + scrapeId?: string; + webhook?: z.infer, + v1: boolean, + data: any | null; + eventType: WebhookEventType, + awaitWebhook?: boolean; +}) => { const logger = _logger.child({ module: "webhook", method: "callWebhook", teamId, team_id: teamId, - crawlId: id, + crawlId, + scrapeId, eventType, awaitWebhook, - webhook: specified, + webhook, isV1: v1, }); - if (specified) { + if (webhook) { let subType = eventType.split(".")[1]; - if (!specified.events.includes(subType as any)) { + if (!webhook.events.includes(subType as any)) { logger.debug("Webhook event type not in specified events", { subType, - specified, + webhook, }); return false; } @@ -41,11 +110,11 @@ export const callWebhook = async ( try { const selfHostedUrl = process.env.SELF_HOSTED_WEBHOOK_URL?.replace( "{{JOB_ID}}", - id, + crawlId, ); const useDbAuthentication = process.env.USE_DB_AUTHENTICATION === "true"; let webhookUrl = - specified ?? + webhook ?? (selfHostedUrl ? webhookSchema.parse({ url: selfHostedUrl }) : undefined); // Only fetch the webhook URL from the database if the self-hosted webhook URL and specified webhook are not set @@ -103,7 +172,7 @@ export const callWebhook = async ( if (awaitWebhook) { try { - await axios.post( + const res = await axios.post( webhookUrl.url, { success: !v1 @@ -112,7 +181,7 @@ export const callWebhook = async ( ? data.success : true, type: eventType, - [v1 ? "id" : "jobId"]: id, + [v1 ? "id" : "jobId"]: crawlId, data: dataToSend, error: !v1 ? data?.error || undefined @@ -129,6 +198,15 @@ export const callWebhook = async ( timeout: v1 ? 10000 : 30000, // 10 seconds timeout (v1) }, ); + logWebhook({ + success: res.status >= 200 && res.status < 300, + teamId, + crawlId, + scrapeId, + url: webhookUrl.url, + event: eventType, + statusCode: res.status, + }); } catch (error) { logger.error( `Failed to send webhook`, @@ -136,6 +214,16 @@ export const callWebhook = async ( error, }, ); + logWebhook({ + success: false, + teamId, + crawlId, + scrapeId, + url: webhookUrl.url, + event: eventType, + error: error instanceof Error ? error.message : (typeof error === "string" ? error : undefined), + statusCode: error instanceof AxiosError ? error.response?.status : undefined, + }); } } else { axios @@ -148,7 +236,7 @@ export const callWebhook = async ( ? data.success : true, type: eventType, - [v1 ? "id" : "jobId"]: id, + [v1 ? "id" : "jobId"]: crawlId, data: dataToSend, error: !v1 ? data?.error || undefined @@ -164,6 +252,17 @@ export const callWebhook = async ( }, }, ) + .then((res) => { + logWebhook({ + success: res.status >= 200 && res.status < 300, + teamId, + crawlId, + scrapeId, + url: webhookUrl.url, + event: eventType, + statusCode: res.status, + }); + }) .catch((error) => { logger.error( `Failed to send webhook`, @@ -171,6 +270,16 @@ export const callWebhook = async ( error, }, ); + logWebhook({ + success: false, + teamId, + crawlId, + scrapeId, + url: webhookUrl.url, + event: eventType, + error: error instanceof Error ? error.message : (typeof error === "string" ? error : undefined), + statusCode: error instanceof AxiosError ? error.response?.status : undefined, + }); }); } } catch (error) {