nomadcode/services/core/internal/gitosync/runner_test.go

package gitosync

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"github.com/nomadcode/nomadcode-core/internal/gitoevents"
	"github.com/nomadcode/nomadcode-core/internal/protosocket"
	"google.golang.org/protobuf/types/known/structpb"
	"nhooyr.io/websocket"
)

// branchUpdatedEnvelope builds a contract-shaped broadcast envelope fixture.
func branchUpdatedEnvelope() protosocket.Envelope {
	return protosocket.Envelope{
		ProtocolVersion: protosocket.ProtocolVersion,
		ID:              "event-id",
		Type:            "event",
		Channel:         "event",
		Action:          "branch.updated",
		Payload: map[string]any{
			"id":          "event-id",
			"type":        "branch.updated",
			"provider":    "forgejo",
			"delivery_id": "delivery-id",
			"repo_id":     "nomadcode",
			"branch":      "develop",
			"before":      "old-sha",
			"after":       "new-sha",
			"changed_files": []any{
				map[string]any{
					"path":        "agent-roadmap/phase/example/milestones/example.md",
					"change_type": "modified",
				},
				map[string]any{
					"path":        "README.md",
					"change_type": "modified",
				},
			},
			"observed_at": "2026-06-13T00:00:00Z",
			"created_at":  "2026-06-13T00:00:00Z",
		},
	}
}

// TestProtoSocketTransportConnectAndClose verifies basic connect/close using toki.WsClient.
func TestProtoSocketTransportConnectAndClose(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := websocket.Accept(w, r, nil)
		if err != nil {
			return
		}
		defer conn.Close(websocket.StatusInternalError, "")

		ctx := r.Context()
		for {
			_, _, err := conn.Read(ctx)
			if err != nil {
				break
			}
		}
	}))
	defer srv.Close()

	wsURL := strings.Replace(srv.URL, "http", "ws", 1)
	transport := &protoSocketTransport{url: wsURL}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	if err := transport.Connect(ctx); err != nil {
		t.Fatalf("Connect failed: %v", err)
	}

	if transport.client == nil || !transport.client.IsAlive() {
		t.Fatal("client should be alive")
	}

	if err := transport.Close(); err != nil {
		t.Fatalf("Close failed: %v", err)
	}
}

// TestProtoSocketTransportOverflowAndDisconnect implements REVIEW_API-1.
// It proves backpressure on overflow and that close/disconnect terminal signals
// are not silently dropped.
func TestProtoSocketTransportOverflowAndDisconnect(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := websocket.Accept(w, r, nil)
		if err != nil {
			return
		}

		serverClient := toki.NewWsClient(conn, 0, 0, protosocket.ParserMap())
		defer serverClient.Close()

		// Send 3 events.
		// Since buffer capacity is 1, the client will block (backpressure) during the second or third send.
		eventEnv := branchUpdatedEnvelope()
		eventStruct, _ := eventEnv.ToStruct()

		_ = serverClient.Send(eventStruct)
		_ = serverClient.Send(eventStruct)
		_ = serverClient.Send(eventStruct)

		// Wait briefly then close the server-side client to trigger a disconnect
		time.Sleep(100 * time.Millisecond)
	}))
	defer srv.Close()

	wsURL := strings.Replace(srv.URL, "http", "ws", 1)
	transport := &protoSocketTransport{
		url:        wsURL,
		inboundCap: 1, // Set small buffer capacity to test overflow/backpressure
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := transport.Connect(ctx); err != nil {
		t.Fatalf("Connect failed: %v", err)
	}
	defer transport.Close()

	// Wait to let server send events and then disconnect
	time.Sleep(200 * time.Millisecond)

	// We expect to read events successfully, and finally observe the disconnect EOF without getting stuck.
	// 1st Read
	env1, err := transport.Read(ctx)
	if err != nil {
		t.Fatalf("Read 1 failed: %v", err)
	}
	if env1.Action != "branch.updated" {
		t.Errorf("expected branch.updated event, got: %s", env1.Action)
	}

	// 2nd Read
	env2, err := transport.Read(ctx)
	if err != nil {
		t.Fatalf("Read 2 failed: %v", err)
	}
	if env2.Action != "branch.updated" {
		t.Errorf("expected branch.updated event, got: %s", env2.Action)
	}

	// 3rd Read (Could either be 3rd event or EOF, depending on toki read timing, but must return either)
	_, err = transport.Read(ctx)
	// Eventually it must return EOF. We do another read just to ensure EOF.
	_, err = transport.Read(ctx)
	if err == nil {
		t.Fatal("expected terminal EOF error, but got nil")
	}
	if !errors.Is(err, io.EOF) && !strings.Contains(err.Error(), "context") {
		t.Fatalf("expected EOF error, got: %v", err)
	}
}

