orisun

package
v0.10.0 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 25 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// Deprecated: use boundary.StatusProvisioning.
	BoundaryStatusProvisioning = boundarymodel.StatusProvisioning
	// Deprecated: use boundary.StatusActive.
	BoundaryStatusActive = boundarymodel.StatusActive
	// Deprecated: use boundary.StatusFailed.
	BoundaryStatusFailed = boundarymodel.StatusFailed
)
View Source 
const (
	IndexCombinatorAND = "AND"
	IndexCombinatorOR  = "OR"

	BoundaryIndexStateBuilding = "building"
	BoundaryIndexStateReady    = "ready"
)
View Source 
const (
	// DefaultReadBatchSize is used by storage backends when an in-process
	// caller does not provide a page size.
	DefaultReadBatchSize uint32 = 1_000
	// MaxReadBatchSize bounds one database read and its result allocation.
	// Larger logical reads must advance by position and fetch another page.
	MaxReadBatchSize uint32 = 10_000
)
View Source 
const (
	EventsSubjectName = "events"
)
View Source 
const (
	StreamDoesNotExist = -1
)

Variables

View Source 
var BuildTime = "unknown"

BuildTime is the time when the binary was built. This will be overridden during build by the -ldflags="-X 'orisun/orisun.BuildTime=<timestamp>'" flag.

View Source 
var ErrQueueFull = errors.New("message handler queue full")
View Source 
var GitCommit = "unknown"

GitCommit is the git commit hash from which the binary was built. This will be overridden during build by the -ldflags="-X 'orisun/orisun.GitCommit=<hash>'" flag.

View Source 
var Roles = []Role{RoleAdmin, RoleOperations}
View Source 
var Version = "dev"

Version is the current version of Orisun. This will be overridden during build by the -ldflags="-X 'orisun/orisun.Version=v1.2.3'" flag.

Functions

func GetBuildInfo 

func GetBuildInfo() (string, string, string)

GetBuildInfo returns the version, build time, and git commit hash.

func GetEventJetstreamSubjectName 

func GetEventJetstreamSubjectName(boundary string, position *Position) string

func GetEventNatsMessageId 

func GetEventNatsMessageId(preparePosition int64, commitPosition int64) string

func GetEventsNatsJetstreamStreamStreamName 

func GetEventsNatsJetstreamStreamStreamName(boundary string) string

func GetEventsStreamSubjectFilterForSubscription 

func GetEventsStreamSubjectFilterForSubscription(boundary string, stream *string) string

func GetEventsSubjectName 

func GetEventsSubjectName(boundary string) string

func GetVersion 

func GetVersion() string

GetVersion returns the current version of Orisun.

func ValidateBoundaryName added in v0.8.0 

func ValidateBoundaryName(name string) error

ValidateBoundaryName is retained for compatibility. Deprecated: use boundary.ValidateName.

Types

type Backoff 

type Backoff struct {
	Base, Max time.Duration
	// contains filtered or unexported fields
}

Backoff implements capped exponential backoff with jitter. Use for retry loops to avoid thundering-herd on shared resources (PG advisory locks, etc).

func (*Backoff) Reset 

func (b *Backoff) Reset()

func (*Backoff) Wait 

func (b *Backoff) Wait(ctx context.Context) error

Wait sleeps for the current interval (with up to 50% jitter), doubles it for next time (capped at Max), and returns ctx.Err if cancelled.

type Boundary deprecated added in v0.8.0 

type Boundary = boundarymodel.Boundary

Deprecated: use boundary.Boundary.

type BoundaryDefinition deprecated added in v0.8.0 

type BoundaryDefinition = boundarymodel.Definition

Deprecated: use boundary.Definition.

type BoundaryIndex added in v0.9.3 

type BoundaryIndex struct {
	Name       string
	Fields     []BoundaryIndexField
	Conditions []BoundaryIndexCondition
	Combinator string
	State      string
}

type BoundaryIndexCondition 

type BoundaryIndexCondition struct {
	Key      string
	Operator string
	Value    string
}

