worker

package

Go to main page

Versions in this module

v0

v0.18.0-rc2

Jul 9, 2026

v0.18.0-rc1

Jul 8, 2026

v0.17.2

Jun 30, 2026

v0.17.1

Jun 26, 2026

v0.17.0

Jun 25, 2026

Changes in this version

+ const DashboardSuggestionMaxAttempts

+ const DashboardSuggestionQueue

+ const JobTypeDashboardSuggestionCell

+ var ErrDashboardSuggestionWorkerDisabled = errors.New("dashboard suggestion worker is disabled")

+ var ErrDashboardSuggestionWorkerNotRegistered = errors.New("dashboard suggestion worker is not registered")

+ func JobInsertOptionsForDashboardSuggestionCell() *river.InsertOpts

+ type DashboardSuggestionCellArgs struct

+ CellIndex int

+ RunID uuid.UUID

+ func (DashboardSuggestionCellArgs) Kind() string

+ func (DashboardSuggestionCellArgs) Timeout() time.Duration

+ type DashboardSuggestionWorker struct

+ func NewDashboardSuggestionWorker(db *gorm.DB, llmClient llm.Client, aiCfg *config.AIConfig, ...) *DashboardSuggestionWorker

+ func (w *DashboardSuggestionWorker) Timeout(job *river.Job[DashboardSuggestionCellArgs]) time.Duration

+ func (w *DashboardSuggestionWorker) Work(ctx context.Context, job *river.Job[DashboardSuggestionCellArgs]) error

type Service

+ func (s *Service) EnqueueDashboardSuggestionCells(ctx context.Context, runID uuid.UUID, cellCount int) error

v0.17.0-rc3

Jun 25, 2026

v0.17.0-rc2

Jun 18, 2026

v0.17.0-rc1

Jun 11, 2026

v0.16.3

May 27, 2026

v0.16.2

May 25, 2026

Changes in this version

+ const JobTypeWorkflowDueSoonChecker

+ type DigestServiceWithSourceJobID interface

+ SendGlobalDigestWithSourceJobID func(ctx context.Context, sourceJobID string) error

v0.16.1

May 22, 2026

v0.16.0

May 7, 2026

Changes in this version

+ const JobTypeSendSlackChannel

+ const JobTypeSendSlackDM

+ type PoamNotificationServiceFactory struct

+ func NewPoamNotificationServiceFactory(notificationRuntime notification.RuntimeProvider) *PoamNotificationServiceFactory

+ func (f *PoamNotificationServiceFactory) New(users notification.UserRepository) (*notification.Service, error)

+ type RiskNotificationServiceFactory struct

+ func NewDirectRiskNotificationServiceFactory(emailService EmailService, slackService SlackService) *RiskNotificationServiceFactory

+ func NewRiskNotificationServiceFactory(notificationRuntime notification.RuntimeProvider) *RiskNotificationServiceFactory

+ func (f *RiskNotificationServiceFactory) New(users notification.UserRepository) (*notification.Service, error)

type SendEmailArgs

+ CorrelationID string

+ NotificationKind string

+ RecipientUserID string

+ SourceJobID string

+ SourceJobKind string

+ type SendSlackArgs struct

+ Blocks slack.Blocks

+ Channel string

+ CorrelationID string

+ NotificationKind string

+ RecipientUserID string

+ SourceJobID string

+ SourceJobKind string

+ TargetType string

+ Text string

+ type SendSlackChannelArgs SendSlackArgs

+ func (SendSlackChannelArgs) Kind() string

+ func (SendSlackChannelArgs) Timeout() time.Duration

+ type SendSlackDMArgs SendSlackArgs

+ func (SendSlackDMArgs) Kind() string

+ func (SendSlackDMArgs) Timeout() time.Duration

+ type SendSlackWorker struct

+ func NewSendSlackWorker(slackService SlackService, logger *zap.SugaredLogger) *SendSlackWorker

