import { Inject, Injectable } from "@nestjs/common";
import {
  NotificationPrefsSchema,
  mergeNotificationPrefs,
  type NotificationPrefs,
  type NotificationTypeValue
} from "@factory/shared";
import { PrismaService } from "../../common/prisma.service.js";
import { EventBus } from "../../common/event-bus.js";
import { ApiError } from "../../common/errors.js";
import { parsePagination } from "../../lib/pagination.js";
import { EVENTS } from "../../constants.js";
import { actorLevel } from "../../lib/authorize.js";

const NOTIF_SELECT = {
  id: true,
  workspaceId: true,
  type: true,
  channel: true,
  title: true,
  body: true,
  link: true,
  readAt: true,
  // The actor (the bell's "who" label — batch-resolved in `list`; no
  // Prisma relation, so a plain userId).
  actorId: true,
  createdAt: true
} as const;

export interface NotifyInput {
  workspaceId: string;
  /** The recipient (the row is per-user, always the caller's own on read). */
  userId: string;
  type: NotificationTypeValue;
  title: string;
  body?: string | null;
  /** Deep link `{ type, id }` → the web deep-link navigation (R5). `pageId`
   *  is an optional ride-along so the web can navigate to a comment/record
   *  link's page without a comment→page lookup endpoint (TASK-0094 R5,
   *  recorded — the worker's link schema is a passthrough and tolerates it). */
  link: { type: string; id: string; pageId?: string };
  /** The acting user (self-suppression for email + the actor label). */
  actorId?: string | null;
}

/**
 * In-app notifications (mentions, assignments, shares, comment activity).
 *
 * Two surfaces per notification (one `notify()` call → both, TASK-0094):
 *   1. the `Notification` row — the durable state (bell count, inbox, the
 *      email decision's input; the row IS the truth for the email channel —
 *      the worker's stream consumer decides + enqueues there),
 *   2. the `notification.created` stream event — the realtime `user:`
 *      control-room push (R7) + the worker's email path (R6).
 *
 * The EMAIL enqueue is deliberately NOT done here (recorded decision, R6):
 * the API already publishes to the `factory:events` stream and does not own
 * the BullMQ queues (those are the worker's — see `apps/worker`). The
 * worker's `worker-email` stream consumer owns the whole email decision
 * (prefs read, self-suppression, instant vs digest, the `email_{id}` job-key
 * idempotency — colon-free: BullMQ 5.x rejects 2-segment colon jobIds — and
 * the `emailSentAt` stamp) so the decision is idempotent
 * end-to-end and testable in one place.
 */
@Injectable()
export class NotificationsService {
  constructor(
    @Inject(PrismaService) private readonly prisma: PrismaService,
    @Inject(EventBus) private readonly events: EventBus
  ) {}

  /**
   * T-157: fan a "your watched page changed" WATCH_UPDATE out to every
   * watcher of `pageId` — with a canView RECHECK at fan-out time, so a
   * membership revoked between watch and event is never notified (the
   * spec's negative case). Best-effort: one watcher's failure does not
   * break the others or the triggering request.
   */
  async notifyWatchersForPageUpdate(pageId: string, actorId: string | null): Promise<number> {
    const watches = await this.prisma.pageWatch.findMany({ where: { pageId } });
    const page = await this.prisma.page.findUnique({
      where: { id: pageId },
      select: { workspaceId: true, title: true }
    });
    if (!watches.length || !page) return 0;
    let sent = 0;
    for (const w of watches) {
      if (w.userId === actorId) continue; // no self-noise
      const level = await actorLevel(this.prisma, { workspaceId: page.workspaceId, userId: w.userId });
      if (level === "NONE") continue; // revoked between watch and event → not notified
      try {
        await this.notify({
          workspaceId: page.workspaceId,
          userId: w.userId,
          type: "WATCH_UPDATE",
          title: `Watched page updated: ${page.title || "Untitled"}`,
          link: { type: "page", id: pageId, pageId },
          actorId: actorId ?? undefined
        });
        sent++;
      } catch {
        // best-effort per watcher (the row creation can only fail on a
        // constraint we already verified) — never fail the trigger.
      }
    }
    return sent;
  }

