workers

package
v0.28.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	UsageSyncCreatedServiceName     = "superplane." + messages.CanvasExchange + "." + messages.OrganizationCreatedRoutingKey + ".worker-consumer"
	UsageSyncPlanChangedServiceName = "superplane." + messages.CanvasExchange + "." + messages.OrganizationPlanChangedRoutingKey + ".worker-consumer"
	UsageSyncConnectionName         = "superplane"
)
View Source 
const MagicCodeEmailConnectionName = "superplane"
View Source 
const MagicCodeEmailServiceName = "superplane" + "." + messages.CanvasExchange + "." + messages.MagicCodeRequestedRoutingKey + ".worker-consumer"

Variables

View Source 
var ErrRecordLocked = errors.New("record locked")

Functions

This section is empty.

Types

type AgentStreamWorker added in v0.21.0 

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

AgentStreamWorker is stateless and safe to run as competing consumers.

func NewAgentStreamWorker added in v0.21.0 

func NewAgentStreamWorker(provider agents.Provider, rabbitMQURL string, customToolExecutor ...agents.CustomToolExecutor) *AgentStreamWorker

func NewAgentStreamWorkerWithUsageService added in v0.27.0 

func NewAgentStreamWorkerWithUsageService(
	provider agents.Provider,
	rabbitMQURL string,
	usageService usage.Service,
	customToolExecutor ...agents.CustomToolExecutor,
) *AgentStreamWorker

func (*AgentStreamWorker) Handle added in v0.21.0 

func (w *AgentStreamWorker) Handle(parentCtx context.Context, body []byte) error

Handle is exported only so tests can drive the worker without RabbitMQ.

func (*AgentStreamWorker) Start added in v0.21.0 

func (w *AgentStreamWorker) Start(ctx context.Context)

type CanvasCleanupWorker added in v0.6.0 

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

func NewCanvasCleanupWorker added in v0.6.0 

func NewCanvasCleanupWorker(gitProvider git.Provider, providers ...agents.Provider) *CanvasCleanupWorker

func (*CanvasCleanupWorker) LockAndProcessCanvas added in v0.6.0 

func (w *CanvasCleanupWorker) LockAndProcessCanvas(canvas models.Canvas) error

func (*CanvasCleanupWorker) Start added in v0.6.0 

func (w *CanvasCleanupWorker) Start(ctx context.Context)

type EventDistributer 

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

EventDistributer coordinates message consumption from RabbitMQ and distributes events to websocket clients

func NewEventDistributer 

func NewEventDistributer(wsHub *ws.Hub) *EventDistributer

NewEventDistributer creates a new event distributer coordinator

func (*EventDistributer) Shutdown 

func (e *EventDistributer) Shutdown(ctx context.Context) error

Shutdown gracefully stops the worker

func (*EventDistributer) Start 

func (e *EventDistributer) Start() error

Start begins consuming messages from RabbitMQ for all relevant routing keys

type EventRetentionWorker added in v0.13.0 

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

func NewEventRetentionWorker added in v0.13.0 

func NewEventRetentionWorker(usageService usage.Service) *EventRetentionWorker

func (*EventRetentionWorker) LockAndProcessRootEvents added in v0.20.0 

func (w *EventRetentionWorker) LockAndProcessRootEvents(referenceTime time.Time, limit int) (int, error)

func (*EventRetentionWorker) Start added in v0.13.0 

func (w *EventRetentionWorker) Start(ctx context.Context)

type EventRouter added in v0.6.0 

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

func NewEventRouter added in v0.6.0 

func NewEventRouter(rabbitMQURL string) *EventRouter

func (*EventRouter) Consume added in v0.12.0 

func (w *EventRouter) Consume(delivery tackle.Delivery) error

func (*EventRouter) LockAndProcessEvent added in v0.6.0 

func (w *EventRouter) LockAndProcessEvent(logger *log.Entry, event models.CanvasEvent, attemptStart time.Time) error

func (*EventRouter) Name added in v0.12.0 

func (w *EventRouter) Name() string

func (*EventRouter) Start added in v0.6.0 

func (w *EventRouter) Start(ctx context.Context)

func (*EventRouter) StartRabbitMQConsumer added in v0.12.0 

func (w *EventRouter) StartRabbitMQConsumer(ctx context.Context)

type IntegrationCleanupWorker added in v0.6.0 

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

func NewIntegrationCleanupWorker added in v0.6.0 

func NewIntegrationCleanupWorker(registry *registry.Registry, encryptor crypto.Encryptor, baseURL string) *IntegrationCleanupWorker

func (*IntegrationCleanupWorker) LockAndProcessIntegration added in v0.6.0 

