mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-30 17:32:26 +00:00
Merge pull request #282 from PR0M3TH3AN/codex/add-feed-engine-module-with-features
Add feed engine module and baseline adapters
This commit is contained in:
8 files changed
+987
No files matched your search
@@ -0,0 +1,111 @@
|
||||
# Feed Engine Architecture
|
||||
|
||||
The feed engine is a lightweight pipeline for composing Bitvid feeds from one
|
||||
or more data sources. It lives in `js/feedEngine/` and exposes factories for
|
||||
sources, stages, sorters, and the engine itself. The goal is to make it easy to
|
||||
register new feeds today while leaving room for future "open algorithm" work
|
||||
that surfaces optional "why this video" metadata.
|
||||
|
||||
## Core Concepts
|
||||
|
||||
- **Feed definition** – A feed consists of a source, zero or more pipeline
|
||||
stages, an optional sorter, and optional decorators. Each feed can expose its
|
||||
own hooks and configuration defaults.
|
||||
- **Feed items** – The engine normalizes everything into a `{ video, pointer?,
|
||||
metadata }` DTO so stages can reason about inputs without caring about the
|
||||
original source.
|
||||
- **Context** – Every source and stage receives a context object containing the
|
||||
feed configuration, runtime helpers, registered hooks, and a `addWhy()`
|
||||
method for collecting metadata about why an item was filtered or prioritised.
|
||||
|
||||
## Getting Started
|
||||
|
||||
```js
|
||||
import {
|
||||
createFeedEngine,
|
||||
createActiveNostrSource,
|
||||
createDedupeByRootStage,
|
||||
createBlacklistFilterStage,
|
||||
createChronologicalSorter,
|
||||
} from "../js/feedEngine/index.js";
|
||||
|
||||
const engine = createFeedEngine();
|
||||
|
||||
engine.registerFeed("recent", {
|
||||
source: createActiveNostrSource(),
|
||||
stages: [
|
||||
createBlacklistFilterStage(),
|
||||
createDedupeByRootStage(),
|
||||
],
|
||||
sorter: createChronologicalSorter(),
|
||||
});
|
||||
|
||||
const { videos, metadata } = await engine.runFeed("recent", {
|
||||
runtime: {
|
||||
blacklistedEventIds: new Set(["..."]),
|
||||
isAuthorBlocked: (pubkey) => false,
|
||||
},
|
||||
});
|
||||
```
|
||||
|
||||
`metadata.why` collects the audit trail from each stage so later phases of the
|
||||
open algorithm project can surface transparency UI.
|
||||
|
||||
## Sources
|
||||
|
||||
| Factory | Description |
|
||||
| --- | --- |
|
||||
| `createActiveNostrSource` | Wraps `nostrService.getFilteredActiveVideos(...)` and emits DTOs with a `metadata.source` of `nostr:active`. |
|
||||
| `createSubscriptionAuthorsSource` | Filters the active video list down to subscribed authors using runtime hooks or `config.actorFilters`. |
|
||||
| `createWatchHistoryPointerSource` | Loads pointer DTOs from `watchHistoryService.getQueuedPointers(...)`. Optional hooks can resolve the backing video if a pointer is missing it. |
|
||||
|
||||
Every source resolves blacklist and author-block runtime helpers so feeds
|
||||
constructed today behave exactly like the existing UI.
|
||||
|
||||
## Stages
|
||||
|
||||
| Stage | Behavior |
|
||||
| --- | --- |
|
||||
| `createDedupeByRootStage` | Reuses the application’s `dedupeVideosByRoot` helper (falling back to `dedupeToNewestByRoot`) to drop older versions of the same videoRoot. Adds "why" metadata for removed entries. |
|
||||
| `createBlacklistFilterStage` | Calls `nostrService.shouldIncludeVideo(...)` so moderators and block lists stay enforced. Each rejection logs a "blacklist" reason in the why-trail. |
|
||||
| `createWatchHistorySuppressionStage` | Invokes feed-provided hooks to optionally suppress watched items. Useful for per-feed watch history preferences. |
|
||||
|
||||
Stages receive `(items, context)` and should return the transformed list. They
|
||||
can rely on `context.addWhy(...)` to annotate decisions without mutating the
|
||||
items in place.
|
||||
|
||||
## Sorting & Decorators
|
||||
|
||||
`createChronologicalSorter` is the baseline sorter that orders DTOs by
|
||||
`video.created_at` (newest first by default). Additional decorators can run
|
||||
after sorting to attach extra metadata or inject presentation hints.
|
||||
|
||||
## Configuration Hooks
|
||||
|
||||
Every feed inherits the default config contract:
|
||||
|
||||
```json
|
||||
{
|
||||
"timeWindow": null,
|
||||
"actorFilters": [],
|
||||
"tagFilters": [],
|
||||
"sortOrder": "recent"
|
||||
}
|
||||
```
|
||||
|
||||
Feeds can override defaults or expose richer schemas through
|
||||
`definition.defaultConfig` and `definition.configSchema`. At execution time the
|
||||
engine merges `options.config` with those defaults and passes them to the
|
||||
pipeline via `context.config`.
|
||||
|
||||
Hooks can be provided globally when registering the feed or per-execution via
|
||||
`engine.runFeed(name, { hooks: { ... } })`. The watch-history suppression stage
|
||||
uses this mechanism so Phase 1 feeds can plug in actor-specific suppression
|
||||
logic later without changing the core pipeline.
|
||||
|
||||
## Why-Metadata
|
||||
|
||||
`context.addWhy()` records structured audit entries. All built-in stages use it
|
||||
for dedupe drops, blacklist filtering, and watch-history suppression. The
|
||||
engine returns these records alongside the final video list so UI components can
|
||||
render transparency affordances when the open algorithm effort ships.
|
||||
@@ -0,0 +1,232 @@
|
||||
// js/feedEngine/engine.js
|
||||
|
||||
import { isPlainObject } from "./utils.js";
|
||||
|
||||
const DEFAULT_FEED_CONFIG = Object.freeze({
|
||||
timeWindow: null,
|
||||
actorFilters: [],
|
||||
tagFilters: [],
|
||||
sortOrder: "recent",
|
||||
});
|
||||
|
||||
const DEFAULT_CONFIG_SCHEMA = Object.freeze({
|
||||
timeWindow: {
|
||||
type: "relative-window",
|
||||
description:
|
||||
"Restrict results to a rolling time window (e.g., last 24 hours).",
|
||||
default: null,
|
||||
},
|
||||
actorFilters: {
|
||||
type: "string[]",
|
||||
description: "Optional list of author pubkeys to include in the feed.",
|
||||
default: [],
|
||||
},
|
||||
tagFilters: {
|
||||
type: "string[]",
|
||||
description: "Optional list of tag identifiers to include in the feed.",
|
||||
default: [],
|
||||
},
|
||||
sortOrder: {
|
||||
type: "enum",
|
||||
values: ["recent"],
|
||||
description:
|
||||
"Controls the final ordering of the feed. Currently only 'recent' is implemented.",
|
||||
default: "recent",
|
||||
},
|
||||
});
|
||||
|
||||
function normalizeLogger(logger) {
|
||||
if (typeof logger === "function") {
|
||||
return logger;
|
||||
}
|
||||
if (logger && typeof logger.log === "function") {
|
||||
return (...args) => logger.log(...args);
|
||||
}
|
||||
return () => {};
|
||||
}
|
||||
|
||||
function normalizeArray(value) {
|
||||
if (!Array.isArray(value)) {
|
||||
return [];
|
||||
}
|
||||
return value.filter((entry) => typeof entry === "function");
|
||||
}
|
||||
|
||||
function normalizeDto(candidate) {
|
||||
if (!candidate || typeof candidate !== "object") {
|
||||
return null;
|
||||
}
|
||||
|
||||
const video = candidate.video ?? null;
|
||||
const pointer = candidate.pointer ?? null;
|
||||
const metadata = isPlainObject(candidate.metadata)
|
||||
? { ...candidate.metadata }
|
||||
: {};
|
||||
|
||||
return { video, pointer, metadata };
|
||||
}
|
||||
|
||||
function normalizeItems(items) {
|
||||
if (!Array.isArray(items)) {
|
||||
return [];
|
||||
}
|
||||
const normalized = [];
|
||||
for (const candidate of items) {
|
||||
const dto = normalizeDto(candidate);
|
||||
if (dto) {
|
||||
normalized.push(dto);
|
||||
}
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
function createExecutionContext(entry, { config, hooks, runtime, logger }) {
|
||||
const whyLog = [];
|
||||
const resolvedLogger = normalizeLogger(logger);
|
||||
|
||||
const context = {
|
||||
feedName: entry.name,
|
||||
config: { ...entry.defaultConfig, ...(isPlainObject(config) ? config : {}) },
|
||||
hooks: isPlainObject(hooks) ? { ...entry.hooks, ...hooks } : { ...entry.hooks },
|
||||
runtime: isPlainObject(runtime) ? { ...runtime } : {},
|
||||
log: (...args) => {
|
||||
try {
|
||||
resolvedLogger(`[feed:${entry.name}]`, ...args);
|
||||
} catch (error) {
|
||||
console.warn(`[feed:${entry.name}] logger threw`, error);
|
||||
}
|
||||
},
|
||||
addWhy: (detail) => {
|
||||
if (!isPlainObject(detail)) {
|
||||
return;
|
||||
}
|
||||
const record = { feed: entry.name, ...detail };
|
||||
whyLog.push(record);
|
||||
return record;
|
||||
},
|
||||
getWhy: () => whyLog.slice(),
|
||||
};
|
||||
|
||||
return context;
|
||||
}
|
||||
|
||||
function normalizeDefinition(name, definition = {}) {
|
||||
if (typeof name !== "string" || !name.trim()) {
|
||||
throw new Error("Feed name must be a non-empty string");
|
||||
}
|
||||
|
||||
if (typeof definition.source !== "function") {
|
||||
throw new Error(`Feed \"${name}\" is missing a source function`);
|
||||
}
|
||||
|
||||
const stages = normalizeArray(definition.stages);
|
||||
const decorators = normalizeArray(definition.decorators);
|
||||
const sorter =
|
||||
typeof definition.sorter === "function" ? definition.sorter : null;
|
||||
const hooks = isPlainObject(definition.hooks) ? { ...definition.hooks } : {};
|
||||
const defaultConfig = {
|
||||
...DEFAULT_FEED_CONFIG,
|
||||
...(isPlainObject(definition.defaultConfig) ? definition.defaultConfig : {}),
|
||||
};
|
||||
const configSchema = isPlainObject(definition.configSchema)
|
||||
? { ...DEFAULT_CONFIG_SCHEMA, ...definition.configSchema }
|
||||
: { ...DEFAULT_CONFIG_SCHEMA };
|
||||
|
||||
return {
|
||||
name,
|
||||
source: definition.source,
|
||||
stages,
|
||||
sorter,
|
||||
decorators,
|
||||
hooks,
|
||||
defaultConfig,
|
||||
configSchema,
|
||||
publicDefinition: Object.freeze({
|
||||
name,
|
||||
configDefaults: { ...defaultConfig },
|
||||
configSchema: { ...configSchema },
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
export function createFeedEngine({ logger } = {}) {
|
||||
const feeds = new Map();
|
||||
const resolvedLogger = normalizeLogger(logger);
|
||||
|
||||
function registerFeed(name, definition) {
|
||||
if (feeds.has(name)) {
|
||||
throw new Error(`Feed \"${name}\" is already registered`);
|
||||
}
|
||||
|
||||
const entry = normalizeDefinition(name, definition);
|
||||
feeds.set(name, entry);
|
||||
return entry.publicDefinition;
|
||||
}
|
||||
|
||||
async function runFeed(name, options = {}) {
|
||||
const entry = feeds.get(name);
|
||||
if (!entry) {
|
||||
throw new Error(`Feed \"${name}\" is not registered`);
|
||||
}
|
||||
|
||||
const context = createExecutionContext(entry, {
|
||||
config: options.config,
|
||||
hooks: options.hooks,
|
||||
runtime: options.runtime,
|
||||
logger: resolvedLogger,
|
||||
});
|
||||
|
||||
const sourceResult = await entry.source(context);
|
||||
let items = normalizeItems(await Promise.resolve(sourceResult));
|
||||
|
||||
for (const stage of entry.stages) {
|
||||
const result = await stage(items, context);
|
||||
if (Array.isArray(result)) {
|
||||
items = normalizeItems(result);
|
||||
} else if (result == null) {
|
||||
items = normalizeItems(items);
|
||||
}
|
||||
}
|
||||
|
||||
if (entry.sorter) {
|
||||
const sorted = await entry.sorter(items, context);
|
||||
if (Array.isArray(sorted)) {
|
||||
items = normalizeItems(sorted);
|
||||
}
|
||||
}
|
||||
|
||||
for (const decorator of entry.decorators) {
|
||||
const decorated = await decorator(items, context);
|
||||
if (Array.isArray(decorated)) {
|
||||
items = normalizeItems(decorated);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
items,
|
||||
videos: items.map((item) => item.video).filter(Boolean),
|
||||
metadata: {
|
||||
why: context.getWhy(),
|
||||
config: context.config,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function listFeeds() {
|
||||
return Array.from(feeds.values()).map((entry) => entry.publicDefinition);
|
||||
}
|
||||
|
||||
function getFeedDefinition(name) {
|
||||
const entry = feeds.get(name);
|
||||
return entry ? entry.publicDefinition : null;
|
||||
}
|
||||
|
||||
return {
|
||||
registerFeed,
|
||||
runFeed,
|
||||
listFeeds,
|
||||
getFeedDefinition,
|
||||
};
|
||||
}
|
||||
|
||||
export { DEFAULT_FEED_CONFIG, DEFAULT_CONFIG_SCHEMA };
|
||||
@@ -0,0 +1,18 @@
|
||||
// js/feedEngine/index.js
|
||||
|
||||
export {
|
||||
createFeedEngine,
|
||||
DEFAULT_FEED_CONFIG,
|
||||
DEFAULT_CONFIG_SCHEMA,
|
||||
} from "./engine.js";
|
||||
export {
|
||||
createDedupeByRootStage,
|
||||
createBlacklistFilterStage,
|
||||
createWatchHistorySuppressionStage,
|
||||
} from "./stages.js";
|
||||
export { createChronologicalSorter } from "./sorters.js";
|
||||
export {
|
||||
createActiveNostrSource,
|
||||
createSubscriptionAuthorsSource,
|
||||
createWatchHistoryPointerSource,
|
||||
} from "./sources.js";
|
||||
@@ -0,0 +1,32 @@
|
||||
// js/feedEngine/sorters.js
|
||||
|
||||
export function createChronologicalSorter({ direction = "desc" } = {}) {
|
||||
const normalizedDirection = direction === "asc" ? "asc" : "desc";
|
||||
|
||||
return function chronologicalSorter(items = []) {
|
||||
if (!Array.isArray(items)) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const copy = [...items];
|
||||
copy.sort((a, b) => {
|
||||
const aTs = Number(a?.video?.created_at);
|
||||
const bTs = Number(b?.video?.created_at);
|
||||
const normalizedATs = Number.isFinite(aTs) ? aTs : Number.NEGATIVE_INFINITY;
|
||||
const normalizedBTs = Number.isFinite(bTs) ? bTs : Number.NEGATIVE_INFINITY;
|
||||
|
||||
if (normalizedATs !== normalizedBTs) {
|
||||
const diff = normalizedATs - normalizedBTs;
|
||||
return normalizedDirection === "asc" ? diff : -diff;
|
||||
}
|
||||
|
||||
const aId = typeof a?.video?.id === "string" ? a.video.id : "";
|
||||
const bId = typeof b?.video?.id === "string" ? b.video.id : "";
|
||||
return normalizedDirection === "asc"
|
||||
? aId.localeCompare(bId)
|
||||
: bId.localeCompare(aId);
|
||||
});
|
||||
|
||||
return copy;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
// 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;
|
||||
}
|
||||
|
||||
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",
|
||||
},
|
||||
}));
|
||||
};
|
||||
}
|
||||
|
||||
export function createSubscriptionAuthorsSource({ service } = {}) {
|
||||
const resolvedService = resolveService(service, nostrService);
|
||||
|
||||
return async function subscriptionAuthorsSource(context = {}) {
|
||||
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)
|
||||
);
|
||||
|
||||
if (!authors.size) {
|
||||
return [];
|
||||
}
|
||||
|
||||
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?.(
|
||||
"[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);
|
||||
});
|
||||
|
||||
return filtered.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;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
// js/feedEngine/stages.js
|
||||
|
||||
import { getApplication } from "../applicationContext.js";
|
||||
import nostrService from "../services/nostrService.js";
|
||||
import { dedupeToNewestByRoot } from "../utils/videoDeduper.js";
|
||||
import { isPlainObject, toSet } from "./utils.js";
|
||||
|
||||
function resolveDedupeFunction(customDedupe) {
|
||||
if (typeof customDedupe === "function") {
|
||||
return customDedupe;
|
||||
}
|
||||
|
||||
const app = getApplication?.();
|
||||
if (app && typeof app.dedupeVideosByRoot === "function") {
|
||||
return (videos) => app.dedupeVideosByRoot(videos);
|
||||
}
|
||||
|
||||
return (videos) => dedupeToNewestByRoot(videos);
|
||||
}
|
||||
|
||||
export function createDedupeByRootStage({
|
||||
stageName = "dedupe-by-root",
|
||||
dedupe: customDedupe,
|
||||
} = {}) {
|
||||
const dedupeFn = resolveDedupeFunction(customDedupe);
|
||||
|
||||
return async function dedupeByRootStage(items = [], context = {}) {
|
||||
const videos = items.map((item) => item?.video).filter(Boolean);
|
||||
const dedupedVideos = dedupeFn(videos) || [];
|
||||
const allowedIds = new Set();
|
||||
|
||||
for (const video of dedupedVideos) {
|
||||
if (video && typeof video.id === "string") {
|
||||
allowedIds.add(video.id);
|
||||
}
|
||||
}
|
||||
|
||||
if (!allowedIds.size) {
|
||||
return items;
|
||||
}
|
||||
|
||||
const keep = [];
|
||||
for (const item of items) {
|
||||
const video = item?.video;
|
||||
const videoId = video && typeof video.id === "string" ? video.id : "";
|
||||
if (!videoId || allowedIds.has(videoId)) {
|
||||
keep.push(item);
|
||||
continue;
|
||||
}
|
||||
|
||||
const rootId =
|
||||
typeof video?.videoRootId === "string" && video.videoRootId
|
||||
? video.videoRootId
|
||||
: videoId;
|
||||
|
||||
if (typeof item?.metadata === "object" && item.metadata) {
|
||||
item.metadata.droppedByStage = stageName;
|
||||
}
|
||||
|
||||
context?.addWhy?.({
|
||||
stage: stageName,
|
||||
type: "dedupe",
|
||||
reason: "older-root-version",
|
||||
videoId,
|
||||
rootId,
|
||||
});
|
||||
}
|
||||
|
||||
return keep;
|
||||
};
|
||||
}
|
||||
|
||||
export function createBlacklistFilterStage({
|
||||
stageName = "blacklist-filter",
|
||||
shouldIncludeVideo,
|
||||
} = {}) {
|
||||
const includeFn =
|
||||
typeof shouldIncludeVideo === "function"
|
||||
? shouldIncludeVideo
|
||||
: (video, options) => nostrService.shouldIncludeVideo(video, options);
|
||||
|
||||
return async function blacklistFilterStage(items = [], context = {}) {
|
||||
const blacklist = toSet(context?.runtime?.blacklistedEventIds);
|
||||
const isAuthorBlocked =
|
||||
typeof context?.runtime?.isAuthorBlocked === "function"
|
||||
? context.runtime.isAuthorBlocked
|
||||
: () => false;
|
||||
|
||||
const options = { blacklistedEventIds: blacklist, isAuthorBlocked };
|
||||
const results = [];
|
||||
|
||||
for (const item of items) {
|
||||
const video = item?.video;
|
||||
if (!video || typeof video !== "object") {
|
||||
results.push(item);
|
||||
continue;
|
||||
}
|
||||
|
||||
let include = false;
|
||||
try {
|
||||
include = includeFn(video, options) !== false;
|
||||
} catch (error) {
|
||||
context?.log?.(`[${stageName}] shouldIncludeVideo threw`, error);
|
||||
include = false;
|
||||
}
|
||||
|
||||
if (!include) {
|
||||
context?.addWhy?.({
|
||||
stage: stageName,
|
||||
type: "filter",
|
||||
reason: "blacklist",
|
||||
videoId: typeof video.id === "string" ? video.id : null,
|
||||
pubkey: typeof video.pubkey === "string" ? video.pubkey : null,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
results.push(item);
|
||||
}
|
||||
|
||||
return results;
|
||||
};
|
||||
}
|
||||
|
||||
function resolveSuppressionHook(context) {
|
||||
if (!context || typeof context !== "object") {
|
||||
return null;
|
||||
}
|
||||
|
||||
const fromHooks = context?.hooks?.watchHistory;
|
||||
if (isPlainObject(fromHooks) && typeof fromHooks.shouldSuppress === "function") {
|
||||
return fromHooks.shouldSuppress;
|
||||
}
|
||||
|
||||
const runtime = context?.runtime?.watchHistory;
|
||||
if (isPlainObject(runtime) && typeof runtime.shouldSuppress === "function") {
|
||||
return runtime.shouldSuppress;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
export function createWatchHistorySuppressionStage({
|
||||
stageName = "watch-history-suppression",
|
||||
shouldSuppress,
|
||||
} = {}) {
|
||||
return async function watchHistorySuppressionStage(items = [], context = {}) {
|
||||
const hook =
|
||||
typeof shouldSuppress === "function" ? shouldSuppress : resolveSuppressionHook(context);
|
||||
|
||||
if (typeof hook !== "function") {
|
||||
return items;
|
||||
}
|
||||
|
||||
const results = [];
|
||||
|
||||
for (const item of items) {
|
||||
let suppress = false;
|
||||
try {
|
||||
suppress = await hook(item, context);
|
||||
} catch (error) {
|
||||
context?.log?.(`[${stageName}] suppression hook threw`, error);
|
||||
suppress = false;
|
||||
}
|
||||
|
||||
if (suppress) {
|
||||
const detail = {
|
||||
stage: stageName,
|
||||
type: "filter",
|
||||
reason: "watch-history",
|
||||
};
|
||||
if (item?.video && typeof item.video.id === "string") {
|
||||
detail.videoId = item.video.id;
|
||||
}
|
||||
if (item?.pointer) {
|
||||
detail.pointer = item.pointer;
|
||||
} else if (item?.metadata && item.metadata.pointerKey) {
|
||||
detail.pointerKey = item.metadata.pointerKey;
|
||||
}
|
||||
context?.addWhy?.(detail);
|
||||
continue;
|
||||
}
|
||||
|
||||
results.push(item);
|
||||
}
|
||||
|
||||
return results;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
// js/feedEngine/utils.js
|
||||
|
||||
export function isPlainObject(value) {
|
||||
if (value === null || typeof value !== "object") {
|
||||
return false;
|
||||
}
|
||||
const prototype = Object.getPrototypeOf(value);
|
||||
return prototype === Object.prototype || prototype === null;
|
||||
}
|
||||
|
||||
export function toSet(values) {
|
||||
if (values instanceof Set) {
|
||||
return new Set(values);
|
||||
}
|
||||
if (Array.isArray(values)) {
|
||||
return new Set(values);
|
||||
}
|
||||
return new Set();
|
||||
}
|
||||
|
||||
export function toArray(value) {
|
||||
if (!value) {
|
||||
return [];
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
return [...value];
|
||||
}
|
||||
return [value];
|
||||
}
|
||||
@@ -0,0 +1,178 @@
|
||||
// Run with: node tests/feed-engine.test.mjs
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
if (typeof globalThis.window === "undefined") {
|
||||
globalThis.window = {};
|
||||
}
|
||||
|
||||
if (typeof globalThis.localStorage === "undefined") {
|
||||
const storage = new Map();
|
||||
globalThis.localStorage = {
|
||||
getItem(key) {
|
||||
return storage.has(key) ? storage.get(key) : null;
|
||||
},
|
||||
setItem(key, value) {
|
||||
storage.set(String(key), String(value));
|
||||
},
|
||||
removeItem(key) {
|
||||
storage.delete(key);
|
||||
},
|
||||
clear() {
|
||||
storage.clear();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (typeof window.localStorage === "undefined") {
|
||||
window.localStorage = globalThis.localStorage;
|
||||
}
|
||||
|
||||
const {
|
||||
createFeedEngine,
|
||||
createDedupeByRootStage,
|
||||
createBlacklistFilterStage,
|
||||
createWatchHistorySuppressionStage,
|
||||
createChronologicalSorter,
|
||||
} = await import("../js/feedEngine/index.js");
|
||||
|
||||
async function testDedupeOrdering() {
|
||||
const engine = createFeedEngine();
|
||||
const feedName = "dedupe";
|
||||
|
||||
const videoOld = { id: "a-old", videoRootId: "rootA", created_at: 100 };
|
||||
const videoNew = { id: "a-new", videoRootId: "rootA", created_at: 200 };
|
||||
const videoOther = { id: "b", created_at: 150 };
|
||||
|
||||
engine.registerFeed(feedName, {
|
||||
source: async () => [
|
||||
{ video: videoOld },
|
||||
{ video: videoOther },
|
||||
{ video: videoNew },
|
||||
],
|
||||
stages: [createDedupeByRootStage()],
|
||||
sorter: createChronologicalSorter(),
|
||||
});
|
||||
|
||||
const result = await engine.runFeed(feedName);
|
||||
assert.equal(result.videos.length, 2, "dedupe stage should drop older root entries");
|
||||
assert.deepEqual(
|
||||
result.videos.map((video) => video.id),
|
||||
["a-new", "b"],
|
||||
"videos should be returned newest first",
|
||||
);
|
||||
|
||||
const dedupeReasons = result.metadata.why.filter(
|
||||
(entry) => entry.reason === "older-root-version",
|
||||
);
|
||||
assert.equal(dedupeReasons.length, 1, "dedupe stage should log one older-root reason");
|
||||
assert.equal(dedupeReasons[0].videoId, "a-old");
|
||||
assert.equal(dedupeReasons[0].rootId, "rootA");
|
||||
}
|
||||
|
||||
async function testBlacklistFiltering() {
|
||||
const engine = createFeedEngine();
|
||||
const feedName = "blacklist";
|
||||
|
||||
engine.registerFeed(feedName, {
|
||||
source: async () => [
|
||||
{ video: { id: "safe", pubkey: "npub1", created_at: 1 } },
|
||||
{ video: { id: "blocked", pubkey: "npub2", created_at: 2 } },
|
||||
{ video: { id: "blocked-author", pubkey: "blocked", created_at: 3 } },
|
||||
],
|
||||
stages: [
|
||||
createBlacklistFilterStage({
|
||||
shouldIncludeVideo(video, { blacklistedEventIds, isAuthorBlocked }) {
|
||||
if (blacklistedEventIds.has(video.id)) {
|
||||
return false;
|
||||
}
|
||||
if (isAuthorBlocked(video.pubkey)) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
},
|
||||
}),
|
||||
],
|
||||
sorter: createChronologicalSorter(),
|
||||
});
|
||||
|
||||
const result = await engine.runFeed(feedName, {
|
||||
runtime: {
|
||||
blacklistedEventIds: new Set(["blocked"]),
|
||||
isAuthorBlocked: (pubkey) => pubkey === "blocked",
|
||||
},
|
||||
});
|
||||
|
||||
assert.deepEqual(result.videos.map((video) => video.id), ["safe"]);
|
||||
|
||||
const blacklistReasons = result.metadata.why.filter(
|
||||
(entry) => entry.reason === "blacklist",
|
||||
);
|
||||
assert.equal(
|
||||
blacklistReasons.length,
|
||||
2,
|
||||
"blacklist stage should log two exclusions (one per blocked item)",
|
||||
);
|
||||
const removedIds = new Set(blacklistReasons.map((entry) => entry.videoId));
|
||||
assert.ok(removedIds.has("blocked"));
|
||||
assert.ok(removedIds.has("blocked-author"));
|
||||
}
|
||||
|
||||
async function testWatchHistoryHookIsolation() {
|
||||
const engine = createFeedEngine();
|
||||
|
||||
engine.registerFeed("feed-a", {
|
||||
source: async () => [
|
||||
{ video: { id: "x1", created_at: 10 }, pointer: { type: "e", value: "x1" } },
|
||||
],
|
||||
stages: [createWatchHistorySuppressionStage()],
|
||||
});
|
||||
|
||||
engine.registerFeed("feed-b", {
|
||||
source: async () => [
|
||||
{ video: { id: "y1", created_at: 20 }, pointer: { type: "e", value: "y1" } },
|
||||
],
|
||||
stages: [createWatchHistorySuppressionStage()],
|
||||
});
|
||||
|
||||
let hookACount = 0;
|
||||
let hookBCount = 0;
|
||||
|
||||
const resultA = await engine.runFeed("feed-a", {
|
||||
hooks: {
|
||||
watchHistory: {
|
||||
async shouldSuppress(item) {
|
||||
hookACount += 1;
|
||||
return item?.video?.id === "x1";
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const resultB = await engine.runFeed("feed-b", {
|
||||
hooks: {
|
||||
watchHistory: {
|
||||
async shouldSuppress(item) {
|
||||
hookBCount += 1;
|
||||
return false;
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(hookACount, 1, "feed A hook should run once");
|
||||
assert.equal(hookBCount, 1, "feed B hook should run once");
|
||||
|
||||
assert.equal(resultA.videos.length, 0, "feed A should suppress the single video");
|
||||
assert.equal(resultB.videos.length, 1, "feed B should keep its video");
|
||||
|
||||
const whyA = resultA.metadata.why.find((entry) => entry.reason === "watch-history");
|
||||
assert.ok(whyA, "feed A should report watch-history suppression metadata");
|
||||
assert.equal(whyA.videoId, "x1");
|
||||
}
|
||||
|
||||
await testDedupeOrdering();
|
||||
await testBlacklistFiltering();
|
||||
await testWatchHistoryHookIsolation();
|
||||
|
||||
console.log("All feed engine tests passed");
|
||||
Reference in new issue
Block a user