browser/platform/api/log_ingest_handlers_test.go

package api

import (
	"bufio"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"browser.local/platform/domain"
	"browser.local/platform/dto"
	"browser.local/platform/validator"
)

func TestLogIngestAPIWorkflow(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 2)

	ackRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch)
	assertStatus(t, ackRecorder, http.StatusOK)
	ack := decodeBody[dto.LogBatchIngestResponse](t, ackRecorder)
	if !ack.Accepted || ack.AcceptedFrom != 1 || ack.AcceptedTo != 2 || ack.LatestSeq != 2 {
		t.Fatalf("unexpected ack: %+v", ack)
	}

	stream := getJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams/log-1")
	if stream.LatestSeq != 2 {
		t.Fatalf("expected latest seq update, got %+v", stream)
	}

	queryRecorder := performJSON(t, router, http.MethodPost, "/api/v1/log-streams/query", dto.LogStreamCursorRequest{LogStreamID: "log-1", AfterSeq: 0, Limit: 1})
	assertStatus(t, queryRecorder, http.StatusOK)
	query := decodeBody[dto.LogStreamCursorResponse](t, queryRecorder)
	if len(query.Entries) != 1 || query.Entries[0].Seq != 1 || query.NextSeq != 1 || query.LatestSeq != 2 {
		t.Fatalf("unexpected query: %+v", query)
	}
}

func TestLogEventsSSEReplaysHistory(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 2)
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch), http.StatusOK)

	server := httptest.NewServer(router)
	defer server.Close()
	client := server.Client()
	client.Timeout = 2 * time.Second
	response, err := client.Get(server.URL + "/api/v1/server-instances/server-1/logs/events?historyLimit=2")
	if err != nil {
		t.Fatalf("open log event stream: %v", err)
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK || !strings.HasPrefix(response.Header.Get("Content-Type"), "text/event-stream") {
		t.Fatalf("unexpected event stream response: status=%d content-type=%q", response.StatusCode, response.Header.Get("Content-Type"))
	}
	body := readSSEUntil(t, response, "event: ready")
	for _, fragment := range []string{"event: stream", "event: log", `"streamId":"log-1"`, `"seq":1`, `"seq":2`} {
		if !strings.Contains(body, fragment) {
			t.Fatalf("expected SSE body to contain %q, got:\n%s", fragment, body)
		}
	}
}

func TestLogIngestAPIDuplicateAndErrors(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	batch := validLogBatchRequest(t, hello.SessionToken, 1, 1)

	first := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch)
	assertStatus(t, first, http.StatusOK)
	duplicate := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", batch)
	assertStatus(t, duplicate, http.StatusOK)
	duplicateAck := decodeBody[dto.LogBatchIngestResponse](t, duplicate)
	if !duplicateAck.Duplicate {
		t.Fatalf("expected duplicate ack, got %+v", duplicateAck)
	}

	gap := validLogBatchRequest(t, hello.SessionToken, 3, 3)
	gapRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", gap)
	assertErrorResponse(t, gapRecorder, http.StatusBadRequest, errorCodeValidation)

	missingQuery := performJSON(t, router, http.MethodPost, "/api/v1/log-streams/query", dto.LogStreamCursorRequest{LogStreamID: "missing", Limit: 1})
	assertErrorResponse(t, missingQuery, http.StatusNotFound, errorCodeNotFound)
}

func createLogIngestAPIFixtures(t *testing.T, router http.Handler) dto.RunControlHelloResponse {
	t.Helper()
	helloRequest := validRunControlHelloRequest()
	helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, "process.install", "process.start", "process.stop", "logs.read")
	helloRequest.CapabilityReport.Fingerprint = "cap-logs"
	hello := decodeBody[dto.RunControlHelloResponse](t, performRunControlHello(t, router, helloRequest))
	adminSession := createAdminSession(t, router)
	postJSON[dto.GamePluginResponse](t, router, "/api/v1/game-plugins", validGamePluginRequest())
	postJSONWithAuth[dto.ServerInstanceResponse](t, router, "/api/v1/server-instances", dto.ServerInstanceCreateRequest{ID: "server-1", PluginID: "server.scum", RunEndpointID: "run-local", Name: "SCUM #1"}, adminSession)
	postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{
		ID:               "log-1",
		ServerInstanceID: "server-1",
		Source:           domain.LogStreamSourceProcess,
		StreamKey:        "stdout",
		StorageBackend:   domain.LogStorageBackendLocalSegments,
		RetentionPolicy:  "default",
	})
	return hello
}

func readSSEUntil(t *testing.T, response *http.Response, marker string) string {
	t.Helper()
	reader := bufio.NewReader(response.Body)
	var body strings.Builder
	for !strings.Contains(body.String(), marker) {
		line, err := reader.ReadString('\n')
		if err != nil {
			t.Fatalf("read event stream: %v\n%s", err, body.String())
		}
		body.WriteString(line)
	}
	return body.String()
}

func validLogBatchRequest(t *testing.T, sessionToken string, firstSeq uint64, lastSeq uint64) dto.LogBatchIngestRequest {
	t.Helper()
	entries := make([]dto.LogEntryBody, 0, lastSeq-firstSeq+1)
	domainEntries := make([]domain.LogEntry, 0, lastSeq-firstSeq+1)
	for seq := firstSeq; seq <= lastSeq; seq++ {
		entry := dto.LogEntryBody{Seq: seq, Timestamp: time.Date(2026, 7, 3, 12, 0, int(seq), 0, time.UTC), Level: "info", Line: "line"}
		entries = append(entries, entry)
		domainEntries = append(domainEntries, domain.LogEntry{Seq: entry.Seq, Timestamp: entry.Timestamp, Level: entry.Level, Line: entry.Line})
	}
	checksum, err := validator.LogEntriesChecksum(domainEntries)
	if err != nil {
		t.Fatalf("checksum entries: %v", err)
	}
	return dto.LogBatchIngestRequest{
		RunEndpointID:    "run-local",
		SessionToken:     sessionToken,
		LogStreamID:      "log-1",
		ServerInstanceID: "server-1",
		StreamKey:        "stdout",
		Source:           domain.LogStreamSourceProcess,
		FirstSeq:         firstSeq,
		LastSeq:          lastSeq,
		Compression:      "none",
		Checksum:         checksum,
		Entries:          entries,
	}
}