From 7b8272476d72e6b42c77a05cd3cbb6ee5ce732c2 Mon Sep 17 00:00:00 2001 From: Marco Beretta <81851188+berry-13@users.noreply.github.com> Date: Sun, 5 Jul 2026 02:43:02 +0200 Subject: [PATCH] fix: cap referenced attachment files that carry no conversationId MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Message-attachment uploads create File rows without a conversationId — they are referenced only from Message.files[].file_id and the conversation's files array (only Code Interpreter outputs set conversationId). The conversation-scoped file caps therefore missed regular attachments entirely: converting a pre-existing chat left their expiredAt null, so storage cleanup never swept them after the conversation and messages expired. Collect the referenced file ids (conversation.files plus one pass over the chat's messages) and extend the file caps to match file_id as well as conversationId, in the single cascade, saveConvo, the bulk tag/project cascade, and both sweep passes. The id scan runs at conversion/alignment time only — post-conversion uploads always receive a deadline at upload, so conforming- parent writes keep the cheap conversationId-scoped cap. A file shared across chats is capped to the earliest converting chat's deadline, consistent with cap-don't-extend. --- .../data-schemas/src/methods/conversation.ts | 26 +++- .../data-schemas/src/methods/message.spec.ts | 83 +++++++++++ packages/data-schemas/src/utils/retention.ts | 131 ++++++++++++++---- 3 files changed, 211 insertions(+), 29 deletions(-) diff --git a/packages/data-schemas/src/methods/conversation.ts b/packages/data-schemas/src/methods/conversation.ts index e24ab2395c..0774e12ec8 100644 --- a/packages/data-schemas/src/methods/conversation.ts +++ b/packages/data-schemas/src/methods/conversation.ts @@ -15,6 +15,8 @@ import { capConversationFiles, capConversationSharedLinks, capForcedRetentionExpiry, + collectConversationFileIds, + conversationNeedsForcedRetention, createFallbackRetentionDate, forceConversationMessagesTemporary, } from '~/utils/retention'; @@ -255,15 +257,20 @@ export function createConversationMethods( Object.prototype.hasOwnProperty.call(update, 'chatProjectId') || Object.prototype.hasOwnProperty.call(unsetFields, 'chatProjectId'); let previousChatProjectId: string | null = null; - let parentRetention: { isTemporary?: boolean | null; expiredAt?: Date | null } | null = null; + let parentRetention: { + isTemporary?: boolean | null; + expiredAt?: Date | null; + files?: string[]; + } | null = null; if (mayChangeProjectMembership || isForcedRetention) { const existing = await Conversation.findOne( { conversationId, user: userId }, - 'chatProjectId isTemporary expiredAt', + 'chatProjectId isTemporary expiredAt files', ).lean<{ chatProjectId?: string | null; isTemporary?: boolean | null; expiredAt?: Date | null; + files?: string[]; } | null>(); previousChatProjectId = existing?.chatProjectId ?? null; parentRetention = existing; @@ -326,9 +333,22 @@ export function createConversationMethods( const Message = mongoose.models.Message as Model; const SharedLink = mongoose.models.SharedLink as Model; const File = mongoose.models.File as Model; + /** + * Referenced file ids (message-attachment rows carry no conversationId) are collected + * only at conversion time: post-conversion uploads always receive a deadline at upload, + * so the id scan over the chat's messages is not needed on every conforming save. + */ + const fileIds = conversationNeedsForcedRetention(parentRetention, forcedExpiredAt) + ? await collectConversationFileIds( + Message, + userId, + [conversationId], + parentRetention.files, + ) + : []; await forceConversationMessagesTemporary(Message, userId, conversationId, forcedExpiredAt); await capConversationSharedLinks(SharedLink, userId, conversationId, forcedExpiredAt); - await capConversationFiles(File, conversationId, forcedExpiredAt); + await capConversationFiles(File, conversationId, forcedExpiredAt, fileIds); } const createdAtOnInsert = diff --git a/packages/data-schemas/src/methods/message.spec.ts b/packages/data-schemas/src/methods/message.spec.ts index 87b1dcf1a2..ca0694cf7b 100644 --- a/packages/data-schemas/src/methods/message.spec.ts +++ b/packages/data-schemas/src/methods/message.spec.ts @@ -1638,6 +1638,47 @@ describe('Message Operations', () => { expect(parent?.expiredAt?.getTime()).toBe(parentDeadline.getTime()); }); + it('aligns referenced attachment files that carry no conversationId', async () => { + const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); + const parentDeadline = new Date(Date.now() + 60 * 60 * 1000); + const conversationId = uuidv4(); + const messageFileId = uuidv4(); + + await Conversation().create({ + conversationId, + user: 'user123', + endpoint: 'openAI', + isTemporary: true, + expiredAt: parentDeadline, + }); + await Message.create({ + messageId: uuidv4(), + conversationId, + user: 'user123', + text: 'with attachment', + isTemporary: true, + expiredAt: parentDeadline, + files: [{ file_id: messageFileId, filename: 'doc.pdf' }], + }); + await File().collection.insertOne({ + file_id: messageFileId, + user: new mongoose.Types.ObjectId(), + expiredAt: null, + }); + + const result = await sweepForcedRetention( + Conversation(), + Message, + SharedLink(), + File(), + forcedExpiredAt, + ); + expect(result).toEqual({ conversations: 0, aligned: 1, errors: 0, projects: [] }); + + const file = await File().findOne({ file_id: messageFileId }).lean(); + expect(file?.expiredAt?.getTime()).toBe(parentDeadline.getTime()); + }); + it('collects converted project memberships so callers can refresh project stats', async () => { const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); const chatProjectId = new mongoose.Types.ObjectId().toString(); @@ -1819,6 +1860,48 @@ describe('Message Operations', () => { await File().deleteMany({}); }); + it('caps referenced attachment files that carry no conversationId when converting', async () => { + const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); + const conversationId = uuidv4(); + const messageFileId = uuidv4(); + const convoFileId = uuidv4(); + + await Conversation().create({ + conversationId, + user: 'user123', + endpoint: 'openAI', + isTemporary: false, + files: [convoFileId], + }); + await Message.create({ + messageId: uuidv4(), + conversationId, + user: 'user123', + text: 'with attachment', + files: [{ file_id: messageFileId, filename: 'doc.pdf' }], + }); + await File().collection.insertMany([ + { file_id: messageFileId, user: new mongoose.Types.ObjectId(), expiredAt: null }, + { file_id: convoFileId, user: new mongoose.Types.ObjectId(), expiredAt: null }, + ]); + + await cascadeForcedConversationRetention( + Conversation(), + Message, + SharedLink(), + File(), + 'user123', + conversationId, + forcedExpiredAt, + ); + + const messageFile = await File().findOne({ file_id: messageFileId }).lean(); + expect(messageFile?.expiredAt?.getTime()).toBe(forcedExpiredAt.getTime()); + + const convoFile = await File().findOne({ file_id: convoFileId }).lean(); + expect(convoFile?.expiredAt?.getTime()).toBe(forcedExpiredAt.getTime()); + }); + it('caps lagging children even when the parent conversation already conforms', async () => { const forcedExpiredAt = new Date(Date.now() + 24 * 60 * 60 * 1000); const parentDeadline = new Date(Date.now() + 60 * 60 * 1000); diff --git a/packages/data-schemas/src/utils/retention.ts b/packages/data-schemas/src/utils/retention.ts index 1f764a4fc1..3c9a6615bd 100644 --- a/packages/data-schemas/src/utils/retention.ts +++ b/packages/data-schemas/src/utils/retention.ts @@ -161,6 +161,46 @@ export const capConversationSharedLinks = async ( return result.modifiedCount ?? 0; }; +/** + * Collects the file ids a set of conversations references. Message-attachment uploads create + * File rows without a `conversationId` — they are referenced only from `Message.files[].file_id` + * and the conversation's own `files` array — so conversation-scoped file caps must also target + * these ids. `seedFileIds` takes the conversations' `files` arrays; message references are read + * in one pass over the conversations' messages. + */ +export const collectConversationFileIds = async ( + Message: Model, + userId: string, + conversationIds: string[], + seedFileIds?: Iterable, +): Promise => { + const fileIds = new Set(); + for (const fileId of seedFileIds ?? []) { + if (typeof fileId === 'string' && fileId.length > 0) { + fileIds.add(fileId); + } + } + if (conversationIds.length > 0) { + const messages = await Message.find( + { + user: userId, + conversationId: { $in: conversationIds }, + files: { $exists: true, $ne: null }, + }, + 'files', + ).lean }>>(); + for (const message of messages) { + for (const file of message.files ?? []) { + const fileId = file?.file_id; + if (typeof fileId === 'string' && fileId.length > 0) { + fileIds.add(fileId); + } + } + } + } + return [...fileIds]; +}; + /** * Caps a conversation's uploaded files to the forced deadline. Files use a retention-scoped * `expiredAt` swept by application code (`getExpiredFiles` only sweeps files whose own `expiredAt` @@ -170,20 +210,27 @@ export const capConversationSharedLinks = async ( * already expires sooner. Under ephemeral retention every conversation-scoped file is meant to * expire (persistent agent files are not retained), so no agent-file exclusion is needed here. * - * Scoped by `conversationId` alone: it is a globally unique per-conversation id, and unlike the - * message/share caps the `File.user` field is an `ObjectId` rather than the string user id, so - * filtering by user would require a cast the callers cannot guarantee. + * Matches by `conversationId` (a globally unique per-conversation id; unlike the message/share + * caps the `File.user` field is an `ObjectId` rather than the string user id, so filtering by + * user would require a cast the callers cannot guarantee) plus any referenced `fileIds` — + * message-attachment rows carry no `conversationId`, so conversion-time callers collect the ids + * via {@link collectConversationFileIds}. A shared file referenced from several chats is capped + * to the earliest converting chat's deadline, consistent with cap-don't-extend. */ export const capConversationFiles = async ( File: Model, conversationId: string, forcedExpiredAt: Date, + fileIds: string[] = [], ): Promise => { + const scope = + fileIds.length > 0 + ? { $or: [{ conversationId }, { file_id: { $in: fileIds } }] } + : { conversationId }; const result = await File.updateMany( { - conversationId, - $or: [{ expiredAt: null }, { expiredAt: { $gt: forcedExpiredAt } }], - }, + $and: [scope, { $or: [{ expiredAt: null }, { expiredAt: { $gt: forcedExpiredAt } }] }], + } as FilterQuery, { $set: { expiredAt: forcedExpiredAt } }, ); return result.modifiedCount ?? 0; @@ -239,12 +286,21 @@ export const cascadeForcedConversationRetention = async ( ): Promise => { const parent = await Conversation.findOne( { conversationId, user: userId }, - 'isTemporary expiredAt', - ).lean(); + 'isTemporary expiredAt files', + ).lean<(RetentionFilterDocument & { files?: string[] }) | null>(); if (parent == null) { return false; } const expiredAt = capForcedRetentionExpiry(parent.expiredAt, forcedExpiredAt); + const needsConversion = conversationNeedsForcedRetention(parent, expiredAt); + /** + * Referenced file ids (message-attachment rows carry no conversationId) are collected only at + * conversion time: post-conversion uploads always receive a deadline at upload, so the id + * scan over the chat's messages is not needed on every conforming-parent write. + */ + const fileIds = needsConversion + ? await collectConversationFileIds(Message, userId, [conversationId], parent.files) + : []; /** * Cap the dependent messages, shares, and files independently of the parent gap check, and * before the parent conversion. An already-conforming parent can still own lagging children @@ -255,8 +311,8 @@ export const cascadeForcedConversationRetention = async ( */ await forceConversationMessagesTemporary(Message, userId, conversationId, expiredAt); await capConversationSharedLinks(SharedLink, userId, conversationId, expiredAt); - await capConversationFiles(File, conversationId, expiredAt); - if (!conversationNeedsForcedRetention(parent, expiredAt)) { + await capConversationFiles(File, conversationId, expiredAt, fileIds); + if (!needsConversion) { return false; } const convoResult = await Conversation.updateOne( @@ -286,13 +342,16 @@ const cascadeForcedRetentionForConversationSet = async ( ): Promise => { const conversations = await Conversation.find( { user: userId, ...conversationMatch } as FilterQuery, - 'conversationId isTemporary expiredAt', - ).lean>(); + 'conversationId isTemporary expiredAt files', + ).lean>(); if (conversations.length === 0) { return; } - const retentionBuckets = new Map(); + const retentionBuckets = new Map< + number, + { expiredAt: Date; conversationIds: string[]; seedFileIds: string[] } + >(); for (const convo of conversations) { if (typeof convo.conversationId !== 'string' || convo.conversationId.length === 0) { continue; @@ -300,15 +359,19 @@ const cascadeForcedRetentionForConversationSet = async ( const expiredAt = capForcedRetentionExpiry(convo.expiredAt, forcedExpiredAt); const key = expiredAt.getTime(); - const bucket = retentionBuckets.get(key); - if (bucket) { - bucket.conversationIds.push(convo.conversationId); - continue; + const bucket = retentionBuckets.get(key) ?? { + expiredAt, + conversationIds: [], + seedFileIds: [], + }; + bucket.conversationIds.push(convo.conversationId); + for (const fileId of convo.files ?? []) { + bucket.seedFileIds.push(fileId); } - retentionBuckets.set(key, { expiredAt, conversationIds: [convo.conversationId] }); + retentionBuckets.set(key, bucket); } - for (const { expiredAt, conversationIds } of retentionBuckets.values()) { + for (const { expiredAt, conversationIds, seedFileIds } of retentionBuckets.values()) { await Conversation.updateMany( { user: userId, @@ -340,11 +403,15 @@ const cascadeForcedRetentionForConversationSet = async ( }, { $set: { expiredAt } }, ); + const fileIds = await collectConversationFileIds(Message, userId, conversationIds, seedFileIds); + const fileScope = + fileIds.length > 0 + ? { $or: [{ conversationId: { $in: conversationIds } }, { file_id: { $in: fileIds } }] } + : { conversationId: { $in: conversationIds } }; await File.updateMany( { - conversationId: { $in: conversationIds }, - $or: [{ expiredAt: null }, { expiredAt: { $gt: expiredAt } }], - }, + $and: [fileScope, { $or: [{ expiredAt: null }, { expiredAt: { $gt: expiredAt } }] }], + } as FilterQuery, { $set: { expiredAt } }, ); } @@ -447,7 +514,7 @@ export const sweepForcedRetention = async ( isTemporary: true, expiredAt: { $ne: null, $lte: forcedExpiredAt }, } as FilterQuery) - .select('conversationId user expiredAt') + .select('conversationId user expiredAt files') .lean() .cursor(); @@ -462,10 +529,16 @@ export const sweepForcedRetention = async ( continue; } try { + const fileIds = await collectConversationFileIds( + Message, + user, + [conversationId], + convo.files, + ); const changed = (await forceConversationMessagesTemporary(Message, user, conversationId, expiredAt)) + (await capConversationSharedLinks(SharedLink, user, conversationId, expiredAt)) + - (await capConversationFiles(File, conversationId, expiredAt)); + (await capConversationFiles(File, conversationId, expiredAt, fileIds)); if (changed > 0) { result.aligned += 1; } @@ -475,7 +548,7 @@ export const sweepForcedRetention = async ( } const cursor = Conversation.find(forcedRetentionGapFilter(forcedExpiredAt)) - .select('_id conversationId user expiredAt chatProjectId') + .select('_id conversationId user expiredAt chatProjectId files') .lean() .cursor(); @@ -486,6 +559,12 @@ export const sweepForcedRetention = async ( } try { const expiredAt = capForcedRetentionExpiry(convo.expiredAt, forcedExpiredAt); + const fileIds = await collectConversationFileIds( + Message, + user, + [conversationId], + convo.files, + ); /** * Convert the dependent messages, shares, and files before marking the conversation itself * conforming. If a child backfill throws, the conversation stays non-conforming so the @@ -493,7 +572,7 @@ export const sweepForcedRetention = async ( */ await forceConversationMessagesTemporary(Message, user, conversationId, expiredAt); await capConversationSharedLinks(SharedLink, user, conversationId, expiredAt); - await capConversationFiles(File, conversationId, expiredAt); + await capConversationFiles(File, conversationId, expiredAt, fileIds); await Conversation.updateOne({ _id: convo._id }, { $set: { isTemporary: true, expiredAt } }); result.conversations += 1; if (typeof chatProjectId === 'string' && chatProjectId.length > 0) {