From bc10cd7400a1f6bb2edb99bd28946aa673222226 Mon Sep 17 00:00:00 2001 From: npc0-hue Date: Fri, 28 Aug 2026 18:00:03 +0800 Subject: [PATCH] Remove legacy plugin query projections --- .../service/plugin_lifecycle_projection.go | 12 +- platform/service/plugin_query_projection.go | 224 ------------------ 2 files changed, 10 insertions(+), 226 deletions(-) delete mode 100644 platform/service/plugin_query_projection.go diff --git a/platform/service/plugin_lifecycle_projection.go b/platform/service/plugin_lifecycle_projection.go index eccb407..30e8ff8 100644 --- a/platform/service/plugin_lifecycle_projection.go +++ b/platform/service/plugin_lifecycle_projection.go @@ -64,7 +64,7 @@ func (svc *CoreService) applyPluginBulkProjection(instance domain.ServerInstance func pluginBulkProjectionValues(fixedValues map[string]string, stamp time.Time, row map[string]any, observedAtField string) map[string]any { value := make(map[string]any, len(fixedValues)+1) for key, fixed := range fixedValues { - value[key] = renderQueryProjectionTemplate(fixed, row) + value[key] = renderPluginValueTemplate(fixed, row) } if observedAtField != "" { value[observedAtField] = stamp.UTC().Format(time.RFC3339Nano) @@ -78,10 +78,18 @@ func pluginBulkActivityValue(target domain.GameClientBridgeBulkActivityTargetDec value[destination] = row[source] } for key, fixed := range target.FixedValues { - value[key] = renderQueryProjectionTemplate(fixed, row) + value[key] = renderPluginValueTemplate(fixed, row) } if target.ObservedAtField != "" { value[target.ObservedAtField] = stamp.UTC().Format(time.RFC3339Nano) } return value } + +func renderPluginValueTemplate(template string, row map[string]any) string { + result := template + for key, value := range row { + result = strings.ReplaceAll(result, "{{"+key+"}}", fmt.Sprint(value)) + } + return result +} diff --git a/platform/service/plugin_query_projection.go b/platform/service/plugin_query_projection.go deleted file mode 100644 index 1c2d09c..0000000 --- a/platform/service/plugin_query_projection.go +++ /dev/null @@ -1,224 +0,0 @@ -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 -}