AfiS-Aktensystem/web/app/api/stream/route.ts

import { getCurrentUser, type CurrentUser } from '@/lib/auth/current'
import { db } from '@/lib/db'
import { visibleCaseWhere } from '@/lib/access/caseFilter'
import { visibleMessageWhere } from '@/lib/access/chatFilter'
import { visiblePersonnelWhere } from '@/lib/access/personnelFilter'
import { visibleAttachmentWhere } from '@/lib/access/attachmentFilter'
import { resolveCasePreviews, type CasePreview } from '@/lib/chat/preview'
import { extractCaseNumbers, type ChatMessageView } from '@/lib/chat/view'
import { ensureListening, subscribe } from '@/lib/realtime/listener'
import type { StreamEvent, Topic } from '@/lib/realtime/publish'

export const runtime = 'nodejs'
export const dynamic = 'force-dynamic'

/** Gegen Zwischenspeicher und Tunnel, die eine stille Verbindung für tot halten. */
const HEARTBEAT_MS = 25_000

/**
 * Ereignisstrom für offene Ansichten.
 *
 *     GET /api/stream?topics=case:<id>,personnel:<id>
 *
 * Server-Sent Events und kein WebSocket: die Richtung ist einseitig, es geht
 * durch jeden Tunnel, und der Browser baut die Verbindung von selbst wieder
 * auf. Siehe docs/06-chat-und-realtime.md.
 *
 * Zwei Prüfungen, beide nötig:
 *  1. **Beim Verbinden**: welche der angefragten Themen darf dieser Nutzer
 *     überhaupt abonnieren. Wer ein fremdes Thema nennt, bekommt es nicht —
 *     und erfährt auch nicht, ob es existiert.
 *  2. **Je Ereignis**: der Inhalt wird mit den Rechten des Empfängers geladen.
 *     Eine Umklassifizierung mitten in einer offenen Verbindung wirkt dadurch
 *     sofort, ohne dass jemand neu laden muss.
 */
export async function GET(request: Request): Promise<Response> {
  const user = await getCurrentUser()
  if (!user) return new Response(null, { status: 404 })

  const url = new URL(request.url)
  const gewuenscht = (url.searchParams.get('topics') ?? '')
    .split(',')
    .map((t) => t.trim())
    .filter(Boolean)
    .slice(0, 20)

  const erlaubt = await erlaubteThemen(user, gewuenscht)
  await ensureListening()

  const encoder = new TextEncoder()
  let heartbeat: ReturnType<typeof setInterval> | null = null
  let unsubscribe: (() => void) | null = null
  let offen = true

  const stream = new ReadableStream({
    async start(controller) {
      const send = (event: string, data: unknown, id?: string) => {
        if (!offen) return
        try {
          const zeilen = [
            ...(id ? [`id: ${id}`] : []),
            `event: ${event}`,
            `data: ${JSON.stringify(data)}`,
            '',
            '',
          ]
          controller.enqueue(encoder.encode(zeilen.join('\n')))
        } catch {
          offen = false
        }
      }

      // Verpasstes nachliefern. Der Browser schickt beim Wiederverbinden von
      // selbst die zuletzt gesehene Kennung mit; die Kennungen sind UUIDv7 und
      // damit zeitlich sortiert, ein Vergleich darauf reicht.
      const seit = request.headers.get('last-event-id') ?? url.searchParams.get('since')
      if (seit) await nachliefern(user, erlaubt, seit, send)

      send('ready', { topics: erlaubt })

      heartbeat = setInterval(() => {
        if (!offen) return
        try {
          controller.enqueue(encoder.encode(': ping\n\n'))
        } catch {
          offen = false
        }
      }, HEARTBEAT_MS)

      unsubscribe = subscribe((event) => {
        if (!erlaubt.includes(event.t)) return
        void zustellen(user, event, send)
      })

      request.signal.addEventListener('abort', () => {
        offen = false
        try {
          controller.close()
        } catch {
          // schon zu
        }
      })
    },
    cancel() {
      offen = false
      if (heartbeat) clearInterval(heartbeat)
      unsubscribe?.()
    },
  })

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
      // Der Tunnel darf nicht puffern, sonst kommt alles am Stück.
      'X-Accel-Buffering': 'no',
    },
  })
}

/**
 * Filtert die angefragten Themen auf die, die der Nutzer sehen darf.
 *
 * Das eigene Nutzerthema ist immer dabei, alles andere muss belegt werden.
 */
async function erlaubteThemen(user: CurrentUser, gewuenscht: string[]): Promise<Topic[]> {
  const themen: Topic[] = [`user:${user.id}`]

  const fallIds = gewuenscht
    .filter((t) => t.startsWith('case:'))
    .map((t) => t.slice(5))
    .filter(Boolean)
  const personalIds = gewuenscht
    .filter((t) => t.startsWith('personnel:'))
    .map((t) => t.slice(10))
    .filter(Boolean)

  if (fallIds.length > 0) {
    const sichtbar = await db.case.findMany({
      where: { AND: [visibleCaseWhere(user.permissions), { id: { in: fallIds } }] },
      select: { id: true },
    })
    themen.push(...sichtbar.map((akte) => `case:${akte.id}` as Topic))
  }

  // Personalakten setzen neben der Stufe ein eigenes Recht voraus.
  const darfPersonal =
    user.permissions.isAdmin || user.permissions.permissions.has('personnel.read')
  if (personalIds.length > 0 && darfPersonal) {
    const sichtbar = await db.personnelFile.findMany({
      where: { AND: [visiblePersonnelWhere(user.permissions), { id: { in: personalIds } }] },
      select: { id: true },
    })
    themen.push(...sichtbar.map((akte) => `personnel:${akte.id}` as Topic))
  }

  return themen
}