type BoundaryIndexField 

type BoundaryIndexField struct {
	JsonKey   string
	ValueType string
}

type BoundaryIndexManager 

type BoundaryIndexManager interface {
	CreateBoundaryIndex(ctx context.Context, boundary, name string, fields []BoundaryIndexField, conditions []BoundaryIndexCondition, combinator string) error
	DropBoundaryIndex(ctx context.Context, boundary, name string) error
	ListBoundaryIndexes(ctx context.Context, boundary string) ([]BoundaryIndex, error)
	GetBoundaryIndex(ctx context.Context, boundary, name string) (*BoundaryIndex, error)
}

type BoundaryPlacement deprecated added in v0.8.0 

type BoundaryPlacement = boundarymodel.Placement

Deprecated: use boundary.Placement.

type BoundaryStatus deprecated added in v0.8.0 

type BoundaryStatus = boundarymodel.Status

Deprecated: use boundary.Status.

type ComparationResult 

type ComparationResult int
const IsEqual ComparationResult = 0
const IsGreaterThan ComparationResult = 1
const IsLessThan ComparationResult = -1

func ComparePositions 

func ComparePositions(p1, p2 *Position) ComparationResult

type ConditionCombinator 

type ConditionCombinator int32
const (
	ConditionCombinator_AND ConditionCombinator = iota
	ConditionCombinator_OR
)

type CreateIndexRequest 

type CreateIndexRequest struct {
	Boundary            string
	Name                string
	Fields              []*IndexField
	Conditions          []*IndexCondition
	ConditionCombinator ConditionCombinator
}

type Criterion 

type Criterion struct {
	Tags []*Tag
}

type CustomEventStream 

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

func NewCustomEventStream 

func NewCustomEventStream(ctx context.Context) *CustomEventStream

func (*CustomEventStream) Context 

func (s *CustomEventStream) Context() context.Context

func (*CustomEventStream) Events 

func (s *CustomEventStream) Events() <-chan *Event

Events returns the channel for consuming events

func (*CustomEventStream) Recv 

func (s *CustomEventStream) Recv() (*Event, error)

func (*CustomEventStream) Send 

func (s *CustomEventStream) Send(event *Event) error

type Direction 

type Direction int32
const (
	Direction_ASC Direction = iota
	Direction_DESC
)

func (Direction) String 

func (d Direction) String() string

type DropIndexRequest 

type DropIndexRequest struct {
	Boundary string
	Name     string
}

type Event 

type Event struct {
	EventId     string    `json:"event_id"`
	EventType   string    `json:"event_type"`
	Data        string    `json:"data"`
	Metadata    string    `json:"metadata"`
	Position    *Position `json:"position"`
	DateCreated time.Time `json:"date_created"`
}

Event is the transport-neutral event shape used by legacy in-process command handlers. Storage backends use ReadEvent directly.

type EventPollingManager added in v0.8.0 

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

func StartEventPolling 

func StartEventPolling(
	ctx context.Context,
	config c.AppConfig,
	lockProvider LockProvider,
	getEvents EventsRetriever,
	js jetstream.JetStream,
	eventPublishingTracker EventPublishingTracker,
	signalProvider func(string) EventSignal,
	logger logging.Logger) *EventPollingManager

func (*EventPollingManager) StartBoundary added in v0.8.0 

func (m *EventPollingManager) StartBoundary(boundary string) error

StartBoundary starts exactly one publishing loop for a boundary in this process. Cluster-wide exclusivity remains enforced by the lock lease.

type EventPublishingTracker 

type EventPublishingTracker interface {
	GetLastPublishedEventPosition(ctx context.Context, boundary string) (Position, error)
	InsertLastPublishedEvent(ctx context.Context, boundary string, transactionID, globalID int64) error
}

EventPublishingTracker stores per-boundary durable publisher checkpoints.

type EventSignal 

type EventSignal interface {
	Wait(ctx context.Context) error
	Stop()
}

EventSignal reports that new events may be available.

type EventStore 

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

func InitializeEventStore 

