config

Documentation

Overview

Package config loads and validates the YAML / environment-variable configuration for the streaming EPO/HUPD processor.

Subcommands

Two streaming subcommands share one Config tree:

• process — EPO XML → Parquet
• process-hupd — HUPD .tar → disk

The legacy download / extract / parse chain has been removed; only the fields used by the streaming pipeline are recognised.

Loading Order

1. Built-in defaults (Config zero values + viper.SetDefault).
2. YAML file (path supplied to [LoadConfig], typically config/config.yaml).
3. Environment variables (uppercase, dot → underscore).

Validation

Every [LoadConfig] call runs the struct through go-playground/validator and fails fast on missing required fields. Invariants beyond struct tags (e.g. spool dir writability) are checked in cmd at start-up.

Environment Variables

All keys are bindable. Examples:

EPO_PROCESSOR_PIPELINE_CHECKPOINT_DB=/var/lib/epo/state.db
EPO_PROCESSOR_TELEMETRY_ENABLED=true

Index

• type Analyze
• type Config
  • func Load(cfgFile string, flags *pflag.FlagSet, binds map[string]string) (Config, error)
• type HUPD
• type Log
• type Pipeline
• type Server

Types

type Analyze

type Analyze struct {
	// HUPDMetaURL is the URL of the HUPD metadata Feather file to download
	// when it is not present at the path given by --hupd-meta.
	// Default: the 2022-02-22 snapshot on HuggingFace.
	HUPDMetaURL string `mapstructure:"hupd_meta_url"`
}

Analyze configures the `analyze` subcommand.

type Config

type Config struct {
	Log      Log      `mapstructure:"log"      validate:"required"`
	Server   Server   `mapstructure:"server"`
	HUPD     HUPD     `mapstructure:"hupd"`
	Analyze  Analyze  `mapstructure:"analyze"`
	Pipeline Pipeline `mapstructure:"pipeline"`
}

Config is the root configuration object. Each subsection is a value type so it can be passed around freely; viper validates on Load().

func Load

func Load(cfgFile string, flags *pflag.FlagSet, binds map[string]string) (Config, error)

Load reads config from file/env/flags/defaults and validates it.

binds maps a CLI flag name to its dotted viper config key (e.g. "archive-concurrency" -> "pipeline.archive_concurrency"). Only flags listed in binds are bound to config; this keeps short, command-local flag names decoupled from the (stable) config-file/env key schema. Either argument may be nil/empty.

type HUPD

type HUPD struct {
	// URL is the full download URL of the HUPD .tar file.
	URL string `mapstructure:"url"`
	// Filename is used as the archive name in log messages and
	// as the key in the checkpoint database.
	Filename string `mapstructure:"filename"`
}

HUPD is the HuggingFace HUPD .tar source consumed by `process-hupd`.

type Log

type Log struct {
	// LogLevel controls the minimum severity emitted.
	// Accepted values (case-insensitive): debug, info, warn, error.
	// Default: info.
	LogLevel string `mapstructure:"log_level" validate:"required,oneof=debug info warn error"`
	// LogDir is the directory where the JSON log file is written.
	// An empty string disables file logging (console logging is unaffected).
	LogDir string `mapstructure:"log_dir"`
}

Log configures the slog logger (level and output directory).

type Pipeline

