browser/platform/service/scum_query_ingest.go

package service

import (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math"
	"sort"
	"strconv"
	"strings"
	"time"

	"browser.local/platform/domain"
	"browser.local/platform/repo"
)

// Plugin-declared SCUM database projection collections. The plugin owns the SQL
// text and the row-to-field mapping; the platform owns the typed tables those
// rows are projected into.
const (
	scumProjectionUsers    = "scum.users"
	scumProjectionVehicles = "scum.vehicles"
)

const scumQueryTemplateIdempotencyPrefix = "scum-query:"

// Steady-state bounds for the recurring database poll. Every dispatched poll
// creates one durable job, so a new poll retires older terminal jobs of the
// same template together with their job log streams instead of leaving an
// unbounded history behind.
const (
	scumQueryJobRetention = 2
	scumQueryJobScanLimit = 64
)

// ReconcileSCUMQueryTemplatesForRunEndpoint dispatches due SCUM database read
// templates for every live server instance registered to one run endpoint.
func (svc *CoreService) ReconcileSCUMQueryTemplatesForRunEndpoint(runEndpointID string) error {
	runEndpointID = strings.TrimSpace(runEndpointID)
	if runEndpointID == "" {
		return nil
	}
	instances, err := svc.store.ServerInstances().List(domain.ServerInstanceFilter{RunEndpointID: runEndpointID})
	if err != nil {
		return err
	}
	for _, instance := range instances {
		if err := svc.ReconcileSCUMQueryTemplates(instance.ID); err != nil {
			return err
		}
	}
	return nil
}

