diff --git a/packages/data-schemas/src/methods/message.spec.ts b/packages/data-schemas/src/methods/message.spec.ts index 85353eb6e38..493783485a9 100644 --- a/packages/data-schemas/src/methods/message.spec.ts +++ b/packages/data-schemas/src/methods/message.spec.ts @@ -59,6 +59,7 @@ let claimBackgroundToolResults: ReturnType< let releaseBackgroundToolResultClaims: ReturnType< typeof createMessageMethods >['releaseBackgroundToolResultClaims']; +let getMessagesByCursor: ReturnType['getMessagesByCursor']; function rejectUpdateArrays(beforeUpdate?: () => Promise) { const originalFindOneAndUpdate = Message.collection.findOneAndUpdate.bind(Message.collection); @@ -98,6 +99,7 @@ beforeAll(async () => { releaseSubagentTaskResultClaim = methods.releaseSubagentTaskResultClaim; claimBackgroundToolResults = methods.claimBackgroundToolResults; releaseBackgroundToolResultClaims = methods.releaseBackgroundToolResultClaims; + getMessagesByCursor = methods.getMessagesByCursor; await mongoose.connect(mongoUri); }); @@ -4162,16 +4164,18 @@ describe('Message Operations', () => { }; /** - * Simulates the pagination logic from api/server/routes/messages.js - * This tests the exact query pattern used in the route + * Maps the route-shaped arguments this suite uses onto production + * `getMessagesByCursor`. Sort-field whitelisting stays here because it + * belongs to the route, not the data method. */ - const getMessagesByCursor = async ({ + const getMessagesByCursorPage = async ({ conversationId, user, pageSize = 25, cursor = null as string | null, sortBy = 'createdAt', sortDirection = 'desc', + select, }: { conversationId: string; user: string; @@ -4179,31 +4183,14 @@ describe('Message Operations', () => { cursor?: string | null; sortBy?: string; sortDirection?: string; + select?: string; }) => { - const sortOrder = sortDirection === 'asc' ? 1 : -1; + const sortOrder: 1 | -1 = sortDirection === 'asc' ? 1 : -1; const sortField = ['createdAt', 'updatedAt'].includes(sortBy) ? sortBy : 'createdAt'; - const cursorOperator = sortDirection === 'asc' ? '$gt' : '$lt'; - - const filter: Record = { conversationId, user }; - if (cursor) { - filter[sortField] = { [cursorOperator]: new Date(cursor) }; - } - - const messages = await Message.find(filter) - .sort({ [sortField]: sortOrder }) - .limit(pageSize + 1) - .lean(); - - let nextCursor: string | null = null; - if (messages.length > pageSize) { - messages.pop(); // Remove extra item used to detect next page - // Create cursor from the last RETURNED item (not the popped one) - nextCursor = (messages[messages.length - 1] as Record)[ - sortField - ] as string; - } - - return { messages, nextCursor }; + return getMessagesByCursor( + { conversationId, user }, + { sortField, sortOrder, limit: pageSize, cursor, select }, + ); }; it('should return messages for a conversation with pagination', async () => { @@ -4217,7 +4204,7 @@ describe('Message Operations', () => { } // Fetch first page (pageSize 25) - const page1 = await getMessagesByCursor({ + const page1 = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 25, @@ -4227,7 +4214,7 @@ describe('Message Operations', () => { expect(page1.nextCursor).toBeTruthy(); // Fetch second page using cursor - const page2 = await getMessagesByCursor({ + const page2 = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 25, @@ -4265,7 +4252,7 @@ describe('Message Operations', () => { const item26 = messages[25]; // Fetch first page with pageSize 25 - const page1 = await getMessagesByCursor({ + const page1 = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 25, @@ -4279,7 +4266,7 @@ describe('Message Operations', () => { expect(page1Ids).not.toContain(item26!.messageId); // Fetch second page - const page2 = await getMessagesByCursor({ + const page2 = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 25, @@ -4311,7 +4298,7 @@ describe('Message Operations', () => { new Date('2026-01-03T00:00:00.000Z'), ); - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', }); @@ -4337,7 +4324,7 @@ describe('Message Operations', () => { new Date('2026-01-02T00:00:00.000Z'), ); - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', sortDirection: 'asc', @@ -4352,7 +4339,7 @@ describe('Message Operations', () => { it('should handle empty conversation', async () => { const conversationId = uuidv4(); - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', }); @@ -4385,7 +4372,7 @@ describe('Message Operations', () => { updatedAt: createdAt, }); - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', }); @@ -4405,7 +4392,7 @@ describe('Message Operations', () => { await createMessageWithTimestamp(i, conversationId, createdAt); } - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 25, @@ -4430,7 +4417,7 @@ describe('Message Operations', () => { const allMessages: unknown[] = []; for (let page = 0; page < 5; page++) { - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 1, @@ -4462,7 +4449,7 @@ describe('Message Operations', () => { messages.push(msg); } - const result = await getMessagesByCursor({ + const result = await getMessagesByCursorPage({ conversationId, user: 'user123', pageSize: 10, @@ -4471,6 +4458,100 @@ describe('Message Operations', () => { // All messages should be returned expect(result?.messages).toHaveLength(5); }); + + const expectNoEqualTimestampGaps = async (sortDirection: 'asc' | 'desc') => { + const conversationId = uuidv4(); + const sameTime = new Date('2026-01-01T12:00:00.000Z'); + + for (let i = 0; i < 30; i++) { + await createMessageWithTimestamp(i, conversationId, sameTime); + } + + const page1 = await getMessagesByCursorPage({ + conversationId, + user: 'user123', + pageSize: 25, + sortDirection, + select: CLIENT_MESSAGE_SELECT, + }); + + expect(page1.messages).toHaveLength(25); + expect(page1.nextCursor).toBeTruthy(); + expect(page1.messages[0]).not.toHaveProperty('_id'); + + const page2 = await getMessagesByCursorPage({ + conversationId, + user: 'user123', + pageSize: 25, + sortDirection, + cursor: page1.nextCursor, + select: CLIENT_MESSAGE_SELECT, + }); + + expect(page2.messages).toHaveLength(5); + + const allMessageIds = [ + ...page1.messages.map((message) => message.messageId), + ...page2.messages.map((message) => message.messageId), + ]; + expect(allMessageIds).toHaveLength(30); + expect(new Set(allMessageIds).size).toBe(30); + }; + + it('should not skip equal-timestamp rows at the page boundary', async () => { + await expectNoEqualTimestampGaps('desc'); + }); + + it('should not skip equal-timestamp rows at the page boundary when sorting ascending', async () => { + await expectNoEqualTimestampGaps('asc'); + }); + + it('should resume from a legacy scalar createdAt cursor', async () => { + const conversationId = uuidv4(); + const newest = new Date('2026-01-03T00:00:00.000Z'); + const middle = new Date('2026-01-02T00:00:00.000Z'); + const oldest = new Date('2026-01-01T00:00:00.000Z'); + const msgNewest = await createMessageWithTimestamp(3, conversationId, newest); + const msgMiddle = await createMessageWithTimestamp(2, conversationId, middle); + const msgOldest = await createMessageWithTimestamp(1, conversationId, oldest); + + for (const legacyCursor of [newest.toISOString(), String(newest)]) { + const result = await getMessagesByCursorPage({ + conversationId, + user: 'user123', + cursor: legacyCursor, + sortDirection: 'desc', + }); + + expect(result.messages.map((message) => message.messageId)).toEqual([ + msgMiddle!.messageId, + msgOldest!.messageId, + ]); + } + }); + + it('should start from the beginning when the cursor is not usable', async () => { + const conversationId = uuidv4(); + await createMessageWithTimestamp(1, conversationId, new Date('2026-01-01T00:00:00.000Z')); + await createMessageWithTimestamp(2, conversationId, new Date('2026-01-02T00:00:00.000Z')); + + const malformed = Buffer.from( + JSON.stringify({ primary: 'not-a-date', id: 'not-an-object-id' }), + ).toString('base64'); + + for (const cursor of [malformed, 'not-a-date']) { + const result = await getMessagesByCursorPage({ + conversationId, + user: 'user123', + cursor, + }); + expect(result.messages).toHaveLength(2); + } + + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('[getMessagesByCursor] Invalid cursor format'), + ); + }); }); describe('tenantId stripping', () => { diff --git a/packages/data-schemas/src/methods/message.ts b/packages/data-schemas/src/methods/message.ts index 1799b90e578..f93b810ba54 100644 --- a/packages/data-schemas/src/methods/message.ts +++ b/packages/data-schemas/src/methods/message.ts @@ -1,9 +1,9 @@ +import { Types, type DeleteResult, type FilterQuery, type Model, type UpdateQuery } from 'mongoose'; import { backgroundResultMetadata, HITL_MESSAGE_FILTER_FIELDS, RetentionMode, } from 'librechat-data-provider'; -import type { DeleteResult, FilterQuery, Model, Types, UpdateQuery } from 'mongoose'; import type { UserSubmittedMessageFieldPath } from 'librechat-data-provider'; import type { SearchParams } from 'meilisearch'; import type { SchemaWithMeiliMethods } from '~/models/plugins/mongoMeili'; @@ -11,6 +11,7 @@ import type { AppConfig, IConversation, IMessage } from '~/types'; import { createChatExpirationDate, createTempChatExpirationDate } from '~/utils/tempChatRetention'; import { activeExpirationFilter, createFallbackRetentionDate } from '~/utils/retention'; import { tenantSafeBulkWrite } from '~/utils/tenantBulkWrite'; +import { isValidObjectIdString } from '~/utils/objectId'; import logger from '~/config/winston'; /** Simple UUID v4 regex to replace zod validation */ @@ -835,6 +836,8 @@ export interface MessageMethods { sortOrder?: 1 | -1; limit?: number; cursor?: string | null; + /** Projection for the page, e.g. `CLIENT_MESSAGE_SELECT` for client-facing reads. */ + select?: string; }, ): Promise<{ messages: IMessage[]; nextCursor: string | null }>; searchMessages( @@ -902,6 +905,99 @@ type SteplessToolCallFallback = { hasResultClaim: boolean; }; + +type MessagesCursor = { primary: string | null; id: string }; + +/** + * Message list cursors carry the sort value and the `_id` that broke its tie, + * base64-encoded so callers treat them as opaque. Older plain-value cursors + * decode to null here and fall back to a single-field boundary. + */ +function decodeMessagesCursor(pageParam: string): MessagesCursor | 'invalid' | null { + try { + const decoded = JSON.parse(Buffer.from(pageParam, 'base64').toString('utf8')) as { + primary?: unknown; + id?: unknown; + }; + if (decoded && typeof decoded === 'object' && ('primary' in decoded || 'id' in decoded)) { + const hasPrimary = typeof decoded.primary === 'string' || decoded.primary === null; + if (hasPrimary && typeof decoded.id === 'string' && isValidObjectIdString(decoded.id)) { + return { primary: decoded.primary as string | null, id: decoded.id }; + } + return 'invalid'; + } + } catch { + /* not a composite cursor */ + } + return null; +} + +function encodeMessagesCursor( + message: IMessage & { _id?: Types.ObjectId }, + sortField: string, + dateSortField: boolean, +): string { + const raw = + sortField === 'createdAt' ? message.createdAt : message[sortField as keyof IMessage]; + let primary: string | null = null; + if (raw instanceof Date) { + primary = raw.toISOString(); + } else if (raw != null) { + if (dateSortField) { + const asDate = new Date(raw as string | number | Date); + primary = Number.isNaN(asDate.getTime()) ? String(raw) : asDate.toISOString(); + } else { + primary = String(raw); + } + } + return Buffer.from(JSON.stringify({ primary, id: String(message._id) })).toString('base64'); +} + +function buildMessagesCursorFilter( + cursor: string, + sortField: string, + op: '$lt' | '$gt', + dateSortField: boolean, +): FilterQuery | null { + const decoded = decodeMessagesCursor(cursor); + if (decoded === 'invalid') { + logger.warn('[getMessagesByCursor] Invalid cursor format, starting from beginning'); + return null; + } + + if (decoded) { + const boundaryId = { [op]: new Types.ObjectId(decoded.id) }; + if (decoded.primary === null) { + return { + $or: [{ [sortField]: null, _id: boundaryId }], + } as FilterQuery; + } + const primaryValue = dateSortField ? new Date(decoded.primary) : decoded.primary; + if (dateSortField && Number.isNaN((primaryValue as Date).getTime())) { + logger.warn('[getMessagesByCursor] Invalid cursor format, starting from beginning'); + return null; + } + return { + $or: [ + { [sortField]: { [op]: primaryValue } }, + { [sortField]: primaryValue, _id: boundaryId }, + ], + } as FilterQuery; + } + + /* Legacy scalar cursor: single-field comparison. */ + if (dateSortField) { + const asDate = new Date(cursor); + if (Number.isNaN(asDate.getTime())) { + logger.warn('[getMessagesByCursor] Invalid cursor format, starting from beginning'); + return null; + } + return { [sortField]: { [op]: asDate } } as FilterQuery; + } + + return { [sortField]: { [op]: cursor } } as FilterQuery; +} + export function createMessageMethods(mongoose: typeof import('mongoose')): MessageMethods { /** * Saves a message in the database. @@ -3700,6 +3796,11 @@ export function createMessageMethods(mongoose: typeof import('mongoose')): Messa /** * Retrieves paginated messages with custom sorting and cursor support. + * + * Cursors are composite `{ primary, id }` values (base64 JSON) so rows that + * share the sort-field timestamp are not skipped at a page boundary. Plain + * scalar cursors from older clients still resume with a single-field + * comparison when they parse cleanly. */ async function getMessagesByCursor( filter: FilterQuery, @@ -3714,28 +3815,49 @@ export function createMessageMethods(mongoose: typeof import('mongoose')): Messa ) { const Message = mongoose.models.Message as Model; const { sortField = 'createdAt', sortOrder = -1, limit = 25, cursor, select } = options; - const queryFilter = { ...filter }; + const descending = sortOrder !== 1; + const op = descending ? '$lt' : '$gt'; + const dateSortField = sortField === 'createdAt' || sortField === 'updatedAt'; + + let queryFilter: FilterQuery = { ...filter }; if (cursor) { - queryFilter[sortField] = sortOrder === 1 ? { $gt: cursor } : { $lt: cursor }; + const cursorFilter = buildMessagesCursorFilter(cursor, sortField, op, dateSortField); + if (cursorFilter) { + queryFilter = { $and: [filter, cursorFilter] } as FilterQuery; + } } + + /* CLIENT_MESSAGE_SELECT excludes `_id`, but the composite cursor needs it + as the tie-breaker. Fetch with `_id`, then strip it before returning. */ + const stripsId = + typeof select === 'string' && /(^|\s)-_id(\s|$)/.test(select); + const querySelect = stripsId + ? select.replace(/(^|\s)-_id(\s|$)/g, ' ').replace(/\s+/g, ' ').trim() + : select; + const query = Message.find(queryFilter); - if (select) { - query.select(select); + if (querySelect) { + query.select(querySelect); } const messages = await query - .sort({ [sortField]: sortOrder }) + .sort({ [sortField]: sortOrder, _id: sortOrder }) .limit(limit + 1) - .lean(); + .lean<(IMessage & { _id?: Types.ObjectId })[]>(); let nextCursor: string | null = null; if (messages.length > limit) { messages.pop(); const last = messages[messages.length - 1]; - const cursorValue = - sortField === 'createdAt' ? last.createdAt : last[sortField as keyof IMessage]; - nextCursor = String(cursorValue ?? ''); + nextCursor = encodeMessagesCursor(last, sortField, dateSortField); } - return { messages, nextCursor }; + + if (stripsId) { + for (const message of messages) { + delete (message as { _id?: Types.ObjectId })._id; + } + } + + return { messages: messages as IMessage[], nextCursor }; } /**