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

Merged
joshunrau merged 9 commits into from
Sep 24, 2026
Merged

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
9 commits
Select commit Hold shift + click to select a range
24e3775
perf(api): batch session creation and scope the upload response
gdevenyi Jul 20, 2026
4082350
refactor(api): resolve batch subjects through the subjects service
gdevenyi Jul 28, 2026
9a40e2c
fix(api): drop the subject read-back and fail loudly on a session rea…
gdevenyi Jul 28, 2026
fa1f4cb
test(api): pin the demographics createMany keeps for a new subject
gdevenyi Sep 11, 2026
fb39d26
test(testing): read back the demographics a first session creates
gdevenyi Sep 11, 2026
371c283
test(testing): refuse a file instrument upload end to end
gdevenyi Sep 11, 2026
aeb0f0f
fix(api): delete the sessions a batch inserted when their read-back f…
gdevenyi Sep 23, 2026
d6966cf
fix(api): scope the group and user a new session names to the caller
gdevenyi Sep 23, 2026
6720e70
test(testing): refuse a session in a group the caller does not belong to
gdevenyi Sep 23, 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
137 changes: 127 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,7 +184,7 @@ 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);
Expand All @@ -208,13 +208,63 @@ describe('InstrumentRecordsService', () => {
expect(result).toStrictEqual([foreignRecord]);
});

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

it('should create the sessions with the caller ability, so their group and user lookups are scoped', async () => {
const ability = createAppAbility([{ action: 'create', subject: 'InstrumentRecord' }]);

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

expect(sessionsService.createMany).toHaveBeenCalledWith(expect.anything(), { ability });
});

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 @@ -227,7 +277,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 @@ -240,7 +290,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 @@ -252,18 +302,85 @@ 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 }),
undefined
);
});

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
121 changes: 62 additions & 59 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`
);
Comment thread
gdevenyi marked this conversation as resolved.
}

if (username) {
const user = await this.usersService.findByUsername(username, options);
Expand All @@ -409,70 +416,66 @@ 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
},
options
);

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

// Only the rows this call wrote. Filtering by instrument alone returned every group's records
// when `groupId` was omitted, and `accessibleQuery` cannot scope this read: a STANDARD uploader
// holds no `read InstrumentRecord` rule, which makes CASL throw rather than filter.
return this.instrumentRecordModel.findMany({
where: { sessionId: { in: createdSessionsArray.map((session) => session.id) } }
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,
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;
}

// Only the rows this call wrote. Filtering by instrument alone returned every group's records
// when `groupId` was omitted, and `accessibleQuery` cannot scope this read: a STANDARD uploader
// holds no `read InstrumentRecord` rule, which makes CASL throw rather than filter.
return this.instrumentRecordModel.findMany({
where: {
sessionId: { in: sessions.map((session) => session.id) }
}
});
}

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