api-server/src/modules/ai-job/ai-job.service.ts

import { Injectable, NotFoundException, ForbiddenException, ConflictException } from '@nestjs/common';
import { PrismaService } from '../../infrastructure/database/prisma.service';
import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository';
import { AiJobStateMachine, LifecycleStatus, TERMINAL_STATUSES } from './ai-job-state-machine';
import { AiJobCreationService } from './ai-job-creation.service';
import { JobDefinitionRegistry } from './job-definition-registry';
import { JobNotCancellableError } from './ai-job.errors';

@Injectable()
export class AiJobService {
  constructor(
    private readonly prisma: PrismaService,
    private readonly lifecycleRepo: AiJobLifecycleRepository,
    private readonly stateMachine: AiJobStateMachine,
    private readonly creationService: AiJobCreationService,
    private readonly registry: JobDefinitionRegistry,
  ) {}

  // ── 用户查询 ──

  async getJobForUser(userId: string, jobId: string) {
    const job = await this.prisma.aiJob.findUnique({
      where: { id: jobId },
      include: { artifacts: { orderBy: { ordinal: 'asc' } } },
    });
    if (!job) throw new NotFoundException('Job not found');
    if (job.userId !== userId) throw new ForbiddenException('Not your job');

    return this.toPublicResponse(job);
  }

  // ── 用户取消 ──

  async cancelJobForUser(userId: string, jobId: string) {
    const job = await this.prisma.aiJob.findUnique({
      where: { id: jobId },
      select: { id: true, userId: true, lifecycleStatus: true },
    });
    if (!job) throw new NotFoundException('Job not found');
    if (job.userId !== userId) throw new ForbiddenException('Not your job');

    const status = this.stateMachine.parse(job.lifecycleStatus);

    // 终态幂等返回
    if (this.stateMachine.isTerminal(status)) {
      return { jobId, status, message: `Job is already in terminal status "${status}"` };
    }

    try {
      const result = await this.lifecycleRepo.requestCancellation(jobId);
      return {
        jobId,
        status: result.status,
        message: result.status === 'cancelled'
          ? 'Job has been cancelled'
          : 'Cancellation requested — job will be cancelled shortly',
      };
    } catch (err: any) {
      if (err instanceof JobNotCancellableError) {
        throw new ConflictException(err.message);
      }
      throw err;
    }
  }

  // ── Admin 查询 ──

  async listJobs(query: {
    jobType?: string;
    lifecycleStatus?: string;
    queueName?: string;
    userId?: string;
    page?: number;
    limit?: number;
  }) {
    const page = query.page ?? 1;
    const limit = Math.min(query.limit ?? 20, 100);
    const skip = (page - 1) * limit;

    const where: any = {};
    if (query.jobType) where.jobType = query.jobType;
    if (query.lifecycleStatus) where.lifecycleStatus = query.lifecycleStatus;
    if (query.queueName) where.queueName = query.queueName;
    if (query.userId) where.userId = query.userId;

    const [items, total] = await Promise.all([
      this.prisma.aiJob.findMany({
        where,
        skip,
        take: limit,
        orderBy: { createdAt: 'desc' },
        include: { artifacts: { orderBy: { ordinal: 'asc' } } },
      }),
      this.prisma.aiJob.count({ where }),
    ]);

    return {
      items: items.map((j: any) => this.toAdminResponse(j)),
      total,
      page,
      limit,
    };
  }

  async getJobDetail(jobId: string) {
    const job = await this.prisma.aiJob.findUnique({
      where: { id: jobId },
      include: {
        artifacts: { orderBy: { ordinal: 'asc' } },
        usageLogs: { orderBy: { createdAt: 'desc' }, take: 10 },
      },
    });
    if (!job) throw new NotFoundException('Job not found');

    // 额外加载 snapshot 和 outbox
    const [snapshot, outboxEvents] = await Promise.all([
      this.prisma.aiJobSnapshot.findUnique({ where: { jobId } }),
      this.prisma.outboxEvent.findMany({
        where: { aggregateId: jobId },
        orderBy: { createdAt: 'asc' },
      }),
    ]);

    const response = this.toAdminResponse(job);
    return {
      ...response,
      snapshot: snapshot ? { id: snapshot.id, schemaVersion: snapshot.schemaVersion, contentHash: snapshot.contentHash } : null,
      outboxEvents: outboxEvents.map((e) => ({
        id: e.id,
        eventType: e.eventType,
        status: e.status,
        attemptCount: e.attemptCount,
        lastErrorCode: e.lastErrorCode,
        createdAt: e.createdAt,
      })),
      usageLogs: (job as any).usageLogs || [],
    };
  }

