feat(provider-b,sdk): automate session reconciliation via Cloudflare cron trigger (#209)

.changeset/tidy-cron-reconciliation.md

@@ -0,0 +1,5 @@
+---
+"@routedock/routedock": patch
+---
+
+Export SessionReconciler helpers and types from the SDK and provider hono entrypoints for automated session reconciliation.

apps/provider-b/src/ChannelSession.ts

@@ -5,6 +5,8 @@ import { createClient, type SupabaseClient } from '@supabase/supabase-js'
 import {
   routedockHono,
   mppSessionWsVerified,
+  reconcileAbandonedSessions,
+  type ReconciliationStats,
 } from '@routedock/routedock/provider/hono'
 import { Store } from '@stellar/mpp/channel/server'
 import {
@@ -272,10 +274,52 @@ export class ChannelSession extends DurableObject<Env> {
       }
     })
 
+    app.post('/__reconcile', async (c) => {
+      const stats = await this.reconcileSessions()
+      return c.json({ status: 'ok', stats })
+    })
+
     return app
   }
 
+  async reconcileSessions(): Promise<ReconciliationStats | null> {
+    const env = this.env
+    if (!env.SUPABASE_URL || !env.SUPABASE_SERVICE_KEY || !env.STELLAR_PAYEE_SECRET) {
+      return null
+    }
+
+    const supabase = createClient(env.SUPABASE_URL, env.SUPABASE_SERVICE_KEY)
+    const network: Network = env.STELLAR_NETWORK === 'mainnet' ? 'mainnet' : 'testnet'
+    const providerUrl = `${env.PUBLIC_BASE_URL ?? 'https://api-b.routedock.xyz'}/stream/orderbook`
+
+    return reconcileAbandonedSessions({
+      supabase,
+      network,
+      payeeSecretKey: env.STELLAR_PAYEE_SECRET,
+      onRecovered: async (channelId: string, txHash: string, totalPaid: string) => {
+        console.log(`[reconcile] Recovered channel ${channelId} with tx ${txHash} for ${totalPaid} USDC`)
+        const { error } = await supabase.from('tx_log').insert({
+          tx_type: 'channel_close',
+          tx_hash: txHash,
+          amount: parseFloat(totalPaid),
+          mode: 'mpp-session',
+          network,
+          provider_url: providerUrl,
+          metadata: { settled_at: new Date().toISOString(), recovered: true },
+        })
+        if (error) {
+          console.error('[supabase] tx_log insert on reconcile failed:', error.message)
+        }
+      },
+    })
+  }
+
   override async fetch(request: Request): Promise<Response> {
+    const url = new URL(request.url)
+    if (url.pathname === '/__reconcile' && request.method === 'POST') {
+      const stats = await this.reconcileSessions()
+      return Response.json({ status: 'ok', stats })
+    }
     this.app ??= this.buildApp()
     return this.app.fetch(request)
   }

apps/provider-b/src/__tests__/worker.test.ts

@@ -120,4 +120,47 @@ describe('provider-b worker payment path & Durable Object routes', () => {
     const body2 = (await res2.json()) as { payee?: string }
     assert.equal(body2.payee, mockEnv.STELLAR_PAYEE_ADDRESS)
   })
+
+  it('scheduled() cron handler forwards trigger to ChannelSession Durable Object', async () => {
+    let reconcileInvoked = false
+    const mockDOBinding = {
+      idFromName(name: string) {
+        return { name }
+      },
+      get(_id: unknown) {
+        return {
+          reconcileSessions: async () => {
+            reconcileInvoked = true
+            return {
+              orphanedCount: 1,
+              recoveredCount: 1,
+              skippedCount: 0,
+              failedCount: 0,
+              errors: [],
+            }
+          },
+          fetch: async () => new Response('ok'),
+        }
+      },
+    }
+
+    const envWithDO: Env = {
+      ...mockEnv,
+      CHANNEL_SESSION: mockDOBinding as unknown as Env['CHANNEL_SESSION'],
+    }
+
+    await worker.scheduled({}, envWithDO)
+    assert.equal(reconcileInvoked, true, 'scheduled() must trigger reconciliation on DO')
+  })
+
+  it('ChannelSession responds to internal POST /__reconcile endpoint', async () => {
+    const session = createTestChannelSession(mockEnv)
+    const req = new Request('http://localhost/__reconcile', { method: 'POST' })
+    const res = await session.fetch(req)
+
+    assert.equal(res.status, 200)
+    const body = (await res.json()) as { status: string; stats: unknown }
+    assert.equal(body.status, 'ok')
+  })
 })
+

apps/provider-b/src/worker.ts

@@ -32,4 +32,28 @@ export default {
     const id = env.CHANNEL_SESSION.idFromName(env.CHANNEL_CONTRACT_ID)
     return env.CHANNEL_SESSION.get(id).fetch(request)
   },
