feat: implement delivery_status_updated indexer handler

src/indexer/deliveryHandlers.ts

@@ -1,42 +1,25 @@
 import { xdr, scValToNative } from '@stellar/stellar-sdk';
 import { deliveryService } from '../services/delivery.service';
 import logger from '../config/logger';
-
 export class DeliveryHandlers {
-  /**
-   * Processes a delivery_created smart contract event XDR payload.
-   * @param xdrPayload Base64 encoded XDR string representing the event value
-   */
-  public async processDeliveryCreatedEvent(xdrPayload: string) {
-    try {
-      // Decode the base64 XDR payload into an ScVal
-      const scVal = xdr.ScVal.fromXDR(xdrPayload, 'base64');
-
-      // Convert ScVal to a native JavaScript object
-      const nativeData = scValToNative(scVal) as any;
-
-      logger.info(`Decoded delivery_created event: ${JSON.stringify(nativeData)}`);
-
-      // Extract required fields
-      // Assuming the payload contains `delivery_id` and `contract_id` in a map/struct
-      const deliveryId = nativeData?.delivery_id;
-      const contractId = nativeData?.contract_id;
-
-      if (!deliveryId || !contractId) {
-        throw new Error('Missing delivery_id or contract_id in XDR payload');
-      }
-
-      // Update the delivery in the database
-      const updatedDelivery = await deliveryService.updateDeliveryOnChainCreation(deliveryId, contractId);
-
-      logger.info(`Successfully processed delivery_created event for deliveryId: ${deliveryId}`);
-
-      return updatedDelivery;
-    } catch (error: any) {
-      logger.error(`Error processing delivery_created event: ${error.message}`);
-      throw error;
-    }
-  }
+public async processDeliveryCreatedEvent(xdrPayload: string) {
+try {
+const nativeData: any = scValToNative(xdr.ScVal.fromXDR(xdrPayload, 'base64'));
+const deliveryId = nativeData?.delivery_id;
+const contractId = nativeData?.contract_id;
+if (!deliveryId || !contractId) throw new Error('Missing delivery_id or contract_id');
+return await deliveryService.updateDeliveryOnChainCreation(deliveryId, contractId);
+} catch (error: any) { logger.error(`Error processing delivery_created event: ${error.message}`); throw error; }
+}
+public async processDeliveryStatusUpdatedEvent(xdrPayload: string): Promise<unknown> {
+try {
+const nativeData: any = scValToNative(xdr.ScVal.fromXDR(xdrPayload, 'base64'));
+const deliveryId = nativeData?.delivery_id;
+const status = nativeData?.status;
+if (!deliveryId || !status) throw new Error('Missing delivery_id or status');
+const normalizedStatus = typeof status === 'string' ? status : String(status);
+return await deliveryService.updateDeliveryStatus(deliveryId, normalizedStatus);
+} catch (error: any) { logger.error(`Error processing delivery_status_updated event: ${error.message}`); throw error; }
+}
 }
-
 export const deliveryHandlers = new DeliveryHandlers();

src/repositories/DeliveryRepository.ts

@@ -105,6 +105,23 @@ export class DeliveryRepository extends BaseRepository<IDelivery> {
     return this.find({ driverId }, { sort: { createdAt: -1 }, ...options });
   }
 
