1+ import { ISubscriptionClient , User , Notification } from '../interfaces/subscription-client.interface' ;
2+ import { NotificationClientFactory } from './notification/notification-factory.service' ;
3+
4+ /**
5+ * Service for handling batch notification processing
6+ * Provides reusable batch operations for sending notifications efficiently
7+ */
8+ export class BatchNotificationService {
9+ constructor (
10+ private readonly subscriptionClient : ISubscriptionClient ,
11+ private readonly notificationFactory : NotificationClientFactory
12+ ) { }
13+
14+ /**
15+ * Prepares batch data by fetching followers and applying deduplication
16+ * @param addresses - Addresses to process (e.g., non-voting addresses)
17+ * @param daoId - DAO identifier
18+ * @param eventIdGenerator - Function to generate unique event IDs for each address
19+ * @returns Prepared batch data with followers and notifications to send
20+ */
21+ async prepareBatchData (
22+ addresses : string [ ] ,
23+ daoId : string ,
24+ eventIdGenerator : ( address : string ) => string
25+ ) : Promise < BatchNotificationData [ ] > {
26+ // Batch fetch all followers for all addresses in a single request
27+ const addressFollowersMap = await this . subscriptionClient . getWalletOwnersBatch ( addresses ) ;
28+ const addressFollowers = addresses
29+ . map ( address => ( { address, followers : addressFollowersMap [ address ] || [ ] } ) )
30+ . filter ( af => af . followers . length > 0 ) ;
31+
32+ if ( addressFollowers . length === 0 ) {
33+ return [ ] ;
34+ }
35+
36+ // Batch check deduplication for all addresses in one request
37+ const shouldSendRequests = addressFollowers . map ( ( { address, followers } ) => ( {
38+ subscribers : followers ,
39+ eventId : eventIdGenerator ( address ) ,
40+ daoId
41+ } ) ) ;
42+
43+ const batchResults = await this . subscriptionClient . shouldSendBatch ( shouldSendRequests ) ;
44+
45+ // Map results back to the original structure
46+ return addressFollowers . map ( ( { address, followers } , index ) => ( {
47+ address,
48+ followers,
49+ notificationsToSend : batchResults [ index ] || [ ]
50+ } ) ) ;
51+ }
52+
53+ /**
54+ * Filters batch data to only include items with notifications to send
55+ * @param batchData - Batch data from prepareBatchData
56+ * @returns Filtered data with valid notifications
57+ */
58+ filterValidNotifications ( batchData : BatchNotificationData [ ] ) : BatchNotificationData [ ] {
59+ return batchData . filter ( result => result . notificationsToSend . length > 0 ) ;
60+ }
61+
62+ /**
63+ * Executes parallel sending of notifications and marks them as sent
64+ * @param validNotifications - Notifications ready to be sent
65+ * @param messageGenerator - Function to generate message for each address
66+ * @param metadataGenerator - Optional function to generate metadata for each notification
67+ */
68+ async executeBatchSend (
69+ validNotifications : BatchNotificationData [ ] ,
70+ messageGenerator : ( address : string ) => string ,
71+ metadataGenerator ?: ( address : string ) => Record < string , any >
72+ ) : Promise < void > {
73+ const sendPromises : Promise < void > [ ] = [ ] ;
74+ const allNotificationsToMark : Notification [ ] = [ ] ;
75+
76+ for ( const { address, followers, notificationsToSend } of validNotifications ) {
77+ const followerMap = new Map ( followers . map ( ( f : User ) => [ f . id , f ] ) ) ;
78+ const message = messageGenerator ( address ) ;
79+ const metadata = metadataGenerator ? metadataGenerator ( address ) : undefined ;
80+
81+ this . queueNotificationSends (
82+ notificationsToSend ,
83+ followerMap ,
84+ message ,
85+ metadata ,
86+ sendPromises
87+ ) ;
88+
89+ allNotificationsToMark . push ( ...notificationsToSend ) ;
90+ }
91+
92+ // Execute all sends in parallel and mark as sent
93+ await Promise . all ( [
94+ Promise . all ( sendPromises ) ,
95+ this . subscriptionClient . markAsSent ( allNotificationsToMark )
96+ ] ) ;
97+ }
98+
99+ /**
100+ * Queues individual notification sends for parallel execution
101+ * @param notificationsToSend - Notifications to queue
102+ * @param followerMap - Map of follower IDs to follower data
103+ * @param message - Formatted notification message
104+ * @param metadata - Optional metadata for the notification
105+ * @param sendPromises - Array to collect send promises
106+ */
107+ private queueNotificationSends (
108+ notificationsToSend : Notification [ ] ,
109+ followerMap : Map < string , User > ,
110+ message : string ,
111+ metadata : Record < string , any > | undefined ,
112+ sendPromises : Promise < void > [ ]
113+ ) : void {
114+ for ( const notification of notificationsToSend ) {
115+ const follower = followerMap . get ( notification . user_id ) ;
116+ if ( ! follower ) continue ;
117+
118+ sendPromises . push (
119+ this . notificationFactory
120+ . getClient ( follower . channel )
121+ . sendNotification ( {
122+ userId : follower . id ,
123+ channel : follower . channel ,
124+ channelUserId : follower . channel_user_id ,
125+ message,
126+ metadata
127+ } )
128+ . catch ( error => {
129+ console . error ( `Failed to send notification to user ${ follower . id } , error ) ;
130+ } )
131+ ) ;
132+ }
133+ }
134+
135+ /**
136+ * Simplified method to send notifications to multiple addresses
137+ * Combines prepare, filter, and execute steps
138+ * @param addresses - Addresses to process
139+ * @param daoId - DAO identifier
140+ * @param eventIdGenerator - Function to generate unique event IDs
141+ * @param messageGenerator - Function to generate messages
142+ * @param metadataGenerator - Optional function to generate metadata
143+ */
144+ async sendBatchNotifications (
145+ addresses : string [ ] ,
146+ daoId : string ,
147+ eventIdGenerator : ( address : string ) => string ,
148+ messageGenerator : ( address : string ) => string ,
149+ metadataGenerator ?: ( address : string ) => Record < string , any >
150+ ) : Promise < void > {
151+ const batchData = await this . prepareBatchData ( addresses , daoId , eventIdGenerator ) ;
152+ const validNotifications = this . filterValidNotifications ( batchData ) ;
153+
154+ if ( validNotifications . length === 0 ) {
155+ return ;
156+ }
157+
158+ await this . executeBatchSend ( validNotifications , messageGenerator , metadataGenerator ) ;
159+ }
160+ }
161+
162+ /**
163+ * Interface for batch notification data
164+ */
165+ export interface BatchNotificationData {
166+ address : string ;
167+ followers : User [ ] ;
168+ notificationsToSend : Notification [ ] ;
169+ }
0 commit comments