Files
stack/apps/api/src/projects/projects.service.ts
Jason Woltje ef25167c24 fix(#196): fix race condition in job status updates
Implemented optimistic locking with version field and SELECT FOR UPDATE
transactions to prevent data corruption from concurrent job status updates.

Changes:
- Added version field to RunnerJob schema for optimistic locking
- Created migration 20260202_add_runner_job_version_for_concurrency
- Implemented ConcurrentUpdateException for conflict detection
- Updated RunnerJobsService methods with optimistic locking:
  * updateStatus() - with version checking and retry logic
  * updateProgress() - with version checking and retry logic
  * cancel() - with version checking and retry logic
- Updated CoordinatorIntegrationService with SELECT FOR UPDATE:
  * updateJobStatus() - transaction with row locking
  * completeJob() - transaction with row locking
  * failJob() - transaction with row locking
  * updateJobProgress() - optimistic locking
- Added retry mechanism (3 attempts) with exponential backoff
- Added comprehensive concurrency tests (10 tests, all passing)
- Updated existing test mocks to support updateMany

Test Results:
- All 10 concurrency tests passing ✓
- Tests cover concurrent status updates, progress updates, completions,
  cancellations, retry logic, and exponential backoff

This fix prevents race conditions that could cause:
- Lost job results (double completion)
- Lost progress updates
- Invalid status transitions
- Data corruption under concurrent access

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-02 12:51:17 -06:00

278 lines
7.0 KiB
TypeScript

import { Injectable, NotFoundException } from "@nestjs/common";
import { Prisma, Project } from "@prisma/client";
import { PrismaService } from "../prisma/prisma.service";
import { ActivityService } from "../activity/activity.service";
import { ProjectStatus } from "@prisma/client";
import type { CreateProjectDto, UpdateProjectDto, QueryProjectsDto } from "./dto";
type ProjectWithRelations = Project & {
creator: { id: string; name: string; email: string };
_count: { tasks: number; events: number };
};
type ProjectWithDetails = Project & {
creator: { id: string; name: string; email: string };
tasks: {
id: string;
title: string;
status: string;
priority: string;
dueDate: Date | null;
}[];
events: {
id: string;
title: string;
startTime: Date;
endTime: Date | null;
}[];
_count: { tasks: number; events: number };
};
/**
* Service for managing projects
*/
@Injectable()
export class ProjectsService {
constructor(
private readonly prisma: PrismaService,
private readonly activityService: ActivityService
) {}
/**
* Create a new project
*/
async create(
workspaceId: string,
userId: string,
createProjectDto: CreateProjectDto
): Promise<ProjectWithRelations> {
const data: Prisma.ProjectCreateInput = {
name: createProjectDto.name,
description: createProjectDto.description ?? null,
color: createProjectDto.color ?? null,
startDate: createProjectDto.startDate ?? null,
endDate: createProjectDto.endDate ?? null,
workspace: { connect: { id: workspaceId } },
creator: { connect: { id: userId } },
status: createProjectDto.status ?? ProjectStatus.PLANNING,
metadata: createProjectDto.metadata
? (createProjectDto.metadata as unknown as Prisma.InputJsonValue)
: {},
};
const project = await this.prisma.project.create({
data,
include: {
creator: {
select: { id: true, name: true, email: true },
},
_count: {
select: { tasks: true, events: true },
},
},
});
// Log activity
await this.activityService.logProjectCreated(workspaceId, userId, project.id, {
name: project.name,
});
return project;
}
/**
* Get paginated projects with filters
*/
async findAll(query: QueryProjectsDto): Promise<{
data: ProjectWithRelations[];
meta: {
total: number;
page: number;
limit: number;
totalPages: number;
};
}> {
const page = query.page ?? 1;
const limit = query.limit ?? 50;
const skip = (page - 1) * limit;
// Build where clause
const where: Prisma.ProjectWhereInput = query.workspaceId
? {
workspaceId: query.workspaceId,
}
: {};
if (query.status) {
where.status = query.status;
}
if (query.startDateFrom || query.startDateTo) {
where.startDate = {};
if (query.startDateFrom) {
where.startDate.gte = query.startDateFrom;
}
if (query.startDateTo) {
where.startDate.lte = query.startDateTo;
}
}
// Execute queries in parallel
const [data, total] = await Promise.all([
this.prisma.project.findMany({
where,
include: {
creator: {
select: { id: true, name: true, email: true },
},
_count: {
select: { tasks: true, events: true },
},
},
orderBy: {
createdAt: "desc",
},
skip,
take: limit,
}),
this.prisma.project.count({ where }),
]);
return {
data,
meta: {
total,
page,
limit,
totalPages: Math.ceil(total / limit),
},
};
}
/**
* Get a single project by ID
*/
async findOne(id: string, workspaceId: string): Promise<ProjectWithDetails> {
const project = await this.prisma.project.findUnique({
where: {
id,
workspaceId,
},
include: {
creator: {
select: { id: true, name: true, email: true },
},
tasks: {
select: {
id: true,
title: true,
status: true,
priority: true,
dueDate: true,
},
orderBy: { sortOrder: "asc" },
},
events: {
select: {
id: true,
title: true,
startTime: true,
endTime: true,
},
orderBy: { startTime: "asc" },
},
_count: {
select: { tasks: true, events: true },
},
},
});
if (!project) {
throw new NotFoundException(`Project with ID ${id} not found`);
}
return project;
}
/**
* Update a project
*/
async update(
id: string,
workspaceId: string,
userId: string,
updateProjectDto: UpdateProjectDto
): Promise<ProjectWithRelations> {
// Verify project exists
const existingProject = await this.prisma.project.findUnique({
where: { id, workspaceId },
});
if (!existingProject) {
throw new NotFoundException(`Project with ID ${id} not found`);
}
// Build update data, only including defined fields
const updateData: Prisma.ProjectUpdateInput = {};
if (updateProjectDto.name !== undefined) updateData.name = updateProjectDto.name;
if (updateProjectDto.description !== undefined)
updateData.description = updateProjectDto.description;
if (updateProjectDto.status !== undefined) updateData.status = updateProjectDto.status;
if (updateProjectDto.startDate !== undefined) updateData.startDate = updateProjectDto.startDate;
if (updateProjectDto.endDate !== undefined) updateData.endDate = updateProjectDto.endDate;
if (updateProjectDto.color !== undefined) updateData.color = updateProjectDto.color;
if (updateProjectDto.metadata !== undefined) {
updateData.metadata = updateProjectDto.metadata as unknown as Prisma.InputJsonValue;
}
const project = await this.prisma.project.update({
where: {
id,
workspaceId,
},
data: updateData,
include: {
creator: {
select: { id: true, name: true, email: true },
},
_count: {
select: { tasks: true, events: true },
},
},
});
// Log activity
await this.activityService.logProjectUpdated(workspaceId, userId, id, {
changes: updateProjectDto as Prisma.JsonValue,
});
return project;
}
/**
* Delete a project
*/
async remove(id: string, workspaceId: string, userId: string): Promise<void> {
// Verify project exists
const project = await this.prisma.project.findUnique({
where: { id, workspaceId },
});
if (!project) {
throw new NotFoundException(`Project with ID ${id} not found`);
}
await this.prisma.project.delete({
where: {
id,
workspaceId,
},
});
// Log activity
await this.activityService.logProjectDeleted(workspaceId, userId, id, {
name: project.name,
});
}
}