diff --git a/.github/workflows/contract-ci.yml b/.github/workflows/contract-ci.yml index 642d9372..57b8ff0a 100644 --- a/.github/workflows/contract-ci.yml +++ b/.github/workflows/contract-ci.yml @@ -2,15 +2,15 @@ name: Contract CI on: push: - branches: [ main, develop ] + branches: [main, develop] paths: - - 'apps/onchain/**' - - '.github/workflows/contract-ci.yml' + - "apps/onchain/**" + - ".github/workflows/contract-ci.yml" pull_request: - branches: [ main, develop ] + branches: [main, develop] paths: - - 'apps/onchain/**' - - '.github/workflows/contract-ci.yml' + - "apps/onchain/**" + - ".github/workflows/contract-ci.yml" jobs: test: @@ -28,8 +28,12 @@ jobs: uses: dtolnay/rust-toolchain@stable with: toolchain: stable + components: rustfmt, clippy targets: wasm32-unknown-unknown + - name: Add wasm target + run: rustup target add wasm32-unknown-unknown + - name: Cache dependencies uses: Swatinem/rust-cache@v2 with: diff --git a/apps/backend/src/app.module.ts b/apps/backend/src/app.module.ts index e3ec58a5..105b6497 100644 --- a/apps/backend/src/app.module.ts +++ b/apps/backend/src/app.module.ts @@ -34,6 +34,7 @@ import { AllowedAsset } from './modules/assets/entities/allowed-asset.entity'; import { IpfsModule } from './modules/ipfs/ipfs.module'; import { HealthModule } from './modules/health/health.module'; import { AppVersionModule } from './app-version/app-version.module'; +import { EventsModule } from './gateways/events.module'; import stellarConfig from './config/stellar.config'; import ipfsConfig from './config/ipfs.config'; @@ -79,6 +80,7 @@ import ipfsConfig from './config/ipfs.config'; AuthModule, UserModule, EscrowModule, + EventsModule, StellarModule, forwardRef(() => AdminModule), WebhookModule, diff --git a/apps/backend/src/gateways/escrow.gateway.spec.ts b/apps/backend/src/gateways/escrow.gateway.spec.ts new file mode 100644 index 00000000..b25bf800 --- /dev/null +++ b/apps/backend/src/gateways/escrow.gateway.spec.ts @@ -0,0 +1,54 @@ +import { EventsGateway } from './escrow.gateway'; + +describe('EventsGateway', () => { + let gateway: EventsGateway; + let jwtService: { verify: jest.Mock }; + + beforeEach(() => { + jwtService = { + verify: jest.fn().mockReturnValue({ sub: 'user-1' }), + }; + gateway = new EventsGateway(jwtService as any); + }); + + it('authenticates a socket and emits a connected event', async () => { + const client = { + id: 'socket-1', + handshake: { auth: { token: 'valid-token' }, headers: {} }, + emit: jest.fn(), + disconnect: jest.fn(), + join: jest.fn(), + leave: jest.fn(), + }; + + await gateway.handleConnection(client as any); + + expect(jwtService.verify).toHaveBeenCalledWith('valid-token'); + expect(client.emit).toHaveBeenCalledWith( + 'connected', + expect.objectContaining({ userId: 'user-1', socketId: 'socket-1' }), + ); + }); + + it('broadcasts escrow and notification events to the correct rooms', () => { + const emit = jest.fn(); + gateway.server = { + to: jest.fn().mockReturnValue({ emit }), + } as any; + + gateway['userSocketMap'].set('user-1', new Set(['socket-1'])); + + gateway.broadcastEscrowStatusChanged('esc-1', { status: 'active' }); + gateway.broadcastNotification('user-1', { message: 'new message' }); + + expect(gateway.server.to).toHaveBeenCalledWith('escrow:esc-1'); + expect(emit).toHaveBeenCalledWith( + 'escrow.status_changed', + expect.objectContaining({ escrowId: 'esc-1' }), + ); + expect(emit).toHaveBeenCalledWith( + 'notification.new', + expect.objectContaining({ message: 'new message' }), + ); + }); +}); diff --git a/apps/backend/src/gateways/escrow.gateway.ts b/apps/backend/src/gateways/escrow.gateway.ts index 86581331..ab1e6d44 100644 --- a/apps/backend/src/gateways/escrow.gateway.ts +++ b/apps/backend/src/gateways/escrow.gateway.ts @@ -1,75 +1,117 @@ +import { Logger } from '@nestjs/common'; +import { JwtService } from '@nestjs/jwt'; import { - WebSocketGateway, - WebSocketServer, - SubscribeMessage, OnGatewayConnection, OnGatewayDisconnect, + SubscribeMessage, + WebSocketGateway, + WebSocketServer, } from '@nestjs/websockets'; import { Server, Socket } from 'socket.io'; -import { Logger } from '@nestjs/common'; -import { JwtService } from '@nestjs/jwt'; interface EscrowEventData { [key: string]: unknown; } +interface JwtPayload { + sub?: string; + userId?: string; + id?: string; +} + +interface ClientAuthPayload { + token?: string; +} + @WebSocketGateway({ + namespace: '/events', cors: { - origin: process.env.FRONTEND_URL || 'http://localhost:3001', + origin: (typeof process !== 'undefined' && + process.env.FRONTEND_URL?.split(',')) || ['http://localhost:3001'], credentials: true, }, - namespace: 'escrow', }) -export class EscrowGateway implements OnGatewayConnection, OnGatewayDisconnect { +export class EventsGateway implements OnGatewayConnection, OnGatewayDisconnect { @WebSocketServer() - server: Server; + server!: Server; - private readonly logger = new Logger(EscrowGateway.name); - private userSocketMap: Map = new Map(); // userId -> socketIds[] - private socketUserMap: Map = new Map(); // socketId -> userId - private socketEscrowMap: Map = new Map(); // socketId -> escrowIds[] + private readonly logger = new Logger(EventsGateway.name); + private readonly userSocketMap = new Map>(); + private readonly socketUserMap = new Map(); + private readonly socketEscrowMap = new Map>(); - constructor(private jwtService: JwtService) {} + constructor(private readonly jwtService: JwtService) {} - handleConnection(client: Socket) { + private extractToken(client: Socket): string | undefined { + const authPayload = client.handshake.auth as ClientAuthPayload | undefined; + const authorizationHeader = client.handshake.headers.authorization as + | string + | undefined; + + if (authPayload?.token) { + return authPayload.token; + } + + const parts = authorizationHeader?.split(' ') ?? []; + return parts[1]; + } + + async handleConnection(client: Socket): Promise { try { - // Extract token from handshake - const token = - (client.handshake.auth.token as string) || - (client.handshake.headers.authorization as string)?.split(' ')[1]; + const token = this.extractToken(client); if (!token) { this.logger.warn( - `Connection rejected: No token provided (${client.id})`, + `Connection rejected: no token provided (${client.id})`, ); client.disconnect(); return; } - // Verify JWT - const decoded: { sub?: string; userId?: string } = - this.jwtService.verify(token); - const userId: string | undefined = decoded?.sub || decoded?.userId; + let decoded: unknown; + try { + decoded = this.jwtService.verify(token); + } catch { + this.logger.warn(`Connection rejected: invalid token (${client.id})`); + client.disconnect(); + return; + } + + if ( + !decoded || + typeof decoded !== 'object' || + !('sub' in decoded || 'userId' in decoded || 'id' in decoded) + ) { + this.logger.warn(`Connection rejected: invalid token (${client.id})`); + client.disconnect(); + return; + } + + const payload = decoded as JwtPayload; + const userId = payload.sub || payload.userId || payload.id; if (!userId) { - this.logger.warn(`Connection rejected: Invalid token (${client.id})`); + this.logger.warn(`Connection rejected: invalid token (${client.id})`); client.disconnect(); return; } - // Store connection mapping this.socketUserMap.set(client.id, userId); - const userSockets: string[] = this.userSocketMap.get(userId) || []; - userSockets.push(client.id); - this.userSocketMap.set(userId, userSockets); + const socketsForUser = + this.userSocketMap.get(userId) || new Set(); + socketsForUser.add(client.id); + this.userSocketMap.set(userId, socketsForUser); + await client.join(`user:${userId}`); this.logger.log(`Client connected: ${client.id} (user: ${userId})`); - - // Send connection success - client.emit('connected', { userId, socketId: client.id }); + client.emit('connected', { + userId, + socketId: client.id, + namespace: '/events', + }); } catch (error: unknown) { this.logger.error( - `Connection rejected: Invalid token (${client.id})`, + `Connection rejected: invalid token (${client.id})`, error, ); client.disconnect(); @@ -78,19 +120,21 @@ export class EscrowGateway implements OnGatewayConnection, OnGatewayDisconnect { handleDisconnect(client: Socket): void { const userId = this.socketUserMap.get(client.id); + if (userId) { - // Remove from user mapping - const userSockets = this.userSocketMap.get(userId) || []; - const updatedSockets = userSockets.filter((id) => id !== client.id); - if (updatedSockets.length === 0) { - this.userSocketMap.delete(userId); - } else { - this.userSocketMap.set(userId, updatedSockets); + const socketsForUser = this.userSocketMap.get(userId); + if (socketsForUser) { + socketsForUser.delete(client.id); + if (socketsForUser.size === 0) { + this.userSocketMap.delete(userId); + } else { + this.userSocketMap.set(userId, socketsForUser); + } } - this.socketUserMap.delete(client.id); - // Clean up escroom subscriptions - const escrowIds = this.socketEscrowMap.get(client.id) || []; + this.socketUserMap.delete(client.id); + const escrowIds = + this.socketEscrowMap.get(client.id) || new Set(); escrowIds.forEach((escrowId) => { void client.leave(`escrow:${escrowId}`); }); @@ -101,140 +145,115 @@ export class EscrowGateway implements OnGatewayConnection, OnGatewayDisconnect { } @SubscribeMessage('joinEscrow') - handleJoinEscrow(client: Socket, escrowId: string): void { + async handleJoinEscrow(client: Socket, escrowId: string): Promise { + if (!escrowId) { + return; + } + const room = `escrow:${escrowId}`; - void client.join(room); + await client.join(room); - // Track subscription - const escrowIds: string[] = this.socketEscrowMap.get(client.id) || []; - if (!escrowIds.includes(escrowId)) { - escrowIds.push(escrowId); - this.socketEscrowMap.set(client.id, escrowIds); - } + const escrowIds = this.socketEscrowMap.get(client.id) || new Set(); + escrowIds.add(escrowId); + this.socketEscrowMap.set(client.id, escrowIds); this.logger.log(`Client ${client.id} joined escrow room: ${escrowId}`); client.emit('joinedEscrow', { escrowId }); } @SubscribeMessage('leaveEscrow') - handleLeaveEscrow(client: Socket, escrowId: string): void { - const room = `escrow:${escrowId}`; - void client.leave(room); + async handleLeaveEscrow(client: Socket, escrowId: string): Promise { + if (!escrowId) { + return; + } - // Remove from tracking - const escrowIds: string[] = this.socketEscrowMap.get(client.id) || []; - const updatedEscrowIds = escrowIds.filter((id) => id !== escrowId); - this.socketEscrowMap.set(client.id, updatedEscrowIds); + await client.leave(`escrow:${escrowId}`); + + const escrowIds = this.socketEscrowMap.get(client.id); + if (escrowIds) { + escrowIds.delete(escrowId); + if (escrowIds.size === 0) { + this.socketEscrowMap.delete(client.id); + } else { + this.socketEscrowMap.set(client.id, escrowIds); + } + } this.logger.log(`Client ${client.id} left escrow room: ${escrowId}`); } - // Broadcast methods - called from EscrowService + @SubscribeMessage('reconnect') + async handleReconnect( + client: Socket, + payload: { escrowIds?: string[] }, + ): Promise { + if (payload?.escrowIds?.length) { + for (const escrowId of payload.escrowIds) { + await this.handleJoinEscrow(client, escrowId); + } + } + + const userId = this.socketUserMap.get(client.id); + client.emit('reconnected', { userId, socketId: client.id }); + } + broadcastEscrowStatusChanged(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:status_changed', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + this.emitToEscrowRoom('escrow.status_changed', escrowId, data); } - broadcastMilestoneReleased(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:milestone_released', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + broadcastConditionFulfilled(escrowId: string, data: EscrowEventData): void { + this.emitToEscrowRoom('escrow.condition_fulfilled', escrowId, data); + } + + broadcastConditionConfirmed(escrowId: string, data: EscrowEventData): void { + this.emitToEscrowRoom('escrow.condition_confirmed', escrowId, data); } broadcastDisputeFiled(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:dispute_filed', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + this.emitToEscrowRoom('escrow.dispute_filed', escrowId, data); } broadcastDisputeResolved(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:dispute_resolved', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + this.emitToEscrowRoom('escrow.dispute_resolved', escrowId, data); } - broadcastPartyJoined(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:party_joined', { - escrowId, + broadcastNotification(userId: string, data: EscrowEventData): void { + const payload = { ...data, + userId, timestamp: new Date().toISOString(), - }); - } + }; - broadcastConditionFulfilled(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:condition_fulfilled', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + this.server.to(`user:${userId}`).emit('notification.new', payload); } - broadcastConditionConfirmed(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:condition_confirmed', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + getOnlineUsers(): Map> { + return this.userSocketMap; } - broadcastNotification(userId: string, data: EscrowEventData): void { - const socketIds = this.userSocketMap.get(userId) || []; - socketIds.forEach((socketId) => { - this.server.to(socketId).emit('notification:new', { - ...data, - timestamp: new Date().toISOString(), - }); - }); + getUserSockets(userId: string): string[] { + return Array.from(this.userSocketMap.get(userId) || []); } - broadcastEscrowFunded(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:funded', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + isUserOnline(userId: string): boolean { + return (this.userSocketMap.get(userId)?.size || 0) > 0; } - broadcastEscrowCompleted(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:completed', { - escrowId, - ...data, - timestamp: new Date().toISOString(), - }); + isHealthy(): boolean { + return this.server !== undefined; } - broadcastEscrowCancelled(escrowId: string, data: EscrowEventData): void { - this.server.to(`escrow:${escrowId}`).emit('escrow:cancelled', { + private emitToEscrowRoom( + eventName: string, + escrowId: string, + data: EscrowEventData, + ): void { + this.server.to(`escrow:${escrowId}`).emit(eventName, { escrowId, ...data, timestamp: new Date().toISOString(), }); } - - // Get online users (for admin/monitoring) - getOnlineUsers(): Map { - return this.userSocketMap; - } - - // Get user's socket IDs - getUserSockets(userId: string): string[] { - return this.userSocketMap.get(userId) || []; - } - - // Check if user is online - isUserOnline(userId: string): boolean { - const sockets = this.userSocketMap.get(userId) || []; - return sockets.length > 0; - } isHealthy(): boolean { return this.server !== undefined; } diff --git a/apps/backend/src/gateways/events.module.ts b/apps/backend/src/gateways/events.module.ts new file mode 100644 index 00000000..0d3173fe --- /dev/null +++ b/apps/backend/src/gateways/events.module.ts @@ -0,0 +1,23 @@ +import { Module } from '@nestjs/common'; +import { ConfigModule, ConfigService } from '@nestjs/config'; +import { JwtModule } from '@nestjs/jwt'; +import { EventsGateway } from './escrow.gateway'; + +@Module({ + imports: [ + ConfigModule, + JwtModule.registerAsync({ + imports: [ConfigModule], + useFactory: (configService: ConfigService) => ({ + secret: + configService.get('JWT_SECRET') || + 'your-secret-key-change-in-production', + signOptions: { expiresIn: '15m' }, + }), + inject: [ConfigService], + }), + ], + providers: [EventsGateway], + exports: [EventsGateway], +}) +export class EventsModule {} diff --git a/apps/backend/src/modules/escrow/escrow.module.ts b/apps/backend/src/modules/escrow/escrow.module.ts index d5f9f0e6..fc60e857 100644 --- a/apps/backend/src/modules/escrow/escrow.module.ts +++ b/apps/backend/src/modules/escrow/escrow.module.ts @@ -23,6 +23,7 @@ import { EscrowLifecycleService } from './escrow-lifecycle.service'; import { EscrowFundingService } from './escrow-funding.service'; import { EscrowDisputeService } from './escrow-dispute.service'; import { EscrowQueryService } from './escrow-query.service'; +import { EventsModule } from '../../gateways/events.module'; import { EscrowEvidenceService } from './services/escrow-evidence.service'; import { EscrowIpfsSyncService } from './services/escrow-ipfs-sync.service'; @@ -38,6 +39,7 @@ import { EscrowIpfsSyncService } from './services/escrow-ipfs-sync.service'; AllowedAsset, ]), AuthModule, + EventsModule, WebhookModule, IpfsModule, NotificationsModule, diff --git a/apps/backend/src/modules/escrow/services/escrow.service.ts b/apps/backend/src/modules/escrow/services/escrow.service.ts index 880d5daa..5cb93da2 100644 --- a/apps/backend/src/modules/escrow/services/escrow.service.ts +++ b/apps/backend/src/modules/escrow/services/escrow.service.ts @@ -5,6 +5,7 @@ import { ForbiddenException, ConflictException, UnprocessableEntityException, + Optional, } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { Brackets, Repository, SelectQueryBuilder } from 'typeorm'; @@ -44,6 +45,7 @@ import { IpfsService } from '../../ipfs/ipfs.service'; import { AllowedAsset } from '../../assets/entities/allowed-asset.entity'; import { NotificationService } from '../../../notifications/notifications.service'; import { NotificationEventType } from '../../../notifications/enums/notification-event.enum'; +import { EventsGateway } from '../../../gateways/escrow.gateway'; @Injectable() export class EscrowService { @@ -67,6 +69,7 @@ export class EscrowService { private readonly webhookService: WebhookService, private readonly ipfsService: IpfsService, private readonly notificationService: NotificationService, + @Optional() private readonly eventsGateway?: EventsGateway, ) {} async create( @@ -446,6 +449,11 @@ export class EscrowService { await this.webhookService.dispatchEvent('escrow.cancelled', { escrowId: id, }); + this.eventsGateway?.broadcastEscrowStatusChanged(id, { + previousStatus: escrow.status, + newStatus: EscrowStatus.CANCELLED, + actorId: userId, + }); return this.findOne(id); } @@ -531,6 +539,7 @@ export class EscrowService { ); const fundedAt = new Date(); + const previousStatus = escrow.status; await this.escrowRepository.update(id, { stellarTxHash, fundedAt, @@ -548,6 +557,11 @@ export class EscrowService { escrowId: id, stellarTxHash, }); + this.eventsGateway?.broadcastEscrowStatusChanged(id, { + previousStatus, + newStatus: EscrowStatus.ACTIVE, + actorId: userId, + }); return this.findOne(id); } @@ -626,6 +640,7 @@ export class EscrowService { escrow.creatorId, ); + const previousStatus = escrow.status; escrow.status = EscrowStatus.COMPLETED; escrow.isReleased = true; escrow.releaseTransactionHash = txHash; @@ -639,6 +654,11 @@ export class EscrowService { escrowId: escrow.id, txHash, }); + this.eventsGateway?.broadcastEscrowStatusChanged(escrow.id, { + previousStatus, + newStatus: EscrowStatus.COMPLETED, + actorId: currentUserId, + }); return escrow; } @@ -721,6 +741,10 @@ export class EscrowService { conditionId, fulfilledBy: userId, }); + this.eventsGateway?.broadcastConditionFulfilled(escrowId, { + conditionId, + fulfilledBy: userId, + }); return condition; } @@ -819,6 +843,11 @@ export class EscrowService { confirmedBy: userId, allConditionsMet, }); + this.eventsGateway?.broadcastConditionConfirmed(escrowId, { + conditionId, + confirmedBy: userId, + allConditionsMet, + }); return condition; } @@ -970,6 +999,15 @@ export class EscrowService { escrowId, disputeId: savedDispute.id, }); + this.eventsGateway?.broadcastDisputeFiled(escrowId, { + disputeId: savedDispute.id, + filedBy: userId, + }); + this.eventsGateway?.broadcastEscrowStatusChanged(escrowId, { + previousStatus: escrow.status, + newStatus: EscrowStatus.DISPUTED, + actorId: userId, + }); return this.disputeRepository.findOne({ where: { id: savedDispute.id }, @@ -1071,6 +1109,16 @@ export class EscrowService { disputeId: resolved.id, outcome: dto.outcome, }); + this.eventsGateway?.broadcastDisputeResolved(escrowId, { + disputeId: resolved.id, + outcome: dto.outcome, + resolvedBy: arbitratorUserId, + }); + this.eventsGateway?.broadcastEscrowStatusChanged(escrowId, { + previousStatus: escrow.status, + newStatus: nextEscrowStatus, + actorId: arbitratorUserId, + }); return this.disputeRepository.findOne({ where: { id: resolved.id }, @@ -1411,6 +1459,12 @@ export class EscrowService { escrowId: escrow.id, reason: options.webhookReason, }); + this.eventsGateway?.broadcastEscrowStatusChanged(escrow.id, { + previousStatus: escrow.status, + newStatus: EscrowStatus.EXPIRED, + actorId: options.actorId, + reason: options.reason, + }); return this.findOne(escrow.id); } diff --git a/apps/backend/src/notifications/notifications.module.ts b/apps/backend/src/notifications/notifications.module.ts index 58852261..d6bafda8 100644 --- a/apps/backend/src/notifications/notifications.module.ts +++ b/apps/backend/src/notifications/notifications.module.ts @@ -9,11 +9,13 @@ import { NotificationService } from './notifications.service'; import { PreferenceService } from './preference.service'; import { EmailSender } from './senders/email.sender'; import { WebhookSender } from './senders/webhook.sender'; +import { EventsModule } from '../gateways/events.module'; @Module({ imports: [ ConfigModule, AuthModule, + EventsModule, TypeOrmModule.forFeature([Notification, NotificationPreference]), ], controllers: [NotificationController], diff --git a/apps/backend/src/notifications/notifications.service.ts b/apps/backend/src/notifications/notifications.service.ts index 14777293..0cbc888b 100644 --- a/apps/backend/src/notifications/notifications.service.ts +++ b/apps/backend/src/notifications/notifications.service.ts @@ -1,4 +1,4 @@ -import { Injectable, Logger } from '@nestjs/common'; +import { Injectable, Logger, Optional } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; import { NotificationChannel, @@ -12,6 +12,7 @@ import { WebhookSender } from './senders/webhook.sender'; import { Repository, IsNull } from 'typeorm'; import { EmailSender } from './senders/email.sender'; import { PreferenceService } from './preference.service'; +import { EventsGateway } from '../gateways/escrow.gateway'; @Injectable() export class NotificationService { @@ -24,6 +25,7 @@ export class NotificationService { private preferenceService: PreferenceService, emailSender: EmailSender, webhookSender: WebhookSender, + @Optional() private readonly eventsGateway?: EventsGateway, ) { this.senders = new Map([ [NotificationChannel.EMAIL, emailSender], @@ -42,7 +44,7 @@ export class NotificationService { if (!pref.enabled) continue; if (!pref.eventTypes.includes(eventType)) continue; - await this.repo.save( + const notification = await this.repo.save( this.repo.create({ userId, eventType, @@ -51,6 +53,12 @@ export class NotificationService { status: NotificationStatus.PENDING, }), ); + + this.eventsGateway?.broadcastNotification(userId, { + notificationId: notification.id, + eventType, + payload, + }); } } diff --git a/apps/backend/tsconfig.json b/apps/backend/tsconfig.json index 286103e8..7c457493 100644 --- a/apps/backend/tsconfig.json +++ b/apps/backend/tsconfig.json @@ -20,6 +20,7 @@ "noImplicitAny": false, "strictBindCallApply": false, "noFallthroughCasesInSwitch": false, + "types": ["node"], "lib": ["es2021"] } } diff --git a/apps/onchain/src/lib.rs b/apps/onchain/src/lib.rs index 4c9aaec4..4bd5a85a 100644 --- a/apps/onchain/src/lib.rs +++ b/apps/onchain/src/lib.rs @@ -398,6 +398,12 @@ pub enum Error { const DEFAULT_FEE_BPS: i128 = 50; const BPS_DENOMINATOR: i128 = 10000; +const FEE_TIERS: &[(i128, i128)] = &[ + (1000, 50), // 0-1,000 XLM => 50 bps + (5000, 30), // 1,001-5,000 XLM => 30 bps + (10000, 20), // 5,001-10,000 XLM => 20 bps + (i128::MAX, 10), // 10,001+ XLM => 10 bps +]; const MAX_BATCH_SIZE: u32 = 20; const ESCROW_ENTRY_STORAGE_VERSION: i128 = 2; const EVENT_NAMESPACE: &str = "Vaultix"; @@ -1690,10 +1696,11 @@ impl VaultixEscrow { if escrow_status(&escrow) == EscrowStatus::Active { let token_client = token::Client::new(&env, &escrow.token_address); refund_amount = if let Ok((treasury, _)) = Self::get_config(env.clone()) { - let fee_bps = resolve_fee_with_escrow_override( + let fee_bps = resolve_effective_fee_bps( &env, &escrow.token_address, escrow_fee_override_opt(&escrow), + escrow.total_amount, )?; fee_amount = calculate_fee(escrow.total_amount, fee_bps)?; if fee_amount > 0 { @@ -1812,10 +1819,11 @@ impl VaultixEscrow { let (treasury, _) = Self::get_config(env.clone())?; // Resolve fee with precedence: escrow > token > global - let fee_bps = resolve_fee_with_escrow_override( + let fee_bps = resolve_effective_fee_bps( &env, &escrow.token_address, escrow_fee_override_opt(&escrow), + escrow.total_amount, )?; // Calculate platform fee using checked arithmetic @@ -1963,6 +1971,42 @@ fn resolve_fee_with_escrow_override( Ok(global_fee) } +fn calculate_tiered_fee_bps(volume: i128) -> i128 { + if volume <= 0 { + return DEFAULT_FEE_BPS; + } + + for &(cap, bps) in FEE_TIERS { + if volume <= cap { + return bps; + } + } + + DEFAULT_FEE_BPS +} + +fn resolve_effective_fee_bps( + env: &Env, + token_address: &Address, + escrow_fee_override: Option, + volume: i128, +) -> Result { + if let Some(escrow_fee) = escrow_fee_override { + return Ok(escrow_fee); + } + + let token_fee_key = get_token_fee_key(token_address); + if let Some(token_fee) = env + .storage() + .persistent() + .get::<(Symbol, Address), i128>(&token_fee_key) + { + return Ok(token_fee); + } + + Ok(calculate_tiered_fee_bps(volume)) +} + fn release_pending_milestone( env: &Env, escrow: &mut EscrowEntryV2, @@ -1981,10 +2025,11 @@ fn release_pending_milestone( } let (treasury, _) = VaultixEscrow::get_config(env.clone())?; - let fee_bps = resolve_fee_with_escrow_override( + let fee_bps = resolve_effective_fee_bps( env, &escrow.token_address, escrow_fee_override_opt(escrow), + escrow.total_amount, )?; let fee_amount = calculate_fee(milestone.amount, fee_bps)?; let payout_amount = milestone diff --git a/gitdiff_lastparts.txt b/gitdiff_lastparts.txt new file mode 100644 index 00000000..eeac382a --- /dev/null +++ b/gitdiff_lastparts.txt @@ -0,0 +1,221 @@ +commit 9a3434cb656bd0c304e2cd8e6c891917aa82a2a4 +Author: Brite +Date: Mon Jun 29 14:13:43 2026 +0100 + + I have made some changes to file + +diff --git a/apps/backend/src/modules/escrow/services/escrow.service.ts b/apps/backend/src/modules/escrow/services/escrow.service.ts +index 880d5da..5cb93da 100644 +--- a/apps/backend/src/modules/escrow/services/escrow.service.ts ++++ b/apps/backend/src/modules/escrow/services/escrow.service.ts +@@ -5,6 +5,7 @@ import { + ForbiddenException, + ConflictException, + UnprocessableEntityException, ++ Optional, + } from '@nestjs/common'; + import { InjectRepository } from '@nestjs/typeorm'; + import { Brackets, Repository, SelectQueryBuilder } from 'typeorm'; +@@ -44,6 +45,7 @@ import { IpfsService } from '../../ipfs/ipfs.service'; + import { AllowedAsset } from '../../assets/entities/allowed-asset.entity'; + import { NotificationService } from '../../../notifications/notifications.service'; + import { NotificationEventType } from '../../../notifications/enums/notification-event.enum'; ++import { EventsGateway } from '../../../gateways/escrow.gateway'; + + @Injectable() + export class EscrowService { +@@ -67,6 +69,7 @@ export class EscrowService { + private readonly webhookService: WebhookService, + private readonly ipfsService: IpfsService, + private readonly notificationService: NotificationService, ++ @Optional() private readonly eventsGateway?: EventsGateway, + ) {} + + async create( +@@ -446,6 +449,11 @@ export class EscrowService { + await this.webhookService.dispatchEvent('escrow.cancelled', { + escrowId: id, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(id, { ++ previousStatus: escrow.status, ++ newStatus: EscrowStatus.CANCELLED, ++ actorId: userId, ++ }); + + return this.findOne(id); + } +@@ -531,6 +539,7 @@ export class EscrowService { + ); + + const fundedAt = new Date(); ++ const previousStatus = escrow.status; + await this.escrowRepository.update(id, { + stellarTxHash, + fundedAt, +@@ -548,6 +557,11 @@ export class EscrowService { + escrowId: id, + stellarTxHash, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(id, { ++ previousStatus, ++ newStatus: EscrowStatus.ACTIVE, ++ actorId: userId, ++ }); + + return this.findOne(id); + } +@@ -626,6 +640,7 @@ export class EscrowService { + escrow.creatorId, + ); + ++ const previousStatus = escrow.status; + escrow.status = EscrowStatus.COMPLETED; + escrow.isReleased = true; + escrow.releaseTransactionHash = txHash; +@@ -639,6 +654,11 @@ export class EscrowService { + escrowId: escrow.id, + txHash, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrow.id, { ++ previousStatus, ++ newStatus: EscrowStatus.COMPLETED, ++ actorId: currentUserId, ++ }); + + return escrow; + } +@@ -721,6 +741,10 @@ export class EscrowService { + conditionId, + fulfilledBy: userId, + }); ++ this.eventsGateway?.broadcastConditionFulfilled(escrowId, { ++ conditionId, ++ fulfilledBy: userId, ++ }); + + return condition; + } +@@ -819,6 +843,11 @@ export class EscrowService { + confirmedBy: userId, + allConditionsMet, + }); ++ this.eventsGateway?.broadcastConditionConfirmed(escrowId, { ++ conditionId, ++ confirmedBy: userId, ++ allConditionsMet, ++ }); + + return condition; + } +@@ -970,6 +999,15 @@ export class EscrowService { + escrowId, + disputeId: savedDispute.id, + }); ++ this.eventsGateway?.broadcastDisputeFiled(escrowId, { ++ disputeId: savedDispute.id, ++ filedBy: userId, ++ }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrowId, { ++ previousStatus: escrow.status, ++ newStatus: EscrowStatus.DISPUTED, ++ actorId: userId, ++ }); + + return this.disputeRepository.findOne({ + where: { id: savedDispute.id }, +@@ -1071,6 +1109,16 @@ export class EscrowService { + disputeId: resolved.id, + outcome: dto.outcome, + }); ++ this.eventsGateway?.broadcastDisputeResolved(escrowId, { ++ disputeId: resolved.id, ++ outcome: dto.outcome, ++ resolvedBy: arbitratorUserId, ++ }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrowId, { ++ previousStatus: escrow.status, ++ newStatus: nextEscrowStatus, ++ actorId: arbitratorUserId, ++ }); + + return this.disputeRepository.findOne({ + where: { id: resolved.id }, +@@ -1411,6 +1459,12 @@ export class EscrowService { + escrowId: escrow.id, + reason: options.webhookReason, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrow.id, { ++ previousStatus: escrow.status, ++ newStatus: EscrowStatus.EXPIRED, ++ actorId: options.actorId, ++ reason: options.reason, ++ }); + + return this.findOne(escrow.id); + } +diff --git a/apps/backend/src/notifications/notifications.module.ts b/apps/backend/src/notifications/notifications.module.ts +index 5885226..d6bafda 100644 +--- a/apps/backend/src/notifications/notifications.module.ts ++++ b/apps/backend/src/notifications/notifications.module.ts +@@ -9,11 +9,13 @@ import { NotificationService } from './notifications.service'; + import { PreferenceService } from './preference.service'; + import { EmailSender } from './senders/email.sender'; + import { WebhookSender } from './senders/webhook.sender'; ++import { EventsModule } from '../gateways/events.module'; + + @Module({ + imports: [ + ConfigModule, + AuthModule, ++ EventsModule, + TypeOrmModule.forFeature([Notification, NotificationPreference]), + ], + controllers: [NotificationController], +diff --git a/apps/backend/src/notifications/notifications.service.ts b/apps/backend/src/notifications/notifications.service.ts +index 1477729..0cbc888 100644 +--- a/apps/backend/src/notifications/notifications.service.ts ++++ b/apps/backend/src/notifications/notifications.service.ts +@@ -1,4 +1,4 @@ +-import { Injectable, Logger } from '@nestjs/common'; ++import { Injectable, Logger, Optional } from '@nestjs/common'; + import { Cron, CronExpression } from '@nestjs/schedule'; + import { + NotificationChannel, +@@ -12,6 +12,7 @@ import { WebhookSender } from './senders/webhook.sender'; + import { Repository, IsNull } from 'typeorm'; + import { EmailSender } from './senders/email.sender'; + import { PreferenceService } from './preference.service'; ++import { EventsGateway } from '../gateways/escrow.gateway'; + + @Injectable() + export class NotificationService { +@@ -24,6 +25,7 @@ export class NotificationService { + private preferenceService: PreferenceService, + emailSender: EmailSender, + webhookSender: WebhookSender, ++ @Optional() private readonly eventsGateway?: EventsGateway, + ) { + this.senders = new Map([ + [NotificationChannel.EMAIL, emailSender], +@@ -42,7 +44,7 @@ export class NotificationService { + if (!pref.enabled) continue; + if (!pref.eventTypes.includes(eventType)) continue; + +- await this.repo.save( ++ const notification = await this.repo.save( + this.repo.create({ + userId, + eventType, +@@ -51,6 +53,12 @@ export class NotificationService { + status: NotificationStatus.PENDING, + }), + ); ++ ++ this.eventsGateway?.broadcastNotification(userId, { ++ notificationId: notification.id, ++ eventType, ++ payload, ++ }); + } + } + diff --git a/gitdiff_remaining.txt b/gitdiff_remaining.txt new file mode 100644 index 00000000..d62eea77 --- /dev/null +++ b/gitdiff_remaining.txt @@ -0,0 +1,451 @@ +- if (!escrowIds.includes(escrowId)) { +- escrowIds.push(escrowId); +- this.socketEscrowMap.set(client.id, escrowIds); +- } ++ const escrowIds = this.socketEscrowMap.get(client.id) || new Set(); ++ escrowIds.add(escrowId); ++ this.socketEscrowMap.set(client.id, escrowIds); + + this.logger.log(`Client ${client.id} joined escrow room: ${escrowId}`); + client.emit('joinedEscrow', { escrowId }); + } + + @SubscribeMessage('leaveEscrow') +- handleLeaveEscrow(client: Socket, escrowId: string): void { +- const room = `escrow:${escrowId}`; +- void client.leave(room); +- +- // Remove from tracking +- const escrowIds: string[] = this.socketEscrowMap.get(client.id) || []; +- const updatedEscrowIds = escrowIds.filter((id) => id !== escrowId); +- this.socketEscrowMap.set(client.id, updatedEscrowIds); ++ async handleLeaveEscrow(client: Socket, escrowId: string): Promise { ++ if (!escrowId) { ++ return; ++ } + +- this.logger.log(`Client ${client.id} left escrow room: ${escrowId}`); +- } ++ await client.leave(`escrow:${escrowId}`); + +- // Broadcast methods - called from EscrowService +- broadcastEscrowStatusChanged(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:status_changed', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); +- } ++ const escrowIds = this.socketEscrowMap.get(client.id); ++ if (escrowIds) { ++ escrowIds.delete(escrowId); ++ if (escrowIds.size === 0) { ++ this.socketEscrowMap.delete(client.id); ++ } else { ++ this.socketEscrowMap.set(client.id, escrowIds); ++ } ++ } + +- broadcastMilestoneReleased(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:milestone_released', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ this.logger.log(`Client ${client.id} left escrow room: ${escrowId}`); + } + +- broadcastDisputeFiled(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:dispute_filed', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); +- } ++ @SubscribeMessage('reconnect') ++ async handleReconnect( ++ client: Socket, ++ payload: { escrowIds?: string[] }, ++ ): Promise { ++ if (payload?.escrowIds?.length) { ++ for (const escrowId of payload.escrowIds) { ++ await this.handleJoinEscrow(client, escrowId); ++ } ++ } + +- broadcastDisputeResolved(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:dispute_resolved', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ const userId = this.socketUserMap.get(client.id); ++ client.emit('reconnected', { userId, socketId: client.id }); + } + +- broadcastPartyJoined(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:party_joined', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ broadcastEscrowStatusChanged(escrowId: string, data: EscrowEventData): void { ++ this.emitToEscrowRoom('escrow.status_changed', escrowId, data); + } + + broadcastConditionFulfilled(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:condition_fulfilled', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ this.emitToEscrowRoom('escrow.condition_fulfilled', escrowId, data); + } + + broadcastConditionConfirmed(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:condition_confirmed', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ this.emitToEscrowRoom('escrow.condition_confirmed', escrowId, data); + } + +- broadcastNotification(userId: string, data: EscrowEventData): void { +- const socketIds = this.userSocketMap.get(userId) || []; +- socketIds.forEach((socketId) => { +- this.server.to(socketId).emit('notification:new', { +- ...data, +- timestamp: new Date().toISOString(), +- }); +- }); ++ broadcastDisputeFiled(escrowId: string, data: EscrowEventData): void { ++ this.emitToEscrowRoom('escrow.dispute_filed', escrowId, data); + } + +- broadcastEscrowFunded(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:funded', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ broadcastDisputeResolved(escrowId: string, data: EscrowEventData): void { ++ this.emitToEscrowRoom('escrow.dispute_resolved', escrowId, data); + } + +- broadcastEscrowCompleted(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:completed', { +- escrowId, ++ broadcastNotification(userId: string, data: EscrowEventData): void { ++ const payload = { + ...data, ++ userId, + timestamp: new Date().toISOString(), +- }); +- } ++ }; + +- broadcastEscrowCancelled(escrowId: string, data: EscrowEventData): void { +- this.server.to(`escrow:${escrowId}`).emit('escrow:cancelled', { +- escrowId, +- ...data, +- timestamp: new Date().toISOString(), +- }); ++ this.server.to(`user:${userId}`).emit('notification.new', payload); + } + +- // Get online users (for admin/monitoring) +- getOnlineUsers(): Map { ++ getOnlineUsers(): Map> { + return this.userSocketMap; + } + +- // Get user's socket IDs + getUserSockets(userId: string): string[] { +- return this.userSocketMap.get(userId) || []; ++ return Array.from(this.userSocketMap.get(userId) || []); + } + +- // Check if user is online + isUserOnline(userId: string): boolean { +- const sockets = this.userSocketMap.get(userId) || []; +- return sockets.length > 0; ++ return (this.userSocketMap.get(userId)?.size || 0) > 0; ++ } ++ ++ private emitToEscrowRoom( ++ eventName: string, ++ escrowId: string, ++ data: EscrowEventData, ++ ): void { ++ this.server.to(`escrow:${escrowId}`).emit(eventName, { ++ escrowId, ++ ...data, ++ timestamp: new Date().toISOString(), ++ }); + } + } +diff --git a/apps/backend/src/gateways/events.module.ts b/apps/backend/src/gateways/events.module.ts +new file mode 100644 +index 0000000..0d3173f +--- /dev/null ++++ b/apps/backend/src/gateways/events.module.ts +@@ -0,0 +1,23 @@ ++import { Module } from '@nestjs/common'; ++import { ConfigModule, ConfigService } from '@nestjs/config'; ++import { JwtModule } from '@nestjs/jwt'; ++import { EventsGateway } from './escrow.gateway'; ++ ++@Module({ ++ imports: [ ++ ConfigModule, ++ JwtModule.registerAsync({ ++ imports: [ConfigModule], ++ useFactory: (configService: ConfigService) => ({ ++ secret: ++ configService.get('JWT_SECRET') || ++ 'your-secret-key-change-in-production', ++ signOptions: { expiresIn: '15m' }, ++ }), ++ inject: [ConfigService], ++ }), ++ ], ++ providers: [EventsGateway], ++ exports: [EventsGateway], ++}) ++export class EventsModule {} +diff --git a/apps/backend/src/modules/escrow/escrow.module.ts b/apps/backend/src/modules/escrow/escrow.module.ts +index d082e45..fd4e00a 100644 +--- a/apps/backend/src/modules/escrow/escrow.module.ts ++++ b/apps/backend/src/modules/escrow/escrow.module.ts +@@ -23,6 +23,7 @@ import { EscrowLifecycleService } from './escrow-lifecycle.service'; + import { EscrowFundingService } from './escrow-funding.service'; + import { EscrowDisputeService } from './escrow-dispute.service'; + import { EscrowQueryService } from './escrow-query.service'; ++import { EventsModule } from '../../gateways/events.module'; + + @Module({ + imports: [ +@@ -36,6 +37,7 @@ import { EscrowQueryService } from './escrow-query.service'; + AllowedAsset, + ]), + AuthModule, ++ EventsModule, + WebhookModule, + IpfsModule, + NotificationsModule, +diff --git a/apps/backend/src/modules/escrow/services/escrow.service.ts b/apps/backend/src/modules/escrow/services/escrow.service.ts +index 880d5da..5cb93da 100644 +--- a/apps/backend/src/modules/escrow/services/escrow.service.ts ++++ b/apps/backend/src/modules/escrow/services/escrow.service.ts +@@ -5,6 +5,7 @@ import { + ForbiddenException, + ConflictException, + UnprocessableEntityException, ++ Optional, + } from '@nestjs/common'; + import { InjectRepository } from '@nestjs/typeorm'; + import { Brackets, Repository, SelectQueryBuilder } from 'typeorm'; +@@ -44,6 +45,7 @@ import { IpfsService } from '../../ipfs/ipfs.service'; + import { AllowedAsset } from '../../assets/entities/allowed-asset.entity'; + import { NotificationService } from '../../../notifications/notifications.service'; + import { NotificationEventType } from '../../../notifications/enums/notification-event.enum'; ++import { EventsGateway } from '../../../gateways/escrow.gateway'; + + @Injectable() + export class EscrowService { +@@ -67,6 +69,7 @@ export class EscrowService { + private readonly webhookService: WebhookService, + private readonly ipfsService: IpfsService, + private readonly notificationService: NotificationService, ++ @Optional() private readonly eventsGateway?: EventsGateway, + ) {} + + async create( +@@ -446,6 +449,11 @@ export class EscrowService { + await this.webhookService.dispatchEvent('escrow.cancelled', { + escrowId: id, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(id, { ++ previousStatus: escrow.status, ++ newStatus: EscrowStatus.CANCELLED, ++ actorId: userId, ++ }); + + return this.findOne(id); + } +@@ -531,6 +539,7 @@ export class EscrowService { + ); + + const fundedAt = new Date(); ++ const previousStatus = escrow.status; + await this.escrowRepository.update(id, { + stellarTxHash, + fundedAt, +@@ -548,6 +557,11 @@ export class EscrowService { + escrowId: id, + stellarTxHash, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(id, { ++ previousStatus, ++ newStatus: EscrowStatus.ACTIVE, ++ actorId: userId, ++ }); + + return this.findOne(id); + } +@@ -626,6 +640,7 @@ export class EscrowService { + escrow.creatorId, + ); + ++ const previousStatus = escrow.status; + escrow.status = EscrowStatus.COMPLETED; + escrow.isReleased = true; + escrow.releaseTransactionHash = txHash; +@@ -639,6 +654,11 @@ export class EscrowService { + escrowId: escrow.id, + txHash, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrow.id, { ++ previousStatus, ++ newStatus: EscrowStatus.COMPLETED, ++ actorId: currentUserId, ++ }); + + return escrow; + } +@@ -721,6 +741,10 @@ export class EscrowService { + conditionId, + fulfilledBy: userId, + }); ++ this.eventsGateway?.broadcastConditionFulfilled(escrowId, { ++ conditionId, ++ fulfilledBy: userId, ++ }); + + return condition; + } +@@ -819,6 +843,11 @@ export class EscrowService { + confirmedBy: userId, + allConditionsMet, + }); ++ this.eventsGateway?.broadcastConditionConfirmed(escrowId, { ++ conditionId, ++ confirmedBy: userId, ++ allConditionsMet, ++ }); + + return condition; + } +@@ -970,6 +999,15 @@ export class EscrowService { + escrowId, + disputeId: savedDispute.id, + }); ++ this.eventsGateway?.broadcastDisputeFiled(escrowId, { ++ disputeId: savedDispute.id, ++ filedBy: userId, ++ }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrowId, { ++ previousStatus: escrow.status, ++ newStatus: EscrowStatus.DISPUTED, ++ actorId: userId, ++ }); + + return this.disputeRepository.findOne({ + where: { id: savedDispute.id }, +@@ -1071,6 +1109,16 @@ export class EscrowService { + disputeId: resolved.id, + outcome: dto.outcome, + }); ++ this.eventsGateway?.broadcastDisputeResolved(escrowId, { ++ disputeId: resolved.id, ++ outcome: dto.outcome, ++ resolvedBy: arbitratorUserId, ++ }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrowId, { ++ previousStatus: escrow.status, ++ newStatus: nextEscrowStatus, ++ actorId: arbitratorUserId, ++ }); + + return this.disputeRepository.findOne({ + where: { id: resolved.id }, +@@ -1411,6 +1459,12 @@ export class EscrowService { + escrowId: escrow.id, + reason: options.webhookReason, + }); ++ this.eventsGateway?.broadcastEscrowStatusChanged(escrow.id, { ++ previousStatus: escrow.status, ++ newStatus: EscrowStatus.EXPIRED, ++ actorId: options.actorId, ++ reason: options.reason, ++ }); + + return this.findOne(escrow.id); + } +diff --git a/apps/backend/src/notifications/notifications.module.ts b/apps/backend/src/notifications/notifications.module.ts +index 5885226..d6bafda 100644 +--- a/apps/backend/src/notifications/notifications.module.ts ++++ b/apps/backend/src/notifications/notifications.module.ts +@@ -9,11 +9,13 @@ import { NotificationService } from './notifications.service'; + import { PreferenceService } from './preference.service'; + import { EmailSender } from './senders/email.sender'; + import { WebhookSender } from './senders/webhook.sender'; ++import { EventsModule } from '../gateways/events.module'; + + @Module({ + imports: [ + ConfigModule, + AuthModule, ++ EventsModule, + TypeOrmModule.forFeature([Notification, NotificationPreference]), + ], + controllers: [NotificationController], +diff --git a/apps/backend/src/notifications/notifications.service.ts b/apps/backend/src/notifications/notifications.service.ts +index 1477729..0cbc888 100644 +--- a/apps/backend/src/notifications/notifications.service.ts ++++ b/apps/backend/src/notifications/notifications.service.ts +@@ -1,4 +1,4 @@ +-import { Injectable, Logger } from '@nestjs/common'; ++import { Injectable, Logger, Optional } from '@nestjs/common'; + import { Cron, CronExpression } from '@nestjs/schedule'; + import { + NotificationChannel, +@@ -12,6 +12,7 @@ import { WebhookSender } from './senders/webhook.sender'; + import { Repository, IsNull } from 'typeorm'; + import { EmailSender } from './senders/email.sender'; + import { PreferenceService } from './preference.service'; ++import { EventsGateway } from '../gateways/escrow.gateway'; + + @Injectable() + export class NotificationService { +@@ -24,6 +25,7 @@ export class NotificationService { + private preferenceService: PreferenceService, + emailSender: EmailSender, + webhookSender: WebhookSender, ++ @Optional() private readonly eventsGateway?: EventsGateway, + ) { + this.senders = new Map([ + [NotificationChannel.EMAIL, emailSender], +@@ -42,7 +44,7 @@ export class NotificationService { + if (!pref.enabled) continue; + if (!pref.eventTypes.includes(eventType)) continue; + +- await this.repo.save( ++ const notification = await this.repo.save( + this.repo.create({ + userId, + eventType, +@@ -51,6 +53,12 @@ export class NotificationService { + status: NotificationStatus.PENDING, + }), + ); ++ ++ this.eventsGateway?.broadcastNotification(userId, { ++ notificationId: notification.id, ++ eventType, ++ payload, ++ }); + } + } +