Reliability & governance helpers

Beyond produce/consume, the Go core ships optional, stdlib-only subpackages that mechanise the reliability and governance contracts the wire spec defines. None touch the frozen envelope (schema_version: 1); each is a tooling layer over the codec. They are the Go face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

idempotency.Wrap(store, handler) makes a handler run at most once per meta.id, even on at-least-once redelivery.

import "github.com/babelqueue/babelqueue-go/idempotency"

store := idempotency.NewInMemoryStore() // tests / single process
app.Handle("urn:babel:orders:created", idempotency.Wrap(store, handler))

For a fleet, two persistent stores ship as separate submodules (so the core stays driver-free) — both also expose an atomic Claim(ctx, id) (bool, error) to close the in-flight race:

import (
	idempgres "github.com/babelqueue/babelqueue-go/idempotency-postgres"
	idempredis "github.com/babelqueue/babelqueue-go/idempotency-redis"
)

pg, _ := idempgres.New(ctx, "postgres://…", idempgres.WithTTL(24*time.Hour))
_ = pg.Migrate(ctx) // CREATE TABLE IF NOT EXISTS
// or: rd, _ := idempredis.New("redis://…", idempredis.WithPrefix("bq:idemp:"))

See Idempotency and the store deep-dive.

Transactional outbox

The …/outbox subpackage removes the producer dual write: persist the encoded envelope in the same DB transaction as your business row, then a relay publishes the durable rows verbatim.

import "github.com/babelqueue/babelqueue-go/outbox"

store := outbox.NewInMemoryStore()  // production: a DB-backed outbox.Store
box := outbox.New(store)

// inside YOUR DB transaction, beside the business write — no commit of its own:
env, _ := babelqueue.Make("urn:babel:orders:created", data, babelqueue.WithQueue("orders"))
id, _ := box.Write(env)             // encodes via the frozen codec, calls Store.Save

// later, a relay drains the durable rows to the broker:
relay := outbox.NewRelay(transport, store, outbox.Options{})
res, _ := relay.Drain(ctx, 0)       // res.Published / res.Failed

outbox.Store is the four-method contract (Save / FetchUnpublished / MarkPublished / MarkFailed); the transaction boundary is yours. See Transactional Outbox.

DLQ redrive & replay-bypass

babelqueue.Redrive(ctx, transport, dlq, RedriveOptions{…}) moves dead-lettered messages back onto a queue — dead_letter removed, attempts reset to 0, everything else preserved — with DryRun, Select, ToQueue (sandbox) and Bypass. The replay guard skips effects that already fired:

res, _ := babelqueue.Redrive(ctx, transport, "orders.dlq", babelqueue.RedriveOptions{
	ToQueue: "orders.sandbox",
	Bypass:  true, // stamp bq-replay-bypass (header-carrying transport)
})

app.Handle("urn:babel:orders:created", func(ctx context.Context, env babelqueue.Envelope) error {
	saveOrder(env) // idempotent core — always runs
	return babelqueue.BypassExternalEffects(ctx, func() error {
		return sendEmail(env) // external effect — skipped on replay
	})
})

See DLQ redrive & replay-bypass.

GDPR field encryption

The …/gdpr subpackage encrypts only the data leaves a schema marks x-gdpr-sensitive, in place. gdpr.Protect / gdpr.Unprotect are free functions over the schema’s SensitivePaths(); Cipher is caller-bound and AESGCMCipher is the stdlib reference. Validate cleartext — protect after validation on produce, unprotect before validation on consume.

import "github.com/babelqueue/babelqueue-go/gdpr"

cipher, _ := gdpr.NewAESGCMCipher(key) // or bind a KMS to gdpr.Cipher
sch, _ := schema.Load("urn:babel:orders:created")

_ = gdpr.Protect(data, sch, cipher)    // producer: encrypt marked leaves
if err := gdpr.Unprotect(data, sch, cipher); errors.Is(err, gdpr.ErrDecrypt) {
	// wrong key / tampered → retry / dead-letter
}

See GDPR field encryption.

OpenTelemetry (traceparent)

The …/otel submodule emits publish <urn> / process <urn> spans and, on a header-carrying transport, injects/extracts the W3C traceparent so a consumer span is a true child of the producer span — degrading to v0.1 trace_id correlation otherwise. All in-tree transports (in-memory, Redis, AMQP, SQS) carry it. otel.WrapHandler / otel.Publish are the entry points. See Observability.

Per-URN schema validation

schema.Check(provider, urn, data) (producer guard) / schema.Wrap(provider, handler) (consumer safety net) validate a message’s data against the JSON Schema registered for its URN, bridged from a babelqueue-registry manifest via schema.NewDirProvider. See Per-URN schema validation.