Keep plugin records out of platform projections

platform/service/job_channel.go

@@ -49,19 +49,9 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
 		return domain.RunJobClaimResult{}, err
 	}
 	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
 	}
-	}
 
 	leaseToken, err := randomToken()
 	if err != nil {
@@ -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
 }
 

platform/service/plugin_data_rows.go

@@ -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
-}

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{{

platform/service/plugin_lifecycle_projection.go

@@ -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
-}

platform/service/plugin_query_projection.go

@@ -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
-}

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