// TestProtoSocketTransportClientRunDeliversBinaryBranchUpdatedEvent implements REVIEW_API-2.
// It verifies that gitoevents.Client.Run works with the production transport and mock server.
func TestProtoSocketTransportClientRunDeliversBinaryBranchUpdatedEvent(t *testing.T) {
	subscribeCh := make(chan *structpb.Struct, 1)

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := websocket.Accept(w, r, nil)
		if err != nil {
			return
		}

		serverClient := toki.NewWsClient(conn, 0, 0, protosocket.ParserMap())
		defer serverClient.Close()

		toki.AddRequestListenerTyped[*structpb.Struct, *structpb.Struct](&serverClient.Communicator, func(req *structpb.Struct) (*structpb.Struct, error) {
			env, err := protosocket.EnvelopeFromStruct(req)
			if err != nil {
				return nil, err
			}

			if env.Channel == "event" && env.Action == "event.subscribe" {
				subscribeCh <- req

				resEnv := protosocket.Envelope{
					ProtocolVersion: protosocket.ProtocolVersion,
					ID:              "res-id",
					CorrelationID:   env.ID,
					Type:            "response",
					Channel:         "event",
					Action:          "event.subscribe",
				}
				resStruct, _ := resEnv.ToStruct()

				go func() {
					time.Sleep(50 * time.Millisecond)
					eventEnv := branchUpdatedEnvelope()
					eventStruct, _ := eventEnv.ToStruct()
					_ = serverClient.Send(eventStruct)
				}()

				return resStruct, nil
			}
			return nil, errors.New("unexpected request")
		})

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

	wsURL := strings.Replace(srv.URL, "http", "ws", 1)
	transport := &protoSocketTransport{url: wsURL}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	handlerCh := make(chan gitoevents.BranchUpdatedEvent, 1)
	handler := func(ctx context.Context, ev gitoevents.BranchUpdatedEvent) {
		handlerCh <- ev
	}

	client, err := gitoevents.NewClient(transport, gitoevents.Options{
		RepoID:  "nomadcode",
		Branch:  "develop",
		Handler: handler,
	})
	if err != nil {
		t.Fatalf("NewClient failed: %v", err)
	}

	runErrCh := make(chan error, 1)
	go func() {
		runErrCh <- client.Run(ctx)
	}()

	// Verify subscribe request was received on the server
	select {
	case reqStruct := <-subscribeCh:
		env, err := protosocket.EnvelopeFromStruct(reqStruct)
		if err != nil {
			t.Fatalf("invalid subscribe request: %v", err)
		}
		if env.Channel != "event" || env.Action != "event.subscribe" {
			t.Fatalf("unexpected subscribe request: %+v", env)
		}
		events, ok := env.Payload["events"].([]any)
		if !ok || len(events) != 1 || events[0] != "branch.updated" {
			t.Fatalf("unexpected subscribe request events: %v", env.Payload["events"])
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for subscribe request")
	}

	// Verify event was received by the handler
	select {
	case ev := <-handlerCh:
		if ev.RepoID != "nomadcode" || ev.Branch != "develop" || ev.After != "new-sha" {
			t.Fatalf("unexpected branch updated event: %+v", ev)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for branch updated event")
	}

	// Cancel context and verify client exits cleanly
	cancel()
	select {
	case runErr := <-runErrCh:
		if runErr != nil {
			t.Fatalf("expected nil error on clean exit, got: %v", runErr)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for Client.Run to exit")
	}
}

// TestProtoSocketTransportClientRunDropsMalformedAndMismatchedStructs implements REVIEW_API-3.
// It verifies that malformed or mismatched structpb.Struct envelopes are safely ignored
// without causing handler side effects or breaking the stream.
func TestProtoSocketTransportClientRunDropsMalformedAndMismatchedStructs(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := websocket.Accept(w, r, nil)
		if err != nil {
			return
		}

		serverClient := toki.NewWsClient(conn, 0, 0, protosocket.ParserMap())
		defer serverClient.Close()

		toki.AddRequestListenerTyped[*structpb.Struct, *structpb.Struct](&serverClient.Communicator, func(req *structpb.Struct) (*structpb.Struct, error) {
			env, _ := protosocket.EnvelopeFromStruct(req)
			if env.Action == "event.subscribe" {
				resEnv := protosocket.Envelope{
					ProtocolVersion: protosocket.ProtocolVersion,
					ID:              "res-id",
					CorrelationID:   env.ID,
					Type:            "response",
					Channel:         "event",
					Action:          "event.subscribe",
				}
				resStruct, _ := resEnv.ToStruct()

				go func() {
					time.Sleep(50 * time.Millisecond)

					// 1. Wrong channel/action
					wrongEnv := branchUpdatedEnvelope()
					wrongEnv.Channel = "wrong"
					wrongEnv.Action = "wrong"
					wrongStruct, _ := wrongEnv.ToStruct()
					_ = serverClient.Send(wrongStruct)

					// 2. Missing payload
					missingEnv := branchUpdatedEnvelope()
					missingEnv.Payload = nil
					missingStruct, _ := missingEnv.ToStruct()
					_ = serverClient.Send(missingStruct)

					// 3. Malformed changed_files
					malformedEnv := branchUpdatedEnvelope()
					malformedEnv.Payload["changed_files"] = "not-a-list"
					malformedStruct, _ := malformedEnv.ToStruct()
					_ = serverClient.Send(malformedStruct)

					// Wait a bit to ensure Client.handleEnvelope processes drops
					time.Sleep(50 * time.Millisecond)

					// 4. Send valid target event
					validEnv := branchUpdatedEnvelope()
					validStruct, _ := validEnv.ToStruct()
					_ = serverClient.Send(validStruct)
				}()

				return resStruct, nil
			}
			return nil, errors.New("unexpected request")
		})

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

	wsURL := strings.Replace(srv.URL, "http", "ws", 1)
	transport := &protoSocketTransport{url: wsURL}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	handlerCh := make(chan gitoevents.BranchUpdatedEvent, 10)
	handler := func(ctx context.Context, ev gitoevents.BranchUpdatedEvent) {
		handlerCh <- ev
	}

	client, err := gitoevents.NewClient(transport, gitoevents.Options{
		RepoID:  "nomadcode",
		Branch:  "develop",
		Handler: handler,
	})
	if err != nil {
		t.Fatalf("NewClient failed: %v", err)
	}

	go func() {
		_ = client.Run(ctx)
	}()

	// We expect exactly one event (the valid on-target one) to reach the handler.
	select {
	case ev := <-handlerCh:
		if ev.RepoID != "nomadcode" || ev.Branch != "develop" {
			t.Fatalf("unexpected event: %+v", ev)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for valid event")
	}

	// Verify no other events were received
	select {
	case extra := <-handlerCh:
		t.Fatalf("received unexpected extra event: %+v", extra)
	default:
		// success
	}
}