Help us shape the future of Go! Take the 2026 Go Developer Survey

eventsequences

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 5, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package eventsequences appends registered events and reads Chronicle sequences. Event logs use the same Sequence implementation. Writes are never retried by the SDK.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func WithOrigin 

func WithOrigin(ctx context.Context, origin Origin) context.Context

WithOrigin returns a context carrying origin for immediate append notifications. A zero origin masks an inherited origin. It does not change correlation or wire metadata. A configured AppendOriginResolver can override it. Unit-of-work commits use the unit's own origin instead, bypassing that resolver.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/cratis/chronicle.go/eventsequences"
)

func main() {
	origin := eventsequences.NewOrigin()
	ctx := eventsequences.WithOrigin(context.Background(), origin)
	// Pass ctx to immediate appends; filter notifications with n.Origin == origin.
	fmt.Println("same execution:", eventsequences.OriginFrom(ctx) == origin)
	fmt.Println("different execution:", eventsequences.NewOrigin() != origin)
	fmt.Println("unattributed:", eventsequences.OriginFrom(context.Background()) == (eventsequences.Origin{}))
}

Output:
same execution: true
different execution: true
unattributed: true

Types

type AppendCallbackPanicError 

type AppendCallbackPanicError struct {
	// Value is the recovered panic value. It is borrowed and must not be mutated.
	Value any
}

AppendCallbackPanicError reports a subscriber panic without changing the append disposition. Other admitted subscribers still receive the result. The append returns this error joined with any operation error; never retry a committed or unknown append because notification delivery failed.

func (*AppendCallbackPanicError) Error 

func (e *AppendCallbackPanicError) Error() string

type AppendError 

type AppendError string

AppendError preserves a kernel error code verbatim, including future codes.

func (AppendError) Error 

func (e AppendError) Error() string

type AppendEvent 

type AppendEvent struct {
	// Source is the attempted event source.
	Source events.SourceID
	// EventType is the registered type and generation submitted for this input.
	EventType events.TypeRef
	// Route is the normalized source/stream classification submitted to the kernel.
	Route Route
	// Position is present only for confirmed persistence; zero is a valid position.
	Position *events.SequenceNumber
}

AppendEvent identifies an original input, not a durable event delivery. It contains no caller-owned event value or payload. Batch entries retain input order.

type AppendNotification 

type AppendNotification struct {
	// CorrelationID is the effective REQUEST correlation, even if the response is
	// absent or carries a different ID. Multiple executions may share this ID;
	// use Origin to distinguish them.
	CorrelationID metadata.CorrelationID
	// Origin identifies the unit of work for a commit. Immediate appends use
	// the handled resolver result (including zero), otherwise the context origin.
	// Zero means unattributed.
	Origin Origin
	// Operation identifies the store, namespace, sequence and distinct exact types.
	Operation OperationMetadata
	// Events identifies each original input and its position when known.
	Events []AppendEvent
	// Result preserves the append disposition and all diagnostics, without
	// inferring success from completion. A single append is represented as a
	// one-entry BatchResult. Its CorrelationID remains the response correlation.
	Result BatchResult
	// Err is the append operation error before notification delivery. Domain
	// rejections may have nil Err: always inspect Result.Disposition/Result.Err().
	Err error
}

AppendNotification describes one local, nonempty append attempt after its result is known. It is not an observer subscription, replay or proof of durable delivery. Each callback owns its Events and Result, including nested collections and positions. Err and any wrapped errors are borrowed and must not be mutated.

func (AppendNotification) Completion 

func (n AppendNotification) Completion() (observation.Completion, error)

Completion snapshots a confirmed append notification, including prepared/UoW batches. The caller must synchronize notification capture across concurrent calls.

type AppendOperationResult 

type AppendOperationResult struct {
	// contains filtered or unexported fields
}

AppendOperationResult adds operation metadata without changing AppendResult's public struct layout. Result retains the existing disposition/diagnostics.

func (AppendOperationResult) Completion 

func (r AppendOperationResult) Completion() (observation.Completion, error)

Completion snapshots this append's exact coordinates and generation. Known rejection yields a trivial target; unknown outcome is an error, never success.

func (AppendOperationResult) Operation 

func (r AppendOperationResult) Operation() OperationMetadata

Operation returns immutable metadata independently of the write outcome.

func (AppendOperationResult) Result 

func (r AppendOperationResult) Result() AppendResult

Result returns the append result. Diagnostics follow AppendResult's ownership contract.

func (AppendOperationResult) WaitForCompletion 

func (r AppendOperationResult) WaitForCompletion(ctx context.Context, observers *observation.Service, timeout time.Duration) (observation.CompletionResult, error)

WaitForCompletion waits on original single-append coordinates without retaining a client in AppendResult. A nil service cannot confirm committed work.

type AppendOption 

type AppendOption func(*appendConfig)

AppendOption configures one append. Scalar options are last-wins. Nil options fail. Slice and pointer inputs are copied when the option is constructed.

func WithCorrelation 

func WithCorrelation(id metadata.CorrelationID) AppendOption

WithCorrelation overrides the context correlation. A zero ID requests a new UUID.

func WithNamedTags 

func WithNamedTags(tags ...events.NamedTag) AppendOption

WithNamedTags supplies exact name/value records. Duplicate records are coalesced; different values under one name remain distinct.

func WithOccurred 

func WithOccurred(occurred time.Time) AppendOption

WithOccurred supplies occurrence time, truncated to .NET-compatible 100ns precision.

func WithRoute 

func WithRoute(route Route) AppendOption

WithRoute overrides routing; it never changes the source argument.

func WithScope 

func WithScope(scope Scope) AppendOption

WithScope overrides the default source-and-route optimistic scope.

func WithSubject 

func WithSubject(subject events.Subject) AppendOption

WithSubject overrides the source-derived compliance subject. Empty subjects and the kernel's reserved confidentiality identifiers fail with compliance.InvalidSubjectError.

func WithTags 

func WithTags(tags ...events.Tag) AppendOption

