package service import ( "bytes" "encoding/json" "fmt" "strings" "time" "browser.local/platform/domain" ) 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 } 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 }