diff --git a/src/deploy-runs/deploy-execution.service.ts b/src/deploy-runs/deploy-execution.service.ts index 4549c21..d4a2b39 100644 --- a/src/deploy-runs/deploy-execution.service.ts +++ b/src/deploy-runs/deploy-execution.service.ts @@ -38,7 +38,11 @@ export class DeployExecutionService { let currentRun = this.requireRun( run.id, - await this.deployRunRepository.updateRun(run.id, { status: 'running' }), + await this.deployRunRepository.updateRun( + run.id, + { status: 'running' }, + run.version, + ), ); let currentStep: string | undefined; @@ -58,7 +62,7 @@ export class DeployExecutionService { try { currentStep = 'gitea-check'; currentRun = await this.completeStep( - currentRun.id, + currentRun, currentStep, 'success', ); @@ -74,7 +78,7 @@ export class DeployExecutionService { currentStep = 'health-check'; currentRun = await this.completeStep( - currentRun.id, + currentRun, currentStep, 'success', ); @@ -84,15 +88,19 @@ export class DeployExecutionService { currentStep = 'finish-release'; currentRun = await this.completeStep( - currentRun.id, + currentRun, currentStep, 'success', ); currentRun = this.requireRun( currentRun.id, - await this.deployRunRepository.updateRun(currentRun.id, { - status: 'success', - }), + await this.deployRunRepository.updateRun( + currentRun.id, + { + status: 'success', + }, + currentRun.version, + ), ); await this.auditService.record({ @@ -121,14 +129,19 @@ export class DeployExecutionService { message: undefined, errorSummary: failureSummary, }, + currentRun.version, )) ?? currentRun; } currentRun = - (await this.deployRunRepository.updateRun(currentRun.id, { - status: 'failed', - failureSummary, - })) ?? currentRun; + (await this.deployRunRepository.updateRun( + currentRun.id, + { + status: 'failed', + failureSummary, + }, + currentRun.version, + )) ?? currentRun; currentRun = await this.advanceNotification(currentRun, 'failed'); @@ -172,26 +185,34 @@ export class DeployExecutionService { if (!queueItem.buildNumber) { return this.requireRun( currentRun.id, - await this.deployRunRepository.updateRun(currentRun.id, { - status: 'queued', - jenkinsQueueId: queueItem.queueId, - jenkinsBuildUrl: queueItem.executableUrl, - }), + await this.deployRunRepository.updateRun( + currentRun.id, + { + status: 'queued', + jenkinsQueueId: queueItem.queueId, + jenkinsBuildUrl: queueItem.executableUrl, + }, + currentRun.version, + ), ); } buildNumber = queueItem.buildNumber; currentRun = this.requireRun( currentRun.id, - await this.deployRunRepository.updateRun(currentRun.id, { - status: 'running', - jenkinsQueueId: queueItem.queueId, - jenkinsBuildNumber: queueItem.buildNumber, - jenkinsBuildUrl: queueItem.executableUrl, - }), + await this.deployRunRepository.updateRun( + currentRun.id, + { + status: 'running', + jenkinsQueueId: queueItem.queueId, + jenkinsBuildNumber: queueItem.buildNumber, + jenkinsBuildUrl: queueItem.executableUrl, + }, + currentRun.version, + ), ); currentRun = await this.markStep( - currentRun.id, + currentRun, 'jenkins-build', 'running', `Jenkins 构建 #${queueItem.buildNumber} 正在运行。`, @@ -221,19 +242,23 @@ export class DeployExecutionService { currentRun = this.requireRun( currentRun.id, - await this.deployRunRepository.updateRun(currentRun.id, { - status: - buildStatus.building || !buildStatus.result - ? 'running' - : currentRun.status, - jenkinsBuildNumber: buildStatus.buildNumber, - jenkinsBuildUrl: buildStatus.url, - }), + await this.deployRunRepository.updateRun( + currentRun.id, + { + status: + buildStatus.building || !buildStatus.result + ? 'running' + : currentRun.status, + jenkinsBuildNumber: buildStatus.buildNumber, + jenkinsBuildUrl: buildStatus.url, + }, + currentRun.version, + ), ); if (buildStatus.building || !buildStatus.result) { return this.markStep( - currentRun.id, + currentRun, 'jenkins-build', 'running', `Jenkins 构建 #${buildStatus.buildNumber} 正在运行。`, @@ -263,7 +288,7 @@ export class DeployExecutionService { health.message ?? 'Jenkins 尚未配置,已完成本地模拟构建。'; const currentRun = await this.completeStep( - run.id, + run, 'jenkins-build', 'success', ); @@ -292,23 +317,27 @@ export class DeployExecutionService { ); } - await this.markStep(run.id, 'jenkins-build', 'running'); + let currentRun = await this.markStep(run, 'jenkins-build', 'running'); const queueItem = await this.jenkinsClient.buildWithParameters( environment.jenkinsJobPath, this.buildParameters(run), ); const nextStatus = queueItem.buildNumber ? 'running' : 'queued'; - let currentRun = this.requireRun( - run.id, - await this.deployRunRepository.updateRun(run.id, { - status: nextStatus, - jenkinsQueueId: queueItem.queueId, - jenkinsBuildNumber: queueItem.buildNumber, - jenkinsBuildUrl: queueItem.executableUrl, - }), + currentRun = this.requireRun( + currentRun.id, + await this.deployRunRepository.updateRun( + currentRun.id, + { + status: nextStatus, + jenkinsQueueId: queueItem.queueId, + jenkinsBuildNumber: queueItem.buildNumber, + jenkinsBuildUrl: queueItem.executableUrl, + }, + currentRun.version, + ), ); currentRun = await this.markStep( - run.id, + currentRun, 'jenkins-build', nextStatus, queueItem.buildNumber @@ -336,17 +365,20 @@ export class DeployExecutionService { run: DeployRunSummary, event: DeployNotificationEvent, ): Promise { - await this.markStep(run.id, 'notify-wecom', 'running'); + let currentRun = await this.markStep(run, 'notify-wecom', 'running'); const notification = - await this.deployNotificationService.sendDeployRunEvent(run, event); + await this.deployNotificationService.sendDeployRunEvent( + currentRun, + event, + ); const stepStatus = notification.status === 'sent' ? 'success' : notification.status === 'failed' ? 'failed' : 'skipped'; - const currentRun = await this.completeStep( - run.id, + currentRun = await this.completeStep( + currentRun, 'notify-wecom', stepStatus, notification.status === 'sent' ? undefined : notification.reason, @@ -372,26 +404,30 @@ export class DeployExecutionService { run: DeployRunSummary, ): Promise { let currentRun = await this.completeStep( - run.id, + run, 'jenkins-build', 'success', ); currentRun = await this.completeStep( - currentRun.id, + currentRun, 'health-check', 'success', ); currentRun = await this.advanceNotification(currentRun, 'succeeded'); currentRun = await this.completeStep( - currentRun.id, + currentRun, 'finish-release', 'success', ); currentRun = this.requireRun( currentRun.id, - await this.deployRunRepository.updateRun(currentRun.id, { - status: 'success', - }), + await this.deployRunRepository.updateRun( + currentRun.id, + { + status: 'success', + }, + currentRun.version, + ), ); await this.auditService.record({ @@ -417,17 +453,26 @@ export class DeployExecutionService { const logExcerpt = await this.readFailureLogExcerpt(run, jobPath); const failureSummary = `Jenkins 构建 #${run.jenkinsBuildNumber ?? '-'} ${result}`; let currentRun = - (await this.deployRunRepository.updateStep(run.id, 'jenkins-build', { - status: 'failed', - message: undefined, - errorSummary: failureSummary, - logExcerpt, - })) ?? run; + (await this.deployRunRepository.updateStep( + run.id, + 'jenkins-build', + { + status: 'failed', + message: undefined, + errorSummary: failureSummary, + logExcerpt, + }, + run.version, + )) ?? run; currentRun = - (await this.deployRunRepository.updateRun(currentRun.id, { - status: 'failed', - failureSummary, - })) ?? currentRun; + (await this.deployRunRepository.updateRun( + currentRun.id, + { + status: 'failed', + failureSummary, + }, + currentRun.version, + )) ?? currentRun; currentRun = await this.advanceNotification(currentRun, 'failed'); await this.auditService.record({ @@ -502,33 +547,43 @@ export class DeployExecutionService { } private async completeStep( - runId: string, + run: DeployRunSummary, bpmnNodeId: string, status: Extract, errorSummary?: string, ): Promise { return this.requireRun( - runId, - await this.deployRunRepository.updateStep(runId, bpmnNodeId, { - status, - message: undefined, - errorSummary, - }), + run.id, + await this.deployRunRepository.updateStep( + run.id, + bpmnNodeId, + { + status, + message: undefined, + errorSummary, + }, + run.version, + ), ); } private async markStep( - runId: string, + run: DeployRunSummary, bpmnNodeId: string, status: Extract, message?: string, ): Promise { return this.requireRun( - runId, - await this.deployRunRepository.updateStep(runId, bpmnNodeId, { - status, - message, - }), + run.id, + await this.deployRunRepository.updateStep( + run.id, + bpmnNodeId, + { + status, + message, + }, + run.version, + ), ); } diff --git a/src/deploy-runs/deploy-run.repository.ts b/src/deploy-runs/deploy-run.repository.ts index fd74ee1..7a3137e 100644 --- a/src/deploy-runs/deploy-run.repository.ts +++ b/src/deploy-runs/deploy-run.repository.ts @@ -11,6 +11,7 @@ import { Prisma, RunStepStatus as PrismaRunStepStatus, } from '@prisma/client'; +import { AppError } from '../common/errors/app-error'; import { EnvConfig } from '../config/env.schema'; import { PrismaService } from '../prisma/prisma.service'; import { seedDeployRuns } from './deploy-run.seed'; @@ -217,9 +218,10 @@ export class DeployRunRepository { async updateRun( id: string, patch: DeployRunMemoryPatch, + expectedVersion?: number, ): Promise { if (!this.useDatabase()) { - return this.updateMemoryRun(id, patch); + return this.updateMemoryRun(id, patch, expectedVersion); } const current = await this.prisma.deployRun.findUnique({ @@ -279,11 +281,16 @@ export class DeployRunRepository { data.finishedAt = this.toDate(patch.finishedAt); } - await this.prisma.deployRun.update({ - where: { id }, + const version = expectedVersion ?? current.version; + const updated = await this.prisma.deployRun.updateMany({ + where: { id, version }, data, }); + if (updated.count === 0) { + await this.raiseRunVersionConflict(id, version); + } + return this.findById(id); } @@ -291,9 +298,15 @@ export class DeployRunRepository { runId: string, bpmnNodeId: string, patch: DeployRunStepMemoryPatch, + expectedRunVersion?: number, ): Promise { if (!this.useDatabase()) { - return this.updateMemoryStep(runId, bpmnNodeId, patch); + return this.updateMemoryStep( + runId, + bpmnNodeId, + patch, + expectedRunVersion, + ); } const current = await this.prisma.runStep.findUnique({ @@ -303,6 +316,9 @@ export class DeployRunRepository { bpmnNodeId, }, }, + include: { + deployRun: true, + }, }); if (!current) { @@ -355,19 +371,89 @@ export class DeployRunRepository { data.metadata = metadata; } - await this.prisma.runStep.update({ - where: { - deployRunId_bpmnNodeId: { - deployRunId: runId, - bpmnNodeId, + const runVersion = expectedRunVersion ?? current.deployRun.version; + + await this.prisma.$transaction(async (tx) => { + const updatedRun = await tx.deployRun.updateMany({ + where: { + id: runId, + version: runVersion, }, - }, - data, + data: { + version: { + increment: 1, + }, + }, + }); + + if (updatedRun.count === 0) { + throw this.createRunVersionConflict(runId, runVersion, { + version: current.deployRun.version, + status: current.deployRun.status, + updatedAt: current.deployRun.updatedAt, + }); + } + + await tx.runStep.update({ + where: { + deployRunId_bpmnNodeId: { + deployRunId: runId, + bpmnNodeId, + }, + }, + data, + }); }); return this.findById(runId); } + private async raiseRunVersionConflict( + id: string, + expectedVersion: number, + ): Promise { + throw await this.buildRunVersionConflict(id, expectedVersion); + } + + private async buildRunVersionConflict( + id: string, + expectedVersion: number, + ): Promise { + const latest = await this.prisma.deployRun.findUnique({ + where: { id }, + select: { + status: true, + version: true, + updatedAt: true, + }, + }); + + return this.createRunVersionConflict(id, expectedVersion, latest); + } + + private createRunVersionConflict( + id: string, + expectedVersion: number, + latest?: { + status?: PrismaDeployRunStatus; + version?: number; + updatedAt?: Date; + } | null, + ): AppError { + return new AppError( + 'CONFLICT', + `发布记录 ${id} 已被其他流程更新,请刷新后重试`, + 409, + { + id, + expectedVersion, + latestVersion: latest?.version, + latestStatus: latest?.status?.toLowerCase(), + latestUpdatedAt: latest?.updatedAt?.toISOString(), + }, + ); + } + private async findRecordById(id: string): Promise { return this.prisma.deployRun.findUnique({ where: { id }, @@ -440,6 +526,7 @@ export class DeployRunRepository { return { id: idempotencyKey, + version: 1, projectKey: input.projectKey, environment: input.environment, ref: input.ref, @@ -497,6 +584,7 @@ export class DeployRunRepository { private updateMemoryRun( id: string, patch: DeployRunMemoryPatch, + expectedVersion?: number, ): DeployRunSummary | null { const run = this.memoryRuns.find((item) => item.id === id); @@ -504,6 +592,7 @@ export class DeployRunRepository { return null; } + this.assertMemoryVersion(run, expectedVersion); const now = new Date().toISOString(); if (patch.status) { @@ -542,6 +631,7 @@ export class DeployRunRepository { run.finishedAt = patch.finishedAt; } + run.version += 1; return run; } @@ -549,6 +639,7 @@ export class DeployRunRepository { runId: string, bpmnNodeId: string, patch: DeployRunStepMemoryPatch, + expectedRunVersion?: number, ): DeployRunSummary | null { const run = this.memoryRuns.find((item) => item.id === runId); const step = run?.steps.find((item) => item.bpmnNodeId === bpmnNodeId); @@ -557,6 +648,7 @@ export class DeployRunRepository { return null; } + this.assertMemoryVersion(run, expectedRunVersion); const now = new Date().toISOString(); if (patch.status) { @@ -591,15 +683,38 @@ export class DeployRunRepository { step.logExcerpt = patch.logExcerpt; } + run.version += 1; return run; } + private assertMemoryVersion( + run: DeployRunSummary, + expectedVersion?: number, + ): void { + if (expectedVersion === undefined || run.version === expectedVersion) { + return; + } + + throw new AppError( + 'CONFLICT', + `发布记录 ${run.id} 已被其他流程更新,请刷新后重试`, + 409, + { + id: run.id, + expectedVersion, + latestVersion: run.version, + latestStatus: run.status, + }, + ); + } + private toSummary(run: DeployRunRecord): DeployRunSummary { const metadata = this.objectMetadata(run.metadata); const sortedSteps = [...run.steps].sort((left, right) => left.order - right.order); return { id: run.id, + version: run.version, projectKey: run.project.key, environment: run.environment.name === 'TEST' ? 'test' : 'production', ref: run.ref, diff --git a/src/deploy-runs/deploy-run.seed.ts b/src/deploy-runs/deploy-run.seed.ts index d570709..717276a 100644 --- a/src/deploy-runs/deploy-run.seed.ts +++ b/src/deploy-runs/deploy-run.seed.ts @@ -5,6 +5,7 @@ const now = new Date().toISOString(); export const seedDeployRuns: DeployRunSummary[] = [ { id: 'run_seed_access_manage_test_001', + version: 1, projectKey: 'access-manage', environment: 'test', ref: 'develop', diff --git a/src/deploy-runs/deploy-run.types.ts b/src/deploy-runs/deploy-run.types.ts index f28bb7c..16d651f 100644 --- a/src/deploy-runs/deploy-run.types.ts +++ b/src/deploy-runs/deploy-run.types.ts @@ -23,6 +23,7 @@ export type DeployRunStepSummary = { export type DeployRunSummary = { id: string; + version: number; projectKey: string; environment: DeployEnvironment; ref: string; diff --git a/src/deploy-runs/deploy-runs.service.ts b/src/deploy-runs/deploy-runs.service.ts index 8b5d707..95ca20a 100644 --- a/src/deploy-runs/deploy-runs.service.ts +++ b/src/deploy-runs/deploy-runs.service.ts @@ -165,17 +165,26 @@ export class DeployRunsService { const jenkinsCancel = await this.cancelJenkinsExecution(source, project); let run = this.requireRun( id, - await this.deployRunRepository.updateRun(id, { - status: 'canceled', - }), + await this.deployRunRepository.updateRun( + id, + { + status: 'canceled', + }, + source.version, + ), ); run = this.requireRun( id, - await this.deployRunRepository.updateStep(id, 'jenkins-build', { - status: 'canceled', - message: undefined, - errorSummary: this.cancelSummary(jenkinsCancel), - }), + await this.deployRunRepository.updateStep( + id, + 'jenkins-build', + { + status: 'canceled', + message: undefined, + errorSummary: this.cancelSummary(jenkinsCancel), + }, + run.version, + ), ); await this.auditService.record({