140 lines
8.6 KiB
TypeScript
140 lines
8.6 KiB
TypeScript
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 }
|
|
}
|
|
}
|