/**
 * TASK-0121 — usage.rollup + budget.notify unit tests (Docker-free, fake
 * prisma/redis — the `as unknown as PrismaClient` convention of
 * ai-embedding.test.ts).
 *
 * Covers the DoD acceptance:
 *   - the rollup sums ONLY the current month (rows before month-start are
 *     excluded — asserted on the createdAt filter + the sum)
 *   - the rollup sums ONLY the requested workspace's rows (scoped to its
 *     own conversation ids; another workspace's ledger is untouched)
 *   - the budget sweep enqueues budget.notify only when crossed AND the 24 h
 *     slot is free (slot set → no enqueue)
 *   - budget.notify: claims the slot (NX), notifies every OWNER/ADMIN,
 *     publishes notification.created per admin, writes the audit row — and
 *     SKIPS when the slot is already held (the 24 h throttle)
 */
import { describe, expect, it, vi } from "vitest";
import type { PrismaClient } from "@prisma/client";
import type { Redis } from "ioredis";
import {
  aiJobSchema,
  aiMonthlyBudget,
  budgetNotifySlot,
  runBudgetNotify,
  runUsageRollup
} from "../processors/ai.processor.js";

// ---------------------------------------------------------------------------
// Fakes
// ---------------------------------------------------------------------------

interface MessageRow {
  conversationId: string;
  createdAt: Date;
  tokensIn: number | null;
  tokensOut: number | number;
}

function fakePrisma(opts: {
  workspaces?: { id: string; settings: unknown; usageLimits: unknown }[];
  conversations?: { id: string; workspaceId: string }[];
  messages?: MessageRow[];
  members?: { userId: string; workspaceId: string; role: string }[];
}) {
  const calls = {
    aggregate: vi.fn(),
    findManyWs: vi.fn(),
    findManyConv: vi.fn(),
    findManyMembers: vi.fn(),
    notificationCreate: vi.fn(),
    auditCreate: vi.fn()
  };

  calls.findManyWs.mockImplementation(async (args: { where: { id?: string } }) => {
    const all = opts.workspaces ?? [];
    if (args.where.id) return all.filter((w) => w.id === args.where.id);
    return all;
  });
  calls.findManyConv.mockImplementation(async (args: { where: { workspaceId: string } }) =>
    (opts.conversations ?? []).filter((c) => c.workspaceId === args.where.workspaceId).map((c) => ({ id: c.id }))
  );
  calls.aggregate.mockImplementation(async (args: {
    where: { conversationId: { in: string[] }; createdAt: { gte: Date } };
  }) => {
    const inConv = new Set(args.where.conversationId.in);
    const rows = (opts.messages ?? []).filter(
      (m) => inConv.has(m.conversationId) && m.createdAt >= args.where.createdAt.gte
    );
    const ti = rows.reduce((s, m) => s + (m.tokensIn ?? 0), 0);
    const to = rows.reduce((s, m) => s + (m.tokensOut ?? 0), 0);
    return { _sum: { tokensIn: ti, tokensOut: to } };
  });
  // Mirrors the DB: both the workspace AND the `role: {in}` where clause
  // apply (the real Prisma filters in Postgres; the fake must too).
  calls.findManyMembers.mockImplementation(async (args: {
    where: { workspaceId: string; role: { in: string[] } };
  }) =>
    (opts.members ?? [])
      .filter((m) => m.workspaceId === args.where.workspaceId && args.where.role.in.includes(m.role))
      .map((m) => ({ userId: m.userId }))
  );
  let notifSeq = 0;
  calls.notificationCreate.mockImplementation(async () => ({ id: `ntf-${++notifSeq}` }));
  calls.auditCreate.mockImplementation(async () => ({ id: "aud-1" }));

  const prisma = {
    workspace: { findMany: calls.findManyWs },
    aiConversation: { findMany: calls.findManyConv },
    aiMessage: { aggregate: calls.aggregate },
    workspaceMember: { findMany: calls.findManyMembers },
    notification: { create: calls.notificationCreate },
    auditLog: { create: calls.auditCreate }
  } as unknown as PrismaClient;

  return { prisma, calls };
}

function fakeRedis(opts: { slotValue?: string | null; setReturns?: string | "OK" } = {}) {
  const calls = {
    get: vi.fn(async (key: string) => (key === budgetNotifySlot("ws-1") ? (opts.slotValue ?? null) : null)),
    set: vi.fn(async () => opts.setReturns ?? "OK"),
    // xadd(key, id, "event", <name>, "payload", <json>, "ts", <iso>) — the
    // real ioredis signature is variadic, so the mock declares it as such
    // (keeps the call-shape assertions below typeable).
    xadd: vi.fn(async (..._args: string[]) => "1234-0")
  };
  return { redis: calls as unknown as Redis, calls };
}

