feat: outgoing webhooks

Owners and admins can point a project at other systems: a signed JSON
POST (HMAC-SHA256 of the body) when tickets, comments, releases or sprints
change, filtered by event or family. Deliveries are retried a couple of
times, the last one is shown, and a ping can be sent from the settings.
Private addresses are refused unless WEBHOOKS_ALLOW_PRIVATE is set.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-08 15:36:19 +02:00
parent 046d77c813
commit d43fc620e9
37 changed files with 3660 additions and 28 deletions

View File

@@ -12,6 +12,9 @@ export * from './lib/notifications';
export * from './lib/saved-filters.service';
export * from './lib/sprints.service';
export * from './lib/git.service';
export * from './lib/webhooks.service';
export * from './lib/webhooks/webhooks';
export * from './lib/webhooks/webhook-events';
export * from './lib/git/git';
export * from './lib/git/git-payloads';
export * from './lib/saved-filters/saved-filters';

View File

@@ -24,3 +24,4 @@ export * from './saved-filters.repository';
export * from './find-sprints.repository';
export * from './upsert-materialized-sprint.repository';
export * from './git-integrations.repository';
export * from './outgoing-webhooks.repository';

View File

@@ -0,0 +1,48 @@
import { UserRefReadModel } from '@task-stream/api/tracker/domain';
export interface WebhookDelivery {
at: string;
event: string;
/** HTTP status, null when it never answered. */
status: number | null;
ok: boolean;
error: string | null;
}
/** Where a project tells other systems what happened in it. */
export interface OutgoingWebhook {
id: string;
projectId: string;
url: string;
/** Signs every delivery (HMAC-SHA256); shown once, when created. */
secret: string;
/** Event names, `issue.*` or `*`. */
events: string[];
active: boolean;
createdBy: UserRefReadModel;
createdAt: string;
lastDelivery: WebhookDelivery | null;
}
/** Configuration, not history: kept outside the event store. */
export abstract class OutgoingWebhooksRepository {
abstract findByProject(projectId: string): Promise<OutgoingWebhook[]>;
abstract findActive(projectId: string): Promise<OutgoingWebhook[]>;
abstract findById(id: string): Promise<OutgoingWebhook | null>;
abstract insert(webhook: OutgoingWebhook): Promise<void>;
abstract update(
id: string,
changes: Partial<Pick<OutgoingWebhook, 'url' | 'events' | 'active'>>,
): Promise<void>;
abstract delete(id: string): Promise<void>;
abstract recordDelivery(id: string, delivery: WebhookDelivery): Promise<void>;
}
/** Sends one delivery. Throws when the receiver cannot be reached. */
export abstract class WebhookSender {
abstract send(request: {
url: string;
body: string;
headers: Record<string, string>;
}): Promise<{ status: number }>;
}

View File

