import { createHmac } from "node:crypto";
import { isIPv4 } from "node:net";
import { config } from "../config";
import {
  KEYLESS_TEAM_PREFIX,
  keylessTeamId,
  keylessTeamUuid,
} from "./keyless-team";
import { db } from "../db/connection";
import * as schema from "../db/schema";
import { redisRateLimitClient } from "../services/rate-limiter";
import { isKeylessIpSuspicious } from "./spur";
import { logger } from "./logger";
import {
  keylessCreditBlocksTotal,
  keylessCreditsTotal,
} from "./keyless-metrics";
import {
  type KeylessPromptReason,
  type KeylessSignupSurface,
  keylessFallbackSignupUrl,
  keylessSignupLink,
  keylessSignupSurface,
} from "./keyless-signup-link";
import { trackKeylessPromptShown } from "./keyless-prompt-analytics";

// Keyless free tier: scrape, search, and interact can be used without an API key
// from the official MCP server, CLI, or SDKs. It's gated per-IP/day by TWO
// limits, both configurable via env: a request count and a credit budget.
// `origin`/`integration` are client-set and spoofable, so they're only a soft
// gate to keep raw API callers on the signup path — the per-IP daily caps plus
// the `keyless/consume` canonical log are the real abuse controls.

// No defaults: the keyless free tier is OFF unless BOTH limits are configured.
const KEYLESS_REQUESTS_PER_DAY = config.KEYLESS_REQUESTS_PER_DAY;
const KEYLESS_CREDITS_PER_DAY = config.KEYLESS_CREDITS_PER_DAY;

// Keyless prompts link to signup at firecrawl.dev/k/<token>, where <token>
// is the encrypted prompt (see keyless-signup-link.ts). The constant uses the
// regular signup link and stays the marker for internal equality checks;
// responses swap in the caller's own link where they leave the API. The URL
// ends its line, so a copied link never picks up punctuation.
function keylessFreeTierLimitMessage(signupUrl: string): string {
  return `You've hit Firecrawl's keyless free tier rate limit. To continue now, create a free API key at ${signupUrl}

Then authenticate with:
Authorization: Bearer YOUR_API_KEY`;
}

export const KEYLESS_FREE_TIER_LIMIT_MESSAGE = keylessFreeTierLimitMessage(
  keylessFallbackSignupUrl("api"),
);

// The tier is "configured" when BOTH limits are set — even to 0. Unset means the
// feature is off (callers get a plain Unauthorized); 0 means it's on but the
// budget is exhausted (callers get the 429 cap message).
export function isKeylessConfigured(): boolean {
  return (
    typeof KEYLESS_REQUESTS_PER_DAY === "number" &&
    typeof KEYLESS_CREDITS_PER_DAY === "number"
  );
}

const DAY_SECONDS = 86400;

// A World ID verified human gets their own bucket, keyed on a hash of their
// World ID subject instead of the client IP, with 20% more requests and
// credits than the per-IP tier.
const WORLD_ID_IDENTITY_PREFIX = "wid_";
const WORLD_ID_LIMIT_MULTIPLIER = 1.2;

export function keylessWorldIdIdentity(subjectHash: string): string {
  return `${WORLD_ID_IDENTITY_PREFIX}${subjectHash}`;
}

/** Daily limits for a keyless identity (a client IP or a World ID identity). */
function keylessLimitsFor(identity: string): {
  requests: number;
  credits: number;
} {
  const requests = KEYLESS_REQUESTS_PER_DAY ?? 0;
  const credits = KEYLESS_CREDITS_PER_DAY ?? 0;
  if (!identity.startsWith(WORLD_ID_IDENTITY_PREFIX)) {
    return { requests, credits };
  }
  return {
    requests: Math.floor(requests * WORLD_ID_LIMIT_MULTIPLIER),
    credits: Math.floor(credits * WORLD_ID_LIMIT_MULTIPLIER),
  };
}

// This value is emitted only on quota exhaustion. It is a versioned, keyed
// pseudonym—not an IP address—so analytics can join the event to the existing
// privacy-controlled signup/OAuth matching pipeline without expanding raw-IP
// logging. Rotation intentionally creates a new cohort namespace.
export const KEYLESS_CONVERSION_COHORT_VERSION = "v1";

