schedule

Documentation

Index

• Constants
• Variables
• func PlanFires(misfire string, due, now time.Time, grace time.Duration) []time.Time
• func RegisterKind(kind string, h KindHandler)
• func ValidMisfire(s string) bool
• type Claimed
• type CreateInput
• type Enqueuer
• type FireArgs
  • func (FireArgs) Kind() string
• type FireRunner
  • func NewFireRunner(client *river.Client[pgx.Tx]) *FireRunner
  • func (r *FireRunner) Name() string
  • func (r *FireRunner) Run(ctx context.Context) error
• type FireWorker
  • func NewFireWorker(pool *pgxpool.Pool, log *slog.Logger) *FireWorker
  • func (w *FireWorker) Kind() string
  • func (w *FireWorker) Work(ctx context.Context, job *river.Job[FireArgs]) error
• type KindHandler
• type PGRepo
  • func NewPGRepo(pool *pgxpool.Pool) *PGRepo
  • func (r *PGRepo) ClaimDue(ctx context.Context, now time.Time, limit int) ([]Claimed, error)
  • func (r *PGRepo) Create(ctx context.Context, p *SchedulePlan) error
  • func (r *PGRepo) Delete(ctx context.Context, workspaceID, id string) error
  • func (r *PGRepo) Get(ctx context.Context, workspaceID, id string) (SchedulePlan, error)
  • func (r *PGRepo) ListByWorkspace(ctx context.Context, workspaceID string) ([]SchedulePlan, error)
  • func (r *PGRepo) Update(ctx context.Context, p *SchedulePlan) error
• type Repo
• type ScannerWorker
  • func (w *ScannerWorker) Name() string
  • func (w *ScannerWorker) Run(ctx context.Context) error
  • func (w *ScannerWorker) ScanOnce(ctx context.Context) (int, error)
• type SchedulePlan
• type Service
  • func NewService(repo Repo, enq Enqueuer, now func() time.Time) *Service
  • func (s *Service) Create(ctx context.Context, workspaceID string, in CreateInput) (SchedulePlan, error)
  • func (s *Service) Delete(ctx context.Context, workspaceID, id string) error
  • func (s *Service) Fire(ctx context.Context, plan SchedulePlan, scheduledFor time.Time) error
  • func (s *Service) List(ctx context.Context, workspaceID string) ([]SchedulePlan, error)
  • func (s *Service) Preview(ctx context.Context, cronExpr, tz string, n int) ([]time.Time, error)
  • func (s *Service) Update(ctx context.Context, workspaceID, id string, in UpdateInput) (SchedulePlan, error)
• type UpdateInput

Constants

const (
	MisfireSkip = "skip"

	MisfireOnce = "once"
)

const Grace = 60 * time.Second

const JobKind = "ploykit.schedule.fire"

Variables

var (
	ErrInvalidCron = errors.New("invalid cron expression")
	ErrInvalidTZ   = errors.New("invalid timezone")
	ErrNotFound    = errors.New("schedule plan not found")
	ErrInvalidKind = errors.New("kind must be 1-100 characters after trim")
	ErrBadMisfire  = errors.New("misfire must be one of: skip, once")
)

Functions

func PlanFires

func PlanFires(misfire string, due, now time.Time, grace time.Duration) []time.Time

func RegisterKind

func RegisterKind(kind string, h KindHandler)

func ValidMisfire

func ValidMisfire(s string) bool

Types

type Claimed

type Claimed struct {
	Plan SchedulePlan
	Due  time.Time
}

type CreateInput

type CreateInput struct {
	Kind     string `json:"kind"`
	CronExpr string `json:"cron_expr"`
	Timezone string `json:"timezone"`
	Misfire  string `json:"misfire"`
}

type Enqueuer

type Enqueuer interface {
	Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)
}

type FireArgs

type FireArgs struct {
	PlanID       string    `json:"plan_id"`
	ScheduledFor time.Time `json:"scheduled_for"`
}

func (FireArgs) Kind

func (FireArgs) Kind() string

type FireRunner

type FireRunner struct {
	StopWait time.Duration
	// contains filtered or unexported fields
}

func NewFireRunner

func NewFireRunner(client *river.Client[pgx.Tx]) *FireRunner

func (*FireRunner) Name

func (r *FireRunner) Name() string

