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

readmodels

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 7, 2026 License: MIT Imports: 39 Imported by: 0

Documentation

Overview

Package readmodels provides explicit model declarations and namespace-bound, presence-aware reads. Models describe kernel-owned state, not local projections.

Index

Examples

Constants

View Source 
const UnlimitedInstances int32 = -1

UnlimitedInstances selects all instances after Skip (up to the kernel int32 limit).

Variables

View Source 
var ErrDecisionReadRefused = errors.New("chronicle: decision read refused")

ErrDecisionReadRefused identifies a decision that cannot safely issue a token.

View Source 
var ErrDecisionRequiresUnitOfWork = errors.New("chronicle: decision read requires an existing unit of work")

ErrDecisionRequiresUnitOfWork means Get has no existing participant. Use GetDetached when the outer orchestrator will explicitly enroll later.

View Source 
var ErrInterrupted = errors.New("chronicle: read-model subscription interrupted")

ErrInterrupted means a best-effort feed ended. There is no durable cursor: re-watch and re-fetch state; neither operation promises gap-free recovery.

View Source 
var ErrOverloaded = errors.New("chronicle: read-model subscription overloaded")

ErrOverloaded means the subscription's bounded queue filled. No changes are silently dropped: the stream terminates and requires a new watch and snapshot.

View Source 
var ErrRelease = errors.New("chronicle: read-model release failed")

ErrRelease identifies a failed protected-value release. No original ciphertext or partially released document is returned on this error.

Functions

func ValidateWatchOptions 

func ValidateWatchOptions(options ...WatchOption) error

ValidateWatchOptions validates options without opening a stream. Registration adapters use it to reject invalid queue configuration before client startup.

Types

type Catalog 

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

Catalog is an immutable, concurrency-safe collection. Use NewCatalog; the zero value is an empty catalog. Descriptors never retain caller-mutable collections.

func NewCatalog 

func NewCatalog(descriptors ...Descriptor) (*Catalog, error)

NewCatalog validates and snapshots declarations, rejecting duplicate types and identifiers. A model has one current generation; this is not a migration catalog.

func (*Catalog) Descriptors 

func (c *Catalog) Descriptors() []Descriptor

Descriptors returns a defensive copy in registration order.

func (*Catalog) Lookup 

func (c *Catalog) Lookup(value any) (Descriptor, bool)

Lookup resolves a value or non-nil pointer to a registered model.

func (*Catalog) LookupIdentifier 

func (c *Catalog) LookupIdentifier(id Identifier) (Descriptor, bool)

LookupIdentifier resolves the persisted model identity.

func (*Catalog) LookupType 

func (c *Catalog) LookupType(typ reflect.Type) (Descriptor, bool)

LookupType resolves a Go struct type or pointer to it, without constructing a value.

type Change 

type Change[T any] struct {
	// Type distinguishes addition, modification and removal.
	Type ChangeType
	// Key identifies the changed model, including when Value is absent.
	Key Key
	// Value is owned by the consumer and usable only when HasValue is true.
	Value T
	// HasValue is false when a removal carries no document.
	HasValue bool
	// Context carries available metadata; unspecified fields remain unknown.
	Context events.Context
}

Change owns a change's key, optional model, and available causing-event metadata. Value is usable only with HasValue. Removed is always explicit in Type; a deletion may carry the old model, or no value. Context.EventType is unknown: the watch protocol does not identify it or provide full event audit metadata.

type ChangeType 

type ChangeType uint8

ChangeType classifies a read-model change, not an event-log delivery. 

const (
	// Added is a new model (or a model entering a materialized window).
	Added ChangeType = iota
	// Modified is a changed model already present.
	Modified
	// Removed is a deleted model, or one leaving a materialized window.
	Removed
)

type CodecPanicError 

type CodecPanicError struct{}

CodecPanicError means an application decoder panicked. It matches ErrProtocol and never retains the panic value, which may contain protected content.

func (*CodecPanicError) Error 

func (*CodecPanicError) Error() string

Error returns a fixed, payload-free description.

func (*CodecPanicError) Unwrap 

func (*CodecPanicError) Unwrap() error

Unwrap preserves the protocol category without exposing the panic payload.

type Collection 

type Collection[T any] struct {
	// Instances is non-nil on success, including an empty collection.
	Instances []Instance[T]
	// ProcessedEventsCount is the actual reported or locally folded event count.
	ProcessedEventsCount events.Count
}

Collection owns present instances and the number of events reported by the operation. Deleted instances are omitted; present zero-valued models remain. ProcessedEventsCount is not a sequence position, sink watermark or freshness guarantee. Materialized retrieval reports zero even when instances exist.

type DecisionCodecPanicError 

type DecisionCodecPanicError struct{}

DecisionCodecPanicError means an application model decoder panicked during a decision read. It matches ErrProtocol and never retains or formats the panic value. The read returns no model or token; cleanup already completed.

func (*DecisionCodecPanicError) Error 

func (*DecisionCodecPanicError) Error() string

Error returns a fixed, payload-free description.

func (*DecisionCodecPanicError) Unwrap 

func (*DecisionCodecPanicError) Unwrap() error

Unwrap preserves the protocol failure category without exposing a panic value.

type DecisionRead 

