From 6105c6299830657c2e1c7bf8149ba44ba90f02de Mon Sep 17 00:00:00 2001 From: "Gabriel A. Devenyi" Date: Mon, 20 Jul 2026 15:05:33 -0400 Subject: [PATCH 1/2] perf(api): scope the upload response and validate before writing Three changes to the bulk record upload path. The response returned every record in the group for that instrument rather than the records the request created: return this.instrumentRecordModel.findMany({ where: { groupId, instrumentId } }); so the response grew with the size of the database instead of the size of the request. Measured against a live instance, uploading 25 records into an instrument that already held 1,200: before 1,300 records, 871,056 bytes (and larger on every upload) after 25 records, 14,691 bytes (flat) Records are now validated up front, before any session is created. A malformed record in the middle of a batch previously rejected the request only after sessions had been created for the records ahead of it, which then had to be rolled back by the catch handler. `SessionsService.create` created a session and then immediately re-read it to attach the subject. The create now returns it directly with the same `include`, removing a round trip per session for every caller -- the upload path, the gateway synchronizer, the demo seeder and the sessions controller. Roughly 25 fewer mongodb operations on a 25-record upload, matching one fewer query per session. Not addressed here: sessions are still created one per record. Batching them means reproducing `SessionsService.create`'s subject resolution and group association inline, including its existing quirk of only setting a session's groupId when the subject was not already a member of that group. That belongs in a refactor of the sessions service with its own review rather than duplicated into the upload path. Refs #1414 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Bfen9VQemZanJLXinJ6MMG --- .../instrument-records.service.spec.ts | 32 +++++++++++++++++ .../instrument-records.service.ts | 35 +++++++++++-------- apps/api/src/sessions/sessions.service.ts | 14 ++++---- 3 files changed, 58 insertions(+), 23 deletions(-) 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..6c8d6cdb7 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,38 @@ 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(); + }); }); 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..627166d30 100644 --- a/apps/api/src/instrument-records/instrument-records.service.ts +++ b/apps/api/src/instrument-records/instrument-records.service.ts @@ -401,6 +401,19 @@ 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) => { + const parseResult = instrument.validationSchema.safeParse(this.parseJson(record.data)); + if (!parseResult.success) { + console.error(parseResult.error.issues); + throw new UnprocessableEntityException( + `Data received for record does not pass validation schema of instrument '${instrument.id}'` + ); + } + return { data: parseResult.data, date: record.date, subjectId: record.subjectId }; + }); + const createdSessionsArray: Session[] = []; try { @@ -413,17 +426,8 @@ 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}'` - ); - } + validatedRecords.map(async (record) => { + const { data, date, subjectId } = record; const session = await this.sessionsService.create({ date: date, @@ -436,12 +440,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, @@ -455,10 +459,11 @@ export class InstrumentRecordsService { 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..a86b8681a 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 rather than re-read afterwards, which cost a second round trip + // for every session created include: { subject: true - }, - where: { id } - }))!; + } + }); } async deleteById(id: string, { ability }: EntityOperationOptions = {}) { From cba4f8f644dc7a094b8cdf837efe2a52b7847ee4 Mon Sep 17 00:00:00 2001 From: "Gabriel A. Devenyi" Date: Mon, 27 Jul 2026 15:05:28 -0400 Subject: [PATCH 2/2] fix(api): settle every upload task before rolling back its sessions Three points from review on this branch: `Promise.all` rejects on the first failure while its siblings are still in flight, so a session created after the catch had begun was missing from `createdSessionsArray` and survived the rollback -- an orphaned session pointing at no record. Settle every task first, then rethrow, so the array is complete before anything is undone. Pinned by a test that fails against `Promise.all`. Validation failure no longer writes the zod issues to stdout, and carries them on the response instead, matching what `create` already does. The message names the offending index, which a batch needs and a single-record create does not. Grammar in the `SessionsService.create` comment. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014ZEJLwa7jwD8sKzcE4PfSc --- .../instrument-records.service.spec.ts | 44 +++++++++++++++++++ .../instrument-records.service.ts | 26 ++++++++--- apps/api/src/sessions/sessions.service.ts | 4 +- 3 files changed, 66 insertions(+), 8 deletions(-) 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 6c8d6cdb7..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 @@ -281,6 +281,50 @@ describe('InstrumentRecordsService', () => { 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 627166d30..8bc6dd16a 100644 --- a/apps/api/src/instrument-records/instrument-records.service.ts +++ b/apps/api/src/instrument-records/instrument-records.service.ts @@ -403,13 +403,15 @@ 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) => { + const validatedRecords = records.map((record, index) => { const parseResult = instrument.validationSchema.safeParse(this.parseJson(record.data)); if (!parseResult.success) { - console.error(parseResult.error.issues); - throw new UnprocessableEntityException( - `Data received for record does not pass validation schema of instrument '${instrument.id}'` - ); + 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 }; }); @@ -425,7 +427,11 @@ export class InstrumentRecordsService { await this.subjectsService.createMany(subjectIdList); - const preProcessedRecords = await Promise.all( + // `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; @@ -455,6 +461,14 @@ export class InstrumentRecordsService { }; }) ); + + const preProcessedRecords = settled.map((result) => { + if (result.status === 'rejected') { + throw result.reason; + } + return result.value; + }); + await this.instrumentRecordModel.createMany({ data: preProcessedRecords }); diff --git a/apps/api/src/sessions/sessions.service.ts b/apps/api/src/sessions/sessions.service.ts index a86b8681a..8ba48b645 100644 --- a/apps/api/src/sessions/sessions.service.ts +++ b/apps/api/src/sessions/sessions.service.ts @@ -69,8 +69,8 @@ export class SessionsService { } : undefined }, - // returned by the create itself rather than re-read afterwards, which cost a second round trip - // for every session created + // Returned by the create itself; re-reading it afterwards cost a second round trip for every + // session created. include: { subject: true }