/**
 * apps/worker — notification → email decision (TASK-0094 R6).
 *
 * The API creates a `Notification` row (channel IN_APP) and publishes
 * `notification.created` to the domain-event stream. THIS consumer owns the
 * whole email channel — the API holds no BullMQ (architecture: the worker owns
 * every queue). On `notification.created` it decides, per user preference:
 *
 *   skip            — unknown type / already emailed (`emailSentAt`) / user
 *                     gone / self-notification (`actorId === userId`) /
 *                     pref "off"
 *   ENQUEUE (instant) — pref "instant" or "all": one email job on the `email`
 *                     queue, jobId `email_{notificationId}` (the BullMQ
 *                     job-key dedup — a double-enqueue produces ONE job, so a
 *                     double-send is impossible at the queue layer). NO colon:
 *                     BullMQ 5.x rejects custom jobIds containing `:` unless
 *                     exactly 3-segment (repeatable compat) — the old
 *                     `email:{id}` threw on every enqueue.
 *   DIGEST          — pref "digest": left for the hourly digest job
 *
 * The HOURLY DIGEST (a repeatable `maintenance` job, `notification-digest`)
 * collects each user's unsent, digest-eligible notifications (≤ 20 per email,
 * "n more" inbox link for the rest), sends one email per (user, workspace),
 * and stamps the collected rows `emailSentAt` so the next run never re-sends
 * them. At-least-once delivery is why the double-send guard is the
 * `emailSentAt` column + the job key, NOT the job key alone: a job redelivered
 * after a partial SMTP failure re-runs the processor, and only the durable
 * stamp (checked here at decision time AND by the email worker before sending)
 * closes that window. (Recorded conflict: the spec's Database-Changes note
 * assumed the job key was sufficient and that a Phase-1 delivery log existed —
 * it does not; see the task file.)
 *
 * Deep links are ABSOLUTE (`PUBLIC_URL` env — the spec's name; the API calls
 * the same value `WEB_ORIGIN`, recorded) and require login on the web side
 * (a link to a deleted target → the standard 404, not a data leak).
 */
import type { Redis } from "ioredis";
import { z } from "zod";
import type { PrismaClient } from "@prisma/client";
import {
  NOTIFICATION_TYPES,
  mergeNotificationPrefs,
  prefIsOff,
  prefSendsInstant,
  type NotificationTypeValue
} from "@factory/shared";
import { QUEUES, enqueue } from "../queues.js";
import { StreamConsumer, type StreamEntry } from "../lib/redis-stream.js";
import { tryParseEvent } from "../events.js";
import { getLogger } from "../logger.js";

const log = getLogger("email-decision");

// ---------------------------------------------------------------------------
// Payload + decision (pure — the unit-testable core)
// ---------------------------------------------------------------------------

/**
 * The notification's deep link. The API's `notification.created` payload
 * carries it as `deepLink`; the row itself stores the same object as `link` —
 * accept either (`.passthrough()` on the payload keeps the rest of the map).
 */
const notificationLinkSchema = z.object({
  deepLink: z.object({ type: z.string().min(1), id: z.string().min(1) }),
  link: z
    .object({ type: z.string().min(1), id: z.string().min(1) })
    .optional()
});
export type NotificationLink = { type: string; id: string };

/** The `notification.created` stream payload (API `notifications.service`). */
export const notificationCreatedPayloadSchema = z
  .object({
    workspaceId: z.string().min(1),
    notificationId: z.string().min(1),
    userId: z.string().min(1),
    type: z.string().min(1),
    title: z.string().min(1)
  })
  .passthrough();
export type NotificationCreatedPayload = z.infer<typeof notificationCreatedPayloadSchema>;

/** The `actorId == notification.userId → no email` suppression (R6). */
export function isSelfNotification(actorId: string | null, userId: string): boolean {
  return actorId !== null && actorId === userId;
}

export type EmailDecision =
  | { kind: "instant" }
  | { kind: "digest" }
  | { kind: "skip"; reason: string };