type Pipeline struct {
	// ArchiveConcurrency is the number of archives fetched and walked
	// in parallel. Default: 4.
	ArchiveConcurrency int `mapstructure:"archive_concurrency"`
	// ExtractorConcurrency is the number of archive entries decoded in
	// parallel (one sequential, CPU-bound XML reader each) — the knob that
	// saturates cores. 0 (or negative) means auto: runtime.NumCPU() clamped
	// by MemoryBudgetGB / PerEntryEstimateMB. Default: 0 (auto).
	ExtractorConcurrency int `mapstructure:"extractor_concurrency"`
	// ParserConcurrency is the number of goroutines that run the CPU-bound
	// per-document parse in parallel within each entry. Since the reader is
	// sequential, a small value (just enough to overlap parse with read) is
	// sufficient. 0 (or negative) defaults to a small internal constant.
	ParserConcurrency int `mapstructure:"parser_concurrency"`
	// MemoryBudgetGB caps the RAM devoted to concurrent readers when
	// ExtractorConcurrency is auto (0). Each in-flight reader buffers a
	// decompressed XML stream plus the current DOM node. Default: 32.
	MemoryBudgetGB int `mapstructure:"memory_budget_gb"`
	// PerEntryEstimateMB is the conservative per-reader working-set estimate
	// used to clamp auto ExtractorConcurrency against MemoryBudgetGB.
	// Default: 256.
	PerEntryEstimateMB int `mapstructure:"per_entry_estimate_mb"`
	// BatchSize is the number of PatentRecords per sink Write call.
	// Default: 1000.
	BatchSize int `mapstructure:"batch_size"`
	// BatchTimeout is the maximum time a partial batch waits before
	// being flushed to the sink. Default: 2s.
	BatchTimeout time.Duration `mapstructure:"batch_timeout"`
	// OutputParquet is the destination Parquet file for `process`.
	// Default: "./data.parquet".
	OutputParquet string `mapstructure:"output_parquet"`
	// RowGroupSize overrides DefaultRowGroupSize for the Parquet writer.
	// 0 uses the library default.
	RowGroupSize int `mapstructure:"row_group_size"`
	// SpoolDir is where zip archives are spooled to disk (zip requires
	// random access). Empty defaults to the OS temp directory.
	SpoolDir string `mapstructure:"spool_dir"`
	// UseLocalDir, when non-empty, replays archives from this local
	// directory instead of fetching them over HTTP. Useful for re-runs
	// after a prior run with KeepArchive=true.
	UseLocalDir string `mapstructure:"use_local_dir"`

	// KeepArchive tees each raw HTTP body to a file under ArchiveDir
	// while the walker reads it. Orthogonal to KeepExtracted.
	// Default: false.
	KeepArchive bool `mapstructure:"keep_archive"`
	// ArchiveDir is the destination directory when KeepArchive is true.
	ArchiveDir string `mapstructure:"archive_dir"`
	// KeepExtracted tees each selected archive entry to a file under
	// ExtractedDir while the consumer reads it. Default: false.
	KeepExtracted bool `mapstructure:"keep_extracted"`
	// ExtractedDir is the destination directory when KeepExtracted is true.
	ExtractedDir string `mapstructure:"extracted_dir"`

	// CheckpointDB, when non-empty, enables resumable processing via a
	// bbolt database at this path. Completed archives are recorded so
	// subsequent runs skip them automatically.
	CheckpointDB string `mapstructure:"checkpoint_db"`
	// ResetCheckpoint, when true, deletes CheckpointDB and any existing
	// OutputParquet shards at startup so the next run starts from scratch.
	ResetCheckpoint bool `mapstructure:"reset_checkpoint"`
}

Pipeline configures the streaming pipeline shared by both subcommands.

type Server

type Server struct {
	// BaseURL is the root of the EPO BDDS REST API, e.g.
	// "https://ops.epo.org/3.2/rest-services".
	BaseURL string `mapstructure:"base_url" validate:"omitempty,url"`
	// Timeout is the per-request HTTP timeout. Default: 30s.
	Timeout time.Duration `mapstructure:"timeout" validate:"omitempty,gt=0"`
	// MaxRetries is the maximum number of HTTP retries per archive.
	// Range: 0–10. Default: 3.
	MaxRetries int `mapstructure:"max_retries" validate:"min=0,max=10"`
	// ProductID selects the EPO bulk-data product to process.
	// Default: 3 (EP front-text exchange data).
	ProductID int `mapstructure:"product_id"`
	// VerifySHA1 enables SHA-1 checksum verification for every download.
	// Mismatches trigger a retry. Default: false.
	VerifySHA1 bool `mapstructure:"verify_sha1"`
}

Server points the EPO product source at the live BDDS API. Only `process` reads it; `process-hupd` is unaffected, hence optional.