spb/fetcha

1import { FetchaError, PLAN_LIMITS } from "@fetcha/core";2import { db, sql, usageEvents } from "@fetcha/db";3import type { ApiPrincipal } from "./auth";4import { getKV } from "./redis";56const CONCURRENCY_TTL_SEC = 150; // safety net if a release is lost78/** Reserve a concurrency slot for the organization. Returns a release function. */9export async function acquireConcurrency(p: ApiPrincipal): Promise<() => Promise<void>> {10  const kv = getKV();11  const limit = PLAN_LIMITS[p.plan].concurrency;12  const key = `fch:conc:${p.organizationId}`;13  const n = await kv.incrWithTtl(key, CONCURRENCY_TTL_SEC);14  if (n > limit) {15    await kv.decr(key);16    throw new FetchaError("CONCURRENCY_LIMIT", `Your plan allows ${limit} concurrent requests.`, { details: { limit } });17  }18  let released = false;19  return async () => {20    if (released) return;21    released = true;22    await kv.decr(key).catch(() => {});23  };24}2526/** Reserve a managed-browser slot for the organization (separate, smaller pool than plain fetches). */27export async function acquireBrowserSlot(p: ApiPrincipal): Promise<() => Promise<void>> {28  const kv = getKV();29  const limit = PLAN_LIMITS[p.plan].browser_concurrency;30  const key = `fch:bconc:${p.organizationId}`;31  const n = await kv.incrWithTtl(key, CONCURRENCY_TTL_SEC);32  if (n > limit) {33    await kv.decr(key);34    throw new FetchaError("CONCURRENCY_LIMIT", `At most ${limit} browser renders can run at once for your organization.`, { details: { limit, scope: "browser" } });35  }36  let released = false;37  return async () => {38    if (released) return;39    released = true;40    await kv.decr(key).catch(() => {});41  };42}4344/** Sliding-window rate limits: per key (burst), per organization (sustained), per IP (anonymous abuse). */45export async function checkRateLimits(p: ApiPrincipal, ip: string | null): Promise<void> {46  const kv = getKV();47  const limits = PLAN_LIMITS[p.plan];48  const perMinute = Math.max(60, limits.concurrency * 60); // generous: concurrency is the real cap49  const checks: Array<Promise<{ allowed: boolean; retryAfterMs: number }>> = [50    kv.slidingWindow(`fch:rl:org:${p.organizationId}`, 60_000, perMinute),51  ];52  if (p.keyId) checks.push(kv.slidingWindow(`fch:rl:key:${p.keyId}`, 1_000, Math.max(10, limits.concurrency * 2)));53  if (ip) checks.push(kv.slidingWindow(`fch:rl:ip:${ip}`, 1_000, 200));54  const results = await Promise.all(checks);55  const blocked = results.find((r) => !r.allowed);56  if (blocked) throw new FetchaError("RATE_LIMITED", undefined, { details: { retry_after_ms: blocked.retryAfterMs } });57}5859export interface MonthlyUsage {60  requests: number;61  spendUsd: number;62  projectRequests: number;63  projectSpendUsd: number;64}6566export async function monthlyUsage(p: ApiPrincipal): Promise<MonthlyUsage> {67  const kv = getKV();68  const cacheKey = `fch:usage:${p.organizationId}:${p.projectId}`;69  const cached = await kv.get(cacheKey).catch(() => null);70  if (cached) return JSON.parse(cached) as MonthlyUsage;71  const start = new Date();72  start.setUTCDate(1);73  start.setUTCHours(0, 0, 0, 0);74  const [org] = await db75    .select({76      requests: sql<number>`coalesce(sum(case when ${usageEvents.metric} = 'request' then ${usageEvents.quantity} else 0 end), 0)::float`,77      spend: sql<number>`coalesce(sum(${usageEvents.costUsd}), 0)::float`,78    })79    .from(usageEvents)80    .where(sql`${usageEvents.organizationId} = ${p.organizationId} and ${usageEvents.createdAt} >= ${start}`);81  const [proj] = await db82    .select({83      requests: sql<number>`coalesce(sum(case when ${usageEvents.metric} = 'request' then ${usageEvents.quantity} else 0 end), 0)::float`,84      spend: sql<number>`coalesce(sum(${usageEvents.costUsd}), 0)::float`,85    })86    .from(usageEvents)87    .where(sql`${usageEvents.projectId} = ${p.projectId} and ${usageEvents.createdAt} >= ${start}`);88  const usage: MonthlyUsage = { requests: Number(org?.requests ?? 0), spendUsd: Number(org?.spend ?? 0), projectRequests: Number(proj?.requests ?? 0), projectSpendUsd: Number(proj?.spend ?? 0) };89  await kv.set(cacheKey, JSON.stringify(usage), 20).catch(() => {});90  return usage;91}9293/** Enforce plan quota and hard spending limits. Soft limits are handled by the alerting job. */94export async function checkMonthlyLimits(p: ApiPrincipal): Promise<MonthlyUsage> {95  const usage = await monthlyUsage(p);96  const limits = PLAN_LIMITS[p.plan];97  if (usage.requests >= limits.monthly_requests) {98    throw new FetchaError("USAGE_LIMIT_REACHED", `Monthly request quota of ${limits.monthly_requests.toLocaleString("en-US")} reached for the ${limits.label} plan.`, {99      details: { limit: limits.monthly_requests, used: usage.requests },100    });101  }102  if (p.projectMonthlyRequestLimit && usage.projectRequests >= p.projectMonthlyRequestLimit) {103    throw new FetchaError("USAGE_LIMIT_REACHED", "Project monthly request limit reached.", { details: { limit: p.projectMonthlyRequestLimit, used: usage.projectRequests } });104  }105  if (p.orgHardLimitUsd !== null && usage.spendUsd >= p.orgHardLimitUsd) {106    throw new FetchaError("USAGE_LIMIT_REACHED", "Organization hard spending limit reached.", { details: { limit_usd: p.orgHardLimitUsd, spent_usd: usage.spendUsd } });107  }108  if (p.projectHardLimitUsd !== null && usage.projectSpendUsd >= p.projectHardLimitUsd) {109    throw new FetchaError("USAGE_LIMIT_REACHED", "Project hard spending limit reached.", { details: { limit_usd: p.projectHardLimitUsd, spent_usd: usage.projectSpendUsd } });110  }111  return usage;112}113114export async function invalidateUsageCache(organizationId: string, projectId: string): Promise<void> {115  await getKV().del(`fch:usage:${organizationId}:${projectId}`).catch(() => {});116}117