worker

package
v0.0.0-...-d25dbbd Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 20, 2026 License: Apache-2.0 Imports: 40 Imported by: 0

Documentation

Overview

Package worker manages the embedded Asynq task worker.

The worker runs as a goroutine inside the PocketBase process, connecting to Redis for persistent async task processing.

Index

Constants

View Source
const (
	// TaskSoftwareInstall is the Asynq task type for a software install action.
	TaskSoftwareInstall = "software:install"
	// TaskSoftwareUpgrade is the Asynq task type for a software upgrade action.
	TaskSoftwareUpgrade = "software:upgrade"
	// TaskSoftwareStart is the Asynq task type for a software start action.
	TaskSoftwareStart = "software:start"
	// TaskSoftwareStop is the Asynq task type for a software stop action.
	TaskSoftwareStop = "software:stop"
	// TaskSoftwareRestart is the Asynq task type for a software restart action.
	TaskSoftwareRestart = "software:restart"
	// TaskSoftwareVerify is the Asynq task type for a software verify action.
	TaskSoftwareVerify = "software:verify"
	// TaskSoftwareReinstall is the Asynq task type for a software reinstall action.
	TaskSoftwareReinstall = "software:reinstall"
	// TaskSoftwareUninstall is the Asynq task type for a software uninstall action.
	TaskSoftwareUninstall = "software:uninstall"
)
View Source
const (
	// Task type constants
	TaskRestartApp    = "restart:app"
	TaskStopApp       = "stop:app"
	TaskDeleteApp     = "delete:app"
	TaskBackupCreate  = "backup:create"
	TaskBackupRestore = "backup:restore"
)
View Source
const TaskDockerImagePull = "docker:image-pull"
View Source
const TaskMonitorAIProviderReachabilitySweep = "monitor:ai_provider_reachability_sweep"
View Source
const TaskMonitorAppHealthSweep = "monitor:app_health_sweep"
View Source
const TaskMonitorConnectorReachabilitySweep = "monitor:connector_reachability_sweep"
View Source
const TaskMonitorControlReachability = "monitor:control_reachability"
View Source
const TaskMonitorCredentialSweep = "monitor:credential_sweep"
View Source
const TaskMonitorFactsPull = "monitor:facts_pull"
View Source
const TaskMonitorMetricsFreshness = "monitor:metrics_freshness"
View Source
const TaskMonitorReachabilitySweep = "monitor:reachability_sweep"
View Source
const TaskMonitorRuntimeSnapshotPull = "monitor:runtime_snapshot_pull"
View Source
const (
	TaskRunOperation = "lifecycle:run_operation"
)
View Source
const TaskSoftwareWarmSnapshot = "software:warm-snapshot"
View Source
const TaskWorkflowRun = "workflow:run"

Variables

View Source
var ErrSoftwareActionUnsupported = errors.New("software action unsupported for component")
View Source
var ErrSoftwareComponentNotFound = errors.New("software component not found in server catalog")
View Source
var ErrSoftwareOperationInFlight = errors.New("software operation already in flight")

Functions

func EnqueueDockerImagePull

func EnqueueDockerImagePull(client *asynq.Client, operationID, serverID, imageName, userID, userEmail string) error

func EnqueueMonitorAIProviderReachabilitySweep

func EnqueueMonitorAIProviderReachabilitySweep(client *asynq.Client) error

func EnqueueMonitorAppHealthSweep

func EnqueueMonitorAppHealthSweep(client *asynq.Client) error

func EnqueueMonitorConnectorReachabilitySweep

func EnqueueMonitorConnectorReachabilitySweep(client *asynq.Client) error

func EnqueueMonitorControlReachability

func EnqueueMonitorControlReachability(client *asynq.Client) error

func EnqueueMonitorCredentialSweep

func EnqueueMonitorCredentialSweep(client *asynq.Client) error

func EnqueueMonitorFactsPull

func EnqueueMonitorFactsPull(client *asynq.Client) error

func EnqueueMonitorMetricsFreshness

func EnqueueMonitorMetricsFreshness(client *asynq.Client) error

func EnqueueMonitorReachabilitySweep