+
+  /**
+   * Cron trigger entry point (runs every 15 minutes).
+   * Forwards reconciliation to the ChannelSession Durable Object so settlement
+   * serializes against live voucher traffic instead of racing it.
+   */
+  async scheduled(
+    _controller: unknown,
+    env: Env,
+    _ctx?: unknown,
+  ): Promise<void> {
+    if (!env.CHANNEL_CONTRACT_ID || !env.CHANNEL_SESSION) return
+    const id = env.CHANNEL_SESSION.idFromName(env.CHANNEL_CONTRACT_ID)
+    const stub = env.CHANNEL_SESSION.get(id) as unknown as {
+      reconcileSessions?: () => Promise<unknown>
+      fetch: (req: Request) => Promise<Response>
+    }
+    if (typeof stub.reconcileSessions === 'function') {
+      await stub.reconcileSessions()
+    } else {
+      await stub.fetch(new Request('http://localhost/__reconcile', { method: 'POST' }))
+    }
+  },
 }
+

apps/provider-b/wrangler.jsonc

@@ -13,6 +13,9 @@
       "custom_domain": true
     }
   ],
+  "triggers": {
+    "crons": ["*/15 * * * *"]
+  },
   "durable_objects": {
     "bindings": [
       {

packages/sdk/src/index.ts

@@ -12,5 +12,6 @@ export * from './client/NulthVault.js'
 export * from './store/SessionStore.js'
 export * from './store/SpendStore.js'
 export * from './provider/routedockMiddleware.js'
+export * from './provider/SessionReconciler.js'
 export * from './registry/index.js'
 export { signManifest, verifyManifestSignature, manifestDigest } from './manifest/sign.js'

packages/sdk/src/provider/__tests__/SessionReconciler.test.ts

@@ -92,3 +92,63 @@ test('reconcileAbandonedSessions records structured close failures readably', as
 
   assert.equal(stats.errors[0]?.reason, '{"code":"scecInvalidAction","status":"FAILED"}')
 })
+
+test('reconcileAbandonedSessions recovers closing session to closed with settlement hash', async () => {
+  const payeeKeypair = Keypair.random()
+  let updatedPayload: any = null
+  let updatedChannelId: string | null = null
+  let onRecoveredCalled = false
+
+  const mockSupabase = {
+    from: (_table: string) => ({
+      select: (_cols: string) => ({
+        eq: (_field: string, _val: string) => ({
+          is: (_field2: string, _val2: any) => ({
+            limit: async (_limit: number) => ({
+              data: [
+                {
+                  channel_id: 'CCK4XOW3YKQUEZFONUTINKMSNW7SNMRQZURME5U3UP7E6WNGK7UHUCAH',
+                  channel_contract: 'CCK4XOW3YKQUEZFONUTINKMSNW7SNMRQZURME5U3UP7E6WNGK7UHUCAH',
+                  cumulative_amount: '0.0050000',
+                  last_signature: '00'.repeat(64),
+                  settlement_tx_hash: null,
+                },
+              ],
+              error: null,
+            }),
+          }),
+        }),
+      }),
+      update: (data: any) => ({
+        eq: (_field: string, val: string) => {
+          updatedPayload = data
+          updatedChannelId = val
+          return Promise.resolve({ error: null })
+        },
+      }),
+    }),
+  } as unknown as SupabaseClient
+
+  const stats = await reconcileAbandonedSessions({
+    supabase: mockSupabase,
+    network: 'testnet',
+    payeeSecretKey: payeeKeypair.secret(),
+    channelClose: async () => 'test_settlement_tx_hash_123',
+    onRecovered: async (channelId, txHash, totalPaid) => {
+      onRecoveredCalled = true
+      assert.equal(channelId, 'CCK4XOW3YKQUEZFONUTINKMSNW7SNMRQZURME5U3UP7E6WNGK7UHUCAH')
+      assert.equal(txHash, 'test_settlement_tx_hash_123')
+      assert.equal(totalPaid, '0.0050000')
+    },
+  })
+
+  assert.equal(stats.orphanedCount, 1)
+  assert.equal(stats.recoveredCount, 1)
+  assert.equal(stats.failedCount, 0)
+  assert.equal(stats.skippedCount, 0)
+  assert.equal(updatedChannelId, 'CCK4XOW3YKQUEZFONUTINKMSNW7SNMRQZURME5U3UP7E6WNGK7UHUCAH')
+  assert.equal(updatedPayload?.status, 'closed')
+  assert.equal(updatedPayload?.settlement_tx_hash, 'test_settlement_tx_hash_123')
+  assert.equal(onRecoveredCalled, true)
+})
+

packages/sdk/src/provider/hono.ts

@@ -783,3 +783,11 @@ export function routedockHono(opts: RouteDockHonoOptions): MiddlewareHandler {
     return handler(c, next)
   }
 }
+
+export {
+  reconcileAbandonedSessions,
+  runStartupReconciliation,
+  type SessionReconcilerOptions,
+  type ReconciliationStats,
+} from './SessionReconciler.js'
+export { signManifest } from '../manifest/sign.js'