/**
 * The per-notification email decision (pure). `prefs` is the user's STORED
 * (partial, unvalidated) prefs — merged over the shared defaults here, the
 * same helper the API `GET /notifications/preferences` uses, so the two can
 * never diverge.
 */
export function decideEmail(
  notification: {
    type: string;
    emailSentAt: Date | null;
    actorId: string | null;
  },
  user: { id: string } | null,
  prefs: unknown
): EmailDecision {
  const type = notification.type;
  if (!(NOTIFICATION_TYPES as readonly string[]).includes(type)) {
    return { kind: "skip", reason: "unknown-type" };
  }
  const t = type as NotificationTypeValue;
  if (notification.emailSentAt) return { kind: "skip", reason: "already-sent" };
  if (!user) return { kind: "skip", reason: "user-not-found" };
  if (isSelfNotification(notification.actorId, user.id)) {
    return { kind: "skip", reason: "self-notification" };
  }
  const merged = mergeNotificationPrefs(prefs);
  if (prefIsOff(t, merged)) return { kind: "skip", reason: "pref-off" };
  if (prefSendsInstant(t, merged)) return { kind: "instant" };
  return { kind: "digest" };
}

/**
 * Build the instant email job's `data` (title + actor + absolute link only —
 * never the comment body: the deep link is the path to it, a smaller leak
 * surface, recorded). `notificationId` is carried so the email worker stamps
 * `Notification.emailSentAt` on a successful send.
 */
export function buildInstantEmailData(input: {
  to: string;
  workspaceName: string;
  title: string;
  actorName: string | null;
  link: NotificationLink | null;
  url: string | null;
  notificationId: string;
}): Record<string, unknown> {
  const { to, workspaceName, title, actorName, url, notificationId } = input;
  const from = actorName ? `From <strong>${esc(actorName)}</strong>. ` : "";
  const line = url
    ? `<p><a href="${esc(url)}">Open in workspace</a></p>`
    : "";
  const html =
    `<p>${from}${esc(title)}</p>` +
    (url ? line : `<p>(open the workspace for details)</p>`) +
    `<p style="color:#999;font-size:12px">${esc(workspaceName)}</p>`;
  const text =
    `${actorName ? `${actorName} — ` : ""}${title}\n` +
    (url ? url : "") +
    `\n${workspaceName}`;
  return {
    to,
    subject: `${workspaceName}: ${title}`,
    html,
    text,
    notificationIds: [notificationId]
  };
}

function esc(s: string): string {
  return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
}

// ---------------------------------------------------------------------------
// Absolute deep links
// ---------------------------------------------------------------------------

/**
 * Resolve a notification deep link to an ABSOLUTE web URL (the spec's
 * `PUBLIC_URL` env). Web routes: `/w/{slugOrId}/page/{pageId}`,
 * `/w/{slugOrId}/databases/{databaseId}`, inbox `/w/{slugOrId}/notifications`.
 *
 *   page     → /w/{ws}/page/{id}
 *   database → /w/{ws}/databases/{id}
 *   record   → its database (the record's home) → /w/{ws}/databases/{dbId}
 *   comment  → its page, else its record's database
 *
 * Unresolvable (deleted target, unknown type, missing workspace) → the
 * workspace root (still requires login; a deleted target → the web's 404).
 * Never throws — an email must never fail because of a stale link.
 */