WithTags sets dynamic tags merged distinctly after static tags.

type AppendOriginResolutionError 

type AppendOriginResolutionError struct {
	// Panicked distinguishes a recovered callback panic from a returned error.
	Panicked bool
}

AppendOriginResolutionError means local attribution failed before dispatch. The result is Rejected, no RPC was made and no append notification was emitted. It deliberately retains neither the callback error nor panic payload, and does not unwrap them: callback failures can contain sensitive context values.

func (*AppendOriginResolutionError) Error 

func (e *AppendOriginResolutionError) Error() string

Error returns a fixed, payload-free diagnostic.

type AppendOriginResolver 

type AppendOriginResolver func(context.Context) (origin Origin, handled bool, err error)

AppendOriginResolver selects local attribution metadata from the actual append context. It cannot replace that context. Handled true selects the returned origin, including zero; false falls back to OriginFrom(ctx). Errors and panics fail before any RPC or append notification, without fallback. SDK-owned unit commits bypass the resolver and always use the unit's origin.

The callback is borrowed, synchronous and invoked outside SDK locks. It must support concurrent calls, honor cancellation and return promptly. It owns no transport or request lifetime. The SDK neither retains ctx nor starts a worker to enforce cancellation. Nil preserves the OriginFrom behavior.

type AppendOriginResolverProvider 

type AppendOriginResolverProvider interface {
	AppendOriginResolver() AppendOriginResolver
}

AppendOriginResolverProvider optionally supplies an immutable callback on a caller-owned connection passed to New. New reads it once; Chronicle clients supply their configured callback automatically. The provider must not perform I/O or invoke the callback while supplying it.

type AppendResult 

type AppendResult struct {
	// Disposition records whether the write committed, was rejected, or is unknown.
	Disposition Disposition
	// Position is present only for a confirmed commit; position zero is valid.
	Position *events.SequenceNumber
	// CorrelationID is the kernel's correlation for this append.
	CorrelationID metadata.CorrelationID
	// ConstraintViolations preserves all kernel constraint details.
	ConstraintViolations []constraints.Violation
	// ConcurrencyViolations preserves the single-append concurrency violation, if any.
	ConcurrencyViolations []ConcurrencyViolation
	// Errors preserves all kernel append error codes.
	Errors []AppendError
	// ConcurrencyCheckPerformed is the kernel's actual check flag, not an assumption.
	ConcurrencyCheckPerformed bool
	// Target identifies committed work for future observer completion APIs.
	Target observation.CompletionTarget
}

AppendResult owns all returned diagnostics and completion coordinates. Domain rejection returns this result with a nil operation error; call Err to promote it.

func (AppendResult) Err 

func (r AppendResult) Err() error

Err returns nil only for Committed results without diagnostics. It exposes both constraint and concurrency errors via errors.As when both occur. Unknown results expose OutcomeUnknownError and preserve diagnostic error identities.

type BatchOperationResult 

type BatchOperationResult struct {
	// contains filtered or unexported fields
}

BatchOperationResult adds operation metadata without changing BatchResult's layout.

func (BatchOperationResult) Completion 

func (r BatchOperationResult) Completion() (observation.Completion, error)

Completion snapshots the batch's exact per-input generations and positions. It uses the greatest actual position per type, not the whole-batch tail.

func (BatchOperationResult) Operation 

func (r BatchOperationResult) Operation() OperationMetadata

Operation returns immutable metadata independently of the write outcome.

func (BatchOperationResult) Result 

func (r BatchOperationResult) Result() BatchResult

Result returns the existing batch result and diagnostics.

func (BatchOperationResult) WaitForCompletion 

func (r BatchOperationResult) WaitForCompletion(ctx context.Context, observers *observation.Service, timeout time.Duration) (observation.CompletionResult, error)

WaitForCompletion waits on original batch coordinates and actual type tails.

type BatchOption 

type BatchOption func(*batchConfig)

BatchOption configures an atomic batch. Scalar options are last-wins; nil options fail. Slice/map/pointer inputs are snapshotted at option construction.

func WithBatchCorrelation 

func WithBatchCorrelation(id metadata.CorrelationID) BatchOption

WithBatchCorrelation overrides context correlation; zero requests a new UUID.

func WithBatchNamedTags 

func WithBatchNamedTags(tags ...events.NamedTag) BatchOption

WithBatchNamedTags sets named tags merged distinctly after every entry's tags.

func WithBatchTags 

func WithBatchTags(tags ...events.Tag) BatchOption

WithBatchTags sets ordinary tags merged into every entry.

func WithScopes 

func WithScopes(scopes ...LabeledScope) BatchOption

WithScopes sets explicit checks. A label matching an appended source replaces its default optimistic check; other labels add independent checks. Resolve queries precisely its Filter. Omitted sources use their first entry's route.

type BatchResult 

type BatchResult struct {
	// Disposition reports confirmed commitment, rejection or unknown outcome.
	Disposition Disposition
	// Positions corresponds one-to-one with input entries, in original order.
	Positions []events.SequenceNumber
	// CorrelationID is the kernel's correlation for this batch.
	CorrelationID metadata.CorrelationID
	// ConstraintViolations contains every rejected constraint and its details.
	ConstraintViolations []constraints.Violation
	// ConcurrencyViolations identifies rejected labels in SourceID, with their tails.
	ConcurrencyViolations []ConcurrencyViolation
	// Errors preserves every append error code, including future codes.
	Errors []AppendError
	// ConcurrencyCheckPerformed means EVERY supplied scope was checked. False is
	// expected when any scope is NoCheck or Resolve finds no matching history.
	ConcurrencyCheckPerformed bool
	// Target identifies committed events for observer completion; empty for eventless checks.
	Target observation.CompletionTarget
}

BatchResult owns an atomic append's disposition and complete diagnostics. Known rejection has no positions; transport and errors-only kernel failures return Unknown, not Rejected.

func (BatchResult) Err 

func (r BatchResult) Err() error

Err exposes the same inspectable rejection and unknown-outcome errors as AppendResult.Err. It returns nil only for a committed result without diagnostics.

