ush-common

USH Common

Shared utilities and libraries for Ultimate Setup Hub microservices.

Overview

This repository contains common packages used across multiple Ultimate Setup Hub services:

• data/ - Access Object pattern for unified data access (DB, cache, API clients)
• httpintercept/ - HTTP interceptor utilities
• validators/ - Custom validation utilities
• criteria/ - Query criteria utilities
• env/ - Environment variable utilities
• auth/ - User token identity, HS256 validation and context accessors
• http/middleware/ - Shared gin middleware
• http/route/ - USH route policies and Traefik route export
• messaging/ - Transactional outbox and event dispatcher
• quota/ - Total-quota enforcement against the subscriptions service's limits

Installation

go get gitlab.com/bitval/ush-common

Usage

import (
    "gitlab.com/bitval/ush-common/data"
    "gitlab.com/bitval/ush-common/validators"
)

Part of Ultimate Setup Hub

This library is part of the Ultimate Setup Hub platform - a microservices-based system for managing car setups and race strategies for racing simulators.

Related Services

• go-ush-content-api - Static resources service
• go-ush-strategy-api - Race strategy calculation
• setups-api - Setup management

License

MIT License - see LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

See CONTRIBUTING.md for guidelines.

Messaging

messaging/ is a transactional outbox: a service writes an event in the same database transaction as the domain change that caused it, and the dispatcher delivers that event later — to handlers inside the same process, to webhooks in other services, or to both.

Event path

Producing an event is unchanged by delivery: it is one row, written in the domain transaction, optionally followed by a NOTIFY.

producer → outbox row, in the domain transaction (+ optional NOTIFY)

dispatcher wakes (NOTIFY or poll)
  ├─ expand:  for each registered subscriber, find the events it subscribes to
  │           that have no delivery row for it yet → insert the rows
  └─ deliver: claim due delivery rows → hand each to its handler → update the row

The event row records what happened. A delivery row per subscriber records who was told and how it went, so retries, errors and backoff are per subscriber and one failing consumer cannot hold up another. Expansion asks only about a dispatcher's own rows, so several services can consume the same table independently. Events whose name nothing subscribes to are never expanded: no row, no retries, nothing to clean up.

Required tables

The library ships no migrations. Each service creates these in its own schema; the names are configurable, the shape is not.

CREATE TABLE <schema>.outbox (
    id            UUID NOT NULL DEFAULT public.uuid_generate_v4() PRIMARY KEY,
    event_name    TEXT NOT NULL,
    resource_type TEXT NOT NULL,
    resource_id   TEXT NOT NULL,
    event_type    TEXT NOT NULL,
    app           TEXT,
    payload       JSONB NOT NULL,
    created_at    TIMESTAMP NOT NULL DEFAULT (now() at time zone 'utc')
);

CREATE TABLE <schema>.outbox_deliveries (
    event_id         UUID NOT NULL REFERENCES <schema>.outbox (id) ON DELETE CASCADE,
    subscriber       TEXT NOT NULL,
    processed_at     TIMESTAMP,
    retry_count      INT NOT NULL DEFAULT 0,
    last_error       TEXT,
    processing_since TIMESTAMP,
    next_attempt_at  TIMESTAMP,
    PRIMARY KEY (event_id, subscriber)
);

CREATE INDEX idx_outbox_deliveries_pending
    ON <schema>.outbox_deliveries (next_attempt_at)
    WHERE processed_at IS NULL;

The event row carries no dispatch state on purpose. A flag on the event would let the first dispatcher to see it hide it from every other one.

Producing events

Implement outbox.Event on the domain event — Name, ResourceType, ResourceID, EventType, Payload — and send it inside the transaction:

err := messagingModule.Outbox().
    NewEventMessage(domainEvent).
    WithTransaction(tx).
    Notify().
    Send(ctx)

A producer serving several applications says which one an event belongs to with .ForApp("ush"). The app is recorded on the event, travels in the envelope, and decides who the event is expanded to — see below. Events of a producer that serves one application, or none in particular, leave it unset.

Consuming in process

