engine

Documentation

Overview

Package engine implements the task scheduler, worker pool, and campaign processor.

Index

• func Backoff(attempt int, base, maxBackoff time.Duration, jitter *JitterSource) time.Duration
• type Engine
  • func New(store storage.Store, tg *telegram.Client, tmplRenderer tmpl.Renderer, ...) *Engine
  • func (e *Engine) PauseCampaign(ctx context.Context, id string) error
  • func (e *Engine) SetJitterSource(j *JitterSource)
  • func (e *Engine) Shutdown(ctx context.Context) error
  • func (e *Engine) Start(ctx context.Context) error
  • func (e *Engine) StartCampaign(ctx context.Context, id string) error
  • func (e *Engine) Stop()
• type JitterSource
  • func NewJitterSource() *JitterSource
  • func NewSeededJitterSource(seed1, seed2 uint64) *JitterSource
• type RateLimiter
  • func NewRateLimiter(globalRPS, perChatRPS float64) *RateLimiter
  • func (rl *RateLimiter) ChatCount() int64
  • func (rl *RateLimiter) Evict(ttl time.Duration)
  • func (rl *RateLimiter) Restore(originalRPS float64)
  • func (rl *RateLimiter) Throttle(newRPS float64)
  • func (rl *RateLimiter) WaitChat(ctx context.Context, chatID int64) error
  • func (rl *RateLimiter) WaitGlobal(ctx context.Context) error

Functions

func Backoff

func Backoff(attempt int, base, maxBackoff time.Duration, jitter *JitterSource) time.Duration

Backoff calculates an exponential backoff duration with ±25% jitter.

jitter must be non-nil; pass a JitterSource owned by the Engine. If jitter is nil, pure exponential backoff without jitter is returned (not recommended for production — thundering herd risk).

attempt starts from 1.

Types

type Engine

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

Engine orchestrates message delivery.

func New

func New(
	store storage.Store,
	tg *telegram.Client,
	tmplRenderer tmpl.Renderer,
	met *metrics.Metrics,
	cfg *config.Config,
	log *slog.Logger,
) *Engine

New creates a new Engine. The jitter source is auto-seeded; for deterministic testing, replace e.jitter after construction.

func (*Engine) PauseCampaign

func (e *Engine) PauseCampaign(ctx context.Context, id string) error

PauseCampaign transitions a campaign to paused state.

func (*Engine) SetJitterSource

func (e *Engine) SetJitterSource(j *JitterSource)

SetJitterSource replaces the jitter RNG. Must be called before Start. Use NewSeededJitterSource(seed1, seed2) for deterministic backoff in tests.

func (*Engine) Shutdown

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

Shutdown cancels all background work and waits for in-flight tasks to complete, or until ctx expires.

func (*Engine) Start

func (e *Engine) Start(ctx context.Context) error

Start launches the scheduler, worker pool, campaign processor, gauge updater, and rate-limiter eviction loop as background goroutines. It returns immediately. Start may only be called once; subsequent calls return an error.

func (*Engine) StartCampaign

func (e *Engine) StartCampaign(ctx context.Context, id string) error

StartCampaign transitions a campaign to running state.

func (*Engine) Stop

func (e *Engine) Stop()

Stop is a convenience that cancels without waiting. Prefer Shutdown.

type JitterSource

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

JitterSource provides a concurrency-safe random number generator for backoff jitter. Each Engine owns its own JitterSource so workers never contend on a global RNG.

Use NewJitterSource for auto-seeded production use, or NewSeededJitterSource for deterministic testing/debugging.

func NewJitterSource

func NewJitterSource() *JitterSource

NewJitterSource creates a JitterSource seeded from the runtime's auto-seeded global source. Safe for production.

func NewSeededJitterSource

func NewSeededJitterSource(seed1, seed2 uint64) *JitterSource

NewSeededJitterSource creates a deterministic JitterSource. Two sources with the same seeds produce identical sequences — useful for reproducing backoff patterns in tests and debugging.

type RateLimiter

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

RateLimiter combines a global token-bucket limiter with per-chat limiters. Per-chat limiters are automatically evicted after a TTL to prevent unbounded memory growth.

func NewRateLimiter

func NewRateLimiter(globalRPS, perChatRPS float64) *RateLimiter

NewRateLimiter creates a combined rate limiter.

func (*RateLimiter) ChatCount

func (rl *RateLimiter) ChatCount() int64

ChatCount returns the current number of tracked per-chat limiters.

func (*RateLimiter) Evict

func (rl *RateLimiter) Evict(ttl time.Duration)

Evict removes per-chat limiters that haven't been used within ttl. Call this periodically (e.g. every 30s) to bound memory.

func (*RateLimiter) Restore

func (rl *RateLimiter) Restore(originalRPS float64)

Restore resets the global limiter to its original rate.

func (*RateLimiter) Throttle

func (rl *RateLimiter) Throttle(newRPS float64)

Throttle temporarily lowers the global rate (e.g. after 429).

func (*RateLimiter) WaitChat

func (rl *RateLimiter) WaitChat(ctx context.Context, chatID int64) error

WaitChat blocks until the per-chat limiter allows an event.

func (*RateLimiter) WaitGlobal

func (rl *RateLimiter) WaitGlobal(ctx context.Context) error

WaitGlobal blocks until the global limiter allows an event.