function monthStart(): Date {
  const d = new Date();
  d.setUTCDate(1);
  d.setUTCHours(0, 0, 0, 0);
  return d;
}

// ---------------------------------------------------------------------------
// payload validation
// ---------------------------------------------------------------------------

describe("aiJobSchema (TASK-0121 job types)", () => {
  it("accepts usage.rollup (workspaceId optional) and budget.notify", () => {
    expect(aiJobSchema.parse({ type: "usage.rollup" }).type).toBe("usage.rollup");
    // (the parse() result is the job union — access the variant field via the
    //  narrow cast now that the union has more members than TASK-0121 added)
    expect((aiJobSchema.parse({ type: "usage.rollup", workspaceId: "ws-1" }) as { workspaceId?: string }).workspaceId).toBe("ws-1");
    expect(
      aiJobSchema.parse({ type: "budget.notify", workspaceId: "ws-1", used: 10, limit: 100 }).type
    ).toBe("budget.notify");
  });

  it("rejects a budget.notify without the numeric budget fields", () => {
    expect(() => aiJobSchema.parse({ type: "budget.notify", workspaceId: "ws-1" })).toThrow();
    expect(() => aiJobSchema.parse({ type: "budget.notify", workspaceId: "ws-1", used: -1, limit: 10 })).toThrow();
    expect(() => aiJobSchema.parse({ type: "budget.notify", workspaceId: "ws-1", used: 10, limit: 0 })).toThrow();
  });
});

// ---------------------------------------------------------------------------
// aiMonthlyBudget
// ---------------------------------------------------------------------------

describe("aiMonthlyBudget (admin override wins over settings)", () => {
  it("uses max(usageLimits, settings.budgets.tokensPerMonth)", () => {
    expect(aiMonthlyBudget({ usageLimits: { aiMonthlyTokens: 500 }, settings: { ai: { budgets: { tokensPerMonth: 900 } } } })).toBe(900);
    expect(aiMonthlyBudget({ usageLimits: { aiMonthlyTokens: 1500 }, settings: {} })).toBe(1500);
  });

  it("unset or 0 = unlimited (0)", () => {
    expect(aiMonthlyBudget({ usageLimits: {}, settings: {} })).toBe(0);
    expect(aiMonthlyBudget({ usageLimits: { aiMonthlyTokens: 0 }, settings: {} })).toBe(0);
    expect(aiMonthlyBudget({ usageLimits: "not-json", settings: null as unknown })).toBe(0);
  });
});

// ---------------------------------------------------------------------------
// runUsageRollup
// ---------------------------------------------------------------------------

