rebreak-monorepo/apps/rebreak-native/lib/sosStream.ts

// SSE Streaming Helper für Lyras SOS-Antworten.
//
// Architektur (bewusst entkoppelt):
// - Step 1: POST /api/coach/sos-session → liefert sessionId
// - Step 2: EventSource auf /api/coach/sos-stream?session=<id>
// - Events: 'message' (chunk), 'chips' (parsed), 'done', 'error'
//
// Phase B (sentence-level TTS): zusätzlich `onSentence?` Callback. Wenn gesetzt,
// feuert er sobald ein vollständiger Satz erkannt wird (live während des
// Streams) + einmal am Stream-Ende für den Tail. Aufrufer kann den Satz dann
// direkt in eine TTS-Queue schieben → erste Audio-Wiedergabe ~3s früher als
// "warten bis fullText fertig".
import EventSource from 'react-native-sse';
import type { BenchOnMetric } from './sosTtsBenchmark';
import type { LlmProvider } from './llmProvider';

type SseEvents = 'message' | 'chips' | 'crisis_chips' | 'done';

export type CrisisLevel = 'crisis' | 'elevated' | 'none';

export type StreamSosLyraOpts = {
  apiBase: string;
  token: string;
  messages: Array<{ role: 'user' | 'assistant'; content: string }>;
  locale: string;
  /** LLM-Provider-Switch: bestimmt welches Modell der Server für diese Session
   *  benutzt. Default (undefined) → openrouter-sonnet auf Server-Seite. */
  llmProvider?: LlmProvider;
  onTextUpdate: (full: string) => void;
  onChips: (chips: Array<{ label: string; action: string }>) => void;
  /** Feuert wenn das Backend Krisenerkennung meldet — vor Stream-Start, aus
   *  POST-Response. Ermöglicht sofortiges UI-Update ohne auf SSE zu warten. */
  onCrisisLevel?: (level: CrisisLevel) => void;
  /** Feuert beim `crisis_chips`-SSE-Event mit Hotline-Chips vom Backend. */
  onCrisisChips?: (chips: Array<{ label: string; action: string }>) => void;
  /** Phase B: feuert pro fertigem Satz live während des Streams + Tail beim
   *  done. Bei aktivem TTS-Streaming sollte der Aufrufer hier seine Queue
   *  füllen statt im onDone den ganzen Text zu sprechen. */
  onSentence?: (sentence: string) => void;
  onDone: (full: string) => void;
  onError: (err: unknown) => void;
  /** Latenz-Benchmark: feuert session-post-start, session-post-done,
   *  sse-first-chunk, sse-done. Siehe lib/sosTtsBenchmark.ts. */
  onMetric?: BenchOnMetric;
};

// Min-Länge für sentence-level TTS — winzige "Hm." / "Ja." kommen mit dem
// nächsten Satz mit, sonst klingt's choppy.
const MIN_SENTENCE_CHARS = 8;

/**
 * Findet vollständige Sätze im Text. Ein Satz endet bei `[.!?]` GEFOLGT VON
 * Whitespace + Großbuchstaben (oder Zitat-Anfang). Das filtert die häufigen
 * deutschen Abkürzungen "z.B. einfach", "d.h. nichts" etc. ohne explizite
 * Abkürzungs-Liste — der nächste Char ist dann meistens lowercase.
 *
 * Returns: { sentences[], consumed } — wieviele chars vom Anfang von `text`
 * bereits in `sentences` enthalten sind (inkl. trailing whitespace).
 */
