gito/services/core/internal/controlplane/router_test.go

package controlplane

import (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"testing"

	"git.toki-labs.com/toki/gito/services/core/internal/config"
	"git.toki-labs.com/toki/gito/services/core/internal/protosocket"
)

func TestHealthz(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())
	req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusOK {
		t.Fatalf("status: got %d", rec.Code)
	}
	var body map[string]string
	if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
		t.Fatalf("decode: %v", err)
	}
	if body["status"] != "ok" {
		t.Fatalf("body: %#v", body)
	}
}

func TestProtoSocketRegistry(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())
	req := httptest.NewRequest(http.MethodGet, "/proto-socket", nil)
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusOK {
		t.Fatalf("status: got %d", rec.Code)
	}
	var body ProtoSocketRegistry
	if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
		t.Fatalf("decode: %v", err)
	}
	if body.Transport != "proto-socket" {
		t.Fatalf("transport: got %q", body.Transport)
	}
	channels := make(map[string]ProtoSocketChannel, len(body.Channels))
	for _, channel := range body.Channels {
		channels[channel.Name] = channel
	}
	assertPlaceholderChannel(t, channels, "operation", []string{"create", "cancel", "get", "stream"})
	assertPlaceholderChannel(t, channels, "change_request", []string{"create", "update", "comment", "request_review", "merge", "close", "list"})
	assertChannelActions(t, channels, "event", "mvp", map[string]string{
		"subscribe": "mvp",
		"list":      "mvp",
		"ack":       "placeholder",
	})
}

func assertPlaceholderChannel(t *testing.T, channels map[string]ProtoSocketChannel, name string, actions []string) {
	t.Helper()
	channel, ok := channels[name]
	if !ok {
		t.Fatalf("missing channel %q", name)
	}
	if channel.Status != "placeholder" {
		t.Fatalf("channel %q status: got %q", name, channel.Status)
	}
	actionStatuses := make(map[string]string, len(channel.Actions))
	for _, action := range channel.Actions {
		actionStatuses[action.Name] = action.Status
	}
	for _, action := range actions {
		if actionStatuses[action] != "placeholder" {
			t.Fatalf("channel %q action %q status: got %q", name, action, actionStatuses[action])
		}
	}
}

func assertChannelActions(t *testing.T, channels map[string]ProtoSocketChannel, name, status string, actions map[string]string) {
	t.Helper()
	channel, ok := channels[name]
	if !ok {
		t.Fatalf("missing channel %q", name)
	}
	if channel.Status != status {
		t.Fatalf("channel %q status: got %q want %q", name, channel.Status, status)
	}
	actionStatuses := make(map[string]string, len(channel.Actions))
	for _, action := range channel.Actions {
		actionStatuses[action.Name] = action.Status
	}
	for action, want := range actions {
		if actionStatuses[action] != want {
			t.Fatalf("channel %q action %q status: got %q want %q", name, action, actionStatuses[action], want)
		}
	}
}

func TestForgejoPushCreatesWatchedBranchEvent(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`)

	req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{
		"ref": "refs/heads/develop",
		"before": "111",
		"after": "222",
		"repository": {"name": "nomadcode", "full_name": "toki/nomadcode"},
		"commits": [{"id": "222", "modified": ["agent-roadmap/phase/a/milestones/b.md"]}]
	}`))
	req.Header.Set("X-Forgejo-Event", "push")
	req.Header.Set("X-Forgejo-Delivery", "delivery-1")
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusAccepted {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
	var pushed map[string]any
	if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil {
		t.Fatalf("decode push response: %v", err)
	}
	if pushed["matched"] != true {
		t.Fatalf("expected push to match listener: %#v", pushed)
	}

	eventsReq := httptest.NewRequest(http.MethodGet, "/api/events", nil)
	eventsRec := httptest.NewRecorder()
	router.ServeHTTP(eventsRec, eventsReq)
	if eventsRec.Code != http.StatusOK {
		t.Fatalf("events status: got %d", eventsRec.Code)
	}
	var body map[string][]map[string]any
	if err := json.NewDecoder(eventsRec.Body).Decode(&body); err != nil {
		t.Fatalf("decode events: %v", err)
	}
	if len(body["events"]) != 1 {
		t.Fatalf("events: %#v", body)
	}
	event := body["events"][0]
	if event["type"] != "branch.updated" || event["provider"] != "forgejo" || event["delivery_id"] != "delivery-1" {
		t.Fatalf("unexpected event: %#v", event)
	}
	revision := event["revision"].(map[string]any)
	if revision["repo_id"] != "nomadcode" || revision["branch"] != "develop" || revision["before"] != "111" || revision["after"] != "222" {
		t.Fatalf("unexpected revision: %#v", revision)
	}
}

