feat(notifications): include federated chat messages and invitations

2026-10-15 17:50:38 +02:00
parent 697f538b6e
commit 7b61857fb8
11 changed files with 413 additions and 16 deletions
--- a/backend/db/migrations/0083_collaboration_notifications.sql
+++ 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');
--- 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
    }
  ]
 }
--- 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
--- 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 = {}) {
--- a/backend/src/modules/collaboration/notifications.ts
+++ 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<string, number>()
+    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<string, unknown>, 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<boolean | "defer"> {
+        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 }
+    }
+}
--- 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<typeof setInterval>
    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) })
--- 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<NodePgDatabase<any>, "execute" | "transaction">
 export const rows = async (db: Pick<Db, "execute">, query: any): Promise<any[]> => (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<void>;
    invited?: (ctx: UserContext, roomId: string, target: Principal, invitationId: string, db: any) => Promise<void>;
+    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 }
--- 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 }
--- a/backend/tests/collaborationNotifications.test.ts
+++ 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<string, ReturnType<typeof Fastify>>()
+    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()
+    }
+})