feat: add the instrument hub and record provenance

apps/api/src/instrument-records/__tests__/export-worker.test.ts

@@ -2,7 +2,7 @@ import { EventEmitter } from 'events';
 
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 
-import type { InitData, ParentMessage, RecordType } from '../thread-types';
+import type { InitPayload, ParentMessage, RecordType } from '../thread-types';
 
 class MockParentPort extends EventEmitter {
   postMessage = vi.fn();
@@ -14,14 +14,18 @@ class MockParentPort extends EventEmitter {
 
 const RUNTIME_INTERNAL_SPECIFIER = '#runtime/v1/@opendatacapture/runtime-internal/index.js';
 
-const initData: InitData = [{ edition: 2, id: 'instrument-1', name: 'Happiness Questionnaire' }];
+const initData: InitPayload = {
+  instruments: [{ edition: 2, id: 'instrument-1', name: 'Happiness Questionnaire' }],
+  seriesNames: { 'series-1': 'Happiness Series' }
+};
 
 const createRecord = (overrides: Partial<RecordType> = {}): RecordType => ({
   computedMeasures: { score: 85 },
   date: '2025-01-02',
   groupId: 'group-1',
   id: 'record-1',
   instrumentId: 'instrument-1',
+  seriesInstrumentId: null,
   session: {
     date: '2025-01-01',
     id: 'session-1',
@@ -42,6 +46,8 @@ const expectedRow = {
   instrumentEdition: 2,
   instrumentName: 'Happiness Questionnaire',
   measure: 'score',
+  seriesId: null,
+  seriesName: null,
   sessionDate: '2025-01-01',
   sessionId: 'session-1',
   sessionType: 'IN_PERSON',
@@ -102,6 +108,28 @@ describe('export-worker', () => {
     expect(parentPort.postMessage).toHaveBeenCalledWith({ success: true });
   });
 
+  // The id is the stable identifier an analysis keys on; the name is carried beside it only because
+  // a bare id in a CSV is unreadable.
+  it('should carry both the series id and its resolved title on a record a series orchestrated', async () => {
+    const parentPort = await loadInitializedWorker();
+    parentPort.send({ data: [createRecord({ seriesInstrumentId: 'series-1' })], type: 'BEGIN_CHUNK_PROCESSING' });
+    expect(parentPort.postMessage).toHaveBeenLastCalledWith({
+      data: [{ ...expectedRow, seriesId: 'series-1', seriesName: 'Happiness Series' }],
+      success: true
+    });
+  });
+
+  // A series deleted since its records were collected still has an id on them, and the export has to
+  // report what it knows rather than failing the whole download.
+  it('should report the series id with a null title when no title was resolved for it', async () => {
+    const parentPort = await loadInitializedWorker();
+    parentPort.send({ data: [createRecord({ seriesInstrumentId: 'series-missing' })], type: 'BEGIN_CHUNK_PROCESSING' });
+    expect(parentPort.postMessage).toHaveBeenLastCalledWith({
+      data: [{ ...expectedRow, seriesId: 'series-missing', seriesName: null }],
+      success: true
+    });
+  });
+
   it('should throw on an unknown message type, so a protocol mismatch is not silently ignored', async () => {
     const parentPort = (await loadWorker())!;
     expect(() => parentPort.emit('message', { type: 'UNKNOWN' })).toThrow('Unexpected message type: UNKNOWN');

apps/api/src/instrument-records/__tests__/instrument-records.controller.test.ts

@@ -44,9 +44,24 @@ describe('InstrumentRecordsController', () => {
 
   it('should pass every query filter to find, scoped to the caller ability', async () => {
     const minDate = new Date('2025-01-01');
-    await instrumentRecordsController.find(ability, 'FORM', minDate, 'group-1', 'instrument-1', 'subject-1');
+    await instrumentRecordsController.find(
+      ability,
+      'FORM',
+      minDate,
+      'group-1',
+      'instrument-1',
+      'series-1',
+      'subject-1'
+    );
     expect(instrumentRecordsService.find).toHaveBeenCalledWith(
-      { groupId: 'group-1', instrumentId: 'instrument-1', kind: 'FORM', minDate, subjectId: 'subject-1' },
+      {
+        groupId: 'group-1',
+        instrumentId: 'instrument-1',
+        kind: 'FORM',
+        minDate,
+        seriesInstrumentId: 'series-1',
+        subjectId: 'subject-1'
+      },
       { ability }
     );
   });
@@ -79,6 +94,21 @@ describe('InstrumentRecordsController', () => {
     expect(instrumentRecordsService.summarizeBySubject).toHaveBeenCalledWith({ groupId: 'group-1' }, { ability });
   });
 
+  it('should summarize the records per instrument, scoped to the caller ability', async () => {
+    const summary = [{ instrumentId: 'instrument-1', lastCollectedAt: null, recordCount: 2, subjectCount: 1 }];
+    instrumentRecordsService.summarizeByInstrument.mockResolvedValueOnce(summary);
+    await expect(instrumentRecordsController.summarizeByInstrument(ability, 'group-1')).resolves.toBe(summary);
+    expect(instrumentRecordsService.summarizeByInstrument).toHaveBeenCalledWith({ groupId: 'group-1' }, { ability });
+  });
+
+  // A series owns no records under its own id, so its totals come from a separate aggregation.
+  it('should summarize the records per series, scoped to the caller ability', async () => {
+    const summary = [{ instrumentId: 'series-1', lastCollectedAt: null, recordCount: 4, subjectCount: 2 }];
+    instrumentRecordsService.summarizeBySeries.mockResolvedValueOnce(summary);
+    await expect(instrumentRecordsController.summarizeBySeries(ability, 'group-1')).resolves.toBe(summary);
+    expect(instrumentRecordsService.summarizeBySeries).toHaveBeenCalledWith({ groupId: 'group-1' }, { ability });
+  });
+
   it('should update the record with the unwrapped data, scoped to the caller ability', async () => {
     await instrumentRecordsController.updateById('record-1', { data: { score: 2 } }, ability);
     expect(instrumentRecordsService.updateById).toHaveBeenCalledWith('record-1', { score: 2 }, { ability });

apps/api/src/instrument-records/__tests__/instrument-records.service.test.ts

@@ -676,6 +676,27 @@ describe('InstrumentRecordsService', () => {
       expect(groupsService.findById).toHaveBeenCalledBefore(instrumentRecordModel.findMany);
     });
 
+    // Asking a scalar instrument for what it orchestrated is a mistake rather than an empty result,
+    // so it is refused instead of silently returning nothing.
+    it('should reject a seriesInstrumentId that names an instrument which is not a series', async () => {
+      instrumentsService.findById.mockResolvedValueOnce({ id: 'form-1', kind: 'FORM' } as any);
+
+      await expect(instrumentRecordsService.find({ seriesInstrumentId: 'form-1' })).rejects.toBeInstanceOf(
+        UnprocessableEntityException
+      );
+      expect(instrumentRecordModel.findMany).not.toHaveBeenCalled();
+    });
+
+    it('should query the records a series orchestrated', async () => {
+      instrumentsService.findById.mockResolvedValueOnce({ id: 'series-1', kind: 'SERIES' } as any);
+
+      await instrumentRecordsService.find({ seriesInstrumentId: 'series-1' });
+
+      expect(instrumentRecordModel.findMany.mock.lastCall?.[0].where.AND).toContainEqual({
+        seriesInstrumentId: 'series-1'
+      });
+    });
+
     it('should restrict the records to instruments of the requested kind', async () => {
       instrumentsService.find.mockResolvedValueOnce([{ id: 'form-1' }, { id: 'form-2' }]);
 
@@ -735,7 +756,7 @@ describe('InstrumentRecordsService', () => {
       await instrumentRecordsService.find({ subjectId: 'subject-1' });
 
       expect(sessionModel.findMany).toHaveBeenCalledWith({
-        select: { id: true, user: { select: { username: true } } },
+        select: { id: true, type: true, user: { select: { username: true } } },
         where: { AND: [{}, { id: { in: ['session-1', 'session-2'] } }] }
       });
     });
@@ -751,7 +772,7 @@ describe('InstrumentRecordsService', () => {
       await instrumentRecordsService.find({ subjectId: 'subject-1' }, { ability });
 
       expect(sessionModel.findMany).toHaveBeenCalledWith({
-        select: { id: true, user: { select: { username: true } } },
+        select: { id: true, type: true, user: { select: { username: true } } },
         where: { AND: [accessibleQuery(ability, 'read', 'Session'), { id: { in: ['session-1'] } }] }
       });
     });
@@ -786,6 +807,7 @@ describe('InstrumentRecordsService', () => {
           groupId: '123',
           id: 'record-1',
           instrumentId: 'instrument-1',
+          seriesInstrumentId: null,
           session: {
             date: '2023-01-01',
             id: 'session-1',
@@ -915,13 +937,55 @@ describe('InstrumentRecordsService', () => {
       expect(matchStage.$match.$expr).toStrictEqual({ $eq: ['$groupId', { $toObjectId: 'group-1' }] });
     });
 
+    // A series carries no language-independent name, so the worker is handed the titles resolved
+    // here rather than the ids — an export naming only ids says nothing about what collected a row.
+    it('should resolve the title of every series the records reference before handing them to the worker', async () => {
+      const record = {
+        computedMeasures: { score: 85 },
+        date: '2023-01-01',
+        groupId: 'group-1',
+        id: 'record-1',
+        instrumentId: 'instrument-1',
+        seriesInstrumentId: 'series-1',
+        session: { date: '2023-01-01', id: 'session-1', type: 'IN_PERSON' as const, user: null },
+        subject: { age: 20, groupIds: ['group-1'], id: 'subject-1', sex: 'MALE' }
+      } satisfies RecordType;
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([record]);
+      instrumentsService.findById
+        .mockResolvedValueOnce({ id: 'instrument-1', internal: { edition: 1, name: 'X' } } as any)
+        .mockResolvedValueOnce({
+          details: { title: 'Happiness Series' },
+          id: 'series-1',
+          kind: 'SERIES',
+          language: 'en'
+        } as any);
+      const worker = useFakeWorker();
+
+      const result = instrumentRecordsService.exportRecords(
+        {},
+        { ability: createAppAbility([{ action: 'manage', subject: 'all' }]) }
+      );
+      await vi.waitFor(() => expect(worker.postMessage).toHaveBeenCalledOnce());
+
+      expect(worker.postMessage).toHaveBeenCalledWith({
+        data: {
+          instruments: [{ edition: 1, id: 'instrument-1', name: 'X' }],
+          seriesNames: { 'series-1': 'Happiness Series' }
+        },
+        type: 'INIT'
+      });
+      worker.emit('error', new Error('settle'));
+      await expect(result).rejects.toThrow('settle');
+    });
+
     describe('when the worker fails', () => {
       const record = {
         computedMeasures: { score: 85 },
         date: '2023-01-01',
         groupId: 'group-1',
         id: 'record-1',
         instrumentId: 'instrument-1',
+        seriesInstrumentId: null,
         session: { date: '2023-01-01', id: 'session-1', type: 'IN_PERSON' as const, user: null },
         subject: { age: 20, groupIds: ['group-1'], id: 'subject-1', sex: 'MALE' }
       } satisfies RecordType;
@@ -1205,13 +1269,20 @@ describe('InstrumentRecordsService', () => {
       const [{ pipeline }] = instrumentRecordModel.aggregateRaw.mock.lastCall as [{ pipeline: any[] }];
       expect(pipeline.find((stage) => stage.$group).$group._id).toStrictEqual({
         groupId: { $ifNull: ['$groupId', null] },
+        key: { $ifNull: ['$subjectId', null] },
         subjectId: { $ifNull: ['$subjectId', null] }
       });
     });
 
     it('should key the summary on the subject, so the subject hub can read a row per subject', async () => {
       instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
-        { groupId: 'group-1', lastCollectedAt: '2025-02-02T00:00:00.000Z', recordCount: 4, subjectId: 'subject-1' }
+        {
+          groupId: 'group-1',
+          key: 'subject-1',
+          lastCollectedAt: '2025-02-02T00:00:00.000Z',
+          recordCount: 4,
+          subjectId: 'subject-1'
+        }
       ]);
 
       const [summary] = await instrumentRecordsService.summarizeBySubject({}, { ability: ADMIN });
@@ -1228,8 +1299,8 @@ describe('InstrumentRecordsService', () => {
         { action: 'read', conditions: { groupId: 'group-1' }, subject: 'InstrumentRecord' }
       ]);
       instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
-        { groupId: 'group-1', lastCollectedAt: null, recordCount: 2, subjectId: 'subject-1' },
-        { groupId: 'group-2', lastCollectedAt: null, recordCount: 7, subjectId: 'subject-1' }
+        { groupId: 'group-1', key: 'subject-1', lastCollectedAt: null, recordCount: 2, subjectId: 'subject-1' },
+        { groupId: 'group-2', key: 'subject-1', lastCollectedAt: null, recordCount: 7, subjectId: 'subject-1' }
       ]);
 
       const [summary] = await instrumentRecordsService.summarizeBySubject({}, { ability });
@@ -1241,8 +1312,20 @@ describe('InstrumentRecordsService', () => {
     // is the later of the two rather than whichever arrived last.
     it('should fold a subject appearing in several readable groups into one row', async () => {
       instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
-        { groupId: 'group-1', lastCollectedAt: '2025-01-01T00:00:00.000Z', recordCount: 2, subjectId: 'subject-1' },
-        { groupId: 'group-2', lastCollectedAt: '2025-03-03T00:00:00.000Z', recordCount: 3, subjectId: 'subject-1' }
+        {
+          groupId: 'group-1',
+          key: 'subject-1',
+          lastCollectedAt: '2025-01-01T00:00:00.000Z',
+          recordCount: 2,
+          subjectId: 'subject-1'
+        },
+        {
+          groupId: 'group-2',
+          key: 'subject-1',
+          lastCollectedAt: '2025-03-03T00:00:00.000Z',
+          recordCount: 3,
+          subjectId: 'subject-1'
+        }
       ]);
 
       const summaries = await instrumentRecordsService.summarizeBySubject({}, { ability: ADMIN });
@@ -1258,10 +1341,98 @@ describe('InstrumentRecordsService', () => {
 
     it('should skip a row whose subject could not be resolved, rather than keying a summary on null', async () => {
       instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
-        { groupId: 'group-1', lastCollectedAt: null, recordCount: 2, subjectId: null }
+        { groupId: 'group-1', key: null, lastCollectedAt: null, recordCount: 2, subjectId: null }
       ]);
 
       expect(await instrumentRecordsService.summarizeBySubject({}, { ability: ADMIN })).toStrictEqual([]);
     });
   });
+
+  describe('summarizeByInstrument', () => {
+    const ADMIN = createAppAbility([{ action: 'manage', subject: 'all' }]);
+
+    it('should group by the instrument, so the hub can read a row per instrument', async () => {
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([]);
+
+      await instrumentRecordsService.summarizeByInstrument({}, { ability: ADMIN });
+
+      const [{ pipeline }] = instrumentRecordModel.aggregateRaw.mock.lastCall as [{ pipeline: any[] }];
+      expect(pipeline.find((stage) => stage.$group).$group._id.key).toStrictEqual({
+        $ifNull: ['$instrumentId', null]
+      });
+    });
+
+    // The count is of subjects, not records, which is the whole reason the pipeline groups by subject
+    // and folds here: one subject completing an instrument four times is still one subject.
+    it('should count a subject once however many records they contributed', async () => {
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
+        { groupId: 'group-1', key: 'instrument-1', lastCollectedAt: null, recordCount: 4, subjectId: 'subject-1' },
+        { groupId: 'group-1', key: 'instrument-1', lastCollectedAt: null, recordCount: 1, subjectId: 'subject-2' }
+      ]);
+
+      const [summary] = await instrumentRecordsService.summarizeByInstrument({}, { ability: ADMIN });
+
+      expect(summary).toStrictEqual({
+        instrumentId: 'instrument-1',
+        lastCollectedAt: null,
+        recordCount: 5,
+        subjectCount: 2
+      });
+    });
+
+    // Summing per-group counts would report this subject twice. Unioning the ids is what makes the
+    // count exact for a caller who can read both groups.
+    it('should count a subject shared by two readable groups once', async () => {
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
+        { groupId: 'group-1', key: 'instrument-1', lastCollectedAt: null, recordCount: 1, subjectId: 'subject-1' },
+        { groupId: 'group-2', key: 'instrument-1', lastCollectedAt: null, recordCount: 1, subjectId: 'subject-1' }
+      ]);
+
+      const [summary] = await instrumentRecordsService.summarizeByInstrument({}, { ability: ADMIN });
+
+      expect(summary).toMatchObject({ recordCount: 2, subjectCount: 1 });
+    });
+
+    // A record written before subjects were required carries no subject, and counting `null` as one
+    // would report a subject nobody can open.
+    it('should count no subject for a row naming none, while still counting its records', async () => {
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
+        { groupId: 'group-1', key: 'instrument-1', lastCollectedAt: null, recordCount: 2, subjectId: null }
+      ]);
+
+      const [summary] = await instrumentRecordsService.summarizeByInstrument({}, { ability: ADMIN });
+
+      expect(summary).toMatchObject({ recordCount: 2, subjectCount: 0 });
+    });
+  });
+
+  describe('summarizeBySeries', () => {
+    const ADMIN = createAppAbility([{ action: 'manage', subject: 'all' }]);
+
+    it('should group by the series instrument, so a series row spans the instruments it composes', async () => {
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([]);
+
+      await instrumentRecordsService.summarizeBySeries({}, { ability: ADMIN });
+
+      const [{ pipeline }] = instrumentRecordModel.aggregateRaw.mock.lastCall as [{ pipeline: any[] }];
+      expect(pipeline.find((stage) => stage.$group).$group._id.key).toStrictEqual({
+        $ifNull: ['$seriesInstrumentId', null]
+      });
+    });
+
+    // Grouping by series buckets every individually collected record under a null key. Those are not
+    // a series and must not become a row, or the hub would show one labelled nothing.
+    it('should drop the null bucket holding every record collected outside a series', async () => {
+      instrumentRecordModel.aggregateRaw.mockResolvedValueOnce([
+        { groupId: 'group-1', key: null, lastCollectedAt: null, recordCount: 90, subjectId: 'subject-1' },
+        { groupId: 'group-1', key: 'series-1', lastCollectedAt: null, recordCount: 2, subjectId: 'subject-1' }
+      ]);
+
+      const summaries = await instrumentRecordsService.summarizeBySeries({}, { ability: ADMIN });
+
+      expect(summaries).toStrictEqual([
+        { instrumentId: 'series-1', lastCollectedAt: null, recordCount: 2, subjectCount: 1 }
+      ]);
+    });
+  });
 });