diff --git a/.changeset/repository-unit-tests.md b/.changeset/repository-unit-tests.md new file mode 100644 index 00000000..a845151c --- /dev/null +++ b/.changeset/repository-unit-tests.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/test/unit/repositories/nip05-verification-repository.spec.ts b/test/unit/repositories/nip05-verification-repository.spec.ts new file mode 100644 index 00000000..03ade9cd --- /dev/null +++ b/test/unit/repositories/nip05-verification-repository.spec.ts @@ -0,0 +1,272 @@ +import { expect } from 'chai' +import Sinon from 'sinon' + +import { DBNip05Verification, Nip05Verification } from '../../../src/@types/nip05' +import { Nip05VerificationRepository } from '../../../src/repositories/nip05-verification-repository' +import { createRecordingKnex } from './recording-knex' + +describe('Nip05VerificationRepository', () => { + const NOW = new Date('2026-06-15T12:00:00.000Z') + const DAY_MS = 86_400_000 + const pubkeyHex = '22e804d26ed16b68db5259e78449e96dab5d464c8f470bda3eb1a70467f2c793' + + let clock: Sinon.SinonFakeTimers + let queries: ReturnType['queries'] + let respondWith: ReturnType['respondWith'] + let repository: Nip05VerificationRepository + + const row = (overrides: Partial = {}): DBNip05Verification => ({ + pubkey: Buffer.from(pubkeyHex, 'hex'), + nip05: 'alice@example.com', + domain: 'example.com', + is_verified: true, + last_verified_at: new Date('2026-06-14T12:00:00.000Z'), + last_checked_at: new Date('2026-06-14T12:00:00.000Z'), + failure_count: 0, + created_at: new Date('2026-06-01T00:00:00.000Z'), + updated_at: new Date('2026-06-14T12:00:00.000Z'), + ...overrides, + }) + + const verification = (overrides: Partial = {}): Nip05Verification => ({ + pubkey: pubkeyHex, + nip05: 'alice@example.com', + domain: 'example.com', + isVerified: true, + lastVerifiedAt: new Date('2026-06-14T12:00:00.000Z'), + lastCheckedAt: new Date('2026-06-14T12:00:00.000Z'), + failureCount: 0, + createdAt: new Date('2026-06-01T00:00:00.000Z'), + updatedAt: new Date('2026-06-14T12:00:00.000Z'), + ...overrides, + }) + + beforeEach(() => { + clock = Sinon.useFakeTimers({ now: NOW, toFake: ['Date'] }) + const recording = createRecordingKnex() + queries = recording.queries + respondWith = recording.respondWith + repository = new Nip05VerificationRepository(recording.db as any) + }) + + afterEach(() => { + clock.restore() + }) + + describe('.findByPubkey', () => { + it('returns undefined for a pubkey that was never verified', async () => { + respondWith([]) + + expect(await repository.findByPubkey(pubkeyHex)).to.equal(undefined) + }) + + it('looks the pubkey up by its bytes and maps the row', async () => { + respondWith([row({ is_verified: false, failure_count: 2, last_verified_at: null })]) + + const found = await repository.findByPubkey(pubkeyHex) + + expect(queries[0].sql).to.equal('select * from "nip05_verifications" where "pubkey" = $1') + expect(queries[0].bindings).to.deep.equal([Buffer.from(pubkeyHex, 'hex')]) + expect(found).to.deep.equal( + verification({ + isVerified: false, + failureCount: 2, + lastVerifiedAt: null, + }), + ) + }) + }) + + describe('.upsert', () => { + it('inserts the verification, and on a second one for the same pubkey updates everything but the pubkey and creation time', async () => { + respondWith(1) + + await repository.upsert(verification({ isVerified: false, failureCount: 3 })) + + const { sql } = queries[0] + expect(sql).to.match(/^insert into "nip05_verifications" /) + expect(sql).to.include('on conflict ("pubkey") do update set') + const update = sql.slice(sql.indexOf('do update set')) + for (const column of [ + 'nip05', + 'domain', + 'is_verified', + 'last_verified_at', + 'last_checked_at', + 'failure_count', + 'updated_at', + ]) { + expect(update, column).to.include(`"${column}" =`) + } + expect(update).to.not.include('"pubkey" =') + expect(update).to.not.include('"created_at" =') + }) + + it('stores the values it was given, stamping creation and update now', async () => { + respondWith(1) + + await repository.upsert(verification({ failureCount: 3 })) + + const { bindings } = queries[0] + expect(bindings).to.deep.include(Buffer.from(pubkeyHex, 'hex')) + expect(bindings).to.include.members(['alice@example.com', 'example.com', true, 3]) + expect(bindings.filter((binding) => binding instanceof Date && binding.getTime() === NOW.getTime())).to.not.be + .empty + }) + + it('checks now when the verification says nothing about when it was last checked, on insert and on conflict', async () => { + respondWith(1) + const checkedAt = new Date('2026-06-10T00:00:00.000Z') + + await repository.upsert(verification({ lastCheckedAt: undefined as unknown as Date })) + await repository.upsert(verification({ lastCheckedAt: checkedAt })) + + const written = (query: (typeof queries)[number]) => { + const insertColumns = /^insert into "nip05_verifications" \(([^)]*)\)/.exec(query.sql)![1].split(', ') + const inserted = insertColumns.indexOf('"last_checked_at"') + const updated = /"last_checked_at" = \$(\d+)/.exec(query.sql.slice(query.sql.indexOf('do update set')))![1] + + return [query.bindings[inserted], query.bindings[Number(updated) - 1]] + } + expect(written(queries[0])).to.deep.equal([NOW, NOW]) + expect(written(queries[1])).to.deep.equal([checkedAt, checkedAt]) + }) + + it('resolves with the number of rows written', async () => { + respondWith(1) + + expect(await repository.upsert(verification())).to.equal(1) + }) + }) + + describe('.findPendingVerifications', () => { + it('selects verifications not checked within the update frequency and not yet failing too often, least recently checked first', async () => { + respondWith([row()]) + + const pending = await repository.findPendingVerifications(DAY_MS, 10, 50) + + expect(queries[0].sql).to.equal( + 'select * from "nip05_verifications" where "last_checked_at" < $1 and "failure_count" < $2 order by "last_checked_at" asc limit $3', + ) + expect(queries[0].bindings).to.deep.equal([new Date(NOW.getTime() - DAY_MS), 10, 50]) + expect(pending).to.deep.equal([verification()]) + }) + + it('gives up on a verification once it has failed the maximum number of times', async () => { + respondWith([]) + + await repository.findPendingVerifications(DAY_MS, 3, 50) + + expect(queries[0].sql).to.include('"failure_count" < $2') + expect(queries[0].bindings[1]).to.equal(3) + }) + }) + + describe('.deleteByPubkey', () => { + it('deletes the verification of the pubkey and returns how many rows went', async () => { + respondWith(1) + + expect(await repository.deleteByPubkey(pubkeyHex)).to.equal(1) + + expect(queries[0].sql).to.equal('delete from "nip05_verifications" where "pubkey" = $1') + expect(queries[0].bindings).to.deep.equal([Buffer.from(pubkeyHex, 'hex')]) + }) + }) + + describe('.listForAdmin', () => { + const query = { verifyExpirationMs: 7 * DAY_MS, limit: 25, offset: 50 } + + // Each list is two statements: the count, then the page. + const list = (filters: Partial[0]> = {}, total = 0) => { + respondWith([{ count: String(total) }], []) + + return repository.listForAdmin({ ...query, ...filters }) + } + + it('counts, then returns the page newest check first, with the limit and offset', async () => { + respondWith([{ count: '120' }], [row()]) + + const result = await repository.listForAdmin(query) + + expect(queries).to.have.length(2) + expect(queries[0].sql).to.equal('select count(*) as "count" from "nip05_verifications" limit $1') + expect(queries[1].sql).to.equal( + 'select * from "nip05_verifications" order by "last_checked_at" desc limit $1 offset $2', + ) + expect(queries[1].bindings).to.deep.equal([25, 50]) + expect(result.total).to.equal(120) + expect(result.rows).to.deep.equal([verification()]) + }) + + it('reports a total of zero for an empty table', async () => { + respondWith([], []) + + expect(await repository.listForAdmin(query)).to.deep.equal({ rows: [], total: 0 }) + }) + + it('filters by domain, in the count and in the page', async () => { + await list({ domain: 'example.com' }) + + for (const { sql, bindings } of queries) { + expect(sql).to.include('where "domain" = $1') + expect(bindings[0]).to.equal('example.com') + } + }) + + it('filters verified ones, and unverified ones', async () => { + await list({ verified: true }) + await list({ verified: false }) + + expect(queries[0].sql).to.include('where "is_verified" = $1') + expect(queries[0].bindings[0]).to.equal(true) + expect(queries[2].bindings[0]).to.equal(false) + }) + + it('filters by a minimum failure count, including zero', async () => { + await list({ minFailures: 3 }) + await list({ minFailures: 0 }) + + expect(queries[0].sql).to.include('where "failure_count" >= $1') + expect(queries[0].bindings[0]).to.equal(3) + expect(queries[2].bindings[0]).to.equal(0) + }) + + it('treats a verification as stale when it is unverified, was never verified, or was verified before the expiry', async () => { + await list({ stale: true }) + + expect(queries[0].sql).to.include( + 'where ("is_verified" = $1 or "last_verified_at" is null or "last_verified_at" < $2)', + ) + expect(queries[0].bindings.slice(0, 2)).to.deep.equal([false, new Date(NOW.getTime() - 7 * DAY_MS)]) + }) + + it('treats a verification as fresh when it is verified and was verified since the expiry', async () => { + await list({ stale: false }) + + expect(queries[0].sql).to.include('where "is_verified" = $1 and "last_verified_at" >= $2') + expect(queries[0].bindings.slice(0, 2)).to.deep.equal([true, new Date(NOW.getTime() - 7 * DAY_MS)]) + }) + + it('does not filter on staleness unless asked to', async () => { + await list() + + expect(queries[0].sql).to.not.include('last_verified_at') + }) + + it('combines the filters with and', async () => { + await list({ domain: 'example.com', verified: true, minFailures: 2, stale: false }) + + const { sql, bindings } = queries[1] + expect(sql).to.include( + 'where "domain" = $1 and "is_verified" = $2 and "failure_count" >= $3 and "is_verified" = $4 and "last_verified_at" >= $5', + ) + expect(bindings.slice(0, 4)).to.deep.equal(['example.com', true, 2, true]) + }) + + it('pages with the filters applied to the count too, so the total matches the filtered list', async () => { + await list({ domain: 'example.com' }, 7) + + expect(queries[0].sql).to.match(/^select count\(\*\) as "count" from "nip05_verifications" where "domain" = \$1/) + }) + }) +}) diff --git a/test/unit/repositories/notification-delivery-log-repository.spec.ts b/test/unit/repositories/notification-delivery-log-repository.spec.ts new file mode 100644 index 00000000..bfbea830 --- /dev/null +++ b/test/unit/repositories/notification-delivery-log-repository.spec.ts @@ -0,0 +1,201 @@ +import { expect } from 'chai' +import Sinon from 'sinon' + +import { DBNotificationDeliveryLogEntry, NotificationDeliveryStatus } from '../../../src/@types/operator-notifications' +import { NotificationDeliveryLogRepository } from '../../../src/repositories/notification-delivery-log-repository' +import { createRecordingKnex } from './recording-knex' + +describe('NotificationDeliveryLogRepository', () => { + const NOW = new Date('2026-06-15T12:00:00.000Z') + let clock: Sinon.SinonFakeTimers + let queries: ReturnType['queries'] + let respondWith: ReturnType['respondWith'] + let repository: NotificationDeliveryLogRepository + + const row = (overrides: Partial = {}): DBNotificationDeliveryLogEntry => ({ + id: '7', + outbox_id: 'outbox-1', + event_type: 'admission.invoice.paid', + target_id: 'ops-slack', + target_type: 'slack', + status: NotificationDeliveryStatus.SUCCESS, + attempt_number: 1, + error_snippet: null, + created_at: new Date('2026-06-15T11:59:00.000Z'), + ...overrides, + }) + + beforeEach(() => { + clock = Sinon.useFakeTimers({ now: NOW, toFake: ['Date'] }) + const recording = createRecordingKnex() + queries = recording.queries + respondWith = recording.respondWith + repository = new NotificationDeliveryLogRepository(recording.db as any) + }) + + afterEach(() => { + clock.restore() + }) + + describe('.append', () => { + it('records a success with the outbox message, the target and the attempt, stamped now', async () => { + respondWith(1) + + await repository.append({ + outboxId: 'outbox-1', + eventType: 'admission.invoice.paid', + targetId: 'ops-slack', + targetType: 'slack', + status: NotificationDeliveryStatus.SUCCESS, + attemptNumber: 1, + errorSnippet: null, + }) + + expect(queries).to.have.length(1) + expect(queries[0].sql).to.match(/^insert into "notification_delivery_log" /) + expect(queries[0].sql).to.include( + '("attempt_number", "created_at", "error_snippet", "event_type", "outbox_id", "status", "target_id", "target_type")', + ) + expect(queries[0].bindings).to.deep.equal([ + 1, + NOW, + null, + 'admission.invoice.paid', + 'outbox-1', + NotificationDeliveryStatus.SUCCESS, + 'ops-slack', + 'slack', + ]) + }) + + it('records a failure with the error it ended in', async () => { + respondWith(1) + + await repository.append({ + outboxId: 'outbox-1', + eventType: 'settings.changed', + targetId: 'ops-http', + targetType: 'http', + status: NotificationDeliveryStatus.FAILED, + attemptNumber: 3, + errorSnippet: 'HTTP 503', + }) + + expect(queries[0].bindings).to.include.members([3, 'HTTP 503', NotificationDeliveryStatus.FAILED, 'ops-http']) + }) + + it('records a test notification, which has no outbox message', async () => { + respondWith(1) + + await repository.append({ + outboxId: null, + eventType: 'test', + targetId: 'ops-http', + targetType: 'http', + status: NotificationDeliveryStatus.SUCCESS, + attemptNumber: 1, + errorSnippet: null, + }) + + expect(queries[0].bindings).to.include(null) + expect(queries[0].bindings).to.not.include('outbox-1') + }) + }) + + describe('.findSuccessfulTargetIds', () => { + it('lists, once each, the targets an outbox message was already delivered to, so a retry skips them', async () => { + respondWith([{ target_id: 'ops-slack' }, { target_id: 'ops-http' }]) + + const targetIds = await repository.findSuccessfulTargetIds('outbox-1') + + expect(targetIds).to.deep.equal(['ops-slack', 'ops-http']) + expect(queries[0].sql).to.include('select distinct "target_id"') + expect(queries[0].sql).to.include('from "notification_delivery_log"') + expect(queries[0].sql).to.include('"outbox_id" = $1 and "status" = $2') + expect(queries[0].bindings).to.deep.equal(['outbox-1', NotificationDeliveryStatus.SUCCESS]) + }) + + it('returns nothing for a message that was never delivered', async () => { + respondWith([]) + + expect(await repository.findSuccessfulTargetIds('outbox-1')).to.deep.equal([]) + }) + }) + + describe('.findRecent', () => { + it('returns the latest 50 entries, newest first, when asked for no filter or limit', async () => { + respondWith([row()]) + + await repository.findRecent() + + expect(queries[0].sql).to.equal('select * from "notification_delivery_log" order by "created_at" desc limit $1') + expect(queries[0].bindings).to.deep.equal([50]) + }) + + it('honours the limit', async () => { + respondWith([]) + + await repository.findRecent(5) + + expect(queries[0].bindings).to.deep.equal([5]) + }) + + it('filters by status', async () => { + respondWith([]) + + await repository.findRecent(50, { status: NotificationDeliveryStatus.FAILED }) + + expect(queries[0].sql).to.include('where "status" = $1') + expect(queries[0].bindings).to.deep.equal([NotificationDeliveryStatus.FAILED, 50]) + }) + + it('filters by event type', async () => { + respondWith([]) + + await repository.findRecent(50, { eventType: 'settings.changed' }) + + expect(queries[0].sql).to.include('where "event_type" = $1') + expect(queries[0].bindings).to.deep.equal(['settings.changed', 50]) + }) + + it('combines both filters', async () => { + respondWith([]) + + await repository.findRecent(50, { status: NotificationDeliveryStatus.SUCCESS, eventType: 'settings.changed' }) + + expect(queries[0].sql).to.include('where "status" = $1 and "event_type" = $2') + expect(queries[0].bindings).to.deep.equal([NotificationDeliveryStatus.SUCCESS, 'settings.changed', 50]) + }) + + it('maps rows to entries', async () => { + respondWith([row({ status: NotificationDeliveryStatus.FAILED, attempt_number: 2, error_snippet: 'HTTP 500' })]) + + const [entry] = await repository.findRecent() + + expect(entry).to.deep.equal({ + id: '7', + outboxId: 'outbox-1', + eventType: 'admission.invoice.paid', + targetId: 'ops-slack', + targetType: 'slack', + status: NotificationDeliveryStatus.FAILED, + attemptNumber: 2, + errorSnippet: 'HTTP 500', + createdAt: new Date('2026-06-15T11:59:00.000Z'), + }) + }) + }) + + describe('.deleteOlderThan', () => { + const cutoff = new Date('2026-05-15T00:00:00.000Z') + + it('deletes entries logged before the cutoff and returns how many', async () => { + respondWith(12) + + expect(await repository.deleteOlderThan(cutoff)).to.equal(12) + + expect(queries[0].sql).to.equal('delete from "notification_delivery_log" where "created_at" < $1') + expect(queries[0].bindings).to.deep.equal([cutoff]) + }) + }) +}) diff --git a/test/unit/repositories/notification-outbox-repository.spec.ts b/test/unit/repositories/notification-outbox-repository.spec.ts new file mode 100644 index 00000000..120bbe31 --- /dev/null +++ b/test/unit/repositories/notification-outbox-repository.spec.ts @@ -0,0 +1,318 @@ +import { expect } from 'chai' +import Sinon from 'sinon' + +import { DBNotificationOutboxMessage, NotificationOutboxStatus } from '../../../src/@types/notification-outbox' +import { + NOTIFICATION_OUTBOX_PROCESSING_LEASE_MS, + NotificationOutboxRepository, +} from '../../../src/repositories/notification-outbox-repository' +import { createRecordingKnex } from './recording-knex' + +describe('NotificationOutboxRepository', () => { + const NOW = new Date('2026-06-15T12:00:00.000Z') + let clock: Sinon.SinonFakeTimers + let queries: ReturnType['queries'] + let steps: ReturnType['steps'] + let stepConnections: ReturnType['stepConnections'] + let respondWith: ReturnType['respondWith'] + let repository: NotificationOutboxRepository + + const row = (overrides: Partial = {}): DBNotificationOutboxMessage => ({ + id: '6d1c9d4e-61a0-4a52-90a4-0c7f9a1c2b11', + event_type: 'admission.invoice.paid', + payload: { invoiceId: 'inv-1' }, + status: NotificationOutboxStatus.PENDING, + attempt_count: 0, + available_at: new Date('2026-06-15T11:00:00.000Z'), + last_error: null, + delivered_at: null, + created_at: new Date('2026-06-15T10:00:00.000Z'), + updated_at: new Date('2026-06-15T10:00:00.000Z'), + ...overrides, + }) + + beforeEach(() => { + clock = Sinon.useFakeTimers({ now: NOW, toFake: ['Date'] }) + const recording = createRecordingKnex() + queries = recording.queries + steps = recording.steps + stepConnections = recording.stepConnections + respondWith = recording.respondWith + repository = new NotificationOutboxRepository(recording.db as any) + }) + + afterEach(() => { + clock.restore() + }) + + describe('.enqueue', () => { + it('inserts a pending message that is available now and has not been attempted', async () => { + respondWith(1) + + const message = await repository.enqueue('admission.invoice.paid', { invoiceId: 'inv-1' }) + + expect(queries).to.have.length(1) + expect(queries[0].sql).to.match(/^insert into "notification_outbox"/) + expect(queries[0].bindings).to.include.members(['admission.invoice.paid', NotificationOutboxStatus.PENDING, 0]) + expect(message).to.include({ + eventType: 'admission.invoice.paid', + status: NotificationOutboxStatus.PENDING, + attemptCount: 0, + lastError: null, + deliveredAt: null, + }) + expect(message.payload).to.deep.equal({ invoiceId: 'inv-1' }) + expect(message.availableAt).to.deep.equal(NOW) + expect(message.createdAt).to.deep.equal(NOW) + expect(message.updatedAt).to.deep.equal(NOW) + }) + + it('gives every message its own id', async () => { + respondWith(1, 1) + + const first = await repository.enqueue('a', {}) + const second = await repository.enqueue('a', {}) + + expect(first.id).to.match(/^[0-9a-f-]{36}$/) + expect(first.id).to.not.equal(second.id) + }) + + it('does not swallow a failed insert, so the caller can roll back with it', async () => { + const failing = createRecordingKnex() + ;(failing.db.client as any).acquireConnection = async () => { + throw new Error('connection refused') + } + + let error: Error | undefined + try { + await new NotificationOutboxRepository(failing.db as any).enqueue('a', {}) + } catch (caught) { + error = caught as Error + } + + expect(error?.message).to.contain('connection refused') + }) + }) + + describe('.claimBatch', () => { + it('returns nothing, and updates nothing, when no message is due', async () => { + respondWith([]) + + expect(await repository.claimBatch(10)).to.deep.equal([]) + + expect(queries).to.have.length(1) + expect(queries[0].sql).to.match(/^select/) + }) + + it('selects due pending messages or processing ones whose lease ran out, oldest first, up to the limit', async () => { + respondWith([]) + + await repository.claimBatch(7) + + const { sql, bindings } = queries[0] + expect(sql).to.include( + 'where (("status" = $1 and "available_at" <= CURRENT_TIMESTAMP) or ("status" = $2 and "updated_at" < $3))', + ) + expect(sql).to.include('order by "created_at" asc') + expect(sql).to.include('limit $4') + expect(bindings[0]).to.equal(NotificationOutboxStatus.PENDING) + expect(bindings[1]).to.equal(NotificationOutboxStatus.PROCESSING) + expect(bindings[3]).to.equal(7) + }) + + it('reclaims a processing message once it has been held for longer than the lease', async () => { + respondWith([]) + + await repository.claimBatch(1) + + const staleBefore = queries[0].bindings[2] as Date + expect(staleBefore).to.deep.equal(new Date(NOW.getTime() - NOTIFICATION_OUTBOX_PROCESSING_LEASE_MS)) + expect(NOTIFICATION_OUTBOX_PROCESSING_LEASE_MS).to.equal(5 * 60 * 1000) + }) + + it('locks the rows it selects and skips rows another worker holds, so no message goes to two workers', async () => { + respondWith([]) + + await repository.claimBatch(1) + + expect(queries[0].sql).to.match(/for update skip locked$/) + }) + + it('marks the selected messages as processing and returns them that way', async () => { + const first = row({ id: 'id-1' }) + const second = row({ id: 'id-2', status: NotificationOutboxStatus.PROCESSING }) + respondWith([first, second], 2) + + const claimed = await repository.claimBatch(10) + + expect(queries).to.have.length(2) + expect(queries[1].sql).to.match( + /^update "notification_outbox" set "status" = \$1, "updated_at" = \$2 where "id" in \(\$3, \$4\)$/, + ) + expect(queries[1].bindings).to.deep.equal([NotificationOutboxStatus.PROCESSING, NOW, 'id-1', 'id-2']) + expect(claimed.map((message) => message.id)).to.deep.equal(['id-1', 'id-2']) + for (const message of claimed) { + expect(message.status).to.equal(NotificationOutboxStatus.PROCESSING) + expect(message.updatedAt).to.deep.equal(NOW) + } + }) + + it('selects and marks the messages in one transaction, so the rows stay locked until they are marked', async () => { + respondWith([row({ id: 'id-1' })], 1) + + await repository.claimBatch(10) + + expect(steps).to.deep.equal(['begin', 'select', 'update', 'commit']) + expect(new Set(stepConnections).size, 'every statement on the same connection').to.equal(1) + }) + + it('commits the transaction when there was nothing to claim', async () => { + respondWith([]) + + await repository.claimBatch(10) + + expect(steps).to.deep.equal(['begin', 'select', 'commit']) + }) + + it('rolls the claim back when marking the messages fails, so they are not left half claimed', async () => { + respondWith([row({ id: 'id-1' })], new Error('update failed')) + + let error: Error | undefined + try { + await repository.claimBatch(10) + } catch (caught) { + error = caught as Error + } + + expect(error?.message).to.contain('update failed') + expect(steps).to.deep.equal(['begin', 'select', 'update', 'rollback']) + }) + + it('maps the row, and leaves the attempt count for the caller to advance', async () => { + respondWith([row({ attempt_count: 2, last_error: 'timeout' })], 1) + + const [message] = await repository.claimBatch(1) + + expect(message).to.include({ + eventType: 'admission.invoice.paid', + attemptCount: 2, + lastError: 'timeout', + }) + expect(message.payload).to.deep.equal({ invoiceId: 'inv-1' }) + expect(message.availableAt).to.deep.equal(new Date('2026-06-15T11:00:00.000Z')) + }) + }) + + describe('.markDelivered', () => { + it('marks the message delivered now and clears its last error', async () => { + respondWith(1) + + await repository.markDelivered('id-1') + + expect(queries).to.have.length(1) + expect(queries[0].sql).to.match(/^update "notification_outbox" set /) + expect(queries[0].sql).to.include('"status" = $1') + expect(queries[0].sql).to.include('"delivered_at" = $2') + expect(queries[0].sql).to.include('"last_error" = $3') + expect(queries[0].sql).to.match(/where "id" = \$5$/) + expect(queries[0].bindings).to.deep.equal([NotificationOutboxStatus.DELIVERED, NOW, null, NOW, 'id-1']) + }) + }) + + describe('.markFailed', () => { + const update = () => { + const [{ sql, bindings }] = queries + const columns = [...sql.matchAll(/"(\w+)" = \$(\d+)/g)].reduce( + (all, [, column, position]) => ({ ...all, [column]: bindings[Number(position) - 1] }), + {} as Record, + ) + + return columns + } + + beforeEach(() => { + respondWith(1) + }) + + it('puts the message back to pending with the attempt count and the error', async () => { + await repository.markFailed('id-1', 'HTTP 500', 1, 5, 1000) + + expect(update()).to.include({ + status: NotificationOutboxStatus.PENDING, + attempt_count: 1, + last_error: 'HTTP 500', + }) + expect(queries[0].bindings[queries[0].bindings.length - 1]).to.equal('id-1') + }) + + it('waits the base delay after the first failure and doubles it after each one', async () => { + const waits: number[] = [] + for (const attempt of [1, 2, 3, 4]) { + queries.length = 0 + respondWith(1) + + await repository.markFailed('id-1', 'e', attempt, 10, 1000) + + waits.push((update().available_at as Date).getTime() - NOW.getTime()) + } + + expect(waits).to.deep.equal([1000, 2000, 4000, 8000]) + }) + + it('never waits more than a minute', async () => { + await repository.markFailed('id-1', 'e', 9, 10, 1000) + + expect((update().available_at as Date).getTime() - NOW.getTime()).to.equal(60_000) + }) + + it('does not wait at all when the base delay is negative', async () => { + await repository.markFailed('id-1', 'e', 3, 10, -500) + + expect(update().available_at).to.deep.equal(NOW) + }) + + it('gives up once the attempts are used up, marking the message dead, available now', async () => { + await repository.markFailed('id-1', 'HTTP 500', 5, 5, 1000) + + expect(update()).to.include({ status: NotificationOutboxStatus.DEAD, attempt_count: 5 }) + expect(update().available_at).to.deep.equal(NOW) + }) + + it('keeps retrying one attempt before the limit', async () => { + await repository.markFailed('id-1', 'e', 4, 5, 1000) + + expect(update().status).to.equal(NotificationOutboxStatus.PENDING) + }) + + it('keeps no more than 2000 characters of the error', async () => { + await repository.markFailed('id-1', 'x'.repeat(5000), 1, 5, 1000) + + expect(update().last_error).to.equal('x'.repeat(2000)) + }) + }) + + describe('.deleteTerminalOlderThan', () => { + const cutoff = new Date('2026-05-15T00:00:00.000Z') + + it('deletes delivered and dead messages last touched before the cutoff, never pending or processing ones', async () => { + respondWith(4) + + await repository.deleteTerminalOlderThan(cutoff) + + expect(queries[0].sql).to.match( + /^delete from "notification_outbox" where "status" in \(\$1, \$2\) and "updated_at" < \$3$/, + ) + expect(queries[0].bindings).to.deep.equal([ + NotificationOutboxStatus.DELIVERED, + NotificationOutboxStatus.DEAD, + cutoff, + ]) + }) + + it('returns how many it deleted', async () => { + respondWith(4) + + expect(await repository.deleteTerminalOlderThan(cutoff)).to.equal(4) + }) + }) +}) diff --git a/test/unit/repositories/recording-knex.ts b/test/unit/repositories/recording-knex.ts new file mode 100644 index 00000000..88517b51 --- /dev/null +++ b/test/unit/repositories/recording-knex.ts @@ -0,0 +1,82 @@ +import knex, { Knex } from 'knex' + +// knex does not export its postgres dialect, and a transaction builds its client from the dialect's prototype, +// so a subclass is the only way to answer queries for the transaction too. +// eslint-disable-next-line @typescript-eslint/no-var-requires +const ClientPG = require('knex/lib/dialects/postgres') + +/** The statements that ran, in order, with a transaction's opening and closing among them. */ +export type RecordedStep = 'begin' | 'commit' | 'rollback' | string + +export interface RecordedQuery { + sql: string + bindings: readonly unknown[] +} + +/** + * A real knex query builder on the postgres dialect that never connects: it records each statement and + * answers it with the next result `respondWith` was given. Rows for a select, a row count for an update or delete, + * or an error to fail the statement with. + */ +export const createRecordingKnex = () => { + const queries: RecordedQuery[] = [] + const steps: RecordedStep[] = [] + // The connection each step ran on, in the same order: statements of one transaction share one. + const stepConnections: number[] = [] + let connections = 0 + const responses: unknown[] = [] + + class RecordingClient extends ClientPG { + public async acquireConnection() { + return { id: ++connections } + } + + public async releaseConnection() { + return undefined + } + + public _query(connection: { id: number }, query: any) { + // A transaction opens and closes with statements of its own. They are kept apart from `queries`, which + // holds only what the code under test asked for, and recorded in `steps` to show what ran inside one. + const boundary = /^\s*(begin|commit|rollback)/i.exec(query.sql) + if (boundary) { + steps.push(boundary[1].toLowerCase()) + stepConnections.push(connection.id) + query.response = { command: 'COMMIT', rows: [], rowCount: 0 } + return Promise.resolve(query) + } + + queries.push({ sql: query.sql, bindings: query.bindings ?? [] }) + steps.push(query.sql.trim().split(' ')[0].toLowerCase()) + stepConnections.push(connection.id) + + const response = responses.length ? responses.shift() : [] + if (response instanceof Error) { + return Promise.reject(response) + } + + const isRows = Array.isArray(response) + query.response = { + command: /^\s*(update|delete|insert)/i.test(query.sql) + ? query.sql.trim().split(' ')[0].toUpperCase() + : 'SELECT', + rows: isRows ? response : [], + rowCount: isRows ? (response as unknown[]).length : response, + } + + return Promise.resolve(query) + } + } + + const db = knex({ client: RecordingClient as any }) + + return { + db: db as Knex, + queries, + steps, + stepConnections, + respondWith: (...results: unknown[]) => { + responses.push(...results) + }, + } +}