diff --git a/apps/api/src/instrument-records/__tests__/instrument-records.service.spec.ts b/apps/api/src/instrument-records/__tests__/instrument-records.service.spec.ts index 6974ccade..da8ae74bc 100644 --- a/apps/api/src/instrument-records/__tests__/instrument-records.service.spec.ts +++ b/apps/api/src/instrument-records/__tests__/instrument-records.service.spec.ts @@ -249,6 +249,82 @@ describe('InstrumentRecordsService', () => { data: [expect.objectContaining({ instrumentId: 'instrument-1', subjectId: 'subject-1' })] }); }); + + it('should return only the records this upload created, not every record in the group', async () => { + await instrumentRecordsService.upload({ ...baseUploadData, groupId: 'group-1' }); + + expect(instrumentRecordModel.findMany).toHaveBeenCalledWith({ + where: { sessionId: { in: ['session-1'] } } + }); + const [call] = instrumentRecordModel.findMany.mock.lastCall as [{ where: { [key: string]: unknown } }]; + expect(call.where).not.toHaveProperty('groupId'); + expect(call.where).not.toHaveProperty('instrumentId'); + }); + + it('should reject an invalid record before creating any sessions', async () => { + instrumentsService.findById.mockResolvedValue({ + ...mockInstrument, + validationSchema: { safeParse: () => ({ error: { issues: [] }, success: false }) } + } as any); + + await expect( + instrumentRecordsService.upload({ + ...baseUploadData, + records: [ + { data: { answer: 1 }, date: new Date(), subjectId: 'subject-1' }, + { data: { answer: 2 }, date: new Date(), subjectId: 'subject-2' } + ] + }) + ).rejects.toBeInstanceOf(UnprocessableEntityException); + + expect(sessionsService.create).not.toHaveBeenCalled(); + expect(subjectsService.createMany).not.toHaveBeenCalled(); + expect(instrumentRecordModel.createMany).not.toHaveBeenCalled(); + }); + + it('should report which record failed and why, so a rejected batch can be corrected', async () => { + const issues = [{ message: 'Required', path: ['answer'] }]; + instrumentsService.findById.mockResolvedValue({ + ...mockInstrument, + validationSchema: { + safeParse: (data: any) => + data.answer === 2 ? { error: { issues }, success: false } : { data, success: true } + } + } as any); + + await expect( + instrumentRecordsService.upload({ + ...baseUploadData, + records: [ + { data: { answer: 1 }, date: new Date(), subjectId: 'subject-1' }, + { data: { answer: 2 }, date: new Date(), subjectId: 'subject-2' } + ] + }) + ).rejects.toMatchObject({ + response: { issues, message: expect.stringContaining('at index 1') } + }); + }); + + it('should roll back a session that was created after an earlier record had already failed', async () => { + sessionsService.create + .mockRejectedValueOnce(new Error('could not create session')) + .mockImplementationOnce(async () => { + await new Promise((resolve) => setTimeout(resolve, 0)); + return { ...mockSession, id: 'session-2' }; + }); + + await expect( + instrumentRecordsService.upload({ + ...baseUploadData, + records: [ + { data: { answer: 1 }, date: new Date(), subjectId: 'subject-1' }, + { data: { answer: 2 }, date: new Date(), subjectId: 'subject-2' } + ] + }) + ).rejects.toThrow('could not create session'); + + expect(sessionsService.deleteByIds).toHaveBeenCalledWith(['session-2']); + }); }); describe('find', () => { diff --git a/apps/api/src/instrument-records/instrument-records.service.ts b/apps/api/src/instrument-records/instrument-records.service.ts index e3fc2b682..8bc6dd16a 100644 --- a/apps/api/src/instrument-records/instrument-records.service.ts +++ b/apps/api/src/instrument-records/instrument-records.service.ts @@ -401,6 +401,21 @@ export class InstrumentRecordsService { } } + // Every record is validated before anything is written, so a malformed record in the middle of a + // batch rejects the request without first creating sessions that then have to be rolled back. + const validatedRecords = records.map((record, index) => { + const parseResult = instrument.validationSchema.safeParse(this.parseJson(record.data)); + if (!parseResult.success) { + throw new UnprocessableEntityException({ + error: 'Unprocessable Entity', + issues: parseResult.error.issues, + message: `Data received for record at index ${index} does not pass validation schema of instrument '${instrument.id}'`, + statusCode: 422 + }); + } + return { data: parseResult.data, date: record.date, subjectId: record.subjectId }; + }); + const createdSessionsArray: Session[] = []; try { @@ -412,18 +427,13 @@ export class InstrumentRecordsService { await this.subjectsService.createMany(subjectIdList); - const preProcessedRecords = await Promise.all( - records.map(async (record) => { - const { data: rawData, date, subjectId } = record; - - // Validate data - const parseResult = instrument.validationSchema.safeParse(this.parseJson(rawData)); - if (!parseResult.success) { - console.error(parseResult.error.issues); - throw new UnprocessableEntityException( - `Data received for record does not pass validation schema of instrument '${instrument.id}'` - ); - } + // `allSettled` rather than `all`: `all` rejects on the first failure while its siblings are + // still in flight, so a session created after the catch had begun would be missing from + // `createdSessionsArray` and survive the rollback. Settling first means every session that + // exists is recorded before anything is undone. + const settled = await Promise.allSettled( + validatedRecords.map(async (record) => { + const { data, date, subjectId } = record; const session = await this.sessionsService.create({ date: date, @@ -436,12 +446,12 @@ export class InstrumentRecordsService { createdSessionsArray.push(session); const computedMeasures = instrument.measures - ? this.instrumentMeasuresService.computeMeasures(instrument.measures, parseResult.data) + ? this.instrumentMeasuresService.computeMeasures(instrument.measures, data) : null; return { computedMeasures, - data: this.serializeData(parseResult.data), + data: this.serializeData(data), date, groupId, instrumentId, @@ -451,14 +461,23 @@ export class InstrumentRecordsService { }; }) ); + + const preProcessedRecords = settled.map((result) => { + if (result.status === 'rejected') { + throw result.reason; + } + return result.value; + }); + await this.instrumentRecordModel.createMany({ data: preProcessedRecords }); + // Scoped to the sessions this upload created, rather than every record in the group for this + // instrument: the response should describe the request, not the size of the database. return this.instrumentRecordModel.findMany({ where: { - groupId, - instrumentId + sessionId: { in: createdSessionsArray.map((session) => session.id) } } }); } catch (err) { diff --git a/apps/api/src/sessions/sessions.service.ts b/apps/api/src/sessions/sessions.service.ts index 2b0889424..8ba48b645 100644 --- a/apps/api/src/sessions/sessions.service.ts +++ b/apps/api/src/sessions/sessions.service.ts @@ -51,7 +51,7 @@ export class SessionsService { } } - const { id } = await this.sessionModel.create({ + return this.sessionModel.create({ data: { date, group: group @@ -68,15 +68,13 @@ export class SessionsService { connect: { id: user.id } } : undefined - } - }); - - return (await this.sessionModel.findUnique({ + }, + // Returned by the create itself; re-reading it afterwards cost a second round trip for every + // session created. include: { subject: true - }, - where: { id } - }))!; + } + }); } async deleteById(id: string, { ability }: EntityOperationOptions = {}) {