// src/lib/billing/quotas.ts — plan-limit resolution + quota assertion.
// Enforcement happens AT CREATION (count-then-block): existing data over the
// limit is never deleted. -1 = unlimited. Fail-closed: expired orgs and
// unknown plans resolve to all-zero limits.
// Spec: docs/superpowers/specs/2026-06-12-billing-quotas-design.md (3B).
import "server-only";
import { eq, sql } from "drizzle-orm";
import { unstable_cache } from "next/cache";
import { cache } from "react";
import { orgLimitOverrides } from "@/lib/db/admin-schema";
import { db } from "@/lib/db/client";
import { getPlan, PLANS, type PlanLimits } from "./plans";
import { type BillingStatus, getOrgBillingInfo } from "./status";
import { getUsage } from "./usage";

export type QuotaResource = keyof PlanLimits;

/** Typed quota violation — server actions map it to a structured return. */
export class QuotaError extends Error {
  readonly code = "quota_exceeded";

  constructor(
    public resource: QuotaResource,
    public limit: number,
    public current: number,
  ) {
    super(`Quota exceeded for ${resource}: ${current}/${limit}`);
    this.name = "QuotaError";
  }
}

/** Serializable QuotaError payload for server-action returns — the client
 *  translates it (i18n `Quota` namespace) instead of showing `error.message`. */
export interface QuotaExceeded {
  resource: QuotaResource;
  limit: number;
  /** Rows still available before the limit (floor 0) — for import messages. */
  remaining: number;
}

export function quotaPayload(e: QuotaError): QuotaExceeded {
  return {
    resource: e.resource,
    limit: e.limit,
    remaining: Math.max(0, e.limit - e.current),
  };
}

const ZERO_LIMITS: PlanLimits = {
  keywords: 0,
  projects: 0,
  seats: 0,
  integrations: 0,
  backlinkRows: 0,
  mcpMonthly: 0,
};

/**
 * Pure resolution (unit-testable):
 * - active + known plan → that plan's limits;
 * - trialing → Pro limits with keywords & mcpMonthly halved (750 / 50 000);
 * - expired or unknown plan → all zeros (fail closed — the billing gate
 *   already blocks the app; MCP keys are also blocked at zero).
 */
export function resolveQuotaLimits(input: {
  status: BillingStatus;
  plan: string | null;
}): PlanLimits {
  if (input.status === "active" && input.plan !== null) {
    const plan = PLANS.find((p) => p.name === input.plan);
    if (plan) return plan.limits;
    return ZERO_LIMITS;
  }
  if (input.status === "trialing") {
    const pro = getPlan("pro").limits;
    return {
      ...pro,
      keywords: pro.keywords / 2,
      mcpMonthly: pro.mcpMonthly / 2,
    };
  }
  return ZERO_LIMITS;
}

/** The override row shape consumed by applyLimitOverrides (a subset of the
 *  org_limit_overrides table — the per-limit fields plus expiry). */
export interface OrgLimitOverride {
  keywords: number | null;
  projects: number | null;
  seats: number | null;
  integrations: number | null;
  backlinkRows: number | null;
  mcpMonthly: number | null;
  expiresAt: Date | null;
}

/**
 * Pure: layer a super-admin override on top of plan-resolved limits. Each
 * non-null override field wins (including 0 and -1=unlimited); null inherits
 * the plan value. An override whose `expiresAt` is in the past is ignored
 * entirely (returns the plan limits unchanged). `null` override = no override.
 */
export function applyLimitOverrides(
  limits: PlanLimits,
  override: OrgLimitOverride | null,
  now: Date,
): PlanLimits {
  if (!override) return limits;
  if (override.expiresAt && override.expiresAt.getTime() <= now.getTime()) {
    return limits;
  }
  const next = { ...limits };
  for (const key of Object.keys(limits) as (keyof PlanLimits)[]) {
    const value = override[key];
    if (value !== null && value !== undefined) next[key] = value;
  }
  return next;
}

/**
 * Pure boundary check: would adding `increment` exceed `limit`?
 * `current + increment === limit` still fits; -1 = unlimited.
 */
export function exceedsQuota(
  limit: number,
  current: number,
  increment: number,
): boolean {
  if (limit === -1) return false;
  return current + increment > limit;
}