export async function resolveDeepLink(
  prisma: PrismaClient,
  env: { PUBLIC_URL: string },
  workspaceId: string,
  link: NotificationLink | null
): Promise<string | null> {
  const ws = await prisma.workspace.findUnique({
    where: { id: workspaceId },
    select: { slug: true, id: true }
  });
  if (!ws) return null;
  const wsPart = ws.slug || ws.id;
  const base = env.PUBLIC_URL.replace(/\/+$/, "");
  if (!link) return `${base}/w/${wsPart}`;
  switch (link.type) {
    case "page":
      return `${base}/w/${wsPart}/page/${link.id}`;
    case "database":
      return `${base}/w/${wsPart}/databases/${link.id}`;
    case "record": {
      const rec = await prisma.databaseRecord.findUnique({
        where: { id: link.id },
        select: { databaseId: true }
      });
      return rec
        ? `${base}/w/${wsPart}/databases/${rec.databaseId}`
        : `${base}/w/${wsPart}`;
    }
    case "comment": {
      const c = await prisma.comment.findUnique({
        where: { id: link.id },
        select: { pageId: true, recordId: true }
      });
      if (c?.pageId) return `${base}/w/${wsPart}/page/${c.pageId}`;
      if (c?.recordId) {
        const rec = await prisma.databaseRecord.findUnique({
          where: { id: c.recordId },
          select: { databaseId: true }
        });
        return rec ? `${base}/w/${wsPart}/databases/${rec.databaseId}` : `${base}/w/${wsPart}`;
      }
      return `${base}/w/${wsPart}`;
    }
    default:
      return `${base}/w/${wsPart}`;
  }
}

/** Absolute inbox URL (the digest "n more" link). */
export async function inboxUrl(
  prisma: PrismaClient,
  env: { PUBLIC_URL: string },
  workspaceId: string
): Promise<string | null> {
  const ws = await prisma.workspace.findUnique({
    where: { id: workspaceId },
    select: { slug: true, id: true }
  });
  if (!ws) return null;
  return `${env.PUBLIC_URL.replace(/\/+$/, "")}/w/${ws.slug || ws.id}/notifications`;
}

// ---------------------------------------------------------------------------
// Digest collection (pure — the ≤ 20 cap + "n more" link)
// ---------------------------------------------------------------------------

export interface DigestItem {
  notificationId: string;
  title: string;
  actorName: string | null;
  url: string | null;
  link: NotificationLink | null;
}

/**
 * Turn a user's (unsent) digest-eligible notifications in ONE workspace into
 * digest email items. Only types whose effective pref mode is exactly "digest"
 * qualify — an "all"/"instant" type already received an INSTANT email
 * (→ `emailSentAt` stamped → excluded by the caller's query anyway), so
 * including it here would risk a double-send in the pre-stamp race window;
 * "off"/unknown are excluded. Each link is resolved to an absolute URL
 * (best-effort). Capped at {@link DIGEST_MAX_ITEMS} per email. Pure except for
 * the link-resolution reads — unit-tested with a fake prisma.
 *
 * (Recorded: the spec's `'all'` mode is interpreted as "send immediately" —
 * matching `prefSendsInstant` which returns true for both "instant" and "all".
 * "all" therefore never batches into the digest; the instant path covers it.)
 */
export const DIGEST_MAX_ITEMS = 20;

export async function collectDigest(
  prisma: PrismaClient,
  env: { PUBLIC_URL: string },
  input: {
    userId: string;
    workspaceId: string;
    storedPrefs: unknown;
    rows: Array<{
      id: string;
      type: string;
      title: string;
      link: unknown;
      actorId: string | null;
    }>;
  }
): Promise<{ items: DigestItem[]; remaining: number }> {
  const merged = mergeNotificationPrefs(input.storedPrefs);
  const eligible: typeof input.rows = [];
  for (const row of input.rows) {
    const t = row.type as NotificationTypeValue;
    if (!(NOTIFICATION_TYPES as readonly string[]).includes(row.type)) continue;
    if (merged[t] !== "digest") continue;
    eligible.push(row);
  }
  const capped = eligible.slice(0, DIGEST_MAX_ITEMS);
  const remaining = eligible.length - capped.length;
  const items: DigestItem[] = [];
  for (const row of capped) {
    const link = (row.link ?? null) as NotificationLink | null;
    const url = link ? await resolveDeepLink(prisma, env, input.workspaceId, link) : null;
    items.push({
      notificationId: row.id,
      title: row.title,
      actorName: row.actorId ?? null,
      url,
      link
    });
  }
  return { items, remaining };
}

