diff --git a/platform/service/job_channel.go b/platform/service/job_channel.go index 1cb31a1..dfc35a0 100644 --- a/platform/service/job_channel.go +++ b/platform/service/job_channel.go @@ -50,17 +50,7 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai } job, ok := firstEligibleSupportedJob(jobs, claim.Capabilities, stamp) if !ok { - if err := svc.scheduleDuePluginQueryProjectionJobs(claim, stamp); err != nil { - return domain.RunJobClaimResult{}, err - } - jobs, err = svc.store.Jobs().List(domain.JobFilter{RunEndpointID: claim.RunEndpointID}) - if err != nil { - return domain.RunJobClaimResult{}, err - } - job, ok = firstEligibleSupportedJob(jobs, claim.Capabilities, stamp) - if !ok { - return emptyJobClaim(claim.RunEndpointID, stamp), nil - } + return emptyJobClaim(claim.RunEndpointID, stamp), nil } leaseToken, err := randomToken() @@ -347,9 +337,6 @@ func (svc *CoreService) CompleteRunJob(result domain.RunJobResult) (domain.RunJo if err := svc.projectPluginOperationsJobResult(job, stamp); err != nil { return domain.RunJobResultResult{}, err } - if err := svc.projectPluginQueryJobResult(job, stamp); err != nil { - return domain.RunJobResultResult{}, err - } return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, result.LeaseToken), ServerTime: stamp}, nil } diff --git a/platform/service/plugin_data_rows.go b/platform/service/plugin_data_rows.go deleted file mode 100644 index 16d597c..0000000 --- a/platform/service/plugin_data_rows.go +++ /dev/null @@ -1,42 +0,0 @@ -package service - -import ( - "encoding/json" - "fmt" - "strings" - - "browser.local/platform/domain" -) - -func mergePluginDataValues(existing, incoming map[string]any) map[string]any { - merged := domain.CopyGameClientBridgePayload(existing) - if merged == nil { - merged = make(map[string]any, len(incoming)) - } - for key, value := range incoming { - merged[key] = value - } - return merged -} - -func pluginDataRowKey(value map[string]any, keys []string) (string, error) { - parts := make([]string, len(keys)) - for index, key := range keys { - item, exists := value[key] - if !exists || item == nil || strings.TrimSpace(fmt.Sprint(item)) == "" { - return "", validationError("declared plugin data row is missing an upsert key") - } - encoded, err := json.Marshal(item) - if err != nil { - return "", validationError("declared plugin data row upsert key is invalid") - } - parts[index] = string(encoded) - } - if len(parts) == 1 { - var key string - if err := json.Unmarshal([]byte(parts[0]), &key); err == nil { - return key, nil - } - } - return strings.Join(parts, "\x1f"), nil -} diff --git a/platform/service/plugin_data_test.go b/platform/service/plugin_data_test.go index e337512..de03a1b 100644 --- a/platform/service/plugin_data_test.go +++ b/platform/service/plugin_data_test.go @@ -158,6 +158,7 @@ func TestRunPollDoesNotSchedulePluginDataProjectionQueries(t *testing.T) { } func TestRunPollSchedulesAndProjectsDeclaredSQLiteQuery(t *testing.T) { + t.Skip("query polling and result projection are plugin-owned") svc, plugin, endpoint, session, instance := createSQLiteQueryBridgeFixture(t) plugin.GameClientBridge.QueryTemplates[0].PollIntervalSeconds = 3 plugin.GameClientBridge.QueryTemplates[0].Projections = []domain.GameClientBridgeQueryProjectionDeclaration{{ diff --git a/platform/service/plugin_lifecycle_projection.go b/platform/service/plugin_lifecycle_projection.go deleted file mode 100644 index eccb407..0000000 --- a/platform/service/plugin_lifecycle_projection.go +++ /dev/null @@ -1,87 +0,0 @@ -package service - -import ( - "fmt" - "strings" - "time" - - "browser.local/platform/domain" -) - -func (svc *CoreService) projectPluginLifecycleState(instance domain.ServerInstance, plugin domain.GamePlugin, capability string, result domain.JobExecutionResult, stamp time.Time) error { - if len(plugin.GameClientBridge.LifecycleProjections) == 0 { - return nil - } - for _, projection := range plugin.GameClientBridge.LifecycleProjections { - if !containsString(projection.Capabilities, capability) || len(projection.Target.FixedValues) == 0 { - continue - } - if len(projection.ProcessStates) > 0 && !containsString(projection.ProcessStates, strings.TrimSpace(result.ProcessState)) { - continue - } - if err := svc.applyPluginBulkProjection(instance, plugin, projection.Target, stamp); err != nil { - return err - } - } - return nil -} - -func (svc *CoreService) applyPluginBulkProjection(instance domain.ServerInstance, plugin domain.GamePlugin, target domain.GameClientBridgeBulkProjectionTargetDeclaration, stamp time.Time) error { - rows, err := svc.store.PluginDataRecords().List(domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: target.Collection}) - if err != nil { - return err - } - mutations := make([]domain.PluginDataMutation, 0, len(rows)) - activityMutations := []domain.PluginDataMutation{} - for _, row := range rows { - if strings.TrimSpace(fmt.Sprint(row.Value[target.MatchField])) != target.MatchValue { - continue - } - value := mergePluginDataValues(row.Value, pluginBulkProjectionValues(target.FixedValues, stamp, row.Value, target.ObservedAtField)) - mutations = append(mutations, domain.PluginDataMutation{Operation: domain.PluginDataMutationPut, Key: row.Key, Value: value}) - if target.ActivityTarget != nil { - activity := pluginBulkActivityValue(*target.ActivityTarget, row.Value, stamp) - activityKey, keyErr := pluginDataRowKey(activity, target.ActivityTarget.UpsertKeys) - if keyErr != nil { - return keyErr - } - activityMutations = append(activityMutations, domain.PluginDataMutation{Operation: domain.PluginDataMutationPut, Key: activityKey, Value: activity}) - } - } - if len(mutations) > 0 { - if _, err := svc.applyPluginDataTransaction(domain.PluginDataTransaction{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: target.Collection, Mutations: mutations}); err != nil { - return err - } - } - if len(activityMutations) > 0 && target.ActivityTarget != nil { - if _, err := svc.applyPluginDataTransaction(domain.PluginDataTransaction{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: target.ActivityTarget.Collection, Mutations: activityMutations}); err != nil { - return err - } - } - return nil -} - -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) - } - if observedAtField != "" { - value[observedAtField] = stamp.UTC().Format(time.RFC3339Nano) - } - return value -} - -func pluginBulkActivityValue(target domain.GameClientBridgeBulkActivityTargetDeclaration, row map[string]any, stamp time.Time) map[string]any { - value := make(map[string]any, len(target.RowMappings)+len(target.FixedValues)+1) - for destination, source := range target.RowMappings { - value[destination] = row[source] - } - for key, fixed := range target.FixedValues { - value[key] = renderQueryProjectionTemplate(fixed, row) - } - if target.ObservedAtField != "" { - value[target.ObservedAtField] = stamp.UTC().Format(time.RFC3339Nano) - } - return value -} 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 -} diff --git a/platform/service/server_lifecycle_projection.go b/platform/service/server_lifecycle_projection.go index 298c2a1..a8ee4d1 100644 --- a/platform/service/server_lifecycle_projection.go +++ b/platform/service/server_lifecycle_projection.go @@ -53,15 +53,6 @@ func (svc *CoreService) ReportRunLifecycle(report domain.RunLifecycleReport) (do } svc.publishLogProcessState(instance) } - if report.State == domain.JobStateSucceeded && !staleObservation { - plugin, pluginErr := svc.store.GamePlugins().Get(instance.PluginID) - if pluginErr != nil { - return domain.RunLifecycleReportResult{}, pluginErr - } - if err := svc.projectPluginLifecycleState(instance, plugin, report.Capability, report.ExecutionResult, stamp); err != nil { - return domain.RunLifecycleReportResult{}, err - } - } return domain.CopyRunLifecycleReportResult(domain.RunLifecycleReportResult{Accepted: true, RunEndpointID: report.RunEndpointID, ServerInstanceID: report.ServerInstanceID, ProjectedState: nextState, ServerTime: stamp}), nil } @@ -107,9 +98,6 @@ func (svc *CoreService) projectLifecycleJobResult(job domain.Job, stamp time.Tim } nextState, ok := lifecycleProjectedState(job.Capability, job.State, job.ExecutionResult) if !ok || job.ServerInstanceID == "" { - if job.State == domain.JobStateSucceeded && job.ServerInstanceID != "" { - return svc.projectPluginLifecycleStateForJob(job, stamp) - } return nil } instance, err := svc.store.ServerInstances().Get(job.ServerInstanceID) @@ -125,30 +113,9 @@ func (svc *CoreService) projectLifecycleJobResult(job domain.Job, stamp time.Tim return err } svc.publishLogProcessState(instance) - if job.State == domain.JobStateSucceeded { - plugin, err := svc.store.GamePlugins().Get(instance.PluginID) - if err != nil { - return err - } - if err := svc.projectPluginLifecycleState(instance, plugin, job.Capability, job.ExecutionResult, stamp); err != nil { - return err - } - } return nil } -func (svc *CoreService) projectPluginLifecycleStateForJob(job domain.Job, stamp time.Time) error { - 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 - } - return svc.projectPluginLifecycleState(instance, plugin, job.Capability, job.ExecutionResult, stamp) -} - func (svc *CoreService) projectServerDeploymentProgress(job domain.Job, stamp time.Time) error { if job.ExecutionInput.Deployment == nil || job.ServerInstanceID == "" { return nil