Function NewManager

internal/senderidentity/manager.go:56–95

NewManager builds the River client with the provision/reconcile/deprovision workers + the periodic orphan reaper. The DB must already have River's schema (call Migrate first). fire may be nil (no events).

(pool *pgxpool.Pool, store Store, provider Provider, fire EventFirer, cfg Config)

// workers + the periodic orphan reaper. The DB must already have River's
// schema (call Migrate first). fire may be nil (no events).
func NewManager(pool *pgxpool.Pool, store Store, provider Provider, fire EventFirer, cfg Config) (*Manager, error) {
	maxWorkers := cfg.MaxWorkers
	if maxWorkers <= 0 {
		maxWorkers = 5
	}
	reaperInterval := cfg.ReaperInterval
	if reaperInterval <= 0 {
		reaperInterval = defaultReaperInterval
	}

	workers := river.NewWorkers()
	river.AddWorker(workers, &ProvisionWorker{store: store, provider: provider, fire: fire, maxReconcileAttempt: cfg.MaxReconcileAttempts})
	river.AddWorker(workers, &ReconcileWorker{store: store, provider: provider, fire: fire})
	river.AddWorker(workers, &DeprovisionWorker{provider: provider})
	river.AddWorker(workers, &ReapWorker{store: store, provider: provider})

	periodic := []*river.PeriodicJob{
		river.NewPeriodicJob(
			river.PeriodicInterval(reaperInterval),
			func() (river.JobArgs, *river.InsertOpts) {
				// No UniqueOpts: River's periodic scheduler already inserts at
				// most one per interval, and a completed reap must not dedup-
				// block the next scheduled run (River can't drop `completed`
				// from a unique state set). The reaper is idempotent anyway.
				return ReapArgs{}, nil
			},
			&river.PeriodicJobOpts{RunOnStart: false},
		),
	}

	client, err := river.NewClient(riverpgxv5.New(pool), &river.Config{
		Queues:       map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: maxWorkers}},
		Workers:      workers,
		PeriodicJobs: periodic,
	})
	if err != nil {
		return nil, fmt.Errorf("river client: %w", err)
	}
	return &Manager{client: client}, nil
}

// Start begins working jobs. Non-blocking.
func (m *Manager) Start(ctx context.Context) error { return m.client.Start(ctx) }