func (w *IntegrationCleanupWorker) LockAndProcessIntegration(integration models.Integration) error

func (*IntegrationCleanupWorker) Start added in v0.6.0 

func (w *IntegrationCleanupWorker) Start(ctx context.Context)

type IntegrationRequestWorker added in v0.6.0 

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

func NewIntegrationRequestWorker added in v0.6.0 

func NewIntegrationRequestWorker(encryptor crypto.Encryptor, registry *registry.Registry, oidcProvider oidc.Provider, baseURL string, webhooksBaseURL string) *IntegrationRequestWorker

func (*IntegrationRequestWorker) LockAndProcessRequest added in v0.6.0 

func (w *IntegrationRequestWorker) LockAndProcessRequest(request models.IntegrationRequest) error

LockAndProcessRequest processes a request in 3 phases so the external HTTP call (Sync/HandleHook) never holds a DB transaction open:

  • Phase 1 (short tx): Lease the request by pushing run_at past the work window
  • Phase 2 (no tx): Run integration.Sync / HandleHook with a non-transactional context
  • Phase 3 (short tx): Persist instance state and mark the request completed

func (*IntegrationRequestWorker) Start added in v0.6.0 

func (w *IntegrationRequestWorker) Start(ctx context.Context)

type MagicCodeEmailConsumer added in v0.14.0 

type MagicCodeEmailConsumer struct {
	Consumer     *tackle.Consumer
	RabbitMQURL  string
	EmailService services.EmailService
	BaseURL      string
}

func NewMagicCodeEmailConsumer added in v0.14.0 

func NewMagicCodeEmailConsumer(rabbitMQURL string, emailService services.EmailService, baseURL string) *MagicCodeEmailConsumer

func (*MagicCodeEmailConsumer) Consume added in v0.14.0 

func (c *MagicCodeEmailConsumer) Consume(delivery tackle.Delivery) error

func (*MagicCodeEmailConsumer) Start added in v0.14.0 

func (c *MagicCodeEmailConsumer) Start() error

func (*MagicCodeEmailConsumer) Stop added in v0.14.0 

func (c *MagicCodeEmailConsumer) Stop()

type NodeExecutor added in v0.6.0 

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

func NewNodeExecutor added in v0.6.0 

func NewNodeExecutor(encryptor crypto.Encryptor, registry *registry.Registry, gitProvider gitprovider.Provider, oidcProvider oidc.Provider, baseURL string, webhookBaseURL string, rabbitMQURL string, authService authorization.Authorization) *NodeExecutor

func (*NodeExecutor) Consume added in v0.12.0 

func (w *NodeExecutor) Consume(delivery tackle.Delivery) error

func (*NodeExecutor) LockAndProcessNodeExecution added in v0.6.0 

func (w *NodeExecutor) LockAndProcessNodeExecution(id uuid.UUID) error

func (*NodeExecutor) Name added in v0.12.0 

func (w *NodeExecutor) Name() string

func (*NodeExecutor) Start added in v0.6.0 

func (w *NodeExecutor) Start(ctx context.Context)

func (*NodeExecutor) StartRabbitMQConsumer added in v0.12.0 

func (w *NodeExecutor) StartRabbitMQConsumer(ctx context.Context)

type NodeQueueWorker added in v0.6.0 

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

func NewNodeQueueWorker added in v0.6.0 

func NewNodeQueueWorker(registry *registry.Registry, gitProvider gitprovider.Provider, rabbitMQURL string) *NodeQueueWorker

func (*NodeQueueWorker) ConsumeExecutionFinished added in v0.26.0 

func (w *NodeQueueWorker) ConsumeExecutionFinished(delivery tackle.Delivery) error

func (*NodeQueueWorker) ConsumeQueueItemCreated added in v0.26.0 

func (w *NodeQueueWorker) ConsumeQueueItemCreated(delivery tackle.Delivery) error

func (*NodeQueueWorker) LockAndProcessNode added in v0.6.0 

func (w *NodeQueueWorker) LockAndProcessNode(logger *log.Entry, node models.CanvasNode, attemptStart time.Time) error

func (*NodeQueueWorker) Name added in v0.12.0 

func (w *NodeQueueWorker) Name() string

func (*NodeQueueWorker) Start added in v0.6.0 

func (w *NodeQueueWorker) Start(ctx context.Context)

type NodeRequestWorker 

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

func NewNodeRequestWorker 

func NewNodeRequestWorker(encryptor crypto.Encryptor, registry *registry.Registry, gitProvider gitprovider.Provider, webhookBaseURL string, authService authorization.Authorization) *NodeRequestWorker

