workers

package
v1.0.0-rc1 Latest Latest
Warning

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

 Go to latest Published: Oct 16, 2026 License: Apache-2.0 Imports: 62 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	FactoryJiraCloseServiceName    = "superplane" + "." + messages.CanvasExchange + "." + messages.FactoryWorkOrderNotificationRoutingKey + ".jira-close-consumer"
	FactoryJiraCloseConnectionName = "superplane"
)
View Source 
const (
	FactorySentryResolveServiceName    = "superplane" + "." + messages.CanvasExchange + "." + messages.FactoryWorkOrderNotificationRoutingKey + ".sentry-resolve-consumer"
	FactorySentryResolveConnectionName = "superplane"
)
View Source 
const FactoryNotificationConnectionName = "superplane"
View Source 
const FactoryNotificationServiceName = "superplane" + "." + messages.CanvasExchange + "." + messages.FactoryWorkOrderNotificationRoutingKey + ".worker-consumer"
View Source 
const MagicCodeEmailConnectionName = "superplane"
View Source 
const MagicCodeEmailServiceName = "superplane" + "." + messages.CanvasExchange + "." + messages.MagicCodeRequestedRoutingKey + ".worker-consumer"
View Source 
const SupportFeedbackConnectionName = "superplane"
View Source 
const SupportFeedbackServiceName = "superplane" + "." + messages.CanvasExchange + "." + messages.SupportFeedbackRequestedRoutingKey + ".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 (*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 AppMessageWorker added in v0.29.0 

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

func NewAppMessageWorker added in v0.29.0 

func NewAppMessageWorker(registry *registry.Registry) *AppMessageWorker

func (*AppMessageWorker) LockAndProcessMessage added in v0.29.0 

func (w *AppMessageWorker) LockAndProcessMessage(message models.AppMessage) error

func (*AppMessageWorker) Start added in v0.29.0 