func EnqueueMonitorReachabilitySweep(client *asynq.Client) error

func EnqueueMonitorRuntimeSnapshotPull

func EnqueueMonitorRuntimeSnapshotPull(client *asynq.Client) error

func EnqueueOperation

func EnqueueOperation(client *asynq.Client, operationID string) error

func EnqueueSoftwareAction

func EnqueueSoftwareAction(client *asynq.Client, operationID, serverID string, componentKey software.ComponentKey, action software.Action, userID, userEmail, apposBaseURL string) error

EnqueueSoftwareAction creates and enqueues an Asynq task for a software delivery action. Returns an error if the client is nil or if task creation fails.

func EnqueueSoftwareSnapshotWarm

func EnqueueSoftwareSnapshotWarm(client *asynq.Client, serverID, userID string, componentKeys []software.ComponentKey) error

func EnqueueWorkflowRun

func EnqueueWorkflowRun(client *asynq.Client, runID string) error

func FindInFlightDockerImagePullOperation

func FindInFlightDockerImagePullOperation(app core.App, serverID, normalizedName string) (*core.Record, error)

func NewDockerImagePullTask

func NewDockerImagePullTask(operationID, serverID, imageName, userID, userEmail string) (*asynq.Task, error)

func NewMonitorAIProviderReachabilitySweepTask

func NewMonitorAIProviderReachabilitySweepTask() (*asynq.Task, error)

func NewMonitorAppHealthSweepTask

func NewMonitorAppHealthSweepTask() (*asynq.Task, error)

func NewMonitorConnectorReachabilitySweepTask

func NewMonitorConnectorReachabilitySweepTask() (*asynq.Task, error)

func NewMonitorControlReachabilityTask

func NewMonitorControlReachabilityTask() (*asynq.Task, error)

func NewMonitorCredentialSweepTask

func NewMonitorCredentialSweepTask() (*asynq.Task, error)

func NewMonitorFactsPullTask

func NewMonitorFactsPullTask() (*asynq.Task, error)

func NewMonitorMetricsFreshnessTask

func NewMonitorMetricsFreshnessTask() (*asynq.Task, error)

func NewMonitorReachabilitySweepTask

func NewMonitorReachabilitySweepTask() (*asynq.Task, error)

func NewMonitorRuntimeSnapshotPullTask

func NewMonitorRuntimeSnapshotPullTask() (*asynq.Task, error)

func NewRunOperationTask

func NewRunOperationTask(operationID string) (*asynq.Task, error)

func NewSoftwareActionTask

func NewSoftwareActionTask(operationID, serverID string, componentKey software.ComponentKey, action software.Action, userID, userEmail, apposBaseURL string) (*asynq.Task, error)

NewSoftwareActionTask creates an Asynq task for a software delivery action. Returns an error if server_id, component_key, or action is empty.

func NewSoftwareSnapshotWarmTask

func NewSoftwareSnapshotWarmTask(serverID, userID string, componentKeys []software.ComponentKey) (*asynq.Task, error)

func NewWorkflowRunTask

func NewWorkflowRunTask(runID string) (*asynq.Task, error)

func NormalizeDockerImageReference

func NormalizeDockerImageReference(name string) string

func PrepareDockerImagePullOperation

func PrepareDockerImagePullOperation(app core.App, serverID, imageName string) (*core.Record, error)

func PrepareSoftwareOperation

func PrepareSoftwareOperation(app core.App, serverID string, componentKey software.ComponentKey, action software.Action) (*core.Record, error)

PrepareSoftwareOperation creates the accepted software operation synchronously so callers can return a stable operation_id to API clients before queue dispatch.

Types

type BackupCreatePayload

type BackupCreatePayload struct {
	UserID    string `json:"user_id"`
	UserEmail string `json:"user_email"`
	Name      string `json:"name"`
}

BackupCreatePayload is the task payload for TaskBackupCreate.

type BackupRestorePayload

type BackupRestorePayload struct {
	UserID    string `json:"user_id"`
	UserEmail string `json:"user_email"`
	Name      string `json:"name"`
}

BackupRestorePayload is the task payload for TaskBackupRestore.

