browser/platform/service/log_ingest_test.go

package service

import (
	"path/filepath"
	"strings"
	"testing"
	"time"

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

func TestCoreServiceIngestsLogBatchAndQueriesCursor(t *testing.T) {
	svc, sessionToken := newRegisteredLogIngestService(t)
	createLogStreamFixture(t, svc)
	batch := validLogBatch(t, sessionToken, 1, 2)

	ack, err := svc.IngestLogBatch(batch)
	if err != nil {
		t.Fatalf("ingest log batch: %v", err)
	}
	if !ack.Accepted || ack.AcceptedFrom != 1 || ack.AcceptedTo != 2 || ack.LatestSeq != 2 {
		t.Fatalf("unexpected ack: %+v", ack)
	}
	stream, err := svc.GetLogStream("log-1")
	if err != nil {
		t.Fatalf("get log stream: %v", err)
	}
	if stream.LatestSeq != 2 {
		t.Fatalf("expected latest seq 2, got %+v", stream)
	}

	query, err := svc.QueryLogStream(domain.LogStreamCursorQuery{LogStreamID: "log-1", AfterSeq: 1, Limit: 10})
	if err != nil {
		t.Fatalf("query log stream: %v", err)
	}
	if len(query.Entries) != 1 || query.Entries[0].Seq != 2 || query.NextSeq != 2 || query.LatestSeq != 2 {
		t.Fatalf("unexpected query result: %+v", query)
	}
}

func TestCoreServiceLogBatchDuplicateAck(t *testing.T) {
	svc, sessionToken := newRegisteredLogIngestService(t)
	createLogStreamFixture(t, svc)
	batch := validLogBatch(t, sessionToken, 1, 2)

	if _, err := svc.IngestLogBatch(batch); err != nil {
		t.Fatalf("ingest first batch: %v", err)
	}
	ack, err := svc.IngestLogBatch(batch)
	if err != nil {
		t.Fatalf("ingest duplicate batch: %v", err)
	}
	if !ack.Duplicate || ack.LatestSeq != 2 {
		t.Fatalf("expected duplicate ack, got %+v", ack)
	}
}

func TestCoreServiceRejectsOutOfOrderAndConflictingLogBatches(t *testing.T) {
	svc, sessionToken := newRegisteredLogIngestService(t)
	createLogStreamFixture(t, svc)
	gap := validLogBatch(t, sessionToken, 2, 2)

	_, err := svc.IngestLogBatch(gap)
	if err == nil || !strings.Contains(err.Error(), "firstSeq") {
		t.Fatalf("expected out-of-order rejection, got %v", err)
	}

	batch := validLogBatch(t, sessionToken, 1, 2)
	if _, err := svc.IngestLogBatch(batch); err != nil {
		t.Fatalf("ingest first batch: %v", err)
	}
	conflict := batch
	conflict.Entries[0].Line = "changed"
	conflict.Checksum = checksumForEntries(t, conflict.Entries)
	_, err = svc.IngestLogBatch(conflict)
	if err == nil || !strings.Contains(err.Error(), "conflicts") {
		t.Fatalf("expected conflicting duplicate rejection, got %v", err)
	}
}

func TestCoreServiceRejectsMissingLogStream(t *testing.T) {
	svc, sessionToken := newRegisteredLogIngestService(t)
	_, err := svc.IngestLogBatch(validLogBatch(t, sessionToken, 1, 1))
	if err == nil {
		t.Fatal("expected missing stream error")
	}
}

func TestFileLogBodyStoreReloadsBatchesAndCursorEntries(t *testing.T) {
	rootDir := filepath.Join(t.TempDir(), "logs")
	store, err := NewFileLogBodyStore(rootDir)
	if err != nil {
		t.Fatalf("create file log store: %v", err)
	}
	entries := []domain.LogEntry{
		{Seq: 1, Timestamp: time.Date(2026, 7, 3, 12, 0, 1, 0, time.UTC), Level: "info", Line: "one"},
		{Seq: 2, Timestamp: time.Date(2026, 7, 3, 12, 0, 2, 0, time.UTC), Level: "warn", Line: "two"},
	}
	record := domain.LogBatchRecord{
		Checksum: checksumForEntries(t, entries),
		FirstSeq: 1,
		LastSeq:  2,
		Entries:  entries,
	}
	if err := store.AppendBatch("log-1", record); err != nil {
		t.Fatalf("append batch: %v", err)
	}

	reloaded, err := NewFileLogBodyStore(rootDir)
	if err != nil {
		t.Fatalf("reload file log store: %v", err)
	}
	got, exists, err := reloaded.GetBatch("log-1", 1)
	if err != nil {
		t.Fatalf("get reloaded batch: %v", err)
	}
	if !exists || got.Checksum != record.Checksum || got.LastSeq != 2 {
		t.Fatalf("unexpected reloaded batch: exists=%v record=%+v", exists, got)
	}
	selected, nextSeq, err := reloaded.Query("log-1", 1, 10)
	if err != nil {
		t.Fatalf("query reloaded entries: %v", err)
	}
	if len(selected) != 1 || selected[0].Seq != 2 || selected[0].Line != "two" || nextSeq != 2 {
		t.Fatalf("unexpected reloaded query: entries=%+v next=%d", selected, nextSeq)
	}
}

func newRegisteredLogIngestService(t *testing.T) (*CoreService, string) {
	t.Helper()
	svc := newTestCoreService()
	plugin, endpoint := createPluginAndRunEndpoint(t, svc)
	if _, err := svc.CreateServerInstance(domain.ServerInstance{
		ID:            "server-1",
		PluginID:      plugin.ID,
		RunEndpointID: endpoint.ID,
		Name:          "SCUM #1",
	}); err != nil {
		t.Fatalf("create server instance: %v", err)
	}
	hello, err := svc.RegisterRunHello(validRunControlHello())
	if err != nil {
		t.Fatalf("register run hello: %v", err)
	}
	return svc, hello.SessionToken
}

func createLogStreamFixture(t *testing.T, svc *CoreService) domain.LogStream {
	t.Helper()
	stream, err := svc.CreateLogStream(domain.LogStream{
		ID:               "log-1",
		ServerInstanceID: "server-1",
		Source:           domain.LogStreamSourceProcess,
		StreamKey:        "stdout",
		StorageBackend:   domain.LogStorageBackendLocalSegments,
		RetentionPolicy:  "default",
	})
	if err != nil {
		t.Fatalf("create log stream: %v", err)
	}
	return stream
}

func validLogBatch(t *testing.T, sessionToken string, firstSeq uint64, lastSeq uint64) domain.LogBatchIngest {
	t.Helper()
	entries := make([]domain.LogEntry, 0, lastSeq-firstSeq+1)
	for seq := firstSeq; seq <= lastSeq; seq++ {
		entries = append(entries, domain.LogEntry{
			Seq:       seq,
			Timestamp: time.Date(2026, 7, 3, 12, 0, int(seq), 0, time.UTC),
			Level:     "info",
			Line:      "line",
		})
	}
	return domain.LogBatchIngest{
		RunEndpointID:    "run-local",
		SessionToken:     sessionToken,
		LogStreamID:      "log-1",
		ServerInstanceID: "server-1",
		StreamKey:        "stdout",
		Source:           domain.LogStreamSourceProcess,
		FirstSeq:         firstSeq,
		LastSeq:          lastSeq,
		Compression:      "none",
		Checksum:         checksumForEntries(t, entries),
		Entries:          entries,
	}
}

func checksumForEntries(t *testing.T, entries []domain.LogEntry) string {
	t.Helper()
	checksum, err := validator.LogEntriesChecksum(entries)
	if err != nil {
		t.Fatalf("checksum entries: %v", err)
	}
	return checksum
}