75 lines
2.3 KiB
JavaScript
75 lines
2.3 KiB
JavaScript
import assert from "node:assert/strict";
|
|
import test from "node:test";
|
|
import { MarketOutbox } from "../src/market-outbox.js";
|
|
|
|
test("keeps the same batch across retries and deletes only after acknowledgement", async () => {
|
|
const store = memoryStore();
|
|
await store.add({ observationId: "one", name: "Stem" });
|
|
await store.add({ observationId: "two", name: "Stem" });
|
|
const sent = [];
|
|
let fail = true;
|
|
const send = async (batch) => {
|
|
sent.push(batch);
|
|
if (fail) throw new Error("offline");
|
|
return { batchId: batch.batchId };
|
|
};
|
|
const options = { now: () => "2026-08-11T12:00:00.000Z", uuid: () => "batch-one" };
|
|
|
|
await assert.rejects(new MarketOutbox(store, send, options).flush(), /offline/);
|
|
fail = false;
|
|
const result = await new MarketOutbox(store, send, options).flush();
|
|
|
|
assert.equal(result.count, 2);
|
|
assert.equal(sent[0].batchId, "batch-one");
|
|
assert.deepEqual(sent[1], sent[0]);
|
|
assert.deepEqual(await store.list(200), []);
|
|
});
|
|
|
|
test("keeps rows pending for a wrong acknowledgement and caps encoded batch size", async () => {
|
|
const store = memoryStore();
|
|
for (let index = 0; index < 80; index += 1) {
|
|
await store.add({ observationId: `observation-${index}`, rawText: "x".repeat(4096), storeRawText: "y".repeat(4096) });
|
|
}
|
|
let sent;
|
|
const outbox = new MarketOutbox(
|
|
store,
|
|
async (batch) => {
|
|
sent = batch;
|
|
return { batchId: "wrong" };
|
|
},
|
|
{ now: () => "2026-08-11T12:00:00.000Z", uuid: () => "batch-two" },
|
|
);
|
|
|
|
await assert.rejects(outbox.flush(), /другой пакет/);
|
|
assert.ok(sent.observations.length < 80);
|
|
assert.ok(new TextEncoder().encode(JSON.stringify(sent)).byteLength <= 80 * 1024);
|
|
assert.equal((await store.list(200)).length, 80);
|
|
});
|
|
|
|
function memoryStore() {
|
|
const observations = new Map();
|
|
let batch = null;
|
|
return {
|
|
async add(value) {
|
|
observations.set(value.observationId, value);
|
|
},
|
|
async list(limit) {
|
|
return [...observations.values()].slice(0, limit);
|
|
},
|
|
async getMany(ids) {
|
|
return ids.map((id) => observations.get(id)).filter(Boolean);
|
|
},
|
|
async getBatch() {
|
|
return batch;
|
|
},
|
|
async setBatch(value) {
|
|
batch = value;
|
|
},
|
|
async ack(value) {
|
|
if (batch?.batchId !== value.batchId) return;
|
|
value.observationIds.forEach((id) => observations.delete(id));
|
|
batch = null;
|
|
},
|
|
};
|
|
}
|