93 lines
3.1 KiB
Go
93 lines
3.1 KiB
Go
package service
|
|
|
|
import (
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
)
|
|
|
|
func (svc *CoreService) scheduleDuePluginQueries(runEndpointID string, capabilities []string, stamp time.Time) error {
|
|
if !containsString(capabilities, domain.JobCapabilityRemoteRunDBSQLiteQuery) {
|
|
return nil
|
|
}
|
|
instances, err := svc.store.ServerInstances().List(domain.ServerInstanceFilter{RunEndpointID: runEndpointID})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: runEndpointID})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, instance := range instances {
|
|
if instance.State == domain.ServerInstanceStateDeleted {
|
|
continue
|
|
}
|
|
plugin, getErr := svc.store.GamePlugins().Get(instance.PluginID)
|
|
if getErr != nil {
|
|
return getErr
|
|
}
|
|
for _, template := range plugin.GameClientBridge.QueryTemplates {
|
|
if template.PollIntervalSeconds <= 0 || template.RowTarget == nil || strings.TrimSpace(template.SQLRef) == "" {
|
|
continue
|
|
}
|
|
if !pluginQueryTemplateDue(jobs, instance.ID, template.Key, time.Duration(template.PollIntervalSeconds)*time.Second, stamp) {
|
|
continue
|
|
}
|
|
bucket := stamp.Unix() / int64(template.PollIntervalSeconds)
|
|
idempotencyKey := fmt.Sprintf("plugin-query:%s:%s:%d", instance.ID, template.Key, bucket)
|
|
job := domain.Job{
|
|
ID: jobIDFromParts("job-plugin-query", instance.ID, idempotencyKey),
|
|
ServerInstanceID: instance.ID,
|
|
RunEndpointID: runEndpointID,
|
|
Capability: domain.JobCapabilityRemoteRunDBSQLiteQuery,
|
|
TargetKey: template.TargetKey,
|
|
InputRef: "input://plugin-query/" + template.Key,
|
|
IdempotencyKey: idempotencyKey,
|
|
Progress: domain.JobProgress{Percent: 0, Message: "declared automatic plugin query queued"},
|
|
RetryPolicy: domain.JobRetryPolicy{MaxAttempts: 1, InitialBackoffSeconds: 1, MaxBackoffSeconds: 1},
|
|
ExecutionInput: domain.JobExecutionInput{
|
|
WorkspaceScope: svc.runtimeProfileScope(instance.ID),
|
|
RemoteAdapterKey: template.TransportKey,
|
|
RemoteAdapterKind: string(domain.RemoteAdapterDatabase),
|
|
TimeoutSeconds: template.TimeoutSeconds,
|
|
Inputs: map[string]string{
|
|
"templateKey": template.Key,
|
|
"sqlRef": template.SQLRef,
|
|
"maxRows": strconv.Itoa(template.MaxRows),
|
|
"limit": strconv.Itoa(template.MaxRows),
|
|
},
|
|
},
|
|
}
|
|
created, createErr := svc.CreateJob(job)
|
|
if createErr != nil {
|
|
return createErr
|
|
}
|
|
jobs = append(jobs, created)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func pluginQueryTemplateDue(jobs []domain.Job, serverInstanceID, templateKey string, interval time.Duration, stamp time.Time) bool {
|
|
var latest time.Time
|
|
for _, job := range jobs {
|
|
if job.ServerInstanceID != serverInstanceID || job.Capability != domain.JobCapabilityRemoteRunDBSQLiteQuery || job.ExecutionInput.Inputs["templateKey"] != templateKey {
|
|
continue
|
|
}
|
|
if !isTerminalJobState(job.State) {
|
|
return false
|
|
}
|
|
attemptedAt := job.TerminalAt
|
|
if attemptedAt.IsZero() {
|
|
attemptedAt = job.UpdatedAt
|
|
}
|
|
if attemptedAt.After(latest) {
|
|
latest = attemptedAt
|
|
}
|
|
}
|
|
return latest.IsZero() || !stamp.Before(latest.Add(interval))
|
|
}
|