// Canonicalizes an IPv4-mapped IPv6 address (`::ffff:1.2.3.4`, what Node's
// dual-stack sockets report) to plain IPv4. Every per-IP artifact (quota
// buckets, team ids, Spur cache) must key off this form, or the same client
// counts as two identities.
export function normalizeKeylessIpv4(ip: string): string {
  const trimmed = ip.trim();
  const lower = trimmed.toLowerCase();
  return lower.startsWith("::ffff:") && isIPv4(trimmed.slice(7))
    ? trimmed.slice(7)
    : trimmed;
}

export function keylessConversionCohort(ip: string): string | undefined {
  const secret = config.KEYLESS_CONVERSION_HMAC_SECRET;
  const normalizedIp = normalizeKeylessIpv4(ip);
  if (!secret || !normalizedIp) return undefined;
  return `${KEYLESS_CONVERSION_COHORT_VERSION}:${createHmac("sha256", secret)
    .update(normalizedIp)
    .digest("base64url")}`;
}

export function keylessExhaustionTelemetry(ip: string): Record<string, string> {
  // The conversion cohort joins on a client IP; a World ID bucket has none.
  if (ip.startsWith(WORLD_ID_IDENTITY_PREFIX)) return {};
  const conversionCohort = keylessConversionCohort(ip);
  return conversionCohort ? { conversionCohort } : {};
}

export { keylessTeamId, keylessTeamUuid };

export function isKeylessTeamId(teamId: string): boolean {
  return teamId.startsWith(KEYLESS_TEAM_PREFIX);
}

// The keyless identity (quota bucket) of a team: a client IP or a World ID
// identity. Quota keys take it as is; prompts take keylessPromptIp.
function keylessIpFromTeamId(teamId: string): string | null {
  return teamId.startsWith(KEYLESS_TEAM_PREFIX)
    ? teamId.slice(KEYLESS_TEAM_PREFIX.length)
    : null;
}

/** The client IP a signup prompt may key on: none for a World ID bucket. */
function keylessPromptIp(identity: string | null): string | null {
  return identity && !identity.startsWith(WORLD_ID_IDENTITY_PREFIX)
    ? identity
    : null;
}

/**
 * Keyless is allowed only for a *valid IPv4* client identity. This both:
 *  - denies IPv6 (a single client controls a huge block — a /64 is ~18
 *    quintillion addresses — so a per-IP cap is trivially bypassed), and
 *  - rejects malformed/unknown values (e.g. a forwarded `x-firecrawl-keyless-ip`
 *    that isn't a real IP), so they can't be used to mint arbitrary buckets and
 *    weaken per-IP quota enforcement.
 * IPv4-mapped IPv6 (e.g. "::ffff:1.2.3.4", how dual-stack sockets surface IPv4)
 * is treated as IPv4.
 */
export function isKeylessIpEligible(ip: string): boolean {
  return isIPv4(normalizeKeylessIpv4(ip));
}

const requestsKey = (ip: string) => `keyless_requests:${ip}`;
const creditsKey = (ip: string) => `keyless_credits:${ip}`;

type KeylessQuotaReason = "requests" | "credits";

type KeylessConsumeResult = {
  ok: boolean;
  reason?: KeylessQuotaReason;
  requestsUsed: number;
  creditsUsed: number;
  retryAfterSeconds?: number;
};

function positiveRedisTtl(pttl: number): number | undefined {
  return pttl > 0 ? Math.ceil(pttl / 1000) : undefined;
}

/**
 * Quota denials remain valid when Redis cannot provide the optional TTL hint.
 * Keep that secondary lookup from changing a controlled 429 into a generic
 * authentication failure.
 */
async function retryAfterSecondsFor(key: string): Promise<number | undefined> {
  try {
    return positiveRedisTtl(await redisRateLimitClient.pttl(key));
  } catch {
    return undefined;
  }
}

/**
 * Signup link for a keyless prompt: the caller's own /k/<token> link for a
 * valid IPv4 identity, the regular signup link for anything else.
 */
