98 lines
5.9 KiB
TypeScript
98 lines
5.9 KiB
TypeScript
import { describe, expect, it } from 'vitest';
|
|
import type { ImportArtifact, ImportProcessingRun, IngestionJob } from '../../src/domain/ingestion-persistence';
|
|
import { LocalAssessorRepository } from '../../src/persistence/local-assessor-repository';
|
|
import { RepositoryConflictError, RepositoryValidationError } from '../../src/persistence/repository-errors';
|
|
|
|
const project = {
|
|
id: 'ingestion-project', name: 'Synthetic ingestion project',
|
|
createdAt: '2026-07-20T12:00:00.000Z', updatedAt: '2026-07-20T12:00:00.000Z', attributes: {}
|
|
};
|
|
|
|
function artifact(id = 'artifact-1', parentArtifactId?: string): ImportArtifact {
|
|
return {
|
|
version: 1, id, projectId: project.id, parentArtifactId, kind: parentArtifactId ? 'container-entry' : 'source-file',
|
|
relativePath: parentArtifactId ? 'entries/report.xml' : 'upload/report.zip', sha256: 'a'.repeat(64),
|
|
sizeBytes: 100, sensitivity: 'confidential', parseStatus: 'pending',
|
|
storage: { kind: 'backend-controlled', storageKey: `project/${id}` }, createdAt: '2026-07-20T12:00:00.000Z'
|
|
};
|
|
}
|
|
|
|
function processingRun(id: string, supersedesRunId?: string): ImportProcessingRun {
|
|
return {
|
|
version: 1, id, projectId: project.id, artifactId: 'artifact-1', supersedesRunId, status: 'complete',
|
|
processors: [{ kind: 'format-adapter', id: 'synthetic-parser', version: id === 'run-1' ? '1.0.0' : '2.0.0' }],
|
|
diagnosticCodes: [], startedAt: '2026-07-20T12:01:00.000Z', completedAt: '2026-07-20T12:01:01.000Z'
|
|
};
|
|
}
|
|
|
|
function job(id: string, maxAttempts = 2): IngestionJob {
|
|
return {
|
|
version: 1, id, projectId: project.id, artifactId: 'artifact-1', kind: 'parse-artifact',
|
|
idempotencyKey: `parse:${id}`, status: 'queued', attempts: 0, maxAttempts,
|
|
availableAt: '2026-07-20T12:00:00.000Z', createdAt: '2026-07-20T12:00:00.000Z', updatedAt: '2026-07-20T12:00:00.000Z'
|
|
};
|
|
}
|
|
|
|
async function repository() {
|
|
const value = new LocalAssessorRepository();
|
|
await value.saveProject(project);
|
|
await value.saveImportArtifact(artifact());
|
|
return value;
|
|
}
|
|
|
|
describe('persistent ingestion state', () => {
|
|
it('stores immutable artifact ancestry and versioned processing history', async () => {
|
|
const value = await repository();
|
|
await value.saveImportArtifact(artifact('artifact-child', 'artifact-1'));
|
|
await value.saveProcessingRun(processingRun('run-1'));
|
|
await value.saveProcessingRun(processingRun('run-2', 'run-1'));
|
|
|
|
await expect(value.saveProcessingRun(processingRun('run-2', 'run-1'))).rejects.toBeInstanceOf(RepositoryConflictError);
|
|
await expect(value.listImportArtifacts(project.id)).resolves.toHaveLength(2);
|
|
await expect(value.listProcessingRuns('artifact-1')).resolves.toMatchObject([
|
|
{ id: 'run-1', processors: [{ version: '1.0.0' }] },
|
|
{ id: 'run-2', supersedesRunId: 'run-1', processors: [{ version: '2.0.0' }] }
|
|
]);
|
|
|
|
await expect(value.saveImportArtifact({ ...artifact('bad-path'), relativePath: '../escape.txt' }))
|
|
.rejects.toBeInstanceOf(RepositoryValidationError);
|
|
});
|
|
|
|
it('deduplicates jobs and enforces leases, retries and dead-letter limits', async () => {
|
|
const value = await repository();
|
|
const queued = job('job-1');
|
|
await expect(value.enqueueIngestionJob(queued)).resolves.toEqual(queued);
|
|
const revisionAfterInsert = value.revision;
|
|
await expect(value.enqueueIngestionJob({ ...queued, id: 'different-id' })).resolves.toEqual(queued);
|
|
expect(value.revision).toBe(revisionAfterInsert);
|
|
|
|
const first = await value.claimNextIngestionJob({ workerId: 'worker-a', now: '2026-07-20T12:00:00.000Z', leaseDurationMs: 1000 });
|
|
expect(first).toMatchObject({ status: 'running', attempts: 1, leaseOwner: 'worker-a' });
|
|
await expect(value.completeIngestionJob('job-1', 'worker-b', '2026-07-20T12:00:00.500Z')).rejects.toBeInstanceOf(RepositoryConflictError);
|
|
await expect(value.failIngestionJob('job-1', 'worker-a', '2026-07-20T12:00:00.500Z', 'synthetic-error', '2026-07-20T12:00:02.000Z'))
|
|
.resolves.toMatchObject({ status: 'queued', attempts: 1, lastErrorCode: 'synthetic-error' });
|
|
await expect(value.claimNextIngestionJob({ workerId: 'worker-b', now: '2026-07-20T12:00:01.000Z', leaseDurationMs: 1000 })).resolves.toBeNull();
|
|
await value.claimNextIngestionJob({ workerId: 'worker-b', now: '2026-07-20T12:00:02.000Z', leaseDurationMs: 1000 });
|
|
await expect(value.failIngestionJob('job-1', 'worker-b', '2026-07-20T12:00:02.500Z', 'still-failing', '2026-07-20T12:00:04.000Z'))
|
|
.resolves.toMatchObject({ status: 'dead-letter', attempts: 2 });
|
|
});
|
|
|
|
it('recovers expired leases without allowing the stale worker to complete', async () => {
|
|
const value = await repository();
|
|
await value.enqueueIngestionJob(job('job-recovery', 3));
|
|
await value.claimNextIngestionJob({ workerId: 'worker-old', now: '2026-07-20T12:00:00.000Z', leaseDurationMs: 1000 });
|
|
const recovered = await value.claimNextIngestionJob({ workerId: 'worker-new', now: '2026-07-20T12:00:01.001Z', leaseDurationMs: 1000 });
|
|
expect(recovered).toMatchObject({ status: 'running', attempts: 2, leaseOwner: 'worker-new' });
|
|
await expect(value.completeIngestionJob('job-recovery', 'worker-old', '2026-07-20T12:00:01.200Z')).rejects.toBeInstanceOf(RepositoryConflictError);
|
|
await expect(value.completeIngestionJob('job-recovery', 'worker-new', '2026-07-20T12:00:01.500Z'))
|
|
.resolves.toMatchObject({ status: 'succeeded', attempts: 2 });
|
|
|
|
await value.enqueueIngestionJob(job('job-final-lease', 1));
|
|
await value.claimNextIngestionJob({ workerId: 'worker-crashed', now: '2026-07-20T12:00:02.000Z', leaseDurationMs: 1000 });
|
|
await expect(value.claimNextIngestionJob({ workerId: 'worker-recovery', now: '2026-07-20T12:00:03.001Z', leaseDurationMs: 1000 })).resolves.toBeNull();
|
|
await expect(value.listIngestionJobs(project.id)).resolves.toContainEqual(expect.objectContaining({
|
|
id: 'job-final-lease', status: 'dead-letter', attempts: 1, lastErrorCode: 'lease-expired'
|
|
}));
|
|
});
|
|
});
|