beacon-server/internal/meshmapper/channels_test.go

// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later

package meshmapper

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/MeshCore-Beacon/beacon-server/internal/config"
	"github.com/MeshCore-Beacon/beacon-server/internal/keystore"
)

func channelList(iata string, names ...string) []byte {
	rows := []map[string]any{}
	for _, name := range names {
		secret, hash, _ := keystore.DeriveHashtagKey(strings.TrimPrefix(name, "#"))
		rows = append(rows, map[string]any{"name": name, "key": hex.EncodeToString(secret), "hash": hex.EncodeToString([]byte{hash}), "zones": []string{iata}})
	}
	b, _ := json.Marshal(map[string]any{"generated_at": "2026-09-30T18:00:00Z", "region": iata, "zones": []string{iata}, "count": len(rows), "channels": rows})
	return b
}

type channelMemoryStore struct {
	iatas   []string
	rows    map[sourceKey]Cache
	members map[string][][]byte
	fail    bool
}

func newChannelMemoryStore(iatas ...string) *channelMemoryStore {
	return &channelMemoryStore{iatas: iatas, rows: map[sourceKey]Cache{}, members: map[string][][]byte{}}
}

func (s *channelMemoryStore) ListKnownIATAs(context.Context) ([]string, error) {
	return slices.Clone(s.iatas), nil
}

func (s *channelMemoryStore) ListChannelCatalogues(context.Context) ([]Catalogue, error) {
	var out []Catalogue
	for key, c := range s.rows {
		out = append(out, Catalogue{IATA: key.iata, URL: key.url, Cache: c})
	}
	return out, nil
}

func (s *channelMemoryStore) SaveChannelCatalogue(_ context.Context, iata, url string, next Cache, fingerprints [][]byte) error {
	if s.fail {
		return context.DeadlineExceeded
	}
	if next.Payload != nil {
		s.members[iata] = fingerprints
	}
	old := s.rows[sourceKey{iata, url}]
	if next.Payload == nil {
		next.Payload = old.Payload
	}
	if next.CheckedAt.IsZero() {
		next.CheckedAt, next.ETag = old.CheckedAt, old.ETag
	}
	s.rows[sourceKey{iata, url}] = next
	return nil
}

func (s *channelMemoryStore) ClearChannelMembers(context.Context) error {
	s.members = map[string][][]byte{}
	return nil
}

type channelSite struct {
	*httptest.Server
	list, body          string
	status              int
	retryAfter          string
	listCalls, channels int
}

func newChannelSite(t *testing.T) *channelSite {
	f := &channelSite{status: 200, body: string(channelList("YOW", "#ottawa-mesh"))}
	f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/get_zones.php":
			f.listCalls++
			_, _ = w.Write([]byte(f.list))
		case "/get_channels.php":
			f.channels++
			w.Header().Set("ETag", `"c1"`)
			w.Header().Set("Retry-After", f.retryAfter)
			w.WriteHeader(f.status)
			if f.status == 200 {
				_, _ = w.Write([]byte(f.body))
			}
		default:
			t.Errorf("unexpected path %s", r.URL.Path)
		}
	}))
	f.list = `{"country":"CA","zones":[{"code":"YOW","url":"` + f.URL + `/","has_boundary":true,"group":null}]}`
	t.Cleanup(f.Close)
	return f
}

type channelHarness struct {
	c     *Channels
	keys  *keystore.MapKeyStore
	added [][]byte
}

func newChannelHarness(t *testing.T, f *channelSite, store *channelMemoryStore, lists *zoneListMemory, enabled bool) *channelHarness {
	t.Helper()
	dir := newTestDirectory(lists)
	dir.listURL = f.URL + "/get_zones.php"
	if err := dir.Restore(context.Background()); err != nil {
		t.Fatal(err)
	}
	h := &channelHarness{keys: keystore.NewMapKeyStore(nil)}
	c, err := NewChannels(context.Background(), config.MeshMapperChannelsConfig{Enabled: enabled}, store, dir, h.keys)
	if err != nil {
		t.Fatal(err)
	}
	c.channelsURL = func(site string) (string, bool) { return site + "get_channels.php", strings.HasPrefix(site, f.URL) }
	c.OnNewKeys(func(hashes [][]byte) { h.added = append(h.added, hashes...) })
	h.c = c
	return h
}

func (h *channelHarness) refresh(t *testing.T, n int) {
	t.Helper()
	for range n {
		if err := h.c.Refresh(context.Background()); err != nil {
			t.Fatal(err)
		}
	}
}

func TestChannelsImportKeysAndMembers(t *testing.T) {
	f := newChannelSite(t)
	store := newChannelMemoryStore("YOW")
	h := newChannelHarness(t, f, store, newZoneListMemory(), true)
	h.refresh(t, 2)
	secret, hash, fingerprint := keystore.DeriveHashtagKey("ottawa-mesh")
	got := h.keys.GetKey([]byte{hash})
	if len(got) != 1 || !bytes.Equal(got[0].Key, secret) || got[0].Name != "#ottawa-mesh" || got[0].Hashtag != "ottawa-mesh" {
		t.Fatalf("key not imported: %+v", got)
	}
	if len(store.members["YOW"]) != 1 || !bytes.Equal(store.members["YOW"][0], fingerprint) {
		t.Fatal("YOW membership not saved", store.members)
	}
	if len(h.added) != 1 || h.added[0][0] != hash {
		t.Fatal("new key not reported for backfill", h.added)
	}
	if b := h.c.sources[0].cache; b.NextAttempt.Sub(b.AttemptedAt) != 24*time.Hour {
		t.Fatal("refresh interval", b.NextAttempt.Sub(b.AttemptedAt))
	}
}