func InitializeEventStore(
	ctx context.Context,
	config c.AppConfig,
	saveEvents EventsSaver,
	getEvents EventsRetriever,
	lockProvider LockProvider,
	indexManager BoundaryIndexManager,
	js jetstream.JetStream,
	logger logging.Logger) *EventStore

func NewEventStoreServer 

func NewEventStoreServer(
	js jetstream.JetStream,
	saveEventsFn EventsSaver,
	getEventsFn EventsRetriever,
	lockProvider LockProvider,
	indexManager BoundaryIndexManager,
	streamCfg EventStreamConfig,
	logger logging.Logger,
) *EventStore

func (*EventStore) ActivateBoundary added in v0.8.0 

func (s *EventStore) ActivateBoundary(boundary string) error

ActivateBoundary exposes a boundary to public requests after its activation event is durable. It is idempotent for replay and clustered delivery.

func (*EventStore) CreateIndex 

func (s *EventStore) CreateIndex(ctx context.Context, req *CreateIndexRequest) error

func (*EventStore) DropIndex 

func (s *EventStore) DropIndex(ctx context.Context, req *DropIndexRequest) error

func (*EventStore) EnableBoundaryActivationGate added in v0.8.0 

func (s *EventStore) EnableBoundaryActivationGate(initiallyActive ...string) error

EnableBoundaryActivationGate makes public event-store operations require a locally observed ACTIVE catalog state. The bootstrap boundary is supplied as initially active so the catalog can replay itself before application boundaries are exposed.

func (*EventStore) EnableOpenTelemetryMetrics added in v0.9.3 

func (s *EventStore) EnableOpenTelemetryMetrics() error

EnableOpenTelemetryMetrics attaches event-store instruments to the current global OpenTelemetry MeterProvider. Metrics are disabled until this is called.

func (*EventStore) EnsureBoundary added in v0.8.0 

func (s *EventStore) EnsureBoundary(ctx context.Context, boundary string) error

EnsureBoundary creates or updates the real-time stream for a boundary. It is idempotent so provisioning retries can safely call it after the durable backend has already been created.

func (*EventStore) GetEvents 

func (s *EventStore) GetEvents(ctx context.Context, req *GetEventsRequest) (*GetEventsResponse, error)

func (*EventStore) GetIndex added in v0.9.3 

func (s *EventStore) GetIndex(ctx context.Context, req *GetIndexRequest) (*GetIndexResponse, error)

func (*EventStore) GetLatestByCriteria 

func (s *EventStore) GetLatestByCriteria(ctx context.Context, req *GetLatestByCriteriaRequest) (*GetLatestByCriteriaResponse, error)

func (*EventStore) ListIndexes added in v0.9.3 

func (s *EventStore) ListIndexes(ctx context.Context, req *ListIndexesRequest) (*ListIndexesResponse, error)

func (*EventStore) Ping 

func (s *EventStore) Ping(ctx context.Context) error

func (*EventStore) RequireBoundaryActive added in v0.8.0 

func (s *EventStore) RequireBoundaryActive(boundary string) error

RequireBoundaryActive rejects unknown, provisioning, and failed catalog boundaries before a public request reaches a backend.

func (*EventStore) SaveEvents 

func (s *EventStore) SaveEvents(ctx context.Context, req *SaveEventsRequest) (resp *WriteResult, err error)

func (*EventStore) SubscribeToAllEvents 

func (s *EventStore) SubscribeToAllEvents(
	ctx context.Context,
	request coreeventstore.SubscribeRequest,
	handler coreeventstore.EventHandler,
) error

type EventStreamConfig 

type EventStreamConfig struct {
	MaxBytes int64
	MaxMsgs  int64
	MaxAge   time.Duration
}

type EventToSave 

type EventToSave struct {
	EventId   string
	EventType string
	Data      string
	Metadata  string
}

type EventWithMapTags 

type EventWithMapTags struct {
	EventId   string `json:"event_id"`
	EventType string `json:"event_type"`
	Data      any    `json:"data"`
	Metadata  any    `json:"metadata"`
}

