routewatch/internal/routewatch/app_integration_internal_test.go

package routewatch

import (
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"net/netip"
	"slices"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/config"
	"git.eeqj.de/sneak/routewatch/internal/database"
	"git.eeqj.de/sneak/routewatch/internal/logger"
	"git.eeqj.de/sneak/routewatch/internal/metrics"
	"git.eeqj.de/sneak/routewatch/internal/ristypes"
	"git.eeqj.de/sneak/routewatch/internal/routetable"
	"git.eeqj.de/sneak/routewatch/internal/server"
	"git.eeqj.de/sneak/routewatch/internal/streamer"
)

// mockStore is a mock implementation of database.Store for testing
type mockStore struct {
	mu sync.Mutex

	// Counters for tracking calls
	ASNCount        int
	PrefixCount     int
	PeeringCount    int
	WithdrawalCount int

	// Track unique items
	ASNs     map[int]*database.ASN
	Prefixes map[string]*database.Prefix
	Peerings map[string]bool // key is "from_to"
	// TablePrefixes holds the prefixes that the written changes of the
	// routing table leave some peer carrying.
	TablePrefixes map[string]bool

	// Track IP versions
	IPv4Prefixes int
	IPv6Prefixes int
}

// newMockStore creates a new mock store
func newMockStore() *mockStore {
	return &mockStore{
		ASNs:          make(map[int]*database.ASN),
		Prefixes:      make(map[string]*database.Prefix),
		Peerings:      make(map[string]bool),
		TablePrefixes: make(map[string]bool),
	}
}

// GetOrCreateASN mock implementation
func (m *mockStore) GetOrCreateASN(
	number int, timestamp time.Time,
) (*database.ASN, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if asn, exists := m.ASNs[number]; exists {
		asn.LastSeen = timestamp

		return asn, nil
	}

	asn := &database.ASN{
		ASN:       number,
		FirstSeen: timestamp,
		LastSeen:  timestamp,
	}
	m.ASNs[number] = asn
	m.ASNCount++

	return asn, nil
}

// UpdatePrefixesBatch mock implementation
func (m *mockStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	for prefix, timestamp := range prefixes {
		if _, exists := m.Prefixes[prefix]; !exists {
			const (
				ipVersionV4 = 4
				ipVersionV6 = 6
			)

			ipVersion := ipVersionV4
			if strings.Contains(prefix, ":") {
				ipVersion = ipVersionV6
			}

			m.Prefixes[prefix] = &database.Prefix{
				Prefix:    prefix,
				IPVersion: ipVersion,
				FirstSeen: timestamp,
			}
			m.PrefixCount++

			if ipVersion == ipVersionV4 {
				m.IPv4Prefixes++
			} else {
				m.IPv6Prefixes++
			}
		}
	}

	return nil
}

// GetOrCreatePrefix mock implementation
func (m *mockStore) GetOrCreatePrefix(
	prefix string, timestamp time.Time,
) (*database.Prefix, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if p, exists := m.Prefixes[prefix]; exists {
		return p, nil
	}

	const (
		ipVersionV4 = 4
		ipVersionV6 = 6
	)

	ipVersion := ipVersionV4
	if strings.Contains(prefix, ":") {
		ipVersion = ipVersionV6
	}

	p := &database.Prefix{
		Prefix:    prefix,
		IPVersion: ipVersion,
		FirstSeen: timestamp,
	}
	m.Prefixes[prefix] = p
	m.PrefixCount++

	if ipVersion == ipVersionV4 {
		m.IPv4Prefixes++
	} else {
		m.IPv6Prefixes++
	}

	return p, nil
}

// RecordAnnouncement mock implementation
func (m *mockStore) RecordAnnouncement(_ *database.Announcement) error {
	// Not tracking announcements in detail for now
	return nil
}

// RecordPeering mock implementation
func (m *mockStore) RecordPeering(asA, asB int, _ database.SeenTimes) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Normalize
	if asA > asB {
		asA, asB = asB, asA
	}

	key := fmt.Sprintf("%d_%d", asA, asB)
	if !m.Peerings[key] {
		m.Peerings[key] = true
		m.PeeringCount++
	}

	return nil
}

