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.

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()
+}

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)
+		}
+	}
+}

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
+}

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
+}

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")
+	}
+}

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
+		}
+	}
+}

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
+}

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")
+	}
+}

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)
+			}
+		}
+	}
+}

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
+}

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)
+		}
+	}
+}