Merge pull request #621 from autonomys/feat/pay-with-ai3/step-10-payment-robustness

Feat/pay with ai3/step 10 payment robustness

Author: EmilFattakhov

apps/backend/.env.sample

@@ -4,5 +4,60 @@ FILES_GATEWAY_URL=http://changeme.com # change to your own (or it'll fail when f
 FILES_GATEWAY_TOKEN=changeme # change to your own (or it'll fail when fetching archived files)
 AUTH_SERVICE_API_KEY=1234567890 # change to your own if updated in auth
 RABBITMQ_URL=amqp://guest:guest@localhost:5672
-EVM_CHAIN_ENDPOINT=http://localhost:8545 # update it you want to simulate/test buy credits feature
-EVM_CHAIN_CONTRACT_ADDRESS=0x0000000000000000000000000000000000000000 # update it you want to simulate/test buy credits feature
+
+# ---------------------------------------------------------------------------
+# Pay-with-AI3 / purchased credits feature
+# ---------------------------------------------------------------------------
+
+# Feature flags — set BUY_CREDITS_ACTIVE=true to enable the purchase flow.
+# BUY_CREDITS_STAFF_ONLY=true restricts it to admin/staff accounts only,
+# useful for a staged rollout before opening to all users.
+BUY_CREDITS_ACTIVE=false
+BUY_CREDITS_STAFF_ONLY=false
+
+# EVM endpoint for the AutoDriveCreditsReceiver contract.
+# Point this at the Auto-EVM RPC for the target network (mainnet or Taurus
+# testnet).  The payment manager watches this chain for deposit events.
+EVM_CHAIN_ENDPOINT=http://localhost:8545
+
+# Address of the deployed AutoDriveCreditsReceiver contract.
+# Must match the chain pointed to by EVM_CHAIN_ENDPOINT.
+EVM_CHAIN_CONTRACT_ADDRESS=0x0000000000000000000000000000000000000000
+
+# Number of block confirmations to wait before treating a payment as final.
+# Higher values reduce the risk of processing a payment that is later
+# reversed by a chain reorganisation. Default: 6.
+EVM_CHAIN_CONFIRMATIONS=6
+
+# How often (in milliseconds) the payment manager polls for CONFIRMED intents
+# that have not yet had credits applied. This is a fallback for cases where
+# the event watcher misses a log. Default: 30000 (30 seconds).
+EVM_CHAIN_CHECK_INTERVAL=30000
+
+# Price multiplier applied on top of the raw Autonomys consensus fee to
+# determine the AI3 cost per byte. A value of 5.0 means users pay 5× the
+# current on-chain transaction byte fee. Default: 5.00.
+CREDITS_PRICE_MULTIPLIER=5.00
+
+# ---------------------------------------------------------------------------
+# Credit lifecycle
+# ---------------------------------------------------------------------------
+
+# Number of days after purchase before a credit batch expires.
+# Users see this value on the purchase confirmation screen and in their credit
+# history. Default: 90.
+CREDIT_EXPIRY_DAYS=90
+
+# Maximum total purchased upload bytes allowed per user across all active
+# (non-expired) credit rows. Attempts to purchase beyond this cap result in
+# an OVER_CAP intent requiring admin review. Default: 107374182400 (100 GiB).
+MAX_CREDITS_PER_USER=107374182400
+
+# How often (in milliseconds) the background job runs to mark expired credit
+# rows and clean up stale PENDING intents. Default: 3600000 (1 hour).
+CREDIT_EXPIRY_CHECK_INTERVAL=3600000
+
+# How many minutes a PENDING intent remains valid before it is treated as
+# expired. Users must submit their on-chain transaction within this window
+# after creating an intent. Default: 10.
+INTENT_EXPIRY_MINUTES=10

apps/backend/src/app/controllers/credits.ts

@@ -117,6 +117,44 @@ creditsController.get(
   }),
 )
 
