feat: reconcile indexed chain state

context/progress-tracker.md

@@ -16,6 +16,15 @@ pure chore/docs commits). Direct pushes to main must also be logged here.
 - Updated existing `test/e2e/modules/liquidity/liquidity-flow.e2e-spec.ts` to mock the new `LiquidityContractClient` structure.
 - Updated `test/unit/modules/liquidity/liquidity.service.spec.ts` unit tests.
 
+## 2026-07-18
+
+- Added scheduled state reconciliation across indexed on-chain loan,
+  liquidity, reputation, and transaction state. The idempotent Cron job
+  resolves provisional loan IDs, repairs stale database state, backfills
+  missed transaction records, marks orphaned pending transactions, exports
+  drift metrics, and logs a structured report without making on-chain writes.
+  Cron is used instead of BullMQ per the API's post-Upstash architecture.
+
 ## 2026-07-16
 
 - Added wallet-bound user roles (sponsor/vendor/mentor): `role` column

src/app.module.ts

@@ -28,6 +28,7 @@ import { MetricsModule } from './modules/metrics/metrics.module';
 import { CreditScoringModule } from './modules/credit-scoring/credit-scoring.module';
 import { AdminModule } from './modules/admin/admin.module';
 import { CorrelationIdMiddleware } from './common/logger/correlation-id.middleware';