+ func (w *SendSlackWorker) WorkChannel(ctx context.Context, job *river.Job[SendSlackChannelArgs]) error

+ func (w *SendSlackWorker) WorkDM(ctx context.Context, job *river.Job[SendSlackDMArgs]) error

type Service

+ func (s *Service) EnqueueNotificationEmail(ctx context.Context, delivery emailprovider.Delivery) error

+ func (s *Service) EnqueueNotificationSlack(ctx context.Context, delivery slackprovider.Delivery) error

v0.16.0-rc2

May 4, 2026

v0.16.0-rc1

Apr 27, 2026

v0.15.0

Apr 15, 2026

Changes in this version

+ const JobTypeMilestoneOverdueReminder

+ const JobTypeMilestoneOverdueScannerScanner

+ const JobTypePoamDeadlineReminder

+ const JobTypePoamDeadlineReminderScanner

+ const JobTypePoamOpenDigest

+ const JobTypePoamOpenDigestScheduler

+ const JobTypePoamOverdueNotification

+ const JobTypePoamOverdueTransitionScanner

+ const JobTypeRiskOrphanedCleanup

+ const JobTypeSendGlobalDigestDelivery

+ func JobInsertOptionsForPoamDigest(byPeriod time.Duration) *river.InsertOpts

+ func JobInsertOptionsForPoamNotification(byPeriod time.Duration) *river.InsertOpts

+ func JobInsertOptionsForPoamWorker(byPeriod time.Duration) *river.InsertOpts

+ func JobInsertOptionsForRiskOrphanedCleanup() *river.InsertOpts

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

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

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

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

type DigestService

+ SendGlobalDigestDelivery func(ctx context.Context, args SendGlobalDigestDeliveryArgs) error

+ type EvidenceDigestItem struct

+ Description string

+ ExpiresAt string

+ ID string

+ Labels []string

+ Status string

+ Title string

+ UUID string

+ type EvidenceDigestSummary struct

+ ExpiredCount int64

+ NotSatisfiedCount int64

+ OtherCount int64

+ SatisfiedCount int64

+ TopExpired []EvidenceDigestItem

+ TopNotSatisfied []EvidenceDigestItem

+ TotalCount int64

+ type MilestoneOverdueReminderArgs struct

+ DueDate string

+ MilestoneID uuid.UUID

+ MilestoneTitle string

+ PoamItemID uuid.UUID

+ PoamTitle string

+ PoamURL string

+ RecipientUserID uuid.UUID

+ SspDisplayName string

+ SspID uuid.UUID

+ WeeklyBucket string

+ func (MilestoneOverdueReminderArgs) Kind() string

+ func (MilestoneOverdueReminderArgs) Timeout() time.Duration

+ type MilestoneOverdueReminderWorker struct

+ func NewMilestoneOverdueReminderWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, ...) *MilestoneOverdueReminderWorker

+ func (w *MilestoneOverdueReminderWorker) Work(ctx context.Context, job *river.Job[MilestoneOverdueReminderArgs]) error

+ type MilestoneOverdueScannerArgs struct

+ func (MilestoneOverdueScannerArgs) Kind() string

+ func (MilestoneOverdueScannerArgs) Timeout() time.Duration

+ type MilestoneOverdueScannerWorker struct

+ func NewMilestoneOverdueScannerWorker(db *gorm.DB, client workflow.RiverClient, userRepo UserRepository, ...) *MilestoneOverdueScannerWorker

+ func (w *MilestoneOverdueScannerWorker) Work(ctx context.Context, job *river.Job[MilestoneOverdueScannerArgs]) error

+ type NotificationSubscription struct

+ Channels []string

+ NotificationType string

type NotificationUser

+ NotificationSubscriptions []NotificationSubscription

+ SlackUserID string

+ func (u NotificationUser) NotificationChannels(notificationType string) []string

+ type PoamDeadlineReminderArgs struct

+ CurrentStatus string

+ Deadline string

+ MilestoneCount int

