mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-05 19:04:02 +00:00
style: adopt formatting for CCTV and radio provider modules
This commit is contained in:
1 parent
79b4929b88
commit
c69416ff6b
13 files changed
+862
-364
No files matched your search
@@ -2,25 +2,53 @@ import { lookup as lookupDns } from 'node:dns/promises';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
|
||||
import { readResponseTextCapped } from '../common/http.js';
|
||||
import { normalizeRadioBrowserStation, publicRadioStation, cleanRadioText } from './stations.js';
|
||||
import { radioMirrorOrigin, radioProxyDestination, resolveRadioProxyAddresses, fetchPinnedRadioResponse } from './transport.js';
|
||||
import { RADIO_DIRECTORY_CACHE_MS, RADIO_DIRECTORY_STALE_MS, RADIO_MIRROR_CACHE_MS, RADIO_FETCH_TIMEOUT_MS, RADIO_RESPONSE_MAX_BYTES, RADIO_DIRECTORY_LIMIT, RADIO_CATALOG_MIN_SUCCESSFUL_QUERIES, RADIO_CATALOG_HEALTHY_MIN_STATIONS, RADIO_USER_AGENT, RADIO_UUID_RE, RADIO_FALLBACK_MIRRORS } from './constants.js';
|
||||
import {
|
||||
normalizeRadioBrowserStation,
|
||||
publicRadioStation,
|
||||
cleanRadioText,
|
||||
} from './stations.js';
|
||||
import {
|
||||
radioMirrorOrigin,
|
||||
radioProxyDestination,
|
||||
resolveRadioProxyAddresses,
|
||||
fetchPinnedRadioResponse,
|
||||
} from './transport.js';
|
||||
import {
|
||||
RADIO_DIRECTORY_CACHE_MS,
|
||||
RADIO_DIRECTORY_STALE_MS,
|
||||
RADIO_MIRROR_CACHE_MS,
|
||||
RADIO_FETCH_TIMEOUT_MS,
|
||||
RADIO_RESPONSE_MAX_BYTES,
|
||||
RADIO_DIRECTORY_LIMIT,
|
||||
RADIO_CATALOG_MIN_SUCCESSFUL_QUERIES,
|
||||
RADIO_CATALOG_HEALTHY_MIN_STATIONS,
|
||||
RADIO_USER_AGENT,
|
||||
RADIO_UUID_RE,
|
||||
RADIO_FALLBACK_MIRRORS,
|
||||
} from './constants.js';
|
||||
export async function mapRadioConcurrent(values, concurrency, mapper) {
|
||||
const results = new Array(values.length);
|
||||
let cursor = 0;
|
||||
const workers = Array.from({ length: Math.min(concurrency, values.length) }, async () => {
|
||||
for (;;) {
|
||||
const index = cursor++;
|
||||
if (index >= values.length) return;
|
||||
results[index] = await mapper(values[index], index);
|
||||
}
|
||||
});
|
||||
const workers = Array.from(
|
||||
{ length: Math.min(concurrency, values.length) },
|
||||
async () => {
|
||||
for (;;) {
|
||||
const index = cursor++;
|
||||
if (index >= values.length) return;
|
||||
results[index] = await mapper(values[index], index);
|
||||
}
|
||||
},
|
||||
);
|
||||
await Promise.all(workers);
|
||||
return results;
|
||||
}
|
||||
|
||||
/** Create the testable Connect middleware backing `/api/radio`. */
|
||||
export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = lookupDns, now = Date.now } = {}) {
|
||||
export function createRadioProxyMiddleware({
|
||||
fetchImpl = null,
|
||||
lookupImpl = lookupDns,
|
||||
now = Date.now,
|
||||
} = {}) {
|
||||
let mirrorCache = { origins: [...RADIO_FALLBACK_MIRRORS], cachedAt: 0 };
|
||||
let mirrorPromise = null;
|
||||
let catalogCache = null;
|
||||
@@ -35,11 +63,15 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
|
||||
async function fetchJson(url, maxBytes = RADIO_RESPONSE_MAX_BYTES) {
|
||||
const destination = radioProxyDestination(url);
|
||||
if (!destination) throw new Error('Radio Browser destination is not permitted');
|
||||
if (!destination)
|
||||
throw new Error('Radio Browser destination is not permitted');
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), RADIO_FETCH_TIMEOUT_MS);
|
||||
try {
|
||||
const addresses = await resolveRadioProxyAddresses(destination.hostname, lookupImpl);
|
||||
const addresses = await resolveRadioProxyAddresses(
|
||||
destination.hostname,
|
||||
lookupImpl,
|
||||
);
|
||||
const options = {
|
||||
headers: { Accept: 'application/json', 'User-Agent': RADIO_USER_AGENT },
|
||||
signal: controller.signal,
|
||||
@@ -49,10 +81,15 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
? await fetchImpl(destination.href, options)
|
||||
: await fetchPinnedRadioResponse(destination, options, addresses);
|
||||
if (response.status >= 300 && response.status < 400) {
|
||||
try { await response.body?.cancel?.(); } catch { /* no-op */ }
|
||||
try {
|
||||
await response.body?.cancel?.();
|
||||
} catch {
|
||||
/* no-op */
|
||||
}
|
||||
throw new Error('Radio Browser redirects are refused');
|
||||
}
|
||||
if (!response.ok) throw new Error(`Radio Browser returned ${response.status}`);
|
||||
if (!response.ok)
|
||||
throw new Error(`Radio Browser returned ${response.status}`);
|
||||
const text = await readResponseTextCapped(response, maxBytes);
|
||||
return JSON.parse(text);
|
||||
} finally {
|
||||
@@ -61,20 +98,40 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
}
|
||||
|
||||
async function mirrors() {
|
||||
if (now() - mirrorCache.cachedAt < RADIO_MIRROR_CACHE_MS) return mirrorCache.origins;
|
||||
if (now() - mirrorCache.cachedAt < RADIO_MIRROR_CACHE_MS)
|
||||
return mirrorCache.origins;
|
||||
if (!mirrorPromise) {
|
||||
mirrorPromise = (async () => {
|
||||
try {
|
||||
const rows = await fetchJson('https://all.api.radio-browser.info/json/servers', 256 * 1024);
|
||||
const discovered = [...new Set((Array.isArray(rows) ? rows : []).map((row) => radioMirrorOrigin(row?.name)).filter(Boolean))];
|
||||
const rows = await fetchJson(
|
||||
'https://all.api.radio-browser.info/json/servers',
|
||||
256 * 1024,
|
||||
);
|
||||
const discovered = [
|
||||
...new Set(
|
||||
(Array.isArray(rows) ? rows : [])
|
||||
.map((row) => radioMirrorOrigin(row?.name))
|
||||
.filter(Boolean),
|
||||
),
|
||||
];
|
||||
if (discovered.length) {
|
||||
mirrorCache = { origins: [...discovered, ...RADIO_FALLBACK_MIRRORS.filter((origin) => !discovered.includes(origin))], cachedAt: now() };
|
||||
mirrorCache = {
|
||||
origins: [
|
||||
...discovered,
|
||||
...RADIO_FALLBACK_MIRRORS.filter(
|
||||
(origin) => !discovered.includes(origin),
|
||||
),
|
||||
],
|
||||
cachedAt: now(),
|
||||
};
|
||||
}
|
||||
} catch {
|
||||
mirrorCache = { ...mirrorCache, cachedAt: now() };
|
||||
}
|
||||
return mirrorCache.origins;
|
||||
})().finally(() => { mirrorPromise = null; });
|
||||
})().finally(() => {
|
||||
mirrorPromise = null;
|
||||
});
|
||||
}
|
||||
return mirrorPromise;
|
||||
}
|
||||
@@ -92,68 +149,113 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
}
|
||||
|
||||
async function refreshCatalog() {
|
||||
const queries = [null, 'news', 'talk', 'weather', 'emergency', 'scanner', 'aviation', 'marine', 'traffic'];
|
||||
const outcomes = await mapRadioConcurrent(queries, 3, async (tag, index) => {
|
||||
const params = new URLSearchParams({
|
||||
has_geo_info: 'true',
|
||||
is_https: 'true',
|
||||
hidebroken: 'true',
|
||||
order: 'clickcount',
|
||||
reverse: 'true',
|
||||
limit: index === 0 ? '1800' : '220',
|
||||
});
|
||||
if (tag) params.set('tag', tag);
|
||||
try {
|
||||
const rows = await fetchPath(`/json/stations/search?${params}`);
|
||||
if (!Array.isArray(rows)) throw new Error('Radio Browser catalog payload was not an array');
|
||||
if (!rows.every((row) => (
|
||||
row
|
||||
&& typeof row === 'object'
|
||||
&& !Array.isArray(row)
|
||||
&& typeof row.stationuuid === 'string'
|
||||
&& typeof row.name === 'string'
|
||||
&& (typeof row.url_resolved === 'string' || typeof row.url === 'string')
|
||||
))) throw new Error('Radio Browser catalog contained a malformed station row');
|
||||
const stations = rows.map(normalizeRadioBrowserStation).filter(Boolean);
|
||||
const requestedTag = cleanRadioText(tag, 80)
|
||||
.toLocaleLowerCase()
|
||||
.replace(/[_-]+/g, ' ')
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim();
|
||||
const requestedTagCovered = !requestedTag || stations.some((station) => (
|
||||
station.tags.some((stationTag) => stationTag === requestedTag || stationTag.includes(requestedTag))
|
||||
));
|
||||
return {
|
||||
// Query coverage is based on accepted rows, not merely a payload that
|
||||
// happens to match the upstream schema. Specialist responses must
|
||||
// also contain an accepted station tagged for the requested category.
|
||||
succeeded: stations.length > 0 && requestedTagCovered,
|
||||
stations,
|
||||
};
|
||||
} catch {
|
||||
return { succeeded: false, stations: [] };
|
||||
}
|
||||
});
|
||||
const queries = [
|
||||
null,
|
||||
'news',
|
||||
'talk',
|
||||
'weather',
|
||||
'emergency',
|
||||
'scanner',
|
||||
'aviation',
|
||||
'marine',
|
||||
'traffic',
|
||||
];
|
||||
const outcomes = await mapRadioConcurrent(
|
||||
queries,
|
||||
3,
|
||||
async (tag, index) => {
|
||||
const params = new URLSearchParams({
|
||||
has_geo_info: 'true',
|
||||
is_https: 'true',
|
||||
hidebroken: 'true',
|
||||
order: 'clickcount',
|
||||
reverse: 'true',
|
||||
limit: index === 0 ? '1800' : '220',
|
||||
});
|
||||
if (tag) params.set('tag', tag);
|
||||
try {
|
||||
const rows = await fetchPath(`/json/stations/search?${params}`);
|
||||
if (!Array.isArray(rows))
|
||||
throw new Error('Radio Browser catalog payload was not an array');
|
||||
if (
|
||||
!rows.every(
|
||||
(row) =>
|
||||
row &&
|
||||
typeof row === 'object' &&
|
||||
!Array.isArray(row) &&
|
||||
typeof row.stationuuid === 'string' &&
|
||||
typeof row.name === 'string' &&
|
||||
(typeof row.url_resolved === 'string' ||
|
||||
typeof row.url === 'string'),
|
||||
)
|
||||
)
|
||||
throw new Error(
|
||||
'Radio Browser catalog contained a malformed station row',
|
||||
);
|
||||
const stations = rows
|
||||
.map(normalizeRadioBrowserStation)
|
||||
.filter(Boolean);
|
||||
const requestedTag = cleanRadioText(tag, 80)
|
||||
.toLocaleLowerCase()
|
||||
.replace(/[_-]+/g, ' ')
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim();
|
||||
const requestedTagCovered =
|
||||
!requestedTag ||
|
||||
stations.some((station) =>
|
||||
station.tags.some(
|
||||
(stationTag) =>
|
||||
stationTag === requestedTag ||
|
||||
stationTag.includes(requestedTag),
|
||||
),
|
||||
);
|
||||
return {
|
||||
// Query coverage is based on accepted rows, not merely a payload that
|
||||
// happens to match the upstream schema. Specialist responses must
|
||||
// also contain an accepted station tagged for the requested category.
|
||||
succeeded: stations.length > 0 && requestedTagCovered,
|
||||
stations,
|
||||
};
|
||||
} catch {
|
||||
return { succeeded: false, stations: [] };
|
||||
}
|
||||
},
|
||||
);
|
||||
const resultSets = outcomes.map((outcome) => outcome.stations);
|
||||
|
||||
const selected = [];
|
||||
const seen = new Set();
|
||||
const take = (station) => {
|
||||
if (!station || seen.has(station.id) || selected.length >= RADIO_DIRECTORY_LIMIT) return;
|
||||
if (
|
||||
!station ||
|
||||
seen.has(station.id) ||
|
||||
selected.length >= RADIO_DIRECTORY_LIMIT
|
||||
)
|
||||
return;
|
||||
seen.add(station.id);
|
||||
selected.push(station);
|
||||
};
|
||||
// Seed specialist station-tag queries before popularity fill so operational
|
||||
// categories remain represented even when global click charts skew musical.
|
||||
for (const rows of resultSets.slice(1)) rows.slice(0, 45).forEach(take);
|
||||
resultSets.flat().sort((a, b) => b.clickCount - a.clickCount || a.name.localeCompare(b.name)).forEach(take);
|
||||
resultSets
|
||||
.flat()
|
||||
.sort(
|
||||
(a, b) => b.clickCount - a.clickCount || a.name.localeCompare(b.name),
|
||||
)
|
||||
.forEach(take);
|
||||
const timestamp = now();
|
||||
const successfulQueries = outcomes.filter((outcome) => outcome.succeeded).length;
|
||||
const broadQueryHealthy = outcomes[0].succeeded && outcomes[0].stations.length > 0;
|
||||
const successfulQueries = outcomes.filter(
|
||||
(outcome) => outcome.succeeded,
|
||||
).length;
|
||||
const broadQueryHealthy =
|
||||
outcomes[0].succeeded && outcomes[0].stations.length > 0;
|
||||
const healthReasons = [];
|
||||
if (!broadQueryHealthy) healthReasons.push('broad-query-unhealthy');
|
||||
if (successfulQueries < RADIO_CATALOG_MIN_SUCCESSFUL_QUERIES) healthReasons.push('query-coverage-below-policy');
|
||||
if (selected.length < RADIO_CATALOG_HEALTHY_MIN_STATIONS) healthReasons.push('station-coverage-below-policy');
|
||||
if (successfulQueries < RADIO_CATALOG_MIN_SUCCESSFUL_QUERIES)
|
||||
healthReasons.push('query-coverage-below-policy');
|
||||
if (selected.length < RADIO_CATALOG_HEALTHY_MIN_STATIONS)
|
||||
healthReasons.push('station-coverage-below-policy');
|
||||
const degraded = healthReasons.length > 0;
|
||||
const coverage = {
|
||||
successfulQueries,
|
||||
@@ -171,14 +273,18 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
coverage,
|
||||
};
|
||||
if (degraded && catalogCache) {
|
||||
const error = new Error('Radio Browser catalog refresh did not meet health policy');
|
||||
const error = new Error(
|
||||
'Radio Browser catalog refresh did not meet health policy',
|
||||
);
|
||||
error.radioCatalogDegraded = true;
|
||||
error.radioDegradedReason = nextCatalog.degradedReason;
|
||||
error.radioCoverage = coverage;
|
||||
throw error;
|
||||
}
|
||||
if (degraded && !selected.length) {
|
||||
const error = new Error('Radio Browser catalog refresh returned no usable stations');
|
||||
const error = new Error(
|
||||
'Radio Browser catalog refresh returned no usable stations',
|
||||
);
|
||||
error.radioCatalogDegraded = true;
|
||||
error.radioDegradedReason = nextCatalog.degradedReason;
|
||||
error.radioCoverage = coverage;
|
||||
@@ -197,16 +303,24 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
}
|
||||
|
||||
async function getCatalog() {
|
||||
if (catalogCache && now() - catalogCache.cachedAt < RADIO_DIRECTORY_CACHE_MS) {
|
||||
if (
|
||||
catalogCache &&
|
||||
now() - catalogCache.cachedAt < RADIO_DIRECTORY_CACHE_MS
|
||||
) {
|
||||
return { ...catalogCache, stale: false };
|
||||
}
|
||||
if (!refreshPromise) {
|
||||
refreshPromise = refreshCatalog().finally(() => { refreshPromise = null; });
|
||||
refreshPromise = refreshCatalog().finally(() => {
|
||||
refreshPromise = null;
|
||||
});
|
||||
}
|
||||
try {
|
||||
return { ...await refreshPromise, stale: false };
|
||||
return { ...(await refreshPromise), stale: false };
|
||||
} catch (error) {
|
||||
if (catalogCache && now() - catalogCache.cachedAt <= RADIO_DIRECTORY_STALE_MS) {
|
||||
if (
|
||||
catalogCache &&
|
||||
now() - catalogCache.cachedAt <= RADIO_DIRECTORY_STALE_MS
|
||||
) {
|
||||
return {
|
||||
...catalogCache,
|
||||
stale: true,
|
||||
@@ -220,7 +334,10 @@ export function createRadioProxyMiddleware({ fetchImpl = null, lookupImpl = look
|
||||
}
|
||||
|
||||
function sendJson(res, status, body) {
|
||||
res.writeHead(status, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store' });
|
||||
res.writeHead(status, {
|
||||
'Content-Type': 'application/json; charset=utf-8',
|
||||
'Cache-Control': 'no-store',
|
||||
});
|
||||
res.end(JSON.stringify(body));
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user