import { describe, expect, it, vi } from 'vitest'; import { DataRoute } from '../../source/routes/resources.ts'; import { UnauthorizedError } from '../../source/errors/index.ts'; import { type BaseBroadcaster } from '../../source/services/broadcaster.ts'; import { ApplicationRouteStream } from '../../source/services/route-stream.ts'; import type { Database } from '../../source/services/storage/database.ts'; import { TestConnection } from '../helpers/test-connection.ts'; import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../../source/constants.ts'; import type { AuthSecp256k1 } from '../../source/auth/auth.ts'; import { Accounts } from '../../source/auth/accounts.ts'; const createBroadcasterStub = (): BaseBroadcaster => { return { subscribe: vi.fn(), unsubscribe: vi.fn().mockResolvedValue(undefined), publish: vi.fn(), sendEvent: vi.fn(), } as unknown as BaseBroadcaster; }; const createAuthStub = (): AuthSecp256k1 => { return { verifySignature: vi.fn().mockImplementation(() => { return true; }), } as unknown as AuthSecp256k1; }; const createAccountsStub = (): Accounts => { return { getBalance: vi.fn().mockImplementation(() => { return 0; }), setBalance: vi.fn().mockImplementation(() => { return; }), deductBalance: vi.fn().mockImplementation(() => { return; }), hasSufficientBalance: vi.fn().mockImplementation(() => { return true; }), } as unknown as Accounts; }; describe('DataRoute subscriptions', (): void => { it('subscribes to future resource changes until removal', async (): Promise => { let resolveRemoved: () => void = () => undefined; const removed = new Promise((resolve) => { resolveRemoved = resolve; }); const storage = { db: { transaction: vi.fn(), }, } as unknown as Database; const broadcaster = createBroadcasterStub(); vi.mocked(broadcaster.subscribe).mockReturnValue(removed); const connection = new TestConnection(true, false); const stream = new ApplicationRouteStream( connection, { resourceId: [ 'a', 'b' ], }, '/data/subscribe', ); const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); const execution = route.subscribeData(stream); expect(broadcaster.subscribe).toHaveBeenCalledWith(stream, [ 'resource:a', 'resource:b' ]); expect(storage.db.transaction).not.toHaveBeenCalled(); expect(connection.messages).toEqual([]); await expect(Promise.race([ execution.then(() => 'settled'), Promise.resolve('pending') ])).resolves.toBe('pending'); resolveRemoved(); await execution; }); it('unsubscribes a bidirectional connection and acknowledges the request', async (): Promise => { const storage = { db: { transaction: vi.fn(), }, } as unknown as Database; const broadcaster = createBroadcasterStub(); const connection = new TestConnection(true, true); const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ] }, '/data/unsubscribe', 'unsubscribe-1'); const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); await route.unsubscribeData(stream); expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [ 'resource:a' ]); expect(connection.messages).toEqual([ { id: 'unsubscribe-1', type: 'response', statusCode: 200, body: {}, }, ]); }); it('rejects selective unsubscribe on a one-way connection', async (): Promise => { const storage = { db: { transaction: vi.fn(), }, } as unknown as Database; const broadcaster = createBroadcasterStub(); const stream = new ApplicationRouteStream( new TestConnection(true, false), { resourceId: [ 'a' ], }, '/data/unsubscribe', ); const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); await expect(route.unsubscribeData(stream)).rejects.toMatchObject({ statusCode: HTTP_STATUS_CODE_NOT_ACCEPTED }); expect(broadcaster.unsubscribe).not.toHaveBeenCalled(); }); }); describe('DataRoute resource write auth', (): void => { it('rejects an invalid resource signature before writing the batch', async (): Promise => { const storage = { db: { transaction: vi.fn(), }, } as unknown as Database; const broadcaster = createBroadcasterStub(); const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); await expect(route.writeData({ connection: new TestConnection(true, true), streaming: true, bidirectional: true, send: vi.fn(), body: { resources: [ { id: 'resource-a', publicKey: 'not-a-public-key', timestamp: Date.now(), signature: 'not-a-signature', value: new Uint8Array([ 1, 2, 3 ]), }, ], }, headers: { publicKey: 'public-key', signature: 'signature', timestamp: Date.now(), }, } as unknown as ApplicationRouteStream)).rejects.toBeInstanceOf(UnauthorizedError); expect(storage.db.transaction).not.toHaveBeenCalled(); expect(broadcaster.publish).not.toHaveBeenCalled(); }); });