func (*NodeRequestWorker) LockAndProcessRequest 

func (w *NodeRequestWorker) LockAndProcessRequest(request models.CanvasNodeRequest) error

func (*NodeRequestWorker) Start 

func (w *NodeRequestWorker) Start(ctx context.Context)

type OrganizationCleanupWorker added in v0.13.0 

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

func NewOrganizationCleanupWorker added in v0.13.0 

func NewOrganizationCleanupWorker(gitProvider git.Provider, providers ...agents.Provider) *OrganizationCleanupWorker

func (*OrganizationCleanupWorker) LockAndProcessOrganization added in v0.13.0 

func (w *OrganizationCleanupWorker) LockAndProcessOrganization(organization models.Organization) error

func (*OrganizationCleanupWorker) Start added in v0.13.0 

func (w *OrganizationCleanupWorker) Start(ctx context.Context)

type RepositoryProvisionerWorker added in v0.24.0 

type RepositoryProvisionerWorker struct {
	Consumer    *tackle.Consumer
	RabbitMQURL string
	Storage     git.Provider
	// contains filtered or unexported fields
}

func NewRepositoryProvisionerWorker added in v0.24.0 

func NewRepositoryProvisionerWorker(rabbitMQURL string, storage git.Provider) *RepositoryProvisionerWorker

func (*RepositoryProvisionerWorker) ConsumeCanvasCreated added in v0.24.0 

func (w *RepositoryProvisionerWorker) ConsumeCanvasCreated(delivery tackle.Delivery) error

func (*RepositoryProvisionerWorker) Start added in v0.24.0 

func (w *RepositoryProvisionerWorker) Start(ctx context.Context)

func (*RepositoryProvisionerWorker) Stop added in v0.24.0 

func (w *RepositoryProvisionerWorker) Stop()

type RunFinalizer added in v0.26.0 

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

func NewRunFinalizer added in v0.26.0 

func NewRunFinalizer(rabbitMQURL string) *RunFinalizer

func (*RunFinalizer) Name added in v0.26.0 

func (w *RunFinalizer) Name() string

func (*RunFinalizer) Start added in v0.26.0 

func (w *RunFinalizer) Start(ctx context.Context)

type UsageSyncWorker added in v0.13.0 

type UsageSyncWorker struct {
	CreatedConsumer     *tackle.Consumer
	PlanChangedConsumer *tackle.Consumer
	RabbitMQURL         string
	UsageService        usage.Service
}

func NewUsageSyncWorker added in v0.13.0 

func NewUsageSyncWorker(rabbitMQURL string, usageService usage.Service) *UsageSyncWorker

func (*UsageSyncWorker) Consume added in v0.13.0 

func (w *UsageSyncWorker) Consume(delivery tackle.Delivery) error

func (*UsageSyncWorker) ConsumeOrganizationPlanChanged added in v0.13.0 

func (w *UsageSyncWorker) ConsumeOrganizationPlanChanged(delivery tackle.Delivery) error

func (*UsageSyncWorker) Start added in v0.13.0 

func (w *UsageSyncWorker) Start(ctx context.Context)

func (*UsageSyncWorker) Stop added in v0.13.0 

func (w *UsageSyncWorker) Stop()

type WebhookCleanupWorker 

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

func NewWebhookCleanupWorker 

func NewWebhookCleanupWorker(encryptor crypto.Encryptor, registry *registry.Registry, baseURL string) *WebhookCleanupWorker

func (*WebhookCleanupWorker) LockAndProcessWebhook 

func (w *WebhookCleanupWorker) LockAndProcessWebhook(logger *log.Entry, webhook models.Webhook) error

func (*WebhookCleanupWorker) Start 

func (w *WebhookCleanupWorker) Start(ctx context.Context)

type WebhookProvisioner 

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

func NewWebhookProvisioner 

func NewWebhookProvisioner(baseURL string, encryptor crypto.Encryptor, registry *registry.Registry) *WebhookProvisioner

func (*WebhookProvisioner) LockAndProcessWebhook 

func (w *WebhookProvisioner) LockAndProcessWebhook(logger *log.Entry, webhook models.Webhook) error

LockAndProcessWebhook processes a webhook in 3 phases to avoid holding a DB connection during potentially long-running external API calls:

  • Phase 1 (short tx): Lock the webhook and set state to "provisioning"
  • Phase 2 (no tx): Run the external handler.Setup() call
  • Phase 3 (short tx): Set state to "ready" or handle errors

func (*WebhookProvisioner) Start 

func (w *WebhookProvisioner) Start(ctx context.Context)

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.