function consumeCompletedSentences(text: string): { sentences: string[]; consumed: number } {
  const sentences: string[] = [];
  const re = /[.!?](?=\s+[A-ZÄÖÜ"„])/g;
  let lastEnd = 0;
  let m: RegExpExecArray | null;
  while ((m = re.exec(text)) !== null) {
    const endOfSentence = m.index + 1;
    sentences.push(text.slice(lastEnd, endOfSentence));
    const wsMatch = text.slice(endOfSentence).match(/^\s+/);
    lastEnd = endOfSentence + (wsMatch ? wsMatch[0].length : 0);
    re.lastIndex = lastEnd;
  }
  return { sentences, consumed: lastEnd };
}

export async function streamSosLyra(opts: StreamSosLyraOpts): Promise<() => void> {
  // Step 1: POST zu /api/coach/sos-session → sessionId holen
  opts.onMetric?.('session-post-start');
  const sessRes = await fetch(`${opts.apiBase}/api/coach/sos-session`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${opts.token}`,
    },
    body: JSON.stringify({ messages: opts.messages, locale: opts.locale, llmProvider: opts.llmProvider }),
  });
  if (!sessRes.ok) throw new Error(`session: ${sessRes.status}`);
  const sessData = await sessRes.json();
  const { sessionId, crisisLevel } = sessData as { sessionId: string; crisisDetected?: boolean; crisisLevel?: CrisisLevel };
  if (crisisLevel && crisisLevel !== 'none' && opts.onCrisisLevel) {
    opts.onCrisisLevel(crisisLevel);
  }
  opts.onMetric?.('session-post-done');

  // Step 2: EventSource für SSE-Stream
  // pollingInterval: 0 → KEIN Auto-Reconnect (Session ist one-time-use)
  const es = new EventSource<SseEvents>(`${opts.apiBase}/api/coach/sos-stream?session=${sessionId}`, {
    headers: { Authorization: `Bearer ${opts.token}` },
    pollingInterval: 0,
    lineEndingCharacter: '\n',
  });

  let fullText = '';
  let sentenceConsumedIndex = 0;
  let firstChunkSeen = false;

  const flushNewSentences = () => {
    if (!opts.onSentence) return;
    const remaining = fullText.slice(sentenceConsumedIndex);
    const { sentences, consumed } = consumeCompletedSentences(remaining);
    for (const s of sentences) {
      const trimmed = s.trim();
      if (trimmed.length >= MIN_SENTENCE_CHARS) {
        opts.onSentence(trimmed);
      }
    }
    sentenceConsumedIndex += consumed;
  };

  es.addEventListener('message', (event) => {
    if (!event.data) return;
    // Backend sendet JSON-encoded String → parse für korrekte Whitespace-Behandlung
    let chunk: string;
    try {
      chunk = JSON.parse(event.data);
    } catch {
      chunk = event.data;
    }
    if (!chunk) return;
    if (!firstChunkSeen) {
      firstChunkSeen = true;
      opts.onMetric?.('sse-first-chunk');
    }
    fullText += chunk;
    opts.onTextUpdate(fullText);
    // Phase B: live sentence-detection für TTS-Queue
    flushNewSentences();
  });

  es.addEventListener('chips', (event) => {
    if (!event.data) return;
    try {
      const chips = JSON.parse(event.data);
      if (Array.isArray(chips)) opts.onChips(chips);
    } catch { /* ignore */ }
  });

  es.addEventListener('crisis_chips', (event) => {
    if (!event.data) return;
    try {
      const chips = JSON.parse(event.data);
      if (Array.isArray(chips) && opts.onCrisisChips) opts.onCrisisChips(chips);
    } catch { /* ignore */ }
  });

  es.addEventListener('done', () => {
    opts.onMetric?.('sse-done');
    // Phase B: Tail flushen (letzter Satz ohne folgendes Capital-Letter wird
    // sonst nie als "complete" erkannt). Trim leeren Tail away.
    if (opts.onSentence) {
      const tail = fullText.slice(sentenceConsumedIndex).trim();
      if (tail.length > 0) {
        opts.onSentence(tail);
      }
    }
    opts.onDone(fullText);
    es.close();
  });

  es.addEventListener('error', (err) => {
    opts.onError(err);
    es.close();
  });

  // Return cancel-Funktion
  return () => { es.close(); };
}