// App-event automations: the store (DATA_DIR/app-triggers.json), their copies on the connected-apps service kept in
// step without the person (reconciliation), and the long-poll that turns each queued event into a routine run. One
// path on desktop, headless and Cloud: Composio posts to the service, this server collects from it.
import { randomBytes, randomUUID } from "node:crypto";
import { mkdirSync, readFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { z } from "zod";

import {
  APP_FILTER_MAX,
  GMAIL_APP,
  gmailSearchQuery,
  type AppEventAttempt,
  type AppEventOutcome,
  type AppTrigger,
  type AppTriggerFilter,
  type AppTriggerService,
  type AppTriggersSnapshot,
} from "../shared/app-triggers.ts";
import { appLabel } from "./app-event-catalog.ts";
import { eventBlock, eventPreview, eventPrompt, matchesFilter, mergeEventBlock } from "./app-events.ts";
import { BrokerError, type AppTriggerBroker, type BrokerEvent, type BrokerInventory, type BrokerTriggerType } from "./app-trigger-broker.ts";
import { writeFileAtomic } from "./atomic.ts";
import { DATA_DIR } from "./config.ts";
import type { RoutineRunOn } from "./routines.ts";

export const DEFAULT_MAX_PENDING_RUNS = 3;
const MAX_ATTEMPTS = 2_000;
const MAX_DELETED = 50;
const RESTORE_WINDOW_MS = 7 * 24 * 60 * 60_000;
const POLL_WAIT_S = 25;
const IDLE_CHECK_MS = 30_000;
const POISON_TRIES = 5;
/** Reconciliation also reads the service's per-trigger drop counts, so it runs more often while events can arrive. */
const RECONCILE_ACTIVE_MS = 15 * 60_000;
const RECONCILE_IDLE_MS = 60 * 60_000;
const NAME_MAX = 80;
const PROMPT_MAX = 20_000;

export class AppTriggerError extends Error {
  readonly status: number;

  constructor(message: string, status = 400) {
    super(message);
    this.status = status;
  }
}

type ConnectedApps = Record<string, { connected: boolean; accounts: Array<{ id: string; alias?: string; status: string }> }>;
type Frame = { kind: "app-trigger"; botId: string; trigger: AppTrigger } | { kind: "app-trigger.deleted"; triggerId: string }
  | { kind: "app-trigger.attempt"; botId: string; attempt: AppEventAttempt } | { kind: "app-trigger.service"; service: AppTriggerService };

export interface AppTriggerManagerOptions {
  file?: string;
  now?: () => number;
  broker: AppTriggerBroker;
  /** Whether app events can work here at all, and why not. Checked on every loop, since access arrives late. */
  service: () => { available: true } | { available: false; reason: "self-hosted" | "unavailable" | "off" };
  connectedApps: () => Promise<ConnectedApps>;
  botState: (botId: string) => "ready" | "busy" | "missing";
  runs: {
    enqueue(input: { webhookId: string; webhookName: string; prompt: string; botId: string; runOn: RoutineRunOn; deliveryId: string; receivedAt: number; eventSource: "app" }): { id: string };
    merge(triggerId: string, eventId: string, merge: (prompt: string) => string | null): { id: string } | null;
    receipt(triggerId: string, eventId: string): { id: string } | null;
    pending(triggerId: string): number;
    cancelQueued(triggerId: string, message: string): void;
  };
  emit?: (frame: Frame) => void;
  log?: (line: string) => void;
  sleep?: (ms: number, signal: AbortSignal) => Promise<void>;
}

const filterSchema = z.object({
  from: z.string().trim().max(APP_FILTER_MAX).optional(),
  subject: z.string().trim().max(APP_FILTER_MAX).optional(),
  label: z.string().trim().max(APP_FILTER_MAX).optional(),
  text: z.string().trim().max(APP_FILTER_MAX).optional(),
}).strict();
const skippedSchema = z.object({ day: z.string(), dailyLimit: z.number(), queueFull: z.number(), expired: z.number(), tooMany: z.number() });
const dropCounts = z.record(z.string(), z.number());
const storedSchema = z.object({
  id: z.string().min(1),
  name: z.string(),
  prompt: z.string(),
  botId: z.string().min(1),
  runOn: z.enum(["maus", "cloud"]),
  enabled: z.boolean(),
  app: z.string().min(1),
  appName: z.string(),
  event: z.string().min(1),
  eventName: z.string(),
  delivery: z.enum(["instant", "poll"]),
  accountId: z.string().min(1),
  accountLabel: z.string().optional(),
  config: z.record(z.string(), z.unknown()),
  filter: filterSchema.optional().catch(undefined),
  remote: z.object({ triggerId: z.string().optional(), state: z.enum(["active", "pending", "error"]), error: z.string().optional(), syncedAt: z.number().optional() }),
  problem: z.union([
    z.object({ kind: z.literal("reconnect"), at: z.number() }),
    z.object({ kind: z.literal("error"), text: z.string(), at: z.number() }),
  ]).optional().catch(undefined),
  sourceThreadId: z.string().optional(),
  received: z.number().int().nonnegative(),
  filtered: z.number().int().nonnegative(),
  skipped: skippedSchema.optional().catch(undefined),
  lastEventAt: z.number().optional(),
  lastRunId: z.string().optional(),
  createdAt: z.number(),
  updatedAt: z.number(),
  /** The service's cumulative drop counts as last seen, and as they stood when this UTC day began. */
  drops: z.object({ day: z.string(), base: dropCounts, last: dropCounts }).optional().catch(undefined),
});
type StoredTrigger = z.output<typeof storedSchema>;
const attemptSchema = z.object({
  id: z.string(),
  triggerId: z.string(),
  eventId: z.string(),
  at: z.number(),
  outcome: z.enum(["started", "joined", "filtered", "duplicate", "skipped", "rejected"]),
  reason: z.string().optional(),
  runId: z.string().optional(),
  preview: z.string().optional(),
});
const fileSchema = z.object({
  version: z.literal(1),
  pollerId: z.string().regex(/^[A-Za-z0-9_-]{8,64}$/),
  cursor: z.number().int().nonnegative(),
  triggers: z.array(storedSchema),
  attempts: z.array(attemptSchema).default([]),
  deleted: z.array(z.object({ at: z.number(), trigger: storedSchema })).default([]),
});

export interface AppTriggerInput {
  botId: string;
  name: string;
  prompt: string;
  app: string;
  event: string;
  account?: string;
  filter?: AppTriggerFilter;
  config?: Record<string, unknown>;
  runOn?: RoutineRunOn;
  enabled?: boolean;
  sourceThreadId?: string;
  /** Another server's automation, made here under its own id. */
  id?: string;
}

export interface AppTriggerPatch {
  name?: string;
  prompt?: string;
  runOn?: RoutineRunOn;
  filter?: AppTriggerFilter | null;
  config?: Record<string, unknown>;
  enabled?: boolean;
  /** Undo: the config exactly as it was saved, filter already folded in. */
  restoreConfig?: Record<string, unknown>;
}

const utcDay = (at: number) => new Date(at).toISOString().slice(0, 10);
const active = (status: string) => /^active$/i.test(status);

function cleanFilter(filter: AppTriggerFilter | null | undefined): AppTriggerFilter | undefined {
  if (!filter) return undefined;
  const parsed = filterSchema.safeParse(filter);
  if (!parsed.success) throw new AppTriggerError(`The filter takes from, subject, label and text, each at most ${APP_FILTER_MAX} characters.`);
  const kept = Object.fromEntries(Object.entries(parsed.data).filter(([, value]) => value)) as AppTriggerFilter;
  return Object.keys(kept).length ? kept : undefined;
}

/** The trigger config Composio gets: the fields the event type takes, with Gmail's plain filter as its search. */
function triggerConfig(type: BrokerTriggerType, config: Record<string, unknown>, filter: AppTriggerFilter | undefined): Record<string, unknown> {
  const properties = (type.config.properties ?? {}) as Record<string, { default?: unknown }>;
  const known = Object.keys(properties);
  const unknown = Object.keys(config).filter((key) => known.length > 0 && !known.includes(key));
  if (unknown.length) throw new AppTriggerError(`${type.name} has no setting ${unknown.join(", ")}. Its settings are: ${known.join(", ") || "none"}.`);
  const next: Record<string, unknown> = { ...config };
  if (type.toolkit === GMAIL_APP && filter) {
    const query = [typeof next.query === "string" ? next.query.trim() : "", gmailSearchQuery(filter)].filter(Boolean).join(" ");
    if (query) next.query = query;
  }
  const required = Array.isArray(type.config.required) ? type.config.required.filter((key): key is string => typeof key === "string") : [];
  const missing = required.filter((key) => next[key] === undefined && properties[key]?.default === undefined);
  if (missing.length) throw new AppTriggerError(`${type.name} needs ${missing.join(", ")} in its settings (event_config).`);
  return next;
}

function publicTrigger(stored: StoredTrigger): AppTrigger {
  const { drops: _drops, ...trigger } = stored;
  return structuredClone(trigger);
}

export class AppTriggerManager {
  private readonly file: string;
  private readonly now: () => number;
  private readonly options: AppTriggerManagerOptions;
  private triggers: StoredTrigger[] = [];
  private attemptLog: AppEventAttempt[] = [];
  private deleted: Array<{ at: number; trigger: StoredTrigger }> = [];
  private pollerId = randomBytes(16).toString("base64url");
  private cursor = 0;
  private unreadable = false;
  private usage: { plan: "desktop" | "cloud"; eventsPerDay: number; eventsToday: number } | null = null;
  private serial: Promise<unknown> = Promise.resolve();
  private failures = new Map<string, number>();
  private stopped = true;
  private abort = new AbortController();
  private nextReconcileAt = 0;
  private loop: Promise<void> | null = null;
  /** A deleted automation's service copy could not be removed yet; reconciliation removes it. */
  private releasePending = false;

  constructor(options: AppTriggerManagerOptions) {
    this.options = options;
    this.file = options.file ?? join(DATA_DIR, "app-triggers.json");
    this.now = options.now ?? Date.now;
    try {
      const parsed = fileSchema.safeParse(JSON.parse(readFileSync(this.file, "utf8")));
      if (!parsed.success) throw parsed.error;
      this.pollerId = parsed.data.pollerId;
      this.cursor = parsed.data.cursor;
      this.triggers = parsed.data.triggers;
      this.attemptLog = parsed.data.attempts.slice(-MAX_ATTEMPTS);
      this.deleted = parsed.data.deleted.filter((entry) => entry.at >= this.now() - RESTORE_WINDOW_MS);
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
        this.unreadable = true;
        console.error(`app triggers: ignoring unreadable ${this.file}: ${error instanceof SyntaxError ? "invalid JSON" : error instanceof Error ? error.message : "unreadable"}`);
      }
    }
  }

  list(): AppTrigger[] {
    return this.triggers.map(publicTrigger);
  }

  get(id: string): AppTrigger | null {
    const stored = this.triggers.find((trigger) => trigger.id === id);
    return stored ? publicTrigger(stored) : null;
  }

  attempts(): AppEventAttempt[] {
    return this.attemptLog.map((attempt) => ({ ...attempt }));
  }

  service(): AppTriggerService {
    const service = this.options.service();
    if (!service.available) return { available: false, reason: service.reason };
    return { available: true, ...this.usage };
  }

  snapshot(): AppTriggersSnapshot {
    return { triggers: this.list(), attempts: this.attempts(), service: this.service() };
  }

  /** Validated against the event type and the person's connected accounts, then saved and copied to the service. A
   *  service that cannot be reached leaves it pending; reconciliation finishes it. */
  async create(input: AppTriggerInput): Promise<AppTrigger> {
    this.assertWritable();
    this.assertAvailable();
    const name = input.name.trim().slice(0, NAME_MAX);
    const prompt = input.prompt.trim();
    if (!name) throw new AppTriggerError("An automation needs a name.");
    if (prompt.length > PROMPT_MAX) throw new AppTriggerError(`Instructions can be at most ${PROMPT_MAX} characters.`);
    if (this.options.botState(input.botId) === "missing") throw new AppTriggerError("That bot no longer exists.", 404);
    const app = input.app.trim().toLowerCase();
    const type = await this.remote(() => this.options.broker.triggerType(input.event.trim()));
    if (!type) throw new AppTriggerError(`${input.event} is not an event OpenMausBot can watch. Call trigger_info action=events to see the events this app offers.`, 404);
    if (type.toolkit !== app) throw new AppTriggerError(`${type.slug} is a ${appLabel(type.toolkit)} event, not ${appLabel(app)}.`);
    const account = await this.resolveAccount(app, input.account);
    const filter = cleanFilter(input.filter);
    const config = triggerConfig(type, input.config ?? {}, filter);
    const duplicate = this.triggers.find((trigger) => trigger.enabled && trigger.botId === input.botId && trigger.event === type.slug &&
      trigger.accountId === account.id && trigger.prompt === prompt && JSON.stringify(trigger.config) === JSON.stringify(config));
    if (duplicate) throw new AppTriggerError(`The same automation already exists (${duplicate.id}). Use trigger_info to review it, then change that one instead.`, 409);
    const now = this.now();
    const trigger: StoredTrigger = {
      id: input.id ?? randomUUID(), name, prompt, botId: input.botId, runOn: input.runOn ?? "maus", enabled: input.enabled ?? true,
      app, appName: appLabel(app), event: type.slug, eventName: type.name || type.slug, delivery: type.type === "poll" ? "poll" : "instant",
      accountId: account.id, ...(account.alias ? { accountLabel: account.alias } : {}), config, ...(filter ? { filter } : {}),
      remote: { state: "pending" }, ...(input.sourceThreadId ? { sourceThreadId: input.sourceThreadId } : {}),
      received: 0, filtered: 0, createdAt: now, updatedAt: now,
    };
    await this.exclusive(async () => {
      try {
        await this.bind(trigger);
        if (!trigger.enabled) await this.applyEnabled(trigger.remote.triggerId);
      } catch (error) {
        // Composio refused this config: nothing to save. Anything else is retried by reconciliation.
        if (error instanceof BrokerError && !error.transient && error.status !== 403) throw new AppTriggerError(error.message, 400);
        this.markRemoteFailure(trigger, error);
      }
      this.triggers.unshift(trigger);
      this.save();
    });
    this.emitTrigger(trigger);
    this.wake();
    return publicTrigger(trigger);
  }

  /** Another server's automation under its own id, against the same connected account: a new one starts switched off;
   * one already here keeps its switch and takes the rest. */
  async adopt(record: AppTrigger): Promise<AppTrigger> {
    const config = Object.fromEntries(Object.entries(record.config).filter(([key, value]) => value !== undefined && !(record.app === GMAIL_APP && key === "query")));
    if (this.triggers.some((candidate) => candidate.id === record.id)) {
      return (await this.update(record.id, { name: record.name, prompt: record.prompt, runOn: record.runOn, filter: record.filter ?? null, config }))!;
    }
    return this.create({ id: record.id, botId: record.botId, name: record.name, prompt: record.prompt, app: record.app, event: record.event, account: record.accountId,
      ...(record.filter ? { filter: record.filter } : {}), config, runOn: record.runOn, enabled: false, ...(record.sourceThreadId ? { sourceThreadId: record.sourceThreadId } : {}) });
  }

  async update(id: string, patch: AppTriggerPatch): Promise<AppTrigger | null> {
    this.assertWritable();
    const trigger = this.triggers.find((candidate) => candidate.id === id);
    if (!trigger) return null;
    const name = patch.name === undefined ? trigger.name : patch.name.trim().slice(0, NAME_MAX);
    if (!name) throw new AppTriggerError("An automation needs a name.");
    if (patch.prompt !== undefined && patch.prompt.trim().length > PROMPT_MAX) throw new AppTriggerError(`Instructions can be at most ${PROMPT_MAX} characters.`);
    const filter = patch.filter === undefined ? trigger.filter : cleanFilter(patch.filter);
    const reconfigure = patch.filter !== undefined || patch.config !== undefined || patch.restoreConfig !== undefined;
    let config = patch.restoreConfig ?? trigger.config;
    if (reconfigure && !patch.restoreConfig) {
      this.assertAvailable();
      const type = await this.remote(() => this.options.broker.triggerType(trigger.event));
      if (!type) throw new AppTriggerError(`${trigger.eventName} can no longer be watched; delete this automation.`, 410);
      const base = patch.config ?? (trigger.app === GMAIL_APP ? { ...trigger.config, query: undefined } : trigger.config);
      config = triggerConfig(type, Object.fromEntries(Object.entries(base).filter(([, value]) => value !== undefined)), filter);
    }
    await this.exclusive(async () => {
      const previous = trigger.remote.triggerId;
      const previousConfig = JSON.stringify(trigger.config);
      Object.assign(trigger, {
        name,
        prompt: patch.prompt === undefined ? trigger.prompt : patch.prompt.trim(),
        runOn: patch.runOn ?? trigger.runOn,
        enabled: patch.enabled ?? trigger.enabled,
        config,
        updatedAt: this.now(),
      });
      if (filter) trigger.filter = filter;
      else delete trigger.filter;
      if (patch.enabled === false) this.options.runs.cancelQueued(trigger.id, "The automation was paused before this event's run started");
      try {
        if (reconfigure && (!previous || JSON.stringify(config) !== previousConfig)) {
          trigger.remote = { state: "pending" };
          await this.bind(trigger);
          await this.release(previous);
        }
        await this.applyEnabled(trigger.remote.triggerId);
      } catch (error) {
        this.markRemoteFailure(trigger, error);
      }
      this.save();
    });
    this.emitTrigger(trigger);
    this.wake();
    return publicTrigger(trigger);
  }

  async remove(id: string): Promise<boolean> {
    this.assertWritable();
    const trigger = this.triggers.find((candidate) => candidate.id === id);
    if (!trigger) return false;
    await this.exclusive(async () => {
      this.triggers = this.triggers.filter((candidate) => candidate !== trigger);
      this.deleted = [...this.deleted.filter((entry) => entry.at >= this.now() - RESTORE_WINDOW_MS && entry.trigger.id !== id),
        { at: this.now(), trigger: structuredClone(trigger) }].slice(-MAX_DELETED);
      this.options.runs.cancelQueued(id, "The automation was deleted before this event's run started");
      this.save();
      try {
        await this.release(trigger.remote.triggerId);
      } catch (error) {
        this.log(`could not remove ${trigger.remote.triggerId ?? id} from the connected-apps service yet: ${String(error)}`);
        this.releasePending = true;
        this.nextReconcileAt = this.now();
      }
    });
    this.options.emit?.({ kind: "app-trigger.deleted", triggerId: id });
    return true;
  }

  /** Brings back one deleted in the last week, with its settings; the service copy is made again. */
  async restore(id: string): Promise<AppTrigger | null> {
    this.assertWritable();
    const entry = this.deleted.find((candidate) => candidate.trigger.id === id && candidate.at >= this.now() - RESTORE_WINDOW_MS);
    if (!entry || this.triggers.some((trigger) => trigger.id === id)) return null;
    if (this.options.botState(entry.trigger.botId) === "missing") throw new AppTriggerError("The assigned bot no longer exists.", 410);
    const trigger: StoredTrigger = { ...structuredClone(entry.trigger), remote: { state: "pending" }, updatedAt: this.now() };
    await this.exclusive(async () => {
      try { await this.bind(trigger); await this.applyEnabled(trigger.remote.triggerId); }
      catch (error) { this.markRemoteFailure(trigger, error); }
      this.deleted = this.deleted.filter((candidate) => candidate !== entry);
      this.triggers.unshift(trigger);
      this.save();
    });
    this.emitTrigger(trigger);
    this.wake();
    return publicTrigger(trigger);
  }

  /** A test event marked as one, sent through the service's real queue when the automation has its copy there. */
  async runNow(id: string): Promise<{ via: "service" | "here"; runId?: string } | null> {
    const trigger = this.triggers.find((candidate) => candidate.id === id);
    if (!trigger) return null;
    const data = { omb_test: true, note: "A test run started from OpenMausBot." };
    if (trigger.remote.triggerId && trigger.remote.state === "active" && this.options.service().available) {
      try {
        await this.options.broker.testEvent(trigger.remote.triggerId, data);
        this.wake();
        return { via: "service" };
      } catch (error) {
        if (!(error instanceof BrokerError) || !error.transient) throw new AppTriggerError(error instanceof Error ? error.message : String(error), 502);
      }
    }
    const eventId = `test_${randomUUID()}`;
    const run = this.options.runs.enqueue({
      webhookId: trigger.id, webhookName: trigger.name, botId: trigger.botId, runOn: trigger.runOn, deliveryId: eventId, receivedAt: this.now(),
      eventSource: "app", prompt: eventPrompt(trigger.prompt, eventBlock(trigger, { id: eventId, receivedAt: this.now(), test: true, data })),
    });
    return { via: "here", runId: run.id };
  }

  start(): void {
    if (!this.stopped) return;
    this.stopped = false;
    this.abort = new AbortController();
    this.nextReconcileAt = 0;
    this.loop = this.run().catch((error) => this.log(`app events stopped: ${error instanceof Error ? error.message : String(error)}`));
  }

  async stop(): Promise<void> {
    this.stopped = true;
    this.abort.abort();
    await this.loop;
    this.loop = null;
  }

  /** Cuts the current wait short: a trigger changed, or access to the service arrived. */
  wake(): void {
    const previous = this.abort;
    this.abort = new AbortController();
    previous.abort();
  }

  /** Compares every automation with the service's copy and fixes what differs: a missing copy is made, one nobody
   *  wants is deleted, and on/off is matched. Idempotent; the person is never asked to set anything up again. */
  reconcile(): Promise<void> {
    return this.exclusive(() => this.reconcileNow());
  }

  /** Handles one batch in order. The cursor (the service's ack) moves only after the runs and this file are saved:
   *  a crash before that brings the same events back, and each event's run receipt stops a second run. */
  receive(events: BrokerEvent[], cursor: number): void {
    let next = this.cursor;
    try {
      for (const event of events) {
        try {
          this.handle(event);
        } catch (error) {
          const tries = (this.failures.get(event.id) ?? 0) + 1;
          this.failures.set(event.id, tries);
          if (tries < POISON_TRIES) throw error;
          this.failures.delete(event.id);
          for (const trigger of this.bound(event.triggerId)) {
            this.record(trigger, event, "rejected", { reason: `Could not be handled: ${error instanceof Error ? error.message : String(error)}`.slice(0, 300) });
          }
        }
        next = event.seq;
      }
      next = Math.max(next, cursor);
    } finally {
      this.save({ cursor: next });
      this.cursor = next;
    }
  }

  private handle(event: BrokerEvent): void {
    if (event.kind === "account_expired") {
      const accountId = event.event.data?.connected_account_id;
      for (const trigger of this.triggers.filter((candidate) => candidate.accountId === accountId)) this.setProblem(trigger, { kind: "reconnect", at: this.now() });
      return;
    }
    if (event.kind === "trigger_disabled") {
      for (const trigger of this.bound(event.triggerId)) this.setProblem(trigger, { kind: "reconnect", at: this.now() });
      return;
    }
    const data = event.event.data ?? {};
    for (const trigger of this.bound(event.triggerId)) {
      if (!trigger.enabled) {
        this.record(trigger, event, "skipped", { reason: "paused" });
        continue;
      }
      if (this.options.botState(trigger.botId) === "missing") {
        this.record(trigger, event, "rejected", { reason: "its bot was deleted" });
        continue;
      }
      if (!matchesFilter(trigger.app, trigger.filter, data)) {
        trigger.filtered += 1;
        this.record(trigger, event, "filtered", { reason: "did not match the filter" });
        continue;
      }
      const seen = this.options.runs.receipt(trigger.id, event.id);
      if (seen) {
        this.record(trigger, event, "duplicate", { runId: seen.id });
        continue;
      }
      const block = eventBlock(trigger, { id: event.id, receivedAt: event.receivedAt, test: event.kind === "test", data });
      const joined = this.options.runs.merge(trigger.id, event.id, (prompt) => mergeEventBlock(prompt, block));
      let runId = joined?.id;
      if (!runId) {
        if (this.options.runs.pending(trigger.id) >= DEFAULT_MAX_PENDING_RUNS) {
          this.countSkipped(trigger, "tooMany");
          this.record(trigger, event, "skipped", { reason: `it already had ${DEFAULT_MAX_PENDING_RUNS} runs waiting` });
          continue;
        }
        runId = this.options.runs.enqueue({
          webhookId: trigger.id, webhookName: trigger.name, prompt: eventPrompt(trigger.prompt, block), botId: trigger.botId,
          runOn: trigger.runOn, deliveryId: event.id, receivedAt: event.receivedAt, eventSource: "app",
        }).id;
      }
      trigger.received += 1;
      trigger.lastEventAt = event.receivedAt;
      trigger.lastRunId = runId;
      if (trigger.problem?.kind === "reconnect" && event.kind === "message") delete trigger.problem;
      this.record(trigger, event, joined ? "joined" : "started", { runId });
    }
  }

  private async run(): Promise<void> {
    let failures = 0;
    while (!this.stopped) {
      const signal = this.abort.signal;
      try {
        if (this.options.service().available && this.now() >= this.nextReconcileAt && (this.triggers.length > 0 || this.releasePending)) {
          this.nextReconcileAt = this.now() + (this.pollable() ? RECONCILE_ACTIVE_MS : RECONCILE_IDLE_MS);
          await this.reconcile().catch((error) => {
            this.nextReconcileAt = this.now() + Math.min(RECONCILE_ACTIVE_MS, 30_000 * 2 ** Math.min(failures, 5));
            this.log(`reconciliation failed: ${error instanceof Error ? error.message : String(error)}`);
          });
        }
        if (!this.pollable()) {
          await this.sleep(IDLE_CHECK_MS, signal);
          continue;
        }
        await this.pollOnce(signal);
        failures = 0;
      } catch (error) {
        if (this.stopped) break;
        if (signal.aborted) continue;
        failures += 1;
        const retry = error instanceof BrokerError && error.status === 409 && error.retryAfterS ? error.retryAfterS * 1000 : 0;
        await this.sleep(retry || Math.min(60_000, 1_000 * 2 ** Math.min(failures, 6)), this.abort.signal);
      }
    }
  }

  /** One long-poll: acks everything handled so far (the cursor) and handles what comes back. */
  async pollOnce(signal?: AbortSignal, waitS = POLL_WAIT_S): Promise<number> {
    const { events, cursor } = await this.options.broker.poll({ after: this.cursor, waitS, poller: this.pollerId, signal });
    if (events.length || cursor > this.cursor) this.receive(events, cursor);
    return events.length;
  }

  private pollable(): boolean {
    return this.options.service().available && this.triggers.some((trigger) => trigger.enabled && trigger.remote.triggerId);
  }

  private async reconcileNow(): Promise<void> {
    if (!this.options.service().available) return;
    const inventory = await this.options.broker.inventory();
    this.usage = { plan: inventory.limits.plan, eventsPerDay: inventory.limits.eventsPerDay, eventsToday: inventory.usage.eventsToday };
    const accounts = await this.options.connectedApps().catch(() => null);
    const bindings = new Map(inventory.triggers.map((binding) => [binding.triggerId, binding]));
    const changed = new Set<StoredTrigger>();
    for (const trigger of this.triggers) {
      const before = JSON.stringify(trigger);
      try {
        await this.reconcileOne(trigger, bindings, accounts);
      } catch (error) {
        this.markRemoteFailure(trigger, error);
      }
      this.noteDrops(trigger, trigger.remote.triggerId ? bindings.get(trigger.remote.triggerId)?.dropped : undefined);
      if (JSON.stringify(trigger) !== before) changed.add(trigger);
    }
    const wanted = new Set(this.triggers.map((trigger) => trigger.remote.triggerId).filter((id): id is string => Boolean(id)));
    for (const triggerId of wanted) {
      const binding = bindings.get(triggerId);
      const desired = this.wantsEnabled(triggerId);
      if (binding && (binding.enabled !== desired || (desired && binding.remote?.state === "disabled"))) {
        await this.options.broker.setEnabled(triggerId, desired).catch((error) => this.log(`could not switch ${triggerId}: ${String(error)}`));
      }
    }
    const orphans = [...inventory.triggers.map((binding) => binding.triggerId), ...(inventory.unbound ?? []).map((instance) => instance.triggerId)]
      .filter((triggerId) => !wanted.has(triggerId));
    let removed = true;
    for (const triggerId of new Set(orphans)) {
      await this.options.broker.remove(triggerId).catch((error) => {
        removed = false;
        this.log(`could not remove ${triggerId}: ${String(error)}`);
      });
    }
    if (removed && inventory.unbound !== null) this.releasePending = false;
    this.save();
    for (const trigger of changed) this.emitTrigger(trigger);
    this.options.emit?.({ kind: "app-trigger.service", service: this.service() });
  }

  private async reconcileOne(trigger: StoredTrigger, bindings: Map<string, BrokerInventory["triggers"][number]>, accounts: ConnectedApps | null): Promise<void> {
    // Its bot is gone: paused here, so Composio stops sending what no one would run.
    if (trigger.enabled && this.options.botState(trigger.botId) === "missing") trigger.enabled = false;
    const account = accounts?.[trigger.app]?.accounts.find((candidate) => candidate.id === trigger.accountId);
    if (accounts && (!account || !active(account.status))) {
      if (trigger.problem?.kind !== "reconnect") trigger.problem = { kind: "reconnect", at: this.now() };
      return;
    }
    if (account && trigger.problem?.kind === "reconnect") delete trigger.problem;
    const binding = trigger.remote.triggerId ? bindings.get(trigger.remote.triggerId) : undefined;
    if (!binding || binding.remote?.state === "missing") {
      trigger.remote = { state: "pending" };
      await this.bind(trigger);
      if (!trigger.enabled && trigger.remote.triggerId) await this.options.broker.setEnabled(trigger.remote.triggerId, false);
      return;
    }
    trigger.remote = { triggerId: binding.triggerId, state: "active", syncedAt: this.now() };
    if (trigger.problem?.kind === "error") delete trigger.problem;
  }

  /** The service's drop counts never reset; today's share is what grew since the UTC day began. */
  private noteDrops(trigger: StoredTrigger, dropped: Record<string, { count: number; lastAt: number }> | undefined): void {
    if (!dropped) return;
    const day = utcDay(this.now());
    const dayStart = Date.parse(`${day}T00:00:00Z`);
    const counts = Object.fromEntries(Object.entries(dropped).map(([reason, drop]) => [reason, drop.count]));
    const base = trigger.drops?.day === day ? trigger.drops.base
      : trigger.drops ? trigger.drops.last
        : Object.fromEntries(Object.entries(dropped).map(([reason, drop]) => [reason, drop.lastAt >= dayStart ? 0 : drop.count]));
    trigger.drops = { day, base, last: counts };
    const today = (reason: string) => Math.max(0, (counts[reason] ?? 0) - (base[reason] ?? 0));
    const tooMany = trigger.skipped?.day === day ? trigger.skipped.tooMany : 0;
    trigger.skipped = { day, dailyLimit: today("daily_limit"), queueFull: today("queue_full"), expired: today("expired"), tooMany };
  }

  private countSkipped(trigger: StoredTrigger, reason: "tooMany"): void {
    const day = utcDay(this.now());
    const current = trigger.skipped?.day === day ? trigger.skipped : { day, dailyLimit: 0, queueFull: 0, expired: 0, tooMany: 0 };
    trigger.skipped = { ...current, [reason]: current[reason] + 1 };
  }

  private async resolveAccount(app: string, wanted: string | undefined): Promise<{ id: string; alias?: string }> {
    let apps: ConnectedApps;
    try {
      apps = await this.options.connectedApps();
    } catch (error) {
      throw new AppTriggerError(`Connected apps could not be checked: ${error instanceof Error ? error.message : String(error)}`, 503);
    }
    const accounts = (apps[app]?.accounts ?? []).filter((account) => active(account.status));
    const label = appLabel(app);
    if (!accounts.length) throw new AppTriggerError(`${label} is not connected. Ask the person to connect ${label} first (COMPOSIO_MANAGE_CONNECTIONS shows them a card for it), then create the automation.`, 409);
    if (wanted?.trim()) {
      const key = wanted.trim().toLowerCase();
      const match = accounts.find((account) => account.id === wanted.trim() || account.alias?.toLowerCase() === key);
      if (!match) throw new AppTriggerError(`No connected ${label} account is called ${wanted}. The accounts are: ${accounts.map((account) => account.alias ?? account.id).join(", ")}.`);
      return match;
    }
    if (accounts.length > 1) {
      throw new AppTriggerError(`The person has ${accounts.length} ${label} accounts (${accounts.map((account) => account.alias ?? account.id).join(", ")}). Ask which one, then pass account.`);
    }
    return accounts[0]!;
  }

  private async bind(trigger: StoredTrigger): Promise<void> {
    if (this.serviceOutOfReach()) return;
    const { triggerId } = await this.options.broker.create({ slug: trigger.event, connectedAccountId: trigger.accountId, config: trigger.config });
    trigger.remote = { triggerId, state: "active", syncedAt: this.now() };
    if (trigger.problem?.kind === "error") delete trigger.problem;
  }

  /** Several automations with the same settings share one service copy; it goes only when the last one does. */
  private async release(triggerId: string | undefined): Promise<void> {
    if (!triggerId || this.triggers.some((trigger) => trigger.remote.triggerId === triggerId)) return;
    if (this.serviceOutOfReach()) {
      this.releasePending = true;
      return;
    }
    await this.options.broker.remove(triggerId);
  }

  private async applyEnabled(triggerId: string | undefined): Promise<void> {
    if (triggerId && !this.serviceOutOfReach()) await this.options.broker.setEnabled(triggerId, this.wantsEnabled(triggerId));
  }

  /** The owner's own key, or the service switched off. A service merely not reached yet still gets its calls and retries. */
  private serviceOutOfReach(): boolean {
    const service = this.options.service();
    return !service.available && service.reason !== "unavailable";
  }

  private wantsEnabled(triggerId: string): boolean {
    return this.triggers.some((trigger) => trigger.remote.triggerId === triggerId && trigger.enabled);
  }

  private markRemoteFailure(trigger: StoredTrigger, error: unknown): void {
    const message = error instanceof Error ? error.message : String(error);
    if (error instanceof BrokerError && error.status === 403) {
      trigger.problem = { kind: "reconnect", at: this.now() };
      trigger.remote = { ...trigger.remote, state: "error", error: message };
    } else if (error instanceof BrokerError && !error.transient) {
      trigger.problem = { kind: "error", text: message.slice(0, 300), at: this.now() };
      trigger.remote = { ...trigger.remote, state: "error", error: message };
    } else {
      trigger.remote = { ...trigger.remote, state: "pending", error: message };
      this.nextReconcileAt = Math.min(this.nextReconcileAt || Infinity, this.now() + 60_000);
    }
  }

  private setProblem(trigger: StoredTrigger, problem: NonNullable<StoredTrigger["problem"]>): void {
    trigger.problem = problem;
    this.emitTrigger(trigger);
  }

  private bound(triggerId: string | null): StoredTrigger[] {
    return triggerId ? this.triggers.filter((trigger) => trigger.remote.triggerId === triggerId) : [];
  }

  private record(trigger: StoredTrigger, event: BrokerEvent, outcome: AppEventOutcome, details: { reason?: string; runId?: string } = {}): void {
    const preview = eventPreview(event.event.data ?? {});
    const attempt: AppEventAttempt = {
      id: randomUUID(), triggerId: trigger.id, eventId: event.id.slice(0, 200), at: event.receivedAt, outcome,
      ...(details.reason ? { reason: details.reason } : {}), ...(details.runId ? { runId: details.runId } : {}), ...(preview ? { preview } : {}),
    };
    this.attemptLog.push(attempt);
    if (this.attemptLog.length > MAX_ATTEMPTS) this.attemptLog.splice(0, this.attemptLog.length - MAX_ATTEMPTS);
    this.options.emit?.({ kind: "app-trigger.attempt", botId: trigger.botId, attempt: { ...attempt } });
    this.emitTrigger(trigger);
  }

  private async remote<T>(work: () => Promise<T>): Promise<T> {
    try {
      return await work();
    } catch (error) {
      if (error instanceof BrokerError) throw new AppTriggerError(error.transient ? `The connected-apps service could not be reached (${error.message}). Try again in a minute.` : error.message, error.transient ? 503 : 400);
      throw error;
    }
  }

  private exclusive<T>(work: () => Promise<T>): Promise<T> {
    const next = this.serial.then(work, work);
    this.serial = next.catch(() => undefined);
    return next;
  }

  private assertAvailable(): void {
    const service = this.options.service();
    if (service.available) return;
    throw new AppTriggerError(service.reason === "self-hosted"
      ? "App events need OpenMausBot's connected apps. This server uses its own Composio key, so it cannot watch apps yet: use a schedule that checks the app, or a webhook."
      : "Connected apps are not set up here, so app events are unavailable.", 409);
  }

  private assertWritable(): void {
    if (this.unreadable) throw new AppTriggerError("Saved app automations could not be read. Repair app-triggers.json before changing them.", 503);
  }

  private emitTrigger(trigger: StoredTrigger): void {
    this.options.emit?.({ kind: "app-trigger", botId: trigger.botId, trigger: publicTrigger(trigger) });
  }

  private sleep(ms: number, signal: AbortSignal): Promise<void> {
    if (this.options.sleep) return this.options.sleep(ms, signal);
    return new Promise((resolve) => {
      if (signal.aborted) return resolve();
      const timer = setTimeout(resolve, ms);
      timer.unref?.();
      signal.addEventListener("abort", () => { clearTimeout(timer); resolve(); }, { once: true });
    });
  }

  private log(line: string): void {
    (this.options.log ?? ((text) => console.warn(`app events: ${text}`)))(line);
  }

  private save(overrides: { cursor?: number } = {}): void {
    if (this.unreadable) return;
    mkdirSync(dirname(this.file), { recursive: true });
    writeFileAtomic(this.file, JSON.stringify({
      version: 1, pollerId: this.pollerId, cursor: overrides.cursor ?? this.cursor, triggers: this.triggers, attempts: this.attemptLog, deleted: this.deleted,
    } satisfies z.input<typeof fileSchema>, null, 2), { mode: 0o600 });
  }
}