PostgreSQL + pgvector: Building Semantic Search and RAG Without a Separate Vector Database

CREATE TABLE documents (
    id          BIGSERIAL PRIMARY KEY,
    title       TEXT NOT NULL,
    content     TEXT NOT NULL,
    category    TEXT,
    author_id   BIGINT,
    tenant_id   BIGINT NOT NULL,
    created_at  TIMESTAMPTZ DEFAULT NOW(),
    updated_at  TIMESTAMPTZ DEFAULT NOW()
);

INSERT INTO documents (title, content, category, author_id, tenant_id) VALUES
('Shopify Webhooks Guide',
 'Shopify sends webhook events to your endpoint whenever specific events occur in a store. To ensure delivery reliability, implement exponential backoff when webhook delivery fails. Verify HMAC signatures on every incoming request using the X-Shopify-Hmac-Sha256 header.',
 'shopify', 1, 1),
('Node.js API Guide',
 'Building REST APIs with Node.js requires handling async errors properly. Use middleware to catch unhandled promise rejections. Implement request validation before business logic.',
 'backend', 1, 1),
('PostgreSQL Transactions Guide',
 'ACID transactions guarantee data integrity. Use BEGIN, COMMIT, and ROLLBACK to wrap operations that must succeed or fail together. Avoid long-running transactions to prevent lock contention.',
 'database', 1, 1),
('Redis Caching Guide',
 'Cache frequently accessed data with appropriate TTL values. Use cache-aside pattern to avoid cache stampede. Invalidate cache entries when underlying data changes.',
 'infrastructure', 1, 1),
('API Security Guide',
 'Protect API endpoints with authentication and authorization. Use JWT tokens with short expiry and refresh token rotation. Rate limiting prevents abuse and protects backend resources.',
 'security', 1, 1);

-- Current keyword search
SELECT id, title, category
FROM documents
WHERE tenant_id = $1
  AND (title ILIKE '%' || $2 || '%' OR content ILIKE '%' || $2 || '%')
ORDER BY updated_at DESC;

-- What FTS sees:
SELECT to_tsvector('english', 'Use exponential backoff when webhook delivery fails');
-- Result: 'backoff':3 'deliveri':6 'exponenti':2 'fail':7 'use':1 'webhook':5

SELECT to_tsquery('english', 'retry & failed & webhook');
-- Result: 'retri' & 'fail' & 'webhook'

"How do I retry failed webhook requests?"
        ↓
  Embedding Model
        ↓
[0.0231, -0.1847, 0.4421, 0.0089, -0.3201, ...]
  (1,536 dimensions for text-embedding-3-small)

import OpenAI from 'openai';

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

async function generateEmbedding(text) {
  const response = await openai.embeddings.create({
    model: 'text-embedding-3-small',
    input: text,
    encoding_format: 'float',
  });
  return response.data[0].embedding; // float[]
}

PostgreSQL
├── users
├── documents         (relational data)
├── document_chunks   (chunks + embeddings)
└── All in one database

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: pgvector/pgvector:pg17
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: docsearch
    ports:
      - '5432:5432'
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

docker-compose up -d

CREATE EXTENSION IF NOT EXISTS vector;

SELECT * FROM pg_extension WHERE extname = 'vector';

Document: "Shopify Webhooks Guide" (2,000 words)
    ├── Chunk 1: Introduction + What are webhooks (chunk_index: 0)
    ├── Chunk 2: HMAC signature verification (chunk_index: 1)
    ├── Chunk 3: Retry logic and exponential backoff (chunk_index: 2)
    └── Chunk 4: Testing with ngrok and CLI (chunk_index: 3)

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
    id              BIGSERIAL PRIMARY KEY,
    title           TEXT NOT NULL,
    content         TEXT NOT NULL,
    category        TEXT,
    author_id       BIGINT,
    tenant_id       BIGINT NOT NULL,
    content_hash    TEXT,          -- For detecting content changes
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    updated_at      TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE document_chunks (
    id              BIGSERIAL PRIMARY KEY,
    document_id     BIGINT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index     INTEGER NOT NULL,
    content         TEXT NOT NULL,
    metadata        JSONB DEFAULT '{}',
    embedding       vector(1536),   -- text-embedding-3-small dimensions
    embedding_model TEXT,           -- Track which model generated this
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    updated_at      TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(document_id, chunk_index)
);

