browser/platform/api/log_ingest_handlers_test.go

package api

import (
	"context"
	"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)

	recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=2")
	assertStatus(t, recorder, http.StatusOK)
	body := recorder.Body.String()
	if !strings.Contains(recorder.Header().Get("Content-Type"), "text/event-stream") || !strings.Contains(body, "event: stream") || !strings.Contains(body, "event: log") || !strings.Contains(body, "event: ready") || !strings.Contains(body, `"seq":1`) || !strings.Contains(body, `"seq":2`) {
		t.Fatalf("expected stream, history log, and ready SSE events, headers=%v body=%s", recorder.Header(), body)
	}
}

func TestLogEventsSSEUsesServerWideNewestHistory(t *testing.T) {
	router := newTestRouter()
	hello := createLogIngestAPIFixtures(t, router)
	postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{
		ID: "log-2", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stderr",
		StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default",
	})
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-1", "stdout", 1, 2, 0)), http.StatusOK)
	assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-2", "stderr", 1, 2, 10)), http.StatusOK)

	recorder := performCancelledSSE(t, router, "/api/v1/server-instances/server-1/logs/events?historyLimit=2")
	assertStatus(t, recorder, http.StatusOK)
	body := recorder.Body.String()
	if strings.Count(body, "event: log") != 2 || !strings.Contains(body, `"streamId":"log-2"`) || strings.Contains(body, `"streamId":"log-1"`) {
		t.Fatalf("expected server-wide newest history across streams, body=%s", body)
	}
}

func performCancelledSSE(t *testing.T, router http.Handler, path string) *httptest.ResponseRecorder {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	req := httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx)
	rec := httptest.NewRecorder()
	router.ServeHTTP(rec, req)
	return rec
}

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 validLogBatchRequest(t *testing.T, sessionToken string, firstSeq uint64, lastSeq uint64) dto.LogBatchIngestRequest {
	return validLogBatchRequestForStream(t, sessionToken, "log-1", "stdout", firstSeq, lastSeq, 0)
}

func validLogBatchRequestForStream(t *testing.T, sessionToken string, streamID string, streamKey string, firstSeq uint64, lastSeq uint64, timestampOffset int) 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, timestampOffset+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:      streamID,
		ServerInstanceID: "server-1",
		StreamKey:        streamKey,
		Source:           domain.LogStreamSourceProcess,
		FirstSeq:         firstSeq,
		LastSeq:          lastSeq,
		Compression:      "none",
		Checksum:         checksum,
		Entries:          entries,
	}
}