+ PoamItemID uuid.UUID

+ PoamTitle string

+ PoamURL string

+ RecipientUserID uuid.UUID

+ ReminderWindowBucket string

+ SspDisplayName string

+ SspID uuid.UUID

+ func (PoamDeadlineReminderArgs) Kind() string

+ func (PoamDeadlineReminderArgs) Timeout() time.Duration

+ type PoamDeadlineReminderScannerArgs struct

+ func (PoamDeadlineReminderScannerArgs) Kind() string

+ func (PoamDeadlineReminderScannerArgs) Timeout() time.Duration

+ type PoamDeadlineReminderScannerWorker struct

+ func NewPoamDeadlineReminderScannerWorker(db *gorm.DB, client workflow.RiverClient, userRepo UserRepository, ...) *PoamDeadlineReminderScannerWorker

+ func (w *PoamDeadlineReminderScannerWorker) Work(ctx context.Context, job *river.Job[PoamDeadlineReminderScannerArgs]) error

+ type PoamDeadlineReminderWorker struct

+ func NewPoamDeadlineReminderWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, ...) *PoamDeadlineReminderWorker

+ func (w *PoamDeadlineReminderWorker) Work(ctx context.Context, job *river.Job[PoamDeadlineReminderArgs]) error

+ type PoamDigestEmailItem struct

+ Deadline string

+ MilestoneCount int

+ POCName string

+ PoamItemID uuid.UUID

+ PoamURL string

+ SSPName string

+ Status string

+ Title string

+ type PoamMilestoneDigestEmailItem struct

+ DueDate string

+ MilestoneTitle string

+ PoamTitle string

+ PoamURL string

+ SSPName string

+ type PoamOpenDigestArgs struct

+ RecipientUserID uuid.UUID

+ WindowEnd string

+ WindowKind string

+ WindowStart string

+ func (PoamOpenDigestArgs) Kind() string

+ func (PoamOpenDigestArgs) Timeout() time.Duration

+ type PoamOpenDigestSchedulerArgs struct

+ func (PoamOpenDigestSchedulerArgs) Kind() string

+ func (PoamOpenDigestSchedulerArgs) Timeout() time.Duration

+ type PoamOpenDigestSchedulerWorker struct

+ func NewPoamOpenDigestSchedulerWorker(db *gorm.DB, client workflow.RiverClient, windowKind string, ...) *PoamOpenDigestSchedulerWorker

+ func (w *PoamOpenDigestSchedulerWorker) Work(ctx context.Context, _ *river.Job[PoamOpenDigestSchedulerArgs]) error

+ type PoamOpenDigestWorker struct

+ func NewPoamOpenDigestWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *PoamOpenDigestWorker

+ func (w *PoamOpenDigestWorker) Work(ctx context.Context, job *river.Job[PoamOpenDigestArgs]) error

+ type PoamOverdueNotificationArgs struct

+ Deadline string

+ OverdueWindow string

+ PoamItemID uuid.UUID

+ PoamTitle string

+ PoamURL string

+ RecipientUserID uuid.UUID

+ SspDisplayName string

+ SspID uuid.UUID

+ func (PoamOverdueNotificationArgs) Kind() string

+ func (PoamOverdueNotificationArgs) Timeout() time.Duration

+ type PoamOverdueNotificationWorker struct

+ func NewPoamOverdueNotificationWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, ...) *PoamOverdueNotificationWorker

+ func (w *PoamOverdueNotificationWorker) Work(ctx context.Context, job *river.Job[PoamOverdueNotificationArgs]) error

+ type PoamOverdueTransitionScannerArgs struct

+ func (PoamOverdueTransitionScannerArgs) Kind() string

+ func (PoamOverdueTransitionScannerArgs) Timeout() time.Duration

+ type PoamOverdueTransitionScannerWorker struct

+ func NewPoamOverdueTransitionScannerWorker(db *gorm.DB, client workflow.RiverClient, userRepo UserRepository, ...) *PoamOverdueTransitionScannerWorker

