138 lines
4.7 KiB
TypeScript
138 lines
4.7 KiB
TypeScript
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||
import { mkdtempSync, rmSync } from 'fs';
|
||
import { join } from 'path';
|
||
import { tmpdir } from 'os';
|
||
import { Repository } from './db/repository.js';
|
||
import { VapidKeyStore } from './vapid-store.js';
|
||
import { PushService, type PushPayload } from './push-service.js';
|
||
|
||
/**
|
||
* Codex review P3 #13: V2 must demonstrate the queue + timeout behavior
|
||
* under a realistic burst, not defer that to V3. This test fires 100
|
||
* payloads at 5 subscriptions and checks the queue:
|
||
* - enqueue() returns synchronously (worker is never blocked)
|
||
* - concurrency limit is honored
|
||
* - no unhandled rejections leak
|
||
* - every send is accounted for in either success or failure markers
|
||
*
|
||
* The mock keeps web-push.sendNotification asynchronous (microtask jump)
|
||
* so the queue actually has to schedule work.
|
||
*/
|
||
|
||
vi.mock('web-push', () => {
|
||
const sendNotification = vi.fn();
|
||
return {
|
||
default: {
|
||
sendNotification,
|
||
setVapidDetails: vi.fn(),
|
||
generateVAPIDKeys: () => ({
|
||
publicKey: 'BPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPubBPub',
|
||
privateKey: 'priv-priv-priv-priv-priv-priv-priv-priv-pr',
|
||
}),
|
||
},
|
||
};
|
||
});
|
||
|
||
import webPush from 'web-push';
|
||
const sendMock = webPush.sendNotification as unknown as ReturnType<typeof vi.fn>;
|
||
|
||
const SUBJECT = 'https://aao.example/';
|
||
|
||
describe('PushService load behavior (Codex P3 #13)', () => {
|
||
let tempDir = '';
|
||
let repo: Repository;
|
||
let store: VapidKeyStore;
|
||
let userId = '';
|
||
let service: PushService;
|
||
|
||
beforeEach(() => {
|
||
tempDir = mkdtempSync(join(tmpdir(), 'maestro-push-load-'));
|
||
repo = new Repository(join(tempDir, 'db.sqlite'));
|
||
store = new VapidKeyStore(join(tempDir, 'vapid.json'), join(tempDir, 'vapid-history'));
|
||
store.loadOrGenerate(SUBJECT);
|
||
service = new PushService(repo, store, { queueConcurrency: 8, perSendTimeoutMs: 1_000 });
|
||
const user = repo.createUser({ email: 'load@example.com', name: 'load', role: 'user', status: 'active' });
|
||
userId = user.id;
|
||
for (let i = 0; i < 5; i++) {
|
||
repo.upsertPushSubscription({
|
||
userId,
|
||
endpoint: `https://push.example/sub-${i}`,
|
||
p256dh: 'p',
|
||
auth: 'a',
|
||
vapidKeyId: store.getCurrent().keyId,
|
||
});
|
||
}
|
||
sendMock.mockReset();
|
||
});
|
||
|
||
afterEach(() => {
|
||
repo.close();
|
||
if (tempDir) {
|
||
rmSync(tempDir, { recursive: true, force: true });
|
||
tempDir = '';
|
||
}
|
||
});
|
||
|
||
const payload = (i: number): PushPayload => ({
|
||
event: i % 2 === 0 ? 'succeeded' : 'running',
|
||
taskId: i,
|
||
taskTitle: `T${i}`,
|
||
pieceName: 'chat',
|
||
ownerId: userId,
|
||
});
|
||
|
||
it('100 events × 5 subscriptions: enqueue is non-blocking, all sends complete', async () => {
|
||
// Real async (next-tick) so the queue actually queues.
|
||
sendMock.mockImplementation(() => Promise.resolve({ statusCode: 201 } as unknown));
|
||
|
||
const t0 = Date.now();
|
||
for (let i = 0; i < 100; i++) {
|
||
service.enqueue(payload(i));
|
||
}
|
||
const enqueueDurationMs = Date.now() - t0;
|
||
|
||
// 100 fire-and-forget enqueues must return effectively instantly
|
||
// (well under a second on any reasonable machine).
|
||
expect(enqueueDurationMs).toBeLessThan(500);
|
||
|
||
await service.waitIdle();
|
||
|
||
// 100 events × 5 subs = 500 sends; every subscription is owned by
|
||
// the same user with all events on.
|
||
expect(sendMock.mock.calls.length).toBe(500);
|
||
|
||
// Every subscription should record a recent success — none silently
|
||
// dropped.
|
||
const subs = repo.listPushSubscriptionsForUser(userId);
|
||
expect(subs).toHaveLength(5);
|
||
for (const sub of subs) {
|
||
expect(sub.lastSuccessAt).toBeTruthy();
|
||
expect(sub.failureCount).toBe(0);
|
||
}
|
||
}, 30_000);
|
||
|
||
it('mix of success + permanent failure: subscriptions clean up via 410', async () => {
|
||
// First 2 endpoints succeed; remaining 3 return 410 (Gone).
|
||
sendMock.mockImplementation((subscription: unknown) => {
|
||
const sub = subscription as { endpoint: string };
|
||
const idx = parseInt(sub.endpoint.split('-').pop()!, 10);
|
||
if (idx < 2) return Promise.resolve({ statusCode: 201 } as unknown);
|
||
const err = Object.assign(new Error('gone'), { statusCode: 410 });
|
||
return Promise.reject(err);
|
||
});
|
||
|
||
for (let i = 0; i < 20; i++) {
|
||
service.enqueue(payload(i));
|
||
}
|
||
await service.waitIdle();
|
||
|
||
const remaining = repo.listPushSubscriptionsForUser(userId);
|
||
// 410 endpoints should have been deleted; 2 survivors remain.
|
||
expect(remaining).toHaveLength(2);
|
||
expect(remaining.map(s => s.endpoint).sort()).toEqual([
|
||
'https://push.example/sub-0',
|
||
'https://push.example/sub-1',
|
||
]);
|
||
}, 30_000);
|
||
});
|