From 7b61857fb8b70ce1df583b4d6371a56675d476a8 Mon Sep 17 00:00:00 2001 From: flfeders Date: Thu, 15 Oct 2026 17:50:38 +0200 Subject: [PATCH] feat(notifications): include federated chat messages and invitations --- .../0083_collaboration_notifications.sql | 7 + backend/db/migrations/meta/_journal.json | 7 + backend/db/schema/notifications_items.ts | 5 +- .../src/modules/collaboration/federation.ts | 6 + .../modules/collaboration/notifications.ts | 139 ++++++++++++ backend/src/modules/collaboration/runtime.ts | 8 +- backend/src/modules/collaboration/service.ts | 4 + backend/src/modules/notification.service.ts | 32 ++- .../tests/collaborationNotifications.test.ts | 201 ++++++++++++++++++ docs/chat-foederation-konzept.md | 9 +- docs/notification-preferences.md | 11 +- 11 files changed, 413 insertions(+), 16 deletions(-) create mode 100644 backend/db/migrations/0083_collaboration_notifications.sql create mode 100644 backend/src/modules/collaboration/notifications.ts create mode 100644 backend/tests/collaborationNotifications.test.ts diff --git a/backend/db/migrations/0083_collaboration_notifications.sql b/backend/db/migrations/0083_collaboration_notifications.sql new file mode 100644 index 00000000..5c422446 --- /dev/null +++ b/backend/db/migrations/0083_collaboration_notifications.sql @@ -0,0 +1,7 @@ +INSERT INTO notifications_event_types (event_key, display_name, description, category, allowed_channels) VALUES +('collaboration.message.new', 'Zusammenarbeit: Chatnachrichten', 'Neue Nachrichten in freigegebenen Unterräumen und föderierten Chats.', 'collaboration', '["inapp","email","push"]'), +('collaboration.room.invited', 'Zusammenarbeit: Einladungen', 'Persönliche Einladungen in freigegebene Unterräume und föderierte Chats.', 'collaboration', '["inapp","email","push"]') +ON CONFLICT (event_key) DO NOTHING; +--> statement-breakpoint +CREATE INDEX notifications_collaboration_queued ON notifications_items(created_at) +WHERE status='queued' AND event_type IN ('collaboration.message.new','collaboration.room.invited'); diff --git a/backend/db/migrations/meta/_journal.json b/backend/db/migrations/meta/_journal.json index 262a7a6b..81c77fba 100644 --- a/backend/db/migrations/meta/_journal.json +++ b/backend/db/migrations/meta/_journal.json @@ -561,6 +561,13 @@ "when": 1792022400000, "tag": "0082_archive_users", "breakpoints": true + }, + { + "idx": 80, + "version": "7", + "when": 1792022400001, + "tag": "0083_collaboration_notifications", + "breakpoints": true } ] } diff --git a/backend/db/schema/notifications_items.ts b/backend/db/schema/notifications_items.ts index d6c6cb3c..b69f0bed 100644 --- a/backend/db/schema/notifications_items.ts +++ b/backend/db/schema/notifications_items.ts @@ -5,7 +5,9 @@ import { text, jsonb, timestamp, + index, } from "drizzle-orm/pg-core" +import { sql } from "drizzle-orm" import { tenants } from "./tenants" import { authUsers } from "./auth_users" @@ -48,7 +50,8 @@ export const notificationsItems = pgTable("notifications_items", { sentAt: timestamp("sent_at", { withTimezone: true }), readAt: timestamp("read_at", { withTimezone: true }), -}) +}, t => ({ collaborationQueued: index("notifications_collaboration_queued").on(t.createdAt) + .where(sql`${t.status}='queued' AND ${t.eventType} IN ('collaboration.message.new','collaboration.room.invited')`) })) export type NotificationItem = typeof notificationsItems.$inferSelect export type NewNotificationItem = typeof notificationsItems.$inferInsert diff --git a/backend/src/modules/collaboration/federation.ts b/backend/src/modules/collaboration/federation.ts index 01555b79..c51592ff 100644 --- a/backend/src/modules/collaboration/federation.ts +++ b/backend/src/modules/collaboration/federation.ts @@ -207,6 +207,7 @@ export class FederationService { if (updated) { await db.execute(sql`DELETE FROM collaboration_remote_messages WHERE binding_id=${updated.id}`) await db.execute(sql`UPDATE collaboration_outbox SET status='failed',last_error='Neue Teilnahme erforderlich' WHERE binding_id=${updated.id} AND status='pending'`) + await this.service.options.notifications?.remoteInvitation(db, principal, updated.id, invitationId, snapshot.name) } }) return { received: true } @@ -322,6 +323,11 @@ export class FederationService { for (const message of projected.filter(m => m.seq >= from)) await db.execute(sql`INSERT INTO collaboration_remote_messages(binding_id,id,seq,payload) VALUES (${binding.id},${message.id},${message.seq},${JSON.stringify(message)}::jsonb) ON CONFLICT(binding_id,id) DO UPDATE SET payload=excluded.payload`) + if (snapshot) { + const unreadAfter = Math.max(Number(current.sync_seq), Number(current.last_read_seq), snapshot.lastReadSeq) + const incoming = projected.filter(m => !m.own && m.seq >= from && m.seq > unreadAfter) + await this.service.options.notifications?.remoteMessages(db, current, snapshot.name, incoming) + } }) } async memberRequest(ctx: UserContext, id: string, action: string, input: any = {}) { diff --git a/backend/src/modules/collaboration/notifications.ts b/backend/src/modules/collaboration/notifications.ts new file mode 100644 index 00000000..2595e44a --- /dev/null +++ b/backend/src/modules/collaboration/notifications.ts @@ -0,0 +1,139 @@ +import type { FastifyInstance } from "fastify" +import { and, asc, eq, inArray, notInArray, sql } from "drizzle-orm" +import { authUsers, notificationsItems } from "../../../db/schema" +import { NotificationService } from "../notification.service" +import { CollaborationService, rows } from "./service" +import type { FederationService } from "./federation" +import type { Principal } from "./types" + +const MESSAGE = "collaboration.message.new" +const INVITATION = "collaboration.room.invited" + +/** Queue inside the source transaction; external delivery never holds up a chat command. */ +export class CollaborationNotifications { + private working = false + private deferredUntil = new Map() + constructor(readonly server: FastifyInstance, readonly service: CollaborationService, readonly federation: FederationService) {} + + delivery(db: any) { + return new NotificationService({ db, log: this.server.log } as FastifyInstance, async (server, userId) => { + const [user] = await server.db.select({ email: authUsers.email }).from(authUsers).where(eq(authUsers.id, userId)).limit(1) + return user || null + }) + } + + async enqueue(db: any, principal: Principal, name: string, payload: Record, message?: { body: string; senderName: string }) { + if (principal.kind !== "user" || principal.disabled) return + try { await this.service.assertPrincipal(principal, db) } + catch (error: any) { if ([403, 404].includes(error.statusCode)) return; throw error } + await this.delivery(db).enqueue({ + tenantId: Number(principal.tenant_id), userId: principal.user_id!, + eventType: message ? MESSAGE : INVITATION, + title: message ? `Neue Nachricht in ${name}` : "Einladung zur Zusammenarbeit", + message: message ? `${message.senderName}: ${message.body || "Bild gesendet"}` : `Du wurdest in den Chat „${name}“ eingeladen.`, + channels: ["inapp", "email", "push"], + payload: { ...payload, principalId: principal.id, + link: message ? `/communication/chat?room=${encodeURIComponent(`collaboration:${payload.bindingId || payload.roomId}`)}` : "/communication/chat" }, + }) + } + + async localMessage(db: any, room: any, sender: Principal, message: any) { + const recipients = await rows(db, sql`SELECT p.* FROM collaboration_members m JOIN collaboration_principals p ON p.id=m.principal_id + WHERE m.room_id=${room.id} AND m.status='active' AND (m.expires_at IS NULL OR m.expires_at>now()) + AND m.history_from_seq<=${message.seq} AND m.last_read_seq<${message.seq} + AND p.kind='user' AND NOT p.disabled AND p.id<>${sender.id}`) + for (const principal of recipients) await this.enqueue(db, principal, room.name, + { roomId: room.id, messageId: message.id, seq: message.seq }, { body: message.body, senderName: sender.display_name }) + } + + async localInvitation(db: any, room: any, target: Principal, invitationId: string) { + await this.enqueue(db, target, room.name, { roomId: room.id, invitationId }) + } + + async remoteInvitation(db: any, principal: Principal, bindingId: string, invitationId: string, name: string) { + await this.enqueue(db, principal, name, { bindingId, invitationId }) + } + + async remoteMessages(db: any, binding: any, name: string, messages: any[]) { + if (!messages.length) return + const [principal] = await rows(db, sql`SELECT * FROM collaboration_principals WHERE id=${binding.principal_id}`) + if (!principal) return + for (const message of new Map(messages.map(message => [message.id, message])).values()) await this.enqueue(db, principal, name, + { bindingId: binding.id, invitationId: binding.invitation_id, messageId: message.id, seq: message.seq }, message) + } + + async eligible(db: any, item: typeof notificationsItems.$inferSelect): Promise { + const payload = item.payload as any + const [principal] = await rows(db, sql`SELECT * FROM collaboration_principals WHERE id=${payload.principalId}`) + if (!principal || principal.kind !== "user" || principal.user_id !== item.userId || Number(principal.tenant_id) !== item.tenantId) return false + try { + await this.service.assertPrincipal(principal, db) + if (payload.bindingId) { + if (!this.federation.options.enabled) return "defer" + const ctx = { principal, userId: item.userId, tenantId: item.tenantId, manage: false } + const binding = await this.federation.binding(ctx, payload.bindingId) + if (!binding || binding.invitation_id !== payload.invitationId) return false + await this.federation.allowed(item.tenantId, binding.origin, db) + if (item.eventType === INVITATION) return binding.status === "pending" && new Date(binding.invitation.expiresAt).getTime() > Date.now() + if (binding.status !== "active") return false + // Do not send cached content while current access cannot be confirmed by its owner. + const detail = await this.federation.sync(ctx, binding) + if (detail.offline) return "defer" + const current = await this.federation.binding(ctx, binding.id) + return current?.status === "active" && current.invitation_id === payload.invitationId + && payload.seq >= detail.historyFromSeq && (item.channel === "inapp" || payload.seq > Number(current.last_read_seq)) + } + const room = await this.service.roomRow(db, payload.roomId) + if (item.eventType === INVITATION) { + this.service.assertOpen(room, true) + const [invite] = await rows(db, sql`SELECT id FROM collaboration_invitations + WHERE id=${payload.invitationId} AND room_id=${room.id} AND target_principal_id=${principal.id} + AND status='pending' AND expires_at>now() AND (access_expires_at IS NULL OR access_expires_at>now())`) + return Boolean(invite) + } + const member = await this.service.membership(db, room, principal) + return payload.seq >= Number(member.history_from_seq) && (item.channel === "inapp" || payload.seq > Number(member.last_read_seq)) + } catch (error: any) { + if ([401, 403, 404, 409, 410].includes(error.statusCode)) return false + throw error + } + } + + async tick() { + if (this.working) return + this.working = true + try { + for (const [id, until] of this.deferredUntil) if (until <= Date.now()) this.deferredUntil.delete(id) + const pending = await this.server.db.select({ id: notificationsItems.id }).from(notificationsItems) + .where(and(eq(notificationsItems.status, "queued"), inArray(notificationsItems.eventType, [MESSAGE, INVITATION]), + this.deferredUntil.size ? notInArray(notificationsItems.id, [...this.deferredUntil.keys()]) : undefined)) + .orderBy(asc(notificationsItems.createdAt)).limit(20) + for (const { id } of pending) { + try { + await this.server.db.transaction(async db => { + const [item] = await db.select().from(notificationsItems) + .where(and(eq(notificationsItems.id, id), eq(notificationsItems.status, "queued"))) + .for("update", { skipLocked: true }) + if (!item) return + const eligible = await this.eligible(db, item) + if (eligible === "defer") { + this.deferredUntil.set(id, Date.now() + 30000) + return + } + const delivery = this.delivery(db) + const enabled = item.channel === "inapp" || (await delivery.preferences(item.tenantId, item.userId)).events + .some(event => event.eventKey === item.eventType && event.channels[item.channel] === true) + if (!eligible || !enabled) { + await db.delete(notificationsItems).where(eq(notificationsItems.id, item.id)) + return + } + await delivery.deliver(item) + }) + } catch (error) { + this.deferredUntil.set(id, Date.now() + 30000) + this.server.log.error({ err: error, notificationId: id }, "Collaboration notification failed") + } + } + } finally { this.working = false } + } +} diff --git a/backend/src/modules/collaboration/runtime.ts b/backend/src/modules/collaboration/runtime.ts index b7005545..05b1ee11 100644 --- a/backend/src/modules/collaboration/runtime.ts +++ b/backend/src/modules/collaboration/runtime.ts @@ -5,6 +5,7 @@ import { CollaborationService } from "./service" import { FederationService } from "./federation" import { encrypt, decrypt } from "../../utils/crypt" import { collaborationConfig } from "./config" +import { CollaborationNotifications } from "./notifications" export function collaborationRuntime(server: FastifyInstance) { const { apiUrl, publicUrl, enabled, disabledReason } = collaborationConfig({ @@ -16,12 +17,17 @@ export function collaborationRuntime(server: FastifyInstance) { sendMail, }) const federation = new FederationService(service, { apiUrl, enabled, disabledReason, seal: encrypt, unseal: decrypt }) + const notifications = new CollaborationNotifications(server, service, federation) + service.options.notifications = notifications service.options.changed = (roomId, db) => federation.changed(roomId, db) service.options.invited = (ctx, roomId, target, invitationId, db) => federation.invite(ctx, roomId, target, invitationId, db) let timer: ReturnType server.addHook("onReady", async () => { if (federation.options.enabled) await federation.keys() - timer = setInterval(() => federation.tick().catch(error => server.log.error({ err: error }, "Collaboration worker failed")), 5000) + timer = setInterval(() => { + void federation.tick().catch(error => server.log.error({ err: error }, "Collaboration worker failed")) + void notifications.tick().catch(error => server.log.error({ err: error }, "Collaboration notification worker failed")) + }, 5000) timer.unref() }) server.addHook("onClose", async () => { clearInterval(timer) }) diff --git a/backend/src/modules/collaboration/service.ts b/backend/src/modules/collaboration/service.ts index db96fe97..3c763885 100644 --- a/backend/src/modules/collaboration/service.ts +++ b/backend/src/modules/collaboration/service.ts @@ -4,6 +4,7 @@ import type { NodePgDatabase } from "drizzle-orm/node-postgres" import { email, fail, futureDate, hash, overview, Principal, sequence, text, UserContext, uuid } from "./types" import { chatContent, chatImage, chatReactions, imageMetadata, setChatReaction } from "../chat-content" +import type { CollaborationNotifications } from "./notifications" type Db = Pick, "execute" | "transaction"> export const rows = async (db: Pick, query: any): Promise => (await db.execute(query)).rows @@ -13,6 +14,7 @@ export type CollaborationOptions = { sendMail: (to: string, subject: string, html: string) => Promise<{ success: boolean }>; changed?: (roomId: string, db: any) => Promise; invited?: (ctx: UserContext, roomId: string, target: Principal, invitationId: string, db: any) => Promise; + notifications?: CollaborationNotifications; } /** Owner-side authority. All membership changes and sends serialize on the room row. */ @@ -201,6 +203,7 @@ export class CollaborationService { const [message] = await rows(db, sql`INSERT INTO collaboration_messages(room_id,seq,principal_id,client_id,body,image) VALUES (${room.id},${next.sequence},${principal.id},${clientId},${body},${image ? JSON.stringify(image) : null}::jsonb) RETURNING id,client_id AS "clientId",seq::float8 AS seq,body,created_at AS "createdAt"`) await this.options.changed?.(room.id, db) + await this.options.notifications?.localMessage(db, room, principal, message) return { ...message, image: imageMetadata(image), own: true, senderName: principal.display_name, senderKind: principal.kind } }) } @@ -302,6 +305,7 @@ export class CollaborationService { if (!this.options.invited) fail("Föderation ist nicht konfiguriert", 503) await this.options.invited(ctx, room.id, target, invite.id, db) } + await this.options.notifications?.localInvitation(db, room, target, invite.id) return invite }) return { ...result, token } diff --git a/backend/src/modules/notification.service.ts b/backend/src/modules/notification.service.ts index 51c1c4cd..0aabceb8 100644 --- a/backend/src/modules/notification.service.ts +++ b/backend/src/modules/notification.service.ts @@ -56,6 +56,24 @@ export class NotificationService { ) {} async trigger(input: TriggerInput) { + const { items, skipped } = await this.createItems(input) + if (skipped !== undefined) return { success: true, created: 0, delivered: 0, skipped } + const results = [] + for (const item of items) results.push(await this.deliver(item)) + return { + success: results.every((result) => result.success), + created: results.length, + delivered: results.filter((result) => result.success).length, + failed: results.filter((result) => !result.success).length, + } + } + + /** Persist first; callers using a transaction can deliver after it commits. */ + async enqueue(input: TriggerInput) { + return (await this.createItems(input)).items + } + + private async createItems(input: TriggerInput): Promise<{ items: (typeof notificationsItems.$inferSelect)[]; skipped?: number }> { const tenantId = input.tenantId const userIds = Array.from(new Set([...(input.userIds || []), input.userId].filter(Boolean))) as string[] @@ -68,7 +86,7 @@ export class NotificationService { const channels = requestedChannels.filter((channel) => allowedChannels.includes(channel)) if (!channels.length) { - return { success: true, created: 0, delivered: 0, skipped: userIds.length } + return { items: [], skipped: userIds.length } } const results = [] @@ -99,16 +117,11 @@ export class NotificationService { const item = itemRows[0] if (!item) continue - results.push(await this.deliver(item)) + results.push(item) } } - return { - success: results.every((result) => result.success), - created: results.length, - delivered: results.filter((result) => result.success).length, - failed: results.filter((result) => !result.success).length, - } + return { items: results } } async preferences(tenantId: number, userId: string) { @@ -295,11 +308,12 @@ export class NotificationService { const defaultPref = defaults.find((pref) => pref.channel === channel) if (defaultPref) return defaultPref.enabled + if (input.eventType === "collaboration.message.new") return channel === "inapp" return channel !== "email" && input.eventType !== "communication.message.room" && input.eventType !== "communication.room.invited" }) } - private async deliver(item: typeof notificationsItems.$inferSelect) { + async deliver(item: typeof notificationsItems.$inferSelect) { if (item.channel === "inapp") { await this.markSent(item.id) return { success: true, id: item.id, channel: item.channel } diff --git a/backend/tests/collaborationNotifications.test.ts b/backend/tests/collaborationNotifications.test.ts new file mode 100644 index 00000000..8a2b0024 --- /dev/null +++ b/backend/tests/collaborationNotifications.test.ts @@ -0,0 +1,201 @@ +import assert from "node:assert/strict" +import test from "node:test" +import { randomUUID } from "node:crypto" +import { readFile } from "node:fs/promises" +import Fastify from "fastify" +import { eq } from "drizzle-orm" +import { notificationsItems } from "../db/schema" +import { CollaborationService } from "../src/modules/collaboration/service" +import { FederationService, federationRoutes } from "../src/modules/collaboration/federation" +import { CollaborationNotifications } from "../src/modules/collaboration/notifications" +import type { Transport } from "../src/modules/collaboration/transport" +import { collaborationDatabase, testDatabaseUrl } from "./helpers/collaboration-db" + +const MESSAGE = "collaboration.message.new", INVITATION = "collaboration.room.invited" + +test("collaboration notifications use real preferences, transactions and signed federation routes", { skip: !testDatabaseUrl }, async t => { + const servers = new Map>() + let offline = false + const transport: Transport = async (url, method, headers, body) => { + if (offline) throw Object.assign(new Error("Offline"), { statusCode: 503 }) + const origin = [...servers.keys()].find(value => url.startsWith(value + "/"))! + const response = await servers.get(origin)!.inject({ method, url: url.slice(origin.length), headers, ...(body ? { payload: body } : {}) }) + if (response.statusCode >= 400) throw Object.assign(new Error(response.json().message), { statusCode: response.statusCode }) + return response.json() + } + const make = async (name: string) => { + const fixture = await collaborationDatabase(["0075_project_collaboration.sql", "0076_collaboration_federation.sql"]) + const original = await readFile(new URL("../db/migrations/0000_brief_dark_beast.sql", import.meta.url), "utf8") + for (const type of ["notification_channel", "notification_status", "notification_severity"]) { + const definition = original.match(new RegExp(`CREATE TYPE "public"\\."${type}" AS ENUM\\([^;]+;`))![0] + await fixture.pool.query(definition.replace('"public".', "")) + } + for (const table of ["notifications_event_types", "notifications_items", "notifications_preferences", "notifications_preferences_defaults"]) { + await fixture.pool.query(original.match(new RegExp(`CREATE TABLE "${table}" \\([\\s\\S]*?\\n\\);`))![0]) + } + await fixture.pool.query("CREATE UNIQUE INDEX ON notifications_preferences(tenant_id,user_id,event_type,channel)") + await fixture.pool.query("CREATE UNIQUE INDEX ON notifications_preferences_defaults(tenant_id,event_key,channel)") + await fixture.pool.query(await readFile(new URL("../db/migrations/0083_collaboration_notifications.sql", import.meta.url), "utf8")) + const userId = randomUUID() + await fixture.pool.query("INSERT INTO tenants VALUES (1,$1,null,null),(2,'Other tenant',null,null)", [name]) + await fixture.pool.query("INSERT INTO auth_users(id,email,is_admin) VALUES ($1,$2,true)", [userId, `${name}@example.test`]) + await fixture.pool.query("INSERT INTO auth_tenant_users(tenant_id,user_id) VALUES (1,$1),(2,$1)", [userId]) + const service = new CollaborationService(fixture.db, { publicUrl: `https://${name}.example`, guestPepper: () => "a".repeat(64), sendMail: async () => ({ success: true }) }) + const federation = new FederationService(service, { apiUrl: `https://${name}.example/backend`, enabled: true, transport, + seal: value => ({ key: value }), unseal: value => value.key }) + service.options.changed = (id, db) => federation.changed(id, db) + service.options.invited = (ctx, roomId, target, invitationId, db) => federation.invite(ctx, roomId, target, invitationId, db) + const app = Fastify() + app.decorate("db", fixture.db) + const notifications = new CollaborationNotifications(app, service, federation) + service.options.notifications = notifications + const delivered: any[] = [] + const delivery = notifications.delivery.bind(notifications) + // Keep persistence and preference resolution real; replace only external channel delivery. + notifications.delivery = db => { + const sender = delivery(db) + sender.deliver = async item => { + delivered.push(item) + await db.update(notificationsItems).set({ status: "sent", sentAt: new Date() }).where(eq(notificationsItems.id, item.id)) + return { success: true, id: item.id, channel: item.channel } + } + return sender + } + await app.register(federationRoutes, { federation }) + await app.ready() + servers.set(federation.origin, app) + return { ...fixture, service, federation, notifications, app, delivered, ctx: await service.userContext(userId, 1), other: await service.userContext(userId, 2), + items: async () => (await fixture.pool.query("SELECT * FROM notifications_items ORDER BY created_at,id")).rows } + } + const a = await make("alpha"), b = await make("beta") + try { + await a.federation.settings(a.ctx, { origin: b.federation.origin, active: true }) + await b.federation.settings(b.ctx, { origin: a.federation.origin, active: true }) + const contact = await a.federation.resolveContact(a.ctx, b.federation.contact(b.ctx.principal)) + const room = await a.service.createRoom(a.ctx, { name: "Gemeinsamer Chat" }) + const before = await a.service.send(a.ctx.principal, room.id, { clientId: randomUUID(), text: "Private history" }) + let bindingId: string + const jobFor = async (invitationId: string) => (await a.pool.query("SELECT * FROM collaboration_outbox WHERE dedup_key=$1", [`invite:${invitationId}`])).rows[0] + const itemsFor = async (instance: typeof a, id: string) => (await instance.items()).filter(item => item.payload.messageId === id) + const enable = async (instance: typeof a, tenantId = 1, enabled = true) => instance.notifications.delivery(instance.db).savePreferences(tenantId, instance.ctx.userId, + ["email", "push"].map(channel => ({ eventType: MESSAGE, channel, enabled }))) + + await t.test("new matrix events have quiet message defaults and per-tenant overrides", async () => { + const preferences = await b.notifications.delivery(b.db).preferences(1, b.ctx.userId) + assert.deepEqual(preferences.events.find(e => e.eventKey === MESSAGE)!.channels, { email: false, push: false }) + assert.deepEqual(preferences.events.find(e => e.eventKey === INVITATION)!.channels, { email: false, push: true }) + await enable(a); await enable(b) + assert.deepEqual((await b.notifications.delivery(b.db).preferences(2, b.ctx.userId)).events.find(e => e.eventKey === MESSAGE)!.channels, { email: false, push: false }) + }) + await t.test("signed invitations notify only their named recipient, once despite redelivery", async () => { + const invitation = await a.service.invite(a.ctx, room.id, contact, {}) + const job = await jobFor(invitation.id) + await a.federation.deliver(job) + bindingId = (await b.federation.list(b.ctx)).invitations[0].id + await a.pool.query("UPDATE collaboration_outbox SET status='pending' WHERE id=$1", [job.id]) + await a.federation.deliver(job) + const items = await b.items() + assert.equal(items.length, 2) + assert.deepEqual(items.map(i => i.channel).sort(), ["inapp", "push"]) + assert.ok(items.every(i => i.user_id === b.ctx.userId && Number(i.tenant_id) === 1 && i.event_type === INVITATION)) + assert.equal((await a.items()).length, 0) + await b.notifications.tick() + assert.equal(b.delivered.length, 2) + await b.federation.accept(b.ctx, bindingId) + assert.equal((await itemsFor(b, before.id)).length, 0) + }) + await t.test("new remote messages notify once with the local room link, including simultaneous syncs", async () => { + const message = await a.service.send(a.ctx.principal, room.id, { clientId: randomUUID(), text: "Hallo Beta" }) + await Promise.all([b.federation.memberRequest(b.ctx, bindingId, "messages"), b.federation.memberRequest(b.ctx, bindingId, "messages")]) + const items = await itemsFor(b, message.id) + assert.equal(items.length, 3) + assert.ok(items.every(i => i.message.includes("Hallo Beta") && i.payload.link === `/communication/chat?room=collaboration%3A${bindingId}`)) + assert.equal((await itemsFor(a, message.id)).length, 0) + await b.federation.memberRequest(b.ctx, bindingId, "messages", { before: message.seq + 1 }) + assert.equal((await itemsFor(b, message.id)).length, 3) + await b.notifications.tick() + assert.equal(b.delivered.filter(i => i.payload.messageId === message.id).length, 3) + }) + await t.test("remote replies notify the owner's local members, never the sender or retry twice", async () => { + const input = { clientId: randomUUID(), text: "Hallo Alpha" } + const reply = (await b.federation.memberRequest(b.ctx, bindingId, "send", input)).result + await b.federation.memberRequest(b.ctx, bindingId, "send", input) + await b.federation.memberRequest(b.ctx, bindingId, "messages") + assert.equal((await itemsFor(a, reply.id)).length, 3) + assert.equal((await itemsFor(b, reply.id)).length, 0) + assert.ok((await itemsFor(a, reply.id)).every(i => Number(i.tenant_id) === 1 && i.user_id === a.ctx.userId)) + await a.notifications.tick() + }) + await t.test("reading before delivery suppresses email and push, and new preferences take effect", async () => { + const read = await a.service.send(a.ctx.principal, room.id, { clientId: randomUUID(), text: "Already read" }) + await b.federation.memberRequest(b.ctx, bindingId, "messages") + await b.federation.memberRequest(b.ctx, bindingId, "read", { seq: read.seq }) + await b.notifications.tick() + assert.deepEqual(b.delivered.filter(i => i.payload.messageId === read.id).map(i => i.channel), ["inapp"]) + const muted = await a.service.send(a.ctx.principal, room.id, { clientId: randomUUID(), text: "Muted before delivery" }) + await b.federation.memberRequest(b.ctx, bindingId, "messages") + await enable(b, 1, false) + await b.notifications.tick() + assert.deepEqual(b.delivered.filter(i => i.payload.messageId === muted.id).map(i => i.channel), ["inapp"]) + await enable(b) + }) + await t.test("offline jobs survive a worker restart and concurrent workers deliver each item once", async () => { + const message = await a.service.send(a.ctx.principal, room.id, { clientId: randomUUID(), text: "Wait for access confirmation" }) + await b.federation.memberRequest(b.ctx, bindingId, "messages") + offline = true + await b.notifications.tick() + assert.equal(b.delivered.filter(i => i.payload.messageId === message.id).length, 0) + assert.ok((await itemsFor(b, message.id)).every(i => i.status === "queued")) + offline = false + const restarted = new CollaborationNotifications(b.app, b.service, b.federation) + restarted.delivery = b.notifications.delivery + await Promise.all([b.notifications.tick(), restarted.tick()]) + const delivered = b.delivered.filter(i => i.payload.messageId === message.id) + assert.equal(delivered.length, 3) + assert.equal(new Set(delivered.map(i => i.id)).size, 3) + }) + await t.test("revoked access cancels queued remote messages and reinvitation cannot replay old ones", async () => { + const message = await a.service.send(a.ctx.principal, room.id, { clientId: randomUUID(), text: "Revoked before delivery" }) + await b.federation.memberRequest(b.ctx, bindingId, "messages") + const member = (await a.service.management(a.ctx, room.id)).members.find(m => m.principalId === contact.id) + await a.service.revoke(a.ctx, room.id, member.id) + await b.notifications.tick() + assert.equal(b.delivered.filter(i => i.payload.messageId === message.id).length, 0) + assert.equal((await itemsFor(b, message.id)).length, 0) + const invitation = await a.service.invite(a.ctx, room.id, contact, {}) + await a.federation.deliver(await jobFor(invitation.id)) + await b.federation.accept(b.ctx, bindingId) + await b.notifications.tick() + assert.equal((await itemsFor(b, message.id)).length, 0) + }) + await t.test("local cross-tenant invitations and messages use recipient scope and honor expiry", async () => { + const local = await a.service.createRoom(a.ctx, { name: "Tenant exchange" }) + const invitation = await a.service.invite(a.ctx, local.id, a.other.principal, {}) + const invites = (await a.items()).filter(i => i.payload.invitationId === invitation.id) + assert.equal(invites.length, 2) + assert.ok(invites.every(i => Number(i.tenant_id) === 2)) + await a.service.accept(a.other.principal, invitation.id) + const first = await a.service.send(a.ctx.principal, local.id, { clientId: randomUUID(), text: "Only tenant two" }) + assert.deepEqual((await itemsFor(a, first.id)).map(i => [Number(i.tenant_id), i.channel]), [[2, "inapp"]]) + await a.pool.query("UPDATE collaboration_members SET expires_at=now()-interval '1 second' WHERE room_id=$1 AND principal_id=$2", [local.id, a.other.principal.id]) + await a.notifications.tick() + assert.equal(a.delivered.filter(i => i.payload.messageId === first.id).length, 0) + const second = await a.service.send(a.ctx.principal, local.id, { clientId: randomUUID(), text: "Expired member" }) + assert.equal((await itemsFor(a, second.id)).length, 0) + }) + await t.test("message and notification records roll back together", async () => { + const hook = a.notifications.localMessage.bind(a.notifications) + a.notifications.localMessage = async (...args) => { await hook(...args); throw new Error("Rollback after enqueue") } + const clientId = randomUUID(), count = (await a.items()).length + try { + await assert.rejects(b.federation.call(1, a.federation.origin, "messages.send", { actorId: b.ctx.principal.id, roomId: room.id, + input: { clientId, text: "Must roll back", revision: Number((await b.federation.binding(b.ctx, bindingId)).revision) } }), /Rollback/) + } finally { a.notifications.localMessage = hook } + assert.equal((await a.items()).length, count) + assert.equal((await a.pool.query("SELECT id FROM collaboration_messages WHERE client_id=$1", [clientId])).rowCount, 0) + }) + } finally { + await a.app.close(); await b.app.close() + await a.close(); await b.close() + } +}) diff --git a/docs/chat-foederation-konzept.md b/docs/chat-foederation-konzept.md index 12d3f770..f549f8d5 100644 --- a/docs/chat-foederation-konzept.md +++ b/docs/chat-foederation-konzept.md @@ -328,18 +328,21 @@ Datenbank und `ENCRYPTION_KEY` gemeinsam sichern; der private Instanzschlüssel ### Prüfung und verbleibender Ausbau -Geprüft wurden 21 Tests ohne Überspringen auf einer isolierten PostgreSQL-17-Testdatenbank. Darunter sind die Absicherung bestehender Chatgrenzen, die Trennung von Gast- und regulären Sitzungen, benachbarte Unterräume, der Historienbeginn, abgelaufene und widerrufene Einladungen sowie zwei unabhängige Instanzdatenbanken mit echten Fastify-Routen und Signaturprüfung. Der Transport zwischen diesen Instanzen ist im Test simuliert: Verbindungsfehler, verlorene Bestätigungen, doppelte Zustellung, veraltete Teilnahmeversionen, verspätete Widerrufe und erneute Einladungen werden gezielt erzeugt. HTTPS/DNS-Einschränkungen und Signaturmanipulationen besitzen zusätzliche Tests. Eine Verbindung zwischen zwei produktiven HTTPS-Installationen wurde nicht ausgeführt. +Geprüft wurden 50 Tests ohne Überspringen auf einer isolierten PostgreSQL-17-Testdatenbank. Darunter sind die Absicherung bestehender Chatgrenzen, die Trennung von Gast- und regulären Sitzungen, benachbarte Unterräume, der Historienbeginn, abgelaufene und widerrufene Einladungen sowie zwei unabhängige Instanzdatenbanken mit echten Fastify-Routen und Signaturprüfung. Die Benachrichtigungstests prüfen zusätzlich Empfängermandanten, persönliche Kanalwahl, Lesestatus, wiederholte Synchronisation, konkurrierende Worker, Neustarts und gemeinsame Transaktions-Rollbacks. Der Transport zwischen diesen Instanzen und die externe Benachrichtigungszustellung sind im Test simuliert: Verbindungsfehler, verlorene Bestätigungen, doppelte Zustellung, veraltete Teilnahmeversionen, verspätete Widerrufe und erneute Einladungen werden gezielt erzeugt. HTTPS/DNS-Einschränkungen und Signaturmanipulationen besitzen zusätzliche Tests. Eine Verbindung zwischen zwei produktiven HTTPS-Installationen wurde nicht ausgeführt. Reproduzierbar aus `backend/` mit einer ausdrücklich bereitgestellten, entbehrlichen Datenbank namens `collaboration_test`: ```sh COLLABORATION_TEST_DATABASE_URL=postgresql://USER:PASSWORD@127.0.0.1:PORT/collaboration_test \ node --import tsx --test tests/communicationAccess.test.ts tests/collaboration.test.ts \ - tests/collaborationRoutes.test.ts tests/collaborationFederation.test.ts tests/chatContent.test.ts + tests/collaborationRoutes.test.ts tests/collaborationFederation.test.ts tests/chatContent.test.ts \ + tests/collaborationNotifications.test.ts tests/notificationPreferences.test.ts ``` Jeder Datenbanktest erzeugt ein eigenes Schema und räumt es anschließend auf. Ohne gesetzte Testadresse werden die Datenbanktests bewusst übersprungen; der Datenbankname wird vor dem Anlegen geprüft. Backend-Typprüfung und Nuxt-Produktionsbuild waren erfolgreich, die abschließenden Vue-Komponenten lassen sich kompilieren. Die spätere Erweiterung um Emojis, Bilder und Reaktionen wurde zusätzlich mit einer lokalen Browser-Vorschau der tatsächlichen Vue-Komponenten und simulierten API-Antworten geprüft. Die automatisierten Medientests prüfen unter anderem Bildvalidierung, Bildnachrichten ohne Text, Reaktionszähler, Historiengrenzen, Gastrechte, deaktivierte Projektchats und signierte Zustellung zwischen zwei Instanzen. -Die nächste Ausbaustufe aus Schritt 3 umfasst insbesondere native Mobile-Parität, Push-Benachrichtigungen, Moderation, feinere Kontingente sowie Schlüsselrotation und abgestimmte Wiederherstellung. Dateifreigaben, Aufgabenfreigaben und E2EE sind weiterhin Erweiterungen. Der hier implementierte Stand ist ein nutzbarer Web-Pilot für Chats und ausdrücklich begrenzte Projektzusammenarbeit; ein produktiver Rollout ist damit noch nicht erfolgt. +Nachrichten und persönliche Einladungen aus der Zusammenarbeit unterstützen jetzt In-App-, E-Mail- und Push-Benachrichtigungen. Die Einstellungen gelten im Heimatmandanten des Empfängers. Die Migration `0083_collaboration_notifications.sql` muss dafür eingespielt sein. Der Versand erfolgt nach dem Speichern über den Hintergrundworker und prüft die aktuellen Zugriffsrechte; wiederholte Synchronisationen erzeugen keine doppelten Benachrichtigungen. Details und Standardwerte stehen in [Persönliche Benachrichtigungen](notification-preferences.md). Die Zustellung auf realen Endgeräten hängt weiterhin von der jeweiligen Push-Konfiguration ab. + +Die nächste Ausbaustufe aus Schritt 3 umfasst insbesondere native Mobile-Parität, Moderation, feinere Kontingente sowie Schlüsselrotation und abgestimmte Wiederherstellung. Dateifreigaben, Aufgabenfreigaben und E2EE sind weiterhin Erweiterungen. Der hier implementierte Stand ist ein nutzbarer Web-Pilot für Chats und ausdrücklich begrenzte Projektzusammenarbeit; ein produktiver Rollout ist damit noch nicht erfolgt. Die automatische Einrichtung wurde zusätzlich mit sechs Konfigurationstests und 58 Updater-Tests geprüft. Die Update-Regressionen verwenden echte Compose-Auswertung mit simuliertem Containerbetrieb und prüfen sowohl Installationen ohne Föderationsvariablen als auch den Erhalt einer ausdrücklichen Abschaltung und eigener URL-Vorgaben. diff --git a/docs/notification-preferences.md b/docs/notification-preferences.md index 63fa01d7..4e9ef90a 100644 --- a/docs/notification-preferences.md +++ b/docs/notification-preferences.md @@ -10,16 +10,23 @@ Unter Einstellungen → Benachrichtigungen (`/settings/notifications`) im Web un | `communication.message.mentioned` | Persönliche @-Erwähnung im Raum oder Projektchat | aus | an | | `communication.message.room` | Sonstige Nachricht in einem Raum oder Projektchat mit Mitgliedschaft | aus | aus | | `communication.room.invited` | Aufnahme in einen neu angelegten allgemeinen Chatraum | aus | aus | +| `collaboration.message.new` | Neue Nachricht in einem freigegebenen Unterraum oder föderierten Chat, ohne Absender | aus | aus | +| `collaboration.room.invited` | Persönliche Einladung in einen Unterraum oder föderierten Chat | aus | an | -Erwähnte Empfänger erhalten nur das Erwähnungsereignis, nicht zusätzlich das normale Raumereignis. Projektchats sind die nativen FEDEO-Chaträume; föderierte Zusammenarbeit wird hier noch nicht erfasst. +Erwähnte Empfänger erhalten nur das Erwähnungsereignis, nicht zusätzlich das normale Raumereignis. Projektchats sind die nativen FEDEO-Chaträume. Die beiden Ereignisse unter „Zusammenarbeit“ gelten für lokale Unterräume, mandantenübergreifende Teilnahmen und föderierte Chats. Sie erscheinen automatisch in den Einstellungen im Web und in der App. -Persönliche Einstellungen haben Vorrang vor Tenant-Standardwerten. Ohne beide Einträge gelten die obigen Standardwerte. In-App bleibt für Direktnachrichten und Erwähnungen aktiv; die neuen optionalen Raumereignisse sind auch In-App standardmäßig aus. Technische Testnachrichten sind nicht Teil der Matrix. +Persönliche Einstellungen haben Vorrang vor Tenant-Standardwerten. Ohne beide Einträge gelten die obigen Standardwerte. In-App bleibt für Direktnachrichten, Erwähnungen und die Zusammenarbeit aktiv; die optionalen nativen Raumereignisse sind auch In-App standardmäßig aus. Technische Testnachrichten sind nicht Teil der Matrix. + +Für die Zusammenarbeit zählt immer der Heimatmandant des empfangenden Benutzers, auch wenn der Raum einem anderen Mandanten oder einer Partnerinstanz gehört. Benachrichtigt werden nur aktive, berechtigte FEDEO-Benutzer. Gäste ohne Benutzerkonto behalten ihren gesonderten Einladungsversand und ihre Anmeldung per E-Mail-Code. + +Nachrichten und Einladungen werden mit ihren Benachrichtigungen atomar gespeichert. Ein Worker verarbeitet die vorgemerkten Kanäle alle fünf Sekunden unabhängig von einem geöffneten Browser. Wiederholte Nachrichtenbefehle, Einladungslieferungen und Synchronisationen erzeugen keine zusätzlichen Einträge; eigene Nachrichten und rückwärts nachgeladener Verlauf lösen keine Benachrichtigung aus. Bei Partnerausfällen wartet der Versand auf eine erneute Zugriffsprüfung. Widerrufene oder abgelaufene Teilnahmen und erledigte Einladungen werden vor dem Versand verworfen. Bereits gelesene Nachrichten lösen keine ausstehenden E-Mails oder Pushs mehr aus. Die persönlichen Einstellungen werden vor dem Versand erneut geprüft. Fehler des eigentlichen E-Mail-/Push-Zustellers werden wie bisher als fehlgeschlagen gespeichert; dafür gibt es noch keinen automatischen Wiederholungsversand. ## API und Migration - `GET /api/notifications/preferences`: wirksame Einstellungen für den Benutzer und Tenant der authentifizierten Session. - `PUT /api/notifications/preferences`: Array aus `{eventType, channel, enabled}`; Kanäle `email` und `push`. Unbekannte Ereignisse, nicht erlaubte Kanäle, nicht-boolsche Werte und Duplikate werden mit 400 abgewiesen. Änderungen werden atomar gespeichert. Benutzer- oder Tenant-IDs im Body ändern den Geltungsbereich nicht. - Migration `0079_notification_matrix` registriert die neuen Ereignisse und übernimmt persönliche Einstellungen und Tenant-Standards von `communication.message.new` für Direktnachrichten und Erwähnungen. Sie deaktiviert den alten Sammelauslöser sowie die bisher nicht angebundenen Anrufereignisse. Historische Benachrichtigungen bleiben erhalten. +- Migration `0083_collaboration_notifications` registriert die beiden Zusammenarbeitsereignisse und indiziert deren wartende Zustellungen. Es werden keine historischen lokalen Nachrichten oder Einladungen nachträglich verschickt. Bei der nächsten Synchronisation können bisher noch nicht abgeholte, ungelesene Partnernachrichten Benachrichtigungen auslösen. - E-Mail verwendet den SMTP-Versand der Instanz, mobiles Push den zentralen Service, Browser-Push die Web-Push-Konfiguration der Instanz. Ein aktivierter Kanal setzt eine funktionierende Zustellkonfiguration voraus. ## Sinnvolle nächste Auslöser