@@ -34,6 +34,8 @@ import { SavedFiltersService } from './saved-filters.service';
import { SprintsService } from './sprints.service';
import { GitService } from './git.service';
import { GIT_HANDLERS } from './git/git';
import { WebhooksService } from './webhooks.service';
import { WEBHOOK_HANDLERS } from './webhooks/webhooks';
/** Notifications: derived from events, kept and dismissed per person. */
const NOTIFICATIONS = [
@@ -56,6 +58,7 @@ const SERVICES = [
SavedFiltersService,
SprintsService,
GitService,
WebhooksService,
];
/**
@@ -79,6 +82,7 @@ const SERVICES = [
...NOTIFICATIONS,
...SAVED_FILTERS,
...GIT_HANDLERS,
...WEBHOOK_HANDLERS,
],
exports: SERVICES,
})

View File

@@ -0,0 +1,62 @@
import { Injectable } from '@nestjs/common';
import { CommandBus, QueryBus } from '@nestjs/cqrs';
import { WebhookDelivery } from './ports/outgoing-webhooks.repository';
import { WEBHOOK_EVENT_NAMES } from './webhooks/webhook-events';
import {
CreateWebhookCommand,
DeleteWebhookCommand,
GetWebhooksQuery,
TestWebhookCommand,
UpdateWebhookCommand,
WebhookView,
} from './webhooks/webhooks';
/** The project's outgoing webhooks (owners and admins). */
@Injectable()
export class WebhooksService {
constructor(
private readonly commandBus: CommandBus,
private readonly queryBus: QueryBus,
) {}
/** Names other systems can subscribe to (plus `family.*` and `*`). */
events(): string[] {
return WEBHOOK_EVENT_NAMES;
}
list(viewerId: string, projectId: string): Promise<WebhookView[]> {
return this.queryBus.execute(new GetWebhooksQuery(viewerId, projectId));
}
/** The secret comes back only here. */
create(
actorId: string,
projectId: string,
url: string,
events: string[],
): Promise<{ id: string; secret: string }> {
return this.commandBus.execute(
new CreateWebhookCommand(actorId, projectId, url, events),
);
}
update(
actorId: string,
webhookId: string,
changes: { url?: string; events?: string[]; active?: boolean },
): Promise<WebhookView> {
return this.commandBus.execute(
new UpdateWebhookCommand(actorId, webhookId, changes),
);
}
delete(actorId: string, webhookId: string): Promise<void> {
return this.commandBus.execute(
new DeleteWebhookCommand(actorId, webhookId),
);
}
test(actorId: string, webhookId: string): Promise<WebhookDelivery> {
return this.commandBus.execute(new TestWebhookCommand(actorId, webhookId));
}
}

View File

@@ -0,0 +1,73 @@
import {
CommentAddedEvent,
CommentDeletedEvent,
CommentEditedEvent,
IssueAssignedEvent,
IssueCreatedEvent,
IssueDeletedEvent,
IssueDevLinkRecordedEvent,
IssueReleaseChangedEvent,
IssueSprintChangedEvent,
IssueStatusChangedEvent,
IssueUpdatedEvent,
ReleaseArchivedEvent,
ReleaseCreatedEvent,
ReleaseShippedEvent,
SprintClosedEvent,
SprintStartedEvent,
TrackerEvent,
} from '@task-stream/api/tracker/domain';
import { Type } from '@nestjs/common';
/** What other systems can subscribe to, by the event that causes it. */
export const WEBHOOK_EVENTS: [Type<TrackerEvent>, string][] = [
[IssueCreatedEvent, 'issue.created'],
[IssueUpdatedEvent, 'issue.updated'],
[IssueStatusChangedEvent, 'issue.status_changed'],
[IssueAssignedEvent, 'issue.assigned'],
[IssueSprintChangedEvent, 'issue.sprint_changed'],
[IssueReleaseChangedEvent, 'issue.release_changed'],
[IssueDevLinkRecordedEvent, 'issue.dev_link'],
[IssueDeletedEvent, 'issue.deleted'],
[CommentAddedEvent, 'comment.added'],
[CommentEditedEvent, 'comment.edited'],
[CommentDeletedEvent, 'comment.deleted'],
[ReleaseCreatedEvent, 'release.created'],
[ReleaseShippedEvent, 'release.shipped'],
[ReleaseArchivedEvent, 'release.archived'],
[SprintStartedEvent, 'sprint.started'],
[SprintClosedEvent, 'sprint.closed'],
];
export const WEBHOOK_EVENT_NAMES = WEBHOOK_EVENTS.map(([, name]) => name);
export function webhookEventOf(event: TrackerEvent): string | null {
return WEBHOOK_EVENTS.find(([type]) => event instanceof type)?.[1] ?? null;
}
/** `*`, an exact name, or a family (`issue.*`). */
export function subscribes(events: string[], name: string): boolean {
return events.some(
(e) =>
e === '*' ||
e === name ||
(e.endsWith('.*') && name.startsWith(e.slice(0, -1))),
);
}
/** Keeps known names and families, once each. */
export function normalizeWebhookEvents(events: string[]): string[] {
const families = new Set(
WEBHOOK_EVENT_NAMES.map((n) => `${n.split('.')[0]}.*`),
);
return [
...new Set(
events
.map((e) => e.trim())
.filter(
(e) =>
e === '*' || families.has(e) || WEBHOOK_EVENT_NAMES.includes(e),
),
),
];
}

View File

@@ -0,0 +1,90 @@
import { createHmac } from 'node:crypto';
import {
OutgoingWebhook,
OutgoingWebhooksRepository,
WebhookDelivery,
WebhookSender,
} from '../ports/outgoing-webhooks.repository';
import { normalizeWebhookEvents, subscribes } from './webhook-events';
import { WebhookDispatcher } from './webhooks';
const webhook: OutgoingWebhook = {
id: 'w1',
projectId: 'p1',
url: 'https://example.org/hook',
secret: 's3cret',
events: ['issue.*'],
active: true,
createdBy: { id: 'u', name: 'U' },
createdAt: '',
lastDelivery: null,
};
function doubles(statuses: (number | Error)[]) {
const recorded: WebhookDelivery[] = [];
const sent: { body: string; headers: Record<string, string> }[] = [];
const repo = {
recordDelivery: async (_id: string, d: WebhookDelivery) => {
recorded.push(d);
},
} as unknown as OutgoingWebhooksRepository;
const sender: WebhookSender = {
send: async (request) => {
sent.push(request);
const next = statuses.shift() ?? 200;
if (next instanceof Error) throw next;
return { status: next };
},
};
return { repo, sender, recorded, sent };
}
describe('outgoing webhooks', () => {
it('subscribes to names, families and everything', () => {
expect(subscribes(['issue.*'], 'issue.created')).toBe(true);
expect(subscribes(['issue.*'], 'comment.added')).toBe(false);
expect(subscribes(['*'], 'sprint.closed')).toBe(true);
expect(
normalizeWebhookEvents(['issue.*', 'nope', ' sprint.closed ', 'issue.*']),
).toEqual(['issue.*', 'sprint.closed']);
});
it('signs the body and retries until it gets through', async () => {
const { repo, sender, recorded, sent } = doubles([
new Error('ECONNRESET'),
503,
204,
]);
const delivery = await new WebhookDispatcher(
repo,
sender,
[0, 0, 0],
).deliver(webhook, 'issue.created', { data: { title: 'x' } });
expect(sent).toHaveLength(3);
const { body, headers } = sent[0];
expect(headers['x-taskstream-signature']).toBe(
`sha256=${createHmac('sha256', 's3cret').update(body).digest('hex')}`,
);
expect(JSON.parse(body)).toMatchObject({
event: 'issue.created',
data: { title: 'x' },
});
expect(delivery).toMatchObject({ ok: true, status: 204 });
expect(recorded).toEqual([delivery]);
});
it('gives up at once on a 4xx', async () => {
const { repo, sender, sent } = doubles([410]);
const delivery = await new WebhookDispatcher(
repo,
sender,
[0, 0, 0],
).deliver(webhook, 'issue.created', {});
expect(sent).toHaveLength(1);
expect(delivery).toMatchObject({
ok: false,
status: 410,
error: 'HTTP 410',
});
});
});

View File

@@ -0,0 +1,411 @@
import { createHmac, randomBytes, randomUUID } from 'node:crypto';
import { Logger } from '@nestjs/common';
import {
Command,
CommandHandler,
EventsHandler,
ICommandHandler,
IEventHandler,
IQueryHandler,
Query,
QueryHandler,
} from '@nestjs/cqrs';
import {
BusinessRuleError,
NotFoundError,
assertThat,
} from '@task-stream/api/shared/domain';
import { TrackerEvent } from '@task-stream/api/tracker/domain';
import { WriteAccess } from '../commands/write-access';
import { FindIssuesRepository } from '../ports/find-issues.repository';
import { FindProjectsRepository } from '../ports/find-projects.repository';
import {
OutgoingWebhook,
OutgoingWebhooksRepository,
WebhookDelivery,
WebhookSender,
} from '../ports/outgoing-webhooks.repository';
import { UserDirectory } from '../ports/user-directory';
import {
WEBHOOK_EVENTS,
normalizeWebhookEvents,
subscribes,
webhookEventOf,
} from './webhook-events';
/** A project keeps at most this many webhooks. */
export const MAX_WEBHOOKS = 20;
/** Waits between attempts: right away, after 2 s, after 10 s. */
export const RETRY_DELAYS_MS = [0, 2_000, 10_000];
/** The webhook as listed: the secret is only ever shown when created. */
export type WebhookView = Omit<OutgoingWebhook, 'secret'>;
const view = ({ secret: _secret, ...rest }: OutgoingWebhook): WebhookView =>
rest;
function assertUrl(url: string): string {
const next = url.trim();
let parsed: URL | null = null;
try {
parsed = new URL(next);
} catch {
parsed = null;
}
assertThat(
!!parsed && ['http:', 'https:'].includes(parsed.protocol),
'A webhook needs an http(s) URL',
);
return next;
}
export class CreateWebhookCommand extends Command<{
id: string;
secret: string;
}> {
constructor(
public readonly actorId: string,
public readonly projectId: string,
public readonly url: string,
public readonly events: string[],
) {
super();
}
}
export class UpdateWebhookCommand extends Command<WebhookView> {
constructor(
public readonly actorId: string,
public readonly webhookId: string,
public readonly changes: {
url?: string;
events?: string[];
active?: boolean;
},
) {
super();
}
}
export class DeleteWebhookCommand extends Command<void> {
constructor(
public readonly actorId: string,
public readonly webhookId: string,
) {
super();
}
}
/** Sends a `ping` now and tells how it went. */
export class TestWebhookCommand extends Command<WebhookDelivery> {
constructor(
public readonly actorId: string,
public readonly webhookId: string,
) {
super();
}
}
export class GetWebhooksQuery extends Query<WebhookView[]> {
constructor(
public readonly viewerId: string,
public readonly projectId: string,
) {
super();
}
}
/** Signs and sends deliveries, retrying a couple of times, and records the outcome. */
export class WebhookDispatcher {
private readonly logger = new Logger(WebhookDispatcher.name);
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
private readonly sender: WebhookSender,
private readonly delays: number[] = RETRY_DELAYS_MS,
) {}
async deliver(
webhook: OutgoingWebhook,
event: string,
payload: Record<string, unknown>,
): Promise<WebhookDelivery> {
const deliveryId = randomUUID();
const body = JSON.stringify({ event, deliveryId, ...payload });
const signature = createHmac('sha256', webhook.secret)
.update(body)
.digest('hex');
let delivery: WebhookDelivery = {
at: new Date().toISOString(),
event,
status: null,
ok: false,
error: null,
};
for (const delay of this.delays) {
if (delay) await new Promise((resolve) => setTimeout(resolve, delay));
try {
const { status } = await this.sender.send({
url: webhook.url,
body,
headers: {
'content-type': 'application/json',
'user-agent': 'TaskStream-Webhooks',
'x-taskstream-event': event,
'x-taskstream-delivery': deliveryId,
'x-taskstream-signature': `sha256=${signature}`,
},
});
delivery = {
...delivery,
at: new Date().toISOString(),
status,
ok: status < 300,
error: status < 300 ? null : `HTTP ${status}`,
};
// A 4xx will not get better by retrying.
if (delivery.ok || (status >= 400 && status < 500)) break;
} catch (error) {
delivery = {
...delivery,
at: new Date().toISOString(),
status: null,
ok: false,
error: error instanceof Error ? error.message : String(error),
};
}
}
if (!delivery.ok)
this.logger.warn(
`Webhook ${webhook.id} (${event}) failed: ${delivery.error}`,
);
await this.webhooks.recordDelivery(webhook.id, delivery);
return delivery;
}
}
/** Whoever manages the project of a webhook. */
async function managedWebhook(
webhooks: OutgoingWebhooksRepository,
access: WriteAccess,
webhookId: string,
actorId: string,
): Promise<OutgoingWebhook> {
const webhook = await webhooks.findById(webhookId);
if (!webhook) throw new NotFoundError('Webhook', webhookId);
(await access.project(webhook.projectId)).assertCanManage(actorId);
return webhook;
}
@CommandHandler(CreateWebhookCommand)
export class CreateWebhookCommandHandler
implements ICommandHandler<CreateWebhookCommand>
{
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
private readonly access: WriteAccess,
private readonly users: UserDirectory,
) {}
async execute(command: CreateWebhookCommand) {
(await this.access.project(command.projectId)).assertCanManage(
command.actorId,
);
const events = normalizeWebhookEvents(command.events);
assertThat(events.length > 0, 'Pick at least one event');
if (
(await this.webhooks.findByProject(command.projectId)).length >=
MAX_WEBHOOKS
)
throw new BusinessRuleError(
`A project keeps at most ${MAX_WEBHOOKS} webhooks`,
);
const users = await this.users.refs([command.actorId]);
const webhook: OutgoingWebhook = {
id: randomUUID(),
projectId: command.projectId,
url: assertUrl(command.url),
secret: randomBytes(24).toString('hex'),
events,
active: true,
createdBy: users.of(command.actorId),
createdAt: new Date().toISOString(),
lastDelivery: null,
};
await this.webhooks.insert(webhook);
return { id: webhook.id, secret: webhook.secret };
}
}
@CommandHandler(UpdateWebhookCommand)
export class UpdateWebhookCommandHandler
implements ICommandHandler<UpdateWebhookCommand>
{
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
private readonly access: WriteAccess,
) {}
async execute(command: UpdateWebhookCommand): Promise<WebhookView> {
const webhook = await managedWebhook(
this.webhooks,
this.access,
command.webhookId,
command.actorId,
);
const changes: { url?: string; events?: string[]; active?: boolean } = {};
if (command.changes.url !== undefined)
changes.url = assertUrl(command.changes.url);
if (command.changes.events !== undefined) {
changes.events = normalizeWebhookEvents(command.changes.events);
assertThat(changes.events.length > 0, 'Pick at least one event');
}
if (command.changes.active !== undefined)
changes.active = command.changes.active;
await this.webhooks.update(webhook.id, changes);
return view({ ...webhook, ...changes });
}
}
@CommandHandler(DeleteWebhookCommand)
export class DeleteWebhookCommandHandler
implements ICommandHandler<DeleteWebhookCommand>
{
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
private readonly access: WriteAccess,
) {}
async execute(command: DeleteWebhookCommand): Promise<void> {
const webhook = await managedWebhook(
this.webhooks,
this.access,
command.webhookId,
command.actorId,
);
await this.webhooks.delete(webhook.id);
}
}
@CommandHandler(TestWebhookCommand)
export class TestWebhookCommandHandler
implements ICommandHandler<TestWebhookCommand>
{
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
private readonly access: WriteAccess,
private readonly sender: WebhookSender,
) {}
async execute(command: TestWebhookCommand): Promise<WebhookDelivery> {
const webhook = await managedWebhook(
this.webhooks,
this.access,
command.webhookId,
command.actorId,
);
// One attempt: someone is waiting for the answer.
return new WebhookDispatcher(this.webhooks, this.sender, [0]).deliver(
webhook,
'ping',
{
occurredAt: new Date().toISOString(),
projectId: webhook.projectId,
actorId: command.actorId,
data: { message: 'Hola desde TaskStream' },
},
);
}
}
@QueryHandler(GetWebhooksQuery)
export class GetWebhooksQueryHandler
implements IQueryHandler<GetWebhooksQuery, WebhookView[]>
{
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
private readonly access: WriteAccess,
) {}
async execute(query: GetWebhooksQuery): Promise<WebhookView[]> {
(await this.access.project(query.projectId)).assertCanManage(
query.viewerId,
);
return (await this.webhooks.findByProject(query.projectId)).map(view);
}
}
/** The event's own fields: what happened, without the envelope. */
function dataOf(event: TrackerEvent): Record<string, unknown> {
const {
actorId: _actor,
aggregateId: _aggregate,
projectId: _project,
occurredAt: _at,
aggregateVersion: _version,
...data
} = event as unknown as Record<string, unknown>;
return data;
}
/**
* Tells the project's webhooks about what happened. Deliveries go out in the
* background: a slow receiver never holds the events bridge.
*/
@EventsHandler(...WEBHOOK_EVENTS.map(([type]) => type))
export class OutgoingWebhooksEventHandler
implements IEventHandler<TrackerEvent>
{
private readonly dispatcher: WebhookDispatcher;
constructor(
private readonly webhooks: OutgoingWebhooksRepository,
sender: WebhookSender,
private readonly findProjects: FindProjectsRepository,
private readonly findIssues: FindIssuesRepository,
) {
this.dispatcher = new WebhookDispatcher(webhooks, sender);
}
async handle(event: TrackerEvent) {
const name = webhookEventOf(event);
if (!name) return;
const targets = (await this.webhooks.findActive(event.projectId)).filter(
(w) => subscribes(w.events, name),
);
if (!targets.length) return;
const project = await this.findProjects.findById(event.projectId);
const issueId = event.constructor.name.startsWith('Issue')
? event.aggregateId
: ((event as { issueId?: string; comment?: { issueId: string } })
.issueId ??
(event as { comment?: { issueId: string } }).comment?.issueId ??
null);
const issue = issueId ? await this.findIssues.findById(issueId) : null;
const payload = {
occurredAt: event.occurredAt,
actorId: event.actorId,
project: project
? { id: project.id, key: project.key }
: { id: event.projectId },
aggregateId: event.aggregateId,
version: event.aggregateVersion ?? null,
...(issue
? { issue: { id: issue.id, key: issue.key, title: issue.title } }
: {}),
data: dataOf(event),
};
for (const webhook of targets)
void this.dispatcher
.deliver(webhook, name, payload)
.catch(() => undefined);
}
}
export const WEBHOOK_HANDLERS = [
CreateWebhookCommandHandler,
UpdateWebhookCommandHandler,
DeleteWebhookCommandHandler,
TestWebhookCommandHandler,
GetWebhooksQueryHandler,
OutgoingWebhooksEventHandler,
];