// ReconcileSCUMQueryTemplates queues the plugin-declared SCUM database read
// templates that are due for one server instance.
//
// Call-count budget for one call: one server instance read, one plugin read, one
// run endpoint read, one in-memory dispatch lookup per declared template, and at
// most one job insert per declared template per poll interval. Job inserts are
// de-duplicated by run endpoint plus idempotency key, so repeated page loads and
// run heartbeats inside the same poll interval add no storage work.
func (svc *CoreService) ReconcileSCUMQueryTemplates(serverInstanceID string) error {
	serverInstanceID = strings.TrimSpace(serverInstanceID)
	if serverInstanceID == "" {
		return nil
	}
	instance, err := svc.store.ServerInstances().Get(serverInstanceID)
	if errors.Is(err, repo.ErrNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	if instance.State == domain.ServerInstanceStateDeleted || strings.TrimSpace(instance.RunEndpointID) == "" {
		return nil
	}
	plugin, err := svc.store.GamePlugins().Get(instance.PluginID)
	if errors.Is(err, repo.ErrNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	templates := scumIngestQueryTemplates(plugin)
	if len(templates) == 0 {
		return nil
	}
	endpoint, err := svc.GetRunEndpoint(instance.RunEndpointID)
	if err != nil {
		return nil
	}
	transports := map[string]domain.RuntimeTransportProfile{}
	for _, transport := range plugin.RuntimeProfiles.TransportProfiles {
		transports[transport.Key] = transport
	}
	stamp := svc.now()
	for _, template := range templates {
		transport, ok := transports[template.TransportKey]
		if !ok || transport.Kind != "sqlite" || transport.TargetKey != template.TargetKey {
			continue
		}
		if !containsString(transport.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
			continue
		}
		if !containsString(plugin.RemoteAccess.RunCapabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
			continue
		}
		if !svc.endpointSupports(endpoint, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
			continue
		}
		interval := int64(template.PollIntervalSeconds)
		if interval <= 0 {
			continue
		}
		idempotencyKey := fmt.Sprintf("%s%s:%d", scumQueryTemplateIdempotencyPrefix, template.Key, stamp.Unix()/interval)
		job := domain.Job{
			ID:               jobIDFromParts("job-scum-query", instance.ID, idempotencyKey),
			ServerInstanceID: instance.ID,
			RunEndpointID:    instance.RunEndpointID,
			Capability:       domain.JobCapabilityRemoteRunDBSQLiteQuery,
			TargetKey:        template.TargetKey,
			InputRef:         fmt.Sprintf("input://scum-queries/%s/%s", instance.ID, template.Key),
			IdempotencyKey:   idempotencyKey,
			Progress:         domain.JobProgress{Percent: 0, Message: "SCUM database read queued"},
			RetryPolicy:      domain.JobRetryPolicy{MaxAttempts: 1, InitialBackoffSeconds: 2, MaxBackoffSeconds: 10},
			ExecutionInput: domain.JobExecutionInput{
				WorkspaceScope:    svc.runtimeProfileScope(instance.ID),
				RemoteAdapterKey:  transport.Key,
				RemoteAdapterKind: string(domain.RemoteAdapterDatabase),
				TimeoutSeconds:    scumQueryTemplateTimeout(template),
				PluginID:          plugin.ID,
				Inputs: map[string]string{
					"templateKey": template.Key,
					"sqlRef":      template.SQLRef,
					"maxRows":     strconv.Itoa(scumQueryTemplateMaxRows(template)),
				},
			},
		}
		created, err := svc.CreateJob(job)
		if err != nil {
			return err
		}
		svc.pruneSCUMQueryJobs(instance.ID, job.InputRef, created.ID)
	}
	return nil
}

// pruneSCUMQueryJobs keeps the newest terminal jobs for one dispatched template
// and removes the rest. It runs once per poll interval, so the caller pays one
// bounded job read per dispatch and no work at all while a poll is in flight.
func (svc *CoreService) pruneSCUMQueryJobs(serverInstanceID string, inputRef string, keepJobID string) {
	jobs, err := svc.store.Jobs().List(domain.JobFilter{ServerInstanceID: serverInstanceID, Limit: scumQueryJobScanLimit})
	if err != nil {
		log.Printf("SCUM query job retention skipped server=%s error=%s", serverInstanceID, err.Error())
		return
	}
	expired := make([]domain.Job, 0, len(jobs))
	for _, candidate := range jobs {
		if candidate.ID == keepJobID || candidate.InputRef != inputRef || !isTerminalJobState(candidate.State) {
			continue
		}
		expired = append(expired, candidate)
	}
	retained := scumQueryJobRetention - 1
	if retained < 0 {
		retained = 0
	}
	if len(expired) <= retained {
		return
	}
	sort.SliceStable(expired, func(i, j int) bool { return expired[i].CreatedAt.After(expired[j].CreatedAt) })
	for _, job := range expired[retained:] {
		if err := svc.deleteJobWithLogStreams(job); err != nil {
			log.Printf("SCUM query job retention failed server=%s job=%s error=%s", serverInstanceID, job.ID, err.Error())
			return
		}
	}
}

func (svc *CoreService) deleteJobWithLogStreams(job domain.Job) error {
	for _, streamKey := range []string{"stdout", "stderr"} {
		err := svc.store.LogStreams().Delete(jobLogStreamID(job.ID, streamKey))
		if err != nil && !errors.Is(err, repo.ErrNotFound) {
			return err
		}
	}
	err := svc.store.Jobs().Delete(job.ID)
	if err != nil && !errors.Is(err, repo.ErrNotFound) {
		return err
	}
	return nil
}

func scumIngestQueryTemplates(plugin domain.GamePlugin) []domain.GameClientBridgeQueryTemplateDeclaration {
	result := make([]domain.GameClientBridgeQueryTemplateDeclaration, 0, len(plugin.GameClientBridge.QueryTemplates))
	for _, template := range plugin.GameClientBridge.QueryTemplates {
		if template.PollIntervalSeconds <= 0 || len(template.Projections) == 0 {
			continue
		}
		if strings.TrimSpace(template.SQLRef) == "" {
			continue
		}
		supported := false
		for _, projection := range template.Projections {
			if scumProjectionCollectionSupported(projection.Collection) {
				supported = true
				break
			}
		}
		if !supported {
			continue
		}
		result = append(result, template)
	}
	return result
}

func scumProjectionCollectionSupported(collection string) bool {
	switch strings.TrimSpace(collection) {
	case scumProjectionUsers, scumProjectionVehicles:
		return true
	default:
		return false
	}
}

func scumQueryTemplateTimeout(template domain.GameClientBridgeQueryTemplateDeclaration) int {
	if template.TimeoutSeconds > 0 {
		return template.TimeoutSeconds
	}
	return 15
}

func scumQueryTemplateMaxRows(template domain.GameClientBridgeQueryTemplateDeclaration) int {
	if template.MaxRows > 0 {
		return template.MaxRows
	}
	return domain.SCUMDefaultListLimit
}

// projectSCUMQueryTemplateResult maps one completed database read into the typed
// SCUM tables through the projection declarations of the owning plugin.
func (svc *CoreService) projectSCUMQueryTemplateResult(job domain.Job, stamp time.Time) error {
	templateKey := strings.TrimSpace(job.ExecutionInput.Inputs["templateKey"])
	if templateKey == "" || job.ExecutionResult.Kind != "sqlite.query" {
		return nil
	}
	instance, err := svc.store.ServerInstances().Get(job.ServerInstanceID)
	if errors.Is(err, repo.ErrNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	plugin, err := svc.store.GamePlugins().Get(instance.PluginID)
	if errors.Is(err, repo.ErrNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	template, ok := scumQueryTemplateByKey(plugin.GameClientBridge.QueryTemplates, templateKey)
	if !ok {
		return nil
	}
	rows, err := decodeSCUMQueryRows(job.ExecutionResult.Content)
	if err != nil {
		log.Printf("SCUM query projection skipped job=%s server=%s template=%s reason=rows_unreadable", job.ID, instance.ID, templateKey)
		return nil
	}
	batch := domain.SCUMFactIngest{ServerInstanceID: instance.ID}
	for _, projection := range template.Projections {
		switch strings.TrimSpace(projection.Collection) {
		case scumProjectionUsers:
			batch.Users = append(batch.Users, scumUserFactsFromRows(rows, projection)...)
		case scumProjectionVehicles:
			batch.Vehicles = append(batch.Vehicles, scumVehicleFactsFromRows(rows, projection)...)
		}
	}
	if len(batch.Users) > domain.SCUMFactBatchLimit {
		batch.Users = batch.Users[:domain.SCUMFactBatchLimit]
	}
	if len(batch.Vehicles) > domain.SCUMFactBatchLimit {
		batch.Vehicles = batch.Vehicles[:domain.SCUMFactBatchLimit]
	}
	if len(batch.Users) == 0 && len(batch.Vehicles) == 0 {
		return nil
	}
	result, err := svc.ingestSCUMFactsForInstance(instance, batch, stamp)
	if err != nil {
		return err
	}
	log.Printf("SCUM query projection applied job=%s server=%s template=%s users=%d vehicles=%d accepted=%d/%d", job.ID, instance.ID, templateKey, len(batch.Users), len(batch.Vehicles), result.AcceptedUserCount, result.AcceptedVehicleCount)
	return nil
}

func scumQueryTemplateByKey(templates []domain.GameClientBridgeQueryTemplateDeclaration, key string) (domain.GameClientBridgeQueryTemplateDeclaration, bool) {
	for _, template := range templates {
		if template.Key == key {
			return template, true
		}
	}
	return domain.GameClientBridgeQueryTemplateDeclaration{}, false
}

type scumQueryRow map[string]any

func decodeSCUMQueryRows(content string) ([]scumQueryRow, error) {
	trimmed := strings.TrimSpace(content)
	if trimmed == "" {
		return nil, errors.New("query result is empty")
	}
	var payload struct {
		Rows []map[string]any `json:"rows"`
	}
	if err := json.Unmarshal([]byte(trimmed), &payload); err != nil {
		return nil, err
	}
	rows := make([]scumQueryRow, 0, len(payload.Rows))
	for _, row := range payload.Rows {
		if len(row) == 0 {
			continue
		}
		rows = append(rows, scumQueryRow(row))
	}
	return rows, nil
}

func scumRowField(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration, field string) (any, bool) {
	if source, ok := projection.FieldMappings[field]; ok {
		value, exists := row[source]
		if !exists {
			return nil, false
		}
		return value, true
	}
	if value, ok := projection.FixedValues[field]; ok {
		return value, true
	}
	value, exists := row[field]
	return value, exists
}

func scumRowText(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration, field string) string {
	value, ok := scumRowField(row, projection, field)
	if !ok || value == nil {
		return ""
	}
	return strings.TrimSpace(scumScalarText(value))
}

func scumScalarText(value any) string {
	switch typed := value.(type) {
	case string:
		return typed
	case bool:
		if typed {
			return "true"
		}
		return "false"
	case float64:
		if typed == math.Trunc(typed) && math.Abs(typed) < 1e15 {
			return strconv.FormatInt(int64(typed), 10)
		}
		return strconv.FormatFloat(typed, 'f', -1, 64)
	case int64:
		return strconv.FormatInt(typed, 10)
	case json.Number:
		return typed.String()
	default:
		return fmt.Sprintf("%v", typed)
	}
}

func scumRowBool(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration, field string) (bool, bool) {
	value, ok := scumRowField(row, projection, field)
	if !ok || value == nil {
		return false, false
	}
	switch typed := value.(type) {
	case bool:
		return typed, true
	case float64:
		return typed != 0, true
	case int64:
		return typed != 0, true
	default:
		text := strings.ToLower(strings.TrimSpace(scumScalarText(typed)))
		switch text {
		case "1", "true", "yes", "online", "active", "connected":
			return true, true
		case "0", "false", "no", "offline", "":
			return false, true
		default:
			return false, false
		}
	}
}

func scumRowFloat(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration, field string) (float64, bool) {
	value, ok := scumRowField(row, projection, field)
	if !ok || value == nil {
		return 0, false
	}
	switch typed := value.(type) {
	case float64:
		return typed, true
	case int64:
		return float64(typed), true
	case json.Number:
		parsed, err := typed.Float64()
		if err != nil {
			return 0, false
		}
		return parsed, true
	default:
		parsed, err := strconv.ParseFloat(strings.TrimSpace(scumScalarText(typed)), 64)
		if err != nil {
			return 0, false
		}
		return parsed, true
	}
}

func scumRowInt64(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration, field string) (int64, bool) {
	value, ok := scumRowField(row, projection, field)
	if !ok || value == nil {
		return 0, false
	}
	switch typed := value.(type) {
	case float64:
		return int64(math.Round(typed)), true
	case int64:
		return typed, true
	case json.Number:
		parsed, err := typed.Int64()
		if err != nil {
			return 0, false
		}
		return parsed, true
	default:
		parsed, err := strconv.ParseInt(strings.TrimSpace(scumScalarText(typed)), 10, 64)
		if err != nil {
			return 0, false
		}
		return parsed, true
	}
}

func scumRowTime(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration, field string) (time.Time, bool) {
	value, ok := scumRowField(row, projection, field)
	if !ok || value == nil {
		return time.Time{}, false
	}
	switch typed := value.(type) {
	case float64:
		seconds, fraction := math.Modf(typed)
		return time.Unix(int64(seconds), int64(fraction*float64(time.Second))).UTC(), true
	case int64:
		return time.Unix(typed, 0).UTC(), true
	default:
		text := strings.TrimSpace(scumScalarText(typed))
		if text == "" {
			return time.Time{}, false
		}
		for _, layout := range []string{time.RFC3339Nano, time.RFC3339, "2006-01-02T15:04:05.000Z", "2006-01-02 15:04:05"} {
			if parsed, err := time.Parse(layout, text); err == nil {
				return parsed.UTC(), true
			}
		}
		if seconds, err := strconv.ParseInt(text, 10, 64); err == nil {
			return time.Unix(seconds, 0).UTC(), true
		}
		return time.Time{}, false
	}
}

func scumUserFactsFromRows(rows []scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration) []domain.SCUMUserFact {
	facts := make([]domain.SCUMUserFact, 0, len(rows))
	for _, row := range rows {
		steamID := scumRowText(row, projection, "steamId")
		if steamID == "" {
			continue
		}
		fact := domain.SCUMUserFact{
			SteamID:             steamID,
			DisplayName:         scumRowText(row, projection, "displayName"),
			LoginIP:             scumRowText(row, projection, "loginIp"),
			RiddenGameVehicleID: scumRowText(row, projection, "riddenGameVehicleId"),
		}
		if online, ok := scumRowBool(row, projection, "online"); ok {
			fact.Online = online
		}
		if login, ok := scumRowBool(row, projection, "login"); ok {
			fact.Login = login
		}
		if observedAt, ok := scumRowTime(row, projection, "observedAt"); ok {
			fact.ObservedAt = observedAt
		}
		if loginAt, ok := scumRowTime(row, projection, "loginObservedAt"); ok {
			fact.LoginObservedAt = loginAt
		}
		if squadID, ok := scumRowInt64(row, projection, "squadId"); ok {
			value := squadID
			fact.SquadID = &value
		}
		if balance, ok := scumRowInt64(row, projection, "bankBalance"); ok {
			value := balance
			fact.BankBalance = &value
		}
		if gold, ok := scumRowInt64(row, projection, "goldBars"); ok {
			value := gold
			fact.GoldBars = &value
		}
		if position, ok := scumRowPosition(row, projection); ok {
			fact.Position = position
		}
		facts = append(facts, fact)
	}
	return facts
}

func scumVehicleFactsFromRows(rows []scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration) []domain.SCUMVehicleFact {
	facts := make([]domain.SCUMVehicleFact, 0, len(rows))
	for _, row := range rows {
		gameVehicleID := scumRowText(row, projection, "gameVehicleId")
		if gameVehicleID == "" {
			continue
		}
		fact := domain.SCUMVehicleFact{
			GameVehicleID:   gameVehicleID,
			VehicleClass:    scumRowText(row, projection, "vehicleClass"),
			DisplayName:     scumRowText(row, projection, "displayName"),
			LockedBySteamID: scumRowText(row, projection, "lockedBySteamId"),
		}
		if exists, ok := scumRowBool(row, projection, "exists"); ok {
			value := exists
			fact.Exists = &value
		}
		if locked, ok := scumRowBool(row, projection, "locked"); ok {
			value := locked
			fact.Locked = &value
		}
		if lockedAt, ok := scumRowTime(row, projection, "lockedAt"); ok {
			fact.LockedAt = lockedAt
		}
		if observedAt, ok := scumRowTime(row, projection, "observedAt"); ok {
			fact.ObservedAt = observedAt
		}
		if position, ok := scumRowPosition(row, projection); ok {
			fact.Position = position
		}
		facts = append(facts, fact)
	}
	return facts
}

func scumRowPosition(row scumQueryRow, projection domain.GameClientBridgeQueryProjectionDeclaration) (*domain.SCUMPosition, bool) {
	x, hasX := scumRowFloat(row, projection, "x")
	y, hasY := scumRowFloat(row, projection, "y")
	z, hasZ := scumRowFloat(row, projection, "z")
	if !hasX || !hasY || !hasZ {
		return nil, false
	}
	return &domain.SCUMPosition{X: x, Y: y, Z: z}, true
}