mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-27 04:07:05 +00:00
🧵 fix: Close Child-Thread Read and Search-Cleanup Gaps (#15055)
* fix: close child thread read and cleanup gaps * fix: preserve child search cleanup invariants * test: complete mocked update result * perf: parallelize scoped message reads * fix: close child thread compatibility gaps * test: expect preserved cleanup failure * fix: reconcile legacy Meili cleanup markers
This commit is contained in:
Danny Avila
GitHub
parent
c7e355b219
commit
d0f9d5625e
14 changed files with 732 additions and 115 deletions
|
|
@ -230,44 +230,54 @@ async function performSync(flowManager, flowId, flowType) {
|
|||
await batchResetMeiliFlags(Conversation.collection);
|
||||
}
|
||||
|
||||
// Check if we need to sync messages
|
||||
logger.info('[indexSync] Requesting message sync progress...');
|
||||
const messageProgress = await Message.getSyncProgress();
|
||||
if (!messageProgress.isComplete || settingsUpdated) {
|
||||
logger.info(
|
||||
`[indexSync] Messages need syncing: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments} indexed`,
|
||||
);
|
||||
|
||||
const messageCount = messageProgress.totalDocuments;
|
||||
const messagesIndexed = messageProgress.totalProcessed;
|
||||
const unindexedMessages = messageCount - messagesIndexed;
|
||||
const messagesPendingCleanup = messageProgress.pendingCleanup ?? 0;
|
||||
const noneIndexed = messagesIndexed === 0 && unindexedMessages > 0;
|
||||
|
||||
if (
|
||||
settingsUpdated ||
|
||||
noneIndexed ||
|
||||
unindexedMessages > syncThreshold ||
|
||||
messagesPendingCleanup > 0
|
||||
) {
|
||||
if (noneIndexed && !settingsUpdated) {
|
||||
logger.info('[indexSync] No messages marked as indexed, forcing full sync');
|
||||
}
|
||||
let messageSyncError;
|
||||
try {
|
||||
// Check if we need to sync messages
|
||||
logger.info('[indexSync] Requesting message sync progress...');
|
||||
const messageProgress = await Message.getSyncProgress();
|
||||
if (!messageProgress.isComplete || settingsUpdated) {
|
||||
logger.info(
|
||||
messagesPendingCleanup > 0
|
||||
? `[indexSync] Starting message sync (${unindexedMessages} unindexed, ${messagesPendingCleanup} pending cleanup)`
|
||||
: `[indexSync] Starting message sync (${unindexedMessages} unindexed)`,
|
||||
`[indexSync] Messages need syncing: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments} indexed`,
|
||||
);
|
||||
await Message.syncWithMeili();
|
||||
messagesSync = true;
|
||||
} else if (unindexedMessages > 0) {
|
||||
|
||||
const messageCount = messageProgress.totalDocuments;
|
||||
const messagesIndexed = messageProgress.totalProcessed;
|
||||
const unindexedMessages = messageCount - messagesIndexed;
|
||||
const messagesPendingCleanup = messageProgress.pendingCleanup ?? 0;
|
||||
const noneIndexed = messagesIndexed === 0 && unindexedMessages > 0;
|
||||
|
||||
if (settingsUpdated || noneIndexed || unindexedMessages > syncThreshold) {
|
||||
if (noneIndexed && !settingsUpdated) {
|
||||
logger.info('[indexSync] No messages marked as indexed, forcing full sync');
|
||||
}
|
||||
logger.info(
|
||||
messagesPendingCleanup > 0
|
||||
? `[indexSync] Starting message sync (${unindexedMessages} unindexed, ${messagesPendingCleanup} pending cleanup)`
|
||||
: `[indexSync] Starting message sync (${unindexedMessages} unindexed)`,
|
||||
);
|
||||
await Message.syncWithMeili();
|
||||
messagesSync = true;
|
||||
} else if (messagesPendingCleanup > 0) {
|
||||
logger.info(
|
||||
`[indexSync] Cleaning ${messagesPendingCleanup} excluded messages from search`,
|
||||
);
|
||||
await Message.cleanupExcludedMeiliIndex();
|
||||
messagesSync = true;
|
||||
} else if (unindexedMessages > 0) {
|
||||
logger.info(
|
||||
`[indexSync] ${unindexedMessages} messages unindexed (below threshold: ${syncThreshold}, skipping)`,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
logger.info(
|
||||
`[indexSync] ${unindexedMessages} messages unindexed (below threshold: ${syncThreshold}, skipping)`,
|
||||
`[indexSync] Messages are fully synced: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments}`,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
logger.info(
|
||||
`[indexSync] Messages are fully synced: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments}`,
|
||||
} catch (error) {
|
||||
messageSyncError = error;
|
||||
logger.error(
|
||||
'[indexSync] Message reconciliation failed; continuing with conversations:',
|
||||
error,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -284,12 +294,7 @@ async function performSync(flowManager, flowId, flowType) {
|
|||
const convosPendingCleanup = convoProgress.pendingCleanup ?? 0;
|
||||
const noneConvosIndexed = convosIndexed === 0 && unindexedConvos > 0;
|
||||
|
||||
if (
|
||||
settingsUpdated ||
|
||||
noneConvosIndexed ||
|
||||
unindexedConvos > syncThreshold ||
|
||||
convosPendingCleanup > 0
|
||||
) {
|
||||
if (settingsUpdated || noneConvosIndexed || unindexedConvos > syncThreshold) {
|
||||
if (noneConvosIndexed && !settingsUpdated) {
|
||||
logger.info('[indexSync] No conversations marked as indexed, forcing full sync');
|
||||
}
|
||||
|
|
@ -300,6 +305,12 @@ async function performSync(flowManager, flowId, flowType) {
|
|||
);
|
||||
await Conversation.syncWithMeili();
|
||||
convosSync = true;
|
||||
} else if (convosPendingCleanup > 0) {
|
||||
logger.info(
|
||||
`[indexSync] Cleaning ${convosPendingCleanup} excluded conversations from search`,
|
||||
);
|
||||
await Conversation.cleanupExcludedMeiliIndex();
|
||||
convosSync = true;
|
||||
} else if (unindexedConvos > 0) {
|
||||
logger.info(
|
||||
`[indexSync] ${unindexedConvos} convos unindexed (below threshold: ${syncThreshold}, skipping)`,
|
||||
|
|
@ -311,6 +322,10 @@ async function performSync(flowManager, flowId, flowType) {
|
|||
);
|
||||
}
|
||||
|
||||
if (messageSyncError) {
|
||||
throw messageSyncError;
|
||||
}
|
||||
|
||||
return { messagesSync, convosSync };
|
||||
} finally {
|
||||
if (indexingDisabled === true) {
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ const createMockModel = (collectionName) => ({
|
|||
collection: { name: collectionName },
|
||||
getSyncProgress: jest.fn(),
|
||||
syncWithMeili: jest.fn(),
|
||||
cleanupExcludedMeiliIndex: jest.fn(),
|
||||
countDocuments: jest.fn(),
|
||||
});
|
||||
|
||||
|
|
@ -528,7 +529,7 @@ describe('performSync() - syncThreshold logic', () => {
|
|||
);
|
||||
});
|
||||
|
||||
test('forces sync when search contains documents that are now excluded', async () => {
|
||||
test('runs bounded cleanup when search contains documents that are now excluded', async () => {
|
||||
Message.getSyncProgress.mockResolvedValue({
|
||||
totalProcessed: 100,
|
||||
totalDocuments: 100,
|
||||
|
|
@ -545,10 +546,61 @@ describe('performSync() - syncThreshold logic', () => {
|
|||
const indexSync = require('./indexSync');
|
||||
await indexSync();
|
||||
|
||||
expect(Message.syncWithMeili).toHaveBeenCalledTimes(1);
|
||||
expect(Message.syncWithMeili).not.toHaveBeenCalled();
|
||||
expect(Message.cleanupExcludedMeiliIndex).toHaveBeenCalledTimes(1);
|
||||
expect(Conversation.syncWithMeili).not.toHaveBeenCalled();
|
||||
expect(mockLogger.info).toHaveBeenCalledWith(
|
||||
'[indexSync] Starting message sync (0 unindexed, 1 pending cleanup)',
|
||||
'[indexSync] Cleaning 1 excluded messages from search',
|
||||
);
|
||||
});
|
||||
|
||||
test('does not start cleanup for excluded documents that were never indexed', async () => {
|
||||
Message.getSyncProgress.mockResolvedValue({
|
||||
totalProcessed: 100,
|
||||
totalDocuments: 100,
|
||||
pendingCleanup: 0,
|
||||
isComplete: true,
|
||||
});
|
||||
Conversation.getSyncProgress.mockResolvedValue({
|
||||
totalProcessed: 50,
|
||||
totalDocuments: 50,
|
||||
pendingCleanup: 0,
|
||||
isComplete: true,
|
||||
});
|
||||
|
||||
const indexSync = require('./indexSync');
|
||||
await indexSync();
|
||||
|
||||
expect(Message.syncWithMeili).not.toHaveBeenCalled();
|
||||
expect(Message.cleanupExcludedMeiliIndex).not.toHaveBeenCalled();
|
||||
expect(Conversation.syncWithMeili).not.toHaveBeenCalled();
|
||||
expect(Conversation.cleanupExcludedMeiliIndex).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('continues conversation cleanup when message cleanup fails transiently', async () => {
|
||||
const cleanupError = new Error('message cleanup timed out');
|
||||
Message.getSyncProgress.mockResolvedValue({
|
||||
totalProcessed: 100,
|
||||
totalDocuments: 100,
|
||||
pendingCleanup: 1,
|
||||
isComplete: false,
|
||||
});
|
||||
Message.cleanupExcludedMeiliIndex.mockRejectedValue(cleanupError);
|
||||
Conversation.getSyncProgress.mockResolvedValue({
|
||||
totalProcessed: 50,
|
||||
totalDocuments: 50,
|
||||
pendingCleanup: 1,
|
||||
isComplete: false,
|
||||
});
|
||||
|
||||
const indexSync = require('./indexSync');
|
||||
await expect(indexSync()).rejects.toThrow(cleanupError);
|
||||
|
||||
expect(Message.cleanupExcludedMeiliIndex).toHaveBeenCalledTimes(1);
|
||||
expect(Conversation.cleanupExcludedMeiliIndex).toHaveBeenCalledTimes(1);
|
||||
expect(mockLogger.error).toHaveBeenCalledWith(
|
||||
'[indexSync] Message reconciliation failed; continuing with conversations:',
|
||||
cleanupError,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -89,6 +89,48 @@ describe('validateMessageReq', () => {
|
|||
expect(next).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('returns not found for a direct child-thread message read', async () => {
|
||||
const req = {
|
||||
method: 'GET',
|
||||
params: { conversationId: 'child-convo', messageId: 'child-message' },
|
||||
body: {},
|
||||
user: { id: userId },
|
||||
};
|
||||
const res = createResponse();
|
||||
const next = jest.fn();
|
||||
getConvoOwnership.mockResolvedValue({
|
||||
user: userId,
|
||||
subagentThread: { parentConversationId: 'parent-convo' },
|
||||
});
|
||||
|
||||
await validateMessageReq(req, res, next);
|
||||
|
||||
expect(res.status).toHaveBeenCalledWith(404);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Conversation not found' });
|
||||
expect(next).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('returns not found for a HEAD request to a child thread', async () => {
|
||||
const req = {
|
||||
method: 'HEAD',
|
||||
params: { conversationId: 'child-convo' },
|
||||
body: {},
|
||||
user: { id: userId },
|
||||
};
|
||||
const res = createResponse();
|
||||
const next = jest.fn();
|
||||
getConvoOwnership.mockResolvedValue({
|
||||
user: userId,
|
||||
subagentThread: { parentConversationId: 'parent-convo' },
|
||||
});
|
||||
|
||||
await validateMessageReq(req, res, next);
|
||||
|
||||
expect(res.status).toHaveBeenCalledWith(404);
|
||||
expect(res.json).toHaveBeenCalledWith({ error: 'Conversation not found' });
|
||||
expect(next).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should allow message reads for an owned active generation job before the conversation is saved', async () => {
|
||||
const req = {
|
||||
method: 'GET',
|
||||
|
|
|
|||
|
|
@ -4,7 +4,11 @@ const validateRegistration = require('./validateRegistration');
|
|||
const buildEndpointOption = require('./buildEndpointOption');
|
||||
const validateEmailLogin = require('./validateEmailLogin');
|
||||
const validateMessageReq = require('./validateMessageReq');
|
||||
const { prepareMessageRequestValidation, sendValidationResponse } = require('./messageValidation');
|
||||
const {
|
||||
canReadActiveJobConversation,
|
||||
prepareMessageRequestValidation,
|
||||
sendValidationResponse,
|
||||
} = require('./messageValidation');
|
||||
const checkDomainAllowed = require('./checkDomainAllowed');
|
||||
const requireLocalAuth = require('./requireLocalAuth');
|
||||
const canDeleteAccount = require('./canDeleteAccount');
|
||||
|
|
@ -49,6 +53,7 @@ module.exports = {
|
|||
configMiddleware,
|
||||
checkDomainAllowed,
|
||||
validateMessageReq,
|
||||
canReadActiveJobConversation,
|
||||
sendValidationResponse,
|
||||
prepareMessageRequestValidation,
|
||||
buildEndpointOption,
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ jest.mock('~/server/services/Endpoints/agents/subagentThreadStore', () =>
|
|||
describe('Convos Routes', () => {
|
||||
let app;
|
||||
let convosRouter;
|
||||
const { deleteToolCalls, deleteConvos, saveConvo } = require('~/models');
|
||||
const { deleteToolCalls, deleteConvos, getConvo, saveConvo } = require('~/models');
|
||||
const {
|
||||
deleteAgentCheckpoints,
|
||||
deleteAllSharedLinksWithCleanup,
|
||||
|
|
@ -52,6 +52,35 @@ describe('Convos Routes', () => {
|
|||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
describe('GET /:conversationId', () => {
|
||||
it('returns an ordinary owned conversation', async () => {
|
||||
getConvo.mockResolvedValue({ conversationId: 'ordinary', title: 'Ordinary' });
|
||||
|
||||
const response = await request(app).get('/api/convos/ordinary');
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body).toEqual({ conversationId: 'ordinary', title: 'Ordinary' });
|
||||
expect(getConvo).toHaveBeenCalledWith('test-user-123', 'ordinary');
|
||||
});
|
||||
|
||||
it('returns the same not-found response for an owned child thread', async () => {
|
||||
getConvo.mockResolvedValue({
|
||||
conversationId: 'child',
|
||||
subagentThread: { parentConversationId: 'parent' },
|
||||
});
|
||||
|
||||
const childResponse = await request(app).get('/api/convos/child');
|
||||
getConvo.mockResolvedValue(null);
|
||||
const missingResponse = await request(app).get('/api/convos/missing');
|
||||
|
||||
expect(childResponse.status).toBe(404);
|
||||
expect(childResponse.text).toBe('');
|
||||
expect(childResponse.status).toBe(missingResponse.status);
|
||||
expect(childResponse.text).toBe(missingResponse.text);
|
||||
expect(getConvo).toHaveBeenNthCalledWith(1, 'test-user-123', 'child');
|
||||
});
|
||||
});
|
||||
|
||||
describe('DELETE /all', () => {
|
||||
it('prunes the deleted conversations’ agent checkpoints (bulk, ids from deleteConvos)', async () => {
|
||||
// HITL: a paused conversation's durable checkpoint must not outlive the conversation.
|
||||
|
|
|
|||
|
|
@ -169,4 +169,30 @@ describe('GET /api/messages/:conversationId with real validation middleware', ()
|
|||
expect(response.status).toBe(200);
|
||||
expect(response.body).toEqual([{ messageId: 'message-1', conversationId: 'convo-1' }]);
|
||||
});
|
||||
|
||||
it('does not return messages for a directly addressed child thread', async () => {
|
||||
getConvoOwnership.mockResolvedValue({
|
||||
conversationId: 'child-convo',
|
||||
user: authenticatedUserId,
|
||||
subagentThread: { parentConversationId: 'parent-convo' },
|
||||
});
|
||||
getMessages.mockResolvedValue([{ messageId: 'child-message', conversationId: 'child-convo' }]);
|
||||
|
||||
const response = await request(app).get('/api/messages/child-convo');
|
||||
|
||||
expect(response.status).toBe(404);
|
||||
expect(response.body).toEqual({ error: 'Conversation not found' });
|
||||
});
|
||||
|
||||
it('does not expose a directly addressed child thread through HEAD', async () => {
|
||||
getConvoOwnership.mockResolvedValue({
|
||||
conversationId: 'child-convo',
|
||||
user: authenticatedUserId,
|
||||
subagentThread: { parentConversationId: 'parent-convo' },
|
||||
});
|
||||
|
||||
const response = await request(app).head('/api/messages/child-convo');
|
||||
|
||||
expect(response.status).toBe(404);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ jest.mock('~/models', () => ({
|
|||
getMessage: jest.fn(),
|
||||
saveMessage: jest.fn(),
|
||||
getMessages: jest.fn(),
|
||||
getConvoOwnership: jest.fn(),
|
||||
updateMessage: jest.fn(),
|
||||
deleteMessages: jest.fn(),
|
||||
getConvosQueried: jest.fn(),
|
||||
|
|
@ -52,6 +53,7 @@ jest.mock('~/server/middleware/requireJwtAuth', () => (req, res, next) => next()
|
|||
|
||||
jest.mock('~/server/middleware', () => {
|
||||
const validateMessageReq = jest.fn((req, res, next) => next());
|
||||
const canReadActiveJobConversation = jest.fn().mockResolvedValue(false);
|
||||
const prepareMessageRequestValidation = jest.fn((req, res, next) => {
|
||||
req.messageRequestValidation = {
|
||||
conversationId: 'convo-1',
|
||||
|
|
@ -70,6 +72,7 @@ jest.mock('~/server/middleware', () => {
|
|||
return {
|
||||
requireJwtAuth: (req, res, next) => next(),
|
||||
validateMessageReq,
|
||||
canReadActiveJobConversation,
|
||||
sendValidationResponse,
|
||||
prepareMessageRequestValidation,
|
||||
configMiddleware: (req, res, next) => next(),
|
||||
|
|
@ -86,8 +89,17 @@ jest.mock('~/db/models', () => ({
|
|||
|
||||
describe('message route conversation ownership filters', () => {
|
||||
let app;
|
||||
const { getMessages, saveConvo, saveMessage } = require('~/models');
|
||||
const { prepareMessageRequestValidation } = require('~/server/middleware');
|
||||
const {
|
||||
getConvoOwnership,
|
||||
getMessages,
|
||||
getMessagesByCursor,
|
||||
saveConvo,
|
||||
saveMessage,
|
||||
} = require('~/models');
|
||||
const {
|
||||
canReadActiveJobConversation,
|
||||
prepareMessageRequestValidation,
|
||||
} = require('~/server/middleware');
|
||||
|
||||
const authenticatedUserId = 'user-owner-123';
|
||||
|
||||
|
|
@ -105,6 +117,7 @@ describe('message route conversation ownership filters', () => {
|
|||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
canReadActiveJobConversation.mockResolvedValue(false);
|
||||
prepareMessageRequestValidation.mockImplementation((req, res, next) => {
|
||||
req.messageRequestValidation = {
|
||||
conversationId: 'convo-1',
|
||||
|
|
@ -180,6 +193,111 @@ describe('message route conversation ownership filters', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('returns indistinguishable not-found responses for child and missing query reads', async () => {
|
||||
getConvoOwnership.mockResolvedValueOnce({
|
||||
user: authenticatedUserId,
|
||||
subagentThread: { parentConversationId: 'parent-convo' },
|
||||
});
|
||||
|
||||
const childResponse = await request(app).get(
|
||||
'/api/messages?conversationId=child-convo&messageId=child-message',
|
||||
);
|
||||
getConvoOwnership.mockResolvedValueOnce(null);
|
||||
const missingResponse = await request(app).get(
|
||||
'/api/messages?conversationId=missing-convo&messageId=missing-message',
|
||||
);
|
||||
|
||||
expect(childResponse.status).toBe(404);
|
||||
expect(childResponse.body).toEqual({ error: 'Conversation not found' });
|
||||
expect(childResponse.status).toBe(missingResponse.status);
|
||||
expect(childResponse.body).toEqual(missingResponse.body);
|
||||
expect(getMessages).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: 'single-message',
|
||||
path: '/api/messages?conversationId=convo-1&messageId=message-1',
|
||||
readMock: getMessages,
|
||||
readResult: [{ messageId: 'message-1', conversationId: 'convo-1' }],
|
||||
},
|
||||
{
|
||||
name: 'cursor',
|
||||
path: '/api/messages?conversationId=convo-1',
|
||||
readMock: getMessagesByCursor,
|
||||
readResult: { messages: [], nextCursor: null },
|
||||
},
|
||||
])(
|
||||
'starts the $name query read before ownership validation resolves',
|
||||
async ({ path, readMock, readResult }) => {
|
||||
const events = [];
|
||||
let resolveOwnership;
|
||||
const ownershipPromise = new Promise((resolve) => {
|
||||
resolveOwnership = resolve;
|
||||
});
|
||||
getConvoOwnership.mockImplementationOnce(() => {
|
||||
events.push('ownership-started');
|
||||
return ownershipPromise;
|
||||
});
|
||||
|
||||
let resolveReadStarted;
|
||||
const readStartedPromise = new Promise((resolve) => {
|
||||
resolveReadStarted = resolve;
|
||||
});
|
||||
readMock.mockImplementationOnce(() => {
|
||||
events.push('messages-started');
|
||||
resolveReadStarted();
|
||||
return Promise.resolve(readResult);
|
||||
});
|
||||
|
||||
const responsePromise = new Promise((resolve, reject) => {
|
||||
request(app)
|
||||
.get(path)
|
||||
.end((error, response) => (error ? reject(error) : resolve(response)));
|
||||
});
|
||||
|
||||
await Promise.race([readStartedPromise, new Promise((resolve) => setTimeout(resolve, 100))]);
|
||||
const eventsBeforeValidation = [...events];
|
||||
resolveOwnership({ user: authenticatedUserId });
|
||||
const response = await responsePromise;
|
||||
|
||||
expect(eventsBeforeValidation).toEqual(['ownership-started', 'messages-started']);
|
||||
expect(response.status).toBe(200);
|
||||
},
|
||||
);
|
||||
|
||||
it('allows an ordinary owned conversation query read', async () => {
|
||||
getConvoOwnership.mockResolvedValue({ user: authenticatedUserId });
|
||||
getMessages.mockResolvedValue([{ messageId: 'message-1', conversationId: 'convo-1' }]);
|
||||
|
||||
const response = await request(app).get(
|
||||
'/api/messages?conversationId=convo-1&messageId=message-1',
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.messages).toEqual([{ messageId: 'message-1', conversationId: 'convo-1' }]);
|
||||
});
|
||||
|
||||
it('allows an owner-scoped active generation before its conversation row exists', async () => {
|
||||
getConvoOwnership.mockResolvedValue(null);
|
||||
canReadActiveJobConversation.mockResolvedValue(true);
|
||||
getMessagesByCursor.mockResolvedValue({
|
||||
messages: [{ messageId: 'prompt-1', conversationId: 'active-convo' }],
|
||||
nextCursor: null,
|
||||
});
|
||||
|
||||
const response = await request(app).get('/api/messages?conversationId=active-convo');
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.messages).toEqual([
|
||||
{ messageId: 'prompt-1', conversationId: 'active-convo' },
|
||||
]);
|
||||
expect(canReadActiveJobConversation).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ user: { id: authenticatedUserId } }),
|
||||
'active-convo',
|
||||
);
|
||||
});
|
||||
|
||||
it('should start conversation message reads before validation resolves', async () => {
|
||||
const events = [];
|
||||
let resolveValidation;
|
||||
|
|
|
|||
|
|
@ -83,7 +83,7 @@ router.get('/:conversationId', async (req, res) => {
|
|||
const { conversationId } = req.params;
|
||||
const convo = await db.getConvo(req.user.id, conversationId);
|
||||
|
||||
if (convo) {
|
||||
if (convo && convo.subagentThread == null) {
|
||||
res.status(200).json(convo);
|
||||
} else {
|
||||
res.status(404).end();
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ const {
|
|||
validateMessageReq,
|
||||
configMiddleware,
|
||||
sendValidationResponse,
|
||||
canReadActiveJobConversation,
|
||||
prepareMessageRequestValidation,
|
||||
} = require('~/server/middleware');
|
||||
const db = require('~/models');
|
||||
|
|
@ -68,14 +69,45 @@ router.get('/', async (req, res) => {
|
|||
: 'createdAt';
|
||||
const sortOrder = sortDirection === 'asc' ? 1 : -1;
|
||||
|
||||
let scopedMessageRead;
|
||||
if (typeof conversationId === 'string') {
|
||||
const ownershipRead = db.getConvoOwnership(user, conversationId);
|
||||
const messageRead = messageId
|
||||
? db.getMessages({ conversationId, messageId, user })
|
||||
: db.getMessagesByCursor(
|
||||
{ conversationId, user },
|
||||
{ sortField, sortOrder, limit: pageSize, cursor },
|
||||
);
|
||||
scopedMessageRead = Promise.resolve(messageRead).then(
|
||||
(value) => ({ ok: true, value }),
|
||||
(error) => ({ ok: false, error }),
|
||||
);
|
||||
|
||||
const conversation = await ownershipRead;
|
||||
const canReadActiveJob =
|
||||
conversation == null &&
|
||||
!messageId &&
|
||||
(await canReadActiveJobConversation(req, conversationId));
|
||||
if ((!conversation && !canReadActiveJob) || conversation?.subagentThread != null) {
|
||||
return res.status(404).json({ error: 'Conversation not found' });
|
||||
}
|
||||
} else if (conversationId) {
|
||||
return res.status(404).json({ error: 'Conversation not found' });
|
||||
}
|
||||
|
||||
if (conversationId && messageId) {
|
||||
const messages = await db.getMessages({ conversationId, messageId, user });
|
||||
const messageResult = await scopedMessageRead;
|
||||
if (!messageResult.ok) {
|
||||
throw messageResult.error;
|
||||
}
|
||||
const messages = messageResult.value;
|
||||
response = { messages: messages?.length ? [messages[0]] : [], nextCursor: null };
|
||||
} else if (conversationId) {
|
||||
response = await db.getMessagesByCursor(
|
||||
{ conversationId, user },
|
||||
{ sortField, sortOrder, limit: pageSize, cursor },
|
||||
);
|
||||
const messageResult = await scopedMessageRead;
|
||||
if (!messageResult.ok) {
|
||||
throw messageResult.error;
|
||||
}
|
||||
response = messageResult.value;
|
||||
} else if (search) {
|
||||
const searchResults = await db.searchMessages(search, { filter: `user = "${user}"` }, true);
|
||||
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ export type MessageValidationRequest = {
|
|||
|
||||
type ConversationRecord = {
|
||||
user?: string;
|
||||
subagentThread?: unknown;
|
||||
} | null;
|
||||
|
||||
type PendingActionRecord = unknown;
|
||||
|
|
@ -71,6 +72,10 @@ export type MessageValidationDeps = {
|
|||
};
|
||||
|
||||
export type MessageRequestMiddleware = {
|
||||
canReadActiveJobConversation: (
|
||||
req: MessageValidationRequest,
|
||||
conversationId?: string,
|
||||
) => Promise<boolean>;
|
||||
createMessageRequestValidation: (req: MessageValidationRequest) => MessageRequestValidation;
|
||||
prepareMessageRequestValidation: (
|
||||
req: MessageValidationRequest,
|
||||
|
|
@ -90,6 +95,10 @@ function hasTenantMismatch(job: GenerationJobRecord, user: MessageValidationUser
|
|||
return job?.metadata?.tenantId != null && job.metadata.tenantId !== user.tenantId;
|
||||
}
|
||||
|
||||
function isPublicReadMethod(method?: string): boolean {
|
||||
return method === 'GET' || method === 'HEAD';
|
||||
}
|
||||
|
||||
export function createMessageRequestMiddleware(
|
||||
deps: MessageValidationDeps,
|
||||
): MessageRequestMiddleware {
|
||||
|
|
@ -97,7 +106,7 @@ export function createMessageRequestMiddleware(
|
|||
req: MessageValidationRequest,
|
||||
conversationId?: string,
|
||||
): Promise<boolean> {
|
||||
if (req.method !== 'GET' || req.params?.messageId) {
|
||||
if (!isPublicReadMethod(req.method) || req.params?.messageId) {
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -153,6 +162,13 @@ export function createMessageRequestMiddleware(
|
|||
};
|
||||
}
|
||||
|
||||
// Child threads are internal execution records, not standalone public
|
||||
// conversations. Keep the same response as a missing conversation so the
|
||||
// read boundary does not disclose whether a supplied child id exists.
|
||||
if (isPublicReadMethod(req.method) && conversation.subagentThread != null) {
|
||||
return { ok: false, status: 404, body: { error: 'Conversation not found' } };
|
||||
}
|
||||
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
|
|
@ -225,6 +241,7 @@ export function createMessageRequestMiddleware(
|
|||
}
|
||||
|
||||
return {
|
||||
canReadActiveJobConversation,
|
||||
createMessageRequestValidation: createMessageRequestValidation,
|
||||
prepareMessageRequestValidation: prepareMessageRequestValidation,
|
||||
sendValidationResponse: sendValidationResponse,
|
||||
|
|
|
|||
|
|
@ -1518,6 +1518,32 @@ describe('Conversation Operations', () => {
|
|||
).toBeNull();
|
||||
expect(await methods.getConvoOwnership('user123', 'non-existent-id')).toBeNull();
|
||||
});
|
||||
|
||||
it('includes child-thread identity without materializing conversation content', async () => {
|
||||
await Conversation.create({
|
||||
conversationId: 'child-conversation',
|
||||
user: 'user123',
|
||||
title: 'Internal child',
|
||||
endpoint: EModelEndpoint.agents,
|
||||
subagentThread: {
|
||||
rootConversationId: 'root-conversation',
|
||||
parentConversationId: 'parent-conversation',
|
||||
parentMessageId: 'parent-message',
|
||||
parentToolCallId: 'parent-tool-call',
|
||||
subagentType: 'child-agent',
|
||||
subagentKind: 'agent',
|
||||
depth: 1,
|
||||
},
|
||||
});
|
||||
|
||||
const result = await methods.getConvoOwnership('user123', 'child-conversation');
|
||||
|
||||
expect(result).toMatchObject({
|
||||
user: 'user123',
|
||||
subagentThread: { parentConversationId: 'parent-conversation' },
|
||||
});
|
||||
expect(result).not.toHaveProperty('title');
|
||||
});
|
||||
});
|
||||
|
||||
describe('getConvoRetention', () => {
|
||||
|
|
|
|||
|
|
@ -192,7 +192,7 @@ export interface ConversationMethods {
|
|||
getConvoOwnership(
|
||||
user: string,
|
||||
conversationId: string,
|
||||
): Promise<Pick<IConversation, 'user'> | null>;
|
||||
): Promise<Pick<IConversation, 'user' | 'subagentThread'> | null>;
|
||||
getConvoRetention(
|
||||
user: string,
|
||||
conversationId: string,
|
||||
|
|
@ -432,15 +432,15 @@ export function createConversationMethods(
|
|||
}
|
||||
|
||||
/**
|
||||
* Ownership probe for request validation: resolves only the owning user id
|
||||
* instead of materializing the full conversation document (preset spread +
|
||||
* Public-read probe: resolves ownership plus the child-thread discriminator
|
||||
* without materializing the full conversation document (preset spread +
|
||||
* message ObjectId array).
|
||||
*/
|
||||
async function getConvoOwnership(user: string, conversationId: string) {
|
||||
try {
|
||||
const Conversation = mongoose.models.Conversation as Model<IConversation>;
|
||||
return await Conversation.findOne({ user, conversationId }, 'user').lean<
|
||||
Pick<IConversation, 'user'>
|
||||
return await Conversation.findOne({ user, conversationId }, 'user subagentThread').lean<
|
||||
Pick<IConversation, 'user' | 'subagentThread'>
|
||||
>();
|
||||
} catch (error) {
|
||||
logger.error('[getConvoOwnership] Error checking conversation ownership', error);
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@ interface DynamicMeiliDocument extends mongoose.Document {
|
|||
isTemporary?: boolean;
|
||||
expiredAt?: Date | null;
|
||||
_meiliIndex?: boolean;
|
||||
_meiliIndexAttempted?: boolean;
|
||||
_meiliCleanupVersion?: number;
|
||||
}
|
||||
|
||||
type DynamicMeiliModel = mongoose.Model<DynamicMeiliDocument> & SchemaWithMeiliMethods;
|
||||
|
|
@ -105,7 +107,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
mockAddDocuments.mockClear();
|
||||
mockAddDocumentsInBatches.mockClear();
|
||||
mockUpdateDocuments.mockClear();
|
||||
mockDeleteDocument.mockClear();
|
||||
mockDeleteDocument.mockReset().mockResolvedValue({ taskUid: 2 });
|
||||
mockDeleteDocuments.mockReset().mockResolvedValue({ taskUid: 1 });
|
||||
mockGetDocument.mockClear();
|
||||
mockGetDocuments.mockReset().mockResolvedValue({ results: [] });
|
||||
|
|
@ -574,6 +576,54 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
expect(storedDoc?._meiliIndex).toBe(true);
|
||||
});
|
||||
|
||||
test('does not clear an indexed marker until an update-hook deletion succeeds', async () => {
|
||||
const conversationModel = createConversationModel(
|
||||
mongoose,
|
||||
) as unknown as SchemaWithMeiliMethods;
|
||||
await conversationModel.deleteMany({});
|
||||
const conversationId = new mongoose.Types.ObjectId().toString();
|
||||
|
||||
await conversationModel.create({
|
||||
conversationId,
|
||||
user: new mongoose.Types.ObjectId().toString(),
|
||||
title: 'Initially searchable conversation',
|
||||
endpoint: EModelEndpoint.agents,
|
||||
});
|
||||
const conversation = await conversationModel
|
||||
.findOne({ conversationId })
|
||||
.select('+_meiliIndex +_meiliIndexAttempted');
|
||||
expect(conversation?._meiliIndex).toBe(true);
|
||||
expect(conversation?._meiliIndexAttempted).toBe(true);
|
||||
|
||||
conversation!.subagentThread = {
|
||||
rootConversationId: 'root-conversation',
|
||||
parentConversationId: 'parent-conversation',
|
||||
parentMessageId: 'parent-message',
|
||||
parentToolCallId: 'parent-tool-call',
|
||||
subagentType: 'agent-child',
|
||||
subagentKind: 'agent',
|
||||
depth: 1,
|
||||
};
|
||||
mockWaitForTask.mockResolvedValueOnce({ status: 'failed' });
|
||||
await conversation!.save();
|
||||
expect((await conversationModel.collection.findOne({ conversationId }))?._meiliIndex).toBe(
|
||||
true,
|
||||
);
|
||||
|
||||
conversation!.title = 'Retry deletion';
|
||||
mockWaitForTask.mockResolvedValueOnce({ status: 'succeeded' });
|
||||
await conversation!.save();
|
||||
const storedDoc = await conversationModel.collection.findOne({ conversationId });
|
||||
|
||||
expect(mockDeleteDocument).toHaveBeenCalledTimes(2);
|
||||
expect(mockWaitForTask).toHaveBeenCalledWith(2, {
|
||||
timeOutMs: 10000,
|
||||
intervalMs: 100,
|
||||
});
|
||||
expect(storedDoc?._meiliIndex).toBeUndefined();
|
||||
expect(storedDoc?._meiliIndexAttempted).toBeUndefined();
|
||||
});
|
||||
|
||||
test('retries cleanup when Meili deletion succeeds before the Mongo flag update fails', async () => {
|
||||
const conversationModel = createConversationModel(
|
||||
mongoose,
|
||||
|
|
@ -705,7 +755,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
expect(storedDoc?._meiliIndex).toBe(false);
|
||||
});
|
||||
|
||||
test('sync removes an existing child message even when its index flag is false', async () => {
|
||||
test('does not schedule cleanup for a child message that was never indexed', async () => {
|
||||
const messageModel = createMessageModel(mongoose) as unknown as SchemaWithMeiliMethods;
|
||||
await messageModel.deleteMany({});
|
||||
const messageId = new mongoose.Types.ObjectId().toString();
|
||||
|
|
@ -721,18 +771,81 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
status: 'completed',
|
||||
},
|
||||
_meiliIndex: false,
|
||||
_meiliCleanupVersion: 1,
|
||||
createdAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
const progress = await messageModel.getSyncProgress();
|
||||
await messageModel.cleanupExcludedMeiliIndex();
|
||||
const storedDoc = await messageModel.collection.findOne({ messageId });
|
||||
|
||||
expect(progress).toMatchObject({ pendingCleanup: 0, isComplete: true });
|
||||
expect(mockDeleteDocuments).not.toHaveBeenCalled();
|
||||
expect(mockGetDocuments).not.toHaveBeenCalled();
|
||||
expect(storedDoc?._meiliIndex).toBe(false);
|
||||
});
|
||||
|
||||
test('reconciles a legacy excluded false marker exactly once', async () => {
|
||||
const messageModel = createMessageModel(mongoose) as unknown as SchemaWithMeiliMethods;
|
||||
await messageModel.deleteMany({});
|
||||
const messageId = new mongoose.Types.ObjectId().toString();
|
||||
|
||||
await messageModel.collection.insertOne({
|
||||
messageId,
|
||||
conversationId: new mongoose.Types.ObjectId().toString(),
|
||||
user: new mongoose.Types.ObjectId().toString(),
|
||||
isCreatedByUser: true,
|
||||
text: 'Legacy child transcript with an ambiguous deletion result',
|
||||
subagentTask: {
|
||||
attemptKey: 'legacy-attempt-key',
|
||||
status: 'completed',
|
||||
},
|
||||
_meiliIndex: false,
|
||||
createdAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
|
||||
const progressBefore = await messageModel.getSyncProgress();
|
||||
await messageModel.cleanupExcludedMeiliIndex();
|
||||
const progressAfter = await messageModel.getSyncProgress();
|
||||
const storedDoc = await messageModel.collection.findOne({ messageId });
|
||||
|
||||
expect(progressBefore).toMatchObject({ pendingCleanup: 1, isComplete: false });
|
||||
expect(progressAfter).toMatchObject({ pendingCleanup: 0, isComplete: true });
|
||||
expect(mockDeleteDocuments).toHaveBeenCalledWith([messageId]);
|
||||
expect(storedDoc?._meiliIndex).toBeUndefined();
|
||||
expect(storedDoc?._meiliCleanupVersion).toBe(1);
|
||||
});
|
||||
|
||||
test('cleans an excluded child when an earlier Meili add was attempted but not acknowledged', async () => {
|
||||
const messageModel = createMessageModel(mongoose) as unknown as SchemaWithMeiliMethods;
|
||||
await messageModel.deleteMany({});
|
||||
const messageId = new mongoose.Types.ObjectId().toString();
|
||||
|
||||
await messageModel.collection.insertOne({
|
||||
messageId,
|
||||
conversationId: new mongoose.Types.ObjectId().toString(),
|
||||
user: new mongoose.Types.ObjectId().toString(),
|
||||
isCreatedByUser: true,
|
||||
text: 'Possibly indexed child transcript',
|
||||
subagentTask: {
|
||||
attemptKey: 'attempt-key',
|
||||
status: 'completed',
|
||||
},
|
||||
_meiliIndex: false,
|
||||
_meiliIndexAttempted: true,
|
||||
createdAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
});
|
||||
mockGetDocuments.mockResolvedValueOnce({ results: [{ messageId }] });
|
||||
|
||||
const progress = await messageModel.getSyncProgress();
|
||||
await messageModel.syncWithMeili();
|
||||
await messageModel.cleanupExcludedMeiliIndex();
|
||||
const storedDoc = await messageModel.collection.findOne({ messageId });
|
||||
|
||||
expect(progress).toMatchObject({ pendingCleanup: 1, isComplete: false });
|
||||
expect(mockDeleteDocuments).toHaveBeenCalledWith([messageId]);
|
||||
expect(storedDoc?._meiliIndex).toBeUndefined();
|
||||
expect(storedDoc?._meiliIndexAttempted).toBeUndefined();
|
||||
});
|
||||
|
||||
test('defines partial indexes for pending excluded-document cleanup', () => {
|
||||
|
|
@ -742,20 +855,62 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
expect(conversationIndexes).toContainEqual([
|
||||
{ _meiliIndex: 1, conversationId: 1 },
|
||||
expect.objectContaining({
|
||||
name: 'meili_excluded_cleanup',
|
||||
name: 'meili_excluded_indexed_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
subagentThread: { $exists: true },
|
||||
_meiliIndex: { $exists: true },
|
||||
_meiliIndex: { $eq: true },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(conversationIndexes).toContainEqual([
|
||||
{ _meiliIndexAttempted: 1, conversationId: 1 },
|
||||
expect.objectContaining({
|
||||
name: 'meili_excluded_attempted_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
subagentThread: { $exists: true },
|
||||
_meiliIndexAttempted: { $eq: true },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(conversationIndexes).toContainEqual([
|
||||
{ _meiliIndex: 1, _meiliCleanupVersion: 1, conversationId: 1 },
|
||||
expect.objectContaining({
|
||||
name: 'meili_excluded_legacy_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
subagentThread: { $exists: true },
|
||||
_meiliIndex: { $eq: false },
|
||||
_meiliCleanupVersion: { $exists: false },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(messageIndexes).toContainEqual([
|
||||
{ _meiliIndex: 1, messageId: 1 },
|
||||
expect.objectContaining({
|
||||
name: 'meili_excluded_cleanup',
|
||||
name: 'meili_excluded_indexed_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
subagentTask: { $exists: true },
|
||||
_meiliIndex: { $exists: true },
|
||||
_meiliIndex: { $eq: true },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(messageIndexes).toContainEqual([
|
||||
{ _meiliIndexAttempted: 1, messageId: 1 },
|
||||
expect.objectContaining({
|
||||
name: 'meili_excluded_attempted_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
subagentTask: { $exists: true },
|
||||
_meiliIndexAttempted: { $eq: true },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(messageIndexes).toContainEqual([
|
||||
{ _meiliIndex: 1, _meiliCleanupVersion: 1, messageId: 1 },
|
||||
expect.objectContaining({
|
||||
name: 'meili_excluded_legacy_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
subagentTask: { $exists: true },
|
||||
_meiliIndex: { $eq: false },
|
||||
_meiliCleanupVersion: { $exists: false },
|
||||
},
|
||||
}),
|
||||
]);
|
||||
|
|
@ -1196,6 +1351,13 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
// Spy on updateMany and make it fail
|
||||
const updateManySpy = jest
|
||||
.spyOn(conversationModel, 'updateMany')
|
||||
.mockResolvedValueOnce({
|
||||
acknowledged: true,
|
||||
matchedCount: 1,
|
||||
modifiedCount: 1,
|
||||
upsertedCount: 0,
|
||||
upsertedId: null,
|
||||
})
|
||||
.mockRejectedValueOnce(new Error('Database connection error'));
|
||||
|
||||
// Sync should throw the error
|
||||
|
|
|
|||
|
|
@ -30,6 +30,8 @@ interface MongoMeiliOptions {
|
|||
interface MeiliIndexable {
|
||||
[key: string]: unknown;
|
||||
_meiliIndex?: boolean;
|
||||
_meiliIndexAttempted?: boolean;
|
||||
_meiliCleanupVersion?: number;
|
||||
}
|
||||
|
||||
interface SyncProgress {
|
||||
|
|
@ -42,6 +44,8 @@ interface SyncProgress {
|
|||
|
||||
interface _DocumentWithMeiliIndex extends Document {
|
||||
_meiliIndex?: boolean;
|
||||
_meiliIndexAttempted?: boolean;
|
||||
_meiliCleanupVersion?: number;
|
||||
isTemporary?: boolean;
|
||||
expiredAt?: Date | null;
|
||||
preprocessObjectForIndex?: () => Record<string, unknown>;
|
||||
|
|
@ -62,6 +66,7 @@ export interface SchemaWithMeiliMethods extends Model<DocumentWithMeiliIndex> {
|
|||
index: Index<MeiliIndexable>,
|
||||
documents: Array<Record<string, unknown>>,
|
||||
): Promise<void>;
|
||||
cleanupExcludedMeiliIndex(): Promise<void>;
|
||||
cleanupMeiliIndex(
|
||||
index: Index<MeiliIndexable>,
|
||||
primaryKey: string,
|
||||
|
|
@ -100,6 +105,7 @@ const hasSchemaPath = (schema: Schema, path: string): boolean =>
|
|||
Object.prototype.hasOwnProperty.call(schema.obj, path);
|
||||
|
||||
const explicitTemporaryFlagKey = 'meiliExplicitTemporaryFlag';
|
||||
const meiliCleanupVersion = 1;
|
||||
|
||||
const buildRetentionIndexableQuery = (schema: Schema): FilterQuery<unknown> => {
|
||||
if (hasSchemaPath(schema, 'isTemporary')) {
|
||||
|
|
@ -133,7 +139,11 @@ const buildExcludedIndexedQuery = (excludeFromIndexPath?: string): FilterQuery<u
|
|||
|
||||
return {
|
||||
[excludeFromIndexPath]: { $exists: true },
|
||||
_meiliIndex: { $exists: true },
|
||||
$or: [
|
||||
{ _meiliIndex: true },
|
||||
{ _meiliIndexAttempted: true },
|
||||
{ _meiliIndex: false, _meiliCleanupVersion: { $exists: false } },
|
||||
],
|
||||
};
|
||||
};
|
||||
|
||||
|
|
@ -356,16 +366,22 @@ const createMeiliMongooseModel = ({
|
|||
);
|
||||
|
||||
try {
|
||||
const docsIds = documents.map((doc) => doc._id);
|
||||
await this.updateMany(
|
||||
{ _id: { $in: docsIds } },
|
||||
{ $set: { _meiliIndexAttempted: true } },
|
||||
{ timestamps: false },
|
||||
);
|
||||
|
||||
// Add documents to MeiliSearch
|
||||
await index.addDocumentsInBatches(formattedDocs, undefined, { primaryKey });
|
||||
|
||||
// Update MongoDB to mark documents as indexed.
|
||||
// { timestamps: false } prevents Mongoose from touching updatedAt, preserving
|
||||
// original conversation/message timestamps (fixes sidebar chronological sort).
|
||||
const docsIds = documents.map((doc) => doc._id);
|
||||
await this.updateMany(
|
||||
{ _id: { $in: docsIds } },
|
||||
{ $set: { _meiliIndex: true } },
|
||||
{ $set: { _meiliIndex: true, _meiliCleanupVersion: meiliCleanupVersion } },
|
||||
{ timestamps: false },
|
||||
);
|
||||
} catch (error) {
|
||||
|
|
@ -374,6 +390,58 @@ const createMeiliMongooseModel = ({
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove documents that are intentionally excluded from search without
|
||||
* scanning the complete Meili index. The Mongo marker is cleared only
|
||||
* after Meili confirms deletion, so interrupted cleanup is retried.
|
||||
*/
|
||||
static async cleanupExcludedMeiliIndex(this: SchemaWithMeiliMethods): Promise<void> {
|
||||
const excludedIndexedQuery = getExcludedIndexedQuery();
|
||||
if (excludedIndexedQuery == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
const { batchSize, delayMs } = syncConfig;
|
||||
while (true) {
|
||||
const pendingExcludedDocuments = await this.find(excludedIndexedQuery)
|
||||
.select(primaryKey)
|
||||
.limit(batchSize)
|
||||
.lean();
|
||||
if (pendingExcludedDocuments.length === 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
const pendingIds = pendingExcludedDocuments.map(
|
||||
(doc: Record<string, unknown>) => doc[primaryKey],
|
||||
);
|
||||
const deletion = await index.deleteDocuments(pendingIds.map(String));
|
||||
const deletionTask = await client.waitForTask(deletion.taskUid, {
|
||||
timeOutMs: 10000,
|
||||
intervalMs: 100,
|
||||
});
|
||||
if (deletionTask.status !== 'succeeded') {
|
||||
throw new Error(
|
||||
`Meili cleanup task ${deletion.taskUid} ended with ${deletionTask.status}`,
|
||||
);
|
||||
}
|
||||
await this.updateMany(
|
||||
{ ...excludedIndexedQuery, [primaryKey]: { $in: pendingIds } },
|
||||
{
|
||||
$set: { _meiliCleanupVersion: meiliCleanupVersion },
|
||||
$unset: { _meiliIndex: '', _meiliIndexAttempted: '' },
|
||||
},
|
||||
{ timestamps: false },
|
||||
);
|
||||
|
||||
if (pendingExcludedDocuments.length < batchSize) {
|
||||
break;
|
||||
}
|
||||
if (delayMs > 0) {
|
||||
await new Promise((resolve) => setTimeout(resolve, delayMs));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up documents in MeiliSearch that no longer exist in MongoDB
|
||||
*/
|
||||
|
|
@ -385,46 +453,7 @@ const createMeiliMongooseModel = ({
|
|||
delayMs: number,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const excludedIndexedQuery = getExcludedIndexedQuery();
|
||||
if (excludedIndexedQuery != null) {
|
||||
let hasPendingExcludedDocuments = true;
|
||||
while (hasPendingExcludedDocuments) {
|
||||
const pendingExcludedDocuments = await this.find(excludedIndexedQuery)
|
||||
.select(primaryKey)
|
||||
.limit(batchSize)
|
||||
.lean();
|
||||
if (pendingExcludedDocuments.length === 0) {
|
||||
hasPendingExcludedDocuments = false;
|
||||
break;
|
||||
}
|
||||
|
||||
const pendingIds = pendingExcludedDocuments.map(
|
||||
(doc: Record<string, unknown>) => doc[primaryKey],
|
||||
);
|
||||
const deletion = await index.deleteDocuments(pendingIds.map(String));
|
||||
const deletionTask = await client.waitForTask(deletion.taskUid, {
|
||||
timeOutMs: 10000,
|
||||
intervalMs: 100,
|
||||
});
|
||||
if (deletionTask.status !== 'succeeded') {
|
||||
throw new Error(
|
||||
`Meili cleanup task ${deletion.taskUid} ended with ${deletionTask.status}`,
|
||||
);
|
||||
}
|
||||
await this.updateMany(
|
||||
{ ...excludedIndexedQuery, [primaryKey]: { $in: pendingIds } },
|
||||
{ $unset: { _meiliIndex: '' } },
|
||||
{ timestamps: false },
|
||||
);
|
||||
|
||||
if (pendingExcludedDocuments.length < batchSize) {
|
||||
break;
|
||||
}
|
||||
if (delayMs > 0) {
|
||||
await new Promise((resolve) => setTimeout(resolve, delayMs));
|
||||
}
|
||||
}
|
||||
}
|
||||
await this.cleanupExcludedMeiliIndex();
|
||||
|
||||
let offset = 0;
|
||||
let moreDocuments = true;
|
||||
|
|
@ -463,7 +492,10 @@ const createMeiliMongooseModel = ({
|
|||
}
|
||||
await this.updateMany(
|
||||
{ [primaryKey]: { $in: toDelete } },
|
||||
{ $unset: { _meiliIndex: '' } },
|
||||
{
|
||||
$set: { _meiliCleanupVersion: meiliCleanupVersion },
|
||||
$unset: { _meiliIndex: '', _meiliIndexAttempted: '' },
|
||||
},
|
||||
{ timestamps: false },
|
||||
);
|
||||
logger.debug(`[cleanupMeiliIndex] Deleted ${toDelete.length} orphaned documents`);
|
||||
|
|
@ -578,6 +610,21 @@ const createMeiliMongooseModel = ({
|
|||
const maxRetries = 3;
|
||||
let retryCount = 0;
|
||||
|
||||
try {
|
||||
// Mark the possible Meili presence before enqueueing the add. If the
|
||||
// later Mongo acknowledgement fails, excluded-record cleanup can still
|
||||
// distinguish this row from a child that was never submitted to Meili.
|
||||
const model = this.constructor as Model<DocumentWithMeiliIndex>;
|
||||
await model.updateOne(
|
||||
{ _id: this._id as Types.ObjectId },
|
||||
{ $set: { _meiliIndexAttempted: true } },
|
||||
{ timestamps: false },
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('[addObjectToMeili] Error marking Meili indexing attempt:', error);
|
||||
return next();
|
||||
}
|
||||
|
||||
while (retryCount < maxRetries) {
|
||||
try {
|
||||
await index.addDocuments([object], { primaryKey });
|
||||
|
|
@ -597,7 +644,7 @@ const createMeiliMongooseModel = ({
|
|||
// eslint-disable-next-line no-restricted-syntax -- _meiliIndex is an internal bookkeeping flag, not tenant-scoped data
|
||||
await this.collection.updateOne(
|
||||
{ _id: this._id as Types.ObjectId },
|
||||
{ $set: { _meiliIndex: true } },
|
||||
{ $set: { _meiliIndex: true, _meiliCleanupVersion: meiliCleanupVersion } },
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('[addObjectToMeili] Error updating _meiliIndex field:', error);
|
||||
|
|
@ -616,11 +663,25 @@ const createMeiliMongooseModel = ({
|
|||
): Promise<void> {
|
||||
try {
|
||||
if (!isIndexableDocument(this, excludeFromIndexPath)) {
|
||||
await index.deleteDocument(String(this[primaryKey as keyof DocumentWithMeiliIndex]));
|
||||
const deletion = await index.deleteDocument(
|
||||
String(this[primaryKey as keyof DocumentWithMeiliIndex]),
|
||||
);
|
||||
const deletionTask = await client.waitForTask(deletion.taskUid, {
|
||||
timeOutMs: 10000,
|
||||
intervalMs: 100,
|
||||
});
|
||||
if (deletionTask.status !== 'succeeded') {
|
||||
throw new Error(
|
||||
`Meili cleanup task ${deletion.taskUid} ended with ${deletionTask.status}`,
|
||||
);
|
||||
}
|
||||
const model = this.constructor as Model<DocumentWithMeiliIndex>;
|
||||
await model.updateOne(
|
||||
{ _id: this._id as Types.ObjectId },
|
||||
{ $set: { _meiliIndex: false } },
|
||||
{
|
||||
$set: { _meiliCleanupVersion: meiliCleanupVersion },
|
||||
$unset: { _meiliIndex: '', _meiliIndexAttempted: '' },
|
||||
},
|
||||
);
|
||||
return next();
|
||||
}
|
||||
|
|
@ -736,16 +797,48 @@ export default function mongoMeili(schema: Schema, options: MongoMeiliOptions):
|
|||
select: false,
|
||||
default: false,
|
||||
},
|
||||
_meiliIndexAttempted: {
|
||||
type: Boolean,
|
||||
required: false,
|
||||
select: false,
|
||||
},
|
||||
_meiliCleanupVersion: {
|
||||
type: Number,
|
||||
required: false,
|
||||
select: false,
|
||||
default: meiliCleanupVersion,
|
||||
},
|
||||
});
|
||||
|
||||
if (options.excludeFromIndexPath != null) {
|
||||
schema.index(
|
||||
{ _meiliIndex: 1, [options.primaryKey]: 1 },
|
||||
{
|
||||
name: 'meili_excluded_cleanup',
|
||||
name: 'meili_excluded_indexed_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
[options.excludeFromIndexPath]: { $exists: true },
|
||||
_meiliIndex: { $exists: true },
|
||||
_meiliIndex: { $eq: true },
|
||||
},
|
||||
},
|
||||
);
|
||||
schema.index(
|
||||
{ _meiliIndexAttempted: 1, [options.primaryKey]: 1 },
|
||||
{
|
||||
name: 'meili_excluded_attempted_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
[options.excludeFromIndexPath]: { $exists: true },
|
||||
_meiliIndexAttempted: { $eq: true },
|
||||
},
|
||||
},
|
||||
);
|
||||
schema.index(
|
||||
{ _meiliIndex: 1, _meiliCleanupVersion: 1, [options.primaryKey]: 1 },
|
||||
{
|
||||
name: 'meili_excluded_legacy_cleanup_v3',
|
||||
partialFilterExpression: {
|
||||
[options.excludeFromIndexPath]: { $exists: true },
|
||||
_meiliIndex: { $eq: false },
|
||||
_meiliCleanupVersion: { $exists: false },
|
||||
},
|
||||
},
|
||||
);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue