XBreach-Worker-Go

A small, focused Go rewrite of the XBreach ingestion worker. It claims pending jobs from PostgreSQL, reads the uploaded leak file from shared storage, parses credential records, and writes searchable rows to ClickHouse. Job status, progress and per-line parse errors are reported back to PostgreSQL.

ingest  ──upload──▶ filesystem (raw files) + Postgres (ingestion_jobs: pending)
worker  ──claim (FOR UPDATE SKIP LOCKED)──▶ Postgres
        ──read & parse──▶ filesystem
        ──batch insert──▶ ClickHouse (leak_records)
        ──progress / status / errors──▶ Postgres

Layout

A single flat package main — intentionally simple:

File	Responsibility
config.go	settings from XBREACH_* env vars
snowflake.go	unique 64-bit ids
parse.go	line → record (the general parser)
postgres.go	claim / progress / status / errors (pgx)
clickhouse.go	migration + batch insert (clickhouse-go)
worker.go	read file, parse, batch, the poll loop
main.go	wiring + graceful shutdown

Parser

The parser is deliberately general. For each line it:

1. skips blank/banner/promo lines;
2. finds a URL anywhere in the line (an explicit scheme://host[:port][/path], or a scheme-less host.tld[:port]/path) and treats the line as ULP — the remaining fields are the identity and password;
3. otherwise treats a leading host.tld:user:pass as ULP, or falls back to identity:password (the password keeps any extra delimiters);
4. classifies the identity as email or username;
5. drops junk identities (URL session/query params containing =, form artifacts starting with () and field-label artifacts (Password/Username).

Supported delimiters: : ; | tab (and space only when the first token is an email).

Configuration

All settings come from XBREACH_* environment variables:

Variable	Default
XBREACH_POSTGRES_HOST / _PORT / _DB / _USER / _PASSWORD	postgres / 5432 / xbreach / xbreach / xbreach
XBREACH_DATA_PATH	/data/xbreach
XBREACH_CLICKHOUSE_HOST / _PORT / _DB / _USER / _PASSWORD	clickhouse / 9000 / xbreach / xbreach / xbreach
XBREACH_APP_ID / XBREACH_NODE_ID	2 / 1
XBREACH_CLICKHOUSE_BATCH_SIZE	50000
XBREACH_PROGRESS_FLUSH_EVERY	10000
XBREACH_MAX_JOB_ERRORS	1000
XBREACH_BANNER_SCAN_LINES	50
XBREACH_STORE_PLAINTEXT_PASSWORD	true
XBREACH_WORKER_POLL_INTERVAL_SECONDS	2

Build & test

go test ./...
go build .
# or a tiny static image:
docker build -t xbreach-worker-go .

It targets the same Postgres (ingestion_jobs, ingestion_job_errors) and ClickHouse (leak_records) as xbreach-ingest, so it is a drop-in replacement for the C++ worker: point it at the same services via the XBREACH_* env vars.