type Send = (event: string, data: unknown, id?: string) => void

/** Was seit `seit` in den abonnierten Themen passiert ist. */
async function nachliefern(
  user: CurrentUser,
  themen: Topic[],
  seit: string,
  send: Send,
): Promise<void> {
  const fallIds = themen.filter((t) => t.startsWith('case:')).map((t) => t.slice(5))
  const personalIds = themen.filter((t) => t.startsWith('personnel:')).map((t) => t.slice(10))
  if (fallIds.length === 0 && personalIds.length === 0) return

  try {
    const verpasst = await db.chatMessage.findMany({
      where: {
        AND: [
          visibleMessageWhere(user.permissions),
          { id: { gt: seit } },
          { OR: [{ caseId: { in: fallIds } }, { personnelFileId: { in: personalIds } }] },
        ],
      },
      orderBy: { id: 'asc' },
      take: 200,
      select: NACHRICHT,
    })

    for (const angereichert of await anreichern(user, verpasst)) {
      send('chat.new', angereichert, angereichert.id)
    }
  } catch (error) {
    console.error('[stream] Nachlieferung fehlgeschlagen', error)
  }
}

const NACHRICHT = {
  id: true,
  body: true,
  authorId: true,
  replyToId: true,
  editedAt: true,
  withdrawnAt: true,
  adoptedEntryId: true,
  mentions: true,
  createdAt: true,
  caseId: true,
  personnelFileId: true,
} as const

/**
 * Ergänzt, was die Anzeige braucht und was **je Empfänger verschieden** ist:
 * den Namen des Autors und die Vorschaukarten zu erwähnten Aktenzeichen. Die
 * Karten werden mit den Rechten des Empfängers aufgelöst, deshalb hier und
 * nicht beim Absender.
 */
function vorschauenFuer(
  body: string,
  alle: Map<string, CasePreview>,
): Record<string, CasePreview> {
  const treffer: Record<string, CasePreview> = {}
  for (const nummer of extractCaseNumbers(body)) {
    const vorschau = alle.get(nummer)
    if (vorschau) treffer[nummer] = vorschau
  }
  return treffer
}

async function anreichern(
  user: CurrentUser,
  nachrichten: {
    id: string
    body: string
    authorId: string
    replyToId: string | null
    editedAt: Date | null
    withdrawnAt: Date | null
    adoptedEntryId: string | null
    mentions: string[]
    createdAt: Date
  }[],
): Promise<ChatMessageView[]> {
  if (nachrichten.length === 0) return []

  const [autoren, vorschauen] = await Promise.all([
    db.user.findMany({
      where: { id: { in: [...new Set(nachrichten.map((n) => n.authorId))] } },
      select: { id: true, displayName: true, username: true },
    }),
    resolveCasePreviews(
      user.permissions,
      nachrichten.flatMap((n) => extractCaseNumbers(n.body)),
    ),
  ])
  const nameVon = new Map(autoren.map((a) => [a.id, a.displayName]))

  return nachrichten.map((nachricht) => ({
    id: nachricht.id,
    body: nachricht.withdrawnAt ? '' : nachricht.body,
    authorId: nachricht.authorId,
    authorName: nameVon.get(nachricht.authorId) ?? nachricht.authorId,
    replyToId: nachricht.replyToId,
    editedAt: nachricht.editedAt?.toISOString() ?? null,
    withdrawnAt: nachricht.withdrawnAt?.toISOString() ?? null,
    adoptedEntryId: nachricht.adoptedEntryId,
    createdAt: nachricht.createdAt.toISOString(),
    previews: vorschauenFuer(nachricht.body, vorschauen),
  }))
}

/**
 * Lädt den Inhalt zu einem Ereignis **mit den Rechten des Empfängers**.
 *
 * Findet die Abfrage nichts, wird nichts geschickt: weder Inhalt noch die
 * Kennung. Sonst verriete schon die Kennung, dass es etwas gibt.
 */
async function zustellen(user: CurrentUser, event: StreamEvent, send: Send): Promise<void> {
  try {
    if (event.k.startsWith('chat.')) {
      const nachricht = await db.chatMessage.findFirst({
        where: { AND: [visibleMessageWhere(user.permissions), { id: event.i }] },
        select: NACHRICHT,
      })
      if (!nachricht) return
      const [angereichert] = await anreichern(user, [nachricht])
      if (angereichert) send(event.k, angereichert, angereichert.id)
      return
    }

    if (event.k === 'attachment.scanned') {
      const anhang = await db.attachment.findFirst({
        where: { AND: [visibleAttachmentWhere(user.permissions), { id: event.i }] },
        select: {
          id: true,
          scanStatus: true,
          scanMessage: true,
          mimeType: true,
          sizeBytes: true,
          thumbKey: true,
          blurhash: true,
          width: true,
          height: true,
        },
      })
      if (anhang) {
        send('attachment.scanned', { ...anhang, hasThumb: anhang.thumbKey !== null, thumbKey: undefined })
      }
      return
    }

    // Verlaufseinträge und Kopfdaten: nur die Kennung. Die Ansicht lädt
    // daraufhin nach, statt dass hier ein zweiter Renderpfad entsteht.
    send(event.k, { id: event.i })
  } catch (error) {
    console.error('[stream] Ereignis nicht zugestellt', error)
  }
}