/** * ChunkIndexingService — Indexation incrémentale d'une note en fragments. * * Inspiré d'AppFlowy flowy-ai/src/embeddings/scheduler.rs. * * Fonctionnement : * 1. Récupère les fragmentIds existants en DB pour la note * 2. Chunk le contenu actuel (via chunkNoteContent) * 3. Compare les hash : * - Inchangés (hash identique) → skip * - Nouveaux (hash absent) → embed + insert * - Supprimés (hash en DB mais plus dans le contenu) → delete * 4. Embed les nouveaux fragments via une queue à concurrence limitée * 5. Retry avec backoff exponentiel en cas d'erreur API * * Garanties : * - Pas de re-embed des fragments inchangés (économie API) * - Pas de fragments orphelins (stale supprimés) * - Pas de race condition (verrou par noteId) */ import PQueue from 'p-queue' import { prisma } from '@/lib/prisma' import { embeddingService } from './embedding.service' import { chunkNoteContent, type NoteChunk, } from '@/lib/text/note-chunking' import { prepareNoteTextForEmbedding, } from '@/lib/text/plain-text' import { Prisma } from '@prisma/client' const EMBEDDING_CONCURRENCY = 4 const MAX_RETRIES = 3 const RETRY_BASE_DELAY_MS = 1000 const embeddingQueue = new PQueue({ concurrency: EMBEDDING_CONCURRENCY }) const noteLocks = new Map>() export interface IndexResult { noteId: string totalFragments: number newFragments: number skipped: number deleted: number durationMs: number } export class ChunkIndexingService { /** * Indexe une note en fragments. Ne re-embed que les fragments modifiés. */ async indexNote( noteId: string, title: string | null | undefined, content: string, ): Promise { while (noteLocks.has(noteId)) { await noteLocks.get(noteId) } const task = this.doIndexNote(noteId, title, content) noteLocks.set(noteId, task.then(() => {}, () => {})) try { return await task } finally { noteLocks.delete(noteId) } } private async doIndexNote( noteId: string, title: string | null | undefined, content: string, ): Promise { const start = Date.now() const plain = prepareNoteTextForEmbedding(title, content) const newChunks = chunkNoteContent(noteId, plain) const newFragmentIds = new Set(newChunks.map((c) => c.fragmentId)) const existing = await prisma.noteEmbeddingChunk.findMany({ where: { noteId }, select: { fragmentId: true }, }) const existingIds = new Set(existing.map((e) => e.fragmentId)) const toDelete = [...existingIds].filter((id) => !newFragmentIds.has(id)) const toEmbed = newChunks.filter((c) => !existingIds.has(c.fragmentId)) const skipped = newChunks.filter((c) => existingIds.has(c.fragmentId)) if (toDelete.length > 0) { await prisma.noteEmbeddingChunk.deleteMany({ where: { noteId, fragmentId: { in: toDelete }, }, }) } const embeddedChunks = await this.embedChunks(toEmbed) if (embeddedChunks.length > 0) { await this.upsertChunks(noteId, embeddedChunks) } return { noteId, totalFragments: newChunks.length, newFragments: embeddedChunks.length, skipped: skipped.length, deleted: toDelete.length, durationMs: Date.now() - start, } } /** * Embed une liste de fragments avec queue à concurrence limitée + retry. */ private async embedChunks(chunks: NoteChunk[]): Promise< Array > { if (chunks.length === 0) return [] const results = await Promise.all( chunks.map((chunk) => embeddingQueue.add(() => this.embedWithRetry(chunk)), ), ) return results.filter( (r): r is NoteChunk & { embedding: number[] } => r !== null, ) } private async embedWithRetry( chunk: NoteChunk, ): Promise<(NoteChunk & { embedding: number[] }) | null> { let lastError: Error | null = null for (let attempt = 0; attempt < MAX_RETRIES; attempt++) { try { const embedding = await embeddingService.embedText(chunk.content) return { ...chunk, embedding } } catch (err: any) { lastError = err if (attempt < MAX_RETRIES - 1) { const delay = RETRY_BASE_DELAY_MS * Math.pow(2, attempt) console.warn( `[ChunkIndexing] Retry ${attempt + 1}/${MAX_RETRIES} for fragment ${chunk.fragmentId} after ${delay}ms: ${err.message}`, ) await new Promise((resolve) => setTimeout(resolve, delay)) } } } console.error( `[ChunkIndexing] Failed to embed fragment ${chunk.fragmentId} after ${MAX_RETRIES} attempts: ${lastError?.message}`, ) return null } /** * Upsert transactionnel des fragments embeddés. * Utilise $executeRawUnsafe pour la colonne vector(1536). */ private async upsertChunks( noteId: string, chunks: Array, ): Promise { for (const chunk of chunks) { const vecStr = `[${chunk.embedding.join(',')}]` await prisma.$executeRaw` INSERT INTO "NoteEmbeddingChunk" ( "id", "noteId", "fragmentId", "chunkIndex", "content", "charCount", "embedding", "embeddingModel", "createdAt", "updatedAt" ) VALUES ( gen_random_uuid(), ${noteId}, ${chunk.fragmentId}, ${chunk.chunkIndex}, ${chunk.content}, ${chunk.charCount}, ${vecStr}::vector, 'text-embedding-3-small', now(), now() ) ON CONFLICT ("noteId", "fragmentId") DO UPDATE SET "content" = EXCLUDED."content", "charCount" = EXCLUDED."charCount", "embedding" = EXCLUDED."embedding", "chunkIndex" = EXCLUDED."chunkIndex", "updatedAt" = now() ` } } /** * Supprime tous les fragments d'une note. */ async deleteNoteChunks(noteId: string): Promise { await prisma.noteEmbeddingChunk.deleteMany({ where: { noteId } }) } /** * Vérifie si une note a déjà des fragments indexés. */ async hasChunks(noteId: string): Promise { const count = await prisma.noteEmbeddingChunk.count({ where: { noteId }, take: 1, }) return count > 0 } } export const chunkIndexingService = new ChunkIndexingService()