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"
	"github.com/nomadcode/nomadcode-core/internal/scheduler"
	"github.com/nomadcode/nomadcode-core/internal/workitem"
	"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
	}
}

// channelEnqueuer captures EnqueueRoadmapCreationSync calls on a buffered
// channel so e2e tests can wait for enqueue completion without polling.
type channelEnqueuer struct {
	ch chan scheduler.RoadmapCreationSyncJobArgs
}

// recordingScanner records the BranchUpdatedEvent it receives and returns
// matchedScanOutput() with Result.Revision.Revision overridden to ev.After.
// This ties the e2e job's RoadmapRevision to the wire event's actual after
// value, so a revision handoff break would be caught by the test assertion.
type recordingScanner struct {
	eventCh chan gitoevents.BranchUpdatedEvent
}

func newRecordingScanner() *recordingScanner {
	return &recordingScanner{eventCh: make(chan gitoevents.BranchUpdatedEvent, 4)}
}

func (s *recordingScanner) Scan(_ context.Context, ev gitoevents.BranchUpdatedEvent) (ScanOutput, error) {
	s.eventCh <- ev
	out := matchedScanOutput()
	out.Result.Revision.Revision = ev.After
	return out, nil
}

func (e *channelEnqueuer) EnqueueRoadmapCreationSync(_ context.Context, args scheduler.RoadmapCreationSyncJobArgs) error {
	e.ch <- args
	return nil
}

// gotoWireServer returns an httptest.Server that speaks binary proto-socket,
// responds to the event.subscribe request, and broadcasts each env in events
// after a 50 ms delay. The caller owns closing the server.
func gotoWireServer(t *testing.T, events []protosocket.Envelope) *httptest.Server {
	t.Helper()
	return 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 != gitoevents.EventChannel || env.Action != gitoevents.SubscribeAction {
				return nil, errors.New("unexpected request")
			}
			resEnv := protosocket.Envelope{
				ProtocolVersion: protosocket.ProtocolVersion,
				ID:              "res-1",
				CorrelationID:   env.ID,
				Type:            "response",
				Channel:         gitoevents.EventChannel,
				Action:          gitoevents.SubscribeAction,
			}
			resStruct, _ := resEnv.ToStruct()
			go func() {
				time.Sleep(50 * time.Millisecond)
				for _, ev := range events {
					s, _ := ev.ToStruct()
					_ = serverClient.Send(s)
					// Give the bridge time to finish Handle (MarkProcessed) before
					// sending the next event so dedup is observable.
					time.Sleep(100 * time.Millisecond)
				}
			}()
			return resStruct, nil
		})
		<-r.Context().Done()
	}))
}