type DecisionRead[T any] struct {
	Instance Instance[T]                // Instance preserves removal progress even when absent.
	Token    transactions.DecisionToken // Token can be enrolled only by one unit/owner.
}

DecisionRead contains explicitly present/absent model state and an opaque optimistic guard. Every failure returns its zero value, never usable evidence.

type DecisionReadAdmission 

type DecisionReadAdmission struct {
	IsAdmitted bool                      // IsAdmitted means the frozen local shape is supported.
	Reason     DecisionReadRefusalReason // Reason is empty on admission.
}

DecisionReadAdmission reports local shape admission without network I/O. It does not establish server agreement, capabilities or a protected read.

type DecisionReadRefusalReason 

type DecisionReadRefusalReason string

DecisionReadRefusalReason is a stable admission or read refusal category. 

const (
	DecisionUnavailable          DecisionReadRefusalReason = "unavailable"            // DecisionUnavailable means no SDK decision provider.
	DecisionReducer              DecisionReadRefusalReason = "reducer"                // DecisionReducer excludes reducer-backed models.
	DecisionAmbiguousProjection  DecisionReadRefusalReason = "ambiguous-projection"   // DecisionAmbiguousProjection requires exactly one projection.
	DecisionNotEventLog          DecisionReadRefusalReason = "not-event-log"          // DecisionNotEventLog excludes other sequences.
	DecisionJoin                 DecisionReadRefusalReason = "join"                   // DecisionJoin excludes joins and removal joins.
	DecisionHierarchy            DecisionReadRefusalReason = "hierarchy"              // DecisionHierarchy excludes children and nested projections.
	DecisionOpenEndedEventTypes  DecisionReadRefusalReason = "open-ended-event-types" // DecisionOpenEndedEventTypes excludes all-event subscriptions.
	DecisionDerivatives          DecisionReadRefusalReason = "derivatives"            // DecisionDerivatives excludes FromEvery derivatives.
	DecisionFromEventProperty    DecisionReadRefusalReason = "from-event-property"    // DecisionFromEventProperty excludes property-derived input.
	DecisionNotEventSourceKeyed  DecisionReadRefusalReason = "not-event-source-keyed" // DecisionNotEventSourceKeyed requires source keys, including parents and All.
	DecisionNoEventTypes         DecisionReadRefusalReason = "no-event-types"         // DecisionNoEventTypes requires a finite nonempty dependency set.
	DecisionUnsupportedEventType DecisionReadRefusalReason = "unsupported-event-type" // DecisionUnsupportedEventType rejects unresolved or comma-containing IDs.
	DecisionKeyConversion        DecisionReadRefusalReason = "key-conversion"         // DecisionKeyConversion requires a string or UUID schema key.
	DecisionInvalidKey           DecisionReadRefusalReason = "invalid-key"            // DecisionInvalidKey rejects noncanonical and wildcard keys.
	DecisionDefinitionMismatch   DecisionReadRefusalReason = "definition-mismatch"    // DecisionDefinitionMismatch means the server disagrees with admission.
	DecisionFoldIncomplete       DecisionReadRefusalReason = "fold-incomplete"        // DecisionFoldIncomplete exhausts three fresh fold attempts.
	DecisionFoldAhead            DecisionReadRefusalReason = "fold-ahead"             // DecisionFoldAhead exhausts three fresh pre-fold boundaries.
)
const DecisionProtectedModel DecisionReadRefusalReason = "protected-model"

DecisionProtectedModel formerly excluded every classified model from session-based decisions.

Deprecated: no longer produced. Kernel 19.32.2 and later release decision session folds with each value's original subject (Chronicle#4561), so classified models are admitted like C#. 

const DecisionProtectionMetadata DecisionReadRefusalReason = "protection-metadata"

DecisionProtectionMetadata formerly meant model protection could not be determined.

Deprecated: no longer produced; admission does not inspect protection metadata.

type DecisionReadRefused 

type DecisionReadRefused struct {
	Model  Identifier                // Model identifies the refused model.
	Reason DecisionReadRefusalReason // Reason describes admission/agreement/fold failure.
}

DecisionReadRefused identifies the model and stable reason; it never contains the potentially sensitive instance key or model content.

func (*DecisionReadRefused) Error 

func (e *DecisionReadRefused) Error() string

func (*DecisionReadRefused) Unwrap 

func (e *DecisionReadRefused) Unwrap() error

type DecisionReader 

type DecisionReader[T any] struct {
	// contains filtered or unexported fields
}

DecisionReader is a concurrency-safe admitted optimistic reader. Construct it with DecisionsFor; its zero value is invalid. It owns no client resources.

func DecisionsFor 

func DecisionsFor[T any](service *Service, model Model[T]) *DecisionReader[T]

DecisionsFor selects a typed decision reader without I/O. Only EventStore's frozen projection catalogs support issuance; low-level/test substitutes do not.

func (*DecisionReader[T]) Admit 

func (r *DecisionReader[T]) Admit() DecisionReadAdmission

Admit inspects the frozen catalog only. Refused declarations never issue RPCs.

func (*DecisionReader[T]) Get 

func (r *DecisionReader[T]) Get(ctx context.Context, key Key) (read DecisionRead[T], err error)

Get reads and enrolls into the existing participant in ctx. It never creates an implicit unit or exposes its owner. Failed enrollment returns no read/token.

func (*DecisionReader[T]) GetDetached 