-- Standard relational indexes
CREATE INDEX idx_document_chunks_document_id ON document_chunks(document_id);
CREATE INDEX idx_documents_tenant_id ON documents(tenant_id);
CREATE INDEX idx_document_chunks_tenant ON document_chunks(document_id)
    INCLUDE (content, embedding);

-- Composite index for tenant-scoped document filtering
CREATE INDEX idx_documents_tenant_category ON documents(tenant_id, category);

// src/documents/chunker.js
const CHUNK_SIZE = 500;       // approximate words (not tokens)
const CHUNK_OVERLAP = 75;     // words of overlap between adjacent chunks

/**
 * Splits document content into overlapping chunks.
 * Attempts to split at paragraph boundaries to preserve coherence.
 */
function chunkDocument(document) {
  const { id, title, content, category, tenant_id } = document;

  // Split on paragraph boundaries first
  const paragraphs = content
    .split(/\n{2,}/)
    .map(p => p.trim())
    .filter(p => p.length > 0);

  const chunks = [];
  let currentChunk = [];
  let currentWordCount = 0;
  let chunkIndex = 0;

  for (const paragraph of paragraphs) {
    const words = paragraph.split(/\s+/);

    // If adding this paragraph would exceed chunk size, flush current chunk
    if (currentWordCount + words.length > CHUNK_SIZE && currentChunk.length > 0) {
      const chunkContent = currentChunk.join('\n\n');

      chunks.push({
        document_id: id,
        chunk_index: chunkIndex++,
        content: chunkContent,
        metadata: {
          document_title: title,
          category,
          tenant_id,
          char_count: chunkContent.length,
          word_count: currentWordCount,
        },
      });

      // Carry over last N words as overlap
      const overlapWords = currentChunk
        .join(' ')
        .split(/\s+/)
        .slice(-CHUNK_OVERLAP);

      currentChunk = [overlapWords.join(' ')];
      currentWordCount = overlapWords.length;
    }

    currentChunk.push(paragraph);
    currentWordCount += words.length;
  }

  // Flush the last chunk
  if (currentChunk.length > 0) {
    const chunkContent = currentChunk.join('\n\n');
    chunks.push({
      document_id: id,
      chunk_index: chunkIndex,
      content: chunkContent,
      metadata: {
        document_title: title,
        category,
        tenant_id,
        char_count: chunkContent.length,
        word_count: currentWordCount,
      },
    });
  }

  return chunks;
}

module.exports = { chunkDocument };

// src/embeddings/embeddingService.js
import OpenAI from 'openai';

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

const EMBEDDING_MODEL = 'text-embedding-3-small';
const MAX_BATCH_SIZE = 100; // OpenAI allows batching multiple inputs
const MAX_RETRIES = 3;
const INITIAL_RETRY_DELAY_MS = 1000;

/**
 * Generate an embedding for a single text string.
 * Returns a float[] of 1,536 dimensions.
 */
async function generateEmbedding(text) {
  return generateEmbeddings([text]).then(results => results[0]);
}

/**
 * Generate embeddings for an array of texts in a single API call.
 * Batching is significantly more efficient than one call per chunk.
 */
