feat: add the instrument hub and record provenance

apps/api/src/instrument-records/__tests__/export-worker.spec.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.spec.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 }
     );
   });

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

@@ -735,7 +735,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 +751,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 +786,7 @@ describe('InstrumentRecordsService', () => {
           groupId: '123',
           id: 'record-1',
           instrumentId: 'instrument-1',
+          seriesInstrumentId: null,
           session: {
             date: '2023-01-01',
             id: 'session-1',
@@ -922,6 +923,7 @@ describe('InstrumentRecordsService', () => {
         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;
@@ -967,7 +969,7 @@ describe('InstrumentRecordsService', () => {
         await expect(Promise.race([result, Promise.resolve('pending')])).resolves.toBe('pending');
         expect(worker.postMessage).toHaveBeenCalledTimes(1);
         expect(worker.postMessage).toHaveBeenCalledWith({
-          data: [{ edition: 1, id: 'instrument-1', name: 'X' }],
+          data: { instruments: [{ edition: 1, id: 'instrument-1', name: 'X' }], seriesNames: {} },
           type: 'INIT'
         });
         worker.emit('error', new Error('settle'));
@@ -1234,13 +1236,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 });
@@ -1257,8 +1266,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 });
@@ -1270,8 +1279,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 });
@@ -1287,10 +1308,86 @@ 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 });
+    });
+  });
+
+  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 }
+      ]);
+    });
+  });
 });

apps/api/src/instrument-records/export-worker.js

@@ -60,12 +60,19 @@ function expandData(listEntry) {
  */
 let initData;
 
+/**
+ * Internal cache mapping a series instrument id to its title.
+ * @type {{ [id: string]: string }}
+ */
+let seriesNames = {};
+
 /**
  * Initializes the worker with instrument metadata.
- * * @param {Array<{id: string, edition: number, name: string}>} data - The initialization payload.
+ * @param {import('./thread-types').InitPayload} data - The initialization payload.
  */
 function handleInit(data) {
-  initData = new Map(data.map((instrument) => [instrument.id, instrument]));
+  initData = new Map(data.instruments.map((instrument) => [instrument.id, instrument]));
+  seriesNames = data.seriesNames;
   parentPort?.postMessage({ success: true });
 }
 
@@ -97,6 +104,9 @@ function handleChunkComplete(_data) {
 
     const rows = [];
 
+    const seriesId = record.seriesInstrumentId ?? null;
+    const seriesName = seriesId ? (seriesNames[seriesId] ?? null) : null;
+
     for (const [measureKey, measureValue] of Object.entries(record.computedMeasures)) {
       if (measureValue == null) continue;
 
@@ -106,6 +116,8 @@ function handleChunkComplete(_data) {
           instrumentEdition: instrument.edition,
           instrumentName: instrument.name,
           measure: measureKey,
+          seriesId,
+          seriesName,
           sessionDate: record.session.date,
           sessionId: record.session.id,
           sessionType: record.session.type,
@@ -131,6 +143,8 @@ function handleChunkComplete(_data) {
           instrumentEdition: instrument.edition,
           instrumentName: instrument.name,
           measure: `${measureKey} - ${entry.measure}`,
+          seriesId,
+          seriesName,
           sessionDate: record.session.date,
           sessionId: record.session.id,
           sessionType: record.session.type,

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

@@ -62,9 +62,13 @@ export class InstrumentRecordsController {
     minDate?: Date,
     @Query('groupId') groupId?: string,
     @Query('instrumentId') instrumentId?: string,
+    @Query('seriesInstrumentId') seriesInstrumentId?: string,
     @Query('subjectId') subjectId?: string
   ) {
-    return this.instrumentRecordsService.find({ groupId, instrumentId, kind, minDate, subjectId }, { ability });
+    return this.instrumentRecordsService.find(
+      { groupId, instrumentId, kind, minDate, seriesInstrumentId, subjectId },
+      { ability }
+    );
   }
 
   @ApiOperation({ summary: 'Delete Record' })
@@ -94,6 +98,20 @@ export class InstrumentRecordsController {
     return this.instrumentRecordsService.linearModel({ groupId, instrumentId }, { ability });
   }
 
+  @ApiOperation({ description: 'Per-instrument record and subject counts', summary: 'Summarize Records By Instrument' })
+  @Get('summary/by-instrument')
+  @RouteAccess({ action: 'read', subject: 'InstrumentRecord' })
+  summarizeByInstrument(@CurrentUser('ability') ability: AppAbility, @Query('groupId') groupId?: string) {
+    return this.instrumentRecordsService.summarizeByInstrument({ groupId }, { ability });
+  }
+
+  @ApiOperation({ description: 'Per-series record counts', summary: 'Summarize Records By Series' })
+  @Get('summary/by-series')
+  @RouteAccess({ action: 'read', subject: 'InstrumentRecord' })
+  summarizeBySeries(@CurrentUser('ability') ability: AppAbility, @Query('groupId') groupId?: string) {
+    return this.instrumentRecordsService.summarizeBySeries({ groupId }, { ability });
+  }
+
   @ApiOperation({ description: 'Per-subject record counts', summary: 'Summarize Records By Subject' })
   @Get('summary/by-subject')
   @RouteAccess({ action: 'read', subject: 'InstrumentRecord' })