EventWithMapTags is the flexible input representation used by embedding APIs.

type EventsRetriever 

type EventsRetriever interface {
	GetBatch(ctx context.Context, req *GetEventsRequest) (ReadEventBatch, error)
	GetLatestByCriteria(ctx context.Context, query LatestByCriteriaQuery) (LatestByCriteriaBatch, error)
}

EventsRetriever reads backend-neutral event batches and carried-state snapshots without depending on a network transport.

type EventsSaver 

type EventsSaver interface {
	SavePrepared(
		ctx context.Context,
		events PreparedEventBatch,
		boundary string,
		expectedPosition *Position,
		subSet *Query,
	) (transactionID string, globalID int64, err error)
}

EventsSaver accepts canonical batches prepared at an API boundary.

type GetEventsRequest 

type GetEventsRequest struct {
	Query        *Query
	FromPosition *Position
	Count        uint32
	Direction    Direction
	Boundary     string
}

type GetEventsResponse 

type GetEventsResponse struct {
	Events []*Event
}

type GetIndexRequest added in v0.9.3 

type GetIndexRequest struct {
	Boundary string
	Name     string
}

type GetIndexResponse added in v0.9.3 

type GetIndexResponse struct {
	Index *BoundaryIndex
}

type GetLatestByCriteriaRequest 

type GetLatestByCriteriaRequest struct {
	Boundary string
	Criteria []*Criterion
}

type GetLatestByCriteriaResponse 

type GetLatestByCriteriaResponse struct {
	Results         []*LatestCriterionResult
	ContextPosition *Position
}

type IndexCondition 

type IndexCondition struct {
	Key      string
	Operator string
	Value    string
}

type IndexField 

type IndexField struct {
	JsonKey   string
	ValueType ValueType
}

type JetStreamLockProvider 

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

JetStreamLockProvider implements a renewable, revision-fenced distributed lock using NATS JetStream KV.

func NewJetStreamLockProvider 

func NewJetStreamLockProvider(ctx context.Context, js jetstream.JetStream, logger logging.Logger) (*JetStreamLockProvider, error)

NewJetStreamLockProvider creates a new JetStreamLockProvider.

func (*JetStreamLockProvider) AcquireLock added in v0.7.0 

func (p *JetStreamLockProvider) AcquireLock(ctx context.Context, lockName string) (LockLease, error)

AcquireLock acquires a renewable, token-fenced lease. An expired lease may be replaced with a revision-guarded update.

func (*JetStreamLockProvider) Lock 

func (p *JetStreamLockProvider) Lock(ctx context.Context, lockName string) error

Lock acquires the named lock for the lifetime of ctx. Callers that need to prove ongoing ownership should use AcquireLock and its returned lease.

type LatestByCriteriaBatch 

type LatestByCriteriaBatch struct {
	Matches                []LatestCriterionMatch
	ContextCommitPosition  int64
	ContextPreparePosition int64
}

LatestByCriteriaBatch is the packed result used by embedded callers and storage backends. Matches is positionally aligned with the input criteria.

type LatestByCriteriaQuery 

type LatestByCriteriaQuery struct {
	Boundary string
	Criteria []ReadCriterion
}

LatestByCriteriaQuery is the protobuf-free request used below the gRPC boundary.

type LatestCriterionMatch 

type LatestCriterionMatch struct {
	Event ReadEvent
	Found bool
}

LatestCriterionMatch is positionally aligned with the corresponding input criterion. Event is valid only when Found is true.

type LatestCriterionResult 

type LatestCriterionResult struct {
	Criterion *Criterion
	Event     *Event
}

type ListIndexesRequest added in v0.9.3 

type ListIndexesRequest struct {
	Boundary string
}

type ListIndexesResponse added in v0.9.3 

type ListIndexesResponse struct {
	Indexes []*BoundaryIndex
}

type LockLease 

type LockLease interface {
	Context() context.Context
	Check(ctx context.Context) error
	Release()
}

LockLease proves ongoing lock ownership until it is released or lost.

type LockLeaseProvider 