+import { StateReconciliationModule } from './jobs/state-reconciliation/state-reconciliation.module';
 
 @Module({
   imports: [
@@ -60,6 +61,7 @@ import { CorrelationIdMiddleware } from './common/logger/correlation-id.middlewa
     TransactionStatusCheckerModule,
     NonceCleanupModule,
     SupabaseKeepAliveModule,
+    StateReconciliationModule,
     CreditScoringModule,
     AdminModule,
     StellarModule,

src/indexer/authoritative-state.reader.ts

@@ -0,0 +1,114 @@
+import { Injectable } from '@nestjs/common';
+import { SupabaseService } from '../database/supabase.client';
+
+export interface AuthoritativeLoan {
+  loanId: string;
+  userWallet: string;
+  status: string;
+  principalAmount: number;
+  transactionHash: string | null;
+}
+
+export interface AuthoritativeLiquidityPosition {
+  userWallet: string;
+  amount: number;
+}
+
+export interface AuthoritativeReputation {
+  wallet: string;
+  score: number;
+}
+
+export interface AuthoritativeState {
+  loans: AuthoritativeLoan[];
+  liquidityPositions: AuthoritativeLiquidityPosition[];
+  reputations: AuthoritativeReputation[];
+  transactionHashes: string[];
+}
+
+interface LoanIndexRow {
+  loan_id: string;
+  user_wallet: string;
+  status: string;
+  principal_amount: number | string;
+  transaction_hash: string | null;
+}
+
+interface InvestmentIndexRow {
+  user_wallet: string;
+  amount: number | string;
+}
+
+interface ReputationHistoryRow {
+  user_wallet: string;
+  new_score: number;
+  transaction_hash: string | null;
+  ledger_sequence: number | null;
+}
+
+interface PaymentIndexRow {
+  tx_hash: string;
+}
+
+@Injectable()
+export class AuthoritativeStateReader {
+  constructor(private readonly supabaseService: SupabaseService) {}
+
+  async read(): Promise<AuthoritativeState> {
+    const db = this.supabaseService.getServiceRoleClient();
+    const [loanResult, investmentResult, reputationResult, paymentResult] = await Promise.all([
+      db.from('loan_index').select('loan_id, user_wallet, status, principal_amount, transaction_hash'),
+      db.from('investments_index').select('user_wallet, amount'),
+      db
+        .from('reputation_history')
+        .select('user_wallet, new_score, transaction_hash, ledger_sequence')
+        .order('ledger_sequence', { ascending: false }),
+      db.from('payment_index').select('tx_hash'),
+    ]);
+
+    const error = loanResult.error ?? investmentResult.error ?? reputationResult.error ?? paymentResult.error;
+    if (error) {
+      throw new Error(`Failed to read authoritative indexed state: ${error.message}`);
+    }
+
+    const loans = (loanResult.data ?? []) as LoanIndexRow[];
+    const investments = (investmentResult.data ?? []) as InvestmentIndexRow[];
+    const reputationRows = (reputationResult.data ?? []) as ReputationHistoryRow[];
+    const payments = (paymentResult.data ?? []) as PaymentIndexRow[];
+    const latestReputation = new Map<string, AuthoritativeReputation>();
+    for (const row of reputationRows) {
+      if (!latestReputation.has(row.user_wallet)) {
+        latestReputation.set(row.user_wallet, { wallet: row.user_wallet, score: row.new_score });
+      }
+    }
+
+    const transactionHashes = new Set<string>();
+    for (const row of loans) if (row.transaction_hash) transactionHashes.add(row.transaction_hash);
+    for (const row of reputationRows) if (row.transaction_hash) transactionHashes.add(row.transaction_hash);
+    for (const row of payments) if (row.tx_hash) transactionHashes.add(row.tx_hash);
+
+    const liquidityByWallet = new Map<string, number>();
+    for (const row of investments) {
+      liquidityByWallet.set(
+        row.user_wallet,
+        (liquidityByWallet.get(row.user_wallet) ?? 0) + Number(row.amount),
+      );
+    }
+
+    return {
+      loans: loans.map((row) => ({
+        loanId: row.loan_id,
+        userWallet: row.user_wallet,
+        status: row.status,
+        principalAmount: Number(row.principal_amount),
+        transactionHash: row.transaction_hash,
+      })),
+      liquidityPositions: [...liquidityByWallet].map(([userWallet, amount]) => ({
+        userWallet,
+        amount,
+      })),
+      reputations: [...latestReputation.values()],
+      transactionHashes: [...transactionHashes],
+    };
+  }
+}

src/indexer/indexer.module.ts

@@ -6,6 +6,7 @@ import { SupabaseService } from '../database/supabase.client';
 import { StellarModule } from '../stellar/stellar.module';
 import { IndexerController } from './indexer.controller';
 import { IndexerStatusService } from './indexer-status.service';
+import { AuthoritativeStateReader } from './authoritative-state.reader';
 
 @Module({
   imports: [ConfigModule, StellarModule],
@@ -15,6 +16,8 @@ import { IndexerStatusService } from './indexer-status.service';
     EventParserService,
     SupabaseService,
     IndexerStatusService,
+    AuthoritativeStateReader,
   ],
+  exports: [AuthoritativeStateReader],
 })
 export class IndexerModule {}

src/indexer/indexer.service.ts

@@ -254,6 +254,8 @@ export class IndexerService {
         interest_amount: payload.interestAmount,
         due_date: payload.dueDate,
         event_id: event.eventId,
+        transaction_hash: event.txHash,
+        ledger_sequence: event.ledgerSequence,
         last_synced_at: new Date().toISOString(),
       },
       { onConflict: 'event_id', ignoreDuplicates: true },

src/jobs/state-reconciliation/state-reconciliation.module.ts

@@ -0,0 +1,12 @@
+import { Module } from '@nestjs/common';
+import { IndexerModule } from '../../indexer/indexer.module';
+import { LoansModule } from '../../modules/loans/loans.module';
+import { TransactionsModule } from '../../modules/transactions/transactions.module';
+import { SupabaseService } from '../../database/supabase.client';
+import { StateReconciliationProcessor } from './state-reconciliation.processor';
+
+@Module({
+  imports: [IndexerModule, LoansModule, TransactionsModule],
+  providers: [StateReconciliationProcessor, SupabaseService],
+})
+export class StateReconciliationModule {}

src/jobs/state-reconciliation/state-reconciliation.processor.ts

@@ -0,0 +1,194 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Cron } from '@nestjs/schedule';
+import { SupabaseService } from '../../database/supabase.client';
+import {
+  AuthoritativeState,
+  AuthoritativeStateReader,
+} from '../../indexer/authoritative-state.reader';
+import { LoansRepository, ReconciliationLoanRow } from '../../modules/loans/loans.repository';
+import { MetricsService } from '../../modules/metrics/metrics.service';
+import {
+  ReconciliationTransaction,
+  TransactionsRepository,
+} from '../../modules/transactions/transactions.repository';
+import { DriftItem, DriftReport, DriftType } from './state-reconciliation.types';
+
+interface LiquidityRow {
+  id: string;
+  provider_wallet: string;
+  deposited_amount: number | string;
+}
+
+interface ReputationRow {
+  wallet_address: string;
+  score: number;
+}
+
+const DRIFT_TYPES: DriftType[] = [
+  'missing_loan_row',
+  'stale_loan_status',
+  'provisional_loan_id',
+  'orphaned_pending_transaction',
+  'missing_transaction_row',
+  'stale_liquidity_position',
+  'stale_reputation',
+];
+
+@Injectable()
+export class StateReconciliationProcessor {
+  private readonly logger = new Logger(StateReconciliationProcessor.name);
+  private isRunning = false;
+
+  constructor(
+    private readonly authoritativeStateReader: AuthoritativeStateReader,
+    private readonly loansRepository: LoansRepository,
+    private readonly transactionsRepository: TransactionsRepository,
+    private readonly supabaseService: SupabaseService,
+    private readonly metricsService: MetricsService,
+  ) {}
+
+  @Cron('0 */15 * * * *')
+  async runScheduled(): Promise<void> {
+    if (this.isRunning) {
+      this.logger.debug('State reconciliation already running; skipping overlapping cycle');
+      return;
+    }
+    this.isRunning = true;
+    try {
+      const report = await this.reconcile();
+      this.logger.log({ event: 'state_reconciliation_completed', report }, 'State reconciliation completed');
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+      this.logger.error({ event: 'state_reconciliation_failed', error: message }, 'State reconciliation failed');
+    } finally {
+      this.isRunning = false;
+    }
+  }
+
+  async reconcile(): Promise<DriftReport> {
+    const startedAt = new Date().toISOString();
+    const [chain, loans, transactions] = await Promise.all([
+      this.authoritativeStateReader.read(),
+      this.loansRepository.findForReconciliation(),
+      this.transactionsRepository.findForReconciliation(),
+    ]);
+    const items: DriftItem[] = [];
+    await this.reconcileLoans(chain, loans, items);
+    await this.reconcileTransactions(chain, transactions, items);
+    await this.reconcileLiquidity(chain, items);
+    await this.reconcileReputation(chain, items);
+
+    const byType = Object.fromEntries(DRIFT_TYPES.map((type) => [type, 0])) as Record<DriftType, number>;
+    for (const item of items) byType[item.type] += 1;
+    for (const type of DRIFT_TYPES) this.metricsService.setReconciliationDrift(type, byType[type]);
+
+    return {
+      startedAt,
+      completedAt: new Date().toISOString(),
+      driftCount: items.length,
+      repairedCount: items.filter((item) => item.repaired).length,
+      byType,
+      items,
+    };
+  }
+
+  private async reconcileLoans(
+    chain: AuthoritativeState,
+    rows: ReconciliationLoanRow[],
+    items: DriftItem[],
+  ): Promise<void> {
+    const rowsByLoanId = new Map(rows.map((row) => [row.loanId, row]));
+    const provisional = rows.filter((row) =>
+      row.loanId.startsWith('loan_') ||
+      row.loanId.startsWith('pending_') ||
+      row.loanId.startsWith('pending-'),
+    );
+    for (const chainLoan of chain.loans) {
+      const row = rowsByLoanId.get(chainLoan.loanId);
+      if (row) {
+        if (row.status !== chainLoan.status) {
+          await this.loansRepository.updateStatus(row.id, chainLoan.status);
+          items.push({ type: 'stale_loan_status', key: chainLoan.loanId, repaired: true, details: { from: row.status, to: chainLoan.status } });
+        }
+        continue;
+      }
+      const candidates = provisional.filter((candidate) => candidate.userWallet === chainLoan.userWallet);
+      const match = candidates.find((candidate) =>
+        Math.abs(candidate.amount - chainLoan.principalAmount) < 0.000001 ||
+        Math.abs(candidate.amount * 0.8 - chainLoan.principalAmount) < 0.000001,
+      ) ?? (candidates.length === 1 ? candidates[0] : undefined);
+      if (match) {
+        await this.loansRepository.resolveProvisionalId(match.id, chainLoan);
+        items.push({ type: 'provisional_loan_id', key: match.loanId, repaired: true, details: { chainLoanId: chainLoan.loanId } });
+      } else {
+        items.push({ type: 'missing_loan_row', key: chainLoan.loanId, repaired: false, details: { userWallet: chainLoan.userWallet } });
+      }
+    }
+  }
+
+  private async reconcileTransactions(
+    chain: AuthoritativeState,
+    rows: ReconciliationTransaction[],
+    items: DriftItem[],
+  ): Promise<void> {
+    const chainHashes = new Set(chain.transactionHashes.map((hash) => hash.toLowerCase()));
+    const rowsByHash = new Map(rows.map((row) => [row.hash.toLowerCase(), row]));
+    for (const hash of chainHashes) {
+      const row = rowsByHash.get(hash);
+      if (!row) {
+        await this.transactionsRepository.backfill(hash);
+        items.push({ type: 'missing_transaction_row', key: hash, repaired: true, details: {} });
+      } else if (row.status === 'pending') {
+        await this.transactionsRepository.markSucceeded(row.id);
+        items.push({ type: 'orphaned_pending_transaction', key: hash, repaired: true, details: { resolution: 'success' } });
+      }
+    }
+    const cutoff = Date.now() - 30 * 60 * 1000;
+    for (const row of rows) {
+      if (row.status === 'pending' && !chainHashes.has(row.hash.toLowerCase()) && Date.parse(row.submittedAt) < cutoff) {
+        await this.transactionsRepository.markOrphaned(row.id);
+        items.push({ type: 'orphaned_pending_transaction', key: row.hash, repaired: true, details: { resolution: 'failed' } });
+      }
+    }
+  }
+
+  private async reconcileLiquidity(chain: AuthoritativeState, items: DriftItem[]): Promise<void> {
+    const db = this.supabaseService.getServiceRoleClient();
+    const { data, error } = await db.from('liquidity_positions').select('id, provider_wallet, deposited_amount');
+    if (error) throw new Error(`Failed to read liquidity positions: ${error.message}`);
+    const rows = (data ?? []) as LiquidityRow[];
+    const byWallet = new Map(rows.map((row) => [row.provider_wallet, row]));
+    for (const position of chain.liquidityPositions) {
+      const row = byWallet.get(position.userWallet);
+      if (!row || Number(row.deposited_amount) !== position.amount) {
+        const payload = { provider_wallet: position.userWallet, deposited_amount: position.amount, updated_at: new Date().toISOString() };
+        const result = row
+          ? await db.from('liquidity_positions').update(payload).eq('id', row.id)
+          : await db.from('liquidity_positions').insert(payload);
+        if (result.error) throw new Error(`Failed to reconcile liquidity position: ${result.error.message}`);
+        items.push({ type: 'stale_liquidity_position', key: position.userWallet, repaired: true, details: { amount: position.amount } });
+      }
+    }
+  }
+
+  private async reconcileReputation(chain: AuthoritativeState, items: DriftItem[]): Promise<void> {
+    const db = this.supabaseService.getServiceRoleClient();
+    const { data, error } = await db.from('reputation_cache').select('wallet_address, score');
+    if (error) throw new Error(`Failed to read reputation cache: ${error.message}`);
+    const rows = (data ?? []) as ReputationRow[];
+    const byWallet = new Map(rows.map((row) => [row.wallet_address, row]));
+    for (const reputation of chain.reputations) {
+      const row = byWallet.get(reputation.wallet);
+      if (!row) {
+        items.push({ type: 'stale_reputation', key: reputation.wallet, repaired: false, details: { score: reputation.score, reason: 'cache row requires user relation' } });
+      } else if (row.score !== reputation.score) {
+        const { error: updateError } = await db
+          .from('reputation_cache')
+          .update({ score: reputation.score, last_synced_at: new Date().toISOString() })
+          .eq('wallet_address', reputation.wallet);
+        if (updateError) throw new Error(`Failed to reconcile reputation: ${updateError.message}`);
+        items.push({ type: 'stale_reputation', key: reputation.wallet, repaired: true, details: { score: reputation.score } });
+      }
+    }
+  }
+}