func (*FireRunner) Run

func (r *FireRunner) Run(ctx context.Context) error

type FireWorker

type FireWorker struct {
	river.WorkerDefaults[FireArgs]
	// contains filtered or unexported fields
}

func NewFireWorker

func NewFireWorker(pool *pgxpool.Pool, log *slog.Logger) *FireWorker

func (*FireWorker) Kind

func (w *FireWorker) Kind() string

func (*FireWorker) Work

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

type KindHandler

type KindHandler func(ctx context.Context, plan SchedulePlan, scheduledFor time.Time) error

type PGRepo

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

func NewPGRepo

func NewPGRepo(pool *pgxpool.Pool) *PGRepo

func (*PGRepo) ClaimDue

func (r *PGRepo) ClaimDue(ctx context.Context, now time.Time, limit int) ([]Claimed, error)

func (*PGRepo) Create

func (r *PGRepo) Create(ctx context.Context, p *SchedulePlan) error

func (*PGRepo) Delete

func (r *PGRepo) Delete(ctx context.Context, workspaceID, id string) error

func (*PGRepo) Get

func (r *PGRepo) Get(ctx context.Context, workspaceID, id string) (SchedulePlan, error)

func (*PGRepo) ListByWorkspace

func (r *PGRepo) ListByWorkspace(ctx context.Context, workspaceID string) ([]SchedulePlan, error)

func (*PGRepo) Update

func (r *PGRepo) Update(ctx context.Context, p *SchedulePlan) error

type Repo

type Repo interface {
	Create(ctx context.Context, p *SchedulePlan) error
	Get(ctx context.Context, workspaceID, id string) (SchedulePlan, error)
	ListByWorkspace(ctx context.Context, workspaceID string) ([]SchedulePlan, error)
	Update(ctx context.Context, p *SchedulePlan) error

	ClaimDue(ctx context.Context, now time.Time, limit int) ([]Claimed, error)

	Delete(ctx context.Context, workspaceID, id string) error
}

type ScannerWorker

type ScannerWorker struct {
	Repo Repo

	Fire func(ctx context.Context, plan SchedulePlan, scheduledFor time.Time) error

	Every time.Duration

	Batch int

	Grace time.Duration

	Now func() time.Time
}

func (*ScannerWorker) Name

func (w *ScannerWorker) Name() string

func (*ScannerWorker) Run

func (w *ScannerWorker) Run(ctx context.Context) error

func (*ScannerWorker) ScanOnce

func (w *ScannerWorker) ScanOnce(ctx context.Context) (int, error)

type SchedulePlan

type SchedulePlan struct {
	ID          string     `json:"id"`
	WorkspaceID string     `json:"workspace_id"`
	Kind        string     `json:"kind"`
	CronExpr    string     `json:"cron_expr"`
	Timezone    string     `json:"timezone"`
	NextFireAt  time.Time  `json:"next_fire_at"`
	Misfire     string     `json:"misfire"`
	LastFiredAt *time.Time `json:"last_fired_at,omitempty"`
	CreatedAt   time.Time  `json:"created_at"`
	Enabled     bool       `json:"enabled"`
}

type Service

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

func NewService

func NewService(repo Repo, enq Enqueuer, now func() time.Time) *Service

func (*Service) Create

func (s *Service) Create(ctx context.Context, workspaceID string, in CreateInput) (SchedulePlan, error)

func (*Service) Delete

func (s *Service) Delete(ctx context.Context, workspaceID, id string) error

func (*Service) Fire

func (s *Service) Fire(ctx context.Context, plan SchedulePlan, scheduledFor time.Time) error

func (*Service) List

func (s *Service) List(ctx context.Context, workspaceID string) ([]SchedulePlan, error)

func (*Service) Preview

func (s *Service) Preview(ctx context.Context, cronExpr, tz string, n int) ([]time.Time, error)

func (*Service) Update

func (s *Service) Update(ctx context.Context, workspaceID, id string, in UpdateInput) (SchedulePlan, error)

type UpdateInput

type UpdateInput struct {
	Kind     *string `json:"kind"`
	CronExpr *string `json:"cron_expr"`
	Timezone *string `json:"timezone"`
	Misfire  *string `json:"misfire"`
	Enabled  *bool   `json:"enabled"`
}