type DeleteAppPayload

type DeleteAppPayload struct {
	UserID        string `json:"user_id"`
	UserEmail     string `json:"user_email"`
	ProjectDir    string `json:"project_dir"`
	RemoveVolumes bool   `json:"remove_volumes"`
}

DeleteAppPayload is the task payload for TaskDeleteApp.

type DockerImagePullPayload

type DockerImagePullPayload struct {
	OperationID string `json:"operation_id"`
	ServerID    string `json:"server_id"`
	ImageName   string `json:"image_name"`
	UserID      string `json:"user_id"`
	UserEmail   string `json:"user_email"`
}

type MonitorAIProviderReachabilitySweepPayload

type MonitorAIProviderReachabilitySweepPayload struct{}

type MonitorAppHealthSweepPayload

type MonitorAppHealthSweepPayload struct{}

type MonitorConnectorReachabilitySweepPayload

type MonitorConnectorReachabilitySweepPayload struct{}

type MonitorControlReachabilityPayload

type MonitorControlReachabilityPayload struct{}

type MonitorCredentialSweepPayload

type MonitorCredentialSweepPayload struct{}

type MonitorFactsPullPayload

type MonitorFactsPullPayload struct{}

type MonitorMetricsFreshnessPayload

type MonitorMetricsFreshnessPayload struct{}

type MonitorReachabilitySweepPayload

type MonitorReachabilitySweepPayload struct{}

type MonitorRuntimeSnapshotPullPayload

type MonitorRuntimeSnapshotPullPayload struct{}

type RestartAppPayload

type RestartAppPayload struct {
	UserID     string `json:"user_id"`
	UserEmail  string `json:"user_email"`
	ProjectDir string `json:"project_dir"`
}

RestartAppPayload is the task payload for TaskRestartApp.

type RunOperationPayload

type RunOperationPayload struct {
	OperationID string `json:"operation_id"`
}

type Snapshot

type Snapshot struct {
	StartedAt         time.Time
	ServerRunning     bool
	SchedulerRunning  bool
	SchedulerLastTick time.Time
	LastDispatchAt    time.Time
	LastServerError   string
	LastDispatchError string
}

type SoftwareActionPayload

type SoftwareActionPayload struct {
	OperationID  string                `json:"operation_id"`
	ServerID     string                `json:"server_id"`
	ComponentKey software.ComponentKey `json:"component_key"`
	Action       software.Action       `json:"action"`
	UserID       string                `json:"user_id"`
	UserEmail    string                `json:"user_email"`
	AppOSBaseURL string                `json:"appos_base_url,omitempty"`
}

SoftwareActionPayload is the Asynq task payload for all software delivery actions.

type SoftwareSnapshotWarmPayload

type SoftwareSnapshotWarmPayload struct {
	ServerID      string                  `json:"server_id"`
	UserID        string                  `json:"user_id"`
	ComponentKeys []software.ComponentKey `json:"component_keys"`
}

type StopAppPayload

type StopAppPayload struct {
	UserID     string `json:"user_id"`
	UserEmail  string `json:"user_email"`
	ProjectDir string `json:"project_dir"`
}

StopAppPayload is the task payload for TaskStopApp.

type Worker

type Worker struct {
	// contains filtered or unexported fields
}

Worker manages the Asynq server and a shared client for enqueuing tasks.

func New

func New(app core.App) (*Worker, error)

New creates a Worker with Asynq server and shared client. app is the PocketBase core.App used for audit writes inside task handlers. Call Start() to begin processing and Shutdown() to stop.

func (*Worker) Client

func (w *Worker) Client() *asynq.Client

Client returns the shared Asynq client for enqueuing tasks.

func (*Worker) Shutdown

func (w *Worker) Shutdown()

Shutdown gracefully stops the worker and closes the client connection.

func (*Worker) Snapshot

func (w *Worker) Snapshot() Snapshot

func (*Worker) Start

func (w *Worker) Start()

Start begins processing tasks in a background goroutine. This should be called only once during the application lifecycle.

type WorkflowRunPayload

type WorkflowRunPayload struct {
	RunID string `json:"run_id"`
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL