diff --git a/app/backend/src/lib/realtime.ts b/app/backend/src/lib/realtime.ts new file mode 100644 index 0000000..2dfdafb --- /dev/null +++ b/app/backend/src/lib/realtime.ts @@ -0,0 +1,171 @@ +import type { FastifyReply, FastifyRequest } from 'fastify'; +import type { NotificationType, Prisma } from '@prisma/client'; +import { redis } from '../redis.js'; +import { prisma } from '../db.js'; +import { sendPushToUser } from './push.js'; + +const REALTIME_CHANNEL = 'teknav:realtime'; +const ACTIVE_GUESTS_KEY = 'visitors:guests'; +const ACTIVE_USERS_KEY = 'visitors:users'; +const COMMENTS_STATS_KEY = 'stats:comments'; +const ACTIVE_WINDOW_MS = 5 * 60 * 1000; + +export type RealtimeEvent = + | { event: 'notification'; userId: string; data: unknown } + | { event: 'activity'; data: unknown } + | { event: 'visitor_update'; data: { count: number; guests: number; users: number; commentsPerMin: number } } + | { event: 'view_update'; data: unknown }; + +export async function publishRealtime(event: RealtimeEvent) { + await redis.publish(REALTIME_CHANNEL, JSON.stringify(event)); +} + +export async function createNotification(userId: string, type: NotificationType, payload: Prisma.InputJsonValue) { + const row = await prisma.notification.create({ data: { userId, type, payload } }); + const data = publicNotification(row); + await publishRealtime({ event: 'notification', userId, data }); + sendPushToUser(userId, type, payload as Record).catch(() => {}); + return data; +} + +export function publicNotification(row: { + id: string; + type: NotificationType; + payload: Prisma.JsonValue; + read: boolean; + createdAt: Date; +}) { + return { + id: row.id, + type: row.type, + payload: row.payload, + read: row.read, + createdAt: row.createdAt.toISOString(), + }; +} + +export function setupSse(reply: FastifyReply) { + reply.hijack(); + reply.raw.writeHead(200, { + 'Content-Type': 'text/event-stream; charset=utf-8', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }); + reply.raw.write('retry: 5000\n\n'); +} + +export function writeSse(reply: FastifyReply, event: string, data: unknown) { + reply.raw.write(`event: ${event}\n`); + reply.raw.write(`data: ${JSON.stringify(data)}\n\n`); +} + +export async function subscribeRealtime( + reply: FastifyReply, + filter: (event: RealtimeEvent) => boolean, +) { + const subscriber = redis.duplicate(); + await subscriber.connect(); + await subscriber.subscribe(REALTIME_CHANNEL); + const onMessage = (_channel: string, message: string) => { + try { + const event = JSON.parse(message) as RealtimeEvent; + if (filter(event)) writeSse(reply, event.event, event.data); + } catch { + // Ignore malformed pub/sub payloads. + } + }; + subscriber.on('message', onMessage); + const heartbeat = setInterval(() => { + reply.raw.write(': heartbeat\n\n'); + }, 25_000); + reply.raw.on('close', () => { + clearInterval(heartbeat); + subscriber.off('message', onMessage); + subscriber.quit().catch(() => undefined); + }); +} + +export async function recordActiveVisitor(req: FastifyRequest) { + if (!req.url.startsWith('/api/') || req.url.includes('/stream')) return; + const now = Date.now(); + if (req.session?.userId) { + await redis.zadd(ACTIVE_USERS_KEY, now, req.session.userId.toString()); + await redis.pexpire(ACTIVE_USERS_KEY, ACTIVE_WINDOW_MS * 2); + } else { + const ip = req.ip || 'unknown'; + await redis.zadd(ACTIVE_GUESTS_KEY, now, ip); + await redis.pexpire(ACTIVE_GUESTS_KEY, ACTIVE_WINDOW_MS * 2); + } +} + +export async function publishVisitorCount() { + const now = Date.now(); + const cutoff = now - ACTIVE_WINDOW_MS; + const commentCutoff = now - 60 * 1000; + + await redis.zremrangebyscore(ACTIVE_GUESTS_KEY, 0, cutoff); + await redis.zremrangebyscore(ACTIVE_USERS_KEY, 0, cutoff); + await redis.zremrangebyscore(COMMENTS_STATS_KEY, 0, commentCutoff); + + const guests = await redis.zcard(ACTIVE_GUESTS_KEY); + const users = await redis.zcard(ACTIVE_USERS_KEY); + const commentsPerMin = await redis.zcard(COMMENTS_STATS_KEY); + const count = guests + users; + + await publishRealtime({ event: 'visitor_update', data: { count, guests, users, commentsPerMin } }); + return { count, guests, users, commentsPerMin }; +} + +export async function recordCommentEvent() { + const now = Date.now(); + await redis.zadd(COMMENTS_STATS_KEY, now, `${now}-${Math.random()}`); + await redis.pexpire(COMMENTS_STATS_KEY, 60 * 1000 * 2); +} + +export async function notifyFollowersOnPublish(article: { + id: string; + slug: string; + title: string; + author: { userId: string | null; name: string }; + category?: { slug: string; name?: string } | null; +}) { + const [writerFollows, topicFollows] = await Promise.all([ + article.author.userId + ? prisma.writerFollow.findMany({ + where: { writerId: article.author.userId }, + select: { followerId: true }, + }) + : Promise.resolve([]), + article.category?.slug + ? prisma.topicFollow.findMany({ + where: { topic: article.category.slug }, + select: { userId: true }, + }) + : Promise.resolve([]), + ]); + const userIds = new Set([ + ...writerFollows.map((follow) => follow.followerId), + ...topicFollows.map((follow) => follow.userId), + ]); + await Promise.all([...userIds].map((userId) => createNotification(userId, 'new_article', { + articleId: article.id, + articleSlug: article.slug, + articleTitle: article.title, + actorName: article.author.name, + topic: article.category?.slug ?? null, + topicName: article.category?.name ?? null, + }))); +} + +export async function notifyEditorsOnSubmit(article: { slug: string, title: string, authorName: string }) { + const editors = await prisma.user.findMany({ + where: { role: { in: ['admin', 'editor', 'reviewer'] } }, + select: { id: true } + }); + await Promise.all(editors.map(e => createNotification(e.id.toString(), 'review_submitted', { + articleSlug: article.slug, + articleTitle: article.title, + actorName: article.authorName, + }))); +}