galaxy-game/gateway/internal/grpcapi/push_heartbeat_test.go

package grpcapi

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"

	edgev1 "galaxy/gateway/proto/edge/v1"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
)

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

	stream := newHeartbeatingStream(newCapturingStream(t), 0, nil)
	assert.Nil(t, stream, "zero interval must not allocate a wrapper")

	negative := newHeartbeatingStream(newCapturingStream(t), -time.Second, nil)
	assert.Nil(t, negative, "negative interval must not allocate a wrapper")
}

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

	inner := newCapturingStream(t)
	hb := newHeartbeatingStream(inner, 30*time.Millisecond, nil)
	require.NotNil(t, hb)
	defer hb.Stop()

	go func() { _ = hb.Run(t.Context()) }()

	event := inner.recv(t, 200*time.Millisecond)
	assert.Equal(t, gatewayHeartbeatEventType, event.GetEventType())
	// Heartbeat envelope: only the event type travels. Every other
	// field stays at proto3 default so the wire frame stays minimal.
	assert.Empty(t, event.GetEventId())
	assert.Zero(t, event.GetTimestampMs())
	assert.Empty(t, event.GetPayloadBytes())
	assert.Empty(t, event.GetPayloadHash())
	assert.Empty(t, event.GetSignature())
	assert.Empty(t, event.GetRequestId())
	assert.Empty(t, event.GetTraceId())
}

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

	inner := newCapturingStream(t)
	hb := newHeartbeatingStream(inner, 50*time.Millisecond, nil)
	require.NotNil(t, hb)
	defer hb.Stop()

	go func() { _ = hb.Run(t.Context()) }()

	// Reset the timer every 20ms for 120ms — the silence window never
	// elapses, so the heartbeat goroutine must stay quiet and the
	// channel must only carry the manual real-event Sends.
	go func() {
		ticker := time.NewTicker(20 * time.Millisecond)
		defer ticker.Stop()
		for range 6 {
			<-ticker.C
			if err := hb.Send(&edgev1.GatewayEvent{EventType: "real.event"}); err != nil {
				t.Errorf("real Send failed: %v", err)
				return
			}
		}
	}()

	for range 6 {
		event := inner.recv(t, 100*time.Millisecond)
		assert.Equal(t, "real.event", event.GetEventType(), "only real events should appear while Send keeps resetting the silence window")
	}
}

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

	inner := newCapturingStream(t)
	hb := newHeartbeatingStream(inner, 20*time.Millisecond, nil)
	require.NotNil(t, hb)

	runDone := make(chan error, 1)
	go func() { runDone <- hb.Run(context.Background()) }()

	hb.Stop()
	select {
	case err := <-runDone:
		require.NoError(t, err)
	case <-time.After(200 * time.Millisecond):
		t.Fatal("Run did not exit after Stop")
	}

	// Stop is idempotent; the second call must not panic on the
	// already-closed done channel.
	assert.NotPanics(t, hb.Stop)
}

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

	inner := newCapturingStream(t)
	hb := newHeartbeatingStream(inner, 20*time.Millisecond, nil)
	require.NotNil(t, hb)
	defer hb.Stop()

	ctx, cancel := context.WithCancel(context.Background())
	runDone := make(chan error, 1)
	go func() { runDone <- hb.Run(ctx) }()

	cancel()
	select {
	case err := <-runDone:
		require.NoError(t, err)
	case <-time.After(200 * time.Millisecond):
		t.Fatal("Run did not exit after context cancel")
	}
}

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

	wantErr := errors.New("send failed")
	inner := newCapturingStream(t)
	inner.sendErr.Store(&errorBox{err: wantErr})

	hb := newHeartbeatingStream(inner, time.Minute, nil)
	require.NotNil(t, hb)
	defer hb.Stop()

	err := hb.Send(&edgev1.GatewayEvent{EventType: "real.event"})
	require.ErrorIs(t, err, wantErr)
}

// capturingStream is a minimal grpc.ServerStreamingServer that pushes
// every Send into a channel so tests can assert on the wire frame.
type capturingStream struct {
	grpc.ServerStreamingServer[edgev1.GatewayEvent]

	events  chan *edgev1.GatewayEvent
	sendErr atomic.Pointer[errorBox]
}

type errorBox struct{ err error }

func newCapturingStream(t *testing.T) *capturingStream {
	t.Helper()

	return &capturingStream{events: make(chan *edgev1.GatewayEvent, 16)}
}

func (s *capturingStream) Send(event *edgev1.GatewayEvent) error {
	if box := s.sendErr.Load(); box != nil {
		return box.err
	}
	s.events <- event

	return nil
}

func (s *capturingStream) Context() context.Context { return context.Background() }

func (s *capturingStream) SetHeader(metadata.MD) error  { return nil }
func (s *capturingStream) SendHeader(metadata.MD) error { return nil }
func (s *capturingStream) SetTrailer(metadata.MD)       {}
func (s *capturingStream) SendMsg(any) error            { return errors.New("capturingStream.SendMsg: unused") }
func (s *capturingStream) RecvMsg(any) error            { return errors.New("capturingStream.RecvMsg: unused") }

func (s *capturingStream) recv(t *testing.T, timeout time.Duration) *edgev1.GatewayEvent {
	t.Helper()

	select {
	case event := <-s.events:
		return event
	case <-time.After(timeout):
		t.Fatalf("no event captured within %s", timeout)
		return nil
	}
}