type CompleteStreamError 

type CompleteStreamError int32

CompleteStreamError is a known kernel refusal. Compare with errors.Is. 

const (
	// StreamAlreadyCompleted means the stream stays closed; no new completion occurred.
	StreamAlreadyCompleted CompleteStreamError = 1
	// DefaultStreamCannotBeCompleted protects the All/Default stream.
	DefaultStreamCannotBeCompleted CompleteStreamError = 2
)

func (CompleteStreamError) Error 

func (e CompleteStreamError) Error() string

type ConcurrencyError 

type ConcurrencyError struct {
	// Violations contains the expected/actual conflicts.
	Violations []ConcurrencyViolation
}

ConcurrencyError promotes concurrency rejections for errors.As callers.

func (*ConcurrencyError) Error 

func (e *ConcurrencyError) Error() string

type ConcurrencyPolicy 

type ConcurrencyPolicy struct {
	// CheckFirstAppendIntoAScope protects an empty matching tail (default false).
	CheckFirstAppendIntoAScope bool
	// Strategy replaces automatic scope selection; nil uses optimistic resolution.
	Strategy ConcurrencyScopeStrategy
}

ConcurrencyPolicy configures default scope resolution. The zero value uses optimistic source-and-route concurrency and leaves empty tails unchecked. CheckFirstAppendIntoAScope protects empty Resolve expectations, including explicit Resolve scopes; explicit Exact, NoMatchingEvent and NoCheck win. Strategy is borrowed for the sequence's lifetime; nil selects the default.

type ConcurrencyPolicyProvider 

type ConcurrencyPolicyProvider interface {
	ConcurrencyPolicy() ConcurrencyPolicy
}

ConcurrencyPolicyProvider optionally supplies immutable policy on a low-level connection passed to New. Chronicle clients implement this automatically.

type ConcurrencyScopeStrategy 

type ConcurrencyScopeStrategy interface {
	GetScope(ctx context.Context, sequence *Sequence, filter ScopeFilter) (Scope, error)
}

ConcurrencyScopeStrategy replaces automatic source-and-route scope selection and is also used by ResolveScope for explicitly selected filters. GetScope may use sequence.Tail with the caller's context, or return a Resolve expectation to defer the one matching tail read to dispatch. Returned scopes are validated like explicit scopes. Implementations must support concurrent calls, must honor cancellation, and must not mutate the sequence or filter. Explicit WithScope/WithScopes bypass this strategy.

type ConcurrencyViolation 

type ConcurrencyViolation struct {
	// SourceID is the source or independent batch scope label named by the kernel.
	SourceID events.SourceID
	// Expected is the supplied sequence-wide expectation.
	Expected events.SequenceNumber
	// Actual is the kernel's matching tail.
	Actual events.SequenceNumber
}

ConcurrencyViolation preserves expected and actual tails, including sentinels.

type ConstraintError 

type ConstraintError struct {
	// Violations contains the rejected constraints.
	Violations []constraints.Violation
}

ConstraintError promotes constraint rejections for errors.As callers.

func (*ConstraintError) Error 

func (e *ConstraintError) Error() string

type Disposition 

type Disposition uint8

Disposition distinguishes unknown, known rejection and committed writes. 

const (
	// Unknown means no authoritative write disposition is available, including
	// errors-only kernel failures that may occur after commit; never assume success.
	Unknown Disposition = iota
	// Rejected means a known pre-storage rejection, such as constraint or
	// concurrency violations (even with errors), or an authorization/validation envelope.
	Rejected
	// Committed means the kernel confirmed persistence.
	Committed
)

type Entry 

type Entry struct {
	// Source is the nonblank event source to append to.
	Source events.SourceID
	// Event is a registered value or non-nil pointer.
	Event any
	// Route selects the append route; zero fields use Default/All/Default.
	Route Route
	// Tags merges after the event type's static tags and before batch tags.
	Tags []events.Tag
	// NamedTags preserves exact name/value pairs, including empty values.
	NamedTags []events.NamedTag
	// Occurred overrides server-assigned occurrence time when non-nil.
	Occurred *time.Time
	// Subject overrides the default compliance subject (Source).
	Subject *events.Subject
	// Causation extends the context's chain for this entry only, in input order.
	Causation []metadata.Causation
}

Entry describes one event in an ordered atomic batch. Inputs are borrowed only during AppendBatch; callers must not mutate them concurrently with the call.

type EventsWithConcurrencyScopes 

type EventsWithConcurrencyScopes struct {
	// Events are appended in order without grouping by source.
	Events []Entry
	// Scopes enroll concurrency checks in the same atomic append.
	Scopes []LabeledScope
}

EventsWithConcurrencyScopes is a reactor return value containing ordered, self-describing events and explicit concurrency checks. It is appended to the event log using AppendBatch, without grouping or reordering. An empty Events slice requires an effective protected scope. Inputs are borrowed until the invocation completes; do not mutate them concurrently.

type Expectation 

type Expectation struct {
	// contains filtered or unexported fields
}

Expectation is a tagged concurrency expectation. The zero value means Resolve, not Exact(0). Reserved numeric positions are rejected before dispatch. ResolveScope may return a resolved unchecked expectation for an empty tail; retain it unchanged to avoid a second tail read at dispatch.

func Exact 

func Exact(position events.SequenceNumber) Expectation

Exact supplies an explicit upper bound: the kernel rejects a matching tail greater than position, but accepts a lower tail or no matching history. It does not require equality. Use NoMatchingEvent to require an empty matching history.

func NoCheck 

func NoCheck() Expectation

NoCheck explicitly disables concurrency checking.

func NoMatchingEvent 

func NoMatchingEvent() Expectation

NoMatchingEvent protects the absence of any matching history.

func Resolve 

func Resolve() Expectation

Resolve asks the optimistic strategy to query the matching tail at append time.

type FromFilter 

type FromFilter struct {
	// SourceID selects a source when present; nil/empty does not narrow.
	SourceID *events.SourceID
	// EventTypes selects type IDs across generations; empty means all types.
	EventTypes []events.TypeRef
}