func (r *DecisionReader[T]) GetDetached(ctx context.Context, key Key) (read DecisionRead[T], err error)

GetDetached uses at most three fresh projection sessions, without enrolling or creating a unit. Classified models are admitted; the kernel releases their folds. It checks server agreement before and after each fold, captures an unfiltered pre-fold log boundary, and probes the source's dependency types separately. LastHandled is progress, never a proof or replacement boundary. Cleanup is awaited, with a five-second cancellation-detached metadata-preserving budget; cleanup/cancellation/generation/epoch failure returns no token. Application codecs run only after cleanup, agreement checks and release of counted RPC work; cancellation, connection generation and catalog epoch are rechecked afterward. The pinned protocol cannot atomically bind definitions or in-place history changes. Codec panics return DecisionCodecPanicError without retaining the panic value. Errors have payload-free messages; underlying returned causes remain deliberately inspectable through errors.Is/As and may contain sensitive transport diagnostics.

type Descriptor 

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

Descriptor is immutable model metadata. Its zero value is invalid.

func BindProjection 

func BindProjection(d Descriptor, id string, sequence events.SequenceID, passive bool) (Descriptor, error)

BindProjection returns a new descriptor associated with a compiled producer. It is a registry-composition hook, not runtime registration. Explicit conflicting observer/sequence settings fail; passive projections use None rather than a materialized sink. Existing handles still refer to the same model declaration.

func BindReducer 

func BindReducer(d Descriptor, id string, sequence events.SequenceID, passive bool) (Descriptor, error)

BindReducer associates a model with exactly one reducer at registry freeze. Existing projection ownership and explicit conflicting sequences are errors. A passive model uses no sink; inactive materialized reducers retain their sink.

func (Descriptor) ContainerName 

func (d Descriptor) ContainerName() string

ContainerName returns the sink container, not the model identifier.

func (Descriptor) DisplayName 

func (d Descriptor) DisplayName() string

DisplayName returns the human-facing name.

func (Descriptor) EventSequence 

func (d Descriptor) EventSequence() events.SequenceID

EventSequence returns the sequence used by immediate reads and session cleanup.

func (Descriptor) Fields 

func (d Descriptor) Fields() []serialization.Field

Fields returns detached metadata from the same plan used by the model schema.

func (Descriptor) Generation 

func (d Descriptor) Generation() Generation

Generation returns the schema generation.

func (Descriptor) GoType 

func (d Descriptor) GoType() reflect.Type

GoType returns the declared, non-pointer struct type.

func (Descriptor) HasExplicitEventSequence 

func (d Descriptor) HasExplicitEventSequence() bool

HasExplicitEventSequence reports whether the model overrides source inference.

func (Descriptor) Identifier 

func (d Descriptor) Identifier() Identifier

Identifier returns the stable persisted identity.

func (Descriptor) Indexes 

func (d Descriptor) Indexes() []string

Indexes returns copied serialized property paths, in declaration order.

func (Descriptor) KeyProperty 

func (d Descriptor) KeyProperty() string

KeyProperty returns the serialized root ID property, or empty when undeclared.

func (Descriptor) Marshal 

func (d Descriptor) Marshal(value any) ([]byte, error)

Marshal serializes a value or non-nil pointer of this model's type. The caller must not mutate it during the call. This does not apply encryption.

func (Descriptor) Observer 

func (d Descriptor) Observer() (ObserverType, string)

Observer returns the producer kind and identifier. An empty identifier means no producer was declared.

func (Descriptor) RebindJSON 

func (d Descriptor) RebindJSON(data []byte, original Descriptor) ([]byte, error)

RebindJSON returns a detached serialized snapshot with this descriptor's names. The original descriptor must describe the same Go type. Explicit nulls and omissions survive; no user callbacks or encryption are invoked.

func (Descriptor) Schema 

func (d Descriptor) Schema() string

Schema returns the shared serialization plan's schema with explicit read-model metadata.

func (Descriptor) Sink 

func (d Descriptor) Sink() Sink

Sink returns a copy of the sink selection.

func (Descriptor) Unmarshal 

func (d Descriptor) Unmarshal(data json.RawMessage) (any, error)

Unmarshal decodes a present model document into a pointer to its registered Go type, normalizing declared collections like Reader.Get. The result is owned. Decoder panics return CodecPanicError without retaining the panic value. Runtime adapters can use this with a catalog; ordinary callers should use Reader.

func (Descriptor) ValidateProducer 

func (d Descriptor) ValidateProducer() error

ValidateProducer checks deferred passive-model requirements at registry freeze.

func (Descriptor) WithDefaultSinkType 

func (d Descriptor) WithDefaultSinkType(kind SinkType) (Descriptor, error)

WithDefaultSinkType returns detached metadata with a materialized client default. Explicit WithSink selections and passive producer bindings win. This registry-composition hook preserves declaration identity, schema and codec snapshots; it performs no I/O and invokes no application callbacks.

func (Descriptor) WithNamingPolicy 

func (d Descriptor) WithNamingPolicy(policy serialization.NamingPolicy) (Descriptor, error)

WithNamingPolicy compiles detached property metadata for registry composition. Container names and declaration identity are unchanged; existing typed handles remain valid when used with the resulting client catalog. Provider classifications are frozen by Go member identity for subsequent naming and producer binding.

