#151: Mention-Benachrichtigungen über die Glocke
Neuer Notification-Typ mentioned; abgeleitete Tabelle page_mentions (Migration), vom Collab-Persist transaktional neu geschrieben — der Diff gegen den Vorzustand wird als pg_notify (page_mentions_changed) emittiert, nur NEU Erwähnte lösen aus (kein Spam bei Folge-Saves). Der api-Listener (erweitert um den zweiten Kanal) ruft NotificationsService.fanoutMentions: Zustellung nur nach canAccessPage-Recheck, die Autoren (pending contributors) benachrich- tigen sich nie selbst; Payload wie gehabt mit Actor-Namen. API-seitig erzeugte Seiten seeden page_mentions aus deriveContent. Glocken-Text de+en; DB-Test (Leser ja / Outsider nein / Autor nein); kompletter Loop live verifiziert (Tippen → Persist → NOTIFY → Notification). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0155v2aT8AG1kZDQEZiCLBWC
This commit is contained in:
parent
e164370691
commit
f9d23bf8d0
@ -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;
|
||||
@ -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")
|
||||
|
||||
111
apps/api/src/notifications/mentions.e2e.db.test.ts
Normal file
111
apps/api/src/notifications/mentions.e2e.db.test.ts
Normal file
@ -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);
|
||||
});
|
||||
});
|
||||
@ -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<void> {
|
||||
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<NotificationListView> {
|
||||
const where = { userId: user.id };
|
||||
const [total, unreadCount] = [
|
||||
|
||||
@ -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<void> {
|
||||
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');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -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),
|
||||
};
|
||||
}
|
||||
|
||||
@ -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 };
|
||||
|
||||
@ -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),
|
||||
};
|
||||
}
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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 {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user