triggerdotdev--trigger.dev/apps/webapp/app/runEngine/concerns/traceEvents.server.ts

import { SemanticInternalAttributes } from "@trigger.dev/core/v3/semanticInternalAttributes";
import type { TaskRun } from "@trigger.dev/database";
import type { IEventRepository } from "~/v3/eventRepository/eventRepository.types";
import { getEventRepository } from "~/v3/eventRepository/index.server";
import { runTriggeredAt } from "~/v3/runTimestamps";
import type { TracedEventSpan, TraceEventConcern, TriggerTaskRequest } from "../types";

export class DefaultTraceEventsConcern implements TraceEventConcern {
  async #getEventRepository(
    request: TriggerTaskRequest,
    parentStore: string | undefined
  ): Promise<{ repository: IEventRepository; store: string }> {
    return await getEventRepository(
      request.environment.organization.id,
      request.environment.organization.featureFlags as Record<string, unknown>,
      parentStore
    );
  }

  async traceRun<T>(
    request: TriggerTaskRequest,
    parentStore: string | undefined,
    callback: (span: TracedEventSpan, store: string) => Promise<T>
  ): Promise<T> {
    const { repository, store } = await this.#getEventRepository(request, parentStore);
    const startTime = request.options?.overrideCreatedAt
      ? runTriggeredAt({
          createdAt: request.options.overrideCreatedAt,
          queueTimestamp: request.options.queueTimestamp,
          scheduleId: request.options.scheduleId,
        })
      : undefined;

    return await repository.traceEvent(
      request.taskId,
      {
        context: request.options?.traceContext,
        spanParentAsLink: request.options?.spanParentAsLink,
        kind: "SERVER",
        environment: request.environment,
        taskSlug: request.taskId,
        attributes: {
          properties: {},
          style: {
            icon: request.options?.customIcon ?? "task",
          },
        },
        incomplete: true,
        immediate: true,
        startTime: startTime ? BigInt(startTime.getTime()) * BigInt(1000000) : undefined,
      },
      async (event, traceContext, traceparent) => {
        return await callback(
          {
            traceId: event.traceId,
            spanId: event.spanId,
            traceContext,
            traceparent,
            setAttribute: (key, value) => event.setAttribute(key as any, value),
            failWithError: event.failWithError.bind(event),
            stop: event.stop.bind(event),
          },
          store
        );
      }
    );
  }

  async traceIdempotentRun<T>(
    request: TriggerTaskRequest,
    parentStore: string | undefined,
    options: {
      existingRun: TaskRun;
      idempotencyKey: string;
      incomplete: boolean;
      isError: boolean;
    },
    callback: (span: TracedEventSpan, store: string) => Promise<T>
  ): Promise<T> {
    const { existingRun, idempotencyKey, incomplete, isError } = options;
    const { repository, store } = await this.#getEventRepository(request, parentStore);

    return await repository.traceEvent(
      `${request.taskId} (cached)`,
      {
        context: request.options?.traceContext,
        spanParentAsLink: request.options?.spanParentAsLink,
        kind: "SERVER",
        environment: request.environment,
        taskSlug: request.taskId,
        attributes: {
          properties: {
            [SemanticInternalAttributes.ORIGINAL_RUN_ID]: existingRun.friendlyId,
          },
          style: {
            icon: "task-cached",
          },
          runId: existingRun.friendlyId,
        },
        incomplete,
        isError,
        immediate: true,
      },
      async (event, traceContext, traceparent) => {
        //log a message
        await repository.recordEvent(
          `There's an existing run for idempotencyKey: ${idempotencyKey}`,
          {
            taskSlug: request.taskId,
            environment: request.environment,
            attributes: {
              runId: existingRun.friendlyId,
            },
            context: request.options?.traceContext,
            parentId: event.spanId,
          }
        );

        return await callback(
          {
            traceId: event.traceId,
            spanId: event.spanId,
            traceContext,
            traceparent,
            setAttribute: (key, value) => event.setAttribute(key as any, value),
            failWithError: event.failWithError.bind(event),
            stop: event.stop.bind(event),
          },
          store
        );
      }
    );
  }

  async traceDebouncedRun<T>(
    request: TriggerTaskRequest,
    parentStore: string | undefined,
    options: {
      existingRun: TaskRun;
      debounceKey: string;
      incomplete: boolean;
      isError: boolean;
    },
    callback: (span: TracedEventSpan, store: string) => Promise<T>
  ): Promise<T> {
    const { existingRun, debounceKey, incomplete, isError } = options;
    const { repository, store } = await this.#getEventRepository(request, parentStore);

    return await repository.traceEvent(
      `${request.taskId} (debounced)`,
      {
        context: request.options?.traceContext,
        spanParentAsLink: request.options?.spanParentAsLink,
        kind: "SERVER",
        environment: request.environment,
        taskSlug: request.taskId,
        attributes: {
          properties: {
            [SemanticInternalAttributes.ORIGINAL_RUN_ID]: existingRun.friendlyId,
          },
          style: {
            icon: "task-cached",
          },
          runId: existingRun.friendlyId,
        },
        incomplete,
        isError,
        immediate: true,
      },
      async (event, traceContext, traceparent) => {
        // Log a message about the debounced trigger
        await repository.recordEvent(`Debounced: using existing run with key "${debounceKey}"`, {
          taskSlug: request.taskId,
          environment: request.environment,
          attributes: {
            runId: existingRun.friendlyId,
          },
          context: request.options?.traceContext,
          parentId: event.spanId,
        });

        return await callback(
          {
            traceId: event.traceId,
            spanId: event.spanId,
            traceContext,
            traceparent,
            setAttribute: (key, value) => event.setAttribute(key as any, value),
            failWithError: event.failWithError.bind(event),
            stop: event.stop.bind(event),
          },
          store
        );
      }
    );
  }
}