From 22527e2604ff78900cceef01c1744ecbbf4018a2 Mon Sep 17 00:00:00 2001 From: Andrew McConnell Date: Tue, 9 Jun 2026 20:53:49 -0500 Subject: [PATCH] api: derive multicast subscriber/publisher counts from live users The device, metro, facility, and contributor handlers read the on-chain multicast_subscribers_count / multicast_publishers_count fields from dz_devices_current. In production these are frequently stale or 0 even when the device has active multicast subscribers, so DZDs showed 0 subscribers and 0 available capacity (appearing oversubscribed) despite real users (#650). Derive the per-device subscriber and publisher counts from dz_users_current instead, counting activated multicast users with a non-empty subscribers / publishers array (a user that both publishes and subscribes counts in each). The metro and contributor aggregates and the facility rollup now sum these live counts, and the effective-max capacity is floored at the live count. --- api/handlers/contributors.go | 23 +++++++-- api/handlers/contributors_test.go | 73 +++++++++++++++++++++++++++ api/handlers/devices.go | 25 ++++++--- api/handlers/devices_test.go | 82 ++++++++++++++++++++++++++++++ api/handlers/facilities.go | 39 +++++++++----- api/handlers/facilities_test.go | 76 ++++++++++++++++++++++++++++ api/handlers/metros.go | 55 ++++++++++++++------ api/handlers/metros_test.go | 84 +++++++++++++++++++++++++++++++ 8 files changed, 417 insertions(+), 40 deletions(-) create mode 100644 api/handlers/contributors_test.go create mode 100644 api/handlers/facilities_test.go diff --git a/api/handlers/contributors.go b/api/handlers/contributors.go index b01b8cd01..88e13e54e 100644 --- a/api/handlers/contributors.go +++ b/api/handlers/contributors.go @@ -284,6 +284,18 @@ func (a *API) GetContributor(w http.ResponseWriter, r *http.Request) { WHERE u.status = 'activated' AND d.contributor_pk = ? GROUP BY d.contributor_pk ), + -- Live per-device multicast subscriber/publisher counts from attached users. + -- On-chain multicast_subscribers_count / multicast_publishers_count on + -- dz_devices_current are frequently stale or 0 even with users attached (#650). + device_multicast_live AS ( + SELECT + device_pk, + countIf(JSONLength(subscribers) > 0) as live_subscribers, + countIf(JSONLength(publishers) > 0) as live_publishers + FROM dz_users_current + WHERE status = 'activated' AND kind = 'multicast' + GROUP BY device_pk + ), onchain_user_counts AS ( SELECT contributor_pk, @@ -299,10 +311,10 @@ func (a *API) GetContributor(w http.ResponseWriter, r *http.Request) { SUM(raw_max_pubs) as raw_max_multicast_publishers FROM ( SELECT - contributor_pk, + d.contributor_pk as contributor_pk, unicast_users_count, - multicast_subscribers_count, - multicast_publishers_count, + toUInt64(COALESCE(dml.live_subscribers, 0)) as multicast_subscribers_count, + toUInt64(COALESCE(dml.live_publishers, 0)) as multicast_publishers_count, max_users, toUInt64(max_unicast_users) as raw_max_unicast, toUInt64(max_multicast_subscribers) as raw_max_subs, @@ -310,8 +322,9 @@ func (a *API) GetContributor(w http.ResponseWriter, r *http.Request) { if(max_unicast_users > 0, toUInt64(max_unicast_users), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_multicast_subscribers) - toInt64(max_multicast_publishers)))) as eff_max_unicast, if(max_multicast_subscribers > 0, toUInt64(max_multicast_subscribers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_publishers)))) as eff_max_subs, if(max_multicast_publishers > 0, toUInt64(max_multicast_publishers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_subscribers)))) as eff_max_pubs - FROM dz_devices_current - WHERE contributor_pk IS NOT NULL + FROM dz_devices_current d + LEFT JOIN device_multicast_live dml ON dml.device_pk = d.pk + WHERE d.contributor_pk IS NOT NULL ) GROUP BY contributor_pk ), diff --git a/api/handlers/contributors_test.go b/api/handlers/contributors_test.go new file mode 100644 index 000000000..45cf3631a --- /dev/null +++ b/api/handlers/contributors_test.go @@ -0,0 +1,73 @@ +package handlers_test + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/malbeclabs/lake/api/handlers" + apitesting "github.com/malbeclabs/lake/api/testing" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// insertContributorMulticastData inserts a contributor with one device whose +// on-chain multicast counts are left at 0, plus activated multicast users that +// subscribe to / publish to groups. The contributor detail must report the live +// per-device counts, not the stale on-chain zeros (#650). +func insertContributorMulticastData(t *testing.T, api *handlers.API) { + ctx := t.Context() + + err := api.DB.Exec(ctx, ` + INSERT INTO dim_dz_contributors_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, code, name) VALUES + ('c-mc', now(), now(), generateUUIDv4(), 0, 1, 'c-mc', 'CMC', 'Multicast Contributor') + `) + require.NoError(t, err) + + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_devices_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, code, status, device_type, contributor_pk, metro_pk, public_ip, max_users) VALUES + ('cdev-1', now(), now(), generateUUIDv4(), 0, 1, 'cdev-1', 'C-DEV-01', 'up', 'switch', 'c-mc', '', '10.8.0.1', 100) + `) + require.NoError(t, err) + + // 3 subscriber-only, 2 publisher-only, 1 both-roles, 1 pending (excluded). + // Live totals: subscribers 4, publishers 3. + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_users_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, status, device_pk, kind, owner_pubkey, client_ip, dz_ip, tunnel_id, publishers, subscribers) VALUES + ('c-s1', now(), now(), generateUUIDv4(), 0, 1, 'c-s1', 'activated', 'cdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('c-s2', now(), now(), generateUUIDv4(), 0, 2, 'c-s2', 'activated', 'cdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('c-s3', now(), now(), generateUUIDv4(), 0, 3, 'c-s3', 'activated', 'cdev-1', 'multicast', 'p', '', '', 0, '[]', '["g2"]'), + ('c-p1', now(), now(), generateUUIDv4(), 0, 4, 'c-p1', 'activated', 'cdev-1', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('c-p2', now(), now(), generateUUIDv4(), 0, 5, 'c-p2', 'activated', 'cdev-1', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('c-b1', now(), now(), generateUUIDv4(), 0, 6, 'c-b1', 'activated', 'cdev-1', 'multicast', 'p', '', '', 0, '["g2"]', '["g2"]'), + ('c-pend', now(), now(), generateUUIDv4(), 0, 7, 'c-pend', 'pending', 'cdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]') + `) + require.NoError(t, err) +} + +func TestGetContributor_MulticastCountsFromLiveUsers(t *testing.T) { + t.Parallel() + api := apitesting.NewTestAPI(t, testChDB) + + insertContributorMulticastData(t, api) + + req := httptest.NewRequest(http.MethodGet, "/api/dz/contributors/c-mc", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("pk", "c-mc") + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + + rr := httptest.NewRecorder() + api.GetContributor(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + var contributor handlers.ContributorDetail + require.NoError(t, json.NewDecoder(rr.Body).Decode(&contributor)) + assert.Equal(t, uint64(4), contributor.MulticastSubscribersCount) + assert.Equal(t, uint64(3), contributor.MulticastPublishersCount) +} diff --git a/api/handlers/devices.go b/api/handlers/devices.go index e54c155b5..fa6040b1a 100644 --- a/api/handlers/devices.go +++ b/api/handlers/devices.go @@ -110,7 +110,11 @@ func (a *API) GetDevices(w http.ResponseWriter, r *http.Request) { GROUP BY device_pk ), multicast_counts AS ( - SELECT device_pk, count(*) as user_count + SELECT + device_pk, + count(*) as user_count, + countIf(JSONLength(subscribers) > 0) as subscriber_count, + countIf(JSONLength(publishers) > 0) as publisher_count FROM dz_users_current WHERE status = 'activated' AND kind = 'multicast' GROUP BY device_pk @@ -158,9 +162,12 @@ func (a *API) GetDevices(w http.ResponseWriter, r *http.Request) { COALESCE(d.max_multicast_subscribers, 0) as max_multicast_subscribers, COALESCE(d.max_multicast_publishers, 0) as max_multicast_publishers, COALESCE(d.unicast_users_count, 0) as unicast_users_count, - COALESCE(d.multicast_subscribers_count, 0) as multicast_subscribers_count, + -- Live subscriber/publisher counts from attached users. The on-chain + -- d.multicast_subscribers_count / d.multicast_publishers_count fields are + -- frequently stale or 0 even when users are attached (#650). + toUInt16(COALESCE(ucm.subscriber_count, 0)) as multicast_subscribers_count, COALESCE(d.reserved_seats, 0) as reserved_seats, - COALESCE(d.multicast_publishers_count, 0) as multicast_publishers_count, + toUInt16(COALESCE(ucm.publisher_count, 0)) as multicast_publishers_count, COALESCE(tr.in_bps, 0) as in_bps, COALESCE(tr.out_bps, 0) as out_bps, COALESCE(pr.peak_in_bps, 0) as peak_in_bps, @@ -360,7 +367,11 @@ func (a *API) GetDevice(w http.ResponseWriter, r *http.Request) { GROUP BY device_pk ), multicast_counts AS ( - SELECT device_pk, count(*) as user_count + SELECT + device_pk, + count(*) as user_count, + countIf(JSONLength(subscribers) > 0) as subscriber_count, + countIf(JSONLength(publishers) > 0) as publisher_count FROM dz_users_current WHERE status = 'activated' AND kind = 'multicast' AND device_pk = ? GROUP BY device_pk @@ -408,9 +419,11 @@ func (a *API) GetDevice(w http.ResponseWriter, r *http.Request) { COALESCE(d.max_multicast_subscribers, 0) as max_multicast_subscribers, COALESCE(d.max_multicast_publishers, 0) as max_multicast_publishers, COALESCE(d.unicast_users_count, 0) as unicast_users_count, - COALESCE(d.multicast_subscribers_count, 0) as multicast_subscribers_count, + -- Live subscriber/publisher counts from attached users; on-chain + -- counts are frequently stale or 0 even with users attached (#650). + toUInt16(COALESCE(ucm.subscriber_count, 0)) as multicast_subscribers_count, COALESCE(d.reserved_seats, 0) as reserved_seats, - COALESCE(d.multicast_publishers_count, 0) as multicast_publishers_count, + toUInt16(COALESCE(ucm.publisher_count, 0)) as multicast_publishers_count, COALESCE(tr.in_bps, 0) as in_bps, COALESCE(tr.out_bps, 0) as out_bps, COALESCE(pr.peak_in_bps, 0) as peak_in_bps, diff --git a/api/handlers/devices_test.go b/api/handlers/devices_test.go index d92f039c4..2f5cd6f8b 100644 --- a/api/handlers/devices_test.go +++ b/api/handlers/devices_test.go @@ -383,3 +383,85 @@ func TestGetDevice_HandlesNullContributor(t *testing.T) { assert.Equal(t, "", device.ContributorPK) assert.Equal(t, "", device.ContributorCode) } + +// insertMulticastCountTestData inserts a device whose on-chain multicast +// subscriber/publisher counts are left at their default of 0 (the stale state +// seen in production), plus activated multicast users that actually subscribe +// to and/or publish to groups. The handler must report the live counts derived +// from these users, not the stale on-chain zeros (#650). +func insertMulticastCountTestData(t *testing.T, api *handlers.API) { + ctx := t.Context() + + err := api.DB.Exec(ctx, ` + INSERT INTO dim_dz_devices_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, code, status, device_type, contributor_pk, metro_pk, public_ip, max_users) VALUES + ('dev-mc', now(), now(), generateUUIDv4(), 0, 1, 'dev-mc', 'MC-DEVICE-01', 'activated', 'switch', '', '', '10.0.9.1', 100) + `) + require.NoError(t, err) + + // 3 subscriber-only, 2 publisher-only, 1 both-roles, 1 pending (must be excluded). + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_users_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, status, device_pk, kind, owner_pubkey, client_ip, dz_ip, tunnel_id, publishers, subscribers) VALUES + ('mc-s1', now(), now(), generateUUIDv4(), 0, 1, 'mc-s1', 'activated', 'dev-mc', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('mc-s2', now(), now(), generateUUIDv4(), 0, 2, 'mc-s2', 'activated', 'dev-mc', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('mc-s3', now(), now(), generateUUIDv4(), 0, 3, 'mc-s3', 'activated', 'dev-mc', 'multicast', 'p', '', '', 0, '[]', '["g2"]'), + ('mc-p1', now(), now(), generateUUIDv4(), 0, 4, 'mc-p1', 'activated', 'dev-mc', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('mc-p2', now(), now(), generateUUIDv4(), 0, 5, 'mc-p2', 'activated', 'dev-mc', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('mc-b1', now(), now(), generateUUIDv4(), 0, 6, 'mc-b1', 'activated', 'dev-mc', 'multicast', 'p', '', '', 0, '["g2"]', '["g2"]'), + ('mc-pend', now(), now(), generateUUIDv4(), 0, 7, 'mc-pend', 'pending', 'dev-mc', 'multicast', 'p', '', '', 0, '[]', '["g1"]') + `) + require.NoError(t, err) +} + +func TestGetDevices_MulticastCountsFromLiveUsers(t *testing.T) { + t.Parallel() + api := apitesting.NewTestAPI(t, testChDB) + + insertMulticastCountTestData(t, api) + + req := httptest.NewRequest(http.MethodGet, "/api/dz/devices", nil) + rr := httptest.NewRecorder() + api.GetDevices(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + var response handlers.PaginatedResponse[handlers.DeviceListItem] + require.NoError(t, json.NewDecoder(rr.Body).Decode(&response)) + + var dev *handlers.DeviceListItem + for i := range response.Items { + if response.Items[i].Code == "MC-DEVICE-01" { + dev = &response.Items[i] + break + } + } + require.NotNil(t, dev) + // On-chain counts are 0, but live users give: 6 multicast users, + // 4 subscribers (mc-s1, mc-s2, mc-s3, mc-b1), 3 publishers (mc-p1, mc-p2, mc-b1). + assert.Equal(t, uint64(6), dev.MulticastUsers) + assert.Equal(t, uint16(4), dev.MulticastSubscribersCount) + assert.Equal(t, uint16(3), dev.MulticastPublishersCount) +} + +func TestGetDevice_MulticastCountsFromLiveUsers(t *testing.T) { + t.Parallel() + api := apitesting.NewTestAPI(t, testChDB) + + insertMulticastCountTestData(t, api) + + req := httptest.NewRequest(http.MethodGet, "/api/dz/devices/dev-mc", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("pk", "dev-mc") + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + + rr := httptest.NewRecorder() + api.GetDevice(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + var device handlers.DeviceDetail + require.NoError(t, json.NewDecoder(rr.Body).Decode(&device)) + + assert.Equal(t, uint64(6), device.MulticastUsers) + assert.Equal(t, uint16(4), device.MulticastSubscribersCount) + assert.Equal(t, uint16(3), device.MulticastPublishersCount) +} diff --git a/api/handlers/facilities.go b/api/handlers/facilities.go index e33a81aef..1643656d4 100644 --- a/api/handlers/facilities.go +++ b/api/handlers/facilities.go @@ -80,20 +80,33 @@ var facilityFilterFields = map[string]FilterFieldConfig{ // facilitiesEnrichedCTE joins facilities with devices (via location_pk), metros, and users. // Requires migration 20260421000002 to add location_pk to dz_devices_current. const facilitiesEnrichedCTE = ` - WITH facility_device_stats AS ( + WITH device_multicast_live AS ( + -- Live per-device multicast subscriber/publisher counts from attached users. + -- On-chain multicast_subscribers_count / multicast_publishers_count on + -- dz_devices_current are frequently stale or 0 even with users attached (#650). SELECT - location_pk, - toUInt32(countDistinct(pk)) AS device_count, - SUM(unicast_users_count) AS unicast_users_count, - SUM(multicast_subscribers_count) AS multicast_subscribers_count, - SUM(multicast_publishers_count) AS multicast_publishers_count, - SUM(max_users) AS max_users, - SUM(max_unicast_users) AS max_unicast_users, - SUM(max_multicast_subscribers) AS max_multicast_subscribers, - SUM(max_multicast_publishers) AS max_multicast_publishers - FROM dz_devices_current - WHERE location_pk != '' - GROUP BY location_pk + device_pk, + countIf(JSONLength(subscribers) > 0) AS live_subscribers, + countIf(JSONLength(publishers) > 0) AS live_publishers + FROM dz_users_current + WHERE status = 'activated' AND kind = 'multicast' + GROUP BY device_pk + ), + facility_device_stats AS ( + SELECT + d.location_pk AS location_pk, + toUInt32(countDistinct(d.pk)) AS device_count, + SUM(d.unicast_users_count) AS unicast_users_count, + SUM(COALESCE(dml.live_subscribers, 0)) AS multicast_subscribers_count, + SUM(COALESCE(dml.live_publishers, 0)) AS multicast_publishers_count, + SUM(d.max_users) AS max_users, + SUM(d.max_unicast_users) AS max_unicast_users, + SUM(d.max_multicast_subscribers) AS max_multicast_subscribers, + SUM(d.max_multicast_publishers) AS max_multicast_publishers + FROM dz_devices_current d + LEFT JOIN device_multicast_live dml ON dml.device_pk = d.pk + WHERE d.location_pk != '' + GROUP BY d.location_pk ), facility_user_counts AS ( SELECT d.location_pk, toUInt32(countDistinct(u.pk)) AS user_count diff --git a/api/handlers/facilities_test.go b/api/handlers/facilities_test.go new file mode 100644 index 000000000..917ac42b3 --- /dev/null +++ b/api/handlers/facilities_test.go @@ -0,0 +1,76 @@ +package handlers_test + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/malbeclabs/lake/api/handlers" + apitesting "github.com/malbeclabs/lake/api/testing" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// insertFacilityMulticastData inserts a facility with one device (via location_pk) +// whose on-chain multicast counts are left at 0, plus activated multicast users +// that subscribe to / publish to groups. The facility aggregate must report the +// live per-device counts, not the stale on-chain zeros (#650). +func insertFacilityMulticastData(t *testing.T, api *handlers.API) { + ctx := t.Context() + + err := api.DB.Exec(ctx, ` + INSERT INTO dim_dz_facilities_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, owner, lat, lng, loc_id, status, code, name, country, reference_count) VALUES + ('fac-mc', now(), now(), generateUUIDv4(), 0, 1, 'fac-mc', '', 40.0, -74.0, 1, 'activated', 'FAC-MC', 'Multicast Facility', 'US', 1) + `) + require.NoError(t, err) + + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_devices_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, code, status, device_type, contributor_pk, metro_pk, location_pk, public_ip, max_users) VALUES + ('fdev-1', now(), now(), generateUUIDv4(), 0, 1, 'fdev-1', 'F-DEV-01', 'up', 'switch', '', '', 'fac-mc', '10.7.0.1', 100) + `) + require.NoError(t, err) + + // 3 subscriber-only, 2 publisher-only, 1 both-roles, 1 pending (excluded). + // Live totals: subscribers 4, publishers 3. + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_users_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, status, device_pk, kind, owner_pubkey, client_ip, dz_ip, tunnel_id, publishers, subscribers) VALUES + ('f-s1', now(), now(), generateUUIDv4(), 0, 1, 'f-s1', 'activated', 'fdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('f-s2', now(), now(), generateUUIDv4(), 0, 2, 'f-s2', 'activated', 'fdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('f-s3', now(), now(), generateUUIDv4(), 0, 3, 'f-s3', 'activated', 'fdev-1', 'multicast', 'p', '', '', 0, '[]', '["g2"]'), + ('f-p1', now(), now(), generateUUIDv4(), 0, 4, 'f-p1', 'activated', 'fdev-1', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('f-p2', now(), now(), generateUUIDv4(), 0, 5, 'f-p2', 'activated', 'fdev-1', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('f-b1', now(), now(), generateUUIDv4(), 0, 6, 'f-b1', 'activated', 'fdev-1', 'multicast', 'p', '', '', 0, '["g2"]', '["g2"]'), + ('f-pend', now(), now(), generateUUIDv4(), 0, 7, 'f-pend', 'pending', 'fdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]') + `) + require.NoError(t, err) +} + +func TestGetFacilities_MulticastCountsFromLiveUsers(t *testing.T) { + t.Parallel() + api := apitesting.NewTestAPI(t, testChDB) + + insertFacilityMulticastData(t, api) + + req := httptest.NewRequest(http.MethodGet, "/api/dz/facilities", nil) + rr := httptest.NewRecorder() + api.GetFacilities(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + var response handlers.PaginatedResponse[handlers.FacilityListItem] + require.NoError(t, json.NewDecoder(rr.Body).Decode(&response)) + + var f *handlers.FacilityListItem + for i := range response.Items { + if response.Items[i].PK == "fac-mc" { + f = &response.Items[i] + break + } + } + require.NotNil(t, f) + assert.Equal(t, uint64(4), f.MulticastSubscribersCount) + assert.Equal(t, uint64(3), f.MulticastPublishersCount) +} diff --git a/api/handlers/metros.go b/api/handlers/metros.go index 583dc7f72..8c9d97e34 100644 --- a/api/handlers/metros.go +++ b/api/handlers/metros.go @@ -81,7 +81,19 @@ func (a *API) GetMetros(w http.ResponseWriter, r *http.Request) { WHERE 1=1` const onchainUserCountsCTE = ` - -- NOTE: keep in sync with the identical CTE in GetMetro below + -- NOTE: keep in sync with the identical CTEs in GetMetro below + -- Live per-device multicast subscriber/publisher counts from attached users. + -- On-chain multicast_subscribers_count / multicast_publishers_count on + -- dz_devices_current are frequently stale or 0 even with users attached (#650). + device_multicast_live AS ( + SELECT + device_pk, + countIf(JSONLength(subscribers) > 0) as live_subscribers, + countIf(JSONLength(publishers) > 0) as live_publishers + FROM dz_users_current + WHERE status = 'activated' AND kind = 'multicast' + GROUP BY device_pk + ), onchain_user_counts AS ( SELECT metro_pk, @@ -97,19 +109,20 @@ func (a *API) GetMetros(w http.ResponseWriter, r *http.Request) { SUM(raw_max_pubs) as raw_max_multicast_publishers FROM ( SELECT - metro_pk, + d.metro_pk as metro_pk, unicast_users_count, - multicast_subscribers_count, - multicast_publishers_count, + toUInt64(COALESCE(dml.live_subscribers, 0)) as multicast_subscribers_count, + toUInt64(COALESCE(dml.live_publishers, 0)) as multicast_publishers_count, max_users, toUInt64(max_unicast_users) as raw_max_unicast, toUInt64(max_multicast_subscribers) as raw_max_subs, toUInt64(max_multicast_publishers) as raw_max_pubs, greatest(if(max_unicast_users > 0, toUInt64(max_unicast_users), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_multicast_subscribers) - toInt64(max_multicast_publishers)))), unicast_users_count) as eff_max_unicast, - greatest(if(max_multicast_subscribers > 0, toUInt64(max_multicast_subscribers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_publishers)))), multicast_subscribers_count) as eff_max_subs, - greatest(if(max_multicast_publishers > 0, toUInt64(max_multicast_publishers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_subscribers)))), multicast_publishers_count) as eff_max_pubs - FROM dz_devices_current - WHERE metro_pk IS NOT NULL + greatest(if(max_multicast_subscribers > 0, toUInt64(max_multicast_subscribers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_publishers)))), toUInt64(COALESCE(dml.live_subscribers, 0))) as eff_max_subs, + greatest(if(max_multicast_publishers > 0, toUInt64(max_multicast_publishers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_subscribers)))), toUInt64(COALESCE(dml.live_publishers, 0))) as eff_max_pubs + FROM dz_devices_current d + LEFT JOIN device_multicast_live dml ON dml.device_pk = d.pk + WHERE d.metro_pk IS NOT NULL ) GROUP BY metro_pk ),` @@ -366,7 +379,16 @@ func (a *API) GetMetro(w http.ResponseWriter, r *http.Request) { WHERE u.status = 'activated' AND d.metro_pk = ? GROUP BY d.metro_pk ), - -- NOTE: keep in sync with the identical CTE in GetMetros above + -- NOTE: keep in sync with the identical CTEs in GetMetros above + device_multicast_live AS ( + SELECT + device_pk, + countIf(JSONLength(subscribers) > 0) as live_subscribers, + countIf(JSONLength(publishers) > 0) as live_publishers + FROM dz_users_current + WHERE status = 'activated' AND kind = 'multicast' + GROUP BY device_pk + ), onchain_user_counts AS ( SELECT metro_pk, @@ -382,19 +404,20 @@ func (a *API) GetMetro(w http.ResponseWriter, r *http.Request) { SUM(raw_max_pubs) as raw_max_multicast_publishers FROM ( SELECT - metro_pk, + d.metro_pk as metro_pk, unicast_users_count, - multicast_subscribers_count, - multicast_publishers_count, + toUInt64(COALESCE(dml.live_subscribers, 0)) as multicast_subscribers_count, + toUInt64(COALESCE(dml.live_publishers, 0)) as multicast_publishers_count, max_users, toUInt64(max_unicast_users) as raw_max_unicast, toUInt64(max_multicast_subscribers) as raw_max_subs, toUInt64(max_multicast_publishers) as raw_max_pubs, greatest(if(max_unicast_users > 0, toUInt64(max_unicast_users), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_multicast_subscribers) - toInt64(max_multicast_publishers)))), unicast_users_count) as eff_max_unicast, - greatest(if(max_multicast_subscribers > 0, toUInt64(max_multicast_subscribers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_publishers)))), multicast_subscribers_count) as eff_max_subs, - greatest(if(max_multicast_publishers > 0, toUInt64(max_multicast_publishers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_subscribers)))), multicast_publishers_count) as eff_max_pubs - FROM dz_devices_current - WHERE metro_pk IS NOT NULL + greatest(if(max_multicast_subscribers > 0, toUInt64(max_multicast_subscribers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_publishers)))), toUInt64(COALESCE(dml.live_subscribers, 0))) as eff_max_subs, + greatest(if(max_multicast_publishers > 0, toUInt64(max_multicast_publishers), toUInt64(greatest(0, toInt64(max_users) - toInt64(max_unicast_users) - toInt64(max_multicast_subscribers)))), toUInt64(COALESCE(dml.live_publishers, 0))) as eff_max_pubs + FROM dz_devices_current d + LEFT JOIN device_multicast_live dml ON dml.device_pk = d.pk + WHERE d.metro_pk IS NOT NULL ) GROUP BY metro_pk ), diff --git a/api/handlers/metros_test.go b/api/handlers/metros_test.go index 3107330ec..343e6d727 100644 --- a/api/handlers/metros_test.go +++ b/api/handlers/metros_test.go @@ -267,3 +267,87 @@ func TestGetMetro_ReturnsDetails(t *testing.T) { assert.Equal(t, uint64(2), metro.DeviceCount) assert.Equal(t, uint64(2), metro.UserCount) } + +// insertMetroMulticastData inserts one metro with two devices whose on-chain +// multicast counts are left at 0, plus activated multicast users that subscribe +// to / publish to groups. The metro aggregate must sum the live per-device +// counts, not the stale on-chain zeros (#650). +func insertMetroMulticastData(t *testing.T, api *handlers.API) { + ctx := t.Context() + + err := api.DB.Exec(ctx, ` + INSERT INTO dim_dz_metros_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, code, name, latitude, longitude) VALUES + ('metro-mc', now(), now(), generateUUIDv4(), 0, 1, 'metro-mc', 'MCM', 'Multicast Metro', 1.0, 2.0) + `) + require.NoError(t, err) + + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_devices_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, code, status, device_type, metro_pk, public_ip, contributor_pk, max_users) VALUES + ('mcdev-1', now(), now(), generateUUIDv4(), 0, 1, 'mcdev-1', 'MC-DEV-01', 'up', 'switch', 'metro-mc', '10.9.0.1', '', 100), + ('mcdev-2', now(), now(), generateUUIDv4(), 0, 2, 'mcdev-2', 'MC-DEV-02', 'up', 'switch', 'metro-mc', '10.9.0.2', '', 100) + `) + require.NoError(t, err) + + // mcdev-1: 2 subscribers, 1 publisher. mcdev-2: 1 subscriber, 1 both-roles. + // Plus 1 pending (excluded). Metro totals: subscribers 4, publishers 2. + err = api.DB.Exec(ctx, ` + INSERT INTO dim_dz_users_history + (entity_id, snapshot_ts, ingested_at, op_id, is_deleted, attrs_hash, pk, status, device_pk, kind, owner_pubkey, client_ip, dz_ip, tunnel_id, publishers, subscribers) VALUES + ('m-s1', now(), now(), generateUUIDv4(), 0, 1, 'm-s1', 'activated', 'mcdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('m-s2', now(), now(), generateUUIDv4(), 0, 2, 'm-s2', 'activated', 'mcdev-1', 'multicast', 'p', '', '', 0, '[]', '["g1"]'), + ('m-p1', now(), now(), generateUUIDv4(), 0, 3, 'm-p1', 'activated', 'mcdev-1', 'multicast', 'p', '', '', 0, '["g1"]', '[]'), + ('m-s3', now(), now(), generateUUIDv4(), 0, 4, 'm-s3', 'activated', 'mcdev-2', 'multicast', 'p', '', '', 0, '[]', '["g2"]'), + ('m-b1', now(), now(), generateUUIDv4(), 0, 5, 'm-b1', 'activated', 'mcdev-2', 'multicast', 'p', '', '', 0, '["g2"]', '["g2"]'), + ('m-pend', now(), now(), generateUUIDv4(), 0, 6, 'm-pend', 'pending', 'mcdev-2', 'multicast', 'p', '', '', 0, '["g2"]', '["g2"]') + `) + require.NoError(t, err) +} + +func TestGetMetros_MulticastCountsFromLiveUsers(t *testing.T) { + t.Parallel() + api := apitesting.NewTestAPI(t, testChDB) + + insertMetroMulticastData(t, api) + + req := httptest.NewRequest(http.MethodGet, "/api/dz/metros", nil) + rr := httptest.NewRecorder() + api.GetMetros(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + var response handlers.PaginatedResponse[handlers.MetroListItem] + require.NoError(t, json.NewDecoder(rr.Body).Decode(&response)) + + var m *handlers.MetroListItem + for i := range response.Items { + if response.Items[i].PK == "metro-mc" { + m = &response.Items[i] + break + } + } + require.NotNil(t, m) + assert.Equal(t, uint64(4), m.MulticastSubscribersCount) + assert.Equal(t, uint64(2), m.MulticastPublishersCount) +} + +func TestGetMetro_MulticastCountsFromLiveUsers(t *testing.T) { + t.Parallel() + api := apitesting.NewTestAPI(t, testChDB) + + insertMetroMulticastData(t, api) + + req := httptest.NewRequest(http.MethodGet, "/api/dz/metros/metro-mc", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("pk", "metro-mc") + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + + rr := httptest.NewRecorder() + api.GetMetro(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + var metro handlers.MetroDetail + require.NoError(t, json.NewDecoder(rr.Body).Decode(&metro)) + assert.Equal(t, uint64(4), metro.MulticastSubscribersCount) + assert.Equal(t, uint64(2), metro.MulticastPublishersCount) +}