messaging.New().Configure(
    messaging.WithOutboxTable("shared.outbox"),
    messaging.WithDeliveriesTable("shared.outbox_deliveries"),
    messaging.WithSubscriberName("discord-notifier"),
    messaging.WithDispatcherEnabled(),
    messaging.WithDispatcherStartFrom(time.Now()),
)

dispatcher.RegisterMessageHandler("go-lmu-setup-api.setup.created", handler)

Every delivery is recorded against a subscriber: whoever it is for. A service that handles events in process is itself that subscriber, so it names itself once with WithSubscriberName rather than repeating it per handler. The name is stored in the subscriber column, so it must stay stable: changing it orphans the existing delivery rows and re-delivers every event still inside the cutoff.

WithDispatcherStartFrom bounds how far back expansion looks, so a newly registered subscriber does not replay the whole history. Rows that already exist are always delivered, whatever their age.

Delivering to other services

messaging.New().Configure(
    messaging.WithEventSigningSecret(settings.EventSigningSecret),
    messaging.WithWebhookHeader("X-Internal-API-Key", settings.InternalAPIKey),
    messaging.WithDispatcherEnabled(),
)

dispatcher.RegisterMessageWebhooks("go-auth-api.user.authenticated",
    map[string]dispatcher.Subscriber{
        "ush-users-api":      {App: "ush", URL: "http://ush-users-api:8000/api/v1/internal/events"},
        "financer-users-api": {App: "financer", URL: "http://financer-users-api:8000/api/v1/internal/events"},
    })

The map key names the subscriber, which is what its delivery rows are recorded under; it never leaves the sender. App is which application's events that subscriber wants: it receives events recorded for that app and events recorded for none, and never another app's. Leave App empty and the subscriber receives everything, which is what a single-application platform wants.

That is what keeps one auth-api serving several applications from handing one app's logins to another's profile service: the wrong app's events are never expanded, so no delivery row is ever written for them.

A service that only sends webhooks needs no WithSubscriberName: it is not the subscriber of anything. WithSubscriberName is required only to register an in process handler, and registering one without it fails.

Deliveries use a client whose timeout is the dispatcher's delivery timeout. WithWebhookHTTPClient(client) replaces it for a service that needs its own transport, timeout or instrumentation.

Reloading webhooks

A service that reads its subscribers from a file can pick up a changed file without a restart. Write the registration against dispatcher.WebhookRegistrar, which the dispatcher satisfies, and run that same code at startup and on reload:

func (s Subscriptions) Register(registrar dispatcher.WebhookRegistrar) error {
    for event, subscribers := range s.byEvent() {
        err := registrar.RegisterMessageWebhooks(event, subscribers)
        if err != nil {
            return fmt.Errorf("register webhooks for event '%s': %w", event, err)
        }
    }

    return nil
}

// at startup
err := subscriptions.Register(d)

// when the file changes
err = d.ReloadMessageWebhooks(reloaded.Register)

ReloadMessageWebhooks hands the function a registrar of its own, and once the function is done replaces every registered webhook with what it registered, all at once. The service never needs to know what was registered before:

• a subscriber no longer registered stops being expanded and claimed, and its delivery rows stay as they are, to resume if it comes back
• a new subscriber is expanded as it would be at startup, so it receives the events still inside the WithDispatcherStartFrom cutoff
• a changed URL applies from the subscriber's next delivery
• if any registration fails, even one the function went past, nothing changes

In-process handlers are never touched by a reload. Registering, of handlers and webhooks alike, is safe while the dispatcher is running.

Webhook contract

POST <url>
Content-Type: application/json
X-Bootstrapify-Messaging-Signature: sha256=<hmac of the raw body>
X-Bootstrapify-Messaging-Event-ID: <uuid>

{
  "event_id":      "…",
  "event_name":    "go-auth-api.user.authenticated",
  "app":           "ush",                      ← omitted when the event has none
  "occurred_at":   "2026-09-04T10:00:00Z",
  "resource_type": "user",
  "resource_id":   "…",
  "event_type":    "authenticated",
  "data":          { … }
}

Any 2xx marks the delivery processed; anything else is retried with backoff until the retry limit, with the response status kept in last_error. Delivery is at-least-once, so receivers must deduplicate on event_id.

