core: implement real-time events via Supabase Realtime

.github/PULL_REQUEST_TEMPLATE.md

@@ -1,30 +1,49 @@
-## 🔗 Related Issue
-Closes #issue-number
+## Summary
 
----
+Closes #[issue number]
 
-## 🔖 Title
-<!-- Brief and clear. Describe the specific task -->
+Briefly describe what this PR does in 2-3 sentences.
 
----
+## This repo is for the NestJS backend API only
 
-## 📝 Description
-<!-- Describe the changes made in this PR. What problem does it solve? -->
+Before submitting, confirm your changes belong here:
 
----
+- [ ] My changes are inside src/ or test/
+- [ ] I have NOT added React, React Native,
+      or frontend component files
+- [ ] I have NOT added Rust or Soroban contract code
+- [ ] This is NestJS/TypeScript backend work
 
-## 🔄 Changes Made
-<!-- List the main changes in this PR -->
-- [ ] <!-- Change 1 -->
-- [ ] <!-- Change 2 -->
-- [ ] <!-- Change 3 -->
+## Type of change
 
----
+- [ ] Bug fix
+- [ ] New endpoint
+- [ ] New service or module
+- [ ] Database migration
+- [ ] Background job
+- [ ] Test coverage
 
-## 📸 Screenshots (if applicable)
-<!-- Add screenshots or GIFs if the changes affect the UI -->
+## Testing
 
----
+- [ ] npm run build passes with zero TypeScript errors
+- [ ] npm test passes — all 184+ existing tests pass
+- [ ] No new `any` types introduced anywhere
+- [ ] Swagger decorators added to every new endpoint
+- [ ] Migration file created for any schema changes
+- [ ] New unit tests written for new service methods
 
-## 🗒️ Additional Notes
-<!-- Any additional information, concerns, or context for reviewers -->
+## Context files reviewed
+
+- [ ] context/architecture-context.md
+- [ ] context/code-standards.md
+- [ ] context/progress-tracker.md updated
+
+## Mandatory before requesting review
+
+Running these must all exit 0:
+npm run build
+npm test
+
+If either fails, fix it before opening this PR.
+PRs with failing CI checks will be closed without review.
+PRs that reduce the test count will be rejected.

docs/realtime-integration.md

