diff --git a/api/db/indexSync.js b/api/db/indexSync.js index 13059033fb5..66781ad4111 100644 --- a/api/db/indexSync.js +++ b/api/db/indexSync.js @@ -33,12 +33,16 @@ class MeiliSearchClient { } /** - * Deletes documents from MeiliSearch index that are missing the user field + * Deletes documents from MeiliSearch index that are missing the user field. + * Without the user field, documents are excluded by the per-user filter and + * become unsearchable; removing them lets the next sync re-index them with the + * user field populated. * @param {import('meilisearch').Index} index - MeiliSearch index instance * @param {string} indexName - Name of the index for logging + * @param {string} primaryKey - Primary key field on the documents (e.g. messageId, conversationId) * @returns {Promise} - Number of documents deleted */ -async function deleteDocumentsWithoutUserField(index, indexName) { +async function deleteDocumentsWithoutUserField(index, indexName, primaryKey) { let deletedCount = 0; let offset = 0; const batchSize = 1000; @@ -47,14 +51,17 @@ async function deleteDocumentsWithoutUserField(index, indexName) { while (true) { const searchResult = await index.search('', { limit: batchSize, - offset: offset, + offset, }); if (searchResult.hits.length === 0) { break; } - const idsToDelete = searchResult.hits.filter((hit) => !hit.user).map((hit) => hit.id); + const idsToDelete = searchResult.hits + .filter((hit) => !hit.user) + .map((hit) => hit[primaryKey]) + .filter((id) => id != null); if (idsToDelete.length > 0) { logger.info( @@ -68,7 +75,8 @@ async function deleteDocumentsWithoutUserField(index, indexName) { break; } - offset += batchSize; + // Deleted documents shrink the index; advance offset only by the kept hits. + offset += searchResult.hits.length - idsToDelete.length; } if (deletedCount > 0) { @@ -81,6 +89,19 @@ async function deleteDocumentsWithoutUserField(index, indexName) { return deletedCount; } +/** + * Returns true if any indexed document is missing the `user` field. Scans up to + * `sampleSize` hits because checking only the first hit can miss orphans when + * newer documents (with the user field) sort first. + * @param {import('meilisearch').Index} index + * @param {number} sampleSize + * @returns {Promise} + */ +async function indexHasOrphanedDocs(index, sampleSize = 200) { + const searchResult = await index.search('', { limit: sampleSize }); + return searchResult.hits.some((hit) => !hit.user); +} + /** * Ensures indexes have proper filterable attributes configured and checks if documents have user field * @param {MeiliSearch} client - MeiliSearch client instance @@ -107,8 +128,7 @@ async function ensureFilterableAttributes(client) { // Check if existing documents have user field indexed try { - const searchResult = await messagesIndex.search('', { limit: 1 }); - if (searchResult.hits.length > 0 && !searchResult.hits[0].user) { + if (await indexHasOrphanedDocs(messagesIndex)) { logger.info( '[indexSync] Existing messages missing user field, will clean up orphaned documents...', ); @@ -139,8 +159,7 @@ async function ensureFilterableAttributes(client) { // Check if existing documents have user field indexed try { - const searchResult = await convosIndex.search('', { limit: 1 }); - if (searchResult.hits.length > 0 && !searchResult.hits[0].user) { + if (await indexHasOrphanedDocs(convosIndex)) { logger.info( '[indexSync] Existing conversations missing user field, will clean up orphaned documents...', ); @@ -155,23 +174,43 @@ async function ensureFilterableAttributes(client) { } } - // If either index has orphaned documents, clean them up (but don't force resync) + // If either index has orphaned documents, remove them. The caller will then + // reset MongoDB _meiliIndex flags so syncWithMeili re-adds them with the + // user field populated; otherwise they'd stay deleted in Meili because + // sync only touches docs marked _meiliIndex !== true. if (hasOrphanedDocs) { + let deletedTotal = 0; try { const messagesIndex = client.index('messages'); - await deleteDocumentsWithoutUserField(messagesIndex, 'messages'); + deletedTotal += await deleteDocumentsWithoutUserField( + messagesIndex, + 'messages', + 'messageId', + ); } catch (error) { logger.debug('[indexSync] Could not clean up messages:', error.message); } try { const convosIndex = client.index('convos'); - await deleteDocumentsWithoutUserField(convosIndex, 'convos'); + deletedTotal += await deleteDocumentsWithoutUserField( + convosIndex, + 'convos', + 'conversationId', + ); } catch (error) { logger.debug('[indexSync] Could not clean up convos:', error.message); } - logger.info('[indexSync] Orphaned documents cleaned up without forcing resync.'); + if (deletedTotal === 0) { + // Detection found orphans but cleanup deleted nothing — treat as no + // orphans so we don't unnecessarily wipe _meiliIndex flags. + hasOrphanedDocs = false; + } else { + logger.info( + `[indexSync] Cleaned up ${deletedTotal} orphaned documents. Forcing re-sync to restore them with the user field.`, + ); + } } if (settingsUpdated) { @@ -213,19 +252,22 @@ async function performSync(flowManager, flowId, flowType) { } /** Ensures indexes have proper filterable attributes configured */ - const { settingsUpdated, orphanedDocsFound: _orphanedDocsFound } = - await ensureFilterableAttributes(client); + const { settingsUpdated, orphanedDocsFound } = await ensureFilterableAttributes(client); let messagesSync = false; let convosSync = false; - // Only reset flags if settings were actually updated (not just for orphaned doc cleanup) - if (settingsUpdated) { + // Reset flags when settings were updated, or when orphaned docs were cleaned + // up — in that case the MongoDB docs still have _meiliIndex: true, so without + // resetting, syncWithMeili would skip them and they'd stay missing from Meili. + const forceResync = settingsUpdated || orphanedDocsFound; + if (forceResync) { logger.info( - '[indexSync] Settings updated. Forcing full re-sync to reindex with new configuration...', + settingsUpdated + ? '[indexSync] Settings updated. Forcing full re-sync to reindex with new configuration...' + : '[indexSync] Orphaned documents removed. Forcing re-sync to restore them with the user field...', ); - // Reset sync flags to force full re-sync await batchResetMeiliFlags(Message.collection); await batchResetMeiliFlags(Conversation.collection); } @@ -233,7 +275,7 @@ async function performSync(flowManager, flowId, flowType) { // Check if we need to sync messages logger.info('[indexSync] Requesting message sync progress...'); const messageProgress = await Message.getSyncProgress(); - if (!messageProgress.isComplete || settingsUpdated) { + if (!messageProgress.isComplete || forceResync) { logger.info( `[indexSync] Messages need syncing: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments} indexed`, ); @@ -243,8 +285,8 @@ async function performSync(flowManager, flowId, flowType) { const unindexedMessages = messageCount - messagesIndexed; const noneIndexed = messagesIndexed === 0 && unindexedMessages > 0; - if (settingsUpdated || noneIndexed || unindexedMessages > syncThreshold) { - if (noneIndexed && !settingsUpdated) { + if (forceResync || noneIndexed || unindexedMessages > syncThreshold) { + if (noneIndexed && !forceResync) { logger.info('[indexSync] No messages marked as indexed, forcing full sync'); } logger.info(`[indexSync] Starting message sync (${unindexedMessages} unindexed)`); @@ -263,7 +305,7 @@ async function performSync(flowManager, flowId, flowType) { // Check if we need to sync conversations const convoProgress = await Conversation.getSyncProgress(); - if (!convoProgress.isComplete || settingsUpdated) { + if (!convoProgress.isComplete || forceResync) { logger.info( `[indexSync] Conversations need syncing: ${convoProgress.totalProcessed}/${convoProgress.totalDocuments} indexed`, ); @@ -273,8 +315,8 @@ async function performSync(flowManager, flowId, flowType) { const unindexedConvos = convoCount - convosIndexed; const noneConvosIndexed = convosIndexed === 0 && unindexedConvos > 0; - if (settingsUpdated || noneConvosIndexed || unindexedConvos > syncThreshold) { - if (noneConvosIndexed && !settingsUpdated) { + if (forceResync || noneConvosIndexed || unindexedConvos > syncThreshold) { + if (noneConvosIndexed && !forceResync) { logger.info('[indexSync] No conversations marked as indexed, forcing full sync'); } logger.info(`[indexSync] Starting convos sync (${unindexedConvos} unindexed)`); diff --git a/api/db/indexSync.spec.js b/api/db/indexSync.spec.js index dbe07c75951..37352555043 100644 --- a/api/db/indexSync.spec.js +++ b/api/db/indexSync.spec.js @@ -527,4 +527,100 @@ describe('performSync() - syncThreshold logic', () => { '[indexSync] 6 convos unindexed (below threshold: 1000, skipping)', ); }); + + test('deletes orphaned docs by their primary key (not by hit.id) and forces re-sync', async () => { + Message.getSyncProgress.mockResolvedValue({ + totalProcessed: 100, + totalDocuments: 100, + isComplete: true, + }); + Conversation.getSyncProgress.mockResolvedValue({ + totalProcessed: 50, + totalDocuments: 50, + isComplete: true, + }); + Message.syncWithMeili.mockResolvedValue(undefined); + Conversation.syncWithMeili.mockResolvedValue(undefined); + + const messagesDeleteDocuments = jest.fn().mockResolvedValue({}); + const convosDeleteDocuments = jest.fn().mockResolvedValue({}); + + // search() is called once during orphan detection and again during cleanup + // (which iterates with offset until a partial/empty page). Returning the + // same page for both calls is fine because hits.length < batchSize ends the loop. + const messagesHits = [ + { messageId: 'm-1', text: 'Barcelona' }, // orphan: no user field + { messageId: 'm-2', user: 'user-A', text: 'Madrid' }, + { messageId: 'm-3', text: 'Barcelona again' }, // orphan + ]; + const messagesSearch = jest.fn().mockResolvedValue({ hits: messagesHits }); + + const convosHits = [{ conversationId: 'c-1', title: 'Trip to Barcelona' }]; + const convosSearch = jest.fn().mockResolvedValue({ hits: convosHits }); + + mockMeiliIndex.mockImplementation((name) => { + if (name === 'messages') { + return { + getSettings: jest.fn().mockResolvedValue({ filterableAttributes: ['user'] }), + updateSettings: jest.fn().mockResolvedValue({}), + search: messagesSearch, + deleteDocuments: messagesDeleteDocuments, + }; + } + return { + getSettings: jest.fn().mockResolvedValue({ filterableAttributes: ['user'] }), + updateSettings: jest.fn().mockResolvedValue({}), + search: convosSearch, + deleteDocuments: convosDeleteDocuments, + }; + }); + + const indexSync = require('./indexSync'); + await indexSync(); + + // The fix: deletion uses primaryKey (messageId/conversationId), not hit.id + expect(messagesDeleteDocuments).toHaveBeenCalledWith(['m-1', 'm-3']); + expect(convosDeleteDocuments).toHaveBeenCalledWith(['c-1']); + + // After cleanup, _meiliIndex flags reset so docs get re-indexed with user field + expect(mockBatchResetMeiliFlags).toHaveBeenCalledWith(Message.collection); + expect(mockBatchResetMeiliFlags).toHaveBeenCalledWith(Conversation.collection); + + // And a full re-sync runs even though progress reported isComplete: true + expect(Message.syncWithMeili).toHaveBeenCalledTimes(1); + expect(Conversation.syncWithMeili).toHaveBeenCalledTimes(1); + expect(mockLogger.info).toHaveBeenCalledWith( + '[indexSync] Orphaned documents removed. Forcing re-sync to restore them with the user field...', + ); + }); + + test('does not reset flags when orphan detection finds nothing to delete', async () => { + Message.getSyncProgress.mockResolvedValue({ + totalProcessed: 100, + totalDocuments: 100, + isComplete: true, + }); + Conversation.getSyncProgress.mockResolvedValue({ + totalProcessed: 50, + totalDocuments: 50, + isComplete: true, + }); + + // All docs have a user field — no orphans + mockMeiliIndex.mockReturnValue({ + getSettings: jest.fn().mockResolvedValue({ filterableAttributes: ['user'] }), + updateSettings: jest.fn().mockResolvedValue({}), + search: jest.fn().mockResolvedValue({ + hits: [{ messageId: 'm-1', user: 'user-A', text: 'Barcelona' }], + }), + deleteDocuments: jest.fn().mockResolvedValue({}), + }); + + const indexSync = require('./indexSync'); + await indexSync(); + + expect(mockBatchResetMeiliFlags).not.toHaveBeenCalled(); + expect(Message.syncWithMeili).not.toHaveBeenCalled(); + expect(Conversation.syncWithMeili).not.toHaveBeenCalled(); + }); });