// UpdatePeer mock implementation
func (m *mockStore) UpdatePeer(_ string, _ int, _ string, _ time.Time) error {
	// Simple mock - just return nil
	return nil
}

// Close mock implementation
func (m *mockStore) Close() error {
	return nil
}

// GetStats returns statistics about the mock store
func (m *mockStore) GetStats() (database.Stats, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	return database.Stats{
		ASNs:         len(m.ASNs),
		Prefixes:     len(m.Prefixes),
		IPv4Prefixes: m.IPv4Prefixes,
		IPv6Prefixes: m.IPv6Prefixes,
		Peerings:     m.PeeringCount,
		Peers:        10, // Mock peer count
	}, nil
}

// GetStatsContext returns statistics about the mock store with context support
func (m *mockStore) GetStatsContext(_ context.Context) (database.Stats, error) {
	return m.GetStats()
}

// GetAS returns the stored AS, or only its number when none is stored.
func (m *mockStore) GetAS(_ context.Context, asn int) (*database.ASN, bool, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if asnInfo, exists := m.ASNs[asn]; exists {
		return asnInfo, true, nil
	}

	return &database.ASN{ASN: asn}, false, nil
}

// GetLastSeenASes returns copies of the stored ASes last seen most recently
// and longest ago, or neither when none is stored.
func (m *mockStore) GetLastSeenASes(
	_ context.Context,
) (*database.ASN, *database.ASN, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	var newest, oldest *database.ASN

	for _, asn := range m.ASNs {
		if newest == nil || asn.LastSeen.After(newest.LastSeen) {
			newest = asn
		}

		if oldest == nil || asn.LastSeen.Before(oldest.LastSeen) {
			oldest = asn
		}
	}

	if newest == nil {
		return nil, nil, nil
	}

	newestCopy, oldestCopy := *newest, *oldest

	return &newestCopy, &oldestCopy, nil
}

// GetASWHOIS returns the stored AS, or only its number when none is stored.
func (m *mockStore) GetASWHOIS(ctx context.Context, asn int) (*database.ASN, error) {
	asnInfo, _, err := m.GetAS(ctx, asn)

	return asnInfo, err
}

// GetPrefixFirstSeen returns the zero time: the mock keeps no prefix times.
func (m *mockStore) GetPrefixFirstSeen(
	_ context.Context, _ netip.Prefix,
) (time.Time, error) {
	return time.Time{}, nil
}

// GetASPeers mock implementation
func (m *mockStore) GetASPeers(_ int) ([]database.ASPeer, error) {
	// Return empty peers for now
	return []database.ASPeer{}, nil
}

// GetASPeersContext mock implementation with context support
func (m *mockStore) GetASPeersContext(
	_ context.Context, asn int,
) ([]database.ASPeer, error) {
	return m.GetASPeers(asn)
}

// GetNextStaleASN mock implementation
func (m *mockStore) GetNextStaleASN(_ context.Context, _ time.Duration) (int, error) {
	return 0, database.ErrNoStaleASN
}

// UpdateASNWHOIS mock implementation
func (m *mockStore) UpdateASNWHOIS(
	_ context.Context, _ *database.ASNWHOISUpdate,
) error {
	return nil
}

// GetWHOISStats mock implementation
func (m *mockStore) GetWHOISStats(
	_ context.Context, _ time.Duration,
) (*database.WHOISStats, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	return &database.WHOISStats{
		TotalASNs:    len(m.ASNs),
		FreshASNs:    0,
		StaleASNs:    0,
		NeverFetched: len(m.ASNs),
	}, nil
}

// WriteRouteChanges records the prefixes the changes leave some peer carrying,
// whatever their origin AS.
func (m *mockStore) WriteRouteChanges(
	changes []routetable.Change, _ map[string]uint,
) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	for _, change := range changes {
		if change.Removed {
			delete(m.TablePrefixes, change.Prefix.String())
		} else {
			m.TablePrefixes[change.Prefix.String()] = true
		}
	}

	return nil
}