+ func (w *PoamOverdueTransitionScannerWorker) Work(ctx context.Context, job *river.Job[PoamOverdueTransitionScannerArgs]) error

+ type ProfileControlResolver interface

+ ResolveProfileControlKeys func(ctx context.Context, profileID uuid.UUID) ([]riskrel.ControlKey, error)

type RiskOpenDigestArgs

+ Channel string

+ type RiskOpenDigestPlan struct

+ RecipientUserIDs []uuid.UUID

+ WindowByPeriod time.Duration

+ WindowEnd time.Time

+ WindowKind string

+ WindowStart time.Time

+ func BuildRiskOpenDigestPlan(ctx context.Context, db *gorm.DB, windowKind string, now time.Time, ...) (*RiskOpenDigestPlan, error)

+ type RiskOrphanedCleanupArgs struct

+ NewProfileID *uuid.UUID

+ OldProfileID *uuid.UUID

+ SSPID uuid.UUID

+ func (RiskOrphanedCleanupArgs) Kind() string

+ func (RiskOrphanedCleanupArgs) Timeout() time.Duration

+ type RiskOrphanedCleanupWorker struct

+ func NewRiskOrphanedCleanupWorker(db *gorm.DB, riskService *riskrel.RiskService, ...) *RiskOrphanedCleanupWorker

+ func (w *RiskOrphanedCleanupWorker) Work(ctx context.Context, job *river.Job[RiskOrphanedCleanupArgs]) error

type RiskReviewDueReminderArgs

+ Channel string

type RiskReviewOverdueEscalationArgs

+ Channel string

type RiskStaleOpenReminderArgs

+ Channel string

+ type SendGlobalDigestDeliveryArgs struct

+ Channel string

+ Email string

+ SlackChannel string

+ Summary EvidenceDigestSummary

+ UserID string

+ UserName string

+ func (SendGlobalDigestDeliveryArgs) Kind() string

+ func (SendGlobalDigestDeliveryArgs) Timeout() time.Duration

+ type SendGlobalDigestDeliveryWorker struct

+ func NewSendGlobalDigestDeliveryWorker(digestService DigestService, logger *zap.SugaredLogger) *SendGlobalDigestDeliveryWorker

+ func (w *SendGlobalDigestDeliveryWorker) Work(ctx context.Context, job *river.Job[SendGlobalDigestDeliveryArgs]) error

type Service

+ func (s *Service) EnqueueOrphanedRiskCleanup(ctx context.Context, sspID uuid.UUID, oldProfileID, newProfileID *uuid.UUID) error

+ func (s *Service) EnqueueSendGlobalDigestDeliveries(ctx context.Context, args []SendGlobalDigestDeliveryArgs) error

+ type SlackService interface

+ IsEnabled func() bool

+ SendMessage func(ctx context.Context, channel string, message *slacktypes.Message) (*slacktypes.SendResult, error)

type WorkflowTaskAssignedArgs

+ Channel string

type WorkflowTaskDigestArgs

+ Channel string

type WorkflowTaskDueSoonArgs

+ Channel string

v0.15.0-rc3

Apr 14, 2026

v0.15.0-rc2

Apr 9, 2026

v0.15.0-rc1

Apr 7, 2026

v0.14.1

Mar 26, 2026

v0.14.0

Mar 26, 2026

Changes in this version

+ const JobTypeRiskOpenDigest

+ const JobTypeRiskOpenDigestScheduler

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

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

+ type RiskDigestEmailItem struct

+ OwnerName string

+ ReviewDeadline string

+ RiskURL string

+ SSPName string

+ Severity string

+ Status string

+ Title string

+ type RiskOpenDigestArgs struct

+ RecipientUserID uuid.UUID

+ WindowEnd string

+ WindowKind string

+ WindowStart string

+ func (RiskOpenDigestArgs) Kind() string

+ func (RiskOpenDigestArgs) Timeout() time.Duration

