diff --git a/config/migrate-ephemeral-retention.js b/config/migrate-ephemeral-retention.js new file mode 100644 index 0000000000..fb86fda2ca --- /dev/null +++ b/config/migrate-ephemeral-retention.js @@ -0,0 +1,96 @@ +const path = require('path'); +const { + logger, + runAsSystem, + createTempChatExpirationDate, + forcedRetentionGapFilter, + sweepForcedRetention, +} = require('@librechat/data-schemas'); +const { RetentionMode } = require('librechat-data-provider'); + +require('module-alias')({ base: path.resolve(__dirname, '..', 'api') }); +const connect = require('./connect'); + +const { getAppConfig } = require('~/server/services/Config'); +const { Conversation, Message, SharedLink } = require('~/db/models'); + +/** + * Backfills forced (ephemeral) retention over conversations that predate the mode. + * + * Convert-on-touch only converts chats that are subsequently written, so enabling ephemeral + * retention on a deployment with existing data leaves untouched permanent chats visible and + * non-expiring. This sweep converts every non-conforming conversation, its messages, and its + * shares to the forced window (capping rather than extending sooner deadlines). It is + * idempotent and safe to re-run. + */ +async function migrateEphemeralRetention({ dryRun = true, force = false } = {}) { + await connect(); + + return runAsSystem(async () => { + const appConfig = await getAppConfig(); + const interfaceConfig = appConfig?.interfaceConfig; + const retentionMode = interfaceConfig?.retentionMode; + + logger.info('Starting Ephemeral Retention Migration', { dryRun, force, retentionMode }); + + if (retentionMode !== RetentionMode.EPHEMERAL && !force) { + logger.error( + `retentionMode is "${retentionMode ?? 'unset'}", not "ephemeral". This migration ` + + 'converts every conversation into a temporary, expiring chat. Enable ephemeral ' + + 'retention first, or pass --force to run anyway.', + ); + return { aborted: true, reason: 'retentionMode is not ephemeral', retentionMode }; + } + + const forcedExpiredAt = createTempChatExpirationDate(interfaceConfig); + const nonConforming = await Conversation.countDocuments( + forcedRetentionGapFilter(forcedExpiredAt), + ); + logger.info(`Found ${nonConforming} non-conforming conversation(s)`, { forcedExpiredAt }); + + if (dryRun) { + return { + dryRun: true, + summary: { nonConformingConversations: nonConforming, forcedExpiredAt }, + }; + } + + const result = await sweepForcedRetention(Conversation, Message, SharedLink, forcedExpiredAt); + logger.info('Ephemeral Retention Migration completed', result); + return { dryRun: false, forcedExpiredAt, ...result }; + }); +} + +if (require.main === module) { + const dryRun = process.argv.includes('--dry-run'); + const force = process.argv.includes('--force'); + + migrateEphemeralRetention({ dryRun, force }) + .then((result) => { + if (result.aborted) { + console.log('\n=== MIGRATION ABORTED ==='); + console.log(`Reason: ${result.reason}`); + console.log(`Current retentionMode: ${result.retentionMode ?? 'unset'}`); + console.log('\nEnable ephemeral retention, or pass --force to run anyway.'); + process.exit(1); + } + + if (result.dryRun) { + console.log('\n=== DRY RUN RESULTS ==='); + console.log(`Non-conforming conversations: ${result.summary.nonConformingConversations}`); + const expiry = result.summary.forcedExpiredAt; + console.log(`Forced expiry: ${expiry?.toISOString?.() ?? expiry}`); + console.log('\nTo run the actual migration, remove the --dry-run flag.'); + } else { + console.log('\n=== MIGRATION RESULTS ==='); + console.log(JSON.stringify(result, null, 2)); + } + process.exit(0); + }) + .catch((error) => { + console.error('Ephemeral retention migration failed:', error); + process.exit(1); + }); +} + +module.exports = { migrateEphemeralRetention }; diff --git a/package.json b/package.json index d7bd425489..dd5b36f629 100644 --- a/package.json +++ b/package.json @@ -109,7 +109,9 @@ "migrate:shared-link-permissions:batch": "node config/migrate-shared-link-permissions.js --batch-size=50", "migrate:orphaned-agent-files:dry-run": "node config/migrate-orphaned-agent-files.js --dry-run", "migrate:orphaned-agent-files": "node config/migrate-orphaned-agent-files.js", - "migrate:orphaned-agent-files:batch": "node config/migrate-orphaned-agent-files.js --batch-size=50" + "migrate:orphaned-agent-files:batch": "node config/migrate-orphaned-agent-files.js --batch-size=50", + "migrate:ephemeral-retention:dry-run": "node config/migrate-ephemeral-retention.js --dry-run", + "migrate:ephemeral-retention": "node config/migrate-ephemeral-retention.js" }, "repository": { "type": "git", diff --git a/packages/data-schemas/src/methods/message.spec.ts b/packages/data-schemas/src/methods/message.spec.ts index 9f82adce53..1f96a97e7c 100644 --- a/packages/data-schemas/src/methods/message.spec.ts +++ b/packages/data-schemas/src/methods/message.spec.ts @@ -4,6 +4,7 @@ import { RetentionMode } from 'librechat-data-provider'; import { MongoMemoryServer } from 'mongodb-memory-server'; import type { IConversation, IMessage, ISharedLink } from '..'; import { tenantStorage, runAsSystem } from '~/config/tenantContext'; +import { sweepForcedRetention } from '../utils/retention'; import { createMessageMethods } from './message'; import { createModels } from '../models'; import logger from '~/config/winston'; @@ -1372,6 +1373,104 @@ describe('Message Operations', () => { }); }); + describe('sweepForcedRetention', () => { + const Conversation = () => mongoose.models.Conversation as mongoose.Model; + const SharedLink = () => mongoose.models.SharedLink as mongoose.Model; + + beforeEach(async () => { + await Conversation().deleteMany({}); + await SharedLink().deleteMany({}); + }); + + it('converts untouched permanent conversations, messages, and shares and skips conforming ones', async () => { + const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); + const permanentId = uuidv4(); + const conformingId = uuidv4(); + const soonerExpiry = new Date(Date.now() + 30 * 60 * 1000); + + await Conversation().create([ + { conversationId: permanentId, user: 'user123', endpoint: 'openAI', isTemporary: false }, + { + conversationId: conformingId, + user: 'user123', + endpoint: 'openAI', + isTemporary: true, + expiredAt: soonerExpiry, + }, + ]); + await Message.create([ + { messageId: uuidv4(), conversationId: permanentId, user: 'user123', text: 'permanent' }, + { + messageId: uuidv4(), + conversationId: conformingId, + user: 'user123', + text: 'conforming', + isTemporary: true, + expiredAt: soonerExpiry, + }, + ]); + await SharedLink().create({ + conversationId: permanentId, + user: 'user123', + shareId: uuidv4(), + }); + + const result = await sweepForcedRetention( + Conversation(), + Message, + SharedLink(), + forcedExpiredAt, + ); + expect(result).toEqual({ conversations: 1, errors: 0 }); + + const permanent = await Conversation().findOne({ conversationId: permanentId }).lean(); + expect(permanent?.isTemporary).toBe(true); + expect(permanent?.expiredAt?.getTime()).toBe(forcedExpiredAt.getTime()); + + const permanentMessages = await getMessages({ conversationId: permanentId, user: 'user123' }); + for (const message of permanentMessages) { + expect(message.isTemporary).toBe(true); + expect(message.expiredAt?.getTime()).toBe(forcedExpiredAt.getTime()); + } + + const share = await SharedLink().findOne({ conversationId: permanentId }).lean(); + expect(share?.expiredAt?.getTime()).toBe(forcedExpiredAt.getTime()); + + const conforming = await Conversation().findOne({ conversationId: conformingId }).lean(); + expect(conforming?.expiredAt?.getTime()).toBe(soonerExpiry.getTime()); + }); + + it('aligns a permanent message to a sooner parent deadline instead of the forced window', async () => { + const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); + const conversationId = uuidv4(); + const soonerExpiry = new Date(Date.now() + 30 * 60 * 1000); + + await Conversation().create({ + conversationId, + user: 'user123', + endpoint: 'openAI', + isTemporary: false, + expiredAt: soonerExpiry, + }); + const permanentMessageId = uuidv4(); + await Message.create({ + messageId: permanentMessageId, + conversationId, + user: 'user123', + text: 'permanent', + isTemporary: false, + }); + + await sweepForcedRetention(Conversation(), Message, SharedLink(), forcedExpiredAt); + + const convo = await Conversation().findOne({ conversationId }).lean(); + expect(convo?.expiredAt?.getTime()).toBe(soonerExpiry.getTime()); + const message = await Message.findOne({ messageId: permanentMessageId }).lean(); + expect(message?.isTemporary).toBe(true); + expect(message?.expiredAt?.getTime()).toBe(soonerExpiry.getTime()); + }); + }); + describe('Message cursor pagination', () => { /** * Helper to create messages with specific timestamps diff --git a/packages/data-schemas/src/utils/retention.ts b/packages/data-schemas/src/utils/retention.ts index adc7961c2d..4fdba966d4 100644 --- a/packages/data-schemas/src/utils/retention.ts +++ b/packages/data-schemas/src/utils/retention.ts @@ -262,3 +262,46 @@ export const cascadeForcedRetentionByTag = async ( ); } }; + +/** + * One-time backfill of forced (ephemeral) retention over pre-existing data. Convert-on-touch + * only converts conversations that are subsequently written, so enabling ephemeral mode on a + * deployment with existing chats leaves untouched permanent rows visible and non-expiring. + * + * Streams every conversation that does not yet conform to the forced window and converts it, + * its messages, and its shares one conversation at a time. Each conversation is capped to the + * earlier of its own deadline and the forced window, and its messages and shares are capped to + * that same per-conversation deadline, so the sweep never extends data that already expires + * sooner and never lets a message outlive its conversation. It is idempotent: re-running skips + * conversations that already conform. + */ +export const sweepForcedRetention = async ( + Conversation: Model, + Message: Model, + SharedLink: Model, + forcedExpiredAt: Date, +): Promise<{ conversations: number; errors: number }> => { + const result = { conversations: 0, errors: 0 }; + const cursor = Conversation.find(forcedRetentionGapFilter(forcedExpiredAt)) + .select('_id conversationId user expiredAt') + .lean() + .cursor(); + + for await (const convo of cursor) { + const { conversationId, user } = convo; + if (typeof conversationId !== 'string' || !conversationId || !user) { + continue; + } + try { + const expiredAt = capForcedRetentionExpiry(convo.expiredAt, forcedExpiredAt); + await Conversation.updateOne({ _id: convo._id }, { $set: { isTemporary: true, expiredAt } }); + await forceConversationMessagesTemporary(Message, user, conversationId, expiredAt); + await capConversationSharedLinks(SharedLink, user, conversationId, expiredAt); + result.conversations += 1; + } catch { + result.errors += 1; + } + } + + return result; +};