func TestChannelsRejectMismatchKeepsLastGood(t *testing.T) {
	f := newChannelSite(t)
	store := newChannelMemoryStore("YOW")
	h := newChannelHarness(t, f, store, newZoneListMemory(), true)
	h.refresh(t, 2)
	good := string(channelList("YOW", "#ottawa-mesh"))
	secret, hash, _ := keystore.DeriveHashtagKey("ottawa-mesh")
	for name, body := range map[string]string{
		"planted key":   strings.Replace(good, hex.EncodeToString(secret), strings.Repeat("0", 32), 1),
		"wrong key":     strings.Replace(string(channelList("YOW", "#other")), `"#other"`, `"#planted"`, 1),
		"wrong hash":    strings.Replace(good, `"hash":"`+hex.EncodeToString([]byte{hash})+`"`, `"hash":"zz"`, 1),
		"other region":  string(channelList("YYZ", "#ottawa-mesh")),
		"bad count":     strings.Replace(good, `"count":1`, `"count":2`, 1),
		"uppercase":     string(channelList("YOW", "#Ottawa")),
		"group zones":   strings.Replace(good, `"zones":["YOW"]}`, `"zones":["YOW","YYZ"]}`, 1),
		"missing count": strings.Replace(good, `"count":1,`, ``, 1),
	} {
		f.body = body
		h.c.sources[0].cache.NextAttempt = time.Time{}
		h.refresh(t, 1)
		s := h.c.sources[0]
		if s.cache.LastError != "invalid response" || len(h.keys.GetKey([]byte{hash})) != 1 || len(store.members["YOW"]) != 1 {
			t.Fatalf("%s: accepted or dropped last good list: %q", name, s.cache.LastError)
		}
		if s.cache.NextAttempt.Sub(s.cache.AttemptedAt) != 24*time.Hour {
			t.Fatalf("%s: retried before the rate limit", name)
		}
	}
}

func TestChannelsRateLimitAndFailuresWait(t *testing.T) {
	for _, tc := range []struct {
		status     int
		retryAfter string
		want       time.Duration
	}{{503, "", 24 * time.Hour}, {429, "60", 24 * time.Hour}, {429, "172800", 48 * time.Hour}} {
		f := newChannelSite(t)
		f.status, f.retryAfter = tc.status, tc.retryAfter
		h := newChannelHarness(t, f, newChannelMemoryStore("YOW"), newZoneListMemory(), true)
		h.refresh(t, 3)
		s := h.c.sources[0]
		if f.channels != 1 || s.cache.NextAttempt.Sub(s.cache.AttemptedAt) != tc.want {
			t.Fatalf("HTTP %d Retry-After %q: calls=%d wait=%v", tc.status, tc.retryAfter, f.channels, s.cache.NextAttempt.Sub(s.cache.AttemptedAt))
		}
	}
}

func TestChannelsRestoreWithoutRequests(t *testing.T) {
	f := newChannelSite(t)
	store, lists := newChannelMemoryStore("YOW"), newZoneListMemory()
	newChannelHarness(t, f, store, lists, true).refresh(t, 2)
	restarted := newChannelHarness(t, f, store, lists, true)
	_, hash, _ := keystore.DeriveHashtagKey("ottawa-mesh")
	if len(restarted.keys.GetKey([]byte{hash})) != 1 {
		t.Fatal("saved channels not restored")
	}
	restarted.refresh(t, 2)
	if f.listCalls != 1 || f.channels != 1 {
		t.Fatal("restart made requests", f.listCalls, f.channels)
	}
}

func TestChannelsDisabledClearsMembers(t *testing.T) {
	f := newChannelSite(t)
	store := newChannelMemoryStore("YOW")
	store.members["YOW"] = [][]byte{{1}}
	h := newChannelHarness(t, f, store, newZoneListMemory(), false)
	if len(store.members) != 0 || len(h.c.sources) != 0 {
		t.Fatal("disabled import kept membership or sources")
	}
}

func TestChannelsRequestCutShortStillCounts(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() }))
	defer server.Close()
	store := newChannelMemoryStore("YOW")
	c, err := NewChannels(context.Background(), config.MeshMapperChannelsConfig{Enabled: true}, store, newTestDirectory(newZoneListMemory()), keystore.NewMapKeyStore(nil))
	if err != nil {
		t.Fatal(err)
	}
	s := &c.sources[0]
	s.url = server.URL
	now := time.Now().UTC()
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()
	_ = c.refresh(ctx, s, now)
	if got := s.cache.NextAttempt.Sub(now); got != channelRetry {
		t.Fatal("abandoned request retried inside the rate limit", got)
	}
	if row := store.rows[sourceKey{"YOW", server.URL}]; row.NextAttempt.Sub(now) != channelRetry {
		t.Fatal("abandoned request not persisted", row)
	}
}