stack/apps/api/src/job-events/job-events.service.ts

import { Injectable, NotFoundException } from "@nestjs/common";
import { Prisma } from "@prisma/client";
import { PrismaService } from "../prisma/prisma.service";
import { CreateEventDto, QueryEventsDto } from "./dto";
import {
  JOB_CREATED,
  JOB_STARTED,
  JOB_COMPLETED,
  JOB_FAILED,
  STEP_STARTED,
  STEP_COMPLETED,
  AI_TOKENS_USED,
} from "./event-types";

/**
 * Service for managing job events
 * Events are immutable once created and provide an audit log of all job activities
 */
@Injectable()
export class JobEventsService {
  constructor(private readonly prisma: PrismaService) {}

  /**
   * Emit a job event
   * Events are stored immutably in PostgreSQL
   */
  async emitEvent(jobId: string, createEventDto: CreateEventDto) {
    // Verify job exists
    const job = await this.prisma.runnerJob.findUnique({
      where: { id: jobId },
      select: { id: true },
    });

    if (!job) {
      throw new NotFoundException(`RunnerJob with ID ${jobId} not found`);
    }

    // Verify step exists if stepId is provided
    if (createEventDto.stepId) {
      const step = await this.prisma.jobStep.findUnique({
        where: { id: createEventDto.stepId },
        select: { id: true },
      });

      if (!step) {
        throw new NotFoundException(`JobStep with ID ${createEventDto.stepId} not found`);
      }
    }

    // Build event data
    const data: Prisma.JobEventCreateInput = {
      job: { connect: { id: jobId } },
      type: createEventDto.type,
      timestamp: new Date(),
      actor: createEventDto.actor,
      payload: createEventDto.payload as unknown as Prisma.InputJsonValue,
    };

    // Add step connection if provided
    if (createEventDto.stepId) {
      data.step = { connect: { id: createEventDto.stepId } };
    }

    // Create and return the event
    return this.prisma.jobEvent.create({ data });
  }

  /**
   * Get events for a specific job with optional filtering
   */
  async getEventsByJobId(jobId: string, query: QueryEventsDto) {
    // Verify job exists
    const job = await this.prisma.runnerJob.findUnique({
      where: { id: jobId },
      select: { id: true },
    });

    if (!job) {
      throw new NotFoundException(`RunnerJob with ID ${jobId} not found`);
    }

    const page = query.page ?? 1;
    const limit = query.limit ?? 50;
    const skip = (page - 1) * limit;

    // Build where clause
    const where: Prisma.JobEventWhereInput = { jobId };

    if (query.type) {
      where.type = query.type;
    }

    if (query.stepId) {
      where.stepId = query.stepId;
    }

    // Execute queries in parallel
    const [data, total] = await Promise.all([
      this.prisma.jobEvent.findMany({
        where,
        orderBy: { timestamp: "asc" },
        skip,
        take: limit,
      }),
      this.prisma.jobEvent.count({ where }),
    ]);

    return {
      data,
      meta: {
        total,
        page,
        limit,
        totalPages: Math.ceil(total / limit),
      },
    };
  }

  /**
   * Convenience method: Emit job.created event
   */
  async emitJobCreated(jobId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: JOB_CREATED,
      actor: "system",
      payload,
    });
  }

  /**
   * Convenience method: Emit job.started event
   */
  async emitJobStarted(jobId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: JOB_STARTED,
      actor: "system",
      payload,
    });
  }

  /**
   * Convenience method: Emit job.completed event
   */
  async emitJobCompleted(jobId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: JOB_COMPLETED,
      actor: "system",
      payload,
    });
  }

  /**
   * Convenience method: Emit job.failed event
   */
  async emitJobFailed(jobId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: JOB_FAILED,
      actor: "system",
      payload,
    });
  }

  /**
   * Convenience method: Emit step.started event
   */
  async emitStepStarted(jobId: string, stepId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: STEP_STARTED,
      actor: "system",
      payload,
      stepId,
    });
  }

  /**
   * Convenience method: Emit step.completed event
   */
  async emitStepCompleted(jobId: string, stepId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: STEP_COMPLETED,
      actor: "system",
      payload,
      stepId,
    });
  }

  /**
   * Convenience method: Emit ai.tokens_used event
   */
  async emitAiTokensUsed(jobId: string, payload: Record<string, unknown> = {}) {
    return this.emitEvent(jobId, {
      type: AI_TOKENS_USED,
      actor: "system",
      payload,
    });
  }

  /**
   * Get all events for a job (no pagination)
   * Alias for getEventsByJobId without pagination
   */
  async findByJob(
    jobId: string
  ): Promise<Awaited<ReturnType<typeof this.prisma.jobEvent.findMany>>> {
    // Verify job exists
    const job = await this.prisma.runnerJob.findUnique({
      where: { id: jobId },
      select: { id: true },
    });

    if (!job) {
      throw new NotFoundException(`RunnerJob with ID ${jobId} not found`);
    }

    return this.prisma.jobEvent.findMany({
      where: { jobId },
      orderBy: { timestamp: "asc" },
    });
  }
}