Documentation
¶
Overview ¶
Package eventstore captures projected events in persistent SQL state and delivers each configured destination independently with leases and retries. Webhook delivery is at least once; consumers deduplicate using EventID. Native destinations using the same SQL pool can atomically append and acknowledge delivery through Worker.TransactionalDestinations.
Index ¶
- Variables
- func MigrationSet(d sqlcommon.Dialect) (sqlcommon.MigrationSet, error)
- type CaptureHealth
- type Delivery
- type DeliveryView
- type Publisher
- type Sender
- type Status
- type Store
- func (s *Store) Capture(ctx context.Context, rec eventsink.Record, destinations []string) error
- func (s *Store) Claim(ctx context.Context, lease time.Duration) (Delivery, error)
- func (s *Store) Finish(ctx context.Context, delivery Delivery, code string, retryAfter time.Duration) error
- func (s *Store) List(ctx context.Context, state, afterEvent, afterDestination string, limit int) ([]DeliveryView, error)
- func (s *Store) Record(ctx context.Context, id string) (eventsink.Record, error)
- func (s *Store) Records(ctx context.Context, kind, after string, limit int) ([]eventsink.Record, error)
- func (s *Store) Retry(ctx context.Context, eventID, destination string) error
- func (s *Store) Run(ctx context.Context, fn func(context.Context) error) error
- func (s *Store) Status(ctx context.Context) (Status, error)
- type Worker
Constants ¶
This section is empty.
Variables ¶
var ( ErrInvalid = errors.New("eventstore: invalid argument") ErrLease = errors.New("eventstore: delivery lease no longer held") ErrEmpty = errors.New("eventstore: no delivery ready") ErrNotFound = errors.New("eventstore: event not found") )
var ErrCapture = event.ErrCapture
ErrCapture requires the associated local mutation to fail and be retried.
Functions ¶
func MigrationSet ¶
func MigrationSet(d sqlcommon.Dialect) (sqlcommon.MigrationSet, error)
MigrationSet exposes the schema to coordinated backup and restore tools.
Types ¶
type CaptureHealth ¶
type CaptureHealth struct {
LastSuccess time.Time `json:"last_success,omitempty"`
LastFailure time.Time `json:"last_failure,omitempty"`
FailedDenials uint64 `json:"failed_denials"`
}
CaptureHealth exposes recording failure without disclosing event contents.
type Delivery ¶
type Delivery struct {
Record eventsink.Record `json:"record"`
Destination string `json:"destination"`
Token string `json:"-"`
Attempts int `json:"attempts"`
}
Delivery contains one leased destination. Token fences late worker results.
type DeliveryView ¶
type DeliveryView struct {
EventID string `json:"event_id"`
Destination string `json:"destination"`
State string `json:"state"`
Attempts int `json:"attempts"`
NextAttempt time.Time `json:"next_attempt"`
LastCode string `json:"last_code,omitempty"`
}
DeliveryView omits lease credentials and event payloads from operator status.
type Publisher ¶
type Publisher struct {
Store *Store
Registry *eventsink.Registry
Destinations []string
Subscribers event.Publisher
Report func(error)
// contains filtered or unexported fields
}
Publisher records events before notifying optional in-process subscribers. Configure it before use and keep Destinations stable for its lifetime.
func (*Publisher) Health ¶
func (p *Publisher) Health() CaptureHealth
Health returns a concurrency-safe snapshot of recording health.
type Sender ¶
Sender delivers a projected record to one destination. A receiver may have accepted an attempt even if its reply was lost; deduplicate using EventID.
type Status ¶
type Status struct {
Records int64 `json:"records"`
Pending int64 `json:"pending"`
Blocked int64 `json:"blocked"`
Delivered int64 `json:"delivered"`
Retries int64 `json:"retries"`
OldestPending time.Time `json:"oldest_pending,omitempty"`
}
Status describes all retained destination records. OldestPending includes blocked deliveries because they still require operator action.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store owns event records and destination acknowledgements, not its SQL pool.
func (*Store) Capture ¶
Capture inserts only the projected record and snapshots its destinations. An existing event ID is an error; delivery retries read the original record.
func (*Store) Claim ¶
Claim leases one ready destination. Expired claims can be reclaimed after a process crash. No network request runs while this transaction is open.
func (*Store) Finish ¶
func (s *Store) Finish( ctx context.Context, delivery Delivery, code string, retryAfter time.Duration, ) error
Finish acknowledges, blocks, or schedules another attempt using the lease token. Codes must be fixed local categories, never remote response text.
func (*Store) List ¶
func (s *Store) List(ctx context.Context, state, afterEvent, afterDestination string, limit int) ([]DeliveryView, error)
List pages destinations by event ID then destination, using the last row as the next cursor. Empty state includes every delivery state.
func (*Store) Record ¶
Record returns the original allowlisted projection, including occurrences that have no external delivery destination.
func (*Store) Records ¶
func (s *Store) Records(ctx context.Context, kind, after string, limit int) ([]eventsink.Record, error)
Records pages captured occurrences by their stable event ID. Delivery state is available separately through List, so local-only events remain visible.
func (*Store) Retry ¶
Retry releases a blocked or waiting delivery for a new attempt. Active leases and completed deliveries cannot be reset by this operation.
type Worker ¶
type Worker struct {
Store *Store
Destinations map[string]Sender
// TransactionalDestinations write only to Store's SQL pool through the
// supplied context. Their write and delivery acknowledgement commit together.
// Never register a network sender or a store backed by another pool here.
TransactionalDestinations map[string]Sender
Timeout time.Duration
PollInterval time.Duration
}
Worker leases records independently across replicas. Destination identifiers must identify immutable receiver configurations, including endpoint changes.
Source Files