devops-platform-api/src/deploy-runs/deploy-run.repository.ts

/**
 * 发布单仓储同时支持本地内存模式和 MySQL/Prisma 持久化模式。
 * 业务层只关心 run/step 状态流转，是否落库由环境配置统一控制。
 */
import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import {
  DeployRunStatus as PrismaDeployRunStatus,
  DeployRunTrigger as PrismaDeployRunTrigger,
  EnvironmentKind,
  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';
import {
  CreateDeployRunInput,
  DeployRunStepStatus,
  DeployRunSummary,
  DeployRunStatus,
} from './deploy-run.types';

type DeployRunMemoryPatch = {
  status?: DeployRunStatus;
  jenkinsQueueId?: string;
  jenkinsBuildNumber?: number;
  jenkinsBuildUrl?: string;
  failureSummary?: string;
  startedAt?: string;
  finishedAt?: string;
};

type DeployRunStepMemoryPatch = {
  status?: DeployRunStepStatus;
  startedAt?: string;
  finishedAt?: string;
  message?: string;
  errorSummary?: string;
  logExcerpt?: string;
};

type DeployRunRecord = Prisma.DeployRunGetPayload<{
  include: {
    project: true;
    environment: true;
    steps: true;
  };
}>;
type DeployRunDatabaseClient = PrismaService | Prisma.TransactionClient;

type DeployRunMetadata = Prisma.InputJsonObject & {
  jenkinsBuildUrl?: string;
  remark?: string;
  productionConfirmation?: Prisma.InputJsonObject;
};

type RunStepMetadata = Prisma.InputJsonObject & {
  message?: string;
  logExcerpt?: string;
};

const terminalRunStatuses = new Set<DeployRunStatus>([
  'success',
  'failed',
  'canceled',
]);
const terminalStepStatuses = new Set<DeployRunStepStatus>([
  'success',
  'failed',
  'canceled',
  'skipped',
]);
const runStatusToPrisma: Record<DeployRunStatus, PrismaDeployRunStatus> = {
  pending: PrismaDeployRunStatus.PENDING,
  queued: PrismaDeployRunStatus.QUEUED,
  running: PrismaDeployRunStatus.RUNNING,
  success: PrismaDeployRunStatus.SUCCESS,
  failed: PrismaDeployRunStatus.FAILED,
  canceled: PrismaDeployRunStatus.CANCELED,
};
const stepStatusToPrisma: Record<DeployRunStepStatus, PrismaRunStepStatus> = {
  pending: PrismaRunStepStatus.PENDING,
  queued: PrismaRunStepStatus.RUNNING,
  running: PrismaRunStepStatus.RUNNING,
  success: PrismaRunStepStatus.SUCCESS,
  failed: PrismaRunStepStatus.FAILED,
  canceled: PrismaRunStepStatus.CANCELED,
  skipped: PrismaRunStepStatus.SKIPPED,
};
const triggerToPrisma: Record<
  DeployRunSummary['trigger'],
  PrismaDeployRunTrigger
> = {
  manual: PrismaDeployRunTrigger.MANUAL,
  gitea_webhook: PrismaDeployRunTrigger.GITEA_WEBHOOK,
  retry: PrismaDeployRunTrigger.RETRY,
};

@Injectable()
export class DeployRunRepository {
  private readonly memoryRuns = [...seedDeployRuns];

  constructor(
    private readonly config: ConfigService<EnvConfig, true>,
    private readonly prisma: PrismaService,
  ) {}

  async findMany(): Promise<DeployRunSummary[]> {
    if (!this.useDatabase()) {
      return [...this.memoryRuns].sort((left, right) =>
        right.createdAt.localeCompare(left.createdAt),
      );
    }

    const runs = await this.prisma.deployRun.findMany({
      include: {
        project: true,
        environment: true,
        steps: {
          orderBy: {
            order: 'asc',
          },
        },
      },
      orderBy: {
        createdAt: 'desc',
      },
      take: 50,
    });

    return runs.map((run) => this.toSummary(run));
  }

  async findById(id: string): Promise<DeployRunSummary | null> {
    if (!this.useDatabase()) {
      return this.memoryRuns.find((run) => run.id === id) ?? null;
    }

    const run = await this.findRecordById(id);

    return run ? this.toSummary(run) : null;
  }

  async create(input: CreateDeployRunInput): Promise<DeployRunSummary> {
    const idempotencyKey = this.resolveIdempotencyKey(input);

    if (!this.useDatabase()) {
      return this.createMemoryRun(input, idempotencyKey);
    }

    const created = await this.createDatabaseRun(input, idempotencyKey);

    return this.toSummary(created);
  }

  async updateStatus(
    id: string,
    status: DeployRunStatus,
  ): Promise<DeployRunSummary | null> {
    return this.updateRun(id, { status });
  }

  async updateRun(
    id: string,
    patch: DeployRunMemoryPatch,
    expectedVersion?: number,
  ): Promise<DeployRunSummary | null> {
    if (!this.useDatabase()) {
      return this.updateMemoryRun(id, patch, expectedVersion);
    }

    const current = await this.prisma.deployRun.findUnique({
      where: { id },
    });

    if (!current) {
      return null;
    }

    const now = new Date();
    const metadata = this.objectMetadata<DeployRunMetadata>(current.metadata);
    const data: Prisma.DeployRunUpdateInput = {
      version: {
        increment: 1,
      },
    };

    if (patch.status) {
      data.status = runStatusToPrisma[patch.status];

      if (patch.status !== 'pending' && !current.startedAt) {
        data.startedAt = this.toDate(patch.startedAt) ?? now;
      }

      if (terminalRunStatuses.has(patch.status)) {
        data.finishedAt = this.toDate(patch.finishedAt) ?? now;
      }
    }

    if ('jenkinsQueueId' in patch) {
      data.jenkinsQueueId = patch.jenkinsQueueId ?? null;
    }

    if ('jenkinsBuildNumber' in patch) {
      data.jenkinsBuildNumber = patch.jenkinsBuildNumber ?? null;
    }

    if ('jenkinsBuildUrl' in patch) {
      if (patch.jenkinsBuildUrl) {
        metadata.jenkinsBuildUrl = patch.jenkinsBuildUrl;
      } else {
        delete metadata.jenkinsBuildUrl;
      }
      data.metadata = metadata;
    }

    if ('failureSummary' in patch) {
      data.failureSummary = patch.failureSummary ?? null;
    }

    if ('startedAt' in patch) {
      data.startedAt = this.toDate(patch.startedAt);
    }

    if ('finishedAt' in patch) {
      data.finishedAt = this.toDate(patch.finishedAt);
    }

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

  async updateStep(
    runId: string,
    bpmnNodeId: string,
    patch: DeployRunStepMemoryPatch,
    expectedRunVersion?: number,
  ): Promise<DeployRunSummary | null> {
    if (!this.useDatabase()) {
      return this.updateMemoryStep(
        runId,
        bpmnNodeId,
        patch,
        expectedRunVersion,
      );
    }

    const current = await this.prisma.runStep.findUnique({
      where: {
        deployRunId_bpmnNodeId: {
          deployRunId: runId,
          bpmnNodeId,
        },
      },
      include: {
        deployRun: true,
      },
    });

    if (!current) {
      return null;
    }

    const now = new Date();
    const metadata = this.objectMetadata<RunStepMetadata>(current.metadata);
    const data: Prisma.RunStepUpdateInput = {};

    if (patch.status) {
      data.status = stepStatusToPrisma[patch.status];

      if (patch.status !== 'pending' && !current.startedAt) {
        data.startedAt = this.toDate(patch.startedAt) ?? now;
      }

      if (terminalStepStatuses.has(patch.status)) {
        data.finishedAt = this.toDate(patch.finishedAt) ?? now;
      }
    }

    if ('startedAt' in patch) {
      data.startedAt = this.toDate(patch.startedAt);
    }

    if ('finishedAt' in patch) {
      data.finishedAt = this.toDate(patch.finishedAt);
    }

    if ('message' in patch) {
      if (patch.message) {
        metadata.message = patch.message;
      } else {
        delete metadata.message;
      }
      data.metadata = metadata;
    }

    if ('errorSummary' in patch) {
      data.errorSummary = patch.errorSummary ?? null;
    }

    if ('logExcerpt' in patch) {
      if (patch.logExcerpt) {
        metadata.logExcerpt = patch.logExcerpt;
      } else {
        delete metadata.logExcerpt;
      }
      data.metadata = metadata;
    }

    const runVersion = expectedRunVersion ?? current.deployRun.version;

    await this.prisma.$transaction(async (tx) => {
      const updatedRun = await tx.deployRun.updateMany({
        where: {
          id: runId,
          version: runVersion,
        },
        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<never> {
    throw await this.buildRunVersionConflict(id, expectedVersion);
  }

  private async buildRunVersionConflict(
    id: string,
    expectedVersion: number,
  ): Promise<AppError> {
    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<DeployRunRecord | null> {
    return this.prisma.deployRun.findUnique({
      where: { id },
      include: {
        project: true,
        environment: true,
        steps: {
          orderBy: {
            order: 'asc',
          },
        },
      },
    });
  }

  private async findRecordByIdempotencyKey(
    idempotencyKey: string,
    client: DeployRunDatabaseClient = this.prisma,
  ): Promise<DeployRunRecord | null> {
    return client.deployRun.findUnique({
      where: { idempotencyKey },
      include: {
        project: true,
        environment: true,
        steps: {
          orderBy: {
            order: 'asc',
          },
        },
      },
    });
  }

  private async createDatabaseRun(
    input: CreateDeployRunInput,
    idempotencyKey: string,
  ): Promise<DeployRunRecord> {
    try {
      return await this.prisma.$transaction(async (tx) => {
        const existing = await this.findRecordByIdempotencyKey(
          idempotencyKey,
          tx,
        );

        if (existing) {
          return existing;
        }

        const target = await this.findTargetProjectEnvironment(input, tx);
        const initialRun = this.buildInitialRun(input, idempotencyKey);
        const metadata = this.inputMetadata(input);

        return tx.deployRun.create({
          data: {
            projectId: target.project.id,
            environmentId: target.environment.id,
            jobId: target.deployJob?.id,
            status: PrismaDeployRunStatus.PENDING,
            trigger: triggerToPrisma[input.trigger ?? 'manual'],
            ref: input.ref,
            operator: input.operator,
            idempotencyKey,
            metadata,
            steps: {
              create: initialRun.steps.map((step) => ({
                bpmnNodeId: step.bpmnNodeId,
                name: step.name,
                status: stepStatusToPrisma[step.status],
                order: step.order,
              })),
            },
          },
          include: {
            project: true,
            environment: true,
            steps: {
              orderBy: {
                order: 'asc',
              },
            },
          },
        });
      });
    } catch (error) {
      if (this.isIdempotencyUniqueConflict(error)) {
        const existing = await this.findRecordByIdempotencyKey(idempotencyKey);

        if (existing) {
          return existing;
        }
      }

      throw error;
    }
  }

  private async findTargetProjectEnvironment(
    input: CreateDeployRunInput,
    client: DeployRunDatabaseClient = this.prisma,
  ) {
    const project = await client.project.findUnique({
      where: { key: input.projectKey },
      include: {
        environments: {
          include: {
            deployJobs: true,
          },
        },
      },
    });

    const targetEnvironmentName =
      input.environment === 'test'
        ? EnvironmentKind.TEST
        : EnvironmentKind.PRODUCTION;
    const environment = project?.environments.find(
      (item) => item.name === targetEnvironmentName,
    );
    const deployJob = environment?.deployJobs[0];

    if (!project || !environment) {
      throw new Error(
        `数据库未初始化项目 ${input.projectKey} 的 ${input.environment} 环境`,
      );
    }

    return {
      project,
      environment,
      deployJob,
    };
  }

  private isIdempotencyUniqueConflict(error: unknown): boolean {
    if (!error || typeof error !== 'object') {
      return false;
    }

    const record = error as {
      code?: unknown;
      meta?: {
        target?: unknown;
      };
    };

    if (record.code !== 'P2002') {
      return false;
    }

    const target = record.meta?.target;

    if (Array.isArray(target)) {
      return target.includes('idempotencyKey');
    }

    return (
      typeof target === 'string' &&
      target.toLowerCase().includes('idempotencykey')
    );
  }

  private createMemoryRun(
    input: CreateDeployRunInput,
    idempotencyKey: string,
  ): DeployRunSummary {
    const existing = this.memoryRuns.find((run) => run.id === idempotencyKey);

    if (existing) {
      return existing;
    }

    const run = this.buildInitialRun(input, idempotencyKey);
    this.memoryRuns.unshift(run);
    return run;
  }

  private buildInitialRun(
    input: CreateDeployRunInput,
    idempotencyKey: string,
  ): DeployRunSummary {
    const now = new Date().toISOString();

    return {
      id: idempotencyKey,
      version: 1,
      projectKey: input.projectKey,
      environment: input.environment,
      ref: input.ref,
      status: 'pending',
      trigger: input.trigger ?? 'manual',
      operator: input.operator,
      createdAt: now,
      steps: [
        {
          id: `${idempotencyKey}:start-release`,
          bpmnNodeId: 'start-release',
          name: '提交发布单',
          status: 'success',
          order: 10,
        },
        {
          id: `${idempotencyKey}:gitea-check`,
          bpmnNodeId: 'gitea-check',
          name: '校验 Gitea ref',
          status: 'pending',
          order: 20,
        },
        {
          id: `${idempotencyKey}:jenkins-build`,
          bpmnNodeId: 'jenkins-build',
          name: '触发 Jenkins 构建',
          status: 'pending',
          order: 30,
        },
        {
          id: `${idempotencyKey}:health-check`,
          bpmnNodeId: 'health-check',
          name: '环境健康检查',
          status: 'pending',
          order: 40,
        },
        {
          id: `${idempotencyKey}:notify-wecom`,
          bpmnNodeId: 'notify-wecom',
          name: '发送通知',
          status: 'pending',
          order: 90,
        },
        {
          id: `${idempotencyKey}:finish-release`,
          bpmnNodeId: 'finish-release',
          name: '归档审计',
          status: 'pending',
          order: 100,
        },
      ],
    };
  }

  private updateMemoryRun(
    id: string,
    patch: DeployRunMemoryPatch,
    expectedVersion?: number,
  ): DeployRunSummary | null {
    const run = this.memoryRuns.find((item) => item.id === id);

    if (!run) {
      return null;
    }

    this.assertMemoryVersion(run, expectedVersion);
    const now = new Date().toISOString();

    if (patch.status) {
      run.status = patch.status;

      if (patch.status !== 'pending' && !run.startedAt) {
        run.startedAt = patch.startedAt ?? now;
      }

      if (terminalRunStatuses.has(patch.status)) {
        run.finishedAt = patch.finishedAt ?? now;
      }
    }

    if ('jenkinsQueueId' in patch) {
      run.jenkinsQueueId = patch.jenkinsQueueId;
    }

    if ('jenkinsBuildNumber' in patch) {
      run.jenkinsBuildNumber = patch.jenkinsBuildNumber;
    }

    if ('jenkinsBuildUrl' in patch) {
      run.jenkinsBuildUrl = patch.jenkinsBuildUrl;
    }

    if ('failureSummary' in patch) {
      run.failureSummary = patch.failureSummary;
    }

    if ('startedAt' in patch) {
      run.startedAt = patch.startedAt;
    }

    if ('finishedAt' in patch) {
      run.finishedAt = patch.finishedAt;
    }

    run.version += 1;
    return run;
  }

  private updateMemoryStep(
    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);

    if (!run || !step) {
      return null;
    }

    this.assertMemoryVersion(run, expectedRunVersion);
    const now = new Date().toISOString();

    if (patch.status) {
      step.status = patch.status;

      if (patch.status !== 'pending' && !step.startedAt) {
        step.startedAt = patch.startedAt ?? now;
      }

      if (terminalStepStatuses.has(patch.status)) {
        step.finishedAt = patch.finishedAt ?? now;
      }
    }

    if ('startedAt' in patch) {
      step.startedAt = patch.startedAt;
    }

    if ('finishedAt' in patch) {
      step.finishedAt = patch.finishedAt;
    }

    if ('message' in patch) {
      step.message = patch.message;
    }

    if ('errorSummary' in patch) {
      step.errorSummary = patch.errorSummary;
    }

    if ('logExcerpt' in patch) {
      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<DeployRunMetadata>(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,
      status: run.status.toLowerCase() as DeployRunStatus,
      trigger: run.trigger.toLowerCase() as DeployRunSummary['trigger'],
      operator: run.operator,
      jenkinsQueueId: run.jenkinsQueueId ?? undefined,
      jenkinsBuildNumber: run.jenkinsBuildNumber ?? undefined,
      jenkinsBuildUrl: metadata.jenkinsBuildUrl,
      failureSummary: run.failureSummary ?? undefined,
      startedAt: run.startedAt?.toISOString(),
      finishedAt: run.finishedAt?.toISOString(),
      createdAt: run.createdAt.toISOString(),
      steps: sortedSteps.map((step) => {
        const stepMetadata = this.objectMetadata<RunStepMetadata>(
          step.metadata,
        );

        return {
          id: step.id,
          bpmnNodeId: step.bpmnNodeId,
          name: step.name,
          status: step.status.toLowerCase() as DeployRunStepStatus,
          order: step.order,
          startedAt: step.startedAt?.toISOString(),
          finishedAt: step.finishedAt?.toISOString(),
          message: stepMetadata.message,
          errorSummary: step.errorSummary ?? undefined,
          logExcerpt: stepMetadata.logExcerpt,
        };
      }),
    };
  }

  private objectMetadata<T extends Record<string, unknown>>(
    value: Prisma.JsonValue | null | undefined,
  ): T {
    if (value && typeof value === 'object' && !Array.isArray(value)) {
      return { ...(value as Record<string, unknown>) } as T;
    }

    return {} as T;
  }

  private inputMetadata(input: CreateDeployRunInput): Prisma.InputJsonObject {
    const metadata: DeployRunMetadata = {};

    if (input.remark) {
      metadata.remark = input.remark;
    }

    if (input.environment === 'production' && input.productionConfirmation) {
      metadata.productionConfirmation = {
        confirmed: input.productionConfirmation.confirmed,
        confirmedAt: input.productionConfirmation.confirmedAt,
        summary: input.productionConfirmation.summary,
        operator: input.operator,
      };
    }

    return metadata;
  }

  private resolveIdempotencyKey(input: CreateDeployRunInput): string {
    return (
      input.idempotencyKey ??
      [
        input.projectKey,
        input.environment,
        input.ref,
        input.operator,
        new Date().toISOString(),
      ].join(':')
    );
  }

  private toDate(value: string | undefined): Date | null | undefined {
    return value ? new Date(value) : undefined;
  }

  private useDatabase(): boolean {
    return (
      this.config.get('USE_DATABASE_READS', { infer: true }) &&
      Boolean(this.config.get('DATABASE_URL', { infer: true }))
    );
  }
}