package service import ( "crypto/sha256" "encoding/hex" "errors" "fmt" "regexp" "strings" "time" "browser.local/platform/domain" "browser.local/platform/repo" ) type pluginLogSequenceState struct { StepIndex int Captures map[string]string LastSeq uint64 } func (svc *CoreService) projectPluginLogBatch(stream domain.LogStream, entries []domain.LogEntry) error { if len(entries) == 0 || (stream.Source != domain.LogStreamSourceProcess && stream.Source != domain.LogStreamSourceFile && stream.Source != domain.LogStreamSourceManagementProgram) { return nil } instance, err := svc.store.ServerInstances().Get(stream.ServerInstanceID) if err != nil { return err } plugin, err := svc.store.GamePlugins().Get(instance.PluginID) if err != nil { return err } for _, projection := range plugin.GameClientBridge.LogProjections { if !containsString(projection.StreamKeys, stream.StreamKey) { continue } for _, entry := range entries { captures, complete, matchErr := svc.advancePluginLogProjection(stream, projection, entry) if matchErr != nil { return matchErr } if complete { observedAt := entry.Timestamp if observedAt.IsZero() { observedAt = svc.now() } if err := svc.applyPluginLogProjection(instance, plugin, projection, captures, observedAt); err != nil { return err } } } } return nil } func (svc *CoreService) advancePluginLogProjection(stream domain.LogStream, projection domain.GameClientBridgeLogProjectionDeclaration, entry domain.LogEntry) (map[string]string, bool, error) { if len(projection.Steps) == 0 { return nil, false, nil } stateKey := strings.Join([]string{stream.ServerInstanceID, stream.ID, stream.LogSessionID, projection.Key}, "\x00") svc.logProjectionMu.Lock() defer svc.logProjectionMu.Unlock() states := svc.logProjectionStates[stateKey] if states == nil { states = map[string]pluginLogSequenceState{} svc.logProjectionStates[stateKey] = states } nextStates := make(map[string]pluginLogSequenceState, len(states)+1) var completed map[string]string for correlationKey, state := range states { if state.StepIndex < 1 || state.StepIndex >= len(projection.Steps) { continue } if projection.MaxInterveningLines >= 0 && state.LastSeq > 0 && entry.Seq > state.LastSeq+uint64(projection.MaxInterveningLines)+1 { continue } match, err := matchLogProjectionStep(projection.Steps[state.StepIndex].Pattern, entry.Line) if err != nil { return nil, false, err } if match == nil { nextStates[correlationKey] = state continue } merged, ok := mergeLogCaptures(state.Captures, match) if !ok || !correlationCapturesAgree(state.Captures, match, projection.CorrelationFields) { continue } if state.StepIndex+1 == len(projection.Steps) { completed = merged continue } nextKey := logCorrelationKey(merged, projection.CorrelationFields) nextStates[nextKey] = pluginLogSequenceState{StepIndex: state.StepIndex + 1, Captures: merged, LastSeq: entry.Seq} } first, err := matchLogProjectionStep(projection.Steps[0].Pattern, entry.Line) if err != nil { return nil, false, err } if first != nil { if len(projection.Steps) == 1 { completed = first } else { key := logCorrelationKey(first, projection.CorrelationFields) nextStates[key] = pluginLogSequenceState{StepIndex: 1, Captures: first, LastSeq: entry.Seq} } } svc.logProjectionStates[stateKey] = nextStates return completed, completed != nil, nil } func matchLogProjectionStep(pattern, line string) (map[string]string, error) { expression, err := regexp.Compile(pattern) if err != nil { return nil, validationError("declared log projection pattern is invalid") } values := expression.FindStringSubmatch(line) if values == nil { return nil, nil } result := make(map[string]string) for index, name := range expression.SubexpNames() { if index > 0 && name != "" && index < len(values) { result[name] = values[index] } } return result, nil } func mergeLogCaptures(existing, incoming map[string]string) (map[string]string, bool) { merged := make(map[string]string, len(existing)+len(incoming)) for key, value := range existing { merged[key] = value } for key, value := range incoming { if previous, exists := merged[key]; exists && previous != value { return nil, false } merged[key] = value } return merged, true } func correlationCapturesAgree(existing, incoming map[string]string, fields []string) bool { for _, field := range fields { left, leftExists := existing[field] right, rightExists := incoming[field] if leftExists && rightExists && left != right { return false } } return true } func logCorrelationKey(captures map[string]string, fields []string) string { parts := make([]string, len(fields)) for index, field := range fields { parts[index] = captures[field] } return strings.Join(parts, "\x1f") } func (svc *CoreService) applyPluginLogProjection(instance domain.ServerInstance, plugin domain.GamePlugin, projection domain.GameClientBridgeLogProjectionDeclaration, captures map[string]string, observedAt time.Time) error { value := pluginLogProjectionValue(instance.ID, projection.Target, captures, observedAt) key, err := pluginDataRowKey(value, projection.Target.UpsertKeys) if err != nil { return err } existing, getErr := svc.store.PluginDataRecords().Get(pluginDataID(instance.ID, plugin.ID, projection.Target.Collection, key)) isNew := errors.Is(getErr, repo.ErrNotFound) if getErr != nil && !isNew { return getErr } insideWindow := false sameObservation := false if projection.Presence != nil && !isNew { if previous, ok := pluginDataTimestamp(existing.Value[projection.Presence.TimestampField]); ok { if observedAt.Before(previous) { return nil } sameObservation = observedAt.Equal(previous) insideWindow = observedAt.Sub(previous) < time.Duration(projection.Presence.ActiveWindowSeconds)*time.Second } } if insideWindow && !sameObservation { return nil } if !isNew { value = mergePluginDataValues(existing.Value, value) } if _, err := svc.applyPluginDataTransaction(domain.PluginDataTransaction{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: projection.Target.Collection, Mutations: []domain.PluginDataMutation{{Operation: domain.PluginDataMutationPut, Key: key, Value: value}}}); err != nil { return err } if projection.Presence != nil && projection.Presence.ActivityTarget != nil { activity := pluginLogProjectionValue(instance.ID, *projection.Presence.ActivityTarget, captures, observedAt) activityKey, keyErr := pluginDataRowKey(activity, projection.Presence.ActivityTarget.UpsertKeys) if keyErr != nil { return keyErr } if _, applyErr := svc.applyPluginDataTransaction(domain.PluginDataTransaction{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: projection.Presence.ActivityTarget.Collection, Mutations: []domain.PluginDataMutation{{Operation: domain.PluginDataMutationPut, Key: activityKey, Value: activity}}}); applyErr != nil { return applyErr } } return nil } func pluginLogProjectionValue(serverID string, target domain.GameClientBridgeLogProjectionTargetDeclaration, captures map[string]string, observedAt time.Time) map[string]any { value := make(map[string]any, len(target.CaptureMappings)+len(target.HashMappings)+len(target.FixedValues)+1) for destination, capture := range target.CaptureMappings { value[destination] = captures[capture] } for destination, capture := range target.HashMappings { value[destination] = logProjectionCorrelationHash(serverID, captures[capture]) } for key, fixed := range target.FixedValues { value[key] = renderLogProjectionTemplate(fixed, captures) } if target.ObservedAtField != "" { value[target.ObservedAtField] = observedAt.UTC().Format(time.RFC3339Nano) } return value } func logProjectionCorrelationHash(serverID, value string) string { digest := sha256.Sum256([]byte(serverID + "\x00" + value)) return hex.EncodeToString(digest[:]) } func renderLogProjectionTemplate(template string, captures map[string]string) string { result := template for key, value := range captures { result = strings.ReplaceAll(result, "{{"+key+"}}", value) } return result } func pluginDataTimestamp(value any) (time.Time, bool) { text := strings.TrimSpace(fmt.Sprint(value)) if text == "" || text == "" { return time.Time{}, false } parsed, err := time.Parse(time.RFC3339Nano, text) return parsed, err == nil }