1"use node";2 3// HTTP half of outbound delivery.4//5// Direction is strictly store → IAPKit → developer backend, server to server:6// destinations are HTTPS endpoints a project owner registered, nothing here is7// reachable from a shipped app, and no client-pullable stream exists.8 9import { v } from "convex/values";10import { lookup as dnsLookup } from "node:dns/promises";11import { request as httpsRequest, type RequestOptions } from "node:https";12import { isIP, type LookupFunction } from "node:net";13 14import { internal } from "../_generated/api";15import { internalAction, type ActionCtx } from "../_generated/server";16import type { ClaimedDelivery } from "./deliveryState";17import {18 REQUEST_TIMEOUT_MS,19 checkDestinationUrl,20 CLAIM_BATCH_LIMIT,21 classifyDeliveryError,22 classifyDeliveryResponse,23 composeDeliveryHeaders,24 isPublicIpAddress,25} from "./signing";26 27export type ResolvedAddress = { address: string; family: number };28export const MAX_FALLBACK_ADDRESSES = 4;29type Resolver = (30 hostname: string,31 options: { all: true; verbatim: true },32) => Promise<ResolvedAddress[]>;33 34export async function resolvePublicAddresses(35 rawHostname: string,36 resolver: Resolver = dnsLookup,37): Promise<ResolvedAddress[]> {38 const hostname = rawHostname.replace(/^\[|\]$/g, "").replace(/\.$/, "");39 const family = isIP(hostname);40 const addresses = family41 ? [{ address: hostname, family }]42 : await resolver(hostname, { all: true, verbatim: true });43 if (44 addresses.length === 0 ||45 addresses.some((entry) => !isPublicIpAddress(entry.address))46 ) {47 throw new Error("destination resolved to a non-public address");48 }49 return [50 ...new Map(51 addresses.map((entry) => [`${entry.family}:${entry.address}`, entry]),52 ).values(),53 ].slice(0, MAX_FALLBACK_ADDRESSES);54}55 56export type DeliveryRequest = {57 url: URL;58 headers: Record<string, string>;59 body: string;60};61 62type AddressPoster = (63 request: DeliveryRequest,64 selected: ResolvedAddress,65 timeoutMs: number,66) => Promise<number>;67 68async function withinTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {69 let timer: ReturnType<typeof setTimeout> | undefined;70 try {71 return await Promise.race([72 promise,73 new Promise<T>((_resolve, reject) => {74 timer = setTimeout(() => reject(new Error("request timed out")), ms);75 }),76 ]);77 } finally {78 if (timer) clearTimeout(timer);79 }80}81 82async function postJsonToAddress(83 request: DeliveryRequest,84 selected: ResolvedAddress,85 timeoutMs: number,86): Promise<number> {87 return await new Promise<number>((resolve, reject) => {88 const outgoing = httpsRequest(89 buildPinnedRequestOptions(request, selected),90 (response) => {91 clearTimeout(timer);92 const status = response.statusCode ?? 0;93 response.destroy();94 resolve(status);95 },96 );97 const timer = setTimeout(() => {98 outgoing.destroy(new Error("request timed out"));99 }, timeoutMs);100 outgoing.once("error", (error) => {101 clearTimeout(timer);102 reject(error);103 });104 outgoing.end(request.body);105 });106}107 108export function buildPinnedRequestOptions(109 request: DeliveryRequest,110 selected: ResolvedAddress,111): RequestOptions {112 const hostname = request.url.hostname113 .replace(/^\[|\]$/g, "")114 .replace(/\.$/, "");115 const pinnedLookup: LookupFunction = (_hostname, _options, callback) => {116 if (_options.all) {117 callback(null, [selected]);118 } else {119 callback(null, selected.address, selected.family);120 }121 };122 123 return {124 protocol: "https:",125 hostname,126 port: request.url.port || undefined,127 path: `${request.url.pathname}${request.url.search}`,128 method: "POST",129 headers: request.headers,130 lookup: pinnedLookup,131 ...(isIP(hostname) === 0 ? { servername: hostname } : {}),132 };133}134 135export async function postJsonPinned(136 request: DeliveryRequest,137 resolver: Resolver = dnsLookup,138 postToAddress: AddressPoster = postJsonToAddress,139): Promise<number> {140 const startedAt = Date.now();141 const hostname = request.url.hostname142 .replace(/^\[|\]$/g, "")143 .replace(/\.$/, "");144 const addresses = await withinTimeout(145 resolvePublicAddresses(hostname, resolver),146 REQUEST_TIMEOUT_MS,147 );148 let lastError: unknown;149 for (let index = 0; index < addresses.length; index += 1) {150 const remaining = REQUEST_TIMEOUT_MS - (Date.now() - startedAt);151 if (remaining <= 0) break;152 const addressesLeft = addresses.length - index;153 const attemptBudget =154 addressesLeft === 1155 ? remaining156 : Math.max(1, Math.min(3_000, Math.floor(remaining / addressesLeft)));157 try {158 return await postToAddress(request, addresses[index], attemptBudget);159 } catch (error) {160 lastError = error;161 }162 }163 throw lastError instanceof Error164 ? lastError165 : new Error("all destination addresses failed");166}167 168export async function deliverPendingEventsHandler(169 ctx: Pick<ActionCtx, "runMutation">,170 post: (request: DeliveryRequest) => Promise<number> = postJsonPinned,171): Promise<{ attempted: number; delivered: number }> {172 let attempted = 0;173 let delivered = 0;174 175 for (let index = 0; index < CLAIM_BATCH_LIMIT; index += 1) {176 const claimed: ClaimedDelivery[] = await ctx.runMutation(177 internal.commerce.deliveryState.claimPendingDeliveries,178 {},179 );180 const [item] = claimed;181 if (!item) break;182 attempted += 1;183 184 const check = checkDestinationUrl(item.url);185 if (!check.ok) {186 await ctx.runMutation(187 internal.commerce.deliveryState.recordDeliveryResult,188 {189 deliveryId: item.deliveryId,190 leaseToken: item.leaseToken,191 ok: false,192 error: `destination rejected: ${check.reason}`,193 errorClass: "destination_rejected",194 retryable: false,195 },196 );197 continue;198 }199 200 const headers = await composeDeliveryHeaders({201 secrets: { current: item.secret, previous: item.previousSecret },202 timestampSeconds: Math.floor(Date.now() / 1000),203 body: item.body,204 eventId: item.eventId,205 deliveryId: item.deliveryId,206 });207 208 const startedAt = Date.now();209 try {210 const status = await post({ url: check.url, headers, body: item.body });211 const outcome = classifyDeliveryResponse(status);212 if (outcome === "delivered") delivered += 1;213 await ctx.runMutation(214 internal.commerce.deliveryState.recordDeliveryResult,215 {216 deliveryId: item.deliveryId,217 leaseToken: item.leaseToken,218 ok: outcome === "delivered",219 statusCode: status,220 latencyMs: Date.now() - startedAt,221 retryable: outcome === "retry",222 },223 );224 } catch (error) {225 // No status arrived (timeout, connection error): the protocol retries.226 const outcome = classifyDeliveryResponse(undefined);227 await ctx.runMutation(228 internal.commerce.deliveryState.recordDeliveryResult,229 {230 deliveryId: item.deliveryId,231 leaseToken: item.leaseToken,232 ok: false,233 error: error instanceof Error ? error.message : "request failed",234 errorClass: classifyDeliveryError(error),235 latencyMs: Date.now() - startedAt,236 retryable: outcome === "retry",237 },238 );239 }240 }241 242 return { attempted, delivered };243}244 245export const deliverPendingEvents = internalAction({246 args: {},247 returns: v.object({ attempted: v.number(), delivered: v.number() }),248 handler: async (ctx) => deliverPendingEventsHandler(ctx),249});250