Skip to content

Commit da7d638

Browse files
joshunraujoshunrau
authored
Merge pull request #1261 from DouglasNeuroInformatics/fix-threading-clone-issue
threading for instrument records export
2 parents fc3e112 + 0816f9a commit da7d638

5 files changed

Lines changed: 496 additions & 122 deletions

File tree

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

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
import type { Model } from '@douglasneuroinformatics/libnest';
2+
import { getModelToken } from '@douglasneuroinformatics/libnest';
3+
import { MockFactory } from '@douglasneuroinformatics/libnest/testing';
4+
import type { MockedInstance } from '@douglasneuroinformatics/libnest/testing';
5+
import { NotFoundException } from '@nestjs/common';
6+
import { Test } from '@nestjs/testing';
7+
import { beforeEach, describe, expect, it } from 'vitest';
8+
9+
import { GroupsService } from '../../groups/groups.service';
10+
import { InstrumentsService } from '../../instruments/instruments.service';
11+
import { SessionsService } from '../../sessions/sessions.service';
12+
import { SubjectsService } from '../../subjects/subjects.service';
13+
import { InstrumentMeasuresService } from '../instrument-measures.service';
14+
import { InstrumentRecordsService } from '../instrument-records.service';
15+
16+
import type { RecordType } from '../thread-types';
17+
18+
describe('InstrumentRecordsService', () => {
19+
let instrumentRecordsService: InstrumentRecordsService;
20+
let instrumentRecordModel: MockedInstance<Model<'InstrumentRecord'>>;
21+
let instrumentsService: MockedInstance<InstrumentsService>;
22+
23+
beforeEach(async () => {
24+
const moduleRef = await Test.createTestingModule({
25+
providers: [
26+
InstrumentRecordsService,
27+
MockFactory.createForModelToken(getModelToken('InstrumentRecord')),
28+
MockFactory.createForService(GroupsService),
29+
MockFactory.createForService(InstrumentMeasuresService),
30+
MockFactory.createForService(InstrumentsService),
31+
MockFactory.createForService(SessionsService),
32+
MockFactory.createForService(SubjectsService)
33+
]
34+
}).compile();
35+
36+
instrumentRecordModel = moduleRef.get(getModelToken('InstrumentRecord'));
37+
instrumentRecordsService = moduleRef.get(InstrumentRecordsService);
38+
instrumentsService = moduleRef.get(InstrumentsService);
39+
});
40+
41+
describe('findById', () => {
42+
it('should throw a NotFoundException if no record is found', async () => {
43+
instrumentRecordModel.findFirst.mockResolvedValueOnce(null);
44+
await expect(instrumentRecordsService.findById('nonexistent-id')).rejects.toBeInstanceOf(NotFoundException);
45+
});
46+
47+
it('should return the instrument record with the correct shape', async () => {
48+
const mockRecord = {
49+
data: { test: 'data' },
50+
date: new Date(),
51+
id: 'test-record-id',
52+
instrumentId: 'test-instrument-id',
53+
sessionId: 'test-session-id',
54+
subjectId: 'test-subject-id'
55+
};
56+
57+
instrumentRecordModel.findFirst.mockResolvedValueOnce(mockRecord);
58+
59+
const result = await instrumentRecordsService.findById('test-record-id');
60+
61+
expect(result).toMatchObject({
62+
data: { test: 'data' },
63+
id: 'test-record-id',
64+
instrumentId: 'test-instrument-id',
65+
sessionId: 'test-session-id',
66+
subjectId: 'test-subject-id'
67+
});
68+
expect(result.date).toBeInstanceOf(Date);
69+
});
70+
});
71+
72+
describe('exportRecords', () => {
73+
it('should return an array of export records with correct shape', async () => {
74+
const mockRecords = [
75+
{
76+
computedMeasures: { score: 85 },
77+
date: '2023-01-01',
78+
groupId: '123',
79+
id: 'record-1',
80+
instrumentId: 'instrument-1',
81+
session: {
82+
date: '2023-01-01',
83+
id: 'session-1',
84+
type: 'IN_PERSON' as const,
85+
user: { username: 'testuser' }
86+
},
87+
subject: {
88+
age: 20,
89+
groupIds: ['group-1'],
90+
id: 'subject-1',
91+
sex: 'MALE' as const
92+
}
93+
}
94+
] satisfies RecordType[];
95+
96+
const mockInstruments = [
97+
{
98+
id: 'instrument-1',
99+
internal: { edition: 1, name: 'Test Instrument' }
100+
}
101+
];
102+
103+
instrumentRecordModel.aggregateRaw.mockResolvedValueOnce(mockRecords);
104+
instrumentsService.findById.mockResolvedValueOnce(mockInstruments[0]);
105+
106+
const ability = { can: () => true } as any;
107+
108+
const result = await instrumentRecordsService.exportRecords({}, { ability });
109+
110+
expect(Array.isArray(result)).toBe(true);
111+
expect(result.length).toBeGreaterThan(0);
112+
expect(result[0]).toMatchObject({
113+
groupId: '123',
114+
instrumentEdition: 1,
115+
instrumentName: 'Test Instrument',
116+
measure: 'score',
117+
sessionId: 'session-1',
118+
sessionType: 'IN_PERSON',
119+
subjectId: expect.any(String),
120+
timestamp: expect.any(String),
121+
username: 'testuser',
122+
value: 85
123+
});
124+
});
125+
});
126+
});