func (w *AppMessageWorker) 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(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() *EventRetentionWorker

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 ExecutionTerminator added in v0.30.0 

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

func NewExecutionTerminator added in v0.30.0 

func NewExecutionTerminator(
	rabbitMQURL string,
	authService authorization.Authorization,
	encryptor crypto.Encryptor,
	registry *registry.Registry,
) *ExecutionTerminator

func (*ExecutionTerminator) LockAndCancelExecution added in v0.30.0 

func (w *ExecutionTerminator) LockAndCancelExecution(execution models.CanvasNodeExecution) error

func (*ExecutionTerminator) Name added in v0.30.0 

func (w *ExecutionTerminator) Name() string

func (*ExecutionTerminator) Start added in v0.30.0 

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

type FactoryCleanupWorker 

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

func NewFactoryCleanupWorker 

func NewFactoryCleanupWorker() *FactoryCleanupWorker

func (*FactoryCleanupWorker) LockAndProcessFactory 

func (w *FactoryCleanupWorker) LockAndProcessFactory(factory models.Factory) error

func (*FactoryCleanupWorker) Start 

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

type FactoryJiraCloseConsumer 

type FactoryJiraCloseConsumer struct {
	Consumer    *tackle.Consumer
	RabbitMQURL string
	Encryptor   crypto.Encryptor
	Registry    *registry.Registry
	BaseURL     string
	// contains filtered or unexported fields
}

FactoryJiraCloseConsumer moves the originating Jira issue after a Jira-intake task closes as completed. It shares the work-order notification routing key with FactoryNotificationConsumer and uses its own queue so Jira failures never block email delivery or SuperPlane Complete.

func NewFactoryJiraCloseConsumer 

func NewFactoryJiraCloseConsumer(
	rabbitMQURL string,
	encryptor crypto.Encryptor,
	componentRegistry *registry.Registry,
	baseURL string,
) *FactoryJiraCloseConsumer

func (*FactoryJiraCloseConsumer) Consume 

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

func (*FactoryJiraCloseConsumer) Start 

func (c *FactoryJiraCloseConsumer) Start() error

func (*FactoryJiraCloseConsumer) Stop 

func (c *FactoryJiraCloseConsumer) Stop()

type FactoryNotificationConsumer 

type FactoryNotificationConsumer struct {
	Consumer     *tackle.Consumer
	RabbitMQURL  string
	EmailService services.EmailService
	BaseURL      string
	// contains filtered or unexported fields
}

FactoryNotificationConsumer turns work order activity messages into notification emails and live browser alerts, honoring each recipient's notification settings.

func NewFactoryNotificationConsumer 

func NewFactoryNotificationConsumer(rabbitMQURL string, emailService services.EmailService, baseURL string) *FactoryNotificationConsumer

func (*FactoryNotificationConsumer) Consume 

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

func (*FactoryNotificationConsumer) Start 

func (c *FactoryNotificationConsumer) Start() error

func (*FactoryNotificationConsumer) Stop 

func (c *FactoryNotificationConsumer) Stop()

type FactorySentryResolveConsumer 

type FactorySentryResolveConsumer struct {
	Consumer    *tackle.Consumer
	RabbitMQURL string
	Encryptor   crypto.Encryptor
	Registry    *registry.Registry
	// contains filtered or unexported fields
}

FactorySentryResolveConsumer resolves the originating Sentry issue after a Sentry-intake task closes as completed. It shares the work-order notification routing key with FactoryNotificationConsumer and uses its own queue so Sentry failures never block email delivery.

func NewFactorySentryResolveConsumer 

func NewFactorySentryResolveConsumer(
	rabbitMQURL string,
	encryptor crypto.Encryptor,
	componentRegistry *registry.Registry,
) *FactorySentryResolveConsumer

func (*FactorySentryResolveConsumer) Consume 

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

func (*FactorySentryResolveConsumer) Start 

func (c *FactorySentryResolveConsumer) Start() error

func (*FactorySentryResolveConsumer) Stop 

func (c *FactorySentryResolveConsumer) Stop()

type FactoryVelocitySyncWorker 

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

FactoryVelocitySyncWorker keeps factory_velocity_repository_merges current for every workspace that selected a GitHub integration and an app repository.

Velocity used to call GitHub inside the request that rendered the page, which made the people half of the report disappear on any API failure. This worker moves that traffic off the request path and makes the data durable.

func NewFactoryVelocitySyncWorker 

func NewFactoryVelocitySyncWorker(rabbitMQURL string, encryptor crypto.Encryptor, reg *registry.Registry) *FactoryVelocitySyncWorker

func (*FactoryVelocitySyncWorker) Name 

func (w *FactoryVelocitySyncWorker) Name() string

func (*FactoryVelocitySyncWorker) Start 

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

func (*FactoryVelocitySyncWorker) SyncFactory 

func (w *FactoryVelocitySyncWorker) SyncFactory(ctx context.Context, factoryID uuid.UUID) (bool, error)

SyncFactory rebuilds the report of one workspace now, whatever its schedule says.

A user who just merged something should not have to wait for the next scheduled run, so this bypasses the staleness check. It keeps a short guard against a sync that is already running, which also stops repeated requests from spending the request budget twice over.

It re-reads the whole history window, not only recent days: a user who asks for a fresh report expects every number on the page to be rebuilt, including merges whose classification changed since they were first stored.

It reports whether there was anything to sync: a workspace with no repository or no version control integration has nothing to read.

type FileCleanupWorker 

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

func NewFileCleanupWorker 

func NewFileCleanupWorker() *FileCleanupWorker

func (*FileCleanupWorker) Start 

func (w *FileCleanupWorker) Start(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 LogArchivingCandidate 

type LogArchivingCandidate struct {
	TaskID           uuid.UUID
	OrganizationID   uuid.UUID
	FleetSlug        string
	FleetScopeType   string
	OrganizationSlug string
	ActiveStore      string
	State            string
	ProcessingUntil  time.Time
}

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 MessageCollector added in v0.30.0 

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

* To avoid using return values to keep track of which messages need to be published * when the transaction is committed, we use a collector.

func NewMessageCollector added in v0.30.0 

func NewMessageCollector(workflowID uuid.UUID, logger *log.Entry) *MessageCollector

func (*MessageCollector) AddExecutionID added in v0.30.0 

func (c *MessageCollector) AddExecutionID(id *uuid.UUID)

func (*MessageCollector) AddQueueItemConsumed added in v0.30.0 

func (c *MessageCollector) AddQueueItemConsumed(item *models.CanvasNodeQueueItem)

func (*MessageCollector) AddQueueItemDeleted added in v0.30.0 

func (c *MessageCollector) AddQueueItemDeleted(item *models.CanvasNodeQueueItem)

func (*MessageCollector) OnNewEvents added in v0.30.0 

func (c *MessageCollector) OnNewEvents(events []models.CanvasEvent)

func (*MessageCollector) Publish added in v0.30.0 

func (c *MessageCollector) Publish()

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, 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, 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 NodeRequestCleanupWorker added in v0.30.0 

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

NodeRequestCleanupWorker deletes old rows that otherwise grow without bound. It removes completed workflow node requests and expired webhook receipts.

func NewNodeRequestCleanupWorker added in v0.30.0 

func NewNodeRequestCleanupWorker() *NodeRequestCleanupWorker

func (*NodeRequestCleanupWorker) Start added in v0.30.0 

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

type NodeRequestWorker 

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

func NewNodeRequestWorker 

func NewNodeRequestWorker(encryptor crypto.Encryptor, registry *registry.Registry, 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(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 PlanningSessionCleanupWorker 

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

PlanningSessionCleanupWorker ends planning sessions whose heartbeat stopped and cancels the hidden canvas run so the fleet machine stops.

func NewPlanningSessionCleanupWorker 

func NewPlanningSessionCleanupWorker() *PlanningSessionCleanupWorker

func (*PlanningSessionCleanupWorker) Start 

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

type PriceBookSyncWorker 

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

PriceBookSyncWorker keeps model rates current from provider catalogs.

func NewPriceBookSyncWorker 

func NewPriceBookSyncWorker(encryptor crypto.Encryptor, reg *registry.Registry) *PriceBookSyncWorker

NewPriceBookSyncWorker builds a worker that syncs OpenRouter catalog prices.

func (*PriceBookSyncWorker) Start 

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

type RunCallbackDispatcher added in v0.30.0 

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

func NewRunCallbackDispatcher added in v0.30.0 

func NewRunCallbackDispatcher(tx *gorm.DB, registry *registry.Registry, run *models.CanvasRun) *RunCallbackDispatcher

func (*RunCallbackDispatcher) DispatchFinished added in v0.30.0 

func (d *RunCallbackDispatcher) DispatchFinished() error

func (*RunCallbackDispatcher) DispatchPending added in v0.30.0 

func (d *RunCallbackDispatcher) DispatchPending() error

func (*RunCallbackDispatcher) WithEventCollector added in v0.30.0 

func (d *RunCallbackDispatcher) WithEventCollector(fn func([]models.CanvasEvent)) *RunCallbackDispatcher

func (*RunCallbackDispatcher) WithExecutionCollector added in v0.30.0 

func (d *RunCallbackDispatcher) WithExecutionCollector(fn func([]models.CanvasNodeExecution)) *RunCallbackDispatcher

type RunCancellationNotifier added in v0.30.0 

type RunCancellationNotifier struct {
	Outcomes []RunCancellationOutcome
}

RunCancellationNotifier collects child run cancellations during a database transaction so callers can publish RabbitMQ messages after the transaction commits.

func (*RunCancellationNotifier) Bind added in v0.30.0 

func (n *RunCancellationNotifier) Bind(ctx *contexts.RunExecutionContext) *contexts.RunExecutionContext

func (*RunCancellationNotifier) Publish added in v0.30.0 

func (n *RunCancellationNotifier) Publish()

func (*RunCancellationNotifier) Record 

func (n *RunCancellationNotifier) Record(workflowID, runID uuid.UUID, drainResult *models.RunCancellationDrainResult)

func (*RunCancellationNotifier) RecordRuns 

func (n *RunCancellationNotifier) RecordRuns(results []*models.RunCancellationResult)

type RunCancellationOutcome added in v0.30.0 

type RunCancellationOutcome struct {
	WorkflowID  uuid.UUID
	RunID       uuid.UUID
	DrainResult *models.RunCancellationDrainResult
}

RunCancellationOutcome records a child run cancellation performed during a transaction.

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, registry *registry.Registry) *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)

func (*RunFinalizer) WithEncryptor 

func (w *RunFinalizer) WithEncryptor(encryptor crypto.Encryptor) *RunFinalizer

type RunInitializer added in v0.30.0 

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

func NewRunInitializer added in v0.30.0 

func NewRunInitializer(rabbitMQURL string, registry *registry.Registry) *RunInitializer

func (*RunInitializer) Name added in v0.30.0 

func (w *RunInitializer) Name() string

func (*RunInitializer) Start added in v0.30.0 

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

type RunnerCleanupWorker 

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

func NewRunnerCleanupWorker 

func NewRunnerCleanupWorker(interval time.Duration) *RunnerCleanupWorker

func (*RunnerCleanupWorker) Process 

func (w *RunnerCleanupWorker) Process() error

func (*RunnerCleanupWorker) Start 

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

type RunnerTaskLogCompactor 

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

func NewRunnerTaskLogCompactor 

func NewRunnerTaskLogCompactor(
	provider blob.Provider,
	activeStore runnerlogs.Store,
	installationID string,
	interval time.Duration,
	liveReaderGrace time.Duration,
) *RunnerTaskLogCompactor

func (*RunnerTaskLogCompactor) Process 

func (w *RunnerTaskLogCompactor) Process(ctx context.Context) error

func (*RunnerTaskLogCompactor) Start 

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

type SupportFeedbackConsumer 

type SupportFeedbackConsumer struct {
	Consumer     *tackle.Consumer
	RabbitMQURL  string
	EmailService services.EmailService
	Discord      *services.DiscordWebhookClient
	// contains filtered or unexported fields
}

func NewSupportFeedbackConsumer 

func NewSupportFeedbackConsumer(
	rabbitMQURL string,
	emailService services.EmailService,
	discord *services.DiscordWebhookClient,
) *SupportFeedbackConsumer

func (*SupportFeedbackConsumer) Consume 

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

func (*SupportFeedbackConsumer) Start 

func (c *SupportFeedbackConsumer) Start() error

func (*SupportFeedbackConsumer) Stop 

func (c *SupportFeedbackConsumer) Stop()

type VCSProviderCatalogWorker 

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

func NewVCSProviderCatalogWorker 

func NewVCSProviderCatalogWorker(provider string, catalog vcsProviderCatalog) *VCSProviderCatalogWorker

func (*VCSProviderCatalogWorker) Start 

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

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.