feat: notifications in the app and, optionally, by e-mail

Derived from events and kept per person: whoever a ticket is assigned to,
whoever reports or works on it when it moves, fails QA or gets a comment,
and whoever joins a project. Never about what you did yourself, once per
event. They arrive live through a subscription to a bell in the header,
and are copied by e-mail to whoever turns it on when SMTP_URL is set.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-08 09:19:59 +02:00
parent 0b65c1c912
commit efa06a0723
60 changed files with 3406 additions and 28 deletions

View File

@@ -0,0 +1,63 @@
import { Injectable } from '@nestjs/common';
import { CommandBus, QueryBus } from '@nestjs/cqrs';
import { NotificationReadModel } from '@task-stream/api/tracker/domain';
import {
MarkNotificationsReadCommand,
UpdateNotificationSettingsCommand,
} from './notifications/notifications.commands';
import {
GetMyNotificationSettingsQuery,
GetMyNotificationsQuery,
MyNotificationSettings,
MyNotifications,
} from './notifications/notifications.queries';
import { ChangeFeed } from './ports/change-feed';
/** The signed-in person's notifications: always their own, nobody else's. */
@Injectable()
export class NotificationsService {
constructor(
private readonly commandBus: CommandBus,
private readonly queryBus: QueryBus,
private readonly changeFeed: ChangeFeed,
) {}
mine(
userId: string,
options: { unreadOnly?: boolean; limit?: number } = {},
): Promise<MyNotifications> {
return this.queryBus.execute(
new GetMyNotificationsQuery(
userId,
options.unreadOnly ?? false,
options.limit ?? 30,
),
);
}
/** `ids` omitted: everything. */
markRead(userId: string, ids?: string[] | null): Promise<void> {
return this.commandBus.execute(
new MarkNotificationsReadCommand(userId, ids?.length ? ids : null),
);
}
settings(userId: string): Promise<MyNotificationSettings> {
return this.queryBus.execute(new GetMyNotificationSettingsQuery(userId));
}
async updateSettings(
userId: string,
changes: { email: boolean },
): Promise<MyNotificationSettings> {
await this.commandBus.execute(
new UpdateNotificationSettingsCommand(userId, changes.email),
);
return this.settings(userId);
}
/** New notifications for this person, as they happen. */
watch(userId: string): AsyncIterableIterator<NotificationReadModel> {
return this.changeFeed.notifications(userId);
}
}

View File

@@ -0,0 +1,3 @@
export * from './notifier';
export * from './notifications.queries';
export * from './notifications.commands';

View File

