Streaming

export type SseFrame = { event: string; data: string };

const FRAME_BOUNDARY = /\r\n\r\n|\n\n|\r\r/;

export function decodeFrame(raw: string): SseFrame | null {
  let event = 'message';
  const data: string[] = [];
  for (const line of raw.split(/\r\n|\r|\n/)) {
    if (!line || line.startsWith(':')) continue; // Blank line or comment.
    const separator = line.indexOf(':');
    const field = separator === -1 ? line : line.slice(0, separator);
    let value = separator === -1 ? '' : line.slice(separator + 1);
    if (value.startsWith(' ')) value = value.slice(1);
    if (field === 'event') event = value;
    if (field === 'data') data.push(value);
  }
  return data.length ? { event, data: data.join('\n') } : null;
}

export async function* parseSse(body: ReadableStream<Uint8Array>): AsyncGenerator<SseFrame> {
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  let finished = false;
  try {
    while (true) {
      const { done, value } = await reader.read();
      buffer += decoder.decode(value, { stream: !done });
      let match: RegExpExecArray | null;
      while ((match = FRAME_BOUNDARY.exec(buffer))) {
        const frame = decodeFrame(buffer.slice(0, match.index));
        buffer = buffer.slice(match.index + match[0].length);
        if (frame) yield frame;
      }
      if (done) {
        finished = true;
        break;
      }
    }
    // Every complete frame ends with a blank line. Leftover text means the stream was cut off
    // mid-frame, so drop it rather than parse a partial event.
  } finally {
    // Runs when the caller breaks out of the loop or aborts: release the connection.
    if (!finished) await reader.cancel('SSE consumer stopped').catch(() => undefined);
    reader.releaseLock();
  }
}

const baseUrl = process.env.DOCTRONIC_API_URL ?? 'https://staging-partners.doctronic.ai/api/v1';

type Message = { role: string; content: string };

export async function streamTurn(
  chatId: string,
  userId: string,
  userInput: string,
  signal: AbortSignal, // Abort on caller disconnect, a cancel action, or your timeout.
): Promise<void> {
  const response = await fetch(`${baseUrl}/chats/${chatId}/stream/`, {
    method: 'POST',
    signal,
    headers: {
      Authorization: `Bearer ${process.env.DOCTRONIC_API_TOKEN}`,
      'X-Doctronic-User-ID': userId,
      'Content-Type': 'application/json',
      Accept: 'text/event-stream',
    },
    body: JSON.stringify({
      userEvent: { type: 'user_message', userInput },
      timezone: 'America/New_York',
    }),
  });
  if (!response.ok || !response.body) {
    // Errors before the stream starts arrive as ordinary JSON responses.
    throw new Error(`Stream failed: ${response.status} ${await response.text()}`);
  }

  const messages = new Map<string, Message>();
  for await (const frame of parseSse(response.body)) {
    const event = JSON.parse(frame.data) as Record<string, unknown>;
    switch (frame.event) {
      case 'message.start':
        messages.set(String(event.messageId), { role: String(event.role), content: '' });
        break;
      case 'message.content': {
        const message = messages.get(String(event.messageId));
        if (message) message.content += String(event.content ?? '');
        else console.warn('message.content before message.start', event.messageId);
        break;
      }
      case 'message.stop':
        console.log('Message complete', messages.get(String(event.messageId)));
        break;
      case 'guardian':
        console.log('Run the approved safety behavior', event.guardianId);
        break;
      case 'conversation.offtopic':
      case 'conversation.end':
      case 'skill_code.cta':
        console.log('Workflow event', frame.event, event);
        break;
      case 'ai_consultation.complete':
        console.log('Store artifact IDs', event.artifacts);
        break;
      case 'error':
        console.log('Stream error', event.errorCode, event.retryAfterSeconds);
        break;
      default:
        // New event names are additive. Ignore what this client doesn't use yet.
        break;
    }
  }
}