  async list(userId: string, query: { cursor?: string; read?: string; limit?: string }) {
    const { cursor, limit } = parsePagination(query);
    const read = query.read;
    const where: Record<string, unknown> = { userId };
    if (read === "true") where.readAt = { not: null };
    else if (read === "false") where.readAt = null;

    const rows = await this.prisma.notification.findMany({
      where,
      ...(cursor ? { cursor: { id: cursor.id }, skip: 1 } : {}),
      orderBy: { createdAt: "desc" },
      take: limit + 1,
      select: NOTIF_SELECT
    });
    const page = rows.slice(0, limit);
    const next = rows.length > limit ? page[page.length - 1] : undefined;

    // Batch the actor name (the `Notification.actorId` has no Prisma
    // relation — the actor is a plain userId) so the inbox can show
    // "who" without an N+1 of requests (TASK-0094 R5). Deleted actors
    // simply don't resolve (the row keeps its `actorId`).
    const actorIds = [...new Set(page.map((r) => r.actorId).filter((id): id is string => Boolean(id)))];
    const actors = actorIds.length
      ? await this.prisma.user.findMany({
          where: { id: { in: actorIds } },
          select: { id: true, name: true, avatarUrl: true }
        })
      : [];
    const actorById = new Map(actors.map((u) => [u.id, u]));

    return {
      data: page.map((r) => ({
        ...r,
        actor: r.actorId ? (actorById.get(r.actorId) ?? null) : null
      })),
      nextCursor: next
        ? Buffer.from(JSON.stringify({ id: next.id, position: next.id })).toString("base64url")
        : null
    };
  }

  async markRead(notificationId: string, userId: string) {
    const row = await this.prisma.notification.findFirst({
      where: { id: notificationId, userId },
      select: { id: true, readAt: true }
    });
    if (!row) throw ApiError.notFound("Notification not found.");
    if (!row.readAt) {
      await this.prisma.notification.update({ where: { id: row.id }, data: { readAt: new Date() } });
    }
    return { read: true, id: row.id };
  }

  async markAllRead(userId: string) {
    const result = await this.prisma.notification.updateMany({
      where: { userId, readAt: null },
      data: { readAt: new Date() }
    });
    return { marked: result.count };
  }

  async unreadCount(userId: string) {
    const count = await this.prisma.notification.count({ where: { userId, readAt: null } });
    return { count };
  }

  /**
   * Create a notification: the row + the `notification.created` stream
   * event (the realtime control-room push, R7 + the worker's email decision,
   * R6). The row is created FIRST (durable) — the event publish is
   * best-effort (a missed push heals on the next poll; the row never
   * depends on the stream).
   */
  async notify(input: NotifyInput) {
    const actor = input.actorId
      ? await this.prisma.user.findUnique({
          where: { id: input.actorId },
          select: { name: true, displayName: true }
        })
      : null;
    const actorName = actor ? actor.displayName ?? actor.name : null;

    const row = await this.prisma.notification.create({
      data: {
        workspaceId: input.workspaceId,
        userId: input.userId,
        type: input.type,
        channel: "IN_APP",
        title: input.title,
        body: input.body ?? null,
        link: input.link as object,
        actorId: input.actorId ?? null
      }
    });

    await this.events.publish(EVENTS.NOTIFICATION_CREATED, {
      workspaceId: input.workspaceId,
      notificationId: row.id,
      userId: input.userId,
      type: input.type,
      title: input.title,
      deepLink: input.link,
      actorId: input.actorId ?? undefined,
      actorName: actorName ?? undefined
    });

    return row;
  }

  // ---------- per-type EMAIL preferences (R6) ----------

  /**
   * The user's effective per-type email prefs: the stored (partial) map
   * merged over the shared defaults. The shape is `NotificationPrefsSchema`
   * (shared — the same schema the worker's decision uses, so API + worker
   * never diverge).
   */
  async getPrefs(userId: string): Promise<NotificationPrefs> {
    const user = await this.prisma.user.findUnique({
      where: { id: userId },
      select: { notificationPrefs: true }
    });
    if (!user) throw ApiError.notFound("User not found.");
    return mergeNotificationPrefs(user.notificationPrefs);
  }

  /**
   * Update the user's per-type email prefs (partial — only the provided
   * types change). The full effective map is returned (stored + defaults).
   */
  async updatePrefs(userId: string, patch: Record<string, string>): Promise<NotificationPrefs> {
    const current = await this.getPrefs(userId);
    const next: Record<string, string> = { ...current };
    for (const [type, mode] of Object.entries(patch)) {
      next[type] = mode as string;
    }
    const validated = NotificationPrefsSchema.parse(next);
    await this.prisma.user.update({
      where: { id: userId },
      data: { notificationPrefs: validated as object }
    });
    return validated as unknown as NotificationPrefs;
  }
}