Turn Documents into Searchable Knowledge with NestJS and pgvector
Part 2 of 4: Build the evidence pipeline behind a document AI application.
A chat interface can make an answer look convincing. The harder question is whether the application found the right document in the first place.
In Part 1, I introduced the roles of RAG, SSE, and MCP. Here we will follow the evidence pipeline in the document-parsing-demo repository: upload a file, process it in the background, preserve its structure, generate embeddings, and search current content.
We will use fictional policy documents. One version says a device request takes ten school days; its replacement says five. That gives us a concrete question to test: can retrieval return the current answer while excluding superseded content?
What we are building
Upload → stored file + version + job record
→ BullMQ → worker
→ parser → canonical blocks → chunks
→ embeddings → PostgreSQL / pgvectorQuestion → query embedding + text search
→ eligible chunks → ranked evidenceThe chat model is downstream of this pipeline. We can inspect retrieval results directly, before asking an LLM to generate an answer.
1. Start the local services
You need Node.js 24+, pnpm 10, and Docker Desktop. From a terminal:
git clone https://github.com/sandylib/document-parsing-demo.git
cd document-parsing-demo
pnpm setupThe setup script installs dependencies, starts PostgreSQL, Redis, and Ollama, pulls the configured demo models, and runs database migrations. The initial model downloads can take time.
Then start each process in its own terminal, from the repository root:
# Terminal 1
pnpm dev:api
# Terminal 2
pnpm worker
# Terminal 3
pnpm dev:webCheck http://localhost:3001/health before uploading. The database, queue, chat, and embedding checks should be healthy. The interface is at http://localhost:3000.
Use a fresh local database for the walkthrough. These commands are for a local demo; the application’s role headers are not production authentication.
2. Register an upload without waiting for embeddings
The POST /documents controller accepts multipart form data and delegates to UploadDocumentUseCase. Its receipt contains documentId, versionId, jobId, and a duplicate flag.
The use case rejects missing or empty files, files over the 10 MB demo limit, and unsupported MIME types. It accepts PDF, Markdown, and plain text. This is basic demo validation, not complete content inspection or malware scanning.
The following excerpt shows how file identity and storage are established:
const checksum = createHash('sha256')
.update(input.file.buffer)
.digest('hex');
const storageKey = `${randomUUID()}${extname(input.file.originalname).toLowerCase()}`;The actual storage key combines a random UUID with the lowercased file extension. The original name remains metadata. This keeps a user-supplied filename from becoming the storage path.
The bytes go to the local storage adapter. PostgreSQL then registers the document version and ingestion job in a transaction. If registration fails, the newly stored file is deleted. If an identical checksum already exists, the current implementation returns the existing receipt and removes the redundant stored copy.
That duplicate check is checksum-based across existing versions; it is not a general identity or tenant-isolation design. Keep it within this local demonstration’s assumptions.
The queue receives identifiers:
await this.queue.enqueue({
jobId: receipt.jobId,
versionId: receipt.versionId,
});It does not receive file bytes or extracted document text. The worker loads the metadata and content using those identifiers.
3. Let the worker coordinate ingestion
BullMQ is configured for three total attempts with exponential backoff starting at one second. The demo worker runs with concurrency one to keep local model usage predictable.
The processor records the attempt, runs the application use case, and updates progress in both BullMQ and PostgreSQL. When processing fails, it records the failure and rethrows the error so BullMQ can apply its retry policy. The database records a terminal failure when attempts are exhausted.
Inside ProcessDocumentUseCase, the core transformation is compact:
const parsed = await parser.parse({
content,
mimeType: version.mimeType,
originalName: version.originalName,
});if (!parsed.blocks.length) {
throw new Error('Parser returned no readable blocks');
}const chunks = this.chunker.chunk(parsed.blocks);
const embeddings = await this.embedding.embed(
chunks.map((chunk) => chunk.text),
);
The reported progress values — 10, 25, 50, 85, and 100 — represent pipeline milestones. They are not a measurement of bytes processed or remaining execution time.
4. Preserve structure before creating chunks
The parsers return canonical blocks rather than one large string. Each block has a type, order, text, section path, and optional location metadata. Blocks can therefore retain a connection to the original document.
Try this Markdown shape:
# Student Device Policy
## Requests
A device request takes five school days.For this simple input, the text parser recognizes two headings and a paragraph. The paragraph carries the section path [“Student Device Policy”, “Requests”]. Blank lines are significant here: this lightweight parser splits input into blank-line-separated pieces and recognizes headings within those pieces.
This is a deliberately small Markdown parser. It is not a complete Markdown syntax engine. The PDF adapter similarly uses deterministic extraction and lightweight heuristics; scanned pages and complex layouts need specialist processing.
Get Sandy Zhang’s stories in your inbox
Join Medium for free to get updates from this writer.
Chunks serve a different purpose from blocks. Blocks preserve extracted structure; chunks group that content into retrieval-sized passages.
ChunkerService starts a new chunk when it encounters another heading or when adding a block would exceed the default target of 1,200 characters. Each chunk retains the indexes of its constituent blocks, its section path, and available page provenance.
There are two limits to understand: the target is character-based, not token-based, and an individual oversized block is not split. Calling it a strict 1,200-character cap would be inaccurate.
5. Generate embeddings and publish the version
The worker passes chunk texts to the EmbeddingModel port. Ollama is the default adapter, configured with qwen3-embedding:0.6b and 1,024 dimensions in the demo.
The application also has an OpenAI embedding adapter. Changing providers is a configuration choice, but existing vectors must be regenerated when the embedding model changes. Matching dimensions alone does not make vectors from different models comparable.
After embedding succeeds, completeProcessing runs a database transaction that:
- Inserts canonical blocks and chunks, including vectors and source references.
- Retires the previous current version for that document.
- Marks the processed version current and ready.
- Marks the ingestion job completed.
Retrieval therefore does not see half of a successfully committed replacement. The old version remains in the database as historical content, while the current flag controls eligibility.
This transaction covers database persistence and version promotion. It does not make the earlier PostgreSQL registration and Redis enqueue a single atomic operation. The demo records an enqueue failure when it catches one; it does not implement a transactional outbox that closes every crash window.
6. Search with both meaning and words
SearchKnowledgeUseCase trims the question, generates its embedding, and delegates to the retrieval repository. The repository first constructs an eligible set:
WHERE v.is_current
AND v.status = 'ready'
AND (c.visibility = 'public' OR $3 = 'staff')This SQL excerpt is the core policy boundary for the demo search. Eligibility is applied before candidate ranking. A later prompt telling the model to ignore restricted or old content would not provide the same boundary.
The current query then takes up to 20 vector candidates and up to 20 lexical candidates. Vector ranking uses cosine distance through pgvector’s <=> operator. Lexical ranking uses PostgreSQL full-text search.
It combines candidate ranks using reciprocal rank fusion: a semantic contribution of 1 / (60 + rank), and a lexical contribution of 2 / (60 + rank). The lexical list therefore has twice the weight in this prototype.
Finally, a result must either have cosine similarity of at least 0.6 or qualify as a lexical match. The lexical qualification requires at least two distinct normalized query terms, or all terms when there are fewer than two. These are implementation choices for this small corpus, not universal retrieval settings.
The returned evidence includes its ID, document and version IDs, title, passage text, section path, page when available, and a score. Its displayed score is not the fusion ranking value or a calibrated probability of correctness.
Read persistence, version promotion, and hybrid search.
7. Exercise retrieval without the chat model
With the services running, seed the fictional documents once:
pnpm seedThe script uploads the older policy, waits for readiness, submits its replacement against the same document ID, and queues the staff notice. Wait until the replacement and staff notice have finished processing before running the checks.
Query the search endpoint directly:
curl -G http://localhost:3001/documents/search --data-urlencode 'query=How long does a device request take?' -H 'x-demo-role: public'Inspect the returned evidence. The expected passage contains “five school days”; the superseded “ten school days” wording should not be returned.
Then ask a public payroll question:
curl -G http://localhost:3001/documents/search --data-urlencode 'query=What is the payroll cut-off?' -H 'x-demo-role: public'The expected result is an empty array. Repeat with x-demo-role: staff to retrieve the restricted notice. This header selects the demo role; a production service must derive authorization from verified identity.
Run the existing retrieval checks:
pnpm evaluateThe script checks three fixture questions for expected text or an empty result. It does not measure Recall@k, citation correctness, or generated-answer faithfulness. These commands describe the intended local exercise, not a production benchmark.
What to check when a result is missing
- The document stays queued: confirm Redis is healthy and the separate worker process is running.
- Processing fails: inspect worker errors and embedding availability. Empty extraction is an explicit failure.
- Evidence is missing: confirm the version is ready and current, and use the expected demo role.
- Results change after switching models: re-index the corpus using the selected embedding model.
As a small exercise, upload a changed policy through the UI’s Version target using the existing document ID. Wait for it to become ready, then repeat the retrieval request. Verify the returned versionId and passage rather than relying on the chat model’s wording.
Next: stream an answer from that evidence
We now have an inspectable evidence pipeline: uploaded files become structured, versioned passages, and retrieval selects eligible content before generation starts.
Part 3 will follow those passages into the answer use case and back to Next.js through SSE. We will inspect text deltas, citations, refusal, cancellation, and the boundary between provider streaming and application events.
Series: 1. Concepts and architecture · 2. Ingestion and retrieval · 3. Streaming chat with SSE (upcoming) · 4. MCP tools (upcoming).