+// ---------------------------------------------------------------------------
+// GET /credits/batches/all
+// Admin-only: all credit batches across every user, newest-first.
+// Each row includes the owner's userPublicId for easy cross-referencing
+// with the admin user table.  Returns 403 for non-admin callers.
+//
+// NOTE: registered BEFORE GET /credits/batches so Express does not attempt
+// to match the literal string "all" against the existing /batches route
+// (they are separate paths and Express won't confuse them, but ordering
+// here keeps the admin routes grouped together).
+// ---------------------------------------------------------------------------
+
+creditsController.get(
+  '/batches/all',
+  asyncSafeHandler(async (req, res) => {
+    const user = await handleAuth(req, res)
+    if (!user) {
+      return
+    }
+
+    const result = await handleInternalErrorResult(
+      CreditsUseCases.getAllBatches(user),
+      'Failed to get all credit batches',
+    )
+    if (result.isErr()) {
+      handleError(result.error, res)
+      return
+    }
+
+    res.status(200).json(
+      result.value.map((batch) => ({
+        ...serializeCredit(batch),
+        userPublicId: batch.userPublicId,
+      })),
+    )
+  }),
+)
+
 // ---------------------------------------------------------------------------
 // GET /credits/economics
 // Admin-only: system-wide credit stats (expiring totals, byte volumes).

apps/backend/src/core/users/credits.ts

@@ -1,5 +1,8 @@
 import { PurchasedCredit, User, UserRole, UserWithOrganization } from '@auto-drive/models'
-import { purchasedCreditsRepository } from '../../infrastructure/repositories/users/purchasedCredits.js'
+import {
+  AdminCreditBatchRow,
+  purchasedCreditsRepository,
+} from '../../infrastructure/repositories/users/purchasedCredits.js'
 import { AccountsUseCases } from './accounts.js'
 import { config } from '../../config.js'
 import { ForbiddenError } from '../../errors/index.js'
@@ -135,9 +138,30 @@ const getEconomics = async (
   })
 }
 
+// ---------------------------------------------------------------------------
+// getAllBatches
+// Admin-only: full purchase history across all users with their publicId.
+// Returns 403 for non-admin callers.
+// ---------------------------------------------------------------------------
+
+const getAllBatches = async (
+  executor: User,
+): Promise<Result<AdminCreditBatchRow[], ForbiddenError>> => {
+  if (executor.role !== UserRole.Admin) {
+    logger.warn('Non-admin user attempted to access all credit batches', {
+      publicId: executor.publicId,
+    })
+    return err(new ForbiddenError('Admin access required'))
+  }
+
+  const rows = await purchasedCreditsRepository.getAllWithUserPublicId()
+  return ok(rows)
+}
+
 export const CreditsUseCases = {
   getSummary,
   getBatches,
   getExpiringBatches,
   getEconomics,
+  getAllBatches,
 }

apps/backend/src/core/users/intents.ts

