browser/platform/service/plugin_query_projection.go

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
}