async function generateEmbeddings(texts) {
  let attempt = 0;

  while (attempt < MAX_RETRIES) {
    try {
      const response = await openai.embeddings.create({
        model: EMBEDDING_MODEL,
        input: texts,
        encoding_format: 'float',
      });

      // Response shape: { data: [{ index: 0, embedding: float[] }, ...] }
      // Sort by index to preserve input order (API guarantees order but explicit is safer)
      return response.data
        .sort((a, b) => a.index - b.index)
        .map(item => item.embedding);

    } catch (error) {
      attempt++;

      if (attempt >= MAX_RETRIES) {
        throw new Error(
          `Embedding generation failed after ${MAX_RETRIES} attempts: ${error.message}`
        );
      }

      // Exponential backoff: 1s, 2s, 4s
      const delay = INITIAL_RETRY_DELAY_MS * Math.pow(2, attempt - 1);
      console.warn(`Embedding attempt ${attempt} failed, retrying in ${delay}ms`);
      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }
}

/**
 * Process chunks in batches to respect API limits.
 */
async function generateChunkEmbeddings(chunks) {
  const results = [];

  for (let i = 0; i < chunks.length; i += MAX_BATCH_SIZE) {
    const batch = chunks.slice(i, i + MAX_BATCH_SIZE);
    const texts = batch.map(chunk => chunk.content);
    const embeddings = await generateEmbeddings(texts);

    results.push(
      ...batch.map((chunk, idx) => ({
        ...chunk,
        embedding: embeddings[idx],
        embedding_model: EMBEDDING_MODEL,
      }))
    );
  }

  return results;
}

export { generateEmbedding, generateEmbeddings, generateChunkEmbeddings, EMBEDDING_MODEL };

// src/db/chunkRepository.js
import { PrismaClient } from '@prisma/client';
import { Prisma } from '@prisma/client';

const prisma = new PrismaClient();

/**
 * Insert document chunks with their embeddings.
 * Uses raw SQL because Prisma requires explicit casting for the vector type.
 */
async function insertChunksWithEmbeddings(chunks) {
  // Process in a transaction to ensure all-or-nothing insertion
  return prisma.$transaction(async (tx) => {
    for (const chunk of chunks) {
      // Format the vector as a PostgreSQL array literal: '[0.1, -0.2, ...]'
      const vectorLiteral = `[${chunk.embedding.join(',')}]`;

      await tx.$executeRaw`
        INSERT INTO document_chunks
          (document_id, chunk_index, content, metadata, embedding, embedding_model)
        VALUES (
          ${chunk.document_id},
          ${chunk.chunk_index},
          ${chunk.content},
          ${chunk.metadata}::jsonb,
          ${vectorLiteral}::vector,
          ${chunk.embedding_model}
        )
        ON CONFLICT (document_id, chunk_index)
        DO UPDATE SET
          content = EXCLUDED.content,
          metadata = EXCLUDED.metadata,
          embedding = EXCLUDED.embedding,
          embedding_model = EXCLUDED.embedding_model,
          updated_at = NOW()
      `;
    }
  });
}

/**
 * Delete all chunks for a document (before re-processing).
 */
async function deleteChunksForDocument(documentId) {
  return prisma.$executeRaw`
    DELETE FROM document_chunks WHERE document_id = ${documentId}
  `;
}

export { insertChunksWithEmbeddings, deleteChunksForDocument };

// src/search/vectorSearch.js
import { generateEmbedding } from '../embeddings/embeddingService.js';
import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();

/**
 * Search for the top-K most semantically similar chunks to a query.
 *
 * The <=> operator is cosine distance.
 * Cosine distance ranges from 0 (identical direction) to 2 (opposite).
 * Cosine similarity = 1 - cosine_distance.
 *
 * text-embedding-3-small vectors are already normalized, so
 * cosine distance is the appropriate metric.
 */
async function semanticSearch({ query, tenantId, limit = 10, category = null }) {
  const queryEmbedding = await generateEmbedding(query);
  const vectorLiteral = `[${queryEmbedding.join(',')}]`;

  const results = await prisma.$queryRaw`
    SELECT
      dc.id               AS chunk_id,
      dc.document_id,
      dc.chunk_index,
      dc.content,
      dc.metadata,
      d.title,
      d.category,
      1 - (dc.embedding <=> ${vectorLiteral}::vector) AS cosine_similarity
    FROM document_chunks dc
    JOIN documents d ON d.id = dc.document_id
    WHERE d.tenant_id = ${tenantId}
      AND dc.embedding IS NOT NULL
      ${category ? Prisma.sql`AND d.category = ${category}` : Prisma.empty}
    ORDER BY dc.embedding <=> ${vectorLiteral}::vector
    LIMIT ${limit}
  `;

  return results;
}

export { semanticSearch };

-- Add HNSW index for cosine distance (matches our <=> operator)
CREATE INDEX idx_chunk_embedding_hnsw
    ON document_chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

-- Before running a search, tune ef_search for this session
SET hnsw.ef_search = 100;  -- default is 40

-- Or per-query inside a transaction:
BEGIN;
SET LOCAL hnsw.ef_search = 100;
SELECT ... FROM document_chunks ORDER BY embedding <=> $1 LIMIT 10;
COMMIT;

EXPLAIN (ANALYZE, BUFFERS)
SELECT dc.id, dc.content, dc.embedding <=> '[...]'::vector AS distance
FROM document_chunks dc
JOIN documents d ON d.id = dc.document_id
WHERE d.tenant_id = 1
ORDER BY dc.embedding <=> '[...]'::vector
LIMIT 10;

-- Build IVFFlat after loading your initial data (k-means requires data)
-- Rule of thumb: lists = rows / 1000 for up to 1M rows
CREATE INDEX idx_chunk_embedding_ivfflat
    ON document_chunks
    USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);