export function keylessSignupUrlForIp(
  ip: string | null | undefined,
  surface: KeylessSignupSurface,
  reason: KeylessPromptReason,
): { url: string; signupRef?: string } {
  return keylessSignupLink(
    ip && isKeylessIpEligible(ip) ? normalizeKeylessIpv4(ip) : null,
    surface,
    reason,
  );
}

/**
 * Report a keyless prompt shown to the caller, for the prompt to signup
 * funnel. Deduplicated per identity, surface and reason per UTC day; never
 * blocks or throws. Only a keyless identity (IPv4) is reported: no other
 * caller can hold a token link or a keyless ledger row to join on.
 */
export function reportKeylessPromptShown(
  ip: string | null | undefined,
  surface: KeylessSignupSurface,
  reason: KeylessPromptReason,
  httpStatus: number,
  signupRef?: string,
): void {
  try {
    const teamUuid =
      ip && isKeylessIpEligible(ip)
        ? keylessTeamUuid(keylessTeamId(normalizeKeylessIpv4(ip)))
        : null;
    if (!teamUuid) return;
    trackKeylessPromptShown({
      keylessTeamId: teamUuid,
      surface,
      reason,
      httpStatus,
      tokenLink: Boolean(signupRef),
    });
  } catch {
    // Analytics only: the prompt itself must still go out.
  }
}

/**
 * The caller's own signup link and the limit message that carries it. Every
 * caller answers 429 with it, so the prompt is reported here.
 */
export function keylessLimitPrompt(
  ip: string | null | undefined,
  surface: KeylessSignupSurface,
): { error: string; signup_url: string; signupRef?: string } {
  const { url, signupRef } = keylessSignupUrlForIp(ip, surface, "limit");
  reportKeylessPromptShown(ip, surface, "limit", 429, signupRef);
  return {
    error: keylessFreeTierLimitMessage(url),
    signup_url: url,
    ...(signupRef ? { signupRef } : {}),
  };
}

/** keylessLimitPrompt for a keyless team id; the regular signup link for other teams. */
export function keylessLimitPromptForTeam(
  teamId: string,
  req: Parameters<typeof keylessSignupSurface>[0],
): ReturnType<typeof keylessLimitPrompt> {
  return keylessLimitPrompt(
    keylessPromptIp(keylessIpFromTeamId(teamId)),
    keylessSignupSurface(req),
  );
}

/** Structured response for projected-credit reservation exhaustion. */
export async function keylessLimitBody(
  teamId: string,
  mode: string,
  req?: Parameters<typeof keylessSignupSurface>[0],
): Promise<{
  success: false;
  error: string;
  reason: "credits";
  signup_url: string;
  retry_after_seconds?: number;
}> {
  const ip = keylessIpFromTeamId(teamId);
  let retryAfterSeconds: number | undefined;
  try {
    retryAfterSeconds = ip
      ? positiveRedisTtl(await redisRateLimitClient.pttl(creditsKey(ip)))
      : undefined;
  } catch {
    // The reservation already proved the limit; missing TTL must not turn its
    // controlled 429 into a server error.
  }
  const prompt = keylessLimitPrompt(
    keylessPromptIp(ip),
    req ? keylessSignupSurface(req) : "api",
  );
  logger.warn("Keyless request blocked", {
    canonicalLog: "keyless/consume",
    event: "keyless_exhausted",
    blocked: true,
    reason: "credits",
    mode,
    retryAfterSeconds,
    ...(prompt.signupRef ? { signupRef: prompt.signupRef } : {}),
    ...keylessExhaustionTelemetry(ip ?? ""),
  });
  return {
    success: false,
    error: prompt.error,
    reason: "credits",
    signup_url: prompt.signup_url,
    ...(retryAfterSeconds ? { retry_after_seconds: retryAfterSeconds } : {}),
  };
}

type KeylessCreditReservationResult = {
  ok: boolean;
  creditsUsed: number;
  limit: number;
};

/**
 * Consume one request from the per-IP daily request budget and check the credit
 * budget (credits are charged after the request completes, in
 * `chargeKeylessCredits`). Returns whether the request may proceed.
 */