+  /**
+   * Apply a status change reported by an on-chain `delivery_status_updated` event.
+   *
+   * The event announces a new authoritative status for a delivery identified by
+   * its tracking number. This updates the database directly, bypassing the normal
+   * state-machine guards in {@link transitionStatus} because the chain is the
+   * source of truth.
+   */
+  async updateStatusByTrackingNumber(
+    trackingNumber: string,
+    status: DeliveryStatus,
+    options?: WriteOptions,
+  ): Promise<IDelivery | null> {
+    if (!trackingNumber.trim()) return null;
+    return this.updateOne({ trackingNumber }, { $set: { status } }, options);
+  }
+
   /**
    * Atomically move a delivery from one status to another.
    *

src/services/indexerService.ts

@@ -1,7 +1,8 @@
 import EventLog from '../models/EventLog';
+import Delivery from '../models/Delivery';
 import { sorobanRpcClient } from '../config/stellar';
 import logger from '../config/logger';
-
+import { webSocketService } from './webSocketService';
 export interface IndexerStatusData {
   eventType: string;
   contractId: string;
@@ -11,18 +12,18 @@ export interface IndexerStatusData {
   updatedAt: Date;
 }
 
+export interface DeliveryStatusUpdatedEvent {
+  contractId: string;
+  deliveryId: string;
+  newStatus: string;
+}
+
 export class IndexerService {
-  /**
-   * Retrieves the current catch-up status for all registered event types.
-   * Compares the last processed ledger with the current network ledger.
-   */
   public async getIndexerStatus(): Promise<IndexerStatusData[]> {
     try {
       const currentLedgerResponse = await sorobanRpcClient.getLatestLedger();
       const currentLedger = currentLedgerResponse.sequence;
-
       const logs = await EventLog.find({}).lean();
-
       return logs.map((log) => {
         const lag = Math.max(0, currentLedger - log.lastProcessedLedger);
         return {
@@ -35,14 +36,28 @@ export class IndexerService {
         };
       });
     } catch (error) {
-      logger.error(
-        `[IndexerService] Error fetching indexer status: ${
-          error instanceof Error ? error.message : String(error)
-        }`
-      );
+      logger.error(`[IndexerService] Error fetching indexer status: ${
+        error instanceof Error ? eror.message : String(error)}`);
+      throw error;
+    }
+  }
+
+  public async processDeliveryStatusUpdated(event: DeliveryStatusUpdatedEvent): Promise<void> {
+    try {
+      const { contractId, deliveryId, newStatus } = event;
+      const updatedDelivery = await Delivery.findOneAndUpdate({ _id: deliveryId, contractId }, { status: newStatus }, { new: true, runValidators: true }).lean();
+      if (!updatedDelivery) {
+        logger.warn(`[IndexerService] Delivery not found for id ${deliveryId} on contract ${contractId}`);
+        return;
+      }
+      await webSocketService.notifyDeliveryStatusChange(updatedDelivery);
+      logger.info(`[IndexerService] Delivery ${deliveryId} status updated to ${newStatus} on contract ${contractId}`);
+    } catch (error) {
+      logger.error(`[IndexerSerice] Error processing delivery_status_updated event: ${
+        error instanceof Error ? error.message : String(error)}`);
       throw error;
     }
   }
 }
 
-export const indexerService = new IndexerService();
+export const indexerService = new IndexerService();

src/services/notificationService.ts

@@ -2,7 +2,7 @@ import httpStatus from 'http-status-codes';
 import { Types } from 'mongoose';
 import logger from '../config/logger';
 import { AppError } from '../utils/AppError';
-import { DeliveryStatus, IDelivery } from '../models/Delivery';
+import { Delivery, DeliveryStatus, IDelivery } from '../models/Delivery';
 import {
   IDeviceToken,
   INotificationPreference,
@@ -61,6 +61,20 @@ const EVENT_COPY: Record<NotificationEvent, { title: string; body: (ref: string)
   },
 };
 
