perf(api): scope the upload response and validate before writing

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', () => {

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) {

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 = {}) {