Receivers verify the signature over the raw body, before decoding it:

if !webhook.Verify([]byte(secret), body, r.Header.Get(webhook.SignatureHeader)) {
    // reject
}

Retries

A failed delivery is retried MaxDeliveryRetries times (default 10). The first retry is immediate and the last waits MaxRetryBackoff (default 10s), with the attempts in between growing exponentially towards it — so a dead subscriber is exhausted in about 20 seconds rather than an hour, and a briefly unavailable one is picked up at once. Both are configurable with WithDispatcherMaxDeliveryRetries and WithDispatcherMaxRetryBackoff. A delivery that exhausts its retries stops being claimed and keeps its last_error for inspection.

Listener resilience

The dispatcher's LISTEN/NOTIFY connection is separate from polling: if it dies — the database restarts, a network partition — the dispatcher keeps delivering on PollInterval alone, and the listener retries with the same backoff curve as delivery retries: immediate on the first attempt, growing towards ListenerMaxBackoff (default 30s) over ListenerMaxRetries (default 10) failures, reconnecting and re-issuing LISTEN once it detects the connection is closed. Unlike a delivery, the listener never gives up — past ListenerMaxRetries it keeps retrying forever, just always waiting ListenerMaxBackoff between attempts — so it recovers on its own once the database comes back, no restart needed. Configurable with WithDispatcherListenerMaxRetries and WithDispatcherListenerMaxBackoff.

Retention

Nothing is deleted unless asked. WithOutboxRetention(ttl) makes the module sweep hourly, deleting events older than the ttl and their delivery rows with them; Outbox().PurgeExpired(ctx, ttl) does the same on demand, for a service that would rather run it from a cron. A sweep does not check whether an event was delivered, so keep the retention comfortably longer than the retry budget.

Routes and middleware

http/route/ has the USH route policies and the Traefik route export, as options for bootstrapify's http module that a service loads as it needs them. http/middleware/ holds the shared gin middleware the policies run, and any other middleware services share. auth/ is the user token identity both build on; it has no gin dependency, so application code can read the user without importing HTTP packages.

Both HTTP packages share their names with bootstrapify's, so services alias ush-common's:

import (
    "gitlab.com/bitval/go-bootstrapify-modules/http/middleware"
    "gitlab.com/bitval/go-bootstrapify-modules/http/route"

    ushMiddleware "gitlab.com/bitval/ush-common/http/middleware"
    ushRoute "gitlab.com/bitval/ush-common/http/route"
)

Policies

A service loads only what its routes use. UserPolicies fails Configure without a header or a valid user token key and InternalKeyPolicy without a header or with an internal API key shorter than 32 characters, so a service never needs a key it has no routes for. A route using a policy the service did not load fails DefineRoutes (unknown policy).

The Traefik middlewares above are the shared policies' defaults (ushRoute.TraefikAuthVerify, …) and must match the names in the USH Traefik configuration. Where one Traefik serves several environments, each with its own forward-auth, TraefikConfig.SharedPolicyMiddlewares (or the TRAEFIK_MIDDLEWARES_ENRICHED, TRAEFIK_MIDDLEWARES_PROTECTED and TRAEFIK_MIDDLEWARES_ADMIN variables) replaces a shared policy's middlewares with the environment's own, e.g. auth-test-forward@docker. The replacement must name at least one middleware, and TraefikExporter rejects one for any other policy, as it rejects a PolicyMiddlewares entry for a shared policy.

Permission checks stay per route, with bootstrapify's middleware.Authorize(authz, permission) after the policy. The user is stored under bootstrapify's authorization key, so Authorize sees it, and is read with auth.UserFromContext; auth.SubscriptionFromContext and auth.TokenFromContext return the subscription claim and the raw token.

Failures are reported with bootstrapify's response sentinels, so the service must register bootstrapify's middleware.ErrorResponse() on the router.

Configuration

ush-common never reads the environment by itself. The helpers take plain config structs (ushRoute.Config holds a HeaderKey per key and the TraefikConfig); ushRoute.ConfigFromEnv() is a helper for the service to call from its own settings. It requires and validates nothing: a missing key fails the policy that needs it, and an empty entry point or middleware name, or one containing whitespace, fails TraefikExporter, whether it came from the environment or was set in code:

