diff --git a/config/migrate-ephemeral-retention.js b/config/migrate-ephemeral-retention.js index ab9f0e33af..ba425fbebc 100644 --- a/config/migrate-ephemeral-retention.js +++ b/config/migrate-ephemeral-retention.js @@ -2,6 +2,7 @@ const path = require('path'); const { logger, runAsSystem, + tenantStorage, createTempChatExpirationDate, forcedRetentionGapFilter, sweepForcedRetention, @@ -14,6 +15,48 @@ const connect = require('./connect'); const { getAppConfig } = require('~/server/services/Config'); const { Conversation, Message, SharedLink, File } = require('~/db/models'); +/** + * Converts one tenant's pre-existing data to the forced (ephemeral) window using that tenant's + * own retention config. Runs inside the caller's tenant context, so every query is scoped to the + * tenant (the untenanted bucket runs in the system context and touches the untenanted rows). A + * tenant whose resolved config is not ephemeral is skipped unless `force` is set. + */ +async function sweepTenant({ tenantId, dryRun, force }) { + const label = tenantId ?? 'default'; + const appConfig = await getAppConfig(tenantId ? { tenantId } : undefined); + const interfaceConfig = appConfig?.interfaceConfig; + const retentionMode = interfaceConfig?.retentionMode; + + if (retentionMode !== RetentionMode.EPHEMERAL && !force) { + logger.info( + `[tenant ${label}] retentionMode is "${retentionMode ?? 'unset'}", not "ephemeral" — skipping.`, + ); + return { tenantId: tenantId ?? null, skipped: true, retentionMode }; + } + + const forcedExpiredAt = createTempChatExpirationDate(interfaceConfig); + const nonConforming = await Conversation.countDocuments( + forcedRetentionGapFilter(forcedExpiredAt), + ); + logger.info(`[tenant ${label}] Found ${nonConforming} non-conforming conversation(s)`, { + forcedExpiredAt, + }); + + if (dryRun) { + return { tenantId: tenantId ?? null, dryRun: true, nonConforming, forcedExpiredAt }; + } + + const result = await sweepForcedRetention( + Conversation, + Message, + SharedLink, + File, + forcedExpiredAt, + ); + logger.info(`[tenant ${label}] completed`, result); + return { tenantId: tenantId ?? null, forcedExpiredAt, ...result }; +} + /** * Backfills forced (ephemeral) retention over conversations that predate the mode. * @@ -22,48 +65,44 @@ const { Conversation, Message, SharedLink, File } = require('~/db/models'); * non-expiring. This sweep converts every non-conforming conversation, its messages, its * shares, and its uploaded files to the forced window (capping rather than extending sooner * deadlines). It is idempotent and safe to re-run. + * + * Each tenant is converted with its OWN retention config: tenants are enumerated and swept inside + * their tenant context (so queries are scoped to that tenant), and a tenant whose config is not + * ephemeral is skipped. This prevents a system/default config from force-expiring a tenant that + * never enabled ephemeral retention. In a mixed deployment, rows without a tenantId cannot be + * scoped to a tenant config, so they are left untouched and must be converted from a + * single-tenant context. */ 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 }); - logger.info('Starting Ephemeral Retention Migration', { dryRun, force, retentionMode }); + const tenantIds = await Conversation.distinct('tenantId'); + const realTenants = tenantIds.filter((tenantId) => tenantId != null && tenantId !== ''); + const hasUntenanted = tenantIds.some((tenantId) => tenantId == null || tenantId === ''); + const skippedUntenanted = realTenants.length > 0 && hasUntenanted; - 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.', + const tenants = realTenants.length > 0 ? realTenants : [undefined]; + if (skippedUntenanted) { + logger.warn( + 'Some conversations have no tenantId; they cannot be scoped to a tenant config and are ' + + 'skipped. Re-run in a single-tenant context to convert them.', ); - 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 results = []; + for (const tenantId of tenants) { + const result = tenantId + ? await tenantStorage.run({ tenantId }, async () => + sweepTenant({ tenantId, dryRun, force }), + ) + : await sweepTenant({ tenantId, dryRun, force }); + results.push(result); } - const result = await sweepForcedRetention( - Conversation, - Message, - SharedLink, - File, - forcedExpiredAt, - ); - logger.info('Ephemeral Retention Migration completed', result); - return { dryRun: false, forcedExpiredAt, ...result }; + return { dryRun, skippedUntenanted, tenants: results }; }); } @@ -73,19 +112,31 @@ if (require.main === module) { 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.'); + if (result.skippedUntenanted) { + console.log('\nNote: conversations without a tenantId were skipped (see log warning).'); + } + + if (result.tenants.length > 0 && result.tenants.every((tenant) => tenant.skipped)) { + console.log('\n=== NOTHING TO MIGRATE ==='); + console.log('No tenant has ephemeral retention enabled.'); + console.log('Enable 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}`); + for (const tenant of result.tenants) { + const label = tenant.tenantId ?? 'default'; + if (tenant.skipped) { + console.log(`[${label}] skipped (retentionMode: ${tenant.retentionMode ?? 'unset'})`); + continue; + } + const expiry = tenant.forcedExpiredAt; + console.log( + `[${label}] non-conforming conversations: ${tenant.nonConforming} ` + + `(forced expiry: ${expiry?.toISOString?.() ?? expiry})`, + ); + } console.log('\nTo run the actual migration, remove the --dry-run flag.'); } else { console.log('\n=== MIGRATION RESULTS ==='); diff --git a/packages/data-schemas/src/methods/message.spec.ts b/packages/data-schemas/src/methods/message.spec.ts index e50631fa68..9bfbb03c73 100644 --- a/packages/data-schemas/src/methods/message.spec.ts +++ b/packages/data-schemas/src/methods/message.spec.ts @@ -1659,6 +1659,44 @@ describe('Message Operations', () => { expect(converted?.isTemporary).toBe(true); expect(converted?.expiredAt?.getTime()).toBe(forcedExpiredAt.getTime()); }); + + it('scopes the sweep to the active tenant context, leaving other tenants untouched', async () => { + const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); + const tenantAConversationId = uuidv4(); + const tenantBConversationId = uuidv4(); + await Conversation().collection.insertMany([ + { + conversationId: tenantAConversationId, + user: 'user123', + endpoint: 'openAI', + isTemporary: false, + tenantId: 'tenant-a', + }, + { + conversationId: tenantBConversationId, + user: 'user123', + endpoint: 'openAI', + isTemporary: false, + tenantId: 'tenant-b', + }, + ]); + + const result = await tenantStorage.run({ tenantId: 'tenant-a' }, async () => + sweepForcedRetention(Conversation(), Message, SharedLink(), File(), forcedExpiredAt), + ); + expect(result).toEqual({ conversations: 1, errors: 0 }); + + const tenantA = await Conversation().collection.findOne({ + conversationId: tenantAConversationId, + }); + expect(tenantA?.isTemporary).toBe(true); + + const tenantB = await Conversation().collection.findOne({ + conversationId: tenantBConversationId, + }); + expect(tenantB?.isTemporary ?? null).not.toBe(true); + expect(tenantB?.expiredAt ?? null).toBeNull(); + }); }); describe('cascadeForcedConversationRetention', () => {