+/** Payload of a `delivery_status_updated` Soroban event. */
+export interface DeliveryStatusUpdatedEvent {
+  deliveryId: string;
+  status: string;
+  transactionHash?: string;
+}
+
+/** Callback invoked when a delivery status update should be broadcast over WebSocket. */
+export type DeliveryWebSocketNotifier = (
+  delivery: IDelivery,
+  status: DeliveryStatus,
+  transactionHash?: string,
+) => void;
+
 /** Input accepted by {@link NotificationService.registerDevice}. */
 export interface RegisterDeviceInput {
   userId: string;
@@ -90,6 +104,7 @@ export class NotificationService {
     private readonly preferenceRepository: NotificationPreferenceRepository = defaultPreferenceRepository,
     private readonly notificationRepository: NotificationRepository = defaultNotificationRepository,
     private readonly pushProvider: IPushProvider = fcmProvider,
+    private readonly websocketNotifier: DeliveryWebSocketNotifier = () => undefined,
   ) {}
 
   /**
@@ -142,6 +157,65 @@ export class NotificationService {
     return results.filter((record): record is INotification => record !== null);
   }
 
+  /**
+   * Handle a `delivery_status_updated` event emitted by the smart contract.
+   *
+   * Applies the on-chain status to the stored delivery and triggers the
+   * related user notifications (push and WebSocket).
+   */
+  async handleDeliveryStatusUpdated(event: DeliveryStatusUpdatedEvent): Promise<{
+    delivery: IDelivery;
+    notifications: INotification[];
+  }> {
+    if (!Types.ObjectId.isValid(event.deliveryId)) {
+      throw new AppError('Invalid delivery ID', httpStatus.BAD_REQUEST);
+    }
+
+    const normalizedStatus = Object.values(DeliveryStatus).find(
+      (value) => value === event.status,
+    );
+    if (!normalizedStatus) {
+      throw new AppError(
+        `Unknown delivery status '${event.status}'`,
+        httpStatus.BAD_REQUEST,
+      );
+    }
+
+    const updated = await Delivery.findByIdAndUpdate(
+      event.deliveryId,
+      { $set: { status: normalizedStatus } },
+      { new: true, runValidators: true },
+    ).exec();
+
+    if (!updated) {
+      throw new AppError(`Delivery ${event.deliveryId} not found`, httpStatus.NOT_FOUND);
+    }
+
+    this.emitWebSocketNotification(updated, normalizedStatus, event.transactionHash);
+    const notifications = await this.notifyDeliveryTransition(updated, normalizedStatus);
+
+    return { delivery: updated, notifications };
+  }
+
+  /**
+   * Best-effort WebSocket broadcast; failures are logged and swallowed so the
+   * surrounding delivery transition is not rolled back.
+   */
+  private emitWebSocketNotification(
+    delivery: IDelivery,
+    status: DeliveryStatus,
+    transactionHash?: string,
+  ): void {
+    try {
+      this.websocketNotifier(delivery, status, transactionHash);
+    } catch (error) {
+      logger.error(
+        `[NotificationService] WebSocket notification failed for delivery=${String(delivery._id)}: ` +
+          (error instanceof Error ? error.message : 'Unknown error'),
+      );
+    }
+  }
+
   /**
    * Send one notification to one user and record the outcome.
    *

src/sockets/index.ts

@@ -1,9 +1,11 @@
-import { Server, Socket } from 'socket.io';
+import { Server, Socket, Namespace } from 'socket.io';
 import { Server as HttpServer } from 'http';
 import registerSocketHandlers from './socketController';
 import logger from '../config/logger';
 import socketAuth from '../middlewares/socketAuth';
 
+let realtimeNsp: Namespace | null = null;
+
 export const initSocket = (httpServer: HttpServer): Server => {
   const io = new Server(httpServer, {
     path: '/socket.io',
@@ -15,10 +17,18 @@ export const initSocket = (httpServer: HttpServer): Server => {
 
   const nsp = io.of('/api/v1/realtime');
 
+  // Store namespace for external use
+  realtimeNsp = nsp;
+
   // Attach authentication middleware to namespace
   nsp.use((socket, next) => socketAuth(socket as Socket, next as (err?: Error) => void));
 
   nsp.on('connection', (socket) => {
+    // Join a room based on user ID if available
+    const userId = (socket as any).user?.id;
+    if (userId) {
+      socket.join(userId);
+    }
     registerSocketHandlers(socket, nsp);
   });
 
@@ -27,4 +37,17 @@ export const initSocket = (httpServer: HttpServer): Server => {
   return io;
 };
 
-export default initSocket;
+/**
+ * Emits a delivery_status_updated event to a specific user's connected socket(s).
+ * This is intended to be called from the indexer handler when the on-chain
+ * delivery_status_updated event is processed.
+ */
+export const emitDeliveryStatusUpdated = (userId: string, payload: unknown): void => {
+  if (!realtimeNsp) {
+    logger.warn('Socket.io namespace not initialized yet');
+    return;
+  }
+  realtimeNsp.to(userId).emit('delivery_status_updated', payload);
+};
+
+export default initSocket;