From a970484b00ac195c4634e68dc4bd3d28ec9c13f9 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 22:51:08 -0400 Subject: [PATCH] feat(meshmapper): import zone groups as regions --- README.md | 1 + cmd/beacon/main.go | 1 + config.yaml.example | 9 ++ db/config.go | 14 ++- db/meshmapper.go | 33 +++++++ db/meshmapper_integration_test.go | 88 +++++++++++++++++ db/migrations/001_baseline.sql | 3 +- db/queries/queries.sql | 31 +++++- db/sqlc/mock/querier.go | 87 ++++++++++++++--- db/sqlc/models.go | 1 + db/sqlc/querier.go | 8 +- db/sqlc/queries.sql.go | 140 +++++++++++++++++++++++---- internal/cache/reader.go | 6 ++ internal/config/meshmapper.go | 4 + internal/config/meshmapper_test.go | 2 + internal/config/seed.go | 9 +- internal/config/seed_test.go | 4 +- internal/meshmapper/directory.go | 74 +++++++++++++-- internal/meshmapper/groups.go | 119 +++++++++++++++++++++++ internal/meshmapper/groups_test.go | 146 +++++++++++++++++++++++++++++ internal/meshmapper/zones.go | 23 ++++- internal/meshmapper/zones_test.go | 40 +++++++- 22 files changed, 785 insertions(+), 58 deletions(-) create mode 100644 internal/meshmapper/groups.go create mode 100644 internal/meshmapper/groups_test.go diff --git a/README.md b/README.md index 119c329..8052707 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index c6f93a1..fa4eab1 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -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) { diff --git a/config.yaml.example b/config.yaml.example index cc863fc..014c9d2 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -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 diff --git a/db/config.go b/db/config.go index ac15b6f..3836a33 100644 --- a/db/config.go +++ b/db/config.go @@ -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}) } diff --git a/db/meshmapper.go b/db/meshmapper.go index 02e0de2..7b4f3cb 100644 --- a/db/meshmapper.go +++ b/db/meshmapper.go @@ -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) +} diff --git a/db/meshmapper_integration_test.go b/db/meshmapper_integration_test.go index e1589a6..5db0161 100644 --- a/db/meshmapper_integration_test.go +++ b/db/meshmapper_integration_test.go @@ -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) + } +} diff --git a/db/migrations/001_baseline.sql b/db/migrations/001_baseline.sql index b858627..ca09fe3 100644 --- a/db/migrations/001_baseline.sql +++ b/db/migrations/001_baseline.sql @@ -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 diff --git a/db/queries/queries.sql b/db/queries/queries.sql index f215ce0..902baca 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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 -- ============================================================ diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 5ab8a15..12e4e2b 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -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() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index bf81a1d..eba458a 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -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 { diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 1b3bbc9..f319a90 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -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 // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index bc83377..e16f105 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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) diff --git a/internal/cache/reader.go b/internal/cache/reader.go index cb49338..737cd37 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -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) { diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go index d4fa642..b1c7094 100644 --- a/internal/config/meshmapper.go +++ b/internal/config/meshmapper.go @@ -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 { diff --git a/internal/config/meshmapper_test.go b/internal/config/meshmapper_test.go index 8cd3264..43c8c22 100644 --- a/internal/config/meshmapper_test.go +++ b/internal/config/meshmapper_test.go @@ -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) { diff --git a/internal/config/seed.go b/internal/config/seed.go index a419573..86d3249 100644 --- a/internal/config/seed.go +++ b/internal/config/seed.go @@ -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 diff --git a/internal/config/seed_test.go b/internal/config/seed_test.go index 04c8e63..6241390 100644 --- a/internal/config/seed_test.go +++ b/internal/config/seed_test.go @@ -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 } diff --git a/internal/meshmapper/directory.go b/internal/meshmapper/directory.go index 0ead347..c4a9d85 100644 --- a/internal/meshmapper/directory.go +++ b/internal/meshmapper/directory.go @@ -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 } diff --git a/internal/meshmapper/groups.go b/internal/meshmapper/groups.go new file mode 100644 index 0000000..464c06f --- /dev/null +++ b/internal/meshmapper/groups.go @@ -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) + } +} diff --git a/internal/meshmapper/groups_test.go b/internal/meshmapper/groups_test.go new file mode 100644 index 0000000..81bd67a --- /dev/null +++ b/internal/meshmapper/groups_test.go @@ -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") + } +} diff --git a/internal/meshmapper/zones.go b/internal/meshmapper/zones.go index 55d2398..b37b49b 100644 --- a/internal/meshmapper/zones.go +++ b/internal/meshmapper/zones.go @@ -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 diff --git a/internal/meshmapper/zones_test.go b/internal/meshmapper/zones_test.go index f93152e..6dc4cf6 100644 --- a/internal/meshmapper/zones_test.go +++ b/internal/meshmapper/zones_test.go @@ -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) {