/**
 * apps/worker — automation worker + trigger matcher consumer (TASK-0113).
 *
 * Two BullMQ/stream surfaces, both thin over the `automations/` package:
 *
 *  1. `createAutomationsWorker` — consumes the `automations` queue.
 *     Two job kinds share the queue:
 *      - `run-automation` jobs (`{ runId, automationId, workspaceId,
 *        triggerEvent, eventId }`) → `handleRunAutomation` (the TASK-0113
 *        pipeline: PENDING → RUNNING → terminal);
 *      - schedule repeatable fires (job NAME `automations:run:{automationId}`,
 *        data `{ automationId, event: "schedule", cron }`) →
 *        `executeScheduleFire` (creates the PENDING run + enqueues a
 *        `run-automation` job).
 *     NO BullMQ attempts on this queue (`attempts: 1`): the pipeline is the
 *     retry manager (worker-side retryCount + backoff, §7). BullMQ's
 *     `lockDuration` 60 s / `stalledInterval` 30 s still reprocess in-process
 *     stalls (§9); the watchdog is the cross-restart backstop.
 *
 *  2. `createAutomationsConsumer` — a StreamConsumer on `factory:events`,
 *     group **`automations-matcher`** (renamed from the old
 *     `worker-automations` — spec §5; a first deploy with the old group
 *     simply gets a second group reading from "0", which the dedupe key
 *     makes idempotent). Maps stream events → trigger types
 *     (`record.updated` fans out to RECORD_UPDATED + PROPERTY_CHANGED) and
 *     delegates to `matchAndEnqueue` (scope match + dedupe + PENDING run +
 *     enqueue).
 *
 * The old processor's `handleAutomationFinalFailure` (BullMQ `failed` hook)
 * is GONE: with attempts:1 the pipeline handles every failure itself
 * (FAILED / RETRYING / DEAD_LETTER are the handler's return values, not
 * BullMQ retries). index.ts drops the `.on("failed")` attachment.
 */
import type { PrismaClient } from "@prisma/client";
import type { Redis } from "ioredis";
import { Worker, type Job } from "bullmq";
import { withJobOtel } from "../lib/job-otel.js";
import { getLogger } from "../logger.js";
import type { WorkerEnv } from "../config.js";
import { QUEUES } from "../queues.js";
import { StreamConsumer } from "../lib/redis-stream.js";
import { tryParseEvent } from "../events.js";
import { handleRunAutomation, runAutomationJobSchema } from "./automations/pipeline.js";
import { matchAndEnqueue } from "./automations/matcher.js";
import { executeScheduleFire, isScheduleFireJobName } from "./automations/schedule.js";

const log = getLogger("automations");

/**
 * Create the `automations` BullMQ worker. `concurrency` comes from
 * `AUTOMATIONS_WORKER_CONCURRENCY` (default 4 — spec §1; schedule fires and
 * run jobs share the pool).
 */
export function createAutomationsWorker(opts: {
  connection: Redis;
  prisma: PrismaClient;
  env: WorkerEnv;
  concurrency?: number;
}): Worker {
  const { connection, prisma, env } = opts;
  return new Worker(
    QUEUES.automations,
    withJobOtel(QUEUES.automations, async (job: Job) => {
      // 1) Schedule repeatable fire (job NAME carries the identity).
      if (isScheduleFireJobName(job.name)) {
        return executeScheduleFire(connection, prisma, env, (job.data ?? {}) as {
          automationId?: string;
          cron?: string;
          timeZone?: string;
        });
      }
      // 2) run-automation job.
      const data = runAutomationJobSchema.safeParse(job.data);
      if (!data.success) {
        log.error({ jobId: job.id, err: data.error.message }, "malformed run-automation job (dropped)");
        return { status: "FAILED" };
      }
      const status = await handleRunAutomation({ prisma, redis: connection, env }, { data: job.data });
      log.info({ runId: data.data.runId, jobId: job.id, status }, "run-automation job settled");
      return { status };
    }),
    {
      connection,
      concurrency: opts.concurrency ?? env.AUTOMATIONS_WORKER_CONCURRENCY,
      // §9: the in-process stall backstop (a wedged worker is reprocessed by
      // a peer after 60 s; the watchdog covers the cross-restart case).
      lockDuration: 60_000,
      stalledInterval: 30_000
    }
  );
}

/**
 * Create the trigger-matcher stream consumer (group `automations-matcher`).
 * Every event the engine can react to is handled here; unknown events are
 * ACKed as not-mine (StreamConsumer semantics: a resolved handler = ACK).
 *
 * `record.updated` fans out to BOTH RECORD_UPDATED and PROPERTY_CHANGED
 * trigger types (PROPERTY_CHANGED is not in EVENT_TO_TRIGGER — it is a
 * property-scoped view of the same event).
 */
export function createAutomationsConsumer(opts: {
  redis: Redis;
  prisma: PrismaClient;
  consumerId: string;
}): StreamConsumer {
  const { redis, prisma, consumerId } = opts;
  return new StreamConsumer(
    redis,
    "factory:events",
    "automations-matcher",
    consumerId,
    async (entry) => {
      const parsed = tryParseEvent(entry);
      if (!parsed) return; // poison entry — StreamConsumer logs + drops
      const { envelope } = parsed;
      const triggerTypes = triggerTypesFor(envelope.event);
      if (triggerTypes.length === 0) return; // not mine — ACK
      const payload = (envelope.payload ?? {}) as Record<string, unknown>;
      for (const tt of triggerTypes) {
        try {
          await matchAndEnqueue({ prisma, redis }, tt, { id: parsed.id, event: envelope.event, payload });
        } catch (err) {
          // A matcher failure for one trigger type must not wedge the group:
          // log + ACK (dedupe keys mean a manual re-drive is safe).
          log.error({ err: (err as Error).message, triggerType: tt, entryId: parsed.id }, "matcher pass failed");
        }
      }
    }
  );
}

/** Stream event → the trigger types it can satisfy. */
function triggerTypesFor(eventName: string): string[] {
  switch (eventName) {
    case "record.created":
      return ["RECORD_CREATED"];
    case "record.updated":
      return ["RECORD_UPDATED", "PROPERTY_CHANGED"];
    case "page.created":
      return ["PAGE_CREATED"];
    case "form.submitted":
      return ["FORM_SUBMITTED"];
    default:
      return [];
  }
}