diff --git a/apps/api/prisma/migrations/20260719232202_page_mentions/migration.sql b/apps/api/prisma/migrations/20260719232202_page_mentions/migration.sql new file mode 100644 index 0000000..45aa2e4 --- /dev/null +++ b/apps/api/prisma/migrations/20260719232202_page_mentions/migration.sql @@ -0,0 +1,16 @@ +-- CreateTable +CREATE TABLE "page_mentions" ( + "page_id" TEXT NOT NULL, + "user_id" TEXT NOT NULL, + + CONSTRAINT "page_mentions_pkey" PRIMARY KEY ("page_id","user_id") +); + +-- CreateIndex +CREATE INDEX "page_mentions_user_id_idx" ON "page_mentions"("user_id"); + +-- AddForeignKey +ALTER TABLE "page_mentions" ADD CONSTRAINT "page_mentions_page_id_fkey" FOREIGN KEY ("page_id") REFERENCES "pages"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "page_mentions" ADD CONSTRAINT "page_mentions_user_id_fkey" FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index a3b88e4..227080c 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -52,6 +52,7 @@ model User { authTokens AuthToken[] apiTokens ApiToken[] feedTokens FeedToken[] + mentionRows PageMention[] ponds Pond[] pages Page[] attachments Attachment[] @@ -282,6 +283,7 @@ model Page { attachments Attachment[] versions PageVersion[] pendingContributors PagePendingContributor[] + mentionRows PageMention[] labels PageLabel[] outgoingLinks PageLink[] @relation("outgoingLinks") comments Comment[] @@ -349,6 +351,21 @@ model PageVersion { /// Collab flushes the current session's contributors here (deduplicated by the /// composite key); version creation on either side reads and clears it in the /// same transaction as writing the snapshot. Cascades on page purge (ADR 0013). +/// Derived mention index (issue #151): one row per user currently +/// mentioned in the page's document. Rewritten on every collab persist; +/// the diff against the previous rows drives the `mentioned` notifications. +model PageMention { + pageId String @map("page_id") + userId String @map("user_id") + + page Page @relation(fields: [pageId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@id([pageId, userId]) + @@index([userId]) + @@map("page_mentions") +} + model PagePendingContributor { pageId String @map("page_id") userId String @map("user_id") diff --git a/apps/api/src/notifications/mentions.e2e.db.test.ts b/apps/api/src/notifications/mentions.e2e.db.test.ts new file mode 100644 index 0000000..74d62ad --- /dev/null +++ b/apps/api/src/notifications/mentions.e2e.db.test.ts @@ -0,0 +1,111 @@ +import { INestApplication } from '@nestjs/common'; +import { PrismaClient } from '@prisma/client'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import { createTestApp } from '../testing/test-app'; +import { createTestPrisma, hasTestDb, uniqueSuffix } from '../testing/test-db'; +import { NotificationsService } from './notifications.service'; + +/** + * Mention notifications (issue #151): newly mentioned users get a + * `mentioned` notification — but only with read access (no leak), and the + * mention's author (a pending contributor) never notifies themselves. + */ +describe.skipIf(!hasTestDb)('mention notifications (e2e, issue #151)', () => { + let app: INestApplication; + let prisma: PrismaClient; + const suffix = uniqueSuffix(); + + let authorId: string; + let readerId: string; + let outsiderId: string; + let pageId: string; + + beforeAll(async () => { + prisma = createTestPrisma(); + app = await createTestApp(); + + const mkUser = async (handle: string) => + ( + await prisma.user.create({ + data: { + username: `mention-${handle}-${suffix}`, + email: `mention-${handle}-${suffix}@example.test`, + displayName: `Mention ${handle}`, + status: 'ACTIVE', + }, + }) + ).id; + authorId = await mkUser('author'); + readerId = await mkUser('reader'); + outsiderId = await mkUser('outsider'); + + const pond = await prisma.pond.create({ + data: { + slug: `mention-pond-${suffix}`, + name: 'Mention Pond', + type: 'SHARED', + ownerId: authorId, + }, + }); + const page = await prisma.page.create({ + data: { + pondId: pond.id, + slug: `notes-${suffix}`, + title: 'Notes', + createdBy: authorId, + sortKey: 'a0', + ydocState: new Uint8Array(), + }, + }); + pageId = page.id; + for (const [userId, role] of [ + [authorId, 'EDITOR'], + [readerId, 'READER'], + ] as const) { + await prisma.roleGrant.create({ + data: { + pondId: pond.id, + subjectType: 'USER', + subjectId: userId, + role, + scopeType: 'POND', + scopeId: null, + effect: 'ALLOW', + createdBy: authorId, + }, + }); + } + // The author edited last — pending contributor, i.e. the acting user. + await prisma.pagePendingContributor.create({ data: { pageId, userId: authorId } }); + }); + + afterAll(async () => { + await prisma.notification.deleteMany({ + where: { userId: { in: [authorId, readerId, outsiderId] } }, + }); + await prisma.pagePendingContributor.deleteMany({ where: { pageId } }); + await prisma.roleGrant.deleteMany({ where: { pond: { ownerId: authorId } } }); + await prisma.page.deleteMany({ where: { pond: { ownerId: authorId } } }); + await prisma.pond.deleteMany({ where: { ownerId: authorId } }); + await prisma.user.deleteMany({ where: { id: { in: [authorId, readerId, outsiderId] } } }); + await prisma.$disconnect(); + await app.close(); + }); + + it('notifies mentioned readers, skips outsiders and the author', async () => { + await app.get(NotificationsService).fanoutMentions(pageId, [readerId, outsiderId, authorId]); + + const readerRows = await prisma.notification.findMany({ where: { userId: readerId } }); + expect(readerRows).toHaveLength(1); + expect(readerRows[0]!.type).toBe('mentioned'); + expect(readerRows[0]!.payload).toMatchObject({ + pageTitle: 'Notes', + actorNames: ['Mention author'], + }); + + // No read access → nothing; the author never notifies themselves. + expect(await prisma.notification.count({ where: { userId: outsiderId } })).toBe(0); + expect(await prisma.notification.count({ where: { userId: authorId } })).toBe(0); + }); +}); diff --git a/apps/api/src/notifications/notifications.service.ts b/apps/api/src/notifications/notifications.service.ts index aac1968..fc46069 100644 --- a/apps/api/src/notifications/notifications.service.ts +++ b/apps/api/src/notifications/notifications.service.ts @@ -94,6 +94,64 @@ export class NotificationsService { } } + /** + * Notifies newly mentioned users (issue #151). Independent of the watch + * table — a mention addresses the person directly — but with the same + * delivery-time read-permission re-check: whoever may not read the page + * gets nothing (no leak). The mention's authors (the page's pending + * contributors at persist time) never notify themselves. + */ + async fanoutMentions(pageId: string, mentionedUserIds: string[]): Promise { + try { + const page = await this.prisma.page.findFirst({ + where: { id: pageId, deletedAt: null }, + select: { id: true, pondId: true, title: true, slug: true }, + }); + if (!page) return; + const pond = await this.prisma.pond.findFirst({ + where: { id: page.pondId, deletedAt: null }, + select: { name: true, slug: true }, + }); + if (!pond) return; + + const contributors = await this.prisma.pagePendingContributor.findMany({ + where: { pageId }, + select: { userId: true }, + }); + const actorIds = contributors.map((row) => row.userId); + const actors = actorIds.length + ? await this.prisma.user.findMany({ + where: { id: { in: actorIds } }, + select: { displayName: true }, + }) + : []; + const payload: NotificationPayload = { + pageId: page.id, + pageTitle: page.title, + pageSlug: page.slug, + pondSlug: pond.slug, + pondName: pond.name, + actorNames: actors.slice(0, 3).map((actor) => actor.displayName), + }; + + const targets = await this.prisma.user.findMany({ + where: { id: { in: mentionedUserIds.filter((id) => !actorIds.includes(id)) } }, + }); + for (const target of targets) { + if (!(await this.permissions.canAccessPage(target, page, 'read'))) continue; + await this.prisma.notification.create({ + data: { + userId: target.id, + type: 'mentioned', + payload: payload as unknown as Prisma.InputJsonObject, + }, + }); + } + } catch (error) { + this.logger.warn({ pageId, err: error }, 'mention fan-out failed'); + } + } + async list(user: User, page: number): Promise { const where = { userId: user.id }; const [total, unreadCount] = [ diff --git a/apps/api/src/notifications/version-event-listener.service.ts b/apps/api/src/notifications/version-event-listener.service.ts index 9c6f805..c4d4b79 100644 --- a/apps/api/src/notifications/version-event-listener.service.ts +++ b/apps/api/src/notifications/version-event-listener.service.ts @@ -1,5 +1,10 @@ import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; -import { PAGE_VERSION_CREATED_CHANNEL, type PageVersionCreatedEvent } from '@dorfteich/shared'; +import { + PAGE_MENTIONS_CHANGED_CHANNEL, + PAGE_VERSION_CREATED_CHANNEL, + type PageMentionsChangedEvent, + type PageVersionCreatedEvent, +} from '@dorfteich/shared'; import { PinoLogger } from 'nestjs-pino'; import { Client } from 'pg'; @@ -49,12 +54,17 @@ export class VersionEventListener implements OnModuleInit, OnModuleDestroy { client.on('error', () => this.scheduleReconnect()); client.on('end', () => this.scheduleReconnect()); client.on('notification', (message) => { - if (message.channel !== PAGE_VERSION_CREATED_CHANNEL || !message.payload) return; - void this.handle(message.payload); + if (!message.payload) return; + if (message.channel === PAGE_VERSION_CREATED_CHANNEL) void this.handle(message.payload); + // Newly added mentions from a collab persist (issue #151). + if (message.channel === PAGE_MENTIONS_CHANGED_CHANNEL) { + void this.handleMentions(message.payload); + } }); try { await client.connect(); await client.query(`LISTEN ${PAGE_VERSION_CREATED_CHANNEL}`); + await client.query(`LISTEN ${PAGE_MENTIONS_CHANGED_CHANNEL}`); this.logger.info({}, 'listening for collab version events'); } catch (error) { this.logger.warn({ err: error }, 'version-event listener could not connect'); @@ -79,4 +89,16 @@ export class VersionEventListener implements OnModuleInit, OnModuleDestroy { this.logger.warn({ err: error }, 'ignoring malformed version event'); } } + + private async handleMentions(payload: string): Promise { + try { + const event = JSON.parse(payload) as PageMentionsChangedEvent; + if (!event.pageId || !Array.isArray(event.addedUserIds) || event.addedUserIds.length === 0) { + return; + } + await this.notifications.fanoutMentions(event.pageId, event.addedUserIds); + } catch (error) { + this.logger.warn({ err: error }, 'ignoring malformed mentions event'); + } + } } diff --git a/apps/api/src/pages/yjs-content.ts b/apps/api/src/pages/yjs-content.ts index c0dc742..a216d27 100644 --- a/apps/api/src/pages/yjs-content.ts +++ b/apps/api/src/pages/yjs-content.ts @@ -5,6 +5,7 @@ import { editorSchema, extractOutline, OutlineEntry, + extractMentionUserIds, extractWikilinkSlugs, } from '@dorfteich/shared'; import { Node } from 'prosemirror-model'; @@ -93,6 +94,9 @@ export interface DerivedPageContent { * api (imports, phantom-create) seed their `page_links` rows from this — * collab, the content writer, rewrites them on every later save. */ wikilinkSlugs: string[]; + /** Resolved user ids of every `@mention` (issue #151) — api-created pages + * seed their `page_mentions` rows from this; collab rewrites on save. */ + mentionUserIds: string[]; } function imageFileIdsOf(doc: Node): string[] { @@ -119,5 +123,6 @@ export function deriveContent(state: Uint8Array): DerivedPageContent { outline: extractOutline(doc), imageFileIds: imageFileIdsOf(doc), wikilinkSlugs: extractWikilinkSlugs(doc), + mentionUserIds: extractMentionUserIds(doc), }; } diff --git a/apps/collab/src/persistence.ts b/apps/collab/src/persistence.ts index b220683..f8bca38 100644 --- a/apps/collab/src/persistence.ts +++ b/apps/collab/src/persistence.ts @@ -1,4 +1,8 @@ -import { MAX_PAGE_DOCUMENT_BYTES, normalizeForSearch } from '@dorfteich/shared'; +import { + MAX_PAGE_DOCUMENT_BYTES, + PAGE_MENTIONS_CHANGED_CHANNEL, + normalizeForSearch, +} from '@dorfteich/shared'; import type { Pool } from 'pg'; import * as Y from 'yjs'; @@ -203,6 +207,31 @@ export class PostgresPagePersistence implements PagePersistence { ); } + // Rewrite the mention index (issue #151); newly added user ids become a + // NOTIFY the api turns into `mentioned` notifications. Inside the + // transaction on purpose — pg_notify only fires on COMMIT. + const previousMentions = await client.query<{ user_id: string }>( + 'SELECT user_id FROM page_mentions WHERE page_id = $1', + [pageId], + ); + await client.query('DELETE FROM page_mentions WHERE page_id = $1', [pageId]); + if (derived.mentionUserIds.length > 0) { + await client.query( + `INSERT INTO page_mentions (page_id, user_id) + SELECT $1, u.id FROM unnest($2::text[]) AS m(user_id) + JOIN users u ON u.id = m.user_id`, + [pageId, derived.mentionUserIds], + ); + } + const known = new Set(previousMentions.rows.map((row) => row.user_id)); + const added = derived.mentionUserIds.filter((id) => !known.has(id)); + if (added.length > 0) { + await client.query('SELECT pg_notify($1, $2)', [ + PAGE_MENTIONS_CHANGED_CHANNEL, + JSON.stringify({ pageId, addedUserIds: added }), + ]); + } + await client.query('COMMIT'); this.lastStoredVector.set(pageId, nextVector); return { outcome: 'stored', bytes: full.byteLength, durationMs: durationOf(), merged }; diff --git a/apps/collab/src/yjs-content.ts b/apps/collab/src/yjs-content.ts index d19c353..b8468e5 100644 --- a/apps/collab/src/yjs-content.ts +++ b/apps/collab/src/yjs-content.ts @@ -4,6 +4,7 @@ import { docToPlainText, editorSchema, extractOutline, + extractMentionUserIds, extractWikilinkSlugs, type OutlineEntry, } from '@dorfteich/shared'; @@ -44,6 +45,9 @@ export interface DerivedPageContent { /** Distinct target slugs of every `[[wikilink]]`, for the `page_links` * index (issue #47). */ wikilinkSlugs: string[]; + /** Distinct resolved user ids of every `@mention`, for the + * `page_mentions` index and the mention notifications (issue #151). */ + mentionUserIds: string[]; } function imageFileIdsOf(doc: Node): string[] { @@ -70,5 +74,6 @@ export function deriveContentFromDoc(ydoc: Y.Doc): DerivedPageContent { outline: extractOutline(doc), imageFileIds: imageFileIdsOf(doc), wikilinkSlugs: extractWikilinkSlugs(doc), + mentionUserIds: extractMentionUserIds(doc), }; } diff --git a/packages/shared/i18n/de/notifications.json b/packages/shared/i18n/de/notifications.json index 7459b8b..ef9614c 100644 --- a/packages/shared/i18n/de/notifications.json +++ b/packages/shared/i18n/de/notifications.json @@ -5,7 +5,8 @@ "someone": "Jemand", "types": { "page_changed": "{{actor}} hat „{{page}}“ in {{pond}} geändert", - "comment_added": "{{actor}} hat „{{page}}“ in {{pond}} kommentiert" + "comment_added": "{{actor}} hat „{{page}}“ in {{pond}} kommentiert", + "mentioned": "{{actor}} hat dich auf „{{page}}“ in {{pond}} erwähnt" }, "digest": { "label": "E-Mail-Digest", diff --git a/packages/shared/i18n/en/notifications.json b/packages/shared/i18n/en/notifications.json index 201bf7d..88799e9 100644 --- a/packages/shared/i18n/en/notifications.json +++ b/packages/shared/i18n/en/notifications.json @@ -5,7 +5,8 @@ "someone": "Someone", "types": { "page_changed": "{{actor}} changed “{{page}}” in {{pond}}", - "comment_added": "{{actor}} commented on “{{page}}” in {{pond}}" + "comment_added": "{{actor}} commented on “{{page}}” in {{pond}}", + "mentioned": "{{actor}} mentioned you on “{{page}}” in {{pond}}" }, "digest": { "label": "E-mail digest", diff --git a/packages/shared/src/collab-token.ts b/packages/shared/src/collab-token.ts index a81fd8b..2cf0578 100644 --- a/packages/shared/src/collab-token.ts +++ b/packages/shared/src/collab-token.ts @@ -56,6 +56,21 @@ export interface PageVersionCreatedEvent { */ export const TASK_TOGGLE_CHANNEL = 'task_toggle'; +/** + * PostgreSQL `NOTIFY` channel over which the collab server announces that a + * persist added new user mentions to a page (issue #151). The api listens + * and creates the `mentioned` notifications — permission-checked there. + * Payload is a JSON {@link PageMentionsChangedEvent}. + */ +export const PAGE_MENTIONS_CHANGED_CHANNEL = 'page_mentions_changed'; + +/** JSON payload carried on {@link PAGE_MENTIONS_CHANGED_CHANNEL}. */ +export interface PageMentionsChangedEvent { + pageId: string; + /** Users newly mentioned by this persist (diff against the stored rows). */ + addedUserIds: string[]; +} + /** JSON payload carried on {@link TASK_TOGGLE_CHANNEL}. */ export interface TaskToggleRequest { pageId: string; diff --git a/packages/shared/src/notifications.ts b/packages/shared/src/notifications.ts index 5dc050f..e097204 100644 --- a/packages/shared/src/notifications.ts +++ b/packages/shared/src/notifications.ts @@ -7,7 +7,7 @@ import { z } from 'zod'; * permission at delivery time. */ -export const NOTIFICATION_TYPES = ['page_changed', 'comment_added'] as const; +export const NOTIFICATION_TYPES = ['page_changed', 'comment_added', 'mentioned'] as const; export type NotificationType = (typeof NOTIFICATION_TYPES)[number]; export interface NotificationPayload {