PolyTools Journal 0.1.1 / src/lib/journal-load.js

217 lines · 9,756 bytes · SHA-256 45e67b3100e2fa196409fef96f5f1b2de5f4b1b7f81aa006374412fc6a014580

1/*
2 * PolyTools — journal data pipeline: Data API v2 + Gamma → normalized positions.
3 * Needs PWL.api and PWL.journal. `cache` is optional: { getMany(keys) → Promise<obj>, setMany(obj) → Promise }.
4 * Cached forever: resolved token payouts ("t:<token>") — they never change once on-chain.
5 * Cached with TTL: event → category ("e:<eventId>" = {c, at}).
6 */
7(function (root) {
8 'use strict';
9
10 const EVENT_TTL_MS = 30 * 86400000;
11 const SERIES_MIN_EVENTS = 8; // a slug pattern with this many events is resolved from a couple of samples
12 const SERIES_SAMPLES = 2;
13 const FLUSH_EVERY = 500; // cache writes happen during the load, so Stop or a failure keeps what was fetched
14 const BREAKER = 25; // consecutive category failures before we stop asking (API down / blocked)
15
16 /** "btc-updown-5m-1793035200" → "btc-updown-#m-#": recurring markets share a category. */
17 function slugStem(slug) { return String(slug || '').toLowerCase().replace(/\d+/g, '#'); }
18
19 async function loadJournal(address, o) {
20 const { api, journal } = root.PWL;
21 // Measured 2026-10-28 (big-wallet run): tags at concurrency 12 → ~35 req/s direct, ~16 via a proxy, no 429s.
22 const opts = Object.assign({ max: 5000, cache: null, signal: null, onProgress: () => {}, concurrency: 8 }, o);
23 const progress = (step, done, total) => opts.onProgress({ step, done, total });
24 const signal = opts.signal;
25 const cache = opts.cache;
26
27 progress('profile');
28 const [profile, stats] = await Promise.all([
29 api.publicProfile(address, { signal }),
30 api.userStats(address, { signal }).catch(() => null),
31 ]);
32
33 progress('pnl');
34 const pnl = await api.userPnl(address, { signal }).catch(() => []);
35
36 // closed and open walk their own cursor chains: run them side by side
37 progress('closed', 0);
38 // Both walks share one controller: if either fails (or the user stops), the other stops paging too.
39 const both = new AbortController();
40 const relay = () => both.abort();
41 if (signal) { if (signal.aborted) both.abort(); else signal.addEventListener('abort', relay, { once: true }); }
42 let closed, open;
43 try {
44 [closed, open] = await Promise.all([
45 api.positions(address, 'CLOSED', { max: opts.max, signal: both.signal, onPage: (n) => progress('closed', n) })
46 .catch((e) => { both.abort(); throw e; }),
47 api.positions(address, 'OPEN', { max: opts.max, signal: both.signal })
48 .catch((e) => { both.abort(); throw e; }),
49 ]);
50 } finally {
51 if (signal) signal.removeEventListener('abort', relay);
52 }
53 progress('open', open.rows.length);
54
55 // ---- token resolutions ----
56 const tokenIds = [...new Set(closed.rows.concat(open.rows).map((r) => String(r.token_id)).filter(Boolean))];
57 const tokenMap = new Map();
58 if (cache) {
59 const hit = await cache.getMany(tokenIds.map((t) => 't:' + t));
60 for (const t of tokenIds) { const v = hit['t:' + t]; if (v) tokenMap.set(t, v); }
61 }
62 const missingTokens = tokenIds.filter((t) => !tokenMap.has(t));
63 progress('tokens', 0, missingTokens.length);
64 let tokenFailures = 0;
65 if (missingTokens.length) {
66 await api.tokens(missingTokens, {
67 signal, concurrency: 8,
68 onProgress: (d, n) => progress('tokens', d, n),
69 onError: (ids) => { tokenFailures += ids.length; }, // those positions simply stay "unresolved"
70 onBatch: async (batch) => {
71 const toCache = {};
72 for (const [t, info] of batch) {
73 const v = { resolved: !!info.resolved, final_price: info.final_price == null ? null : Number(info.final_price) };
74 tokenMap.set(t, v);
75 if (v.resolved && v.final_price != null) toCache['t:' + t] = v;
76 }
77 if (cache && Object.keys(toCache).length) await cache.setMany(toCache);
78 },
79 });
80 }
81
82 // ---- event categories ----
83 const allRows = closed.rows.concat(open.rows);
84 const slugOf = new Map();
85 for (const r of allRows) if (r.event_id) slugOf.set(String(r.event_id), r.event_slug || '');
86 const eventIds = [...slugOf.keys()];
87 const catMap = new Map();
88 const now = Date.now();
89 if (cache) {
90 const hit = await cache.getMany(eventIds.map((e) => 'e:' + e));
91 for (const e of eventIds) { const v = hit['e:' + e]; if (v && now - v.at < EVENT_TTL_MS) catMap.set(e, v.c); }
92 }
93 let missingEvents = eventIds.filter((e) => !catMap.has(e));
94 const total = missingEvents.length;
95 let done = 0, categoryFailures = 0, streak = 0, broken = false;
96 let evCache = {}, pending = 0;
97 const flush = async () => {
98 if (cache && pending) { const part = evCache; evCache = {}; pending = 0; await cache.setMany(part); }
99 };
100 const put = (id, c) => { catMap.set(id, c); evCache['e:' + id] = { c, at: now }; pending++; };
101 const fetchCat = async (id) => {
102 if (broken) { categoryFailures++; catMap.set(id, 'other'); return null; }
103 try {
104 const c = journal.categorize(await api.eventTags(id, { signal, retries: 3 }));
105 streak = 0;
106 put(id, c);
107 return c;
108 } catch (e) {
109 if (signal && signal.aborted) throw e;
110 categoryFailures++; // shown as "other" and reported on the page; not cached
111 if (++streak >= BREAKER) broken = true;
112 catMap.set(id, 'other');
113 return null;
114 }
115 };
116 progress('categories', 0, total);
117 try {
118 // 1) recurring series (5-minute crypto, daily weather, …): resolve a couple of samples per slug pattern
119 const groups = new Map();
120 for (const id of missingEvents) {
121 const stem = slugStem(slugOf.get(id));
122 if (!groups.has(stem)) groups.set(stem, []);
123 groups.get(stem).push(id);
124 }
125 const series = [...groups.entries()].filter(([stem, ids]) => stem.includes('#') && ids.length >= SERIES_MIN_EVENTS);
126 if (series.length) {
127 const known = cache ? await cache.getMany(series.map(([stem]) => 's:' + stem)) : {};
128 await api.pool(series, opts.concurrency, async ([stem, ids]) => {
129 let c = null;
130 const k = known['s:' + stem];
131 if (k && now - k.at < EVENT_TTL_MS) c = k.c;
132 else {
133 const picks = [ids[0], ids[Math.floor(ids.length / 2)], ids[ids.length - 1]].slice(0, SERIES_SAMPLES);
134 const got = [];
135 for (const id of picks) got.push(await fetchCat(id));
136 if (got.every((x) => x && x === got[0])) {
137 c = got[0];
138 if (cache) await cache.setMany({ ['s:' + stem]: { c, at: now } });
139 }
140 }
141 if (c) { for (const id of ids) if (!catMap.has(id)) put(id, c); }
142 done = [...missingEvents].filter((id) => catMap.has(id)).length;
143 progress('categories', done, total);
144 });
145 missingEvents = missingEvents.filter((id) => !catMap.has(id));
146 await flush();
147 }
148 // 2) everything else, one event at a time
149 done = 0;
150 await api.pool(missingEvents, opts.concurrency, async (id) => {
151 await fetchCat(id);
152 done++;
153 if (pending >= FLUSH_EVERY) await flush();
154 if (done % 10 === 0 || done === total) progress('categories', Math.min(done + (total - missingEvents.length), total), total);
155 });
156 } finally {
157 await flush(); // keep what we already have, also on Stop
158 }
159
160 progress('build');
161 const positions = journal.buildPositions(closed.rows, open.rows, tokenMap, catMap);
162 return {
163 address, profile, stats, pnl, positions,
164 truncated: closed.truncated || open.truncated, categoryFailures, tokenFailures,
165 counts: { closed: closed.rows.length, open: open.rows.length, tokens: tokenIds.length, events: eventIds.length },
166 loadedAt: Date.now(),
167 };
168 }
169
170 /**
171 * chrome.storage.local adapter for the cache interface. The cache is an optimisation: reads/writes are chunked
172 * and any storage error (quota, context gone) is swallowed so it can never break a journal load.
173 */
174 function chromeCache() {
175 const CHUNK = 2000;
176 return {
177 getMany: async (keys) => {
178 const out = {};
179 for (let i = 0; i < keys.length; i += CHUNK) {
180 const part = await new Promise((resolve) => {
181 try {
182 chrome.storage.local.get(keys.slice(i, i + CHUNK), (r) => resolve(chrome.runtime.lastError ? {} : (r || {})));
183 } catch (e) { resolve({}); }
184 });
185 Object.assign(out, part);
186 }
187 return out;
188 },
189 setMany: async (obj) => {
190 const entries = Object.entries(obj);
191 for (let i = 0; i < entries.length; i += CHUNK) {
192 const ok = await new Promise((resolve) => {
193 try {
194 chrome.storage.local.set(Object.fromEntries(entries.slice(i, i + CHUNK)), () => resolve(!chrome.runtime.lastError));
195 } catch (e) { resolve(false); }
196 });
197 if (!ok) break; // quota or context error: stop caching, keep going without it
198 }
199 },
200 };
201 }
202
203 /** In-memory adapter (Node tests). */
204 function memoryCache() {
205 const m = new Map();
206 return {
207 getMany: async (keys) => Object.fromEntries(keys.filter((k) => m.has(k)).map((k) => [k, m.get(k)])),
208 setMany: async (obj) => { for (const [k, v] of Object.entries(obj)) m.set(k, v); },
209 size: () => m.size,
210 };
211 }
212
213 const loader = { loadJournal, chromeCache, memoryCache, slugStem };
214 root.PWL = root.PWL || {};
215 root.PWL.loader = loader;
216 if (typeof module !== 'undefined' && module.exports) module.exports = loader;
217})(typeof globalThis !== 'undefined' ? globalThis : this);