+ type RiskOpenDigestSchedulerArgs struct

+ func (RiskOpenDigestSchedulerArgs) Kind() string

+ func (RiskOpenDigestSchedulerArgs) Timeout() time.Duration

+ type RiskOpenDigestSchedulerWorker struct

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

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

+ type RiskOpenDigestWorker struct

+ func NewRiskOpenDigestWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *RiskOpenDigestWorker

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

v0.13.0

Mar 20, 2026

Changes in this version

+ const JobTypeRiskEvidenceReconciliationScanner

+ const JobTypeRiskProcessEvidence

+ const JobTypeRiskReconcileDuplicates

+ const JobTypeRiskReviewDeadlineReminderScanner

+ const JobTypeRiskReviewDueReminder

+ const JobTypeRiskReviewOverdueEscalation

+ const JobTypeRiskReviewOverdueEscalationScanner

+ const JobTypeRiskReviewOverdueReopen

+ const JobTypeRiskStaleOpenReminder

+ const JobTypeRiskStaleRiskScanner

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

+ func JobInsertOptionsForRiskProcessEvidence() *river.InsertOpts

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

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

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

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

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

type NotificationUser

+ RiskNotificationsSubscribed bool

+ type RiskEvidenceReconciliationScannerArgs struct

+ func (RiskEvidenceReconciliationScannerArgs) Kind() string

+ func (RiskEvidenceReconciliationScannerArgs) Timeout() time.Duration

+ type RiskEvidenceReconciliationScannerWorker struct

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

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

+ type RiskEvidenceWorker struct

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

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

+ type RiskProcessEvidenceArgs struct

+ EvidenceEnd string

+ EvidenceID uuid.UUID

+ Status string

+ func (RiskProcessEvidenceArgs) Kind() string

+ func (RiskProcessEvidenceArgs) Timeout() time.Duration

+ type RiskReconcileDuplicatesArgs struct

+ DedupeKey string

+ func (RiskReconcileDuplicatesArgs) Kind() string

+ func (RiskReconcileDuplicatesArgs) Timeout() time.Duration

+ type RiskReconcileDuplicatesWorker struct

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

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

+ type RiskReviewDeadlineReminderScannerArgs struct

+ func (RiskReviewDeadlineReminderScannerArgs) Kind() string

+ func (RiskReviewDeadlineReminderScannerArgs) Timeout() time.Duration

+ type RiskReviewDeadlineReminderScannerWorker struct

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

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

+ type RiskReviewDueReminderArgs struct

+ OwnerUserID uuid.UUID

+ ReminderWindow string

+ ReviewDeadline string

+ RiskID uuid.UUID

+ func (RiskReviewDueReminderArgs) Kind() string

+ func (RiskReviewDueReminderArgs) Timeout() time.Duration

+ type RiskReviewDueReminderWorker struct

+ func NewRiskReviewDueReminderWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *RiskReviewDueReminderWorker

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

+ type RiskReviewOverdueEscalationArgs struct

+ OverdueWindow string

+ OwnerUserID uuid.UUID

+ ReviewDeadline string

+ RiskID uuid.UUID

+ func (RiskReviewOverdueEscalationArgs) Kind() string

+ func (RiskReviewOverdueEscalationArgs) Timeout() time.Duration

+ type RiskReviewOverdueEscalationScannerArgs struct

+ func (RiskReviewOverdueEscalationScannerArgs) Kind() string

+ func (RiskReviewOverdueEscalationScannerArgs) Timeout() time.Duration

+ type RiskReviewOverdueEscalationScannerWorker struct

+ func NewRiskReviewOverdueEscalationScannerWorker(db *gorm.DB, client workflow.RiverClient, logger *zap.SugaredLogger, ...) *RiskReviewOverdueEscalationScannerWorker

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

+ type RiskReviewOverdueEscalationWorker struct

+ func NewRiskReviewOverdueEscalationWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *RiskReviewOverdueEscalationWorker

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