@@ -0,0 +1,51 @@
import { Command, CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { NotificationsRepository } from '../ports/notifications.repository';
/*
* Per-person state, outside the event store: whether a notification was read
* and whether to copy them by e-mail concern nobody else and need no history.
*/
export class MarkNotificationsReadCommand extends Command<void> {
constructor(
public readonly userId: string,
/** Null: all of them. */
public readonly ids: string[] | null,
) {
super();
}
}
export class UpdateNotificationSettingsCommand extends Command<void> {
constructor(
public readonly userId: string,
public readonly email: boolean,
) {
super();
}
}
@CommandHandler(MarkNotificationsReadCommand)
export class MarkNotificationsReadCommandHandler
implements ICommandHandler<MarkNotificationsReadCommand>
{
constructor(private readonly notifications: NotificationsRepository) {}
async execute(command: MarkNotificationsReadCommand): Promise<void> {
await this.notifications.markRead(command.userId, command.ids);
}
}
@CommandHandler(UpdateNotificationSettingsCommand)
export class UpdateNotificationSettingsCommandHandler
implements ICommandHandler<UpdateNotificationSettingsCommand>
{
constructor(private readonly notifications: NotificationsRepository) {}
async execute(command: UpdateNotificationSettingsCommand): Promise<void> {
await this.notifications.saveSettings({
userId: command.userId,
email: command.email,
});
}
}

View File

@@ -0,0 +1,143 @@
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import {
CommentAddedEvent,
IssueAssignedEvent,
IssueCreatedEvent,
IssueStatusChangedEvent,
NotificationKind,
ProjectMemberAddedEvent,
StatusCategory,
TrackerEvent,
} from '@task-stream/api/tracker/domain';
import { FindIssuesRepository } from '../ports/find-issues.repository';
import { FindProjectsRepository } from '../ports/find-projects.repository';
import { UserDirectory } from '../ports/user-directory';
import { NotificationDraft, Notifier } from './notifier';
/* Read by people, like the activity feed: Spanish. */
const excerpt = (text: string, max = 120) => {
const flat = text.replace(/\s+/g, ' ').trim();
return flat.length > max ? `${flat.slice(0, max - 1)}…` : flat;
};
/**
* Who is told what, for the events that concern someone in particular:
* whoever a ticket is assigned to, whoever reported or works on it when it
* moves or gets a comment, whoever joins a project.
*/
@EventsHandler(
IssueCreatedEvent,
IssueAssignedEvent,
IssueStatusChangedEvent,
CommentAddedEvent,
ProjectMemberAddedEvent,
)
export class NotificationsEventHandler implements IEventHandler<TrackerEvent> {
constructor(
private readonly notifier: Notifier,
private readonly findIssues: FindIssuesRepository,
private readonly findProjects: FindProjectsRepository,
private readonly users: UserDirectory,
) {}
async handle(event: TrackerEvent): Promise<void> {
const project = await this.findProjects.findById(event.projectId);
if (!project) return;
const actor = (await this.users.refs([event.actorId])).of(
event.actorId,
).name;
if (event instanceof ProjectMemberAddedEvent) {
return this.notifier.deliver(
event,
{
projectId: project.id,
projectKey: project.key,
issueId: null,
issueKey: null,
},
[
{
userId: event.userId,
kind: NotificationKind.ADDED_TO_PROJECT,
summary: `${actor} te añadió al proyecto ${project.name} (${project.key})`,
},
],
);
}
const issueId =
event instanceof CommentAddedEvent
? event.comment.issueId
: event.aggregateId;
const issue =
event instanceof IssueCreatedEvent
? event.issue
: await this.findIssues.findById(issueId);
if (!issue) return;
const reporterId =
'reporterId' in issue ? issue.reporterId : issue.reporter.id;
const assigneeId =
'assigneeId' in issue ? issue.assigneeId : (issue.assignee?.id ?? null);
const ticket = `${issue.key}: ${issue.title}`;
const drafts: NotificationDraft[] = [];
if (event instanceof IssueCreatedEvent && assigneeId) {
drafts.push({
userId: assigneeId,
kind: NotificationKind.ASSIGNED,
summary: `${actor} te asignó ${ticket}`,
});
}
if (event instanceof IssueAssignedEvent && event.assigneeId) {
drafts.push({
userId: event.assigneeId,
kind: NotificationKind.ASSIGNED,
summary: `${actor} te asignó ${ticket}`,
});
}
if (event instanceof IssueStatusChangedEvent) {
const to = event.meta?.toName ?? event.to;
// Back from QA without passing: the one move whoever did the work must hear about.
const failedQa =
event.meta?.fromCategory === StatusCategory.QA &&
event.meta?.toCategory !== StatusCategory.QA &&
event.meta?.toCategory !== StatusCategory.DONE;
const reason = event.reason ? `: ${excerpt(event.reason)}` : '';
for (const userId of [assigneeId, reporterId]) {
if (!userId) continue;
drafts.push({
userId,
kind: failedQa
? NotificationKind.QA_FAILED
: NotificationKind.STATUS_CHANGED,
summary: failedQa
? `${actor} devolvió ${ticket} desde QA${reason}`
: `${actor} movió ${ticket} a ${to}${reason}`,
});
}
}
if (event instanceof CommentAddedEvent) {
for (const userId of [assigneeId, reporterId]) {
if (!userId) continue;
drafts.push({
userId,
kind: NotificationKind.COMMENTED,
summary: `${actor} comentó en ${issue.key}: «${excerpt(event.comment.body)}»`,
});
}
}
await this.notifier.deliver(
event,
{
projectId: project.id,
projectKey: project.key,
issueId: issue.id,
issueKey: issue.key,
},
drafts,
);
}
}

View File

@@ -0,0 +1,71 @@
import { IQueryHandler, Query, QueryHandler } from '@nestjs/cqrs';
import { NotificationReadModel } from '@task-stream/api/tracker/domain';
import { NotificationMailer } from '../ports/notification-mailer';
import { NotificationsRepository } from '../ports/notifications.repository';
export interface MyNotifications {
items: NotificationReadModel[];
unread: number;
}
export interface MyNotificationSettings {
/** Copies by e-mail (the user's choice). */
email: boolean;
/** Whether this installation can send e-mail at all. */
emailAvailable: boolean;
}
export class GetMyNotificationsQuery extends Query<MyNotifications> {
constructor(
public readonly userId: string,
public readonly unreadOnly: boolean,
public readonly limit: number,
) {
super();
}
}
export class GetMyNotificationSettingsQuery extends Query<MyNotificationSettings> {
constructor(public readonly userId: string) {
super();
}
}
@QueryHandler(GetMyNotificationsQuery)
export class GetMyNotificationsQueryHandler
implements IQueryHandler<GetMyNotificationsQuery, MyNotifications>
{
constructor(private readonly notifications: NotificationsRepository) {}
async execute(query: GetMyNotificationsQuery): Promise<MyNotifications> {
const [items, unread] = await Promise.all([
this.notifications.findForUser(query.userId, {
unreadOnly: query.unreadOnly,
limit: Math.min(Math.max(query.limit, 1), 100),
}),
this.notifications.countUnread(query.userId),
]);
return { items, unread };
}
}
@QueryHandler(GetMyNotificationSettingsQuery)
export class GetMyNotificationSettingsQueryHandler
implements
IQueryHandler<GetMyNotificationSettingsQuery, MyNotificationSettings>
{
constructor(
private readonly notifications: NotificationsRepository,
private readonly mailer: NotificationMailer,
) {}
async execute(
query: GetMyNotificationSettingsQuery,
): Promise<MyNotificationSettings> {
const settings = await this.notifications.settings(query.userId);
return {
email: this.mailer.enabled && settings.email,
emailAvailable: this.mailer.enabled,
};
}
}

View File

@@ -0,0 +1,176 @@
import {
CommentAddedEvent,
IssueAssignedEvent,
IssueStatusChangedEvent,
NotificationKind,
NotificationReadModel,
NotificationSettingsReadModel,
ProjectMemberAddedEvent,
ProjectRole,
StatusCategory,
TrackerEvent,
} from '@task-stream/api/tracker/domain';
import { FakeUserDirectory } from '../../testing/in-memory';
import { ChangeFeed } from '../ports/change-feed';
import { FindIssuesRepository } from '../ports/find-issues.repository';
import { FindProjectsRepository } from '../ports/find-projects.repository';
import { NotificationMailer } from '../ports/notification-mailer';
import { NotificationsRepository } from '../ports/notifications.repository';
import { NotificationsEventHandler } from './notifications.event-handler';
import { Notifier } from './notifier';
class InMemoryNotifications extends NotificationsRepository {
readonly all: NotificationReadModel[] = [];
readonly emailOn = new Set<string>();
async insert(n: NotificationReadModel) {
if (this.all.some((x) => x.id === n.id)) return false;
this.all.push(n);
return true;
}
async findForUser(userId: string) {
return this.all.filter((n) => n.userId === userId);
}
async countUnread(userId: string) {
return this.all.filter((n) => n.userId === userId && !n.readAt).length;
}
async markRead() {
return;
}
async settings(userId: string): Promise<NotificationSettingsReadModel> {
return { userId, email: this.emailOn.has(userId) };
}
async saveSettings() {
return;
}
}
const at = <T extends TrackerEvent>(event: T, version = 2): T =>
Object.assign(event, {
aggregateVersion: version,
occurredAt: '2026-10-08T10:00:00.000Z',
});
function setup(mailEnabled = false) {
const repo = new InMemoryNotifications();
const published: NotificationReadModel[] = [];
const mailed: string[] = [];
const users = new FakeUserDirectory().add('ana').add('bob').add('qa');
const issue = {
id: 'i-1',
key: 'WEB-1',
title: 'Login roto',
reporter: { id: 'qa', name: 'QA' },
assignee: { id: 'bob', name: 'Bob' },
};
const notifier = new Notifier(
repo,
{
publishNotification: async (n: NotificationReadModel) => {
published.push(n);
},
} as unknown as ChangeFeed,
{
enabled: mailEnabled,
send: async (to: { email: string }) => {
mailed.push(to.email);
},
} as unknown as NotificationMailer,
users,
);
const handler = new NotificationsEventHandler(
notifier,
{ findById: async () => issue } as unknown as FindIssuesRepository,
{
findById: async () => ({ id: 'p-1', key: 'WEB', name: 'Web' }),
} as unknown as FindProjectsRepository,
users,
);
return { repo, published, mailed, handler };
}
describe('notifications', () => {
it('tells whoever a ticket is assigned to, not whoever assigned it', async () => {
const { handler, repo, published } = setup();
await handler.handle(
at(new IssueAssignedEvent('ana', 'i-1', 'p-1', 'bob', null)),
);
await handler.handle(
at(new IssueAssignedEvent('ana', 'i-1', 'p-1', 'ana', null), 3),
);
expect(repo.all.map((n) => [n.userId, n.kind])).toEqual([
['bob', NotificationKind.ASSIGNED],
]);
expect(published).toHaveLength(1);
expect(repo.all[0].summary).toContain('WEB-1');
});
it('marks a ticket sent back from QA as such, for whoever works on it and reported it', async () => {
const { handler, repo } = setup();
await handler.handle(
at(
new IssueStatusChangedEvent(
'ana',
'i-1',
'p-1',
'IN_QA',
'QA_FAILED',
'Falla en Safari',
{
fromName: 'En QA',
toName: 'QA fallido',
fromCategory: StatusCategory.QA,
toCategory: StatusCategory.IN_PROGRESS,
},
),
),
);
expect(repo.all.map((n) => [n.userId, n.kind])).toEqual([
['bob', NotificationKind.QA_FAILED],
['qa', NotificationKind.QA_FAILED],
]);
expect(repo.all[0].summary).toContain('Falla en Safari');
});
it('tells about a comment everyone but its author, once even when redelivered', async () => {
const { handler, repo } = setup();
const comment = at(
new CommentAddedEvent('bob', {
id: 'c-1',
issueId: 'i-1',
projectId: 'p-1',
authorId: 'bob',
body: 'Ya está arreglado',
deleted: false,
createdAt: '2026-10-08T10:00:00.000Z',
editedAt: null,
}),
1,
);
await handler.handle(comment);
await handler.handle(comment);
expect(repo.all.map((n) => n.userId)).toEqual(['qa']);
});
it('copies by e-mail only to whoever asked for it, when mail is configured', async () => {
const { handler, repo, mailed } = setup(true);
repo.emailOn.add('bob');
await handler.handle(
at(new ProjectMemberAddedEvent('ana', 'p-1', 'bob', ProjectRole.MEMBER)),
);
await handler.handle(
at(
new ProjectMemberAddedEvent('ana', 'p-1', 'qa', ProjectRole.MEMBER),
3,
),
);
expect(mailed).toEqual(['bob@example.org']);
expect(repo.all.map((n) => n.kind)).toEqual([
NotificationKind.ADDED_TO_PROJECT,
NotificationKind.ADDED_TO_PROJECT,
]);
});
});

View File

@@ -0,0 +1,110 @@
import { Injectable, Logger } from '@nestjs/common';
import {
NotificationKind,
NotificationReadModel,
SYSTEM_ACTOR_ID,
SYSTEM_ACTOR_NAME,
TrackerEvent,
} from '@task-stream/api/tracker/domain';
import { ChangeFeed } from '../ports/change-feed';
import { NotificationMailer } from '../ports/notification-mailer';
import { NotificationsRepository } from '../ports/notifications.repository';
import { UserDirectory } from '../ports/user-directory';
import { versionOf } from '../event-handlers/read-model-notifier';
export interface NotificationDraft {
userId: string;
kind: NotificationKind;
summary: string;
}
export interface NotificationContext {
projectId: string;
projectKey: string;
issueId: string | null;
issueKey: string | null;
}
/**
* Delivers notifications: stored for the recipient, pushed live, and copied
* by e-mail to whoever asked for it. Nobody is told about what they did
* themselves, and each person is told once per event — the first reason wins.
*/
@Injectable()
export class Notifier {
private readonly logger = new Logger(Notifier.name);
constructor(
private readonly notifications: NotificationsRepository,
private readonly changeFeed: ChangeFeed,
private readonly mailer: NotificationMailer,
private readonly users: UserDirectory,
) {}
async deliver(
event: TrackerEvent,
context: NotificationContext,
drafts: NotificationDraft[],
): Promise<void> {
const seen = new Set<string>([event.actorId]);
const wanted = drafts.filter((draft) => {
if (seen.has(draft.userId)) return false;
seen.add(draft.userId);
return true;
});
if (!wanted.length) return;
const refs = await this.users.refs([event.actorId]);
const actor =
event.actorId === SYSTEM_ACTOR_ID
? { id: SYSTEM_ACTOR_ID, name: SYSTEM_ACTOR_NAME }
: refs.of(event.actorId);
const delivered: NotificationReadModel[] = [];
for (const draft of wanted) {
const notification: NotificationReadModel = {
id: `${event.aggregateId}:${versionOf(event)}:${draft.userId}`,
userId: draft.userId,
kind: draft.kind,
...context,
actor,
summary: draft.summary,
readAt: null,
createdAt: event.occurredAt,
};
// Delivered again (at-least-once): already told, nothing to repeat.
if (!(await this.notifications.insert(notification))) continue;
await this.changeFeed.publishNotification(notification);
delivered.push(notification);
}
if (this.mailer.enabled && delivered.length) {
await this.email(delivered);
}
}
/** A failed e-mail never blocks the event stream: it is logged and skipped. */
private async email(notifications: NotificationReadModel[]): Promise<void> {
const contacts = new Map(
(await this.users.contacts(notifications.map((n) => n.userId))).map(
(c) => [c.id, c],
),
);
for (const notification of notifications) {
const contact = contacts.get(notification.userId);
if (!contact?.email) continue;
if (!(await this.notifications.settings(notification.userId)).email)
continue;
try {
await this.mailer.send(
{ name: contact.name, email: contact.email },
notification,
);
} catch (error) {
this.logger.warn(
`E-mail to ${notification.userId} failed: ${(error as Error).message}`,
);
}
}
}
}

View File

@@ -1,6 +1,7 @@
import {
ActivityReadModel,
CommentReadModel,
NotificationReadModel,
IssueReadModel,
ProjectReadModel,
ReleaseReadModel,
@@ -48,6 +49,10 @@ export abstract class ChangeFeed {
notifyUserIds?: string[],
): Promise<void>;
abstract publishNotification(
notification: NotificationReadModel,
): Promise<void>;
abstract issues(projectId: string): AsyncIterableIterator<IssueChange>;
abstract comments(issueId: string): AsyncIterableIterator<CommentChange>;
abstract activity(
@@ -56,4 +61,7 @@ export abstract class ChangeFeed {
abstract releases(projectId: string): AsyncIterableIterator<ReleaseChange>;
abstract project(projectId: string): AsyncIterableIterator<ProjectChange>;
abstract userProjects(userId: string): AsyncIterableIterator<ProjectChange>;
abstract notifications(
userId: string,
): AsyncIterableIterator<NotificationReadModel>;
}

View File

@@ -18,3 +18,5 @@ export * from './upsert-materialized-timeline.repository';
export * from './find-timeline.repository';
export * from './tracker-streams.repository';
export * from './attachment-storage';
export * from './notifications.repository';
export * from './notification-mailer';

View File

@@ -0,0 +1,16 @@
import { NotificationReadModel } from '@task-stream/api/tracker/domain';
export interface MailRecipient {
name: string;
email: string;
}
/** E-mail copies of notifications, for whoever asked for them. */
export abstract class NotificationMailer {
/** False when no mail server is configured: nobody is offered e-mail. */
abstract get enabled(): boolean;
abstract send(
to: MailRecipient,
notification: NotificationReadModel,
): Promise<void>;
}

View File

@@ -0,0 +1,23 @@
import {
NotificationReadModel,
NotificationSettingsReadModel,
} from '@task-stream/api/tracker/domain';
/**
* Notifications are per-person state, not part of any aggregate: they are
* derived from events and read and dismissed by their recipient alone.
*/
export abstract class NotificationsRepository {
/** False when it was already there (the event was delivered again). */
abstract insert(notification: NotificationReadModel): Promise<boolean>;
/** Newest first. */
abstract findForUser(
userId: string,
options: { unreadOnly: boolean; limit: number },
): Promise<NotificationReadModel[]>;
abstract countUnread(userId: string): Promise<number>;
/** `ids` null: all of them. */
abstract markRead(userId: string, ids: string[] | null): Promise<void>;
abstract settings(userId: string): Promise<NotificationSettingsReadModel>;
abstract saveSettings(settings: NotificationSettingsReadModel): Promise<void>;
}

View File

@@ -6,6 +6,12 @@ export interface DirectoryUser {
isActive: boolean;
}
export interface UserContact {
id: string;
name: string;
email: string | null;
}
/** Outbound port to the identity context. */
export abstract class UserDirectory {
/**
@@ -17,6 +23,9 @@ export abstract class UserDirectory {
accessToken?: string,
): Promise<DirectoryUser | null>;
/** Names and e-mail addresses, for the users this application knows. */
abstract contacts(userIds: string[]): Promise<UserContact[]>;
/** Resolves user ids to {id, name} for read models, in one lookup. */
abstract refs(userIds: (string | null | undefined)[]): Promise<UserRefs>;
}

View File

@@ -16,6 +16,27 @@ import { ReadAccess } from './queries/read-access';
import { ReleasesService } from './releases.service';
import { SAGAS } from './sagas/sagas';
import { WorkflowsService } from './workflows.service';
import { NotificationsService } from './notifications.service';
import { Notifier } from './notifications/notifier';
import { NotificationsEventHandler } from './notifications/notifications.event-handler';
import {
GetMyNotificationSettingsQueryHandler,
GetMyNotificationsQueryHandler,
} from './notifications/notifications.queries';
import {
MarkNotificationsReadCommandHandler,
UpdateNotificationSettingsCommandHandler,
} from './notifications/notifications.commands';
/** Notifications: derived from events, kept and dismissed per person. */
const NOTIFICATIONS = [
Notifier,
NotificationsEventHandler,
GetMyNotificationsQueryHandler,
GetMyNotificationSettingsQueryHandler,
MarkNotificationsReadCommandHandler,
UpdateNotificationSettingsCommandHandler,
];
const SERVICES = [
ProjectsService,
@@ -24,6 +45,7 @@ const SERVICES = [
ReleasesService,
ChangesService,
WorkflowsService,
NotificationsService,
];
/**
@@ -43,6 +65,7 @@ const SERVICES = [
...QUERY_HANDLERS,
...EVENT_HANDLERS,
...SAGAS,
...NOTIFICATIONS,
],
exports: SERVICES,
})