feat(providers): phases 7.3/7.4/7.6/8.1 — provenance, health, NDJSON, contrat unique
CI / build-and-test (push) Successful in 14m43s

7.3: capturedAt/source au registre + 6 adaptateurs + module provenance.ts + ?debug=1 (search-transport.mjs). 7.4: ProviderHealthService + badge source degradee. 7.6: squelettes par provider + snapshots progressifs + transport NDJSON /api/search. 8.1: ProviderAdapter unifie (search enveloppe + channelContent/channelMeta/capabilities) via getProviderAdapter + test de contrat offline.
This commit is contained in:
2026-09-30 07:57:11 -04:00
parent 37681c4f53
commit 665a0f0ebd
87 changed files with 8850 additions and 550 deletions
+232 -34
View File
@@ -15,8 +15,10 @@ import ffmpegPath from 'ffmpeg-static';
import * as cheerio from 'cheerio';
import axios from 'axios';
import rumbleRouter from './rumble.mjs';
import { providerRegistry, validateProviders } from './providers/registry.mjs';
import { providerRegistry, getProviderAdapter, validateProviders, SUGGESTION_CONTRACT_VERSION } from './providers/registry.mjs';
import { applyProviderFlags, providerFlag } from './providers/feature-flags.mjs';
import { dedupeSuggestGroups } from './suggest.mjs';
import { buildDebugPayload, APPLICATION_NDJSON } from './search-transport.mjs';
import { parseSearchFilters, activeFilters as activeSearchFilters, applySearchFilters, filtersCacheKey } from './search-filters.mjs';
import { fetchWebSuggest, fetchOdyseeLighthouseSuggest } from './suggest-web.mjs';
import { pickTrack, parseTrackText, parseVtt, dedupeTranscriptLines, orderedTracks, translatedFallbacks, firstPerLanguage, normalizeTranscriptProvider, transcriptTrackExt, looksLikeHtmlError, ensureFmtParam, isEmptyTimedTextBody, isBlobTranscript, mergeTranscriptCandidates } from './transcript.mjs';
@@ -102,7 +104,7 @@ import {
deleteTranscriptHistoryById,
deleteAllTranscriptHistory,
} from './db.mjs';
import { getChannelAdapter, setTwitchTokenProvider } from './providers/channel-registry.mjs';
import { setTwitchTokenProvider } from './providers/channel-registry.mjs';
import { fetchChannelContent } from './providers/channel-content.mjs';
import {
oauthStatus, buildAuthUrl, createOAuthState, consumeOAuthState, exchangeCode,
@@ -275,11 +277,11 @@ function requireProviderId(value) {
}
async function resolveChannel(provider, externalId, { forceRefresh = false } = {}) {
const adapterEntry = getChannelAdapter(provider) || providerRegistry[provider];
if (!adapterEntry || typeof adapterEntry.fetchChannelById !== 'function') {
const adapterEntry = getProviderAdapter(provider);
if (!adapterEntry || typeof adapterEntry.channelMeta !== 'function') {
throw Object.assign(new Error('provider_not_supported'), { status: 501 });
}
return ensureChannelFresh(provider, externalId, () => adapterEntry.fetchChannelById(externalId), { force: forceRefresh });
return ensureChannelFresh(provider, externalId, () => adapterEntry.channelMeta(externalId), { force: forceRefresh });
}
r.post('/channels/resolve', authMiddlewareCookieAware, channelsLimiter, async (req, res) => {
@@ -325,8 +327,12 @@ r.get('/channels/:provider/:externalId/content', channelsLimiter, async (req, re
const limit = Math.min(50, Math.max(1, Number(req.query.limit || 24)));
const sort = req.query.sort === 'popular' ? 'popular' : req.query.sort === 'relevance' ? 'relevance' : 'recent';
const q = typeof req.query.q === 'string' ? req.query.q.slice(0, 200) : '';
const data = await fetchChannelContent(provider, externalId, { type, page, limit, sort, q }, { searchRegistry: providerRegistry });
return res.json({ ...data, page, limit, sort, type });
// Phase 3.1 - curseur opaque pour les API paginées par curseur (Twitch Helix).
// Sans lui, `page=2` re-demandait la page 1 et la pagination bouclait à
// l'infini. Le front le renvoie tel quel dans `?cursor=`.
const cursor = typeof req.query.cursor === 'string' ? req.query.cursor.slice(0, 512) : '';
const data = await getProviderAdapter(provider).channelContent(externalId, { type, page, limit, sort, q, cursor }, { searchRegistry: providerRegistry });
return res.json({ ...data, page, limit, sort, type, ...(data?.nextCursor ? { nextCursor: data.nextCursor } : {}) });
} catch (error) {
const status = error?.status || 500;
return res.status(status).json({ error: error?.message || 'channel_content_failed', items: [], nextPage: null });
@@ -1761,9 +1767,9 @@ r.get('/details/:provider/:videoId', async (req, res) => {
if (!uploaderAvatar && channelExternalId) {
const shortToRegistry = { youtube: 'yt', dailymotion: 'dm', twitch: 'tw', peertube: 'pt', odysee: 'od', rumble: 'ru' };
const regProvider = shortToRegistry[String(provider)] || String(provider);
const adapter = (typeof getChannelAdapter === 'function') ? getChannelAdapter(regProvider) : null;
if (adapter && typeof adapter.fetchChannelById === 'function') {
const ch = await adapter.fetchChannelById(String(channelExternalId));
const adapter = getProviderAdapter(regProvider);
if (adapter && typeof adapter.channelMeta === 'function') {
const ch = await adapter.channelMeta(String(channelExternalId));
if (ch?.avatarUrl) uploaderAvatar = ch.avatarUrl;
if (typeof ch?.subsCount === 'number' && !subscribers) subscribers = ch.subsCount;
}
@@ -2873,13 +2879,124 @@ r.get('/img/odysee', async (req, res) => {
});
// Mount API router (prod) and alias for dev proxy
// Phase 4.5 : purge de fond du cache de recherche (toutes les 10 min).
// `unref()` est indispensable : sans lui le timer empeche le process de
// s'arreter, y compris dans les tests qui demarrent puis ferment le serveur.
const SEARCH_CACHE_JANITOR_MS = Math.max(60e3, Number(process.env.SEARCH_CACHE_PRUNE_MS || 10 * 60 * 1000));
let searchCacheJanitorStarted = false;
function startSearchCacheJanitor() {
if (searchCacheJanitorStarted) return;
searchCacheJanitorStarted = true;
const tick = async () => {
try {
const { pruneSearchCache, purgeProviderMetrics } = await import('./db.mjs');
const purged = pruneSearchCache();
const metrics = purgeProviderMetrics();
if (purged || metrics) console.log(`[cache] purge: ${purged} entrées search_cache, ${metrics} lignes provider_metrics`);
} catch (e) {
console.warn('[cache] purge échouée:', e?.message || e);
}
};
const timer = setInterval(tick, SEARCH_CACHE_JANITOR_MS);
timer.unref?.();
// Une purge immédiate au boot : une base qui a tourné des semaines hors ligne
// peut accumuler des lignes expirees en masse.
setTimeout(tick, 2000).unref?.();
}
// Phase 4.2 : migration `youtube_search_cache` -> `search_cache`, une fois au boot.
// Idempotente et non destructive : l'ancienne table reste lue en repli.
(async () => {
try {
const { migrateYoutubeCacheToSearchCache } = await import('./db.mjs');
const r = migrateYoutubeCacheToSearchCache();
if (r?.migrated > 0) console.log(`[cache] migration youtube_search_cache -> search_cache: ${r.migrated} ligne(s) copiée(s)`);
} catch (e) {
console.warn('[cache] migration ignorée:', e?.message || e);
}
})();
// Phase 4.3 : observabilité fournisseurs. Aucun secret exposé.
const PROVIDER_IDS = ['yt', 'dm', 'tw', 'pt', 'od', 'ru'];
// Sonde par provider : une recherche `limit=1`, résultat mis en cache 60 s
// (sans ce garde-fou, `/providers/health` créerait l'inverse du problème qu'il
// mesure en martelant les upstreams).
const healthProbeCache = new Map();
const HEALTH_PROBE_TTL_MS = Math.max(5e3, Number(process.env.PROVIDER_HEALTH_TTL_MS || 60e3));
async function probeProvider(pid) {
const cached = healthProbeCache.get(pid);
if (cached && Date.now() - cached.at < HEALTH_PROBE_TTL_MS) return cached.value;
const t0 = Date.now();
let value;
try {
const { providerRegistry } = await import('./providers/registry.mjs');
const adapter = providerRegistry[pid];
if (!adapter || typeof adapter.search !== 'function') throw new Error('adapter_absent');
const items = await adapter.search('test', { limit: 1, page: 1, sort: 'relevance' });
value = {
ok: true, latencyMs: Date.now() - t0, itemCount: Array.isArray(items) ? items.length : 0,
lastSuccessAt: new Date().toISOString(), lastError: null,
};
} catch (e) {
value = { ok: false, latencyMs: Date.now() - t0, itemCount: 0, lastSuccessAt: null, lastError: String(e?.message || e).slice(0, 300) };
}
healthProbeCache.set(pid, { at: Date.now(), value });
return value;
}
r.get('/providers/health', async (req, res) => {
try {
const only = String(req.query.provider || '').trim();
const ids = only ? (PROVIDER_IDS.includes(only) ? [only] : []) : PROVIDER_IDS;
if (!ids.length) return res.status(400).json({ error: 'unknown_provider' });
const snapshot = (await import('./db.mjs')).providerMetricsSnapshot({ hours: 1 });
const entries = await Promise.all(ids.map(async (pid) => {
// Phase 8.3 : on ne sonde PAS un provider désactivé par feature flag.
// Sonder un upstream volontairement éteint renverrait `ok: false`, alors
// qu'il n'est pas en panne : `disabled` distingue « éteint » de « cassé ».
const flag = providerFlag(pid);
if (!flag.enabled) {
return [pid, {
ok: false, disabled: true, flag: flag.flag, flagValue: flag.raw,
latencyMs: 0, itemCount: 0, lastSuccessAt: null, lastError: 'disabled_by_ff',
consecutiveFailures: 0, errorRate: 0,
}];
}
const probe = await probeProvider(pid);
const row = snapshot.find((x) => x.provider === pid);
return [pid, {
...probe,
disabled: false,
flag: flag.flag, flagValue: flag.raw,
consecutiveFailures: row?.errors || 0, errorRate: row?.errorRate || 0,
}];
}));
res.json({ providers: Object.fromEntries(entries) });
} catch (e) {
res.status(500).json({ error: String(e?.message || e) });
}
});
r.get('/providers/metrics', async (_req, res) => {
try {
const dbm = await import('./db.mjs');
res.json({
cache: dbm.searchCacheStats?.() || [],
providers: dbm.providerMetricsSnapshot?.({ hours: 1 }) || [],
youtube: { today: dbm.getYoutubeMetricsToday?.() || null },
});
} catch (e) {
res.status(500).json({ error: String(e?.message || e) });
}
});
app.use('/api', r);
// Health endpoint for container checks
app.get('/api/health', (_req, res) => res.json({ status: 'ok' }));
// Step 17 : observabilité YouTube (mode, yt-dlp, cache, quota). Aucun secret exposé.
app.get(['/healthz', '/api/healthz'], async (_req, res) => {
try {
const [{ getYoutubeMetricsToday, countYoutubeCacheRows }, common] =
const [{ getYoutubeMetricsToday, countYoutubeCacheRows, searchCacheStats, providerMetricsSnapshot, countSearchCacheRows }, common] =
await Promise.all([import('./db.mjs'), import('./providers/youtube-common.mjs')]);
let ytdlpVersion = null;
let resolvedBin = null;
@@ -2895,6 +3012,9 @@ app.get(['/healthz', '/api/healthz'], async (_req, res) => {
status: 'ok',
youtube: {
mode: getSearchMode(),
// Phase 4.4 : le mode reel peut differer de l'intention (bascule auto).
effectiveMode: common.getEffectiveSearchMode?.() || getSearchMode(),
failover: common.getFailoverState?.() || null,
ytdlp: { bin: resolvedBin || getYtDlpBin(), version: ytdlpVersion, info: ytDlpInfo, binOk: Boolean(ytdlpVersion) },
antiban: {
cookiesFile: hasCookiesFile(),
@@ -2905,6 +3025,12 @@ app.get(['/healthz', '/api/healthz'], async (_req, res) => {
metrics: { ...metricsSnapshot(), today: getYoutubeMetricsToday() },
keys: { count: keys.length, banned },
},
// Phase 4.3 : résumé multi-fournisseurs. Volontairement compact — le
// détail est sur `/api/providers/metrics`.
providers: {
cache: { rows: countSearchCacheRows?.() || 0, byProvider: searchCacheStats?.() || [] },
lastHour: providerMetricsSnapshot?.({ hours: 1 }) || [],
},
});
} catch (e) {
res.status(500).json({ status: 'error', error: String(e?.message || e) });
@@ -3117,6 +3243,14 @@ app.all('/proxy/twitch-api/*', (req, res) => forwardJson(req, res, 'https://api.
app.all('/proxy/twitch-auth/*', (req, res) => forwardJson(req, res, 'https://id.twitch.tv'));
// -------------------- Unified search endpoint (GET) --------------------
// Phase 7.3 : le payload `?debug=1` (provenance + `raw` tronque et redige) est
// construit par `search-transport.mjs` : la troncature et la redaction sont une
// barriere de securite, elles meritent un module et des tests dedies.
// Phase 7.6 : deux TRANSPORTS pour un SEUL fan-out. `Accept: application/x-ndjson`
// (ou `?stream=1`) fait écrire une ligne JSON dès qu'un provider répond ; sinon
// la réponse atomique historique est renvoyée à l'identique. Les deux modes
// partagent le même code d'agrégation : impossible qu'ils divergent sur le
// filtrage ou la forme des erreurs.
app.get('/api/search', async (req, res) => {
try {
console.log('[SEARCH] Requête reçue - Query:', req.query);
@@ -3140,46 +3274,102 @@ app.get('/api/search', async (req, res) => {
sort = filters.sort;
// Validate and normalize providers list (default to all supported when none/invalid)
const requested = typeof providers === 'string' ? String(providers) : '';
const validProviders = validateProviders(requested);
// Phase 8.3 : on retire les providers désactivés par feature flag AVANT le
// fan-out (inutile d'appeler un upstream qu'on a choisi d'éteindre), et on
// le signale dans `errors` pour que la colonne vide soit explicable.
const { providerIds: validProviders, errors: flagErrors } = applyProviderFlags(validateProviders(requested));
for (const [pid, err] of Object.entries(flagErrors)) {
console.warn(`[search] provider ${pid} désactivé (${err.code})`);
}
// Execute search for each provider in parallel
const results = await Promise.allSettled(
validProviders.map((providerId) => {
const mod = providerRegistry[/** @type {any} */(providerId)];
if (!mod || typeof mod.search !== 'function') return Promise.resolve([]);
// Basic options include pagination, sort hints and the search filters
return Promise.resolve().then(() => mod.search(q, { limit: pageSize, page: pageNum, sort, filters }));
})
);
// Phase 7.6 : transport incrémental. `Accept: text/event-stream` n'est pas
// utilisé volontairement — NDJSON se parse ligne à ligne sans parseur SSE, et
// reste rejouable / testable avec un simple client HTTP.
const wantsStream = String(req.headers.accept || '').includes(APPLICATION_NDJSON)
|| req.query.stream === '1';
let streamStarted = false;
if (wantsStream) {
// `X-Accel-Buffering: no` : derrière nginx (configuré par défaut), un
// buffer de réponse différerait TOUT le NDJSON jusqu'à la fin — on
// obtiendrait un streaming qui streame en un seul coup, c'est-à-dire rien.
res.setHeader('Content-Type', `${APPLICATION_NDJSON}; charset=utf-8`);
res.setHeader('Cache-Control', 'no-store');
res.setHeader('X-Accel-Buffering', 'no');
res.flushHeaders?.();
streamStarted = true;
}
/** Écrit une ligne du flux, si le client a demandé le streaming. */
const emit = (line) => {
if (!streamStarted) return;
// `res.writableEnded` : le client a pu annuler (nouvelle frappe) pendant
// qu'un provider répondait ; écrire dans le vide lèverait une erreur.
if (res.writableEnded || res.destroyed) return;
res.write(`${JSON.stringify(line)}\n`);
};
// Group results by provider id (+ per-provider errors for diagnosable UI)
const groups = /** @type {Record<string, any[]>} */ ({});
const errors = /** @type {Record<string, { message: string, status?: number, code?: string }>} */ ({});
results.forEach((result, index) => {
const providerId = validProviders[index];
if (result.status === 'fulfilled') {
// Les providers désactivés par FF ont bien un groupe vide : le front
// affiche une pastille « aucun résultat » cohérente avec le reste.
for (const pid of Object.keys(flagErrors)) groups[pid] = [];
const errors = /** @type {Record<string, { message: string, status?: number, code?: string }>} */ ({ ...flagErrors });
// Exécution en parallèle, avec émission IMMÉDIATE de chaque groupe dès
// qu'il est résolu. L'agrégation est écrite UNE fois : le mode atomique et
// le mode NDJSON consomment exactement les mêmes objets.
await Promise.all(validProviders.map(async (providerId) => {
try {
const mod = providerRegistry[/** @type {any} */(providerId)];
const raw = (!mod || typeof mod.search !== 'function')
? []
: await mod.search(q, { limit: pageSize, page: pageNum, sort, filters });
// Filtres non supportés nativement par le provider -> affinage ici.
groups[providerId] = applySearchFilters(providerId, Array.isArray(result.value) ? result.value : [], filters);
} else {
console.warn(`Search failed for provider ${providerId}:`, result.reason?.message || result.reason);
const items = applySearchFilters(providerId, Array.isArray(raw) ? raw : [], filters);
groups[providerId] = items;
emit({ type: 'provider', provider: providerId, ok: true, items });
} catch (r) {
console.warn(`Search failed for provider ${providerId}:`, r?.message || r);
groups[providerId] = [];
try {
const r = /** @type {any} */ (result.reason);
const rr = /** @type {any} */ (r);
errors[providerId] = {
message: String(r?.message || r || 'search_failed'),
...(typeof r?.ytStatus === 'number' ? { status: r.ytStatus } : {}),
...(r?.code ? { code: String(r.code) } : {}),
message: String(rr?.message || rr || 'search_failed'),
...(typeof rr?.ytStatus === 'number' ? { status: rr.ytStatus } : {}),
...(rr?.code ? { code: String(rr.code) } : {}),
};
} catch {}
emit({ type: 'provider', provider: providerId, ok: false, error: errors[providerId] || { message: 'search_failed' } });
}
});
}));
const debug = req.query.debug === '1' || req.query.debug === 'true';
if (streamStarted) {
// Ligne de clôture : tout le contrat (version, pagination, filtres), puis
// `end`. Le front peut donc valider l'arrivée complète sans deviner quand
// le flux se termine.
emit({ type: 'done', v: SUGGESTION_CONTRACT_VERSION, q, providers: validProviders, page: pageNum, pageSize, sort, filters: activeSearchFilters(filters) });
if (!res.writableEnded) res.end();
return undefined;
}
return res.json({
// Phase 0.4 - le contrat Suggestion est versionne : le front peut
// detecter un serveur ancien et degrader proprement au lieu de lire des
// champs `undefined` sans le savoir. 2 = champs optionnels (views,
// publishedAt, avatars, channelId, …). 1 = contrat d'origine.
v: SUGGESTION_CONTRACT_VERSION,
q, providers: validProviders, groups, errors,
page: pageNum, pageSize, sort,
filters: activeSearchFilters(filters),
// Phase 7.3 - mode debug : on expose le `raw` provider tronque. Volontairement
// HORS du contrat normal : un payload brut de 6 providers ferait exploser la
// reponse et peut contenir des cles d'API internes, donc uniquement si
// demande explicitement.
...(debug ? { debug: buildDebugPayload(groups, errors) } : {}),
});
} catch (e) {
// Une coupure du client ne doit pas logger une trace d'erreur serveur : en
// flux incrémental, abandonner la requête est le comportement NORMAL.
if (streamStarted && (res.writableEnded || res.destroyed)) return undefined;
return res.status(500).json({ error: 'search_failed', details: String(e?.message || e) });
}
});
@@ -3230,7 +3420,10 @@ r.get('/search/suggest', suggestLimiter, async (req, res) => {
}
const limit = Math.min(20, Math.max(1, Number(req.query.limit || 10)));
const requested = typeof req.query.providers === 'string' ? String(req.query.providers) : '';
const validProviders = validateProviders(requested);
// Phase 8.3 : flags appliqués AVANT la clé de cache, sinon un résultat
// calculé pendant que le provider était éteint serait resservi après
// réactivation (et l'inverse). Inclus dans la clé de toute façon.
const { providerIds: validProviders, errors: flagErrors } = applyProviderFlags(validateProviders(requested));
const cacheKey = `suggest:${validProviders.join(',')}:${q.toLowerCase()}:${limit}`;
const cached = suggestCacheGet(cacheKey);
if (cached) return res.json(cached);
@@ -3250,6 +3443,9 @@ r.get('/search/suggest', suggestLimiter, async (req, res) => {
);
const [webRes, lightRes] = await Promise.allSettled([webPromise, lighthousePromise]);
const groups = {};
// Cohérence avec /api/search : un provider éteint a un groupe vide ET une
// raison, sinon le front ne peut pas distinguer « éteint » de « muet ».
for (const pid of Object.keys(flagErrors)) groups[pid] = [];
results.forEach((result, index) => {
const providerId = validProviders[index];
if (result.status === 'fulfilled' && Array.isArray(result.value)) {
@@ -3269,6 +3465,7 @@ r.get('/search/suggest', suggestLimiter, async (req, res) => {
}
// La même occurrence n'est renvoyée qu'une fois (providers d'abord, web en dernier).
const data = { q, groups: dedupeSuggestGroups(groups, [...validProviders, 'web']) };
if (Object.keys(flagErrors).length > 0) data.errors = flagErrors;
suggestCacheSet(cacheKey, data);
return res.json(data);
} catch (e) {
@@ -3760,6 +3957,7 @@ app.listen(PORT, () => {
console.log(`[newtube-api] distRoot=${distRoot} exists=${hasDistRoot}`);
console.log(`[newtube-api] distBrowser=${distBrowser} exists=${hasDistBrowser}`);
console.log(`[newtube-api] staticDir=${staticDir} indexExists=${hasIndex}`);
startSearchCacheJanitor();
});
// --- Playlists ---