package api import ( "net/http" "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) assertStatus(t, performJSON(t, router, http.MethodGet, "/api/v1/server-instances/server-1/logs/events?historyLimit=2", nil), http.StatusNotFound) } 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) assertStatus(t, performJSON(t, router, http.MethodGet, "/api/v1/server-instances/server-1/logs/events?historyLimit=2", nil), http.StatusNotFound) } 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, } }