stats: where scraper, bot, and scanner traffic comes from

deploy/consolidated/README.md

@@ -220,7 +220,12 @@ COMPOSE_PROFILES=stats
 
 deploy.sh runs compose in `~`, where compose reads that file, so every normal deploy includes
 the trio. Alongside it live `STATS_AWS_ACCESS_KEY_ID`, `STATS_AWS_SECRET_ACCESS_KEY`,
-`STATS_S3_BUCKET`, `STATS_S3_REGION`, and `STATS_DB_PASSWORD` (same URL-safe rules as the
+`STATS_S3_BUCKET`, `STATS_S3_REGION`, optionally `STATS_GEO_DB_KEY` (the key of
+DB-IP's country CSV in that bucket — upload `dbip-country-lite-YYYY-MM.csv.gz`
+under `geo/` and restart `stats` for a new month; unset, the geo rows all read
+`--`, and a key that will not load is an error in the log, not a boot
+failure), and
+`STATS_DB_PASSWORD` (same URL-safe rules as the
 other database passwords: it rides in a libpq URL and a single-quoted SQL literal). Without
 the `COMPOSE_PROFILES` line the containers keep running after a deploy but silently stop
 being updated — compose ignores profile-gated services on an unflagged `up -d`.

deploy/consolidated/compose.yaml

@@ -662,6 +662,9 @@ services:
       - AWS_SECRET_ACCESS_KEY=${STATS_AWS_SECRET_ACCESS_KEY}
       - S3_BUCKET=${STATS_S3_BUCKET}
       - S3_REGION=${STATS_S3_REGION:-us-east-1}
+      # DB-IP's country CSV in the same bucket (#1467); empty means every
+      # geo row reads "--", and so does a key that will not load.
+      - GEO_DB_KEY=${STATS_GEO_DB_KEY:-}
       - PORT=8092
     networks:
       - app_network

deploy/consolidated/deploy_config_test.go

@@ -2116,6 +2116,11 @@ func TestTheStatsPairIsProfileGatedTogether(t *testing.T) {
 	if !strings.Contains(serviceBlock(t, "compose.yaml", "stats"), "postgresql://stats:") {
 		t.Errorf("stats names no stats database URL; the aggregates have nowhere to land")
 	}
+	// The geo database key rides the stats block (#1467); dropping it is an
+	// all-"--" table with nothing else to say so.
+	if !strings.Contains(serviceBlock(t, "compose.yaml", "stats"), "GEO_DB_KEY=") {
+		t.Errorf("the stats service does not pass GEO_DB_KEY; the geo rollup has no database to load")
+	}
 }
 
 // catchAllIsLastHandle reports whether the site block ends its handle

domains/platform/apis/stats/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "aggregate.go",
         "api.go",
         "classify.go",
+        "geo.go",
         "loop.go",
         "queries.go",
         "store.go",
@@ -33,6 +34,7 @@ go_test(
         "aggregate_test.go",
         "api_test.go",
         "classify_test.go",
+        "geo_test.go",
         "loop_test.go",
         "queries_test.go",
         "store_test.go",

domains/platform/apis/stats/README.md

@@ -56,9 +56,23 @@ The pass runs while the API serves, so for its length the counts climb
 back up from zero — minutes at this scale, and the log says when it is
 done.
 
-What stays ad hoc: IP-range clusters (a /24 key is caller-shaped and
-unbounded) and geography (nothing in the repo maps addresses to
-countries). Both are one query over the raw partitions in S3, which keep
+The geo rollup (#1467) places each request's `client_ip` (`remote_ip` on
+older lines) in a country and keys `geo_stats` on day, host, agent class,
+and the two-letter code, with request, 403, and probe counts — where the
+scrapers, bots, and scanners come from. The database is DB-IP's free
+country CSV (`dbip-country-lite-YYYY-MM.csv.gz`, CC BY 4.0; muchq.com's
+stats page carries the attribution), uploaded by the operator to the stats
+bucket under the key `GEO_DB_KEY` names; the service loads it at boot
+into a sorted range table and binary-searches it, no library. Overlapping
+rows lose to the range they sit in. An address outside every range, or no
+database at all, files under `--`. A key that will not load, after a few
+retries for a bucket still waking up, is an error in the log and an
+all-`--` table, never a boot failure. A new monthly file is a restart.
+Rows aggregated before the database was uploaded stay `--` until a
+re-aggregation (bump `RollupVersion`).
+
+What stays ad hoc: IP-range clusters — a /24 key is caller-shaped and
+unbounded — which is one query over the raw partitions in S3, keeping
 `request.remote_ip` and `request.client_ip` per line.
 
 ## The API
@@ -77,6 +91,8 @@ countries). Both are one query over the raw partitions in S3, which keep
   Rows from before #1468 landed on `api.muchq.com` and `gpt.muchq.com`
   overcount served, and the summary's error count there rose with the
   change, because scanner traffic now gets the 404 it always deserved.
+- `GET /stats/v1/countries?days=30&limit=2000` — per host/class/country
+  request, 403, and probe counts, busiest first
 - `GET /stats/v1/one_d4/queries?days=30` — one_d4 queries per
   day/entry/source/outcome/cache
 - `GET /stats/v1/one_d4/terms?days=30&limit=200` — which fields, motifs,
@@ -89,9 +105,10 @@ stay in the log, not on the wire.
 ## Configuration
 
 `STATS_DB_URL` (postgres), `S3_BUCKET`, `S3_REGION`, `AWS_ACCESS_KEY_ID`,
-`AWS_SECRET_ACCESS_KEY` — the same stats IAM user the shipper writes with,
+`AWS_SECRET_ACCESS_KEY`, and optionally `GEO_DB_KEY` — the same stats IAM user the shipper writes with,
 which therefore needs `s3:GetObject` and `s3:ListBucket` on the `logs/*`
-prefix as well as `s3:PutObject`. `AGGREGATE_INTERVAL` and `PORT`
+prefix as well as `s3:PutObject`, and `s3:GetObject` on the geo key's
+prefix (`geo/*` in the deployment). `AGGREGATE_INTERVAL` and `PORT`
 (default 8092) are optional.
 
 The store integration test needs `STATS_TEST_DB_URL` and skips without it,

domains/platform/apis/stats/aggregate.go

@@ -59,11 +59,15 @@ const (
 // so a crash between the two reprocesses the object rather than losing or
 // double-counting it.
 type Rollup struct {
-	Requests map[RequestKey]int64
-	Slugs    map[SlugKey]int64
-	Probes   map[ProbeKey]int64
-	Queries  map[QueryKey]int64
-	Terms    map[TermKey]int64
+	Requests  map[RequestKey]int64
+	Slugs     map[SlugKey]int64
+	Probes    map[ProbeKey]int64
+	Queries   map[QueryKey]int64
+	Terms     map[TermKey]int64
+	Countries map[GeoKey]GeoStat
+
+	// Geo places client addresses; NoLocator when no database is loaded.
+	Geo Locator
 
 	tailAgents map[string]bool
 }
@@ -75,6 +79,8 @@ func NewRollup() *Rollup {
 		Probes:     map[ProbeKey]int64{},
 		Queries:    map[QueryKey]int64{},
 		Terms:      map[TermKey]int64{},
+		Countries:  map[GeoKey]GeoStat{},
+		Geo:        NoLocator{},
 		tailAgents: map[string]bool{},
 	}
 }
@@ -100,13 +106,24 @@ func (r *Rollup) boundedAgent(class, agent string) string {
 type caddyLine struct {
 	Status  int `json:"status"`
 	Request struct {
-		Host    string              `json:"host"`
-		Method  string              `json:"method"`
-		URI     string              `json:"uri"`
-		Headers map[string][]string `json:"headers"`
+		Host     string              `json:"host"`
+		Method   string              `json:"method"`
+		URI      string              `json:"uri"`
+		ClientIP string              `json:"client_ip"`
+		RemoteIP string              `json:"remote_ip"`
+		Headers  map[string][]string `json:"headers"`
 	} `json:"request"`
 }
 
+// Caddy is the edge, so client_ip and remote_ip agree; older lines carry
+// only remote_ip.
+func (l *caddyLine) clientIP() string {
+	if l.Request.ClientIP != "" {
+		return l.Request.ClientIP
+	}
+	return l.Request.RemoteIP
+}
+
 func (l *caddyLine) userAgent() string {
 	values := l.Request.Headers["User-Agent"]
 	if len(values) == 0 {
@@ -158,9 +175,11 @@ func (r *Rollup) Consume(reader io.Reader, date string) (skipped int, err error)
 			AgentClass: agentClass,
 			Agent:      r.boundedAgent(agentClass, agent),
 		}]++
-		if probe := ProbeOf(parsed.Request.URI); probe != "" {
+		probe := ProbeOf(parsed.Request.URI)
+		if probe != "" {
 			r.Probes[ProbeKey{Date: date, Host: parsed.Request.Host, Probe: probe, Status: parsed.Status}]++
 		}
+		r.addGeo(date, parsed.Request.Host, agentClass, parsed.clientIP(), parsed.Status, probe != "")
 		if slug := SlugOf(parsed.Request.Host, parsed.Request.Method, parsed.Request.URI); slug != "" {
 			r.Slugs[SlugKey{Date: date, Slug: slug, Status: parsed.Status}]++
 		}

domains/platform/apis/stats/aggregate_test.go

@@ -112,6 +112,67 @@ func TestConsumeCapsTheAnonymousAgentTailPerObject(t *testing.T) {
 	}
 }
 
+type fakeLocator map[string]string
+
+func (f fakeLocator) Country(ip string) string { return f[ip] }
+
+func TestConsumeRollsUpWhereEachClassCameFromWithBlocksAndProbes(t *testing.T) {
+	lines := strings.Join([]string{
+		`{"status":200,"request":{"host":"h","method":"GET","uri":"/","client_ip":"1.0.0.7","headers":{"User-Agent":["Mozilla/5.0 (Macintosh) Chrome/126.0"]}}}`,
+		`{"status":403,"request":{"host":"h","method":"GET","uri":"/x","client_ip":"57.141.3.4","headers":{"User-Agent":["meta-externalagent/1.1"]}}}`,
+		`{"status":404,"request":{"host":"h","method":"GET","uri":"/.env","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`,
+		`{"status":404,"request":{"host":"h","method":"GET","uri":"/wp-login.php","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`,
+		// Refused and a probe at once: the production shape once refuse_bots answers a scanner.
+		`{"status":403,"request":{"host":"h","method":"GET","uri":"/.git/config","client_ip":"195.178.110.199","headers":{"User-Agent":["TLM-Audit-Scanner/1.0"]}}}`,
+		// Older lines carry remote_ip only; an address the locator does not know is "--".
+		`{"status":200,"request":{"host":"h","method":"GET","uri":"/","remote_ip":"10.1.2.3","headers":{"User-Agent":["curl/8.6.0"]}}}`,
+	}, "\n")
+	rollup := NewRollup()
+	rollup.Geo = fakeLocator{"1.0.0.7": "AU", "57.141.3.4": "US", "195.178.110.199": "GB"}
+
+	if _, err := rollup.Consume(strings.NewReader(lines), "2026-08-30"); err != nil {
+		t.Fatal(err)
+	}
+
+	want := map[GeoKey]GeoStat{
+		{"2026-08-30", "h", AgentBrowser, "AU"}:   {1, 0, 0},
+		{"2026-08-30", "h", AgentAIScraper, "US"}: {1, 1, 0},
+		{"2026-08-30", "h", AgentBot, "GB"}:       {3, 1, 3},
+		{"2026-08-30", "h", AgentBot, "--"}:       {1, 0, 0},
+	}
+	if len(rollup.Countries) != len(want) {
+		t.Errorf("geo rows = %v, want %v", rollup.Countries, want)
+	}
+	for key, stat := range want {
+		if rollup.Countries[key] != stat {
+			t.Errorf("geo %+v = %+v, want %+v", key, rollup.Countries[key], stat)
+		}
+	}
+}
+
+// Without a database every row is "--"; the class split still holds, so the
+// table is honest rather than empty.
+func TestConsumeWithoutAGeoDatabaseFilesEverythingUnderUnknown(t *testing.T) {
+	rollup := NewRollup()
+	if _, err := rollup.Consume(strings.NewReader(sampleLines), "2026-08-30"); err != nil {
+		t.Fatal(err)
+	}
+	var total int64
+	for key, stat := range rollup.Countries {
+		if key.Country != UnknownCountry {
+			t.Errorf("row %+v placed without a database", key)
+		}
+		total += stat.Requests
+	}
+	var requests int64
+	for _, count := range rollup.Requests {
+		requests += count
+	}
+	if total != requests {
+		t.Errorf("geo rows count %d requests, the request rollup %d; every request is somewhere", total, requests)
+	}
+}
+
 func TestConsumeSurvivesOversizedLines(t *testing.T) {
 	huge := `{"status":200,"request":{"host":"x","method":"GET","uri":"/` +
 		strings.Repeat("a", 2*1024*1024) + `","headers":{}}}`

domains/platform/apis/stats/api.go

@@ -17,6 +17,7 @@ type Reader interface {
 	Probes(ctx context.Context, days int) ([]ProbeRow, error)
 	Queries(ctx context.Context, days int) ([]QueryRow, error)
 	QueryTerms(ctx context.Context, days, limit int) ([]TermRow, error)
+	Countries(ctx context.Context, days, limit int) ([]CountryRow, error)
 }
 
 type Handlers struct {
@@ -113,6 +114,17 @@ func (h *Handlers) GetQueryTerms(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)})
 }
 
+func (h *Handlers) GetCountries(w http.ResponseWriter, r *http.Request) {
+	days := queryInt(r, "days", 30, 365)
+	limit := queryInt(r, "limit", 2000, 5000)
+	rows, err := h.reader.Countries(r.Context(), days, limit)
+	if err != nil {
+		h.serverError(w, "countries", err)
+		return
+	}
+	writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)})
+}
+
 func (h *Handlers) serverError(w http.ResponseWriter, what string, err error) {
 	h.logger.Error("stats query failed", "query", what, "error", err)
 	// The reason goes to the log, not the wire: these endpoints are public.

domains/platform/apis/stats/api_test.go

@@ -19,6 +19,7 @@ type fakeReader struct {
 	probes    []ProbeRow
 	queries   []QueryRow
 	terms     []TermRow
+	countries []CountryRow
 	lastDays  int
 	lastLimit int
 	fail      bool
@@ -72,6 +73,14 @@ func (f *fakeReader) QueryTerms(_ context.Context, days, limit int) ([]TermRow,
 	return f.terms, nil
 }
 
+func (f *fakeReader) Countries(_ context.Context, days, limit int) ([]CountryRow, error) {
+	if f.fail {
+		return nil, errors.New("db is having a day")
+	}
+	f.lastDays, f.lastLimit = days, limit
+	return f.countries, nil
+}
+
 func handlersWith(reader *fakeReader) *Handlers {
 	return NewHandlers(reader, slog.New(slog.NewTextHandler(io.Discard, nil)))
 }
@@ -247,3 +256,29 @@ func TestOneD4QueryEndpointsShareTheWindowRules(t *testing.T) {
 		}
 	}
 }