‎apps/api/src/instrument-records/export-worker.ts‎

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
import { parentPort } from 'worker_threads';
2+
3+
import type { FormTypes, InstrumentMeasureValue } from '@opendatacapture/runtime-core';
4+
import { DEFAULT_GROUP_NAME } from '@opendatacapture/schemas/core';
5+
import type { InstrumentRecordsExport } from '@opendatacapture/schemas/instrument-records';
6+
import { removeSubjectIdScope } from '@opendatacapture/subject-utils';
7+
8+
import type { BeginChunkProcessingData, InitData, ParentMessage, RecordType, WorkerMessage } from './thread-types';
9+
10+
type ExpandDataType =
11+
| {
12+
measure: string;
13+
measureValue: FormTypes.RecordArrayFieldValue | InstrumentMeasureValue;
14+
success: true;
15+
}
16+
| {
17+
message: string;
18+
success: false;
19+
};
20+
21+
function expandData(listEntry: any[]): ExpandDataType[] {
22+
const validRecordArrayList: ExpandDataType[] = [];
23+
if (listEntry.length < 1) {
24+
throw new Error('Record Array is Empty');
25+
}
26+
for (const objectEntry of Object.values(listEntry)) {
27+
for (const [dataKey, dataValue] of Object.entries(
28+
objectEntry as { [key: string]: FormTypes.RecordArrayFieldValue }
29+
)) {
30+
validRecordArrayList.push({
31+
measure: dataKey,
32+
measureValue: dataValue,
33+
success: true
34+
});
35+
}
36+
}
37+
return validRecordArrayList;
38+
}
39+
40+
let initData: Map<
41+
string | undefined,
42+
{
43+
edition: number;
44+
id: string;
45+
name: string;
46+
}
47+
>;
48+
49+
function handleInit(data: InitData) {
50+
initData = new Map(data.map((instrument) => [instrument.id, instrument]));
51+
52+
parentPort?.postMessage({ success: true });
53+
}
54+
55+
function handleChunkComplete(_data: BeginChunkProcessingData) {
56+
if (!initData) {
57+
throw new Error('Expected init data to be defined');
58+
}
59+
const instrumentsMap = initData;
60+
61+
const processRecord = (record: RecordType): InstrumentRecordsExport => {
62+
const instrument = instrumentsMap.get(record.instrumentId)!;
63+
64+
if (!record.computedMeasures) return [];
65+
66+
//const instrument = instrumentsMap.get(record.instrumentId)!;
67+
const rows: InstrumentRecordsExport = [];
68+
69+
for (const [measureKey, measureValue] of Object.entries(record.computedMeasures)) {
70+
if (measureValue == null) continue;
71+
72+
if (!Array.isArray(measureValue)) {
73+
rows.push({
74+
groupId: record.groupId ?? DEFAULT_GROUP_NAME,
75+
instrumentEdition: instrument.edition,
76+
instrumentName: instrument.name,
77+
measure: measureKey,
78+
sessionDate: record.session.date,
79+
sessionId: record.session.id,
80+
sessionType: record.session.type,
81+
subjectAge: record.subject.age,
82+
subjectId: removeSubjectIdScope(record.subject.id),
83+
subjectSex: record.subject.sex,
84+
timestamp: record.date,
85+
username: record.session.user?.username ?? 'N/A',
86+
value: measureValue as InstrumentMeasureValue
87+
});
88+
continue;
89+
}
90+
91+
if (measureValue.length < 1) continue;
92+
93+
const expanded = expandData(measureValue);
94+
for (const entry of expanded) {
95+
if (!entry.success) {
96+
throw new Error(`exportRecords: ${instrument.name}.${measureKey} — ${entry.message}`);
97+
}
98+
rows.push({
99+
groupId: record.groupId ?? DEFAULT_GROUP_NAME,
100+
instrumentEdition: instrument.edition,
101+
instrumentName: instrument.name,
102+
measure: `${measureKey} - ${entry.measure}`,
103+
sessionDate: record.session.date,
104+
sessionId: record.session.id,
105+
sessionType: record.session.type,
106+
subjectAge: record.subject.age,
107+
subjectId: removeSubjectIdScope(record.subject.id),
108+
subjectSex: record.subject.sex,
109+
timestamp: record.date,
110+
username: record.session.user?.username ?? 'N/A',
111+
value: entry.measureValue
112+
});
113+
}
114+
}
115+
return rows;
116+
};
117+
118+
try {
119+
const results = _data.map(processRecord);
120+
parentPort?.postMessage({ data: results.flat(), success: true } satisfies WorkerMessage);
121+
} catch (error) {
122+
parentPort?.postMessage({ error: (error as Error).message, success: false } satisfies WorkerMessage);
123+
}
124+
}
125+
126+
parentPort!.on('message', (message: ParentMessage) => {
127+
switch (message.type) {
128+
case 'BEGIN_CHUNK_PROCESSING':
129+
return handleChunkComplete(message.data);
130+
case 'INIT':
131+
return handleInit(message.data);
132+
default:
133+
throw new Error(`Unexpected message type: ${(message satisfies never as { [key: string]: any }).type}`);
134+
}
135+
});

0 commit comments

Comments
 (0)