FromFilter includes only dimensions supported by the inclusive from-position RPC.

type History 

type History struct {
	// Events owns the ordered loaded events.
	Events []events.Appended
	// Filter is the normalized filter used for this read, preserving the exact source ID.
	Filter ScopeFilter
	// Expectation is NoMatchingEvent for empty history, otherwise Exact(last loaded).
	Expectation Expectation
}

History is one complete matching read, ordered by sequence position. It is not a reservation or a server snapshot across multiple calls. Expectation is derived ONLY from Events, never from a later tail query.

func (History) Scope 

func (h History) Scope() Scope

Scope returns a defensive copy suitable for WithScope or WithScopes. Mutating Events cannot advance this expectation to a newer tail.

Example 
package main

import (
	"fmt"

	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/eventsequences"
)

func main() {
	// ReadHistory supplies these fields from one successful read. This empty
	// fixture illustrates converting protected absence into a batch check.
	source := events.SourceID("customer-42")
	history := eventsequences.History{
		Filter:      eventsequences.ScopeFilter{SourceID: &source},
		Expectation: eventsequences.NoMatchingEvent(),
	}
	check := eventsequences.LabeledScope{Label: string(source), Scope: history.Scope()}
	fmt.Println(check.Label, check.Scope.Expectation == eventsequences.NoMatchingEvent())
}

Output:
customer-42 true

type LabeledScope 

type LabeledScope struct {
	// Label identifies the check in returned concurrency violations.
	Label string
	// Scope specifies the expectation and exact narrowing to check.
	Scope Scope
}

LabeledScope enrolls a concurrency check in the same atomic batch. Labels must be nonblank and unique. If Scope narrows SourceID, it must equal Label. Labels need not have an appended event, so independent histories can be protected.

type MutationOutcomeUnknownError 

type MutationOutcomeUnknownError struct {
	// Cause is the failure after dispatch, not proof that nothing changed.
	Cause error
}

MutationOutcomeUnknownError means a history mutation or stream completion may have been accepted despite the error. Never blindly retry: revision can add another revision, and source redaction can affect newly appended events. Unwrap preserves transport, context and kernel envelope error identities.

func (*MutationOutcomeUnknownError) Error 

func (e *MutationOutcomeUnknownError) Error() string

func (*MutationOutcomeUnknownError) Unwrap 

func (e *MutationOutcomeUnknownError) Unwrap() error

Unwrap returns the underlying failure.

type OperationMetadata 

type OperationMetadata struct {
	// contains filtered or unexported fields
}

OperationMetadata identifies attempted work, not confirmed commitment. It is available even for rejected/unknown results and contains no completion target. Its zero value describes no operation. Event types retain their generations.

func (OperationMetadata) EventTypes 

func (m OperationMetadata) EventTypes() []events.TypeRef

EventTypes returns a defensive copy of distinct exact types in input order.

func (OperationMetadata) Namespace 

func (m OperationMetadata) Namespace() metadata.Namespace

Namespace returns the attempted namespace.

func (OperationMetadata) Sequence 

func (m OperationMetadata) Sequence() events.SequenceID

Sequence returns the attempted event sequence.

func (OperationMetadata) Store 

func (m OperationMetadata) Store() metadata.StoreName

Store returns the attempted logical store.

type Origin 

type Origin struct {
	// contains filtered or unexported fields
}

Origin is an opaque, comparable identity for local append attribution. The zero value means unattributed. Copy and compare values directly; an origin is not a correlation ID, durable identity, wire value or authorization credential. Origins remain valid for the lifetime of the process, not across restarts.

func NewOrigin 

func NewOrigin() Origin

NewOrigin returns a nonzero identity never reused within this process, even after callers release it. It is safe for concurrent use. It panics if all uint64 identities have been allocated rather than reusing an identity.

func OriginFrom 

func OriginFrom(ctx context.Context) Origin

OriginFrom returns the installed origin, or zero when none is installed. It does not invoke a configured AppendOriginResolver.

type OutcomeUnknownError 

type OutcomeUnknownError struct {
	// Cause is the original failure.
	Cause error
}

OutcomeUnknownError means an append may have committed despite the operation error or an errors-only kernel result. Do not blindly retry. Unwrap preserves kernel diagnostics, transport status and context identity.

func (*OutcomeUnknownError) Error 

func (e *OutcomeUnknownError) Error() string

func (*OutcomeUnknownError) Unwrap 

func (e *OutcomeUnknownError) Unwrap() error

type PreparedBatch 

type PreparedBatch struct {
	// contains filtered or unexported fields
}

PreparedBatch is an immutable, serialized batch bound to its originating Sequence handle. Use PrepareBatch; its zero value is invalid. It owns its content, metadata and scopes, but not the connection. It is not a transaction: use transactions.Begin for once-only completion and staging ownership.

func (*PreparedBatch) GetEvents 

func (b *PreparedBatch) GetEvents() []json.RawMessage

GetEvents returns owned copies of the serialized events in enrollment order. These are JSON snapshots, not the original Go values or persisted events.

func (*PreparedBatch) Merge 

func (b *PreparedBatch) Merge(next *PreparedBatch) (*PreparedBatch, error)

Merge returns an ordered snapshot of b followed by next, without changing either input. Both must use the same Sequence handle, correlation and actor. Explicit scopes replace implicit defaults. Repeated identical scopes join; conflicting scopes fail atomically. Event-type filters compare as sets.

func (*PreparedBatch) Operation 

func (b *PreparedBatch) Operation() OperationMetadata

Operation returns attempted coordinates and exact types for an immutable prepared snapshot, without I/O. Nil/zero snapshots return zero metadata.

func (*PreparedBatch) WithDecisionGuard 

func (b *PreparedBatch) WithDecisionGuard(guard *decision.Guard) (*PreparedBatch, error)

WithDecisionGuard attaches SDK-only enrolled evidence without reconstructing or serializing staged events. Applications cannot manufacture a Guard.