func TestForgejoPushIgnoresUnwatchedBranch(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`)

	req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{
		"ref": "refs/heads/feature/a",
		"before": "111",
		"after": "222",
		"repository": {"name": "nomadcode", "full_name": "toki/nomadcode"}
	}`))
	req.Header.Set("X-Forgejo-Event", "push")
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusAccepted {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
	var pushed map[string]any
	if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil {
		t.Fatalf("decode push response: %v", err)
	}
	if pushed["matched"] != false {
		t.Fatalf("expected push to be ignored: %#v", pushed)
	}
	if _, ok := pushed["event"]; ok {
		t.Fatalf("unmatched push must not expose an emitted event: %#v", pushed)
	}
}

func TestForgejoPushRequiresSignatureBeforeIgnoringUnsupportedEvent(t *testing.T) {
	router := NewRouter(config.Config{
		AppEnv:               "test",
		ProtoSocketPath:      "/proto-socket",
		ForgejoWebhookSecret: "forgejo-secret",
	}, slog.Default())

	req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push", bytes.NewBufferString(`{"unsupported":true}`))
	req.Header.Set("X-Forgejo-Event", "repository")
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusUnauthorized {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
}

func TestProviderCallbackRecordsReceivedWebhookEvent(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	req := httptest.NewRequest(http.MethodPost, "/callbacks/providers/fake", bytes.NewBufferString(`{
		"action": "ping",
		"token": "top-secret"
	}`))
	req.Header.Set("X-Gito-Provider-Event", "ping")
	req.Header.Set("X-Gito-Delivery", "fake-delivery-1")
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusAccepted {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
	if bytes.Contains(rec.Body.Bytes(), []byte("top-secret")) ||
		bytes.Contains(rec.Body.Bytes(), []byte(`"token"`)) {
		t.Fatalf("provider callback response leaked raw payload: %s", rec.Body.String())
	}
	var created map[string]any
	if err := json.NewDecoder(rec.Body).Decode(&created); err != nil {
		t.Fatalf("decode callback response: %v", err)
	}
	event, ok := created["event"].(map[string]any)
	if !ok {
		t.Fatalf("missing event: %#v", created)
	}
	if event["type"] != "provider.webhook.received" ||
		event["provider"] != "fake" ||
		event["delivery_id"] != "fake-delivery-1" {
		t.Fatalf("unexpected event: %#v", event)
	}
	webhook, ok := event["webhook"].(map[string]any)
	if !ok {
		t.Fatalf("missing webhook metadata: %#v", event)
	}
	if webhook["event_type"] != "ping" || webhook["external_id"] != "fake-delivery-1" {
		t.Fatalf("unexpected webhook metadata: %#v", webhook)
	}
	if _, ok := webhook["payload_size"].(float64); !ok {
		t.Fatalf("payload_size missing: %#v", webhook)
	}

	eventsReq := httptest.NewRequest(http.MethodGet, "/api/events", nil)
	eventsRec := httptest.NewRecorder()
	router.ServeHTTP(eventsRec, eventsReq)
	if eventsRec.Code != http.StatusOK {
		t.Fatalf("events status: got %d", eventsRec.Code)
	}
	if bytes.Contains(eventsRec.Body.Bytes(), []byte("top-secret")) ||
		bytes.Contains(eventsRec.Body.Bytes(), []byte(`"token"`)) {
		t.Fatalf("event list leaked raw payload: %s", eventsRec.Body.String())
	}
	var listed map[string][]map[string]any
	if err := json.NewDecoder(eventsRec.Body).Decode(&listed); err != nil {
		t.Fatalf("decode events response: %v", err)
	}
	if len(listed["events"]) != 1 {
		t.Fatalf("events: %#v", listed)
	}
	if listed["events"][0]["type"] != "provider.webhook.received" {
		t.Fatalf("unexpected listed event: %#v", listed["events"][0])
	}
}

func TestWebhookSubscriptionRegistersFilterWithoutLeakingSecretRef(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	req := httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{
		"name": "nomadcode-develop-wakeup",
		"target_url": "https://consumer.example.invalid/gito/events",
		"events": ["branch.updated"],
		"repo_id": "nomadcode",
		"branch": "develop",
		"secret_ref": "secret://gito/webhooks/nomadcode"
	}`))
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusCreated {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
	if bytes.Contains(rec.Body.Bytes(), []byte("secret://gito/webhooks/nomadcode")) ||
		bytes.Contains(rec.Body.Bytes(), []byte("secret_ref")) ||
		bytes.Contains(rec.Body.Bytes(), []byte("consumer.example.invalid")) {
		t.Fatalf("subscription response leaked private subscription input: %s", rec.Body.String())
	}
	var created map[string]map[string]any
	if err := json.NewDecoder(rec.Body).Decode(&created); err != nil {
		t.Fatalf("decode subscription response: %v", err)
	}
	subscription := created["subscription"]
	if subscription["id"] == "" {
		t.Fatalf("missing subscription id: %#v", subscription)
	}
	if subscription["name"] != "nomadcode-develop-wakeup" ||
		subscription["repo_id"] != "nomadcode" ||
		subscription["branch"] != "develop" {
		t.Fatalf("unexpected subscription payload: %#v", subscription)
	}
	events, ok := subscription["events"].([]any)
	if !ok || len(events) != 1 || events[0] != "branch.updated" {
		t.Fatalf("events: %#v", subscription["events"])
	}

	listReq := httptest.NewRequest(http.MethodGet, "/api/webhook-subscriptions", nil)
	listRec := httptest.NewRecorder()
	router.ServeHTTP(listRec, listReq)
	if listRec.Code != http.StatusOK {
		t.Fatalf("list status: got %d body=%s", listRec.Code, listRec.Body.String())
	}
	if bytes.Contains(listRec.Body.Bytes(), []byte("secret://gito/webhooks/nomadcode")) ||
		bytes.Contains(listRec.Body.Bytes(), []byte("secret_ref")) ||
		bytes.Contains(listRec.Body.Bytes(), []byte("consumer.example.invalid")) {
		t.Fatalf("subscription list leaked private subscription input: %s", listRec.Body.String())
	}
}

func TestWebhookSubscriptionEnvSecretRefSignsDelivery(t *testing.T) {
	const secret = "router-secret"
	t.Setenv("GITO_WEBHOOK_SECRET", secret)

	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())
	registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`)

	subReq := httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{
		"name": "nomadcode-develop-wakeup",
		"target_url": "`+server.URL+`",
		"events": ["branch.updated"],
		"repo_id": "nomadcode",
		"branch": "develop",
		"secret_ref": "env:GITO_WEBHOOK_SECRET"
	}`))
	subRec := httptest.NewRecorder()
	router.ServeHTTP(subRec, subReq)
	if subRec.Code != http.StatusCreated {
		t.Fatalf("subscription status: got %d body=%s", subRec.Code, subRec.Body.String())
	}

	pushReq := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{
		"ref": "refs/heads/develop",
		"before": "111",
		"after": "222",
		"repository": {"name": "nomadcode", "full_name": "toki/nomadcode"},
		"commits": [{"id": "222", "modified": ["agent-roadmap/phase/a/milestones/b.md"]}]
	}`))
	pushReq.Header.Set("X-Forgejo-Event", "push")
	pushReq.Header.Set("X-Forgejo-Delivery", "delivery-1")
	pushRec := httptest.NewRecorder()
	router.ServeHTTP(pushRec, pushReq)
	if pushRec.Code != http.StatusAccepted {
		t.Fatalf("push status: got %d body=%s", pushRec.Code, pushRec.Body.String())
	}

	consumer.mu.Lock()
	received := append([]webhookReception(nil), consumer.received...)
	consumer.mu.Unlock()
	if len(received) != 1 {
		t.Fatalf("expected one webhook delivery, got %d", len(received))
	}
	rec := received[0]
	if rec.eventType != "branch.updated" || rec.deliveryID == "" {
		t.Fatalf("unexpected delivery headers: event=%q delivery=%q", rec.eventType, rec.deliveryID)
	}
	mac := hmac.New(sha256.New, []byte(secret))
	_, _ = mac.Write(rec.bodyRaw)
	wantSignature := "sha256=" + hex.EncodeToString(mac.Sum(nil))
	if rec.signature != wantSignature {
		t.Fatalf("signature: got %q want %q", rec.signature, wantSignature)
	}
}