type Generation 

type Generation uint32

Generation identifies a schema generation, starting at one.

type Identifier 

type Identifier string

Identifier is the persisted, store-wide model identity, independent of its sink container.

type Instance 

type Instance[T any] struct {
	// Value is usable only when Exists is true.
	Value T
	// Exists distinguishes absent state from a legitimate zero-valued model.
	Exists bool
	// LastHandled is the reported position, or nil when unavailable.
	LastHandled *events.SequenceNumber
}

Instance distinguishes absence from a valid zero-valued model. Value is usable only when Exists is true. LastHandled is nil only for the unavailable sentinel; an absent (removed) model can still have a last-handled position.

type Key 

type Key string

Key identifies a model instance. Values are preserved verbatim.

type MaterializedReader 

type MaterializedReader[T any] struct {
	// contains filtered or unexported fields
}

MaterializedReader is the typed materialized-window API. Construct through Reader.Materialized.

func (*MaterializedReader[T]) GetInstances 

func (m *MaterializedReader[T]) GetInstances(ctx context.Context, window *Window) ([]T, error)

GetInstances is the typed equivalent of MaterializedService.GetInstances.

func (*MaterializedReader[T]) ObserveInstances 

func (m *MaterializedReader[T]) ObserveInstances(ctx context.Context, window *Window, options ...WatchOption) (*Subscription[[]T], error)

ObserveInstances is the typed equivalent of MaterializedService.ObserveInstances.

type MaterializedService 

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

MaterializedService reads sink windows. It does not execute passive projections or reducers, and leaving a window does not prove that a document was deleted.

func (*MaterializedService) GetInstances 

func (m *MaterializedService) GetInstances(ctx context.Context, model Identifier, window *Window) ([]json.RawMessage, error)

GetInstances reads the exact requested window. Nil means 0/50. Results own their JSON. The kernel releases stored values; the SDK validates the final representation without decrypting again. Protected windows are limited to MongoDB root string PII and namespace-encrypted properties excluding sink identity aliases; other classified profiles fail with ErrUnsupported before RPC.

func (*MaterializedService) ObserveInstances 

func (m *MaterializedService) ObserveInstances(ctx context.Context, model Identifier, window *Window, options ...WatchOption) (*Subscription[[]json.RawMessage], error)

ObserveInstances returns after the first snapshot has been decoded. Unlike Watch this RPC has no Subscribed marker; the first Recv returns that snapshot. Each subsequent value replaces the entire window. Server-side coalescing can omit intermediate states. No durable changes, cursor or gap-free resume exist. Cancellation, overload, terminal errors and Close follow Subscription's contract.

type Model 

type Model[T any] struct {
	// contains filtered or unexported fields
}

Model is a typed, immutable declaration, normally returned by chronicle.RegisterReadModel.

func Define 

func Define[T any](options ...ModelOption) (Model[T], error)

Define compiles T without registration or I/O. T must be a named, non-pointer struct accepted by serialization.Compile. Generic instantiations require explicit WithIdentifier and WithContainerName options. Most callers use RegisterReadModel.

func (Model[T]) Descriptor 

func (m Model[T]) Descriptor() Descriptor

Descriptor returns the immutable untyped model metadata.

func (Model[T]) Identifier 

func (m Model[T]) Identifier() Identifier

Identifier returns the persisted model identity.

type ModelOption 

type ModelOption func(*modelConfig)

ModelOption configures a declaration. Scalars are last-wins; property path options accumulate and reject duplicates. Slices are copied. Nil options and invalid final configurations fail before registry admission.

func Passive 

func Passive() ModelOption

Passive declares an on-demand reducer model. Register its reducer before NewClient; for projections use projections.Passive on the producer instead.

func WithCodecs 

func WithCodecs(codecs *serialization.Codecs) ModelOption

WithCodecs selects an immutable explicit derived codec set. The last option wins; nil clears it. Classification providers are frozen at declaration.

func WithContainerName 

func WithContainerName(name string) ModelOption

WithContainerName overrides the case-preserving pluralized simple type name, corresponding to C# ReadModelNameAttribute. It does not change the identifier.

func WithDisplayName 

func WithDisplayName(name string) ModelOption

WithDisplayName overrides the simple type name used for display.

func WithEventSequence 

func WithEventSequence(sequence events.SequenceID) ModelOption

WithEventSequence selects the sequence for immediate reads (default event-log).

func WithGeneration 

func WithGeneration(generation Generation) ModelOption

WithGeneration selects a positive schema generation (default one).

func WithIdentifier 

func WithIdentifier(id Identifier) ModelOption

WithIdentifier overrides the default full Go import path plus type name. Use an explicit C# full name to share a persisted model across languages.

func WithIndexes 

func WithIndexes(paths ...string) ModelOption

WithIndexes declares nested serialized property paths, including paths through collection items. Calls accumulate paths in declaration order. Unknown paths and duplicates within or across calls fail declaration with ErrInvalidConfiguration.

func WithObserver 

func WithObserver(kind ObserverType, id string) ModelOption

WithObserver associates a registered producer. It does not register that producer.

func WithPII 

func WithPII(paths ...string) ModelOption

WithPII marks properties as personal data in the model schema. Paths use serialized names and accumulate across calls. Duplicate paths within or across calls fail declaration with ErrInvalidConfiguration. Composite values propagate to leaves; explicitly classified collections retain their container classification. Tags and WithProtection use the same pipeline.