export async function consumeKeylessRequest(
  ip: string,
): Promise<KeylessConsumeResult> {
  const { requests: requestLimit, credits: creditLimit } = keylessLimitsFor(ip);

  const rKey = requestsKey(ip);
  const requestsUsed = await redisRateLimitClient.incr(rKey);
  if (requestsUsed === 1) {
    await redisRateLimitClient.expire(rKey, DAY_SECONDS);
  }

  const creditsUsed = parseInt(
    (await redisRateLimitClient.get(creditsKey(ip))) ?? "0",
    10,
  );

  if (requestsUsed > requestLimit) {
    return {
      ok: false,
      reason: "requests",
      requestsUsed,
      creditsUsed,
      retryAfterSeconds: await retryAfterSecondsFor(rKey),
    };
  }
  if (creditsUsed >= creditLimit) {
    return {
      ok: false,
      reason: "credits",
      requestsUsed,
      creditsUsed,
      retryAfterSeconds: await retryAfterSecondsFor(creditsKey(ip)),
    };
  }
  return { ok: true, requestsUsed, creditsUsed };
}

/**
 * Atomically reserve projected credits from the keyless per-IP daily credit
 * budget. No-op for non-keyless teams.
 */
export async function reserveKeylessCredits(
  teamId: string,
  projectedCredits: number,
): Promise<KeylessCreditReservationResult> {
  const ip = keylessIpFromTeamId(teamId);
  const limit = ip ? keylessLimitsFor(ip).credits : 0;
  if (!ip || !Number.isFinite(projectedCredits) || projectedCredits <= 0) {
    return { ok: true, creditsUsed: 0, limit };
  }

  const projected = Math.ceil(projectedCredits);
  const key = creditsKey(ip);
  const result = (await redisRateLimitClient.eval(
    `
local current = tonumber(redis.call("GET", KEYS[1]) or "0")
local projected = tonumber(ARGV[1])
local limit = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
if current + projected > limit then
  return {0, current}
end
local total = redis.call("INCRBY", KEYS[1], projected)
if redis.call("TTL", KEYS[1]) < 0 then
  redis.call("EXPIRE", KEYS[1], ttl)
end
return {1, total}
`,
    1,
    key,
    projected,
    limit,
    DAY_SECONDS,
  )) as [number, number];

  if (result[0] !== 1) keylessCreditBlocksTotal.inc();
  return {
    ok: result[0] === 1,
    creditsUsed: Number(result[1] ?? 0),
    limit,
  };
}

/**
 * Reconcile a prior keyless reservation to actual credits. Delta may be
 * negative; the counter is clamped at zero to tolerate retries or races.
 * The daily window keeps the expiry set on first use.
 */
export async function adjustKeylessCredits(
  teamId: string,
  deltaCredits: number,
): Promise<number | null> {
  const ip = keylessIpFromTeamId(teamId);
  if (!ip || !Number.isFinite(deltaCredits) || deltaCredits === 0) return null;

  const delta =
    deltaCredits > 0
      ? Math.ceil(deltaCredits)
      : -Math.ceil(Math.abs(deltaCredits));
  const key = creditsKey(ip);
  const total = (await redisRateLimitClient.eval(
    `
local current = tonumber(redis.call("GET", KEYS[1]) or "0")
local delta = tonumber(ARGV[1])
local ttl = tonumber(ARGV[2])
if delta <= 0 and redis.call("EXISTS", KEYS[1]) == 0 then
  return 0
end
local next = current + delta
if next < 0 then
  next = 0
end
redis.call("SET", KEYS[1], next, "KEEPTTL")
if redis.call("TTL", KEYS[1]) < 0 then
  redis.call("EXPIRE", KEYS[1], ttl)
end
return next
`,
    1,
    key,
    delta,
    DAY_SECONDS,
  )) as number;

  return Number(total);
}