-- Set probes at query time (default: 1, start with sqrt(lists))
SET ivfflat.probes = 10;

-- Correct: filter is part of the query
SELECT
  dc.id, dc.content, d.title, d.category,
  1 - (dc.embedding <=> '[...]'::vector) AS similarity
FROM document_chunks dc
JOIN documents d ON d.id = dc.document_id
WHERE d.tenant_id = $1              -- <-- MANDATORY
  AND dc.embedding IS NOT NULL
ORDER BY dc.embedding <=> '[...]'::vector
LIMIT 10;

-- Enable iterative scanning (pgvector 0.8+)
SET hnsw.iterative_scan = relaxed_order;
-- Options: off (default), relaxed_order (faster, approx), strict_order (more precise, slower)

-- Safety valve: limit maximum tuples scanned during iteration
SET hnsw.max_scan_tuples = 20000;

-- Partition document_chunks by tenant_id
-- Each partition gets its own HNSW index — smaller index = faster search per tenant
CREATE TABLE document_chunks (
    id          BIGSERIAL,
    document_id BIGINT NOT NULL,
    tenant_id   BIGINT NOT NULL,
    chunk_index INTEGER NOT NULL,
    content     TEXT NOT NULL,
    metadata    JSONB DEFAULT '{}',
    embedding   vector(1536),
    ...
) PARTITION BY HASH (tenant_id);

CREATE TABLE document_chunks_0 PARTITION OF document_chunks
    FOR VALUES WITH (modulus 4, remainder 0);
-- ... (create additional partitions)

-- Add a generated tsvector column for FTS
-- Using 'english' configuration: tokenization + stemming + stop words
ALTER TABLE documents ADD COLUMN search_vector tsvector
    GENERATED ALWAYS AS (
        setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
        setweight(to_tsvector('english', coalesce(content, '')), 'B')
    ) STORED;

-- GIN index for fast full-text search
CREATE INDEX idx_documents_search_vector ON documents USING gin(search_vector);

// src/search/ftsSearch.js
import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();

/**
 * PostgreSQL full-text search using tsvector/tsquery.
 * Uses plainto_tsquery which handles natural language input
 * (vs. to_tsquery which requires explicit operators).
 */
async function fullTextSearch({ query, tenantId, limit = 20, category = null }) {
  // plainto_tsquery converts plain text to tsquery safely
  // 'english' configuration: stemming + stop words
  const results = await prisma.$queryRaw`
    SELECT
      d.id          AS document_id,
      d.title,
      d.category,
      dc.id         AS chunk_id,
      dc.chunk_index,
      dc.content,
      ts_rank(d.search_vector, plainto_tsquery('english', ${query})) AS fts_rank
    FROM documents d
    JOIN document_chunks dc ON dc.document_id = d.id
    WHERE d.tenant_id = ${tenantId}
      AND d.search_vector @@ plainto_tsquery('english', ${query})
      ${category ? `AND d.category = '${category}'` : ''}
    ORDER BY fts_rank DESC
    LIMIT ${limit}
  `;

  return results;
}

export { fullTextSearch };

// src/search/hybridSearch.js
import { semanticSearch } from './vectorSearch.js';
import { fullTextSearch } from './ftsSearch.js';

/**
 * Reciprocal Rank Fusion combines results from multiple ranked lists.
 *
 * RRF score for a document = sum of 1 / (k + rank_in_each_list)
 * where k is a smoothing constant (60 is the standard value from the original paper).
 *
 * Documents appearing high in both lists score highest.
 * Documents appearing in only one list still contribute.
 */
function reciprocalRankFusion(rankedLists, k = 60) {
  const scores = new Map(); // chunk_id -> RRF score
  const metadata = new Map(); // chunk_id -> chunk data

  for (const rankedList of rankedLists) {
    rankedList.forEach((item, rank) => {
      const id = String(item.chunk_id);
      const current = scores.get(id) || 0;
      scores.set(id, current + 1 / (k + rank + 1));

      if (!metadata.has(id)) {
        metadata.set(id, item);
      }
    });
  }

  // Sort by combined RRF score (descending)
  return Array.from(scores.entries())
    .sort((a, b) => b[1] - a[1])
    .map(([id, score]) => ({
      ...metadata.get(id),
      rrf_score: score,
    }));
}

