Class LokiLogAdapter

worker/src/adapters/loki-log.adapter.ts:83–263

}

export class LokiLogAdapter implements WorkflowLogSink {
  constructor(
    private readonly client: LokiPushClient,
    private readonly db?: NodePgDatabase<typeof schema>,
  ) {}
  private ensureIndexPromise?: Promise<void>;

  async append(entry: WorkflowLogEntry): Promise<void> {
    if (!entry.message || entry.message.trim().length === 0) {
      return;
    }

    const timestamp = entry.timestamp ?? new Date();
    const labels = this.buildLabels(entry);
    const lines = this.buildLines(entry.message, timestamp);
    const lineCount = lines.length;

    try {
      await this.client.push(labels, lines);
    } catch (error) {
      console.error('[LOKI] Failed to push log entry', error);
      return;
    }

    if (this.db) {
      await this.persistMetadata({
        runId: entry.runId,
        nodeRef: entry.nodeRef,
        stream: entry.stream,
        labels,
        lineCount,
        timestamp,
        organizationId: entry.organizationId ?? null,
      });
    }
  }

  private buildLabels(entry: WorkflowLogEntry): Record<string, string> {
    const labels: Record<string, string> = {
      run_id: entry.runId,
      node: entry.nodeRef,
      stream: entry.stream,
    };

    if (entry.level) {
      labels.level = entry.level;
    }

    if (entry.metadata?.activityId) {
      labels.activity_id = entry.metadata.activityId;
    }

    if (entry.metadata?.attempt !== undefined) {
      labels.attempt = String(entry.metadata.attempt);
    }

    if (entry.metadata?.correlationId) {
      labels.correlation_id = entry.metadata.correlationId;