+ type RiskReviewOverdueReopenArgs struct

+ ReviewDeadline string

+ RiskID uuid.UUID

+ ThresholdDays int

+ func (RiskReviewOverdueReopenArgs) Kind() string

+ func (RiskReviewOverdueReopenArgs) Timeout() time.Duration

+ type RiskReviewOverdueReopenWorker struct

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

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

+ type RiskStaleOpenReminderArgs struct

+ LastSeenAt string

+ OwnerUserID uuid.UUID

+ RiskID uuid.UUID

+ StaleBucketDate string

+ func (RiskStaleOpenReminderArgs) Kind() string

+ func (RiskStaleOpenReminderArgs) Timeout() time.Duration

+ type RiskStaleOpenReminderWorker struct

+ func NewRiskStaleOpenReminderWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *RiskStaleOpenReminderWorker

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

+ type RiskStaleRiskScannerArgs struct

+ func (RiskStaleRiskScannerArgs) Kind() string

+ func (RiskStaleRiskScannerArgs) Timeout() time.Duration

+ type RiskStaleRiskScannerWorker struct

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

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

type Service

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

v0.13.0-rc3

Mar 20, 2026

v0.13.0-rc2

Mar 19, 2026

v0.13.0-rc1

Mar 12, 2026

v0.12.0

Feb 23, 2026

Changes in this version

+ const JobTypeWorkflowExecutionFailed

+ const JobTypeWorkflowTaskAssigned

+ const JobTypeWorkflowTaskDigest

+ const JobTypeWorkflowTaskDueSoon

+ func JobInsertOptionsForWorkflowNotification() *river.InsertOpts

+ func JobInsertOptionsForWorkflowTaskAssignedNotification() *river.InsertOpts

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

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

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

+ type DigestTask struct

+ DueDate *string

+ StepTitle string

+ StepURL string

+ WorkflowInstanceTitle string

+ WorkflowTitle string

+ type DueSoonCheckerArgs struct

+ func (DueSoonCheckerArgs) Kind() string

+ func (DueSoonCheckerArgs) Timeout() time.Duration

+ type DueSoonCheckerWorker struct

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

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

type EmailService

+ GetDefaultFromAddress func() string

+ UseTemplate func(templateName string, data map[string]interface{}) (htmlContent, textContent string, err error)

+ type GORMUserRepository struct

+ func NewGORMUserRepository(db *gorm.DB) *GORMUserRepository

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

+ type NotificationUser struct

+ Email string

+ FirstName string

+ ID string

+ LastName string

+ TaskAvailableEmailSubscribed bool

+ TaskDailyDigestSubscribed bool

+ func (u NotificationUser) FullName() string

type Service

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

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

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

+ type UserRepository interface

+ FindUserByID func(ctx context.Context, userID string) (NotificationUser, error)

+ type WorkflowExecutionFailedArgs struct

+ WorkflowExecutionID string

+ func (WorkflowExecutionFailedArgs) Kind() string

+ func (WorkflowExecutionFailedArgs) Timeout() time.Duration

+ type WorkflowExecutionFailedWorker struct

+ func NewWorkflowExecutionFailedWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *WorkflowExecutionFailedWorker

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

+ type WorkflowTaskAssignedArgs struct

+ AssignedToType string

+ DueDate *time.Time

+ StepExecutionID string

+ StepTitle string

+ StepURL string

+ UserID string

+ WorkflowInstanceTitle string

+ WorkflowTitle string

+ func (WorkflowTaskAssignedArgs) Kind() string

+ func (WorkflowTaskAssignedArgs) Timeout() time.Duration

+ type WorkflowTaskAssignedWorker struct

+ func NewWorkflowTaskAssignedWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, ...) *WorkflowTaskAssignedWorker

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

+ type WorkflowTaskDigestArgs struct

+ UserID string

+ func (WorkflowTaskDigestArgs) Kind() string

