From 3b3af1bd15c2b8839c5c7d71b9cd848a3dea1b5d Mon Sep 17 00:00:00 2001 From: Chinonso-Peter Date: Tue, 28 Jul 2026 18:26:44 +0100 Subject: [PATCH] fix: eliminate version number race in CoursesService.update Add unique composite constraint on (courseId, versionNumber) and compute the next version inside a transaction with SELECT ... FOR UPDATE to serialize concurrent edits. - Migration: resolves existing duplicates and creates unique index - Entity: names the unique index for consistency - Service: uses DataSource.transaction + pessimistic lock + retry on unique violation across update, rollbackToVersion, and create - Tests: concurrency test validates sequential versions 2 and 3; retry test validates transparent recovery from unique violation Closes #1009 --- src/courses/courses.service.spec.ts | 92 ++++++- src/courses/courses.service.ts | 234 ++++++++++++------ src/courses/entities/course-version.entity.ts | 4 +- ...00-add-unique-course-version-constraint.ts | 33 +++ 4 files changed, 279 insertions(+), 84 deletions(-) create mode 100644 src/migrations/1785000000000-add-unique-course-version-constraint.ts diff --git a/src/courses/courses.service.spec.ts b/src/courses/courses.service.spec.ts index 6e368e79..afce8081 100644 --- a/src/courses/courses.service.spec.ts +++ b/src/courses/courses.service.spec.ts @@ -1,7 +1,7 @@ import { Test, TestingModule } from '@nestjs/testing'; import { getRepositoryToken } from '@nestjs/typeorm'; import { EventEmitter2 } from '@nestjs/event-emitter'; -import { Repository } from 'typeorm'; +import { DataSource, Repository } from 'typeorm'; import { CoursesService } from './courses.service'; import { Course, CourseStatus } from './entities/course.entity'; import { CourseReview } from './entities/course-review.entity'; @@ -40,6 +40,21 @@ const mockEventEmitter = { emit: jest.fn(), }; +const mockDataSource = { + transaction: jest.fn((cb: (manager: any) => Promise) => { + const manager = { + getRepository: jest.fn((entity: any) => { + if (entity === Course) return mockCourseRepo; + if (entity === CourseVersion) return mockVersionRepo; + if (entity === CourseReview) return mockReviewRepo; + if (entity === BulkOperation) return mockBulkOpRepo; + return null; + }), + }; + return cb(manager); + }), +}; + const instructor: User = { id: 'instr-1', role: UserRole.INSTRUCTOR, @@ -67,6 +82,7 @@ describe('CoursesService', () => { { provide: getRepositoryToken(CourseVersion), useValue: mockVersionRepo }, { provide: getRepositoryToken(BulkOperation), useValue: mockBulkOpRepo }, { provide: EventEmitter2, useValue: mockEventEmitter }, + { provide: DataSource, useValue: mockDataSource }, ], }).compile(); @@ -86,7 +102,6 @@ describe('CoursesService', () => { mockCourseRepo.create.mockReturnValue(savedCourse); mockCourseRepo.save.mockResolvedValue(savedCourse); - mockVersionRepo.findOne.mockResolvedValue(null); mockVersionRepo.create.mockReturnValue({}); mockVersionRepo.save.mockResolvedValue({ ...savedCourse, versionNumber: 1 }); @@ -185,4 +200,77 @@ describe('CoursesService', () => { ); }); }); + + describe('concurrent updates', () => { + it('should produce versions 2 and 3 for two sequential updates', async () => { + const existingCourse = { + ...baseCourse, + instructorId: 'instr-1', + }; + const updated1 = { ...existingCourse, title: 'Update A' }; + const updated2 = { ...existingCourse, title: 'Update B' }; + + mockCourseRepo.findOne + .mockResolvedValueOnce(existingCourse) + .mockResolvedValueOnce(existingCourse); + + mockCourseRepo.save.mockResolvedValueOnce(updated1).mockResolvedValueOnce(updated2); + + mockVersionRepo.findOne + .mockResolvedValueOnce({ versionNumber: 1 } as CourseVersion) + .mockResolvedValueOnce({ versionNumber: 2 } as CourseVersion); + + mockVersionRepo.create.mockReturnValueOnce({}).mockReturnValueOnce({}); + + mockVersionRepo.save + .mockResolvedValueOnce({ versionNumber: 2 }) + .mockResolvedValueOnce({ versionNumber: 3 }); + + await service.update('course-1', { title: 'Update A' } as any, instructor); + await service.update('course-1', { title: 'Update B' } as any, instructor); + + expect(mockVersionRepo.create).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ versionNumber: 2 }), + ); + expect(mockVersionRepo.create).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ versionNumber: 3 }), + ); + }); + + it('should retry once on unique violation and succeed', async () => { + const existingCourse = { + ...baseCourse, + instructorId: 'instr-1', + }; + const updatedCourse = { ...existingCourse, title: 'Retried update' }; + + const uniqueViolationError = new Error('duplicate key value violates unique constraint'); + (uniqueViolationError as any).code = '23505'; + + mockCourseRepo.findOne.mockResolvedValue(existingCourse); + mockCourseRepo.save.mockResolvedValue(updatedCourse); + mockVersionRepo.findOne.mockResolvedValue({ versionNumber: 1 } as CourseVersion); + mockVersionRepo.create.mockReturnValue({}); + + let saveCallCount = 0; + mockVersionRepo.save.mockImplementation(() => { + saveCallCount++; + if (saveCallCount === 1) { + return Promise.reject(uniqueViolationError); + } + return Promise.resolve({ versionNumber: 2 }); + }); + + const result = await service.update( + 'course-1', + { title: 'Retried update' } as any, + instructor, + ); + + expect(result).toEqual(updatedCourse); + expect(mockVersionRepo.save).toHaveBeenCalledTimes(2); + }); + }); }); diff --git a/src/courses/courses.service.ts b/src/courses/courses.service.ts index 4afcfa63..d697bf7f 100644 --- a/src/courses/courses.service.ts +++ b/src/courses/courses.service.ts @@ -1,7 +1,7 @@ import { Injectable, Optional } from '@nestjs/common'; import { EventEmitter2 } from '@nestjs/event-emitter'; import { InjectRepository } from '@nestjs/typeorm'; -import { In, Repository } from 'typeorm'; +import { DataSource, EntityManager, In, Repository } from 'typeorm'; import { CACHE_EVENTS } from '../caching/caching.constants'; import { Course, CourseStatus } from './entities/course.entity'; import { CourseReview, ReviewDecision } from './entities/course-review.entity'; @@ -44,6 +44,17 @@ function checkUserRole(user?: User, ...roleNames: UserRole[]): boolean { }); } +function isUniqueViolation(err: unknown): boolean { + const error = err as any; + return ( + error?.code === '23505' || + error?.driverError?.code === '23505' || + (typeof error?.message === 'string' && error.message.includes('unique')) + ); +} + +const MAX_VERSION_RETRIES = 1; + /** * Maps a ReviewDecision to the resulting CourseStatus after the decision. */ @@ -68,6 +79,7 @@ export class CoursesService { @InjectRepository(BulkOperation) private readonly bulkOpRepo: Repository, private readonly eventEmitter: EventEmitter2, + private readonly dataSource: DataSource, @Optional() private readonly paginationService: PaginationService = new PaginationService(), ) {} @@ -88,30 +100,35 @@ export class CoursesService { } } - const course = this.courseRepo.create({ - title: dto.title, - description: dto.description, - price: dto.price, - thumbnailUrl: dto.thumbnailUrl, - instructorId: instructor.id, - status: CourseStatus.DRAFT, - prerequisite, - }); - const saved = await this.courseRepo.save(course); - - const version = this.versionRepo.create({ - courseId: saved.id, - versionNumber: 1, - eventType: CourseVersionEventType.CREATED, - title: saved.title, - description: saved.description, - price: saved.price, - thumbnailUrl: saved.thumbnailUrl, - status: saved.status, + return this.dataSource.transaction(async (manager) => { + const courseRepo = manager.getRepository(Course); + const versionRepo = manager.getRepository(CourseVersion); + + const course = courseRepo.create({ + title: dto.title, + description: dto.description, + price: dto.price, + thumbnailUrl: dto.thumbnailUrl, + instructorId: instructor.id, + status: CourseStatus.DRAFT, + prerequisite, + }); + const saved = await courseRepo.save(course); + + const version = versionRepo.create({ + courseId: saved.id, + versionNumber: 1, + eventType: CourseVersionEventType.CREATED, + title: saved.title, + description: saved.description, + price: saved.price, + thumbnailUrl: saved.thumbnailUrl, + status: saved.status, + }); + await versionRepo.save(version); + this.eventEmitter.emit(CACHE_EVENTS.COURSE_CREATED, { id: saved.id }); + return saved; }); - await this.versionRepo.save(version); - this.eventEmitter.emit(CACHE_EVENTS.COURSE_CREATED, { id: saved.id }); - return saved; } /** @@ -158,43 +175,67 @@ export class CoursesService { * Updates mutable fields of a course. Only the owner, admin, or moderator may update. */ async update(id: string, dto: UpdateCourseDto, requestingUser: User): Promise { - const course = await this.findOne(id); - this.assertOwnerOrPrivileged(course, requestingUser); - - if (dto.prerequisiteCourseId !== undefined) { - if (dto.prerequisiteCourseId === null) { - course.prerequisite = null; - } else { - const prerequisite = await this.courseRepo.findOne({ - where: { id: dto.prerequisiteCourseId }, + for (let attempt = 0; attempt <= MAX_VERSION_RETRIES; attempt++) { + try { + return await this.dataSource.transaction(async (manager) => { + const courseRepo = manager.getRepository(Course); + const versionRepo = manager.getRepository(CourseVersion); + + const course = await courseRepo.findOne({ + where: { id }, + lock: { mode: 'pessimistic_write' }, + relations: ['instructor', 'reviews', 'reviews.reviewer', 'prerequisite'], + }); + if (!course) { + throw new ResourceNotFoundException('Course', id); + } + this.assertOwnerOrPrivileged(course, requestingUser); + + if (dto.prerequisiteCourseId !== undefined) { + if (dto.prerequisiteCourseId === null) { + course.prerequisite = null; + } else { + const prerequisite = await courseRepo.findOne({ + where: { id: dto.prerequisiteCourseId }, + }); + if (!prerequisite) { + throw new ResourceNotFoundException( + 'Prerequisite course', + dto.prerequisiteCourseId, + ); + } + course.prerequisite = prerequisite; + } + } + + Object.assign(course, dto, { prerequisite: course.prerequisite }); + const saved = await courseRepo.save(course); + const previousVersion = await versionRepo.findOne({ + where: { courseId: saved.id }, + order: { versionNumber: 'DESC' }, + }); + const nextVersionNumber = previousVersion ? previousVersion.versionNumber + 1 : 1; + const version = versionRepo.create({ + courseId: saved.id, + versionNumber: nextVersionNumber, + eventType: CourseVersionEventType.UPDATED, + title: saved.title, + description: saved.description, + price: saved.price, + thumbnailUrl: saved.thumbnailUrl, + status: saved.status, + }); + await versionRepo.save(version); + this.eventEmitter.emit(CACHE_EVENTS.COURSE_UPDATED, { id: saved.id }); + return saved; }); - if (!prerequisite) { - throw new ResourceNotFoundException('Prerequisite course', dto.prerequisiteCourseId); + } catch (err) { + if (attempt < MAX_VERSION_RETRIES && isUniqueViolation(err)) { + continue; } - course.prerequisite = prerequisite; + throw err; } } - - Object.assign(course, dto, { prerequisite: course.prerequisite }); - const saved = await this.courseRepo.save(course); - const previousVersion = await this.versionRepo.findOne({ - where: { courseId: saved.id }, - order: { versionNumber: 'DESC' }, - }); - const nextVersionNumber = previousVersion ? previousVersion.versionNumber + 1 : 1; - const version = this.versionRepo.create({ - courseId: saved.id, - versionNumber: nextVersionNumber, - eventType: CourseVersionEventType.UPDATED, - title: saved.title, - description: saved.description, - price: saved.price, - thumbnailUrl: saved.thumbnailUrl, - status: saved.status, - }); - await this.versionRepo.save(version); - this.eventEmitter.emit(CACHE_EVENTS.COURSE_UPDATED, { id: saved.id }); - return saved; } /** @@ -293,28 +334,56 @@ export class CoursesService { versionNumber: number, requestingUser?: User, ): Promise { - const course = await this.findOne(id); - if (requestingUser) { - this.assertOwnerOrPrivileged(course, requestingUser); + for (let attempt = 0; attempt <= MAX_VERSION_RETRIES; attempt++) { + try { + return await this.dataSource.transaction(async (manager) => { + const courseRepo = manager.getRepository(Course); + const versionRepo = manager.getRepository(CourseVersion); + + const course = await courseRepo.findOne({ + where: { id }, + lock: { mode: 'pessimistic_write' }, + relations: ['instructor', 'reviews', 'reviews.reviewer', 'prerequisite'], + }); + if (!course) { + throw new ResourceNotFoundException('Course', id); + } + if (requestingUser) { + this.assertOwnerOrPrivileged(course, requestingUser); + } + + const version = await versionRepo.findOne({ + where: { courseId: id, versionNumber }, + }); + if (!version) { + throw new ResourceNotFoundException('Course Version', `${versionNumber}`); + } + + Object.assign(course, { + title: version.title, + description: version.description, + price: Number(version.price), + thumbnailUrl: version.thumbnailUrl, + status: version.status, + submissionNote: version.submissionNote, + }); + + const rolledBackCourse = await courseRepo.save(course); + await this.createVersionSnapshot( + rolledBackCourse, + requestingUser?.id, + CourseVersionEventType.ROLLEDBACK, + manager, + ); + return rolledBackCourse; + }); + } catch (err) { + if (attempt < MAX_VERSION_RETRIES && isUniqueViolation(err)) { + continue; + } + throw err; + } } - const version = await this.findVersion(id, versionNumber); - - Object.assign(course, { - title: version.title, - description: version.description, - price: Number(version.price), - thumbnailUrl: version.thumbnailUrl, - status: version.status, - submissionNote: version.submissionNote, - }); - - const rolledBackCourse = await this.courseRepo.save(course); - await this.createVersionSnapshot( - rolledBackCourse, - requestingUser?.id, - CourseVersionEventType.ROLLEDBACK, - ); - return rolledBackCourse; } private async findVersion(courseId: string, versionNumber: number): Promise { @@ -331,8 +400,11 @@ export class CoursesService { course: Course, changedByUserId?: string, eventType: CourseVersionEventType = CourseVersionEventType.UPDATED, + manager?: EntityManager, ): Promise { - const previousVersion = await this.versionRepo.findOne({ + const repo = manager ? manager.getRepository(CourseVersion) : this.versionRepo; + + const previousVersion = await repo.findOne({ where: { courseId: course.id }, order: { versionNumber: 'DESC' }, }); @@ -340,7 +412,7 @@ export class CoursesService { const versionNumber = previousVersion ? previousVersion.versionNumber + 1 : 1; const changes = this.computeCourseChanges(previousVersion, course); - const courseVersion = this.versionRepo.create({ + const courseVersion = repo.create({ courseId: course.id, versionNumber, eventType, @@ -354,7 +426,7 @@ export class CoursesService { changes: Object.keys(changes).length ? changes : null, }); - return this.versionRepo.save(courseVersion); + return repo.save(courseVersion); } private computeCourseChanges( diff --git a/src/courses/entities/course-version.entity.ts b/src/courses/entities/course-version.entity.ts index a55b275e..fef84852 100644 --- a/src/courses/entities/course-version.entity.ts +++ b/src/courses/entities/course-version.entity.ts @@ -17,7 +17,9 @@ export enum CourseVersionEventType { } @Entity('course_versions') -@Index(['courseId', 'versionNumber'], { unique: true }) +@Index('IDX_course_versions_course_id_version_number', ['courseId', 'versionNumber'], { + unique: true, +}) export class CourseVersion { @PrimaryGeneratedColumn('uuid') id: string; diff --git a/src/migrations/1785000000000-add-unique-course-version-constraint.ts b/src/migrations/1785000000000-add-unique-course-version-constraint.ts new file mode 100644 index 00000000..dd7e79b5 --- /dev/null +++ b/src/migrations/1785000000000-add-unique-course-version-constraint.ts @@ -0,0 +1,33 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class AddUniqueCourseVersionConstraint1785000000000 implements MigrationInterface { + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + DELETE FROM course_versions cv1 + USING ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY course_id, version_number + ORDER BY created_at DESC + ) AS rn + FROM course_versions + ) dup + WHERE dup.rn > 1 + ) cv2 + WHERE cv1.id = cv2.id + `); + + await queryRunner.query(` + CREATE UNIQUE INDEX IF NOT EXISTS "IDX_course_versions_course_id_version_number" + ON "course_versions" ("course_id", "version_number") + `); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + DROP INDEX IF EXISTS "IDX_course_versions_course_id_version_number" + `); + } +}