import * as Sentry from "@sentry/cloudflare"; // --------------------------------------------------------------------------- // Router puro. El Worker YA NO convierte PDFs: recibe la peticion, autentica, // aplica rate-limit por consumer y la reenvia (proxy) a uno de los upstreams // del pool, balanceando carga con failover. // // Los backends (APIs reales) se anaden mas tarde via la env UPSTREAMS. Mientras // el pool este vacio, el router responde 503 (no hay donde enrutar). // --------------------------------------------------------------------------- const DEFAULT_SENTRY_DSN = "https://f888d18229479849c4cde8d260ce4598@o4511293736091648.ingest.de.sentry.io/4511593877340240"; // Headers que NUNCA se reenvian (hop-by-hop, RFC 7230) ni se devuelven: son de // la conexion concreta, no del mensaje. Reenviarlos rompe el proxy. const HOP_BY_HOP = new Set([ "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade", "host", "content-length", ]); // Headers de respuesta que delatan la arquitectura interna multi-cloud. Se // borran antes de devolver al cliente. const STRIP_RESPONSE_HEADERS = [ "x-conversion-engine", "x-recommended-provider", "x-route-class", // Senales de carga del backend concreto que sirvio ESTA peticion. El router // las consume (passive health) para enrutar; no deben llegar al cliente (delatan // que backend respondio). El estado agregado del pool se expone aparte en // X-Pool-Health / /healthz. "x-svc-provider", "x-svc-active", "x-svc-waiting", "x-svc-limit", "x-svc-load", "x-svc-state", "via", "server", ]; // 5xx que justifican reintentar en otro upstream (el backend esta caido/saturado, // no es culpa del cliente). 500 NO se reintenta: suele ser error real del backend. const RETRYABLE_STATUS = new Set([502, 503, 504]); // 413 tambien se reintenta en otro upstream: el tope real es por-backend (p.ej. // Vercel corta a ~4.5 MB en su funcion serverless) y no refleja el limite // contractual del router (RAW_UPLOAD_MAX_BYTES, 12 MB). Un body que YA paso el // gate de tamano del edge no deberia fallar solo porque el primer backend // elegido tiene un techo mas bajo que otro del pool. const UPSTREAM_RETRYABLE_STATUS = new Set([...RETRYABLE_STATUS, 413]); // Tope de subida por defecto (100 MB). El router bufferiza el body para poder // reintentarlo en otro upstream, y el isolate de CF tiene 128 MB de RAM dura => // el tope debe quedar holgadamente por debajo. Ajustable por env MAX_UPLOAD_BYTES. const DEFAULT_MAX_UPLOAD_BYTES = 100 * 1024 * 1024; // Tope ESTRICTO de subida CRUDA por la ruta /convert/raw|/convert|/convert/stream // (mismo contrato que el backend: RAW_UPLOAD_MAX_BYTES=12 MB). Si el cliente // intenta subir mas de esto por raw, le devolvemos 413 con un mensaje accionable // que recomienda usar /convert/from-url o /convert/accelerated. Asi el edge // rechaza temprano (sin gastar bandwidth al backend) y replica exactamente el // el edge rechaza temprano sin gastar bandwidth al backend. const DEFAULT_RAW_UPLOAD_MAX_BYTES = 12 * 1024 * 1024; // Rutas a las que aplica el tope estricto de subida cruda. const RAW_UPLOAD_PATHS = new Set(["/convert/raw", "/convert", "/convert/stream"]); // Rutas que NUNCA llevan body grande (health/status/route/...): saltamos // totalmente el buffer body para latencia minima en el camino caliente. const NO_BODY_PATHS = new Set([ "/", "/healthz", "/status", "/warmup", "/me", "/usage", "/route/estimate", "/route/plan", ]); function sentryConfig(env) { return { dsn: env.SENTRY_DSN || DEFAULT_SENTRY_DSN, environment: env.SENTRY_ENVIRONMENT || env.ENVIRONMENT || "production", release: env.SENTRY_RELEASE || "pdf-to-markdown-api-router", tracesSampleRate: Number(env.SENTRY_TRACES_SAMPLE_RATE || "0.05"), sendDefaultPii: false, beforeSend(event) { if (event.request?.url) event.request.url = event.request.url.split("?")[0]; if (event.request) delete event.request.query_string; return event; }, }; } function captureException(error, tags = {}) { try { Sentry.withScope((scope) => { Object.entries(tags).forEach(([key, value]) => scope.setTag(key, String(value))); Sentry.captureException(error); }); } catch { // Sentry nunca debe romper el enrutado. } } // Parsea UPSTREAMS: lista separada por comas de "https://host[#peso]". // Ej: "https://a.example.com#70, https://b.example.com#30". Sin peso => 1. // Se valida que sea https y se descartan entradas invalidas. function parseUpstreams(raw) { if (!raw) return []; return raw .split(",") .map((entry) => entry.trim()) .filter(Boolean) .map((entry) => { const [urlPart, weightPart] = entry.split("#"); const weight = Math.max(1, Number(weightPart) || 1); let url; try { url = new URL(urlPart.trim()); } catch { return null; } if (url.protocol !== "https:") return null; return { origin: url.origin, weight }; }) .filter(Boolean); } // --- Auto-async: degrada a 202+job_id cuando todos los backends estan saturados. // Usa Cloudflare Queue (cola real FIFO con retry) + Cache API (estado gratis). const DEGRADABLE_PATHS = new Set([ "/convert", "/convert/raw", "/convert/from-url", "/convert/stream", "/convert/stream-from-url", "/convert/accelerated", ]); // Rutas de streaming pagina a pagina. En AWS, la Lambda principal (Mangum) // bufferiza toda la respuesta antes de devolverla -> streaming real inutil ahi. // Si AWS_STREAMING_ORIGIN esta configurado (Function URL de la Lambda separada // con Web Adapter + InvokeMode=RESPONSE_STREAM), la reemplazamos SOLO para // estas rutas cuando el upstream elegido es AWS. El resto de proveedores // (Cloudflare Workers, Scaleway, Cloud Run) ya streamean real sin este truco. // Detalle: docs/2026-07-01-aws-lambda-streaming-web-adapter.md const STREAMING_PATHS = new Set(["/convert/stream", "/convert/stream-from-url"]); const OVERLOAD_THRESHOLD = 0.85; const JOB_CACHE_TTL = 3600; const JOB_CACHE_PREFIX = "https://job-status.internal/"; function allBackendsSaturated(upstreams) { if (upstreams.length === 0) return false; let knownCount = 0; let saturatedCount = 0; for (const upstream of upstreams) { const health = freshHealth(upstream.origin); if (!health) continue; knownCount++; if (health.load >= OVERLOAD_THRESHOLD || health.state === "saturated") { saturatedCount++; } } if (knownCount === 0) return false; return saturatedCount === knownCount; } async function cacheJobStatus(jobId, data) { const cache = caches.default; const url = JOB_CACHE_PREFIX + jobId; const body = JSON.stringify({ ...data, updated_at: Date.now() }); await cache.put( new Request(url), new Response(body, { headers: { "content-type": "application/json", "cache-control": `public, max-age=${JOB_CACHE_TTL}`, }, }) ); } async function readJobStatus(jobId) { const cache = caches.default; const match = await cache.match(new Request(JOB_CACHE_PREFIX + jobId)); if (!match) return null; return match.json(); } // Cloudflare Queues tiene tope duro de 128KB por mensaje. Los bodies JSON // (from-url/accelerated: solo llevan una URL) entran de sobra en base64 // inline. Los uploads binarios (PDF crudo, hasta RAW_UPLOAD_MAX_BYTES) casi // siempre lo superan -> van a R2, la cola solo lleva la referencia. const INLINE_BODY_MAX_BYTES = 64 * 1024; const DEFAULT_MAX_QUEUE_DEPTH = 500; // Devuelve null si se encolo, o una Response 503 si la cola ya esta al tope // (backpressure real, cierra C9). El chequeo de cupo va ANTES de escribir a // R2/Queue para no gastar esas escrituras en un job que se va a rechazar. async function enqueueJob(env, jobId, request, url, body, auth, requestId, declaredInputBytes) { const maxDepth = Number(env.MAX_QUEUE_DEPTH || DEFAULT_MAX_QUEUE_DEPTH); const accepted = await incrQueueDepth(env, maxDepth); if (!accepted) return queueSaturatedResponse(requestId); try { let bodyB64 = null; let bodyR2Key = null; if (body) { if (body.byteLength > INLINE_BODY_MAX_BYTES && env.PDF_OVERFLOW) { bodyR2Key = `overflow/${jobId}`; await env.PDF_OVERFLOW.put(bodyR2Key, body); } else { bodyB64 = arrayBufferToBase64(body); } } const job = { job_id: jobId, created_at: Date.now(), request_id: requestId, method: request.method, path: url.pathname + url.search, auth_user_id: auth?.userId || "", auth_key_id: auth?.keyId || "", declared_input_bytes: declaredInputBytes || 0, headers: Object.fromEntries( [...request.headers].filter( ([k]) => !HOP_BY_HOP.has(k.toLowerCase()) && k.toLowerCase() !== "x-api-key" && k.toLowerCase() !== "apikey" && k.toLowerCase() !== "authorization" ) ), body_b64: bodyB64, body_r2_key: bodyR2Key, }; await env.CONVERSION_QUEUE.send(job); await cacheJobStatus(jobId, { job_id: jobId, status: "queued", created_at: job.created_at }); return null; } catch (err) { // El cupo ya se reservo (incrQueueDepth) pero el job nunca llego a la cola: // liberarlo, si no el contador queda inflado para siempre. await decrQueueDepth(env); throw err; } } // Borra el body en R2 apenas el job llega a un estado TERMINAL (exito o // fallo definitivo). Nunca se llama antes de un msg.retry(): el proximo // intento todavia necesita leer el archivo. async function deleteOverflowBody(env, job) { if (!job.body_r2_key || !env.PDF_OVERFLOW) return; try { await env.PDF_OVERFLOW.delete(job.body_r2_key); } catch { // best-effort: si falla, el archivo lo limpia igual la vida util corta // que aplica sobre el bucket (defensa en profundidad, no lo primario). } } function arrayBufferToBase64(buffer) { const bytes = new Uint8Array(buffer); let binary = ""; for (let i = 0; i < bytes.length; i++) binary += String.fromCharCode(bytes[i]); return btoa(binary); } function base64ToArrayBuffer(b64) { const binary = atob(b64); const bytes = new Uint8Array(binary.length); for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i); return bytes.buffer; } function queueSaturatedResponse(requestId) { return Response.json( { error: "Queue at capacity", detail: "All backends are saturated and the overflow queue is already at its depth cap. Retry shortly.", }, { status: 503, headers: { "x-request-id": requestId, "retry-after": "10" }, } ); } function asyncAcceptedResponse(jobId, requestId) { return Response.json( { accepted: true, auto_async: true, job_id: jobId, status: "queued", status_url: `/jobs/${jobId}`, retry_after_seconds: 5, message: "All backends are at capacity. Your request has been queued and will be processed " + "as soon as a backend becomes available. Poll /jobs/{job_id} for status.", }, { status: 202, headers: { "x-request-id": requestId, "x-job-id": jobId, "x-auto-async": "true", "retry-after": "5", }, } ); } // --- Passive health: carga viva de cada upstream ------------------------------ // Cada respuesta del backend trae su carga en cabeceras X-Svc-* (active/waiting/ // limit/load/state). El router las lee y guarda aqui, asi APRENDE cual backend // esta al limite SIN sondear /status aparte. Map EN MEMORIA del isolate (como // RATE_BUCKETS): cada isolate aprende de su propio trafico; basta para repartir // hacia el menos colapsado. Para una vista global exacta entre isolates haria // falta KV/Durable Object (mas latencia); el passive per-isolate es el rapido. const UPSTREAM_HEALTH = new Map(); // Tras este tiempo sin noticias, la salud se considera obsoleta y el upstream // vuelve a tratarse como "desconocido" (se le manda trafico para re-sondearlo). const HEALTH_TTL_MS = 15000; // Infiere el proveedor por hostname (fallback si el backend no manda X-Svc-Provider). function providerFromOrigin(origin) { const host = origin.toLowerCase(); if (host.includes("lambda-url") || host.includes("amazonaws")) return "aws"; if (host.includes("azurewebsites") || host.includes(".azure")) return "azure"; if (host.includes("scw.cloud") || host.includes("scaleway")) return "scaleway"; if (host.includes("vercel.app") || host.includes("vercel")) return "vercel"; if (host.includes("run.app") || host.includes("cloudfunctions")) return "gcp"; if (host.includes("fly.dev")) return "fly"; if (host.includes("fcapp.run") || host.includes("aliyuncs")) return "alibaba"; return "unknown"; } // Registra la carga reportada por un upstream tras responder. function recordUpstreamHealth(origin, headers) { const load = Number(headers.get("x-svc-load")); if (!Number.isFinite(load)) return; // backend viejo sin cabeceras X-Svc-* UPSTREAM_HEALTH.set(origin, { provider: headers.get("x-svc-provider") || providerFromOrigin(origin), load, active: Number(headers.get("x-svc-active") || 0), waiting: Number(headers.get("x-svc-waiting") || 0), limit: Number(headers.get("x-svc-limit") || 0), state: headers.get("x-svc-state") || "unknown", ts: Date.now(), }); } // Salud fresca de un origin, o null si falta/obsoleta. function freshHealth(origin) { const health = UPSTREAM_HEALTH.get(origin); if (!health || Date.now() - health.ts > HEALTH_TTL_MS) return null; return health; } // Penalidad fija por proveedor con cold start conocido (Azure Functions duerme // el isolate sin trafico reciente; retomarlo cuesta segundos reales de mas). // Es un empujon PROBABILISTICO, no una exclusion: debe quedar por debajo del // jitter (0.01) para que Azure siga ganando una fraccion real de los empates // "ambos ociosos/desconocidos", no 0 de N. Con 0.05 (> jitter) Azure perdia // SIEMPRE ese empate -- quedaba fuera del pool de facto. Con 0.004 gana ~1 de // cada 6-8 empates: sigue de verdad en rotacion, solo un poco menos preferido. // El caso de load YA conocido (mas abajo) usa un penalty mayor a proposito: // ahi el load es una medida real (0..1), no un empate ciego. const PROVIDER_COLD_START_PENALTY_UNKNOWN = { azure: 0.004 }; const PROVIDER_COLD_START_PENALTY_KNOWN = { azure: 0.05 }; // Score de enrutado (menor = se prueba antes): // - Desconocido/obsoleto -> ~0 con jitter (+ penalidad chica de cold start): // se trata como ocioso y se sondea, con leve preferencia por always-on. // - Saturado (cola) -> se hunde al fondo: solo failover si no hay nada mejor. // - Resto -> su load real + penalidad + jitter minimo para romper empates. function routingScore(origin) { const provider = providerFromOrigin(origin); const health = freshHealth(origin); if (!health) return (PROVIDER_COLD_START_PENALTY_UNKNOWN[provider] || 0) + Math.random() * 0.01; if (health.state === "saturated" || health.load >= 1) return 10 + health.load; return health.load + (PROVIDER_COLD_START_PENALTY_KNOWN[provider] || 0) + Math.random() * 0.001; } // Probabilidad de EXPLORAR en vez de explotar el ganador conocido. Sin esto, // un backend con load=0 YA CONOCIDO le gana casi siempre a uno "desconocido" // (su jitter es 10x mas chico), asi que el PRIMERO en responder en un isolate // queda pegado ahi mientras su load siga bajo -- los demas nunca reciben // trafico real en ESE isolate (confirmado: 5/5 requests al mismo backend pese // a haber 5 backends sanos). Con probabilidad EXPLORE_PROBABILITY se promueve // un upstream al azar al frente del orden, ignorando el score: garantiza que // TODOS los backends reciban trafico real con el tiempo, no solo el ganador // de la primera carrera. El resto del orden (fallback si el explorado falla) // sigue siendo el ranking normal por score. const EXPLORE_PROBABILITY = 0.2; // Orden de prueba: el MENOS cargado primero (least-loaded), peso mayor como // desempate (mas capacidad = preferido). El jitter de routingScore evita que // todos los isolates apunten al mismo backend ocioso a la vez (estampida). function buildAttemptOrder(upstreams) { const ranked = [...upstreams] .map((upstream) => ({ ...upstream, score: routingScore(upstream.origin) })) .sort((a, b) => a.score - b.score || b.weight - a.weight); if (ranked.length > 1 && Math.random() < EXPLORE_PROBABILITY) { const exploreIndex = Math.floor(Math.random() * ranked.length); const [explored] = ranked.splice(exploreIndex, 1); ranked.unshift(explored); } return ranked; } // Tope de body conocido POR PROVEEDOR (no configurable por request): Vercel // corta funciones serverless a ~4.5 MB de payload sin importar RAW_UPLOAD_MAX_BYTES. // Reintentar contra el ya sabemos que dara 413 cuesta un upload entero del PDF // desperdiciado (segundos en un body de varios MB). Mejor sacarlo del orden de // intento ANTES de subir un solo byte cuando el body ya supera su tope conocido. const KNOWN_UPSTREAM_BODY_LIMITS = [{ suffix: "vercel.app", maxBytes: 4 * 1024 * 1024 }]; function upstreamBodyLimit(origin) { const hostname = (() => { try { return new URL(origin).hostname; } catch { return ""; } })(); const match = KNOWN_UPSTREAM_BODY_LIMITS.find((entry) => hostname.endsWith(entry.suffix)); return match ? match.maxBytes : Infinity; } // Filtra del orden los backends cuyo tope conocido ya sabemos que el body supera. // Si TODOS quedarian filtrados (body enorme incluso para el mas grande), se // mantiene el orden original: mejor un intento real (y el fallback 413->cola) // que rechazar sin darle chance a un backend sin tope documentado. function filterOrderByBodySize(order, bodyBytes) { if (!bodyBytes) return order; const fits = order.filter((upstream) => bodyBytes <= upstreamBodyLimit(upstream.origin)); return fits.length > 0 ? fits : order; } // Sustituye el origin de AWS por la Lambda de streaming SOLO en rutas de // streaming y SOLO si esta configurada. No-op en cualquier otro caso: no // cambia el orden, el peso, ni el resto de proveedores del pool. function applyStreamingOverride(order, pathname, env) { if (!env.AWS_STREAMING_ORIGIN || !STREAMING_PATHS.has(pathname)) return order; return order.map((upstream) => providerFromOrigin(upstream.origin) === "aws" ? { ...upstream, origin: env.AWS_STREAMING_ORIGIN } : upstream, ); } // Foto del pool para exponer en X-Pool-Health / /healthz: que proveedor esta // ocioso y cual al limite. `load`/`state` null = aun sin datos en este isolate. function poolHealthSnapshot(upstreams) { return upstreams.map((upstream) => { const health = freshHealth(upstream.origin); return { provider: health ? health.provider : providerFromOrigin(upstream.origin), load: health ? health.load : null, state: health ? health.state : "unknown", active: health ? health.active : null, limit: health ? health.limit : null, age_ms: health ? Date.now() - health.ts : null, }; }); } function shouldExposePoolHealth(request, env) { if (String(env.POOL_HEALTH_PUBLIC || "").toLowerCase() === "true") return true; const operatorSecret = env.POOL_HEALTH_SECRET || env.ADMIN_SECRET || ""; return Boolean(operatorSecret && request.headers.get("x-operator-secret") === operatorSecret); } async function apiKeyFingerprint(apiKey) { const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(apiKey)); return Array.from(new Uint8Array(digest)) .map((byte) => byte.toString(16).padStart(2, "0")) .join("") .slice(0, 16); } // SHA-256 completo en hex (64 chars). Se usa como `key_hash` en D1: la API key // real del usuario NUNCA se guarda, solo su hash. En cada peticion hasheamos la // key entrante y buscamos por hash (lookup constante, sin exponer la clave). async function sha256Hex(value) { const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(value)); return Array.from(new Uint8Array(digest)) .map((byte) => byte.toString(16).padStart(2, "0")) .join(""); } // --- Validacion de API key contra D1 (store real de keys en el edge) -------- // Cache de validacion de API key EN MEMORIA del isolate (mismo patron que // ACCOUNT_CACHE/RATE_BUCKETS). Sin esto, D1 Read Replication no aporta nada // al camino MAS caliente de todos (se valida la key en CADA request): con // cache, la mayoria de requests de un mismo isolate ni siquiera tocan D1. // TTL corto (no el de ACCOUNT_CACHE_TTL_MS): revocar una key debe notarse // rapido -- es el gate de seguridad, no de facturacion. Tambien cachea // resultados null (key invalida) para no pegarle a D1 en cada intento de un // atacante probando keys al azar. const API_KEY_CACHE = new Map(); const DEFAULT_API_KEY_CACHE_TTL_MS = 5000; // Devuelve la identidad del usuario (user_id, key_id, limite por minuto, nombre) // o null si la key no existe o esta revocada. D1 Read Replication (activado // en el dashboard) sirve esta lectura desde la region mas cercana via // Sessions API; "first-unconstrained" porque cada request es una consulta // independiente, no hace falta bookmark de una sesion previa. async function validateApiKeyD1(env, apiKey) { if (!env.DB || !apiKey) return null; // Cache por HASH (no por la key cruda): asi handleAdminRevokeKey puede // invalidar activamente la entrada exacta al revocar, en vez de depender // solo del TTL. El hash ya hay que calcularlo si no hay cache hit, y es // barato (WebCrypto, sub-ms) si lo hay -- no vale la pena evitarlo. const hash = await sha256Hex(apiKey); const ttl = Number(env.API_KEY_CACHE_TTL_MS ?? DEFAULT_API_KEY_CACHE_TTL_MS); if (ttl > 0) { const hit = API_KEY_CACHE.get(hash); if (hit && Date.now() - hit.ts <= ttl) return hit.ident; } const session = env.DB.withSession("first-unconstrained"); const row = await session .prepare( "SELECT id, usuario_id, nombre, limite_por_minuto FROM claves_api WHERE hash_clave = ? AND estado = 'activa'", ) .bind(hash) .first(); const ident = row ? { apiKey, keyId: row.id, userId: row.usuario_id, name: row.nombre || "", rateLimit: Number(row.limite_por_minuto) || 0, } : null; if (ttl > 0) { API_KEY_CACHE.set(hash, { ident, ts: Date.now() }); if (API_KEY_CACHE.size > 10000) { const now = Date.now(); for (const [key, value] of API_KEY_CACHE) { if (now - value.ts > ttl) API_KEY_CACHE.delete(key); } } } return ident; } // Marca usado_en sin bloquear (fire-and-forget desde ctx.waitUntil). function touchApiKey(env, keyId) { return env.DB .prepare("UPDATE claves_api SET usado_en = ? WHERE id = ?") .bind(Date.now(), keyId) .run() .catch(() => {}); } // --- Appwrite: plan + saldo + registro de uso ------------------------------- // El Worker habla con Appwrite con una API key de servidor (bypassa permisos). // El saldo y el plan son la fuente de verdad de "puede pasar esta peticion"; el // uso se escribe en eventos_uso y un flujo server-side descuenta saldo (prepago) // + actualiza el rollup (resumen_uso) que el dashboard observa. // La base de datos es `public`; en la coleccion `cuentas` el document id ($id) // ES el usuario_id (relacion 1:1), asi que la cuenta se lee/crea por id directo. function appwriteHeaders(env) { return { "X-Appwrite-Project": env.APPWRITE_PROJECT_ID, "X-Appwrite-Key": env.APPWRITE_API_KEY, "content-type": "application/json", }; } // Lee la cuenta (plan + saldo + estado) desde Appwrite por id = usuario_id. // Devuelve null si no se puede leer (incl. 404 = cuenta inexistente): fail-open // (no romper el servicio por un fallo de red de billing). Appwrite devuelve el // documento directamente (no un array como PostgREST); los campos de negocio // (plan, saldo_creditos, ...) llegan al mismo nivel junto a $id/$createdAt/... async function fetchAccount(env, userId) { if (!env.APPWRITE_ENDPOINT || !env.APPWRITE_API_KEY || !env.APPWRITE_PROJECT_ID || !userId) { return null; } try { const url = `${env.APPWRITE_ENDPOINT}/databases/public/collections/cuentas/documents/` + encodeURIComponent(userId); const res = await fetch(url, { headers: appwriteHeaders(env) }); if (!res.ok) return null; return await res.json(); } catch { return null; } } // Cache de cuentas EN MEMORIA del isolate (como RATE_BUCKETS/UPSTREAM_HEALTH): // evita pegar a Appwrite en cada peticion. Map usuario_id -> { account, ts }. // TTL configurable (ACCOUNT_CACHE_TTL_MS, default 10s; 0 = desactivado). // Saldo bajo => se ignora el cache y se valida en vivo, para que el corte por // saldo en prepago sea exacto cuando importa (cerca de 0). const ACCOUNT_CACHE = new Map(); const DEFAULT_ACCOUNT_CACHE_TTL_MS = 10000; const ACCOUNT_CACHE_LIVE_THRESHOLD = 25; // creditos: por debajo, no se cachea async function fetchAccountCached(env, userId) { const ttl = Number(env.ACCOUNT_CACHE_TTL_MS ?? DEFAULT_ACCOUNT_CACHE_TTL_MS); if (ttl <= 0) return fetchAccount(env, userId); const now = Date.now(); const hit = ACCOUNT_CACHE.get(userId); if (hit && now - hit.ts <= ttl) return hit.account; const account = await fetchAccount(env, userId); // No cachear cuentas cerca de su tope: queremos el saldo real en cada peticion // para no dejar colar gasto de mas dentro del TTL. // prepago -> margen = saldo restante. // pago_por_uso -> margen = limite_postpago + saldo (saldo es negativo). const lowMargin = account && (account.plan === "prepago" ? Number(account.saldo_creditos) < ACCOUNT_CACHE_LIVE_THRESHOLD : Number(account.limite_postpago) > 0 && Number(account.limite_postpago) + Number(account.saldo_creditos) < ACCOUNT_CACHE_LIVE_THRESHOLD); if (account && !lowMargin) { ACCOUNT_CACHE.set(userId, { account, ts: now }); if (ACCOUNT_CACHE.size > 10000) { for (const [key, value] of ACCOUNT_CACHE) { if (now - value.ts > ttl) ACCOUNT_CACHE.delete(key); } } } else { ACCOUNT_CACHE.delete(userId); } return account; } // Decide si la peticion pasa segun plan/saldo (corta peticiones por cuota): // estado != activa -> 403 (cuenta suspendida) // tarjeta no verificada -> 403 (anti-fraude: exige 1 pago real confirmado) // prepago sin saldo -> 402 (creditos insuficientes; recargar) // pago_por_uso en su techo -> 402 (deuda >= limite_postpago; facturar/subir tope) // pago_por_uso bajo techo -> pasa (postpago medido) // cuenta ilegible (null) -> pasa (fail-open: no romper el servicio por billing) function accountGate(account) { if (!account) return null; if (account.estado && account.estado !== "activa") { return Response.json({ error: "Account suspended", estado: account.estado }, { status: 403 }); } if (!account.tarjeta_verificada) { return Response.json( { error: "Card verification required", detail: "Verifica una tarjeta antes de usar la API.", verify_url: "https://markpdf.tech/billing", }, { status: 403 }, ); } const saldo = Number(account.saldo_creditos ?? 0); if (account.plan === "prepago") { if (saldo <= 0) { return Response.json( { error: "Insufficient credits", plan: "prepago", saldo_creditos: saldo, recharge_url: "https://markpdf.tech/billing", }, { status: 402 }, ); } } else { // pago_por_uso: deuda permitida hasta limite_postpago (0 = sin techo). const limite = Number(account.limite_postpago ?? 0); if (limite > 0 && saldo <= -limite) { return Response.json( { error: "Usage limit reached", plan: "pago_por_uso", saldo_creditos: saldo, limite_postpago: limite, manage_url: "https://markpdf.tech/billing", }, { status: 402 }, ); } } return null; } function normalizeBalanceAccount(account) { if (!account) return null; return { plan: account.plan === "pago_por_uso" ? "pago_por_uso" : "prepago", saldo_creditos: Number(account.saldo_creditos ?? 0), limite_postpago: Number(account.limite_postpago ?? 1000), estado: account.estado || "activa", // Default false a proposito (a diferencia del resto de estos campos): // es un gate anti-fraude, un valor ausente/desconocido debe bloquear, // no dejar pasar. tarjeta_verificada: account.tarjeta_verificada === true || account.tarjeta_verificada === 1, }; } function balanceGateResponse(account, credits) { const normalized = normalizeBalanceAccount(account); const charge = Math.max(0, Number(credits) || 0); if (!normalized) { return Response.json( { error: "Billing account unavailable", detail: "No local balance mirror row" }, { status: 503 }, ); } if (normalized.estado !== "activa") { return Response.json( { error: "Account suspended", estado: normalized.estado }, { status: 403 }, ); } if (charge <= 0) return null; if (normalized.plan === "prepago") { if (normalized.saldo_creditos < charge) { return Response.json( { error: "Insufficient credits", plan: "prepago", saldo_creditos: normalized.saldo_creditos, creditos_requeridos: charge, recharge_url: "https://markpdf.tech/billing", }, { status: 402 }, ); } return null; } const limite = normalized.limite_postpago; if (limite > 0 && normalized.saldo_creditos - charge <= -limite) { return Response.json( { error: "Usage limit reached", plan: "pago_por_uso", saldo_creditos: normalized.saldo_creditos, limite_postpago: limite, creditos_requeridos: charge, manage_url: "https://markpdf.tech/billing", }, { status: 402 }, ); } return null; } async function upsertBalanceD1(env, userId, account, now = Date.now()) { const normalized = normalizeBalanceAccount(account); if (!env.DB || !userId || !normalized) return; await env.DB.prepare( "INSERT INTO saldo_cuentas (usuario_id, plan, saldo_creditos, limite_postpago, estado, tarjeta_verificada, actualizado_en) " + "VALUES (?, ?, ?, ?, ?, ?, ?) " + "ON CONFLICT(usuario_id) DO UPDATE SET " + "plan=excluded.plan, saldo_creditos=excluded.saldo_creditos, " + "limite_postpago=excluded.limite_postpago, estado=excluded.estado, " + "tarjeta_verificada=excluded.tarjeta_verificada, actualizado_en=excluded.actualizado_en", ) .bind( userId, normalized.plan, normalized.saldo_creditos, normalized.limite_postpago, normalized.estado, normalized.tarjeta_verificada ? 1 : 0, now, ) .run(); } async function fetchBalanceD1(env, userId) { if (!env.DB || !userId) return null; return env.DB.prepare( "SELECT plan, saldo_creditos, limite_postpago, estado, tarjeta_verificada FROM saldo_cuentas WHERE usuario_id = ?", ) .bind(userId) .first(); } async function getOrSeedBalanceD1(env, userId) { const mirrored = await fetchBalanceD1(env, userId); if (mirrored) return normalizeBalanceAccount(mirrored); const account = await fetchAccount(env, userId); if (!account) return null; await upsertBalanceD1(env, userId, account); return normalizeBalanceAccount(account); } async function gateBalanceD1(env, userId, credits) { if (!env.DB) { return Response.json({ error: "Billing mirror (D1) not configured" }, { status: 503 }); } const account = await getOrSeedBalanceD1(env, userId); if (!account) { return Response.json( { error: "Billing unavailable", detail: "No D1 balance row and Appwrite fallback failed", }, { status: 503 }, ); } return balanceGateResponse(account, credits); } async function gateAndChargeD1(env, userId, credits) { const charge = Math.max(0, Number(credits) || 0); const gate = await gateBalanceD1(env, userId, charge); if (gate || charge <= 0) return gate; // D1 descuenta de forma provisional para cerrar la carrera en el edge. El // ledger real sigue siendo Appwrite: recordUsageAppwrite inserta eventos_uso, // un flujo server-side descuenta alla, y el webhook vuelve a corregir este // espejo con el saldo canonico. const res = await env.DB.prepare( "UPDATE saldo_cuentas " + "SET saldo_creditos = saldo_creditos - ?, actualizado_en = ? " + "WHERE usuario_id = ? AND estado = 'activa' AND " + "((plan = 'prepago' AND saldo_creditos >= ?) OR " + "(plan != 'prepago' AND (limite_postpago <= 0 OR saldo_creditos - ? > -limite_postpago)))", ) .bind(charge, Date.now(), userId, charge, charge) .run(); if ((res.meta?.changes ?? 0) !== 1) { return Response.json( { error: "Insufficient credits", detail: "Balance changed while this request was being processed", creditos_requeridos: charge, }, { status: 402 }, ); } return null; } // Version idempotente de gateAndChargeD1 para el consumer de cola: Cloudflare // Queues entrega "at-least-once" -- si el Worker crashea (OOM/wall-time) justo // despues de cobrar pero antes de msg.ack(), el mensaje se reentrega y sin esta // reserva se cobraria dos veces. cargos_encolados(job_id) es la marca de "ya // cobrado": el INSERT OR IGNORE es atomico, asi que una reentrega concurrente // o posterior no vuelve a descontar saldo. async function gateAndChargeD1Idempotent(env, userId, credits, jobId) { const charge = Math.max(0, Number(credits) || 0); if (charge <= 0 || !env.DB) return gateAndChargeD1(env, userId, charge); const insert = await env.DB.prepare( "INSERT OR IGNORE INTO cargos_encolados (job_id, usuario_id, creditos, creado_en) VALUES (?, ?, ?, ?)", ) .bind(jobId, userId, charge, Date.now()) .run(); const alreadyCharged = (insert.meta?.changes ?? 0) === 0; if (alreadyCharged) return null; // cobrado en un intento anterior: no repetir const gate = await gateAndChargeD1(env, userId, charge); if (gate) { // El cobro fallo (saldo insuficiente): libera la marca para no bloquear un // reintento legitimo si el saldo se repone antes del siguiente intento. await env.DB.prepare("DELETE FROM cargos_encolados WHERE job_id = ?").bind(jobId).run(); } return gate; } async function handleIngestBalance(request, env) { if ( !env.BALANCE_INGEST_SECRET || request.headers.get("x-ingest-secret") !== env.BALANCE_INGEST_SECRET ) { return new Response("Forbidden", { status: 403 }); } let body; try { body = await request.json(); } catch { return Response.json({ error: "Invalid JSON body" }, { status: 400 }); } const userId = String(body.usuario_id || "").trim(); if (!userId) return Response.json({ error: "usuario_id is required" }, { status: 400 }); await upsertBalanceD1( env, userId, { plan: body.plan, saldo_creditos: body.saldo_creditos, limite_postpago: body.limite_postpago, estado: body.estado, tarjeta_verificada: body.tarjeta_verificada, }, Date.now(), ); return Response.json({ ok: true }); } // Inserta una fila de uso en eventos_uso. Un flujo server-side descuenta saldo // (prepago) y actualiza resumen_uso. Best-effort: nunca rompe la respuesta. // Invalida el cache de la cuenta para que el siguiente gate vea el saldo nuevo. // Appwrite no tiene default now(): se incluye `creado_en` explicito. El id del // documento se autogenera con "unique()". async function recordUsageAppwrite(env, event) { if (!env.APPWRITE_ENDPOINT || !env.APPWRITE_API_KEY || !env.APPWRITE_PROJECT_ID || !event.usuario_id) { return; } ACCOUNT_CACHE.delete(event.usuario_id); try { await fetch(`${env.APPWRITE_ENDPOINT}/databases/public/collections/eventos_uso/documents`, { method: "POST", headers: appwriteHeaders(env), body: JSON.stringify({ documentId: "unique()", data: { ...event, creado_en: new Date().toISOString() }, // Equivale a la RLS "eventos dueno lectura" (SELECT auth.uid()=usuario_id): // sin esto el dashboard (listDocuments con la sesion del usuario) y la // suscripcion realtime del cliente no ven sus propios eventos. permissions: [`read("user:${event.usuario_id}")`], }), }); } catch { // best-effort } } // --- Emision de API keys (panel/admin) -------------------------------------- // Genera una key opaca `mk_live_<48 hex>`. Solo se devuelve UNA vez en claro; // en D1 se guarda su hash + un prefijo corto para mostrarla en el dashboard. function generateApiKey() { const bytes = new Uint8Array(24); crypto.getRandomValues(bytes); const rand = Array.from(bytes) .map((b) => b.toString(16).padStart(2, "0")) .join(""); return `mk_live_${rand}`; } function decodeBase64Url(value) { try { const padded = value.replace(/-/g, "+").replace(/_/g, "/").padEnd(Math.ceil(value.length / 4) * 4, "="); return atob(padded); } catch { return ""; } } function getAdminSecretHeader(request) { const raw = request.headers.get("x-admin-secret"); if (raw) return raw; const encoded = request.headers.get("x-admin-secret-b64"); return encoded ? decodeBase64Url(encoded) : ""; } // Comparacion en tiempo constante: `!==` sobre strings corta en el primer // byte distinto, filtrando por timing cuanto del secreto adivino un // atacante. Recorre SIEMPRE el largo completo (del mas largo de los dos) // sin early-return, para que el tiempo no dependa de en que byte difieren. function timingSafeEqualStr(a, b) { const bufA = new TextEncoder().encode(String(a ?? "")); const bufB = new TextEncoder().encode(String(b ?? "")); const len = Math.max(bufA.length, bufB.length, 1); let diff = bufA.length ^ bufB.length; for (let i = 0; i < len; i++) { diff |= (i < bufA.length ? bufA[i] : 0) ^ (i < bufB.length ? bufB[i] : 0); } return diff === 0; } const ADMIN_THROTTLE_PER_MINUTE = 30; // C10: /admin/keys puede emitir/revocar API keys de CUALQUIER usuario. Antes // solo dependia de ADMIN_SECRET sin rotacion ni limite de intentos -- un // secreto filtrado o adivinado por fuerza bruta no tenia freno. Ahora: (1) // comparacion en tiempo constante (no revela por timing cuanto acertaste), // (2) throttle por IP reusando el mismo rate-limiter global (DO/binding // nativo) que ya protege /convert -- sube el costo de un ataque de fuerza // bruta sin necesitar infraestructura nueva. async function requireAdmin(request, env) { const ip = request.headers.get("cf-connecting-ip") || "unknown"; const throttle = await checkRateLimit(env, `admin:${ip}`, ADMIN_THROTTLE_PER_MINUTE); if (!throttle.allowed) { return Response.json( { error: "Too many admin requests from this IP" }, { status: 429, headers: { "retry-after": String(Math.max(1, throttle.reset - Math.floor(Date.now() / 1000))) } }, ); } if (!env.ADMIN_SECRET || !timingSafeEqualStr(getAdminSecretHeader(request), env.ADMIN_SECRET)) { return new Response("Forbidden", { status: 403 }); } if (!env.DB) return new Response("Key store (D1) not configured", { status: 501 }); return null; } // POST /admin/keys { user_id, name?, plan?, rate_limit_per_minute? } // Crea la key real, guarda su hash en D1 y asegura el account en Appwrite. async function handleAdminCreateKey(request, env) { const denied = await requireAdmin(request, env); if (denied) return denied; let body; try { body = await request.json(); } catch { return Response.json({ error: "Invalid JSON body" }, { status: 400 }); } const userId = String(body.user_id || "").trim(); if (!userId) return Response.json({ error: "user_id is required" }, { status: 400 }); const name = String(body.name || "").slice(0, 80); // Acepta tanto los valores en espanol como los alias en ingles. const plan = body.plan === "pago_por_uso" || body.plan === "pay_per_use" ? "pago_por_uso" : "prepago"; const rateLimit = Number.isFinite(body.rate_limit_per_minute) ? Math.max(0, Math.trunc(body.rate_limit_per_minute)) : 120; // Limite de keys por cuenta: no-pro = 1. Se lee de cuentas.max_claves (0 = // ilimitado). Si Appwrite no esta o no hay fila, default 1 (lo mas estricto). // tarjeta_verificada es anti-fraude: fail-CLOSED (a diferencia de max_claves) // -- si no se puede leer Appwrite, NO se emite la key. Mejor romper la UX // un momento que dejar pasar una cuenta sin tarjeta real verificada. let maxKeys = 1; let tarjetaVerificada = false; if (env.APPWRITE_ENDPOINT && env.APPWRITE_API_KEY && env.APPWRITE_PROJECT_ID) { try { const res = await fetch( `${env.APPWRITE_ENDPOINT}/databases/public/collections/cuentas/documents/` + encodeURIComponent(userId), { headers: appwriteHeaders(env) }, ); if (res.ok) { const doc = await res.json(); if (doc && doc.max_claves != null) maxKeys = Number(doc.max_claves); tarjetaVerificada = doc?.tarjeta_verificada === true; } } catch { // best-effort: queda el default 1 / tarjetaVerificada=false } } if (!tarjetaVerificada) { return Response.json( { error: "Card verification required", detail: "El usuario debe verificar una tarjeta antes de generar API keys.", verify_url: "https://markpdf.tech/billing", }, { status: 403 }, ); } if (maxKeys > 0) { const { results } = await env.DB.prepare( "SELECT COUNT(*) AS n FROM claves_api WHERE usuario_id = ? AND estado = 'activa'", ) .bind(userId) .all(); const activeKeys = Number(results?.[0]?.n ?? 0); if (activeKeys >= maxKeys) { return Response.json( { error: "API key limit reached", max_claves: maxKeys, active_keys: activeKeys, upgrade_url: "https://markpdf.tech/billing", }, { status: 409 }, ); } } const apiKey = generateApiKey(); const keyHash = await sha256Hex(apiKey); const keyId = crypto.randomUUID(); const keyPrefix = apiKey.slice(0, 12); await env.DB.prepare( "INSERT INTO claves_api (id, hash_clave, prefijo_clave, usuario_id, nombre, limite_por_minuto, estado, creado_en) " + "VALUES (?, ?, ?, ?, ?, ?, 'activa', ?)", ) .bind(keyId, keyHash, keyPrefix, userId, name, rateLimit, Date.now()) .run(); // Asegura la cuenta en Appwrite (idempotente). El doc id ES el usuario_id: // se comprueba con GET y solo se crea si devuelve 404. Si ya existe NO se // toca (ni saldo ni plan). La cuenta tambien nace en el signup del panel. // Un 409 en el POST (creada por una carrera concurrente) se ignora. if (env.APPWRITE_ENDPOINT && env.APPWRITE_API_KEY && env.APPWRITE_PROJECT_ID) { try { const docUrl = `${env.APPWRITE_ENDPOINT}/databases/public/collections/cuentas/documents/` + encodeURIComponent(userId); const existing = await fetch(docUrl, { headers: appwriteHeaders(env) }); if (existing.status === 404) { const nowIso = new Date().toISOString(); await fetch(`${env.APPWRITE_ENDPOINT}/databases/public/collections/cuentas/documents`, { method: "POST", headers: appwriteHeaders(env), body: JSON.stringify({ documentId: userId, data: { usuario_id: userId, plan, creado_en: nowIso, actualizado_en: nowIso }, }), }); } } catch { // best-effort } } return Response.json( { api_key: apiKey, key_id: keyId, key_prefix: keyPrefix, user_id: userId, plan, rate_limit_per_minute: rateLimit, warning: "Store this key now — it will not be shown again.", }, { status: 201 }, ); } // GET /admin/keys?user_id= -> lista keys (solo prefijo, nunca hash ni clave). async function handleAdminListKeys(request, env, url) { const denied = await requireAdmin(request, env); if (denied) return denied; const cols = "id, prefijo_clave, usuario_id, nombre, limite_por_minuto, estado, creado_en, usado_en"; const userId = url.searchParams.get("user_id"); const stmt = userId ? env.DB.prepare( `SELECT ${cols} FROM claves_api WHERE usuario_id = ? ORDER BY creado_en DESC`, ).bind(userId) : env.DB.prepare(`SELECT ${cols} FROM claves_api ORDER BY creado_en DESC LIMIT 200`); const { results } = await stmt.all(); return Response.json({ keys: results || [] }); } // DELETE /admin/keys/:id -> revoca (estado='revocada'; no se borra para auditoria). async function handleAdminRevokeKey(request, env, keyId) { const denied = await requireAdmin(request, env); if (denied) return denied; // Hash ANTES del UPDATE: hace falta para invalidar la entrada exacta en // API_KEY_CACHE (cacheada por hash, no por keyId). Best-effort: si esto // falla, la revocacion sigue igual, solo queda a merced del TTL corto. const row = await env.DB.prepare("SELECT hash_clave FROM claves_api WHERE id = ?") .bind(keyId) .first(); const res = await env.DB.prepare("UPDATE claves_api SET estado = 'revocada' WHERE id = ?") .bind(keyId) .run(); if (row?.hash_clave) API_KEY_CACHE.delete(row.hash_clave); return Response.json({ revoked: keyId, changes: res.meta?.changes ?? 0 }); } function getApiKey(request) { const authorization = request.headers.get("authorization") || ""; const bearer = authorization.toLowerCase().startsWith("bearer ") ? authorization.slice(7).trim() : ""; return request.headers.get("x-api-key") || request.headers.get("apikey") || bearer; } // Auth opcional: sin API_KEYS configurada => acceso abierto. Si se define, // exige una key valida. Devuelve la key (o "" en modo abierto). function requireApiKey(request, env) { const configured = (env.API_KEYS || "") .split(",") .map((value) => value.trim()) .filter(Boolean); if (configured.length === 0) return ""; const apiKey = getApiKey(request); if (!apiKey || !configured.includes(apiKey)) { throw new Response("Invalid API key", { status: 401 }); } return apiKey; } // Identidad para rate-limit: la API key si existe; si no, la IP del cliente. function rateLimitIdentity(request, apiKey) { if (apiKey) return `key:${apiKey}`; const ip = request.headers.get("cf-connecting-ip") || "unknown"; return `ip:${ip}`; } // Durable Object para rate-limit CUSTOM (C1): contador global REAL, una // instancia por identity (API key o IP), en vez de un Map por-isolate. Se usa // para keys con limite distinto al default (D1 permite un limite por cliente, // ej. enterprise) porque el binding nativo de Cloudflare (DEFAULT_RATE_LIMITER // abajo) tiene el limite fijo en wrangler.toml y no puede leer el limite // dinamico de D1. Storage SQLite (Workers Free plan compatible) para que el // contador sobreviva si el DO se evicta de memoria entre requests espaciadas. export class RateLimiterDO { constructor(state) { this.state = state; this.windowStart = null; this.count = 0; } async fetch(request) { const { limitPerMinute, windowStart } = await request.json(); if (this.windowStart !== windowStart) { const stored = await this.state.storage.get("bucket"); if (stored && stored.windowStart === windowStart) { this.windowStart = stored.windowStart; this.count = stored.count; } else { this.windowStart = windowStart; this.count = 0; } } this.count += 1; await this.state.storage.put("bucket", { windowStart: this.windowStart, count: this.count }); return Response.json({ allowed: this.count <= limitPerMinute, count: this.count, }); } } // Durable Object GLOBAL (una sola instancia, idFromName("global")) que cuenta // jobs en vuelo en la cola auto-async (C9): sin esto, bajo saturacion // sostenida la cola y R2 (PDF_OVERFLOW) crecen sin techo porque Cloudflare // Queues no expone su profundidad al Worker. incr() se llama al encolar // (enqueueJob), decr() al llegar a un estado TERMINAL en el consumer // (worker.queue: exito, fallo definitivo, o excepcion no recuperable) -- // NUNCA antes de un msg.retry(), el job sigue en vuelo. export class QueueDepthDO { constructor(state) { this.state = state; this.count = null; } async _load() { if (this.count === null) { this.count = (await this.state.storage.get("count")) || 0; } } async fetch(request) { await this._load(); const { op, maxDepth } = await request.json(); if (op === "incr") { if (maxDepth > 0 && this.count >= maxDepth) { return Response.json({ ok: false, count: this.count }); } this.count += 1; await this.state.storage.put("count", this.count); return Response.json({ ok: true, count: this.count }); } if (op === "decr") { this.count = Math.max(0, this.count - 1); await this.state.storage.put("count", this.count); return Response.json({ ok: true, count: this.count }); } return Response.json({ ok: true, count: this.count }); } } async function incrQueueDepth(env, maxDepth) { if (!env.QUEUE_DEPTH_DO) return true; // fallback: sin binding desplegado, no bloquea const id = env.QUEUE_DEPTH_DO.idFromName("global"); const stub = env.QUEUE_DEPTH_DO.get(id); const res = await stub.fetch("https://queue-depth/incr", { method: "POST", body: JSON.stringify({ op: "incr", maxDepth }), }); const { ok } = await res.json(); return ok; } async function decrQueueDepth(env) { if (!env.QUEUE_DEPTH_DO) return; const id = env.QUEUE_DEPTH_DO.idFromName("global"); const stub = env.QUEUE_DEPTH_DO.get(id); await stub.fetch("https://queue-depth/decr", { method: "POST", body: JSON.stringify({ op: "decr" }), }); } // Rate-limit EN MEMORIA del isolate (sin estado externo). Map de identity -> // { windowStart, count }, ventana fija de 60s. Fallback SOLO si el binding // RATE_LIMITER_DO no esta desplegado todavia (p.ej. mientras se aplica la // migracion de wrangler.toml) -- en ese caso el limite efectivo global vuelve // a ser ~limite x nº isolates activos, igual que antes del DO. const RATE_BUCKETS = new Map(); // Rate limit REAL y compartido para el caso comun (limite default, sin // personalizar por cliente): usa el binding nativo `ratelimit` de Cloudflare // Workers (GA desde sept 2025). A diferencia de RATE_BUCKETS, este contador // es compartido de verdad entre TODOS los isolates y colos -- un cliente (o // atacante con una key robada) no lo esquiva pegandole a distintos PoPs. // Como esta API exige API key en TODAS las rutas de conversion (D1 bind // siempre presente), practicamente todo el trafico pasa por esta identidad // "key:", no por IP -- por eso el binding se aplica aca, al caso comun, // no a un caso anonimo que en la practica nunca ocurre. // Limitacion real del binding: el limite es fijo en wrangler.toml (no puede // leer env vars ni el limite dinamico de D1), asi que las keys con limite // CUSTOM (distinto al default) siguen con el contador por-isolate de arriba, // respaldado ademas por el gate de creditos (D1/Appwrite) como limite real // de abuso cuando ya sabemos quien es el cliente. const DEFAULT_RATE_LIMIT = 120; async function checkRateLimit(env, identity, limitPerMinute) { if (limitPerMinute <= 0) { return { allowed: true, limit: limitPerMinute, remaining: limitPerMinute, reset: 0 }; } const now = Math.floor(Date.now() / 1000); const windowStart = now - (now % 60); const reset = windowStart + 60; if (env.DEFAULT_RATE_LIMITER && limitPerMinute >= DEFAULT_RATE_LIMIT) { const { success } = await env.DEFAULT_RATE_LIMITER.limit({ key: identity }); if (!success || limitPerMinute === DEFAULT_RATE_LIMIT) { return { allowed: success, limit: Math.min(limitPerMinute, DEFAULT_RATE_LIMIT), remaining: success ? 1 : 0, // el binding no expone el remaining exacto reset, }; } } // Limite CUSTOM: contador global real via Durable Object (una instancia // por identity), en vez del Map por-isolate. Cierra C1. if (env.RATE_LIMITER_DO) { const id = env.RATE_LIMITER_DO.idFromName(identity); const stub = env.RATE_LIMITER_DO.get(id); const res = await stub.fetch("https://rate-limiter/check", { method: "POST", body: JSON.stringify({ limitPerMinute, windowStart }), }); const { allowed, count } = await res.json(); return { allowed, limit: limitPerMinute, remaining: Math.max(0, limitPerMinute - count), reset, }; } let bucket = RATE_BUCKETS.get(identity); if (!bucket || bucket.windowStart !== windowStart) { bucket = { windowStart, count: 0 }; RATE_BUCKETS.set(identity, bucket); // Poda perezosa: al rotar ventana, descarta entradas de ventanas viejas para // que el Map no crezca sin limite con identities que ya no llegan. if (RATE_BUCKETS.size > 10000) { for (const [key, value] of RATE_BUCKETS) { if (value.windowStart !== windowStart) RATE_BUCKETS.delete(key); } } } bucket.count += 1; return { allowed: bucket.count <= limitPerMinute, limit: limitPerMinute, remaining: Math.max(0, limitPerMinute - bucket.count), reset, }; } // Construye los headers que van al upstream: copia los del cliente menos los // hop-by-hop y las credenciales (hide_credentials), e inyecta request-id + // el secreto de gateway que el backend valida. function buildUpstreamHeaders(request, env, requestId, auth) { const headers = new Headers(); for (const [name, value] of request.headers) { const lower = name.toLowerCase(); if (HOP_BY_HOP.has(lower)) continue; if (lower === "x-api-key" || lower === "apikey" || lower === "authorization") continue; // El cliente NO puede falsificar su identidad: descartamos cualquier // X-Consumer-* entrante; solo el Worker la inyecta tras validar en D1. if (lower.startsWith("x-consumer-") || lower === "x-anonymous-consumer") continue; headers.set(name, value); } headers.set("x-request-id", requestId); if (env.GATEWAY_SECRET) headers.set("x-gateway-secret", env.GATEWAY_SECRET); // Identidad del consumer para el seam de billing del backend (get_consumer). if (auth && auth.userId) { headers.set("x-consumer-id", auth.userId); headers.set("x-consumer-custom-id", auth.userId); if (auth.name) headers.set("x-consumer-username", auth.name); } return headers; } function stripResponseHeaders(headers) { for (const name of STRIP_RESPONSE_HEADERS) headers.delete(name); } // Devuelve true si la peticion viene de un navegador (Accept incluye text/html). // Usado para servir la landing humana en / cuando alguien entra con el browser. // Agentes IA, curl y monitoreo mandan Accept: application/json o */* y siguen // recibiendo el JSON tecnico de /healthz. function wantsHtml(request) { const accept = (request.headers.get("accept") || "").toLowerCase(); return accept.includes("text/html"); } // Landing minima en HTML para humanos. Sin tracking, sin frameworks: HTML // puro inline. Tres mensajes: que es la API, donde sacar la key, donde // leer docs. Ingles. Cacheable 5 min en edge porque es estatico. function landingResponse() { const html = ` markpdf.tech - PDF to Markdown API