/** Absolute browser reservations, including prompt upgrades, with atomic receipts. */
export async function updateKeylessBrowserCredits(
  teamId: string,
  sessionId: string,
  credits: number,
  finalize = false,
): Promise<boolean> {
  const ip = keylessIpFromTeamId(teamId);
  if (!ip) return true;
  const result = await redisRateLimitClient.eval(
    `
    local now = tonumber(ARGV[1])
    local target = tonumber(ARGV[2])
    local final = ARGV[3] == '1'
    local reserved = tonumber(redis.call('HGET', KEYS[2], 'credits') or '0')
    local deadline = tonumber(redis.call('HGET', KEYS[2], 'deadline') or '0')
    if redis.call('HGET', KEYS[2], 'final') == '1' then
      if final then return 1 else return 0 end
    end
    -- An expired budget cannot be refunded into a different day's allowance.
    if deadline > 0 and deadline <= now then
      if final then return 1 end
      reserved = 0
      deadline = 0
    end
    if final and deadline == 0 then return 1 end
    local delta = target - reserved
    if not final and delta <= 0 then return 1 end
    local current = tonumber(redis.call('GET', KEYS[1]) or '0')
    if not final and current + delta > tonumber(ARGV[4]) then return 0 end
    local ttl = redis.call('PTTL', KEYS[1])
    if ttl <= 0 then ttl = 86400000 end
    if deadline == 0 then deadline = now + ttl end
    redis.call('SET', KEYS[1], math.max(0, current + delta), 'PX', ttl)
    redis.call('HSET', KEYS[2], 'credits', target, 'deadline', deadline, 'final', ARGV[3])
    redis.call('EXPIRE', KEYS[2], 172800)
    return 1
  `,
    2,
    creditsKey(ip),
    `keyless_browser:${sessionId}`,
    Date.now(),
    Math.ceil(credits),
    finalize ? "1" : "0",
    keylessLimitsFor(ip).credits,
  );
  if (result !== 1 && !finalize) keylessCreditBlocksTotal.inc();
  return result === 1;
}

/**
 * Read-only check of whether an IP could currently use the keyless tier (no
 * consumption). Used by the hosted MCP before a keyless tool call so an
 * ineligible caller receives structured recovery without an OAuth challenge.
 * A World ID identity is checked against its own bucket and skips the IP
 * checks, as it does in auth.
 */
export async function checkKeylessEligibility(
  ip: string,
  worldIdIdentity?: string,
): Promise<{
  eligible: boolean;
  reason?:
    | KeylessQuotaReason
    | "disabled"
    | "ineligible_ip"
    | "suspicious"
    | "error";
  retryAfterSeconds?: number;
}> {
  if (!isKeylessConfigured()) return { eligible: false, reason: "disabled" };
  if (worldIdIdentity) {
    ip = worldIdIdentity;
  } else {
    if (!ip || !isKeylessIpEligible(ip)) {
      return { eligible: false, reason: "ineligible_ip" };
    }
    // Key the Spur cache and quota buckets below off the canonical IPv4 form.
    ip = normalizeKeylessIpv4(ip);
    // Optional Spur Context check (only when SPUR_API_KEY is set): treat IPs on
    // anonymizing/rotating infrastructure as ineligible so the hosted MCP can
    // return a bounded recovery result instead of serving a request auth
    // rejects.
    if (await isKeylessIpSuspicious(ip)) {
      return { eligible: false, reason: "suspicious" };
    }
  }
  const limits = keylessLimitsFor(ip);
  try {
    const requestsUsed = parseInt(
      (await redisRateLimitClient.get(requestsKey(ip))) ?? "0",
      10,
    );
    if (requestsUsed >= limits.requests) {
      return {
        eligible: false,
        reason: "requests",
        retryAfterSeconds: await retryAfterSecondsFor(requestsKey(ip)),
      };
    }
    const creditKey = creditsKey(ip);
    const creditsUsed = parseInt(
      (await redisRateLimitClient.get(creditKey)) ?? "0",
      10,
    );
    if (creditsUsed >= limits.credits) {
      return {
        eligible: false,
        reason: "credits",
        retryAfterSeconds: await retryAfterSecondsFor(creditKey),
      };
    }
    return { eligible: true };
  } catch {
    // Limiter store unavailable — fail closed so the MCP returns structured
    // recovery rather than granting unbounded keyless access.
    return { eligible: false, reason: "error" };
  }
}

