Add comprehensive documentation for employee learning platform

- Created handover document outlining design decisions and application functionality.
- Developed implementation plan detailing phased approach for service development.
- Specified ingestion service responsibilities, API surface, and processing pipeline.

app/services/ingestion/src/pipeline/chunk.ts

@@ -0,0 +1,246 @@
+import { v4 as uuid } from 'uuid';
+import type { Chunk, DocumentFormat } from '../types.js';
+
+const MD_MIN = 100;
+const MD_MAX = 1500;
+const TXT_WINDOW = 800;
+const TXT_OVERLAP = 150;
+const PDF_MIN = 100;
+const PDF_MAX = 1200;
+
+// ---------------------------------------------------------------------------
+// MD chunking — heading-based
+// ---------------------------------------------------------------------------
+
+interface HeadingSection {
+  level: number;
+  heading: string;
+  parent: string | null;
+  content: string;
+}
+
+function parseMdSections(text: string): HeadingSection[] {
+  const lines = text.split('\n');
+  const sections: HeadingSection[] = [];
+  let current: HeadingSection | null = null;
+  const parentStack: { level: number; heading: string }[] = [];
+
+  for (const line of lines) {
+    const h3 = line.match(/^### (.+)/);
+    const h2 = line.match(/^## (.+)/);
+    const h1 = line.match(/^# (.+)/);
+    const match = h3 ?? h2 ?? h1;
+
+    if (match) {
+      if (current) sections.push(current);
+      const level = h1 ? 1 : h2 ? 2 : 3;
+      const heading = match[1] ?? line;
+
+      // Maintain parent stack
+      while (parentStack.length > 0 && (parentStack[parentStack.length - 1]?.level ?? 0) >= level) {
+        parentStack.pop();
+      }
+      const parent = parentStack[parentStack.length - 1]?.heading ?? null;
+      parentStack.push({ level, heading });
+
+      current = { level, heading, parent, content: '' };
+    } else if (current) {
+      current.content += line + '\n';
+    }
+  }
+  if (current) sections.push(current);
+  return sections;
+}
+
+function splitOnParagraphs(text: string, maxSize: number): string[] {
+  const paragraphs = text.split(/\n\n+/);
+  const parts: string[] = [];
+  let current = '';
+
+  for (const para of paragraphs) {
+    if ((current + para).length > maxSize && current.length > 0) {
+      parts.push(current.trim());
+      current = para;
+    } else {
+      current = current ? current + '\n\n' + para : para;
+    }
+  }
+  if (current.trim()) parts.push(current.trim());
+  return parts;
+}
+
+function chunkMd(text: string, documentId: string): Chunk[] {
+  const sections = parseMdSections(text);
+  const merged: HeadingSection[] = [];
+
+  for (let i = 0; i < sections.length; i++) {
+    const sec = sections[i];
+    if (sec === undefined) continue;
+    const fullText = `${'#'.repeat(sec.level)} ${sec.heading}\n\n${sec.content}`.trim();
+
+    if (fullText.length < MD_MIN && i + 1 < sections.length) {
+      // Merge with next sibling by appending to next's content prefix
+      const next = sections[i + 1];
+      if (next !== undefined) {
+        next.content = fullText + '\n\n' + next.content;
+        continue;
+      }
+    }
+    merged.push(sec);
+  }
+
+  const chunks: Chunk[] = [];
+  let globalIndex = 0;
+
+  for (const sec of merged) {
+    const fullText = `${'#'.repeat(sec.level)} ${sec.heading}\n\n${sec.content}`.trim();
+    const parts = fullText.length > MD_MAX ? splitOnParagraphs(fullText, MD_MAX) : [fullText];
+
+    for (const part of parts) {
+      if (part.length < 1) continue;
+      chunks.push({
+        id: uuid(),
+        documentId,
+        text: part,
+        format: 'md',
+        index: globalIndex++,
+        metadata: {
+          headingLevel: sec.level,
+          headingText: sec.heading,
+          parentHeading: sec.parent ?? undefined,
+        },
+      });
+    }
+  }
+
+  return chunks;
+}
+
+// ---------------------------------------------------------------------------
+// TXT chunking — sliding window
+// ---------------------------------------------------------------------------
+
+function splitSentences(text: string): string[] {
+  return text.match(/[^.!?]+[.!?]+\s*/g) ?? [text];
+}
+
+function chunkTxt(text: string, documentId: string): Chunk[] {
+  const paragraphs = text.split(/\n\n+/).filter(p => p.trim().length > 0);
+  const chunks: Chunk[] = [];
+  let current = '';
+  let index = 0;
+  const total = text.length;
+
+  const pushChunk = (t: string) => {
+    const pos = index * TXT_WINDOW;
+    chunks.push({
+      id: uuid(),
+      documentId,
+      text: t.trim(),
+      format: 'txt',
+      index: index++,
+      metadata: {
+        approximatePosition:
+          pos < total * 0.25 ? 'start' : pos > total * 0.75 ? 'end' : 'middle',
+      },
+    });
+  };
+
+  for (const para of paragraphs) {
+    if ((current + ' ' + para).trim().length <= TXT_WINDOW) {
+      current = (current + ' ' + para).trim();
+    } else {
+      if (current.length >= MD_MIN) pushChunk(current);
+      // Para may itself exceed window — split on sentences
+      if (para.length > TXT_WINDOW) {
+        const sentences = splitSentences(para);
+        let buf = '';
+        for (const sent of sentences) {
+          if ((buf + sent).length > TXT_WINDOW && buf.length >= MD_MIN) {
+            pushChunk(buf);
+            // Keep overlap
+            const words = buf.split(' ');
+            buf = words.slice(-Math.floor(TXT_OVERLAP / 5)).join(' ') + ' ' + sent;
+          } else {
+            buf += sent;
+          }
+        }
+        current = buf;
+      } else {
+        current = para;
+      }
+    }
+  }
+  if (current.trim().length >= MD_MIN) pushChunk(current);
+
+  return chunks;
+}
+
+// ---------------------------------------------------------------------------
+// PDF chunking — page + paragraph
+// ---------------------------------------------------------------------------
+
+function chunkPdf(text: string, documentId: string): Chunk[] {
+  const pages = text.split('---PAGE---');
+  const chunks: Chunk[] = [];
+  let globalIndex = 0;
+
+  for (let pageIdx = 0; pageIdx < pages.length; pageIdx++) {
+    const page = pages[pageIdx];
+    if (page === undefined || !page.trim()) continue;
+
+    const paragraphs = page.split(/\n\n+/).filter(p => p.trim().length > 0);
+    let chunkOnPage = 0;
+    let accumulator = '';
+
+    const flushAccumulator = () => {
+      const t = accumulator.trim();
+      if (t.length >= PDF_MIN) {
+        chunks.push({
+          id: uuid(),
+          documentId,
+          text: t,
+          format: 'pdf',
+          index: globalIndex++,
+          metadata: { pageNumber: pageIdx + 1, chunkIndexOnPage: chunkOnPage++ },
+        });
+      }
+      accumulator = '';
+    };
+
+    for (const para of paragraphs) {
+      if ((accumulator + '\n\n' + para).trim().length > PDF_MAX) {
+        flushAccumulator();
+        // Para may exceed max on its own — hard split at sentence boundary
+        if (para.length > PDF_MAX) {
+          const sentences = splitSentences(para);
+          for (const sent of sentences) {
+            if ((accumulator + sent).length > PDF_MAX && accumulator.length >= PDF_MIN) {
+              flushAccumulator();
+            }
+            accumulator += sent;
+          }
+        } else {
+          accumulator = para;
+        }
+      } else {
+        accumulator = accumulator ? accumulator + '\n\n' + para : para;
+      }
+    }
+    flushAccumulator();
+  }
+
+  return chunks;
+}
+
+// ---------------------------------------------------------------------------
+// Public entry
+// ---------------------------------------------------------------------------
+
+export function chunk(text: string, format: DocumentFormat, documentId: string): Chunk[] {
+  switch (format) {
+    case 'md':  return chunkMd(text, documentId);
+    case 'txt': return chunkTxt(text, documentId);
+    case 'pdf': return chunkPdf(text, documentId);
+  }
+}

app/services/ingestion/src/pipeline/clean.ts

@@ -0,0 +1,20 @@
+import type { Chunk } from '../types.js';
+
+const MIN_CLEAN_LENGTH = 80;
+
+export function clean(chunks: Chunk[]): Chunk[] {
+  return chunks
+    .map(c => ({ ...c, text: cleanText(c.text) }))
+    .filter(c => c.text.length >= MIN_CLEAN_LENGTH);
+}
+
+function cleanText(text: string): string {
+  return text
+    .normalize('NFC')
+    // Remove null bytes and non-printable characters (keep tabs + newlines)
+    .replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, '')
+    // Collapse 3+ consecutive newlines to 2
+    .replace(/\n{3,}/g, '\n\n')
+    // Trim leading/trailing whitespace
+    .trim();
+}

app/services/ingestion/src/pipeline/embed.ts

@@ -0,0 +1,102 @@
+import { v4 as uuid } from 'uuid';
+import { openai, EMBEDDING_MODEL, EMBEDDING_BATCH_SIZE } from '../lib/openai.js';
+import { qdrant, QDRANT_COLLECTIONS } from '../lib/qdrant.js';
+import { updateTopicQdrantIds } from './write.js';
+import type { Chunk, WrittenTopic, SourceChunkPayload, TopicSummaryPayload } from '../types.js';
+
+async function embedTexts(texts: string[]): Promise<number[][]> {
+  const vectors: number[][] = [];
+
+  for (let i = 0; i < texts.length; i += EMBEDDING_BATCH_SIZE) {
+    const batch = texts.slice(i, i + EMBEDDING_BATCH_SIZE);
+    const response = await openai.embeddings.create({
+      model: EMBEDDING_MODEL,
+      input: batch,
+    });
+    for (const item of response.data) {
+      vectors.push(item.embedding);
+    }
+  }
+
+  return vectors;
+}
+
+export async function embedAndStore(
+  chunks: Chunk[],
+  writtenTopics: WrittenTopic[],
+  onProgress: (embedded: number) => void,
+): Promise<void> {
+  // Build chunk → topic mapping
+  const chunkTopicMap = new Map<string, string>();
+  const chunkThemeMap = new Map<string, string>();
+  for (const topic of writtenTopics) {
+    for (const chunkId of topic.sourceChunkIds) {
+      chunkTopicMap.set(chunkId, topic.id);
+      chunkThemeMap.set(chunkId, topic.themeId);
+    }
+  }
+
+  // -------------------------------------------------------------------------
+  // Source chunks
+  // -------------------------------------------------------------------------
+  const chunkTexts = chunks.map(c => c.text);
+  const chunkVectors = await embedTexts(chunkTexts);
+
+  const sourcePoints = chunks.map((c, i) => {
+    const vector = chunkVectors[i];
+    if (!vector) throw new Error(`Missing embedding for chunk index ${i}`);
+
+    const payload: SourceChunkPayload = {
+      source_document_id: c.documentId,
+      chunk_index: c.index,
+      text: c.text,
+      theme_id: chunkThemeMap.get(c.id) ?? null,
+      topic_id: chunkTopicMap.get(c.id) ?? null,
+      format: c.format,
+    };
+
+    return { id: c.id, vector, payload };
+  });
+
+  // Upsert in batches
+  for (let i = 0; i < sourcePoints.length; i += EMBEDDING_BATCH_SIZE) {
+    const batch = sourcePoints.slice(i, i + EMBEDDING_BATCH_SIZE);
+    await qdrant.upsert(QDRANT_COLLECTIONS.SOURCE_CHUNKS, { points: batch });
+    onProgress(i + batch.length);
+  }
+
+  // -------------------------------------------------------------------------
+  // Topic summaries
+  // -------------------------------------------------------------------------
+  const topicTexts = writtenTopics.map(t => t.body);
+  const topicVectors = await embedTexts(topicTexts);
+
+  const summaryPoints = writtenTopics.map((topic, i) => {
+    const vector = topicVectors[i];
+    if (!vector) throw new Error(`Missing embedding for topic index ${i}`);
+
+    const payload: TopicSummaryPayload = {
+      topic_id: topic.id,
+      theme_id: topic.themeId,
+      title: topic.title,
+      text: topic.body,
+    };
+
+    return { id: uuid(), vector, payload };
+  });
+
+  for (let i = 0; i < summaryPoints.length; i += EMBEDDING_BATCH_SIZE) {
+    const batch = summaryPoints.slice(i, i + EMBEDDING_BATCH_SIZE);
+    await qdrant.upsert(QDRANT_COLLECTIONS.TOPIC_SUMMARIES, { points: batch });
+  }
+
+  // -------------------------------------------------------------------------
+  // Update topics.qdrant_chunk_ids in PocketBase
+  // -------------------------------------------------------------------------
+  for (const topic of writtenTopics) {
+    const qdrantIds = topic.sourceChunkIds.filter(id => chunkTopicMap.get(id) === topic.id);
+    if (qdrantIds.length > 0) {
+      await updateTopicQdrantIds(topic.id, qdrantIds);
+    }
+  }
+}

app/services/ingestion/src/pipeline/extract.ts

@@ -0,0 +1,40 @@
+import fs from 'node:fs/promises';
+import type { DocumentFormat } from '../types.js';
+
+// Lazy import — pdf-parse has side effects on module load
+async function getPdfParse() {
+  const mod = await import('pdf-parse');
+  return mod.default ?? mod;
+}
+
+export async function extract(filePath: string, format: DocumentFormat): Promise<string> {
+  let fileBuffer: Buffer;
+  try {
+    fileBuffer = await fs.readFile(filePath);
+  } catch {
+    throw new Error('file_not_found');
+  }
+
+  if (format === 'txt' || format === 'md') {
+    return fileBuffer.toString('utf-8');
+  }
+
+  // PDF — collect one text string per page via pagerender callback
+  const pdfParse = await getPdfParse();
+  const pageTexts: string[] = [];
+
+  await pdfParse(fileBuffer, {
+    pagerender: (pageData: { getTextContent: () => Promise<{ items: Array<{ str: string }> }> }) =>
+      pageData.getTextContent().then(tc => {
+        const text = tc.items.map(i => i.str).join(' ').trim();
+        if (text) pageTexts.push(text);
+        return text;
+      }),
+  });
+
+  if (pageTexts.length === 0) {
+    throw new Error('pdf_extraction_empty');
+  }
+
+  return pageTexts.join('\n\n---PAGE---\n\n');
+}

app/services/ingestion/src/pipeline/structure.ts

@@ -0,0 +1,181 @@
+import { anthropic, MODELS } from '../lib/anthropic.js';
+import { DraftKBSchema, type Chunk, type DraftKB, type DraftTheme, type DraftTopic, type DocumentFormat } from '../types.js';
+
+const BATCH_SIZE = 40;
+const BATCH_OVERLAP = 5;
+const LARGE_DOC_THRESHOLD = 60;
+
+const SYSTEM_PROMPT = `You are a knowledge architect. Your task is to analyse a set of text chunks from a source document and extract a structured knowledge base.
+
+Output ONLY valid JSON matching the schema provided. No preamble, no explanation, no markdown fences.
+
+Rules:
+- Group related content into Themes. A Theme is a broad subject area.
+- Under each Theme, identify discrete Topics. A Topic covers one specific concept.
+- Identify relationships between Topics: related, prerequisite, or contrast.
+  - related: Topics that complement each other
+  - prerequisite: Topic A must be understood before Topic B
+  - contrast: Topics that represent opposing approaches or concepts
+- For each Topic, extract key terms suitable for a glossary.
+- Assign a complexity weight (1–5) to each Topic.
+  1 = introductory, 5 = advanced
+- Draft a body for each Topic (2–4 paragraphs) based on the source chunks.
+- Draft a description for each Theme (1–2 sentences).
+- Every Topic must reference the chunk IDs that contributed to it.
+
+Output schema:
+{
+  "themes": [
+    {
+      "title": "string",
+      "description": "string",
+      "topics": [
+        {
+          "title": "string",
+          "body": "string",
+          "difficulty": "introductory" | "intermediate" | "advanced",
+          "complexityWeight": 1-5,
+          "keyTerms": ["string"],
+          "sourceChunkIds": ["chunk-id"],
+          "relationships": {
+            "related": ["topic title"],
+            "prerequisites": ["topic title"],
+            "contrasts": ["topic title"]
+          }
+        }
+      ]
+    }
+  ]
+}`;
+
+const STRICT_SUFFIX = '\n\nCRITICAL: Your entire response must be valid JSON only. No text before or after.';
+
+function buildUserPrompt(chunks: Chunk[], filename: string, format: DocumentFormat, strict: boolean): string {
+  const chunkText = chunks
+    .map(c => `[CHUNK-${c.id}]\n${c.text}`)
+    .join('\n\n');
+
+  return `Source document: ${filename}\nFormat: ${format}\n\nChunks:\n${chunkText}\n\nExtract the knowledge base structure from these chunks.${strict ? STRICT_SUFFIX : ''}`;
+}
+
+async function callClaude(chunks: Chunk[], filename: string, format: DocumentFormat, strict: boolean): Promise<DraftKB> {
+  const response = await anthropic.messages.create({
+    model: MODELS.SONNET,
+    max_tokens: 8000,
+    temperature: 0,
+    system: SYSTEM_PROMPT,
+    messages: [{ role: 'user', content: buildUserPrompt(chunks, filename, format, strict) }],
+  });
+
+  const textBlock = response.content.find(b => b.type === 'text');
+  if (!textBlock || textBlock.type !== 'text') {
+    throw new Error('structure_extraction_failed: no text block in response');
+  }
+
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(textBlock.text);
+  } catch {
+    if (strict) throw new Error('structure_extraction_failed');
+    return callClaude(chunks, filename, format, true);
+  }
+
+  const result = DraftKBSchema.safeParse(parsed);
+  if (!result.success) {
+    if (strict) throw new Error('structure_extraction_failed');
+    return callClaude(chunks, filename, format, true);
+  }
+
+  if (result.data.themes.length === 0) {
+    throw new Error('no_structure_found');
+  }
+
+  return result.data;
+}
+
+// ---------------------------------------------------------------------------
+// DraftKB merge (for large documents processed in batches)
+// ---------------------------------------------------------------------------
+
+interface MergedTheme {
+  title: string;
+  description: string;
+  topicMap: Map<string, DraftTopic>;
+}
+
+function mergeDraftKBs(batches: DraftKB[]): DraftKB {
+  const themeMap = new Map<string, MergedTheme>();
+
+  for (const kb of batches) {
+    for (const theme of kb.themes) {
+      const key = theme.title.toLowerCase().trim();
+      const existing = themeMap.get(key);
+
+      if (!existing) {
+        themeMap.set(key, {
+          title: theme.title,
+          description: theme.description,
+          topicMap: new Map(theme.topics.map(t => [t.title.toLowerCase().trim(), { ...t }])),
+        });
+      } else {
+        if (theme.description.length > existing.description.length) {
+          existing.description = theme.description;
+        }
+        for (const topic of theme.topics) {
+          const tKey = topic.title.toLowerCase().trim();
+          const existingTopic = existing.topicMap.get(tKey);
+          if (!existingTopic) {
+            existing.topicMap.set(tKey, { ...topic });
+          } else {
+            existingTopic.body =
+              existingTopic.body.length >= topic.body.length ? existingTopic.body : topic.body;
+            existingTopic.sourceChunkIds = [
+              ...new Set([...existingTopic.sourceChunkIds, ...topic.sourceChunkIds]),
+            ];
+            existingTopic.relationships = {
+              related: [...new Set([...existingTopic.relationships.related, ...topic.relationships.related])],
+              prerequisites: [...new Set([...existingTopic.relationships.prerequisites, ...topic.relationships.prerequisites])],
+              contrasts: [...new Set([...existingTopic.relationships.contrasts, ...topic.relationships.contrasts])],
+            };
+          }
+        }
+      }
+    }
+  }
+
+  const themes: DraftTheme[] = [...themeMap.values()].map(t => ({
+    title: t.title,
+    description: t.description,
+    topics: [...t.topicMap.values()],
+  }));
+
+  return { themes };
+}
+
+// ---------------------------------------------------------------------------
+// Public entry
+// ---------------------------------------------------------------------------
+
+export async function extractStructure(
+  chunks: Chunk[],
+  filename: string,
+  format: DocumentFormat,
+): Promise<DraftKB> {
+  if (chunks.length <= LARGE_DOC_THRESHOLD) {
+    return callClaude(chunks, filename, format, false);
+  }
+
+  const batches: DraftKB[] = [];
+  let start = 0;
+
+  while (start < chunks.length) {
+    const end = Math.min(start + BATCH_SIZE, chunks.length);
+    const batchChunks = chunks.slice(start, end);
+    const batchKB = await callClaude(batchChunks, filename, format, false);
+    batches.push(batchKB);
+    if (end >= chunks.length) break;
+    start = end - BATCH_OVERLAP;
+  }
+
+  return mergeDraftKBs(batches);
+}

app/services/ingestion/src/pipeline/write.ts

@@ -0,0 +1,91 @@
+import { getPocketBase } from '../lib/pocketbase.js';
+import type { Chunk, DraftKB, WrittenTopic } from '../types.js';
+
+export async function writeToPocketBase(
+  draftKB: DraftKB,
+  documentId: string,
+  chunkCount: number,
+): Promise<WrittenTopic[]> {
+  const pb = await getPocketBase();
+
+  await pb.collection('source_documents').update(documentId, {
+    status: 'processed',
+    chunk_count: chunkCount,
+    ingested_at: new Date().toISOString(),
+  });
+
+  const writtenTopics: WrittenTopic[] = [];
+  // title → PocketBase topic ID (for relationship resolution)
+  const topicIdByTitle = new Map<string, string>();
+
+  // Pass 1: create all themes and topics (without relationships)
+  for (const theme of draftKB.themes) {
+    const themeRecord = await pb.collection('themes').create({
+      title: theme.title,
+      description: theme.description,
+      status: 'draft',
+      source_documents: [documentId],
+    });
+
+    for (const topic of theme.topics) {
+      const topicRecord = await pb.collection('topics').create({
+        theme: themeRecord.id,
+        title: topic.title,
+        body: topic.body,
+        difficulty: topic.difficulty,
+        complexity_weight: topic.complexityWeight,
+        status: 'draft',
+        key_terms: topic.keyTerms,
+        qdrant_chunk_ids: [],
+        related_topics: [],
+        prerequisite_topics: [],
+        contrast_topics: [],
+      });
+
+      topicIdByTitle.set(topic.title.toLowerCase().trim(), topicRecord.id);
+
+      writtenTopics.push({
+        id: topicRecord.id,
+        title: topic.title,
+        themeId: themeRecord.id,
+        body: topic.body,
+        sourceChunkIds: topic.sourceChunkIds,
+      });
+    }
+  }
+
+  // Pass 2: resolve relationships (title → ID lookup)
+  for (const theme of draftKB.themes) {
+    for (const topic of theme.topics) {
+      const topicId = topicIdByTitle.get(topic.title.toLowerCase().trim());
+      if (!topicId) continue;
+
+      const resolve = (titles: string[]): string[] =>
+        titles
+          .map(t => topicIdByTitle.get(t.toLowerCase().trim()))
+          .filter((id): id is string => id !== undefined);
+
+      await pb.collection('topics').update(topicId, {
+        related_topics: resolve(topic.relationships.related),
+        prerequisite_topics: resolve(topic.relationships.prerequisites),
+        contrast_topics: resolve(topic.relationships.contrasts),
+      });
+    }
+  }
+
+  return writtenTopics;
+}
+
+export async function updateTopicQdrantIds(
+  topicId: string,
+  qdrantChunkIds: string[],
+): Promise<void> {
+  const pb = await getPocketBase();
+  await pb.collection('topics').update(topicId, { qdrant_chunk_ids: qdrantChunkIds });
+}
+
+export async function markDocumentFailed(documentId: string, reason: string): Promise<void> {
+  console.error(`[write] document ${documentId} failed: ${reason}`);
+  const pb = await getPocketBase();
+  await pb.collection('source_documents').update(documentId, { status: 'failed' });
+}