mirror of
https://github.com/EvoMap/evolver.git
synced 2026-09-18 21:47:53 +08:00
evolver-publish
758 lines
25 KiB
JavaScript
758 lines
25 KiB
JavaScript
'use strict';
|
|
|
|
const { describe, it, before, after } = require('node:test');
|
|
const assert = require('node:assert/strict');
|
|
const fs = require('fs');
|
|
const os = require('os');
|
|
const path = require('path');
|
|
|
|
const { MailboxStore, generateUUIDv7, DEFAULT_CHANNEL, SCHEMA_VERSION, PROXY_PROTOCOL_VERSION } = require('../src/proxy/mailbox/store');
|
|
|
|
function tmpDataDir() {
|
|
return fs.mkdtempSync(path.join(os.tmpdir(), 'mailbox-test-'));
|
|
}
|
|
|
|
function withEnv(name, value, fn) {
|
|
const saved = process.env[name];
|
|
process.env[name] = value;
|
|
try {
|
|
return fn();
|
|
} finally {
|
|
if (saved === undefined) delete process.env[name];
|
|
else process.env[name] = saved;
|
|
}
|
|
}
|
|
|
|
function withFixedNow(now, fn) {
|
|
const originalNow = Date.now;
|
|
Date.now = () => now;
|
|
try {
|
|
return fn();
|
|
} finally {
|
|
Date.now = originalNow;
|
|
}
|
|
}
|
|
|
|
function inboundRow({ id, type, text, now }) {
|
|
return {
|
|
id,
|
|
channel: DEFAULT_CHANNEL,
|
|
direction: 'inbound',
|
|
type,
|
|
status: 'pending',
|
|
payload: { text },
|
|
priority: 'normal',
|
|
ref_id: null,
|
|
created_at: now,
|
|
synced_at: null,
|
|
expires_at: null,
|
|
retry_count: 0,
|
|
next_retry_at: null,
|
|
error: null,
|
|
};
|
|
}
|
|
|
|
function inboundTextForLineBytes(maxBytes, { id, type, now }) {
|
|
const baseLine = JSON.stringify(inboundRow({ id, type, text: '', now })) + '\n';
|
|
const fillerBytes = maxBytes - Buffer.byteLength(baseLine, 'utf8');
|
|
assert.ok(fillerBytes >= 0, 'test fixture base row must fit below max line bytes');
|
|
return 'a'.repeat(fillerBytes);
|
|
}
|
|
|
|
describe('generateUUIDv7', () => {
|
|
it('returns a valid UUID v7 format', () => {
|
|
const id = generateUUIDv7();
|
|
assert.match(id, /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/);
|
|
});
|
|
|
|
it('generates unique IDs', () => {
|
|
const ids = new Set();
|
|
for (let i = 0; i < 1000; i++) ids.add(generateUUIDv7());
|
|
assert.equal(ids.size, 1000);
|
|
});
|
|
|
|
it('generates IDs with non-decreasing timestamp prefix', () => {
|
|
const ids = [];
|
|
for (let i = 0; i < 10; i++) ids.push(generateUUIDv7());
|
|
for (let i = 1; i < ids.length; i++) {
|
|
const prevTs = ids[i - 1].slice(0, 13);
|
|
const currTs = ids[i].slice(0, 13);
|
|
assert.ok(currTs >= prevTs, `timestamp prefix should be non-decreasing: ${prevTs} <= ${currTs}`);
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('MailboxStore', () => {
|
|
let store;
|
|
let dataDir;
|
|
|
|
before(() => {
|
|
dataDir = tmpDataDir();
|
|
store = new MailboxStore(dataDir);
|
|
});
|
|
|
|
after(() => {
|
|
store.close();
|
|
try { fs.rmSync(dataDir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
describe('send()', () => {
|
|
it('creates an outbound message with correct fields', () => {
|
|
const result = store.send({ type: 'asset_submit', payload: { data: 'test' } });
|
|
assert.ok(result.message_id);
|
|
assert.equal(result.status, 'pending');
|
|
|
|
const msg = store.getById(result.message_id);
|
|
assert.equal(msg.direction, 'outbound');
|
|
assert.equal(msg.type, 'asset_submit');
|
|
assert.equal(msg.status, 'pending');
|
|
assert.equal(msg.channel, DEFAULT_CHANNEL);
|
|
assert.deepEqual(msg.payload, { data: 'test' });
|
|
assert.equal(msg.priority, 'normal');
|
|
});
|
|
|
|
it('supports custom channel and priority', () => {
|
|
const result = store.send({
|
|
type: 'dm',
|
|
payload: { text: 'hello' },
|
|
channel: 'custom-channel',
|
|
priority: 'high',
|
|
});
|
|
const msg = store.getById(result.message_id);
|
|
assert.equal(msg.channel, 'custom-channel');
|
|
assert.equal(msg.priority, 'high');
|
|
});
|
|
|
|
it('supports string payload', () => {
|
|
const result = store.send({ type: 'test', payload: '{"raw": true}' });
|
|
const msg = store.getById(result.message_id);
|
|
assert.deepEqual(msg.payload, { raw: true });
|
|
});
|
|
});
|
|
|
|
describe('writeInbound()', () => {
|
|
it('creates an inbound message', () => {
|
|
const id = store.writeInbound({ type: 'task_available', payload: { task_id: 't1' } });
|
|
assert.ok(id);
|
|
|
|
const msg = store.getById(id);
|
|
assert.equal(msg.direction, 'inbound');
|
|
assert.equal(msg.type, 'task_available');
|
|
assert.equal(msg.status, 'pending');
|
|
});
|
|
|
|
it('accepts a custom id', () => {
|
|
const customId = generateUUIDv7();
|
|
const id = store.writeInbound({ id: customId, type: 'hub_event', payload: {} });
|
|
assert.equal(id, customId);
|
|
});
|
|
|
|
it('ignores duplicate inbound ids (community PR #515)', () => {
|
|
// At-least-once delivery from the Hub or retry loops can replay the
|
|
// same message id. The second write must be a no-op: the stored
|
|
// payload stays as the first write, countPending does not double,
|
|
// and poll() still sees exactly one row.
|
|
const customId = generateUUIDv7();
|
|
store.writeInbound({ id: customId, type: 'hub_event', payload: { n: 1 } });
|
|
const duplicateId = store.writeInbound({ id: customId, type: 'hub_event', payload: { n: 2 } });
|
|
assert.equal(duplicateId, customId, 'duplicate write must still return the original id');
|
|
const pendingMatches = store.poll({ type: 'hub_event' }).filter(m => m.id === customId);
|
|
assert.equal(pendingMatches.length, 1, 'duplicate write must not produce a second row');
|
|
assert.deepEqual(pendingMatches[0].payload, { n: 1 }, 'first write wins (idempotent)');
|
|
});
|
|
});
|
|
|
|
describe('writeInboundBatch()', () => {
|
|
it('writes multiple inbound messages', () => {
|
|
const ids = store.writeInboundBatch([
|
|
{ type: 'dm', payload: { text: 'a' } },
|
|
{ type: 'dm', payload: { text: 'b' } },
|
|
{ type: 'dm', payload: { text: 'c' } },
|
|
]);
|
|
assert.equal(ids.length, 3);
|
|
for (const id of ids) {
|
|
const msg = store.getById(id);
|
|
assert.ok(msg);
|
|
assert.equal(msg.direction, 'inbound');
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('poll()', () => {
|
|
it('returns pending inbound messages', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
store2.writeInbound({ type: 'task_available', payload: { id: 1 } });
|
|
store2.writeInbound({ type: 'task_available', payload: { id: 2 } });
|
|
store2.writeInbound({ type: 'hub_event', payload: { id: 3 } });
|
|
|
|
const all = store2.poll();
|
|
assert.equal(all.length, 3);
|
|
|
|
const tasks = store2.poll({ type: 'task_available' });
|
|
assert.equal(tasks.length, 2);
|
|
|
|
store2.close();
|
|
});
|
|
|
|
it('respects limit', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
for (let i = 0; i < 10; i++) {
|
|
store2.writeInbound({ type: 'test', payload: { i } });
|
|
}
|
|
const limited = store2.poll({ limit: 3 });
|
|
assert.equal(limited.length, 3);
|
|
store2.close();
|
|
});
|
|
});
|
|
|
|
describe('pollOutbound()', () => {
|
|
it('returns pending outbound messages ordered by priority', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
store2.send({ type: 'low', payload: {}, priority: 'low' });
|
|
store2.send({ type: 'high', payload: {}, priority: 'high' });
|
|
store2.send({ type: 'normal', payload: {} });
|
|
|
|
const msgs = store2.pollOutbound();
|
|
assert.equal(msgs[0].priority, 'high');
|
|
store2.close();
|
|
});
|
|
});
|
|
|
|
describe('ack()', () => {
|
|
it('marks inbound messages as delivered', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
const id = store2.writeInbound({ type: 'test', payload: {} });
|
|
|
|
const count = store2.ack(id);
|
|
assert.equal(count, 1);
|
|
|
|
const msg = store2.getById(id);
|
|
assert.equal(msg.status, 'delivered');
|
|
store2.close();
|
|
});
|
|
|
|
it('does not ack outbound messages', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
const { message_id } = store2.send({ type: 'test', payload: {} });
|
|
|
|
const count = store2.ack(message_id);
|
|
assert.equal(count, 0);
|
|
|
|
const msg = store2.getById(message_id);
|
|
assert.equal(msg.status, 'pending');
|
|
store2.close();
|
|
});
|
|
});
|
|
|
|
describe('updateStatus()', () => {
|
|
it('updates status and synced_at', () => {
|
|
const { message_id } = store.send({ type: 'status_test', payload: {} });
|
|
store.updateStatus(message_id, 'synced');
|
|
|
|
const msg = store.getById(message_id);
|
|
assert.equal(msg.status, 'synced');
|
|
assert.ok(msg.synced_at);
|
|
});
|
|
|
|
it('records error on failure', () => {
|
|
const { message_id } = store.send({ type: 'fail_test', payload: {} });
|
|
store.updateStatus(message_id, 'failed', { error: 'timeout' });
|
|
|
|
const msg = store.getById(message_id);
|
|
assert.equal(msg.status, 'failed');
|
|
assert.equal(msg.error, 'timeout');
|
|
});
|
|
});
|
|
|
|
describe('incrementRetry()', () => {
|
|
it('increments retry count and records error', () => {
|
|
const { message_id } = store.send({ type: 'retry_test', payload: {} });
|
|
store.incrementRetry(message_id, 'first error');
|
|
store.incrementRetry(message_id, 'second error');
|
|
|
|
const msg = store.getById(message_id);
|
|
assert.equal(msg.retry_count, 2);
|
|
assert.equal(msg.error, 'second error');
|
|
});
|
|
});
|
|
|
|
describe('list()', () => {
|
|
it('lists messages by type', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
store2.send({ type: 'list_test', payload: { n: 1 } });
|
|
store2.send({ type: 'list_test', payload: { n: 2 } });
|
|
store2.send({ type: 'other', payload: {} });
|
|
|
|
const results = store2.list({ type: 'list_test' });
|
|
assert.equal(results.length, 2);
|
|
store2.close();
|
|
});
|
|
|
|
it('requires type parameter', () => {
|
|
assert.throws(() => store.list({}), /type is required/);
|
|
});
|
|
});
|
|
|
|
describe('countPending()', () => {
|
|
it('counts pending messages', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
store2.send({ type: 'a', payload: {} });
|
|
store2.send({ type: 'b', payload: {} });
|
|
store2.writeInbound({ type: 'c', payload: {} });
|
|
|
|
assert.equal(store2.countPending({ direction: 'outbound' }), 2);
|
|
assert.equal(store2.countPending({ direction: 'inbound' }), 1);
|
|
store2.close();
|
|
});
|
|
|
|
it('counts pending messages by type', () => {
|
|
const store2 = new MailboxStore(tmpDataDir());
|
|
store2.send({ type: 'proxy_trace', payload: {} });
|
|
store2.send({ type: 'asset_submit', payload: {} });
|
|
store2.writeInbound({ type: 'proxy_trace_config', payload: {} });
|
|
|
|
assert.equal(store2.countPending({ direction: 'outbound', type: 'proxy_trace' }), 1);
|
|
assert.equal(store2.countPending({ direction: 'outbound', type: 'asset_submit' }), 1);
|
|
assert.equal(store2.countPending({ direction: 'inbound', type: 'proxy_trace_config' }), 1);
|
|
store2.close();
|
|
});
|
|
});
|
|
|
|
describe('sync cursors', () => {
|
|
it('gets and sets cursors', () => {
|
|
store.setCursor('evomap-hub:inbound_cursor', 'cursor_123');
|
|
assert.equal(store.getCursor('evomap-hub:inbound_cursor'), 'cursor_123');
|
|
|
|
store.setCursor('evomap-hub:inbound_cursor', 'cursor_456');
|
|
assert.equal(store.getCursor('evomap-hub:inbound_cursor'), 'cursor_456');
|
|
});
|
|
|
|
it('returns null for missing cursor', () => {
|
|
assert.equal(store.getCursor('nonexistent'), null);
|
|
});
|
|
});
|
|
|
|
describe('local state', () => {
|
|
it('gets and sets state', () => {
|
|
store.setState('node_id', 'node_abc123');
|
|
assert.equal(store.getState('node_id'), 'node_abc123');
|
|
});
|
|
|
|
it('overwrites existing state', () => {
|
|
store.setState('counter', '1');
|
|
store.setState('counter', '2');
|
|
assert.equal(store.getState('counter'), '2');
|
|
});
|
|
|
|
it('prevents stale explicit node_secret writes from overwriting disk hub_rotate tuple', () => {
|
|
const dir = tmpDataDir();
|
|
const oldStore = new MailboxStore(dir);
|
|
try {
|
|
oldStore.setNodeSecretState({
|
|
secret: 'a'.repeat(64),
|
|
version: '1',
|
|
source: 'env_seed',
|
|
envSuppressed: '',
|
|
});
|
|
|
|
const freshStore = new MailboxStore(dir);
|
|
freshStore.setNodeSecretState({
|
|
secret: 'b'.repeat(64),
|
|
version: '9',
|
|
source: 'hub_rotate',
|
|
envSuppressed: '',
|
|
});
|
|
freshStore.close();
|
|
|
|
oldStore.setState('node_secret', 'c'.repeat(64));
|
|
oldStore.setState('node_secret_version', '2');
|
|
oldStore.setState('node_secret_source', 'env_seed');
|
|
|
|
assert.equal(oldStore.getState('node_secret'), 'b'.repeat(64));
|
|
assert.equal(oldStore.getState('node_secret_version'), '9');
|
|
assert.equal(oldStore.getState('node_secret_source'), 'hub_rotate');
|
|
const stateRaw = JSON.parse(fs.readFileSync(path.join(dir, 'state.json'), 'utf8'));
|
|
assert.equal(stateRaw.node_secret, 'b'.repeat(64));
|
|
assert.equal(stateRaw.node_secret_version, '9');
|
|
assert.equal(stateRaw.node_secret_source, 'hub_rotate');
|
|
} finally {
|
|
oldStore.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
}
|
|
});
|
|
|
|
it('creates mailbox state files with owner-only permissions', {
|
|
skip: process.platform === 'win32' ? 'chmod not enforced on Windows' : false,
|
|
}, () => {
|
|
const parent = tmpDataDir();
|
|
const dir = path.join(parent, 'mailbox');
|
|
const oldUmask = process.umask(0o022);
|
|
let s;
|
|
try {
|
|
s = new MailboxStore(dir);
|
|
s.setState('node_secret', 'a'.repeat(64));
|
|
s.send({ type: 'permission_probe', payload: { ok: true } });
|
|
|
|
assert.equal(fs.statSync(dir).mode & 0o777, 0o700);
|
|
assert.equal(fs.statSync(s._stateFile).mode & 0o777, 0o600);
|
|
assert.equal(fs.statSync(s._messagesFile).mode & 0o777, 0o600);
|
|
} finally {
|
|
process.umask(oldUmask);
|
|
if (s) s.close();
|
|
try { fs.rmSync(parent, { recursive: true }); } catch {}
|
|
}
|
|
});
|
|
|
|
it('tightens existing mailbox directory and state file permissions', {
|
|
skip: process.platform === 'win32' ? 'chmod not enforced on Windows' : false,
|
|
}, () => {
|
|
const parent = tmpDataDir();
|
|
const dir = path.join(parent, 'mailbox');
|
|
const stateFile = path.join(dir, 'state.json');
|
|
fs.mkdirSync(dir, { recursive: true, mode: 0o755 });
|
|
fs.chmodSync(dir, 0o755);
|
|
fs.writeFileSync(stateFile, JSON.stringify({ node_secret: 'b'.repeat(64) }) + '\n', {
|
|
encoding: 'utf8',
|
|
mode: 0o644,
|
|
});
|
|
fs.chmodSync(stateFile, 0o644);
|
|
|
|
const s = new MailboxStore(dir);
|
|
try {
|
|
assert.equal(fs.statSync(dir).mode & 0o777, 0o700);
|
|
assert.equal(fs.statSync(s._stateFile).mode & 0o777, 0o600);
|
|
assert.equal(s.getState('node_secret'), 'b'.repeat(64));
|
|
} finally {
|
|
s.close();
|
|
try { fs.rmSync(parent, { recursive: true }); } catch {}
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('persistence', () => {
|
|
it('survives restart by re-reading JSONL', () => {
|
|
const dir = tmpDataDir();
|
|
const s1 = new MailboxStore(dir);
|
|
s1.send({ type: 'persist_test', payload: { val: 42 } });
|
|
s1.setState('my_key', 'my_val');
|
|
s1.setCursor('test:cursor', 'c1');
|
|
s1.close();
|
|
|
|
const s2 = new MailboxStore(dir);
|
|
const msgs = s2.pollOutbound();
|
|
assert.ok(msgs.length >= 1);
|
|
assert.equal(msgs.find(m => m.type === 'persist_test').payload.val, 42);
|
|
assert.equal(s2.getState('my_key'), 'my_val');
|
|
assert.equal(s2.getCursor('test:cursor'), 'c1');
|
|
s2.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
it('rebuilds from JSONL without reading the whole mailbox into one string', () => {
|
|
const dir = tmpDataDir();
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const msgFile = path.join(dir, 'messages.jsonl');
|
|
const rows = [];
|
|
for (let i = 0; i < 5000; i++) {
|
|
rows.push(JSON.stringify({
|
|
id: `msg-${i}`,
|
|
channel: DEFAULT_CHANNEL,
|
|
direction: 'outbound',
|
|
type: 'stream_rebuild',
|
|
status: 'pending',
|
|
payload: { i, text: 'x'.repeat(256) },
|
|
priority: 'normal',
|
|
created_at: Date.now() + i,
|
|
}));
|
|
}
|
|
fs.writeFileSync(msgFile, rows.join('\n') + '\n', 'utf8');
|
|
|
|
const s = new MailboxStore(dir);
|
|
try {
|
|
assert.equal(s.countPending({ direction: 'outbound', type: 'stream_rebuild' }), 5000);
|
|
assert.equal(s._lastRebuildStats.parsed, 5000);
|
|
} finally {
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
}
|
|
});
|
|
|
|
it('skips overlong mailbox JSONL rows during startup rebuild', () => {
|
|
const dir = tmpDataDir();
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const msgFile = path.join(dir, 'messages.jsonl');
|
|
const good = {
|
|
id: 'msg-good',
|
|
channel: DEFAULT_CHANNEL,
|
|
direction: 'inbound',
|
|
type: 'startup_good',
|
|
status: 'pending',
|
|
payload: { ok: true },
|
|
priority: 'normal',
|
|
created_at: Date.now(),
|
|
};
|
|
fs.writeFileSync(
|
|
msgFile,
|
|
'x'.repeat(2048) + '\n' + JSON.stringify(good) + '\n',
|
|
'utf8'
|
|
);
|
|
const saved = process.env.EVOMAP_MAILBOX_MAX_LINE_BYTES;
|
|
process.env.EVOMAP_MAILBOX_MAX_LINE_BYTES = '512';
|
|
const s = new MailboxStore(dir);
|
|
try {
|
|
assert.equal(s._lastRebuildStats.overlong, 1);
|
|
assert.equal(s.poll({ type: 'startup_good' }).length, 1);
|
|
} finally {
|
|
if (saved === undefined) delete process.env.EVOMAP_MAILBOX_MAX_LINE_BYTES;
|
|
else process.env.EVOMAP_MAILBOX_MAX_LINE_BYTES = saved;
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
}
|
|
});
|
|
|
|
it('keeps public API writes at the JSONL line limit visible after restart', () => {
|
|
const dir = tmpDataDir();
|
|
const maxBytes = 512;
|
|
const now = 1700000000000;
|
|
const id = 'msg-close-to-limit';
|
|
const type = 'line_limit_close';
|
|
const text = inboundTextForLineBytes(maxBytes, { id, type, now });
|
|
|
|
withEnv('EVOMAP_MAILBOX_MAX_LINE_BYTES', String(maxBytes), () => {
|
|
withFixedNow(now, () => {
|
|
const s1 = new MailboxStore(dir);
|
|
try {
|
|
assert.equal(s1.writeInbound({ id, type, payload: { text } }), id);
|
|
} finally {
|
|
s1.close();
|
|
}
|
|
});
|
|
|
|
const raw = fs.readFileSync(path.join(dir, 'messages.jsonl'));
|
|
assert.equal(raw.length, maxBytes);
|
|
|
|
const s2 = new MailboxStore(dir);
|
|
try {
|
|
assert.equal(s2._lastRebuildStats.overlong, 0);
|
|
const msg = s2.getById(id);
|
|
assert.ok(msg);
|
|
assert.deepEqual(msg.payload, { text });
|
|
} finally {
|
|
s2.close();
|
|
}
|
|
});
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
it('rejects public API writes over the JSONL line limit without indexing them', () => {
|
|
const dir = tmpDataDir();
|
|
const maxBytes = 512;
|
|
const now = 1700000000000;
|
|
const id = 'msg-over-limit';
|
|
const type = 'line_limit_over';
|
|
const text = inboundTextForLineBytes(maxBytes, { id, type, now }) + 'a';
|
|
|
|
withEnv('EVOMAP_MAILBOX_MAX_LINE_BYTES', String(maxBytes), () => {
|
|
withFixedNow(now, () => {
|
|
const s1 = new MailboxStore(dir);
|
|
try {
|
|
assert.throws(
|
|
() => s1.writeInbound({ id, type, payload: { text } }),
|
|
(err) => err
|
|
&& err.code === 'MAILBOX_JSONL_LINE_TOO_LARGE'
|
|
&& /maximum is 512 bytes/.test(err.message)
|
|
);
|
|
assert.equal(s1.getById(id), null);
|
|
assert.equal(s1.poll({ type }).length, 0);
|
|
} finally {
|
|
s1.close();
|
|
}
|
|
});
|
|
|
|
const s2 = new MailboxStore(dir);
|
|
try {
|
|
assert.equal(s2.getById(id), null);
|
|
assert.equal(s2.poll({ type }).length, 0);
|
|
} finally {
|
|
s2.close();
|
|
}
|
|
});
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
it('fails startup rebuild on mailbox JSONL read errors', () => {
|
|
const dir = tmpDataDir();
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const msgFile = path.join(dir, 'messages.jsonl');
|
|
fs.writeFileSync(
|
|
msgFile,
|
|
JSON.stringify({
|
|
id: 'msg-unreadable',
|
|
channel: DEFAULT_CHANNEL,
|
|
direction: 'outbound',
|
|
type: 'read_failure',
|
|
status: 'pending',
|
|
payload: {},
|
|
priority: 'normal',
|
|
created_at: Date.now(),
|
|
}) + '\n',
|
|
'utf8'
|
|
);
|
|
|
|
const originalReadSync = fs.readSync;
|
|
fs.readSync = function failReadSync(...args) {
|
|
const err = new Error('simulated read failure');
|
|
err.code = 'EIO';
|
|
throw err;
|
|
};
|
|
try {
|
|
assert.throws(
|
|
() => new MailboxStore(dir),
|
|
(err) => err
|
|
&& err.code === 'MAILBOX_JSONL_READ_FAILED'
|
|
&& err.stats
|
|
&& err.stats.read_failed === 1
|
|
&& /simulated read failure/.test(err.message)
|
|
);
|
|
} finally {
|
|
fs.readSync = originalReadSync;
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('compact()', () => {
|
|
it('reduces file size by collapsing updates', () => {
|
|
const dir = tmpDataDir();
|
|
const s = new MailboxStore(dir);
|
|
const { message_id } = s.send({ type: 'compact_test', payload: {} });
|
|
s.updateStatus(message_id, 'synced');
|
|
s.updateStatus(message_id, 'delivered');
|
|
s.incrementRetry(message_id, 'err');
|
|
|
|
const sizeBefore = fs.statSync(s._messagesFile).size;
|
|
s.compact();
|
|
const sizeAfter = fs.statSync(s._messagesFile).size;
|
|
assert.ok(sizeAfter < sizeBefore, 'compaction should reduce file size');
|
|
|
|
const msg = s.getById(message_id);
|
|
assert.equal(msg.status, 'delivered');
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
});
|
|
|
|
describe('schema version and migration', () => {
|
|
it('writes schema version to state on fresh init', () => {
|
|
const dir = tmpDataDir();
|
|
const s = new MailboxStore(dir);
|
|
const stateRaw = JSON.parse(fs.readFileSync(s._stateFile, 'utf8'));
|
|
assert.equal(stateRaw._schema_version, SCHEMA_VERSION);
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
it('preserves schema version across restart', () => {
|
|
const dir = tmpDataDir();
|
|
const s1 = new MailboxStore(dir);
|
|
s1.close();
|
|
const s2 = new MailboxStore(dir);
|
|
const stateRaw = JSON.parse(fs.readFileSync(s2._stateFile, 'utf8'));
|
|
assert.equal(stateRaw._schema_version, SCHEMA_VERSION);
|
|
s2.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
it('runs migrations when state has older schema version', () => {
|
|
const dir = tmpDataDir();
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
fs.writeFileSync(
|
|
path.join(dir, 'state.json'),
|
|
JSON.stringify({ _schema_version: 0 }) + '\n',
|
|
'utf8'
|
|
);
|
|
const s = new MailboxStore(dir);
|
|
const stateRaw = JSON.parse(fs.readFileSync(s._stateFile, 'utf8'));
|
|
assert.equal(stateRaw._schema_version, SCHEMA_VERSION);
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
});
|
|
|
|
describe('exports', () => {
|
|
it('exports PROXY_PROTOCOL_VERSION as a semver string', () => {
|
|
assert.match(PROXY_PROTOCOL_VERSION, /^\d+\.\d+\.\d+$/);
|
|
});
|
|
|
|
it('exports SCHEMA_VERSION as a positive integer', () => {
|
|
assert.equal(typeof SCHEMA_VERSION, 'number');
|
|
assert.ok(SCHEMA_VERSION >= 1);
|
|
});
|
|
});
|
|
|
|
describe('prototype pollution hardening (GHSA-2cjr-5v3h-v2w4)', () => {
|
|
it('strips __proto__ from update rows when rebuilding from JSONL', () => {
|
|
const dir = tmpDataDir();
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const msgFile = path.join(dir, 'messages.jsonl');
|
|
fs.writeFileSync(
|
|
msgFile,
|
|
JSON.stringify({
|
|
id: 'msg-1',
|
|
channel: DEFAULT_CHANNEL,
|
|
direction: 'inbound',
|
|
type: 'test',
|
|
status: 'pending',
|
|
payload: {},
|
|
priority: 'normal',
|
|
created_at: Date.now(),
|
|
}) + '\n' +
|
|
JSON.stringify({
|
|
_op: 'update',
|
|
id: 'msg-1',
|
|
fields: {
|
|
__proto__: { polluted: true, isAdmin: true },
|
|
status: 'synced',
|
|
},
|
|
}) + '\n',
|
|
'utf8'
|
|
);
|
|
|
|
const s = new MailboxStore(dir);
|
|
const probe = {};
|
|
assert.equal(probe.polluted, undefined, 'Object.prototype must not be polluted');
|
|
assert.equal(probe.isAdmin, undefined, 'Object.prototype must not be polluted');
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
|
|
it('strips constructor/prototype from raw message rows', () => {
|
|
const dir = tmpDataDir();
|
|
fs.mkdirSync(dir, { recursive: true });
|
|
const msgFile = path.join(dir, 'messages.jsonl');
|
|
fs.writeFileSync(
|
|
msgFile,
|
|
JSON.stringify({
|
|
id: 'msg-1',
|
|
channel: DEFAULT_CHANNEL,
|
|
direction: 'inbound',
|
|
type: 'test',
|
|
status: 'pending',
|
|
payload: {},
|
|
priority: 'normal',
|
|
created_at: Date.now(),
|
|
constructor: { prototype: { evil: true } },
|
|
prototype: { evil: true },
|
|
}) + '\n',
|
|
'utf8'
|
|
);
|
|
|
|
const s = new MailboxStore(dir);
|
|
const probe = {};
|
|
assert.equal(probe.evil, undefined, 'Object.prototype must not be polluted');
|
|
s.close();
|
|
try { fs.rmSync(dir, { recursive: true }); } catch {}
|
|
});
|
|
});
|
|
});
|