Conversation
WalkthroughThis update introduces a modular, database-backed notification producer system with a focus on improved lifecycle management, batch notification handling, and robust retry logic. The notification producer is refactored to use producer classes for different notification types (CMS and trade events), each running concurrently and managed via a standardized Changes
Sequence Diagram(s)sequenceDiagram
participant Main as mainLoop
participant DB as PostgreSQL
participant CMS as CmsNotificationProducer
participant Trade as TradeNotificationProducer
participant Repo as NotificationsRepository
participant Queue as Notification Queue
Main->>DB: Initialize connection pool
Main->>CMS: Start CMS producer
Main->>Trade: Start Trade producers (per chain)
CMS->>Repo: Connect to queue
CMS->>Queue: Send batch notifications
Trade->>Repo: Connect to queue
Trade->>Queue: Send batch notifications
Trade->>DB: Persist last indexed block state
Main-->>CMS: On shutdown, stop producer
Main-->>Trade: On shutdown, stop producers
Main->>DB: Close connection pool
sequenceDiagram
participant Telegram as Telegram Consumer
participant Queue as Notification Queue
Telegram->>Queue: Receive message (batch of notifications)
Telegram->>Telegram: Parse notifications[]
loop For each notification
Telegram->>Telegram: Send notification to all subscribers
end
Telegram->>Telegram: Track retries
Telegram->>Queue: Ack/Nack message based on success or retry count
Possibly related PRs
Suggested reviewers
Poem
Tip ⚡💬 Agentic Chat (Pro Plan, General Availability)
📜 Recent review detailsConfiguration used: CodeRabbit UI ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (18)
💤 Files with no reviewable changes (1)
🧰 Additional context used🧬 Code Graph Analysis (7)apps/notification-producer/src/postToQueueTest.test.ts (1)
apps/notification-producer/src/repositories/SubscriptionsRepository.ts (1)
apps/notification-producer/src/utils.ts (3)
apps/notification-producer/src/producers/CmsNotificationProducer.ts (7)
apps/notification-producer/src/repositories/NotificationsRepository.ts (1)
apps/telegram/src/main.ts (2)
libs/cms-api/src/index.ts (1)
🪛 Biome (1.9.4)apps/notification-producer/src/main.ts[error] 89-89: void is confusing outside a return type or a type parameter. Unsafe fix: Use undefined instead. (lint/suspicious/noConfusingVoidType) 🔇 Additional comments (13)
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
|
Review the following changes in direct dependencies. Learn more about Socket for GitHub.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 17
🔭 Outside diff range comments (2)
libs/cms-api/src/index.ts (1)
163-199:⚠️ Potential issueFirst page is fetched twice – increment
pagebefore the 2nd request
getAllPages()retrieves page 0 again inside thewhileloop becausepageis only incremented after the repeated call togetPage.
That duplicates the first batch and can lead to an infinite loop if the API always returnspageSize + 1items.- while (subscriptions.length > pageSize) { - subscriptions = await getPage({ - page, - pageSize: pageSize + 1, - }); + while (subscriptions.length > pageSize) { + page++; // ✅ advance page *before* fetching + subscriptions = await getPage({ + page, + pageSize: pageSize + 1, + });Same pattern appears in
getAllNotifications; consider fixing both for consistency.apps/telegram/src/main.ts (1)
47-55:⚠️ Potential issueSubscription cache never refreshed – timestamp is not updated
After fetching fresh subscriptions you store them, but you forget to record the fetch time.
LAST_SUBSCRIPTION_CHECKis therefore alwaysundefined, causing a fetch on every call and defeating the cache.SUBSCRIPTION_CACHE.set(account, subscriptionForAccount); + LAST_SUBSCRIPTION_CHECK.set(account, new Date()); // ✅ mark refresh time
🧹 Nitpick comments (10)
apps/notification-producer/types.ts (1)
1-14: Well-designed interface for standardizing component lifecyclesThe
Runnableinterface cleanly defines the contract for components with start/stop lifecycle methods. The comments clearly document the expected behavior that methods should not throw or complete immediately.Consider adding a method to query the current state of the runnable component, which could help with monitoring and management:
export interface Runnable { /** * Start the program, this method should not throw or finish. */ start(): Promise<void>; /** * Stop the program, this method should not throw or finish. */ stop(): Promise<void>; + + /** + * Returns the current state of the program. + */ + getStatus(): 'starting' | 'running' | 'stopping' | 'stopped'; }apps/notification-producer/src/utils.ts (1)
3-28: Robust implementation for continuous background processingThe
doForeverutility provides a clean pattern for implementing resilient background processes that can gracefully handle errors and shutdown requests. The function correctly implements async/await patterns and has good logging.Consider these enhancements to make the utility more flexible:
import { sleep } from '@cowprotocol/notifications'; export async function doForever( name: string, callback: (stop: () => void) => Promise<void>, waitTimeMilliseconds: number, + options?: { + onError?: (error: unknown) => void; + errorWaitTimeMilliseconds?: number; + } ) { // eslint-disable-next-line no-constant-condition let running = true; while (running) { const stop = () => { console.log(`[${name}] Stopping...`); running = false; }; try { await callback(stop); } catch (error) { console.error(`[${name}] Error `, error); + if (options?.onError) { + options.onError(error); + } console.log( `[${name}] Reconnecting in ${ - waitTimeMilliseconds / 1000 + (options?.errorWaitTimeMilliseconds || waitTimeMilliseconds) / 1000 }s...` ); + await sleep(options?.errorWaitTimeMilliseconds || waitTimeMilliseconds); + continue; } finally { - await sleep(waitTimeMilliseconds); + if (running) { + await sleep(waitTimeMilliseconds); + } } } console.log(`[${name}] Stopped`); }This refactoring allows for:
- Custom error handling logic
- Different wait times after errors versus successful executions
- Avoiding unnecessary waits after the loop is stopped
apps/notification-producer/src/postToQueueTest.test.ts (1)
19-21: Avoid JSON string for the message property.The
messageproperty contains a JSON string with redundant data already present in other properties of the notification.- message: - '{"title": "my title", "message": "my message", "account": "0x79063d9173C09887d536924E2F6eADbaBAc099f5", "url": "https://www.google.com"}', + message: "my message",If the payload format needs to maintain backward compatibility, consider adding a comment explaining why the message is structured this way.
apps/notification-producer/src/repositories/SubscriptionsRepository.ts (2)
5-6: Consider updating the TODO comment with a more specific action item.The TODO comment is somewhat vague. It would be helpful to add more specific details about which dependency injection framework will be used and what the "proper cached repository" structure looks like.
25-27: Consider moving the helper function inside the class.The
uniqueLowercasefunction is only used within theSubscriptionRepositoryclass. Consider moving it inside the class as a private static method for better encapsulation.-function uniqueLowercase(items: string[]): string[] { - return Array.from(new Set(items.map((item) => item.toLowerCase()))); -} export class SubscriptionRepository { private lastCheck: number | null = null; private cachedAccounts: string[] | null = null; + private static uniqueLowercase(items: string[]): string[] { + return Array.from(new Set(items.map((item) => item.toLowerCase()))); + } async getAllSubscribedAccounts(): Promise<string[]> { const now = Date.now(); if ( !this.cachedAccounts || !this.lastCheck || now - this.lastCheck > CACHE_TIME ) { - this.cachedAccounts = uniqueLowercase(await getAllSubscribedAccounts()); + this.cachedAccounts = SubscriptionRepository.uniqueLowercase(await getAllSubscribedAccounts()); this.lastCheck = now; return this.cachedAccounts; } return this.cachedAccounts || []; } }apps/notification-producer/src/repositories/NotificationsRepository.ts (1)
55-64: Improve error handling in pingConnection method.The
pingConnectionmethod catches all errors without logging them, which might mask specific connection issues and make debugging harder.async pingConnection(): Promise<boolean> { if (!this.connection) return false; try { const ch = await this.connection.connection.createChannel(); await ch.close(); return true; - } catch (error) { + } catch (error: unknown) { + const errorMessage = error instanceof Error ? error.message : String(error); + console.debug(`Connection ping failed: ${errorMessage}`); return false; } }libs/notifications/src/index.ts (2)
29-31: Add validation before stringifying notifications.The
stringifyNotificationsfunction doesn't validate the notification objects before stringification. This could lead to unexpected behavior if the objects don't match the expected format.export function stringifyNotifications(notifications: Notification[]): string { + // Ensure notifications is an array + if (!Array.isArray(notifications)) { + console.warn('Expected notifications to be an array, got:', typeof notifications); + return JSON.stringify([]); + } + + // Filter out invalid notifications + const validNotifications = notifications.filter(n => + n && typeof n === 'object' && + typeof n.id === 'string' && + typeof n.account === 'string' && + typeof n.title === 'string' && + typeof n.message === 'string' + ); + + if (validNotifications.length !== notifications.length) { + console.warn(`Filtered out ${notifications.length - validNotifications.length} invalid notifications`); + } + + return JSON.stringify(validNotifications); - return JSON.stringify(notifications); }
78-81: Add validation for empty notification arrays.The
sendNotificationsToQueuefunction will send empty arrays to the queue. Consider adding a check to avoid unnecessary queue operations when there are no notifications to send.export function sendNotificationsToQueue(params: SendToQueueParams) { const { channel, queue, notifications } = params; + // Don't send empty notification arrays + if (!notifications || notifications.length === 0) { + return; + } const message = stringifyNotifications(notifications); channel.sendToQueue(queue, Buffer.from(message)); }apps/notification-producer/src/producers/TradeNotificationProducer.ts (1)
161-164:title: 'Trade'reused forOrderInvalidatedNotifications for cancelled orders are labelled “Trade”, which is misleading.
- title: 'Trade', + title: 'Order invalidated',apps/telegram/src/main.ts (1)
58-67: Guard clause could drop valid empty batches
parseNewMessagetreats an empty array ([]) as an error becauseisNotificationArray([])is true but the calling code later loops over the array.
Consider returning an empty array and lettingonNewMessageacknowledge it immediately, rather than triggering retries for a benign case.
🛑 Comments failed to post (17)
apps/notification-producer/src/postToQueueTest.test.ts (1)
7-26: 🛠️ Refactor suggestion
Test lacks assertions and connection cleanup.
While the test exercises the notification queue functionality, it doesn't verify the operation succeeded or properly clean up resources.
-it('Post to queue', async () => { +it('Post to queue', async () => { + let connection; + let channel; + + try { - const { channel } = await connectToChannel({ + ({ connection, channel } = await connectToChannel({ channel: NOTIFICATIONS_QUEUE, - }); + })); sendNotificationsToQueue({ channel, queue: NOTIFICATIONS_QUEUE, notifications: [ { id: '1', title: 'title', message: '{"title": "my title", "message": "my message", "account": "0x79063d9173C09887d536924E2F6eADbaBAc099f5", "url": "https://www.google.com"}', account: '0x79063d9173C09887d536924E2F6eADbaBAc099f5', url: 'url', }, ], }); + + // Add assertion to verify queue operation + expect(channel.sendToQueue).toHaveBeenCalledWith( + NOTIFICATIONS_QUEUE, + expect.any(Buffer) + ); + } finally { + // Clean up resources + if (channel) await channel.close(); + if (connection) await connection.close(); + } });Also, consider mocking the RabbitMQ channel to avoid external dependencies during unit testing:
// At the top of the file jest.mock('@cowprotocol/notifications', () => ({ NOTIFICATIONS_QUEUE: 'notifications', connectToChannel: jest.fn().mockResolvedValue({ connection: { close: jest.fn() }, channel: { sendToQueue: jest.fn(), close: jest.fn() } }), sendNotificationsToQueue: jest.fn(), }));apps/notification-producer/src/repositories/SubscriptionsRepository.ts (1)
10-22: 🛠️ Refactor suggestion
Add error handling for the external API call.
The
getAllSubscribedAccountsmethod doesn't handle potential errors from the external API call. If the API call fails, the error will propagate and potentially disrupt the application flow.async getAllSubscribedAccounts(): Promise<string[]> { const now = Date.now(); if ( !this.cachedAccounts || !this.lastCheck || now - this.lastCheck > CACHE_TIME ) { - this.cachedAccounts = uniqueLowercase(await getAllSubscribedAccounts()); - this.lastCheck = now; - return this.cachedAccounts; + try { + this.cachedAccounts = uniqueLowercase(await getAllSubscribedAccounts()); + this.lastCheck = now; + return this.cachedAccounts; + } catch (error) { + console.error('Failed to fetch subscribed accounts:', error); + // Return cached accounts if available, otherwise empty array + return this.cachedAccounts || []; + } } return this.cachedAccounts || []; }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async getAllSubscribedAccounts(): Promise<string[]> { const now = Date.now(); if ( !this.cachedAccounts || !this.lastCheck || now - this.lastCheck > CACHE_TIME ) { try { this.cachedAccounts = uniqueLowercase(await getAllSubscribedAccounts()); this.lastCheck = now; return this.cachedAccounts; } catch (error) { console.error('Failed to fetch subscribed accounts:', error); // Return cached accounts if available, otherwise empty array return this.cachedAccounts || []; } } return this.cachedAccounts || []; }apps/notification-producer/src/repositories/NotificationsRepository.ts (2)
14-29: 🛠️ Refactor suggestion
Fix potential event listener memory leak.
The
connectmethod adds a new 'close' event listener each time it's called without removing previous listeners. This could lead to multiple duplicate handlers if the method is called repeatedly after disconnections.async connect(): Promise<ConnectToChannelResponse> { if (!this.connection || !this.pingConnection()) { + // Remove any existing listeners if the connection exists + if (this.connection) { + this.connection.connection.removeAllListeners('close'); + } // Connect to the queue this.connection = await connectToChannel({ channel: this.queueName, }); // Watch for connection close this.connection.connection.on('close', () => this.close()); return this.connection; } // Return connection return this.connection; }
41-49: 🛠️ Refactor suggestion
Add error handling for queue operations.
The
sendNotificationsmethod doesn't handle potential errors when sending notifications to the queue. If the operation fails, there's no retry mechanism or error logging.async sendNotifications(notifications: Notification[]) { - const connection = await this.connect(); - - await sendNotificationsToQueue({ - channel: connection.channel, - queue: this.queueName, - notifications, - }); + try { + const connection = await this.connect(); + + await sendNotificationsToQueue({ + channel: connection.channel, + queue: this.queueName, + notifications, + }); + } catch (error) { + console.error('Failed to send notifications to queue:', error); + throw error; // Re-throw or handle based on your error handling strategy + } }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async sendNotifications(notifications: Notification[]) { try { const connection = await this.connect(); await sendNotificationsToQueue({ channel: connection.channel, queue: this.queueName, notifications, }); } catch (error) { console.error('Failed to send notifications to queue:', error); throw error; // Re-throw or handle based on your error handling strategy } }apps/notification-producer/src/repositories/NotificationsIndexerStateRepository.ts (3)
38-47: 🛠️ Refactor suggestion
Handle null chainId values in the upsert method.
The
chainIdparameter can be undefined, but there's no explicit handling for this case in the SQL query. This could lead to inserting undefined or invalid values.async upsert<T>(key: string, state: T, chainId?: number): Promise<void> { const query = ` INSERT INTO notifications_indexer_state (key, chainId, state) VALUES ($1, $2, $3) ON CONFLICT (key, chainId) DO UPDATE SET state = $3 `; - await this.db.query(query, [key, chainId, state]); + // Ensure chainId is null when undefined + await this.db.query(query, [key, chainId ?? null, state]); }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async upsert<T>(key: string, state: T, chainId?: number): Promise<void> { const query = ` INSERT INTO notifications_indexer_state (key, chainId, state) VALUES ($1, $2, $3) ON CONFLICT (key, chainId) DO UPDATE SET state = $3 `; // Ensure chainId is null when undefined await this.db.query(query, [key, chainId ?? null, state]); }
18-33:
⚠️ Potential issueFix SQL query to match return type.
The SQL query only selects the
statecolumn, but the return type isNotificationsIndexerState<T>. The returned object won't have all the expected properties (key, chainId, created_at, updated_at).async get<T>( key: string, chainId?: SupportedChainId ): Promise<NotificationsIndexerState<T> | null> { const query = ` - SELECT state + SELECT key, chainId, state, created_at, updated_at FROM notifications_indexer_state WHERE key = $1 ${chainId !== undefined ? 'AND chainId = $2' : ''} LIMIT 1 `; const params = chainId !== undefined ? [key, chainId] : [key]; const result = await this.db.query(query, params); return result.rows[0] || null; }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async get<T>( key: string, chainId?: SupportedChainId ): Promise<NotificationsIndexerState<T> | null> { const query = ` SELECT key, chainId, state, created_at, updated_at FROM notifications_indexer_state WHERE key = $1 ${chainId !== undefined ? 'AND chainId = $2' : ''} LIMIT 1 `; const params = chainId !== undefined ? [key, chainId] : [key]; const result = await this.db.query(query, params); return result.rows[0] || null; }
35-47: 🛠️ Refactor suggestion
Add serialization of the state object.
The state parameter might be a complex object that needs to be serialized to JSON before storing in the database. Without explicit serialization, PostgreSQL might not store it correctly.
async upsert<T>(key: string, state: T, chainId?: number): Promise<void> { const query = ` INSERT INTO notifications_indexer_state (key, chainId, state) VALUES ($1, $2, $3) ON CONFLICT (key, chainId) DO UPDATE SET state = $3 `; - await this.db.query(query, [key, chainId, state]); + // Serialize state to JSON + await this.db.query(query, [key, chainId ?? null, JSON.stringify(state)]); }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements./** * Update or insert indexer state */ async upsert<T>(key: string, state: T, chainId?: number): Promise<void> { const query = ` INSERT INTO notifications_indexer_state (key, chainId, state) VALUES ($1, $2, $3) ON CONFLICT (key, chainId) DO UPDATE SET state = $3 `; // Serialize state to JSON await this.db.query(query, [key, chainId ?? null, JSON.stringify(state)]); }libs/notifications/src/index.ts (1)
23-27: 🛠️ Refactor suggestion
Add error handling for JSON parsing.
The
parseNotificationsfunction doesn't handle potential parsing errors. If the JSON is invalid, it will throw an error that might disrupt the application flow.export function parseNotifications( notificationsString: string ): Notification[] { - return JSON.parse(notificationsString); + try { + return JSON.parse(notificationsString); + } catch (error) { + console.error('Failed to parse notifications:', error); + return []; + } }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.export function parseNotifications( notificationsString: string ): Notification[] { try { return JSON.parse(notificationsString); } catch (error) { console.error('Failed to parse notifications:', error); return []; } }apps/notification-producer/src/producers/CmsNotificationProducer.ts (2)
60-67:
⚠️ Potential issueAddress‑matching is case‑sensitive – risk of silently dropping legitimate notifications
accountsis already normalized to lower‑case bySubscriptionRepository, however theaccountfield coming from the CMS is compared verbatim.
If the CMS happens to return mixed‑case addresses (0xAbC…) the strictincludescheck will fail, causing valid notifications to be ignored.Suggestion:
-({ account }) => accounts.includes(account) +({ account }) => accounts.includes(account.toLowerCase())
82-93: 🛠️ Refactor suggestion
⚠️ Potential issue
sendNotificationsisasyncbut its result is not awaited – notifications may be lost & errors unhandled
connect()is awaited, but the actual publish operation is fired‑and‑forgotten.
If the Promise rejects (network blip, queue unavailable, …) the exception will be unhandled andpendingNotificationswill still be cleared, losing those messages.-// Post notifications to queue -this.props.notificationsRepository.sendNotifications(pushNotifications); -this.pendingNotifications.clear(); +try { + // Post notifications to queue + await this.props.notificationsRepository.sendNotifications( + pushNotifications + ); + // Clear only after a successful publish + this.pendingNotifications.clear(); +} catch (err) { + console.error('[CmsNotificationProducer] Failed to send', err); + // keep notifications in memory for the next retry +}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.// Save notifications in-memory, so they are not lost if there's an issue with the queue pushNotifications.forEach((notification) => this.pendingNotifications.set(notification.id, notification) ); // Connect await this.props.notificationsRepository.connect(); try { // Post notifications to queue await this.props.notificationsRepository.sendNotifications( pushNotifications ); // Clear only after a successful publish this.pendingNotifications.clear(); } catch (err) { console.error('[CmsNotificationProducer] Failed to send', err); // keep notifications in memory for the next retry } }apps/notification-producer/src/producers/TradeNotificationProducer.ts (3)
145-153: 🛠️ Refactor suggestion
User‑facing message shows raw wei amounts and token addresses
log.args.sellAmount/buyAmountareuint256in wei and the token fields are raw addresses.
End‑users will receive unreadable messages like
“Trade 1000000000000000000 0x6B… for 2000000000000000000 0xA0…”.Consider resolving token symbols/decimals and formatting with
formatUnits.
196-204:
⚠️ Potential issueDatabase state update is asynchronous but not awaited
upsertreturns aPromise; ignoring it risks losing the last processed block on crash.-await notificationsIndexerStateRepository.upsert<TradeNotificationProducerState>( +await notificationsIndexerStateRepository.upsert<TradeNotificationProducerState>(📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.await notificationsIndexerStateRepository.upsert<TradeNotificationProducerState>( PRODUCER_NAME, { lastBlock: lastBlock.number.toString(), lastBlockTimestamp: lastBlock.timestamp.toString(), lastBlockHash: lastBlock.hash, }, chainId );
188-193:
⚠️ Potential issue
sendNotificationsis not awaited – same reliability issueRepeat of the CMS producer issue.
-await this.props.notificationsRepository.connect(); -// Post notifications to queue -this.props.notificationsRepository.sendNotifications(notifications); +await this.props.notificationsRepository.connect(); +await this.props.notificationsRepository.sendNotifications(notifications);Committable suggestion skipped: line range outside the PR's diff.
apps/notification-producer/src/main.ts (2)
60-70: 🛠️ Refactor suggestion
Producer auto‑restart logic drops the replacement Promise – shutdown & monitoring issues
Inside the
.then/.catchhandlers you callproducer.start()again but do not return the new promise, sopromisesonly tracks the first run.
producersPromisemay resolve even though the restarted producer is still running, breaking graceful shutdown logic.Refactor to a small helper that wraps each producer with a self‑restarting loop and always returns the current promise.
76-82:
⚠️ Potential issueCannot catch
SIGKILL– useSIGINT/SIGTERMinsteadNode.js cannot intercept
SIGKILL; the handler will never execute, so resources won’t be cleaned up.-process.on('SIGKILL', () => { +process.on('SIGTERM', () => { + gracefulShutdown(pool, producers, producersPromise).catch(…); +}); +process.on('SIGINT', () => { gracefulShutdown(pool, producers, producersPromise).catch((error) => { … }); -}); +});📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.process.on('SIGTERM', () => { gracefulShutdown(pool, producers, producersPromise).catch((error) => { console.error('Error during shutdown', error); process.exit(1); }); }); process.on('SIGINT', () => { gracefulShutdown(pool, producers, producersPromise).catch((error) => { console.error('Error during shutdown', error); process.exit(1); }); });libs/cms-api/src/index.ts (1)
210-217: 🛠️ Refactor suggestion
Potentially over‑long URL – switch to query/body param
/accounts/${accounts.join(',')}/subscriptions/telegramcan exceed typical URL‐length limits whenaccountsis large (easily >2 kB).
Consider POSTing the list in the body or passing it as a query string (accounts=0x…&accounts=0x…) to avoid truncated requests.apps/telegram/src/main.ts (1)
127-134:
⚠️ Potential issueRetry logic ineffective – message is
nack‑ed without requeue
channel.nack(msg, false, false)discards the message, so it will never be delivered again despite incrementingMESSAGE_RETRIES.
Passtrueas the third argument to requeue the message and allow the retry loop to work.- channel.nack(msg, false, false); + channel.nack(msg, false, true); // requeue = true📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.// Increment retry count and NACK the message const newRetryCount = retryCount + 1; MESSAGE_RETRIES.set(messageId, newRetryCount); console.warn( `[telegram:main] Retry attempt ${newRetryCount}/${MAX_RETRIES} for message ${messageId}` ); channel.nack(msg, false, true); // requeue = true }
| import { getAddress, parseAbi } from 'viem'; | ||
|
|
||
| const EVENTS = parseAbi([ | ||
| 'event OrderInvalidated(address indexed owner, bytes orderUid)', |
There was a problem hiding this comment.
BTW, this is tracked in the backend DB as well.
There's a table called onchain_order_invalidations.
So in theory we don't need to watch for those events.
3 participants
This PR:
Get the addresses of users that are subscribed
Use it later for filtering logs
Get logs of trades
Get logs of events for the settlement contract
filters by subscribed users
Get logs of cancelations
Get logs of events for the settlement contract
filters by subscribed users
Convert logs to notifications
Converts both the cancelation and the trade into a notification
Push notification to queue
Send the notifications to the queue
Summary by CodeRabbit
New Features
Bug Fixes
Refactor
Tests
Documentation
Chores