import undici from "undici";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 { createHmac } from "crypto";import { getSecureDispatcher, isIPPrivate,} from "../scraper/scrapeURL/engines/utils/safeFetch";configDotenv(); const WEBHOOK_INSERT_QUEUE_KEY = "webhook-insert-queue";const WEBHOOK_INSERT_BATCH_SIZE = 1000; interface WebhookLogData { success: boolean; error?: string; teamId: string; crawlId: string; scrapeId?: string; url: string; statusCode?: number; event: WebhookEventType;} interface CallWebhookParams { teamId: string; crawlId: string; scrapeId?: string; webhook?: z.infer<typeof webhookSchema>; v1: boolean; data: any | null; eventType: WebhookEventType; awaitWebhook?: boolean;} async function addWebhookInsertJob(data: any) { await redisEvictConnection.rpush( WEBHOOK_INSERT_QUEUE_KEY, JSON.stringify(data), );} export async function getWebhookInsertQueueLength(): Promise<number> { return (await redisEvictConnection.llen(WEBHOOK_INSERT_QUEUE_KEY)) ?? 0;} async function getWebhookInsertJobs(): Promise<any[]> { 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: WebhookLogData) { 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", }); }} function generateHmacSignature(payload: string, secret: string): string { const hmac = createHmac("sha256", secret); hmac.update(payload); return hmac.digest("hex");} export const callWebhook = async ({ teamId, crawlId, scrapeId, data, webhook, v1, eventType, awaitWebhook = false,}: CallWebhookParams) => { const logger = _logger.child({ module: "webhook", method: "callWebhook", teamId, team_id: teamId, crawlId, scrapeId, eventType, awaitWebhook, webhook, isV1: v1, }); if (webhook) { let subType = eventType.split(".")[1]; if (!webhook.events.includes(subType as any)) { logger.debug("Webhook event type not in specified events", { subType, webhook, }); return false; } } try { const selfHostedUrl = process.env.SELF_HOSTED_WEBHOOK_URL?.replace( "{{JOB_ID}}", crawlId, ); const useDbAuthentication = process.env.USE_DB_AUTHENTICATION === "true"; let webhookUrl = webhook ?? (selfHostedUrl ? webhookSchema.parse({ url: selfHostedUrl }) : undefined); // TODO: do these queries upstream so we don't have to query the DB multiple times per crawl // Only fetch the webhook URL from the database if the self-hosted webhook URL and specified webhook are not set // and the USE_DB_AUTHENTICATION environment variable is set to true if (!webhookUrl && useDbAuthentication) { const { data: webhooksData, error } = await supabase_rr_service .from("webhooks") .select("url") .eq("team_id", teamId) .limit(1); if (error) { logger.error(`Error fetching webhook URL for team`, { error, }); return null; } if (!webhooksData || webhooksData.length === 0) { return null; } webhookUrl = webhooksData[0].url; } let hmacSecret: string | undefined = process.env.SELF_HOSTED_WEBHOOK_HMAC_SECRET; if (useDbAuthentication) { const { data: teamData, error: teamError } = await supabase_rr_service .from("teams") .select("hmac_secret") .eq("id", teamId) .limit(1) .single(); if (teamError) { logger.error(`Error fetching team HMAC secret`, { error: teamError, }); } if (teamData?.hmac_secret) hmacSecret = teamData.hmac_secret; } logger.debug("Calling webhook...", { webhookUrl, }); if (!webhookUrl) { return null; } // check if the webhook URL is a private IP address *before* making the request // the dispatcher also performs a check once connected, however this prevents unnecessary connections const webhookHost = new URL(webhookUrl.url).hostname; if (isIPPrivate(webhookHost)) { logger.warn("Aborting webhook call to private IP address", { url: webhookUrl.url, }); return null; } let dataToSend: any[] = []; if ( data && data.result && data.result.links && data.result.links.length !== 0 ) { for (let i = 0; i < data.result.links.length; i++) { if (v1) { dataToSend.push(data.result.links[i].content); } else { dataToSend.push({ content: data.result.links[i].content.content, markdown: data.result.links[i].content.markdown, metadata: data.result.links[i].content.metadata, }); } } } const payload = { success: !v1 ? data.success : eventType === "crawl.page" ? data.success : true, type: eventType, [v1 ? "id" : "jobId"]: crawlId, data: dataToSend, error: !v1 ? data?.error || undefined : eventType === "crawl.page" ? data?.error || undefined : undefined, metadata: webhookUrl.metadata || undefined, }; const payloadString = JSON.stringify(payload); const headers: Record<string, string> = { "Content-Type": "application/json", ...webhookUrl.headers, }; if (hmacSecret) { const signature = generateHmacSignature(payloadString, hmacSecret); headers["X-Firecrawl-Signature"] = `sha256=${signature}`; } const executeWebhookRequest = async () => { try { const res = await undici.fetch(webhookUrl.url, { method: "POST", headers, body: payloadString, dispatcher: getSecureDispatcher(), signal: AbortSignal.timeout(v1 ? 10000 : 30000), // 10 seconds timeout (v1) }); if (!res.ok) { throw { status: res.status }; } logWebhook({ success: res.status >= 200 && res.status < 300, teamId, crawlId, scrapeId, url: webhookUrl.url, event: eventType, statusCode: res.status, }); return res; } catch (error) { logger.error(`Failed to send webhook`, { error, }); logWebhook({ success: false, teamId, crawlId, scrapeId, url: webhookUrl.url, event: eventType, error: error instanceof Error ? error.message : typeof error === "string" ? error : undefined, statusCode: typeof (error as any)?.status === "number" ? (error as any).status : undefined, }); throw error; } }; if (awaitWebhook) { await executeWebhookRequest(); } else { executeWebhookRequest().catch(() => { // already logged in executeWebhookRequest }); } } catch (error) { logger.warn(`Error sending webhook`, { error, }); }};
import undici from "undici";
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 { createHmac } from "crypto";
import {
getSecureDispatcher,
isIPPrivate,
} from "../scraper/scrapeURL/engines/utils/safeFetch";
configDotenv();
const WEBHOOK_INSERT_QUEUE_KEY = "webhook-insert-queue";
const WEBHOOK_INSERT_BATCH_SIZE = 1000;
interface WebhookLogData {
success: boolean;
error?: string;
teamId: string;
crawlId: string;
scrapeId?: string;
url: string;
statusCode?: number;
event: WebhookEventType;
}
interface CallWebhookParams {
teamId: string;
crawlId: string;
scrapeId?: string;
webhook?: z.infer<typeof webhookSchema>;
v1: boolean;
data: any | null;
eventType: WebhookEventType;
awaitWebhook?: boolean;
}
async function addWebhookInsertJob(data: any) {
await redisEvictConnection.rpush(
WEBHOOK_INSERT_QUEUE_KEY,
JSON.stringify(data),
);
}
export async function getWebhookInsertQueueLength(): Promise<number> {
return (await redisEvictConnection.llen(WEBHOOK_INSERT_QUEUE_KEY)) ?? 0;
}
async function getWebhookInsertJobs(): Promise<any[]> {
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: WebhookLogData) {
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",
});
}
}
function generateHmacSignature(payload: string, secret: string): string {
const hmac = createHmac("sha256", secret);
hmac.update(payload);
return hmac.digest("hex");
}
export const callWebhook = async ({
teamId,
crawlId,
scrapeId,
data,
webhook,
v1,
eventType,
awaitWebhook = false,
}: CallWebhookParams) => {
const logger = _logger.child({
module: "webhook",
method: "callWebhook",
teamId,
team_id: teamId,
crawlId,
scrapeId,
eventType,
awaitWebhook,
webhook,
isV1: v1,
});
if (webhook) {
let subType = eventType.split(".")[1];
if (!webhook.events.includes(subType as any)) {
logger.debug("Webhook event type not in specified events", {
subType,
webhook,
});
return false;
}
}
try {
const selfHostedUrl = process.env.SELF_HOSTED_WEBHOOK_URL?.replace(
"{{JOB_ID}}",
crawlId,
);
const useDbAuthentication = process.env.USE_DB_AUTHENTICATION === "true";
let webhookUrl =
webhook ??
(selfHostedUrl ? webhookSchema.parse({ url: selfHostedUrl }) : undefined);
// TODO: do these queries upstream so we don't have to query the DB multiple times per crawl
// Only fetch the webhook URL from the database if the self-hosted webhook URL and specified webhook are not set
// and the USE_DB_AUTHENTICATION environment variable is set to true
if (!webhookUrl && useDbAuthentication) {
const { data: webhooksData, error } = await supabase_rr_service
.from("webhooks")
.select("url")
.eq("team_id", teamId)
.limit(1);
if (error) {
logger.error(`Error fetching webhook URL for team`, {
error,
});
return null;
}
if (!webhooksData || webhooksData.length === 0) {
return null;
}
webhookUrl = webhooksData[0].url;
}
let hmacSecret: string | undefined =
process.env.SELF_HOSTED_WEBHOOK_HMAC_SECRET;
if (useDbAuthentication) {
const { data: teamData, error: teamError } = await supabase_rr_service
.from("teams")
.select("hmac_secret")
.eq("id", teamId)
.limit(1)
.single();
if (teamError) {
logger.error(`Error fetching team HMAC secret`, {
error: teamError,
});
}
if (teamData?.hmac_secret) hmacSecret = teamData.hmac_secret;
}
logger.debug("Calling webhook...", {
webhookUrl,
});
if (!webhookUrl) {
return null;
}
// check if the webhook URL is a private IP address *before* making the request
// the dispatcher also performs a check once connected, however this prevents unnecessary connections
const webhookHost = new URL(webhookUrl.url).hostname;
if (isIPPrivate(webhookHost)) {
logger.warn("Aborting webhook call to private IP address", {
url: webhookUrl.url,
});
return null;
}
let dataToSend: any[] = [];
if (
data &&
data.result &&
data.result.links &&
data.result.links.length !== 0
) {
for (let i = 0; i < data.result.links.length; i++) {
if (v1) {
dataToSend.push(data.result.links[i].content);
} else {
dataToSend.push({
content: data.result.links[i].content.content,
markdown: data.result.links[i].content.markdown,
metadata: data.result.links[i].content.metadata,
});
}
}
}
const payload = {
success: !v1
? data.success
: eventType === "crawl.page"
? data.success
: true,
type: eventType,
[v1 ? "id" : "jobId"]: crawlId,
data: dataToSend,
error: !v1
? data?.error || undefined
: eventType === "crawl.page"
? data?.error || undefined
: undefined,
metadata: webhookUrl.metadata || undefined,
};
const payloadString = JSON.stringify(payload);
const headers: Record<string, string> = {
"Content-Type": "application/json",
...webhookUrl.headers,
};
if (hmacSecret) {
const signature = generateHmacSignature(payloadString, hmacSecret);
headers["X-Firecrawl-Signature"] = `sha256=${signature}`;
}
const executeWebhookRequest = async () => {
try {
const res = await undici.fetch(webhookUrl.url, {
method: "POST",
headers,
body: payloadString,
dispatcher: getSecureDispatcher(),
signal: AbortSignal.timeout(v1 ? 10000 : 30000), // 10 seconds timeout (v1)
});
if (!res.ok) {
throw { status: res.status };
}
logWebhook({
success: res.status >= 200 && res.status < 300,
teamId,
crawlId,
scrapeId,
url: webhookUrl.url,
event: eventType,
statusCode: res.status,
});
return res;
} catch (error) {
logger.error(`Failed to send webhook`, {
error,
});
logWebhook({
success: false,
teamId,
crawlId,
scrapeId,
url: webhookUrl.url,
event: eventType,
error:
error instanceof Error
? error.message
: typeof error === "string"
? error
: undefined,
statusCode:
typeof (error as any)?.status === "number"
? (error as any).status
: undefined,
});
throw error;
}
};
if (awaitWebhook) {
await executeWebhookRequest();
} else {
executeWebhookRequest().catch(() => {
// already logged in executeWebhookRequest
});
}
} catch (error) {
logger.warn(`Error sending webhook`, {
error,
});
}
};