@@ -0,0 +1,116 @@
+# Real-Time Notifications Integration Guide
+
+This document explains how web and mobile clients can subscribe to real-time events from the StepFi API. Two channels are provided for real-time updates:
+1. **Supabase Realtime (Database Replication)**: Recommended for simple database table synchronization.
+2. **WebSocket Gateway (Standalone Port)**: Best for dedicated real-time event-driven updates.
+
+---
+
+## 1. Supabase Realtime
+
+We have enabled Postgres changes replication on the following tables:
+- `loan_index`: Emits changes when a loan is created, updated, or defaulted.
+- `payment_index`: Emits insertions when a payment is processed.
+
+### How to Subscribe (JS/TS Example)
+
+```javascript
+import { createClient } from '@supabase/supabase-js';
+
+const supabase = createClient('SUPABASE_URL', 'SUPABASE_ANON_KEY');
+
+// 1. Subscribe to Loan Status changes
+const loanChannel = supabase
+  .channel('loan-status-changes')
+  .on(
+    'postgres_changes',
+    {
+      event: 'UPDATE', // Listen for updates (active -> paid, defaulted, etc)
+      schema: 'public',
+      table: 'loan_index',
+    },
+    (payload) => {
+      console.log('Loan status updated:', payload.new);
+      // payload.new.status contains 'paid' or 'defaulted'
+    }
+  )
+  .subscribe();
+
+// 2. Subscribe to Payment Confirmations
+const paymentChannel = supabase
+  .channel('payment-confirmations')
+  .on(
+    'postgres_changes',
+    {
+      event: 'INSERT', // Listen for insertions of new payments
+      schema: 'public',
+      table: 'payment_index',
+    },
+    (payload) => {
+      console.log('Payment confirmed:', payload.new);
+      // payload.new contains loan_id, tx_hash, amount, paid_at
+    }
+  )
+  .subscribe();
+```
+
+---
+
+## 2. WebSocket Gateway
+
+The API exposes a WebSocket Gateway running on a dedicated port (`3005` by default, configurable via `WEBSOCKET_PORT`).
+
+- **Endpoint**: `ws://localhost:3005` (or custom host/port)
+
+### Gateway Events
+
+#### `loan.status_changed`
+Emitted immediately after a ledger event updates a loan's status in the index.
+- **Payload Schema**:
+  ```json
+  {
+    "loanId": "string",
+    "status": "active" | "paid" | "defaulted",
+    "userWallet": "string (optional)",
+    "principalAmount": "string (optional)",
+    "interestAmount": "string (optional)",
+    "dueDate": "string (optional)"
+  }
+  ```
+
+#### `payment.confirmed`
+Emitted immediately when a new payment confirmation block is parsed and written to the DB index.
+- **Payload Schema**:
+  ```json
+  {
+    "loanId": "string",
+    "txHash": "string",
+    "amount": "string",
+    "paidAt": "string"
+  }
+  ```
+
+### How to Subscribe (JS Example)
+
+```javascript
+const ws = new WebSocket('ws://localhost:3005');
+
+ws.onopen = () => {
+  console.log('Connected to StepFi WebSocket Gateway');
+};
+
+ws.onmessage = (event) => {
+  const data = JSON.parse(event.data);
+  console.log(`Received event ${data.event}:`, data.payload);
+
+  if (data.event === 'loan.status_changed') {
+    handleLoanStatusUpdate(data.payload.loanId, data.payload.status);
+  } else if (data.event === 'payment.confirmed') {
+    showPaymentToast(data.payload.amount, data.payload.txHash);
+  }
+};
+
+ws.onclose = () => {
+  console.log('Disconnected from WebSocket Gateway');
+};
+```

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 { RealtimeModule } from './realtime/realtime.module';
 
 @Module({
   imports: [
@@ -63,6 +64,7 @@ import { CorrelationIdMiddleware } from './common/logger/correlation-id.middlewa
     CreditScoringModule,
     AdminModule,
     StellarModule,
+    RealtimeModule,
   ],
   controllers: [],
   providers: [

src/indexer/event-handlers/realtime.handler.ts

@@ -0,0 +1,36 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { RealtimeService } from '../../realtime/realtime.service';
+import { ParsedContractEvent, LoanCreatedPayload, LoanRepaidPayload, LoanDefaultedPayload } from '../interfaces';
+
+@Injectable()
+export class RealtimeEventHandler {
+  private readonly logger = new Logger(RealtimeEventHandler.name);
+
+  constructor(private readonly realtimeService: RealtimeService) {}
+
+  handleLoanCreated(event: ParsedContractEvent<LoanCreatedPayload>): void {
+    this.logger.log(`Handling LOAN_CREATED event for realtime: ${event.payload.loanId}`);
+    this.realtimeService.broadcastLoanStatusChanged(event.payload.loanId, 'active', {
+      userWallet: event.payload.userWallet,
+      principalAmount: String(event.payload.principalAmount),
+      interestAmount: String(event.payload.interestAmount),
+      dueDate: event.payload.dueDate,
+    });
+  }
+
+  handleLoanRepaid(event: ParsedContractEvent<LoanRepaidPayload>, newStatus: string): void {
+    this.logger.log(`Handling LOAN_REPAID event for realtime: ${event.payload.loanId}`);
+    this.realtimeService.broadcastPaymentConfirmed(
+      event.payload.loanId,
+      event.payload.txHash,
+      String(event.payload.amount),
+      event.payload.paidAt,
+    );
+    this.realtimeService.broadcastLoanStatusChanged(event.payload.loanId, newStatus);
+  }
+
+  handleLoanDefaulted(event: ParsedContractEvent<LoanDefaultedPayload>): void {
+    this.logger.log(`Handling LOAN_DEFAULTED event for realtime: ${event.payload.loanId}`);
+    this.realtimeService.broadcastLoanStatusChanged(event.payload.loanId, 'defaulted');
+  }
+}

src/indexer/indexer.module.ts

@@ -6,15 +6,18 @@ import { SupabaseService } from '../database/supabase.client';
 import { StellarModule } from '../stellar/stellar.module';
 import { IndexerController } from './indexer.controller';
 import { IndexerStatusService } from './indexer-status.service';
+import { RealtimeModule } from '../realtime/realtime.module';
+import { RealtimeEventHandler } from './event-handlers/realtime.handler';
 
 @Module({
-  imports: [ConfigModule, StellarModule],
+  imports: [ConfigModule, StellarModule, RealtimeModule],
   controllers: [IndexerController],
   providers: [
     IndexerService,
     EventParserService,
     SupabaseService,
     IndexerStatusService,
+    RealtimeEventHandler,
   ],
 })
 export class IndexerModule {}

src/indexer/indexer.service.ts

@@ -5,6 +5,7 @@ import * as StellarSdk from 'stellar-sdk';
 import { SupabaseService } from '../database/supabase.client';
 import { SorobanService } from '../blockchain/soroban/soroban.service';
 import { EventParserService } from './event-parser.service';
+import { RealtimeEventHandler } from './event-handlers/realtime.handler';
 import {
   ParsedContractEvent,
   LoanEventType,
@@ -31,6 +32,7 @@ export class IndexerService {
     private readonly sorobanService: SorobanService,
     private readonly supabaseService: SupabaseService,
     private readonly eventParser: EventParserService,
+    private readonly realtimeEventHandler: RealtimeEventHandler,
   ) {
     this.loanContractId =
       this.configService.get<string>('CREDIT_LINE_CONTRACT_ID') || '';
@@ -266,6 +268,8 @@ export class IndexerService {
       }
       throw new Error(`Failed to persist LOAN_CREATED: ${error.message}`);
     }
+
+    this.realtimeEventHandler.handleLoanCreated(event);
   }
 
   private async persistLoanRepaid(
@@ -325,6 +329,8 @@ export class IndexerService {
           last_synced_at: new Date().toISOString(),
         })
         .eq('loan_id', payload.loanId);
+
+      this.realtimeEventHandler.handleLoanRepaid(event, newStatus);
     }
   }
 
@@ -344,6 +350,8 @@ export class IndexerService {
     if (error) {
       throw new Error(`Failed to persist LOAN_DEFAULTED: ${error.message}`);
     }
+
+    this.realtimeEventHandler.handleLoanDefaulted(event);
   }
 
   private async persistScoreChanged(

src/modules/health/health.service.ts

@@ -53,10 +53,20 @@ export class HealthService {
       if (error && error.message !== 'Invalid Refresh Token' && !error.message.includes('JWT')) {
         throw error;
       }
-      return { status: 'ok', database: 'connected', message: 'Supabase reachable' };
+      return {
+        status: 'ok',
+        database: 'connected',
+        message: 'Supabase reachable',
+        timestamp: new Date().toISOString(),
+      };
     } catch (error) {
       this.logger.error({ context: 'HealthService', action: 'checkDatabase', error: error.message });
-      return { status: 'error', database: 'disconnected', message: error.message };
+      return {
+        status: 'error',
+        database: 'disconnected',
+        message: error.message,
+        timestamp: new Date().toISOString(),
+      };
     }
   }
 

src/realtime/realtime.gateway.ts

@@ -0,0 +1,51 @@
+import { Injectable, OnModuleInit, OnModuleDestroy, Logger } from '@nestjs/common';
+import { WebSocketServer, WebSocket } from 'ws';
+import { ConfigService } from '@nestjs/config';
+
+@Injectable()
+export class RealtimeGateway implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(RealtimeGateway.name);
+  private wss: WebSocketServer;
+
+  constructor(private readonly configService: ConfigService) {}
+
+  onModuleInit() {
+    const port = Number(this.configService.get<number>('WEBSOCKET_PORT', 3005));
+    this.wss = new WebSocketServer({ port });
+    this.logger.log(`WebSocket Gateway initialized on port ${port}`);
+
+    this.wss.on('connection', (ws: WebSocket) => {
+      this.logger.log('Client connected to WebSocket Gateway');
+
+      ws.on('error', (err) => {
+        this.logger.error(`WebSocket error: ${err.message}`);
+      });
+
+      ws.on('close', () => {
+        this.logger.log('Client disconnected');
+      });
+    });
+  }
+
+  onModuleDestroy() {
+    if (this.wss) {
+      this.wss.close(() => {
+        this.logger.log('WebSocket Gateway server closed');
+      });
+    }
+  }
+
+  broadcast(event: string, payload: unknown): void {
+    if (!this.wss) return;
+    const message = JSON.stringify({ event, payload });
+    this.wss.clients.forEach((client) => {
+      if (client.readyState === WebSocket.OPEN) {
+        client.send(message);
+      }
+    });
+  }
+
+  getClientsCount(): number {
+    return this.wss ? this.wss.clients.size : 0;
+  }
+}

src/realtime/realtime.module.ts

@@ -0,0 +1,11 @@
+import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
+import { RealtimeGateway } from './realtime.gateway';
+import { RealtimeService } from './realtime.service';
+
+@Module({
+  imports: [ConfigModule],
+  providers: [RealtimeGateway, RealtimeService],
+  exports: [RealtimeService, RealtimeGateway],
+})
+export class RealtimeModule {}