/**
 * Hybrid search: runs FTS and semantic search in parallel,
 * then fuses results using Reciprocal Rank Fusion.
 */
async function hybridSearch({ query, tenantId, limit = 10, category = null }) {
  // Run both searches in parallel
  const [semanticResults, ftsResults] = await Promise.all([
    semanticSearch({ query, tenantId, limit: 20, category }),
    fullTextSearch({ query, tenantId, limit: 20, category }),
  ]);

  // Fuse the two ranked lists
  const fused = reciprocalRankFusion([semanticResults, ftsResults]);

  return fused.slice(0, limit);
}

export { hybridSearch, reciprocalRankFusion };

// src/rag/ragPipeline.js
import OpenAI from 'openai';
import { hybridSearch } from '../search/hybridSearch.js';

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

/**
 * Build the context string from retrieved chunks.
 * Each chunk is labeled with its document title and index for citation.
 */
function buildContext(chunks) {
  return chunks
    .map((chunk, idx) => (
      `[Source ${idx + 1}: ${chunk.title} (chunk ${chunk.chunk_index})]\n${chunk.content}`
    ))
    .join('\n\n---\n\n');
}

/**
 * Extract source attribution from retrieved chunks.
 * Returns unique documents, deduplicated by document_id.
 */
function extractSources(chunks) {
  const seen = new Set();
  return chunks
    .filter(chunk => {
      if (seen.has(chunk.document_id)) return false;
      seen.add(chunk.document_id);
      return true;
    })
    .map(chunk => ({
      document_id: chunk.document_id,
      title: chunk.title,
      category: chunk.category,
      chunk_index: chunk.chunk_index,
    }));
}

/**
 * RAG pipeline:
 * 1. Retrieve relevant chunks via hybrid search
 * 2. Build LLM context from chunks
 * 3. Call LLM with context + user question
 * 4. Return answer + source attribution
 */
async function ragQuery({ question, tenantId, category = null, topK = 5 }) {
  // Step 1: Retrieve relevant chunks
  const chunks = await hybridSearch({
    query: question,
    tenantId,
    limit: topK,
    category,
  });

  if (chunks.length === 0) {
    return {
      answer: "I don't have relevant documentation to answer this question. Please check the available documentation categories.",
      sources: [],
      chunks_used: 0,
    };
  }

  // Step 2: Build context
  const context = buildContext(chunks);

  // Step 3: Call LLM
  // The prompt explicitly instructs the model to:
  // - Use only the provided context
  // - Acknowledge when context is insufficient
  // - Not invent information
  const response = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    messages: [
      {
        role: 'system',
        content: `You are a technical documentation assistant. 
Answer questions using ONLY the provided documentation excerpts below.
If the documentation does not contain enough information to answer the question, 
say so clearly — do not invent information not present in the context.
When referencing specific information, indicate which source it comes from.`,
      },
      {
        role: 'user',
        content: `Documentation context:\n\n${context}\n\nQuestion: ${question}`,
      },
    ],
    temperature: 0.1, // Low temperature for factual retrieval tasks
    max_tokens: 800,
  });

  const answer = response.choices[0].message.content;

  // Step 4: Return answer + sources
  return {
    answer,
    sources: extractSources(chunks),
    chunks_used: chunks.length,
    usage: {
      prompt_tokens: response.usage.prompt_tokens,
      completion_tokens: response.usage.completion_tokens,
    },
  };
}

export { ragQuery };

// src/api/searchRoutes.js
import express from 'express';
import { ragQuery } from '../rag/ragPipeline.js';

const router = express.Router();

// All routes require authentication middleware (authMiddleware checks tenant_id)
router.post('/rag', async (req, res) => {
  const { question, category } = req.body;
  const tenantId = req.user.tenant_id; // Never accept tenant_id from client

  if (!question || typeof question !== 'string' || question.trim().length === 0) {
    return res.status(400).json({ error: 'question is required' });
  }

  try {
    const result = await ragQuery({ question, tenantId, category });
    return res.json(result);
  } catch (error) {
    console.error('RAG query error:', error);
    return res.status(500).json({ error: 'Search temporarily unavailable' });
  }
});

export default router;