func TestWebhookSubscriptionValidatesTargetAndEvents(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	req := httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{
		"name": "invalid",
		"target_url": "file:///tmp/event",
		"events": ["branch.updated"]
	}`))
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusBadRequest {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}

	req = httptest.NewRequest(http.MethodPost, "/api/webhook-subscriptions", bytes.NewBufferString(`{
		"name": "missing-events",
		"target_url": "https://consumer.example.invalid/gito/events"
	}`))
	rec = httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusBadRequest {
		t.Fatalf("missing events status: got %d body=%s", rec.Code, rec.Body.String())
	}
}

func TestEventSubscribeRegistersConnectionFilter(t *testing.T) {
	dispatcher := protosocket.NewDispatcher()
	runtime := NewRuntime(nil)
	subscriber := &recordingSubscriber{}
	registerProtoSocketHandlers(dispatcher, runtime, subscriber)

	response := dispatcher.Dispatch(t.Context(), protosocket.Envelope{
		ID:      "msg-1",
		Type:    "request",
		Channel: "event",
		Action:  "event.subscribe",
		Payload: map[string]any{
			"events":  []any{"branch.updated"},
			"repo_id": "nomadcode",
			"branch":  "develop",
		},
		Meta: map[string]any{
			"connection_id": "conn-1",
		},
	})

	if response.Type != "response" {
		t.Fatalf("response type: got %q error=%+v", response.Type, response.Error)
	}
	if subscriber.connectionID != "conn-1" {
		t.Fatalf("connection id: got %q", subscriber.connectionID)
	}
	if len(subscriber.subscription.Events) != 1 || subscriber.subscription.Events[0] != "branch.updated" {
		t.Fatalf("events: %#v", subscriber.subscription.Events)
	}
	if subscriber.subscription.RepoID != "nomadcode" || subscriber.subscription.Branch != "develop" {
		t.Fatalf("subscription: %+v", subscriber.subscription)
	}
}

func TestForgejoPushTreatsMissingEventHeaderAsPush(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`)

	// X-Forgejo-Event header missing but valid push payload should still be processed.
	req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{
		"ref": "refs/heads/develop",
		"before": "111",
		"after": "222",
		"repository": {"name": "nomadcode", "full_name": "toki/nomadcode"},
		"commits": [{"id": "222", "modified": ["README.md"]}]
	}`))
	req.Header.Set("X-Forgejo-Delivery", "delivery-missing-event")
	// Note: X-Forgejo-Event header is intentionally absent.
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusAccepted {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
	var pushed map[string]any
	if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil {
		t.Fatalf("decode push response: %v", err)
	}
	if pushed["matched"] != true {
		t.Fatalf("expected push to match listener with missing event header: %#v", pushed)
	}
}

func TestForgejoPushRejectsMalformedPayload(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(`{malformed json`))
	req.Header.Set("X-Forgejo-Event", "push")
	req.Header.Set("X-Forgejo-Delivery", "delivery-malformed")
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusBadRequest {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
}

func TestForgejoPushReturns400WhenNoRepoIDSource(t *testing.T) {
	router := NewRouter(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, slog.Default())

	// Valid payload but repository has no name or full_name.
	req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push", bytes.NewBufferString(`{
		"ref": "refs/heads/develop",
		"before": "111",
		"after": "222",
		"repository": {},
		"commits": [{"id": "222"}]
	}`))
	req.Header.Set("X-Forgejo-Event", "push")
	req.Header.Set("X-Forgejo-Delivery", "delivery-no-repo")
	rec := httptest.NewRecorder()

	router.ServeHTTP(rec, req)

	if rec.Code != http.StatusBadRequest {
		t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
	}
}

func TestEventSubscribeRequiresConnectionID(t *testing.T) {
	dispatcher := protosocket.NewDispatcher()
	runtime := NewRuntime(nil)
	registerProtoSocketHandlers(dispatcher, runtime, &recordingSubscriber{})

	response := dispatcher.Dispatch(t.Context(), protosocket.Envelope{
		ID:      "msg-1",
		Type:    "request",
		Channel: "event",
		Action:  "event.subscribe",
	})

	if response.Type != "error" || response.Error == nil || response.Error.Code != "event.missing_connection" {
		t.Fatalf("expected missing connection error, got %+v", response)
	}
}

func registerBranchListener(t *testing.T, router http.Handler, body string) {
	t.Helper()
	req := httptest.NewRequest(http.MethodPost, "/api/listeners/branches", bytes.NewBufferString(body))
	rec := httptest.NewRecorder()
	router.ServeHTTP(rec, req)
	if rec.Code != http.StatusCreated {
		t.Fatalf("listener status: got %d body=%s", rec.Code, rec.Body.String())
	}
}

type recordingSubscriber struct {
	connectionID string
	subscription protosocket.EventSubscription
}

func (r *recordingSubscriber) Subscribe(connectionID string, subscription protosocket.EventSubscription) protosocket.EventSubscription {
	r.connectionID = connectionID
	r.subscription = subscription.Normalized()
	return r.subscription
}