202 lines
15 KiB
TypeScript
202 lines
15 KiB
TypeScript
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()
|
|
}
|
|
})
|