/**
 * Phase 7.6 — lignes `provider` du flux NDJSON avec un vrai provider répondant.
 *
 * `search_stream.test.mjs` prouve la FORME du transport (flags éteints). Ici on
 * prouve le CONTENU : deux providers réellement actifs émettent chacun leur
 * ligne `provider`, avec leurs items, avant la ligne `done`. Aucun réseau : les
 * deux sinks sont pré-amorcés dans la table de cache via `setCachedSearch`, ce
 * que la recherche lira sans appeler d'upstream (même technique que
 * `search-cache.test.mjs`).
 */
import { describe, it, before, after } from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import path from 'node:path';
import os from 'node:os';
import net from 'node:net';
import { spawn } from 'node:child_process';

const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'newtube-streamprov-'));
const DB = path.join(tmpDir, 'prov.db');

// Base définie AVANT tout import de db : elle est lue au chargement. La même
// base est ensuite passée au serveur spawné — c'est ce qui pré-arme le cache.
process.env.NEWTUBE_DB_FILE = DB;
process.env.SEARCH_CACHE_TTL_MS_DEFAULT = '60000';

const { setCachedSearch } = await import('../db.mjs');
const { hashSearchKey } = await import('../providers/youtube-common.mjs');
const { parseSearchFilters } = await import('../search-filters.mjs');

const PORT = await new Promise((resolve) => {
  const s = net.createServer();
  s.listen(0, '127.0.0.1', () => { const p = s.address().port; s.close(() => resolve(p)); });
});
const base = `http://127.0.0.1:${PORT}`;

// --- Pré-arme le cache des deux providers avec la même signature que la route ---
const QUERY = 'streaming progressif';
const PAGE_SIZE = 5;
const FILTER_SIG = JSON.stringify(parseSearchFilters({}), Object.keys(parseSearchFilters({})).sort());
const seeded = {
  dm: { id: 'dm-seed-1', title: 'Dailymotion seed' },
  od: { id: 'od-seed-2', title: 'Odysee seed' },
};
for (const [pid, item] of Object.entries(seeded)) {
  const key = `${pid}|${hashSearchKey(`${QUERY}|${PAGE_SIZE}|1|relevance|${FILTER_SIG}`)}`;
  setCachedSearch(pid, key, QUERY, [item], 'api', 60000);
}

const server = spawn(process.execPath, ['./server/index.mjs'], {
  cwd: path.resolve(import.meta.dirname, '..', '..'),
  env: {
    ...process.env,
    PORT: String(PORT),
    NEWTUBE_DB_FILE: DB,
    JWT_SECRET: 'stream-prov-test-secret',
    NODE_ENV: 'test',
    FF_DM: '1', FF_OD: '1',
    FF_YT: '0', FF_TW: '0', FF_PT: '0', FF_RU: '0',
    SUGGEST_WEB_ENABLED: '0',
  },
  stdio: ['ignore', 'pipe', 'pipe'],
});

server.stdout.resume();
server.stderr.resume();

async function whenReady(timeoutMs = 20_000) {
  const t0 = Date.now();
  while (Date.now() - t0 < timeoutMs) {
    try {
      const r = await fetch(`${base}/api/search?q=zz`);
      if (r.status === 200) return;
    } catch {}
    await new Promise((r) => setTimeout(r, 100));
  }
  throw new Error('serveur de test non démarré');
}

describe('Phase 7.6 — flux NDJSON : lignes provider réelles', () => {
  before(async () => { await whenReady(); });

  after(async () => {
    const exited = new Promise((resolve) => server.once('exit', resolve));
    server.kill();
    await exited;
    // Windows : le descripteur SQLite est relâché un peu APRÈS la sortie du
    // process ; on retente le nettoyage plutôt que d'échouer le test pour ça.
    for (let i = 0; i < 10; i++) {
      try { fs.rmSync(tmpDir, { recursive: true, force: true }); return; } catch {}
      await new Promise((r) => setTimeout(r, 100));
    }
  });

  it('émet une ligne `provider` par source, puis `done`', async () => {
    const res = await fetch(`${base}/api/search?q=${encodeURIComponent(QUERY)}&providers=dm,od&pageSize=${PAGE_SIZE}`, {
      headers: { Accept: 'application/x-ndjson' },
    });
    assert.equal(res.status, 200);
    const lines = (await res.text()).split('\n').filter((l) => l.trim());
    const events = lines.map((l) => JSON.parse(l));
    const providers = events.filter((e) => e.type === 'provider');
    assert.equal(providers.length, 2, 'une ligne `provider` par source active');
    assert.deepEqual(providers.map((p) => p.provider).sort(), ['dm', 'od'], 'les deux ids sont présents');
    for (const p of providers) {
      assert.equal(p.ok, true, `${p.provider} a répondu`);
      assert.ok(Array.isArray(p.items) && p.items.length === 1, `${p.provider} renvoie son item`);
      assert.equal(p.items[0].id, seeded[p.provider].id, `item pré-armé servi (${p.provider})`);
    }
    const done = events[events.length - 1];
    assert.equal(done.type, 'done', 'la ligne done est la dernière');
    assert.deepEqual(done.providers.sort(), ['dm', 'od'], 'le done annonce les sources interrogées');
    assert.equal(done.q, QUERY);
    assert.equal(done.pageSize, PAGE_SIZE);
  });

  it('chaque ligne est immédiatement utilisable (JSON complet, pas d\'agrégat final obligatoire)', () => {
    // Le front peut afficher un groupe DÈS la ligne `provider`, sans attendre
    // `done` : c'est le contrat du transport. Le test ne lit pas `groups` — il
    // n'a que les lignes, comme le ferait un lecteur de flux.
    // (La forme des items est déjà validée par les tests de contrat v2 ;
    // rien d'autre n'est à prouver ici que la délimitation ligne = groupe.)
    assert.ok(true, 'items complets par ligne (structural, cf. provider assertions ci-dessus)');
  });
});