/**
 * apps/worker — automation trigger matcher (TASK-0113 / TASK-0110 matcher).
 *
 * For each `factory:events` stream entry, find the ACTIVE automations whose
 * trigger(s) of the matching type scope-match the event, dedupe on the
 * event id, create a PENDING `automation_runs` row, and enqueue a
 * `run-automation` job on the `automations` queue.
 *
 * Stream→trigger map (events.ts `EVENT_TO_TRIGGER`):
 *   record.created  → RECORD_CREATED
 *   record.updated  → RECORD_UPDATED | PROPERTY_CHANGED
 *   page.created    → PAGE_CREATED
 *   form.submitted  → FORM_SUBMITTED
 * MANUAL / SCHEDULE / WEBHOOK are enqueued by other paths (API manual run,
 * schedule reconciler, incoming webhook) — the matcher ignores them.
 *
 * Scope rules:
 *  - the event's workspaceId must equal the automation's (when present),
 *  - record/form triggers: the automation's databaseId (or the trigger's)
 *    must equal the event's databaseId (when a database is set),
 *  - record.updated: an OPTIONAL `config.properties` list narrows the match
 *    to events whose `changedProperties` intersect it (a create event has
 *    `changedProperties = ["*"]`, which matches any list);
 *  - property.changed: `config.properties` is REQUIRED (the intersection
 *    must be non-empty).
 *
 * Dedupe: `auto:dedupe:{automationId}:{eventId}` (shared `dedupeKey`),
 * SET NX EX 600 — a redelivered stream entry (consumer-group retry) does not
 * re-fire the same automation for the same event.
 */
import type { PrismaClient } from "@prisma/client";
import type { Redis } from "ioredis";
import { Queue } from "bullmq";
import { dedupeKey, truncateJson } from "@factory/shared";
import { asJson } from "../../lib/prisma.js";
import { getLogger } from "../../logger.js";
import { QUEUES } from "../../queues.js";

const log = getLogger("automation-matcher");

/** The run row's stored `input` cap (spec: 64 KB, truncated). */
export const RUN_INPUT_MAX_BYTES = 64 * 1024;
const DEDUPE_TTL_S = 600;

const DB_TRIGGER_TYPES = new Set(["RECORD_CREATED", "RECORD_UPDATED", "PROPERTY_CHANGED", "FORM_SUBMITTED"]);

/** Build the run's stored triggerEvent (the job carries the same object). */
export function buildTriggerEvent(eventName: string, payload: Record<string, unknown>): Record<string, unknown> {
  // Flatten the event payload to the top level (recordId / databaseId /
  // changedProperties / workspaceId become first-class) AND keep `payload`
  // intact (so `event.payload.<x>` interpolation + condition refs work).
  return {
    type: eventName,
    payload,
    ...payload
  };
}

/** Does this trigger scope-match the event? */
export function triggerScopeMatches(
  trigger: { type: string; config: unknown },
  automation: { workspaceId: string; databaseId: string | null },
  payload: Record<string, unknown>
): boolean {
  const p = payload as Record<string, unknown>;
  if (typeof p.workspaceId === "string" && p.workspaceId !== automation.workspaceId) return false;

  if (DB_TRIGGER_TYPES.has(trigger.type)) {
    const dbId = automation.databaseId ?? ((trigger.config as Record<string, unknown> | null)?.databaseId as string | undefined);
    if (dbId && p.databaseId !== dbId) return false;
  }

  if (trigger.type === "RECORD_UPDATED" || trigger.type === "PROPERTY_CHANGED") {
    const want: string[] = Array.isArray((trigger.config as Record<string, unknown> | null)?.properties)
      ? ((trigger.config as Record<string, unknown>).properties as string[])
      : [];
    if (want.length > 0) {
      // Fail closed: an event WITHOUT changedProperties never fires a
      // property-scoped trigger (spec: "an event lacking changedProperties
      // never fires PROPERTY_CHANGED"). A create event carries
      // changedProperties = ["*"], which matches any list.
      const changedRaw = p.changedProperties;
      if (!Array.isArray(changedRaw)) return false;
      const changed = changedRaw as string[];
      const hit = changed.includes("*") || want.some((w) => changed.includes(w));
      if (!hit) return false;
    }
  }
  return true;
}