type Route 

type Route struct {
	// SourceType classifies the event source.
	SourceType events.SourceType
	// StreamType classifies the stream.
	StreamType events.StreamType
	// StreamID identifies the stream independently of source ID.
	StreamID events.StreamID
}

Route selects an append route. Empty fields resolve to Default/All/Default.

type Scope 

type Scope struct {
	// Expectation defaults to Resolve.
	Expectation Expectation
	// Filter selects matching history.
	Filter ScopeFilter
}

Scope combines a concurrency expectation and history filter.

type ScopeFilter 

type ScopeFilter struct {
	// SourceID narrows to this source when present.
	SourceID *events.SourceID
	// SourceType narrows by source classification (Default is a kernel wildcard).
	SourceType *events.SourceType
	// StreamType narrows by stream classification (All is a kernel wildcard).
	StreamType *events.StreamType
	// StreamID narrows by stream identity (Default is a kernel wildcard).
	StreamID *events.StreamID
	// EventTypes narrows by type identity; generation is validated but the kernel tail query filters IDs.
	EventTypes []events.TypeRef
}

ScopeFilter describes the exact matching history. Nil dimensions do not narrow. A single Append can only bind SourceID to its own target; other sources are rejected.

type Sequence 

type Sequence struct {
	// contains filtered or unexported fields
}

Sequence is a concurrency-safe handle with immutable coordinates and local subscriptions. It must not be copied. Obtain it from EventStore; it does not own the client and cannot outlive the client's Close.

func New 

func New(store metadata.StoreName, namespace metadata.Namespace, id events.SequenceID, catalog *events.Catalog, conn grpc.ClientConnInterface) (*Sequence, error)

New creates a low-level sequence over a caller-owned connection and registered catalog. Each call creates independent, handle-local append subscriptions; this is the low-level escape hatch from EventStore's shared sequence handles. It does not ensure registration or compatibility; prefer EventStore's handles, which enforce those barriers. It snapshots optional ConcurrencyPolicyProvider and AppendOriginResolverProvider configuration once from conn, without invoking the resolver. Invalid coordinates or nil dependencies fail.

func (*Sequence) Append 

func (s *Sequence) Append(ctx context.Context, source events.SourceID, event any, options ...AppendOption) (result AppendResult, err error)

Append serializes and persists one registered event. Default concurrency queries the source/route tail; an empty tail is unchecked unless the client opts into first-append protection. Use NoMatchingEvent to protect first append, or Exact to protect an earlier read. Known domain rejections are results, not operation errors. Any post-dispatch failure is conservatively OutcomeUnknownError. The SDK never retries writes. Caller-owned event data must not be mutated concurrently with this call.

func (*Sequence) AppendBatch 

func (s *Sequence) AppendBatch(ctx context.Context, entries []Entry, options ...BatchOption) (BatchResult, error)

AppendBatch atomically appends entries without grouping or reordering sources. All values/metadata are serialized before any scope read or write. Missing source checks resolve optimistically using each source's first entry route. An empty batch must carry a protected check. Transport failures are unknown outcomes and never retried; known atomic rejections are returned as results.

func (*Sequence) AppendBatchWithMetadata 

func (s *Sequence) AppendBatchWithMetadata(ctx context.Context, entries []Entry, options ...BatchOption) (BatchOperationResult, error)

AppendBatchWithMetadata behaves like AppendBatch with outcome-independent metadata.

func (*Sequence) AppendMany 

func (s *Sequence) AppendMany(ctx context.Context, source events.SourceID, values []any, options ...AppendOption) (BatchResult, error)

AppendMany atomically appends registered values for one source, in input order. Static tags from every type are unioned across the whole batch, like C#. The legacy RPC is used only for the default route; other routes use the ordered heterogeneous RPC. Options and failure semantics are the same as Append. Empty values require an effective protected scope, not an unchecked no-op.

func (*Sequence) AppendManyWithMetadata 

func (s *Sequence) AppendManyWithMetadata(ctx context.Context, source events.SourceID, values []any, options ...AppendOption) (BatchOperationResult, error)

AppendManyWithMetadata behaves like AppendMany with outcome-independent metadata.

func (*Sequence) AppendPreparedBatch 

func (s *Sequence) AppendPreparedBatch(ctx context.Context, snapshot *PreparedBatch) (BatchResult, error)

AppendPreparedBatch resolves remaining optimistic scopes and dispatches one atomic heterogeneous batch. No event is serialized again. The snapshot must originate from s; commit metadata comes from the snapshot, not ctx. This low- level method does not prevent repeated calls; transaction owners do. Local notification origin is resolved from this append ctx, not the snapshot. PrepareBatch never invokes the resolver; SDK-owned unit commits bypass it.

func (*Sequence) AppendWithMetadata 

func (s *Sequence) AppendWithMetadata(ctx context.Context, source events.SourceID, event any, options ...AppendOption) (AppendOperationResult, error)

AppendWithMetadata behaves like Append and also returns attempted coordinates and registered event types on every outcome, including operation errors. Unregistered inputs have no type reference; they still fail as in Append.

func (*Sequence) BindAuditMetadata 

func (s *Sequence) BindAuditMetadata(_ outgoing.Binding, ctx context.Context) (*PreparedBatch, outgoing.Audit, error)

BindAuditMetadata is the internal Begin bridge. Empty preparation resolves metadata without enriching any events; no caller context is retained.

func (*Sequence) CompleteStream 

func (s *Sequence) CompleteStream(ctx context.Context, streamType events.StreamType, streamID events.StreamID) (events.SequenceNumber, error)

CompleteStream permanently closes a stream type/ID pair in this sequence, across ALL event sources. Both parameters must be explicit and nonblank. Success returns the sequence-wide tail at completion, not the stream's tail; Unavailable is valid for an empty sequence. Subsequent appends to the pair are rejected by the kernel with StreamClosed. Repeated completion returns StreamAlreadyCompleted. All/Default returns DefaultStreamCannotBeCompleted.

