Repository Public
1import { newId, redactHeaders, type ConcreteNetwork, type FetchRequest, type Plan } from "@fetcha/core";2import { db, domainProfiles, fetchRequests, requestAttempts, routingMetrics, usageEvents, sql } from "@fetcha/db";3import { foldRouteStat, preferredRoute, routeKey, type AttemptRecord, type ExecutionResult } from "@fetcha/routing";4import { priceRequest } from "./pricing";56export interface RequestContext {7 requestId: string;8 organizationId: string;9 projectId: string;10 apiKeyId: string | null;11 source: "api" | "playground" | "sdk";12 plan: Plan;13 clientIp: string | null;14 userAgent: string | null;15 logLevel: "none" | "metadata" | "headers" | "full";16}1718export async function createRequestRow(ctx: RequestContext, req: FetchRequest, domain: string): Promise<void> {19 await db.insert(fetchRequests).values({20 id: ctx.requestId,21 organizationId: ctx.organizationId,22 projectId: ctx.projectId,23 apiKeyId: ctx.apiKeyId,24 source: ctx.source,25 url: req.url,26 domain,27 method: req.method,28 requestedNetwork: req.network,29 country: req.country ?? null,30 region: req.region ?? null,31 city: req.city ?? null,32 sessionId: req.session ?? null,33 browser: req.browser,34 format: req.format,35 status: "pending",36 requestHeaders: ctx.logLevel === "headers" || ctx.logLevel === "full" ? redactHeaders(req.headers) : null,37 clientIp: ctx.clientIp,38 userAgent: ctx.userAgent,39 });40}4142export async function recordAttempt(requestId: string, a: AttemptRecord): Promise<void> {43 await db.insert(requestAttempts).values({44 id: a.attemptId,45 requestId,46 attemptNo: a.attemptNo,47 provider: a.provider,48 network: a.network,49 country: a.country,50 sessionKey: a.sessionKey,51 outcome: a.outcome,52 httpStatus: a.httpStatus,53 errorCode: a.errorCode,54 errorDetail: a.errorDetail,55 blockReason: a.blockReason,56 durationMs: a.durationMs,57 bytesIn: a.bytesIn,58 bytesOut: a.bytesOut,59 unitPricePerGb: a.unitPricePerGb,60 costUsd: a.costUsd,61 routingScore: a.routingScore,62 timing: a.timing ? { ...a.timing } : null,63 });64}6566export interface CompletionSummary {67 priceUsd: number;68 costUsd: number;69}7071/** Persist a successful (or blocked-but-answered) execution: request row, usage ledger, domain intelligence, routing metrics. */72export async function completeRequest(ctx: RequestContext, req: FetchRequest, result: ExecutionResult, latencyMs: number): Promise<CompletionSummary> {73 const network = result.network;74 const price = priceRequest({ plan: ctx.plan, network, bytes: result.bytesIn + result.bytesOut, upstreamCostUsd: result.costUsd, attempts: result.attempts.length, success: result.body.success });75 const finalStatus = result.body.success ? "success" : "failed";7677 await db78 .update(fetchRequests)79 .set({80 status: finalStatus,81 httpStatus: result.body.status,82 errorCode: result.body.success ? null : "TARGET_BLOCKED",83 errorMessage: result.body.success ? null : "The target blocked every route we tried.",84 finalUrl: result.finalUrl,85 network,86 attempts: result.attempts.length,87 latencyMs,88 bytesIn: result.bytesIn,89 bytesOut: result.bytesOut,90 costUsd: result.costUsd,91 priceUsd: price.totalUsd,92 responseHeaders: ctx.logLevel === "headers" || ctx.logLevel === "full" ? redactHeaders(result.body.headers) : null,93 timing: result.body.metadata.timing ? { ...result.body.metadata.timing } : null,94 completedAt: new Date(),95 })96 .where(sql`${fetchRequests.id} = ${ctx.requestId}`);9798 await writeUsage(ctx, result, price.totalUsd, network);99 await Promise.all([updateDomainProfile(result.domain, result.attempts, result.body.success), updateRoutingMetrics(result.attempts)]);100 return { priceUsd: price.totalUsd, costUsd: result.costUsd };101}102103export async function failRequest(ctx: RequestContext, code: string, message: string, attempts: AttemptRecord[], latencyMs: number, domain: string): Promise<void> {104 const costUsd = attempts.reduce((s, a) => s + a.costUsd, 0);105 await db106 .update(fetchRequests)107 .set({ status: "failed", errorCode: code, errorMessage: message, attempts: attempts.length, latencyMs, costUsd, completedAt: new Date() })108 .where(sql`${fetchRequests.id} = ${ctx.requestId}`);109 // Failed requests still count against the request quota (abuse control) but are not priced.110 await db.insert(usageEvents).values({111 id: newId("usage"),112 organizationId: ctx.organizationId,113 projectId: ctx.projectId,114 requestId: ctx.requestId,115 metric: "request",116 quantity: 1,117 unit: "count",118 costUsd: 0,119 upstreamCostUsd: costUsd,120 });121 if (attempts.length) await Promise.all([updateDomainProfile(domain, attempts, false), updateRoutingMetrics(attempts)]);122}123124async function writeUsage(ctx: RequestContext, result: ExecutionResult, priceUsd: number, network: ConcreteNetwork | null) {125 const bytes = result.bytesIn + result.bytesOut;126 const rows = [127 { metric: "request", quantity: 1, unit: "count", costUsd: priceUsd, upstreamCostUsd: result.costUsd },128 { metric: "bandwidth", quantity: bytes, unit: "bytes", costUsd: 0, upstreamCostUsd: 0 },129 ];130 if (network === "residential" || network === "isp") rows.push({ metric: "residential_bandwidth", quantity: bytes, unit: "bytes", costUsd: 0, upstreamCostUsd: 0 });131 if (network === "mobile") rows.push({ metric: "mobile_bandwidth", quantity: bytes, unit: "bytes", costUsd: 0, upstreamCostUsd: 0 });132 await db.insert(usageEvents).values(133 rows.map((r) => ({ id: newId("usage"), organizationId: ctx.organizationId, projectId: ctx.projectId, requestId: ctx.requestId, ...r })),134 );135}136137async function updateDomainProfile(domain: string, attempts: AttemptRecord[], success: boolean) {138 if (!domain) return;139 const [existing] = await db.select().from(domainProfiles).where(sql`${domainProfiles.domain} = ${domain}`).limit(1);140 const stats = { ...(existing?.routeStats ?? {}) };141 let blocks = 0;142 let captchas = 0;143 for (const a of attempts) {144 foldRouteStat(stats, routeKey(a.provider, a.network), { ok: a.outcome === "success", blocked: a.outcome === "blocked", latencyMs: a.durationMs, costUsd: a.costUsd });145 if (a.outcome === "blocked") blocks++;146 if (a.blockReason === "captcha" || a.blockReason === "cloudflare_challenge") captchas++;147 }148 const pref = preferredRoute(stats);149 const totalLatency = attempts.reduce((s, a) => s + a.durationMs, 0);150 const n = (existing?.requests ?? 0) + 1;151 const avgLatency = existing ? existing.avgLatencyMs + (totalLatency - existing.avgLatencyMs) / n : totalLatency;152 await db153 .insert(domainProfiles)154 .values({155 domain,156 preferredNetwork: pref?.network ?? null,157 preferredProvider: pref?.provider ?? null,158 requests: 1,159 successes: success ? 1 : 0,160 blocks,161 captchas,162 avgLatencyMs: totalLatency,163 routeStats: stats,164 lastSeenAt: new Date(),165 })166 .onConflictDoUpdate({167 target: domainProfiles.domain,168 set: {169 preferredNetwork: pref?.network ?? null,170 preferredProvider: pref?.provider ?? null,171 requests: sql`${domainProfiles.requests} + 1`,172 successes: sql`${domainProfiles.successes} + ${success ? 1 : 0}`,173 blocks: sql`${domainProfiles.blocks} + ${blocks}`,174 captchas: sql`${domainProfiles.captchas} + ${captchas}`,175 avgLatencyMs: avgLatency,176 routeStats: stats,177 lastSeenAt: new Date(),178 updatedAt: new Date(),179 },180 });181}182183async function updateRoutingMetrics(attempts: AttemptRecord[]) {184 const bucket = new Date();185 bucket.setMinutes(0, 0, 0);186 for (const a of attempts) {187 const ok = a.outcome === "success" ? 1 : 0;188 const blocked = a.outcome === "blocked" ? 1 : 0;189 const errors = ok || blocked ? 0 : 1;190 await db191 .insert(routingMetrics)192 .values({193 id: newId("evt"),194 bucket,195 provider: a.provider,196 network: a.network,197 country: a.country ?? "",198 requests: 1,199 successes: ok,200 blocked,201 errors,202 latencySumMs: a.durationMs,203 bytes: a.bytesIn + a.bytesOut,204 costUsd: a.costUsd,205 })206 .onConflictDoUpdate({207 target: [routingMetrics.bucket, routingMetrics.provider, routingMetrics.network, routingMetrics.country],208 set: {209 requests: sql`${routingMetrics.requests} + 1`,210 successes: sql`${routingMetrics.successes} + ${ok}`,211 blocked: sql`${routingMetrics.blocked} + ${blocked}`,212 errors: sql`${routingMetrics.errors} + ${errors}`,213 latencySumMs: sql`${routingMetrics.latencySumMs} + ${a.durationMs}`,214 bytes: sql`${routingMetrics.bytes} + ${a.bytesIn + a.bytesOut}`,215 costUsd: sql`${routingMetrics.costUsd} + ${a.costUsd}`,216 },217 });218 }219}220