balanceworker

Documentation

Index

• Constants
• type BalanceWorkerRepository
• type BatchedIngestEventHandler
• type IngestEventDataResponse
• type IngestEventQueryFilter
• type NamespacedID
• type Recalculator
  • func NewRecalculator(opts RecalculatorOptions) (*Recalculator, error)
  • func (r *Recalculator) Recalculate(ctx context.Context, ns string) error
• type RecalculatorOptions
  • func (o RecalculatorOptions) Validate() error
• type SubjectResolver
• type Worker
  • func New(opts WorkerOptions) (*Worker, error)
  • func (w *Worker) AddHandler(handler grouphandler.GroupEventHandler)
  • func (w *Worker) Close() error
  • func (w *Worker) GetEntitlementsAffectedByMeterSubject(ctx context.Context, namespace string, meterSlugs []string, subject string) ([]NamespacedID, error)
  • func (w *Worker) Run(ctx context.Context) error
• type WorkerOptions

Constants

const (
	// DefaultIncludeDeletedDuration is the default duration for which deleted entitlements are included in recalculation.
	// This ensures that the recent deleted snapshot events are also resent.
	DefaultIncludeDeletedDuration = 24 * time.Hour
)

Types

type BalanceWorkerRepository

type BalanceWorkerRepository interface {
	ListAffectedEntitlements(ctx context.Context, filterPairs []IngestEventQueryFilter) ([]IngestEventDataResponse, error)
}

type BatchedIngestEventHandler

type BatchedIngestEventHandler = func(ctx context.Context, event ingestevents.EventBatchedIngest) error

type IngestEventDataResponse

type IngestEventDataResponse struct {
	Namespace     string
	EntitlementID string
	SubjectKey    string
	// not all entitlements have a meter associated
	MeterSlug *string
}

type IngestEventQueryFilter

type IngestEventQueryFilter struct {
	Namespace  string
	SubjectKey string
	MeterSlugs []string
}

type NamespacedID

type NamespacedID = pkgmodels.NamespacedID

type Recalculator

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

func NewRecalculator

func NewRecalculator(opts RecalculatorOptions) (*Recalculator, error)

func (*Recalculator) Recalculate

func (r *Recalculator) Recalculate(ctx context.Context, ns string) error

type RecalculatorOptions

type RecalculatorOptions struct {
	Entitlement     *registry.Entitlement
	SubjectResolver SubjectResolver
	EventBus        eventbus.Publisher
	MetricMeter     metric.Meter
}

func (RecalculatorOptions) Validate

func (o RecalculatorOptions) Validate() error

type SubjectResolver

type SubjectResolver interface {
	GetSubjectByKey(ctx context.Context, namespace, key string) (models.Subject, error)
}

type Worker

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

func New

func New(opts WorkerOptions) (*Worker, error)

func (*Worker) AddHandler

func (w *Worker) AddHandler(handler grouphandler.GroupEventHandler)

AddHandler adds an additional handler to the list of batched ingest event handlers. Handlers are called in the order they are added and run after the riginal balance worker handler. In the case of any handler returning an error, the event will be retried so it is important that all handlers are idempotent.

func (*Worker) Close

func (w *Worker) Close() error

func (*Worker) GetEntitlementsAffectedByMeterSubject

func (w *Worker) GetEntitlementsAffectedByMeterSubject(ctx context.Context, namespace string, meterSlugs []string, subject string) ([]NamespacedID, error)

func (*Worker) Run

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

type WorkerOptions

type WorkerOptions struct {
	SystemEventsTopic string
	IngestEventsTopic string

	Router   router.Options
	EventBus eventbus.Publisher

	Entitlement *registry.Entitlement
	Repo        BalanceWorkerRepository
	// External connectors
	SubjectResolver SubjectResolver

	Logger *slog.Logger
}