routewatch/internal/routewatch/app_integration_internal_test.go

package routewatch

import (
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"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/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
	RouteCount      int
	WithdrawalCount int

	// Track unique items
	ASNs     map[int]*database.ASN
	Prefixes map[string]*database.Prefix
	Peerings map[string]bool // key is "from_to"
	Routes   map[string]bool // key is "prefix_origin_peer"

	// 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),
		Routes:   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,
			}
		}
	}

	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, _ time.Time) 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()
}

// UpsertLiveRoute mock implementation
func (m *mockStore) UpsertLiveRoute(_ *database.LiveRoute) error {
	// Simple mock - just return nil
	return nil
}

// DeleteLiveRoute mock implementation
func (m *mockStore) DeleteLiveRoute(_, _ string) error {
	// Simple mock - just return nil
	return nil
}

// GetPrefixDistribution mock implementation
func (m *mockStore) GetPrefixDistribution() (
	[]database.PrefixDistribution, []database.PrefixDistribution, error,
) {
	// Return empty distributions for now
	return nil, nil, nil
}

// GetPrefixDistributionContext mock implementation with context support
func (m *mockStore) GetPrefixDistributionContext(_ context.Context) (
	[]database.PrefixDistribution, []database.PrefixDistribution, error,
) {
	return m.GetPrefixDistribution()
}

// GetLiveRouteCounts mock implementation
func (m *mockStore) GetLiveRouteCounts() (int, int, error) {
	// Return mock counts
	return m.RouteCount / 2, m.RouteCount / 2, nil
}

// GetLiveRouteCountsContext mock implementation with context support
func (m *mockStore) GetLiveRouteCountsContext(_ context.Context) (int, int, error) {
	return m.GetLiveRouteCounts()
}

// GetASDetails mock implementation
func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Check if ASN exists
	if asnInfo, exists := m.ASNs[asn]; exists {
		// Return empty prefixes for now
		return asnInfo, []database.LiveRoute{}, nil
	}

	return nil, nil, database.ErrNoRoute
}

// GetASDetailsContext mock implementation with context support
func (m *mockStore) GetASDetailsContext(
	_ context.Context, asn int,
) (*database.ASN, []database.LiveRoute, error) {
	return m.GetASDetails(asn)
}

// GetPrefixDetails mock implementation
func (m *mockStore) GetPrefixDetails(
	_ string,
) ([]database.LiveRoute, map[int]database.ASN, error) {
	// Return empty routes for now
	return []database.LiveRoute{}, map[int]database.ASN{}, nil
}

// GetPrefixDetailsContext mock implementation with context support
func (m *mockStore) GetPrefixDetailsContext(
	_ context.Context, prefix string,
) ([]database.LiveRoute, map[int]database.ASN, error) {
	return m.GetPrefixDetails(prefix)
}

func (m *mockStore) GetRandomPrefixesByLength(
	_, _, _ int,
) ([]database.LiveRoute, error) {
	// Return empty routes for now
	return []database.LiveRoute{}, nil
}

// GetRandomPrefixesByLengthContext mock implementation with context support
func (m *mockStore) GetRandomPrefixesByLengthContext(
	_ context.Context, maskLength, ipVersion, limit int,
) ([]database.LiveRoute, error) {
	return m.GetRandomPrefixesByLength(maskLength, ipVersion, limit)
}

// 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)
}

// GetIPInfo mock implementation
func (m *mockStore) GetIPInfo(ip string) (*database.IPInfo, error) {
	return m.GetIPInfoContext(context.Background(), ip)
}

// GetIPInfoContext mock implementation with context support
func (m *mockStore) GetIPInfoContext(
	_ context.Context, ip string,
) (*database.IPInfo, error) {
	now := time.Now()

	return &database.IPInfo{
		IP:          ip,
		Netblock:    "8.8.8.0/24",
		MaskLength:  24,
		IPVersion:   4,
		NumPeers:    3,
		ASN:         15169,
		Handle:      "GOOGLE",
		Description: "Google LLC",
		CountryCode: "US",
		FirstSeen:   now.Add(-24 * time.Hour),
		LastSeen:    now,
	}, nil
}

// 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
}

// UpsertLiveRouteBatch mock implementation
func (m *mockStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	for _, route := range routes {
		// Track prefix
		if _, exists := m.Prefixes[route.Prefix]; !exists {
			m.Prefixes[route.Prefix] = &database.Prefix{
				Prefix:    route.Prefix,
				IPVersion: route.IPVersion,
				FirstSeen: route.LastUpdated,
			}
			m.PrefixCount++

			if route.IPVersion == 4 {
				m.IPv4Prefixes++
			} else {
				m.IPv6Prefixes++
			}
		}

		m.RouteCount++
	}

	return nil
}

// DeleteLiveRouteBatch mock implementation
func (m *mockStore) DeleteLiveRouteBatch(_ []database.LiveRouteDeletion) error {
	// Simple mock - just return nil
	return nil
}

// DeleteLiveRoutesOfPeer mock implementation
func (m *mockStore) DeleteLiveRoutesOfPeer(_ string) (int, error) {
	// Simple mock - deletes nothing
	return 0, nil
}

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

	for number, timestamp := range asns {
		if _, exists := m.ASNs[number]; !exists {
			m.ASNs[number] = &database.ASN{
				ASN:       number,
				FirstSeen: timestamp,
				LastSeen:  timestamp,
			}
			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)

	// Create test config with empty state dir (no snapshot loading)
	cfg := &config.Config{
		StateDir:                    "",
		MaxRuntime:                  5 * time.Second,
		EnableBatchedDatabaseWrites: true,
	}

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

	// Create RouteWatch with 5 second limit
	deps := Dependencies{
		DB:       mockDB,
		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.
func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
	t.Parallel()

	const maxWait = 10 * time.Second

	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, s, log, cfg),
		Logger:   log,
		Config:   cfg,
	})
	rw.prefixHandler = NewPrefixHandler(store, log)
	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(maxWait):
			t.Fatalf("the streamer did not read %d lines within %s",
				cap(linesSeen), maxWait)
		}
	}

	shutDown := make(chan struct{})

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

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

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

	for _, prefix := range prefixes {
		if _, ok := store.Prefixes[prefix]; !ok {
			t.Errorf("%s was queued in the streamer when Shutdown started "+
				"but is not in the database", 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
}

// TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked 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 within its deadline
// plus a small margin, 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 TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
	t.Parallel()

	const (
		deadline = 100 * time.Millisecond
		// margin is how long Shutdown may take after its deadline, to stop
		// the HTTP server.
		margin = 2 * time.Second
	)

	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())

	rw := New(Dependencies{
		DB:       store,
		Streamer: s,
		Server:   server.New(store, 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)
	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
	startBlockedWrite(t, blocking, rw.dbHandler,
		[]*ristypes.RISMessage{{Path: fill}})

	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 + margin):
		t.Fatalf("Shutdown did not return within %s of its deadline while "+
			"a write was blocked", margin)
	}

	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]time.Time) 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()

	const deadline = 10 * time.Second

	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, 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(), deadline)
	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)
	}
}