+ func (WorkflowTaskDigestArgs) Timeout() time.Duration

+ type WorkflowTaskDigestCheckerArgs struct

+ func (WorkflowTaskDigestCheckerArgs) Kind() string

+ func (WorkflowTaskDigestCheckerArgs) Timeout() time.Duration

+ type WorkflowTaskDigestCheckerWorker struct

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

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

+ type WorkflowTaskDigestWorker struct

+ func NewWorkflowTaskDigestWorker(db *gorm.DB, emailService EmailService, userRepo UserRepository, ...) *WorkflowTaskDigestWorker

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

+ type WorkflowTaskDueSoonArgs struct

+ DueDate time.Time

+ StepExecutionID string

+ StepTitle string

+ StepURL string

+ UserID string

+ WorkflowInstanceTitle string

+ WorkflowTitle string

+ func (WorkflowTaskDueSoonArgs) Kind() string

+ func (WorkflowTaskDueSoonArgs) Timeout() time.Duration

+ type WorkflowTaskDueSoonWorker struct

+ func NewWorkflowTaskDueSoonWorker(emailService EmailService, userRepo UserRepository, webBaseURL string, ...) *WorkflowTaskDueSoonWorker

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

v0.11.0

Jan 27, 2026

v0.10.0

Jan 21, 2026

Changes in this version

+ const JobTypeSendEmail

+ const JobTypeSendEmailFrom

+ const JobTypeSendGlobalDigest

+ func JobInsertOptions() *river.InsertOpts

+ func JobInsertOptionsWithQueue(queue string) *river.InsertOpts

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

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

+ func Workers(emailService EmailService, digestService DigestService, logger Logger) *river.Workers

+ type DigestService interface

+ SendGlobalDigest func(ctx context.Context) error

+ type EmailService interface

+ Send func(ctx context.Context, message *types.Message) (*types.SendResult, error)

+ SendWithProvider func(ctx context.Context, providerName string, message *types.Message) (*types.SendResult, error)

+ type Logger interface

+ Debugw func(msg string, keysAndValues ...interface{})

+ Errorw func(msg string, keysAndValues ...interface{})

+ Infow func(msg string, keysAndValues ...interface{})

+ Warnw func(msg string, keysAndValues ...interface{})

+ type SendEmailArgs struct

+ Attachments []types.Attachment

+ Bcc []string

+ Cc []string

+ From string

+ HTMLBody string

+ Headers map[string]string

+ Subject string

+ TextBody string

+ To []string

+ func (SendEmailArgs) Kind() string

+ func (SendEmailArgs) Timeout() time.Duration

+ type SendEmailFromArgs struct

+ Attachments []types.Attachment

+ Bcc []string

+ Cc []string

+ From string

+ HTMLBody string

+ Headers map[string]string

+ Provider string

+ Subject string

+ TextBody string

+ To []string

+ func (SendEmailFromArgs) Kind() string

+ func (SendEmailFromArgs) Timeout() time.Duration

+ type SendEmailFromWorker struct

+ func NewSendEmailFromWorker(emailService EmailService, logger Logger) *SendEmailFromWorker

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

+ type SendEmailWorker struct

+ func NewSendEmailWorker(emailService EmailService, logger Logger) *SendEmailWorker

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

+ type SendGlobalDigestArgs struct

+ func (SendGlobalDigestArgs) Kind() string

+ func (SendGlobalDigestArgs) Timeout() time.Duration

+ type SendGlobalDigestWorker struct

+ func NewSendGlobalDigestWorker(digestService DigestService, logger Logger) *SendGlobalDigestWorker

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

+ type Service struct

+ func NewService(cfg *config.WorkerConfig, db *gorm.DB, emailSvc *email.Service, ...) (*Service, error)

+ func NewServiceWithDigest(cfg *config.WorkerConfig, db *gorm.DB, emailSvc *email.Service, ...) (*Service, error)

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

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

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

+ func (s *Service) IsStarted() bool

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

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

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

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