browser/platform/service/log_ingest.go

package service

import (
	"errors"
	"strings"
	"time"

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

const defaultLogQueryLimit = 100

func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogBatchIngestResult, error) {
	batch = domain.CopyLogBatchIngest(batch)
	projectionBatch := domain.CopyLogBatchIngest(batch)
	if err := validator.ValidateLogBatchIngest(batch); err != nil {
		return domain.LogBatchIngestResult{}, err
	}
	if err := svc.validateRunSession(batch.RunEndpointID, batch.SessionToken); err != nil {
		return domain.LogBatchIngestResult{}, err
	}

	stamp := svc.now()
	stream, err := svc.store.LogStreams().Get(batch.LogStreamID)
	if errors.Is(err, repo.ErrNotFound) {
		if repairErr := svc.ensureJobLogStreamForBatch(batch, stamp); repairErr == nil {
			stream, err = svc.store.LogStreams().Get(batch.LogStreamID)
		}
	}
	if err != nil {
		return domain.LogBatchIngestResult{}, err
	}
	if err := validateLogBatchStream(batch, stream); err != nil {
		return domain.LogBatchIngestResult{}, err
	}

	if batch.LastSeq <= stream.LatestSeq {
		record, exists, err := svc.logStore.GetBatch(batch.LogStreamID, batch.FirstSeq)
		if err != nil {
			return domain.LogBatchIngestResult{}, err
		}
		if exists && record.LastSeq == batch.LastSeq && logBatchRecordMatches(record, batch) {
			if err := svc.projectGamePlayerEvents(projectionBatch); err != nil {
				return domain.LogBatchIngestResult{}, err
			}
			if err := svc.projectGameMapTrajectoryEvents(projectionBatch); err != nil {
				return domain.LogBatchIngestResult{}, err
			}
			return domain.LogBatchIngestResult{
				Accepted:     true,
				LogStreamID:  batch.LogStreamID,
				AcceptedFrom: batch.FirstSeq,
				AcceptedTo:   batch.LastSeq,
				LatestSeq:    stream.LatestSeq,
				Duplicate:    true,
				ServerTime:   stamp,
			}, nil
		}
		return domain.LogBatchIngestResult{}, validationError("log batch conflicts with acknowledged range")
	}
	if stream.LatestSeq > 0 && batch.FirstSeq != stream.LatestSeq+1 {
		return domain.LogBatchIngestResult{}, validationError("log batch firstSeq must follow latest acknowledged sequence")
	}

	storedBatch := domain.CopyLogBatchIngest(batch)
	sanitizeGamePlayerNetworkFields(&storedBatch)
	record := domain.CopyLogBatchRecord(domain.LogBatchRecord{
		Checksum: batch.Checksum,
		FirstSeq: batch.FirstSeq,
		LastSeq:  batch.LastSeq,
		Entries:  storedBatch.Entries,
	})
	if err := svc.logStore.AppendBatch(batch.LogStreamID, record); err != nil {
		return domain.LogBatchIngestResult{}, err
	}
	stream.LatestSeq = batch.LastSeq
	stream.UpdatedAt = stamp
	if err := svc.store.LogStreams().Update(stream); err != nil {
		return domain.LogBatchIngestResult{}, err
	}
	if err := svc.projectGamePlayerEvents(projectionBatch); err != nil {
		return domain.LogBatchIngestResult{}, err
	}
	if err := svc.projectGameMapTrajectoryEvents(projectionBatch); err != nil {
		return domain.LogBatchIngestResult{}, err
	}
	svc.publishLogEvents(stream, storedBatch.Entries)
	return domain.LogBatchIngestResult{
		Accepted:     true,
		LogStreamID:  batch.LogStreamID,
		AcceptedFrom: batch.FirstSeq,
		AcceptedTo:   batch.LastSeq,
		LatestSeq:    stream.LatestSeq,
		ServerTime:   stamp,
	}, nil
}

func (svc *CoreService) ensureJobLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error {
	jobID, ok := jobIDFromLogBatch(batch)
	if !ok {
		return repo.ErrNotFound
	}
	job, err := svc.store.Jobs().Get(jobID)
	if err != nil {
		return err
	}
	if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID {
		return validationError("log batch job scope does not match stream")
	}
	return svc.ensureJobLogStreams(job, stamp)
}

func jobIDFromLogBatch(batch domain.LogBatchIngest) (string, bool) {
	streamKey := strings.TrimSpace(batch.StreamKey)
	if streamKey == "" || !strings.HasPrefix(batch.LogStreamID, "job.") {
		return "", false
	}
	suffix := "." + streamKey
	body := strings.TrimPrefix(batch.LogStreamID, "job.")
	if !strings.HasSuffix(body, suffix) {
		return "", false
	}
	jobID := strings.TrimSuffix(body, suffix)
	return jobID, strings.TrimSpace(jobID) != ""
}

func logBatchRecordMatches(record domain.LogBatchRecord, batch domain.LogBatchIngest) bool {
	if record.Checksum == batch.Checksum {
		return true
	}
	if len(record.Entries) == 1 && len(batch.Entries) == 1 {
		return batch.Checksum == validator.LogLineChecksum(record.Entries[0].Line)
	}
	return false
}

// sanitizeGamePlayerNetworkFields removes raw network material before the durable log body is written.
func sanitizeGamePlayerNetworkFields(batch *domain.LogBatchIngest) {
	for index := range batch.Entries {
		fields := batch.Entries[index].Fields
		if fields == nil {
			continue
		}
		if fields["eventType"] == "scum.login" {
			delete(fields, "networkFingerprint")
			delete(fields, "ip")
			delete(fields, "ipAddress")
		}
	}
}

func (svc *CoreService) QueryLogStream(query domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error) {
	if err := validator.ValidateLogStreamCursorQuery(query); err != nil {
		return domain.LogStreamCursorResult{}, err
	}
	stream, err := svc.store.LogStreams().Get(query.LogStreamID)
	if err != nil {
		return domain.LogStreamCursorResult{}, err
	}
	limit := query.Limit
	if limit == 0 {
		limit = defaultLogQueryLimit
	}

	selected, nextSeq, err := svc.logStore.Query(query.LogStreamID, query.AfterSeq, limit)
	if err != nil {
		return domain.LogStreamCursorResult{}, err
	}
	return domain.CopyLogStreamCursorResult(domain.LogStreamCursorResult{
		LogStreamID: query.LogStreamID,
		Entries:     selected,
		NextSeq:     nextSeq,
		LatestSeq:   stream.LatestSeq,
	}), nil
}

func validateLogBatchStream(batch domain.LogBatchIngest, stream domain.LogStream) error {
	if stream.ID != batch.LogStreamID {
		return validationError("logStreamId must match stream")
	}
	if stream.ServerInstanceID != batch.ServerInstanceID {
		return validationError("serverInstanceId must match stream")
	}
	if stream.StreamKey != batch.StreamKey {
		return validationError("streamKey must match stream")
	}
	if stream.Source != batch.Source {
		return validationError("source must match stream")
	}
	return nil
}