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 }