type LockLeaseProvider interface {
	AcquireLock(ctx context.Context, lockName string) (LockLease, error)
}

LockLeaseProvider exposes explicit lock lifecycles to embedded subscribers and server-side publishers.

type LockProvider 

type LockProvider interface {
	Lock(ctx context.Context, lockName string) error
}

LockProvider coordinates named work within a backend runtime.

type MessageHandler 

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

func NewMessageHandler 

func NewMessageHandler[T any](ctx context.Context) *MessageHandler[T]

NewMessageHandler Add constructor and methods for the generic stream

func NewMessageHandlerWithBuffer 

func NewMessageHandlerWithBuffer[T any](ctx context.Context, bufferSize int) *MessageHandler[T]

NewMessageHandlerWithBuffer allows configuring the internal buffer size

func (*MessageHandler[T]) Close 

func (s *MessageHandler[T]) Close()

Close marks the handler as closed to prevent further sends

func (*MessageHandler[T]) Context 

func (s *MessageHandler[T]) Context() context.Context

func (*MessageHandler[T]) Recv 

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

func (*MessageHandler[T]) Send 

func (s *MessageHandler[T]) Send(event *T) error

func (*MessageHandler[T]) TrySend 

func (s *MessageHandler[T]) TrySend(event *T) error

TrySend attempts to enqueue without blocking and returns ErrQueueFull if buffer is full

type OrisunServer 

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

OrisunServer provides a high-level interface to interact with the Orisun event store

func NewOrisunServer 

func NewOrisunServer(
	ctx context.Context,
	saveEvents EventsSaver,
	getEvents EventsRetriever,
	lockProvider LockProvider,
	js jetstream.JetStream,
	logger logging.Logger,
) (*OrisunServer, error)

NewOrisunServer creates a new Orisun client with the provided configuration

func (*OrisunServer) ActivateBoundary added in v0.8.0 

func (c *OrisunServer) ActivateBoundary(_ context.Context, boundary string) error

ActivateBoundary exposes a durably activated boundary to public operations.

func (*OrisunServer) EnableBoundaryActivationGate added in v0.8.0 

func (c *OrisunServer) EnableBoundaryActivationGate(initiallyActive ...string) error

EnableBoundaryActivationGate makes public operations require an ACTIVE boundary catalog state. The admin boundary is supplied during bootstrap.

func (*OrisunServer) EnsureBoundary added in v0.8.0 

func (c *OrisunServer) EnsureBoundary(ctx context.Context, boundary string) error

EnsureBoundary prepares the real-time stream for a newly provisioned boundary. Durable backend creation remains the backend adapter's concern.

func (*OrisunServer) GetEvents 

func (c *OrisunServer) GetEvents(ctx context.Context, req *GetEventsRequest) (ReadEventBatch, error)

GetEvents retrieves events from the event store based on the request.

func (*OrisunServer) GetLatestByCriteria 

func (c *OrisunServer) GetLatestByCriteria(ctx context.Context, query LatestByCriteriaQuery) (LatestByCriteriaBatch, error)

GetLatestByCriteria returns the latest event per criterion from one backend read snapshot, plus the max observed position as the optimistic-lock token for the combined context.

func (*OrisunServer) RequireBoundaryActive added in v0.8.0 

func (c *OrisunServer) RequireBoundaryActive(boundary string) error

RequireBoundaryActive validates a boundary against the local catalog gate.

func (*OrisunServer) SaveEvents 

func (c *OrisunServer) SaveEvents(ctx context.Context, events []EventWithMapTags, boundary string,
	expectedPosition *Position, streamSubSet *Query) (*Position, error)

SaveEvents saves a batch of events to the event store

func (*OrisunServer) SubscribeToEvents 

func (c *OrisunServer) SubscribeToEvents(
	ctx context.Context,
	request coreeventstore.SubscribeRequest,
	handler coreeventstore.EventHandler,
) error

SubscribeToEvents subscribes to events from a boundary with the given handler

type PollingSignal 

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

func NewPollingSignal 

