/**
 * apps/worker — schedule triggers (TASK-0110 §4 + TASK-0113 schedule path).
 *
 * A BullMQ repeatable job per ACTIVE SCHEDULE automation, keyed
 * `automations:run:{automationId}` (spec §4: the repeatable job id).
 * `reconcileScheduleAutomations` is a set-difference over the queue's
 * repeatables (idempotent — BullMQ upserts by name); it runs at boot and
 * every 5 min (index.ts) so API-side activate/pause edits converge even if
 * a reconcile was missed.
 *
 * When a repeatable fires, the worker handler calls `executeScheduleFire`:
 * it creates a PENDING run with `triggerEvent = { type: "schedule",
 * firedAt, cron, timeZone? }` (spec §4) and enqueues a `run-automation`
 * job — the SAME pipeline as every other trigger.
 */
import type { PrismaClient } from "@prisma/client";
import type { Redis } from "ioredis";
import { Queue } from "bullmq";
import { asJson } from "../../lib/prisma.js";
import { getLogger } from "../../logger.js";
import { QUEUES } from "../../queues.js";
import { enqueueRunJob } from "./matcher.js";
import { truncateJson } from "@factory/shared";
import type { WorkerEnv } from "../../config.js";

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

/** Run `input` storage cap (spec: 64 KB). */
const RUN_INPUT_MAX_BYTES = 64 * 1024;

/** The repeatable-job key for a schedule automation (spec §4). */
export function scheduleJobKey(automationId: string): string {
  return `automations:run:${automationId}`;
}

/** True when a job name is one of ours (repeatable schedule fires). */
export function isScheduleFireJobName(name: string): boolean {
  return name.startsWith("automations:run:");
}

export function cronFromConfig(config: unknown): string | null {
  if (config && typeof config === "object") {
    const c = config as Record<string, unknown>;
    if (typeof c.cron === "string" && c.cron.trim()) return c.cron.trim();
  }
  return null;
}

function timeZoneFromConfig(config: unknown): string | undefined {
  if (config && typeof config === "object") {
    const tz = (config as Record<string, unknown>).timeZone;
    if (typeof tz === "string" && tz.trim()) return tz.trim();
  }
  return undefined;
}

/**
 * Register repeatable jobs for active SCHEDULE automations and remove stale
 * ones (set difference by name — a shared literal name would make the
 * removal loop delete every schedule on each reconcile). Idempotent.
 */
export async function reconcileScheduleAutomations(
  connection: Redis,
  prisma: PrismaClient
): Promise<{ added: number; removed: number }> {
  const q = new Queue(QUEUES.automations, { connection });
  let added = 0;
  let removed = 0;
  try {
    const automations = await prisma.automation.findMany({
      where: { status: "ACTIVE" },
      include: { triggers: { where: { type: "SCHEDULE" } } }
    });
    const wanted = new Map<string, { automationId: string; cron: string; tz?: string }>();
    for (const a of automations) {
      for (const t of a.triggers) {
        const cron = cronFromConfig(t.config);
        if (cron) {
          const tz = timeZoneFromConfig(t.config);
          wanted.set(scheduleJobKey(a.id), { automationId: a.id, cron, ...(tz ? { tz } : {}) });
        }
      }
    }

    const existing = await q.getRepeatableJobs();
    const existingNames = new Set(existing.map((r) => r.name));
    for (const [name, spec] of wanted) {
      if (!existingNames.has(name)) {
        await q.add(
          name,
          { automationId: spec.automationId, event: "schedule", cron: spec.cron, ...(spec.tz ? { timeZone: spec.tz } : {}) },
          {
            repeat: { pattern: spec.cron, ...(spec.tz ? { tz: spec.tz } : {}) },
            removeOnComplete: true,
            removeOnFail: false
          }
        );
        added++;
        log.info({ automationId: spec.automationId, cron: spec.cron, tz: spec.tz ?? null }, "schedule registered");
      }
    }
    for (const r of existing) {
      if (!isScheduleFireJobName(r.name)) continue; // not ours
      if (!wanted.has(r.name)) {
        // Every one of our repeatables is added with a pattern (line above);
        // BullMQ's removeRepeatable requires one — skip the unpatterned
        // stragglers (another consumer's repeatable that happens to share
        // the name prefix is not ours to remove).
        if (!r.pattern) continue;
        await q.removeRepeatable(r.name, { pattern: r.pattern });
        removed++;
        log.info({ name: r.name }, "schedule removed");
      }
    }
    if (added || removed) log.info({ added, removed }, "schedule reconcile done");
  } finally {
    await q.close();
  }
  return { added, removed };
}

/**
 * A repeatable fired: create the PENDING run + enqueue the pipeline job.
 * `data` is the repeatable's payload `{ automationId, event, cron, timeZone? }`.
 * A deleted automation (row gone) is skipped — the next reconcile removes
 * the repeatable.
 */
export async function executeScheduleFire(
  connection: Redis,
  prisma: PrismaClient,
  env: WorkerEnv,
  data: { automationId?: string; cron?: string; timeZone?: string }
): Promise<{ runId: string } | null> {
  void env;
  if (!data.automationId) return null;
  const automation = await prisma.automation.findUnique({
    where: { id: data.automationId },
    select: { id: true, workspaceId: true, status: true }
  });
  if (!automation || automation.status !== "ACTIVE") {
    log.warn({ automationId: data.automationId }, "schedule fire for a missing/inactive automation (skipped)");
    return null;
  }

  const firedAt = new Date().toISOString();
  const triggerEvent: Record<string, unknown> = {
    type: "schedule",
    firedAt,
    cron: data.cron ?? null,
    ...(data.timeZone ? { timeZone: data.timeZone } : {})
  };
  const { value: truncated, truncated: truncatedFlag } = truncateJson(triggerEvent, RUN_INPUT_MAX_BYTES);
  const inputObj: Record<string, unknown> = {
    triggerEvent: truncatedFlag ? { ...(truncated as Record<string, unknown>), truncated: true } : truncated,
    conditions: null,
    actionCount: 0
  };

  const run = await prisma.automationRun.create({
    data: {
      workspaceId: automation.workspaceId,
      automationId: automation.id,
      status: "PENDING",
      triggerEvent: asJson(triggerEvent),
      input: asJson(inputObj),
      retryCount: 0
    }
  });
  await enqueueRunJob(connection, {
    runId: run.id,
    automationId: automation.id,
    workspaceId: automation.workspaceId,
    triggerEvent,
    eventId: run.id
  });
  log.info({ runId: run.id, automationId: automation.id, cron: data.cron ?? null }, "schedule fire → run enqueued");
  return { runId: run.id };
}