feat(stats): roll hourly scope observers and advertising nodes

This commit is contained in:
MrAlders0n authored and MrAlders0n committed 2026-10-02 15:24:33 -04:00
1 parent 53d9f86dd4
commit 2042c1617f
19 files changed
+271 -66

No files matched your search

+18
View File
@@ -547,6 +547,24 @@ CREATE TABLE analytics_hourly_scope_sets (
PRIMARY KEY (hour, iatas, scope_id)
);
-- Observers that heard a scoped packet, per IATA; distinct counts span any IATA filter.
CREATE TABLE analytics_hourly_scope_observers (
hour timestamp with time zone NOT NULL,
iata character(3) NOT NULL,
scope_id integer NOT NULL,
observer_id uuid NOT NULL,
PRIMARY KEY (hour, scope_id, iata, observer_id)
);
-- Nodes whose ADVERTs were heard in a scope, per IATA.
CREATE TABLE analytics_hourly_scope_nodes (
hour timestamp with time zone NOT NULL,
iata character(3) NOT NULL,
scope_id integer NOT NULL,
origin_pubkey bytea NOT NULL,
PRIMARY KEY (hour, scope_id, iata, origin_pubkey)
);
CREATE TABLE analytics_rollup_hours (
hour timestamp with time zone PRIMARY KEY,
status text NOT NULL CHECK (status IN ('complete', 'partial', 'missing')),
+34 -9
View File
@@ -1243,15 +1243,40 @@ WHERE ($1::text = '' OR preset = $1::text)
ORDER BY preset, iata, source_type;
-- name: GetScopeStatsHourly :many
-- GetScopeStats packet counts split by hour (same window and IATA-set filter); zero hours omitted.
SELECT ts.name, s.hour, SUM(s.packets)::bigint AS packets
FROM analytics_hourly_scope_sets s
JOIN transport_scopes ts ON ts.id = s.scope_id
WHERE s.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR s.iatas && @iatas::bpchar[])
GROUP BY ts.name, s.hour
HAVING SUM(s.packets) > 0
ORDER BY ts.name, s.hour;
-- GetScopeStats packet counts split by hour (same window and IATA-set filter), with the
-- distinct observers and advertising nodes active in each scope that hour. Empty hours omitted.
WITH pk AS (
SELECT s.scope_id, s.hour, SUM(s.packets)::bigint AS n
FROM analytics_hourly_scope_sets s
WHERE s.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR s.iatas && @iatas::bpchar[])
GROUP BY s.scope_id, s.hour
), ob AS (
SELECT o.scope_id, o.hour, COUNT(DISTINCT o.observer_id)::bigint AS n
FROM analytics_hourly_scope_observers o
WHERE o.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR o.iata = ANY(@iatas::bpchar[]))
GROUP BY o.scope_id, o.hour
), nd AS (
SELECT d.scope_id, d.hour, COUNT(DISTINCT d.origin_pubkey)::bigint AS n
FROM analytics_hourly_scope_nodes d
WHERE d.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR d.iata = ANY(@iatas::bpchar[]))
GROUP BY d.scope_id, d.hour
), k AS (
SELECT scope_id, hour FROM pk UNION SELECT scope_id, hour FROM ob UNION SELECT scope_id, hour FROM nd
)
SELECT ts.name, k.hour,
COALESCE(pk.n, 0)::bigint AS packets,
COALESCE(ob.n, 0)::bigint AS observers,
COALESCE(nd.n, 0)::bigint AS nodes
FROM k
JOIN transport_scopes ts ON ts.id = k.scope_id
LEFT JOIN pk ON pk.scope_id = k.scope_id AND pk.hour = k.hour
LEFT JOIN ob ON ob.scope_id = k.scope_id AND ob.hour = k.hour
LEFT JOIN nd ON nd.scope_id = k.scope_id AND nd.hour = k.hour
WHERE COALESCE(pk.n, 0) > 0 OR COALESCE(ob.n, 0) > 0 OR COALESCE(nd.n, 0) > 0
ORDER BY ts.name, k.hour;
-- name: GetScopeStats :many
-- Packets since the given hour come from the IATA-set rollup (counted once per hour heard).
+19 -3
View File
@@ -94,8 +94,10 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour = @ho
g AS (DELETE FROM analytics_hourly_advert_hearings tg WHERE tg.hour = @hour::timestamptz),
h AS (DELETE FROM analytics_hourly_advert_sets th WHERE th.hour = @hour::timestamptz),
i AS (DELETE FROM analytics_hourly_talker_sets ti WHERE ti.hour = @hour::timestamptz),
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = @hour::timestamptz)
DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = @hour::timestamptz;
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = @hour::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = @hour::timestamptz),
l AS (DELETE FROM analytics_hourly_scope_observers tl WHERE tl.hour = @hour::timestamptz)
DELETE FROM analytics_hourly_scope_nodes tm WHERE tm.hour = @hour::timestamptz;
-- name: RollIATAObservations :exec
INSERT INTO analytics_hourly_iata_observations (hour, iata, observation_count)
@@ -221,6 +223,16 @@ FROM (SELECT scope_id, array_agg(DISTINCT iata ORDER BY iata) AS iatas
GROUP BY packet_hash, scope_id) p
GROUP BY iatas, scope_id;
-- name: RollScopeObservers :exec
INSERT INTO analytics_hourly_scope_observers (hour, iata, scope_id, observer_id)
SELECT DISTINCT @hour::timestamptz, iata, scope_id, observer_id
FROM rollup_obs WHERE scope_id IS NOT NULL;
-- name: RollScopeNodes :exec
INSERT INTO analytics_hourly_scope_nodes (hour, iata, scope_id, origin_pubkey)
SELECT DISTINCT @hour::timestamptz, iata, scope_id, origin_pubkey
FROM rollup_obs WHERE payload_type = 4 AND origin_pubkey IS NOT NULL AND scope_id IS NOT NULL;
-- name: RollupContentHash :one
-- Order-independent fingerprint of every family row for the hour.
SELECT md5(concat_ws('|',
@@ -234,7 +246,9 @@ SELECT md5(concat_ws('|',
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_advert_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_talker_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_packet_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = @hour::timestamptz)
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_observers t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_nodes t WHERE t.hour = @hour::timestamptz)
))::text AS content_hash;
-- name: FinishRollupHour :one
@@ -275,5 +289,7 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour < @cu
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour < @cutoff::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour < @cutoff::timestamptz),
l AS (DELETE FROM analytics_dirty_hours tl WHERE tl.hour < @cutoff::timestamptz),
n AS (DELETE FROM analytics_hourly_scope_observers tn WHERE tn.hour < @cutoff::timestamptz),
o AS (DELETE FROM analytics_hourly_scope_nodes tob WHERE tob.hour < @cutoff::timestamptz),
m AS (DELETE FROM analytics_rollup_hours tm WHERE tm.hour < @cutoff::timestamptz RETURNING 1)
UPDATE analytics_raw_state SET coverage = coverage + 1 WHERE EXISTS (SELECT 1 FROM m);
+2
View File
@@ -153,6 +153,8 @@ func rollHour(ctx context.Context, tx pgx.Tx, hour time.Time) (RollOutcome, bool
"talker sets": q.RollTalkerSets,
"packet sets": q.RollPacketSets,
"scope sets": q.RollScopeSets,
"scope observers": q.RollScopeObservers,
"scope nodes": q.RollScopeNodes,
} {
if err := roll(ctx, h); err != nil {
return RollSkipped, false, fmt.Errorf("roll %s: %w", name, err)
+10 -1
View File
@@ -37,7 +37,7 @@ INSERT INTO channels (id, channel_hash) VALUES (1, '\x11');
INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, origin_pubkey, scope_id, raw_payload, raw_header, first_heard_at, last_heard_at)
SELECT hash, pt, rt, 0, origin, scope, '\x00', '\x00', $1::timestamptz, $1::timestamptz FROM (VALUES
('\x01'::bytea, 4, 1, '\xee'::bytea, NULL::int), -- flood advert, two IATAs
('\x02', 4, 2, '\xee', NULL), -- direct advert, one IATA
('\x02', 4, 2, '\xee', 1), -- direct advert, scoped, one IATA
('\x03', 5, 1, NULL, 1), -- scoped channel message, two IATAs
('\x04', 5, 1, NULL, 1), -- scoped channel message, one IATA
('\x05', 2, 1, NULL, NULL), -- straddles h0 and h0+1h
@@ -129,6 +129,15 @@ var rollupOracles = map[string][2]string{
`SELECT hour, iata, observer_id, observation_count, display_name, observer_type FROM analytics_hourly_observer_identity`,
`SELECT date_trunc('hour', po.heard_at, 'UTC'), po.iata, po.observer_id, count(*), o.display_name, o.observer_type
FROM packet_observations po JOIN observers o ON o.id = po.observer_id GROUP BY 1, 2, 3, 5, 6`},
"scope observers": {
`SELECT hour, iata, scope_id, observer_id FROM analytics_hourly_scope_observers`,
`SELECT DISTINCT date_trunc('hour', po.heard_at, 'UTC'), po.iata, p.scope_id, po.observer_id
FROM packet_observations po JOIN packets p ON p.packet_hash = po.packet_hash WHERE p.scope_id IS NOT NULL`},
"scope nodes": {
`SELECT hour, iata, scope_id, origin_pubkey FROM analytics_hourly_scope_nodes`,
`SELECT DISTINCT date_trunc('hour', po.heard_at, 'UTC'), po.iata, p.scope_id, p.origin_pubkey
FROM packet_observations po JOIN packets p ON p.packet_hash = po.packet_hash
WHERE po.payload_type = 4 AND p.origin_pubkey IS NOT NULL AND p.scope_id IS NOT NULL`},
"advert hearings": {
`SELECT hour, iata, origin_pubkey, observations, last_heard, name, node_type FROM analytics_hourly_advert_hearings`,
`SELECT date_trunc('hour', po.heard_at, 'UTC'), po.iata, p.origin_pubkey, count(*), max(po.heard_at), n.name, n.node_type
+25 -22
View File
@@ -31,15 +31,16 @@ FROM (VALUES (1,'#a'),(2,'#b'),(3,'#unused')) v(i,name);
INSERT INTO observers (id,public_key,last_iata)
SELECT md5(i::text)::uuid,decode(lpad(to_hex(i),2,'0'),'hex'),iata
FROM (VALUES (1,'YVR'),(2,'YYJ'),(3,'YVR'),(4,'YVR'),(5,'YYZ')) v(i,iata);
INSERT INTO packets (packet_hash,scope_id,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT decode(lpad(to_hex(i),2,'0'),'hex'),scope_id,4,0,1,'\x00','\x00',$1::timestamptz,$1::timestamptz
-- every packet is an advert from its own node, so per-scope nodes equal per-scope packets
INSERT INTO packets (packet_hash,scope_id,payload_type,payload_version,route_type,origin_pubkey,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT decode(lpad(to_hex(i),2,'0'),'hex'),scope_id,4,0,1,decode(lpad(to_hex(100+i),64,'0'),'hex'),'\x00','\x00',$1::timestamptz,$1::timestamptz
FROM (VALUES (1,1),(2,1),(3,1),(4,1),(5,2),(6,2),(7,2),(8,NULL)) v(i,scope_id);
INSERT INTO observer_scopes (observer_id,scope_id)
SELECT md5(i::text)::uuid,scope_id FROM (VALUES (1,1),(2,1),(4,1),(5,1),(1,2),(2,2),(3,2)) v(i,scope_id);
-- packet 4 is never heard, so it never reaches the rollup
INSERT INTO packet_observations (id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count)
INSERT INTO packet_observations (id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,payload_type)
SELECT id,decode(lpad(to_hex(packet),2,'0'),'hex'),md5(observer::text)::uuid,iata,
$1::timestamptz+id*interval '1 second',0,1,0
$1::timestamptz+id*interval '1 second',0,1,0,4
FROM (VALUES (1,1,1,'YVR'),(2,1,4,'YVR'),(3,1,2,'YYJ'),(4,2,2,'YVR'),
(5,3,1,'YYZ'),(6,5,3,'YVR'),(7,6,1,'YYJ'),(8,7,2,'YYZ'),(9,8,1,'YVR')) v(id,packet,observer,iata);
INSERT INTO nodes (id,public_key,node_type,default_scope_id)
@@ -62,20 +63,21 @@ SELECT md5(i::text)::uuid,iata FROM (VALUES (1,'YVR'),(1,'YYJ'),(2,'YVR'),(3,'YY
r.Close(ctx)
for _, tc := range []struct {
name string
iatas []string
since time.Time
a, b [3]int64 // packets, observers, nodes
name string
iatas []string
since time.Time
a, b [3]int64 // packets, observers, nodes
ah, bh int64 // distinct observers heard carrying the scope in hour h
}{
{"global", nil, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}},
{"empty filter", []string{}, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}},
{"YVR", []string{"YVR"}, time.Time{}, [3]int64{2, 2, 2}, [3]int64{1, 2, 1}},
{"YYJ", []string{"YYJ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 1, 2}},
{"YYZ", []string{"YYZ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 0, 0}},
{"overlapping IATAs", []string{"YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}},
{"duplicate IATAs", []string{"YYJ", "YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}},
{"unknown IATA", []string{"ZZZ"}, time.Time{}, [3]int64{}, [3]int64{}},
{"window after the traffic", nil, h.Add(time.Hour), [3]int64{0, 4, 4}, [3]int64{0, 3, 2}},
{"global", nil, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}, 3, 3},
{"empty filter", []string{}, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}, 3, 3},
{"YVR", []string{"YVR"}, time.Time{}, [3]int64{2, 2, 2}, [3]int64{1, 2, 1}, 3, 1},
{"YYJ", []string{"YYJ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 1, 2}, 1, 1},
{"YYZ", []string{"YYZ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 0, 0}, 1, 1},
{"overlapping IATAs", []string{"YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}, 3, 2},
{"duplicate IATAs", []string{"YYJ", "YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}, 3, 2},
{"unknown IATA", []string{"ZZZ"}, time.Time{}, [3]int64{}, [3]int64{}, 0, 0},
{"window after the traffic", nil, h.Add(time.Hour), [3]int64{0, 4, 4}, [3]int64{0, 3, 2}, 0, 0},
} {
t.Run(tc.name, func(t *testing.T) {
rows, err := store.GetScopeStats(ctx, tc.iatas, tc.since)
@@ -83,15 +85,16 @@ SELECT md5(i::text)::uuid,iata FROM (VALUES (1,'YVR'),(1,'YYJ'),(2,'YVR'),(3,'YY
t.Fatal(err)
}
// All fixture traffic is in hour h; hours with no packets are omitted.
hourly := func(packets int64) []api.ScopeHour {
if packets == 0 {
// Each packet has its own origin, so active nodes per hour equal packets per hour.
hourly := func(packets, observers int64) []api.ScopeHour {
if packets == 0 && observers == 0 {
return []api.ScopeHour{}
}
return []api.ScopeHour{{Hour: h.UnixMilli(), Packets: packets}}
return []api.ScopeHour{{Hour: h.UnixMilli(), Packets: packets, Observers: observers, Nodes: packets}}
}
want := []api.ScopeStats{
{Name: "#a", PacketCount: tc.a[0], ObserverCount: tc.a[1], NodeCount: tc.a[2], Hourly: hourly(tc.a[0])},
{Name: "#b", PacketCount: tc.b[0], ObserverCount: tc.b[1], NodeCount: tc.b[2], Hourly: hourly(tc.b[0])},
{Name: "#a", PacketCount: tc.a[0], ObserverCount: tc.a[1], NodeCount: tc.a[2], Hourly: hourly(tc.a[0], tc.ah)},
{Name: "#b", PacketCount: tc.b[0], ObserverCount: tc.b[1], NodeCount: tc.b[2], Hourly: hourly(tc.b[0], tc.bh)},
{Name: "#unused", Hourly: []api.ScopeHour{}},
}
if !reflect.DeepEqual(rows, want) {
+28
View File
@@ -1978,6 +1978,34 @@ func (mr *MockQuerierMockRecorder) RollPayloadBreakdown(ctx, hour any) *gomock.C
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RollPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).RollPayloadBreakdown), ctx, hour)
}
// RollScopeNodes mocks base method.
func (m *MockQuerier) RollScopeNodes(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "RollScopeNodes", ctx, hour)
ret0, _ := ret[0].(error)
return ret0
}
// RollScopeNodes indicates an expected call of RollScopeNodes.
func (mr *MockQuerierMockRecorder) RollScopeNodes(ctx, hour any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RollScopeNodes", reflect.TypeOf((*MockQuerier)(nil).RollScopeNodes), ctx, hour)
}
// RollScopeObservers mocks base method.
func (m *MockQuerier) RollScopeObservers(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "RollScopeObservers", ctx, hour)
ret0, _ := ret[0].(error)
return ret0
}
// RollScopeObservers indicates an expected call of RollScopeObservers.
func (mr *MockQuerierMockRecorder) RollScopeObservers(ctx, hour any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RollScopeObservers", reflect.TypeOf((*MockQuerier)(nil).RollScopeObservers), ctx, hour)
}
// RollScopeSets mocks base method.
func (m *MockQuerier) RollScopeSets(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
+14
View File
@@ -91,6 +91,20 @@ type AnalyticsHourlyPayloadBreakdown struct {
Count int64 `json:"count"`
}
type AnalyticsHourlyScopeNode struct {
Hour pgtype.Timestamptz `json:"hour"`
Iata string `json:"iata"`
ScopeID int32 `json:"scope_id"`
OriginPubkey []byte `json:"origin_pubkey"`
}
type AnalyticsHourlyScopeObserver struct {
Hour pgtype.Timestamptz `json:"hour"`
Iata string `json:"iata"`
ScopeID int32 `json:"scope_id"`
ObserverID uuid.UUID `json:"observer_id"`
}
type AnalyticsHourlyScopeSet struct {
Hour pgtype.Timestamptz `json:"hour"`
Iatas []string `json:"iatas"`
+4 -1
View File
@@ -122,7 +122,8 @@ type Querier interface {
// Packets since the given hour come from the IATA-set rollup (counted once per hour heard).
// Observer and node counts are current memberships; observers filter by their latest IATA.
GetScopeStats(ctx context.Context, arg GetScopeStatsParams) ([]GetScopeStatsRow, error)
// GetScopeStats packet counts split by hour (same window and IATA-set filter); zero hours omitted.
// GetScopeStats packet counts split by hour (same window and IATA-set filter), with the
// distinct observers and advertising nodes active in each scope that hour. Empty hours omitted.
GetScopeStatsHourly(ctx context.Context, arg GetScopeStatsHourlyParams) ([]GetScopeStatsHourlyRow, error)
// Read the hourly rollup, never observations on an HTTP request.
// Weight averages by sample counts instead of averaging regional/hourly means.
@@ -317,6 +318,8 @@ type Querier interface {
// category: 0 = routed, 1 = zero-hop, 2 = TRACE, 3 = invalid (meshcore-go IsValidPathLen).
RollPaths(ctx context.Context, hour pgtype.Timestamptz) error
RollPayloadBreakdown(ctx context.Context, hour pgtype.Timestamptz) error
RollScopeNodes(ctx context.Context, hour pgtype.Timestamptz) error
RollScopeObservers(ctx context.Context, hour pgtype.Timestamptz) error
RollScopeSets(ctx context.Context, hour pgtype.Timestamptz) error
// kind = grouping(snr_bin, rssi_bin): 3 = per IATA totals, 1 = per SNR bin, 2 = per RSSI bin.
// rssi=0 AND snr=0 means "not reported".
+46 -13
View File
@@ -1726,14 +1726,38 @@ func (q *Queries) GetScopeStats(ctx context.Context, arg GetScopeStatsParams) ([
}
const getScopeStatsHourly = `-- name: GetScopeStatsHourly :many
SELECT ts.name, s.hour, SUM(s.packets)::bigint AS packets
FROM analytics_hourly_scope_sets s
JOIN transport_scopes ts ON ts.id = s.scope_id
WHERE s.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR s.iatas && $2::bpchar[])
GROUP BY ts.name, s.hour
HAVING SUM(s.packets) > 0
ORDER BY ts.name, s.hour
WITH pk AS (
SELECT s.scope_id, s.hour, SUM(s.packets)::bigint AS n
FROM analytics_hourly_scope_sets s
WHERE s.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR s.iatas && $2::bpchar[])
GROUP BY s.scope_id, s.hour
), ob AS (
SELECT o.scope_id, o.hour, COUNT(DISTINCT o.observer_id)::bigint AS n
FROM analytics_hourly_scope_observers o
WHERE o.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR o.iata = ANY($2::bpchar[]))
GROUP BY o.scope_id, o.hour
), nd AS (
SELECT d.scope_id, d.hour, COUNT(DISTINCT d.origin_pubkey)::bigint AS n
FROM analytics_hourly_scope_nodes d
WHERE d.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR d.iata = ANY($2::bpchar[]))
GROUP BY d.scope_id, d.hour
), k AS (
SELECT scope_id, hour FROM pk UNION SELECT scope_id, hour FROM ob UNION SELECT scope_id, hour FROM nd
)
SELECT ts.name, k.hour,
COALESCE(pk.n, 0)::bigint AS packets,
COALESCE(ob.n, 0)::bigint AS observers,
COALESCE(nd.n, 0)::bigint AS nodes
FROM k
JOIN transport_scopes ts ON ts.id = k.scope_id
LEFT JOIN pk ON pk.scope_id = k.scope_id AND pk.hour = k.hour
LEFT JOIN ob ON ob.scope_id = k.scope_id AND ob.hour = k.hour
LEFT JOIN nd ON nd.scope_id = k.scope_id AND nd.hour = k.hour
WHERE COALESCE(pk.n, 0) > 0 OR COALESCE(ob.n, 0) > 0 OR COALESCE(nd.n, 0) > 0
ORDER BY ts.name, k.hour
`
type GetScopeStatsHourlyParams struct {
@@ -1742,12 +1766,15 @@ type GetScopeStatsHourlyParams struct {
}
type GetScopeStatsHourlyRow struct {
Name string `json:"name"`
Hour pgtype.Timestamptz `json:"hour"`
Packets int64 `json:"packets"`
Name string `json:"name"`
Hour pgtype.Timestamptz `json:"hour"`
Packets int64 `json:"packets"`
Observers int64 `json:"observers"`
Nodes int64 `json:"nodes"`
}
// GetScopeStats packet counts split by hour (same window and IATA-set filter); zero hours omitted.
// GetScopeStats packet counts split by hour (same window and IATA-set filter), with the
// distinct observers and advertising nodes active in each scope that hour. Empty hours omitted.
func (q *Queries) GetScopeStatsHourly(ctx context.Context, arg GetScopeStatsHourlyParams) ([]GetScopeStatsHourlyRow, error) {
rows, err := q.db.Query(ctx, getScopeStatsHourly, arg.Since, arg.Iatas)
if err != nil {
@@ -1757,7 +1784,13 @@ func (q *Queries) GetScopeStatsHourly(ctx context.Context, arg GetScopeStatsHour
items := []GetScopeStatsHourlyRow{}
for rows.Next() {
var i GetScopeStatsHourlyRow
if err := rows.Scan(&i.Name, &i.Hour, &i.Packets); err != nil {
if err := rows.Scan(
&i.Name,
&i.Hour,
&i.Packets,
&i.Observers,
&i.Nodes,
); err != nil {
return nil, err
}
items = append(items, i)
+31 -3
View File
@@ -24,6 +24,8 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour < $1:
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour < $1::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour < $1::timestamptz),
l AS (DELETE FROM analytics_dirty_hours tl WHERE tl.hour < $1::timestamptz),
n AS (DELETE FROM analytics_hourly_scope_observers tn WHERE tn.hour < $1::timestamptz),
o AS (DELETE FROM analytics_hourly_scope_nodes tob WHERE tob.hour < $1::timestamptz),
m AS (DELETE FROM analytics_rollup_hours tm WHERE tm.hour < $1::timestamptz RETURNING 1)
UPDATE analytics_raw_state SET coverage = coverage + 1 WHERE EXISTS (SELECT 1 FROM m)
`
@@ -44,8 +46,10 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour = $1:
g AS (DELETE FROM analytics_hourly_advert_hearings tg WHERE tg.hour = $1::timestamptz),
h AS (DELETE FROM analytics_hourly_advert_sets th WHERE th.hour = $1::timestamptz),
i AS (DELETE FROM analytics_hourly_talker_sets ti WHERE ti.hour = $1::timestamptz),
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = $1::timestamptz)
DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = $1::timestamptz
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = $1::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = $1::timestamptz),
l AS (DELETE FROM analytics_hourly_scope_observers tl WHERE tl.hour = $1::timestamptz)
DELETE FROM analytics_hourly_scope_nodes tm WHERE tm.hour = $1::timestamptz
`
func (q *Queries) DeleteRollupHour(ctx context.Context, hour pgtype.Timestamptz) error {
@@ -379,6 +383,28 @@ func (q *Queries) RollPayloadBreakdown(ctx context.Context, hour pgtype.Timestam
return err
}
const rollScopeNodes = `-- name: RollScopeNodes :exec
INSERT INTO analytics_hourly_scope_nodes (hour, iata, scope_id, origin_pubkey)
SELECT DISTINCT $1::timestamptz, iata, scope_id, origin_pubkey
FROM rollup_obs WHERE payload_type = 4 AND origin_pubkey IS NOT NULL AND scope_id IS NOT NULL
`
func (q *Queries) RollScopeNodes(ctx context.Context, hour pgtype.Timestamptz) error {
_, err := q.db.Exec(ctx, rollScopeNodes, hour)
return err
}
const rollScopeObservers = `-- name: RollScopeObservers :exec
INSERT INTO analytics_hourly_scope_observers (hour, iata, scope_id, observer_id)
SELECT DISTINCT $1::timestamptz, iata, scope_id, observer_id
FROM rollup_obs WHERE scope_id IS NOT NULL
`
func (q *Queries) RollScopeObservers(ctx context.Context, hour pgtype.Timestamptz) error {
_, err := q.db.Exec(ctx, rollScopeObservers, hour)
return err
}
const rollScopeSets = `-- name: RollScopeSets :exec
INSERT INTO analytics_hourly_scope_sets (hour, iatas, scope_id, packets)
SELECT $1::timestamptz, iatas, scope_id, count(*)
@@ -451,7 +477,9 @@ SELECT md5(concat_ws('|',
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_advert_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_talker_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_packet_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = $1::timestamptz)
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_observers t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_nodes t WHERE t.hour = $1::timestamptz)
))::text AS content_hash
`
+1 -1
View File
@@ -236,7 +236,7 @@ func (s *Store) GetScopeStats(ctx context.Context, iatas []string, since time.Ti
}
byScope := make(map[string][]api.ScopeHour)
for _, h := range hours {
byScope[h.Name] = append(byScope[h.Name], api.ScopeHour{Hour: h.Hour.Time.UnixMilli(), Packets: h.Packets})
byScope[h.Name] = append(byScope[h.Name], api.ScopeHour{Hour: h.Hour.Time.UnixMilli(), Packets: h.Packets, Observers: h.Observers, Nodes: h.Nodes})
}
items := make([]api.ScopeStats, 0, len(rows))
for _, r := range rows {
+9 -1
View File
@@ -2798,7 +2798,7 @@ const docTemplate = `{
},
"/stats/scopes": {
"get": {
"description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour (zero hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.",
"description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour and adds the distinct observers and advertising nodes active in the scope each hour (empty hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.",
"produces": [
"application/json"
],
@@ -5179,6 +5179,14 @@ const docTemplate = `{
"description": "epoch ms, start of the UTC hour",
"type": "integer"
},
"nodes": {
"description": "distinct nodes whose adverts were heard in the scope that hour",
"type": "integer"
},
"observers": {
"description": "distinct observers that heard the scope's packets that hour",
"type": "integer"
},
"packets": {
"type": "integer"
}
+1
View File
@@ -48,6 +48,7 @@ sub-hour observer activity.
| Observations, receptions, payload counts | Sum |
| Unique packets, adverts, messages, scoped packets | Sum of hourly IATA-set counts: each is counted once per hour it was heard, so one heard across an hour boundary counts twice (measured +1.1%) |
| Active observers, IATAs, scopes | Distinct across the window |
| Active scope observers and nodes (per scope, per hour) | Distinct within the hour, across the requested IATAs |
| Max path entries | Max |
| SNR / RSSI averages | Σ sum / Σ samples, never an average of averages |
+9 -1
View File
@@ -2796,7 +2796,7 @@
},
"/stats/scopes": {
"get": {
"description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour (zero hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.",
"description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour and adds the distinct observers and advertising nodes active in the scope each hour (empty hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.",
"produces": [
"application/json"
],
@@ -5177,6 +5177,14 @@
"description": "epoch ms, start of the UTC hour",
"type": "integer"
},
"nodes": {
"description": "distinct nodes whose adverts were heard in the scope that hour",
"type": "integer"
},
"observers": {
"description": "distinct observers that heard the scope's packets that hour",
"type": "integer"
},
"packets": {
"type": "integer"
}
+12 -5
View File
@@ -1251,6 +1251,12 @@ definitions:
hour:
description: epoch ms, start of the UTC hour
type: integer
nodes:
description: distinct nodes whose adverts were heard in the scope that hour
type: integer
observers:
description: distinct observers that heard the scope's packets that hour
type: integer
packets:
type: integer
type: object
@@ -3523,11 +3529,12 @@ paths:
description: 'Packets are those heard since the window start, each counted once
per hour however many of the requested IATAs heard it. Observers and nodes
are current memberships: observers filter by the IATA they last reported from,
nodes by their IATA memberships. hourly splits packetCount by UTC hour (zero
hours omitted; outage gaps come from /stats/series hour status). With filters,
lists only manual scopes configured for a matching region and imported scopes
whose current MeshMapper catalogue includes a matching IATA; those remain
listed with zero counts. An empty region returns an empty array.'
nodes by their IATA memberships. hourly splits packetCount by UTC hour and
adds the distinct observers and advertising nodes active in the scope each
hour (empty hours omitted; outage gaps come from /stats/series hour status).
With filters, lists only manual scopes configured for a matching region and
imported scopes whose current MeshMapper catalogue includes a matching IATA;
those remain listed with zero counts. An empty region returns an empty array.'
parameters:
- description: Comma-separated IATA codes
in: query
+1 -1
View File
@@ -457,7 +457,7 @@ func getStatsRadioPresets(reader api.Reader) http.HandlerFunc {
// getStatsScopes godoc
//
// @Summary Scope statistics
// @Description Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour (zero hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.
// @Description Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour and adds the distinct observers and advertising nodes active in the scope each hour (empty hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.
// @Tags Stats
// @Produce json
// @Param iatas query string false "Comma-separated IATA codes"
+5 -3
View File
@@ -99,10 +99,12 @@ type ScopeStats struct {
Hourly []ScopeHour `json:"hourly"` // packetCount by hour, oldest first; hours with none omitted
}
// ScopeHour is one UTC hour of a scope's packets; a scope's hours sum to its packetCount.
// ScopeHour is one UTC hour of a scope's activity; a scope's hours' packets sum to its packetCount.
type ScopeHour struct {
Hour int64 `json:"hour"` // epoch ms, start of the UTC hour
Packets int64 `json:"packets"`
Hour int64 `json:"hour"` // epoch ms, start of the UTC hour
Packets int64 `json:"packets"`
Observers int64 `json:"observers"` // distinct observers that heard the scope's packets that hour
Nodes int64 `json:"nodes"` // distinct nodes whose adverts were heard in the scope that hour
}
// TopNode is a node ranked by how often its adverts were heard in the window.
+2 -2
View File
@@ -158,8 +158,8 @@ func (cr *CachedReader) GetStatsTopTalkers(ctx context.Context, iatas []string,
// GetScopeStats implements [api.Reader].
func (cr *CachedReader) GetScopeStats(ctx context.Context, iatas []string, since time.Time) ([]api.ScopeStats, error) {
// ":h" keeps entries cached before hourly existed from being served without it.
return cachedRollup(ctx, cr, keyScopeStats+":h:", iatas, func(ctx context.Context) ([]api.ScopeStats, error) {
// The suffix changes with the hourly shape so older cached entries aren't served.
return cachedRollup(ctx, cr, keyScopeStats+":h2:", iatas, func(ctx context.Context) ([]api.ScopeStats, error) {
return cr.inner.GetScopeStats(ctx, iatas, since)
}, hourKey(since))
}