// ---------------------------------------------------------------------------
// Digest runner (hourly)
// ---------------------------------------------------------------------------

/**
 * The hourly digest: for every (user, workspace) pair with ≥1 unread, unsent,
 * digest-eligible notification, enqueue ONE email (template "digest", subject
 * "Your Factory digest", ≤ {@link DIGEST_MAX_ITEMS} items + "n more" inbox
 * link) carrying the collected `notificationIds` so the email worker stamps
 * `emailSentAt` on a successful send (never re-sent next hour). Returns the
 * number of digest emails enqueued.
 *
 * `enqueue` is injectable for tests. Idempotent across the redelivery of the
 * repeatable job itself: a second run finds no unsent rows (already stamped)
 * → enqueues nothing.
 */
export async function runDigestRun(
  prisma: PrismaClient,
  connection: Redis,
  env: { PUBLIC_URL: string },
  enqueueFn?: typeof enqueue
): Promise<{ enqueued: number }> {
  const enq = enqueueFn ?? enqueue;
  const rows = await prisma.notification.findMany({
    where: { readAt: null, emailSentAt: null },
    select: {
      id: true,
      userId: true,
      workspaceId: true,
      type: true,
      title: true,
      link: true,
      actorId: true
    }
  });
  // Group by (userId, workspaceId).
  const byUser = new Map<
    string,
    { userId: string; workspaceId: string; rows: typeof rows }
  >();
  for (const r of rows) {
    const key = `${r.userId}\u0000${r.workspaceId}`;
    let g = byUser.get(key);
    if (!g) {
      g = { userId: r.userId, workspaceId: r.workspaceId, rows: [] };
      byUser.set(key, g);
    }
    g.rows.push(r);
  }

  let enqueued = 0;
  for (const g of byUser.values()) {
    const user = await prisma.user.findUnique({
      where: { id: g.userId },
      select: { email: true, notificationPrefs: true }
    });
    if (!user) continue;
    const { items, remaining } = await collectDigest(prisma, env, {
      userId: g.userId,
      workspaceId: g.workspaceId,
      storedPrefs: user.notificationPrefs,
      rows: g.rows
    });
    if (items.length === 0) continue;

    const inbox = await inboxUrl(prisma, env, g.workspaceId);
    const bodyItems = items.map((it) => (it.actorName ? `${it.actorName} — ${it.title}` : it.title));
    const vars: Record<string, unknown> = {
      items: bodyItems,
      url: inbox ?? ""
    };
    if (remaining > 0) {
      vars.items = [...bodyItems, `+ ${remaining} more (open your inbox)`];
    }
    await enq(
      connection,
      QUEUES.email,
      {
        to: user.email,
        subject: "Your Factory digest",
        template: "digest",
        vars,
        notificationIds: items.map((it) => it.notificationId)
      },
      // Stable job key: one digest email per (user, workspace, run-window) — a
      // redelivered repeatable tick must not double-enqueue. The window is the
      // collected set's ids (the stamp makes a repeat a no-op regardless).
      { jobId: `digest:${g.userId}:${g.workspaceId}` }
    );
    enqueued++;
    log.info(
      { userId: g.userId, workspaceId: g.workspaceId, items: items.length, remaining },
      "digest email enqueued"
    );
  }
  return { enqueued };
}

// ---------------------------------------------------------------------------
// Stream consumer (worker-email)
// ---------------------------------------------------------------------------

/**
 * The per-event handler (extracted from the StreamConsumer for unit-testing the
 * full decision→enqueue path with fakes). `enqueueFn` is injectable; the real
 * one is the BullMQ `enqueue` helper (dedup by `jobId`). Returns the action
 * taken (for logs/tests). Never throws on a business skip — a throw here would
 * leave the stream entry pending for redelivery.
 */
