File size: 6,209 Bytes
ee888e1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | import { createHash } from 'node:crypto';
import { buildDedupMaterial } from '../shared/notification-dedup.cjs';
import { buildWatchlistStoryEvents, resolveWatchlistScoreMin, WATCHLIST_STORY_EVENT_TYPE } from './watchlist-story-events.mjs';
export const WATCHLIST_SCAN_DEDUP_TTL_SECONDS = 24 * 60 * 60;
const WATCHLIST_SCAN_WINDOW_MS = 24 * 60 * 60 * 1000;
const WATCHLIST_SCAN_ACCUMULATORS = [
'digest:accumulator:v1:full:en',
'digest:accumulator:v1:finance:en',
];
function flatArrayToObject(flat) {
const obj = Object.create(null);
if (!Array.isArray(flat)) return obj;
for (let i = 0; i + 1 < flat.length; i += 2) {
obj[flat[i]] = flat[i + 1];
}
return obj;
}
export async function publishWatchlistNotificationEvent(
{ eventType, payload, severity, dedupTtl = WATCHLIST_SCAN_DEDUP_TTL_SECONDS },
{ upstashRest, nowMs = () => Date.now(), logger = console } = {},
) {
if (typeof upstashRest !== 'function') throw new Error('upstashRest dependency is required');
const dedupMaterial = buildDedupMaterial(eventType, payload?.title, payload?.coalesceKey);
const dedupHash = createHash('sha256').update(dedupMaterial).digest('hex').slice(0, 16);
const dedupKey = `wm:notif:scan-dedup:${eventType}:${dedupHash}`;
const isNew = (await upstashRest('SET', dedupKey, '1', 'NX', 'EX', String(dedupTtl))) === 'OK';
if (!isNew) return false;
const msg = JSON.stringify({ eventType, payload, severity, publishedAt: nowMs() });
let pushed;
try {
pushed = await upstashRest('LPUSH', 'wm:events:queue', msg);
} catch (err) {
logger.warn?.(`[digest] watchlist LPUSH failed for ${eventType} - rolling back dedup key: ${err?.message ?? err}`);
try { await upstashRest('DEL', dedupKey); } catch {}
return false;
}
if (typeof pushed !== 'number') {
logger.warn?.(`[digest] watchlist LPUSH failed for ${eventType} - rolling back dedup key`);
try { await upstashRest('DEL', dedupKey); } catch {}
return false;
}
logger.log?.(
`[digest] watchlist queued ${severity} ${eventType}: ` +
`${String(payload?.title ?? '').slice(0, 60)} tickers=${(payload?.tickers ?? []).join(',')}`,
);
return true;
}
export async function scanAndEnqueueWatchlistStoryEvents(nowMs, {
env = process.env,
upstashRest,
upstashPipeline,
readStoryTracksChunked,
tickerDictionary,
publishNotificationEvent,
logger = console,
accumulators = WATCHLIST_SCAN_ACCUMULATORS,
scanWindowMs = WATCHLIST_SCAN_WINDOW_MS,
} = {}) {
try {
if (typeof upstashRest !== 'function') throw new Error('upstashRest dependency is required');
if (typeof upstashPipeline !== 'function') throw new Error('upstashPipeline dependency is required');
if (typeof readStoryTracksChunked !== 'function') throw new Error('readStoryTracksChunked dependency is required');
const scoreMin = resolveWatchlistScoreMin(env);
const windowStart = String(nowMs - scanWindowMs);
const seenHashes = new Set();
const hashes = [];
const memberLists = await Promise.all(
accumulators.map((accKey) =>
upstashRest('ZRANGEBYSCORE', accKey, windowStart, String(nowMs)),
),
);
for (const members of memberLists) {
if (!Array.isArray(members)) continue;
for (const h of members) {
if (typeof h === 'string' && h.length > 0 && !seenHashes.has(h)) {
seenHashes.add(h);
hashes.push(h);
}
}
}
if (hashes.length === 0) return { hashes: 0, candidates: 0, events: 0, enqueued: 0, scoreMin };
const trackResults = await readStoryTracksChunked(hashes, upstashPipeline);
if (trackResults === null) {
logger.warn?.('[digest] watchlist scan: story-track read failed - skipping this tick');
return { hashes: hashes.length, candidates: 0, events: 0, enqueued: 0, scoreMin, skipped: 'track_read_failed' };
}
const candidates = [];
for (let i = 0; i < hashes.length; i++) {
const raw = trackResults[i]?.result;
if (!Array.isArray(raw) || raw.length === 0) continue;
const track = flatArrayToObject(raw);
if (!track.title) continue;
const currentScore = parseInt(track.currentScore ?? '0', 10);
if (!Number.isFinite(currentScore) || currentScore < scoreMin) continue;
candidates.push({
hash: hashes[i],
title: track.title,
description: typeof track.description === 'string' ? track.description : '',
link: track.link ?? '',
source: '',
currentScore,
});
}
if (candidates.length === 0) return { hashes: hashes.length, candidates: 0, events: 0, enqueued: 0, scoreMin };
const eventEntries = [];
for (const candidate of candidates) {
for (const event of buildWatchlistStoryEvents([candidate], tickerDictionary, scoreMin)) {
eventEntries.push({ event, sourceKey: `story:sources:v1:${candidate.hash}` });
}
}
if (eventEntries.length > 0) {
try {
const srcResults = await upstashPipeline(
eventEntries.map(({ sourceKey }) => ['SMEMBERS', sourceKey]),
);
for (let i = 0; i < eventEntries.length; i++) {
const arr = srcResults[i]?.result;
if (Array.isArray(arr) && typeof arr[0] === 'string') eventEntries[i].event.payload.source = arr[0];
}
} catch { /* best-effort */ }
}
const events = eventEntries.map(({ event }) => event);
const publish = publishNotificationEvent ??
((event) => publishWatchlistNotificationEvent(event, { upstashRest, logger }));
let enqueued = 0;
for (const ev of events) {
if (await publish(ev)) enqueued++;
}
logger.log?.(
`[digest] watchlist scan: hashes=${hashes.length} candidates=${candidates.length} ` +
`events=${events.length} enqueued=${enqueued} score_min=${scoreMin} ` +
`event_type=${WATCHLIST_STORY_EVENT_TYPE}`,
);
return { hashes: hashes.length, candidates: candidates.length, events: events.length, enqueued, scoreMin };
} catch (err) {
logger.warn?.(`[digest] watchlist scan failed (non-fatal): ${err?.message ?? err}`);
return { hashes: 0, candidates: 0, events: 0, enqueued: 0, error: err?.message ?? String(err) };
}
}
|