/** Load the org's limit-override row (or null). Fail-open: a read error returns
 *  null so a transient DB hiccup can never tighten/loosen quotas unexpectedly —
 *  the caller falls back to plain plan limits. */
async function loadLimitOverride(
  orgId: string,
): Promise<OrgLimitOverride | null> {
  try {
    const [row] = await db
      .select({
        keywords: orgLimitOverrides.keywords,
        projects: orgLimitOverrides.projects,
        seats: orgLimitOverrides.seats,
        integrations: orgLimitOverrides.integrations,
        backlinkRows: orgLimitOverrides.backlinkRows,
        mcpMonthly: orgLimitOverrides.mcpMonthly,
        expiresAt: orgLimitOverrides.expiresAt,
      })
      .from(orgLimitOverrides)
      .where(eq(orgLimitOverrides.organizationId, orgId))
      .limit(1);
    return row ?? null;
  } catch (err) {
    console.error("[quotas] override read failed, using plan limits", err);
    return null;
  }
}

/** Cache tag for `getOrgQuotaLimits(orgId)` — invalidated wherever a plan
 *  change, trial change, or limit-override write could shift the resolved
 *  limits (see revalidateBillingCache()/setOrgLimitOverrideAction). */
export function orgLimitsTag(orgId: string): string {
  return `org-limits-${orgId}`;
}

async function fetchOrgQuotaLimits(orgId: string): Promise<PlanLimits> {
  const planLimits = resolveQuotaLimits(await getOrgBillingInfo(orgId));
  const override = await loadLimitOverride(orgId);
  return applyLimitOverrides(planLimits, override, new Date());
}

// Two cache layers, deliberately different lifetimes:
//  - React `cache()` (outer): one resolution per REQUEST — free, exact,
//    scoped to a single render/action.
//  - `unstable_cache` (inner): survives ACROSS requests, 60s TTL, tagged per
//    org. This is the MCP quota chokepoint's slowest-changing read (plan +
//    trial status + super-admin override), so a cross-request cache is the
//    highest-leverage cut of the ~8 round-trips/call (QW2-A). A stale window
//    can (rarely) let an admin override's `expiresAt` linger up to 60s past
//    expiry, or a just-upgraded plan take up to 60s to raise the ceiling —
//    acceptable for a soft quota gate; revalidateTag calls below bound it
//    tighter on the write paths that matter.
export const getOrgQuotaLimits = cache(
  async (orgId: string): Promise<PlanLimits> => {
    return unstable_cache(fetchOrgQuotaLimits, ["org-quota-limits", orgId], {
      tags: [orgLimitsTag(orgId)],
      revalidate: 60,
    })(orgId);
  },
);

/** Throws QuotaError when `current + increment` would exceed the org's limit.
 *
 * TOCTOU warning (sec audit A2/P8): this reads the current usage
 * (COUNT/SUM) and decides in JS — it does NOT lock anything and does NOT
 * write anything itself. Two concurrent callers for the SAME org can both
 * read the same pre-write usage and both pass, together overshooting the
 * limit.
 *
 * Callers that only ever read/decide (e.g. `checkInvitationSeatBlock`'s
 * notification check, the `invite-member` hook in auth.ts) can keep calling
 * this directly: there's nothing here to serialize against a write we don't
 * own (see A2 in the sec2 report — Better Auth owns those inserts).
 *
 * Callers that go on to actually WRITE the counted rows (imports,
 * `createSiteAction`, `addKeywordAction`, …) must NOT call this from inside
 * their write transaction — use `assertWithinQuotaTx` instead. This
 * function resolves `getOrgQuotaLimits`/`getUsage` via the plain `db`
 * handle, i.e. a POOL CONNECTION OF ITS OWN; calling it from inside an
 * already-open `db.transaction` deadlocks the pool (see
 * `assertWithinQuotaTx`'s doc — this is not theoretical, it reproduces
 * under `bun run test` once concurrency reaches the pool size). */
export async function assertWithinQuota(
  orgId: string,
  resource: QuotaResource,
  increment = 1,
): Promise<void> {
  const limits = await getOrgQuotaLimits(orgId);
  const limit = limits[resource];
  if (limit === -1) return;
  const current = await getUsage(orgId, resource);
  if (exceedsQuota(limit, current, increment)) {
    throw new QuotaError(resource, limit, current);
  }
}

