worker

package
v0.14.1 Latest Latest
Warning

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

 Go to latest Published: Mar 26, 2026 License: AGPL-3.0 Imports: 32 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	JobTypeSendEmail        = "send_email"
	JobTypeSendEmailFrom    = "send_email_from"
	JobTypeSendGlobalDigest = "send_global_digest"
)

Job types for email processing

View Source 
const (
	JobTypeWorkflowTaskAssigned    = "workflow_task_assigned"
	JobTypeWorkflowTaskDueSoon     = "workflow_task_due_soon"
	JobTypeWorkflowTaskDigest      = "workflow_task_digest"
	JobTypeWorkflowExecutionFailed = "workflow_execution_failed"
)

Job types for workflow notifications

View Source 
const (
	JobTypeRiskReviewDeadlineReminderScanner  = "risk_review_deadline_reminder_scanner"
	JobTypeRiskReviewOverdueEscalationScanner = "risk_review_overdue_escalation_scanner"
	JobTypeRiskStaleRiskScanner               = "risk_stale_risk_scanner"
	JobTypeRiskEvidenceReconciliationScanner  = "risk_evidence_reconciliation_scanner"
	JobTypeRiskOpenDigestScheduler            = "risk_open_digest_scheduler"

	JobTypeRiskReviewDueReminder       = "risk_review_due_reminder"
	JobTypeRiskReviewOverdueEscalation = "risk_review_overdue_escalation"
	JobTypeRiskStaleOpenReminder       = "risk_stale_open_reminder"
	JobTypeRiskReconcileDuplicates     = "risk_reconcile_duplicates"
	JobTypeRiskReviewOverdueReopen     = "risk_review_overdue_reopen"
	JobTypeRiskOpenDigest              = "risk_open_digest"
)
View Source 
const (
	JobTypeRiskProcessEvidence = "risk_process_evidence"
)

Job types for risk processing

Variables

This section is empty.

Functions

func JobInsertOptionsForRiskDigest added in v0.14.0 

func JobInsertOptionsForRiskDigest(byPeriod time.Duration) *river.InsertOpts

func JobInsertOptionsForRiskNotification added in v0.13.0 

func JobInsertOptionsForRiskNotification(byPeriod time.Duration) *river.InsertOpts

func JobInsertOptionsForRiskProcessEvidence added in v0.13.0 

func JobInsertOptionsForRiskProcessEvidence() *river.InsertOpts

JobInsertOptionsForRiskProcessEvidence returns insert options for risk processing jobs

func JobInsertOptionsForRiskWorkerUnique added in v0.13.0 

func JobInsertOptionsForRiskWorkerUnique(byPeriod time.Duration) *river.InsertOpts

func JobInsertOptionsForWorkflowNotification added in v0.12.0 

func JobInsertOptionsForWorkflowNotification() *river.InsertOpts

JobInsertOptionsForWorkflowNotification returns insert options for workflow notification email jobs

func JobInsertOptionsForWorkflowTaskAssignedNotification added in v0.12.0 

func JobInsertOptionsForWorkflowTaskAssignedNotification() *river.InsertOpts

func JobInsertOptionsWithRetry 

func JobInsertOptionsWithRetry(queue string, maxAttempts int) *river.InsertOpts

JobInsertOptionsWithRetry returns insert options for jobs with custom retry policy

func NewDigestPeriodicJob 

func NewDigestPeriodicJob(cronSchedule string, logger *zap.SugaredLogger) *river.PeriodicJob

NewDigestPeriodicJob creates a periodic job for digest scheduling

func NewDueSoonCheckerPeriodicJob added in v0.12.0 