The defaults are applied by ConfigFromEnv only; the helpers use the config as given.

Usage

// settings.Configure()
RouteConfig = ushRoute.ConfigFromEnv()

// application setup
hc := httpModule.New()
err = hc.Configure(
    ushRoute.UserPolicies(settings.RouteConfig.UserToken),
    ushRoute.InternalKeyPolicy(settings.RouteConfig.InternalKey),
    ushRoute.TraefikExporter(settings.RouteConfig.Traefik),
    httpModule.WithServerPort(port),
)
hc.Router().Use(middleware.ErrorResponse())

err = hc.DefineRoutes("/api/v1", func(g *route.Group) {
    api := g.Group("", ushMiddleware.ClientKeyAuth(
        settings.RouteConfig.ClientKey.Header, settings.RouteConfig.ClientKey.Key))

    api.GET("/setups", h.List).Policy(ushRoute.PolicyEnriched)
    api.POST("/setups", h.Create).Policy(ushRoute.PolicyProtected)
    api.DELETE("/setups/:setup_id", middleware.Authorize(authz, perm.DeleteAny), h.Delete).
        Policy(ushRoute.PolicyAdmin)

    g.Internal("/internal").Policy(ushRoute.PolicyInternalKey).
        GET("/setups/:setup_id", h.Internal)
})

// handler
user, ok := auth.UserFromContext(c.Request.Context())

Policies a service needs alone are registered with bootstrapify's httpModule.WithPolicy and listed after the shared one on a route, which runs them in that order — .Policy(ushRoute.PolicyProtected, "quota"). The service maps them to Traefik middlewares in code, with TraefikConfig.PolicyMiddlewares.

Quota

quota/ enforces total quotas: a cap on how many of something a subscriber holds at once (private_setups, later strategies). The subscriptions service (auth-api) owns only the limit; the service that owns the rows counts them inside its own write transaction, so the count can never drift from the data.

The package fetches the limit, serialises concurrent writes into one pool with an advisory lock, calls the service's count, compares, and maps the outcome to a standard HTTP error. It does not decide which writes enter a pool (e.g. which visibility transitions make a setup private); that stays in the service's domain logic. It holds no database connection and does no logging: the service passes its write transaction, and the controller logs the error it gets back.

Configuration

quota.ConfigFromEnv() is called from the service's settings. It fails only on an unparseable base URL; quota.New fails with ErrInvalidConfig without a base URL or internal key, or with a timeout that is not positive (0s would disable it), so a misconfigured service fails at startup.

The internal API key variables are the ones http/route reads, so a service already sending the internal key needs nothing new for it.

Usage

// settings.Configure()
QuotaConfig, err = quota.ConfigFromEnv()

// application setup
enforcer, err := quota.New(settings.QuotaConfig)

// use case
tx, _, err := dao.GetDB().Begin(ctx)
if err != nil {
    return fmt.Errorf("begin save setup: %w", err)
}
defer tx.Rollback(ctx)
setupsRepo := factory.SetupsRepository(tx)

subscriberID := setup.UserID
qf := criteria.NewFilter("")
qf.Add("visibility", entities.SetupVisibility_Hidden, entities.SetupVisibility_Private)
if setup.TeamID != nil {
    qf.Add("team_id", setup.TeamID)
    subscriberID = teamSvc.GetTeamOwner(ctx, setup.TeamID)
} else {
    qf.Add("user_id", setup.UserID)
}

err = enforcer.CheckTotal(ctx, tx, "private_setups", subscriberID,
    func(ctx context.Context) (int, error) { return setupsRepo.Count(ctx, qf) })
if err != nil {
    return fmt.Errorf("save setup: %w", err)
}
// setupsRepo.Create(ctx, setup) on the same tx, then:
if err := tx.Commit(ctx); err != nil {
    return fmt.Errorf("commit save setup: %w", err)
}

// controller
if resp := quota.ErrorResponse(err); resp != nil {
    _ = c.Error(resp)
    return
}

