feat(communication): add chat reactions emojis and images
This commit is contained in:
90
backend/src/modules/chat-content.ts
Normal file
90
backend/src/modules/chat-content.ts
Normal file
@@ -0,0 +1,90 @@
|
||||
import sharp from "sharp"
|
||||
import { sql } from "drizzle-orm"
|
||||
import type { NodePgDatabase } from "drizzle-orm/node-postgres"
|
||||
import { fail, text, uuid } from "./collaboration/types"
|
||||
|
||||
export const reactionEmojis = ["👍", "❤️", "😂", "🎉", "😮", "😢", "🙏", "🔥", "✅", "👎"]
|
||||
export type ChatImage = { dataUrl: string; name: string; width: number; height: number }
|
||||
type Db = Pick<NodePgDatabase<any>, "execute">
|
||||
type Source = "communication" | "collaboration"
|
||||
|
||||
/** Only re-encoded raster images are stored; neither URLs nor original metadata are retained. */
|
||||
export async function chatContent(input: any): Promise<{ body: string; image: ChatImage | null }> {
|
||||
const body = text(input?.text, 10000, true)
|
||||
let image: ChatImage | null = null
|
||||
if (input?.image != null) {
|
||||
const value = input.image
|
||||
if (typeof value.dataUrl !== "string" || value.dataUrl.length > 700_000) fail("Das Bild ist zu groß (maximal 512 KB nach Verkleinerung)")
|
||||
const match = /^data:image\/(?:jpeg|png|webp);base64,([A-Za-z0-9+/]+={0,2})$/.exec(value.dataUrl)
|
||||
if (!match) fail("Nur JPEG-, PNG- und WebP-Bilder sind erlaubt")
|
||||
try {
|
||||
const buffer = Buffer.from(match[1], "base64")
|
||||
if (buffer.length > 512 * 1024) fail("Das Bild ist zu groß")
|
||||
const decoder = sharp(buffer, { limitInputPixels: 16_000_000, failOn: "warning" })
|
||||
const metadata = await decoder.metadata()
|
||||
if (!["jpeg", "png", "webp"].includes(metadata.format || "") || (metadata.pages || 1) > 1) fail("Bitte ein einzelnes JPEG-, PNG- oder WebP-Bild verwenden")
|
||||
const result = await decoder.rotate().resize({ width: 1600, height: 1600, fit: "inside", withoutEnlargement: true })
|
||||
.webp({ quality: 82 }).toBuffer({ resolveWithObject: true })
|
||||
if (result.data.length > 512 * 1024) fail("Das Bild ist zu groß. Bitte eine kleinere Datei wählen")
|
||||
image = { dataUrl: `data:image/webp;base64,${result.data.toString("base64")}`,
|
||||
name: text(value.name || "Bild", 200), width: result.info.width, height: result.info.height }
|
||||
} catch (error: any) {
|
||||
if (error.statusCode) throw error
|
||||
fail("Das Bild konnte nicht gelesen werden")
|
||||
}
|
||||
}
|
||||
if (!body && !image) fail("Nachricht darf nicht leer sein")
|
||||
return { body, image }
|
||||
}
|
||||
|
||||
export function imageMetadata(value: any) {
|
||||
if (value == null) return null
|
||||
if (!Number.isInteger(value.width) || !Number.isInteger(value.height) || value.width < 1 || value.height < 1
|
||||
|| value.width > 1600 || value.height > 1600) fail("Ungültige Bildabmessungen")
|
||||
return { name: text(value.name, 200), width: value.width, height: value.height }
|
||||
}
|
||||
|
||||
function messageId(source: Source, value: unknown) {
|
||||
if (source === "collaboration") return uuid(value)
|
||||
const id = Number(value)
|
||||
if (!Number.isSafeInteger(id) || id < 1) fail("Ungültige Nachricht")
|
||||
return id
|
||||
}
|
||||
const tables = (source: Source) => ({ messages: sql.identifier(`${source}_messages`), reactions: sql.identifier(`${source}_message_reactions`) })
|
||||
const history = (source: Source, from: number) => source === "collaboration" ? sql`AND m.seq>=${from}` : sql``
|
||||
|
||||
/** Callers must authorize room membership first, including the current history boundary. */
|
||||
export async function chatImage(db: Db, source: Source, roomId: string, id: unknown, from = 0) {
|
||||
const { messages } = tables(source)
|
||||
const result = await db.execute(sql`SELECT m.image FROM ${messages} m WHERE m.room_id=${roomId}
|
||||
AND m.id=${messageId(source, id)} ${history(source, from)}`)
|
||||
const image = result.rows[0]?.image as ChatImage | null
|
||||
if (!image) fail("Bild nicht gefunden oder kein Zugriff", 404)
|
||||
return { dataUrl: image.dataUrl }
|
||||
}
|
||||
|
||||
export async function chatReactions(db: Db, source: Source, roomId: string, actorId: string, value: unknown, from = 0) {
|
||||
const ids = typeof value === "string" ? value.split(",").filter(Boolean) : value
|
||||
if (!Array.isArray(ids) || ids.length > 100) fail("Ungültige Nachrichtenliste")
|
||||
const unique = [...new Set(ids.map(id => messageId(source, id)))]
|
||||
const reactions: Record<string, { emoji: string; count: number; own: boolean }[]> = {}
|
||||
if (!unique.length) return { reactions }
|
||||
const { messages, reactions: table } = tables(source)
|
||||
const result = await db.execute(sql`SELECT r.message_id AS id,r.emoji,count(*)::int AS count,bool_or(r.actor_id=${actorId}) AS own
|
||||
FROM ${table} r JOIN ${messages} m ON m.id=r.message_id
|
||||
WHERE m.room_id=${roomId} AND m.id IN (${sql.join(unique.map(id => sql`${id}`), sql`,`)}) ${history(source, from)}
|
||||
GROUP BY r.message_id,r.emoji ORDER BY r.emoji`)
|
||||
for (const row of result.rows as any[]) (reactions[row.id] ||= []).push({ emoji: row.emoji, count: row.count, own: row.own })
|
||||
return { reactions }
|
||||
}
|
||||
|
||||
export async function setChatReaction(db: Db, source: Source, roomId: string, actorId: string, input: any, from = 0) {
|
||||
const id = messageId(source, input?.messageId)
|
||||
if (!reactionEmojis.includes(input?.emoji) || typeof input?.active !== "boolean") fail("Ungültige Reaktion")
|
||||
const { messages, reactions } = tables(source)
|
||||
const result = await db.execute(sql`SELECT m.id FROM ${messages} m WHERE m.room_id=${roomId} AND m.id=${id} ${history(source, from)}`)
|
||||
if (!result.rows.length) fail("Nachricht nicht gefunden oder kein Zugriff", 404)
|
||||
if (input.active) await db.execute(sql`INSERT INTO ${reactions}(message_id,actor_id,emoji) VALUES (${id},${actorId},${input.emoji}) ON CONFLICT DO NOTHING`)
|
||||
else await db.execute(sql`DELETE FROM ${reactions} WHERE message_id=${id} AND actor_id=${actorId} AND emoji=${input.emoji}`)
|
||||
return chatReactions(db, source, roomId, actorId, [id], from)
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { chatContent, imageMetadata } from "../chat-content"
|
||||
import { generateKeyPairSync, randomUUID } from "node:crypto"
|
||||
import { sql } from "drizzle-orm"
|
||||
import type { FastifyInstance } from "fastify"
|
||||
@@ -122,7 +123,9 @@ export class FederationService {
|
||||
ON CONFLICT(dedup_key) DO UPDATE SET dedup_key=excluded.dedup_key RETURNING *`)
|
||||
if (JSON.stringify(job.payload) !== JSON.stringify(entry.payload)) {
|
||||
// JSONB key order differs: compare the actual command values instead of its serialization.
|
||||
if (job.payload.input?.text !== entry.payload.input?.text || job.payload.actorId !== entry.payload.actorId || job.payload.roomId !== entry.payload.roomId || job.payload.input?.revision !== entry.payload.input?.revision) fail("Auftrags-ID wurde bereits anders verwendet", 409)
|
||||
if (job.payload.input?.text !== entry.payload.input?.text || job.payload.actorId !== entry.payload.actorId || job.payload.roomId !== entry.payload.roomId || job.payload.input?.revision !== entry.payload.input?.revision
|
||||
|| (job.payload.input?.image?.dataUrl || null) !== (entry.payload.input?.image?.dataUrl || null)
|
||||
|| (job.payload.input?.image?.name || null) !== (entry.payload.input?.image?.name || null)) fail("Auftrags-ID wurde bereits anders verwendet", 409)
|
||||
}
|
||||
return job
|
||||
}
|
||||
@@ -223,6 +226,9 @@ export class FederationService {
|
||||
if (action === "room.get") return this.service.room(principal, room.id)
|
||||
if (action === "messages.get") return this.service.messages(principal, room.id, request.input)
|
||||
if (action === "messages.send") return this.service.send(principal, room.id, request.input)
|
||||
if (action === "image.get") return this.service.image(principal, room.id, request.input)
|
||||
if (action === "reactions.get") return this.service.reactions(principal, room.id, request.input)
|
||||
if (action === "reactions.set") return this.service.react(principal, room.id, request.input)
|
||||
if (action === "members.get") return { members: await this.service.members(principal, room.id) }
|
||||
if (action === "read") return this.service.markRead(principal, room.id, request.input?.seq)
|
||||
fail("Unbekannte Föderationsaktion")
|
||||
@@ -300,7 +306,7 @@ export class FederationService {
|
||||
}
|
||||
async cache(binding: any, messages: any[], snapshot?: any) {
|
||||
if (!Array.isArray(messages) || messages.length > 50) fail("Ungültige Nachrichtenantwort", 502)
|
||||
const projected = messages.map(m => ({ id: uuid(m.id), clientId: uuid(m.clientId), seq: sequence(m.seq), body: text(m.body, 10000),
|
||||
const projected = messages.map(m => ({ id: uuid(m.id), clientId: uuid(m.clientId), seq: sequence(m.seq), body: text(m.body, 10000, Boolean(m.image)), image: imageMetadata(m.image),
|
||||
senderName: text(m.senderName, 400), senderKind: ['user','guest','remote'].includes(m.senderKind) ? m.senderKind : 'remote',
|
||||
own: m.own === true, createdAt: new Date(m.createdAt).toISOString() }))
|
||||
await this.service.db.transaction(async db => {
|
||||
@@ -326,8 +332,9 @@ export class FederationService {
|
||||
await this.allowed(ctx.tenantId, binding.origin)
|
||||
if (action === "send") {
|
||||
if (!binding.snapshot?.canSend) fail("Unterraum ist nicht zum Schreiben freigegeben", 403)
|
||||
const content = await chatContent(input)
|
||||
const command = { actorId: ctx.principal.id, roomId: binding.room_id,
|
||||
input: { clientId: uuid(input.clientId), text: text(input.text, 10000), revision: Number(binding.revision) } }
|
||||
input: { clientId: uuid(input.clientId), text: content.body, image: content.image, revision: Number(binding.revision) } }
|
||||
const job = await this.enqueue(this.service.db, { key: `message:${binding.id}:${command.input.clientId}`, ctx, origin: binding.origin,
|
||||
action: 'messages.send', payload: command, bindingId: binding.id })
|
||||
await this.deliver(job)
|
||||
@@ -336,6 +343,20 @@ export class FederationService {
|
||||
if (currentJob.status === 'failed') fail(currentJob.last_error || "Nachricht konnte nicht gesendet werden", 409)
|
||||
return { found: true, result: message?.payload || this.pendingMessage(currentJob) }
|
||||
}
|
||||
if (["image", "reactions", "react"].includes(action)) {
|
||||
const rpc = action === "image" ? "image.get" : action === "react" ? "reactions.set" : "reactions.get"
|
||||
try {
|
||||
const result = await this.call(ctx.tenantId, binding.origin, rpc, { actorId: ctx.principal.id, roomId: binding.room_id,
|
||||
input: { ...input, revision: Number(binding.revision) } })
|
||||
return { found: true, result }
|
||||
} catch (error: any) {
|
||||
if ([401, 403, 404, 410].includes(error.statusCode)) {
|
||||
// A missing image/message is not by itself evidence of revoked room access.
|
||||
await this.sync(ctx, binding)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
const detail = await this.sync(ctx, binding)
|
||||
binding = await this.binding(ctx, id)
|
||||
if (action === "room") return { found: true, result: detail }
|
||||
@@ -368,7 +389,7 @@ export class FederationService {
|
||||
return { found: true, result: { messages, hasMore: page.length === 50, offline: detail.offline } }
|
||||
}
|
||||
pendingMessage(job: any) {
|
||||
return { id: job.payload.input.clientId, clientId: job.payload.input.clientId, seq: null, body: job.payload.input.text,
|
||||
return { id: job.payload.input.clientId, clientId: job.payload.input.clientId, seq: null, body: job.payload.input.text, image: imageMetadata(job.payload.input.image),
|
||||
createdAt: job.created_at, own: true, senderName: 'Du', senderKind: 'user', pending: job.status === 'pending', failed: job.status === 'failed' }
|
||||
}
|
||||
async deliver(job: any) {
|
||||
@@ -433,12 +454,12 @@ export class FederationService {
|
||||
|
||||
export async function federationRoutes(server: FastifyInstance, { federation }: { federation: FederationService }) {
|
||||
server.removeContentTypeParser('application/json')
|
||||
server.addContentTypeParser('application/json', { parseAs: 'string', bodyLimit: 128 * 1024 }, (req: any, body: string, done) => {
|
||||
server.addContentTypeParser('application/json', { parseAs: 'string', bodyLimit: 1024 * 1024 }, (req: any, body: string, done) => {
|
||||
req.federationBody = body
|
||||
try { done(null, JSON.parse(body)) } catch { done(Object.assign(new Error('Ungültiges JSON'), { statusCode: 400 })) }
|
||||
})
|
||||
server.get('/.well-known/fedeo-federation', async () => federation.discovery())
|
||||
server.post('/federation/v1/rpc', { bodyLimit: 128 * 1024 }, async (req: any, reply) => {
|
||||
server.post('/federation/v1/rpc', { bodyLimit: 1024 * 1024 }, async (req: any, reply) => {
|
||||
reply.header('Cache-Control','no-store')
|
||||
// Duplicate signature-bearing fields are rejected rather than silently combined.
|
||||
const names = req.raw.rawHeaders.filter((_: string, index: number) => index % 2 === 0).map((name: string) => name.toLowerCase())
|
||||
|
||||
@@ -3,6 +3,8 @@ import { sql } from "drizzle-orm"
|
||||
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"
|
||||
|
||||
type Db = Pick<NodePgDatabase<any>, "execute" | "transaction">
|
||||
export const rows = async (db: Pick<Db, "execute">, query: any): Promise<any[]> => (await db.execute(query)).rows
|
||||
export type CollaborationOptions = {
|
||||
@@ -172,7 +174,7 @@ export class CollaborationService {
|
||||
const room = await this.roomRow(this.db, roomId)
|
||||
const member = await this.membership(this.db, room, principal)
|
||||
const after = sequence(query.after), before = sequence(query.before)
|
||||
const result = await rows(this.db, sql`SELECT m.id,m.client_id AS "clientId",m.seq::float8 AS seq,m.body,m.created_at AS "createdAt",
|
||||
const result = await rows(this.db, sql`SELECT m.id,m.client_id AS "clientId",m.seq::float8 AS seq,m.body,m.image - 'dataUrl' AS image,m.created_at AS "createdAt",
|
||||
p.display_name AS "senderName",p.kind AS "senderKind", m.principal_id=${principal.id} AS own
|
||||
FROM collaboration_messages m JOIN collaboration_principals p ON p.id=m.principal_id
|
||||
WHERE m.room_id=${room.id} AND m.seq>=${member.history_from_seq}
|
||||
@@ -182,22 +184,45 @@ export class CollaborationService {
|
||||
}
|
||||
|
||||
async send(principal: Principal, roomId: string, input: any) {
|
||||
const body = text(input.text, 10000), clientId = uuid(input.clientId)
|
||||
const clientId = uuid(input.clientId)
|
||||
await this.membership(this.db, await this.roomRow(this.db, roomId), principal, true)
|
||||
const { body, image } = await chatContent(input)
|
||||
return this.db.transaction(async (db) => {
|
||||
const room = await this.roomRow(db, roomId, true)
|
||||
const member = await this.membership(db, room, principal, true)
|
||||
if (principal.kind === "remote" && sequence(input.revision) !== Number(member.revision)) fail("Teilnahmerechte haben sich geändert", 409)
|
||||
const [existing] = await rows(db, sql`SELECT id,client_id AS "clientId",seq::float8 AS seq,body,created_at AS "createdAt" FROM collaboration_messages
|
||||
const [existing] = await rows(db, sql`SELECT id,client_id AS "clientId",seq::float8 AS seq,body,image,created_at AS "createdAt" FROM collaboration_messages
|
||||
WHERE room_id=${room.id} AND principal_id=${principal.id} AND client_id=${clientId}`)
|
||||
if (existing) {
|
||||
if (existing.body !== body) fail("Diese Auftrags-ID wurde bereits für eine andere Nachricht verwendet", 409)
|
||||
return { ...existing, own: true, senderName: principal.display_name, senderKind: principal.kind }
|
||||
if (existing.body !== body || (existing.image?.dataUrl || null) !== (image?.dataUrl || null) || (existing.image?.name || null) !== (image?.name || null)) fail("Diese Auftrags-ID wurde bereits für eine andere Nachricht verwendet", 409)
|
||||
return { ...existing, image: imageMetadata(existing.image), own: true, senderName: principal.display_name, senderKind: principal.kind }
|
||||
}
|
||||
const [next] = await rows(db, sql`UPDATE collaboration_rooms SET sequence=sequence+1,updated_at=now() WHERE id=${room.id} RETURNING sequence`)
|
||||
const [message] = await rows(db, sql`INSERT INTO collaboration_messages(room_id,seq,principal_id,client_id,body)
|
||||
VALUES (${room.id},${next.sequence},${principal.id},${clientId},${body}) RETURNING id,client_id AS "clientId",seq::float8 AS seq,body,created_at AS "createdAt"`)
|
||||
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)
|
||||
return { ...message, own: true, senderName: principal.display_name, senderKind: principal.kind }
|
||||
return { ...message, image: imageMetadata(image), own: true, senderName: principal.display_name, senderKind: principal.kind }
|
||||
})
|
||||
}
|
||||
|
||||
async image(principal: Principal, roomId: string, input: any) {
|
||||
const room = await this.roomRow(this.db, roomId)
|
||||
const member = await this.membership(this.db, room, principal)
|
||||
return chatImage(this.db, "collaboration", room.id, input?.messageId, Number(member.history_from_seq))
|
||||
}
|
||||
|
||||
async reactions(principal: Principal, roomId: string, input: any) {
|
||||
const room = await this.roomRow(this.db, roomId)
|
||||
const member = await this.membership(this.db, room, principal)
|
||||
return chatReactions(this.db, "collaboration", room.id, principal.id, input?.ids, Number(member.history_from_seq))
|
||||
}
|
||||
|
||||
async react(principal: Principal, roomId: string, input: any) {
|
||||
return this.db.transaction(async db => {
|
||||
const room = await this.roomRow(db, roomId, true)
|
||||
const member = await this.membership(db, room, principal, true)
|
||||
if (principal.kind === "remote" && sequence(input?.revision) !== Number(member.revision)) fail("Teilnahmerechte haben sich geändert", 409)
|
||||
return setChatReaction(db, "collaboration", room.id, principal.id, input, Number(member.history_from_seq))
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user