Completion is synchronous, unlike Redact/Revise. There is no reopen operation or stream-status RPC in the pinned contract. It carries no actor/reason fields on the wire: consumers must authorize and audit it externally. No retries, local append notifications or unit-of-work staging occur. Transport failures and malformed returned tails are MutationOutcomeUnknownError, not proof the stream remains open. Unavailable means absence; Max and BeforeFirst are invalid.

func (*Sequence) DecisionTarget 

func (s *Sequence) DecisionTarget() decision.Target

DecisionTarget is an internal SDK composition seam, not a guard constructor.

func (*Sequence) HasEvents 

func (s *Sequence) HasEvents(ctx context.Context, source events.SourceID) (bool, error)

HasEvents reports whether the source has any events in this sequence/namespace.

func (*Sequence) ID 

func (s *Sequence) ID() events.SequenceID

ID returns the selected event sequence.

func (*Sequence) Next 

func (s *Sequence) Next(ctx context.Context) (events.SequenceNumber, error)

Next returns a non-reserved candidate position: zero for an empty sequence, otherwise tail plus one. A concurrent writer may consume it immediately.

func (*Sequence) OnAppend 

func (s *Sequence) OnAppend(callback func(AppendNotification)) (unsubscribe func())

OnAppend subscribes to this Sequence. EventStore caches handles, so subscriptions are shared per client, store, namespace and sequence. Low-level New handles keep independent subscriptions. Nil callbacks return a no-op unsubscribe. The caller must unsubscribe to release the callback and its captured state. Unsubscribe is concurrency-safe, idempotent and nonblocking, including inside callbacks. It prevents new callback admissions but does not join an already admitted invocation. Join command-owned appends before disposing command state.

Subscribers run synchronously in subscription order, after a validated, nonempty request is handed to the RPC client and its result becomes known, before Append/AppendMany/AppendBatch/AppendPreparedBatch returns. Staging, pre-dispatch failures, rollback and eventless operations do not notify. Unit-of-work commits notify through AppendPreparedBatch, before unit completion. The subscriber list is snapshotted at delivery; subscriptions added during a callback start with the next delivery. Disposal before admission skips it.

No internal locks are held during callbacks. Callbacks may subscribe, dispose or append recursively (the caller must bound recursion). Concurrent appends may invoke the same callback concurrently, with no cross-operation ordering; protect shared command state and filter Origin. Slow callbacks delay the caller. Panics become AppendCallbackPanicError, not a changed write disposition or a silent failure. Notification Err excludes subscriber failures.

Example 
package main

import (
	"context"
	"errors"
	"fmt"
	"sync"

	"github.com/cratis/chronicle.go/contracts/sequences"
	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/eventsequences"
	"github.com/cratis/chronicle.go/metadata"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// lostAcknowledgmentConnection is an offline fixture, not a production transport.
// Normally obtain the sequence from store.EventLog().
type lostAcknowledgmentConnection struct{}

func (lostAcknowledgmentConnection) Invoke(_ context.Context, method string, _, _ any, _ ...grpc.CallOption) error {
	if method != sequences.EventSequences_Append_FullMethodName {
		return fmt.Errorf("unexpected RPC: %s", method)
	}
	return status.Error(codes.Unavailable, "append acknowledgment lost")
}

func (lostAcknowledgmentConnection) NewStream(context.Context, *grpc.StreamDesc, string, ...grpc.CallOption) (grpc.ClientStream, error) {
	return nil, fmt.Errorf("unexpected stream")
}

func main() {
	type OrderPlaced struct{ Product string }
	definition, err := events.Define[OrderPlaced](events.WithID("OrderPlaced"))
	if err != nil {
		panic(err)
	}
	catalog, err := events.NewCatalog(definition.Descriptor())
	if err != nil {
		panic(err)
	}
	sequence, err := eventsequences.New("orders", "Default", events.EventLog, catalog, lostAcknowledgmentConnection{})
	if err != nil {
		panic(err)
	}
	correlation, err := metadata.NewCorrelationID()
	if err != nil {
		panic(err)
	}
	origin := eventsequences.NewOrigin()
	ctx := eventsequences.WithOrigin(metadata.WithCorrelation(context.Background(), correlation), origin)

	// A command adapter subscribes on the same handle handed to its handlers.
	// Other concurrent commands may use this handle, so filter and synchronize.
	var mu sync.Mutex
	var attempts []eventsequences.AppendNotification
	unsubscribe := sequence.OnAppend(func(n eventsequences.AppendNotification) {
		if n.Origin != origin {
			return
		}
		mu.Lock()
		defer mu.Unlock()
		attempts = append(attempts, n)
	})
	defer unsubscribe()

	_, err = sequence.Append(ctx, "order-42", OrderPlaced{Product: "book"},
		eventsequences.WithScope(eventsequences.Scope{Expectation: eventsequences.NoCheck()}))
	var unknown *eventsequences.OutcomeUnknownError
	if !errors.As(err, &unknown) {
		panic(fmt.Sprintf("expected an unknown outcome, got %v", err))
	}

	// All command-owned appends have returned; dispose before releasing state.
	// To track ONLY immediate writes, also dispose before committing the unit.
	unsubscribe()
	mu.Lock()
	defer mu.Unlock()
	for _, n := range attempts {
		fmt.Println(n.Operation.Store(), n.Operation.Namespace(), n.Events[0].Source)
		fmt.Println("outcome unknown:", n.Result.Disposition == eventsequences.Unknown)
	}
}

Output:
orders Default order-42
outcome unknown: true

func (*Sequence) PrepareBatch 

func (s *Sequence) PrepareBatch(ctx context.Context, entries []Entry, options ...BatchOption) (*PreparedBatch, error)

PrepareBatch snapshots entries, context metadata and scopes without I/O. Serialization and all local validation happen now; Resolve tails are read only by AppendPreparedBatch. Empty snapshots are allowed for incremental enrollment. Inputs are borrowed until return and must not be mutated concurrently.

func (*Sequence) PrepareBoundBatch 

func (s *Sequence) PrepareBoundBatch(_ outgoing.Binding, ctx context.Context, bound outgoing.Audit, entries []Entry, scopes []LabeledScope) (*PreparedBatch, error)

PrepareBoundBatch resolves Stage's operational metadata outside unit locks, verifies its bound actor/correlation before encoding, and snapshots all entries.

func (*Sequence) PrepareReturnedEvents 

func (s *Sequence) PrepareReturnedEvents(ctx context.Context, effect reactoreffects.Preparation[Entry, LabeledScope]) (func(context.Context) error, error)

PrepareReturnedEvents is the module-private reactor seam. It freezes all local content/audit/origin work before any built-in append or custom effect executes. Dispatch preserves the original single, legacy-many or heterogeneous route.

func (*Sequence) ReadFrom 

func (s *Sequence) ReadFrom(ctx context.Context, from events.SequenceNumber, filter FromFilter) ([]events.Appended, error)

ReadFrom returns matching events at or after from (inclusive), in ascending sequence order. The RPC is a finite unpaged read, not a live subscription.

func (*Sequence) ReadHistory 

func (s *Sequence) ReadHistory(ctx context.Context, source events.SourceID, filter SourceFilter) (History, error)

ReadHistory returns ordered events and a normalized filter/expectation suitable for saving work based on precisely that history. Empty history protects absence; nonempty history uses the greatest loaded position. No tail RPC is performed.

func (*Sequence) ReadSource 

func (s *Sequence) ReadSource(ctx context.Context, source events.SourceID, filter SourceFilter) ([]events.Appended, error)

ReadSource returns complete matching source history in ascending sequence order. Empty filters do not narrow; returned data is owned by the caller. A malformed response or failed envelope returns an error, never a partially trusted history.

func (*Sequence) Redact 

func (s *Sequence) Redact(ctx context.Context, position events.SequenceNumber, reason events.RedactionReason) error

Redact requests irreversible replacement of one event's payload and revisions with an EventRedacted marker. Zero is valid; the three highest positions are reserved (Unavailable, Max and BeforeFirst) and are invalid targets. A nonblank reason is required. Actor and causation come from metadata in ctx. Nil error confirms request acceptance, NOT completion of the asynchronous mutation or observer replay. Authorize this operation at the consumer boundary. It never emits AppendOperations, participates in a unit of work, or retries.

Example 
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/cratis/chronicle.go/contracts/sequences"
	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/eventsequences"
	"github.com/cratis/chronicle.go/identities"
	"github.com/cratis/chronicle.go/metadata"
	"google.golang.org/grpc"
)

