// js/feedEngine/sources.js import nostrService from "../services/nostrService.js"; import watchHistoryService from "../watchHistoryService.js"; import { isPlainObject, toArray, toSet } from "./utils.js"; function resolveService(candidate, fallback) { if (candidate && typeof candidate === "object") { return candidate; } return fallback; } function normalizeAuthor(value) { if (typeof value !== "string") { return ""; } const trimmed = value.trim(); return trimmed ? trimmed.toLowerCase() : ""; } function normalizeActorCandidate(...values) { for (const value of values) { if (typeof value === "string") { const trimmed = value.trim(); if (trimmed) { return trimmed; } } } return ""; } export function createActiveNostrSource({ service } = {}) { const resolvedService = resolveService(service, nostrService); return async function activeNostrSource(context = {}) { const options = { blacklistedEventIds: toSet(context?.runtime?.blacklistedEventIds), isAuthorBlocked: typeof context?.runtime?.isAuthorBlocked === "function" ? context.runtime.isAuthorBlocked : () => false, }; let videos = []; try { videos = await Promise.resolve( resolvedService.getFilteredActiveVideos(options) ); } catch (error) { context?.log?.("[active-source] Failed to resolve active videos", error); videos = []; } return (Array.isArray(videos) ? videos : []).map((video) => ({ video, metadata: { source: "nostr:active", }, })); }; } // Bounded discovery source for curated surfaces such as Most Zapped. It never // falls back to an unscoped relay request: the caller must provide approved // creator pubkeys in runtime.whitelistedAuthors. The successful remote result // is retained briefly so an asynchronous ranking update does not turn into a // second full relay scan for the same author set. export function createWhitelistedAuthorsNostrSource({ service, refreshTtlMs = 2 * 60 * 1000, now = () => Date.now(), } = {}) { const resolvedService = resolveService(service, nostrService); let cachedAuthorsKey = ""; let cachedVideos = []; let fetchedAt = 0; let inFlight = null; const normalizeAuthors = (input) => Array.from( new Set( toArray(input) .map(normalizeAuthor) .filter(Boolean), ), ).sort(); return async function whitelistedAuthorsNostrSource(context = {}) { const authors = normalizeAuthors(context?.runtime?.whitelistedAuthors); const authorsKey = authors.join(","); if (!authors.length) { cachedAuthorsKey = ""; cachedVideos = []; fetchedAt = 0; return []; } const cacheIsCurrent = authorsKey === cachedAuthorsKey && now() - fetchedAt < Math.max(0, Number(refreshTtlMs) || 0); if (!cacheIsCurrent && !inFlight) { const fetchByAuthors = resolvedService?.fetchVideosByAuthors; inFlight = Promise.resolve( typeof fetchByAuthors === "function" ? fetchByAuthors.call(resolvedService, authors) : [], ) .then((videos) => { const allowed = new Set(authors); cachedAuthorsKey = authorsKey; cachedVideos = (Array.isArray(videos) ? videos : []).filter( (video) => allowed.has(normalizeAuthor(video?.pubkey)), ); fetchedAt = now(); return cachedVideos; }) .catch((error) => { context?.log?.( "[whitelisted-authors-source] Failed to fetch approved creators", error, ); return authorsKey === cachedAuthorsKey ? cachedVideos : []; }) .finally(() => { inFlight = null; }); } const videos = cacheIsCurrent ? cachedVideos : await inFlight; // The whitelist may change while a prior author-scoped request is still in // flight. Filter again at return time so that stale results can never // briefly render after an administrator removes a creator. const allowed = new Set(authors); return (Array.isArray(videos) ? videos : []) .filter((video) => allowed.has(normalizeAuthor(video?.pubkey))) .map((video) => ({ video, metadata: { source: "nostr:whitelisted-authors", }, })); }; } export function createSubscriptionAuthorsSource({ service } = {}) { const resolvedService = resolveService(service, nostrService); return async function subscriptionAuthorsSource(context = {}) { context?.log?.("[subscriptions-source] START"); const runtimeAuthors = toArray( context?.runtime?.subscriptionAuthors || context?.runtime?.authors ).map(normalizeAuthor); const configAuthors = toArray(context?.config?.actorFilters).map( normalizeAuthor ); const hookAuthors = []; const hook = context?.hooks?.subscriptions; if (isPlainObject(hook) && typeof hook.resolveAuthors === "function") { try { const resolved = await hook.resolveAuthors(context); hookAuthors.push(...toArray(resolved).map(normalizeAuthor)); } catch (error) { context?.log?.("[subscriptions-source] resolveAuthors threw", error); } } const authors = new Set( [...runtimeAuthors, ...configAuthors, ...hookAuthors].filter(Boolean) ); context?.log?.(`[subscriptions-source] Resolved ${authors.size} authors.`); if (!authors.size) { return []; } const limitCandidate = Number(context?.runtime?.limit); const limit = Number.isFinite(limitCandidate) && limitCandidate > 0 ? Math.floor(limitCandidate) : null; const options = { blacklistedEventIds: toSet(context?.runtime?.blacklistedEventIds), isAuthorBlocked: typeof context?.runtime?.isAuthorBlocked === "function" ? context.runtime.isAuthorBlocked : () => false, limit, }; const authorList = Array.from(authors); const hasTargetedLookup = resolvedService && typeof resolvedService.getActiveVideosByAuthors === "function"; let videos = []; try { if (hasTargetedLookup) { context?.log?.("[subscriptions-source] Calling getActiveVideosByAuthors..."); videos = await Promise.resolve( resolvedService.getActiveVideosByAuthors(authorList, options) ); context?.log?.(`[subscriptions-source] getActiveVideosByAuthors returned ${videos.length} videos.`); if (!videos.length && typeof resolvedService.fetchVideosByAuthors === "function") { context?.log?.("[subscriptions-source] Cache empty. Calling fetchVideosByAuthors..."); const fetched = await resolvedService.fetchVideosByAuthors(authorList, options); context?.log?.(`[subscriptions-source] fetchVideosByAuthors returned ${fetched ? fetched.length : 0} videos.`); if (Array.isArray(fetched) && fetched.length > 0) { videos = fetched; } } } else { videos = await Promise.resolve( resolvedService.getFilteredActiveVideos(options) ); } } catch (error) { context?.log?.( "[subscriptions-source] Failed to resolve videos from nostrService", error ); videos = []; } const filtered = (Array.isArray(videos) ? videos : []).filter((video) => { const author = normalizeAuthor(video?.pubkey); return author && authors.has(author); }); const sorted = filtered .slice() .sort( (a, b) => (Number(b?.created_at) || 0) - (Number(a?.created_at) || 0) ); const limited = limit ? sorted.slice(0, limit) : sorted; return limited.map((video) => ({ video, metadata: { source: "nostr:subscriptions", matchedAuthor: normalizeAuthor(video?.pubkey), }, })); }; } export function createWatchHistoryPointerSource({ service } = {}) { const resolvedService = resolveService(service, watchHistoryService); return async function watchHistoryPointerSource(context = {}) { const actor = normalizeActorCandidate( context?.config?.actor, context?.runtime?.watchHistory?.actor, context?.runtime?.actor ); if (!resolvedService || typeof resolvedService.getQueuedPointers !== "function") { return []; } let pointers = []; try { pointers = await Promise.resolve( resolvedService.getQueuedPointers(actor || undefined) ); } catch (error) { context?.log?.( "[watch-history-source] Failed to load queued pointers", error ); pointers = []; } const results = []; const hook = context?.hooks?.watchHistory; const resolveVideoHook = isPlainObject(hook) && typeof hook.resolveVideo === "function" ? hook.resolveVideo : null; for (const pointer of Array.isArray(pointers) ? pointers : []) { const dto = { video: null, pointer, metadata: { source: "watch-history", actor: actor || null, }, }; if (pointer && isPlainObject(pointer) && pointer.video) { dto.video = pointer.video; } if (!dto.video && resolveVideoHook) { try { const resolvedVideo = await resolveVideoHook(pointer, context); if (resolvedVideo) { dto.video = resolvedVideo; } } catch (error) { context?.log?.( "[watch-history-source] resolveVideo hook threw", error ); } } results.push(dto); } return results; }; }