/**
 * Record a completed keyless request — the only place a keyless caller's IP is
 * persisted (`requests.team_id` collapses all keyless traffic onto one preview
 * team).
 *
 * Billable requests append a row to `keyless_credit_usage` exactly as before.
 *
 * Zero-credit operations are recorded too, but as a canonical log line rather
 * than a DB row: the free Research Index paper endpoints bill nothing, so
 * skipping them meant their client IP was never written anywhere and abuse on
 * them was invisible (during the corpus-harvest incident only 3.0% of the
 * traffic had a resolvable IP). Zero-credit volume is orders of magnitude
 * higher than billable volume, so the durable row for it is deferred until the
 * companion firecrawl-db migration lands; until then the log line carries the
 * same fields.
 *
 * No-op for non-keyless teams. Best-effort — never throws.
 */
export async function logKeylessCreditUsage(
  teamId: string,
  credits: number,
): Promise<void> {
  const ip = keylessIpFromTeamId(teamId);
  if (!ip || !Number.isFinite(credits)) return;
  const teamUuid = keylessTeamUuid(teamId);
  if (!teamUuid) return;

  // The column records consumption, so a negative reconciliation delta must
  // not land as a negative value.
  const creditsUsed = Math.max(0, Math.ceil(credits));

  // Both branches sit behind the same gate as every other keyless usage
  // record: self-hosted deployments without DB auth track nothing.
  if (config.USE_DB_AUTHENTICATION !== true) return;

  keylessCreditsTotal.inc(creditsUsed);

  // `keyless_credit_usage.ip` and the zero-credit line hold the raw client
  // IP, which a World ID bucket isn't keyed on. Its usage goes to the same
  // canonical log under the team id instead; the client IP of each World ID
  // request is already on its `keyless/consume` line.
  if (ip.startsWith(WORLD_ID_IDENTITY_PREFIX)) {
    logger.info("Keyless World ID usage", {
      canonicalLog: "keyless/usage",
      teamId: teamUuid,
      worldId: true,
      creditsUsed,
    });
    return;
  }

  if (creditsUsed <= 0) {
    // TODO(firecrawl-db): switch to a `keyless_credit_usage` row once the
    // zero-credit usage migration is merged. The IP is repeated in the
    // message body because the console transport only serializes metadata
    // for warn/error lines; the structured fields are the contract for
    // Cloud Logging queries.
    logger.info(`Keyless zero-credit usage ip=${ip} team=${teamUuid}`, {
      canonicalLog: "keyless/usage",
      ip,
      teamId: teamUuid,
      creditsUsed: 0,
    });
    return;
  }

  try {
    await db.insert(schema.keyless_credit_usage).values({
      team_id: teamUuid,
      ip,
      credits_used: creditsUsed,
    });
  } catch {
    // Logging is best-effort.
  }
}

/**
 * Add the actual credits a completed request consumed to the IP's daily credit
 * counter, and record the request in `keyless_credit_usage`. The counter is only
 * touched for positive credits, so a zero-credit operation becomes observable
 * without drawing down any budget. No-op for non-keyless teams. Best-effort;
 * never throws. Used by the worker for the non-reserved path; the controllers
 * reserve up front and call `logKeylessCreditUsage` directly at reconciliation.
 */
export async function chargeKeylessCredits(
  teamId: string,
  credits: number,
): Promise<void> {
  const ip = keylessIpFromTeamId(teamId);
  if (!ip || !Number.isFinite(credits)) return;

  if (credits > 0) {
    const inc = Math.ceil(credits);
    try {
      const key = creditsKey(ip);
      await redisRateLimitClient.eval(
        `
redis.call("INCRBY", KEYS[1], ARGV[1])
if redis.call("TTL", KEYS[1]) < 0 then
  redis.call("EXPIRE", KEYS[1], ARGV[2])
end
`,
        1,
        key,
        inc,
        DAY_SECONDS,
      );
    } catch {
      // Counter is best-effort; a missed charge just means the IP gets a few
      // extra free credits today.
    }
  }

  // Log the usage to keyless_credit_usage for abuse monitoring. Best-effort.
  await logKeylessCreditUsage(teamId, credits);
}