feat: implement pbUpsert helper for streamlined database operations and update related functions

src/lib/extractionPipeline.js

@@ -1,5 +1,5 @@
 import * as db from './db';
-import { callLLM } from './llm';
+import { callLLM, cachedSystem } from './llm';
 import { extractionLimiter } from './llmRetry';
 import { EMIT_KNOWLEDGE_GRAPH_TOOL } from './llmTools';
 
@@ -42,8 +42,6 @@ Assign a learning_relevance to every topic:
 Relation types: related_to | depends_on | part_of | executed_by.
 `;
 
-const cachedSystem = (text) => [{ type: 'text', text, cache_control: { type: 'ephemeral' } }];
-
 /**
  * Sentence-aware chunker with overlap.
  *
@@ -171,7 +169,7 @@ export async function processSourceText(textContent, sourceName, { signal } = {}
       }
     }
 
-    await mergeKnowledgeGraph({ topics: allExtractedTopics, relations: allExtractedRelations });
+    await mergeKnowledgeGraph(existingTopics, { topics: allExtractedTopics, relations: allExtractedRelations });
     await db.updateSourceStatus(sourceId, 'completed');
 
     return { success: true, data: { topics: allExtractedTopics, relations: allExtractedRelations } };
@@ -183,8 +181,7 @@ export async function processSourceText(textContent, sourceName, { signal } = {}
   }
 }
 
-async function mergeKnowledgeGraph(newData) {
-  const existingTopics = await db.getTopics();
+async function mergeKnowledgeGraph(existingTopics, newData) {
   const existingRelations = await db.getRelations();
 
   const topicsMap = new Map(existingTopics.map(t => [t.id, t]));