Document ingestion systems face a recurring challenge: content lives in many places—local filesystems, web pages, Git repositories, S3 buckets—and arrives in many formats (PDF, HTML, Markdown, plain text, DOCX). A well-designed processor abstraction unifies access to these diverse sources behind a common streaming interface, routes each item through format-specific parsers via a registry, and produces normalized output suitable for indexing, search, or further analysis.
A well-implemented document processor follows this pattern. It provides a single Process() entry point that accepts a source configuration (filesystem path, URL list, Git remote, or S3 prefix), streams items from that source, identifies each item’s format, runs it through the appropriate parser, and emits processed items via a streaming callback. An WithIDTransform option rewrites document IDs during processing, enabling cross-site deduplication when the same content appears at different URLs or file paths across multiple sources.
This architecture decouples source access from format parsing and output handling, making it straightforward to add new sources (e.g., a Confluence API source) or new format parsers (e.g., an EPUB parser) without modifying the core processing loop. The streaming design means memory usage stays constant regardless of corpus size—critical when processing millions of documents.
Each content source implements a common interface that produces a stream of items. An item contains metadata (ID, path, content type, last modified timestamp) and a reader for the raw bytes. The source abstraction handles connection management, pagination, rate limiting, and error recovery internally, presenting a simple iterator to the processing loop.
The processor supports four built-in source types: Filesystem (recursive directory traversal with glob filtering), Web (HTTP fetching with configurable concurrency and politeness delays), Git (clone or fetch a repository and walk its tree), and S3 (list and fetch objects from an S3 bucket/prefix). Each source translates its native listing/fetching API into the common item stream interface.
The format registry maps MIME types and file extensions to parser implementations. When an item arrives, the processor determines its format through a priority chain: explicit content-type header (for web sources), file extension mapping, and finally magic-byte sniffing of the content. The registry returns a parser that converts the raw bytes into structured output—extracted text, metadata fields, and optionally embedded resources (images, links).
Every format parser implements a Parse(ctx context.Context, r io.Reader, meta ItemMeta) (*Document, error) method. The Document output contains the extracted text body, a title, structured metadata (author, date, language), and a list of embedded resources. This uniform output makes downstream indexing code format-agnostic—it processes Document objects regardless of whether they originated from PDF, HTML, or Markdown.
Rather than collecting all processed documents into a slice (which would require unbounded memory), the processor accepts a callback function that is invoked for each successfully processed item. The callback receives the processed document and can write it to an index, store it in a database, or forward it to another pipeline stage. If the callback returns an error, the processor stops and propagates the error to the caller.
When ingesting the same content from multiple sources (e.g., a document mirrored on two websites), each source assigns a different ID (typically the URL or file path). The WithIDTransform option applies a transformation function to document IDs before they reach the callback, allowing normalization (strip URL scheme, remove trailing slashes) or domain-scoping (prefix with a canonical site identifier). This enables the downstream deduplication layer to recognize identical content regardless of its source location.
The processor uses a bounded worker pool to process items concurrently. The source streams items into a channel, and N worker goroutines consume from the channel, parse each item, and invoke the callback. A semaphore controls the maximum number of concurrent HTTP requests (for web sources) or S3 API calls to respect rate limits and avoid overwhelming external services.
Processor with source configuration and optionsWithIDTransform is set, the document ID is rewritten before callback invocationpackage docs
import (
"context"
"fmt"
"io"
"sync"
)
// Source streams raw items from an external location.
type Source interface {
// Items returns a channel of raw items. Closed when exhausted.
Items(ctx context.Context) (<-chan RawItem, error)
}
// Parser converts raw bytes into a structured Document.
type Parser interface {
Parse(ctx context.Context, r io.Reader, meta ItemMeta) (*Document, error)
}
// Callback is invoked for each successfully processed document.
type Callback func(ctx context.Context, doc *Document) error
// Option configures the Processor.
type Option func(*Processor)
// WithIDTransform rewrites document IDs before invoking the callback.
func WithIDTransform(fn func(string) string) Option {
return func(p *Processor) { p.idTransform = fn }
}
// WithWorkers sets the number of concurrent processing goroutines.
func WithWorkers(n int) Option {
return func(p *Processor) { p.workers = n }
}
type Processor struct {
source Source
registry *FormatRegistry
callback Callback
idTransform func(string) string
workers int
}
func New(src Source, reg *FormatRegistry, cb Callback, opts ...Option) *Processor {
p := &Processor{
source: src,
registry: reg,
callback: cb,
workers: 4,
}
for _, opt := range opts {
opt(p)
}
return p
}
func (p *Processor) Run(ctx context.Context) error {
items, err := p.source.Items(ctx)
if err != nil {
return fmt.Errorf("source init: %w", err)
}
var (
wg sync.WaitGroup
once sync.Once
pErr error
)
for i := 0; i < p.workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for item := range items {
if err := p.processItem(ctx, item); err != nil {
once.Do(func() { pErr = err })
return
}
}
}()
}
wg.Wait()
return pErr
}
func (p *Processor) processItem(ctx context.Context, item RawItem) error {
parser, err := p.registry.ParserFor(item.ContentType, item.Path)
if err != nil {
return fmt.Errorf("no parser for %s: %w", item.Path, err)
}
doc, err := parser.Parse(ctx, item.Body, item.Meta)
if err != nil {
return fmt.Errorf("parse %s: %w", item.ID, err)
}
if p.idTransform != nil {
doc.ID = p.idTransform(doc.ID)
}
return p.callback(ctx, doc)
}package docs
import (
"net/url"
"strings"
)
// NormalizeURL strips scheme, trailing slashes, and query parameters
// to produce a canonical ID for deduplication.
func NormalizeURL(rawID string) string {
u, err := url.Parse(rawID)
if err != nil {
return rawID
}
// Strip scheme
canonical := u.Host + u.Path
// Remove trailing slash
canonical = strings.TrimRight(canonical, "/")
// Lowercase for case-insensitive matching
return strings.ToLower(canonical)
}
// ScopedTransform prefixes the normalized ID with a site identifier,
// enabling per-site deduplication scopes.
func ScopedTransform(siteID string) func(string) string {
return func(rawID string) string {
return siteID + "::" + NormalizeURL(rawID)
}
}
// Usage:
// processor := docs.New(webSource, registry, indexCallback,
// docs.WithIDTransform(ScopedTransform("docs-main")),
// docs.WithWorkers(8),
// )A common use of such a processor is building search indexes. The processor ingests content from multiple sources (documentation sites, Git repositories, internal wikis), extracts text and metadata, and feeds the results into the indexing pipeline. The streaming callback writes directly to the index writer, avoiding the need to buffer the entire corpus in memory.
When migrating content between systems (e.g., from a file-based CMS to a database-backed one), the processor serves as the extraction layer. Configure a filesystem source pointing at the old content directory, register parsers for each format in use, and use the callback to write normalized documents to the new system. The ID transform rewrites legacy paths to the new URL structure.
Documentation often appears at multiple URLs (versioned docs, regional mirrors, CDN caches). By processing each source with a scoped ID transform, the downstream deduplication layer can identify when the same document appears at docs.example.com/guide and mirror.example.com/guide, keeping only the canonical version while recording the alternate URLs as aliases.
The format registry uses a two-level lookup: first by MIME type (exact match), then by file extension (prefix match). This allows registering a single PDF parser that handles both application/pdf MIME types and .pdf extensions. The registry is populated at initialization and is read-only during processing, eliminating the need for synchronization.
package docs
import (
"path/filepath"
"strings"
)
type FormatRegistry struct {
byMIME map[string]Parser
byExt map[string]Parser
}
func NewRegistry() *FormatRegistry {
return &FormatRegistry{
byMIME: make(map[string]Parser),
byExt: make(map[string]Parser),
}
}
func (r *FormatRegistry) Register(mime string, ext string, p Parser) {
if mime != "" {
r.byMIME[strings.ToLower(mime)] = p
}
if ext != "" {
r.byExt[strings.ToLower(ext)] = p
}
}
func (r *FormatRegistry) ParserFor(mime, path string) (Parser, error) {
if mime != "" {
if p, ok := r.byMIME[strings.ToLower(mime)]; ok {
return p, nil
}
}
ext := strings.ToLower(filepath.Ext(path))
if p, ok := r.byExt[ext]; ok {
return p, nil
}
return nil, fmt.Errorf("no parser registered for mime=%q ext=%q", mime, ext)
}In batch processing, individual item failures should not abort the entire run. The processor supports an OnError callback that receives parse errors and decides whether to skip the item (return nil) or abort (return the error). This allows logging failures and continuing, or halting on the first error depending on the use case.
The item channel between the source and workers is bounded (typically 2× the worker count). If workers are slow (e.g., parsing large PDFs), the channel fills up and the source blocks, naturally throttling the rate at which new items are fetched. This prevents the source from fetching faster than the workers can process, keeping memory usage predictable.
ctx.Done() periodically. Without context propagation, canceling the processor leaves orphaned goroutines that continue consuming resources until they finish or the process exits./docs?lang=en and /docs?lang=fr collapse to /docs. Test ID transforms against real data and preserve distinguishing components.RawItem.Body is an io.ReadCloser. Failing to close it after parsing leaks file descriptors (filesystem source) or HTTP connections (web source). Use defer item.Body.Close() at the start of processItem.io.TeeReader or buffer the sniffed bytes and prepend them with io.MultiReader before passing to the parser.