func WithProtection 

func WithProtection(declarations ...compliance.Declaration) ModelOption

WithProtection adds copied type/property classifications and providers. Tags need no option. Invalid or conflicting metadata fails before any connection I/O.

func WithSink 

func WithSink(sink Sink) ModelOption

WithSink replaces the default MongoDB/default-configuration sink.

func WithSubjectProperty 

func WithSubjectProperty(path string) ModelOption

WithSubjectProperty selects a top-level serialized scalar property for Release. Empty/unset values fall back to the model's ID field, as in C#.

type ObserverType 

type ObserverType uint8

ObserverType identifies the definition that produces this model. 

const (
	// Projection is the C# default observer type.
	Projection ObserverType = 2
	// Reducer identifies a reducer-backed model. Passive instances fold locally;
	// reducer sessions are not supported.
	Reducer ObserverType = 1
)

type PassiveReader 

type PassiveReader func(context.Context, Descriptor, Key) (Instance[json.RawMessage], error)

PassiveReader is the store's in-process reducer seam. It returns raw state; Service performs protected-value release before returning it to the caller.

type ProjectionReplayValidator 

type ProjectionReplayValidator func(context.Context, Descriptor) (known bool, err error)

ProjectionReplayValidator admits a model only when its producer's defaults and key/relationship shape are faithfully supported by kernel collection/history replay. It returns known=false, err=nil when no local producer is known. New collection/history APIs refuse unknown producers; legacy ReplayProjection preserves its caller-owned admission contract in that case.

type Reader 

type Reader[T any] struct {
	// contains filtered or unexported fields
}

Reader is a typed, concurrency-safe reader for one immutable model declaration. Its zero value is invalid; construct it with For.

func For 

func For[T any](service *Service, model Model[T]) *Reader[T]

For creates a lazy typed reader. A nil service, empty or foreign declaration is rejected by operations with ErrNotRegistered; no network I/O occurs here.

func (*Reader[T]) Get 

func (r *Reader[T]) Get(ctx context.Context, key Key) (Instance[T], error)

Get reads one instance. Missing models remain Exists=false; declared slice properties in present models normalize missing/null to empty (pointer-to-slice declarations retain nullable presence). LastHandled is never folded into T.

func (*Reader[T]) GetAll 

func (r *Reader[T]) GetAll(ctx context.Context, count *events.Count) (Collection[T], error)

GetAll decodes the entire collection with the registered serialization plan. No partial collection or progress is returned after decoding or cancellation fails. Codecs run outside transport leases and may close their client.

Example 
package main

import (
	"context"
	"errors"
	"fmt"

	contracts "github.com/cratis/chronicle.go/contracts/readmodels"
	"github.com/cratis/chronicle.go/events"
	"github.com/cratis/chronicle.go/readmodels"
	"google.golang.org/grpc"
)

// collectionExampleTransport is an offline response fixture, not kernel replay
// or storage. The real-kernel specs establish the routing and state semantics.
type collectionExampleTransport struct{}

func (collectionExampleTransport) Invoke(_ context.Context, method string, request, response any, _ ...grpc.CallOption) error {
	if method != contracts.ReadModels_GetAllInstances_FullMethodName {
		return errors.New("unexpected RPC")
	}
	r := response.(*contracts.GetAllInstancesResponse)
	r.Instances = []string{`{"id":"account-1","name":"Open"}`}
	if request.(*contracts.GetAllInstancesRequest).EventCount != uint64(events.UnlimitedCount) {
		r.ProcessedEventsCount = 2
	}
	return nil
}
func (collectionExampleTransport) NewStream(context.Context, *grpc.StreamDesc, string, ...grpc.CallOption) (grpc.ClientStream, error) {
	return nil, errors.New("unexpected stream")
}

