feat: Webhook (#8457)

* feat: introduce webhook

* wip

* wip

* wip

* Update CHANGELOG.md

packages/backend/src/queue/index.ts

@@ -8,13 +8,15 @@ import processInbox from './processors/inbox.js';
 import processDb from './processors/db/index.js';
 import processObjectStorage from './processors/object-storage/index.js';
 import processSystemQueue from './processors/system/index.js';
+import processWebhookDeliver from './processors/webhook-deliver.js';
 import { endedPollNotification } from './processors/ended-poll-notification.js';
 import { queueLogger } from './logger.js';
 import { DriveFile } from '@/models/entities/drive-file.js';
 import { getJobInfo } from './get-job-info.js';
-import { systemQueue, dbQueue, deliverQueue, inboxQueue, objectStorageQueue, endedPollNotificationQueue } from './queues.js';
+import { systemQueue, dbQueue, deliverQueue, inboxQueue, objectStorageQueue, endedPollNotificationQueue, webhookDeliverQueue } from './queues.js';
 import { ThinUser } from './types.js';
 import { IActivity } from '@/remote/activitypub/type.js';
+import { Webhook } from '@/models/entities/webhook.js';
 
 function renderError(e: Error): any {
 	return {
@@ -26,6 +28,7 @@ function renderError(e: Error): any {
 
 const systemLogger = queueLogger.createSubLogger('system');
 const deliverLogger = queueLogger.createSubLogger('deliver');
+const webhookLogger = queueLogger.createSubLogger('webhook');
 const inboxLogger = queueLogger.createSubLogger('inbox');
 const dbLogger = queueLogger.createSubLogger('db');
 const objectStorageLogger = queueLogger.createSubLogger('objectStorage');
@@ -70,6 +73,14 @@ objectStorageQueue
 	.on('error', (job: any, err: Error) => objectStorageLogger.error(`error ${err}`, { job, e: renderError(err) }))
 	.on('stalled', (job) => objectStorageLogger.warn(`stalled id=${job.id}`));
 
+webhookDeliverQueue
+	.on('waiting', (jobId) => webhookLogger.debug(`waiting id=${jobId}`))
+	.on('active', (job) => webhookLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+	.on('completed', (job, result) => webhookLogger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+	.on('failed', (job, err) => webhookLogger.warn(`failed(${err}) ${getJobInfo(job)} to=${job.data.to}`))
+	.on('error', (job: any, err: Error) => webhookLogger.error(`error ${err}`, { job, e: renderError(err) }))
+	.on('stalled', (job) => webhookLogger.warn(`stalled ${getJobInfo(job)} to=${job.data.to}`));
+
 export function deliver(user: ThinUser, content: unknown, to: string | null) {
 	if (content == null) return null;
 	if (to == null) return null;
@@ -251,12 +262,32 @@ export function createCleanRemoteFilesJob() {
 	});
 }
 
+export function webhookDeliver(webhook: Webhook, content: unknown) {
+	const data = {
+		content,
+		webhookId: webhook.id,
+		to: webhook.url,
+		secret: webhook.secret,
+	};
+
+	return webhookDeliverQueue.add(data, {
+		attempts: 4,
+		timeout: 1 * 60 * 1000,	// 1min
+		backoff: {
+			type: 'apBackoff',
+		},
+		removeOnComplete: true,
+		removeOnFail: true,
+	});
+}
+
 export default function() {
 	if (envOption.onlyServer) return;
 
 	deliverQueue.process(config.deliverJobConcurrency || 128, processDeliver);
 	inboxQueue.process(config.inboxJobConcurrency || 16, processInbox);
 	endedPollNotificationQueue.process(endedPollNotification);
+	webhookDeliverQueue.process(64, processWebhookDeliver);
 	processDb(dbQueue);
 	processObjectStorage(objectStorageQueue);
 

packages/backend/src/queue/processors/webhook-deliver.ts

@@ -0,0 +1,56 @@
+import { URL } from 'node:url';
+import Bull from 'bull';
+import Logger from '@/services/logger.js';
+import { WebhookDeliverJobData } from '../types.js';
+import { getResponse, StatusError } from '@/misc/fetch.js';
+import { Webhooks } from '@/models/index.js';
+import config from '@/config/index.js';
+
+const logger = new Logger('webhook');
+
+let latest: string | null = null;
+
+export default async (job: Bull.Job<WebhookDeliverJobData>) => {
+	try {
+		if (latest !== (latest = JSON.stringify(job.data.content, null, 2))) {
+			logger.debug(`delivering ${latest}`);
+		}
+
+		const res = await getResponse({
+			url: job.data.to,
+			method: 'POST',
+			headers: {
+				'User-Agent': 'Misskey-Hooks',
+				'X-Misskey-Host': config.host,
+				'X-Misskey-Hook-Id': job.data.webhookId,
+				'X-Misskey-Hook-Secret': job.data.secret,
+			},
+			body: JSON.stringify(job.data.content),
+		});
+
+		Webhooks.update({ id: job.data.webhookId }, {
+			latestSentAt: new Date(),
+			latestStatus: res.status,
+		});
+
+		return 'Success';
+	} catch (res) {
+		Webhooks.update({ id: job.data.webhookId }, {
+			latestSentAt: new Date(),
+			latestStatus: res instanceof StatusError ? res.statusCode : 1,
+		});
+
+		if (res instanceof StatusError) {
+			// 4xx
+			if (res.isClientError) {
+				return `${res.statusCode} ${res.statusMessage}`;
+			}
+
+			// 5xx etc.
+			throw `${res.statusCode} ${res.statusMessage}`;
+		} else {
+			// DNS error, socket error, timeout ...
+			throw res;
+		}
+	}
+};

packages/backend/src/queue/queues.ts

@@ -1,6 +1,6 @@
 import config from '@/config/index.js';
 import { initialize as initializeQueue } from './initialize.js';
-import { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData } from './types.js';
+import { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData, WebhookDeliverJobData } from './types.js';
 
 export const systemQueue = initializeQueue<Record<string, unknown>>('system');
 export const endedPollNotificationQueue = initializeQueue<EndedPollNotificationJobData>('endedPollNotification');
@@ -8,6 +8,7 @@ export const deliverQueue = initializeQueue<DeliverJobData>('deliver', config.de
 export const inboxQueue = initializeQueue<InboxJobData>('inbox', config.inboxJobPerSec || 16);
 export const dbQueue = initializeQueue<DbJobData>('db');
 export const objectStorageQueue = initializeQueue<ObjectStorageJobData>('objectStorage');
+export const webhookDeliverQueue = initializeQueue<WebhookDeliverJobData>('webhookDeliver', 64);
 
 export const queues = [
 	systemQueue,
@@ -16,4 +17,5 @@ export const queues = [
 	inboxQueue,
 	dbQueue,
 	objectStorageQueue,
+	webhookDeliverQueue,
 ];

packages/backend/src/queue/types.ts

@@ -1,6 +1,7 @@
 import { DriveFile } from '@/models/entities/drive-file.js';
 import { Note } from '@/models/entities/note';
 import { User } from '@/models/entities/user.js';
+import { Webhook } from '@/models/entities/webhook';
 import { IActivity } from '@/remote/activitypub/type.js';
 import httpSignature from 'http-signature';
 
@@ -46,6 +47,13 @@ export type EndedPollNotificationJobData = {
 	noteId: Note['id'];
 };
 
+export type WebhookDeliverJobData = {
+	content: unknown;
+	webhookId: Webhook['id'];
+	to: string;
+	secret: string;
+};
+
 export type ThinUser = {
 	id: User['id'];
 };