// TestWireConsumerEventEnqueuesCreationSyncOnce verifies the full wire path:
// binary proto-socket transport → gitoevents.Client → Bridge.Handle → Enqueuer.
// EnqueueRoadmapCreationSync must be called exactly once with the expected args.
// The recordingScanner reflects ev.After into RoadmapRevision, so this test
// also verifies that the wire event's revision is forwarded through the path.
func TestWireConsumerEventEnqueuesCreationSyncOnce(t *testing.T) {
	jobCh := make(chan scheduler.RoadmapCreationSyncJobArgs, 4)
	enq := &channelEnqueuer{ch: jobCh}
	reader := &fakeReader{item: workitem.WorkItem{DescriptionHTML: "<p>html body</p>"}}
	scanner := newRecordingScanner()

	bridge, err := NewBridge(scanner, reader, enq,
		NewInMemoryRevisionStore(), BridgeConfig{
			RepoID:      "nomadcode",
			Branch:      "develop",
			TodoStateID: "state-todo",
		}, nil)
	if err != nil {
		t.Fatalf("NewBridge: %v", err)
	}

	srv := gotoWireServer(t, []protosocket.Envelope{branchUpdatedEnvelope()})
	defer srv.Close()

	wsURL := strings.Replace(srv.URL, "http", "ws", 1)
	client, err := NewRunner(wsURL, "nomadcode", "develop", bridge, nil)
	if err != nil {
		t.Fatalf("NewRunner: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = client.Run(ctx) }()

	var job scheduler.RoadmapCreationSyncJobArgs
	select {
	case job = <-jobCh:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for EnqueueRoadmapCreationSync")
	}

	// Validate the scanner received the wire event with expected fields.
	// By the time jobCh has a value, scanner.eventCh already has the event.
	var scannedEv gitoevents.BranchUpdatedEvent
	select {
	case scannedEv = <-scanner.eventCh:
	default:
		t.Fatal("recordingScanner was not called with the wire event")
	}
	if scannedEv.RepoID != "nomadcode" || scannedEv.Branch != "develop" {
		t.Fatalf("scanner event repo/branch: got %q/%q", scannedEv.RepoID, scannedEv.Branch)
	}
	if scannedEv.Before != "old-sha" || scannedEv.After != "new-sha" {
		t.Fatalf("scanner event before/after: got %q/%q", scannedEv.Before, scannedEv.After)
	}
	if len(scannedEv.MilestoneChangedFiles()) == 0 {
		t.Fatal("scanner event: expected at least one milestone changed file")
	}

	// Job args assertions.
	if job.Expected.WorkItemID != "wi-123" {
		t.Fatalf("Expected.WorkItemID: got %q, want wi-123", job.Expected.WorkItemID)
	}
	if job.Ref.Provider != "plane" || job.Ref.ID != "wi-123" || job.Ref.Project != "proj-1" {
		t.Fatalf("Ref: got %+v", job.Ref)
	}
	if job.TodoStateID != "state-todo" {
		t.Fatalf("TodoStateID: got %q", job.TodoStateID)
	}
	// recordingScanner sets Revision.Revision to ev.After; the bridge copies it
	// into RoadmapRevision, so this asserts the wire revision is forwarded.
	if job.RoadmapRevision != "new-sha" {
		t.Fatalf("RoadmapRevision: got %q, want new-sha (wire event after)", job.RoadmapRevision)
	}
	if job.MilestoneMarkdown != milestoneMarkdownWithIdentity {
		t.Fatalf("MilestoneMarkdown mismatch")
	}
	if job.OriginalBody != "<p>html body</p>" {
		t.Fatalf("OriginalBody: got %q", job.OriginalBody)
	}
	if job.ExternalSource != "nomadcode" || job.ExternalID != "wi-123" {
		t.Fatalf("ExternalTags: source=%q id=%q", job.ExternalSource, job.ExternalID)
	}
	if job.ProviderRevision != "" {
		t.Fatalf("ProviderRevision: got %q, want empty (mutable Plane state must not be a revision)", job.ProviderRevision)
	}

	// One event was sent; verify no second enqueue occurred.
	select {
	case extra := <-jobCh:
		t.Fatalf("unexpected second enqueue: %+v", extra)
	default:
	}
}

// TestWireConsumerDuplicateEventDoesNotReEnqueue sends the same branch.updated
// event twice over the wire. The bridge's InMemoryRevisionStore must deduplicate
// at the (repo, branch, after) key so only one creation sync job is enqueued.
func TestWireConsumerDuplicateEventDoesNotReEnqueue(t *testing.T) {
	jobCh := make(chan scheduler.RoadmapCreationSyncJobArgs, 4)
	enq := &channelEnqueuer{ch: jobCh}
	reader := &fakeReader{item: workitem.WorkItem{DescriptionText: "body"}}

	bridge, err := NewBridge(newRecordingScanner(), reader, enq,
		NewInMemoryRevisionStore(), BridgeConfig{
			RepoID:      "nomadcode",
			Branch:      "develop",
			TodoStateID: "state-todo",
		}, nil)
	if err != nil {
		t.Fatalf("NewBridge: %v", err)
	}

	ev := branchUpdatedEnvelope()
	srv := gotoWireServer(t, []protosocket.Envelope{ev, ev})
	defer srv.Close()

	wsURL := strings.Replace(srv.URL, "http", "ws", 1)
	client, err := NewRunner(wsURL, "nomadcode", "develop", bridge, nil)
	if err != nil {
		t.Fatalf("NewRunner: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = client.Run(ctx) }()

	// First event must produce exactly one enqueue.
	select {
	case <-jobCh:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for first EnqueueRoadmapCreationSync")
	}

	// The duplicate (same after="new-sha") must be silently dropped by the
	// InMemoryRevisionStore. Wait long enough for the wire + bridge path to
	// complete, then assert no second job arrived.
	select {
	case extra := <-jobCh:
		t.Fatalf("duplicate event must not re-enqueue: %+v", extra)
	case <-time.After(500 * time.Millisecond):
		// No second enqueue — expected.
	}
}