// ReadRouteTable returns an empty routing table.
func (m *mockStore) ReadRouteTable(_ context.Context) (*routetable.Table, error) {
	return routetable.New(routetable.Params{}), nil
}

// GetOrCreateASNBatch mock implementation
func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	for number, seen := range asns {
		if _, exists := m.ASNs[number]; !exists {
			m.ASNs[number] = &database.ASN{
				ASN:       number,
				FirstSeen: seen.First,
				LastSeen:  seen.Last,
			}
			m.ASNCount++
		}
	}

	return nil
}

// UpdatePeerBatch mock implementation
func (m *mockStore) UpdatePeerBatch(_ map[string]database.PeerUpdate) error {
	// Simple mock - just return nil
	return nil
}

// Vacuum mock implementation
func (m *mockStore) Vacuum(_ context.Context) error {
	return nil
}

// Analyze mock implementation
func (m *mockStore) Analyze(_ context.Context) error {
	return nil
}

// Checkpoint mock implementation
func (m *mockStore) Checkpoint(_ context.Context) error {
	return nil
}

// Ping mock implementation
func (m *mockStore) Ping(_ context.Context) error {
	return nil
}

func TestRouteWatchLiveFeed(t *testing.T) {
	t.Parallel()

	if testing.Short() {
		t.Skip("skipping live RIPE RIS network feed test in short mode; " +
			"run without -short to include it")
	}

	// Create mock database
	mockDB := newMockStore()

	defer func() {
		err := mockDB.Close()
		if err != nil {
			t.Errorf("failed to close mock database: %v", err)
		}
	}()

	logger := logger.New()

	// Create metrics tracker
	metricsTracker := metrics.New()

	// Create streamer
	s := streamer.New(logger, metricsTracker)

	// The state directory holds the blocklist copies Run downloads.
	cfg := &config.Config{
		StateDir:                    t.TempDir(),
		MaxRuntime:                  5 * time.Second,
		EnableBatchedDatabaseWrites: true,
	}

	table := routetable.New(routetable.Params{})

	// Create server
	srv := server.New(mockDB, table, s, logger, cfg)

	// Create RouteWatch with 5 second limit
	deps := Dependencies{
		DB:       mockDB,
		Table:    table,
		Streamer: s,
		Server:   srv,
		Logger:   logger,
		Config:   cfg,
	}
	rw := New(deps)

	// Run with context
	ctx := context.Background()
	go func() {
		_ = rw.Run(ctx)
	}()

	// Wait for the configured duration
	time.Sleep(5 * time.Second)

	// Force peering processing for test
	if rw.peeringHandler != nil {
		rw.peeringHandler.ProcessPeeringsNow()
	}

	// Get statistics
	stats, err := mockDB.GetStats()
	if err != nil {
		t.Fatalf("Failed to get stats: %v", err)
	}

	if stats.ASNs == 0 {
		t.Error("Expected to receive some ASNs from live feed")
	}

	t.Logf("Received %d unique ASNs in 5 seconds", stats.ASNs)

	if stats.Prefixes == 0 {
		t.Error("Expected to receive some prefixes from live feed")
	}

	t.Logf("Received %d unique prefixes (%d IPv4, %d IPv6) in 5 seconds",
		stats.Prefixes, stats.IPv4Prefixes, stats.IPv6Prefixes)

	if stats.Peerings == 0 {
		t.Error("Expected to receive some peerings from live feed")
	}

	t.Logf("Recorded %d AS peering relationships in 5 seconds", stats.Peerings)
}

// gatedPrefixHandler is a PrefixHandler whose worker in the streamer waits on
// the first message until release is closed, so the messages after it stay on
// the streamer's queue.
type gatedPrefixHandler struct {
	*PrefixHandler

	release chan struct{}
}

func (h gatedPrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
	<-h.release
	h.PrefixHandler.HandleMessage(msg)
}

