Documentation
¶
Overview ¶
Package clio integrates a clio event store as a server-registered Atlas connector: a BPMN clio "write-events" connector task appends an event to a configured clio instance through the job path (ADR-0036), mirroring how the dmn package delegates a decision to temis (ADR-0014). The integration inherits the job protocol's durability and non-blocking properties (ADR-0007):
- A connector task creates a job carrying the reserved compiler.ClioWriteJobType. The processor never performs the outbound call itself, so it stays allocation-free (invariant I1) and free of any HTTP dependency.
- The in-process Handler — a job worker — pulls those jobs, appends the event to clio off the processor goroutine and after fsync (invariant I2, never inside applyToState / I4), and completes the job, which drives the token onward.
- The clio endpoint and credentials live in a server-side Registry keyed by connector name, so a model refers to a connector by name only and never carries a URL or secret (ADR-0036).
Delivery is at-least-once (a crash between "clio accepted" and "job completed" replays the write); every event carries the job key as an idempotency key so clio de-duplicates a replayed write rather than doubling the event.
Index ¶
- Constants
- func Handler(store *state.Store, lookup ProcessLookup, reg *Registry) job.Handler
- func MintKey(ctx context.Context, endpoint, adminToken string, req KeyRequest) (string, error)
- func QueryHandler(store *state.Store, lookup ProcessLookup, reg *Registry) job.OutputHandler
- func ReadHandler(store *state.Store, lookup ProcessLookup, reg *Registry) job.OutputHandler
- type Client
- type Connector
- type Event
- type HTTPClient
- func (c *HTTPClient) GetState(ctx context.Context, subject, reduceSpec string) (map[string]any, error)
- func (c *HTTPClient) Query(ctx context.Context, subject, where string) (any, error)
- func (c *HTTPClient) ReadEvents(ctx context.Context, r ReadEventsRequest) ([]InboundEvent, error)
- func (c *HTTPClient) WriteEvent(ctx context.Context, e Event) error
- type InboundEvent
- type KeyRequest
- type ProcessLookup
- type ReadEventsRequest
- type Registry
Constants ¶
const DefaultEventSource = "atlas"
DefaultEventSource is the CloudEvents `source` an outbound write carries when a task does not set one. clio rejects a write with an empty source, so this keeps a model that only names a subject and type working.
Variables ¶
This section is empty.
Functions ¶
func Handler ¶
Handler builds a job handler that performs a clio "write-events" connector task. Register it with a job.Runner for the reserved ClioWriteJobType index; the runner then pulls activatable clio jobs, and for each the handler resolves the connector task's connector/subject/event-type from the compiled process, resolves the connector's client from reg, and appends an event carrying the instance's variables as its body — keyed by the job key so an at-least-once retry de-duplicates (ADR-0036). Returning an error leaves the job pending, exactly as for any worker; the runner completes it only on success.
func MintKey ¶
MintKey creates a new clio API key and returns its full, once-shown secret (clio's "kid.secret"). It is a standalone call — not a Client method — because it authenticates with a clio **admin** token, distinct from a connector's read token, and exists only to provision a connector's credential (ADR-0092) without an operator copy-pasting one. The caller must never persist adminToken; only the returned scoped key is stored (sealed in the vault). Delivery is not idempotent, so a caller should mint once per provisioning action.
func QueryHandler ¶
func QueryHandler(store *state.Store, lookup ProcessLookup, reg *Registry) job.OutputHandler
QueryHandler builds a job handler for a clio "query" connector task: it reads projected state (get_state) or runs a stored query (run_query) on the task's connector and writes the result back into the task's result variable. Register it with a job.Runner for the reserved ClioQueryJobType index via HandleWithOutput, like the REST worker (ADR-0036/0067): when the task carries a query the handler runs it, otherwise it reads get_state for the task's subject (with the optional reduce spec). Returning an error leaves the job pending (retry, then an incident), exactly as for the write handler.
func ReadHandler ¶
func ReadHandler(store *state.Store, lookup ProcessLookup, reg *Registry) job.OutputHandler
ReadHandler builds a job handler for a clio "read" connector task: it reads the task's subject events (up to the task's limit) from the connector and writes them back into the task's result variable as a JSON array. Register it for the reserved ClioReadJobType index via HandleWithOutput (ADR-0036).
Types ¶
type Client ¶
type Client interface {
WriteEvent(ctx context.Context, e Event) error
GetState(ctx context.Context, subject, reduceSpec string) (map[string]any, error)
Query(ctx context.Context, subject, where string) (any, error)
ReadEvents(ctx context.Context, req ReadEventsRequest) ([]InboundEvent, error)
}
Client talks to one clio instance. It is an interface so the worker and the inbound bridge are testable without a live clio and so a connector name binds to exactly one endpoint. WriteEvent appends a domain event; GetState reads a projection; Query runs a stored query; ReadEvents reads a subject's events.
type Connector ¶
Connector is the server-side configuration of one clio connector: the base endpoint of the clio instance and an optional bearer token for it.
type Event ¶
type Event struct {
Source string // CloudEvents source (clio requires it); defaults to DefaultEventSource when empty
Subject string
Type string
Data map[string]any
IdempotencyKey string
}
Event is one event a connector task appends to clio. IdempotencyKey is deterministic (the job key), so an at-least-once retry is de-duplicated by clio rather than appended twice.
type HTTPClient ¶
type HTTPClient struct {
// contains filtered or unexported fields
}
HTTPClient talks to a real clio instance over its HTTP API (clio v1). Each method targets clio's documented route: write-events, read-events, run-query (all POST with a JSON body) and state (GET with the subject in the path). clio subjects are absolute ("/orders/42"); a leading slash is added if a task omits it. Reads stream NDJSON, one event JSON per line, oldest first.
func NewHTTPClient ¶
func NewHTTPClient(conn Connector) *HTTPClient
NewHTTPClient builds a clio HTTP client for a configured connector.
func (*HTTPClient) GetState ¶
func (c *HTTPClient) GetState(ctx context.Context, subject, reduceSpec string) (map[string]any, error)
GetState reads a subject's folded state and returns its state object. clio's state route is GET {Endpoint}/api/v1/state/<subject> with the subject in the path; the effective reduce spec is chosen server-side by registered prefix (ADR-0041), so reduceSpec is accepted for interface symmetry but not sent. The response is a state envelope; we return its `state` object.
func (*HTTPClient) Query ¶
Query runs a clio filter query over a subject and returns the matching events. Wire format: POST {Endpoint}/api/v1/run-query with {subject, where}, where `where` is clio's CEL predicate ("" = every event in the scope). The response streams NDJSON (one event per line); we collect the events into a slice so the result canonicalizes into a process variable like any other JSON array.
func (*HTTPClient) ReadEvents ¶
func (c *HTTPClient) ReadEvents(ctx context.Context, r ReadEventsRequest) ([]InboundEvent, error)
ReadEvents reads a subject's events oldest-first. Wire format: POST {Endpoint}/api/v1/read-events with a JSON body {subject, recursive, lowerBound, types, limit}, returning NDJSON (one event JSON per line). clio's lowerBound is inclusive but AfterID is the last-consumed id, so the first line equal to AfterID is dropped, making AfterID an exclusive cursor. `recursive` includes the subject's whole subtree — the setting that lets a watch on /employees catch an event written to /employees/E-123456.
func (*HTTPClient) WriteEvent ¶
func (c *HTTPClient) WriteEvent(ctx context.Context, e Event) error
type InboundEvent ¶
type InboundEvent struct {
ID string `json:"id"`
Subject string `json:"subject"`
Type string `json:"type"`
Data map[string]any `json:"data"`
Partition int `json:"partition,omitempty"`
}
InboundEvent is one event read back from clio (a read/query result, or an event the inbound bridge consumes, ADR-0075). ID is clio's event id — a per-partition monotonic sequence rendered as a decimal string (clio's `strconv.FormatUint`), which is both the resume cursor and, parsed as a uint64, the inbound bridge's dedup sequence. clio events carry no separate `seq` field. Partition is a server-derived view attribute set only on reads (0 for a single-partition clio, the recommended production configuration).
type KeyRequest ¶
KeyRequest describes a clio API key to mint (clio's POST /api/v1/keys). Scopes are clio scope strings — e.g. "read:/employees/*" grants read on the /employees subtree (the recursive grant a subtree watch needs), "read:/employees" only the exact subject. ExpiresAt is an optional RFC3339 timestamp ("" = no expiry).
type ProcessLookup ¶
type ProcessLookup func(defKey uint64) *compiler.CompiledProcess
ProcessLookup resolves a process-definition key to its compiled process. The worker uses it to find the connector, subject, and event type a clio job belongs to, so one handler serves every deployed process.
type ReadEventsRequest ¶
type ReadEventsRequest struct {
Subject string
AfterID string
Recursive bool
Types []string
Limit int
}
ReadEventsRequest selects the events a read returns: a Subject, an optional exclusive AfterID cursor ("" reads from the start), whether to include the subject's subtree, an optional type filter, and a Limit (0 = the connector's default).
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry resolves a connector name to the Client for its clio instance. Connectors are registered at the server from configuration (endpoint plus credentials), so a model refers to a connector by name only (ADR-0036). A Registry is read-only once populated and safe for concurrent use by workers.
func (*Registry) Client ¶
Client returns the client bound to name, or nil and false if none is registered.
func (*Registry) Register ¶
Register binds a connector name to its client. Registering the same name again replaces the earlier binding (last write wins), so reconfiguration is simple. Populate the registry before the processes that use it start running.
func (*Registry) Replace ¶
Replace swaps the whole set of registered connectors at once, so a server can rebuild the registry from managed configuration after a change (ADR-0041). The caller must serialize Replace with the workers that read the registry — the Atlas server does both on its run-loop goroutine — so no lock is needed. A nil map clears the registry.
Source Files