Restore durable log ingest and typed plugin projections

2026-09-02 10:20:30 +08:00
parent 40ac46ba17
commit 6018d8f0fc
61 changed files with 809 additions and 3157 deletions
@@ -0,0 +1,224 @@
+package service
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"browser.local/platform/domain"
+	"browser.local/platform/repo"
+)
+
+func (svc *CoreService) scheduleDuePluginQueryProjectionJobs(claim domain.RunJobClaim, stamp time.Time) error {
+	if !containsString(claim.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
+		return nil
+	}
+	endpoint, err := svc.store.RunEndpoints().Get(claim.RunEndpointID)
+	if err != nil || !containsString(endpoint.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
+		return err
+	}
+	jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: claim.RunEndpointID})
+	if err != nil {
+		return err
+	}
+	instances, err := svc.store.ServerInstances().List(domain.ServerInstanceFilter{RunEndpointID: claim.RunEndpointID})
+	if err != nil {
+		return err
+	}
+	for _, instance := range instances {
+		if instance.State != domain.ServerInstanceStateRunning || strings.TrimSpace(instance.PluginID) == "" {
+			continue
+		}
+		plugin, pluginErr := svc.store.GamePlugins().Get(instance.PluginID)
+		if pluginErr != nil || !plugin.Permissions.RemoteAccess || !containsString(plugin.RequiredRunCapabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) || !containsString(plugin.RemoteAccess.RunCapabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
+			continue
+		}
+		for _, template := range plugin.GameClientBridge.QueryTemplates {
+			if template.PollIntervalSeconds <= 0 || len(template.Projections) == 0 || !pluginQueryTemplateTransportReady(plugin, endpoint, template) {
+				continue
+			}
+			interval := time.Duration(template.PollIntervalSeconds) * time.Second
+			prefix := pluginQueryPollPrefix(instance.ID, plugin.ID, template.Key)
+			if pluginQueryPollActiveOrFresh(jobs, prefix, stamp, interval) {
+				continue
+			}
+			bucket := stamp.Unix() / int64(template.PollIntervalSeconds)
+			idempotencyKey := fmt.Sprintf("%s%d", prefix, bucket)
+			inputs := map[string]string{"templateKey": template.Key, "maxRows": fmt.Sprint(template.MaxRows)}
+			if template.SQLRef != "" {
+				inputs["sqlRef"] = template.SQLRef
+			}
+			job := domain.Job{
+				ID:               jobIDFromParts("job-plugin-query-poll", instance.ID, idempotencyKey),
+				ServerInstanceID: instance.ID,
+				RunEndpointID:    instance.RunEndpointID,
+				Capability:       domain.JobCapabilityRemoteRunDBSQLiteQuery,
+				TargetKey:        template.TargetKey,
+				InputRef:         fmt.Sprintf("input://plugin-query-poll/%s/%s", instance.ID, template.Key),
+				IdempotencyKey:   idempotencyKey,
+				Progress:         domain.JobProgress{Percent: 0, Message: "plugin query projection poll queued"},
+				RetryPolicy:      domain.JobRetryPolicy{MaxAttempts: 1, InitialBackoffSeconds: 2, MaxBackoffSeconds: 2},
+				ExecutionInput:   domain.JobExecutionInput{WorkspaceScope: svc.runtimeProfileScope(instance.ID), RemoteAdapterKey: template.TransportKey, RemoteAdapterKind: string(domain.RemoteAdapterDatabase), TimeoutSeconds: template.TimeoutSeconds, PluginID: plugin.ID, Inputs: inputs},
+			}
+			if _, createErr := svc.CreateJob(job); createErr != nil {
+				return createErr
+			}
+		}
+	}
+	return nil
+}
+
+func pluginQueryTemplateTransportReady(plugin domain.GamePlugin, endpoint domain.RunEndpoint, template domain.GameClientBridgeQueryTemplateDeclaration) bool {
+	if template.Engine != "sqlite" || template.TransportKey == "" || template.TargetKey == "" {
+		return false
+	}
+	for _, profile := range plugin.RuntimeProfiles.TransportProfiles {
+		if profile.Key == template.TransportKey && profile.Kind == "sqlite" && profile.TargetKey == template.TargetKey && containsString(profile.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
+			return containsString(endpoint.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery)
+		}
+	}
+	return false
+}
+
+func pluginQueryPollPrefix(serverID, pluginID, templateKey string) string {
+	return fmt.Sprintf("plugin-query-poll:%s:%s:%s:", serverID, pluginID, templateKey)
+}
+
+func pluginQueryPollActiveOrFresh(jobs []domain.Job, prefix string, stamp time.Time, interval time.Duration) bool {
+	for _, job := range jobs {
+		if !strings.HasPrefix(job.IdempotencyKey, prefix) {
+			continue
+		}
+		if !isTerminalJobState(job.State) {
+			return true
+		}
+		freshAt := job.TerminalAt
+		if freshAt.IsZero() {
+			freshAt = job.UpdatedAt
+		}
+		if !freshAt.IsZero() && stamp.Sub(freshAt) < interval {
+			return true
+		}
+	}
+	return false
+}
+
+func (svc *CoreService) projectPluginQueryJobResult(job domain.Job, stamp time.Time) error {
+	if job.State != domain.JobStateSucceeded || job.Capability != domain.JobCapabilityRemoteRunDBSQLiteQuery || job.ExecutionResult.Kind != "sqlite.query" {
+		return nil
+	}
+	templateKey := strings.TrimSpace(job.ExecutionInput.Inputs["templateKey"])
+	if templateKey == "" || strings.TrimSpace(job.ServerInstanceID) == "" {
+		return nil
+	}
+	instance, err := svc.store.ServerInstances().Get(job.ServerInstanceID)
+	if err != nil {
+		return err
+	}
+	plugin, err := svc.store.GamePlugins().Get(instance.PluginID)
+	if err != nil {
+		return err
+	}
+	template, ok := pluginQueryTemplateByKey(plugin, templateKey)
+	if !ok || len(template.Projections) == 0 {
+		return nil
+	}
+	rows, err := pluginQueryRows(job.ExecutionResult.Content)
+	if err != nil {
+		return err
+	}
+	mutationsByCollection := map[string]map[string]domain.PluginDataMutation{}
+	for _, projection := range template.Projections {
+		if projection.RowPath != "rows" {
+			continue
+		}
+		collectionMutations := mutationsByCollection[projection.Collection]
+		if collectionMutations == nil {
+			collectionMutations = map[string]domain.PluginDataMutation{}
+			mutationsByCollection[projection.Collection] = collectionMutations
+		}
+		for _, row := range rows {
+			if projection.MatchField != "" && strings.TrimSpace(fmt.Sprint(row[projection.MatchField])) != projection.MatchValue {
+				continue
+			}
+			value := pluginQueryProjectionValue(projection, row, stamp)
+			key, keyErr := pluginDataRowKey(value, projection.UpsertKeys)
+			if keyErr != nil {
+				return keyErr
+			}
+			if projection.MergeExisting {
+				if existing, existingErr := svc.store.PluginDataRecords().Get(pluginDataID(instance.ID, plugin.ID, projection.Collection, key)); existingErr == nil {
+					value = mergePluginDataValues(existing.Value, value)
+				} else if !errors.Is(existingErr, repo.ErrNotFound) {
+					return existingErr
+				}
+			}
+			collectionMutations[key] = domain.PluginDataMutation{Operation: domain.PluginDataMutationPut, Key: key, Value: value}
+		}
+	}
+	for collection, keyed := range mutationsByCollection {
+		mutations := make([]domain.PluginDataMutation, 0, len(keyed))
+		for _, mutation := range keyed {
+			mutations = append(mutations, mutation)
+		}
+		if len(mutations) == 0 {
+			continue
+		}
+		if _, err := svc.applyPluginDataTransaction(domain.PluginDataTransaction{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: collection, Mutations: mutations}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func pluginQueryTemplateByKey(plugin domain.GamePlugin, templateKey string) (domain.GameClientBridgeQueryTemplateDeclaration, bool) {
+	for _, template := range plugin.GameClientBridge.QueryTemplates {
+		if template.Key == templateKey {
+			return template, true
+		}
+	}
+	return domain.GameClientBridgeQueryTemplateDeclaration{}, false
+}
+
+func pluginQueryRows(content string) ([]map[string]any, error) {
+	var payload struct {
+		Rows []map[string]any `json:"rows"`
+	}
+	decoder := json.NewDecoder(bytes.NewBufferString(content))
+	decoder.UseNumber()
+	if err := decoder.Decode(&payload); err != nil {
+		return nil, validationError("sqlite query result content is not a row payload")
+	}
+	return payload.Rows, nil
+}
+
+func pluginQueryProjectionValue(projection domain.GameClientBridgeQueryProjectionDeclaration, row map[string]any, observedAt time.Time) map[string]any {
+	value := map[string]any{}
+	if len(projection.FieldMappings) == 0 {
+		for key, item := range row {
+			value[key] = item
+		}
+	} else {
+		for destination, source := range projection.FieldMappings {
+			value[destination] = row[source]
+		}
+	}
+	for key, fixed := range projection.FixedValues {
+		value[key] = renderQueryProjectionTemplate(fixed, row)
+	}
+	if projection.ObservedAtField != "" {
+		value[projection.ObservedAtField] = observedAt.UTC().Format(time.RFC3339Nano)
+	}
+	return value
+}
+
+func renderQueryProjectionTemplate(template string, row map[string]any) string {
+	result := template
+	for key, value := range row {
+		result = strings.ReplaceAll(result, "{{"+key+"}}", fmt.Sprint(value))
+	}
+	return result
+}