browser/platform/service/plugin_log_projection.go

package service

import (
	"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 stream.Source != domain.LogStreamSourceProcess || len(entries) == 0 {
		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(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
	}
	announcementAlreadyQueued := false
	announcementIdempotencyKey := ""
	if projection.Presence != nil {
		announcementIdempotencyKey = fmt.Sprintf("log-projection:%s:%s:%d", projection.Key, key, observedAt.Unix()/int64(projection.Presence.ActiveWindowSeconds))
		_, commandErr := svc.store.GameClientBridgeCommands().GetByIdempotency(instance.ID, "system:log-projection", projection.Presence.Announcement.CommandType, announcementIdempotencyKey)
		if commandErr == nil {
			announcementAlreadyQueued = true
		} else if !errors.Is(commandErr, repo.ErrNotFound) {
			return commandErr
		}
	}
	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(*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
		}
	}
	if projection.Presence != nil && !announcementAlreadyQueued {
		announcement := projection.Presence.Announcement
		template := announcement.ReturningTextTemplate
		if isNew || sameObservation {
			template = announcement.NewTextTemplate
		}
		requestText := renderLogProjectionTemplate(template, captures)
		expiresAt := svc.now().Add(gameClientBridgeCommandTimeout(plugin, announcement.CommandType))
		if _, err := svc.queueGameClientBridgeCommand("system:log-projection", domain.GameClientBridgeQueueRequest{
			ServerInstanceID: instance.ID,
			PluginID:         plugin.ID,
			ProfileKey:       announcement.ProfileKey,
			CommandType:      announcement.CommandType,
			Payload:          map[string]any{announcement.TextField: requestText},
			IdempotencyKey:   announcementIdempotencyKey,
			Priority:         100,
			ExpiresAt:        expiresAt,
		}); err != nil {
			return err
		}
	}
	return nil
}

func gameClientBridgeCommandTimeout(plugin domain.GamePlugin, commandType string) time.Duration {
	for _, declaration := range plugin.GameClientBridge.Commands {
		if declaration.Type == commandType && declaration.TimeoutSeconds > 0 {
			return time.Duration(declaration.TimeoutSeconds) * time.Second
		}
	}
	return time.Minute
}

func pluginLogProjectionValue(target domain.GameClientBridgeLogProjectionTargetDeclaration, captures map[string]string, observedAt time.Time) map[string]any {
	value := make(map[string]any, len(target.CaptureMappings)+len(target.FixedValues)+1)
	for destination, capture := range target.CaptureMappings {
		value[destination] = 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 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 == "<nil>" {
		return time.Time{}, false
	}
	parsed, err := time.Parse(time.RFC3339Nano, text)
	return parsed, err == nil
}