Skip to content

perf(api): batch session creation and scope the upload response (alternative to #1424) #1427

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gdevenyi wants to merge 3 commits into
base: main
Choose a base branch
from
Open

perf(api): batch session creation and scope the upload response (alternative to #1424) #1427

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bac6420
perf(api): batch session creation and scope the upload response
gdevenyi Jul 20, 2026
806dbf8
refactor(api): resolve batch subjects through the subjects service
gdevenyi Jul 28, 2026
e6150dd
fix(api): drop the subject read-back and fail loudly on a session rea…
gdevenyi Jul 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 122 additions & 10 deletions apps/api/src/instrument-records/__tests__/instrument-records.service.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,19 +184,60 @@ describe('InstrumentRecordsService', () => {
beforeEach(() => {
instrumentsService.findById.mockResolvedValue(mockInstrument as any);
subjectsService.createMany.mockResolvedValue([] as any);
sessionsService.create.mockResolvedValue(mockSession as any);
sessionsService.createMany.mockResolvedValue([mockSession] as any);
sessionsService.deleteByIds.mockResolvedValue(undefined as any);
instrumentRecordModel.createMany.mockResolvedValue([] as any);
instrumentRecordModel.findMany.mockResolvedValue([] as any);
});

it('should call sessionsService.create with the provided username', async () => {
it('should create the sessions in one batched call carrying the provided username', async () => {
usersService.findByUsername.mockResolvedValueOnce({ groups: [{ id: 'group-1' }], username: 'validuser' } as any);

await instrumentRecordsService.upload({ ...baseUploadData, groupId: 'group-1', username: 'validuser' });

expect(usersService.findByUsername).toHaveBeenCalledWith('validuser', undefined);
expect(sessionsService.create).toHaveBeenCalledWith(expect.objectContaining({ username: 'validuser' }));
expect(sessionsService.create).not.toHaveBeenCalled();
expect(sessionsService.createMany).toHaveBeenCalledTimes(1);
expect(sessionsService.createMany).toHaveBeenCalledWith(
expect.objectContaining({ groupId: 'group-1', type: 'RETROSPECTIVE', username: 'validuser' })
);
});

it('should batch every record into a single session creation call', async () => {
const records = Array.from({ length: 25 }, (_, i) => ({
data: { answer: i },
date: new Date(),
subjectId: `subject-${i}`
}));
sessionsService.createMany.mockResolvedValueOnce(
records.map((_, i) => ({ ...mockSession, id: `session-${i}` })) as any
);

await instrumentRecordsService.upload({ ...baseUploadData, records });

expect(sessionsService.createMany).toHaveBeenCalledTimes(1);
const [call] = sessionsService.createMany.mock.lastCall as [{ entries: unknown[] }];
expect(call.entries).toHaveLength(25);
});

it('should pair each record with the session created for it, by position', async () => {
const records = [
{ data: { answer: 1 }, date: new Date(), subjectId: 'subject-a' },
{ data: { answer: 2 }, date: new Date(), subjectId: 'subject-b' }
];
sessionsService.createMany.mockResolvedValueOnce([
{ ...mockSession, id: 'session-a' },
{ ...mockSession, id: 'session-b' }
] as any);

await instrumentRecordsService.upload({ ...baseUploadData, records });

expect(instrumentRecordModel.createMany.mock.lastCall?.[0]).toMatchObject({
data: [
{ sessionId: 'session-a', subjectId: 'subject-a' },
{ sessionId: 'session-b', subjectId: 'subject-b' }
]
});
});

it('should throw a ForbiddenException when a non-admin user uploads without a group', async () => {
Expand All @@ -209,7 +250,7 @@ describe('InstrumentRecordsService', () => {
instrumentRecordsService.upload({ ...baseUploadData, username: 'validuser' })
).rejects.toBeInstanceOf(ForbiddenException);

expect(sessionsService.create).not.toHaveBeenCalled();
expect(sessionsService.createMany).not.toHaveBeenCalled();
});

it('should throw a ForbiddenException when a user uploads to a group they are not a member of', async () => {
Expand All @@ -222,7 +263,7 @@ describe('InstrumentRecordsService', () => {
instrumentRecordsService.upload({ ...baseUploadData, groupId: 'group-1', username: 'validuser' })
).rejects.toBeInstanceOf(ForbiddenException);

expect(sessionsService.create).not.toHaveBeenCalled();
expect(sessionsService.createMany).not.toHaveBeenCalled();
});

it('should reject and not create any sessions when an unknown username is provided', async () => {
Expand All @@ -234,18 +275,89 @@ describe('InstrumentRecordsService', () => {
NotFoundException
);

expect(sessionsService.create).not.toHaveBeenCalled();
expect(sessionsService.createMany).not.toHaveBeenCalled();
});

it('should call sessionsService.create with username undefined when no username is provided', async () => {
it('should create the sessions with username undefined when no username is provided', async () => {
await instrumentRecordsService.upload({ ...baseUploadData });

expect(usersService.findByUsername).not.toHaveBeenCalled();
expect(sessionsService.create).toHaveBeenCalledWith(expect.objectContaining({ username: undefined }));
expect(sessionsService.createMany).toHaveBeenCalledWith(expect.objectContaining({ username: undefined }));
});

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'] } }
});
});

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 })).rejects.toBeInstanceOf(
UnprocessableEntityException
);

expect(sessionsService.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 the sessions when the record insert fails, so none is left without records', async () => {
instrumentRecordModel.createMany.mockRejectedValueOnce(new Error('insert failed'));

await expect(instrumentRecordsService.upload({ ...baseUploadData })).rejects.toThrow('insert failed');

expect(sessionsService.deleteByIds).toHaveBeenCalledWith(['session-1']);
});

it('should keep the sessions when only the read-back fails, since the records already reference them', async () => {
instrumentRecordModel.findMany.mockRejectedValueOnce(new Error('read-back failed'));

await expect(instrumentRecordsService.upload({ ...baseUploadData })).rejects.toThrow('read-back failed');

expect(sessionsService.deleteByIds).not.toHaveBeenCalled();
});

// The bulk payload cannot carry a file and this path never attaches one, so such a record could
// only ever be incomplete. Refusing it is the same call `create` makes for series instruments.
it('should reject a file instrument rather than write a record its files can never reach', async () => {
instrumentsService.findById.mockResolvedValue({ ...mockInstrument, kind: 'FILE' } as any);

await expect(instrumentRecordsService.upload({ ...baseUploadData })).rejects.toBeInstanceOf(
UnprocessableEntityException
);

expect(sessionsService.createMany).not.toHaveBeenCalled();
expect(instrumentRecordModel.createMany).not.toHaveBeenCalled();
});

// `pending` is intentionally not written on create; the find-side OR filter treats missing and
// false `pending` alike (see the 'find' describe block), so records stay query-visible without it.
it('should create records via createMany with the processed record data', async () => {
await instrumentRecordsService.upload({ ...baseUploadData });

Expand Down
113 changes: 55 additions & 58 deletions apps/api/src/instrument-records/instrument-records.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import type {
UploadInstrumentRecordsData
} from '@opendatacapture/schemas/instrument-records';
import { Prisma } from '@prisma/client';
import type { InstrumentRecord as PrismaInstrumentRecord, Session } from '@prisma/client';
import type { InstrumentRecord as PrismaInstrumentRecord } from '@prisma/client';
import { isNumber, mergeWith, pickBy } from 'lodash-es';

import { accessibleQuery } from '@/auth/ability.utils';
Expand All @@ -34,7 +34,6 @@ import { GroupsService } from '@/groups/groups.service';
import { InstrumentsService } from '@/instruments/instruments.service';
import { SessionsService } from '@/sessions/sessions.service';
import { StorageService } from '@/storage/storage.service';
import { CreateSubjectDto } from '@/subjects/dto/create-subject.dto';
import { SubjectsService } from '@/subjects/subjects.service';
import { UsersService } from '@/users/users.service';

Expand Down Expand Up @@ -398,6 +397,14 @@ export class InstrumentRecordsService {
`Cannot create instrument record for series instrument '${instrument.id}'`
);
}
// A bulk-uploaded file record could never be completed: the payload schema has nowhere to carry a
// file, this path never attaches one, and the response ids the client would need to attach one
// afterwards are discarded. Refuse it rather than write a record that can only ever be pending.
if (instrument.kind === 'FILE') {
throw new UnprocessableEntityException(
`Cannot create instrument record for file instrument '${instrument.id}': files cannot be attached to a bulk upload`
);
}

if (username) {
const user = await this.usersService.findByUsername(username, options);
Expand All @@ -409,70 +416,60 @@ export class InstrumentRecordsService {
}
}

const createdSessionsArray: Session[] = [];

try {
const subjectIdList = records.map(({ subjectId }) => {
const subjectToAdd: CreateSubjectDto = { id: subjectId };

return subjectToAdd;
});

await this.subjectsService.createMany(subjectIdList);

const preProcessedRecords = await Promise.all(
records.map(async (record) => {
const { data: rawData, date, subjectId } = record;
// 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 };
});

// 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}'`
);
}
// One batched call rather than one session creation per record, which cost several queries each.
// Returned in input order, so each record can be paired with its session by index.
const sessions = await this.sessionsService.createMany({
entries: validatedRecords.map((record) => ({
date: record.date,
subjectData: { id: record.subjectId }
})),
groupId: groupId ?? null,
type: 'RETROSPECTIVE',
username: username ?? undefined
});

const session = await this.sessionsService.create({
date: date,
groupId: groupId ?? null,
subjectData: { id: subjectId },
type: 'RETROSPECTIVE',
username: username ?? undefined
});

createdSessionsArray.push(session);

const computedMeasures = instrument.measures
? this.instrumentMeasuresService.computeMeasures(instrument.measures, parseResult.data)
: null;

return {
computedMeasures,
data: this.serializeData(parseResult.data),
date,
groupId,
instrumentId,
pending: false,
sessionId: session.id,
subjectId
};
})
);
// Only the insert is rolled back on failure. Deleting the sessions after it has succeeded would
// strand the records that now reference them, so the read-back below sits outside the catch.
try {
await this.instrumentRecordModel.createMany({
data: preProcessedRecords
});

return this.instrumentRecordModel.findMany({
where: {
data: validatedRecords.map((record, index) => ({
computedMeasures: instrument.measures
? this.instrumentMeasuresService.computeMeasures(instrument.measures, record.data)
: null,
data: this.serializeData(record.data),
date: record.date,
groupId,
instrumentId
}
instrumentId,
pending: false,
sessionId: sessions[index]!.id,
subjectId: record.subjectId
}))
});
} catch (err) {
await this.sessionsService.deleteByIds(createdSessionsArray.map((session) => session.id));
await this.sessionsService.deleteByIds(sessions.map((session) => session.id));
throw err;
}

return this.instrumentRecordModel.findMany({
where: {
sessionId: { in: sessions.map((session) => session.id) }
}
});
}

private getInstrumentById(instrumentId: string) {
Expand Down
Loading
Loading