export interface MatcherEntry {
  /** The stream entry id (the dedupe eventId). */
  id: string;
  event: string;
  payload: Record<string, unknown>;
}

export interface MatcherDeps {
  prisma: PrismaClient;
  redis: Redis;
}

/**
 * Match one stream entry against all ACTIVE automations and enqueue the
 * resulting runs. Returns the number of runs enqueued. Never throws for a
 * single-automation failure (logged + the entry is still ACKed by the
 * consumer so one bad row cannot wedge the group).
 *
 * §7 observability: every decision bumps a Redis counter —
 * `automations:matched:{type}` and `automations:skipped:{reason}`
 * (filter_mismatch / deduped).
 */
export async function matchAndEnqueue(
  deps: MatcherDeps,
  triggerType: string,
  entry: MatcherEntry
): Promise<number> {
  const automations = await deps.prisma.automation.findMany({
    where: { status: "ACTIVE" },
    include: { triggers: { where: { type: triggerType as never } } }
  });

  const bump = (key: string) => {
    void deps.redis.incr(key).catch(() => undefined);
  };

  let enqueued = 0;
  for (const a of automations) {
    const matching = a.triggers.filter((t) => triggerScopeMatches(t, a, entry.payload));
    if (matching.length === 0) {
      bump("automations:skipped:filter_mismatch");
      continue;
    }

    // Dedupe on (automationId, eventId): a redelivery must not re-fire.
    try {
      const got = await deps.redis.set(dedupeKey(a.id, entry.id), "1", "EX", DEDUPE_TTL_S, "NX");
      if (got === null) {
        bump("automations:skipped:deduped");
        continue;
      }
    } catch (err) {
      log.warn({ err: (err as Error).message, automationId: a.id }, "dedupe check failed (proceeding)");
    }

    try {
      // Load the automation's condition tree + action count for the run's
      // stored `input` (spec: { triggerEvent (truncated), conditions,
      // actionCount }).
      const full = await deps.prisma.automation.findUnique({
        where: { id: a.id },
        select: {
          conditions: { select: { expression: true } },
          _count: { select: { actions: true } }
        }
      });
      const triggerEvent = buildTriggerEvent(entry.event, entry.payload);
      const { value: truncatedTrigger, truncated } = truncateJson(triggerEvent, RUN_INPUT_MAX_BYTES);
      const inputObj: Record<string, unknown> = {
        triggerEvent: truncated ? { ...(truncatedTrigger as Record<string, unknown>), truncated: true } : truncatedTrigger,
        conditions: full?.conditions.length ? full.conditions[0]?.expression ?? null : null,
        actionCount: full?._count.actions ?? 0
      };
      const run = await deps.prisma.automationRun.create({
        data: {
          workspaceId: a.workspaceId,
          automationId: a.id,
          status: "PENDING",
          triggerEvent: asJson(triggerEvent),
          input: asJson(inputObj),
          retryCount: 0
        }
      });
      await enqueueRunJob(deps.redis, {
        runId: run.id,
        automationId: a.id,
        workspaceId: a.workspaceId,
        triggerEvent,
        eventId: entry.id
      });
      enqueued++;
      bump(`automations:matched:${triggerType.toLowerCase()}`);
    } catch (err) {
      log.error({ err: (err as Error).message, automationId: a.id, entryId: entry.id }, "matcher enqueue failed for automation");
    }
  }
  return enqueued;
}

/** Enqueue a `run-automation` job (best-effort; the watchdog reclaims orphans). */
export async function enqueueRunJob(
  redis: Redis,
  job: { runId: string; automationId: string; workspaceId: string; triggerEvent: Record<string, unknown>; eventId: string }
): Promise<void> {
  const q = new Queue(QUEUES.automations, { connection: redis });
  try {
    await q.add("run-automation", job, { removeOnComplete: 1000, removeOnFail: 500 });
  } finally {
    await q.close();
  }
}