mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-27 07:52:24 +00:00
watch-history: bucket timestamp-less items by d-tag time, not 1970-01
Legacy/migrated items with no watchedAt were siloed into a literal "1970-01" month event that self-perpetuated (read back as 0 -> re-bucketed -> re-published with 0). Derive a stable watch time from the pointer d-tag's embedded ms-timestamp and backfill it so they land in a real month and the loop stops. Bucketing extracted to watchHistoryBucketing.js to keep watchHistory.js under its size cap. Also fixes three stale tests in watch-history.test.mjs (partial-acceptance, actor alignment, session flag) so the file is un-quarantined. Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
1 parent
7e67471a4b
commit
d8e1b12992
4 files changed
+207
-124
No files matched your search
@@ -41,6 +41,7 @@ import {
|
||||
import { pMap } from "../utils/asyncUtils.js";
|
||||
import { selectNewestListEvent } from "./listEventOrdering.js";
|
||||
import { dedupeNewestPerReplaceableAddress } from "./watchHistoryDedup.js";
|
||||
import { bucketWatchHistoryItemsByMonth } from "./watchHistoryBucketing.js";
|
||||
import {
|
||||
getCachedChunkPlaintext,
|
||||
setCachedChunkPlaintext,
|
||||
@@ -306,7 +307,7 @@ export function normalizeActorKey(actor) {
|
||||
return trimmed.toLowerCase();
|
||||
}
|
||||
|
||||
function canonicalizeWatchHistoryItems(rawItems, maxItems = WATCH_HISTORY_MAX_ITEMS) {
|
||||
export function canonicalizeWatchHistoryItems(rawItems, maxItems = WATCH_HISTORY_MAX_ITEMS) {
|
||||
const seen = new Map();
|
||||
if (Array.isArray(rawItems)) {
|
||||
for (const candidate of rawItems) {
|
||||
@@ -360,20 +361,7 @@ function canonicalizeWatchHistoryItems(rawItems, maxItems = WATCH_HISTORY_MAX_IT
|
||||
? deduped
|
||||
: deduped.slice(0, Math.max(0, Math.floor(maxItems)));
|
||||
|
||||
const buckets = {};
|
||||
for (const item of limited) {
|
||||
const watchedAt = Number.isFinite(item.watchedAt) ? item.watchedAt : 0;
|
||||
const date = new Date(watchedAt * 1000);
|
||||
const year = date.getUTCFullYear();
|
||||
const month = String(date.getUTCMonth() + 1).padStart(2, '0');
|
||||
const key = (watchedAt > 0) ? `${year}-${month}` : "1970-01";
|
||||
|
||||
if (!buckets[key]) {
|
||||
buckets[key] = [];
|
||||
}
|
||||
buckets[key].push(item);
|
||||
}
|
||||
return buckets;
|
||||
return bucketWatchHistoryItemsByMonth(limited);
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
// Watch-history month bucketing. Items are grouped into "YYYY-MM" buckets by their
|
||||
// watch time. Legacy/migrated items can arrive with no real watch time
|
||||
// (watchedAt 0); those used to silo into a literal "1970-01" month that
|
||||
// self-perpetuated (read back as 0 -> re-bucketed -> re-published with 0). bitvid
|
||||
// pointer values look like "<kind>:<pubkey>:<dTag>" and the dTag usually embeds the
|
||||
// video's creation time as a 13-digit millisecond timestamp, so we derive a STABLE
|
||||
// fallback from it instead.
|
||||
|
||||
// Returns the embedded creation time (seconds) from a pointer's d-tag, or 0 if no
|
||||
// plausible 13-digit ms timestamp is present (e.g. a hash-only d-tag).
|
||||
export function deriveWatchedAtFromPointer(pointer) {
|
||||
const value = typeof pointer?.value === "string" ? pointer.value : "";
|
||||
if (!value) {
|
||||
return 0;
|
||||
}
|
||||
const dTag = value.includes(":") ? value.slice(value.lastIndexOf(":") + 1) : value;
|
||||
const matches = dTag.match(/\d{13}/g);
|
||||
if (!matches) {
|
||||
return 0;
|
||||
}
|
||||
// Constrain to a sane ms range (~2014..~2033) so we don't latch onto 13 random
|
||||
// digits inside a hex d-tag.
|
||||
const MIN_MS = 1_400_000_000_000;
|
||||
const MAX_MS = 2_000_000_000_000;
|
||||
for (const match of matches) {
|
||||
const ms = Number(match);
|
||||
if (Number.isFinite(ms) && ms >= MIN_MS && ms <= MAX_MS) {
|
||||
return Math.floor(ms / 1000);
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Group already-deduped/sorted/limited pointer items into { "YYYY-MM": [items] }.
|
||||
// For items with no watch time, derive a stable one from the d-tag and BACKFILL it
|
||||
// (this is what breaks the 0 -> "1970-01" -> 0 loop). Only truly undecodable items
|
||||
// still land in the stable epoch bucket.
|
||||
export function bucketWatchHistoryItemsByMonth(items) {
|
||||
const buckets = {};
|
||||
const list = Array.isArray(items) ? items : [];
|
||||
for (const item of list) {
|
||||
let watchedAt = Number.isFinite(item.watchedAt) ? item.watchedAt : 0;
|
||||
if (watchedAt <= 0) {
|
||||
const derived = deriveWatchedAtFromPointer(item);
|
||||
if (derived > 0) {
|
||||
watchedAt = derived;
|
||||
item.watchedAt = derived;
|
||||
}
|
||||
}
|
||||
const date = new Date(watchedAt * 1000);
|
||||
const year = date.getUTCFullYear();
|
||||
const month = String(date.getUTCMonth() + 1).padStart(2, "0");
|
||||
const key = watchedAt > 0 ? `${year}-${month}` : "1970-01";
|
||||
if (!buckets[key]) {
|
||||
buckets[key] = [];
|
||||
}
|
||||
buckets[key].push(item);
|
||||
}
|
||||
return buckets;
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
// Watch-history month bucketing: legacy/migrated items with no watch time
|
||||
// (watchedAt 0) used to silo into a literal "1970-01" month that self-perpetuated
|
||||
// (read back as 0 -> re-bucketed to 1970-01 -> re-published with 0). bitvid
|
||||
// pointers embed the video's creation time in their d-tag; derive a stable
|
||||
// fallback from it so those items land in a real month and the value is backfilled.
|
||||
import "./test-helpers/setup-localstorage.mjs";
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
import { canonicalizeWatchHistoryItems } from "../js/nostr/watchHistory.js";
|
||||
import { deriveWatchedAtFromPointer } from "../js/nostr/watchHistoryBucketing.js";
|
||||
|
||||
const PK = "a".repeat(64);
|
||||
|
||||
test("deriveWatchedAtFromPointer: extracts the ms timestamp from a timestamp-first d-tag", () => {
|
||||
// 1761508697162 ms = 2025-10-26 ~ in seconds 1761508697
|
||||
const seconds = deriveWatchedAtFromPointer({
|
||||
value: `30078:${PK}:1761508697162-z7ax37o3nqm`,
|
||||
});
|
||||
assert.equal(seconds, 1761508697);
|
||||
});
|
||||
|
||||
test("deriveWatchedAtFromPointer: extracts the ms timestamp from a timestamp-last d-tag", () => {
|
||||
const seconds = deriveWatchedAtFromPointer({
|
||||
value: `30078:${PK}:goblinbox-1782226234472`,
|
||||
});
|
||||
assert.equal(seconds, 1782226234);
|
||||
});
|
||||
|
||||
test("deriveWatchedAtFromPointer: returns 0 for a hash-only d-tag (no plausible timestamp)", () => {
|
||||
const seconds = deriveWatchedAtFromPointer({
|
||||
value: `30078:${PK}:fd11d796f68d6cb02ae46bcdf0a1ee2f032dea747652d3e96fd13338c845a9d5`,
|
||||
});
|
||||
assert.equal(seconds, 0);
|
||||
});
|
||||
|
||||
test("canonicalize: a watchedAt:0 item buckets into a REAL month (not 1970-01) and backfills", () => {
|
||||
const item = {
|
||||
type: "a",
|
||||
value: `30078:${PK}:1761508697162-z7ax37o3nqm`,
|
||||
watchedAt: 0,
|
||||
};
|
||||
const buckets = canonicalizeWatchHistoryItems([item]);
|
||||
const months = Object.keys(buckets);
|
||||
assert.ok(!months.includes("1970-01"), "must not silo a decodable item into 1970-01");
|
||||
assert.deepEqual(months, ["2025-10"], "item buckets by its derived creation month");
|
||||
// Backfilled so the published watchedAt map carries a real value (stops the loop).
|
||||
assert.equal(buckets["2025-10"][0].watchedAt, 1761508697);
|
||||
});
|
||||
|
||||
test("canonicalize: a real watchedAt is unaffected", () => {
|
||||
const item = {
|
||||
type: "a",
|
||||
value: `30078:${PK}:goblinbox-1782226234472`,
|
||||
watchedAt: 1782237450, // a genuine, later watch time
|
||||
};
|
||||
const buckets = canonicalizeWatchHistoryItems([item]);
|
||||
// 1782237450 s = 2026-06; uses the real watch time, not the derived creation time.
|
||||
assert.deepEqual(Object.keys(buckets), ["2026-06"]);
|
||||
assert.equal(buckets["2026-06"][0].watchedAt, 1782237450);
|
||||
});
|
||||
|
||||
test("canonicalize: a truly undecodable timestamp-less item still buckets stably (1970-01)", () => {
|
||||
const item = {
|
||||
type: "a",
|
||||
value: `30078:${PK}:fd11d796f68d6cb02ae46bcdf0a1ee2f032dea747652d3e96fd13338c845a9d5`,
|
||||
watchedAt: 0,
|
||||
};
|
||||
const buckets = canonicalizeWatchHistoryItems([item]);
|
||||
assert.deepEqual(Object.keys(buckets), ["1970-01"]);
|
||||
});
|
||||
+74
-109
@@ -1183,142 +1183,98 @@ async function testWatchHistoryPartialRelayRetry() {
|
||||
"write relays should include all configured endpoints",
|
||||
);
|
||||
|
||||
const attemptPlans = [
|
||||
{
|
||||
"wss://relay.one": true,
|
||||
"wss://relay.two": true,
|
||||
"wss://relay.three": false,
|
||||
},
|
||||
{
|
||||
"wss://relay.one": true,
|
||||
"wss://relay.two": false,
|
||||
"wss://relay.three": true,
|
||||
},
|
||||
{
|
||||
"wss://relay.one": true,
|
||||
"wss://relay.two": true,
|
||||
"wss://relay.three": true,
|
||||
},
|
||||
];
|
||||
// Partial acceptance: 2 of 3 write relays accept, one rejects — on a single
|
||||
// attempt (no escalation). The relay set is stable across calls.
|
||||
const acceptPlan = {
|
||||
"wss://relay.one": true,
|
||||
"wss://relay.two": true,
|
||||
"wss://relay.three": false,
|
||||
};
|
||||
|
||||
poolHarness.setResolver(({ relays }) => {
|
||||
const relayUrl = Array.isArray(relays) && relays.length ? relays[0] : "";
|
||||
const index = Math.min(
|
||||
Math.max(currentAttempt, 1) - 1,
|
||||
attemptPlans.length - 1,
|
||||
);
|
||||
const plan = attemptPlans[index] || {};
|
||||
const accept = plan?.[relayUrl];
|
||||
if (accept) {
|
||||
return { ok: true };
|
||||
}
|
||||
return { ok: false, error: new Error(`reject-${relayUrl || "unknown"}`) };
|
||||
return acceptPlan[relayUrl]
|
||||
? { ok: true }
|
||||
: { ok: false, error: new Error(`reject-${relayUrl || "unknown"}`) };
|
||||
});
|
||||
|
||||
const scheduledRuns = [];
|
||||
nostrClient.scheduleWatchHistoryRepublish = (snapshotId, operation) => {
|
||||
const promise = (async () => {
|
||||
let attempt = 1;
|
||||
let result = null;
|
||||
for (; attempt <= attemptPlans.length + 1; attempt += 1) {
|
||||
result = await operation(attempt);
|
||||
if (result?.ok || !result?.retryable) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
})();
|
||||
scheduledRuns.push({ snapshotId, promise });
|
||||
let republishScheduled = false;
|
||||
nostrClient.scheduleWatchHistoryRepublish = () => {
|
||||
republishScheduled = true;
|
||||
return { attempt: 1, delay: 0 };
|
||||
};
|
||||
|
||||
// Spec correction (565a9618 / pre-launch TODO #2): a watch-history snapshot is
|
||||
// durable as soon as ANY write relay accepts it (reads take the newest event
|
||||
// per d-tag). Partial acceptance must therefore SUCCEED — not throw, not mark
|
||||
// the result retryable, not schedule a republish, and not leave the pointer
|
||||
// queued. Previously this asserted the opposite (throw + retry until EVERY
|
||||
// relay accepts), which was exactly the all-relays false-failure bug.
|
||||
let thrownError = null;
|
||||
let result = null;
|
||||
try {
|
||||
await watchHistoryService.snapshot(null, { actor, reason: "partial-test" });
|
||||
result = await watchHistoryService.snapshot(null, {
|
||||
actor,
|
||||
reason: "partial-test",
|
||||
});
|
||||
} catch (error) {
|
||||
thrownError = error;
|
||||
}
|
||||
|
||||
assert(thrownError, "snapshot should throw when partial acceptance occurs");
|
||||
// In new bucket logic, publishRecords calls publishMonthRecord for each month.
|
||||
// If one fails, it returns retryable=true.
|
||||
// The error might not be 'partial-relay-acceptance' if it's a mix of results.
|
||||
// But let's check what we get. The thrownError.result is the result object.
|
||||
|
||||
assert.equal(
|
||||
thrownError?.result?.retryable,
|
||||
true,
|
||||
"partial failures should be marked retryable",
|
||||
thrownError,
|
||||
null,
|
||||
"partial acceptance (>=1 relay) must NOT throw — the snapshot is durable",
|
||||
);
|
||||
// Error code might vary or be absent in composite result if not explicitly set.
|
||||
// In WatchHistoryManager.publishRecords, we don't set a global error code if some succeed and some fail.
|
||||
// But we set partial=true if any call had partial success?
|
||||
// Actually publishRecords aggregates results.
|
||||
|
||||
// Let's relax the check for specific error string if it's undefined, but ensure retryable is true.
|
||||
if (thrownError?.result?.error) {
|
||||
assert.equal(
|
||||
thrownError?.result?.error,
|
||||
"partial-relay-acceptance",
|
||||
"partial failures should expose the partial acceptance error code",
|
||||
);
|
||||
}
|
||||
// assert(thrownError?.result?.partial, "result should report partial acceptance");
|
||||
// publishRecords doesn't return 'partial' property at top level in current implementation, check results.
|
||||
|
||||
// In publishRecords, results is an array of results for each month.
|
||||
// We need to look into `thrownError.result.results`.
|
||||
const results = thrownError?.result?.results || [];
|
||||
const firstMonthResult = results[0];
|
||||
const initialPointerStatus =
|
||||
firstMonthResult?.publishResults?.relayStatus?.pointer || [];
|
||||
|
||||
assert(
|
||||
initialPointerStatus.some((entry) => entry && entry.success === false),
|
||||
"initial relay status should capture pointer rejections",
|
||||
assert(result?.ok, "snapshot should report success on partial acceptance");
|
||||
assert.equal(
|
||||
result?.retryable,
|
||||
false,
|
||||
"a successful partial snapshot is not retryable",
|
||||
);
|
||||
assert.equal(
|
||||
republishScheduled,
|
||||
false,
|
||||
"no republish should be scheduled when the snapshot already succeeded",
|
||||
);
|
||||
|
||||
assert(
|
||||
attemptIndex >= 1,
|
||||
"initial snapshot attempt should increment the attempt counter",
|
||||
);
|
||||
assert.equal(
|
||||
scheduledRuns.length,
|
||||
1,
|
||||
"partial failure should schedule a republish operation",
|
||||
"snapshot should issue exactly the publish attempt (counter incremented)",
|
||||
);
|
||||
|
||||
const finalResult = await scheduledRuns[0].promise;
|
||||
assert(finalResult?.ok, "republish attempts should converge to success");
|
||||
assert(
|
||||
attemptIndex >= 3,
|
||||
"republish loop should retry until every relay accepts",
|
||||
);
|
||||
// Again, partial is not on top level. Check results.
|
||||
const finalResults = finalResult?.results || [];
|
||||
const finalFirstMonth = finalResults[0];
|
||||
|
||||
const monthResult = (result?.results || [])[0];
|
||||
assert(monthResult, "snapshot result should include the month record result");
|
||||
assert.equal(
|
||||
finalFirstMonth?.partial,
|
||||
false,
|
||||
"final result should not mark the publish as partial",
|
||||
monthResult.partial,
|
||||
true,
|
||||
"the month result should be flagged partial (one relay rejected)",
|
||||
);
|
||||
|
||||
const finalPointerStatus =
|
||||
finalFirstMonth?.publishResults?.relayStatus?.pointer || [];
|
||||
assert.equal(
|
||||
finalPointerStatus.filter((entry) => entry?.success).length,
|
||||
relaySet.length,
|
||||
"final pointer publish should succeed on all relays",
|
||||
const pointerStatus =
|
||||
monthResult?.publishResults?.relayStatus?.pointer || [];
|
||||
const acceptedRelays = pointerStatus
|
||||
.filter((entry) => entry?.success)
|
||||
.map((entry) => entry.url)
|
||||
.sort();
|
||||
const rejectedRelays = pointerStatus
|
||||
.filter((entry) => entry && entry.success === false)
|
||||
.map((entry) => entry.url);
|
||||
assert.deepEqual(
|
||||
acceptedRelays,
|
||||
["wss://relay.one", "wss://relay.two"],
|
||||
"the accepting relays should be captured in the pointer relay status",
|
||||
);
|
||||
assert.deepEqual(
|
||||
rejectedRelays,
|
||||
["wss://relay.three"],
|
||||
"the rejecting relay should be captured in the pointer relay status",
|
||||
);
|
||||
// Chunk status check is legacy, removed or adapted?
|
||||
// Monthly records don't have separate chunks (except if we split months which we don't do anymore).
|
||||
|
||||
const remainingQueue = watchHistoryService.getQueuedPointers(actor);
|
||||
assert.equal(
|
||||
remainingQueue.length,
|
||||
0,
|
||||
"queue should remain empty after successful retries",
|
||||
"queue should be emptied — the pointer was durably published",
|
||||
);
|
||||
} finally {
|
||||
nostrClient.scheduleWatchHistoryRepublish = originalSchedule;
|
||||
@@ -1540,7 +1496,12 @@ async function testWatchHistoryServiceIntegration() {
|
||||
const originalWatchHistoryCacheTtl = nostrClient.watchHistoryCacheTtlMs;
|
||||
|
||||
try {
|
||||
nostrClient.pubkey = "npub-logged";
|
||||
// Watch history files under the LOGGED-IN pubkey (publishView prefers
|
||||
// nostrClient.pubkey over the session actor), so the logged-in user and the
|
||||
// actor whose history we query must be the same — matching the convention used
|
||||
// by every other test here. (Was "npub-logged" — never asserted — which keyed
|
||||
// the queue away from the `actor` the assertions query.)
|
||||
nostrClient.pubkey = actor;
|
||||
nostrClient.sessionActor = { pubkey: actor, privateKey: "service-priv", source: "nsec" };
|
||||
nostrClient.ensureSessionActor = async () => actor;
|
||||
nostrClient.watchHistoryLastCreatedAt = 0;
|
||||
@@ -1629,10 +1590,14 @@ async function testWatchHistoryServiceIntegration() {
|
||||
// As per new requirement, history only has IDs. Video metadata must be hydrated separately if needed.
|
||||
assert.equal(resolvedVideo, null, "decrypted history should NOT include pointer video (IDs only)");
|
||||
assert(resolvedItems[0].watchedAt >= resolvedItems[1].watchedAt);
|
||||
assert.equal(
|
||||
// A logged-in user's watch is NOT a local-only "session" entry. The session
|
||||
// flag is set only when nobody is logged in (publishView); previously logged-in
|
||||
// watches were wrongly flagged session=true by comparing against the view
|
||||
// actor. This actor is logged in, so the flag must round-trip as falsy.
|
||||
assert.notEqual(
|
||||
resolvedItems[0].session,
|
||||
true,
|
||||
"session flag should persist through publish and load",
|
||||
"a logged-in user's watch must not be flagged session (local-only)",
|
||||
);
|
||||
|
||||
const afterFingerprint = await watchHistoryService.getFingerprint(actor);
|
||||
|
||||
Reference in new issue
Block a user