pwa/backend/tests/chat-stream.test.ts

import assert from 'node:assert/strict';
import { after, before, describe, it } from 'node:test';
import { prisma } from '../src/lib/prisma.js';
import { PCQ_QUESTION_IDS } from '../src/modules/assessments/scoring/index.js';
import { seedCourseSessions } from '../src/modules/sessions/sessions.service.js';
import {
  authHeader,
  createTestApp,
  isDatabaseAvailable,
  requestJson,
} from './helpers.js';

const app = createTestApp();
const createdUserIds: bigint[] = [];

function pcqAnswers(value: number) {
  return PCQ_QUESTION_IDS.map((questionId) => ({
    questionId,
    selectedValue: String(value),
  }));
}

async function registerUser(suffix: string) {
  const email = `stream-${suffix}-${Date.now()}@example.com`;
  const { status, body } = await requestJson(app, '/auth/register', {
    method: 'POST',
    body: JSON.stringify({ email, password: 'password12' }),
  });
  assert.equal(status, 201);
  const token = body.data?.sessionToken as string;
  const externalId = (body.data?.user as { id: string }).id;
  const user = await prisma.user.findUnique({ where: { externalId } });
  assert.ok(user);
  createdUserIds.push(user.id);
  return { token, user };
}

async function completePreTest(token: string) {
  const pre = await requestJson(app, '/assessments/attempts', {
    method: 'POST',
    headers: authHeader(token),
    body: JSON.stringify({ phase: 'pre', answers: pcqAnswers(4) }),
  });
  assert.equal(pre.status, 201);
}

describe('chat SSE progressive streaming', () => {
  let dbAvailable = false;

  before(async () => {
    dbAvailable = await isDatabaseAvailable();
    if (dbAvailable) {
      await seedCourseSessions();
    }
  });

  after(async () => {
    if (dbAvailable && createdUserIds.length > 0) {
      await prisma.user.deleteMany({ where: { id: { in: createdUserIds } } });
    }
  });

  it('preserves non-streaming fallback when Accept is not text/event-stream', async (t) => {
    if (!dbAvailable) {
      t.skip('MySQL is not running');
      return;
    }

    const { token } = await registerUser('non-stream');
    await completePreTest(token);

    const created = await requestJson(app, '/chat/sessions', {
      method: 'POST',
      headers: authHeader(token),
      body: JSON.stringify({ sessionOrder: 1 }),
    });
    assert.equal(created.status, 201);
    const chatId = created.body.data?.id as string;

    const res = await requestJson(app, `/chat/sessions/${chatId}/messages`, {
      method: 'POST',
      headers: authHeader(token),
      body: JSON.stringify({ content: 'سلام' }),
    });
    assert.equal(res.status, 201);
    assert.ok(res.body.data?.userMessage);
    assert.ok(res.body.data?.assistantMessage);
  });

  it('delivers progressive chunks via SSE without leaking [COACH_SESSION_COMPLETE] control markers', async (t) => {
    if (!dbAvailable) {
      t.skip('MySQL is not running');
      return;
    }

    const { token } = await registerUser('sse-delivery');
    await completePreTest(token);

    const created = await requestJson(app, '/chat/sessions', {
      method: 'POST',
      headers: authHeader(token),
      body: JSON.stringify({ sessionOrder: 1 }),
    });
    assert.equal(created.status, 201);
    const chatId = created.body.data?.id as string;

    const response = await app.request(`/chat/sessions/${chatId}/messages`, {
      method: 'POST',
      headers: {
        ...authHeader(token),
        Accept: 'text/event-stream',
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ content: 'سلام' }),
    });

    assert.equal(response.status, 200);
    assert.match(response.headers.get('content-type') || '', /text\/event-stream/);

    const reader = response.body?.getReader();
    assert.ok(reader, 'Response body must be readable stream');

    const decoder = new TextDecoder();
    let buffer = '';
    const events: Array<{ event: string; data: any }> = [];
    let receivedFirstChunk = false;

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const blocks = buffer.split('\n\n');
      buffer = blocks.pop() || '';

      for (const block of blocks) {
        if (!block.trim()) continue;
        const lines = block.split('\n');
        let eventName = 'message';
        let dataStr = '';
        for (const line of lines) {
          if (line.startsWith('event:')) eventName = line.slice(6).trim();
          else if (line.startsWith('data:')) dataStr = line.slice(5).trim();
        }
        if (dataStr) {
          try {
            const parsed = JSON.parse(dataStr);
            events.push({ event: eventName, data: parsed });
            if (eventName === 'chunk') {
              receivedFirstChunk = true;
            }
          } catch {
            events.push({ event: eventName, data: dataStr });
          }
        }
      }
    }

    assert.ok(receivedFirstChunk, 'Must have received at least one chunk before stream completion');
    const chunkEvents = events.filter((e) => e.event === 'chunk');
    assert.ok(chunkEvents.length > 0, 'Must have delivered chunk events');

    // Verify control marker never leaks in visible chunks
    for (const chunk of chunkEvents) {
      assert.doesNotMatch(
        chunk.data.text || '',
        /\[COACH_SESSION_COMPLETE\]/,
        'Visible chunk must NEVER contain [COACH_SESSION_COMPLETE]',
      );
    }

    const doneEvent = events.find((e) => e.event === 'done');
    assert.ok(doneEvent, 'Must emit done event');
    assert.ok(doneEvent.data.assistantMessage, 'Done event must contain assistantMessage');
    assert.ok(doneEvent.data.assistantMessage.content, 'Assistant content must be non-empty');

    // Verify DB consistency
    const dbMessage = await prisma.chatMessage.findUnique({
      where: { id: BigInt(doneEvent.data.assistantMessage.id) },
    });
    assert.ok(dbMessage, 'Message must be saved in database');
    assert.equal(
      dbMessage.content,
      doneEvent.data.assistantMessage.content,
      'Database content must match done event content',
    );
    assert.doesNotMatch(
      dbMessage.content,
      /\[COACH_SESSION_COMPLETE\]/,
      'Database content must not contain raw control markers',
    );
  });

  it('supports stream start on /chat/sessions/:id/start', async (t) => {
    if (!dbAvailable) {
      t.skip('MySQL is not running');
      return;
    }

    const { token } = await registerUser('sse-start');
    await completePreTest(token);

    const created = await requestJson(app, '/chat/sessions', {
      method: 'POST',
      headers: authHeader(token),
      body: JSON.stringify({ sessionOrder: 1 }),
    });
    assert.equal(created.status, 201);
    const chatId = created.body.data?.id as string;

    const response = await app.request(`/chat/sessions/${chatId}/start`, {
      method: 'POST',
      headers: {
        ...authHeader(token),
        Accept: 'text/event-stream',
      },
    });

    assert.equal(response.status, 200);
    assert.match(response.headers.get('content-type') || '', /text\/event-stream/);

    const reader = response.body?.getReader();
    if (reader) {
      while (true) {
        const { done } = await reader.read();
        if (done) break;
      }
    }
  });
});