// src/documents/documentService.js
import crypto from 'crypto';
import { PrismaClient } from '@prisma/client';
import { chunkDocument } from './chunker.js';
import { generateChunkEmbeddings } from '../embeddings/embeddingService.js';
import { insertChunksWithEmbeddings, deleteChunksForDocument } from '../db/chunkRepository.js';
import { embeddingQueue } from '../queue/embeddingQueue.js';

const prisma = new PrismaClient();

/**
 * Compute content hash to detect actual changes.
 * If content hasn't changed, there's no need to regenerate embeddings.
 */
function computeContentHash(content) {
  return crypto.createHash('sha256').update(content).digest('hex');
}

/**
 * Update a document and queue embedding regeneration if content changed.
 */
async function updateDocument(documentId, { title, content, category }) {
  const newHash = computeContentHash(content);

  const existing = await prisma.document.findUnique({
    where: { id: documentId },
    select: { content_hash: true },
  });

  await prisma.document.update({
    where: { id: documentId },
    data: { title, content, category, content_hash: newHash, updated_at: new Date() },
  });

  // Only queue re-embedding if content actually changed
  if (!existing || existing.content_hash !== newHash) {
    await embeddingQueue.add('process-document', {
      documentId,
      operation: 'update',
    }, {
      attempts: 3,
      backoff: { type: 'exponential', delay: 2000 },
    });
  }

  return { documentId, embeddingQueued: !existing || existing.content_hash !== newHash };
}

export { updateDocument, computeContentHash };

POST /documents
    ↓
Save document to PostgreSQL
    ↓
Add job to Redis/BullMQ queue
    ↓
Return 202 Accepted immediately

[Background Worker]
    ↓
Dequeue job
    ↓
Chunk document
    ↓
Generate embeddings (OpenAI API)
    ↓
Store vectors in PostgreSQL
    ↓
Mark document as indexed

// src/queue/embeddingQueue.js
import { Queue } from 'bullmq';
import Redis from 'ioredis';

const connection = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: Number(process.env.REDIS_PORT) || 6379,
  maxRetriesPerRequest: null, // Required by BullMQ
});

export const embeddingQueue = new Queue('document-embeddings', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000, // 2s, 4s, 8s
    },
    removeOnComplete: 100, // Keep last 100 completed jobs
    removeOnFail: 500,     // Keep last 500 failed jobs for debugging
  },
});

// src/workers/embeddingWorker.js
import { Worker } from 'bullmq';
import Redis from 'ioredis';
import { PrismaClient } from '@prisma/client';
import { chunkDocument } from '../documents/chunker.js';
import { generateChunkEmbeddings } from '../embeddings/embeddingService.js';
import { insertChunksWithEmbeddings, deleteChunksForDocument } from '../db/chunkRepository.js';

const prisma = new PrismaClient();

const connection = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: Number(process.env.REDIS_PORT) || 6379,
  maxRetriesPerRequest: null,
});

const worker = new Worker(
  'document-embeddings',
  async (job) => {
    const { documentId, operation } = job.data;

    await job.updateProgress(10);

    // Fetch the document
    const document = await prisma.document.findUnique({
      where: { id: documentId },
    });

    if (!document) {
      // Document was deleted between enqueue and processing — not an error
      console.log(`Document ${documentId} no longer exists, skipping.`);
      return { skipped: true };
    }

    await job.updateProgress(20);

    // Chunk the document
    const chunks = chunkDocument(document);
    await job.updateProgress(40);

    // Generate embeddings in batches
    const chunksWithEmbeddings = await generateChunkEmbeddings(chunks);
    await job.updateProgress(70);

    // Replace old chunks: delete first, then insert
    // This ensures atomicity — no period with partial data
    await deleteChunksForDocument(documentId);
    await insertChunksWithEmbeddings(chunksWithEmbeddings);

    await job.updateProgress(90);

    // Mark document as indexed
    await prisma.document.update({
      where: { id: documentId },
      data: { indexed_at: new Date() },
    });

    await job.updateProgress(100);

    return { documentId, chunksProcessed: chunks.length };
  },
  {
    connection,
    concurrency: 3, // Process up to 3 documents in parallel
  }
);

worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed: ${JSON.stringify(result)}`);
});

worker.on('failed', (job, error) => {
  console.error(`Job ${job.id} failed (attempt ${job.attemptsMade}): ${error.message}`);
});

export { worker };

// src/eval/searchEval.js

// Example evaluation dataset
const evalDataset = [
  {
    query: "How do I retry failed webhook requests?",
    expectedDocumentIds: [1], // Shopify Webhooks Guide
  },
  {
    query: "prevent API abuse and rate limiting",
    expectedDocumentIds: [5], // API Security Guide
  },
  {
    query: "database consistency and ACID",
    expectedDocumentIds: [3], // PostgreSQL Transactions Guide
  },
];

/**
 * Precision@K: What fraction of top-K results are relevant?
 * Recall@K: What fraction of relevant docs appear in top-K results?
 */
async function evaluateSearch(searchFn, dataset, k = 5) {
  const results = [];

  for (const item of dataset) {
    const searchResults = await searchFn({ query: item.query, tenantId: 1, limit: k });
    const returnedIds = searchResults.map(r => r.document_id);

    const relevantReturned = returnedIds.filter(id => item.expectedDocumentIds.includes(id));

    const precisionAtK = relevantReturned.length / k;
    const recallAtK = relevantReturned.length / item.expectedDocumentIds.length;

    results.push({ query: item.query, precisionAtK, recallAtK });
  }

  const avgPrecision = results.reduce((s, r) => s + r.precisionAtK, 0) / results.length;
  const avgRecall = results.reduce((s, r) => s + r.recallAtK, 0) / results.length;

  return { avgPrecision, avgRecall, details: results };
}

export { evaluateSearch };

// Cache query embeddings for 1 hour
async function getCachedQueryEmbedding(query) {
  const cacheKey = `qembed:${crypto.createHash('md5').update(query).digest('hex')}`;
  const cached = await redis.get(cacheKey);
  if (cached) return JSON.parse(cached);

  const embedding = await generateEmbedding(query);
  await redis.setex(cacheKey, 3600, JSON.stringify(embedding));
  return embedding;
}

SET maintenance_work_mem = '2GB';
CREATE INDEX idx_chunk_embedding_hnsw ON document_chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

// Always derive tenantId from the authenticated session
// NEVER accept it from the request body or query parameters
const tenantId = req.session.tenantId; // from auth middleware

src/
├── api/
│   ├── searchRoutes.js       # /search and /rag endpoints
│   └── documentRoutes.js     # Document CRUD + trigger embedding jobs
├── db/
│   ├── chunkRepository.js    # Raw SQL for vector insert/delete
│   └── prisma.js             # PrismaClient singleton
├── embeddings/
│   └── embeddingService.js   # OpenAI SDK calls, batching, retries
├── documents/
│   ├── chunker.js            # Text chunking logic
│   └── documentService.js    # Business logic + content hash
├── search/
│   ├── vectorSearch.js       # pgvector cosine similarity search
│   ├── ftsSearch.js          # PostgreSQL full-text search
│   └── hybridSearch.js       # RRF fusion of both
├── rag/
│   └── ragPipeline.js        # Context construction + LLM call
├── workers/
│   └── embeddingWorker.js    # BullMQ worker
├── queue/
│   └── embeddingQueue.js     # BullMQ queue definition
├── eval/
│   └── searchEval.js         # Evaluation dataset + metrics
└── utils/
    └── contentHash.js        # SHA-256 content hashing

[ ] PostgreSQL 13+ installed
[ ] pgvector extension installed (CREATE EXTENSION vector)
[ ] documents table created with content_hash and tenant_id
[ ] document_chunks table created with vector(1536) column
[ ] Embedding model chosen and documented (text-embedding-3-small)
[ ] Chunking implemented with configurable size and overlap
[ ] Embedding generation with batching and retry logic
[ ] Vectors stored with cast (::vector) and model tracking
[ ] First semantic search working with <=> operator
[ ] HNSW index created with vector_cosine_ops
[ ] ef_search tuned and tested
[ ] Tenant filter applied in ALL search queries
[ ] Iterative index scan enabled for filtered search
[ ] Generated tsvector column with GIN index
[ ] Full-text search with plainto_tsquery working
[ ] Hybrid search with RRF fusion implemented
[ ] RAG pipeline with system prompt constraints
[ ] Source attribution in RAG response
[ ] Background embedding queue with BullMQ + Redis
[ ] Exponential backoff on embedding API failures
[ ] Content hash for change detection
[ ] Evaluation dataset created (50+ questions)
[ ] Precision@K and Recall@K measured
[ ] Query embedding caching in Redis
[ ] Prompt injection mitigations reviewed
[ ] Tenant isolation verified by code review and testing
[ ] Data handling policy reviewed with legal/compliance