/** The transaction type `db.transaction(async (tx) => …)` hands its callback
 *  — extracted generically so helpers like `lockOrgForQuota` don't need a
 *  hard dependency on drizzle-orm's postgres-js internals. */
export type QuotaTx = Parameters<Parameters<typeof db.transaction>[0]>[0];

/**
 * Serializes quota-guarded writes for one org (sec audit A2/P8 fix, volet 1).
 *
 * `assertWithinQuota` is a plain read-then-decide with no lock (see its
 * doc). To close the race, wrap the read + the write in ONE transaction and
 * call this FIRST: a Postgres transaction-level advisory lock keyed on
 * `hashtext(orgId)`. A concurrent transaction trying to acquire the SAME
 * key blocks until this one commits or rolls back (`pg_advisory_xact_lock`
 * auto-releases then — no manual unlock, no leak risk on early
 * return/throw). By the time the second transaction gets the lock, the
 * first one's write (if it landed) is already committed and visible to its
 * own usage read (READ COMMITTED), so only as many concurrent callers as
 * the true remaining quota allows can ever pass.
 *
 * Advisory lock, not `SELECT … FROM organization WHERE id = $1 FOR UPDATE`:
 * a real row lock on `organization` would also serialize against UNRELATED
 * concurrent writes to that row — `assertMcpQuota`'s own
 * `UPDATE organization SET mcp_monthly_usage = …` (src/lib/mcp/quota.ts),
 * Stripe webhooks writing `stripeCustomerId`, admin god-mode trial edits.
 * None of those touch project/keyword/backlink-row counts, so blocking them
 * on an unrelated import would be needless contention. The advisory lock's
 * key namespace is scoped to quota-guarded writes only — nothing else in the
 * codebase takes a pg_advisory_xact_lock.
 *
 * One key per org (not per resource): imports are not a hot path, so the
 * simplicity of a single org-wide serialization point outweighs the (small)
 * extra contention between e.g. a concurrent site-create and keyword-import
 * for the same org.
 *
 * Call `assertWithinQuotaTx` (below) next, INSIDE the same transaction —
 * never plain `assertWithinQuota` (see its doc for why that deadlocks).
 */
export async function lockOrgForQuota(
  tx: QuotaTx,
  orgId: string,
): Promise<void> {
  await tx.execute(sql`select pg_advisory_xact_lock(hashtext(${orgId}))`);
}

/**
 * `assertWithinQuota`'s twin for use INSIDE an already-open transaction
 * (right after `lockOrgForQuota`, sharing the same `tx`) — sec audit A2/P8.
 *
 * Why this isn't just "`assertWithinQuota(orgId, resource, increment, tx)`":
 * that function resolves the LIMIT via `getOrgQuotaLimits`, which is cached
 * (react `cache()` + `unstable_cache`) but on a cache MISS queries through
 * the plain `db` handle — a connection request of its own. Requesting a
 * second connection while the caller's transaction already holds one
 * deadlocks the pool once concurrency reaches the pool size (default 10; 1 on
 * Vercel — see src/lib/db/client.ts) — confirmed by hanging
 * `quotas.test.ts`'s concurrency test until the limit was resolved outside
 * the transaction. So: resolve `limits = await getOrgQuotaLimits(orgId)`
 * BEFORE opening the transaction (its own connection, released before the
 * transaction begins — safe, the plan/override read is fine on a 60s-stale
 * cache same as everywhere else) and pass the single resolved `limit`
 * number in here. This then only touches the DB via `tx` — the live usage
 * COUNT (`getUsage(orgId, resource, tx)`), same connection as the lock, no
 * second connection ever requested.
 */
export async function assertWithinQuotaTx(
  tx: QuotaTx,
  orgId: string,
  resource: QuotaResource,
  limit: number,
  increment = 1,
): Promise<void> {
  if (limit === -1) return;
  const current = await getUsage(orgId, resource, tx);
  if (exceedsQuota(limit, current, increment)) {
    throw new QuotaError(resource, limit, current);
  }
}