func main() {
	type Account struct {
		ID   string `json:"id"`
		Name string `json:"name"`
	}
	model, err := readmodels.Define[Account](readmodels.WithObserver(readmodels.Projection, "accounts"))
	if err != nil {
		panic(err)
	}
	catalog, err := readmodels.NewCatalog(model.Descriptor())
	if err != nil {
		panic(err)
	}
	// Production uses store.ReadModels(). This offline adapter owns the fixture's
	// simple source-key/empty-default producer evidence.
	service, err := readmodels.New("accounts", "Default", catalog, collectionExampleTransport{}, readmodels.WithProjectionReplayValidator(allowReplay))
	if err != nil {
		panic(err)
	}
	reader := readmodels.For(service, model)
	ctx := context.Background()
	current, err := reader.GetAll(ctx, nil)
	if err != nil {
		panic(err)
	}
	fmt.Println(current.Instances[0].Value.ID, current.ProcessedEventsCount)
	count := events.Count(2)
	historical, err := reader.GetAll(ctx, &count)
	if err != nil {
		panic(err)
	}
	fmt.Println(historical.ProcessedEventsCount, historical.Instances[0].LastHandled == nil)
	// Count two does not invent last-handled position one.

func (*Reader[T]) GetSnapshots 

func (r *Reader[T]) GetSnapshots(ctx context.Context, key Key) ([]Snapshot[T], error)

GetSnapshots decodes each state through the frozen model plan; contributions remain raw and can be decoded using events.Decode and the event catalog. No partial snapshot or metadata is returned on late codec failure/cancellation.

func (*Reader[T]) Materialized 

func (r *Reader[T]) Materialized() *MaterializedReader[T]

Materialized selects sink pages rather than projection changes or passive reads.

func (*Reader[T]) NewSession 

func (r *Reader[T]) NewSession(key Key) (*Session[T], error)

NewSession creates a lazy hydration session with an internally generated UUID. The first Get hydrates; each Get uses that same session. Reducers and models without a declared projection fail with ErrUnsupported. Classified values are released by the kernel with the subject each value was written under (kernel 19.32.2 or later, Chronicle#4561); against an older or unverified kernel, Get of a classified model fails with ErrUnsupported before RPC. Call Close even after failed Get, because the kernel may already have allocated hydration state.

func (*Reader[T]) Release 

func (r *Reader[T]) Release(ctx context.Context, value T) (T, error)

Release decrypts a value fetched directly from a sink. No value is returned on failure. T must carry its subject; use Service.Release for raw stored lineage.

func (*Reader[T]) ReleaseMany 

func (r *Reader[T]) ReleaseMany(ctx context.Context, values []T) ([]T, error)

ReleaseMany releases unreleased sink values in input order. It is not idempotent: never pass server-released read results, since legitimate plaintext can resemble ciphertext. Input values are never modified. Any failure returns a nil slice, never partially released data. Nil input remains nil.

func (*Reader[T]) Watch 

func (r *Reader[T]) Watch(ctx context.Context, options ...WatchOption) (*Subscription[Change[T]], error)

Watch is the typed equivalent of Service.Watch, including its ready barrier, bounded queue, terminal errors, namespace isolation and Close obligation.

type ReducerCollectionReader 

type ReducerCollectionReader func(context.Context, Descriptor, events.Count) (Collection[json.RawMessage], error)

ReducerCollectionReader folds released matching history locally. It returns owned plaintext, actual progress and present instances only. The callback must honor cancellation and support concurrent calls. It must not publish changes or acknowledge an observer. No counted transport lease spans this callback.

type ReductionChanges 

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

ReductionChanges is a store/namespace-local notification source for successful in-process folds, not a durable kernel change feed. The client/store coordinate owns it; adapters may construct its zero value, bind a generation, and publish folds. It is concurrency-safe and invokes no callbacks while holding its lock.

func (*ReductionChanges) BindGeneration 

func (h *ReductionChanges) BindGeneration(ctx context.Context)

BindGeneration binds future local watches to a client generation. Rebinding terminates old feeds; already returned watches never silently switch generation. The context must be a client lifetime, not a short registration/request context.

func (*ReductionChanges) Publish 

func (h *ReductionChanges) Publish(generation context.Context, model Identifier, key Key, value json.RawMessage)

Publish reports a successful local fold in its final plaintext representation, not an encrypted sink document. Nil means removed. Documents are snapshotted; the source never waits for slow watchers. Each overloaded watcher terminates independently and no failure is attributed to the durable reducer. generation must be the exact context passed to BindGeneration; late folds from retired generations are ignored. Notifications do not prove that the kernel persisted the returned fold yet.

type ReleaseError 

type ReleaseError struct {
	// Cause is the inspectable underlying error. Do not log transport diagnostics
	// without considering whether the server included sensitive data.
	Cause error
}

ReleaseError preserves the underlying transport/protocol cause without including model data, subject values or potentially sensitive kernel error messages.

func (*ReleaseError) Error 

func (e *ReleaseError) Error() string

Error returns a payload-free description.

func (*ReleaseError) Unwrap 

func (e *ReleaseError) Unwrap() []error

Unwrap exposes ErrRelease and the underlying failure to errors.Is/As.

type Service 

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

Service is a concurrency-safe namespace-bound reader. It borrows its transport and catalog. Use EventStore.ReadModels; New is for adapters and test transports.

func New 

func New(store metadata.StoreName, namespace metadata.Namespace, catalog *Catalog, conn grpc.ClientConnInterface, options ...ServiceOption) (*Service, error)

New constructs a service without I/O. The caller owns the channel and any registration/readiness barriers; the store facade supplies those automatically.

func (*Service) Catalog 

func (s *Service) Catalog() *Catalog

Catalog returns the immutable set of registered models.

func (*Service) Get 

func (s *Service) Get(ctx context.Context, model Identifier, key Key) (Instance[json.RawMessage], error)

Get reads raw JSON by registered model identity and key. JSON null means absent; an empty/malformed/non-object response fails with ErrProtocol. Raw JSON is owned by the caller and retains kernel metadata except the root ID alias, which is normalized to the model's serialized property name. Materialized, projection immediate and session reads are server-released and validated without another decrypt. Classified projection immediate/session reads need Chronicle 19.32.2 or later (Chronicle#4561) and fail with ErrUnsupported before RPC unless the connection reports such a kernel.

func (*Service) GetAll 

func (s *Service) GetAll(ctx context.Context, model Identifier, count *events.Count) (Collection[json.RawMessage], error)

GetAll reads a collection. A nil count selects materialized state unless the model is passive. Every explicit reducer count folds locally, even Unlimited. Finite counts select the first N matching events globally, not per instance. Zero returns empty without I/O or reducer activation; finite counts above MaxInt32 fail rather than clamp. Local folds fetch complete matching history: the count bounds computation, not transport volume. Failures return no partial instances or progress. This operation allocates no hydration session.

func (*Service) GetSnapshots 

func (s *Service) GetSnapshots(ctx context.Context, model Identifier, key Key) ([]Snapshot[json.RawMessage], error)

GetSnapshots reads correlation-grouped projection history for a nonblank key. Reducers fail with ErrUnsupported before any RPC: kernel 19.29.4 returns an uninformative empty history for every non-projection. Replay fidelity admission also refuses unknown/defaulted/relationship producers. Classified states are released by the kernel with their original subjects from 19.32.2; older or unverified kernels are refused with ErrUnsupported. Success returns a non-nil slice. Any error or cancellation discards all results. No sessions are created.

func (*Service) GetValue 

func (s *Service) GetValue(ctx context.Context, typ reflect.Type, key Key) (any, error)

GetValue reads a registered model for a runtime-discovered Go parameter type. It uses the same collection normalization as Reader.Get. typ must name the model struct or *model. Missing models return a typed nil pointer; a value-type request fails with ErrNotRegistered rather than manufacturing present state. Most applications should prefer the statically typed Reader.Get API.

func (*Service) Materialized 

func (s *Service) Materialized() *MaterializedService

Materialized returns the raw materialized-window API for this namespace.

func (*Service) Release 

func (s *Service) Release(ctx context.Context, model Identifier, document json.RawMessage) (json.RawMessage, error)

Release resolves protected values in a raw sink document using its registered schema. Top-level properties are released in groups selected by __subjects, falling back to the resolved default subject. Persisted lineage is preserved. Missing subjects, kernel release errors and invalid replies fail closed. Unprotected documents are copied without a release RPC. This API accepts unreleased sink documents, not results from server-released reads. Release is not idempotent: legitimate plaintext can resemble ciphertext and must not be released twice. Namespace/global confidentiality does not require a subject. Subject-dependent values without an owner fail closed, even in mixed-scope models.

func (*Service) ReplayProjection 

func (s *Service) ReplayProjection(ctx context.Context, model Identifier, eventCount uint64) ([]json.RawMessage, error)

ReplayProjection computes all instances by replaying at most eventCount events through the kernel projection. This does not wait for or prove sink/observer catch-up. Counts above math.MaxInt32 are invalid; zero returns an empty non-nil result without an RPC. Reducers are explicitly unsupported here. The kernel releases classified values with the subject each was written under (19.32.2 or later, Chronicle#4561); the SDK never decrypts them again, and refuses classified models with ErrUnsupported unless the connection reports such a kernel. Returned JSON is owned and ID aliases are normalized by the same path as Get. No local projection engine is used.

func (*Service) Watch 

func (s *Service) Watch(ctx context.Context, model Identifier, options ...WatchOption) (*Subscription[Change[json.RawMessage]], error)

Watch opens a projection change feed and returns only after Subscribed. This barrier is NOT a current model snapshot. Changes are ordered as received; an interruption is terminal, never resumed. The context owns the full lifetime. Watches cannot join hydration sessions: the wire has no session selector. Classified models fail with ErrUnsupported before opening a stream: the pinned kernel cannot reliably release projection changes, and local reducer changes have no delivery-time erasure fence. For unclassified reducers, Watch attaches to this client's plaintext fold notifications (local readiness, no server Subscribed or sink-write guarantee). Local changes are Modified/Removed; reactors infer first-seen Added.

type ServiceOption 

type ServiceOption func(*Service) error

ServiceOption configures an immutable read service at construction.

func WithPassiveReader 

func WithPassiveReader(reader PassiveReader) ServiceOption

WithPassiveReader installs the store's reducer reader. The callback is borrowed and must support concurrent calls and cancellation. Repeated options fail.

func WithProjectionReplayValidator 

func WithProjectionReplayValidator(validator ProjectionReplayValidator) ServiceOption

WithProjectionReplayValidator installs immutable producer admission for new collection/history reads. Repeated options fail. The callback must support concurrent calls and cancellation; it must not evaluate projections locally.

func WithReducerCollectionReader 

func WithReducerCollectionReader(reader ReducerCollectionReader) ServiceOption

WithReducerCollectionReader installs the store's local collection fold. Unlike the legacy PassiveReader seam, results are already plaintext. Repeats fail.

func WithReductionChanges 

func WithReductionChanges(source *ReductionChanges) ServiceOption

WithReductionChanges installs the store's local reducer notification source. It is borrowed, and must be bound to a generation before Watch. Each watch has two bounded queues: raw local notifications and decoded consumer deliveries.

func WithReleasedPassiveReader 

func WithReleasedPassiveReader(reader PassiveReader) ServiceOption

WithReleasedPassiveReader installs a passive reader that folds already released events into plaintext. It preserves the legacy PassiveReader signature without decrypting its result again. Do not combine with WithPassiveReader.

func WithSnapshotEventCatalog 

func WithSnapshotEventCatalog(catalog *events.Catalog) ServiceOption

WithSnapshotEventCatalog installs frozen event schemas for protected snapshot contribution validation. Known protected generations validate their released representation before delivery; unknown types stay raw. The store supplies this automatically. Nil/repeated catalogs fail; no application codec runs here.

type Session 

type Session[T any] struct {
	// contains filtered or unexported fields
}

Session owns one projection hydration session, bound to one reader/key. Get and Close serialize with cancellable admission. It must not be copied. Close before closing the client; the kernel session is not a protected decision token or a snapshot guarantee, and is not automatically replayed after connection loss.

func (*Session[T]) Close 

func (s *Session[T]) Close(ctx context.Context) error

Close dehydrates using the exact model, key, namespace, sequence and session ID. It accepts a cleanup context rather than hiding I/O behind an unbounded Close. Success is idempotent; failed cleanup is reported and may be retried with a fresh context. No further Get is admitted once cleanup starts.

func (*Session[T]) Get 

func (s *Session[T]) Get(ctx context.Context) (Instance[T], error)

Get reads in this session. After cleanup begins it returns ErrClosed.

type Sink 

type Sink struct {
	// Type selects one of the supported kernel sink providers.
	Type SinkType
	// ConfigurationID selects a server-managed connection configuration.
	ConfigurationID string
}

Sink selects kernel-owned storage. ConfigurationID is a canonical UUID string; empty means the all-zero default configuration. No database driver is required.

type SinkType 

type SinkType string

SinkType identifies a kernel sink provider. 

const (
	// MongoDB is the default sink.
	MongoDB SinkType = "MongoDB"
	// SQL selects the kernel's SQL sink.
	SQL SinkType = "SQL"
	// InMemory selects the kernel's in-memory sink.
	InMemory SinkType = "InMemory"
	// NoSink selects on-demand projection or in-process reduction.
	NoSink SinkType = "None"
)

type Snapshot 

type Snapshot[T any] struct {
	// Instance is the projected state, not a presence-bearing Instance[T].
	Instance T
	// Events contains raw, server-released contributions with their full context.
	// Event IDs, original content, revisions and alternate generations are absent
	// from this RPC and are not fabricated. Unknown event types remain raw.
	Events []events.Appended
	// Occurred is the first contributing event's timestamp, not commit time.
	Occurred time.Time
	// CorrelationID belongs to this group, not the response envelope.
	CorrelationID metadata.CorrelationID
}

Snapshot owns a correlation-grouped projection state and its contributions. The kernel groups correlation globally: A/B/A produces groups A then B, not three contiguous groups. Neither existence nor last-handled progress is present on this wire; an empty object after removal is indistinguishable from a present empty model. Snapshots are not optimistic-concurrency or decision-read evidence.

type Subscription 

type Subscription[T any] struct {
	// contains filtered or unexported fields
}

Subscription owns one best-effort stream and its receive worker. One consumer may call Recv or iterate Values. Close is concurrency-safe and joins the worker; always close after early consumption. Buffered items precede terminal errors. Its zero value is invalid. It never reconnects or implies a current snapshot.

func (*Subscription[T]) Close 

func (s *Subscription[T]) Close() error

Close cancels and joins the receive worker. Repeated calls are safe. It returns nil for successful cleanup; inspect Err or Recv for the terminal stream outcome.

func (*Subscription[T]) Done 

func (s *Subscription[T]) Done() <-chan struct{}

Done closes when the receive worker has exited (buffered items may remain).

func (*Subscription[T]) Err 

func (s *Subscription[T]) Err() error

Err returns nil while running and the retained error after termination.

func (*Subscription[T]) Recv 

func (s *Subscription[T]) Recv() (T, error)

Recv returns the next owned value, or the retained terminal error. Cancellation is controlled by the context supplied when opening the subscription.

func (*Subscription[T]) Values 

func (s *Subscription[T]) Values() iter.Seq2[T, error]

Values iterates in receive order, yields the terminal error once, and closes on exhaustion or early break. Like Recv, it permits only one consumer.

type WatchOption 

type WatchOption func(*watchConfig)

WatchOption configures a subscription. Scalar options are last-wins.

func WithWatchBuffer 

func WithWatchBuffer(messages, bytes int) WatchOption

WithWatchBuffer bounds queued messages and their serialized bytes (defaults: 64 messages, 8 MiB). Both must be positive. A single oversized message also terminates with ErrOverloaded. Decoded Go values and gRPC's own buffers are extra.

type Window 

type Window struct {
	// Skip is the number of instances to skip; negative values mean zero.
	Skip int32
	// Take is the maximum count; -1 means unlimited, other nonpositive values empty.
	Take int32
}

Window selects a skip/take range, not an event-count replay. A nil *Window uses Skip=0, Take=50. Negative Skip becomes zero; Take<=0 is empty except -1 (unlimited). Non-aligned ranges fetch a covering page from the start, capped at MaxInt32.

type WindowDiffer 

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

WindowDiffer compares released, normalized model JSON by key. It is single- consumer state, scoped to one subscription. A removed item has no Value and only means it left this window, not that it was deleted from the sink. Unlike C#'s silent skip/overwrite, missing or duplicate keys fail explicitly. The zero value is usable; Reset is achieved by constructing a new differ.

func (*WindowDiffer) Diff 

func (d *WindowDiffer) Diff(model Descriptor, window []json.RawMessage) ([]Change[json.RawMessage], error)

Diff emits additions/modifications in current window order, then removals in previous order. Invalid windows leave the prior state intact. The registered key property takes precedence over C# id/_id/Id fallbacks.

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.