subscriberID is the user whose subscription sets the limit: the user for a personal pool, the team owner for a team pool (auth-api has no notion of teams). enforcer.Limit(ctx, quotaType, subscriberID) returns the quota on its own, with Limit and IsUnlimited, e.g. for a usage endpoint, or the same *NotAllowedError / ErrResourceNotFound / ErrUnexpectedResponse / ErrUnavailable as CheckTotal.

Quota states

auth-api answers GET /internal/subscriptions/{user_id}/quotas with exactly the quotas the user's active tier defines, nothing filled in:

{"quotas":[{"quota_type":"private_setups","kind":"total","quota_limit":5,"unlimited":false}]}

kind is total for the quotas this package enforces; periodic quotas are counted by auth-api itself. A 404 is ErrResourceNotFound; a 500 (e.g. an existing user without an active subscription, a broken invariant on auth-api's side) is ErrUnexpectedResponse.

Unlimited is decided by unlimited, never by quota_limit == 0. Decoding is strict: an entry without quota_limit or unlimited, or a response without quotas, is ErrUnexpectedResponse, because a missing number would otherwise read as 0, which means unlimited.

Locking

CheckTotal first fetches the limit, so the lock is never held across an HTTP call; an unlimited quota returns nil and a quota the tier is not allowed returns *NotAllowedError, both without locking or counting. It then takes pg_advisory_xact_lock on quota:<quotaType>:<subscriberID> on tx, counts, and returns *LimitReachedError when used >= limit.

• Saves into one pool serialise: team saves by different members resolve to the same owner and queue; personal saves of different users, and different quota types of one subscriber, do not block each other.
• The lock is held until tx commits or rolls back, so the count and the write must run on tx: build the repository on it, and commit after the write. Outside a transaction the lock would be released as soon as the statement returns.
• The count must run in READ COMMITTED (the pgx default). Under REPEATABLE READ or SERIALIZABLE the snapshot can predate a competitor's commit, and the count would be stale.

Errors

quota.ErrorResponse(err) returns the response.Error to pass to c.Error, or nil for an error that is not a quota error:

403 means the subscriptions service answered and the save is refused: the pool is full (limit_reached) or the tier has no such quota (quota_not_allowed; not limit_reached with a limit of 0, since 0 means unlimited). Both are upsell cases for the client.

The limit is never guessed; when it cannot be obtained, the save is refused and the error wraps the cause:

• 502 means the subscriptions service answered, but not with a quota: a 404 (ErrResourceNotFound), any other non-200, an undecodable body, or an entry without quota_limit or unlimited (ErrUnexpectedResponse). It is neither an authorisation failure nor something the user lacks.
• 503 means it did not answer: unreachable or timed out (ErrUnavailable).

Testing with quotatest

quotatest.NewServer(t, internalKey) stands in for the subscriptions service's quotas endpoint and is closed when the test ends:

server := quotatest.NewServer(t, settings.RouteConfig.InternalKey)
server.SetDefaultQuotas(quota.Quota{Type: "private_setups", Kind: quota.Kind_Total, Limit: 3})
server.SetQuotas(supporterID, quota.Quota{Type: "private_setups", Kind: quota.Kind_Total, Limit: 0, IsUnlimited: true})
server.SetQuotas(noPrivateSetupsID) // a tier without the quota: CheckTotal returns *NotAllowedError
// any other user gets the default quotas; without SetDefaultQuotas, 404 (ErrResourceNotFound)

enforcer, err := quota.New(server.Config())

It answers 401 without the internal key, so a passing test proves the service sends it. RespondWith(status) forces every response to that status with an empty body, RespondWithBody(body) forces a 200 with a raw body (e.g. an entry without quota_limit), Close() makes it unreachable, and Requests() counts the requests it received.

A suite that bootstraps once (e.g. in TestMain) shares one server instead: quotatest.Start(internalKey) starts it without a testing.TB, and the suite closes it in its teardown. RespondUnreachable() drops every connection without a response (ErrUnavailable), which, unlike Close(), can be undone. Reset() clears the per-user quotas and every Respond* mode and keeps the default quotas, so each test registers it with t.Cleanup(server.Reset).