Hello from the MarkPDF API.

Get your API key at markpdf.tech. Looking for documentation? Visit docs.markpdf.tech.

Get your API key Read the docs

Calling from a script? Add Accept: application/json for the technical health endpoint.

`; return new Response(html, { status: 200, headers: { "content-type": "text/html; charset=utf-8", "cache-control": "public, max-age=300", "x-robots-tag": "noindex", }, }); } // Lee el body contando bytes y ABORTA en cuanto supera maxBytes, sin esperar a // bufferizar el upload entero. Evita el pico de RAM de cargar un archivo gigante // solo para rechazarlo (clave: el isolate tiene 128 MB) y libera el socket antes. // Lanza Response 413 si se pasa. Devuelve un ArrayBuffer reutilizable (failover). async function readBodyWithLimit(stream, maxBytes) { const reader = stream.getReader(); const chunks = []; let total = 0; try { for (;;) { const { done, value } = await reader.read(); if (done) break; total += value.byteLength; if (total > maxBytes) { await reader.cancel(); throw new Response(`Upload too large (>${maxBytes} bytes)`, { status: 413 }); } chunks.push(value); } } finally { reader.releaseLock?.(); } const bytes = new Uint8Array(total); let offset = 0; for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; } return bytes.buffer; } // Mensaje accionable cuando un PDF supera el limite de raw upload. Replica el // contrato del backend (main.py:_raw_too_large_response): nunca filtra URL // firmada del request ni rutas internas. function rawTooLargeResponse(maxRawBytes, requestId) { const maxMb = Math.round((maxRawBytes / (1024 * 1024)) * 10) / 10; return new Response( JSON.stringify({ error: "PDF too large for raw upload fast path", max_raw_upload_mb: maxMb, why: "Raw upload sends the full PDF through the request body. Large PDFs spend seconds " + "only in network transfer before conversion starts.", recommendation: "Upload the PDF to S3/R2/GCS/Blob Storage and call /convert/from-url or " + "/convert/accelerated with signed URLs.", recommended_endpoint: "/convert/accelerated", }), { status: 413, headers: { "content-type": "application/json", "x-request-id": requestId, }, }, ); } async function proxy(request, env, auth, requestId) { const upstreams = parseUpstreams(env.UPSTREAMS); if (upstreams.length === 0) { return new Response("No upstreams configured", { status: 503 }); } const incomingUrl = new URL(request.url); const hasBody = !["GET", "HEAD"].includes(request.method); const maxUploadBytes = Number(env.MAX_UPLOAD_BYTES || DEFAULT_MAX_UPLOAD_BYTES); const rawUploadMaxBytes = Number(env.RAW_UPLOAD_MAX_BYTES || DEFAULT_RAW_UPLOAD_MAX_BYTES); const isRawRoute = RAW_UPLOAD_PATHS.has(incomingUrl.pathname); const noBodyRoute = NO_BODY_PATHS.has(incomingUrl.pathname); const effectiveLimit = isRawRoute ? Math.min(rawUploadMaxBytes, maxUploadBytes) : maxUploadBytes; const declaredInputBytes = Number(request.headers.get("content-length") || "0"); // --- Auto-async: si todos los backends estan saturados y la ruta es de // conversion, degradamos a 202+job_id. El request se encola en CF Queue y // el consumer lo procesa cuando un backend se libere. Coste: $0.40/M msgs. if ( DEGRADABLE_PATHS.has(incomingUrl.pathname) && request.method === "POST" && allBackendsSaturated(upstreams) && env.CONVERSION_QUEUE ) { let bodyBuf = null; if (hasBody && !noBodyRoute && request.body) { if (declaredInputBytes > effectiveLimit) { if (isRawRoute) return rawTooLargeResponse(rawUploadMaxBytes, requestId); return new Response(`Upload too large (>${effectiveLimit} bytes)`, { status: 413 }); } try { bodyBuf = await readBodyWithLimit(request.body, effectiveLimit); } catch (error) { if (isRawRoute && error instanceof Response && error.status === 413) { return rawTooLargeResponse(rawUploadMaxBytes, requestId); } if (error instanceof Response) return error; throw error; } } const jobId = crypto.randomUUID().replace(/-/g, ""); const rejected = await enqueueJob(env, jobId, request, incomingUrl, bodyBuf, auth, requestId, declaredInputBytes); if (rejected) return rejected; return asyncAcceptedResponse(jobId, requestId); } const headers = buildUpstreamHeaders(request, env, requestId, auth); // El body se bufferiza una vez para poder reintentarlo en otro upstream: un // ReadableStream solo se consume una vez. Solo aplica si hay body (POST/PUT). // Limite GLOBAL del isolate (proteccion de RAM). Por encima de esto, rechazo // duro independiente del endpoint. // Limite ESPECIFICO de raw (~12 MB). Se aplica SOLO a las rutas que aceptan // upload directo de PDF. Las rutas from-url/accelerated/pdf-index no llevan // PDF en el body (solo JSON pequeno con URLs firmadas), asi que no aplica. // Tope efectivo para esta peticion: en raw, manda el limite estricto. let body; if (hasBody && !noBodyRoute) { // Rechazo rapido si el cliente declara un tamano que ya supera el tope, sin // leer un solo byte del cuerpo. En rutas raw devolvemos JSON accionable; // en el resto, plain text generico. const declared = declaredInputBytes; if (declared > effectiveLimit) { if (isRawRoute) return rawTooLargeResponse(rawUploadMaxBytes, requestId); return new Response(`Upload too large (>${effectiveLimit} bytes)`, { status: 413 }); } if (request.body) { // Corte real por streaming: aborta aunque content-length falte o mienta. try { body = await readBodyWithLimit(request.body, effectiveLimit); } catch (error) { // readBodyWithLimit lanza Response 413 generico al pasarse del limite. // En rutas raw lo enriquecemos con el mensaje accionable del backend. if (error instanceof Response) { if (isRawRoute && error.status === 413) { return rawTooLargeResponse(rawUploadMaxBytes, requestId); } return error; } throw error; } } } // Orden de intento: filtrado por el tope de body conocido POR PROVEEDOR (ver // KNOWN_UPSTREAM_BODY_LIMITS) usando el tamano YA leido del body. Evita subir // el PDF entero a un backend que sabemos de antemano que lo va a rechazar con // 413 (Vercel ~4.5 MB): ese upload completo desperdiciado es el grueso del // tiempo extra quel el reintento entre backends le suma a la peticion. const order = filterOrderByBodySize( applyStreamingOverride(buildAttemptOrder(upstreams), incomingUrl.pathname, env), body ? body.byteLength : 0, ); const maxRetries = Math.min(order.length, Number(env.MAX_UPSTREAM_RETRIES || order.length)); let lastError = null; let lastRetryableResponse = null; for (let attempt = 0; attempt < maxRetries; attempt += 1) { const target = order[attempt]; const targetUrl = target.origin + incomingUrl.pathname + incomingUrl.search; try { const upstreamResponse = await fetch(targetUrl, { method: request.method, headers, body, redirect: "manual", }); // Passive health: aprende la carga que reporto este backend (X-Svc-*), // tanto si sirvio como si dio un 5xx reintentable (saturado = util saberlo). recordUpstreamHealth(target.origin, upstreamResponse.headers); if (UPSTREAM_RETRYABLE_STATUS.has(upstreamResponse.status)) { lastError = `upstream ${target.origin} -> ${upstreamResponse.status}`; // 413 no siempre es "reintentable" de verdad: puede ser el tope de ESE // backend (Vercel ~4.5 MB), no del body en si. Guardamos la respuesta // por si es el ultimo intento y no hay fallback de cola disponible. lastRetryableResponse = upstreamResponse; if (attempt < maxRetries - 1) continue; break; } const responseHeaders = new Headers(upstreamResponse.headers); stripResponseHeaders(responseHeaders); responseHeaders.set("x-request-id", requestId); if (shouldExposePoolHealth(request, env)) { responseHeaders.set("x-pool-health", JSON.stringify(poolHealthSnapshot(upstreams))); } return new Response(upstreamResponse.body, { status: upstreamResponse.status, statusText: upstreamResponse.statusText, headers: responseHeaders, }); } catch (error) { lastError = error?.message || String(error); if (attempt >= maxRetries - 1) { captureException(error, { component: "proxy", target: target.origin }); } } } // Ultimo recurso: si todos los upstreams fallaron (network) o el unico backend // que quedaba respondio 413/502/503/504, y la ruta es degradable, encolar como // async (R2 + CF Queue) en vez de devolver el error al cliente. El body viaja // a R2 si supera INLINE_BODY_MAX_BYTES; el consumer procesa cuando haya cupo. if (DEGRADABLE_PATHS.has(incomingUrl.pathname) && request.method === "POST" && env.CONVERSION_QUEUE) { const jobId = crypto.randomUUID().replace(/-/g, ""); const rejected = await enqueueJob(env, jobId, request, incomingUrl, body); if (rejected) return rejected; return asyncAcceptedResponse(jobId, requestId); } if (lastRetryableResponse) { const responseHeaders = new Headers(lastRetryableResponse.headers); stripResponseHeaders(responseHeaders); responseHeaders.set("x-request-id", requestId); if (shouldExposePoolHealth(request, env)) { responseHeaders.set("x-pool-health", JSON.stringify(poolHealthSnapshot(upstreams))); } return new Response(lastRetryableResponse.body, { status: lastRetryableResponse.status, statusText: lastRetryableResponse.statusText, headers: responseHeaders, }); } return new Response(`No upstream reachable: ${lastError || "unknown"}`, { status: 502 }); } const worker = { async fetch(request, env, ctx) { const url = new URL(request.url); const requestId = request.headers.get("x-request-id") || crypto.randomUUID(); // Landing amistosa para humanos en /: si el cliente pide text/html (browsers // por defecto), servimos una pagina chiquita con CTA hacia markpdf.tech / // docs.markpdf.tech. NO exige API key (es solo marketing/redireccion). // curl, agentes y monitores que pidan JSON o no envien Accept se llevan // el JSON tecnico de mas abajo. if (request.method === "GET" && url.pathname === "/" && wantsHtml(request)) { return landingResponse(); } if (url.pathname === "/internal/ingest-balance") { if (request.method !== "POST") return new Response("Method not allowed", { status: 405 }); return handleIngestBalance(request, env); } // --- Admin API: emision/gestion de API keys reales. Auth propia via // x-admin-secret (la usa el panel/dashboard server-side, no clientes). if (url.pathname === "/admin/keys") { if (request.method === "POST") return handleAdminCreateKey(request, env); if (request.method === "GET") return handleAdminListKeys(request, env, url); return new Response("Method not allowed", { status: 405 }); } const adminKeyMatch = url.pathname.match(/^\/admin\/keys\/([0-9a-f-]+)$/); if (adminKeyMatch && request.method === "DELETE") { return handleAdminRevokeKey(request, env, adminKeyMatch[1]); } // /, /healthz quedan SIN auth (monitoreo y smoke tests CI/CD pegan aqui). const openPath = url.pathname === "/" || url.pathname === "/healthz"; // --- Auth: D1 (store real de keys, validacion en el edge) si hay binding; // si no, fallback al API_KEYS estatico de env (modo legacy / dev sin D1). let auth = { apiKey: "", userId: "", keyId: "", name: "", rateLimit: 0 }; if (!openPath) { if (env.DB) { const ident = await validateApiKeyD1(env, getApiKey(request)); if (!ident) { return new Response("Invalid API key", { status: 401, headers: { "x-request-id": requestId }, }); } auth = ident; ctx.waitUntil(touchApiKey(env, ident.keyId)); } else { try { auth.apiKey = requireApiKey(request, env); } catch (error) { if (error instanceof Response) return error; return new Response("Invalid API key", { status: 401 }); } } } if (url.pathname === "/" || url.pathname === "/healthz") { const upstreams = parseUpstreams(env.UPSTREAMS); const health = { name: "pdf-to-markdown-api-router", mode: "router", role: "edge-router-load-balancer", upstreams: upstreams.length, routing: "least-loaded", rate_limit_per_minute: Number(env.RATE_LIMIT_PER_MINUTE || "0"), // Tope global por isolate (RAM). Aplica a TODAS las rutas con body. max_upload_bytes: Number(env.MAX_UPLOAD_BYTES || DEFAULT_MAX_UPLOAD_BYTES), // Tope estricto en /convert/raw|/convert|/convert/stream. PDFs mayores // deben ir por /convert/from-url|/convert/accelerated con URL firmada. raw_upload_max_bytes: Number(env.RAW_UPLOAD_MAX_BYTES || DEFAULT_RAW_UPLOAD_MAX_BYTES), raw_upload_paths: [...RAW_UPLOAD_PATHS], get_api_key_url: "https://markpdf.tech", docs_url: "https://docs.markpdf.tech", }; if (shouldExposePoolHealth(request, env)) { // Estado vivo de cada servicio segun lo ultimo reportado a ESTE isolate. health.pool = poolHealthSnapshot(upstreams); } return Response.json(health); } // --- Edge job status: lee estado del job auto-async desde Cache API (gratis). // Si el job no esta en cache (fue procesado por un backend directamente), // proxia GET /jobs/{id} a todos los backends hasta encontrarlo. const jobMatch = url.pathname.match(/^\/jobs\/([a-f0-9-]+)$/); if (jobMatch && request.method === "GET") { const jobId = jobMatch[1]; const cached = await readJobStatus(jobId); if (cached) { return Response.json(cached, { headers: { "x-request-id": requestId, "x-job-source": "edge-cache" } }); } // Fallback: el job puede estar en _instant_jobs del backend (async manual). // Proxia al primer backend que responda 200. const upstreams = parseUpstreams(env.UPSTREAMS); const headers = buildUpstreamHeaders(request, env, requestId); for (const upstream of upstreams) { try { const res = await fetch(`${upstream.origin}/jobs/${jobId}`, { headers }); if (res.ok) { const data = await res.json(); return Response.json(data, { headers: { "x-request-id": requestId, "x-job-source": "backend" } }); } } catch { /* try next */ } } return Response.json( { error: "Job not found or expired", job_id: jobId }, { status: 404, headers: { "x-request-id": requestId } } ); } // Limite por minuto: el de la key (D1) si lo define; si no, el global de env. const limitPerMinute = auth.rateLimit > 0 ? auth.rateLimit : Number(env.RATE_LIMIT_PER_MINUTE || "0"); const identity = auth.keyId ? `key:${auth.keyId}` : rateLimitIdentity(request, auth.apiKey); const rate = await checkRateLimit(env, identity, limitPerMinute); if (!rate.allowed) { return new Response("Rate limit exceeded", { status: 429, headers: { "retry-after": String(Math.max(1, rate.reset - Math.floor(Date.now() / 1000))), "x-ratelimit-limit": String(rate.limit), "x-ratelimit-remaining": "0", "x-ratelimit-reset": String(rate.reset), "x-request-id": requestId, }, }); } // --- Gate de plan/saldo (D1). Solo en rutas de conversion y solo si // conocemos al usuario (auth via D1). Este pre-gate usa 1 credito como // margen minimo porque el costo exacto lo informa el backend en la // respuesta; el cobro atomico real ocurre mas abajo con x-request-credits. const isConversion = DEGRADABLE_PATHS.has(url.pathname); if (auth.userId && isConversion) { const gate = await gateBalanceD1(env, auth.userId, 1); if (gate) { gate.headers.set("x-request-id", requestId); return gate; } } // Bytes de entrada declarados (fallback si el backend no emite X-Input-Bytes). const declaredInputBytes = Number(request.headers.get("content-length") || "0"); const response = await proxy(request, env, auth, requestId); if (rate.limit > 0) { response.headers.set("x-ratelimit-limit", String(rate.limit)); response.headers.set("x-ratelimit-remaining", String(Math.max(0, rate.remaining))); response.headers.set("x-ratelimit-reset", String(rate.reset)); } // --- Cobro D1 + registro de uso en Appwrite. D1 corta carreras en el edge; // Appwrite sigue siendo el ledger real y su flujo/webhook reconcilia D1. if (auth.userId && isConversion && response.status >= 200 && response.status < 300) { const creditsCharged = Number(response.headers.get("x-request-credits") || 0); const chargeGate = await gateAndChargeD1(env, auth.userId, creditsCharged); if (chargeGate) { chargeGate.headers.set("x-request-id", requestId); return chargeGate; } const inputBytes = Number(response.headers.get("x-input-bytes") || declaredInputBytes || 0); const outputBytes = Number(response.headers.get("x-markdown-bytes") || 0); ctx.waitUntil( recordUsageAppwrite(env, { usuario_id: auth.userId, clave_api_id: auth.keyId || null, peticion_id: requestId, endpoint: url.pathname, bytes_entrada: inputBytes, bytes_salida: outputBytes, nivel: response.headers.get("x-price-tier") || null, creditos_cobrados: creditsCharged, precio_usd: Number(response.headers.get("x-request-price-usd") || 0), estado_http: response.status, }), ); } // Log JSON para observabilidad. ctx.waitUntil( (async () => { console.log( JSON.stringify({ type: "router_request", timestamp: new Date().toISOString(), request_id: requestId, api_key_fingerprint: auth.apiKey ? await apiKeyFingerprint(auth.apiKey) : null, user_id: auth.userId || null, method: request.method, path: url.pathname, status: response.status, }), ); })(), ); return response; }, }; // --- Queue consumer: procesa jobs encolados por auto-async degradation. // CF invoca queue() push-based cuando hay mensajes. max_batch_size=1 en // wrangler.toml para que cada job se procese individualmente con retry. worker.queue = async function (batch, env) { for (const msg of batch.messages) { const job = msg.body; const jobId = job.job_id; try { await cacheJobStatus(jobId, { job_id: jobId, status: "processing", started_at: Date.now() }); const upstreams = parseUpstreams(env.UPSTREAMS); if (upstreams.length === 0) { await cacheJobStatus(jobId, { job_id: jobId, status: "failed", error: "No upstreams configured" }); await deleteOverflowBody(env, job); await decrQueueDepth(env); msg.ack(); continue; } const ordered = buildAttemptOrder(upstreams); const headers = new Headers(job.headers || {}); headers.set("x-request-id", `queue-${jobId}`); if (env.GATEWAY_SECRET) headers.set("x-gateway-secret", env.GATEWAY_SECRET); let bodyBuf = job.body_b64 ? base64ToArrayBuffer(job.body_b64) : null; if (!bodyBuf && job.body_r2_key && env.PDF_OVERFLOW) { const obj = await env.PDF_OVERFLOW.get(job.body_r2_key); bodyBuf = obj ? await obj.arrayBuffer() : null; } let lastError = null; for (const upstream of ordered) { const targetUrl = upstream.origin + job.path; try { const res = await fetch(targetUrl, { method: job.method || "POST", headers, body: bodyBuf, }); recordUpstreamHealth(upstream.origin, res.headers); if (RETRYABLE_STATUS.has(res.status)) { lastError = `${upstream.origin} returned ${res.status}`; continue; } const resultBody = await res.text(); const resultHeaders = {}; for (const [k, v] of res.headers) { if (!STRIP_RESPONSE_HEADERS.includes(k.toLowerCase())) resultHeaders[k] = v; } const isConversion = DEGRADABLE_PATHS.has(job.path.split("?")[0]); if (job.auth_user_id && isConversion && res.status >= 200 && res.status < 300) { const creditsCharged = Number(res.headers.get("x-request-credits") || 0); const chargeGate = await gateAndChargeD1Idempotent(env, job.auth_user_id, creditsCharged, jobId); if (chargeGate) { await cacheJobStatus(jobId, { job_id: jobId, status: "failed", error: "Billing charge failed after queued conversion", http_status: chargeGate.status, failed_at: Date.now(), }); await deleteOverflowBody(env, job); await decrQueueDepth(env); msg.ack(); lastError = null; break; } await recordUsageAppwrite(env, { usuario_id: job.auth_user_id, clave_api_id: job.auth_key_id || null, peticion_id: job.request_id || `queue-${jobId}`, endpoint: job.path.split("?")[0], bytes_entrada: Number(res.headers.get("x-input-bytes") || job.declared_input_bytes || 0), bytes_salida: Number(res.headers.get("x-markdown-bytes") || 0), nivel: res.headers.get("x-price-tier") || null, creditos_cobrados: creditsCharged, precio_usd: Number(res.headers.get("x-request-price-usd") || 0), estado_http: res.status, }); } await cacheJobStatus(jobId, { job_id: jobId, status: "completed", http_status: res.status, headers: resultHeaders, body: resultBody.length <= 512_000 ? resultBody : resultBody.slice(0, 512_000), truncated: resultBody.length > 512_000, completed_at: Date.now(), }); await deleteOverflowBody(env, job); await decrQueueDepth(env); msg.ack(); lastError = null; break; } catch (err) { lastError = `${upstream.origin}: ${err.message}`; } } if (lastError) { if (msg.attempts < 3) { msg.retry({ delaySeconds: 10 * msg.attempts }); } else { await cacheJobStatus(jobId, { job_id: jobId, status: "failed", error: lastError, failed_at: Date.now(), }); await deleteOverflowBody(env, job); await decrQueueDepth(env); msg.ack(); } } } catch (err) { await cacheJobStatus(jobId, { job_id: jobId, status: "failed", error: err.message }); await deleteOverflowBody(env, job); await decrQueueDepth(env); msg.ack(); } } }; // GC de objetos R2 huerfanos (C9, cierre completo): deleteOverflowBody borra // el body apenas el job llega a un estado terminal, pero si el Worker // crashea a mitad de worker.queue (OOM/wall-time) ANTES de ese borrado, el // objeto en PDF_OVERFLOW queda huerfano para siempre -- nadie mas lo borra. // Cron Trigger (ver [triggers] en wrangler.toml) barre la cola cada cierto // tiempo y borra lo que sobrevivio mas de ORPHAN_OVERFLOW_MAX_AGE_MS: un job // normal tarda segundos/minutos, asi que cualquier objeto mas viejo que eso // ya perdio su job y es basura segura de borrar. const ORPHAN_OVERFLOW_MAX_AGE_MS = 30 * 60 * 1000; worker.scheduled = async function (_event, env, _ctx) { if (!env.PDF_OVERFLOW) return; const cutoff = Date.now() - ORPHAN_OVERFLOW_MAX_AGE_MS; let cursor; do { const listing = await env.PDF_OVERFLOW.list({ prefix: "overflow/", cursor }); for (const obj of listing.objects) { if (obj.uploaded.getTime() < cutoff) { await env.PDF_OVERFLOW.delete(obj.key); } } cursor = listing.truncated ? listing.cursor : undefined; } while (cursor); }; export default Sentry.withSentry(sentryConfig, worker); // Exports nombrados SOLO para tests unitarios (tests/worker/*.test.js, `node // --test`). No cambian el runtime: el Worker sigue usando unicamente el // default export de arriba. Cubre justo la logica pura que hoy no tenia // ningun test (billing gate y routing/failover -- ahi vivieron los bugs // reales de esta semana: tabla D1 faltante, 413 no reintentable, backend // pegado por scoring). export { normalizeBalanceAccount, balanceGateResponse, accountGate, routingScore, buildAttemptOrder, providerFromOrigin, filterOrderByBodySize, upstreamBodyLimit, recordUpstreamHealth, UPSTREAM_RETRYABLE_STATUS, RETRYABLE_STATUS, enqueueJob, incrQueueDepth, decrQueueDepth, gateAndChargeD1Idempotent, timingSafeEqualStr, requireAdmin, worker, };

​n8n

​HTTP Request node

n8n

HTTP Request node