/**
 * apps/worker — domain event catalog + envelope.
 *
 * The API publishes domain events onto the Redis stream `factory:events`
 * (ARCHITECTURE.md §3/§9). Every entry is a JSON envelope:
 *
 *   { event: "page.created", payload: { ... }, ts: "2025-01-01T00:00:00.000Z" }
 *
 * This module is the single source of truth inside the worker for:
 *  - the event names (EVENTS catalog),
 *  - the zod envelope schema (`parseEvent`),
 *  - per-consumer subscriptions (SEARCH_INDEX_EVENTS / WEBHOOK_EVENT_CATALOG /
 *    AUTOMATION_EVENT_CATALOG) and the trigger mapping used by the
 *    automation stream consumer.
 *
 * Note: the worker declares the catalog locally (rather than importing
 * @factory/shared) so the app stays self-contained while shared keeps
 * evolving; keep both in sync when adding event names.
 */
import { z } from "zod";
import { getLogger } from "./logger.js";

const log = getLogger("events");

/** All domain events the platform can emit (ARCHITECTURE.md §9). */
export const EVENTS = [
  "page.created",
  "page.updated",
  "page.deleted",
  "page.shared",
  "database.created",
  "database.updated",
  "record.created",
  "record.updated",
  "comment.created",
  "comment.deleted",
  "notification.created",
  "form.submitted",
  "automation.executed",
  "automation.manual",
  "file.uploaded",
  // TASK-0096: worker publishes this at S3 finalization (File row mime/size
  // are set after `file.uploaded` fires at presign) → the search consumer
  // re-upserts the file doc with the final metadata.
  "file.finalized",
  // TASK-0102: the files pipeline flagged the object infected (ClamAV).
  // Stream-internal: the worker notifies workspace admins; NO search index
  // update and NO `file.uploaded` is ever emitted for this file.
  "file.virus_detected",
  // TASK-0104 §3: a file was trashed (soft delete). The search index removes
  // the doc; the object + version rows are retained until purge.
  "file.deleted",
  // TASK-0104 §8: a file's searchable fields changed (rename). The search
  // index re-upserts the file doc so the title converges immediately.
  "file.updated",
  "api_key.created",
  "user.login",
  "user.logout",
  "user.register",
  "permission.changed",
  "workspace.created"
] as const;

export type DomainEventName = (typeof EVENTS)[number];

/**
 * Stream envelope. `payload` is validated here only as a plain JSON object;
 * each consumer zod-parses its own payload shape (defense in depth).
 */
export const eventEnvelopeSchema = z.object({
  event: z.string().min(1),
  payload: z.record(z.string(), z.unknown()).default({}),
  ts: z.string().min(1)
});

export type EventEnvelope = z.infer<typeof eventEnvelopeSchema>;

/**
 * Parse one raw stream entry. Throws (zod) on a malformed envelope.
 */
export function parseEvent(raw: {
  id: string;
  fields: Record<string, string | undefined>;
}): { id: string; envelope: EventEnvelope } {
  let envelope: EventEnvelope;
  try {
    const event = raw.fields["event"];
    const payload = raw.fields["payload"];
    const ts = raw.fields["ts"];
    envelope = eventEnvelopeSchema.parse({
      event,
      payload: payload ? JSON.parse(payload) : {},
      ts: ts ?? new Date().toISOString()
    });
  } catch (err) {
    throw new Error(`poison stream entry ${raw.id}: ${(err as Error).message}`);
  }
  return { id: raw.id, envelope };
}

/**
 * Parse-or-drop: returns the envelope, or null when the entry is malformed
 * (logged). Consumers return early on null so the entry is XACKed — a poison
 * entry must not be redelivered forever.
 */
export function tryParseEvent(raw: {
  id: string;
  fields: Record<string, string | undefined>;
}): { id: string; envelope: EventEnvelope } | null {
  try {
    return parseEvent(raw);
  } catch (err) {
    log.error({ err: (err as Error).message }, "dropping poison stream entry");
    return null;
  }
}

// ---------------------------------------------------------------------------
// Per-consumer subscriptions (XREADGROUP consumer groups are per-queue, so a
// consumer only needs the events its queue acts on — everything else is
// XACKed as not-mine).
// ---------------------------------------------------------------------------

/** search-index consumer: content + permission + user/workspace events.
 * TASK-0096: user.register / user.login (with a workspaceId) upsert a
 * per-workspace user doc; workspace.created triggers the initial index. */
export const SEARCH_INDEX_EVENTS: readonly DomainEventName[] = [
  "page.created",
  "page.updated",
  "page.deleted",
  "page.shared",
  "database.created",
  "database.updated",
  "record.created",
  "record.updated",
  "comment.created",
  "comment.deleted",
  "file.uploaded",
  "file.finalized",
  "file.deleted",
  "file.updated",
  "permission.changed",
  "user.register",
  "user.login",
  "workspace.created"
];

/**
 * webhooks consumer: what owners can subscribe their endpoints to.
 *
 * TASK-0130: the SINGLE SOURCE is `@factory/shared` `WEBHOOK_EVENT_NAMES`
 * (packages/shared/src/events/catalog.ts) — the 13-event public catalog. The
 * old local list (17 names, incl. user.login/user.register/permission.changed/
 * workspace.created) is gone: those platform-internal events were never
 * offered to third-party subscribers (docs §3 lists exactly 13).
 */
import { WEBHOOK_EVENT_NAMES } from "@factory/shared";
export const WEBHOOK_EVENT_CATALOG: readonly DomainEventName[] = WEBHOOK_EVENT_NAMES as readonly DomainEventName[];

/**
 * automations consumer: events that can start an automation run.
 *
 * NOTE: `comment.created` is intentionally NOT triggerable — the Prisma
 * AutomationTriggerType enum has no COMMENT_CREATED value, and the worker
 * must not fabricate trigger types the schema doesn't support. The stream
 * consumer XACKs comment events as "not automatable" until the schema adds
 * the value (per AGENTS.md: schema wins over the brief; conflict recorded).
 */
export const AUTOMATION_EVENT_CATALOG: readonly DomainEventName[] = [
  "record.created",
  "record.updated",
  "page.created",
  "form.submitted",
  "automation.manual"
];

/** Domain event → AutomationTriggerType (for stream→trigger matching). */
export const EVENT_TO_TRIGGER: Record<string, string> = {
  "record.created": "RECORD_CREATED",
  "record.updated": "RECORD_UPDATED",
  "page.created": "PAGE_CREATED",
  "form.submitted": "FORM_SUBMITTED",
  "automation.manual": "MANUAL"
};