  async retryJob(adminUserId: string, jobId: string) {
    // 1. Load original Job
    const original = await this.prisma.aiJob.findUnique({
      where: { id: jobId },
      include: { snapshot: true },
    });
    if (!original) throw new NotFoundException('Original job not found');

    // 2. Validate jobType via Registry
    this.registry.get(original.jobType); // throws UnknownJobTypeError if not registered

    // 3. Validate retryable state (only terminal jobs can be retried)
    const status = this.stateMachine.parse(original.lifecycleStatus);
    if (!this.stateMachine.isTerminal(status)) {
      throw new ConflictException(
        `Cannot retry job in "${status}" status. Only terminal jobs (succeeded/failed/cancelled) can be retried.`,
      );
    }

    // 4. Minimal validation: original must have targetType + targetId
    if (!original.targetType || !original.targetId) {
      throw new ConflictException('Original job missing targetType/targetId — cannot retry');
    }

    // 5. Idempotency key: allows same admin operation to return same retry job,
    //    but different retry operations produce different jobs
    const retryIdempotencyKey = `admin-retry:${jobId}:${Date.now()}`;

    // 6. Delegate to AiJobCreationService (Registry + Snapshot + Outbox + atomic transaction)
    const newJob = await this.creationService.createJob({
      userId: original.userId,
      jobType: original.jobType,
      triggerType: 'admin_manual',
      targetType: original.targetType!,
      targetId: original.targetId!,
      parentJobId: original.id,
      idempotencyKey: retryIdempotencyKey,
      retrySnapshotContent: original.snapshot?.content as Record<string, unknown> | undefined,
      overrides: {
        credentialMode: original.credentialMode as 'platform_key' | 'user_deepseek_key',
        credentialId: original.credentialId ?? undefined,
      },
    });

    this.logger.log(
      `Admin retry: original=${jobId} → new=${newJob.id} ` +
      `type=${newJob.jobType} parentJobId=${original.id}`,
    );

    return this.toAdminResponse(newJob);
  }

  private logger = new (require('@nestjs/common').Logger)(AiJobService.name);

  // ── 可观测性 ──

  async getOutboxStatus() {
    const counts = await Promise.all([
      this.prisma.outboxEvent.count({ where: { status: 'pending' } }),
      this.prisma.outboxEvent.count({ where: { status: 'processing' } }),
      this.prisma.outboxEvent.count({ where: { status: 'published' } }),
      this.prisma.outboxEvent.count({ where: { status: 'failed' } }),
    ]);
    return { pending: counts[0], processing: counts[1], published: counts[2], failed: counts[3] };
  }

  // ── 响应构造 ──

  private toPublicResponse(job: any) {
    return {
      id: job.id,
      jobType: job.jobType,
      lifecycleStatus: job.lifecycleStatus,
      queueName: job.queueName,
      priority: job.priority,
      progress: job.progress,
      createdAt: job.createdAt,
      queuedAt: job.queuedAt,
      startedAt: job.startedAt,
      finishedAt: job.finishedAt,
      errorCode: job.errorCode,
      publicErrorMessage: job.publicErrorMessage,
      artifacts: (job.artifacts || []).map((a: any) => ({
        artifactType: a.artifactType,
        artifactId: a.artifactId,
        ordinal: a.ordinal,
      })),
      // 禁止返回：validatedOutput, internalErrorMessage, snapshot, credential, outbox
    };
  }

  private toAdminResponse(job: any) {
    return {
      id: job.id,
      userId: job.userId,
      jobType: job.jobType,
      lifecycleStatus: job.lifecycleStatus,
      triggerType: job.triggerType,
      queueName: job.queueName,
      priority: job.priority,
      progress: job.progress,
      targetType: job.targetType,
      targetId: job.targetId,
      parentJobId: job.parentJobId,
      attemptCount: job.attemptCount,
      maxAttempts: job.maxAttempts,
      timeoutMs: job.timeoutMs,
      credentialMode: job.credentialMode,
      modelProvider: job.modelProvider,
      modelName: job.modelName,
      createdAt: job.createdAt,
      queuedAt: job.queuedAt,
      startedAt: job.startedAt,
      finishedAt: job.finishedAt,
      errorCode: job.errorCode,
      publicErrorMessage: job.publicErrorMessage,
      internalErrorMessage: job.internalErrorMessage,
      validatedOutput: job.validatedOutput,
      outputHash: job.outputHash,
      artifacts: (job.artifacts || []).map((a: any) => ({
        artifactType: a.artifactType,
        artifactId: a.artifactId,
        ordinal: a.ordinal,
      })),
    };
  }
}