dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-16 00:49:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(stats): rank top advertisers by flood or direct (#218)

Browse Source 
Top advertisers ranked by flood + direct, so chatty companions with few flood adverts topped the list. sort=flood|direct is now required and applied before the limit; client floors raised to app 1.0.0 / web 2.0.4.
This commit is contained in:
MrAlders0n
2026-10-10 09:38:36 -04:00
committed by GitHub
parent 9bbf8efee5
commit d788e82ab4
20 changed files with 276 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/queries/queries.sql
+6 -4
View File
@@ -1191,8 +1191,8 @@ LEFT JOIN observers o ON o.id = r.observer_id
ORDER BY r.observation_count DESC, r.observer_id;
-- name: GetStatsTopAdvertisers :many
-- Top N advertisers since the given hour. Each ADVERT packet counts once per hour heard,
-- however many of the requested IATAs heard it (IATA-set rollup). Live names win.
-- Top N advertisers since the given hour, ranked by flood or direct adverts (@sort). Each ADVERT
-- packet counts once per hour heard, however many of the requested IATAs heard it. Live names win.
WITH ranked AS (
SELECT origin_pubkey, SUM(advert_packets)::bigint AS advert_count,
SUM(flood_packets)::bigint AS flood_advert_count, SUM(direct_packets)::bigint AS direct_advert_count
@@ -1200,7 +1200,8 @@ WITH ranked AS (
WHERE hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR iatas && @iatas::bpchar[])
GROUP BY origin_pubkey
ORDER BY advert_count DESC, origin_pubkey
ORDER BY CASE WHEN @sort::text = 'flood' THEN SUM(flood_packets) ELSE SUM(direct_packets) END DESC,
advert_count DESC, origin_pubkey
LIMIT @row_limit
), heard AS (
SELECT h.origin_pubkey, MAX(h.last_heard) AS last_heard, MAX(h.iata) AS iata,
@@ -1217,7 +1218,8 @@ SELECT n.id AS node_id, encode(r.origin_pubkey, 'hex') AS public_key,
FROM ranked r
LEFT JOIN heard h ON h.origin_pubkey = r.origin_pubkey
LEFT JOIN nodes n ON n.public_key = r.origin_pubkey
ORDER BY r.advert_count DESC, r.origin_pubkey;
ORDER BY CASE WHEN @sort::text = 'flood' THEN r.flood_advert_count ELSE r.direct_advert_count END DESC,
r.advert_count DESC, r.origin_pubkey;
-- name: GetStatsClockDrift :many
-- Repeaters/room servers (node_type 2/3) whose current advert-derived clock drift exceeds
db/sqlc/querier.go
+2 -2
View File
@@ -147,8 +147,8 @@ type Querier interface {
// Counts sum; observers, IATAs and scopes are distinct across the window; packets count
// once per hour heard. Each branch pair keeps "all IATAs" and "these IATAs" plans separate.
GetStatsSeries(ctx context.Context, arg GetStatsSeriesParams) ([]GetStatsSeriesRow, error)
// Top N advertisers since the given hour. Each ADVERT packet counts once per hour heard,
// however many of the requested IATAs heard it (IATA-set rollup). Live names win.
// Top N advertisers since the given hour, ranked by flood or direct adverts (@sort). Each ADVERT
// packet counts once per hour heard, however many of the requested IATAs heard it. Live names win.
GetStatsTopAdvertisers(ctx context.Context, arg GetStatsTopAdvertisersParams) ([]GetStatsTopAdvertisersRow, error)
// Top N observers since the given hour. Counts sum across matched IATAs; iata is a
// representative one. Live names win over the rolled snapshot.
db/sqlc/queries.sql.go
+18 -10
View File
@@ -1969,17 +1969,18 @@ WITH ranked AS (
SELECT origin_pubkey, SUM(advert_packets)::bigint AS advert_count,
SUM(flood_packets)::bigint AS flood_advert_count, SUM(direct_packets)::bigint AS direct_advert_count
FROM analytics_hourly_advert_sets
WHERE hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR iatas && $2::bpchar[])
WHERE hour >= $2::timestamptz
AND (COALESCE(cardinality($3::bpchar[]), 0) = 0 OR iatas && $3::bpchar[])
GROUP BY origin_pubkey
ORDER BY advert_count DESC, origin_pubkey
LIMIT $3
ORDER BY CASE WHEN $1::text = 'flood' THEN SUM(flood_packets) ELSE SUM(direct_packets) END DESC,
advert_count DESC, origin_pubkey
LIMIT $4
), heard AS (
SELECT h.origin_pubkey, MAX(h.last_heard) AS last_heard, MAX(h.iata) AS iata,
MAX(h.name) AS name, MAX(h.node_type) AS node_type
FROM analytics_hourly_advert_hearings h
WHERE h.hour >= $1::timestamptz AND h.origin_pubkey IN (SELECT origin_pubkey FROM ranked)
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR h.iata = ANY($2::bpchar[]))
WHERE h.hour >= $2::timestamptz AND h.origin_pubkey IN (SELECT origin_pubkey FROM ranked)
AND (COALESCE(cardinality($3::bpchar[]), 0) = 0 OR h.iata = ANY($3::bpchar[]))
GROUP BY h.origin_pubkey
)
SELECT n.id AS node_id, encode(r.origin_pubkey, 'hex') AS public_key,
@@ -1989,10 +1990,12 @@ SELECT n.id AS node_id, encode(r.origin_pubkey, 'hex') AS public_key,
FROM ranked r
LEFT JOIN heard h ON h.origin_pubkey = r.origin_pubkey
LEFT JOIN nodes n ON n.public_key = r.origin_pubkey
ORDER BY r.advert_count DESC, r.origin_pubkey
ORDER BY CASE WHEN $1::text = 'flood' THEN r.flood_advert_count ELSE r.direct_advert_count END DESC,
r.advert_count DESC, r.origin_pubkey
`
type GetStatsTopAdvertisersParams struct {
Sort string `json:"sort"`
Since pgtype.Timestamptz `json:"since"`
Iatas []string `json:"iatas"`
RowLimit int32 `json:"row_limit"`
@@ -2010,10 +2013,15 @@ type GetStatsTopAdvertisersRow struct {
Iata string `json:"iata"`
}
// Top N advertisers since the given hour. Each ADVERT packet counts once per hour heard,
// however many of the requested IATAs heard it (IATA-set rollup). Live names win.
// Top N advertisers since the given hour, ranked by flood or direct adverts (@sort). Each ADVERT
// packet counts once per hour heard, however many of the requested IATAs heard it. Live names win.
func (q *Queries) GetStatsTopAdvertisers(ctx context.Context, arg GetStatsTopAdvertisersParams) ([]GetStatsTopAdvertisersRow, error) {
rows, err := q.db.Query(ctx, getStatsTopAdvertisers, arg.Since, arg.Iatas, arg.RowLimit)
rows, err := q.db.Query(ctx, getStatsTopAdvertisers,
arg.Sort,
arg.Since,
arg.Iatas,
arg.RowLimit,
)
if err != nil {
return nil, err
}
db/stats.go
+2 -2
View File
@@ -130,8 +130,8 @@ func (s *Store) GetStatsTopObservers(ctx context.Context, iatas []string, since
return items, nil
}
func (s *Store) GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopAdvertiser, error) {
rows, err := s.q.GetStatsTopAdvertisers(ctx, sqlc.GetStatsTopAdvertisersParams{Since: rollupSince(since, 24*time.Hour), Iatas: iatas, RowLimit: limit})
func (s *Store) GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error) {
rows, err := s.q.GetStatsTopAdvertisers(ctx, sqlc.GetStatsTopAdvertisersParams{Sort: string(sort), Since: rollupSince(since, 24*time.Hour), Iatas: iatas, RowLimit: limit})
if err != nil {
return nil, err
}
db/stats_test.go
+3 -2
View File
@@ -10,6 +10,7 @@ import (
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"go.uber.org/mock/gomock"
@@ -133,7 +134,7 @@ func TestGetStatsTopAdvertisers_FloodDirectSplit(t *testing.T) {
heardAt := pgtype.Timestamptz{Time: time.Now(), Valid: true}
mock.EXPECT().
GetStatsTopAdvertisers(gomock.Any(), gomock.Any()).
GetStatsTopAdvertisers(gomock.Any(), gomock.Cond(func(p sqlc.GetStatsTopAdvertisersParams) bool { return p.Sort == "direct" })).
Return([]sqlc.GetStatsTopAdvertisersRow{
{
NodeID: pgtype.UUID{Bytes: nodeID, Valid: true},
@@ -149,7 +150,7 @@ func TestGetStatsTopAdvertisers_FloodDirectSplit(t *testing.T) {
}, nil)
store := &Store{q: mock}
items, err := store.GetStatsTopAdvertisers(context.Background(), []string{"YVR"}, time.Now().Add(-time.Hour), 5)
items, err := store.GetStatsTopAdvertisers(context.Background(), []string{"YVR"}, time.Now().Add(-time.Hour), 5, api.AdvertiserSortDirect)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
db/top_advertisers_integration_test.go
+69
View File
@@ -0,0 +1,69 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
"os"
"slices"
"testing"
"time"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/jackc/pgx/v5"
)
func TestTopAdvertisersSortPostgres(t *testing.T) {
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL regression")
}
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
conn, err := pgx.Connect(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer conn.Close(context.Background())
tx, err := conn.Begin(ctx)
if err != nil {
t.Fatal(err)
}
defer tx.Rollback(context.Background())
applyBaseline(t, ctx, tx)
store := &Store{q: sqlc.New(tx)}
// a1 is a chatty companion (mostly direct); a3 ties a2 on flood and wins on total.
if _, err := tx.Exec(ctx, `
INSERT INTO analytics_hourly_advert_sets (hour, origin_pubkey, iatas, advert_packets, flood_packets, direct_packets)
SELECT date_trunc('hour', now(), 'UTC') - INTERVAL '2 hours', pk, '{YVR}', f + d, f, d FROM (VALUES
('\xa1'::bytea, 10, 8300), ('\xa2', 50, 0), ('\xa3', 50, 5)
) v(pk, f, d)`); err != nil {
t.Fatal(err)
}
for _, tc := range []struct {
sort api.AdvertiserSort
limit int32
want []string
}{
{api.AdvertiserSortFlood, 1, []string{"a3"}},
{api.AdvertiserSortFlood, 3, []string{"a3", "a2", "a1"}},
{api.AdvertiserSortDirect, 1, []string{"a1"}},
{api.AdvertiserSortDirect, 3, []string{"a1", "a3", "a2"}},
} {
rows, err := store.GetStatsTopAdvertisers(ctx, []string{"YVR"}, time.Time{}, tc.limit, tc.sort)
if err != nil {
t.Fatal(err)
}
got := make([]string, len(rows))
for i, r := range rows {
got[i] = r.PublicKey
}
if !slices.Equal(got, tc.want) {
t.Errorf("sort=%s limit=%d: got %v, want %v", tc.sort, tc.limit, got, tc.want)
}
}
}
docs/docs.go
+19 -2
View File
@@ -3215,14 +3215,14 @@ const docTemplate = `{
},
"/stats/top-advertisers": {
"get": {
"description": "Each advert counts once per hour heard, however many requested IATAs heard it. nodeId is null when the node row has been deleted.",
"description": "Ranks by the sort count, ties broken by total adverts then public key. Each advert counts once per hour heard, however many requested IATAs heard it. nodeId is null when the node row has been deleted.",
"produces": [
"application/json"
],
"tags": [
"Stats"
],
"summary": "Top N nodes by distinct ADVERT packet count (last 24h by default)",
"summary": "Top N nodes by distinct flood or direct ADVERT packet count (last 24h by default)",
"parameters": [
{
"type": "string",
@@ -3255,6 +3255,17 @@ const docTemplate = `{
"description": "Max results (default 10); must be positive, values above 200 are clamped",
"name": "limit",
"in": "query"
},
{
"enum": [
"flood",
"direct"
],
"type": "string",
"description": "Rank by flood or direct advert count",
"name": "sort",
"in": "query",
"required": true
}
],
"responses": {
@@ -3267,6 +3278,12 @@ const docTemplate = `{
}
}
},
"400": {
"description": "Bad Request",
"schema": {
"$ref": "#/definitions/internal_api_handlers.APIError"
}
},
"500": {
"description": "Internal Server Error",
"schema": {
docs/swagger.json
+19 -2
View File
@@ -3213,14 +3213,14 @@
},
"/stats/top-advertisers": {
"get": {
"description": "Each advert counts once per hour heard, however many requested IATAs heard it. nodeId is null when the node row has been deleted.",
"description": "Ranks by the sort count, ties broken by total adverts then public key. Each advert counts once per hour heard, however many requested IATAs heard it. nodeId is null when the node row has been deleted.",
"produces": [
"application/json"
],
"tags": [
"Stats"
],
"summary": "Top N nodes by distinct ADVERT packet count (last 24h by default)",
"summary": "Top N nodes by distinct flood or direct ADVERT packet count (last 24h by default)",
"parameters": [
{
"type": "string",
@@ -3253,6 +3253,17 @@
"description": "Max results (default 10); must be positive, values above 200 are clamped",
"name": "limit",
"in": "query"
},
{
"enum": [
"flood",
"direct"
],
"type": "string",
"description": "Rank by flood or direct advert count",
"name": "sort",
"in": "query",
"required": true
}
],
"responses": {
@@ -3265,6 +3276,12 @@
}
}
},
"400": {
"description": "Bad Request",
"schema": {
"$ref": "#/definitions/internal_api_handlers.APIError"
}
},
"500": {
"description": "Internal Server Error",
"schema": {
docs/swagger.yaml
+17 -3
View File
@@ -3958,8 +3958,9 @@ paths:
- Stats
/stats/top-advertisers:
get:
description: Each advert counts once per hour heard, however many requested
IATAs heard it. nodeId is null when the node row has been deleted.
description: Ranks by the sort count, ties broken by total adverts then public
key. Each advert counts once per hour heard, however many requested IATAs
heard it. nodeId is null when the node row has been deleted.
parameters:
- description: Comma-separated IATA codes
in: query
@@ -3984,6 +3985,14 @@ paths:
minimum: 1
name: limit
type: integer
- description: Rank by flood or direct advert count
enum:
- flood
- direct
in: query
name: sort
required: true
type: string
produces:
- application/json
responses:
@@ -3993,11 +4002,16 @@ paths:
items:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopAdvertiser'
type: array
"400":
description: Bad Request
schema:
$ref: '#/definitions/internal_api_handlers.APIError'
"500":
description: Internal Server Error
schema:
$ref: '#/definitions/internal_api_handlers.APIError'
summary: Top N nodes by distinct ADVERT packet count (last 24h by default)
summary: Top N nodes by distinct flood or direct ADVERT packet count (last 24h
by default)
tags:
- Stats
/stats/top-nodes:
internal/api/handlers/limits_test.go
+2 -2
View File
@@ -79,7 +79,7 @@ func TestListLimits(t *testing.T) {
got = limit
return nil, nil
},
getStatsTopAdvertisers: func(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopAdvertiser, error) {
getStatsTopAdvertisers: func(ctx context.Context, iatas []string, since time.Time, limit int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error) {
calls++
got = limit
return nil, nil
@@ -133,7 +133,7 @@ func TestListLimits(t *testing.T) {
{"/traces", 50},
{"/stats/top-nodes", 10},
{"/stats/top-observers", 10},
{"/stats/top-advertisers", 10},
{"/stats/top-advertisers?sort=flood", 10},
{"/stats/clock-drift", 10},
{"/stats/top-talkers", 10},
} {
internal/api/handlers/stats.go
+10 -3
View File
@@ -284,8 +284,8 @@ func getStatsTopObservers(reader api.Reader) http.HandlerFunc {
// getStatsTopAdvertisers godoc
//
// @Summary Top N nodes by distinct ADVERT packet count (last 24h by default)
// @Description Each advert counts once per hour heard, however many requested IATAs heard it. nodeId is null when the node row has been deleted.
// @Summary Top N nodes by distinct flood or direct ADVERT packet count (last 24h by default)
// @Description Ranks by the sort count, ties broken by total adverts then public key. Each advert counts once per hour heard, however many requested IATAs heard it. nodeId is null when the node row has been deleted.
// @Tags Stats
// @Produce json
// @Param iatas query string false "Comma-separated IATA codes"
@@ -293,7 +293,9 @@ func getStatsTopObservers(reader api.Reader) http.HandlerFunc {
// @Param region query string false "Filter by region slug, expands to member IATAs"
// @Param since query int false "Start of window epoch ms (default last 24h)"
// @Param limit query int false "Max results (default 10); must be positive, values above 200 are clamped" minimum(1) maximum(200)
// @Param sort query string true "Rank by flood or direct advert count" Enums(flood,direct)
// @Success 200 {array} api.TopAdvertiser
// @Failure 400 {object} handlers.APIError
// @Failure 500 {object} handlers.APIError
// @Router /stats/top-advertisers [get]
func getStatsTopAdvertisers(reader api.Reader) http.HandlerFunc {
@@ -320,7 +322,12 @@ func getStatsTopAdvertisers(reader api.Reader) http.HandlerFunc {
respondError(w, http.StatusBadRequest, err.Error())
return
}
advertisers, err := reader.GetStatsTopAdvertisers(r.Context(), iatas, since, limit)
sort := api.AdvertiserSort(r.URL.Query().Get("sort"))
if sort != api.AdvertiserSortFlood && sort != api.AdvertiserSortDirect {
respondError(w, http.StatusBadRequest, "sort must be flood or direct")
return
}
advertisers, err := reader.GetStatsTopAdvertisers(r.Context(), iatas, since, limit, sort)
if err != nil {
slog.Error("api: GetStatsTopAdvertisers failed", "component", "api", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
internal/api/handlers/stats_region_test.go
+6 -2
View File
@@ -41,7 +41,7 @@ func (r iataRecorder) GetStatsTopObservers(_ context.Context, iatas []string, _
r.rec(iatas)
return nil, nil
}
func (r iataRecorder) GetStatsTopAdvertisers(_ context.Context, iatas []string, _ time.Time, _ int32) ([]api.TopAdvertiser, error) {
func (r iataRecorder) GetStatsTopAdvertisers(_ context.Context, iatas []string, _ time.Time, _ int32, _ api.AdvertiserSort) ([]api.TopAdvertiser, error) {
r.rec(iatas)
return nil, nil
}
@@ -81,7 +81,11 @@ func TestStatsEmptyRegionMatchesNothing(t *testing.T) {
return &api.Region{IATAs: []string{"YVR"}}, nil
}}}
w := httptest.NewRecorder()
StatsRouter(reader, StatsOptions{}).ServeHTTP(w, httptest.NewRequest("GET", "/"+path+"?region="+tc.region, nil))
query := "?region=" + tc.region
if path == "top-advertisers" {
query += "&sort=flood"
}
StatsRouter(reader, StatsOptions{}).ServeHTTP(w, httptest.NewRequest("GET", "/"+path+query, nil))
if w.Code != 200 || !reflect.DeepEqual(got, tc.want) {
t.Errorf("/stats/%s?region=%s: status %d, iatas %q; want 200, %q", path, tc.region, w.Code, got, tc.want)
}
internal/api/handlers/stats_test.go
+38 -13
View File
@@ -147,7 +147,7 @@ func TestGetStatsTopObservers_OK(t *testing.T) {
func TestGetStatsTopAdvertisers_InvalidSince(t *testing.T) {
r := chi.NewRouter()
r.Get("/stats/top-advertisers", getStatsTopAdvertisers(stubReader{}))
req := httptest.NewRequest(http.MethodGet, "/stats/top-advertisers?since=notanint", nil)
req := httptest.NewRequest(http.MethodGet, "/stats/top-advertisers?sort=flood&since=notanint", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
@@ -158,7 +158,7 @@ func TestGetStatsTopAdvertisers_InvalidSince(t *testing.T) {
func TestGetStatsTopAdvertisers_InvalidLimit(t *testing.T) {
r := chi.NewRouter()
r.Get("/stats/top-advertisers", getStatsTopAdvertisers(stubReader{}))
req := httptest.NewRequest(http.MethodGet, "/stats/top-advertisers?limit=notanint", nil)
req := httptest.NewRequest(http.MethodGet, "/stats/top-advertisers?sort=flood&limit=notanint", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
@@ -167,17 +167,42 @@ func TestGetStatsTopAdvertisers_InvalidLimit(t *testing.T) {
}
func TestGetStatsTopAdvertisers_OK(t *testing.T) {
r := chi.NewRouter()
r.Get("/stats/top-advertisers", getStatsTopAdvertisers(stubReader{
getStatsTopAdvertisers: func(_ context.Context, _ []string, _ time.Time, _ int32) ([]api.TopAdvertiser, error) {
return []api.TopAdvertiser{{IATA: "YVR", AdvertCount: 5}}, nil
},
}))
req := httptest.NewRequest(http.MethodGet, "/stats/top-advertisers", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d", w.Code)
for _, want := range []api.AdvertiserSort{api.AdvertiserSortFlood, api.AdvertiserSortDirect} {
var got api.AdvertiserSort
r := chi.NewRouter()
r.Get("/stats/top-advertisers", getStatsTopAdvertisers(stubReader{
getStatsTopAdvertisers: func(_ context.Context, _ []string, _ time.Time, _ int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error) {
got = sort
return []api.TopAdvertiser{{IATA: "YVR", AdvertCount: 5}}, nil
},
}))
req := httptest.NewRequest(http.MethodGet, "/stats/top-advertisers?sort="+string(want), nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("sort=%s: expected 200, got %d", want, w.Code)
}
if got != want {
t.Errorf("reader got sort %q, want %q", got, want)
}
}
}
func TestGetStatsTopAdvertisers_InvalidSort(t *testing.T) {
for _, query := range []string{"", "?sort=", "?sort=total", "?sort=FLOOD", "?sort=bogus"} {
called := false
r := chi.NewRouter()
r.Get("/stats/top-advertisers", getStatsTopAdvertisers(stubReader{
getStatsTopAdvertisers: func(_ context.Context, _ []string, _ time.Time, _ int32, _ api.AdvertiserSort) ([]api.TopAdvertiser, error) {
called = true
return nil, nil
},
}))
w := httptest.NewRecorder()
r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/stats/top-advertisers"+query, nil))
if w.Code != http.StatusBadRequest || called {
t.Errorf("%q: got %d (reader called=%v), want 400 without a reader call", query, w.Code, called)
}
}
}
internal/api/handlers/stub_reader_test.go
+3 -3
View File
@@ -51,7 +51,7 @@ type stubReader struct {
getStatsPayloadBreakdown func(ctx context.Context, iatas []string, since time.Time) ([]api.PayloadBreakdownItem, error)
getStatsTopNodes func(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopNode, error)
getStatsTopObservers func(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopObserver, error)
getStatsTopAdvertisers func(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopAdvertiser, error)
getStatsTopAdvertisers func(ctx context.Context, iatas []string, since time.Time, limit int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error)
getStatsClockDrift func(ctx context.Context, iatas []string, limit int32) ([]api.ClockDriftEntry, error)
getStatsTopTalkers func(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopTalker, error)
getScopeStats func(ctx context.Context, iatas []string, since time.Time) ([]api.ScopeStats, error)
@@ -305,9 +305,9 @@ func (s stubReader) GetStatsTopObservers(ctx context.Context, iatas []string, si
return nil, nil
}
func (s stubReader) GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopAdvertiser, error) {
func (s stubReader) GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error) {
if s.getStatsTopAdvertisers != nil {
return s.getStatsTopAdvertisers(ctx, iatas, since, limit)
return s.getStatsTopAdvertisers(ctx, iatas, since, limit, sort)
}
return nil, nil
}
internal/api/info.go
+2 -2
View File
@@ -11,8 +11,8 @@ import (
// Client floors this server code needs. Bump the matching one when a release breaks older
// clients; config can raise them, never lower them. Empty means no requirement.
const (
MinAppVersion = "" // BEACON Mobile
MinWebVersion = "2.0.3" // Beacon Web: batched route search
MinAppVersion = "1.0.0" // BEACON Mobile: top advertisers sort
MinWebVersion = "2.0.4" // Beacon Web: top advertisers sort
)
// Info is the public server description clients check before using a server.
internal/api/reader.go
+3 -2
View File
@@ -187,10 +187,11 @@ type Reader interface {
// since defines the start of the window; pass zero time for default (last 24h).
GetStatsTopObservers(ctx context.Context, iatas []string, since time.Time, limit int32) ([]TopObserver, error)
// GetStatsTopAdvertisers returns the top N nodes by distinct ADVERT packet count.
// GetStatsTopAdvertisers returns the top N nodes by distinct flood or direct ADVERT count,
// ties broken by total adverts then public key.
// Pass nil for iatas to return stats across all IATAs.
// since defines the start of the window; pass zero time for default (last 24h).
GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32) ([]TopAdvertiser, error)
GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32, sort AdvertiserSort) ([]TopAdvertiser, error)
// GetStatsClockDrift returns repeaters/room servers whose most recent advert-derived
// clock drift exceeds the configured threshold (nodes.clock_drift_threshold), worst
internal/api/stats.go
+10 -2
View File
@@ -128,8 +128,16 @@ type TopObserver struct {
ObservationCount int64 `json:"observationCount"`
}
// TopAdvertiser is a node ranked by distinct ADVERT packet count within the requested
// window. Count is per-advert, not per-hearing -- see GetStatsTopAdvertisers.
// AdvertiserSort picks which advert count ranks top advertisers.
type AdvertiserSort string
const (
AdvertiserSortFlood AdvertiserSort = "flood"
AdvertiserSortDirect AdvertiserSort = "direct"
)
// TopAdvertiser is a node ranked by distinct flood or direct ADVERT packet count within the
// requested window. Counts are per-advert, not per-hearing -- see GetStatsTopAdvertisers.
type TopAdvertiser struct {
NodeID *uuid.UUID `json:"nodeId"` // null once the node row has been deleted
PublicKey string `json:"publicKey"` // hex
internal/cache/cache_test.go
Vendored
+1 -1
View File
@@ -104,7 +104,7 @@ func (s *stubReader) GetStatsTopObservers(_ context.Context, _ []string, _ time.
return nil, nil
}
func (s *stubReader) GetStatsTopAdvertisers(_ context.Context, _ []string, _ time.Time, _ int32) ([]api.TopAdvertiser, error) {
func (s *stubReader) GetStatsTopAdvertisers(_ context.Context, _ []string, _ time.Time, _ int32, _ api.AdvertiserSort) ([]api.TopAdvertiser, error) {
return nil, nil
}
internal/cache/rollup_stats.go
Vendored
+3 -3
View File
@@ -156,10 +156,10 @@ func (cr *CachedReader) GetStatsTopObservers(ctx context.Context, iatas []string
}
// GetStatsTopAdvertisers implements [api.Reader].
func (cr *CachedReader) GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32) ([]api.TopAdvertiser, error) {
func (cr *CachedReader) GetStatsTopAdvertisers(ctx context.Context, iatas []string, since time.Time, limit int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error) {
return cachedRollup(ctx, cr, keyStatsTopAdvPrefix, iatas, func(ctx context.Context) ([]api.TopAdvertiser, error) {
return cr.inner.GetStatsTopAdvertisers(ctx, iatas, since, limit)
}, cr.sinceKey(since), limit)
return cr.inner.GetStatsTopAdvertisers(ctx, iatas, since, limit, sort)
}, cr.sinceKey(since), limit, sort)
}
// GetStatsTopTalkers implements [api.Reader].
internal/cache/top_advertisers_test.go
Vendored
+43
View File
@@ -0,0 +1,43 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package cache
import (
"context"
"testing"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
)
type topAdvertisersReader struct {
api.Reader
calls int64
}
func (r *topAdvertisersReader) GetStatsTopAdvertisers(_ context.Context, _ []string, _ time.Time, _ int32, sort api.AdvertiserSort) ([]api.TopAdvertiser, error) {
r.calls++
return []api.TopAdvertiser{{PublicKey: string(sort), AdvertCount: r.calls}}, nil
}
func (r *topAdvertisersReader) AnalyticsRevision(context.Context) (int64, error) { return 0, nil }
func TestTopAdvertisersCacheSeparatesSorts(t *testing.T) {
c, _ := newTestClient(t)
inner := &topAdvertisersReader{}
reader := NewCachedReader(inner, c, CacheTTLs{Stats: time.Hour})
since := time.UnixMilli(3600000)
for _, sort := range []api.AdvertiserSort{api.AdvertiserSortFlood, api.AdvertiserSortDirect, api.AdvertiserSortFlood, api.AdvertiserSortDirect} {
rows, err := reader.GetStatsTopAdvertisers(context.Background(), nil, since, 10, sort)
if err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0].PublicKey != string(sort) {
t.Fatalf("sort %s served %+v", sort, rows)
}
}
if inner.calls != 2 {
t.Fatalf("underlying calls = %d, want 2", inner.calls)
}
}