export async function handleNotificationCreated(
  prisma: PrismaClient,
  connection: Redis,
  env: { PUBLIC_URL: string },
  payload: NotificationCreatedPayload,
  enqueueFn?: typeof enqueue
): Promise<{ action: "instant" | "digest" | "skip"; reason?: string }> {
  const enq = enqueueFn ?? enqueue;

  // Load the authoritative row (the stream payload is a fan-out hint).
  const row = await prisma.notification.findUnique({
    where: { id: payload.notificationId },
    select: { type: true, emailSentAt: true, actorId: true, userId: true }
  });
  if (!row) {
    return { action: "skip", reason: "row-not-found" };
  }
  const user = await prisma.user.findUnique({
    where: { id: row.userId },
    select: { id: true, email: true, notificationPrefs: true }
  });

  const decision = decideEmail(row, user, user?.notificationPrefs ?? null);
  if (decision.kind === "skip") return { action: "skip", reason: decision.reason };
  if (decision.kind === "digest") return { action: "digest" }; // hourly digest collects it
  if (!user) return { action: "skip", reason: "user-not-found" };

  // instant:
  const actorName =
    typeof payload.actorName === "string" && payload.actorName ? payload.actorName : null;
  const link = (payload.deepLink ?? payload.link) as NotificationLink | undefined;
  const url = await resolveDeepLink(prisma, env, payload.workspaceId, link ?? null);
  const ws = await prisma.workspace.findUnique({
    where: { id: payload.workspaceId },
    select: { name: true }
  });
  const data = buildInstantEmailData({
    to: user.email,
    workspaceName: ws?.name ?? "Factory",
    title: payload.title,
    actorName,
    link: link ?? null,
    url,
    notificationId: payload.notificationId
  });
  await enq(connection, QUEUES.email, data, {
    // Dedup key = the notification id. NO colon: BullMQ 5.x `Job.validateOptions`
    // rejects any custom jobId containing `:` (only a 3-segment id is legal —
    // repeatable compatibility). The old `email:${id}` (2-segment) made EVERY
    // instant-enqueue throw `Custom Id cannot contain :` (recorded conflict,
    // FEATURE_TRACKER T-176/177/178 — fixed 2026-10-12). The underscore keeps
    // the human-readable `email` prefix in the Jobs-tab DLQ view; nothing
    // parses the jobId prefix (health tab correlates on payload).
    jobId: `email_${payload.notificationId}`
  });
  return { action: "instant" };
}

/**
 * The `worker-email` StreamConsumer (its own consumer group, so email lag can't
 * block the other queues). Only `notification.created` is actionable. The
 * consumer only ENqueues — the real send is the `email` queue's job (stamping
 * `emailSentAt` happens there), so there is no SMTP dependency here.
 *
 * `connection` is the BullMQ Redis (for `enqueue`); `redis` drives the stream.
 */
export function createNotificationEmailConsumer(opts: {
  redis: Redis;
  connection: Redis;
  prisma: PrismaClient;
  env: { PUBLIC_URL: string };
  consumerId: string;
}): StreamConsumer {
  const { redis, connection, prisma, env } = opts;
  return new StreamConsumer(
    redis,
    "factory:events",
    "worker-email",
    opts.consumerId,
    async (entry: StreamEntry) => {
      const parsed = tryParseEvent(entry);
      if (!parsed) return; // poison entry: already logged; drop + XACK
      const { envelope } = parsed;
      if (envelope.event !== "notification.created") return; // XACK: not ours
      const payload = notificationCreatedPayloadSchema.safeParse(
        envelope.payload as Record<string, unknown>
      );
      if (!payload.success) {
        log.warn(
          { id: entry.id, err: payload.error.message },
          "dropping malformed notification.created payload"
        );
        return;
      }
      const res = await handleNotificationCreated(prisma, connection, env, payload.data);
      if (res.action === "instant") {
        log.info({ notificationId: payload.data.notificationId }, "instant email enqueued");
      }
    }
  );
}