From 415bf0cce18dfc4363b4f1795fbe3abc575010cf Mon Sep 17 00:00:00 2001
From: unkin-agent <unkin-agent@unkin.net>
Date: Sun, 23 Aug 2026 16:43:05 +1000
Subject: [PATCH] Add chlog CLI with chcat/chtail/chgrep entrypoints

Single Go binary for the ClickHouse log store (logs.raw): chlog with
cat/tail/grep subcommands, plus chcat/chtail/chgrep argv[0]-dispatched
symlink entrypoints. Every query is time-bounded and fully parameterized;
chgrep guards wide unfiltered scans. Ships nfpm RPM with completions and
woodpecker PR/tag pipelines mirroring node-lookup.
---
 .gitignore                    |   8 +
 .pre-commit-config.yaml       |  17 +++
 .woodpecker/build.yaml        |  18 +++
 .woodpecker/pre-commit.yaml   |  18 +++
 .woodpecker/release.yaml      | 142 +++++++++++++++++
 .woodpecker/test.yaml         |  33 ++++
 Makefile                      |  67 ++++++++
 README.md                     |  64 +++++++-
 go.mod                        |  10 ++
 go.sum                        |  11 ++
 internal/chlog/client.go      | 109 +++++++++++++
 internal/chlog/client_test.go | 279 ++++++++++++++++++++++++++++++++++
 internal/chlog/config.go      |  29 ++++
 internal/chlog/format.go      | 149 ++++++++++++++++++
 internal/chlog/format_test.go | 150 ++++++++++++++++++
 internal/chlog/pager.go       |  78 ++++++++++
 internal/chlog/query.go       | 131 ++++++++++++++++
 internal/chlog/query_test.go  | 204 +++++++++++++++++++++++++
 internal/chlog/tail.go        |  84 ++++++++++
 internal/chlog/time.go        |  38 +++++
 internal/chlog/time_test.go   |  51 +++++++
 main.go                       | 249 ++++++++++++++++++++++++++++++
 main_test.go                  |  96 ++++++++++++
 packaging/nfpm.yaml           |  86 +++++++++++
 scripts/build-rpm.sh          |  51 +++++++
 25 files changed, 2171 insertions(+), 1 deletion(-)
 create mode 100644 .gitignore
 create mode 100644 .pre-commit-config.yaml
 create mode 100644 .woodpecker/build.yaml
 create mode 100644 .woodpecker/pre-commit.yaml
 create mode 100644 .woodpecker/release.yaml
 create mode 100644 .woodpecker/test.yaml
 create mode 100644 Makefile
 create mode 100644 go.mod
 create mode 100644 go.sum
 create mode 100644 internal/chlog/client.go
 create mode 100644 internal/chlog/client_test.go
 create mode 100644 internal/chlog/config.go
 create mode 100644 internal/chlog/format.go
 create mode 100644 internal/chlog/format_test.go
 create mode 100644 internal/chlog/pager.go
 create mode 100644 internal/chlog/query.go
 create mode 100644 internal/chlog/query_test.go
 create mode 100644 internal/chlog/tail.go
 create mode 100644 internal/chlog/time.go
 create mode 100644 internal/chlog/time_test.go
 create mode 100644 main.go
 create mode 100644 main_test.go
 create mode 100644 packaging/nfpm.yaml
 create mode 100644 scripts/build-rpm.sh

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..48c6e5b
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,8 @@
+# built binaries (repo root only)
+/chlog
+/chcat
+/chtail
+/chgrep
+# cross-compiled release artifacts (e.g. chlog-linux-amd64)
+/chlog-*
+dist/
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
new file mode 100644
index 0000000..2e63b82
--- /dev/null
+++ b/.pre-commit-config.yaml
@@ -0,0 +1,17 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.6.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-merge-conflict
+      - id: mixed-line-ending
+        args: [--fix=lf]
+
+  - repo: https://github.com/dnephin/pre-commit-golang
+    rev: v0.5.1
+    hooks:
+      - id: go-fmt
+      - id: go-vet
+      - id: go-unit-tests
diff --git a/.woodpecker/build.yaml b/.woodpecker/build.yaml
new file mode 100644
index 0000000..74123b5
--- /dev/null
+++ b/.woodpecker/build.yaml
@@ -0,0 +1,18 @@
+when:
+  - event: pull_request
+
+steps:
+  - name: build
+    image: golang:1.25
+    commands:
+      - make build
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
diff --git a/.woodpecker/pre-commit.yaml b/.woodpecker/pre-commit.yaml
new file mode 100644
index 0000000..d57b508
--- /dev/null
+++ b/.woodpecker/pre-commit.yaml
@@ -0,0 +1,18 @@
+when:
+  - event: pull_request
+
+steps:
+  - name: pre-commit
+    image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
+    commands:
+      - uvx pre-commit run --all-files
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
diff --git a/.woodpecker/release.yaml b/.woodpecker/release.yaml
new file mode 100644
index 0000000..729b942
--- /dev/null
+++ b/.woodpecker/release.yaml
@@ -0,0 +1,142 @@
+when:
+  - event: tag
+
+steps:
+  - name: test
+    image: golang:1.25
+    commands:
+      - go test -race ./...
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
+
+  # Build the single chlog binary into dist/ (consumed by the RPM step) plus
+  # cross-platform binaries attached to the Gitea release.
+  - name: build
+    image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
+    commands:
+      - make build VERSION=${CI_COMMIT_TAG}
+      # $$ escapes shell vars so Woodpecker leaves them for the shell;
+      # ${CI_COMMIT_TAG} is a real Woodpecker var and stays single-$.
+      - |
+        for osarch in linux/amd64 linux/arm64 darwin/amd64 darwin/arm64; do
+          os="$${osarch%/*}"; arch="$${osarch#*/}"
+          GOOS="$$os" GOARCH="$$arch" \
+            go build -ldflags="-s -w -X main.version=${CI_COMMIT_TAG}" \
+            -o "chlog-$${os}-$${arch}" .
+        done
+    depends_on: [test]
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
+
+  # Package the built binary + symlinks + shell completions into an RPM.
+  - name: package
+    image: git.unkin.net/unkin/almalinux9-rpmbuilder:latest
+    commands:
+      - ./scripts/build-rpm.sh ${CI_COMMIT_TAG}
+    depends_on: [build]
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
+
+  # Publish the RPM to the artifactapi local rpm repo (a real yum repo;
+  # repodata regenerates automatically).
+  - name: upload-rpm
+    image: git.unkin.net/unkin/almalinux9-base:20260606
+    commands:
+      - |
+        HOST="https://artifactapi.k8s.syd1.au.unkin.net"
+        REPO="rpm-internal"
+        for rpm in dist/*.rpm; do
+          FILE=$$(basename "$$rpm")
+          # artifactapi has no HEAD route (returns 405); probe with GET against
+          # the served path (RPMs are stored under Packages/) to avoid re-upload.
+          code=$$(curl -s -o /dev/null -w '%{http_code}' "$$HOST/api/v2/remotes/$$REPO/files/Packages/$$FILE" || true)
+          if [ "$$code" = "200" ]; then
+            echo "$$FILE already exists in $$REPO (HTTP $$code); skipping upload"
+            continue
+          fi
+          echo "Uploading $$FILE to $$REPO (existence probe returned $$code)"
+          curl -f -X PUT \
+            "$$HOST/api/v2/remotes/$$REPO/files/$$FILE" \
+            -H "Content-Type: application/x-rpm" \
+            --data-binary @"$$rpm"
+        done
+    depends_on: [package]
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 128Mi
+            cpu: 100m
+          limits:
+            memory: 512Mi
+            cpu: 500m
+
+  # Cut a Gitea release with the cross-platform binaries attached.
+  - name: release
+    image: git.unkin.net/unkin/almalinux9-base:20260606
+    environment:
+      RELEASER_TOKEN:
+        from_secret: RELEASER_TOKEN
+    commands:
+      - |
+        curl --output /usr/local/bin/tea https://artifactapi.k8s.syd1.au.unkin.net/api/v1/remote/gitea-dl/tea/0.12.0/tea-0.12.0-linux-amd64 && chmod +x /usr/local/bin/tea
+        tea logins add --name gitea --url https://git.unkin.net --token "$${RELEASER_TOKEN}" --no-version-check
+        # Several tags can point at the same commit, so skip tags on the current
+        # commit and pick the newest semver tag that is a real ancestor.
+        CUR_SHA=$$(git rev-list -n1 "${CI_COMMIT_TAG}")
+        PREV_TAG=""
+        for t in $$(git tag --sort=-v:refname); do
+          [ "$$t" = "${CI_COMMIT_TAG}" ] && continue
+          [ "$$(git rev-list -n1 "$$t")" = "$$CUR_SHA" ] && continue
+          if git merge-base --is-ancestor "$$t" "${CI_COMMIT_TAG}" 2>/dev/null; then
+            PREV_TAG="$$t"; break
+          fi
+        done
+        if [ -n "$$PREV_TAG" ]; then
+          NOTES=$$(git log "$${PREV_TAG}..${CI_COMMIT_TAG}" --pretty=format:"- %s")
+        else
+          NOTES=$$(git log --pretty=format:"- %s")
+        fi
+        tea releases create --tag "${CI_COMMIT_TAG}" --title "${CI_COMMIT_TAG}" --note "$${NOTES}" --login gitea --repo "${CI_REPO}"
+        RPM=$$(ls dist/*.rpm 2>/dev/null | head -1)
+        ASSETS="chlog-linux-amd64 chlog-linux-arm64 chlog-darwin-amd64 chlog-darwin-arm64"
+        [ -n "$$RPM" ] && ASSETS="$$ASSETS $$RPM"
+        sha256sum $$ASSETS > sha256sums.txt
+        tea releases assets create "${CI_COMMIT_TAG}" $$ASSETS sha256sums.txt \
+          --login gitea --repo "${CI_REPO}"
+    depends_on: [upload-rpm]
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 128Mi
+            cpu: 100m
+          limits:
+            memory: 512Mi
+            cpu: 500m
diff --git a/.woodpecker/test.yaml b/.woodpecker/test.yaml
new file mode 100644
index 0000000..5e179a7
--- /dev/null
+++ b/.woodpecker/test.yaml
@@ -0,0 +1,33 @@
+when:
+  - event: pull_request
+
+steps:
+  - name: lint
+    image: golangci/golangci-lint:latest
+    commands:
+      - golangci-lint run ./...
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
+
+  - name: test
+    image: golang:1.25
+    commands:
+      - go test -v -race ./...
+    backend_options:
+      kubernetes:
+        serviceAccountName: default
+        resources:
+          requests:
+            memory: 512Mi
+            cpu: 1
+          limits:
+            memory: 2Gi
+            cpu: 2
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..50ff968
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,67 @@
+BINARY  := chlog
+# chcat/chtail/chgrep are argv[0]-dispatched symlinks to the single chlog binary.
+LINKS   := chcat chtail chgrep
+DIST    := dist
+VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo dev)
+GOFLAGS := -ldflags="-s -w -X main.version=$(VERSION)"
+OS      ?= $(shell go env GOOS)
+ARCH    ?= $(shell go env GOARCH)
+
+.PHONY: all build test lint fmt clean install completions rpm rpm-package patch minor major _tag
+
+all: build
+
+build:
+	CGO_ENABLED=0 GOOS=$(OS) GOARCH=$(ARCH) go build $(GOFLAGS) -o $(DIST)/$(BINARY) .
+	@for l in $(LINKS); do ln -sf $(BINARY) $(DIST)/$$l; done
+
+test:
+	go test -v -race ./...
+
+lint:
+	golangci-lint run ./...
+
+fmt:
+	gofmt -w .
+
+clean:
+	rm -rf $(DIST)
+
+install:
+	go install $(GOFLAGS) .
+
+# Generate bash/zsh/fish completions for chlog and each symlink entrypoint.
+completions: build
+	@mkdir -p $(DIST)/completions
+	@for b in $(BINARY) $(LINKS); do \
+		$(DIST)/$$b completion bash > $(DIST)/completions/$$b.bash; \
+		$(DIST)/$$b completion zsh  > $(DIST)/completions/_$$b; \
+		$(DIST)/$$b completion fish > $(DIST)/completions/$$b.fish; \
+	done
+
+rpm: build rpm-package
+
+rpm-package:
+	./scripts/build-rpm.sh $(VERSION)
+
+# Bump helpers — reads the latest semver tag and creates the next one.
+_LATEST := $(shell git tag --sort=-v:refname | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$$' | head -1)
+_BASE   := $(if $(_LATEST),$(_LATEST),v0.0.0)
+_MAJ    := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f1)
+_MIN    := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f2)
+_PAT    := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f3)
+
+patch:
+	@NEW=v$(_MAJ).$(_MIN).$(shell expr $(_PAT) + 1); \
+	git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
+
+minor:
+	@NEW=v$(_MAJ).$(shell expr $(_MIN) + 1).0; \
+	git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
+
+major:
+	@NEW=v$(shell expr $(_MAJ) + 1).0.0; \
+	git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
+
+_tag:
+	git push origin $(TAG)
diff --git a/README.md b/README.md
index f0c1b2b..a4e163d 100644
--- a/README.md
+++ b/README.md
@@ -1,3 +1,65 @@
 # clickhouse-tools
 
-CLI tools (chcat, chtail, chgrep) for searching, filtering and tailing logs in the ClickHouse log store
\ No newline at end of file
+CLI tools for the ClickHouse log store (`logs.raw`): one binary, `chlog`, with
+three entrypoints installed as symlinks:
+
+| Command  | Also as       | Does |
+|----------|---------------|------|
+| `chcat`  | `chlog cat`   | Print logs oldest-first over a bounded time range |
+| `chtail` | `chlog tail`  | Follow logs live (2s poll, overlap + dedupe so nothing is lost or repeated) |
+| `chgrep` | `chlog grep`  | Search log messages (substring, `-i`, `--regex`) |
+
+## Why time bounds everywhere
+
+`logs.raw` has no text index and holds ~281M rows/day (3-day TTL). An unbounded
+message scan takes ~1 minute and the server kills queries at 120s. Every query
+these tools issue is therefore time-bounded — the default range is the last
+hour (`--since 1h`) — and `chgrep` refuses a search wider than 6h with no
+`--namespace`/`--host`/`--app` filter unless you pass `--force`.
+
+All user input travels as ClickHouse HTTP `{name:Type}` parameters; nothing is
+ever interpolated into SQL text.
+
+## Usage
+
+```sh
+chcat -n logging --since 30m
+chcat --host web01 --since 2h --until 1h --format logfmt
+chtail -n media --app jellyfin
+chgrep -n kube-system -i "connection refused" --since 4h
+chgrep --app vector --regex 'timed? ?out' --since 1d
+chgrep --fields req_id=42 -n api "payment"
+```
+
+### Common flags
+
+- `--since` / `--until` — duration ago (`15m`, `1h`, `2d`, `1w`) or RFC3339;
+  default `--since 1h`, `--until` now
+- `-n/--namespace`, `--host`, `--pod`, `--container`, `--app` (labels['app']),
+  `--severity` (case-insensitive), `--stream`, `--source`
+- `--limit` — max rows (default 10000 for cat/grep; tail is unlimited)
+- `--format text|json|logfmt` — text is `ts ns/pod msg` (host for vm rows),
+  colored only on a TTY (`NO_COLOR` respected)
+
+### chgrep extras
+
+- pattern is a substring by default; `-i` case-insensitive; `--regex` RE2 (`match()`)
+- `--fields key=value` (repeatable) filters the structured `fields` map
+- `--force` overrides the wide-unfiltered-search guard
+
+## Connection
+
+| Env | Default |
+|-----|---------|
+| `CH_URL` | `http://clickhouse-logs.logging.svc.cluster.local:8123` |
+| `CH_USER` | `logreader` |
+| `CH_PASSWORD` | (empty) |
+
+## Build and release
+
+```sh
+make build        # dist/chlog + symlinks
+make test         # go test -race ./...
+make rpm          # nfpm RPM with binary, symlinks, bash/zsh/fish completions
+make patch        # tag + push next vX.Y.Z → CI releases RPM to artifactapi rpm-internal
+```
diff --git a/go.mod b/go.mod
new file mode 100644
index 0000000..b88c48f
--- /dev/null
+++ b/go.mod
@@ -0,0 +1,10 @@
+module git.unkin.net/unkin/clickhouse-tools
+
+go 1.25
+
+require github.com/spf13/cobra v1.10.2
+
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.10 // indirect
+)
diff --git a/go.sum b/go.sum
new file mode 100644
index 0000000..ef5d78d
--- /dev/null
+++ b/go.sum
@@ -0,0 +1,11 @@
+github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
+github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU=
+github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4=
+github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
+github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/internal/chlog/client.go b/internal/chlog/client.go
new file mode 100644
index 0000000..2451511
--- /dev/null
+++ b/internal/chlog/client.go
@@ -0,0 +1,109 @@
+package chlog
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"hash/fnv"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const chTimeLayout = "2006-01-02 15:04:05.999"
+
+type Row struct {
+	Timestamp string            `json:"timestamp"`
+	Host      string            `json:"host"`
+	Source    string            `json:"source"`
+	Namespace string            `json:"namespace"`
+	Pod       string            `json:"pod"`
+	Container string            `json:"container"`
+	Stream    string            `json:"stream"`
+	Severity  string            `json:"severity"`
+	Message   string            `json:"message"`
+	Labels    map[string]string `json:"labels"`
+	Fields    map[string]string `json:"fields"`
+}
+
+func (r Row) Time() time.Time {
+	t, err := time.Parse(chTimeLayout, r.Timestamp)
+	if err != nil {
+		return time.Time{}
+	}
+	return t.UTC()
+}
+
+// Key identifies a row for overlap dedupe during paging and tailing.
+func (r Row) Key() uint64 {
+	h := fnv.New64a()
+	for _, s := range []string{r.Timestamp, r.Host, r.Source, r.Namespace, r.Pod, r.Container, r.Stream, r.Message} {
+		io.WriteString(h, s)
+		h.Write([]byte{0})
+	}
+	return h.Sum64()
+}
+
+type Client struct {
+	cfg  Config
+	http *http.Client
+}
+
+func NewClient(cfg Config) *Client {
+	return &Client{cfg: cfg, http: &http.Client{Timeout: 130 * time.Second}}
+}
+
+// Run executes the query and streams each result row to fn. The SQL travels
+// in the request body; every value goes as a param_* HTTP parameter.
+func (c *Client) Run(ctx context.Context, q Query, fn func(Row) error) error {
+	v := url.Values{}
+	v.Set("default_format", "JSONEachRow")
+	for name, value := range q.Params {
+		v.Set("param_"+name, value)
+	}
+	u := strings.TrimRight(c.cfg.URL, "/") + "/?" + v.Encode()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, strings.NewReader(q.SQL))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "text/plain")
+	req.Header.Set("X-ClickHouse-User", c.cfg.User)
+	if c.cfg.Password != "" {
+		req.Header.Set("X-ClickHouse-Key", c.cfg.Password)
+	}
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("clickhouse request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+
+	sc := bufio.NewScanner(resp.Body)
+	sc.Buffer(make([]byte, 0, 64*1024), 16*1024*1024)
+	line := 0
+	for sc.Scan() {
+		line++
+		b := sc.Bytes()
+		if len(b) == 0 {
+			continue
+		}
+		var r Row
+		if err := json.Unmarshal(b, &r); err != nil {
+			return fmt.Errorf("parse result row %s: %w", strconv.Itoa(line), err)
+		}
+		if err := fn(r); err != nil {
+			return err
+		}
+	}
+	return sc.Err()
+}
diff --git a/internal/chlog/client_test.go b/internal/chlog/client_test.go
new file mode 100644
index 0000000..bf18bea
--- /dev/null
+++ b/internal/chlog/client_test.go
@@ -0,0 +1,279 @@
+package chlog
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func chTS(t time.Time) string {
+	return t.UTC().Format("2006-01-02 15:04:05.000")
+}
+
+func mkRow(ts time.Time, msg string) Row {
+	return Row{
+		Timestamp: chTS(ts),
+		Host:      "node1",
+		Source:    "k8s",
+		Namespace: "logging",
+		Pod:       "vector-abc",
+		Container: "vector",
+		Stream:    "stdout",
+		Message:   msg,
+	}
+}
+
+// fakeCH serves the ClickHouse HTTP contract used by Client: JSONEachRow rows
+// from an in-memory store, honoring the since/until/limit query parameters.
+type fakeCH struct {
+	mu       sync.Mutex
+	rows     []Row
+	queries  int
+	lastUser string
+	lastPass string
+	lastSQL  string
+}
+
+func (f *fakeCH) add(rows ...Row) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.rows = append(f.rows, rows...)
+}
+
+func (f *fakeCH) handler(t *testing.T) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		f.queries++
+		f.lastUser = r.Header.Get("X-ClickHouse-User")
+		f.lastPass = r.Header.Get("X-ClickHouse-Key")
+		body := make([]byte, r.ContentLength)
+		r.Body.Read(body)
+		f.lastSQL = string(body)
+
+		q := r.URL.Query()
+		since, err1 := strconv.ParseInt(q.Get("param_since_ms"), 10, 64)
+		until, err2 := strconv.ParseInt(q.Get("param_until_ms"), 10, 64)
+		if err1 != nil || err2 != nil {
+			t.Errorf("query missing time bound params: %s", r.URL.RawQuery)
+			http.Error(w, "unbounded query", http.StatusBadRequest)
+			return
+		}
+		var limit int64 = -1
+		if s := q.Get("param_limit"); s != "" {
+			limit, _ = strconv.ParseInt(s, 10, 64)
+		}
+		var sent int64
+		enc := json.NewEncoder(w)
+		for _, row := range f.rows {
+			ms := row.Time().UnixMilli()
+			if ms < since || ms >= until {
+				continue
+			}
+			if limit >= 0 && sent >= limit {
+				break
+			}
+			enc.Encode(row)
+			sent++
+		}
+	}
+}
+
+func TestClientRun(t *testing.T) {
+	base := time.Date(2026, 8, 23, 5, 0, 0, 0, time.UTC)
+	fake := &fakeCH{}
+	fake.add(mkRow(base, "hello"), mkRow(base.Add(time.Second), "world"))
+	srv := httptest.NewServer(fake.handler(t))
+	defer srv.Close()
+
+	c := NewClient(Config{URL: srv.URL, User: "logreader", Password: "secret"})
+	q, err := Build(Filter{Since: base.Add(-time.Minute), Until: base.Add(time.Minute)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []Row
+	if err := c.Run(context.Background(), q, func(r Row) error {
+		got = append(got, r)
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 2 || got[0].Message != "hello" || got[1].Message != "world" {
+		t.Fatalf("rows = %+v", got)
+	}
+	if fake.lastUser != "logreader" || fake.lastPass != "secret" {
+		t.Errorf("auth headers: user=%q pass set=%v", fake.lastUser, fake.lastPass != "")
+	}
+	if !strings.Contains(fake.lastSQL, "FROM logs.raw") {
+		t.Errorf("SQL body not sent: %q", fake.lastSQL)
+	}
+	if got[0].Time().IsZero() {
+		t.Error("timestamp did not parse")
+	}
+}
+
+func TestClientRunHTTPError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "Code: 159. DB::Exception: Timeout exceeded", http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	c := NewClient(Config{URL: srv.URL, User: "logreader"})
+	q, _ := Build(Filter{Since: time.Now().Add(-time.Hour), Until: time.Now()})
+	err := c.Run(context.Background(), q, func(Row) error { return nil })
+	if err == nil || !strings.Contains(err.Error(), "HTTP 500") || !strings.Contains(err.Error(), "Timeout exceeded") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestPagePagesAndDedupes(t *testing.T) {
+	base := time.Date(2026, 8, 23, 5, 0, 0, 0, time.UTC)
+	fake := &fakeCH{}
+	// 25 rows, including three rows sharing one boundary timestamp.
+	for i := 0; i < 25; i++ {
+		ts := base.Add(time.Duration(i) * time.Second)
+		if i >= 10 && i < 13 {
+			ts = base.Add(10 * time.Second)
+		}
+		fake.add(mkRow(ts, fmt.Sprintf("msg-%d", i)))
+	}
+	srv := httptest.NewServer(fake.handler(t))
+	defer srv.Close()
+
+	c := NewClient(Config{URL: srv.URL, User: "logreader"})
+	f := Filter{Since: base.Add(-time.Minute), Until: base.Add(time.Hour)}
+	var got []string
+	n, err := Page(context.Background(), c, f, 7, func(r Row) error {
+		got = append(got, r.Message)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 25 || len(got) != 25 {
+		t.Fatalf("emitted %d rows (%d reported), want 25: %v", len(got), n, got)
+	}
+	seen := map[string]bool{}
+	for _, m := range got {
+		if seen[m] {
+			t.Fatalf("duplicate row %q emitted", m)
+		}
+		seen[m] = true
+	}
+	if fake.queries < 4 {
+		t.Errorf("expected multiple paged queries, got %d", fake.queries)
+	}
+}
+
+func TestPageHonorsLimit(t *testing.T) {
+	base := time.Date(2026, 8, 23, 5, 0, 0, 0, time.UTC)
+	fake := &fakeCH{}
+	for i := 0; i < 30; i++ {
+		fake.add(mkRow(base.Add(time.Duration(i)*time.Second), fmt.Sprintf("msg-%d", i)))
+	}
+	srv := httptest.NewServer(fake.handler(t))
+	defer srv.Close()
+
+	c := NewClient(Config{URL: srv.URL, User: "logreader"})
+	f := Filter{Since: base.Add(-time.Minute), Until: base.Add(time.Hour), Limit: 12}
+	var got []string
+	n, err := Page(context.Background(), c, f, 5, func(r Row) error {
+		got = append(got, r.Message)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 12 || len(got) != 12 || got[0] != "msg-0" || got[11] != "msg-11" {
+		t.Fatalf("emitted %d rows: %v", len(got), got)
+	}
+}
+
+func TestPageSingleMillisecondBurst(t *testing.T) {
+	base := time.Date(2026, 8, 23, 5, 0, 0, 0, time.UTC)
+	fake := &fakeCH{}
+	for i := 0; i < 6; i++ {
+		fake.add(mkRow(base, fmt.Sprintf("burst-%d", i)))
+	}
+	fake.add(mkRow(base.Add(time.Second), "after"))
+	srv := httptest.NewServer(fake.handler(t))
+	defer srv.Close()
+
+	c := NewClient(Config{URL: srv.URL, User: "logreader"})
+	f := Filter{Since: base.Add(-time.Minute), Until: base.Add(time.Hour)}
+	var got []string
+	_, err := Page(context.Background(), c, f, 3, func(r Row) error {
+		got = append(got, r.Message)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 7 || got[6] != "after" {
+		t.Fatalf("rows = %v", got)
+	}
+}
+
+func TestTailDedupesAcrossPolls(t *testing.T) {
+	base := time.Date(2026, 8, 23, 5, 0, 0, 0, time.UTC)
+	fake := &fakeCH{}
+	fake.add(mkRow(base, "initial-1"), mkRow(base.Add(time.Second), "initial-2"))
+	srv := httptest.NewServer(fake.handler(t))
+	defer srv.Close()
+
+	c := NewClient(Config{URL: srv.URL, User: "logreader"})
+
+	var mu sync.Mutex
+	var got []string
+	polls := 0
+	now := func() time.Time {
+		return base.Add(time.Duration(10+polls) * time.Second)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	emit := func(r Row) error {
+		mu.Lock()
+		defer mu.Unlock()
+		got = append(got, r.Message)
+		return nil
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- Tail(ctx, c, Filter{Since: base.Add(-time.Minute)}, func() time.Time {
+			mu.Lock()
+			defer mu.Unlock()
+			polls++
+			return now()
+		}, 10*time.Millisecond, emit)
+	}()
+
+	// Late-arriving row inside the overlap window plus a genuinely new row.
+	time.Sleep(50 * time.Millisecond)
+	fake.add(mkRow(base.Add(500*time.Millisecond), "late"), mkRow(base.Add(2*time.Second), "new"))
+	time.Sleep(100 * time.Millisecond)
+	cancel()
+	if err := <-done; err != context.Canceled {
+		t.Fatalf("tail err = %v", err)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	counts := map[string]int{}
+	for _, m := range got {
+		counts[m]++
+	}
+	for _, want := range []string{"initial-1", "initial-2", "late", "new"} {
+		if counts[want] != 1 {
+			t.Errorf("row %q emitted %d times; all: %v", want, counts[want], got)
+		}
+	}
+}
diff --git a/internal/chlog/config.go b/internal/chlog/config.go
new file mode 100644
index 0000000..a7f6ae8
--- /dev/null
+++ b/internal/chlog/config.go
@@ -0,0 +1,29 @@
+package chlog
+
+import "os"
+
+const (
+	DefaultURL  = "http://clickhouse-logs.logging.svc.cluster.local:8123"
+	DefaultUser = "logreader"
+)
+
+type Config struct {
+	URL      string
+	User     string
+	Password string
+}
+
+func ConfigFromEnv() Config {
+	cfg := Config{
+		URL:      os.Getenv("CH_URL"),
+		User:     os.Getenv("CH_USER"),
+		Password: os.Getenv("CH_PASSWORD"),
+	}
+	if cfg.URL == "" {
+		cfg.URL = DefaultURL
+	}
+	if cfg.User == "" {
+		cfg.User = DefaultUser
+	}
+	return cfg
+}
diff --git a/internal/chlog/format.go b/internal/chlog/format.go
new file mode 100644
index 0000000..ece6763
--- /dev/null
+++ b/internal/chlog/format.go
@@ -0,0 +1,149 @@
+package chlog
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+)
+
+const (
+	colReset = "\x1b[0m"
+	colDim   = "\x1b[2m"
+	colCyan  = "\x1b[36m"
+	colGreen = "\x1b[32m"
+	colRed   = "\x1b[31m"
+	colYel   = "\x1b[33m"
+)
+
+type Formatter func(w io.Writer, r Row) error
+
+func NewFormatter(format string, color bool) (Formatter, error) {
+	switch format {
+	case "text":
+		return textFormatter(color), nil
+	case "json":
+		return jsonFormatter, nil
+	case "logfmt":
+		return logfmtFormatter, nil
+	default:
+		return nil, fmt.Errorf("unknown format %q: want text, json or logfmt", format)
+	}
+}
+
+// origin renders the row's source identity: ns/pod for k8s rows, host for vm.
+func origin(r Row) string {
+	if r.Pod != "" {
+		return r.Namespace + "/" + r.Pod
+	}
+	if r.Namespace != "" {
+		return r.Namespace
+	}
+	return r.Host
+}
+
+func severityColor(sev string) string {
+	switch strings.ToLower(sev) {
+	case "error", "err", "fatal", "critical", "crit":
+		return colRed
+	case "warn", "warning":
+		return colYel
+	default:
+		return ""
+	}
+}
+
+func textFormatter(color bool) Formatter {
+	return func(w io.Writer, r Row) error {
+		ts := r.Time().Format("2006-01-02T15:04:05.000Z")
+		msg := r.Message
+		if !color {
+			_, err := fmt.Fprintf(w, "%s %s %s\n", ts, origin(r), msg)
+			return err
+		}
+		if c := severityColor(r.Severity); c != "" {
+			msg = c + msg + colReset
+		}
+		_, err := fmt.Fprintf(w, "%s%s%s %s%s%s %s\n",
+			colDim, ts, colReset, colCyan, origin(r), colReset, msg)
+		return err
+	}
+}
+
+type jsonRow struct {
+	Timestamp string            `json:"timestamp"`
+	Host      string            `json:"host"`
+	Source    string            `json:"source"`
+	Namespace string            `json:"namespace,omitempty"`
+	Pod       string            `json:"pod,omitempty"`
+	Container string            `json:"container,omitempty"`
+	Stream    string            `json:"stream,omitempty"`
+	Severity  string            `json:"severity,omitempty"`
+	Message   string            `json:"message"`
+	Labels    map[string]string `json:"labels,omitempty"`
+	Fields    map[string]string `json:"fields,omitempty"`
+}
+
+func jsonFormatter(w io.Writer, r Row) error {
+	b, err := json.Marshal(jsonRow{
+		Timestamp: r.Time().Format(time.RFC3339Nano),
+		Host:      r.Host,
+		Source:    r.Source,
+		Namespace: r.Namespace,
+		Pod:       r.Pod,
+		Container: r.Container,
+		Stream:    r.Stream,
+		Severity:  r.Severity,
+		Message:   r.Message,
+		Labels:    r.Labels,
+		Fields:    r.Fields,
+	})
+	if err != nil {
+		return err
+	}
+	_, err = fmt.Fprintf(w, "%s\n", b)
+	return err
+}
+
+func logfmtValue(s string) string {
+	if s == "" {
+		return `""`
+	}
+	if strings.ContainsAny(s, " \t\"=\n") {
+		return fmt.Sprintf("%q", s)
+	}
+	return s
+}
+
+func logfmtFormatter(w io.Writer, r Row) error {
+	var b strings.Builder
+	pair := func(k, v string) {
+		if v == "" {
+			return
+		}
+		if b.Len() > 0 {
+			b.WriteByte(' ')
+		}
+		b.WriteString(k)
+		b.WriteByte('=')
+		b.WriteString(logfmtValue(v))
+	}
+	pair("ts", r.Time().Format(time.RFC3339Nano))
+	pair("source", r.Source)
+	pair("host", r.Host)
+	pair("ns", r.Namespace)
+	pair("pod", r.Pod)
+	pair("container", r.Container)
+	pair("stream", r.Stream)
+	pair("severity", r.Severity)
+	pair("app", r.Labels["app"])
+	if b.Len() > 0 {
+		b.WriteByte(' ')
+	}
+	b.WriteString("msg=")
+	b.WriteString(logfmtValue(r.Message))
+	b.WriteByte('\n')
+	_, err := io.WriteString(w, b.String())
+	return err
+}
diff --git a/internal/chlog/format_test.go b/internal/chlog/format_test.go
new file mode 100644
index 0000000..823c1cf
--- /dev/null
+++ b/internal/chlog/format_test.go
@@ -0,0 +1,150 @@
+package chlog
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+)
+
+func sampleRow() Row {
+	return Row{
+		Timestamp: "2026-08-23 05:00:01.234",
+		Host:      "node1",
+		Source:    "k8s",
+		Namespace: "logging",
+		Pod:       "vector-abc",
+		Container: "vector",
+		Stream:    "stdout",
+		Severity:  "Error",
+		Message:   "something broke",
+		Labels:    map[string]string{"app": "vector"},
+		Fields:    map[string]string{"req_id": "42"},
+	}
+}
+
+func render(t *testing.T, format string, color bool, r Row) string {
+	t.Helper()
+	f, err := NewFormatter(format, color)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var b strings.Builder
+	if err := f(&b, r); err != nil {
+		t.Fatal(err)
+	}
+	return b.String()
+}
+
+func TestTextNoColor(t *testing.T) {
+	out := render(t, "text", false, sampleRow())
+	want := "2026-08-23T05:00:01.234Z logging/vector-abc something broke\n"
+	if out != want {
+		t.Errorf("got %q, want %q", out, want)
+	}
+	if strings.Contains(out, "\x1b[") {
+		t.Error("no-color output contains ANSI escapes")
+	}
+}
+
+func TestTextColor(t *testing.T) {
+	out := render(t, "text", true, sampleRow())
+	if !strings.Contains(out, "\x1b[") {
+		t.Error("color output missing ANSI escapes")
+	}
+	if !strings.Contains(out, "logging/vector-abc") || !strings.Contains(out, "something broke") {
+		t.Errorf("content missing: %q", out)
+	}
+}
+
+func TestTextVMRowUsesHost(t *testing.T) {
+	r := sampleRow()
+	r.Source = "vm"
+	r.Namespace = ""
+	r.Pod = ""
+	out := render(t, "text", false, r)
+	if !strings.Contains(out, " node1 ") {
+		t.Errorf("vm row should show host: %q", out)
+	}
+}
+
+func TestJSONFormat(t *testing.T) {
+	out := render(t, "json", false, sampleRow())
+	var m map[string]any
+	if err := json.Unmarshal([]byte(out), &m); err != nil {
+		t.Fatalf("not valid JSON: %v: %q", err, out)
+	}
+	if m["timestamp"] != "2026-08-23T05:00:01.234Z" {
+		t.Errorf("timestamp = %v", m["timestamp"])
+	}
+	if m["message"] != "something broke" || m["namespace"] != "logging" {
+		t.Errorf("fields wrong: %v", m)
+	}
+	labels, _ := m["labels"].(map[string]any)
+	if labels["app"] != "vector" {
+		t.Errorf("labels = %v", m["labels"])
+	}
+}
+
+func TestLogfmtFormat(t *testing.T) {
+	out := render(t, "logfmt", false, sampleRow())
+	for _, want := range []string{
+		"ts=2026-08-23T05:00:01.234Z",
+		"ns=logging",
+		"pod=vector-abc",
+		"severity=Error",
+		"app=vector",
+		`msg="something broke"`,
+	} {
+		if !strings.Contains(out, want) {
+			t.Errorf("logfmt missing %q: %q", want, out)
+		}
+	}
+	if !strings.HasSuffix(out, "\n") {
+		t.Error("missing trailing newline")
+	}
+}
+
+func TestLogfmtQuoting(t *testing.T) {
+	r := sampleRow()
+	r.Message = `plain`
+	out := render(t, "logfmt", false, r)
+	if !strings.Contains(out, "msg=plain") {
+		t.Errorf("unquoted simple value expected: %q", out)
+	}
+	r.Message = "has \"quotes\" and = signs"
+	out = render(t, "logfmt", false, r)
+	if !strings.Contains(out, `msg="has \"quotes\" and = signs"`) {
+		t.Errorf("quoted value wrong: %q", out)
+	}
+}
+
+func TestUnknownFormat(t *testing.T) {
+	if _, err := NewFormatter("yaml", false); err == nil {
+		t.Fatal("expected error for unknown format")
+	}
+}
+
+func TestRowKeyStableAndDistinct(t *testing.T) {
+	a := sampleRow()
+	b := sampleRow()
+	if a.Key() != b.Key() {
+		t.Error("identical rows should share a key")
+	}
+	b.Message = "different"
+	if a.Key() == b.Key() {
+		t.Error("different rows should not share a key")
+	}
+}
+
+func TestRowTimeParses(t *testing.T) {
+	r := sampleRow()
+	want := time.Date(2026, 8, 23, 5, 0, 1, 234000000, time.UTC)
+	if !r.Time().Equal(want) {
+		t.Errorf("Time() = %s, want %s", r.Time(), want)
+	}
+	r.Timestamp = "2026-08-23 05:00:01"
+	if r.Time().IsZero() {
+		t.Error("timestamp without fraction should still parse")
+	}
+}
diff --git a/internal/chlog/pager.go b/internal/chlog/pager.go
new file mode 100644
index 0000000..81ba5cd
--- /dev/null
+++ b/internal/chlog/pager.go
@@ -0,0 +1,78 @@
+package chlog
+
+import (
+	"context"
+	"time"
+)
+
+const DefaultPageSize = 10000
+
+type runner interface {
+	Run(ctx context.Context, q Query, fn func(Row) error) error
+}
+
+// Page walks the filter's time range as a series of bounded keyset-paged
+// queries: each page re-queries from the last-seen timestamp (inclusive, so
+// nothing on the boundary millisecond is lost) and dedupes the overlap. Each
+// page requests len(seen) extra rows on top of the wanted count, so known
+// boundary duplicates can never starve progress. Returns rows emitted.
+func Page(ctx context.Context, c runner, f Filter, pageSize uint64, emit func(Row) error) (uint64, error) {
+	if pageSize == 0 {
+		pageSize = DefaultPageSize
+	}
+	budget := f.Limit
+
+	cursor := f.Since
+	seen := map[uint64]struct{}{}
+	var emitted uint64
+
+	for {
+		want := pageSize
+		if budget > 0 && budget-emitted < want {
+			want = budget - emitted
+		}
+		pf := f
+		pf.Since = cursor
+		pf.Limit = want + uint64(len(seen))
+		q, err := Build(pf)
+		if err != nil {
+			return emitted, err
+		}
+
+		var got uint64
+		var lastTS time.Time
+		pageSeen := map[uint64]struct{}{}
+		err = c.Run(ctx, q, func(r Row) error {
+			got++
+			if budget > 0 && emitted >= budget {
+				return nil
+			}
+			ts := r.Time()
+			k := r.Key()
+			if ts.Equal(cursor) {
+				if _, dup := seen[k]; dup {
+					return nil
+				}
+				seen[k] = struct{}{}
+			} else {
+				if !ts.Equal(lastTS) {
+					lastTS = ts
+					pageSeen = map[uint64]struct{}{}
+				}
+				pageSeen[k] = struct{}{}
+			}
+			emitted++
+			return emit(r)
+		})
+		if err != nil {
+			return emitted, err
+		}
+		if got < pf.Limit || (budget > 0 && emitted >= budget) {
+			return emitted, nil
+		}
+		if lastTS.After(cursor) {
+			cursor = lastTS
+			seen = pageSeen
+		}
+	}
+}
diff --git a/internal/chlog/query.go b/internal/chlog/query.go
new file mode 100644
index 0000000..f0add3f
--- /dev/null
+++ b/internal/chlog/query.go
@@ -0,0 +1,131 @@
+package chlog
+
+import (
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const Table = "logs.raw"
+
+// Filter describes one bounded query against logs.raw. Since/Until are
+// mandatory: the table has no text index, so every query must be time-bounded.
+type Filter struct {
+	Since time.Time
+	Until time.Time
+
+	Namespace string
+	Host      string
+	Pod       string
+	Container string
+	App       string
+	Severity  string
+	Stream    string
+	Source    string
+
+	Pattern    string
+	Regex      bool
+	IgnoreCase bool
+	Fields     map[string]string
+
+	Limit uint64
+}
+
+// Selective reports whether the filter narrows the scan enough to be cheap:
+// any of namespace, host, or app restricts to a small slice of the table.
+func (f Filter) Selective() bool {
+	return f.Namespace != "" || f.Host != "" || f.App != ""
+}
+
+type Query struct {
+	SQL    string
+	Params map[string]string
+}
+
+const selectColumns = "timestamp, host, source, namespace, pod, container, stream, severity, message, labels, fields"
+
+func chTime(t time.Time) string {
+	return strconv.FormatInt(t.UnixMilli(), 10)
+}
+
+// Build renders a fully parameterized query. All user-supplied values travel
+// as HTTP {name:Type} parameters, never interpolated into the SQL text.
+func Build(f Filter) (Query, error) {
+	if f.Since.IsZero() || f.Until.IsZero() {
+		return Query{}, fmt.Errorf("query must be time-bounded: since/until missing")
+	}
+	if !f.Since.Before(f.Until) {
+		return Query{}, fmt.Errorf("empty time range: since %s is not before until %s",
+			f.Since.UTC().Format(time.RFC3339), f.Until.UTC().Format(time.RFC3339))
+	}
+
+	params := map[string]string{
+		"since_ms": chTime(f.Since),
+		"until_ms": chTime(f.Until),
+	}
+	where := []string{
+		"timestamp >= fromUnixTimestamp64Milli({since_ms:Int64})",
+		"timestamp < fromUnixTimestamp64Milli({until_ms:Int64})",
+	}
+
+	addEq := func(column, name, value string) {
+		if value == "" {
+			return
+		}
+		where = append(where, fmt.Sprintf("%s = {%s:String}", column, name))
+		params[name] = value
+	}
+	addEq("namespace", "ns", f.Namespace)
+	addEq("host", "host", f.Host)
+	addEq("pod", "pod", f.Pod)
+	addEq("container", "container", f.Container)
+	addEq("stream", "stream", f.Stream)
+	addEq("source", "source", f.Source)
+	addEq("labels['app']", "app", f.App)
+
+	if f.Severity != "" {
+		where = append(where, "lowerUTF8(severity) = {severity:String}")
+		params["severity"] = strings.ToLower(f.Severity)
+	}
+
+	if f.Pattern != "" {
+		switch {
+		case f.Regex:
+			pat := f.Pattern
+			if f.IgnoreCase {
+				pat = "(?i)" + pat
+			}
+			where = append(where, "match(message, {pattern:String})")
+			params["pattern"] = pat
+		case f.IgnoreCase:
+			where = append(where, "positionCaseInsensitive(message, {pattern:String}) > 0")
+			params["pattern"] = f.Pattern
+		default:
+			where = append(where, "position(message, {pattern:String}) > 0")
+			params["pattern"] = f.Pattern
+		}
+	}
+
+	keys := make([]string, 0, len(f.Fields))
+	for k := range f.Fields {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for i, k := range keys {
+		kn := fmt.Sprintf("fk%d", i)
+		vn := fmt.Sprintf("fv%d", i)
+		where = append(where, fmt.Sprintf("fields[{%s:String}] = {%s:String}", kn, vn))
+		params[kn] = k
+		params[vn] = f.Fields[k]
+	}
+
+	sql := fmt.Sprintf("SELECT %s FROM %s WHERE %s ORDER BY timestamp ASC",
+		selectColumns, Table, strings.Join(where, " AND "))
+	if f.Limit > 0 {
+		sql += " LIMIT {limit:UInt64}"
+		params["limit"] = strconv.FormatUint(f.Limit, 10)
+	}
+	return Query{SQL: sql, Params: params}, nil
+}
diff --git a/internal/chlog/query_test.go b/internal/chlog/query_test.go
new file mode 100644
index 0000000..546f67d
--- /dev/null
+++ b/internal/chlog/query_test.go
@@ -0,0 +1,204 @@
+package chlog
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+var (
+	tSince = time.Date(2026, 8, 23, 5, 0, 0, 0, time.UTC)
+	tUntil = time.Date(2026, 8, 23, 6, 0, 0, 0, time.UTC)
+)
+
+func baseFilter() Filter {
+	return Filter{Since: tSince, Until: tUntil}
+}
+
+func TestBuildRequiresBounds(t *testing.T) {
+	if _, err := Build(Filter{Until: tUntil}); err == nil {
+		t.Fatal("expected error when since missing")
+	}
+	if _, err := Build(Filter{Since: tSince}); err == nil {
+		t.Fatal("expected error when until missing")
+	}
+	if _, err := Build(Filter{Since: tUntil, Until: tSince}); err == nil {
+		t.Fatal("expected error when since >= until")
+	}
+}
+
+func TestBuildAlwaysTimeBounded(t *testing.T) {
+	q, err := Build(baseFilter())
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{
+		"timestamp >= fromUnixTimestamp64Milli({since_ms:Int64})",
+		"timestamp < fromUnixTimestamp64Milli({until_ms:Int64})",
+		"ORDER BY timestamp ASC",
+	} {
+		if !strings.Contains(q.SQL, want) {
+			t.Errorf("SQL missing %q:\n%s", want, q.SQL)
+		}
+	}
+	if q.Params["since_ms"] != strconv.FormatInt(tSince.UnixMilli(), 10) {
+		t.Errorf("since_ms = %s", q.Params["since_ms"])
+	}
+	if q.Params["until_ms"] != strconv.FormatInt(tUntil.UnixMilli(), 10) {
+		t.Errorf("until_ms = %s", q.Params["until_ms"])
+	}
+}
+
+func TestBuildNoUserInputInSQL(t *testing.T) {
+	f := baseFilter()
+	f.Namespace = "evil'; DROP TABLE logs.raw; --"
+	f.Pattern = "inject{p:String}"
+	f.App = "x' OR 1=1"
+	f.Fields = map[string]string{"k'": "v\""}
+	q, err := Build(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, needle := range []string{f.Namespace, f.Pattern, f.App, "k'", "v\"", "DROP"} {
+		if strings.Contains(q.SQL, needle) {
+			t.Errorf("user input %q leaked into SQL:\n%s", needle, q.SQL)
+		}
+	}
+	if q.Params["ns"] != f.Namespace || q.Params["pattern"] != f.Pattern {
+		t.Errorf("params missing user values: %v", q.Params)
+	}
+}
+
+func TestBuildFilters(t *testing.T) {
+	f := baseFilter()
+	f.Namespace = "logging"
+	f.Host = "node1"
+	f.Pod = "vector-abc"
+	f.Container = "vector"
+	f.Stream = "stderr"
+	f.Source = "k8s"
+	f.App = "vector"
+	f.Severity = "ERROR"
+	f.Limit = 100
+	q, err := Build(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for clause, param := range map[string]string{
+		"namespace = {ns:String}":                 "ns",
+		"host = {host:String}":                    "host",
+		"pod = {pod:String}":                      "pod",
+		"container = {container:String}":          "container",
+		"stream = {stream:String}":                "stream",
+		"source = {source:String}":                "source",
+		"labels['app'] = {app:String}":            "app",
+		"lowerUTF8(severity) = {severity:String}": "severity",
+		"LIMIT {limit:UInt64}":                    "limit",
+	} {
+		if !strings.Contains(q.SQL, clause) {
+			t.Errorf("SQL missing %q", clause)
+		}
+		if _, ok := q.Params[param]; !ok {
+			t.Errorf("param %q missing", param)
+		}
+	}
+	if q.Params["severity"] != "error" {
+		t.Errorf("severity not lowercased: %q", q.Params["severity"])
+	}
+	if q.Params["limit"] != "100" {
+		t.Errorf("limit = %q", q.Params["limit"])
+	}
+}
+
+func TestBuildEmptyFiltersOmitted(t *testing.T) {
+	q, err := Build(baseFilter())
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, where, ok := strings.Cut(q.SQL, " WHERE ")
+	if !ok {
+		t.Fatalf("no WHERE clause:\n%s", q.SQL)
+	}
+	for _, clause := range []string{"namespace =", "host =", "pod =", "labels[", "lowerUTF8", "position", "match", "LIMIT"} {
+		if strings.Contains(where, clause) {
+			t.Errorf("unexpected clause %q in WHERE:\n%s", clause, where)
+		}
+	}
+	if len(q.Params) != 2 {
+		t.Errorf("want only time params, got %v", q.Params)
+	}
+}
+
+func TestBuildGrepVariants(t *testing.T) {
+	cases := []struct {
+		regex, ignoreCase bool
+		wantClause        string
+		wantPattern       string
+	}{
+		{false, false, "position(message, {pattern:String}) > 0", "Timeout"},
+		{false, true, "positionCaseInsensitive(message, {pattern:String}) > 0", "Timeout"},
+		{true, false, "match(message, {pattern:String})", "Timeout"},
+		{true, true, "match(message, {pattern:String})", "(?i)Timeout"},
+	}
+	for _, c := range cases {
+		f := baseFilter()
+		f.Pattern = "Timeout"
+		f.Regex = c.regex
+		f.IgnoreCase = c.ignoreCase
+		q, err := Build(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !strings.Contains(q.SQL, c.wantClause) {
+			t.Errorf("regex=%v i=%v: SQL missing %q:\n%s", c.regex, c.ignoreCase, c.wantClause, q.SQL)
+		}
+		if q.Params["pattern"] != c.wantPattern {
+			t.Errorf("regex=%v i=%v: pattern param = %q, want %q", c.regex, c.ignoreCase, q.Params["pattern"], c.wantPattern)
+		}
+	}
+}
+
+func TestBuildFieldsDeterministic(t *testing.T) {
+	f := baseFilter()
+	f.Fields = map[string]string{"b": "2", "a": "1"}
+	q1, err := Build(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+	q2, _ := Build(f)
+	if q1.SQL != q2.SQL {
+		t.Error("field clause order not deterministic")
+	}
+	if !strings.Contains(q1.SQL, "fields[{fk0:String}] = {fv0:String}") ||
+		!strings.Contains(q1.SQL, "fields[{fk1:String}] = {fv1:String}") {
+		t.Errorf("field clauses missing:\n%s", q1.SQL)
+	}
+	if q1.Params["fk0"] != "a" || q1.Params["fv0"] != "1" || q1.Params["fk1"] != "b" || q1.Params["fv1"] != "2" {
+		t.Errorf("field params wrong: %v", q1.Params)
+	}
+}
+
+func TestSelective(t *testing.T) {
+	f := baseFilter()
+	if f.Selective() {
+		t.Error("empty filter should not be selective")
+	}
+	for _, set := range []func(*Filter){
+		func(f *Filter) { f.Namespace = "x" },
+		func(f *Filter) { f.Host = "x" },
+		func(f *Filter) { f.App = "x" },
+	} {
+		g := baseFilter()
+		set(&g)
+		if !g.Selective() {
+			t.Errorf("filter %+v should be selective", g)
+		}
+	}
+	g := baseFilter()
+	g.Pod = "x"
+	g.Container = "x"
+	if g.Selective() {
+		t.Error("pod/container alone should not count as selective")
+	}
+}
diff --git a/internal/chlog/tail.go b/internal/chlog/tail.go
new file mode 100644
index 0000000..7723757
--- /dev/null
+++ b/internal/chlog/tail.go
@@ -0,0 +1,84 @@
+package chlog
+
+import (
+	"context"
+	"time"
+)
+
+const (
+	TailInterval = 2 * time.Second
+	tailOverlap  = 5 * time.Second
+)
+
+// Tail streams the initial window then polls every interval. Each poll
+// re-queries from a little before the last-seen timestamp and drops rows
+// already emitted, so late-arriving rows inside the overlap still surface.
+func Tail(ctx context.Context, c runner, f Filter, now func() time.Time, interval time.Duration, emit func(Row) error) error {
+	if now == nil {
+		now = time.Now
+	}
+	if interval <= 0 {
+		interval = TailInterval
+	}
+
+	seen := map[uint64]time.Time{}
+	lastTS := f.Since
+	track := func(r Row) error {
+		if ts := r.Time(); ts.After(lastTS) {
+			lastTS = ts
+		}
+		seen[r.Key()] = r.Time()
+		return emit(r)
+	}
+
+	first := f
+	first.Until = now().UTC()
+	first.Limit = 0
+	if _, err := Page(ctx, c, first, 0, track); err != nil {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		return err
+	}
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+
+		since := lastTS.Add(-tailOverlap)
+		if since.Before(f.Since) {
+			since = f.Since
+		}
+		pf := f
+		pf.Since = since
+		pf.Until = now().UTC()
+		pf.Limit = 0
+		if !pf.Since.Before(pf.Until) {
+			continue
+		}
+		_, err := Page(ctx, c, pf, 0, func(r Row) error {
+			if _, dup := seen[r.Key()]; dup {
+				return nil
+			}
+			return track(r)
+		})
+		if err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			return err
+		}
+
+		floor := lastTS.Add(-2 * tailOverlap)
+		for k, ts := range seen {
+			if ts.Before(floor) {
+				delete(seen, k)
+			}
+		}
+	}
+}
diff --git a/internal/chlog/time.go b/internal/chlog/time.go
new file mode 100644
index 0000000..3373989
--- /dev/null
+++ b/internal/chlog/time.go
@@ -0,0 +1,38 @@
+package chlog
+
+import (
+	"fmt"
+	"regexp"
+	"strconv"
+	"time"
+)
+
+var durationRe = regexp.MustCompile(`^(\d+)([smhdw])$`)
+
+var durationUnits = map[string]time.Duration{
+	"s": time.Second,
+	"m": time.Minute,
+	"h": time.Hour,
+	"d": 24 * time.Hour,
+	"w": 7 * 24 * time.Hour,
+}
+
+// ParseTimeSpec accepts a relative duration (15m, 1h, 2d, 1w) meaning "that
+// long before now", or an absolute RFC3339 timestamp.
+func ParseTimeSpec(spec string, now time.Time) (time.Time, error) {
+	if spec == "" {
+		return time.Time{}, fmt.Errorf("empty time spec")
+	}
+	if m := durationRe.FindStringSubmatch(spec); m != nil {
+		n, err := strconv.ParseInt(m[1], 10, 64)
+		if err != nil {
+			return time.Time{}, fmt.Errorf("invalid duration %q: %w", spec, err)
+		}
+		return now.Add(-time.Duration(n) * durationUnits[m[2]]), nil
+	}
+	t, err := time.Parse(time.RFC3339, spec)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("invalid time %q: use a duration (15m, 1h, 2d) or RFC3339", spec)
+	}
+	return t.UTC(), nil
+}
diff --git a/internal/chlog/time_test.go b/internal/chlog/time_test.go
new file mode 100644
index 0000000..88d3498
--- /dev/null
+++ b/internal/chlog/time_test.go
@@ -0,0 +1,51 @@
+package chlog
+
+import (
+	"testing"
+	"time"
+)
+
+func TestParseTimeSpecDurations(t *testing.T) {
+	now := time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC)
+	cases := map[string]time.Time{
+		"15m": now.Add(-15 * time.Minute),
+		"1h":  now.Add(-time.Hour),
+		"2d":  now.Add(-48 * time.Hour),
+		"90s": now.Add(-90 * time.Second),
+		"1w":  now.Add(-7 * 24 * time.Hour),
+	}
+	for spec, want := range cases {
+		got, err := ParseTimeSpec(spec, now)
+		if err != nil {
+			t.Errorf("%s: %v", spec, err)
+			continue
+		}
+		if !got.Equal(want) {
+			t.Errorf("%s: got %s, want %s", spec, got, want)
+		}
+	}
+}
+
+func TestParseTimeSpecRFC3339(t *testing.T) {
+	now := time.Now()
+	got, err := ParseTimeSpec("2026-08-23T10:30:00+10:00", now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := time.Date(2026, 8, 23, 0, 30, 0, 0, time.UTC)
+	if !got.Equal(want) {
+		t.Errorf("got %s, want %s", got, want)
+	}
+	if got.Location() != time.UTC {
+		t.Errorf("not normalized to UTC: %s", got.Location())
+	}
+}
+
+func TestParseTimeSpecInvalid(t *testing.T) {
+	now := time.Now()
+	for _, spec := range []string{"", "abc", "1x", "-5m", "2026-13-99", "1.5h"} {
+		if _, err := ParseTimeSpec(spec, now); err == nil {
+			t.Errorf("%q: expected error", spec)
+		}
+	}
+}
diff --git a/main.go b/main.go
new file mode 100644
index 0000000..34affc9
--- /dev/null
+++ b/main.go
@@ -0,0 +1,249 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"git.unkin.net/unkin/clickhouse-tools/internal/chlog"
+)
+
+var version = "dev"
+
+const guardWindow = 6 * time.Hour
+
+type commonFlags struct {
+	since     string
+	until     string
+	namespace string
+	host      string
+	pod       string
+	container string
+	app       string
+	severity  string
+	stream    string
+	source    string
+	limit     uint64
+	format    string
+}
+
+func (cf *commonFlags) register(cmd *cobra.Command, defaultLimit uint64) {
+	fl := cmd.Flags()
+	fl.StringVar(&cf.since, "since", "1h", "start of time range: duration ago (15m, 1h, 2d) or RFC3339")
+	fl.StringVar(&cf.until, "until", "", "end of time range: duration ago or RFC3339 (default now)")
+	fl.StringVarP(&cf.namespace, "namespace", "n", "", "filter: k8s namespace")
+	fl.StringVar(&cf.host, "host", "", "filter: host")
+	fl.StringVar(&cf.pod, "pod", "", "filter: pod name")
+	fl.StringVar(&cf.container, "container", "", "filter: container name")
+	fl.StringVar(&cf.app, "app", "", "filter: labels['app']")
+	fl.StringVar(&cf.severity, "severity", "", "filter: severity (case-insensitive)")
+	fl.StringVar(&cf.stream, "stream", "", "filter: stream (stdout/stderr)")
+	fl.StringVar(&cf.source, "source", "", "filter: source (k8s/vm)")
+	fl.Uint64Var(&cf.limit, "limit", defaultLimit, "maximum rows to print (0 = unlimited)")
+	fl.StringVar(&cf.format, "format", "text", "output format: text, json or logfmt")
+}
+
+func (cf *commonFlags) filter(now time.Time) (chlog.Filter, error) {
+	since, err := chlog.ParseTimeSpec(cf.since, now)
+	if err != nil {
+		return chlog.Filter{}, fmt.Errorf("--since: %w", err)
+	}
+	until := now
+	if cf.until != "" {
+		until, err = chlog.ParseTimeSpec(cf.until, now)
+		if err != nil {
+			return chlog.Filter{}, fmt.Errorf("--until: %w", err)
+		}
+	}
+	return chlog.Filter{
+		Since:     since.UTC(),
+		Until:     until.UTC(),
+		Namespace: cf.namespace,
+		Host:      cf.host,
+		Pod:       cf.pod,
+		Container: cf.container,
+		App:       cf.app,
+		Severity:  cf.severity,
+		Stream:    cf.stream,
+		Source:    cf.source,
+		Limit:     cf.limit,
+	}, nil
+}
+
+func stdoutIsTTY() bool {
+	fi, err := os.Stdout.Stat()
+	return err == nil && fi.Mode()&os.ModeCharDevice != 0
+}
+
+func (cf *commonFlags) formatter() (chlog.Formatter, error) {
+	color := cf.format == "text" && stdoutIsTTY() && os.Getenv("NO_COLOR") == ""
+	return chlog.NewFormatter(cf.format, color)
+}
+
+func newCatCmd(use string) *cobra.Command {
+	cf := &commonFlags{}
+	cmd := &cobra.Command{
+		Use:   use,
+		Short: "Print logs from the ClickHouse log store, oldest first",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			f, err := cf.filter(time.Now().UTC())
+			if err != nil {
+				return err
+			}
+			out, err := cf.formatter()
+			if err != nil {
+				return err
+			}
+			c := chlog.NewClient(chlog.ConfigFromEnv())
+			_, err = chlog.Page(cmd.Context(), c, f, 0, func(r chlog.Row) error {
+				return out(os.Stdout, r)
+			})
+			return err
+		},
+	}
+	cf.register(cmd, 10000)
+	return cmd
+}
+
+func newTailCmd(use string) *cobra.Command {
+	cf := &commonFlags{}
+	cmd := &cobra.Command{
+		Use:   use,
+		Short: "Follow logs from the ClickHouse log store",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			cf.limit = 0
+			f, err := cf.filter(time.Now().UTC())
+			if err != nil {
+				return err
+			}
+			out, err := cf.formatter()
+			if err != nil {
+				return err
+			}
+			c := chlog.NewClient(chlog.ConfigFromEnv())
+			err = chlog.Tail(cmd.Context(), c, f, nil, chlog.TailInterval, func(r chlog.Row) error {
+				return out(os.Stdout, r)
+			})
+			if errors.Is(err, context.Canceled) {
+				return nil
+			}
+			return err
+		},
+	}
+	cf.register(cmd, 0)
+	cmd.Flags().MarkHidden("until")
+	cmd.Flags().MarkHidden("limit")
+	return cmd
+}
+
+func newGrepCmd(use string) *cobra.Command {
+	cf := &commonFlags{}
+	var (
+		regex      bool
+		ignoreCase bool
+		fields     []string
+		force      bool
+	)
+	cmd := &cobra.Command{
+		Use:   use + " <pattern>",
+		Short: "Search log messages in the ClickHouse log store",
+		Args:  cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			f, err := cf.filter(time.Now().UTC())
+			if err != nil {
+				return err
+			}
+			f.Pattern = args[0]
+			f.Regex = regex
+			f.IgnoreCase = ignoreCase
+			f.Fields, err = parseFields(fields)
+			if err != nil {
+				return err
+			}
+			if !f.Selective() && f.Until.Sub(f.Since) > guardWindow && !force {
+				return fmt.Errorf("unfiltered search over %s scans the whole table (no message index, ~281M rows/day); add --namespace/--host/--app, shrink --since to 6h or less, or pass --force",
+					f.Until.Sub(f.Since).Round(time.Minute))
+			}
+			out, err := cf.formatter()
+			if err != nil {
+				return err
+			}
+			c := chlog.NewClient(chlog.ConfigFromEnv())
+			_, err = chlog.Page(cmd.Context(), c, f, 0, func(r chlog.Row) error {
+				return out(os.Stdout, r)
+			})
+			return err
+		},
+	}
+	cf.register(cmd, 10000)
+	fl := cmd.Flags()
+	fl.BoolVar(&regex, "regex", false, "treat pattern as an RE2 regular expression")
+	fl.BoolVarP(&ignoreCase, "ignore-case", "i", false, "case-insensitive match")
+	fl.StringArrayVar(&fields, "fields", nil, "filter on structured fields: key=value (repeatable)")
+	fl.BoolVar(&force, "force", false, "allow an unfiltered search wider than 6h")
+	return cmd
+}
+
+func parseFields(kvs []string) (map[string]string, error) {
+	if len(kvs) == 0 {
+		return nil, nil
+	}
+	m := make(map[string]string, len(kvs))
+	for _, kv := range kvs {
+		k, v, ok := strings.Cut(kv, "=")
+		if !ok || k == "" {
+			return nil, fmt.Errorf("--fields %q: want key=value", kv)
+		}
+		m[k] = v
+	}
+	return m, nil
+}
+
+func newRootCmd() *cobra.Command {
+	root := &cobra.Command{
+		Use:           "chlog",
+		Short:         "CLI for the ClickHouse log store (logs.raw)",
+		Version:       version,
+		SilenceUsage:  true,
+		SilenceErrors: true,
+	}
+	root.AddCommand(newCatCmd("cat"), newTailCmd("tail"), newGrepCmd("grep"))
+	return root
+}
+
+func entrypoint() *cobra.Command {
+	var cmd *cobra.Command
+	switch filepath.Base(os.Args[0]) {
+	case "chcat":
+		cmd = newCatCmd("chcat")
+	case "chtail":
+		cmd = newTailCmd("chtail")
+	case "chgrep":
+		cmd = newGrepCmd("chgrep")
+	default:
+		return newRootCmd()
+	}
+	cmd.Version = version
+	cmd.SilenceUsage = true
+	cmd.SilenceErrors = true
+	return cmd
+}
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	if err := entrypoint().ExecuteContext(ctx); err != nil {
+		fmt.Fprintln(os.Stderr, "error:", err)
+		os.Exit(1)
+	}
+}
diff --git a/main_test.go b/main_test.go
new file mode 100644
index 0000000..db68ca0
--- /dev/null
+++ b/main_test.go
@@ -0,0 +1,96 @@
+package main
+
+import (
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestParseFields(t *testing.T) {
+	m, err := parseFields([]string{"a=1", "b=x=y"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if m["a"] != "1" || m["b"] != "x=y" {
+		t.Errorf("m = %v", m)
+	}
+	for _, bad := range []string{"noequals", "=v"} {
+		if _, err := parseFields([]string{bad}); err == nil {
+			t.Errorf("%q: expected error", bad)
+		}
+	}
+	if m, _ := parseFields(nil); m != nil {
+		t.Error("nil input should give nil map")
+	}
+}
+
+func TestCommonFlagsFilterDefaults(t *testing.T) {
+	cf := &commonFlags{since: "1h", format: "text"}
+	now := time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC)
+	f, err := cf.filter(now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !f.Since.Equal(now.Add(-time.Hour)) {
+		t.Errorf("since = %s", f.Since)
+	}
+	if !f.Until.Equal(now) {
+		t.Errorf("until = %s", f.Until)
+	}
+}
+
+func TestGrepGuardBlocksWideUnfilteredSearch(t *testing.T) {
+	cmd := newGrepCmd("chgrep")
+	cmd.SetArgs([]string{"--since", "24h", "needle"})
+	var out strings.Builder
+	cmd.SetOut(&out)
+	cmd.SetErr(&out)
+	err := cmd.Execute()
+	if err == nil || !strings.Contains(err.Error(), "--force") {
+		t.Fatalf("expected guard error mentioning --force, got %v", err)
+	}
+}
+
+func TestGrepGuardAllowsFilteredSearch(t *testing.T) {
+	// A namespace filter disables the guard; the query then fails on the
+	// unreachable server rather than the guard, proving the guard passed.
+	t.Setenv("CH_URL", "http://127.0.0.1:1")
+	cmd := newGrepCmd("chgrep")
+	cmd.SetArgs([]string{"--since", "24h", "--namespace", "logging", "needle"})
+	var out strings.Builder
+	cmd.SetOut(&out)
+	cmd.SetErr(&out)
+	err := cmd.Execute()
+	if err == nil {
+		t.Fatal("expected connection error")
+	}
+	if strings.Contains(err.Error(), "--force") {
+		t.Fatalf("guard should not trigger with a namespace filter: %v", err)
+	}
+}
+
+func TestGrepGuardAllowsShortWindow(t *testing.T) {
+	t.Setenv("CH_URL", "http://127.0.0.1:1")
+	cmd := newGrepCmd("chgrep")
+	cmd.SetArgs([]string{"--since", "1h", "needle"})
+	err := cmd.Execute()
+	if err == nil {
+		t.Fatal("expected connection error")
+	}
+	if strings.Contains(err.Error(), "--force") {
+		t.Fatalf("guard should not trigger for 1h window: %v", err)
+	}
+}
+
+func TestEntrypointDispatch(t *testing.T) {
+	root := newRootCmd()
+	names := map[string]bool{}
+	for _, c := range root.Commands() {
+		names[c.Name()] = true
+	}
+	for _, want := range []string{"cat", "tail", "grep"} {
+		if !names[want] {
+			t.Errorf("chlog missing subcommand %q", want)
+		}
+	}
+}
diff --git a/packaging/nfpm.yaml b/packaging/nfpm.yaml
new file mode 100644
index 0000000..ab44a09
--- /dev/null
+++ b/packaging/nfpm.yaml
@@ -0,0 +1,86 @@
+# nfpm config for the clickhouse-tools RPM.
+# Rendered through envsubst (see scripts/build-rpm.sh) then fed to `nfpm pkg`.
+
+name: ${PACKAGE_NAME}
+version: ${PACKAGE_VERSION}
+release: ${PACKAGE_RELEASE}
+arch: ${PACKAGE_ARCH}
+platform: ${PACKAGE_PLATFORM}
+section: default
+priority: extra
+description: "${PACKAGE_DESCRIPTION}"
+
+maintainer: ${PACKAGE_MAINTAINER}
+homepage: ${PACKAGE_HOMEPAGE}
+license: ${PACKAGE_LICENSE}
+
+disable_globbing: false
+
+contents:
+  - src: dist/chlog
+    dst: /usr/bin/chlog
+    file_info:
+      mode: 0755
+      owner: root
+      group: root
+
+  # chcat/chtail/chgrep dispatch on argv[0] inside the chlog binary.
+  - src: /usr/bin/chlog
+    dst: /usr/bin/chcat
+    type: symlink
+  - src: /usr/bin/chlog
+    dst: /usr/bin/chtail
+    type: symlink
+  - src: /usr/bin/chlog
+    dst: /usr/bin/chgrep
+    type: symlink
+
+  # Shell completions (generated by scripts/build-rpm.sh before packaging).
+  - src: dist/completions/chlog.bash
+    dst: /usr/share/bash-completion/completions/chlog
+    file_info:
+      mode: 0644
+  - src: dist/completions/_chlog
+    dst: /usr/share/zsh/site-functions/_chlog
+    file_info:
+      mode: 0644
+  - src: dist/completions/chlog.fish
+    dst: /usr/share/fish/vendor_completions.d/chlog.fish
+    file_info:
+      mode: 0644
+  - src: dist/completions/chcat.bash
+    dst: /usr/share/bash-completion/completions/chcat
+    file_info:
+      mode: 0644
+  - src: dist/completions/_chcat
+    dst: /usr/share/zsh/site-functions/_chcat
+    file_info:
+      mode: 0644
+  - src: dist/completions/chcat.fish
+    dst: /usr/share/fish/vendor_completions.d/chcat.fish
+    file_info:
+      mode: 0644
+  - src: dist/completions/chtail.bash
+    dst: /usr/share/bash-completion/completions/chtail
+    file_info:
+      mode: 0644
+  - src: dist/completions/_chtail
+    dst: /usr/share/zsh/site-functions/_chtail
+    file_info:
+      mode: 0644
+  - src: dist/completions/chtail.fish
+    dst: /usr/share/fish/vendor_completions.d/chtail.fish
+    file_info:
+      mode: 0644
+  - src: dist/completions/chgrep.bash
+    dst: /usr/share/bash-completion/completions/chgrep
+    file_info:
+      mode: 0644
+  - src: dist/completions/_chgrep
+    dst: /usr/share/zsh/site-functions/_chgrep
+    file_info:
+      mode: 0644
+  - src: dist/completions/chgrep.fish
+    dst: /usr/share/fish/vendor_completions.d/chgrep.fish
+    file_info:
+      mode: 0644
diff --git a/scripts/build-rpm.sh b/scripts/build-rpm.sh
new file mode 100644
index 0000000..2db597c
--- /dev/null
+++ b/scripts/build-rpm.sh
@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+#
+# Package the (already built) chlog binary into an RPM with nfpm, bundling
+# chcat/chtail/chgrep symlinks and generated bash/zsh/fish shell completions.
+# Usage: scripts/build-rpm.sh [version]   (version defaults to $CI_COMMIT_TAG)
+#
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${ROOT_DIR}"
+
+VERSION="${1:-${CI_COMMIT_TAG:-0.0.0-dev}}"
+VERSION="${VERSION#v}" # strip a leading v
+BINARY="chlog"
+NAMES=(chlog chcat chtail chgrep)
+DIST="dist"
+
+if [ ! -f "${DIST}/${BINARY}" ]; then
+  echo "ERROR: ${DIST}/${BINARY} not found; run 'make build' first" >&2
+  exit 1
+fi
+
+for l in chcat chtail chgrep; do
+  ln -sf "${BINARY}" "${DIST}/${l}"
+done
+
+# Generate shell completions per entrypoint name so they always match the
+# shipped flags/subcommands (each name dispatches to its own command tree).
+COMP_DIR="${DIST}/completions"
+mkdir -p "${COMP_DIR}"
+for b in "${NAMES[@]}"; do
+  "./${DIST}/${b}" completion bash >"${COMP_DIR}/${b}.bash"
+  "./${DIST}/${b}" completion zsh >"${COMP_DIR}/_${b}"
+  "./${DIST}/${b}" completion fish >"${COMP_DIR}/${b}.fish"
+done
+
+export PACKAGE_NAME="clickhouse-tools"
+export PACKAGE_VERSION="${VERSION}"
+export PACKAGE_RELEASE="1"
+export PACKAGE_ARCH="amd64"
+export PACKAGE_PLATFORM="linux"
+export PACKAGE_DESCRIPTION="CLI tools for the ClickHouse log store: chlog with chcat (print), chtail (follow) and chgrep (search) entrypoints"
+export PACKAGE_MAINTAINER="Ben Vincent <ben@unkin.net>"
+export PACKAGE_HOMEPAGE="https://git.unkin.net/unkin/clickhouse-tools"
+export PACKAGE_LICENSE="MIT"
+
+envsubst <packaging/nfpm.yaml >"${DIST}/nfpm.yaml"
+nfpm pkg --config "${DIST}/nfpm.yaml" --target "${DIST}" --packager rpm
+
+echo "Built:"
+ls -1 "${DIST}"/*.rpm