@@ -155,8 +155,30 @@ const markIntentAsConfirmed = async ({
     return err(new ObjectNotFoundError('Intent not found'))
   }
 
+  // Idempotency guard — do not overwrite an intent that is already in a
+  // post-PENDING state.  Duplicate calls arise from:
+  //   • chain reorgs causing the same event to be re-emitted
+  //   • the payment manager reconnecting and re-processing already-seen logs
+  //   • watchTransaction and the _checkConfirmedIntents polling loop racing
+  //
+  // We return ok() rather than an error so the caller does not treat a
+  // duplicate as a failure and does not retry indefinitely.
+  if (
+    intent.status === IntentStatus.CONFIRMED ||
+    intent.status === IntentStatus.COMPLETED ||
+    intent.status === IntentStatus.OVER_CAP ||
+    intent.status === IntentStatus.FAILED ||
+    intent.status === IntentStatus.EXPIRED
+  ) {
+    logger.info('markIntentAsConfirmed: intent already processed — skipping', {
+      intentId,
+      currentStatus: intent.status,
+    })
+    return ok(intent)
+  }
+
   return ok(
-    intentsRepository.updateIntent({
+    await intentsRepository.updateIntent({
       ...intent,
       status: IntentStatus.CONFIRMED,
       paymentAmount,
@@ -189,9 +211,37 @@ const onConfirmedIntent = async (intentId: string) => {
     return err(new Error('Intent has no deposit amount'))
   }
 
+  // Guard: reject payments whose value is too small to purchase even a single
+  // byte of storage.  getIntentCredits divides paymentAmount by shannonsPerByte
+  // using BigInt integer division, so a dust payment (paymentAmount <
+  // shannonsPerByte) yields 0 credits.  Granting 0 credits would mark the
+  // intent COMPLETED while giving the user nothing — a misleading outcome that
+  // wastes a DB row and silently discards the payment.
+  //
+  // Both paymentAmount and shannonsPerByte are immutable on a confirmed intent,
+  // so this condition is permanent.  We mark the intent FAILED (terminal) so
+  // the polling loop stops retrying.  The on-chain payment is irreversible;
+  // resolution requires admin review (similar to OVER_CAP handling).
+  const creditBytes = IntentsUseCases.getIntentCredits(intent)
+  if (creditBytes === BigInt(0)) {
+    logger.warn(
+      'onConfirmedIntent: payment too small to yield any credits — marking FAILED',
+      {
+        intentId,
+        paymentAmount: intent.paymentAmount.toString(),
+        shannonsPerByte: intent.shannonsPerByte.toString(),
+      },
+    )
+    await intentsRepository.updateIntent({
+      ...intent,
+      status: IntentStatus.FAILED,
+    })
+    return ok()
+  }
+
   const addResult = await AccountsUseCases.addCreditsToAccount(
     intent.userPublicId,
-    IntentsUseCases.getIntentCredits(intent),
+    creditBytes,
     intentId,
   )
 
@@ -326,6 +376,13 @@ const getPrice = async (): Promise<{ price: number; pricePerGB: number }> => {
   }
 }
 
+// Returns PENDING intents that already have a tx_hash — used by the payment
+// manager startup sweep to re-watch transactions that were submitted but never
+// confirmed due to a service restart or RPC outage.
+const getPendingWithTxHash = async (): Promise<Intent[]> => {
+  return intentsRepository.getPendingWithTxHash()
+}
+
 export const IntentsUseCases = {
   createIntent,
   getIntent,
@@ -335,6 +392,7 @@ export const IntentsUseCases = {
   markIntentAsConfirmed,
   getConfirmedIntents,
   getOverCapIntents,
+  getPendingWithTxHash,
   reprocessOverCapIntent,
   getIntentCredits,
   getPrice,

apps/backend/src/infrastructure/repositories/users/intents.ts

@@ -117,6 +117,23 @@ const expireIntentIfPending = async (intentId: string): Promise<boolean> => {
   return (result.rowCount ?? 0) > 0
 }
 
+// Returns PENDING intents that already have an on-chain tx_hash.
+// These are intents where the user submitted a transaction but the payment
+// manager did not process the confirmation event — typically because the
+// service was restarted or the EVM RPC was temporarily unavailable.
+// Used by the startup recovery sweep so that no paid transaction is silently
+// abandoned across a service restart.
+const getPendingWithTxHash = async (): Promise<Intent[]> => {
+  const db = await getDatabase()
+  const result = await db.query<DBIntent>(
+    `SELECT * FROM intents
+     WHERE status = $1
+       AND tx_hash IS NOT NULL`,
+    [IntentStatus.PENDING],
+  )
+  return mapRows(result.rows)
+}
+
 // Returns all intents that were blocked by the per-user cap.
 // These are terminal — the polling loop skips them — and require admin review.
 const getOverCapIntents = async (): Promise<Intent[]> => {
@@ -136,4 +153,5 @@ export const intentsRepository = {
   getExpiredPendingIntents,
   expireIntentIfPending,
   getOverCapIntents,
+  getPendingWithTxHash,
 }

apps/backend/src/infrastructure/repositories/users/purchasedCredits.ts

@@ -547,6 +547,30 @@ const createPurchasedCreditWithCapCheck = async (
   }
 }
 
+// ---------------------------------------------------------------------------
+// getAllWithUserPublicId
+// Admin view: every credit batch across all users, joined with the
+// user_public_id from the originating intent row. Ordered newest-first.
+// ---------------------------------------------------------------------------
+
+export type AdminCreditBatchRow = PurchasedCredit & {
+  userPublicId: string
+}
+
+const getAllWithUserPublicId = async (): Promise<AdminCreditBatchRow[]> => {
+  const db = await getDatabase()
+  const result = await db.query<DBPurchasedCredit & { user_public_id: string }>(
+    `SELECT pc.*, i.user_public_id
+     FROM purchased_credits pc
+     JOIN intents i ON i.id = pc.intent_id
+     ORDER BY pc.purchased_at DESC`,
+  )
+  return result.rows.map((row) => ({
+    ...mapRow(row),
+    userPublicId: row.user_public_id,
+  }))
+}
+
 // ---------------------------------------------------------------------------
 // Public API
 // ---------------------------------------------------------------------------
@@ -563,4 +587,5 @@ export const purchasedCreditsRepository = {
   createPurchasedCreditWithCapCheck,
   markExpiredCredits,
   getByAccountId,
+  getAllWithUserPublicId,
 }

apps/backend/src/infrastructure/services/paymentManager/index.ts

@@ -122,8 +122,56 @@ const parseEventLogs = <
 let checkInterval: NodeJS.Timeout | null = null
 let unwatchContractEvent: (() => void) | null = null
 
+// On startup, re-watch any PENDING intents that already have a tx_hash.
+// These represent transactions submitted by users before the last service
+// restart or during an EVM RPC outage.  The cleanup job explicitly skips
+// PENDING+txHash rows (they are not abandoned — they are actively watched),
+// so without this sweep they would sit in limbo indefinitely: the user paid
+// on-chain but receives no credits.
+//
+// Re-calling watchTransaction for each orphan is safe:
+//   • waitForTransactionReceipt returns immediately for already-mined txs
+//   • markIntentAsConfirmed is idempotent — a duplicate CONFIRMED write is a
+//     no-op if the intent was already processed before the restart
+const _recoverOrphanedTransactions = async () => {
+  const pending = await IntentsUseCases.getPendingWithTxHash()
+  if (pending.length === 0) {
+    logger.info('Startup recovery: no orphaned transactions found')
+    return
+  }
+
+  logger.info('Startup recovery: re-watching orphaned transactions', {
+    count: pending.length,
+    intentIds: pending.map((i) => i.id),
+  })
+
+  await Promise.allSettled(
+    pending.map(async (intent) => {
+      if (!intent.txHash) return
+      try {
+        await paymentManager.watchTransaction(intent.txHash)
+        logger.info('Startup recovery: transaction recovered', {
+          intentId: intent.id,
+          txHash: intent.txHash,
+        })
+      } catch (err) {
+        logger.error('Startup recovery: failed to recover transaction', {
+          intentId: intent.id,
+          txHash: intent.txHash,
+          err,
+        })
+      }
+    }),
+  )
+}
+
 const start = () => {
   logger.info('Starting payment manager')
+
+  // Run the recovery sweep asynchronously so it does not block startup.
+  // Errors inside the sweep are caught per-intent and logged individually.
+  safeCallback(paymentManager._recoverOrphanedTransactions)()
+
   checkInterval = setInterval(
     safeCallback(paymentManager._checkConfirmedIntents),
     config.paymentManager.checkInterval,
@@ -154,6 +202,7 @@ export const paymentManager = {
   stop,
   _onLogs: onLogs,
   _checkConfirmedIntents: _checkConfirmedIntents,
+  _recoverOrphanedTransactions: _recoverOrphanedTransactions,
   _viemClient: viemClient,
   _parseEventLogs: parseEventLogs,
 }