func NewDueSoonCheckerPeriodicJob(schedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewRiskEvidenceReconciliationPeriodicJob added in v0.13.0 

func NewRiskEvidenceReconciliationPeriodicJob(schedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewRiskOpenDigestPeriodicJob added in v0.14.0 

func NewRiskOpenDigestPeriodicJob(schedule string, windowKind string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewRiskReviewDeadlineReminderPeriodicJob added in v0.13.0 

func NewRiskReviewDeadlineReminderPeriodicJob(schedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewRiskReviewOverdueEscalationPeriodicJob added in v0.13.0 

func NewRiskReviewOverdueEscalationPeriodicJob(schedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewRiskStaleScannerPeriodicJob added in v0.13.0 

func NewRiskStaleScannerPeriodicJob(schedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewWorkflowSchedulerPeriodicJob added in v0.12.0 

func NewWorkflowSchedulerPeriodicJob(cronSchedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func NewWorkflowTaskDigestPeriodicJob added in v0.12.0 

func NewWorkflowTaskDigestPeriodicJob(schedule string, logger *zap.SugaredLogger) *river.PeriodicJob

func Workers 

func Workers(emailService EmailService, digestService DigestService, userRepo UserRepository, db *gorm.DB, webBaseURL string, logger *zap.SugaredLogger) *river.Workers

Workers returns all workers as work functions with dependencies injected

Types

type DigestService 

type DigestService interface {
	SendGlobalDigest(ctx context.Context) error
}

DigestService interface for dependency injection

type DigestTask added in v0.12.0 

type DigestTask struct {
	StepTitle             string
	WorkflowTitle         string
	WorkflowInstanceTitle string
	DueDate               *string
	StepURL               string
}

DigestTask represents a single task entry in the digest email

type DueSoonCheckerArgs added in v0.12.0 

type DueSoonCheckerArgs struct{}

DueSoonCheckerArgs represents the arguments for the periodic due-soon checker job

func (DueSoonCheckerArgs) Kind added in v0.12.0 

func (DueSoonCheckerArgs) Kind() string

Kind returns the job kind for River

func (DueSoonCheckerArgs) Timeout added in v0.12.0 

func (DueSoonCheckerArgs) Timeout() time.Duration

Timeout returns the timeout for the due-soon checker job

type DueSoonCheckerWorker added in v0.12.0 

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

DueSoonCheckerWorker scans for step executions due in a week and enqueues reminder emails

func NewDueSoonCheckerWorker added in v0.12.0 

func NewDueSoonCheckerWorker(db *gorm.DB, client workflow.RiverClient, logger *zap.SugaredLogger) *DueSoonCheckerWorker

NewDueSoonCheckerWorker creates a new DueSoonCheckerWorker

func (*DueSoonCheckerWorker) Work added in v0.12.0 

func (w *DueSoonCheckerWorker) Work(ctx context.Context, job *river.Job[DueSoonCheckerArgs]) error

Work scans for step executions due in ~1 week and enqueues WorkflowTaskDueSoonArgs jobs

type EmailService 

type EmailService interface {
	Send(ctx context.Context, message *types.Message) (*types.SendResult, error)
	SendWithProvider(ctx context.Context, providerName string, message *types.Message) (*types.SendResult, error)
	UseTemplate(templateName string, data map[string]interface{}) (htmlContent, textContent string, err error)
	GetDefaultFromAddress() string
}

EmailService interface for dependency injection

type GORMUserRepository added in v0.12.0 

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

GORMUserRepository implements UserRepository using GORM

func NewGORMUserRepository added in v0.12.0 

func NewGORMUserRepository(db *gorm.DB) *GORMUserRepository

NewGORMUserRepository creates a new GORMUserRepository

func (*GORMUserRepository) FindUserByID added in v0.12.0 

func (r *GORMUserRepository) FindUserByID(ctx context.Context, userID string) (NotificationUser, error)

FindUserByID looks up a user by UUID string and returns a NotificationUser

type NotificationUser added in v0.12.0 

type NotificationUser struct {
	ID                           string
	Email                        string
	FirstName                    string
	LastName                     string
	TaskAvailableEmailSubscribed bool
	TaskDailyDigestSubscribed    bool
	RiskNotificationsSubscribed  bool
}

NotificationUser holds the user fields needed for sending notification emails

func (NotificationUser) FullName added in v0.12.0 

func (u NotificationUser) FullName() string

type RiskDigestEmailItem added in v0.14.0 

type RiskDigestEmailItem struct {
	Title          string
	SSPName        string
	Status         string
	Severity       string
	OwnerName      string
	ReviewDeadline string
	RiskURL        string
}

type RiskEvidenceReconciliationScannerArgs added in v0.13.0 

type RiskEvidenceReconciliationScannerArgs struct{}

func (RiskEvidenceReconciliationScannerArgs) Kind added in v0.13.0 

func (RiskEvidenceReconciliationScannerArgs) Kind() string

func (RiskEvidenceReconciliationScannerArgs) Timeout added in v0.13.0 

func (RiskEvidenceReconciliationScannerArgs) Timeout() time.Duration

type RiskEvidenceReconciliationScannerWorker added in v0.13.0 

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

func NewRiskEvidenceReconciliationScannerWorker added in v0.13.0 

func NewRiskEvidenceReconciliationScannerWorker(db *gorm.DB, client workflow.RiverClient, logger *zap.SugaredLogger) *RiskEvidenceReconciliationScannerWorker

func (*RiskEvidenceReconciliationScannerWorker) Work added in v0.13.0 

func (w *RiskEvidenceReconciliationScannerWorker) Work(ctx context.Context, _ *river.Job[RiskEvidenceReconciliationScannerArgs]) error

type RiskEvidenceWorker added in v0.13.0 

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

RiskEvidenceWorker handles processing evidence and creating risks

func NewRiskEvidenceWorker added in v0.13.0 

func NewRiskEvidenceWorker(db *gorm.DB, logger *zap.SugaredLogger) *RiskEvidenceWorker

NewRiskEvidenceWorker creates a new RiskEvidenceWorker

func (*RiskEvidenceWorker) Work added in v0.13.0 

func (w *RiskEvidenceWorker) Work(ctx context.Context, job *river.Job[RiskProcessEvidenceArgs]) error

Work is the River work function for processing evidence and creating risks

type RiskOpenDigestArgs added in v0.14.0 

type RiskOpenDigestArgs struct {
	RecipientUserID uuid.UUID `json:"recipient_user_id"`
	WindowStart     string    `json:"window_start"`
	WindowEnd       string    `json:"window_end"`
	WindowKind      string    `json:"window_kind"`
}

func (RiskOpenDigestArgs) Kind added in v0.14.0 

func (RiskOpenDigestArgs) Kind() string

func (RiskOpenDigestArgs) Timeout added in v0.14.0 

func (RiskOpenDigestArgs) Timeout() time.Duration

type RiskOpenDigestSchedulerArgs added in v0.14.0 

type RiskOpenDigestSchedulerArgs struct{}

func (RiskOpenDigestSchedulerArgs) Kind added in v0.14.0 

func (RiskOpenDigestSchedulerArgs) Kind() string

func (RiskOpenDigestSchedulerArgs) Timeout added in v0.14.0 

func (RiskOpenDigestSchedulerArgs) Timeout() time.Duration

type RiskOpenDigestSchedulerWorker added in v0.14.0 

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

func NewRiskOpenDigestSchedulerWorker added in v0.14.0 

func NewRiskOpenDigestSchedulerWorker(db *gorm.DB, client workflow.RiverClient, windowKind string, logger *zap.SugaredLogger) *RiskOpenDigestSchedulerWorker

func (*RiskOpenDigestSchedulerWorker) Work added in v0.14.0 

func (w *RiskOpenDigestSchedulerWorker) Work(ctx context.Context, _ *river.Job[RiskOpenDigestSchedulerArgs]) error

type RiskOpenDigestWorker added in v0.14.0 

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

func NewRiskOpenDigestWorker added in v0.14.0 

func NewRiskOpenDigestWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *RiskOpenDigestWorker

func (*RiskOpenDigestWorker) Work added in v0.14.0 

func (w *RiskOpenDigestWorker) Work(ctx context.Context, job *river.Job[RiskOpenDigestArgs]) error

type RiskProcessEvidenceArgs added in v0.13.0 

type RiskProcessEvidenceArgs struct {
	EvidenceID  uuid.UUID `json:"evidence_id"`
	EvidenceEnd string    `json:"evidence_end"`
	Status      string    `json:"status"`
}

RiskProcessEvidenceArgs represents the arguments for processing evidence and creating risks. EvidenceEnd and Status are included alongside EvidenceID intentionally: River uses ByArgs uniqueness to deduplicate jobs within the 5-minute window. Including the end time and status ensures that two different evidence records for the same stream (different end times or states) each get their own independent deduplication window, preventing the second record from being silently dropped.

func (RiskProcessEvidenceArgs) Kind added in v0.13.0 

func (RiskProcessEvidenceArgs) Kind() string

Kind returns the job kind for River

func (RiskProcessEvidenceArgs) Timeout added in v0.13.0 

func (RiskProcessEvidenceArgs) Timeout() time.Duration

Timeout returns the timeout for risk process evidence jobs

type RiskReconcileDuplicatesArgs added in v0.13.0 

type RiskReconcileDuplicatesArgs struct {
	DedupeKey string `json:"dedupe_key"`
}

func (RiskReconcileDuplicatesArgs) Kind added in v0.13.0 

func (RiskReconcileDuplicatesArgs) Kind() string

func (RiskReconcileDuplicatesArgs) Timeout added in v0.13.0 

func (RiskReconcileDuplicatesArgs) Timeout() time.Duration

type RiskReconcileDuplicatesWorker added in v0.13.0 

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

func NewRiskReconcileDuplicatesWorker added in v0.13.0 

func NewRiskReconcileDuplicatesWorker(db *gorm.DB, logger *zap.SugaredLogger) *RiskReconcileDuplicatesWorker

func (*RiskReconcileDuplicatesWorker) Work added in v0.13.0 

func (w *RiskReconcileDuplicatesWorker) Work(ctx context.Context, job *river.Job[RiskReconcileDuplicatesArgs]) error

type RiskReviewDeadlineReminderScannerArgs added in v0.13.0 

type RiskReviewDeadlineReminderScannerArgs struct{}

func (RiskReviewDeadlineReminderScannerArgs) Kind added in v0.13.0 

func (RiskReviewDeadlineReminderScannerArgs) Kind() string

func (RiskReviewDeadlineReminderScannerArgs) Timeout added in v0.13.0 

func (RiskReviewDeadlineReminderScannerArgs) Timeout() time.Duration

type RiskReviewDeadlineReminderScannerWorker added in v0.13.0 

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

func NewRiskReviewDeadlineReminderScannerWorker added in v0.13.0 

func NewRiskReviewDeadlineReminderScannerWorker(db *gorm.DB, client workflow.RiverClient, logger *zap.SugaredLogger) *RiskReviewDeadlineReminderScannerWorker

func (*RiskReviewDeadlineReminderScannerWorker) Work added in v0.13.0 

func (w *RiskReviewDeadlineReminderScannerWorker) Work(ctx context.Context, _ *river.Job[RiskReviewDeadlineReminderScannerArgs]) error

type RiskReviewDueReminderArgs added in v0.13.0 

type RiskReviewDueReminderArgs struct {
	RiskID         uuid.UUID `json:"risk_id"`
	OwnerUserID    uuid.UUID `json:"owner_user_id"`
	ReviewDeadline string    `json:"review_deadline"`
	ReminderWindow string    `json:"reminder_window"`
}

func (RiskReviewDueReminderArgs) Kind added in v0.13.0 

func (RiskReviewDueReminderArgs) Kind() string

func (RiskReviewDueReminderArgs) Timeout added in v0.13.0 

func (RiskReviewDueReminderArgs) Timeout() time.Duration

type RiskReviewDueReminderWorker added in v0.13.0 

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

func NewRiskReviewDueReminderWorker added in v0.13.0 

func NewRiskReviewDueReminderWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *RiskReviewDueReminderWorker

func (*RiskReviewDueReminderWorker) Work added in v0.13.0 

func (w *RiskReviewDueReminderWorker) Work(ctx context.Context, job *river.Job[RiskReviewDueReminderArgs]) error

type RiskReviewOverdueEscalationArgs added in v0.13.0 

type RiskReviewOverdueEscalationArgs struct {
	RiskID         uuid.UUID `json:"risk_id"`
	OwnerUserID    uuid.UUID `json:"owner_user_id"`
	ReviewDeadline string    `json:"review_deadline"`
	OverdueWindow  string    `json:"overdue_window"`
}

func (RiskReviewOverdueEscalationArgs) Kind added in v0.13.0 

func (RiskReviewOverdueEscalationArgs) Kind() string

func (RiskReviewOverdueEscalationArgs) Timeout added in v0.13.0 

func (RiskReviewOverdueEscalationArgs) Timeout() time.Duration

type RiskReviewOverdueEscalationScannerArgs added in v0.13.0 

type RiskReviewOverdueEscalationScannerArgs struct{}

func (RiskReviewOverdueEscalationScannerArgs) Kind added in v0.13.0 

func (RiskReviewOverdueEscalationScannerArgs) Kind() string

func (RiskReviewOverdueEscalationScannerArgs) Timeout added in v0.13.0 

func (RiskReviewOverdueEscalationScannerArgs) Timeout() time.Duration

type RiskReviewOverdueEscalationScannerWorker added in v0.13.0 

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

func NewRiskReviewOverdueEscalationScannerWorker added in v0.13.0 

func NewRiskReviewOverdueEscalationScannerWorker(
	db *gorm.DB,
	client workflow.RiverClient,
	logger *zap.SugaredLogger,
	autoReopenEnabled bool,
	autoReopenThresholdDays int,
) *RiskReviewOverdueEscalationScannerWorker

func (*RiskReviewOverdueEscalationScannerWorker) Work added in v0.13.0 

func (w *RiskReviewOverdueEscalationScannerWorker) Work(ctx context.Context, _ *river.Job[RiskReviewOverdueEscalationScannerArgs]) error

type RiskReviewOverdueEscalationWorker added in v0.13.0 

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

func NewRiskReviewOverdueEscalationWorker added in v0.13.0 

func NewRiskReviewOverdueEscalationWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *RiskReviewOverdueEscalationWorker

func (*RiskReviewOverdueEscalationWorker) Work added in v0.13.0 

func (w *RiskReviewOverdueEscalationWorker) Work(ctx context.Context, job *river.Job[RiskReviewOverdueEscalationArgs]) error

type RiskReviewOverdueReopenArgs added in v0.13.0 

type RiskReviewOverdueReopenArgs struct {
	RiskID         uuid.UUID `json:"risk_id"`
	ReviewDeadline string    `json:"review_deadline"`
	ThresholdDays  int       `json:"threshold_days"`
}

func (RiskReviewOverdueReopenArgs) Kind added in v0.13.0 

func (RiskReviewOverdueReopenArgs) Kind() string

func (RiskReviewOverdueReopenArgs) Timeout added in v0.13.0 

func (RiskReviewOverdueReopenArgs) Timeout() time.Duration

type RiskReviewOverdueReopenWorker added in v0.13.0 

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

func NewRiskReviewOverdueReopenWorker added in v0.13.0 

func NewRiskReviewOverdueReopenWorker(db *gorm.DB, logger *zap.SugaredLogger) *RiskReviewOverdueReopenWorker

func (*RiskReviewOverdueReopenWorker) Work added in v0.13.0 

func (w *RiskReviewOverdueReopenWorker) Work(ctx context.Context, job *river.Job[RiskReviewOverdueReopenArgs]) error

type RiskStaleOpenReminderArgs added in v0.13.0 

type RiskStaleOpenReminderArgs struct {
	RiskID          uuid.UUID `json:"risk_id"`
	OwnerUserID     uuid.UUID `json:"owner_user_id"`
	LastSeenAt      string    `json:"last_seen_at"`
	StaleBucketDate string    `json:"stale_bucket_date"`
}

func (RiskStaleOpenReminderArgs) Kind added in v0.13.0 

func (RiskStaleOpenReminderArgs) Kind() string

func (RiskStaleOpenReminderArgs) Timeout added in v0.13.0 

func (RiskStaleOpenReminderArgs) Timeout() time.Duration

type RiskStaleOpenReminderWorker added in v0.13.0 

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

func NewRiskStaleOpenReminderWorker added in v0.13.0 

func NewRiskStaleOpenReminderWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *RiskStaleOpenReminderWorker

func (*RiskStaleOpenReminderWorker) Work added in v0.13.0 

func (w *RiskStaleOpenReminderWorker) Work(ctx context.Context, job *river.Job[RiskStaleOpenReminderArgs]) error

type RiskStaleRiskScannerArgs added in v0.13.0 

type RiskStaleRiskScannerArgs struct{}

func (RiskStaleRiskScannerArgs) Kind added in v0.13.0 

func (RiskStaleRiskScannerArgs) Kind() string

func (RiskStaleRiskScannerArgs) Timeout added in v0.13.0 

func (RiskStaleRiskScannerArgs) Timeout() time.Duration

type RiskStaleRiskScannerWorker added in v0.13.0 

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

func NewRiskStaleRiskScannerWorker added in v0.13.0 

func NewRiskStaleRiskScannerWorker(db *gorm.DB, client workflow.RiverClient, logger *zap.SugaredLogger) *RiskStaleRiskScannerWorker

func (*RiskStaleRiskScannerWorker) Work added in v0.13.0 

func (w *RiskStaleRiskScannerWorker) Work(ctx context.Context, _ *river.Job[RiskStaleRiskScannerArgs]) error

type SendEmailArgs 

type SendEmailArgs struct {
	// Email message fields
	From        string             `json:"from"`
	To          []string           `json:"to"`
	Cc          []string           `json:"cc,omitempty"`
	Bcc         []string           `json:"bcc,omitempty"`
	Subject     string             `json:"subject"`
	HTMLBody    string             `json:"html_body,omitempty"`
	TextBody    string             `json:"text_body,omitempty"`
	Attachments []types.Attachment `json:"attachments,omitempty"`
	Headers     map[string]string  `json:"headers,omitempty"`
}

SendEmailArgs represents the arguments for sending an email

func (SendEmailArgs) Kind 

func (SendEmailArgs) Kind() string

Kind returns the job kind for River

func (SendEmailArgs) Timeout 

func (SendEmailArgs) Timeout() time.Duration

Timeout returns the timeout for email jobs

type SendEmailFromArgs 

type SendEmailFromArgs struct {
	// Provider to use for sending
	Provider string `json:"provider"`

	// Email message fields
	From        string             `json:"from"`
	To          []string           `json:"to"`
	Cc          []string           `json:"cc,omitempty"`
	Bcc         []string           `json:"bcc,omitempty"`
	Subject     string             `json:"subject"`
	HTMLBody    string             `json:"html_body,omitempty"`
	TextBody    string             `json:"text_body,omitempty"`
	Attachments []types.Attachment `json:"attachments,omitempty"`
	Headers     map[string]string  `json:"headers,omitempty"`
}

SendEmailFromArgs represents the arguments for sending an email from a specific provider

func (SendEmailFromArgs) Kind 

func (SendEmailFromArgs) Kind() string

Kind returns the job kind for River

func (SendEmailFromArgs) Timeout 

func (SendEmailFromArgs) Timeout() time.Duration

Timeout returns the timeout for email jobs

type SendEmailFromWorker 

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

SendEmailFromWorker handles sending email from provider jobs

func NewSendEmailFromWorker 

func NewSendEmailFromWorker(emailService EmailService, logger *zap.SugaredLogger) *SendEmailFromWorker

NewSendEmailFromWorker creates a new SendEmailFromWorker

func (*SendEmailFromWorker) Work 

func (w *SendEmailFromWorker) Work(ctx context.Context, job *river.Job[SendEmailFromArgs]) error

Work is the River work function for sending emails from a provider

type SendEmailWorker 

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

SendEmailWorker handles sending email jobs

func NewSendEmailWorker 

func NewSendEmailWorker(emailService EmailService, logger *zap.SugaredLogger) *SendEmailWorker

NewSendEmailWorker creates a new SendEmailWorker

func (*SendEmailWorker) Work 

func (w *SendEmailWorker) Work(ctx context.Context, job *river.Job[SendEmailArgs]) error

Work is the River work function for sending emails

type SendGlobalDigestArgs 

type SendGlobalDigestArgs struct {
}

SendGlobalDigestArgs represents the arguments for sending global digest

func (SendGlobalDigestArgs) Kind 

func (SendGlobalDigestArgs) Kind() string

Kind returns the job kind for River

func (SendGlobalDigestArgs) Timeout 

func (SendGlobalDigestArgs) Timeout() time.Duration

Timeout returns the timeout for digest jobs (longer due to multiple emails)

type SendGlobalDigestWorker 

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

SendGlobalDigestWorker handles sending global digest jobs

func NewSendGlobalDigestWorker 

func NewSendGlobalDigestWorker(digestService DigestService, logger *zap.SugaredLogger) *SendGlobalDigestWorker

NewSendGlobalDigestWorker creates a new SendGlobalDigestWorker

func (*SendGlobalDigestWorker) Work 

func (w *SendGlobalDigestWorker) Work(ctx context.Context, job *river.Job[SendGlobalDigestArgs]) error

Work is the River work function for sending global digest

type Service 

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

Service manages the River client and workers

func NewServiceWithDigest 

func NewServiceWithDigest(
	cfg *config.WorkerConfig,
	db *gorm.DB,
	emailSvc *email.Service,
	digestSvc DigestService,
	digestCfg *config.Config,
	logger *zap.SugaredLogger,
) (*Service, error)

NewServiceWithDigest creates a new worker service with digest support

func (*Service) EnqueueRiskProcessEvidence added in v0.13.0 

func (s *Service) EnqueueRiskProcessEvidence(ctx context.Context, evidenceID uuid.UUID, evidenceEnd, status string) error

EnqueueRiskProcessEvidence enqueues a risk process evidence job

func (*Service) EnqueueSendEmail 

func (s *Service) EnqueueSendEmail(ctx context.Context, args *SendEmailArgs) error

EnqueueSendEmail enqueues a send email job

func (*Service) EnqueueSendEmailFrom 

func (s *Service) EnqueueSendEmailFrom(ctx context.Context, args *SendEmailFromArgs) error

EnqueueSendEmailFrom enqueues a send email from provider job

func (*Service) EnqueueWorkflowExecutionFailed added in v0.12.0 

func (s *Service) EnqueueWorkflowExecutionFailed(ctx context.Context, execution *workflows.WorkflowExecution) error

EnqueueWorkflowExecutionFailed enqueues a workflow-execution-failed notification email job. Implements the workflow.NotificationEnqueuer interface.

func (*Service) EnqueueWorkflowTaskAssigned added in v0.12.0 

func (s *Service) EnqueueWorkflowTaskAssigned(ctx context.Context, stepExecution *workflows.StepExecution) error

EnqueueWorkflowTaskAssigned enqueues a workflow-task-assigned notification email job. Implements the workflow.NotificationEnqueuer interface.

func (*Service) GetClient 

func (s *Service) GetClient() *river.Client[pgx.Tx]

GetClient returns the River client for job insertion

func (*Service) GetDAGExecutor added in v0.12.0 

func (s *Service) GetDAGExecutor() *workflow.DAGExecutor

GetDAGExecutor returns the shared DAG executor used by workflow River workers. Returns nil when the worker service is disabled.

func (*Service) IsStarted 

func (s *Service) IsStarted() bool

IsStarted returns true if the worker service is started

func (*Service) Migrate 

func (s *Service) Migrate(ctx context.Context) error

Migrate runs River migrations

func (*Service) Start 

func (s *Service) Start(ctx context.Context) error

Start starts the worker service

func (*Service) Stop 

func (s *Service) Stop(ctx context.Context) error

Stop stops the worker service

type UserRepository added in v0.12.0 

type UserRepository interface {
	FindUserByID(ctx context.Context, userID string) (NotificationUser, error)
}

UserRepository is the minimal DB interface needed by notification workers

type WorkflowExecutionFailedArgs added in v0.12.0 

type WorkflowExecutionFailedArgs struct {
	WorkflowExecutionID string `json:"workflow_execution_id"`
}

WorkflowExecutionFailedArgs represents the arguments for a workflow-execution-failed notification email

func (WorkflowExecutionFailedArgs) Kind added in v0.12.0 

func (WorkflowExecutionFailedArgs) Kind() string

Kind returns the job kind for River

func (WorkflowExecutionFailedArgs) Timeout added in v0.12.0 

func (WorkflowExecutionFailedArgs) Timeout() time.Duration

Timeout returns the timeout for workflow execution failed jobs

type WorkflowExecutionFailedWorker added in v0.12.0 

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

WorkflowExecutionFailedWorker sends a failure notification email to the workflow instance creator

func NewWorkflowExecutionFailedWorker added in v0.12.0 

func NewWorkflowExecutionFailedWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *WorkflowExecutionFailedWorker

NewWorkflowExecutionFailedWorker creates a new WorkflowExecutionFailedWorker

func (*WorkflowExecutionFailedWorker) Work added in v0.12.0 

func (w *WorkflowExecutionFailedWorker) Work(ctx context.Context, job *river.Job[WorkflowExecutionFailedArgs]) error

Work sends a failure notification email for the workflow execution identified by job.Args.WorkflowExecutionID

type WorkflowTaskAssignedArgs added in v0.12.0 

type WorkflowTaskAssignedArgs struct {
	AssignedToType        string     `json:"assigned_to_type"`
	UserID                string     `json:"user_id"`
	StepExecutionID       string     `json:"step_execution_id"`
	StepTitle             string     `json:"step_title"`
	WorkflowTitle         string     `json:"workflow_title"`
	WorkflowInstanceTitle string     `json:"workflow_instance_title"`
	StepURL               string     `json:"step_url"`
	DueDate               *time.Time `json:"due_date,omitempty"`
}

WorkflowTaskAssignedArgs represents the arguments for a new-task-assigned notification email

func (WorkflowTaskAssignedArgs) Kind added in v0.12.0 

func (WorkflowTaskAssignedArgs) Kind() string

Kind returns the job kind for River

func (WorkflowTaskAssignedArgs) Timeout added in v0.12.0 

func (WorkflowTaskAssignedArgs) Timeout() time.Duration

Timeout returns the timeout for workflow task assigned jobs

type WorkflowTaskAssignedWorker added in v0.12.0 

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

WorkflowTaskAssignedWorker handles new-task-assigned notification email jobs

func NewWorkflowTaskAssignedWorker added in v0.12.0 

func NewWorkflowTaskAssignedWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *WorkflowTaskAssignedWorker

NewWorkflowTaskAssignedWorker creates a new WorkflowTaskAssignedWorker

func (*WorkflowTaskAssignedWorker) Work added in v0.12.0 

func (w *WorkflowTaskAssignedWorker) Work(ctx context.Context, job *river.Job[WorkflowTaskAssignedArgs]) error

Work is the River work function for sending new-task-assigned notification emails

type WorkflowTaskDigestArgs added in v0.12.0 

type WorkflowTaskDigestArgs struct {
	UserID string `json:"user_id"`
}

WorkflowTaskDigestArgs represents the arguments for a per-user task digest email

func (WorkflowTaskDigestArgs) Kind added in v0.12.0 

func (WorkflowTaskDigestArgs) Kind() string

Kind returns the job kind for River

func (WorkflowTaskDigestArgs) Timeout added in v0.12.0 

func (WorkflowTaskDigestArgs) Timeout() time.Duration

Timeout returns the timeout for workflow task digest jobs

type WorkflowTaskDigestCheckerArgs added in v0.12.0 

type WorkflowTaskDigestCheckerArgs struct{}

WorkflowTaskDigestCheckerArgs represents the arguments for the periodic digest checker job

func (WorkflowTaskDigestCheckerArgs) Kind added in v0.12.0 

func (WorkflowTaskDigestCheckerArgs) Kind() string

Kind returns the job kind for River

func (WorkflowTaskDigestCheckerArgs) Timeout added in v0.12.0 

func (WorkflowTaskDigestCheckerArgs) Timeout() time.Duration

Timeout returns the timeout for the digest checker job

type WorkflowTaskDigestCheckerWorker added in v0.12.0 

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

WorkflowTaskDigestCheckerWorker queries all subscribed users and enqueues per-user digest jobs

func NewWorkflowTaskDigestCheckerWorker added in v0.12.0 

func NewWorkflowTaskDigestCheckerWorker(db *gorm.DB, client workflow.RiverClient, logger *zap.SugaredLogger) *WorkflowTaskDigestCheckerWorker

NewWorkflowTaskDigestCheckerWorker creates a new WorkflowTaskDigestCheckerWorker

func (*WorkflowTaskDigestCheckerWorker) Work added in v0.12.0 

func (w *WorkflowTaskDigestCheckerWorker) Work(ctx context.Context, job *river.Job[WorkflowTaskDigestCheckerArgs]) error

Work queries all users with TaskDailyDigestSubscribed=true and enqueues a WorkflowTaskDigestArgs job for each

type WorkflowTaskDigestWorker added in v0.12.0 

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

WorkflowTaskDigestWorker sends a per-user digest of pending and overdue workflow tasks

func NewWorkflowTaskDigestWorker added in v0.12.0 

func NewWorkflowTaskDigestWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *WorkflowTaskDigestWorker

NewWorkflowTaskDigestWorker creates a new WorkflowTaskDigestWorker

func (*WorkflowTaskDigestWorker) Work added in v0.12.0 

func (w *WorkflowTaskDigestWorker) Work(ctx context.Context, job *river.Job[WorkflowTaskDigestArgs]) error

Work sends a digest email for the user identified by job.Args.UserID

type WorkflowTaskDueSoonArgs added in v0.12.0 

type WorkflowTaskDueSoonArgs struct {
	UserID                string    `json:"user_id"`
	StepExecutionID       string    `json:"step_execution_id"`
	StepTitle             string    `json:"step_title"`
	WorkflowTitle         string    `json:"workflow_title"`
	WorkflowInstanceTitle string    `json:"workflow_instance_title"`
	StepURL               string    `json:"step_url"`
	DueDate               time.Time `json:"due_date"`
}

WorkflowTaskDueSoonArgs represents the arguments for a task-due-in-1-day reminder email

func (WorkflowTaskDueSoonArgs) Kind added in v0.12.0 

func (WorkflowTaskDueSoonArgs) Kind() string

Kind returns the job kind for River

func (WorkflowTaskDueSoonArgs) Timeout added in v0.12.0 

func (WorkflowTaskDueSoonArgs) Timeout() time.Duration

Timeout returns the timeout for workflow task due soon jobs

type WorkflowTaskDueSoonWorker added in v0.12.0 

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

WorkflowTaskDueSoonWorker handles task-due-soon reminder email jobs

func NewWorkflowTaskDueSoonWorker added in v0.12.0 

func NewWorkflowTaskDueSoonWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, logger *zap.SugaredLogger) *WorkflowTaskDueSoonWorker

NewWorkflowTaskDueSoonWorker creates a new WorkflowTaskDueSoonWorker

func (*WorkflowTaskDueSoonWorker) Work added in v0.12.0 

func (w *WorkflowTaskDueSoonWorker) Work(ctx context.Context, job *river.Job[WorkflowTaskDueSoonArgs]) error

Work is the River work function for sending task-due-in-1-day reminder emails

Source Files

View all Source files

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.