func NewPollingSignal(interval time.Duration) *PollingSignal

func (*PollingSignal) Stop 

func (s *PollingSignal) Stop()

func (*PollingSignal) Wait 

func (s *PollingSignal) Wait(ctx context.Context) error

type Position 

type Position struct {
	CommitPosition  int64 `json:"commit_position"`
	PreparePosition int64 `json:"prepare_position"`
}

Position identifies one event in a boundary's total order.

func FirstPosition 

func FirstPosition() Position

FirstPosition returns the first persisted event position.

func NotExistsPosition 

func NotExistsPosition() Position

NotExistsPosition is the optimistic-concurrency position before any event.

type PreparedEvent 

type PreparedEvent struct {
	EventId      string
	EventType    string
	DataJSON     string
	MetadataJSON string
}

PreparedEvent is the canonical backend-facing event representation.

func (PreparedEvent) MarshalJSON 

func (e PreparedEvent) MarshalJSON() ([]byte, error)

MarshalJSON preserves the canonical data and metadata as JSON values.

type PreparedEventBatch 

type PreparedEventBatch []PreparedEvent

PreparedEventBatch keeps canonical event descriptors contiguous.

func PrepareEventsForSave 

func PrepareEventsForSave(events []EventWithMapTags) (PreparedEventBatch, error)

PrepareEventsForSave normalizes flexible embedding input exactly once.

type Query 

type Query struct {
	Criteria []*Criterion
}

type ReadCriterion 

type ReadCriterion struct {
	Tags []ReadTag
}

ReadCriterion is one conjunction of equality tags. Criteria in a LatestByCriteriaQuery are independent and retain their input order.

type ReadEvent 

type ReadEvent struct {
	EventId         string
	EventType       string
	Data            string
	Metadata        string
	CommitPosition  int64
	PreparePosition int64
	DateCreated     time.Time
}

ReadEvent is the backend-neutral, contiguous read representation. Positions and timestamps are scalar values so storage and internal consumers do not allocate an object graph for every row.

func (*ReadEvent) Event added in v0.8.0 

func (e *ReadEvent) Event() *Event

Event materializes the legacy in-process event representation.

func (ReadEvent) MarshalJSON 

func (e ReadEvent) MarshalJSON() ([]byte, error)

MarshalJSON preserves the stable NATS event envelope while keeping the publisher on the packed representation.

type ReadEventBatch 

type ReadEventBatch []ReadEvent

ReadEventBatch keeps database results in one value slab.

func (ReadEventBatch) Response added in v0.8.0 

func (b ReadEventBatch) Response() *GetEventsResponse

Response materializes the legacy in-process response shape.

type ReadTag 

type ReadTag struct {
	Key   string
	Value string
}

ReadTag is the protobuf-free criterion tag used by embedded callers and storage backends.

type Role 

type Role string
const (
	RoleAdmin      Role = "ADMIN"
	RoleOperations Role = "OPERATIONS"
)

func (Role) String 

func (r Role) String() string

type SaveEventsRequest 

type SaveEventsRequest struct {
	Boundary string
	Query    *SaveQuery
	Events   []*EventToSave
}

type SaveQuery 

type SaveQuery struct {
	ExpectedPosition *Position
	SubsetQuery      *Query
}

type Tag 

type Tag struct {
	Key   string
	Value string
}

type User 

type User struct {
	Id             string `json:"id"`
	Name           string `json:"name"`
	Username       string `json:"username"`
	HashedPassword string `json:"password_hash"`
	Roles          []Role `json:"roles"`
}

type UserContextKeyType 

type UserContextKeyType string
const UserContextKey UserContextKeyType = "user"

type ValueType 

type ValueType int32
const (
	ValueType_TEXT ValueType = iota
	ValueType_NUMERIC
	ValueType_BOOLEAN
	ValueType_TIMESTAMPTZ
)

func (ValueType) String 

func (v ValueType) String() string

type WriteResult 

type WriteResult struct {
	LogPosition *Position
}

Source Files

View all Source files

Directories

Path Synopsis

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.