repository

func (*Repository) AckOutbox ¶

func (r *Repository) AckOutbox(ctx context.Context, batch uuid.UUID) error

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 ¶

func (r *Repository) OutboxBacklog(ctx context.Context) (int64, time.Duration, error)

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.