// acceptedHistoryConnection is an offline protocol fixture, not a kernel or a
// storage substitute. It acknowledges requests without applying any mutations.
type acceptedHistoryConnection struct{}

func (acceptedHistoryConnection) Invoke(_ context.Context, method string, _, reply any, _ ...grpc.CallOption) error {
	switch method {
	case sequences.EventSequences_Redact_FullMethodName, sequences.EventSequences_RedactForEventSource_FullMethodName, sequences.EventSequences_Revise_FullMethodName:
		reply.(*sequences.CommandResult).IsAuthorized = true
	case sequences.EventSequences_CompleteStream_FullMethodName:
		result := reply.(*sequences.CommandResult_CompleteStreamResponse)
		result.IsAuthorized = true
		result.Response = &sequences.CompleteStreamResponse{IsSuccess: true, SequenceNumber: 12}
	default:
		return fmt.Errorf("unexpected method %s", method)
	}
	return nil
}

func (acceptedHistoryConnection) NewStream(context.Context, *grpc.StreamDesc, string, ...grpc.CallOption) (grpc.ClientStream, error) {
	return nil, errors.New("unexpected stream")
}

func main() {
	type OrderNamed struct{ Name string }
	definition, err := events.Define[OrderNamed]()
	if err != nil {
		panic(err)
	}
	catalog, err := events.NewCatalog(definition.Descriptor())
	if err != nil {
		panic(err)
	}
	// In production use store.EventLog(), after authorization and approval.
	sequence, err := eventsequences.New("orders", "Default", events.EventLog, catalog, acceptedHistoryConnection{})
	if err != nil {
		panic(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	ctx = metadata.WithIdentity(ctx, identities.Identity{Subject: "approved-operator"})
	ctx = metadata.WithCausation(ctx, metadata.Causation{Occurred: time.Now().UTC(), Type: "history-correction", Properties: map[string]string{"ticket": "case-40"}})
	if err = sequence.Revise(ctx, 12, OrderNamed{Name: "corrected non-personal label"}); err != nil {
		panic(err)
	}
	if err = sequence.Redact(ctx, 12, "approved removal"); err != nil {
		var unknown *eventsequences.MutationOutcomeUnknownError
		if errors.As(err, &unknown) {
			fmt.Println("reconcile; do not retry")
		}
		return
	}
	fmt.Println("redaction request accepted; application and replay may still be pending")
	// This explicitly targets every generation of OrderNamed for order-42,
	// across all routes. Omitting the type argument targets ALL event types.
	if err = sequence.RedactForEventSource(ctx, "order-42", "approved source cleanup", definition.Ref().ID); err != nil {
		panic(err)
	}
	// Stream completion is a separate synchronous operation, not source deletion.
	if _, err = sequence.CompleteStream(ctx, "Orders", "closed-order"); err != nil {
		panic(err)
	}
	fmt.Println("stream closed")
}

Output:
redaction request accepted; application and replay may still be pending
stream closed

func (*Sequence) RedactForEventSource 

func (s *Sequence) RedactForEventSource(ctx context.Context, source events.SourceID, reason events.RedactionReason, eventTypes ...events.TypeID) error

RedactForEventSource requests irreversible redaction of all matching events for source in this sequence, across ALL stream/source types. With no eventTypes it targets every type; supplied IDs must be registered and select all generations. The kernel chooses matching events when it executes the asynchronous request, not at call time. This is not a prohibition on future appends. Reason and audit metadata follow Redact. Nil error means accepted, not completed. No retries, transaction staging, local append notifications, or client-side emulation occur.

func (*Sequence) ResolveScope 

func (s *Sequence) ResolveScope(ctx context.Context, filter ScopeFilter) (Scope, error)

ResolveScope applies the configured custom or default concurrency strategy to filter and resolves any returned Resolve expectation using its matching tail. It honors CheckFirstAppendIntoAScope: an empty tail becomes NoMatchingEvent when enabled, otherwise a resolved unchecked expectation retaining the filter. Explicit strategy expectations win. The returned scope can be staged or passed to WithScope/WithScopes without another strategy invocation or tail read.

Supply normalized dimensions (nil means no narrowing); no append-route defaults are added. Empty optional dimensions are non-narrowing. Inputs and returned filter collections are copied; do not mutate inputs during the call. The method is safe for concurrent use, honors ctx, never appends and never retries. It does not protect an earlier history read; use ReadHistory for that. Invalid filters or strategy scopes return ErrInvalidConfiguration; strategy/RPC errors retain their identities. The Sequence must have been constructed with New.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/cratis/chronicle.go/contracts/sequences"
	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/eventsequences"
	"github.com/cratis/chronicle.go/transactions"
	"google.golang.org/grpc"
)

// emptyScopeConnection is an executable fixture for an empty kernel tail.
// Applications normally obtain the sequence from EventStore.EventLog().
type emptyScopeConnection struct{ calls int }

func (*emptyScopeConnection) ConcurrencyPolicy() eventsequences.ConcurrencyPolicy {
	return eventsequences.ConcurrencyPolicy{CheckFirstAppendIntoAScope: true}
}
func (c *emptyScopeConnection) Invoke(_ context.Context, method string, _ any, response any, _ ...grpc.CallOption) error {
	if method != sequences.EventSequences_TailSequenceNumber_FullMethodName {
		return fmt.Errorf("unexpected RPC: %s", method)
	}
	c.calls++
	*response.(*sequences.QueryResult_EventSequenceTailResponse) = sequences.QueryResult_EventSequenceTailResponse{
		IsAuthorized: true, Data: &sequences.EventSequenceTailResponse{SequenceNumber: uint64(events.Unavailable)},
	}
	return nil
}
func (*emptyScopeConnection) NewStream(context.Context, *grpc.StreamDesc, string, ...grpc.CallOption) (grpc.ClientStream, error) {
	return nil, fmt.Errorf("unexpected stream")
}

func main() {
	catalog, err := events.NewCatalog()
	if err != nil {
		panic(err)
	}
	connection := &emptyScopeConnection{}
	sequence, err := eventsequences.New("orders", "Default", events.EventLog, catalog, connection)
	if err != nil {
		panic(err)
	}
	ctx := context.Background()
	source, stream := events.SourceID("order-42"), events.StreamID("sales")
	scope, err := sequence.ResolveScope(ctx, eventsequences.ScopeFilter{SourceID: &source, StreamID: &stream})
	if err != nil {
		panic(err)
	}
	unit, owner, err := transactions.Begin(ctx, sequence)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := owner.Rollback(); err != nil {
			panic(err)
		}
	}()
	// Stage a resolved, independently labeled check. No append or second tail read.
	if err := unit.Stage(ctx, nil, eventsequences.LabeledScope{Label: string(source), Scope: scope}); err != nil {
		panic(err)
	}
	fmt.Println(scope.Expectation == eventsequences.NoMatchingEvent(), connection.calls)
}

