Fix: 4 issues (#83, #84, #85, #86)

drizzle/0011_onchain_attempts.sql

@@ -0,0 +1,2 @@
+ALTER TABLE resources ADD COLUMN onchain_attempts integer NOT NULL DEFAULT 0;
+ALTER TABLE payments ADD COLUMN delivery_status text NOT NULL DEFAULT 'delivered';

src/config.ts

@@ -244,6 +244,11 @@ const envSchema = z.object({
   WEBHOOK_TIMEOUT_MS: z.coerce.number().default(10_000),
   WEBHOOK_MAX_ATTEMPTS: z.coerce.number().default(3),
 
+  // Retry worker (#85): max items processed per tick and per-item timeout so a
+  // slow RPC can't make a tick run into the next one.
+  RETRY_WORKER_BATCH_SIZE: z.coerce.number().int().min(1).default(25),
+  RETRY_WORKER_ITEM_TIMEOUT_MS: z.coerce.number().int().min(1).default(30_000),
+
   // Optional Redis URL for a shared sliding-window rate-limit store (multi-instance).
   REDIS_URL: z.string().url().optional(),
 

src/db/schema.ts

@@ -91,6 +91,9 @@ export const resources = pgTable(
     adminDelistedAt: timestamp("admin_delisted_at"),
     onchainStatus: onchainStatusEnum("onchain_status").notNull().default("none"),
     onchainTxHash: text("onchain_tx_hash"),
+    // Number of on-chain registration attempts made so far. Persisted so the
+    // retry budget survives worker restarts (#84).
+    onchainAttempts: integer("onchain_attempts").notNull().default(0),
     // Set when onchainStatus transitions to "pending". The retry worker uses
     // this (not createdAt) for its cutoff so a freshly pending old resource
     // isn't retried before a live registration can finish (#82).
@@ -175,6 +178,9 @@ export const payments = pgTable(
     payerAddress: text("payer_address").notNull(),
     recipientAddress: text("recipient_address").notNull(),
     amount: text("amount").notNull(), // USDC amount
+    // Delivery state of the purchased content: "delivered" once the file was
+    // served, "undelivered" when the download failed after payment (#86).
+    deliveryStatus: text("delivery_status").notNull().default("delivered"),
     // Settlement state of the payment. Only settled payments count toward
     // earnings (#99).
     status: paymentStatusEnum("status").notNull().default("pending"),

src/routes/resources.ts

@@ -307,29 +307,39 @@ router.get("/resources/:id", dynamicPaywall, async (req, res) => {
     }
   }
 
-  const [payment] = await db
-    .insert(payments)
-    .values({
-      resourceId: resource.id,
-      payerAddress,
-      recipientAddress: resource.walletAddress,
-      amount: resource.price,
-    })
-    .returning();
-
-  // Fire-and-forget: webhook delivery never delays or fails content access.
-  const purchase = {
-    paymentId: payment.id,
-    resourceId: payment.resourceId,
-    amount: payment.amount,
-    payerAddress: payment.payerAddress,
-    recipientAddress: payment.recipientAddress,
-    paidAt: payment.paidAt,
+  // Records a payment row with an explicit delivery status so an undelivered
+  // purchase can be reconciled later (#86).
+  const recordPayment = async (deliveryStatus: "delivered" | "undelivered") => {
+    const [payment] = await db
+      .insert(payments)
+      .values({
+        resourceId: resource.id,
+        payerAddress,
+        recipientAddress: resource.walletAddress,
+        amount: resource.price,
+        deliveryStatus,
+      })
+      .returning();
+    return payment;
+  };
+
+  const emitPurchase = (payment: typeof payments.$inferSelect) => {
+    // Fire-and-forget: webhook delivery never delays or fails content access.
+    const purchase = {
+      paymentId: payment.id,
+      resourceId: payment.resourceId,
+      amount: payment.amount,
+      payerAddress: payment.payerAddress,
+      recipientAddress: payment.recipientAddress,
+      paidAt: payment.paidAt,
+    };
+    void emitToPublisher(resource.publisherId, "resource.purchased", purchase);
+    void emitToPublisher(resource.publisherId, "payment.received", purchase);
   };
-  void emitToPublisher(resource.publisherId, "resource.purchased", purchase);
-  void emitToPublisher(resource.publisherId, "payment.received", purchase);
 
   if (resource.resourceType === "link") {
+    const payment = await recordPayment("delivered");
+    emitPurchase(payment);
     res.json({
       url: resource.externalUrl,
       receipt: {
@@ -348,12 +358,39 @@ router.get("/resources/:id", dynamicPaywall, async (req, res) => {
     throw new AppError("INTERNAL_ERROR", "Resource file not found");
   }
 
+  // Fetch the content *before* recording the payment so a storage failure
+  // never leaves a silently undelivered purchase.
+  let fileContent: Awaited<ReturnType<typeof downloadFile>>;
+  try {
+    fileContent = await downloadFile(resource.storagePath);
+  } catch (err) {
+    const payment = await recordPayment("undelivered");
+    getLogger().error(
+      {
+        event: "resource_delivery_failed",
+        resourceId: resource.id,
+        paymentId: payment.id,
+        payerAddress,
+        storagePath: resource.storagePath,
+        err,
+      },
+      "file download failed after payment; payment marked undelivered for reconciliation",
+    );
+    throw new AppError("UPSTREAM_ERROR", "Failed to deliver resource content", {
+      paymentId: payment.id,
+      deliveryStatus: "undelivered",
+    });
+  }
+
+  const payment = await recordPayment("delivered");
+  emitPurchase(payment);
+
   // Add receipt info in headers for file downloads
   res.setHeader("X-Payment-Id", payment.id);
   res.setHeader("X-Payment-Amount", `${payment.amount} USDC`);
   res.setHeader("X-Payment-Recipient", payment.recipientAddress);
 
-  const { buffer, mimeType } = await downloadFile(resource.storagePath);
+  const { buffer, mimeType } = fileContent;
   res.setHeader("Content-Type", mimeType);
   res.setHeader(
     "Content-Disposition",

src/workers/retryPendingWorker.ts

@@ -35,6 +35,7 @@ export async function enqueueThumbnail(resourceId: string): Promise<void> {
 }
 
 let intervalHandle: ReturnType<typeof setInterval> | null = null;
+let inFlight = false;
 
 export function startRetryPendingWorker(): void {
   getLogger().info(
@@ -51,19 +52,31 @@ export function stopRetryPendingWorker(): void {
     clearInterval(intervalHandle);
     intervalHandle = null;
   }
-  retryCounts.clear();
 }
 
 async function tick(): Promise<void> {
+  if (inFlight) {
+    getLogger().debug(
+      { event: "retry_worker_tick_skipped" },
+      "previous retry worker tick still running; skipping",
+    );
+    return;
+  }
+
+  inFlight = true;
   try {
     const cutoff = sql`NOW() - INTERVAL '5 minutes'`;
 
     const stuck = await db
       .select()
       .from(resources)
       .where(
-        and(eq(resources.onchainStatus, "pending"), lt(resources.onchainPendingSince, cutoff)),
-      );
+        and(
+          eq(resources.onchainStatus, "pending"),
+          lt(resources.onchainPendingSince, cutoff),
+        ),
+      )
+      .limit(config.RETRY_WORKER_BATCH_SIZE);
 
     if (stuck.length === 0) return;
 
@@ -73,16 +86,39 @@ async function tick(): Promise<void> {
     );
 
     for (const resource of stuck) {
-      await retryResource(resource);
+      await withTimeout(
+        retryResource(resource),
+        config.RETRY_WORKER_ITEM_TIMEOUT_MS,
+        resource.id,
+      );
     }
   } catch (err) {
     getLogger().error({ event: "retry_worker_tick_error", err }, "retry worker tick failed");
+  } finally {
+    inFlight = false;
   }
 
   await processCleanupJobs();
   await processThumbnailJobs();
 }
 
+async function withTimeout<T>(promise: Promise<T>, ms: number, resourceId: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_, reject) => {
+        timer = setTimeout(
+          () => reject(new Error(`retryResource timed out after ${ms}ms for ${resourceId}`)),
+          ms,
+        );
+      }),
+    ]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
 async function processCleanupJobs(): Promise<void> {
   const log = getLogger();
   try {
@@ -182,7 +218,30 @@ async function processThumbnailJobs(): Promise<void> {
 async function retryResource(resource: typeof resources.$inferSelect): Promise<void> {
   const log = getLogger().child({ resourceId: resource.id });
 
+  let attempts = 0;
+
   try {
+    const [incremented] = await db
+      .update(resources)
+      .set({ onchainAttempts: sql`${resources.onchainAttempts} + 1` })
+      .where(eq(resources.id, resource.id))
+      .returning({ onchainAttempts: resources.onchainAttempts });
+
+    attempts = incremented?.onchainAttempts ?? 0;
+
+    if (attempts > MAX_RETRY_ATTEMPTS) {
+      log.warn(
+        { event: "retry_exhausted", attempts },
+        "max retry attempts exhausted; marking as failed",
+      );
+      await db
+        .update(resources)
+        .set({ onchainStatus: "failed" })
+        .where(eq(resources.id, resource.id));
+      retryCounts.delete(resource.id);
+      return;
+    }
+
     if (resource.onchainTxHash) {
       const status = await pollTransactionStatus(resource.onchainTxHash);
 
@@ -229,23 +288,6 @@ async function retryResource(resource: typeof resources.$inferSelect): Promise<v
         .update(resources)
         .set({ onchainStatus: "registered" })
         .where(eq(resources.id, resource.id));
-      retryCounts.delete(resource.id);
-      return;
-    }
-
-    const attempts = (retryCounts.get(resource.id) ?? 0) + 1;
-    retryCounts.set(resource.id, attempts);
-
-    if (attempts > MAX_RETRY_ATTEMPTS) {
-      log.warn(
-        { event: "retry_exhausted", attempts: attempts - 1 },
-        "max retry attempts exhausted; marking as failed",
-      );
-      await db
-        .update(resources)
-        .set({ onchainStatus: "failed" })
-        .where(eq(resources.id, resource.id));
-      retryCounts.delete(resource.id);
       return;
     }
 
@@ -316,7 +358,6 @@ async function retryResource(resource: typeof resources.$inferSelect): Promise<v
       })
       .where(eq(resources.id, resource.id));
 
-    retryCounts.delete(resource.id);
     log.info(
       { event: "retry_success", txHash: txHash || undefined },
       "on-chain registration retry succeeded",
@@ -327,17 +368,15 @@ async function retryResource(resource: typeof resources.$inferSelect): Promise<v
       "on-chain registration retry failed",
     );
 
-    const attempts = retryCounts.get(resource.id) ?? 1;
     if (attempts > MAX_RETRY_ATTEMPTS) {
       log.warn(
-        { event: "retry_exhausted_on_error", attempts: attempts - 1 },
+        { event: "retry_exhausted_on_error", attempts },
         "max retry attempts exhausted after error; marking as failed",
       );
       await db
         .update(resources)
         .set({ onchainStatus: "failed" })
         .where(eq(resources.id, resource.id));
-      retryCounts.delete(resource.id);
     }
   }
 }