+
+func TestCountriesShareTheWindowRules(t *testing.T) {
+	reader := &fakeReader{countries: []CountryRow{
+		{Host: "git.muchq.com", AgentClass: AgentAIScraper, Country: "US", Requests: 9, Blocked: 4, Probes: 2},
+	}}
+	handlers := handlersWith(reader)
+
+	_, body := get(t, handlers.GetCountries, "/stats/v1/countries")
+	if reader.lastDays != 30 || reader.lastLimit != 2000 {
+		t.Errorf("default countries (days, limit) = (%d, %d), want (30, 2000)", reader.lastDays, reader.lastLimit)
+	}
+	row := body["rows"].([]any)[0].(map[string]any)
+	if row["country"] != "US" || row["requests"] != float64(9) || row["blocked"] != float64(4) || row["probes"] != float64(2) {
+		t.Errorf("country row = %v", row)
+	}
+	get(t, handlers.GetCountries, "/stats/v1/countries?days=99999&limit=99999")
+	if reader.lastDays != 365 || reader.lastLimit != 5000 {
+		t.Errorf("clamped countries (days, limit) = (%d, %d), want (365, 5000)", reader.lastDays, reader.lastLimit)
+	}
+	if _, body := get(t, handlersWith(&fakeReader{}).GetCountries, "/x"); body["rows"] == nil {
+		t.Error("countries rows serialized as null; want []")
+	}
+	if recorder, _ := get(t, handlersWith(&fakeReader{fail: true}).GetCountries, "/x"); recorder.Code != http.StatusInternalServerError {
+		t.Errorf("countries on a failing store = %d, want 500", recorder.Code)
+	}
+}