describe("runUsageRollup (TASK-0121 nightly reconcile)", () => {
  it("sums ONLY the current month and ONLY the workspace's own rows", async () => {
    const ms = monthStart();
    const before = new Date(ms.getTime() - 60_000); // last month
    const inMonth = new Date(ms.getTime() + 60_000);
    const { prisma, calls } = fakePrisma({
      workspaces: [{ id: "ws-1", usageLimits: { aiMonthlyTokens: 10_000 }, settings: {} }],
      conversations: [
        { id: "conv-a", workspaceId: "ws-1" },
        { id: "conv-b", workspaceId: "ws-1" }
      ],
      messages: [
        { conversationId: "conv-a", createdAt: before, tokensIn: 500, tokensOut: 500 }, // previous month — excluded
        { conversationId: "conv-a", createdAt: inMonth, tokensIn: 100, tokensOut: 50 },
        { conversationId: "conv-b", createdAt: inMonth, tokensIn: null, tokensOut: 150 }
      ]
    });
    const { redis } = fakeRedis();
    const enqueueBudgetNotify = vi.fn(async () => ({}));

    const res = await runUsageRollup(
      { prisma, redis, enqueueBudgetNotify },
      { type: "usage.rollup", workspaceId: "ws-1" },
      "job-1"
    );

    expect(res).toMatchObject({ reconciled: 1, budgetsExceeded: 0, enqueued: 0 });
    // The aggregate was scoped to ws-1's conversation ids AND month start.
    const aggArg = calls.aggregate.mock.calls[0]![0] as {
      where: { conversationId: { in: string[] }; createdAt: { gte: Date } };
    };
    expect(aggArg.where.conversationId.in.sort()).toEqual(["conv-a", "conv-b"]);
    expect(aggArg.where.createdAt.gte.getTime()).toBe(ms.getTime());
    // used = 100+50+150 = 300 (the 1000 previous-month tokens are excluded)
    expect(enqueueBudgetNotify).not.toHaveBeenCalled();
  });

  it("scopes to ONE workspace: another workspace's ledger never enters the sum", async () => {
    const inMonth = new Date();
    const { prisma } = fakePrisma({
      // only ws-1 is in scope (single-workspace run); ws-2 has a full ledger
      workspaces: [{ id: "ws-1", usageLimits: {}, settings: {} }],
      conversations: [{ id: "conv-1", workspaceId: "ws-1" }],
      messages: [
        { conversationId: "conv-1", createdAt: inMonth, tokensIn: 10, tokensOut: 10 }
      ]
    });
    const { redis } = fakeRedis();
    const res = await runUsageRollup(
      { prisma, redis, enqueueBudgetNotify: vi.fn(async () => ({})) },
      { type: "usage.rollup", workspaceId: "ws-1" },
      null
    );
    expect(res.reconciled).toBe(1);
  });

  it("enqueues budget.notify when the budget was crossed and the slot is free", async () => {
    const inMonth = new Date();
    const { prisma } = fakePrisma({
      workspaces: [{ id: "ws-1", usageLimits: { aiMonthlyTokens: 100 }, settings: {} }],
      conversations: [{ id: "conv-1", workspaceId: "ws-1" }],
      messages: [{ conversationId: "conv-1", createdAt: inMonth, tokensIn: 60, tokensOut: 60 }]
    });
    const { redis } = fakeRedis({ slotValue: null });
    const enqueueBudgetNotify = vi.fn(async () => ({}));

    const res = await runUsageRollup(
      { prisma, redis, enqueueBudgetNotify },
      { type: "usage.rollup", workspaceId: "ws-1" },
      null
    );

    expect(res).toMatchObject({ reconciled: 1, budgetsExceeded: 1, enqueued: 1 });
    expect(enqueueBudgetNotify).toHaveBeenCalledWith({
      type: "budget.notify",
      workspaceId: "ws-1",
      used: 120,
      limit: 100
    });
  });

  it("does NOT re-enqueue when the 24 h slot is already set", async () => {
    const inMonth = new Date();
    const { prisma } = fakePrisma({
      workspaces: [{ id: "ws-1", usageLimits: { aiMonthlyTokens: 100 }, settings: {} }],
      conversations: [{ id: "conv-1", workspaceId: "ws-1" }],
      messages: [{ conversationId: "conv-1", createdAt: inMonth, tokensIn: 60, tokensOut: 60 }]
    });
    const { redis } = fakeRedis({ slotValue: "1" });
    const enqueueBudgetNotify = vi.fn(async () => ({}));

    const res = await runUsageRollup(
      { prisma, redis, enqueueBudgetNotify },
      { type: "usage.rollup", workspaceId: "ws-1" },
      null
    );

    expect(res).toMatchObject({ budgetsExceeded: 1, enqueued: 0 });
    expect(enqueueBudgetNotify).not.toHaveBeenCalled();
  });

  it("unlimited budget (0) never counts as exceeded", async () => {
    const inMonth = new Date();
    const { prisma } = fakePrisma({
      workspaces: [{ id: "ws-1", usageLimits: {}, settings: {} }],
      conversations: [{ id: "conv-1", workspaceId: "ws-1" }],
      messages: [{ conversationId: "conv-1", createdAt: inMonth, tokensIn: 1e9, tokensOut: 1 }]
    });
    const { redis } = fakeRedis();
    const enqueueBudgetNotify = vi.fn(async () => ({}));
    const res = await runUsageRollup(
      { prisma, redis, enqueueBudgetNotify },
      { type: "usage.rollup", workspaceId: "ws-1" },
      null
    );
    expect(res.budgetsExceeded).toBe(0);
    expect(enqueueBudgetNotify).not.toHaveBeenCalled();
  });
});

// ---------------------------------------------------------------------------
// runBudgetNotify
// ---------------------------------------------------------------------------

