Merge pull request #856 from PR0M3TH3AN/codex/implement-comment-thread-service-with-utils

Add comment thread service with shared video pointer helper
This commit is contained in:
thePR0M3TH3AN authored and GitHub committed 2025-10-24 15:09:54 -04:00
commit b79a5c2a16
5 files changed
+838 -71

No files matched your search

+4 -36
View File
@@ -96,7 +96,9 @@ import initEditModal from "./ui/initEditModal.js";
import initDeleteModal from "./ui/initDeleteModal.js";
import { MediaLoader } from "./utils/mediaLoader.js";
import { pointerArrayToKey } from "./utils/pointer.js";
import { resolveVideoPointer } from "./utils/videoPointer.js";
import resolveVideoPointer, {
buildVideoAddressPointer,
} from "./utils/videoPointer.js";
import { isValidMagnetUri } from "./utils/magnetValidators.js";
import { dedupeToNewestByRoot } from "./utils/videoDeduper.js";
import { buildServiceWorkerFallbackStatus } from "./utils/serviceWorkerFallbackMessages.js";
@@ -6177,41 +6179,7 @@ class Application {
return this.discussionCountService.getVideoAddressPointer(video);
}
if (!video || typeof video !== "object") {
return "";
}
const tags = Array.isArray(video.tags) ? video.tags : [];
const dTag = tags.find(
(tag) =>
Array.isArray(tag) &&
tag.length >= 2 &&
tag[0] === "d" &&
typeof tag[1] === "string" &&
tag[1].trim()
);
if (!dTag) {
return "";
}
const pubkey =
typeof video.pubkey === "string" ? video.pubkey.trim() : "";
if (!pubkey) {
return "";
}
const identifier = dTag[1].trim();
if (!identifier) {
return "";
}
const kind =
Number.isFinite(video.kind) && video.kind > 0
? Math.floor(video.kind)
: VIDEO_EVENT_KIND;
return `${kind}:${pubkey}:${identifier}`;
return buildVideoAddressPointer(video, { defaultKind: VIDEO_EVENT_KIND });
}
bindThumbnailFallbacks(container) {
+534
View File
@@ -0,0 +1,534 @@
// js/services/commentThreadService.js
import { devLogger } from "../utils/logger.js";
import { buildVideoAddressPointer } from "../utils/videoPointer.js";
const ROOT_PARENT_KEY = "__root__";
const DEFAULT_INITIAL_LIMIT = 40;
const DEFAULT_HYDRATION_DEBOUNCE_MS = 25;
function toPositiveInteger(value, fallback) {
const numeric = Number(value);
if (!Number.isFinite(numeric) || numeric <= 0) {
return fallback;
}
return Math.floor(numeric);
}
function normalizeString(value) {
return typeof value === "string" ? value.trim() : "";
}
function normalizePubkey(pubkey) {
const normalized = normalizeString(pubkey);
return normalized ? normalized.toLowerCase() : "";
}
function toParentKey(parentId) {
return parentId ? parentId : ROOT_PARENT_KEY;
}
function fromParentKey(parentKey) {
return parentKey === ROOT_PARENT_KEY ? null : parentKey;
}
function safeCall(handler, payload) {
if (typeof handler !== "function") {
return;
}
try {
handler(payload);
} catch (error) {
devLogger.warn("[commentThread] callback threw", error);
}
}
export default class CommentThreadService {
constructor({
nostrClient = null,
app = null,
limit = DEFAULT_INITIAL_LIMIT,
hydrationDebounceMs = DEFAULT_HYDRATION_DEBOUNCE_MS,
logger = devLogger,
} = {}) {
this.nostrClient = nostrClient;
this.app = app;
this.defaultLimit = toPositiveInteger(limit, DEFAULT_INITIAL_LIMIT);
this.hydrationDelay = Math.max(
0,
toPositiveInteger(hydrationDebounceMs, DEFAULT_HYDRATION_DEBOUNCE_MS)
);
this.logger =
logger && typeof logger.warn === "function" ? logger : devLogger;
this.callbacks = {
onThreadReady: null,
onCommentsAppended: null,
onError: null,
};
this.resetInternalState();
}
resetInternalState() {
this.eventsById = new Map();
this.metaById = new Map();
this.childrenByParent = new Map();
this.childrenByParent.set(ROOT_PARENT_KEY, []);
this.profileCache = new Map();
this.profileQueue = new Set();
this.profileHydrationTimer = null;
this.profileHydrationPromise = null;
this.subscriptionCleanup = null;
this.videoEventId = "";
this.videoAddressPointer = "";
this.parentCommentId = "";
this.activeRelays = null;
}
setCallbacks({
onThreadReady = this.callbacks.onThreadReady,
onCommentsAppended = this.callbacks.onCommentsAppended,
onError = this.callbacks.onError,
} = {}) {
this.callbacks.onThreadReady =
typeof onThreadReady === "function" ? onThreadReady : null;
this.callbacks.onCommentsAppended =
typeof onCommentsAppended === "function" ? onCommentsAppended : null;
this.callbacks.onError = typeof onError === "function" ? onError : null;
}
async loadThread({
video = null,
parentCommentId = "",
limit = this.defaultLimit,
relays = null,
} = {}) {
this.teardown();
this.videoEventId = normalizeString(video?.id);
this.videoAddressPointer = buildVideoAddressPointer(video);
this.parentCommentId = normalizeString(parentCommentId);
this.activeRelays = Array.isArray(relays) ? [...relays] : null;
if (!this.videoEventId || !this.videoAddressPointer) {
this.emitError(
new Error("Unable to resolve video pointer for comment thread."),
);
return { success: false };
}
if (!this.nostrClient ||
typeof this.nostrClient.fetchVideoComments !== "function") {
this.emitError(
new Error("Nostr client is missing fetchVideoComments implementation."),
);
return { success: false };
}
const fetchLimit = toPositiveInteger(limit, this.defaultLimit);
const target = {
videoEventId: this.videoEventId,
videoDefinitionAddress: this.videoAddressPointer,
parentCommentId: this.parentCommentId,
};
let events = [];
try {
events = await this.nostrClient.fetchVideoComments(target, {
limit: fetchLimit,
relays: this.activeRelays,
});
} catch (error) {
this.emitError(error);
return { success: false, error };
}
if (Array.isArray(events)) {
events.forEach((event) => {
this.processIncomingEvent(event, { isInitial: true });
});
}
this.emitThreadReady();
this.startSubscription(target);
if (this.profileHydrationTimer) {
clearTimeout(this.profileHydrationTimer);
this.profileHydrationTimer = null;
}
if (this.profileQueue.size) {
await this.flushProfileQueue();
}
return this.getSnapshot();
}
startSubscription(target) {
if (
!this.nostrClient ||
typeof this.nostrClient.subscribeVideoComments !== "function"
) {
return;
}
const options = {
relays: this.activeRelays,
onEvent: (event) => this.processIncomingEvent(event),
};
try {
const cleanup = this.nostrClient.subscribeVideoComments(target, options);
if (typeof cleanup === "function") {
this.subscriptionCleanup = () => {
try {
cleanup();
} catch (error) {
if (this.logger?.warn) {
this.logger.warn(
"[commentThread] Failed to unsubscribe from comment feed:",
error,
);
}
}
};
} else {
this.subscriptionCleanup = null;
}
} catch (error) {
this.emitError(error);
}
}
processIncomingEvent(event, { isInitial = false } = {}) {
const result = this.applyEvent(event);
if (!result) {
return;
}
if (!isInitial && result.type === "insert") {
this.emitAppend(result);
}
}
applyEvent(event) {
if (!event || typeof event !== "object") {
return null;
}
const eventId = normalizeString(event.id);
if (!eventId) {
return null;
}
const parentId = this.extractParentId(event);
const parentKey = toParentKey(parentId);
const createdAt = Number.isFinite(event.created_at) ? event.created_at : 0;
const pubkey = normalizePubkey(event.pubkey);
const existingMeta = this.metaById.get(eventId);
this.eventsById.set(eventId, event);
this.metaById.set(eventId, { parentKey, createdAt });
if (!existingMeta) {
this.insertIntoParentList(parentKey, eventId, createdAt);
if (pubkey) {
this.queueProfileForHydration(pubkey);
}
return { type: "insert", parentId, parentKey, eventId, event };
}
if (existingMeta.parentKey !== parentKey) {
this.removeFromParentList(existingMeta.parentKey, eventId);
this.insertIntoParentList(parentKey, eventId, createdAt);
} else if (existingMeta.createdAt !== createdAt) {
this.reorderParentList(parentKey, eventId, createdAt);
}
if (pubkey) {
this.queueProfileForHydration(pubkey);
}
return { type: "update", parentId, parentKey, eventId, event };
}
extractParentId(event) {
const tags = Array.isArray(event.tags) ? event.tags : [];
const values = [];
for (const tag of tags) {
if (!Array.isArray(tag) || tag.length < 2) {
continue;
}
const [name, value] = tag;
if (name !== "e") {
continue;
}
const normalizedValue = normalizeString(value);
if (!normalizedValue) {
continue;
}
values.push(normalizedValue);
}
for (let index = values.length - 1; index >= 0; index -= 1) {
const candidate = values[index];
if (candidate && candidate !== this.videoEventId) {
return candidate;
}
}
return "";
}
insertIntoParentList(parentKey, eventId, createdAt) {
const list = this.getParentList(parentKey);
if (list.includes(eventId)) {
return;
}
const insertIndex = list.findIndex((existingId) => {
const meta = this.metaById.get(existingId);
const existingCreated = meta ? meta.createdAt : 0;
if (createdAt === existingCreated) {
return eventId.localeCompare(existingId) < 0;
}
return createdAt < existingCreated;
});
if (insertIndex === -1) {
list.push(eventId);
} else {
list.splice(insertIndex, 0, eventId);
}
}
removeFromParentList(parentKey, eventId) {
const list = this.childrenByParent.get(parentKey);
if (!Array.isArray(list)) {
return;
}
const index = list.indexOf(eventId);
if (index >= 0) {
list.splice(index, 1);
}
}
reorderParentList(parentKey, eventId, createdAt) {
this.removeFromParentList(parentKey, eventId);
this.insertIntoParentList(parentKey, eventId, createdAt);
}
getParentList(parentKey) {
if (!this.childrenByParent.has(parentKey)) {
this.childrenByParent.set(parentKey, []);
}
return this.childrenByParent.get(parentKey);
}
queueProfileForHydration(pubkey) {
if (!pubkey) {
return;
}
if (this.profileCache.has(pubkey)) {
return;
}
const cachedProfile = this.getProfileFromApp(pubkey);
if (cachedProfile) {
this.profileCache.set(pubkey, cachedProfile);
return;
}
this.profileQueue.add(pubkey);
this.scheduleProfileHydration();
}
getProfileFromApp(pubkey) {
if (!this.app || typeof this.app.getProfileCacheEntry !== "function") {
return null;
}
try {
const entry = this.app.getProfileCacheEntry(pubkey);
if (entry && typeof entry === "object" && entry.profile) {
return entry.profile;
}
} catch (error) {
if (this.logger?.warn) {
this.logger.warn(
"[commentThread] Failed to read profile cache entry:",
error,
);
}
}
return null;
}
scheduleProfileHydration() {
if (this.profileQueue.size === 0) {
return;
}
if (this.profileHydrationTimer) {
return;
}
this.profileHydrationTimer = setTimeout(() => {
this.profileHydrationTimer = null;
this.flushProfileQueue();
}, this.hydrationDelay);
}
async flushProfileQueue() {
if (!this.profileQueue.size) {
return;
}
const pubkeys = Array.from(this.profileQueue);
this.profileQueue.clear();
if (!this.app || typeof this.app.batchFetchProfiles !== "function") {
return;
}
const hydrationPromise = Promise.resolve(
this.app.batchFetchProfiles(pubkeys),
)
.catch((error) => {
if (this.logger?.warn) {
this.logger.warn(
"[commentThread] Profile hydration failed:",
error,
);
}
this.emitError(error);
})
.then(() => {
pubkeys.forEach((pubkey) => {
const profile = this.getProfileFromApp(pubkey);
if (profile) {
this.profileCache.set(pubkey, profile);
}
});
})
.finally(() => {
if (this.profileHydrationPromise === hydrationPromise) {
this.profileHydrationPromise = null;
}
});
this.profileHydrationPromise = hydrationPromise;
await hydrationPromise;
}
emitThreadReady() {
const payload = {
videoEventId: this.videoEventId,
parentCommentId: this.parentCommentId || null,
topLevelIds: this.getCommentIdsForParent(null),
commentsById: this.cloneCommentsMap(),
childrenByParent: this.cloneTreeMap(),
profiles: this.getProfilesSnapshot(),
};
safeCall(this.callbacks.onThreadReady, payload);
}
emitAppend({ parentId, eventId }) {
const payload = {
videoEventId: this.videoEventId,
parentCommentId: parentId || null,
commentIds: [eventId],
commentsById: this.cloneCommentsMap(),
childrenByParent: this.cloneTreeMap(),
profiles: this.getProfilesSnapshot(),
};
safeCall(this.callbacks.onCommentsAppended, payload);
}
emitError(error) {
if (this.logger?.warn && error) {
this.logger.warn("[commentThread]", error);
}
safeCall(this.callbacks.onError, error);
}
getSnapshot() {
return {
videoEventId: this.videoEventId,
parentCommentId: this.parentCommentId || null,
commentsById: this.cloneCommentsMap(),
childrenByParent: this.cloneTreeMap(),
profiles: this.getProfilesSnapshot(),
};
}
cloneCommentsMap() {
return new Map(this.eventsById);
}
cloneTreeMap() {
const tree = new Map();
this.childrenByParent.forEach((ids, parentKey) => {
tree.set(fromParentKey(parentKey), [...ids]);
});
return tree;
}
getProfilesSnapshot() {
return new Map(this.profileCache);
}
getProfile(pubkey) {
const normalized = normalizePubkey(pubkey);
if (!normalized) {
return null;
}
if (this.profileCache.has(normalized)) {
return this.profileCache.get(normalized);
}
const profile = this.getProfileFromApp(normalized);
if (profile) {
this.profileCache.set(normalized, profile);
return profile;
}
return null;
}
getCommentIdsForParent(parentId) {
const key = toParentKey(normalizeString(parentId));
const list = this.childrenByParent.get(key);
return Array.isArray(list) ? [...list] : [];
}
waitForProfileHydration() {
return this.profileHydrationPromise || Promise.resolve();
}
teardown() {
if (this.subscriptionCleanup) {
try {
this.subscriptionCleanup();
} catch (error) {
if (this.logger?.warn) {
this.logger.warn(
"[commentThread] Failed during teardown unsubscribe:",
error,
);
}
}
}
this.subscriptionCleanup = null;
if (this.profileHydrationTimer) {
clearTimeout(this.profileHydrationTimer);
this.profileHydrationTimer = null;
}
this.profileQueue.clear();
this.profileHydrationPromise = null;
this.resetInternalState();
}
}
+3 -35
View File
@@ -1,4 +1,5 @@
import { devLogger } from "../utils/logger.js";
import { buildVideoAddressPointer } from "../utils/videoPointer.js";
const DEFAULT_MAX_DISCUSSION_COUNT_VIDEOS = 24;
const COUNT_UNSUPPORTED_TITLE = "Relay does not support NIP-45 COUNT queries.";
@@ -132,7 +133,7 @@ export default class DiscussionCountService {
filters.push({ kinds: [1], "#e": [eventId] });
}
const address = this.getVideoAddressPointer(video);
const address = buildVideoAddressPointer(video);
if (address) {
filters.push({ kinds: [1], "#a": [address] });
}
@@ -141,40 +142,7 @@ export default class DiscussionCountService {
}
getVideoAddressPointer(video) {
if (!video || typeof video !== "object") {
return "";
}
const tags = Array.isArray(video.tags) ? video.tags : [];
const dTag = tags.find(
(tag) =>
Array.isArray(tag) &&
tag.length >= 2 &&
tag[0] === "d" &&
typeof tag[1] === "string" &&
tag[1].trim()
);
if (!dTag) {
return "";
}
const pubkey = typeof video.pubkey === "string" ? video.pubkey.trim() : "";
if (!pubkey) {
return "";
}
const identifier = dTag[1].trim();
if (!identifier) {
return "";
}
const kind =
Number.isFinite(video.kind) && video.kind > 0
? Math.floor(video.kind)
: 30078;
return `${kind}:${pubkey}:${identifier}`;
return buildVideoAddressPointer(video);
}
setDiscussionCountPending(element) {
+58
View File
@@ -1,6 +1,7 @@
// js/utils/videoPointer.js
import { pointerArrayToKey } from "./pointer.js";
import { devLogger } from "./logger.js";
const DEFAULT_VIDEO_KIND = 30078;
@@ -49,3 +50,60 @@ export function resolveVideoPointer({
}
export default resolveVideoPointer;
export function buildVideoAddressPointer(
video,
{ defaultKind = DEFAULT_VIDEO_KIND, logger = devLogger } = {}
) {
if (!video || typeof video !== "object") {
return "";
}
const tags = Array.isArray(video.tags) ? video.tags : [];
const dTag = tags.find(
(tag) =>
Array.isArray(tag) &&
tag.length >= 2 &&
tag[0] === "d" &&
typeof tag[1] === "string" &&
tag[1].trim()
);
if (!dTag) {
return "";
}
const pubkey = typeof video.pubkey === "string" ? video.pubkey.trim() : "";
if (!pubkey) {
return "";
}
const identifier = dTag[1].trim();
if (!identifier) {
return "";
}
let kind = defaultKind;
if (Number.isFinite(video.kind) && video.kind > 0) {
kind = Math.floor(video.kind);
} else if (typeof video.kind === "string" && video.kind.trim()) {
const parsed = Number(video.kind);
if (Number.isFinite(parsed) && parsed > 0) {
kind = Math.floor(parsed);
}
}
if (!Number.isFinite(kind) || kind <= 0) {
if (logger?.warn) {
logger.warn(
"[videoPointer] Invalid kind detected while building video pointer; defaulting to 30078.",
{ kind: video.kind }
);
}
kind = DEFAULT_VIDEO_KIND;
}
return `${kind}:${pubkey}:${identifier}`;
}
export { DEFAULT_VIDEO_KIND };
+239
View File
@@ -0,0 +1,239 @@
// Run with: node tests/comment-thread-service.test.mjs
import assert from "node:assert/strict";
import CommentThreadService from "../js/services/commentThreadService.js";
function createBaseVideo() {
return {
id: "video123",
pubkey: "authorpk",
kind: 30078,
tags: [["d", "video-root"], ["alt", "ignored"]],
};
}
function createComment({ id, pubkey, createdAt, parentId = null }) {
const baseTags = [
["e", "video123"],
["a", "30078:authorpk:video-root"],
];
const tags = parentId
? [...baseTags, ["e", parentId]]
: [...baseTags];
return {
id,
kind: 1,
pubkey,
created_at: createdAt,
content: `${id} content`,
tags,
};
}
const tick = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms));
async function testThreadFlow() {
const video = createBaseVideo();
const initialEvents = [
createComment({ id: "comment-1", pubkey: "pk1", createdAt: 100 }),
createComment({ id: "comment-2", pubkey: "pk2", createdAt: 90 }),
createComment({
id: "reply-1",
pubkey: "pk3",
createdAt: 110,
parentId: "comment-1",
}),
];
const profileCache = new Map();
const batchCalls = [];
const app = {
getProfileCacheEntry: (pubkey) =>
profileCache.has(pubkey) ? { profile: profileCache.get(pubkey) } : null,
batchFetchProfiles: async (pubkeys) => {
batchCalls.push([...pubkeys]);
pubkeys.forEach((pubkey) => {
profileCache.set(pubkey, {
name: `Profile ${pubkey}`,
picture: `https://example.com/${pubkey}.png`,
});
});
},
};
let subscriptionHandler = null;
const nostrClient = {
fetchVideoComments: async () => initialEvents,
subscribeVideoComments: (_target, options = {}) => {
subscriptionHandler = typeof options.onEvent === "function"
? options.onEvent
: null;
return () => {
subscriptionHandler = null;
};
},
};
const received = {
initial: [],
appended: [],
errors: [],
};
const service = new CommentThreadService({
nostrClient,
app,
hydrationDebounceMs: 0,
});
service.setCallbacks({
onThreadReady: (payload) => received.initial.push(payload),
onCommentsAppended: (payload) => received.appended.push(payload),
onError: (error) => received.errors.push(error),
});
await service.loadThread({ video });
await tick(10);
await service.waitForProfileHydration();
assert.equal(received.initial.length, 1, "initial callback should fire once");
assert.equal(received.errors.length, 0, "no errors should be reported");
assert.equal(batchCalls.length, 1, "profiles should be fetched once for unknown authors");
assert.deepEqual(
new Set(batchCalls[0]),
new Set(["pk1", "pk2", "pk3"]),
"hydration should request each unique pubkey",
);
const initialPayload = received.initial[0];
assert.ok(initialPayload.childrenByParent instanceof Map);
assert.deepEqual(
initialPayload.topLevelIds,
["comment-2", "comment-1"],
"top-level comments should be sorted by created_at",
);
assert.deepEqual(
initialPayload.childrenByParent.get(null),
["comment-2", "comment-1"],
"root mapping should expose top-level ids",
);
assert.deepEqual(
initialPayload.childrenByParent.get("comment-1"),
["reply-1"],
"parent mapping should include replies",
);
assert.ok(
service.getProfile("pk3"),
"profiles should be cached for retrieval",
);
assert.ok(subscriptionHandler, "subscription handler should be registered");
const newReply = createComment({
id: "reply-2",
pubkey: "pk4",
createdAt: 120,
parentId: "comment-2",
});
subscriptionHandler(newReply);
await tick(10);
await service.waitForProfileHydration();
for (let attempt = 0; attempt < 5 && batchCalls.length < 2; attempt += 1) {
await tick(20);
}
assert.equal(
batchCalls.length,
2,
"hydration should fetch new reply authors",
);
assert.equal(
service.profileQueue.size,
0,
"profile queue should be cleared after hydration",
);
assert.equal(
received.appended.length,
1,
"one append callback should fire for the new reply",
);
const appendPayload = received.appended[0];
assert.equal(
appendPayload.parentCommentId,
"comment-2",
"append payload should identify the parent comment",
);
assert.deepEqual(
appendPayload.commentIds,
["reply-2"],
"append payload should surface the new reply id",
);
assert.ok(
service.getProfilesSnapshot().has("pk4"),
"new reply author should be hydrated and cached",
);
// Duplicate events should be ignored.
subscriptionHandler(newReply);
await tick();
assert.equal(
received.appended.length,
1,
"duplicate events should not trigger additional append callbacks",
);
service.teardown();
}
async function testTeardownCancelsHydration() {
const video = createBaseVideo();
const delayedReply = createComment({
id: "delayed-1",
pubkey: "pk9",
createdAt: 200,
});
let unsubscribed = false;
let batchCallCount = 0;
let subscriptionHandler = null;
const app = {
getProfileCacheEntry: () => null,
batchFetchProfiles: async () => {
batchCallCount += 1;
},
};
const nostrClient = {
fetchVideoComments: async () => [],
subscribeVideoComments: (_target, options = {}) => {
subscriptionHandler = typeof options.onEvent === "function"
? options.onEvent
: null;
return () => {
unsubscribed = true;
};
},
};
const service = new CommentThreadService({
nostrClient,
app,
hydrationDebounceMs: 50,
});
await service.loadThread({ video });
assert.ok(subscriptionHandler, "subscription handler should be available");
subscriptionHandler(delayedReply);
service.teardown();
await tick(80);
assert.equal(batchCallCount, 0, "teardown should cancel pending hydration timers");
assert.equal(unsubscribed, true, "teardown should unsubscribe from nostr feed");
}
await testThreadFlow();
await testTeardownCancelsHydration();
console.log("comment-thread-service tests passed");