Fill the SCUM user and vehicle tables from the plugin-declared database read path
The SCUM 用户管理 list stayed empty because the typed scum_user and scum_vehicle tables had no producer: the run ingest endpoint is signed-run only, and the previously registered plugin templates asked for a Run database capability the endpoint never advertises. The plugin now declares bounded, read-only SQLite projections for players and vehicles (sql/scum-db-v57/*.sql with query schemas), and the platform dispatches those templates as durable remote.run.db.sqlite.query jobs on run heartbeat and page open, then projects the returned rows into the typed SCUM tables through the same shared ingest path used by signed run facts.
This commit is contained in:
npc0-hue
@@ -0,0 +1,510 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"math"
|
||||
"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:"
|
||||
|
||||
// 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)),
|
||||
},
|
||||
},
|
||||
}
|
||||
if _, err := svc.CreateJob(job); err != nil {
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user