Documentation
¶
Index ¶
- type OutboxBatch
- type Repository
- func (r *Repository) AckOutbox(ctx context.Context, batch uuid.UUID) error
- func (r *Repository) BackfillOutbox(ctx context.Context, chEmpty func(context.Context) (bool, error)) (int64, error)
- func (r *Repository) BackfillPending(ctx context.Context) (bool, error)
- func (r *Repository) ClaimOutbox(ctx context.Context, owner uuid.UUID, limit int, lease time.Duration) (OutboxBatch, []*domain.UsageEvent, error)
- func (r *Repository) InsertBatch(ctx context.Context, events []*domain.UsageEvent) (created map[uuid.UUID]struct{}, err error)
- func (r *Repository) InsertOne(ctx context.Context, event *domain.UsageEvent) (bool, error)
- func (r *Repository) OutboxBacklog(ctx context.Context) (int64, time.Duration, error)
- func (r *Repository) Ping(ctx context.Context) error
- func (r *Repository) QuerySummary(ctx context.Context, params domain.SummaryParams) ([]domain.SummaryRow, bool, error)
- func (r *Repository) QueryUsageByMeta(ctx context.Context, params domain.MetaUsageParams) ([]domain.MetaUsageRow, bool, error)
- func (r *Repository) RenewOutbox(ctx context.Context, owner, batch uuid.UUID, lease time.Duration) (bool, error)
- func (r *Repository) RetryOutboxLater(ctx context.Context, owner, batch uuid.UUID, delay time.Duration) (bool, error)
- func (r *Repository) WithOutbox() *Repository
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type OutboxBatch ¶
OutboxBatch is a claimed batch: its ID is ClickHouse's deduplication token, Attempts counts ClickHouse's refusals of it so far.
type Repository ¶
type Repository struct {
// contains filtered or unexported fields
}
Repository handles database operations for usage events
func New ¶
func New(pool *pgxpool.Pool) *Repository
New creates a new repository with the given connection pool
func (*Repository) BackfillOutbox ¶
func (r *Repository) BackfillOutbox(ctx context.Context, chEmpty func(context.Context) (bool, error)) (int64, error)
BackfillOutbox queues the stored events ClickHouse keeps (its 90-day TTL), once per database, when chEmpty says ClickHouse has none: history recorded before ClickHouse was turned on would otherwise be missing from its reports.
func (*Repository) BackfillPending ¶
func (r *Repository) BackfillPending(ctx context.Context) (bool, error)
BackfillPending reports whether history queued by BackfillOutbox still waits for ClickHouse; until it is copied, reports must come from Postgres.
func (*Repository) ClaimOutbox ¶
func (r *Repository) ClaimOutbox(ctx context.Context, owner uuid.UUID, limit int, lease time.Duration) (OutboxBatch, []*domain.UsageEvent, error)
ClaimOutbox leases the next batch for ClickHouse. A batch whose lease ended (its replica died, or a refusal postponed it) comes first and keeps its ID for good, so ClickHouse drops a repeat of an insert it already stored.
func (*Repository) InsertBatch ¶
func (r *Repository) InsertBatch(ctx context.Context, events []*domain.UsageEvent) (created map[uuid.UUID]struct{}, err error)
InsertBatch inserts multiple usage events efficiently using COPY protocol Returns the count of newly created events and duplicate events It returns the IDs actually inserted: duplicates of stored events are not, so downstream sinks fed from them see each event once, however often it is sent.
func (*Repository) InsertOne ¶
func (r *Repository) InsertOne(ctx context.Context, event *domain.UsageEvent) (bool, error)
InsertOne inserts a single usage event with idempotency (ON CONFLICT DO NOTHING)
func (*Repository) OutboxBacklog ¶
OutboxBacklog reports how many events wait for ClickHouse, counted up to a million so a large backlog stays cheap to sample, and the age of the oldest.
func (*Repository) Ping ¶
func (r *Repository) Ping(ctx context.Context) error
Ping checks database connectivity
func (*Repository) QuerySummary ¶
func (r *Repository) QuerySummary(ctx context.Context, params domain.SummaryParams) ([]domain.SummaryRow, bool, error)
QuerySummary returns aggregated usage statistics
func (*Repository) QueryUsageByMeta ¶
func (r *Repository) QueryUsageByMeta(ctx context.Context, params domain.MetaUsageParams) ([]domain.MetaUsageRow, bool, error)
QueryUsageByMeta aggregates usage over events carrying params.MetaKey = params.MetaValue, split by (group meta value, vendor, model).
func (*Repository) RenewOutbox ¶
func (r *Repository) RenewOutbox(ctx context.Context, owner, batch uuid.UUID, lease time.Duration) (bool, error)
RenewOutbox extends the lease while ClickHouse is retried. False: the batch is gone or was taken over, and the caller must let it go.
func (*Repository) RetryOutboxLater ¶
func (r *Repository) RetryOutboxLater(ctx context.Context, owner, batch uuid.UUID, delay time.Duration) (bool, error)
RetryOutboxLater counts a refusal and makes the batch claimable again after delay, so a refused batch waits while the events behind it flow. False: the batch was taken over meanwhile and is not the caller's to postpone.
func (*Repository) WithOutbox ¶
func (r *Repository) WithOutbox() *Repository
WithOutbox makes every new event also land in clickhouse_outbox, in the same statement, for the ClickHouse replicator to copy.
Source Files