Output:
true 1

func (*Sequence) Revise 

func (s *Sequence) Revise(ctx context.Context, position events.SequenceNumber, replacement any) error

Revise requests a new content revision of an existing event, retaining its original content and history. The registered replacement supplies the type ID, generation and shared serialization plan. The kernel requires the same type ID as the original; it may reject this asynchronously AFTER accepting the request. Revision is not schema migration or PII erasure. The pinned kernel does NOT apply append's PII/encryption processing to revisions or their system requests. Any protected schema in the selected catalog sharing the replacement's type ID, or failure to inspect that metadata, returns ErrUnsupported before serialization or dispatch (Chronicle#4525). This cannot detect undeclared sensitive data, audit metadata or unknown server-only classifications, nor repair old leaks. Zero is valid; the three highest positions are reserved and invalid targets. It has no reason field on the wire: record the non-sensitive reason in metadata.WithCausation. Nil error means accepted, not applied or replayed. Inputs must not be mutated during the call. No retries, local append notifications or unit-of-work staging occur.

func (*Sequence) Tail 

func (s *Sequence) Tail(ctx context.Context, filter TailFilter) (events.SequenceNumber, bool, error)

Tail returns the greatest matching position and whether one exists. An empty history returns (0, false, nil), distinct from a first event at position zero. The returned position is not reserved and cannot protect an earlier read.

func (*Sequence) TailForObserver 

func (s *Sequence) TailForObserver(ctx context.Context, types []events.TypeRef) (events.SequenceNumber, bool, error)

TailForObserver reads the source tail filtered by the observer's exact declared event types, like C# GetTailSequenceNumberForObserver. It is NOT the observer's handled checkpoint. Pass plan.EventTypes() or Information.EventTypes(); empty types fail rather than accidentally querying the unfiltered sequence tail.

type SourceFilter 

type SourceFilter struct {
	// SourceType narrows source classification; Default does not narrow.
	SourceType events.SourceType
	// StreamType narrows stream classification; All does not narrow.
	StreamType events.StreamType
	// StreamID narrows stream identity; Default does not narrow.
	StreamID events.StreamID
	// EventTypes selects type IDs; empty means every event type.
	EventTypes []events.TypeRef
}

SourceFilter narrows a source read. Empty values and Default/All/Default are non-narrowing kernel sentinels, not exact routes. Event types match IDs across generations. Inputs are borrowed during a read and copied into History.

type TailFilter 

type TailFilter ScopeFilter

TailFilter selects matching history, using the same fields and sentinel semantics as ScopeFilter. Nil/empty dimensions do not narrow reads.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.