diff --git a/.github/workflows/keeper.yml b/.github/workflows/keeper.yml index 1bf0537a9..17d51b84b 100644 --- a/.github/workflows/keeper.yml +++ b/.github/workflows/keeper.yml @@ -1,16 +1,18 @@ name: keeper -# InvoFi keeper automation (Task 12). -# Runs every 6 hours against the live testnet deployment: +# InvoFi keeper automation (Task 12 & Event-driven upgrade). +# Runs against the live testnet deployment: +# - reacts to inv_reg / off_acc contract events # - marks Financed invoices overdue once their due_date passes # - best-effort TTL extension for active invoice storage +# - retains full-sweep fallback # See invofi/scripts/keeper.ts for details and env var docs. # # Required repo configuration (once): # Secret: KEEPER_SECRET_KEY — secret key of a funded testnet account that # pays transaction fees. Ask in the repo's # issues if you need a funded key. -# Variable: REGISTRY_CONTRACT_ID, REPAYMENT_CONTRACT_ID — override the +# Variable: REGISTRY_CONTRACT_ID, REPAYMENT_CONTRACT_ID, FINANCING_CONTRACT_ID — override the # defaults below if you deployed your own contract set. on: @@ -39,12 +41,32 @@ jobs: cache: npm cache-dependency-path: invofi/scripts/package-lock.json - run: npm ci + - run: npm test - run: npm run type-check + - name: Download checkpoint artifact + uses: actions/download-artifact@fa0a91b85d4f404e444e00e005971372dc801d16 # v4 + with: + name: keeper-checkpoint + path: invofi/scripts + continue-on-error: true - name: Run keeper + # Note: continuous long-running daemon (KEEPER_MODE=event-driven) is intended + # for persistent container/VPS hosting. In scheduled GitHub Actions runs, we + # execute KEEPER_MODE=event-catchup to process recent contract events before the sweep. env: + KEEPER_MODE: event-catchup RPC_URL: https://soroban-testnet.stellar.org NETWORK_PASSPHRASE: Test SDF Network ; September 2015 REGISTRY_CONTRACT_ID: ${{ vars.REGISTRY_CONTRACT_ID || 'CAXNTWSKDVSB3GPJMU3RTSDTAIFF4A6FFRAAI35B4AE7LZLLI4VXMCF7' }} REPAYMENT_CONTRACT_ID: ${{ vars.REPAYMENT_CONTRACT_ID || 'CCDATW5GMVDOPK55Q4MLXV5SGA3VLXPD67ABLBNMHWFF6BLL2IZBUVEP' }} + FINANCING_CONTRACT_ID: ${{ vars.FINANCING_CONTRACT_ID || 'CBGRA3457ZFXYZNEQLO4YGUQ3OBEWOE6US6ZREHK6NF2DLZYBO73IFVW' }} KEEPER_SECRET_KEY: ${{ secrets.KEEPER_SECRET_KEY }} run: npm run keeper + - name: Upload checkpoint artifact + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: keeper-checkpoint + path: invofi/scripts/.keeper-checkpoint.json + overwrite: true + continue-on-error: true diff --git a/README.md b/README.md index fbd0f6f3a..8f5e4ce7a 100644 --- a/README.md +++ b/README.md @@ -62,8 +62,8 @@ InvoFi lives across **two repositories**, split so the fast-moving app layer and > - reputation: [`CCHKVUWGTQ56U53C5U7ZSOFDTTMGLMOFCL22DME5UMXIYWQNUYXOYPDN`](https://stellar.expert/explorer/testnet/contract/CCHKVUWGTQ56U53C5U7ZSOFDTTMGLMOFCL22DME5UMXIYWQNUYXOYPDN) > - position token: `POS` minted to lenders on acceptance ([`CBIXYAJPEOOVIALBUTA7X2H26WXSI5JDZCTE23RUMQR4QFJNMPL6767Z`](https://stellar.expert/explorer/testnet/contract/CBIXYAJPEOOVIALBUTA7X2H26WXSI5JDZCTE23RUMQR4QFJNMPL6767Z)) > -> A keeper automation (6-hourly GitHub Action) scans testnet, bumps contract-data TTLs, -> and marks past-due Financed invoices Overdue — see `invofi/scripts/keeper.ts`. +> A keeper automation (event-driven Soroban RPC getEvents polling for `inv_reg`/`off_acc` + 6-hourly fallback sweep) +> bumps contract-data TTLs and marks past-due Financed invoices Overdue — see `invofi/scripts/keeper.ts`. > > Deploy your own via the **Deploy Contracts to Testnet** workflow in [invofi-contracts](https://github.com/Stellar-VaultLink/invofi-contracts) and set the three `NEXT_PUBLIC_*_CONTRACT_ID` variables in Vercel. Without a contract configured the app runs in alpha mode (off-chain only). @@ -151,7 +151,7 @@ npm install && npm run dev │ reads protocol_stats │ │ checkpointed event replay → protocol_stats │ └──────────────────────────┘ └──────────────────────────────────────────────┘ -keeper (T12) — 6-hourly GitHub Action: mark_overdue + TTL bumps (Soroban RPC) +keeper (T12) — Event-driven RPC subscriptions (`inv_reg`/`off_acc`) + 6-hourly fallback sweep: mark_overdue + TTL bumps ``` No always-on backend server to manage. 100% free hosting. @@ -530,7 +530,7 @@ Both identities are auto-funded via Friendbot on testnet. See - [x] Marketplace sorting (newest, amount, due date) and Stellar Expert explorer links - [x] Insurance coverage pool with **payout on default** - [x] On-chain **reputation scoring** for originators -- [x] Keeper automation — 6-hourly TTL bump + overdue marking +- [x] Keeper automation — event-driven Soroban RPC subscriptions (`inv_reg`, `off_acc`) + 6-hourly fallback sweep - [x] SEP-41 token movement — `accept_offer` funds the business, `repay_invoice` repays principal + yield - [x] Split into 5 auditable contract crates — registry / financing / repayment / insurance / reputation - [x] Emergency pause / circuit breaker — admin-gated `pause` on every state-mutating function diff --git a/docs/10-roadmap.md b/docs/10-roadmap.md index 843580652..88754949f 100644 --- a/docs/10-roadmap.md +++ b/docs/10-roadmap.md @@ -36,7 +36,7 @@ Last updated: August 2026. Checkbox status reflects what is merged to `main` ### Infrastructure & Automation - [x] Frontend testnet config — registry/financing/repayment contract IDs (`NEXT_PUBLIC_{REGISTRY,FINANCING,REPAYMENT}_CONTRACT_ID`) -- [x] Keeper automation — 6-hourly overdue marking + TTL bumps +- [x] Keeper automation — event-driven Soroban RPC getEvents polling (`inv_reg`, `off_acc`) + 6-hourly fallback sweep - [x] Event indexer — checkpointed replay → `protocol_stats`; **temporarily bypassed pending the Supabase/DB migration** — re-enabled when the indexer is rewired to the new database - [x] Contributors auto-table on merge (no opt-in comment needed), bot-driven PRs, issues open to all - [x] One-click Testnet deploy via GitHub Actions (invofi-contracts) @@ -59,5 +59,4 @@ Last updated: August 2026. Checkbox status reflects what is merged to `main` - [ ] Lender verification (threshold-based SEP-12 onboarding, Phase 4 of compliance.md) - [ ] On-chain (event-sourced) position listings — the alternative deferred in ADR-0004, if listings ever need to be readable without InvoFi's frontend -- [ ] Event-driven keeper (Soroban RPC event subscriptions instead of polling) - [ ] Historical time-series charts on `/stats` diff --git a/docs/adr/0005-event-driven-keeper.md b/docs/adr/0005-event-driven-keeper.md new file mode 100644 index 000000000..68e09b273 --- /dev/null +++ b/docs/adr/0005-event-driven-keeper.md @@ -0,0 +1,36 @@ +# ADR-0005: Event-driven keeper upgrade (RPC event subscriptions) + +**Status:** Accepted (2026-08-18) + +## Context + +The InvoFi keeper (`invofi/scripts/keeper.ts`) performs two critical protocol maintenance tasks: +1. `mark_overdue`: Calling `repayment.mark_overdue` on past-due Financed invoices. +2. `bump_ttl`: Extending contract storage footprint TTL for active invoices so state entries never expire on Soroban testnet/mainnet. + +Previously (Task 12), the keeper operated exclusively as a 6-hourly batch job via GitHub Actions. As invoice volume increases, scanning all pages of invoices on every run introduces up to a 6-hour delay before past-due invoices are marked overdue or newly registered/financed invoices receive TTL bumps. + +## Decision + +1. **Soroban RPC Event Subscriptions (`getEvents`)**: + Upgrade the keeper to run in an event-driven mode (`KEEPER_MODE=event-driven` or `--mode=event-driven`). The keeper continuously polls Soroban RPC `getEvents` with ledger cursor tracking for contract events emitted by `invofi-registry` and `invofi-financing`. + +2. **Targeted Event Handlers**: + - **`inv_reg`** (Invoice Registered): When a new invoice is created on-chain, the keeper instantly receives the event and performs an immediate best-effort TTL bump (`bumpTtl(invoiceId)`). + - **`off_acc`** (Offer Accepted -> Financed): When an offer is accepted and an invoice transitions to `Financed`, the keeper instantly performs a TTL bump, checks if `due_date < now`, and calls `repayment.mark_overdue` immediately if past-due. + - **`off_def`** (Offer Defaulted): Default events published by repayment contract are recognized by event parsing. + +3. **Polling Fallback Retained**: + The full-sweep paginated invoice scan is retained both: + - As a periodic background fallback loop (defaulting to every 6 hours) in continuous daemon mode. + - As a one-shot execution mode (`KEEPER_MODE=sweep`, default fallback) for scheduled cron jobs. + This provides bounded recovery based on the configured fallback sweep interval (default 6h) against network partitions, process restarts, or missing events beyond RPC retention limits. + +4. **Ledger Cursor Checkpointing**: + In event-driven mode, the keeper maintains a disk checkpoint (`.keeper-checkpoint.json`) of the last processed ledger sequence, allowing seamless catch-up after restarts without missed events. + +## Consequences + +- Reaction latency for newly registered or financed invoices drops from hours (up to 6h) to under a minute (~10s ledger poll). +- Incremental event processing provides rapid reaction time while the full sweep runs only as a periodic fallback. +- The keeper remains backward compatible with existing 6-hourly GitHub Actions cron jobs (`npm run keeper` in `sweep` mode). diff --git a/invofi/scripts/keeper.test.ts b/invofi/scripts/keeper.test.ts new file mode 100644 index 000000000..74d65a64a --- /dev/null +++ b/invofi/scripts/keeper.test.ts @@ -0,0 +1,163 @@ +process.env.NODE_ENV = 'test'; + +import assert from 'node:assert/strict'; +import test, { describe } from 'node:test'; +import { + parseRawEvent, + statusNum, + parseKeeperMode, + parseStartLedger, + STATUS, + processEvents, +} from './keeper.js'; +import { nativeToScVal } from '@stellar/stellar-sdk'; + +describe('Keeper Unit Tests', () => { + test('statusNum parses status variants correctly', () => { + assert.equal(statusNum('Pending'), STATUS.Pending); + assert.equal(statusNum('Financed'), STATUS.Financed); + assert.equal(statusNum(1), STATUS.Financed); + assert.equal(statusNum('3'), STATUS.Overdue); + assert.equal(statusNum('InvalidStatus'), -1); + }); + + test('parseRawEvent correctly decodes inv_reg event', () => { + const topic0 = nativeToScVal('inv_reg', { type: 'symbol' }); + const topic1 = nativeToScVal('INV-101', { type: 'symbol' }); + const value = nativeToScVal(['GABC...', 5000n, 1700000000n]); + + const rawEvent = { + type: 'contract', + contractId: 'CCREGISTRY...', + topic: [topic0, topic1], + value, + ledger: 12345, + ledgerClosedAt: '2026-08-18T10:00:00Z', + id: 'evt-1', + pagingToken: 'pt-1', + inSuccessfulContractCall: true, + txHash: 'hash-1', + } as any; + + const parsed = parseRawEvent(rawEvent); + assert.notEqual(parsed, null); + assert.equal(parsed?.type, 'inv_reg'); + assert.equal(parsed?.invoiceId, 'INV-101'); + assert.equal(parsed?.ledger, 12345); + }); + + test('parseRawEvent correctly decodes off_acc event', () => { + const topic0 = nativeToScVal('off_acc', { type: 'symbol' }); + const topic1 = nativeToScVal('INV-303', { type: 'symbol' }); + const value = nativeToScVal(['INV-303', 'GLENDER...', 10000n]); + + const rawEvent = { + type: 'contract', + contractId: 'CCFINANCING...', + topic: [topic0, topic1], + value, + ledger: 12346, + ledgerClosedAt: '2026-08-18T10:00:05Z', + id: 'evt-2', + pagingToken: 'pt-2', + inSuccessfulContractCall: true, + txHash: 'hash-2', + } as any; + + const parsed = parseRawEvent(rawEvent); + assert.notEqual(parsed, null); + assert.equal(parsed?.type, 'off_acc'); + assert.equal(parsed?.invoiceId, 'INV-303'); + assert.equal(parsed?.ledger, 12346); + }); + + test('parseRawEvent returns null on decoder failure paths', () => { + const validTopic0 = nativeToScVal('inv_reg', { type: 'symbol' }); + const validTopic1 = nativeToScVal('INV-100', { type: 'symbol' }); + const validValue = nativeToScVal(['GABC...', 5000n]); + + // 1. Unrecognized event name + const unrecTopic0 = nativeToScVal('unknown_event', { type: 'symbol' }); + assert.equal( + parseRawEvent({ topic: [unrecTopic0, validTopic1], value: validValue } as any), + null, + ); + + // 2. Missing topic 1 + assert.equal( + parseRawEvent({ topic: [validTopic0], value: validValue } as any), + null, + ); + + // 3. Non-string topic 1 (e.g. u32 ScVal) + const intTopic1 = nativeToScVal(9999, { type: 'u32' }); + assert.equal( + parseRawEvent({ topic: [validTopic0, intTopic1], value: validValue } as any), + null, + ); + + // 4. Undecodable value + const badValue = { _switch: { value: -9999 } } as any; + assert.equal( + parseRawEvent({ topic: [validTopic0, validTopic1], value: badValue } as any), + null, + ); + }); + + test('processEvents with empty event list preserves counter contract', async () => { + const dummyKp = { publicKey: () => 'GBDUMMY...' } as any; + const result = await processEvents([], dummyKp); + assert.deepEqual(result, { processed: 0, ttlBumps: 0, markedOverdue: 0 }); + }); + + test('parseKeeperMode handles CLI flags, env vars, and default fallbacks', () => { + const originalArgv = process.argv; + const originalEnvMode = process.env.KEEPER_MODE; + + try { + // Clear inputs -> default fallback 'sweep' + process.argv = ['node', 'keeper.js']; + delete process.env.KEEPER_MODE; + assert.equal(parseKeeperMode(), 'sweep'); + + // ENV var fallback + process.env.KEEPER_MODE = 'event-driven'; + assert.equal(parseKeeperMode(), 'event-driven'); + + // CLI flag overrides ENV var + process.argv = ['node', 'keeper.js', '--mode=event-catchup']; + assert.equal(parseKeeperMode(), 'event-catchup'); + } finally { + process.argv = originalArgv; + if (originalEnvMode !== undefined) { + process.env.KEEPER_MODE = originalEnvMode; + } else { + delete process.env.KEEPER_MODE; + } + } + }); + + test('parseStartLedger handles CLI flags, env vars, and default fallbacks', () => { + const originalArgv = process.argv; + const originalEnvStart = process.env.KEEPER_START_LEDGER; + + try { + process.argv = ['node', 'keeper.js']; + delete process.env.KEEPER_START_LEDGER; + assert.equal(parseStartLedger(), undefined); + + process.env.KEEPER_START_LEDGER = '50000'; + assert.equal(parseStartLedger(), 50000); + + process.argv = ['node', 'keeper.js', '--start-ledger=60000']; + assert.equal(parseStartLedger(), 60000); + } finally { + process.argv = originalArgv; + if (originalEnvStart !== undefined) { + process.env.KEEPER_START_LEDGER = originalEnvStart; + } else { + delete process.env.KEEPER_START_LEDGER; + } + } + }); +}); diff --git a/invofi/scripts/keeper.ts b/invofi/scripts/keeper.ts index 2e102fc5b..c2dced591 100644 --- a/invofi/scripts/keeper.ts +++ b/invofi/scripts/keeper.ts @@ -1,38 +1,41 @@ #!/usr/bin/env tsx /** - * InvoFi Keeper (Task 12) - * ======================= - * Off-chain automation that runs on a schedule (GitHub Actions cron, every - * 6h — see .github/workflows/keeper.yml). It does two jobs: + * InvoFi Keeper (Task 12 & Event-Driven Upgrade) + * ============================================= + * Off-chain automation that maintains protocol health on Soroban: * * 1. mark_overdue — for every Financed invoice whose due_date has passed, * call repayment.mark_overdue (a public, permissionless transition). * 2. bump_ttl — best-effort TTL extension for the storage of active * invoices, so contract state never expires on the Soroban network. * - * Design notes (scalability-first, per Task 4's guidance): - * - Invoice discovery is PAGINATED from the start via - * registry.get_invoices_paginated(offset, limit) — bounded pages, never - * an unbounded on-chain list read. This stays cheap as the invoice count - * grows (Task 13's indexer will eventually supersede on-chain discovery). - * - TTL bumps are capped per run (MAX_TTL_BUMPS) and best-effort: a failure - * logs and continues. Polling is fine at this scale; the event-driven - * upgrade path (Soroban RPC event subscriptions) is a documented - * follow-up, not a blocker. + * Operating Modes: + * - `event-driven`: (Daemon mode) Continuously subscribes to Soroban RPC events + * (`inv_reg`, `off_acc`) via cursor polling every 10s. Triggers immediate + * TTL bumps on `inv_reg` and immediate overdue checks + TTL bumps on `off_acc`. + * Runs a full paginated sweep every 6 hours as a fallback. + * - `event-catchup`: Single-pass execution that queries RPC events since the + * last checkpoint/start ledger to handle recent events, then finishes. + * - `sweep`: (Default / Classic fallback) Single-pass paginated sweep over all + * invoices on the registry contract (ideal for 6-hourly cron jobs). * - * Env vars: + * Env vars / CLI flags: + * KEEPER_MODE / --mode Operating mode: 'event-driven' | 'event-catchup' | 'sweep' (default: 'sweep') + * KEEPER_START_LEDGER / --start-ledger Target starting ledger for event mode (optional) * RPC_URL Soroban RPC endpoint (default: soroban-testnet) * NETWORK_PASSPHRASE network passphrase (default: testnet) * REGISTRY_CONTRACT_ID registry contract (required) * REPAYMENT_CONTRACT_ID repayment contract (required) + * FINANCING_CONTRACT_ID financing contract (required for event-driven) * KEEPER_SECRET_KEY secret key of the funded keeper account (required) - * PAGE_SIZE invoices per page (default 50) - * MAX_TTL_BUMPS max TTL extensions per run (default 50) + * PAGE_SIZE invoices per page for sweep (default 50) + * MAX_TTL_BUMPS max TTL extensions per sweep (default 50) * TTL_EXTEND_LEDGERS how many ledgers to extend TTL to (default ~30 days) - * - * On testnet the keeper account is auto-funded via Friendbot if missing. + * EVENT_POLL_INTERVAL_MS polling interval for event listener (default 10000ms = 10s) + * FALLBACK_SWEEP_INTERVAL_MS interval for full sweep fallback in daemon (default 6h) */ +import fs from 'fs'; import { Contract, Keypair, @@ -49,24 +52,27 @@ import { // ── Config ─────────────────────────────────────────────────────────────────── -const RPC_URL = process.env.RPC_URL ?? 'https://soroban-testnet.stellar.org'; -const NETWORK_PASSPHRASE = +export const RPC_URL = process.env.RPC_URL ?? 'https://soroban-testnet.stellar.org'; +export const NETWORK_PASSPHRASE = process.env.NETWORK_PASSPHRASE ?? Networks.TESTNET; -const REGISTRY_ID = process.env.REGISTRY_CONTRACT_ID; -const REPAYMENT_ID = process.env.REPAYMENT_CONTRACT_ID; -const KEEPER_SECRET_KEY = process.env.KEEPER_SECRET_KEY; -const PAGE_SIZE = Number(process.env.PAGE_SIZE ?? 50); -const MAX_TTL_BUMPS = Number(process.env.MAX_TTL_BUMPS ?? 50); -const TTL_EXTEND_LEDGERS = Number(process.env.TTL_EXTEND_LEDGERS ?? 311_040); // ~30 days -const FEE = '100'; -const MAX_PAGES = 2_000; // safety cap: 2_000 pages x PAGE_SIZE invoices max - -const rpc = new SorobanRpc.Server(RPC_URL, { allowHttp: false }); - -// Contract enum discriminants (must match invofi-common): -// InvoiceStatus: Pending=0, Financed=1, Repaid=2, Overdue=3, Cancelled=4, -// Disputed=5, Defaulted=6 -const STATUS: Record = { +export const REGISTRY_ID = process.env.REGISTRY_CONTRACT_ID; +export const REPAYMENT_ID = process.env.REPAYMENT_CONTRACT_ID; +export const FINANCING_ID = process.env.FINANCING_CONTRACT_ID; +export const KEEPER_SECRET_KEY = process.env.KEEPER_SECRET_KEY; +export const PAGE_SIZE = Number(process.env.PAGE_SIZE ?? 50); +export const MAX_TTL_BUMPS = Number(process.env.MAX_TTL_BUMPS ?? 50); +export const TTL_EXTEND_LEDGERS = Number(process.env.TTL_EXTEND_LEDGERS ?? 311_040); // ~30 days +export const EVENT_POLL_INTERVAL_MS = Number(process.env.EVENT_POLL_INTERVAL_MS ?? 10_000); // 10s +export const FALLBACK_SWEEP_INTERVAL_MS = Number( + process.env.FALLBACK_SWEEP_INTERVAL_MS ?? 6 * 3600 * 1000, +); // 6 hours +export const CHECKPOINT_FILE = process.env.CHECKPOINT_FILE ?? '.keeper-checkpoint.json'; +export const FEE = '100'; +export const MAX_PAGES = 2_000; + +export const rpc = new SorobanRpc.Server(RPC_URL, { allowHttp: false }); + +export const STATUS: Record = { Pending: 0, Financed: 1, Repaid: 2, @@ -76,26 +82,28 @@ const STATUS: Record = { Defaulted: 6, }; +export type KeeperMode = 'event-driven' | 'event-catchup' | 'sweep'; + // ── Helpers ────────────────────────────────────────────────────────────────── -function log(msg: string): void { +export function log(msg: string): void { console.log(`[${new Date().toISOString()}] ${msg}`); } -function sleep(ms: number): Promise { +export function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function encodeSymbol(value: string): xdr.ScVal { +export function encodeSymbol(value: string): xdr.ScVal { return nativeToScVal(value, { type: 'symbol' }); } -function encodeU32(value: number): xdr.ScVal { +export function encodeU32(value: number): xdr.ScVal { return nativeToScVal(value, { type: 'u32' }); } /** Defensively parse the serialized status (number | numeric string | name). */ -function statusNum(value: unknown): number { +export function statusNum(value: unknown): number { if (typeof value === 'number') return value; const s = String(value); if (s in STATUS) return STATUS[s]; @@ -103,12 +111,12 @@ function statusNum(value: unknown): number { return Number.isNaN(n) ? -1 : n; } -async function getAccount(pub: string) { +export async function getAccount(pub: string) { return rpc.getAccount(pub); } /** Fund the keeper account on testnet if it doesn't exist yet. */ -async function ensureAccount(pub: string): Promise { +export async function ensureAccount(pub: string): Promise { try { await rpc.getAccount(pub); return; @@ -126,7 +134,6 @@ async function ensureAccount(pub: string): Promise { if (!res.ok) { throw new Error(`Friendbot funding failed (${res.status}): ${await res.text()}`); } - // Friendbot can take a couple of ledgers to apply. for (let i = 0; i < 10; i++) { await sleep(1_000); try { @@ -139,7 +146,7 @@ async function ensureAccount(pub: string): Promise { throw new Error('Friendbot funded but account still not visible after 10s.'); } -async function sendAndConfirm(tx: Transaction): Promise { +export async function sendAndConfirm(tx: Transaction): Promise { const resp = await rpc.sendTransaction(tx); if (resp.status === 'ERROR') { log(` send ERROR: ${resp.errorResult?.result().toXDR('base64')}`); @@ -160,7 +167,7 @@ async function sendAndConfirm(tx: Transaction): Promise { } /** Read one page of invoices (bounded — the scalability-safe query). */ -async function readInvoicePage(offset: number, pub: string): Promise { +export async function readInvoicePage(offset: number, pub: string): Promise { const contract = new Contract(REGISTRY_ID!); const account = await rpc.getAccount(pub); const tx = new TransactionBuilder(account, { @@ -180,8 +187,33 @@ async function readInvoicePage(offset: number, pub: string): Promise return Array.isArray(parsed) ? parsed : []; } +/** Fetch a single invoice from registry contract to inspect status & due_date. */ +export async function fetchInvoiceDetails(invoiceId: string, pub: string): Promise | null> { + try { + const contract = new Contract(REGISTRY_ID!); + const account = await rpc.getAccount(pub); + const tx = new TransactionBuilder(account, { + fee: FEE, + networkPassphrase: NETWORK_PASSPHRASE, + }) + .addOperation(contract.call('get_invoice', encodeSymbol(invoiceId))) + .setTimeout(30) + .build(); + const sim = await rpc.simulateTransaction(tx); + if (SorobanRpc.Api.isSimulationError(sim)) { + return null; + } + const retval = sim.result?.retval; + if (!retval) return null; + const parsed = scValToNative(retval); + return parsed && typeof parsed === 'object' ? (parsed as Record) : null; + } catch { + return null; + } +} + /** Mark a single invoice overdue (public transition). Returns true on success. */ -async function markOverdue(invoiceId: string, kp: Keypair): Promise { +export async function markOverdue(invoiceId: string, kp: Keypair): Promise { const account = await getAccount(kp.publicKey()); const contract = new Contract(REPAYMENT_ID!); let tx = new TransactionBuilder(account, { @@ -203,24 +235,12 @@ async function markOverdue(invoiceId: string, kp: Keypair): Promise { /** * Best-effort TTL extension: probe a read of the invoice to learn the - * storage footprint, then submit an extendFootprintTtl transaction covering - * exactly that footprint (contract code + instance + invoice entry). - * - * Follows the Stellar docs pattern (extend-persistent-entry-js): fresh - * SorobanDataBuilder with only the read-only footprint keys, a DELTA - * `extendTo` (ledgers past LCL), then `server.prepareTransaction` to fill in - * real resources + fee. Verified against soroban-testnet: the earlier naive - * approaches (absolute extendTo, reusing the probe's SorobanTransactionData, - * assembleTransaction) all failed with txBadSeq / extendFootprintTtlMalformed. + * storage footprint, then submit an extendFootprintTtl transaction. */ -async function bumpTtl(invoiceId: string, kp: Keypair): Promise { +export async function bumpTtl(invoiceId: string, kp: Keypair): Promise { try { const contract = new Contract(REGISTRY_ID!); - // 1. Probe read — the simulated footprint is what we want to extend. - // NOTE: TransactionBuilder consumes + increments the account's sequence on - // every build, so each builder gets a FRESH account fetch (reuse would - // shift the sequence and cause txBadSeq). const probe = new TransactionBuilder(await getAccount(kp.publicKey()), { fee: FEE, networkPassphrase: NETWORK_PASSPHRASE, @@ -235,8 +255,6 @@ async function bumpTtl(invoiceId: string, kp: Keypair): Promise { } const readOnlyKeys = sim.transactionData.build().resources().footprint().readOnly(); - // 2. extendFootprintTtl with the probed footprint, extendTo = ledgers past - // LCL (a DELTA, per the Stellar docs — not an absolute ledger number). let bump = new TransactionBuilder(await getAccount(kp.publicKey()), { fee: FEE, networkPassphrase: NETWORK_PASSPHRASE, @@ -246,7 +264,6 @@ async function bumpTtl(invoiceId: string, kp: Keypair): Promise { .setTimeout(30) .build(); - // 3. prepareTransaction simulates and fills in the real resources + fee. bump = await rpc.prepareTransaction(bump); bump.sign(kp); return sendAndConfirm(bump); @@ -256,21 +273,137 @@ async function bumpTtl(invoiceId: string, kp: Keypair): Promise { } } -// ── Main ───────────────────────────────────────────────────────────────────── +// ── Event Processing Core ─────────────────────────────────────────────────── -async function main(): Promise { - if (!REGISTRY_ID || !REPAYMENT_ID) { - throw new Error('REGISTRY_CONTRACT_ID and REPAYMENT_CONTRACT_ID are required.'); +export interface ParsedEvent { + type: string; + invoiceId: string; + ledger: number; + data: Record; +} + +export function parseRawEvent(rawEvent: SorobanRpc.Api.EventResponse): ParsedEvent | null { + try { + const topic0 = rawEvent.topic?.[0]; + if (!topic0) return null; + const name = scValToNative(topic0); + if (typeof name !== 'string') return null; + + const topic1 = rawEvent.topic?.[1]; + if (!topic1) return null; + const topic1Dec = scValToNative(topic1); + if (typeof topic1Dec !== 'string' || !topic1Dec) return null; + const invoiceId = topic1Dec; + + let valDec: unknown; + valDec = scValToNative(rawEvent.value); + + const arr = Array.isArray(valDec) ? valDec : []; + const map = typeof valDec === 'object' && valDec !== null && !Array.isArray(valDec) + ? (valDec as Record) + : {}; + + if (name === 'inv_reg') { + return { + type: 'inv_reg', + invoiceId, + ledger: rawEvent.ledger, + data: { valDec }, + }; + } + + if (name === 'off_acc') { + return { + type: 'off_acc', + invoiceId, + ledger: rawEvent.ledger, + data: { + invoiceId, + lender: arr[1] ?? map['lender'], + amount: arr[2] ?? map['amount'], + }, + }; + } + + if (name === 'off_def') { + return { + type: 'off_def', + invoiceId, + ledger: rawEvent.ledger, + data: { valDec }, + }; + } + + return null; + } catch { + return null; } - if (!KEEPER_SECRET_KEY) { - throw new Error('KEEPER_SECRET_KEY is required (funded keeper account).'); +} + +export function loadCheckpoint(): number | undefined { + try { + if (fs.existsSync(CHECKPOINT_FILE)) { + const content = fs.readFileSync(CHECKPOINT_FILE, 'utf-8'); + const data = JSON.parse(content); + if (typeof data.lastLedger === 'number') return data.lastLedger; + } + } catch { + /* ignore error */ } + return undefined; +} - const kp = Keypair.fromSecret(KEEPER_SECRET_KEY); +export function saveCheckpoint(ledger: number): void { + try { + fs.writeFileSync(CHECKPOINT_FILE, JSON.stringify({ lastLedger: ledger, updatedAt: new Date().toISOString() }, null, 2)); + } catch { + /* ignore error */ + } +} + +/** Process a list of decoded contract events. */ +export async function processEvents( + events: ParsedEvent[], + kp: Keypair, +): Promise<{ processed: number; ttlBumps: number; markedOverdue: number }> { + let processed = 0; + let ttlBumps = 0; + let markedOverdue = 0; + const now = Math.floor(Date.now() / 1000); const pub = kp.publicKey(); - await ensureAccount(pub); - log(`keeper ${pub} — registry=${REGISTRY_ID.slice(0, 8)}… repayment=${REPAYMENT_ID.slice(0, 8)}…`); - log(`scanning invoices in pages of ${PAGE_SIZE} (max ${MAX_PAGES} pages)`); + + for (const evt of events) { + if (!evt.invoiceId) continue; + processed++; + + if (evt.type === 'inv_reg') { + log(`[event:inv_reg] New invoice registered: ${evt.invoiceId} -> triggering instant TTL bump`); + if (await bumpTtl(evt.invoiceId, kp)) ttlBumps++; + } else if (evt.type === 'off_acc') { + log(`[event:off_acc] Offer accepted for invoice: ${evt.invoiceId} -> triggering instant TTL bump & overdue check`); + if (await bumpTtl(evt.invoiceId, kp)) ttlBumps++; + + const inv = await fetchInvoiceDetails(evt.invoiceId, pub); + if (inv) { + const st = statusNum(inv.status); + const due = Number(inv.due_date ?? inv.dueDate ?? 0); + if (st === STATUS.Financed && due > 0 && due < now) { + log(`[event:off_acc -> overdue] ${evt.invoiceId} due=${due} now=${now} -> mark_overdue`); + if (await markOverdue(evt.invoiceId, kp)) markedOverdue++; + } + } + } + } + + return { processed, ttlBumps, markedOverdue }; +} + +// ── Keeper Execution Modes ─────────────────────────────────────────────────── + +/** Paginated full sweep over registry contract invoices. */ +export async function runFullSweep(kp: Keypair): Promise<{ scanned: number; marked: number; bumped: number }> { + const pub = kp.publicKey(); + log(`[sweep] scanning invoices in pages of ${PAGE_SIZE} (max ${MAX_PAGES} pages)`); let offset = 0; let scanned = 0; @@ -290,15 +423,15 @@ async function main(): Promise { const id = String(inv.id); const st = statusNum(inv.status); const due = Number(inv.due_date); - const active = st === 0 /* Pending */ || st === 1 /* Financed */; + const active = st === STATUS.Pending || st === STATUS.Financed; - if (st === 1 && due < now) { - log(`[overdue] ${id} due=${due} now=${now} → mark_overdue`); + if (st === STATUS.Financed && due < now) { + log(`[sweep:overdue] ${id} due=${due} now=${now} → mark_overdue`); if (await markOverdue(id, kp)) marked += 1; } if (active && bumped < MAX_TTL_BUMPS) { - log(`[ttl] ${id} (status=${st}) → extend ${TTL_EXTEND_LEDGERS} ledgers`); + log(`[sweep:ttl] ${id} (status=${st}) → extend ${TTL_EXTEND_LEDGERS} ledgers`); if (await bumpTtl(id, kp)) bumped += 1; } } @@ -307,16 +440,236 @@ async function main(): Promise { offset += page.length; } - log(`summary: pages=${pages} scanned=${scanned} marked_overdue=${marked} ttl_bumps=${bumped}`); - if (scanned === 0) log('no invoices found — nothing to do (this is fine on a fresh deployment)'); + log(`[sweep summary] pages=${pages} scanned=${scanned} marked_overdue=${marked} ttl_bumps=${bumped}`); + return { scanned, marked, bumped }; } -main() - .then(() => { - log('keeper run complete'); - process.exit(0); - }) - .catch((err: unknown) => { - log(`keeper run FAILED: ${(err as Error).message}`); - process.exit(1); - }); +/** Poll contract events from starting ledger cursor up to latest ledger with pagination. */ +export async function pollEventsOnce( + currentLedger: number, + kp: Keypair, +): Promise<{ nextLedger: number; processed: number; ttlBumps: number; markedOverdue: number }> { + const contractIds = [REGISTRY_ID, FINANCING_ID].filter(Boolean) as string[]; + if (contractIds.length === 0) { + return { nextLedger: currentLedger, processed: 0, ttlBumps: 0, markedOverdue: 0 }; + } + + const latestRes = await rpc.getLatestLedger(); + const latestSeq = latestRes.sequence; + + if (latestSeq < currentLedger) { + return { nextLedger: currentLedger, processed: 0, ttlBumps: 0, markedOverdue: 0 }; + } + + let cursor: string | undefined = undefined; + let latestSeenLedger = currentLedger; + let totalProcessed = 0; + let totalTtlBumps = 0; + let totalMarkedOverdue = 0; + + while (true) { + const params: SorobanRpc.Api.GetEventsRequest = cursor + ? { cursor, filters: [{ type: 'contract', contractIds }], limit: 100 } + : { startLedger: currentLedger, filters: [{ type: 'contract', contractIds }], limit: 100 }; + + const eventRes = await rpc.getEvents(params); + if (!eventRes.events || eventRes.events.length === 0) { + if (eventRes.latestLedger && eventRes.latestLedger > latestSeenLedger) { + latestSeenLedger = eventRes.latestLedger; + } + break; + } + + const parsedEvents: ParsedEvent[] = []; + for (const rawEvt of eventRes.events) { + const parsed = parseRawEvent(rawEvt); + if (parsed) { + parsedEvents.push(parsed); + if (parsed.ledger > latestSeenLedger) { + latestSeenLedger = parsed.ledger; + } + } + } + + const result = await processEvents(parsedEvents, kp); + totalProcessed += result.processed; + totalTtlBumps += result.ttlBumps; + totalMarkedOverdue += result.markedOverdue; + + if (eventRes.latestLedger && eventRes.latestLedger > latestSeenLedger) { + latestSeenLedger = eventRes.latestLedger; + } + + if (eventRes.cursor && eventRes.cursor !== cursor) { + cursor = eventRes.cursor; + } else { + break; + } + } + + const nextLedger = latestSeenLedger >= currentLedger ? latestSeenLedger + 1 : currentLedger; + saveCheckpoint(nextLedger); + + return { + nextLedger, + processed: totalProcessed, + ttlBumps: totalTtlBumps, + markedOverdue: totalMarkedOverdue, + }; +} + +/** Continuous daemon mode: RPC event subscriptions + periodic fallback sweep. */ +export async function runEventDrivenDaemon(kp: Keypair): Promise { + log(`starting event-driven keeper daemon (poll interval: ${EVENT_POLL_INTERVAL_MS}ms, fallback sweep: ${FALLBACK_SWEEP_INTERVAL_MS}ms)`); + + let currentLedger = parseStartLedger() ?? loadCheckpoint(); + if (currentLedger === undefined) { + const latestRes = await rpc.getLatestLedger(); + currentLedger = latestRes.sequence; + log(`no checkpoint found — initializing starting ledger to ${currentLedger}`); + saveCheckpoint(currentLedger); + } else { + log(`resumed from starting/checkpoint ledger ${currentLedger}`); + } + + let running = true; + const shutdown = () => { + log('shutdown signal received — stopping event-driven daemon'); + running = false; + }; + process.on('SIGINT', shutdown); + process.on('SIGTERM', shutdown); + + await runFullSweep(kp); + let lastSweepTime = Date.now(); + let consecutiveFailures = 0; + + while (running) { + try { + const res = await pollEventsOnce(currentLedger, kp); + currentLedger = res.nextLedger; + consecutiveFailures = 0; + } catch (err) { + consecutiveFailures++; + log(`[daemon error] event poll failed (attempt ${consecutiveFailures}): ${(err as Error).message}`); + + if (consecutiveFailures >= 3) { + try { + const health = await rpc.getHealth(); + log(`[daemon recovery] rpc health: ${health.status}`); + const oldestLedger = (health as { oldestLedger?: number }).oldestLedger; + if (oldestLedger !== undefined && oldestLedger > currentLedger) { + log(`[daemon recovery] cursor ${currentLedger} expired below oldest ledger ${oldestLedger} -> advancing cursor`); + currentLedger = oldestLedger; + saveCheckpoint(currentLedger); + } else { + log(`[daemon recovery] preserving current cursor ${currentLedger}`); + } + consecutiveFailures = 0; + } catch (healthErr) { + log(`[daemon recovery error] health check failed: ${(healthErr as Error).message}`); + } + } + } + + if (Date.now() - lastSweepTime >= FALLBACK_SWEEP_INTERVAL_MS) { + log('[daemon] running periodic fallback sweep'); + try { + await runFullSweep(kp); + } catch (sweepErr) { + log(`[daemon sweep error]: ${(sweepErr as Error).message}`); + } + lastSweepTime = Date.now(); + } + + await sleep(EVENT_POLL_INTERVAL_MS); + } +} + +// ── Main Entrypoint ────────────────────────────────────────────────────────── + +export function parseKeeperMode(): KeeperMode { + const args = process.argv.slice(2); + for (const arg of args) { + if (arg.startsWith('--mode=')) { + const modeStr = arg.split('=')[1]; + if (modeStr === 'event-driven' || modeStr === 'event-catchup' || modeStr === 'sweep') { + return modeStr; + } + } + } + const envMode = process.env.KEEPER_MODE; + if (envMode === 'event-driven' || envMode === 'event-catchup' || envMode === 'sweep') { + return envMode; + } + return 'sweep'; +} + +export function parseStartLedger(): number | undefined { + const args = process.argv.slice(2); + for (const arg of args) { + if (arg.startsWith('--start-ledger=')) { + const val = Number(arg.split('=')[1]); + if (Number.isInteger(val) && val > 0) return val; + } + } + const envVal = Number(process.env.KEEPER_START_LEDGER); + if (Number.isInteger(envVal) && envVal > 0) return envVal; + return undefined; +} + +export async function main(): Promise { + const mode = parseKeeperMode(); + + if (!REGISTRY_ID || !REPAYMENT_ID) { + throw new Error('REGISTRY_CONTRACT_ID and REPAYMENT_CONTRACT_ID are required.'); + } + if ((mode === 'event-driven' || mode === 'event-catchup') && !FINANCING_ID) { + throw new Error('FINANCING_CONTRACT_ID is required for event-driven keeper modes.'); + } + if (!KEEPER_SECRET_KEY) { + throw new Error('KEEPER_SECRET_KEY is required (funded keeper account).'); + } + + const kp = Keypair.fromSecret(KEEPER_SECRET_KEY); + const pub = kp.publicKey(); + await ensureAccount(pub); + + log(`keeper ${pub} — mode=${mode} registry=${REGISTRY_ID.slice(0, 8)}… repayment=${REPAYMENT_ID.slice(0, 8)}…`); + + if (mode === 'event-driven') { + await runEventDrivenDaemon(kp); + } else if (mode === 'event-catchup') { + let ledger = parseStartLedger() ?? loadCheckpoint(); + if (ledger === undefined) { + const latest = await rpc.getLatestLedger(); + ledger = Math.max(1, latest.sequence - 1_000); + } + log(`[event-catchup] polling events starting at ledger ${ledger}`); + const res = await pollEventsOnce(ledger, kp); + log(`[event-catchup summary] processed=${res.processed} ttl_bumps=${res.ttlBumps} marked_overdue=${res.markedOverdue}`); + await runFullSweep(kp); + } else { + // Mode: 'sweep' + await runFullSweep(kp); + } +} + +const isDirectExecution = Boolean( + process.argv[1] && + (process.argv[1].endsWith('keeper.ts') || + process.argv[1].endsWith('keeper.js') || + process.argv[1].endsWith('keeper')), +); + +if (isDirectExecution && process.env.NODE_ENV !== 'test' && !process.env.VITEST) { + main() + .then(() => { + log('keeper run complete'); + process.exit(0); + }) + .catch((err: unknown) => { + log(`keeper run FAILED: ${(err as Error).message}`); + process.exit(1); + }); +} diff --git a/invofi/scripts/package.json b/invofi/scripts/package.json index 3ec744a5f..0663423f1 100644 --- a/invofi/scripts/package.json +++ b/invofi/scripts/package.json @@ -6,7 +6,8 @@ "scripts": { "keeper": "tsx keeper.ts", "e2e:onchain": "tsx e2e-onchain.ts", - "type-check": "tsc --noEmit" + "type-check": "tsc --noEmit", + "test": "tsx --test keeper.test.ts" }, "dependencies": { "@stellar/stellar-sdk": "^16.0.1" diff --git a/invofi/scripts/tsconfig.json b/invofi/scripts/tsconfig.json index 67022bb69..7e0c7aee8 100644 --- a/invofi/scripts/tsconfig.json +++ b/invofi/scripts/tsconfig.json @@ -11,5 +11,5 @@ "resolveJsonModule": true, "types": ["node"] }, - "include": ["keeper.ts", "e2e-onchain.ts"] + "include": ["keeper.ts", "keeper.test.ts", "e2e-onchain.ts"] }