From 384500a4d599824b3b95e9bd7bae0b7a025566b6 Mon Sep 17 00:00:00 2001 From: Hasmong Date: Mon, 31 Aug 2026 10:46:57 +0100 Subject: [PATCH 1/5] feat: Backend: Implement indexer handler for delivery_status_updat (#32) --- src/services/notificationService.ts | 76 ++++++++++++++++++++++++++++- 1 file changed, 75 insertions(+), 1 deletion(-) diff --git a/src/services/notificationService.ts b/src/services/notificationService.ts index dad3ba8..d0126cb 100644 --- a/src/services/notificationService.ts +++ b/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 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. * From a59f99b92ad6e2ab73e74bfb5f8409b8e132c70d Mon Sep 17 00:00:00 2001 From: Hasmong Date: Mon, 31 Aug 2026 10:46:59 +0100 Subject: [PATCH 2/5] feat: Backend: Implement indexer handler for delivery_status_updat (#32) --- src/indexer/deliveryHandlers.ts | 55 ++++++++++++--------------------- 1 file changed, 19 insertions(+), 36 deletions(-) diff --git a/src/indexer/deliveryHandlers.ts b/src/indexer/deliveryHandlers.ts index 5ea406f..1b142d1 100644 --- a/src/indexer/deliveryHandlers.ts +++ b/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 { +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(); From b55281adb1e1eda28b2bd80f819e7992175a29b4 Mon Sep 17 00:00:00 2001 From: Hasmong Date: Mon, 31 Aug 2026 10:47:00 +0100 Subject: [PATCH 3/5] feat: Backend: Implement indexer handler for delivery_status_updat (#32) --- src/services/indexerService.ts | 41 +++++++++++++++++++++++----------- 1 file changed, 28 insertions(+), 13 deletions(-) diff --git a/src/services/indexerService.ts b/src/services/indexerService.ts index 8654921..a93f1d4 100644 --- a/src/services/indexerService.ts +++ b/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 { 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 { + 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(); \ No newline at end of file From 4fe1efd8255de7c8ad752b72bbf649db1233a76d Mon Sep 17 00:00:00 2001 From: Hasmong Date: Mon, 31 Aug 2026 10:47:02 +0100 Subject: [PATCH 4/5] feat: Backend: Implement indexer handler for delivery_status_updat (#32) --- src/sockets/index.ts | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/src/sockets/index.ts b/src/sockets/index.ts index d2e7220..2c946b3 100644 --- a/src/sockets/index.ts +++ b/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; \ No newline at end of file From 0986eabc9576fb617296169fbcc42488eb4b281d Mon Sep 17 00:00:00 2001 From: Hasmong Date: Mon, 31 Aug 2026 10:47:05 +0100 Subject: [PATCH 5/5] feat: Backend: Implement indexer handler for delivery_status_updat (#32) --- src/repositories/DeliveryRepository.ts | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/repositories/DeliveryRepository.ts b/src/repositories/DeliveryRepository.ts index 77015fb..4bf4d4d 100644 --- a/src/repositories/DeliveryRepository.ts +++ b/src/repositories/DeliveryRepository.ts @@ -105,6 +105,23 @@ export class DeliveryRepository extends BaseRepository { 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 { + if (!trackingNumber.trim()) return null; + return this.updateOne({ trackingNumber }, { $set: { status } }, options); + } + /** * Atomically move a delivery from one status to another. *