import { cronJobs } from '@rocket.chat/cron'; import { Logger } from '@rocket.chat/logger'; import { ReadReceipts, ReadReceiptsArchive, Messages } from '@rocket.chat/models'; import { sleep } from '../../../lib/utils/sleep'; import { settings } from '../../../server/settings'; const logger = new Logger('ReadReceiptsArchive'); const MILLISECONDS_PER_DAY = 24 * 60 * 60 * 1000; const BATCH_DELAY_MS = 1000; // 1 second delay between batches export async function archiveOldReadReceipts(): Promise { const retentionDays = settings.get('Message_Read_Receipt_Archive_Retention_Days') || 30; const batchSize = settings.get('Message_Read_Receipt_Archive_Batch_Size') || 10000; const cutoffDate = new Date(Date.now() - retentionDays * MILLISECONDS_PER_DAY); logger.info({ msg: 'Starting to archive old read receipts', batchSize, cutoffDate }); let totalProcessed = 0; let batchNumber = 0; let hasMore = true; while (hasMore) { batchNumber++; logger.info({ msg: 'Processing batch', batchNumber }); // Find receipts older than the retention period, limited by batch size const oldReceipts = await ReadReceipts.findOlderThan(cutoffDate).limit(batchSize).toArray(); if (oldReceipts.length === 0) { logger.info({ msg: 'No more read receipts to archive', totalProcessed }); break; } logger.info({ msg: 'Found read receipts in batch', count: oldReceipts.length, batchNumber }); // Get unique message IDs from the receipts to be archived const messageIds = [...new Set(oldReceipts.map((receipt) => receipt.messageId))]; try { // Insert receipts into archive collection (using insertMany with ordered: false to continue on duplicate key errors) try { await ReadReceiptsArchive.insertMany(oldReceipts, { ordered: false }); logger.info({ msg: 'Successfully archived read receipts', count: oldReceipts.length, batchNumber }); } catch (error: unknown) { // If we get duplicate key errors, some receipts were already archived, which is fine // We'll continue to mark messages and delete from hot storage if (error && typeof error === 'object' && ('code' in error || 'writeErrors' in error)) { const mongoError = error as { code?: number; result?: { insertedCount?: number }; writeErrors?: Array<{ code?: number }>; }; const onlyDuplicateErrors = mongoError.writeErrors?.length ? mongoError.writeErrors.every((writeError) => writeError.code === 11000) : mongoError.code === 11000; if (onlyDuplicateErrors) { const insertedCount = mongoError.result?.insertedCount || 0; logger.info({ msg: 'Archived read receipts (some were already archived)', insertedCount, batchNumber }); } else { throw error; } } else { throw error; } } // Mark messages as having archived receipts const updateResult = await Messages.setReceiptsArchivedById(messageIds, true); logger.info({ msg: 'Marked messages as having archived receipts', modifiedCount: updateResult.modifiedCount, batchNumber }); // Delete old receipts from hot storage for this batch const receiptIds = oldReceipts.map((receipt) => receipt._id); const deleteResult = await ReadReceipts.removeByIds(receiptIds); logger.info({ msg: 'Deleted old receipts from hot storage', deletedCount: deleteResult.deletedCount, batchNumber }); totalProcessed += oldReceipts.length; // If we processed a full batch, there might be more, so wait and continue if (oldReceipts.length === batchSize) { logger.info({ msg: 'Batch complete, waiting before next batch', batchNumber, delayMs: BATCH_DELAY_MS }); await sleep(BATCH_DELAY_MS); } else { // This was the last batch (partial batch) logger.info({ msg: 'Final batch complete', batchNumber, totalProcessed }); hasMore = false; } } catch (error) { logger.error({ msg: 'Error during read receipts archiving', batchNumber, err: error }); throw error; } } } export async function readReceiptsArchiveCron(): Promise { const cronSchedule = settings.get('Message_Read_Receipt_Archive_Cron') || '0 2 * * *'; // Remove existing job if it exists if (await cronJobs.has('ReadReceiptsArchive')) { await cronJobs.remove('ReadReceiptsArchive'); } if (!settings.get('Message_Read_Receipt_Archive_Enabled')) { return; } logger.info({ msg: 'Scheduling read receipts archive cron job', cronSchedule }); return cronJobs.add('ReadReceiptsArchive', cronSchedule, async () => archiveOldReadReceipts()); }