describe("runBudgetNotify (TASK-0121 admin notification)", () => {
  const JOB = { type: "budget.notify" as const, workspaceId: "ws-1", used: 150, limit: 100 };

  it("notifies every OWNER/ADMIN + publishes notification.created + audits", async () => {
    const { prisma, calls } = fakePrisma({
      members: [
        { userId: "owner-1", workspaceId: "ws-1", role: "OWNER" },
        { userId: "admin-1", workspaceId: "ws-1", role: "ADMIN" },
        { userId: "member-1", workspaceId: "ws-1", role: "MEMBER" } // not an admin
      ]
    });
    const { redis, calls: redisCalls } = fakeRedis({ setReturns: "OK" });

    const res = await runBudgetNotify({ prisma, redis: redis as unknown as Redis }, JOB, "job-9");

    expect(res).toEqual({ notified: 2 });
    // slot claimed with NX + 24 h TTL
    expect(redisCalls.set).toHaveBeenCalledWith(budgetNotifySlot("ws-1"), "1", "EX", 24 * 3600, "NX");
    expect(calls.notificationCreate).toHaveBeenCalledTimes(2);
    const first = calls.notificationCreate.mock.calls[0]![0] as { data: Record<string, unknown> };
    expect(first.data).toMatchObject({
      workspaceId: "ws-1",
      userId: "owner-1",
      type: "AUTOMATION",
      channel: "IN_APP",
      title: "AI budget exceeded"
    });
    expect(String(first.data.body)).toContain("150 used of 100");
    // one stream event per admin notification (the email channel's input)
    expect(redisCalls.xadd).toHaveBeenCalledTimes(2);
    // xadd(key, id, "event", <name>, "payload", <json>, "ts", <iso>)
    const call = redisCalls.xadd.mock.calls[0]!;
    expect(call[0]).toBe("factory:events");
    expect(call[3]).toBe("notification.created");
    expect(call[4]).toBe("payload");
    const payload = JSON.parse(call[5] as string);
    expect(payload).toMatchObject({ notificationId: "ntf-1", type: "AUTOMATION", workspaceId: "ws-1" });
    expect(payload.deepLink).toEqual({ type: "workspace", id: "ws-1" });
    // the audit row (system actor)
    expect(calls.auditCreate).toHaveBeenCalledTimes(1);
    const audit = calls.auditCreate.mock.calls[0]![0] as { data: Record<string, unknown> };
    expect(audit.data).toMatchObject({
      actorType: "SYSTEM",
      action: "ai.budget_exceeded",
      workspaceId: "ws-1"
    });
    expect(audit.data.metadata).toMatchObject({ used: 150, limit: 100, adminsNotified: 2 });
  });

  it("skips when the slot is already held (24 h throttle — the API hook won the race)", async () => {
    const { prisma, calls } = fakePrisma({ members: [{ userId: "owner-1", workspaceId: "ws-1", role: "OWNER" }] });
    const { redis } = fakeRedis({ setReturns: "PENDING" }); // NX lost → not "OK"

    const res = await runBudgetNotify({ prisma, redis: redis as unknown as Redis }, JOB, null);

    expect(res).toEqual({ skipped: true, reason: "already notified" });
    expect(calls.notificationCreate).not.toHaveBeenCalled();
    expect(calls.auditCreate).not.toHaveBeenCalled();
  });

  it("no OWNER/ADMIN → no rows, no crash (the slot stays set for 24 h)", async () => {
    const { prisma, calls } = fakePrisma({ members: [{ userId: "member-1", workspaceId: "ws-1", role: "MEMBER" }] });
    const { redis } = fakeRedis({ setReturns: "OK" });

    const res = await runBudgetNotify({ prisma, redis: redis as unknown as Redis }, JOB, null);

    expect(res).toEqual({ notified: 0 });
    expect(calls.notificationCreate).not.toHaveBeenCalled();
    expect(calls.auditCreate).not.toHaveBeenCalled();
  });

  it("a redis slot failure fails OPEN (notifies) — a duplicate is cheaper than a miss", async () => {
    const { prisma, calls } = fakePrisma({ members: [{ userId: "owner-1", workspaceId: "ws-1", role: "OWNER" }] });
    const redisCalls = {
      get: vi.fn(async () => null),
      set: vi.fn(async () => {
        throw new Error("redis down");
      }),
      xadd: vi.fn(async () => "1-0")
    };
    const res = await runBudgetNotify({ prisma, redis: redisCalls as unknown as Redis }, JOB, null);
    expect(res).toEqual({ notified: 1 });
    expect(calls.notificationCreate).toHaveBeenCalledTimes(1);
  });
});