mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-22 17:41:26 +00:00
feat(meshmapper): import zone groups as regions
This commit is contained in:
1 parent
bfda0d5281
commit
a970484b00
22 files changed
+785
-58
No files matched your search
@@ -201,6 +201,7 @@ even when the same bearer-authenticated request works with curl.
|
||||
# enabled: true
|
||||
# zones:
|
||||
# enabled: true
|
||||
# import_groups: true # MeshMapper zone groups become regions
|
||||
|
||||
# Optional IATA overrides — auto-created on first packet arrival,
|
||||
# only needed if you want to customise display name or coordinates.
|
||||
|
||||
@@ -214,6 +214,7 @@ func main() {
|
||||
zones := meshmapper.NewZones(cfg.MeshMapper.Zones, store, directory)
|
||||
if cr, ok := reader.(*cache.CachedReader); ok {
|
||||
zones.OnChange(cr.InvalidateIATABorder)
|
||||
zones.OnRegionsChange(cr.InvalidateRegions)
|
||||
}
|
||||
if localBorders != nil {
|
||||
zones.OnUpdate(func(imported map[string]json.RawMessage) {
|
||||
|
||||
@@ -70,8 +70,15 @@ ratelimit:
|
||||
# A missing or failed boundary keeps the last good import, or the borderFile when
|
||||
# there is none. Disabling drops every import on restart. Logged under
|
||||
# component=meshmapper.zones.
|
||||
# import_groups turns MeshMapper zone groups (e.g. ONQC: YOW, YQB, YUL) into
|
||||
# regions, from the zone lists already fetched. A group is imported when at least
|
||||
# one member is a known IATA, with its full member list; slug = lowercased group
|
||||
# code, listed after the regions below. A region below with the same slug wins.
|
||||
# Imported regions are removed when the group disappears or stops qualifying,
|
||||
# and all of them when import_groups is turned off. Requires zones.enabled.
|
||||
# zones:
|
||||
# enabled: false
|
||||
# import_groups: false
|
||||
# refresh_interval: 24h # 1h-168h; one request per 15s tick, ETag-conditional
|
||||
|
||||
iatas:
|
||||
@@ -104,6 +111,8 @@ iatas:
|
||||
lat: 43.6777
|
||||
lng: -79.6248
|
||||
|
||||
# Hand-written regions; optional with meshmapper.zones.import_groups. Seeding
|
||||
# sets each region's members to exactly this list.
|
||||
regions:
|
||||
- slug: western-canada
|
||||
name: Western Canada
|
||||
|
||||
+9
-5
@@ -166,9 +166,13 @@ func (s *Store) GetRegionBySlug(ctx context.Context, slug string) (*api.Region,
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpsertRegionIATA(ctx context.Context, regionID int32, iata string) error {
|
||||
return s.q.UpsertRegionIATA(ctx, sqlc.UpsertRegionIATAParams{
|
||||
RegionID: regionID,
|
||||
Iata: iata,
|
||||
})
|
||||
// SetRegionIATAs makes iatas the region's exact member list.
|
||||
func (s *Store) SetRegionIATAs(ctx context.Context, regionID int32, iatas []string) error {
|
||||
if iatas == nil {
|
||||
iatas = []string{}
|
||||
}
|
||||
if err := s.q.DeleteRegionIATAsNotIn(ctx, sqlc.DeleteRegionIATAsNotInParams{RegionID: regionID, Keep: iatas}); err != nil {
|
||||
return err
|
||||
}
|
||||
return s.q.AddRegionIATAs(ctx, sqlc.AddRegionIATAsParams{RegionID: regionID, Iatas: iatas})
|
||||
}
|
||||
@@ -113,3 +113,36 @@ func (s *Store) SaveZoneBoundary(ctx context.Context, b meshmapper.Boundary) err
|
||||
}
|
||||
return s.q.SaveZoneBoundary(ctx, params)
|
||||
}
|
||||
|
||||
func (s *Store) ListRegionState(ctx context.Context) ([]meshmapper.RegionState, error) {
|
||||
rows, err := s.q.ListRegionState(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]meshmapper.RegionState, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
out = append(out, meshmapper.RegionState{Slug: row.Slug, Name: row.Name, DisplayOrder: int(row.DisplayOrder),
|
||||
Imported: row.Imported, IATAs: row.Iatas})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SaveImportedRegion reports false when a hand-written region owns the slug.
|
||||
func (s *Store) SaveImportedRegion(ctx context.Context, r meshmapper.RegionState) (bool, error) {
|
||||
order := int32(r.DisplayOrder)
|
||||
id, err := s.q.UpsertImportedRegion(ctx, sqlc.UpsertImportedRegionParams{Slug: r.Slug, Name: r.Name, DisplayOrder: &order})
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, s.SetRegionIATAs(ctx, id, r.IATAs)
|
||||
}
|
||||
|
||||
func (s *Store) PruneImportedRegions(ctx context.Context, keep []string) ([]string, error) {
|
||||
if keep == nil {
|
||||
keep = []string{}
|
||||
}
|
||||
return s.q.PruneImportedRegions(ctx, keep)
|
||||
}
|
||||
@@ -226,3 +226,91 @@ func TestMeshMapperZoneBoundariesPostgres(t *testing.T) {
|
||||
t.Fatal("manual border did not return after pruning")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMeshMapperRegionsPostgres(t *testing.T) {
|
||||
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("set BEACON_TEST_POSTGRES_DSN")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := RunMigrations(ctx, pool); err != nil {
|
||||
pool.Close()
|
||||
t.Fatal(err)
|
||||
}
|
||||
pool.Close()
|
||||
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())
|
||||
for _, table := range []string{"regions", "region_iatas"} {
|
||||
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
store := &Store{q: sqlc.New(tx)}
|
||||
state := func() map[string]meshmapper.RegionState {
|
||||
t.Helper()
|
||||
rows, err := store.ListRegionState(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out := map[string]meshmapper.RegionState{}
|
||||
for _, r := range rows {
|
||||
out[r.Slug] = r
|
||||
}
|
||||
return out
|
||||
}
|
||||
hand, err := store.UpsertRegion(ctx, "east", "East", "", 3, nil, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.SetRegionIATAs(ctx, hand, []string{"YOW", "YUL"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.SetRegionIATAs(ctx, hand, []string{"YOW"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := state()["east"]; got.Imported || got.DisplayOrder != 3 || strings.Join(got.IATAs, ",") != "YOW" {
|
||||
t.Fatal("configured members not replaced", got)
|
||||
}
|
||||
group := meshmapper.RegionState{Slug: "onqc", Name: "Corridor", DisplayOrder: 4, IATAs: []string{"YOW", "YQB"}}
|
||||
if saved, err := store.SaveImportedRegion(ctx, group); err != nil || !saved {
|
||||
t.Fatal(saved, err)
|
||||
}
|
||||
group.IATAs = []string{"YQB", "YUL"}
|
||||
if saved, err := store.SaveImportedRegion(ctx, group); err != nil || !saved {
|
||||
t.Fatal(saved, err)
|
||||
}
|
||||
if got := state()["onqc"]; !got.Imported || strings.Join(got.IATAs, ",") != "YQB,YUL" {
|
||||
t.Fatal("imported region not updated", got)
|
||||
}
|
||||
if saved, err := store.SaveImportedRegion(ctx, meshmapper.RegionState{Slug: "east", Name: "Group", IATAs: []string{"YYZ"}}); err != nil || saved {
|
||||
t.Fatal("imported group replaced a configured region", saved, err)
|
||||
}
|
||||
if got := state()["east"]; got.Name != "East" || strings.Join(got.IATAs, ",") != "YOW" {
|
||||
t.Fatal("configured region changed by a clash", got)
|
||||
}
|
||||
if _, err := store.UpsertRegion(ctx, "onqc", "Mine", "", 1, nil, nil, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if state()["onqc"].Imported {
|
||||
t.Fatal("config did not take over an imported slug")
|
||||
}
|
||||
if _, err := store.SaveImportedRegion(ctx, meshmapper.RegionState{Slug: "golm", Name: "Lakes", IATAs: []string{"YYZ"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if pruned, err := store.PruneImportedRegions(ctx, nil); err != nil || strings.Join(pruned, ",") != "golm" {
|
||||
t.Fatal("prune touched configured regions", pruned, err)
|
||||
}
|
||||
}
|
||||
@@ -911,7 +911,8 @@ CREATE TABLE regions (
|
||||
center_lng double precision,
|
||||
zoom_level integer DEFAULT 8,
|
||||
created_at timestamp with time zone DEFAULT now() NOT NULL,
|
||||
updated_at timestamp with time zone DEFAULT now() NOT NULL
|
||||
updated_at timestamp with time zone DEFAULT now() NOT NULL,
|
||||
imported boolean DEFAULT false NOT NULL
|
||||
);
|
||||
|
||||
CREATE SEQUENCE regions_id_seq
|
||||
|
||||
+29
-2
@@ -1200,14 +1200,41 @@ ON CONFLICT (slug) DO UPDATE SET
|
||||
center_lat = EXCLUDED.center_lat,
|
||||
center_lng = EXCLUDED.center_lng,
|
||||
zoom_level = EXCLUDED.zoom_level,
|
||||
imported = FALSE, -- config owns the slug from now on
|
||||
updated_at = NOW()
|
||||
RETURNING id;
|
||||
|
||||
-- name: UpsertRegionIATA :exec
|
||||
-- name: DeleteRegionIATAsNotIn :exec
|
||||
DELETE FROM region_iatas WHERE region_id = @region_id AND NOT (iata = ANY(@keep::bpchar[]));
|
||||
|
||||
-- name: AddRegionIATAs :exec
|
||||
INSERT INTO region_iatas (region_id, iata)
|
||||
VALUES ($1, $2)
|
||||
SELECT @region_id, unnest(@iatas::bpchar[])
|
||||
ON CONFLICT (region_id, iata) DO NOTHING;
|
||||
|
||||
-- name: ListRegionState :many
|
||||
-- Every region with its members, for reconciling imported MeshMapper groups.
|
||||
SELECT r.slug, r.name, COALESCE(r.display_order, 0)::int AS display_order, r.imported,
|
||||
COALESCE(array_agg(ri.iata::text ORDER BY ri.iata) FILTER (WHERE ri.iata IS NOT NULL), '{}')::text[] AS iatas
|
||||
FROM regions r
|
||||
LEFT JOIN region_iatas ri ON ri.region_id = r.id
|
||||
GROUP BY r.id
|
||||
ORDER BY r.slug;
|
||||
|
||||
-- name: UpsertImportedRegion :one
|
||||
-- A hand-written region owns its slug: the WHERE turns a clash into no row.
|
||||
INSERT INTO regions (slug, name, display_order, zoom_level, imported, updated_at)
|
||||
VALUES (@slug, @name, @display_order, NULL, TRUE, NOW())
|
||||
ON CONFLICT (slug) DO UPDATE SET
|
||||
name = EXCLUDED.name,
|
||||
display_order = EXCLUDED.display_order,
|
||||
updated_at = NOW()
|
||||
WHERE regions.imported
|
||||
RETURNING id;
|
||||
|
||||
-- name: PruneImportedRegions :many
|
||||
DELETE FROM regions WHERE imported AND NOT (slug = ANY(@keep::text[])) RETURNING slug;
|
||||
|
||||
-- ============================================================
|
||||
-- TRACES
|
||||
-- ============================================================
|
||||
|
||||
+73
-14
@@ -44,6 +44,20 @@ func (m *MockQuerier) EXPECT() *MockQuerierMockRecorder {
|
||||
return m.recorder
|
||||
}
|
||||
|
||||
// AddRegionIATAs mocks base method.
|
||||
func (m *MockQuerier) AddRegionIATAs(ctx context.Context, arg db.AddRegionIATAsParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "AddRegionIATAs", ctx, arg)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// AddRegionIATAs indicates an expected call of AddRegionIATAs.
|
||||
func (mr *MockQuerierMockRecorder) AddRegionIATAs(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddRegionIATAs", reflect.TypeOf((*MockQuerier)(nil).AddRegionIATAs), ctx, arg)
|
||||
}
|
||||
|
||||
// AmbiguousPrefixes mocks base method.
|
||||
func (m *MockQuerier) AmbiguousPrefixes(ctx context.Context) ([]db.AmbiguousPrefixesRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -190,6 +204,20 @@ func (mr *MockQuerierMockRecorder) DeleteOldTraceIATAs(ctx, lastHeard any) *gomo
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldTraceIATAs", reflect.TypeOf((*MockQuerier)(nil).DeleteOldTraceIATAs), ctx, lastHeard)
|
||||
}
|
||||
|
||||
// DeleteRegionIATAsNotIn mocks base method.
|
||||
func (m *MockQuerier) DeleteRegionIATAsNotIn(ctx context.Context, arg db.DeleteRegionIATAsNotInParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "DeleteRegionIATAsNotIn", ctx, arg)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// DeleteRegionIATAsNotIn indicates an expected call of DeleteRegionIATAsNotIn.
|
||||
func (mr *MockQuerierMockRecorder) DeleteRegionIATAsNotIn(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteRegionIATAsNotIn", reflect.TypeOf((*MockQuerier)(nil).DeleteRegionIATAsNotIn), ctx, arg)
|
||||
}
|
||||
|
||||
// GetAccount mocks base method.
|
||||
func (m *MockQuerier) GetAccount(ctx context.Context, id uuid.UUID) (db.Account, error) {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1224,6 +1252,21 @@ func (mr *MockQuerierMockRecorder) ListPacketsByIATAs(ctx, arg any) *gomock.Call
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListPacketsByIATAs", reflect.TypeOf((*MockQuerier)(nil).ListPacketsByIATAs), ctx, arg)
|
||||
}
|
||||
|
||||
// ListRegionState mocks base method.
|
||||
func (m *MockQuerier) ListRegionState(ctx context.Context) ([]db.ListRegionStateRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "ListRegionState", ctx)
|
||||
ret0, _ := ret[0].([]db.ListRegionStateRow)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// ListRegionState indicates an expected call of ListRegionState.
|
||||
func (mr *MockQuerierMockRecorder) ListRegionState(ctx any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListRegionState", reflect.TypeOf((*MockQuerier)(nil).ListRegionState), ctx)
|
||||
}
|
||||
|
||||
// ListRegions mocks base method.
|
||||
func (m *MockQuerier) ListRegions(ctx context.Context) ([]db.ListRegionsRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1314,6 +1357,21 @@ func (mr *MockQuerierMockRecorder) ListZoneBoundaries(ctx any) *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListZoneBoundaries", reflect.TypeOf((*MockQuerier)(nil).ListZoneBoundaries), ctx)
|
||||
}
|
||||
|
||||
// PruneImportedRegions mocks base method.
|
||||
func (m *MockQuerier) PruneImportedRegions(ctx context.Context, keep []string) ([]string, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "PruneImportedRegions", ctx, keep)
|
||||
ret0, _ := ret[0].([]string)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// PruneImportedRegions indicates an expected call of PruneImportedRegions.
|
||||
func (mr *MockQuerierMockRecorder) PruneImportedRegions(ctx, keep any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PruneImportedRegions", reflect.TypeOf((*MockQuerier)(nil).PruneImportedRegions), ctx, keep)
|
||||
}
|
||||
|
||||
// PruneZoneBoundaries mocks base method.
|
||||
func (m *MockQuerier) PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error) {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1844,6 +1902,21 @@ func (mr *MockQuerierMockRecorder) UpsertIATADetails(ctx, arg any) *gomock.Call
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpsertIATADetails", reflect.TypeOf((*MockQuerier)(nil).UpsertIATADetails), ctx, arg)
|
||||
}
|
||||
|
||||
// UpsertImportedRegion mocks base method.
|
||||
func (m *MockQuerier) UpsertImportedRegion(ctx context.Context, arg db.UpsertImportedRegionParams) (int32, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "UpsertImportedRegion", ctx, arg)
|
||||
ret0, _ := ret[0].(int32)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// UpsertImportedRegion indicates an expected call of UpsertImportedRegion.
|
||||
func (mr *MockQuerierMockRecorder) UpsertImportedRegion(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpsertImportedRegion", reflect.TypeOf((*MockQuerier)(nil).UpsertImportedRegion), ctx, arg)
|
||||
}
|
||||
|
||||
// UpsertKnownRoute mocks base method.
|
||||
func (m *MockQuerier) UpsertKnownRoute(ctx context.Context, arg db.UpsertKnownRouteParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1988,20 +2061,6 @@ func (mr *MockQuerierMockRecorder) UpsertRegion(ctx, arg any) *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpsertRegion", reflect.TypeOf((*MockQuerier)(nil).UpsertRegion), ctx, arg)
|
||||
}
|
||||
|
||||
// UpsertRegionIATA mocks base method.
|
||||
func (m *MockQuerier) UpsertRegionIATA(ctx context.Context, arg db.UpsertRegionIATAParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "UpsertRegionIATA", ctx, arg)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// UpsertRegionIATA indicates an expected call of UpsertRegionIATA.
|
||||
func (mr *MockQuerierMockRecorder) UpsertRegionIATA(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpsertRegionIATA", reflect.TypeOf((*MockQuerier)(nil).UpsertRegionIATA), ctx, arg)
|
||||
}
|
||||
|
||||
// UpsertTraceIATA mocks base method.
|
||||
func (m *MockQuerier) UpsertTraceIATA(ctx context.Context, arg db.UpsertTraceIATAParams) error {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@@ -529,6 +529,7 @@ type Region struct {
|
||||
ZoomLevel *int32 `json:"zoom_level"`
|
||||
CreatedAt pgtype.Timestamptz `json:"created_at"`
|
||||
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
||||
Imported bool `json:"imported"`
|
||||
}
|
||||
|
||||
type RegionIata struct {
|
||||
|
||||
+7
-1
@@ -13,6 +13,7 @@ import (
|
||||
)
|
||||
|
||||
type Querier interface {
|
||||
AddRegionIATAs(ctx context.Context, arg AddRegionIATAsParams) error
|
||||
// Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run.
|
||||
AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error)
|
||||
CreateAccount(ctx context.Context, name string) (Account, error)
|
||||
@@ -41,6 +42,7 @@ type Querier interface {
|
||||
DeleteOldTelemetry(ctx context.Context, reportedAt pgtype.Timestamptz) error
|
||||
// Keeps the trace IATA filter in step with packet retention.
|
||||
DeleteOldTraceIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error
|
||||
DeleteRegionIATAsNotIn(ctx context.Context, arg DeleteRegionIATAsNotInParams) error
|
||||
GetAccount(ctx context.Context, id uuid.UUID) (Account, error)
|
||||
GetChannelByID(ctx context.Context, id int32) (Channel, error)
|
||||
// Returns neighbors of a node that are in a different IATA.
|
||||
@@ -199,6 +201,8 @@ type Querier interface {
|
||||
// A site that filled scan_depth still has unread history below its floor.
|
||||
// The newest such floor is the point above which every site is covered.
|
||||
ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAsParams) ([]ListPacketsByIATAsRow, error)
|
||||
// Every region with its members, for reconciling imported MeshMapper groups.
|
||||
ListRegionState(ctx context.Context) ([]ListRegionStateRow, error)
|
||||
// ============================================================
|
||||
// REGIONS
|
||||
// ============================================================
|
||||
@@ -221,6 +225,7 @@ type Querier interface {
|
||||
// internal/ingest.BackfillChannelMessages.
|
||||
ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error)
|
||||
ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBoundary, error)
|
||||
PruneImportedRegions(ctx context.Context, keep []string) ([]string, error)
|
||||
// Drops imports for IATAs no longer configured, so their manual border returns.
|
||||
PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error)
|
||||
// Delete node_neighbors where the neighbor has departed from node_short_ids
|
||||
@@ -302,6 +307,8 @@ type Querier interface {
|
||||
// bbox already computed -- see internal/config/border.go.
|
||||
UpsertIATABorder(ctx context.Context, arg UpsertIATABorderParams) error
|
||||
UpsertIATADetails(ctx context.Context, arg UpsertIATADetailsParams) error
|
||||
// A hand-written region owns its slug: the WHERE turns a clash into no row.
|
||||
UpsertImportedRegion(ctx context.Context, arg UpsertImportedRegionParams) (int32, error)
|
||||
// ============================================================
|
||||
// ROUTES
|
||||
// ============================================================
|
||||
@@ -343,7 +350,6 @@ type Querier interface {
|
||||
// ============================================================
|
||||
UpsertPacket(ctx context.Context, arg UpsertPacketParams) (UpsertPacketRow, error)
|
||||
UpsertRegion(ctx context.Context, arg UpsertRegionParams) (int32, error)
|
||||
UpsertRegionIATA(ctx context.Context, arg UpsertRegionIATAParams) error
|
||||
// Refreshes at most hourly so repeat hears don't churn the row.
|
||||
UpsertTraceIATA(ctx context.Context, arg UpsertTraceIATAParams) error
|
||||
// ============================================================
|
||||
|
||||
+124
-16
@@ -13,6 +13,22 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
const addRegionIATAs = `-- name: AddRegionIATAs :exec
|
||||
INSERT INTO region_iatas (region_id, iata)
|
||||
SELECT $1, unnest($2::bpchar[])
|
||||
ON CONFLICT (region_id, iata) DO NOTHING
|
||||
`
|
||||
|
||||
type AddRegionIATAsParams struct {
|
||||
RegionID int32 `json:"region_id"`
|
||||
Iatas []string `json:"iatas"`
|
||||
}
|
||||
|
||||
func (q *Queries) AddRegionIATAs(ctx context.Context, arg AddRegionIATAsParams) error {
|
||||
_, err := q.db.Exec(ctx, addRegionIATAs, arg.RegionID, arg.Iatas)
|
||||
return err
|
||||
}
|
||||
|
||||
const ambiguousPrefixes = `-- name: AmbiguousPrefixes :many
|
||||
SELECT iata::text AS iata, 1::int AS len, prefix_1 AS prefix FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1
|
||||
UNION ALL
|
||||
@@ -235,6 +251,20 @@ func (q *Queries) DeleteOldTraceIATAs(ctx context.Context, lastHeard pgtype.Time
|
||||
return err
|
||||
}
|
||||
|
||||
const deleteRegionIATAsNotIn = `-- name: DeleteRegionIATAsNotIn :exec
|
||||
DELETE FROM region_iatas WHERE region_id = $1 AND NOT (iata = ANY($2::bpchar[]))
|
||||
`
|
||||
|
||||
type DeleteRegionIATAsNotInParams struct {
|
||||
RegionID int32 `json:"region_id"`
|
||||
Keep []string `json:"keep"`
|
||||
}
|
||||
|
||||
func (q *Queries) DeleteRegionIATAsNotIn(ctx context.Context, arg DeleteRegionIATAsNotInParams) error {
|
||||
_, err := q.db.Exec(ctx, deleteRegionIATAsNotIn, arg.RegionID, arg.Keep)
|
||||
return err
|
||||
}
|
||||
|
||||
const getAccount = `-- name: GetAccount :one
|
||||
SELECT id, name, created_at, deactivated_at FROM accounts WHERE id = $1
|
||||
`
|
||||
@@ -3687,6 +3717,50 @@ func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAs
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listRegionState = `-- name: ListRegionState :many
|
||||
SELECT r.slug, r.name, COALESCE(r.display_order, 0)::int AS display_order, r.imported,
|
||||
COALESCE(array_agg(ri.iata::text ORDER BY ri.iata) FILTER (WHERE ri.iata IS NOT NULL), '{}')::text[] AS iatas
|
||||
FROM regions r
|
||||
LEFT JOIN region_iatas ri ON ri.region_id = r.id
|
||||
GROUP BY r.id
|
||||
ORDER BY r.slug
|
||||
`
|
||||
|
||||
type ListRegionStateRow struct {
|
||||
Slug string `json:"slug"`
|
||||
Name string `json:"name"`
|
||||
DisplayOrder int32 `json:"display_order"`
|
||||
Imported bool `json:"imported"`
|
||||
Iatas []string `json:"iatas"`
|
||||
}
|
||||
|
||||
// Every region with its members, for reconciling imported MeshMapper groups.
|
||||
func (q *Queries) ListRegionState(ctx context.Context) ([]ListRegionStateRow, error) {
|
||||
rows, err := q.db.Query(ctx, listRegionState)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []ListRegionStateRow{}
|
||||
for rows.Next() {
|
||||
var i ListRegionStateRow
|
||||
if err := rows.Scan(
|
||||
&i.Slug,
|
||||
&i.Name,
|
||||
&i.DisplayOrder,
|
||||
&i.Imported,
|
||||
&i.Iatas,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listRegions = `-- name: ListRegions :many
|
||||
|
||||
SELECT id, slug, name
|
||||
@@ -3926,6 +4000,30 @@ func (q *Queries) ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBound
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const pruneImportedRegions = `-- name: PruneImportedRegions :many
|
||||
DELETE FROM regions WHERE imported AND NOT (slug = ANY($1::text[])) RETURNING slug
|
||||
`
|
||||
|
||||
func (q *Queries) PruneImportedRegions(ctx context.Context, keep []string) ([]string, error) {
|
||||
rows, err := q.db.Query(ctx, pruneImportedRegions, keep)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []string{}
|
||||
for rows.Next() {
|
||||
var slug string
|
||||
if err := rows.Scan(&slug); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, slug)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const pruneZoneBoundaries = `-- name: PruneZoneBoundaries :many
|
||||
DELETE FROM meshmapper_zone_boundaries WHERE NOT (iata = ANY($1::text[])) RETURNING iata
|
||||
`
|
||||
@@ -4941,6 +5039,31 @@ func (q *Queries) UpsertIATADetails(ctx context.Context, arg UpsertIATADetailsPa
|
||||
return err
|
||||
}
|
||||
|
||||
const upsertImportedRegion = `-- name: UpsertImportedRegion :one
|
||||
INSERT INTO regions (slug, name, display_order, zoom_level, imported, updated_at)
|
||||
VALUES ($1, $2, $3, NULL, TRUE, NOW())
|
||||
ON CONFLICT (slug) DO UPDATE SET
|
||||
name = EXCLUDED.name,
|
||||
display_order = EXCLUDED.display_order,
|
||||
updated_at = NOW()
|
||||
WHERE regions.imported
|
||||
RETURNING id
|
||||
`
|
||||
|
||||
type UpsertImportedRegionParams struct {
|
||||
Slug string `json:"slug"`
|
||||
Name string `json:"name"`
|
||||
DisplayOrder *int32 `json:"display_order"`
|
||||
}
|
||||
|
||||
// A hand-written region owns its slug: the WHERE turns a clash into no row.
|
||||
func (q *Queries) UpsertImportedRegion(ctx context.Context, arg UpsertImportedRegionParams) (int32, error) {
|
||||
row := q.db.QueryRow(ctx, upsertImportedRegion, arg.Slug, arg.Name, arg.DisplayOrder)
|
||||
var id int32
|
||||
err := row.Scan(&id)
|
||||
return id, err
|
||||
}
|
||||
|
||||
const upsertKnownRoute = `-- name: UpsertKnownRoute :exec
|
||||
|
||||
INSERT INTO known_routes (path_key, node_ids, hash_prefix, iata, hop_count)
|
||||
@@ -5325,6 +5448,7 @@ ON CONFLICT (slug) DO UPDATE SET
|
||||
center_lat = EXCLUDED.center_lat,
|
||||
center_lng = EXCLUDED.center_lng,
|
||||
zoom_level = EXCLUDED.zoom_level,
|
||||
imported = FALSE, -- config owns the slug from now on
|
||||
updated_at = NOW()
|
||||
RETURNING id
|
||||
`
|
||||
@@ -5354,22 +5478,6 @@ func (q *Queries) UpsertRegion(ctx context.Context, arg UpsertRegionParams) (int
|
||||
return id, err
|
||||
}
|
||||
|
||||
const upsertRegionIATA = `-- name: UpsertRegionIATA :exec
|
||||
INSERT INTO region_iatas (region_id, iata)
|
||||
VALUES ($1, $2)
|
||||
ON CONFLICT (region_id, iata) DO NOTHING
|
||||
`
|
||||
|
||||
type UpsertRegionIATAParams struct {
|
||||
RegionID int32 `json:"region_id"`
|
||||
Iata string `json:"iata"`
|
||||
}
|
||||
|
||||
func (q *Queries) UpsertRegionIATA(ctx context.Context, arg UpsertRegionIATAParams) error {
|
||||
_, err := q.db.Exec(ctx, upsertRegionIATA, arg.RegionID, arg.Iata)
|
||||
return err
|
||||
}
|
||||
|
||||
const upsertTraceIATA = `-- name: UpsertTraceIATA :exec
|
||||
INSERT INTO trace_iatas (trace_tag, iata, last_heard)
|
||||
VALUES ($1, $2, $3)
|
||||
|
||||
Vendored
+6
@@ -90,6 +90,12 @@ func (cr *CachedReader) InvalidateIATABorder(ctx context.Context, iata string) {
|
||||
cr.c.del(ctx, keyIATABorderPrefix+iata)
|
||||
}
|
||||
|
||||
// InvalidateRegions makes imported MeshMapper regions visible to every region read.
|
||||
func (cr *CachedReader) InvalidateRegions(ctx context.Context) {
|
||||
cr.c.del(ctx, keyRegions)
|
||||
cr.c.delPrefix(ctx, keyRegionPrefix)
|
||||
}
|
||||
|
||||
// InvalidateNode removes the cached entries for a node by UUID.
|
||||
// Should be called from the ingest path after a node upsert.
|
||||
func (cr *CachedReader) InvalidateNode(ctx context.Context, nodeID uuid.UUID) {
|
||||
|
||||
@@ -17,6 +17,7 @@ type MeshMapperConfig struct {
|
||||
// MeshMapperZonesConfig imports each known IATA's published boundary, overriding borderFile.
|
||||
type MeshMapperZonesConfig struct {
|
||||
Enabled bool `yaml:"enabled"`
|
||||
ImportGroups bool `yaml:"import_groups"` // zone groups become regions
|
||||
RefreshInterval duration `yaml:"refresh_interval"`
|
||||
}
|
||||
|
||||
@@ -43,6 +44,9 @@ func (c MeshMapperScopesConfig) Interval() time.Duration {
|
||||
}
|
||||
|
||||
func (c *Config) validateMeshMapper() error {
|
||||
if z := c.MeshMapper.Zones; z.ImportGroups && !z.Enabled {
|
||||
return fmt.Errorf("meshmapper.zones.import_groups requires meshmapper.zones.enabled")
|
||||
}
|
||||
if z := c.MeshMapper.Zones; z.Enabled {
|
||||
// The Zones API asks clients not to poll more than once an hour.
|
||||
if z.Interval() < time.Hour || z.Interval() > 7*24*time.Hour {
|
||||
|
||||
@@ -53,6 +53,8 @@ func TestMeshMapperZonesConfig(t *testing.T) {
|
||||
{"too often", valid + " refresh_interval: 59m\n", true, 0},
|
||||
{"too rare", valid + " refresh_interval: 169h\n", true, 0},
|
||||
{"no regions", "meshmapper:\n zones:\n enabled: true\n", false, 24 * time.Hour},
|
||||
{"groups", valid + " import_groups: true\n", false, 24 * time.Hour},
|
||||
{"groups without zones", "meshmapper:\n zones:\n import_groups: true\n", true, 0},
|
||||
{"disabled", "meshmapper:\n zones:\n enabled: false\n", false, 24 * time.Hour},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
|
||||
@@ -19,7 +19,7 @@ type Seeder interface {
|
||||
UpsertIATADetails(ctx context.Context, iata string, name string, lat, lng *float64) error
|
||||
UpsertIATABorder(ctx context.Context, iata string, border json.RawMessage) error
|
||||
UpsertRegion(ctx context.Context, slug, name, description string, displayOrder int, centerLat, centerLng *float64, zoomLevel *int) (int32, error)
|
||||
UpsertRegionIATA(ctx context.Context, regionID int32, iata string) error
|
||||
SetRegionIATAs(ctx context.Context, regionID int32, iatas []string) error
|
||||
UpsertTransportScope(ctx context.Context, name, displayName string, transportKey, keyFingerprint []byte) error
|
||||
}
|
||||
|
||||
@@ -56,9 +56,10 @@ func Seed(ctx context.Context, cfg *Config, db Seeder) error {
|
||||
if err := db.UpsertIATA(ctx, iata); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := db.UpsertRegionIATA(ctx, id, iata); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Config owns the member list, including when it takes over an imported slug.
|
||||
if err := db.SetRegionIATAs(ctx, id, r.IATAs); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Transport Codes
|
||||
|
||||
@@ -52,8 +52,8 @@ func (s *stubSeeder) UpsertRegion(_ context.Context, slug, _, _ string, _ int, _
|
||||
return id, s.upsertErr
|
||||
}
|
||||
|
||||
func (s *stubSeeder) UpsertRegionIATA(_ context.Context, regionID int32, iata string) error {
|
||||
s.regionIATAs[regionID] = append(s.regionIATAs[regionID], iata)
|
||||
func (s *stubSeeder) SetRegionIATAs(_ context.Context, regionID int32, iatas []string) error {
|
||||
s.regionIATAs[regionID] = iatas
|
||||
return s.upsertErr
|
||||
}
|
||||
|
||||
|
||||
@@ -10,9 +10,12 @@ import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"unicode"
|
||||
)
|
||||
|
||||
type zoneEntry struct {
|
||||
@@ -20,8 +23,14 @@ type zoneEntry struct {
|
||||
hasBoundary bool
|
||||
}
|
||||
|
||||
type zoneGroup struct {
|
||||
name string
|
||||
members []string
|
||||
}
|
||||
|
||||
type zoneList struct {
|
||||
zones map[string]zoneEntry
|
||||
groups map[string]zoneGroup
|
||||
etag string
|
||||
fetchedAt, nextAttempt time.Time
|
||||
}
|
||||
@@ -33,6 +42,7 @@ type Directory struct {
|
||||
listURL string
|
||||
lists map[string]*zoneList
|
||||
retryAfter time.Time
|
||||
version int // bumped whenever a list is replaced
|
||||
}
|
||||
|
||||
func NewDirectory() *Directory {
|
||||
@@ -82,11 +92,12 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er
|
||||
problem := ""
|
||||
switch status {
|
||||
case http.StatusOK:
|
||||
zones, decodeErr := decodeZones(body, country)
|
||||
zones, groups, decodeErr := decodeZones(body, country)
|
||||
if decodeErr != nil {
|
||||
problem = "invalid response"
|
||||
} else {
|
||||
list.zones, list.etag = zones, header.Get("ETag")
|
||||
list.zones, list.groups, list.etag = zones, groups, header.Get("ETag")
|
||||
d.version++
|
||||
}
|
||||
case http.StatusNotModified:
|
||||
if list.zones == nil {
|
||||
@@ -116,7 +127,36 @@ func (d *Directory) fetch(ctx context.Context, country string, now time.Time) er
|
||||
return nil
|
||||
}
|
||||
|
||||
func decodeZones(body []byte, country string) (map[string]zoneEntry, error) {
|
||||
// groups merges every loaded list's groups by code. complete is false until each
|
||||
// country has a list, so a cold start can't look like every group disappearing.
|
||||
func (d *Directory) groups(countries []string) (merged map[string]zoneGroup, version int, complete bool) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
merged = map[string]zoneGroup{}
|
||||
for _, country := range countries {
|
||||
list := d.lists[country]
|
||||
if list == nil || list.zones == nil {
|
||||
return nil, d.version, false
|
||||
}
|
||||
for code, g := range list.groups {
|
||||
if have, ok := merged[code]; ok {
|
||||
g.members = append(slices.Clone(have.members), g.members...)
|
||||
slices.Sort(g.members)
|
||||
g.members = slices.Compact(g.members)
|
||||
g.name = have.name
|
||||
}
|
||||
merged[code] = g
|
||||
}
|
||||
}
|
||||
return merged, d.version, true
|
||||
}
|
||||
|
||||
var (
|
||||
groupCode = regexp.MustCompile(`^[A-Z0-9]{2,16}$`)
|
||||
memberIATA = regexp.MustCompile(`^[A-Z]{3}$`)
|
||||
)
|
||||
|
||||
func decodeZones(body []byte, country string) (map[string]zoneEntry, map[string]zoneGroup, error) {
|
||||
var document struct {
|
||||
Country string `json:"country"`
|
||||
Zones *[]struct {
|
||||
@@ -124,16 +164,38 @@ func decodeZones(body []byte, country string) (map[string]zoneEntry, error) {
|
||||
URL string `json:"url"`
|
||||
HasBoundary bool `json:"has_boundary"`
|
||||
} `json:"zones"`
|
||||
Groups []struct {
|
||||
Code string `json:"code"`
|
||||
Name string `json:"name"`
|
||||
Members []string `json:"members"`
|
||||
} `json:"groups"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &document); err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
if !strings.EqualFold(document.Country, country) || document.Zones == nil {
|
||||
return nil, fmt.Errorf("invalid zone list")
|
||||
return nil, nil, fmt.Errorf("invalid zone list")
|
||||
}
|
||||
zones := make(map[string]zoneEntry, len(*document.Zones))
|
||||
for _, zone := range *document.Zones {
|
||||
zones[strings.ToUpper(zone.Code)] = zoneEntry{url: zone.URL, hasBoundary: zone.HasBoundary}
|
||||
}
|
||||
return zones, nil
|
||||
groups := map[string]zoneGroup{}
|
||||
for _, g := range document.Groups {
|
||||
code, name := strings.ToUpper(g.Code), strings.TrimSpace(g.Name)
|
||||
if !groupCode.MatchString(code) || name == "" || len(name) > 128 || strings.ContainsFunc(name, unicode.IsControl) {
|
||||
continue // a bad group never blocks the zone list
|
||||
}
|
||||
var members []string
|
||||
for _, m := range g.Members {
|
||||
if m = strings.ToUpper(m); memberIATA.MatchString(m) {
|
||||
members = append(members, m)
|
||||
}
|
||||
}
|
||||
slices.Sort(members)
|
||||
if members = slices.Compact(members); len(members) > 0 {
|
||||
groups[code] = zoneGroup{name: name, members: members}
|
||||
}
|
||||
}
|
||||
return zones, groups, nil
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package meshmapper
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// RegionState is a stored region; imported ones are also what SaveImportedRegion writes.
|
||||
type RegionState struct {
|
||||
Slug, Name string
|
||||
DisplayOrder int
|
||||
Imported bool
|
||||
IATAs []string
|
||||
}
|
||||
|
||||
// OnRegionsChange is told whenever the imported regions change, for cache invalidation.
|
||||
func (z *Zones) OnRegionsChange(fn func(ctx context.Context)) { z.onRegions = fn }
|
||||
|
||||
// pruneGroups drops every imported region when group import is off.
|
||||
func (z *Zones) pruneGroups(ctx context.Context) error {
|
||||
pruned, err := z.store.PruneImportedRegions(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("prune MeshMapper regions: %w", err)
|
||||
}
|
||||
z.regionsRemoved(pruned)
|
||||
if len(pruned) > 0 {
|
||||
z.regionsChanged(ctx)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// syncGroups reconciles imported regions with the loaded zone lists. It only runs
|
||||
// once every tracked country has a list, and again when a list or the IATA set changes.
|
||||
func (z *Zones) syncGroups(ctx context.Context, iatas []string) error {
|
||||
if !z.importGroups {
|
||||
return nil
|
||||
}
|
||||
var countries []string
|
||||
for _, r := range z.regions {
|
||||
countries = append(countries, r.country)
|
||||
}
|
||||
slices.Sort(countries)
|
||||
groups, version, complete := z.dir.groups(slices.Compact(countries))
|
||||
if !complete || (z.groupsSynced && version == z.groupsVersion && len(iatas) == z.groupsKnown) {
|
||||
return nil
|
||||
}
|
||||
known := make(map[string]bool, len(iatas))
|
||||
for _, iata := range iatas {
|
||||
known[iata] = true
|
||||
}
|
||||
state, err := z.store.ListRegionState(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list regions for MeshMapper groups: %w", err)
|
||||
}
|
||||
stored, handOrder := map[string]RegionState{}, 0
|
||||
for _, r := range state {
|
||||
stored[r.Slug] = r
|
||||
if !r.Imported {
|
||||
handOrder = max(handOrder, r.DisplayOrder)
|
||||
}
|
||||
}
|
||||
var want []RegionState
|
||||
for code, g := range groups {
|
||||
if slices.ContainsFunc(g.members, func(m string) bool { return known[m] }) {
|
||||
want = append(want, RegionState{Slug: strings.ToLower(code), Name: g.name, Imported: true, IATAs: g.members})
|
||||
}
|
||||
}
|
||||
slices.SortFunc(want, func(a, b RegionState) int { return cmp.Or(cmp.Compare(a.Name, b.Name), cmp.Compare(a.Slug, b.Slug)) })
|
||||
keep, changed := []string{}, false
|
||||
for n, r := range want {
|
||||
r.DisplayOrder = handOrder + 1 + n
|
||||
if cur, ok := stored[r.Slug]; ok && !cur.Imported {
|
||||
slog.Info("MeshMapper group skipped: a configured region uses its slug", "component", "meshmapper.zones", "slug", r.Slug)
|
||||
continue
|
||||
} else if ok && cur.Name == r.Name && cur.DisplayOrder == r.DisplayOrder && slices.Equal(cur.IATAs, r.IATAs) {
|
||||
keep = append(keep, r.Slug)
|
||||
continue
|
||||
}
|
||||
saved, err := z.store.SaveImportedRegion(ctx, r)
|
||||
if err != nil {
|
||||
return fmt.Errorf("save MeshMapper region %s: %w", r.Slug, err)
|
||||
}
|
||||
if !saved {
|
||||
slog.Info("MeshMapper group skipped: a configured region uses its slug", "component", "meshmapper.zones", "slug", r.Slug)
|
||||
continue
|
||||
}
|
||||
keep, changed = append(keep, r.Slug), true
|
||||
slog.Info("MeshMapper region imported", "component", "meshmapper.zones", "slug", r.Slug, "name", r.Name, "iatas", r.IATAs)
|
||||
}
|
||||
pruned, err := z.store.PruneImportedRegions(ctx, keep)
|
||||
if err != nil {
|
||||
return fmt.Errorf("prune MeshMapper regions: %w", err)
|
||||
}
|
||||
z.regionsRemoved(pruned)
|
||||
if changed || len(pruned) > 0 {
|
||||
z.regionsChanged(ctx)
|
||||
}
|
||||
z.groupsSynced, z.groupsVersion, z.groupsKnown = true, version, len(iatas)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (z *Zones) regionsRemoved(slugs []string) {
|
||||
for _, slug := range slugs {
|
||||
slog.Info("MeshMapper region removed", "component", "meshmapper.zones", "slug", slug)
|
||||
}
|
||||
}
|
||||
|
||||
func (z *Zones) regionsChanged(ctx context.Context) {
|
||||
if z.onRegions != nil {
|
||||
z.onRegions(ctx)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package meshmapper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
)
|
||||
|
||||
const groupedList = `{"country":"CA","zones":[{"code":"YOW","url":"URL/","has_boundary":true,"group":"ONQC"}],"groups":[` +
|
||||
`{"code":"ONQC","name":"Ottawa-Quebec Corridor","members":["YUL","YOW","YQB"],"url":"https://onqc.meshmapper.net/"},` +
|
||||
`{"code":"GOLM","name":"Great Ontario Lake Mesh","members":["YYZ","YKF"],"url":"https://golm.meshmapper.net/"}]}`
|
||||
|
||||
type groupHarness struct {
|
||||
z *Zones
|
||||
f *fakeMeshMapper
|
||||
store *zoneMemoryStore
|
||||
invalidated int
|
||||
}
|
||||
|
||||
func newGroupHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, importGroups bool) *groupHarness {
|
||||
t.Helper()
|
||||
if store.iatas == nil {
|
||||
store.iatas = []string{"YOW"}
|
||||
}
|
||||
if store.rows == nil {
|
||||
store.rows = map[string]Boundary{}
|
||||
}
|
||||
h := &groupHarness{f: f, store: store}
|
||||
dir := NewDirectory()
|
||||
dir.listURL = f.URL + "/get_zones.php"
|
||||
h.z = NewZones(config.MeshMapperZonesConfig{Enabled: true, ImportGroups: importGroups}, store, dir)
|
||||
h.z.boundsURL = func(site string) (string, bool) { return site + "get_geojson.php", true }
|
||||
h.z.OnRegionsChange(func(context.Context) { h.invalidated++ })
|
||||
if err := h.z.Restore(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func groupedFake(t *testing.T) *fakeMeshMapper {
|
||||
f := newFakeMeshMapper(t)
|
||||
f.list = strings.ReplaceAll(groupedList, "URL", f.URL)
|
||||
return f
|
||||
}
|
||||
|
||||
// refetch makes the country list and YOW due, then runs the fetch and the sync tick.
|
||||
func (h *groupHarness) refetch(t *testing.T) {
|
||||
t.Helper()
|
||||
if list := h.z.dir.lists["CA"]; list != nil {
|
||||
list.fetchedAt, list.nextAttempt = time.Time{}, time.Time{}
|
||||
}
|
||||
for _, r := range h.z.regions {
|
||||
r.b.NextAttempt = time.Time{}
|
||||
}
|
||||
for range 2 {
|
||||
if err := h.z.Refresh(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGroupImportedWhenAMemberIsKnown(t *testing.T) {
|
||||
h := newGroupHarness(t, groupedFake(t), &zoneMemoryStore{}, true)
|
||||
h.refetch(t)
|
||||
got, ok := h.store.regions["onqc"]
|
||||
if !ok || got.Name != "Ottawa-Quebec Corridor" || !slices.Equal(got.IATAs, []string{"YOW", "YQB", "YUL"}) || got.DisplayOrder != 1 {
|
||||
t.Fatalf("group not imported with full members: %+v", h.store.regions)
|
||||
}
|
||||
if _, ok := h.store.regions["golm"]; ok {
|
||||
t.Fatal("group with no known member imported")
|
||||
}
|
||||
if h.invalidated != 1 {
|
||||
t.Fatal("region caches not invalidated", h.invalidated)
|
||||
}
|
||||
h.refetch(t)
|
||||
if h.invalidated != 1 {
|
||||
t.Fatal("unchanged groups invalidated again", h.invalidated)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGroupSlugClashKeepsConfiguredRegion(t *testing.T) {
|
||||
hand := RegionState{Slug: "onqc", Name: "Mine", DisplayOrder: 4, IATAs: []string{"YOW"}}
|
||||
store := &zoneMemoryStore{regions: map[string]RegionState{"onqc": hand}}
|
||||
h := newGroupHarness(t, groupedFake(t), store, true)
|
||||
h.refetch(t)
|
||||
if got := store.regions["onqc"]; got.Imported || got.Name != "Mine" || !slices.Equal(got.IATAs, []string{"YOW"}) || h.invalidated != 0 {
|
||||
t.Fatalf("configured region replaced: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGroupDroppedRemovesRegion(t *testing.T) {
|
||||
h := newGroupHarness(t, groupedFake(t), &zoneMemoryStore{}, true)
|
||||
h.refetch(t)
|
||||
h.f.list = strings.ReplaceAll(`{"country":"CA","zones":[{"code":"YOW","url":"URL/","has_boundary":true,"group":null}],"groups":[]}`, "URL", h.f.URL)
|
||||
h.refetch(t)
|
||||
if len(h.store.regions) != 0 || h.invalidated != 2 {
|
||||
t.Fatalf("dropped group kept: %+v invalidated=%d", h.store.regions, h.invalidated)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGroupImportOffRemovesImportedRegions(t *testing.T) {
|
||||
hand := RegionState{Slug: "east", Name: "East", IATAs: []string{"YOW"}}
|
||||
imported := RegionState{Slug: "onqc", Name: "Ottawa-Quebec Corridor", Imported: true, IATAs: []string{"YOW"}}
|
||||
store := &zoneMemoryStore{regions: map[string]RegionState{"east": hand, "onqc": imported}}
|
||||
h := newGroupHarness(t, groupedFake(t), store, false)
|
||||
if _, ok := store.regions["onqc"]; ok || h.invalidated != 1 {
|
||||
t.Fatal("imported region kept with import_groups off")
|
||||
}
|
||||
if _, ok := store.regions["east"]; !ok {
|
||||
t.Fatal("configured region removed")
|
||||
}
|
||||
h.refetch(t)
|
||||
if len(store.regions) != 1 {
|
||||
t.Fatal("groups imported with import_groups off", store.regions)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGroupFailedFetchKeepsGroups(t *testing.T) {
|
||||
h := newGroupHarness(t, groupedFake(t), &zoneMemoryStore{}, true)
|
||||
h.refetch(t)
|
||||
h.f.listStatus = 500
|
||||
h.z.groupsSynced = false // force a reconcile against the retained list
|
||||
h.refetch(t)
|
||||
if _, ok := h.store.regions["onqc"]; !ok {
|
||||
t.Fatal("failed fetch dropped imported region")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGroupColdStartWaitsForEveryList(t *testing.T) {
|
||||
old := RegionState{Slug: "old", Name: "Old", Imported: true, IATAs: []string{"YOW"}}
|
||||
store := &zoneMemoryStore{regions: map[string]RegionState{"old": old}}
|
||||
f := groupedFake(t)
|
||||
f.listStatus = 500
|
||||
h := newGroupHarness(t, f, store, true)
|
||||
h.refetch(t)
|
||||
if _, ok := store.regions["old"]; !ok {
|
||||
t.Fatal("imported regions pruned before any list loaded")
|
||||
}
|
||||
}
|
||||
@@ -46,6 +46,9 @@ type ZoneStore interface {
|
||||
PruneZoneBoundaries(ctx context.Context, keep []string) ([]string, error)
|
||||
ListZoneBoundaries(ctx context.Context) ([]Boundary, error)
|
||||
SaveZoneBoundary(ctx context.Context, b Boundary) error
|
||||
ListRegionState(ctx context.Context) ([]RegionState, error)
|
||||
SaveImportedRegion(ctx context.Context, r RegionState) (bool, error)
|
||||
PruneImportedRegions(ctx context.Context, keep []string) ([]string, error)
|
||||
}
|
||||
|
||||
type zoneRegion struct {
|
||||
@@ -66,10 +69,15 @@ type Zones struct {
|
||||
retryAfter time.Time
|
||||
onChange func(ctx context.Context, iata string)
|
||||
onUpdate func(imported map[string]json.RawMessage)
|
||||
|
||||
importGroups bool
|
||||
groupsSynced bool
|
||||
groupsVersion, groupsKnown int
|
||||
onRegions func(ctx context.Context)
|
||||
}
|
||||
|
||||
func NewZones(cfg config.MeshMapperZonesConfig, store ZoneStore, dir *Directory) *Zones {
|
||||
return &Zones{store: store, enabled: cfg.Enabled, seen: map[string]bool{}, interval: cfg.Interval(), dir: dir,
|
||||
return &Zones{store: store, enabled: cfg.Enabled, importGroups: cfg.Enabled && cfg.ImportGroups, seen: map[string]bool{}, interval: cfg.Interval(), dir: dir,
|
||||
boundsURL: boundaryEndpoint, client: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
|
||||
@@ -82,9 +90,15 @@ func (z *Zones) OnChange(fn func(ctx context.Context, iata string)) { z.onChange
|
||||
// OnUpdate receives every imported boundary whenever the set changes.
|
||||
func (z *Zones) OnUpdate(fn func(imported map[string]json.RawMessage)) { z.onUpdate = fn }
|
||||
|
||||
// Restore prunes imports for unknown IATAs (all of them when disabled) and
|
||||
// loads saved boundaries, without making HTTP requests.
|
||||
// Restore prunes imports for unknown IATAs (all of them when disabled), and
|
||||
// imported regions when group import is off, then loads saved boundaries,
|
||||
// without making HTTP requests.
|
||||
func (z *Zones) Restore(ctx context.Context) error {
|
||||
if !z.importGroups {
|
||||
if err := z.pruneGroups(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
iatas, err := z.store.ListKnownIATAs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("list IATAs for MeshMapper boundaries: %w", err)
|
||||
@@ -163,6 +177,9 @@ func (z *Zones) Refresh(ctx context.Context) (err error) {
|
||||
return fmt.Errorf("list IATAs for MeshMapper boundaries: %w", err)
|
||||
}
|
||||
z.track(iatas, nil)
|
||||
if err := z.syncGroups(ctx, iatas); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, r := range z.regions {
|
||||
if now.Before(r.b.NextAttempt) {
|
||||
continue
|
||||
|
||||
@@ -25,10 +25,42 @@ func boundaryBody(code, geometry string) string {
|
||||
}
|
||||
|
||||
type zoneMemoryStore struct {
|
||||
iatas []string
|
||||
rows map[string]Boundary
|
||||
pruned []string
|
||||
fail bool
|
||||
iatas []string
|
||||
rows map[string]Boundary
|
||||
pruned []string
|
||||
fail bool
|
||||
regions map[string]RegionState
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) ListRegionState(context.Context) ([]RegionState, error) {
|
||||
var out []RegionState
|
||||
for _, r := range s.regions {
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) SaveImportedRegion(_ context.Context, r RegionState) (bool, error) {
|
||||
if s.regions == nil {
|
||||
s.regions = map[string]RegionState{}
|
||||
}
|
||||
if cur, ok := s.regions[r.Slug]; ok && !cur.Imported {
|
||||
return false, nil
|
||||
}
|
||||
r.Imported = true
|
||||
s.regions[r.Slug] = r
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) PruneImportedRegions(_ context.Context, keep []string) ([]string, error) {
|
||||
var removed []string
|
||||
for slug, r := range s.regions {
|
||||
if r.Imported && !slices.Contains(keep, slug) {
|
||||
delete(s.regions, slug)
|
||||
removed = append(removed, slug)
|
||||
}
|
||||
}
|
||||
return removed, nil
|
||||
}
|
||||
|
||||
func (s *zoneMemoryStore) ListKnownIATAs(context.Context) ([]string, error) {
|
||||
|
||||
Reference in new issue
Block a user