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