// announcementFeed returns the lines RIS Live would send for one UPDATE per
// prefix, each announcing that prefix from testPeerIP with origin testASNA,
// followed by a KEEPALIVE. The streamer hands a KEEPALIVE to no handler, so
// once a raw handler has seen it, every update before it is queued.
func announcementFeed(prefixes []string) string {
	var feed strings.Builder

	for _, prefix := range prefixes {
		fmt.Fprintf(&feed, `{"type":"ris_message","data":{"type":"UPDATE",`+
			`"peer":"%s","peer_asn":"%d","timestamp":1700000000,"path":[%d],`+
			`"announcements":[{"next_hop":"%s","prefixes":["%s"]}]}}`+"\n",
			testPeerIP, testASNA, testASNA, testPeerIP, prefix)
	}

	fmt.Fprintf(&feed, `{"type":"ris_message","data":{"type":"KEEPALIVE",`+
		`"peer":"%s","peer_asn":"%d","timestamp":1700000000}}`+"\n",
		testPeerIP, testASNA)

	return feed.String()
}

// TestShutdownWritesTheUpdatesQueuedInTheStreamer has a local server send
// announcements of three prefixes and keep the response open. The prefix
// handler's worker waits on the first one until the server sees the streamer
// close the connection, which only Shutdown does, so all three are still in
// the streamer when Shutdown starts. Each prefix must be in the database when
// Shutdown returns, both in the prefix tables and in the routing table's
// writes.
func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
	t.Parallel()

	prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
	feed := announcementFeed(prefixes)

	release := make(chan struct{})

	srv := httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, r *http.Request) {
			_, _ = io.WriteString(w, feed)
			_ = http.NewResponseController(w).Flush()

			<-r.Context().Done()
			close(release)
		}))
	defer srv.Close()

	store := newMockStore()
	log := logger.New()
	cfg := &config.Config{}
	s := streamer.New(log, metrics.New())
	s.SetURL(srv.URL)

	rw := New(Dependencies{
		DB:       store,
		Streamer: s,
		Server:   server.New(store, nil, s, log, cfg),
		Logger:   log,
		Config:   cfg,
	})
	rw.prefixHandler = NewPrefixHandler(store, log, routetable.New(routetable.Params{}))
	s.RegisterHandler(gatedPrefixHandler{
		PrefixHandler: rw.prefixHandler, release: release,
	})

	linesSeen := make(chan struct{}, len(prefixes)+1)

	s.RegisterRawHandler(func(string) { linesSeen <- struct{}{} })

	err := s.Start(t.Context())
	if err != nil {
		t.Fatalf("Start returned error: %v", err)
	}

	for range cap(linesSeen) {
		select {
		case <-linesSeen:
		case <-time.After(stepTimeout):
			t.Fatalf("the streamer did not read %d lines within %s",
				cap(linesSeen), stepTimeout)
		}
	}

	shutDown := make(chan struct{})

	go func() {
		rw.Shutdown(t.Context())
		close(shutDown)
	}()

	select {
	case <-shutDown:
	case <-time.After(stepTimeout):
		t.Fatalf("Shutdown did not return within %s", stepTimeout)
	}

	store.mu.Lock()
	defer store.mu.Unlock()

	for _, prefix := range prefixes {
		_, inPrefixTables := store.Prefixes[prefix]
		if !inPrefixTables || !store.TablePrefixes[prefix] {
			t.Errorf("%s was queued in the streamer when Shutdown started; "+
				"written to the prefix tables %t, in the routing table's "+
				"writes %t", prefix, inPrefixTables, store.TablePrefixes[prefix])
		}
	}
}

// freePort returns a TCP port that was free a moment ago, for a server the
// test starts.
func freePort(t *testing.T) int {
	t.Helper()

	var listenConfig net.ListenConfig

	listener, err := listenConfig.Listen(t.Context(), "tcp", ":0")
	if err != nil {
		t.Fatalf("failed to find a free port: %v", err)
	}
	defer func() { _ = listener.Close() }()

	addr, ok := listener.Addr().(*net.TCPAddr)
	if !ok {
		t.Fatalf("listener address %v is not a TCP address", listener.Addr())
	}

	return addr.Port
}

// TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the
// AS handler start a write that the store holds until the test ends, past
// Shutdown's deadline. Shutdown must return without waiting for the write, with
// the HTTP server stopped, and must neither checkpoint nor close the database:
// the real one would wait for the write.
// Before the fix Shutdown waited for the write.
func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
	t.Parallel()

	const deadline = 100 * time.Millisecond

	blocking := newBlockingStore()
	defer blocking.releaseWrites()

	store := &closeOrderStore{Store: blocking}
	log := logger.New()
	port := freePort(t)
	cfg := &config.Config{Port: port}
	s := streamer.New(log, metrics.New())

	// A started server reads its routing table in the background, so it
	// needs one.
	rw := New(Dependencies{
		DB:       store,
		Streamer: s,
		Server: server.New(store, routetable.New(routetable.Params{}), s, log,
			cfg),
		Logger: log,
		Config: cfg,
	})

	err := rw.server.Start(t.Context())
	if err != nil {
		t.Fatalf("failed to start the HTTP server: %v", err)
	}

	rw.dbHandler = NewASHandler(store, log)
	startBlockedWrite(t, blocking, rw.dbHandler,
		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})

	ctx, cancel := context.WithTimeout(t.Context(), deadline)
	defer cancel()

	shutDown := make(chan struct{})

	go func() {
		rw.Shutdown(ctx)
		close(shutDown)
	}()

	select {
	case <-shutDown:
	case <-time.After(deadline + stepTimeout):
		t.Fatalf("Shutdown did not return within %s of its deadline while "+
			"a write was blocked", stepTimeout)
	}

	var dialer net.Dialer

	addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))

	conn, err := dialer.DialContext(t.Context(), "tcp", addr)
	if err == nil {
		_ = conn.Close()

		t.Errorf("the HTTP server still accepts connections on %s after "+
			"Shutdown returned", addr)
	}

	store.callsMu.Lock()
	defer store.callsMu.Unlock()

	if slices.Contains(store.calls, callCheckpoint) ||
		slices.Contains(store.calls, callClose) {
		t.Errorf("Shutdown made the calls %q past its deadline, want no %s "+
			"and no %s", store.calls, callCheckpoint, callClose)
	}
}

// closeOrderStore is a store that records, in order, the AS writes,
// checkpoints and closes it is given. It passes the AS writes on to Store.
type closeOrderStore struct {
	database.Store

	callsMu sync.Mutex
	calls   []string
}

// The calls closeOrderStore records.
const (
	callWrite      = "AS write"
	callCheckpoint = "checkpoint"
	callClose      = "close"
)

func (s *closeOrderStore) GetOrCreateASNBatch(
	asns map[int]database.SeenTimes,
) error {
	s.record(callWrite)

	return s.Store.GetOrCreateASNBatch(asns)
}

func (s *closeOrderStore) Checkpoint(_ context.Context) error {
	s.record(callCheckpoint)

	return nil
}

func (s *closeOrderStore) Close() error {
	s.record(callClose)

	return nil
}

func (s *closeOrderStore) record(call string) {
	s.callsMu.Lock()
	defer s.callsMu.Unlock()

	s.calls = append(s.calls, call)
}

// TestShutdownCheckpointsAndClosesTheDatabaseLast hands the AS handler one
// message and shuts down with time to spare. The handler's last write must
// come first, then a checkpoint, then the close. Before the fix Shutdown
// neither checkpointed nor closed the database.
func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
	t.Parallel()

	store := &closeOrderStore{Store: newMockStore()}
	log := logger.New()
	cfg := &config.Config{}
	s := streamer.New(log, metrics.New())

	rw := New(Dependencies{
		DB:       store,
		Streamer: s,
		Server:   server.New(store, nil, s, log, cfg),
		Logger:   log,
		Config:   cfg,
	})
	rw.dbHandler = NewASHandler(store, log)
	rw.dbHandler.HandleMessage(&ristypes.RISMessage{
		Path: ristypes.ASPath{Sequence: []int{testASNA}},
	})

	ctx, cancel := context.WithTimeout(t.Context(), stepTimeout)
	defer cancel()

	rw.Shutdown(ctx)

	store.callsMu.Lock()
	defer store.callsMu.Unlock()

	want := []string{callWrite, callCheckpoint, callClose}
	if !slices.Equal(store.calls, want) {
		t.Errorf("Shutdown made the calls %q, want %q", store.calls, want)
	}
}