README
¶
Events and delivery providers
kit/events owns transactional publication, SQL outbox relay, tenant-scoped
handling claims and terminal records. A broker acceptance alone does not prove
that a handler committed. The existing Publish, Consume, Relay, Purge,
Subscription and Handler APIs retain those responsibilities.
For an application that owns its own persistence, import the parts directly:
Import beneath github.com/septagon-oss/platformkit/ |
Responsibility |
|---|---|
kit/events/transport |
Event, Transport, Sink, ValidName; only standard library and UUID dependencies |
kit/events/providers/memory |
New() delivers within one process and waits for handling or terminal recording |
kit/events/providers/nats |
JetStream(url, options...) and Connect(config.NATS) select the existing NATS transport |
events.Event, events.Transport and events.Sink are aliases; ValidName
forwards to the shared grammar. events.Memory() is gone: call memory.New(),
so the SQL outbox package imports none of its own providers and the package gate
holds it there. Existing outbox consumers can migrate those imports independently. The retry
ladder, handler-attempt cap and seven-day retention have one internal owner;
moving packages does not change stored subjects, stream names or durables.
The NATS constructors have moved out of the SQL package. Replace
events.JetStream(...) with nats.JetStream(...) and
events.ConnectJetStream(settings) with nats.Connect(settings), importing
kit/events/providers/nats under an unambiguous alias when also using the SDK.
The root constructors are removed so SQL outbox consumers no longer import the
NATS SDK. The reference application's transport selection uses the new owner.
Removing these constructors breaks stable v1 source APIs. This change belongs to
the planned /v2 release in RELEASE, and must not ship as a
compatible v1 minor or patch release.
kit/app imports neither provider: the composing application passes both
constructors as app.Transports{Memory: memory.New, JetStream: nats.Connect}
and the kernel selects one by nats.transport and the role, refusing at New
when the selected name has no constructor.
Connect validates the existing config.NATS settings and then connects;
JetStream accepts official NATS options. Both can create the existing
PLATFORMKIT stream on first connection. Subscribe reconciles stream settings
and durable consumers. These are broker operations, not read-only readiness
checks. The caller drains work and closes the returned io.Closer connection.
No new configuration namespace, broker or lifecycle service is introduced.
Delivery is at least once. Independent sinks must provide their own durable idempotency and tenant checks; the transport cannot supply database isolation. Consumer reconciliation may replay events when an incompatible durable is recreated. Memory has no restart persistence and does not coordinate duplicate durables across processes. When using the SQL outbox, handling and terminal claims commit atomically, and unfinished memory deliveries leave rows pending. External effects still require provider idempotency.
Run go test -race ./kit/events/transport ./kit/events/providers/... for portable
envelope, memory delivery and local NATS TLS/credential checks. The SQL/broker
recovery tests remain in kit/events; run them with the development PostgreSQL
and NATS settings described in Contributing.
Reserve the broker for that suite: its stream-drift test changes and reconciles
shared stream settings. These checks do not qualify a deployed broker or a
downstream application's handler idempotency.
Documentation
¶
Overview ¶
Package events is the one door a module's events leave through.
A module writes an event with Publish, inside the same transaction as the state change that caused it, so the row and the change commit together or not at all: there is no window in which the state moved and the event was lost. The relay in the worker role reads those rows and hands them to a transport — in-process for a single-process run, JetStream for a fleet.
Delivery is at-least-once, and Consume is what turns that into exactly-once handling: it claims each (event, subscription) pair in platformkit_handled inside the handler's own transaction, so a redelivery of work already done finds the claim taken and skips the handler.
This is also the job queue: durable, retried, transactional background work is what an outbox is, and asking for it twice buys nothing. Periodic work is kit/jobs. Both arguments are docs/adr/0004.
Index ¶
- func Consume(ctx context.Context, conn *db.Conn, t Transport, subs []Subscription) error
- func Publish(ctx context.Context, tx db.Tx[db.Tenant], name string, payload any) error
- func PublishFor(ctx context.Context, tx db.Tx[db.System], tenantID uuid.UUID, name string, ...) error
- func Purge(ctx context.Context, conn *db.Conn) error
- func Relay(ctx context.Context, conn *db.Conn, t Transport) error
- func ValidName(name string) bool
- type Event
- type Handler
- type Sink
- type Subscription
- type Transport
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Consume ¶
Consume subscribes every handler in subs to its event. Each delivery opens a transaction in the event's own tenant, so a handler reaches the tenant's rows the same way a request handler does and can publish events of its own into the same transaction.
Each delivery is claimed before the handler runs, so a handler sees each event once however many times the transport delivers it. See claim.
func Publish ¶
Publish writes an event into the outbox inside tx. It is not a network call and it cannot fail because a broker is down: the row commits with the state change and the relay carries it from there.
It takes a context only to read the actor off it. A db.Tx carries none — a transaction is a handle, not a scope somebody can cancel through — so the caller's own context is the one thing that knows whose request this is.
func PublishFor ¶
func PublishFor(ctx context.Context, tx db.Tx[db.System], tenantID uuid.UUID, name string, payload any) error
PublishFor writes an event from a cross-tenant transaction, naming the tenant it belongs to.
It exists for one shape of work and is the only place in the program where a tenant is an argument to an event. The control plane creates a tenant, and the event that says so has to be written in the transaction that created it — a transaction that belongs to no tenant, about a tenant that did not exist a statement earlier. Publish cannot express that, and publishing afterwards in a second transaction would be an event that can be lost while its cause is kept, which is the exact failure the outbox exists to remove.
A caller needs a db.Tx[db.System] to reach it, so the audience is the modules that already hold the capability. See docs/adr/0006.
func Purge ¶
Purge removes published outbox history after a week. Unpublished rows remain recoverable regardless of age. An old handling claim remains while either its outbox row or terminal failure record exists: losing that claim would replay completed work when the pending row is relayed. Dead letters and their claims require explicit operator review; they are never automatically purged. The database clock supplies the cutoff for all workers.
func Relay ¶
Relay moves every unpublished row to the transport, a batch at a time, and returns when the queue is empty or ctx is done. kit/jobs calls it every second in the worker role.
It takes no lock, and that is deliberate: FOR UPDATE SKIP LOCKED is the concurrency control, so several workers relaying at once each take rows nobody else holds and none of them waits. An advisory lock around it would add nothing and would take something away — a transport that blocks inside one relay pass would hold that lock, and every other periodic job on every replica would stop behind it.
One transaction per batch rather than one for the whole drain, so the row locks are held for a batch and not for a backlog. The caller bounds the whole thing with a deadline; a pass that runs out of time leaves its rows unstamped and the next tick takes them.
Types ¶
type Handler ¶
Handler is what a module does with an event. It runs inside a transaction scoped to the event's tenant, so anything it writes commits with the acknowledgement of the event and rolls back with a redelivery.
type Subscription ¶
Subscription is one module's interest in one event. A module lists its subscriptions in its manifest; kit/app refuses to start when one names an event no module publishes.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
internal
|
|
|
delivery
Package delivery owns the retry ladder and retention shared by event adapters.
|
Package delivery owns the retry ladder and retention shared by event adapters. |
|
providers
|
|
|
memory
Package memory delivers committed events inside one process.
|
Package memory delivers committed events inside one process. |
|
nats
Package nats provides the existing JetStream event transport.
|
Package nats provides the existing JetStream event transport. |
|
Package transport describes committed event delivery without a database or broker.
|
Package transport describes committed event delivery without a database or broker. |