diff --git a/test/usage.test.js b/test/usage.test.js index 3f00317..4b2e4c9 100644 --- a/test/usage.test.js +++ b/test/usage.test.js @@ -71,3 +71,27 @@ test('usage log: an MCP reply sent as an event stream is read (the verdict is lo assert.equal(e.route, 'check_wallet_approvals'); assert.deepEqual(e.result, { verdict: 'RISK' }); }); + +test('usage log: writes are batched (one read and one write for several calls), and a day file over 1 MB keeps its lines', async () => { + const { createUsageLog } = require('../usage-log'); + const path = 'events/plaintext/2026-10-14.jsonl'; + const old = `${'x'.repeat(1100 * 1024)}\n`; + let file = old; + const calls = []; + const fetchFn = async (url, opts = {}) => { + const method = opts.method || 'GET'; + calls.push(`${method} ${new URL(url).pathname}`); + // Like GitHub: the JSON form has no content for files over 1 MB, the raw form has the bytes. + if (method === 'GET') return /raw/.test(opts.headers?.accept || '') ? new Response(file) : Response.json({ sha: 'x', content: '' }); + file = Buffer.from(JSON.parse(opts.body).content, 'base64').toString('utf8'); + return Response.json({}, { status: 200 }); + }; + const log = createUsageLog({ service: 'plaintext', env: { USAGE_LOG_TOKEN: 't' }, fetchFn, now: () => new Date('2026-10-14T10:00:00Z'), log: { warn() {}, error() {} } }); + log.record({ route: 'check-wallet', status: 402 }); + log.record({ route: 'explain', status: 200 }); + assert.equal(calls.length, 0, 'nothing written until the flush'); + await log.flush(); + assert.ok(file.startsWith(old), 'the old lines are still there'); + assert.equal(file.slice(old.length).trim().split('\n').length, 2); + assert.deepEqual(calls, [`GET /repos/Fizzl13/usage-log/contents/${path}`, `PUT /repos/Fizzl13/usage-log/contents/${path}`]); +}); diff --git a/usage-log.js b/usage-log.js index 1df3531..0e6ed33 100644 --- a/usage-log.js +++ b/usage-log.js @@ -12,6 +12,12 @@ const crypto = require('node:crypto'); +// Git's blob id of a file's bytes: what the contents API calls its sha. +const blobSha = (buf) => crypto.createHash('sha1').update(`blob ${buf.length}\0`).update(buf).digest('hex'); +const MAX_QUEUE = 200; +// Tests pass their own fetch; only the real service exits on SIGTERM. +const fetchFnGiven = (fn) => fn !== globalThis.fetch; + const DEFAULT_REPO = 'Fizzl13/usage-log'; const API = 'https://api.github.com'; const MAX_TEXT = 300; @@ -141,12 +147,20 @@ function x402Refusal(res) { return null; } -function createUsageLog({ service, env = process.env, fetchFn = globalThis.fetch, now = () => new Date(), log = console } = {}) { +function createUsageLog({ service, env = process.env, fetchFn = globalThis.fetch, now = () => new Date(), log = console, flushDelayMs = 10 * 60 * 1000, exitOnSignal = !fetchFnGiven(fetchFn) } = {}) { const token = env.USAGE_LOG_TOKEN; const repo = env.USAGE_LOG_REPO || DEFAULT_REPO; const visitorSecret = env.USAGE_LOG_SALT || token; const queue = []; let flushing = null; + let timer = null; + // Every write re-sends the whole day file to GitHub, so writes are batched: at most one per flushMs + // (USAGE_LOG_FLUSH_MS, default 10 minutes). On shutdown (a deploy, or the free plan going to sleep) the queue is + // written first. + const flushMs = Number(env.USAGE_LOG_FLUSH_MS) || flushDelayMs; + if (token && exitOnSignal) { + process.once('SIGTERM', () => { flush().finally(() => process.exit(0)); }); + } let warned = false; const headers = { @@ -160,13 +174,15 @@ function createUsageLog({ service, env = process.env, fetchFn = globalThis.fetch async function append(path, lines) { for (let attempt = 0; attempt < 4; attempt++) { const url = `${API}/repos/${repo}/contents/${path}`; - const current = await fetchFn(url, { headers }); + // The raw file: the JSON form leaves out the content of files over 1 MB, which would then be overwritten + // with only the new lines. The sha GitHub wants back is the git blob sha of exactly these bytes. + const current = await fetchFn(url, { headers: { ...headers, accept: 'application/vnd.github.raw+json' } }); let sha; let text = ''; if (current.status === 200) { - const file = await current.json(); - sha = file.sha; - text = Buffer.from(file.content || '', 'base64').toString('utf8'); + const raw = Buffer.from(await current.arrayBuffer()); + sha = blobSha(raw); + text = raw.toString('utf8'); } else if (current.status !== 404) { throw new Error(`read ${path}: HTTP ${current.status}`); } @@ -184,6 +200,7 @@ function createUsageLog({ service, env = process.env, fetchFn = globalThis.fetch } async function flush() { + if (timer) { clearTimeout(timer); timer = null; } if (flushing) return flushing; flushing = (async () => { while (queue.length) { @@ -215,7 +232,11 @@ function createUsageLog({ service, env = process.env, fetchFn = globalThis.fetch return; } queue.push({ t: now().toISOString(), service, ...event, input: clipAll(event.input), result: clipAll(event.result) }); - flush(); + if (queue.length >= MAX_QUEUE) flush(); + else if (!timer) { + timer = setTimeout(() => { timer = null; flush(); }, flushMs); + timer.unref?.(); + } } // Express middleware. describe(req, res, body) returns { route, input, result, via, payment? }