Documentation
¶
Index ¶
- func AdminBoundaryDefinition(adminConfig config.AdminConfig) boundarymodel.Definition
- type BoundaryPools
- func OpenBoundaryPools(ctx context.Context, dir, boundary, adminBoundary string) (*BoundaryPools, error)
- func OpenBoundaryPoolsWithConfig(ctx context.Context, sqliteCfg config.SqliteConfig, ...) (*BoundaryPools, error)
- func OpenMetadataPoolsWithConfig(ctx context.Context, sqliteCfg config.SqliteConfig, boundary string) (*BoundaryPools, error)
- type BoundaryRegistry
- type DatabaseRuntime
- type SqliteAdminDB
- func (a *SqliteAdminDB) CreateBoundaryIndex(ctx context.Context, boundary, name string, ...) (err error)
- func (a *SqliteAdminDB) DeleteUser(ctx context.Context, id string) error
- func (a *SqliteAdminDB) DropBoundaryIndex(ctx context.Context, boundary, name string) (err error)
- func (a *SqliteAdminDB) GetBoundaryIndex(ctx context.Context, boundary, name string) (*eventstore.BoundaryIndex, error)
- func (a *SqliteAdminDB) GetEventsCount(ctx context.Context, boundary string) (int, error)
- func (a *SqliteAdminDB) GetProjectorLastPosition(ctx context.Context, projectorName string) (*eventstore.Position, error)
- func (a *SqliteAdminDB) GetUserById(ctx context.Context, id string) (eventstore.User, error)
- func (a *SqliteAdminDB) GetUserByUsername(ctx context.Context, username string) (eventstore.User, error)
- func (a *SqliteAdminDB) GetUsersCount(ctx context.Context) (uint32, error)
- func (a *SqliteAdminDB) ListAdminUsers(ctx context.Context) ([]*eventstore.User, error)
- func (a *SqliteAdminDB) ListBoundaryIndexes(ctx context.Context, boundary string) ([]eventstore.BoundaryIndex, error)
- func (a *SqliteAdminDB) SaveEventCount(ctx context.Context, count int, boundary string) error
- func (a *SqliteAdminDB) SaveUsersCount(ctx context.Context, count uint32) error
- func (a *SqliteAdminDB) UpdateProjectorPosition(ctx context.Context, name string, position *eventstore.Position) error
- func (a *SqliteAdminDB) UpsertUser(ctx context.Context, user eventstore.User) error
- type SqliteBoundaryProvisioner
- type SqliteEventNotifier
- type SqliteEventPublishing
- type SqliteGetEvents
- type SqliteSaveEvents
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func AdminBoundaryDefinition ¶ added in v0.9.0
func AdminBoundaryDefinition(adminConfig config.AdminConfig) boundarymodel.Definition
Types ¶
type BoundaryPools ¶
type BoundaryPools struct {
Boundary string
Write *sqlitex.Pool
Read *sqlitex.Pool
// contains filtered or unexported fields
}
BoundaryPools holds the write and read connection pools backing one boundary's SQLite file.
Write pool is sized to 1: SQLite serializes writers anyway via its file lock, so a single in-flight writer queues callers in Go and avoids SQLITE_BUSY churn. Read pool is sized to runtime.NumCPU() — under WAL, readers run concurrently with the single writer.
func OpenBoundaryPools ¶
func OpenBoundaryPools(ctx context.Context, dir, boundary, adminBoundary string) (*BoundaryPools, error)
OpenBoundaryPools opens write+read pools for one boundary at {dir}/{boundary}.db. Migrations are applied on the first connection drawn from the write pool. adminBoundary is retained for API compatibility; admin metadata now lives in the admin boundary's metadata DB opened by OpenMetadataPoolsWithConfig.
func OpenBoundaryPoolsWithConfig ¶
func OpenBoundaryPoolsWithConfig(ctx context.Context, sqliteCfg config.SqliteConfig, boundary, adminBoundary string) (*BoundaryPools, error)
func OpenMetadataPoolsWithConfig ¶
func OpenMetadataPoolsWithConfig(ctx context.Context, sqliteCfg config.SqliteConfig, boundary string) (*BoundaryPools, error)
func (*BoundaryPools) Close ¶
func (b *BoundaryPools) Close() error
type BoundaryRegistry ¶ added in v0.8.0
type BoundaryRegistry struct {
// contains filtered or unexported fields
}
BoundaryRegistry is the shared, concurrency-safe source of SQLite event and metadata pools. Every boundary-aware adapter in one runtime uses the same registry so provisioning becomes visible atomically.
func NewBoundaryRegistry ¶ added in v0.8.0
func NewBoundaryRegistry(pools, metadataPools map[string]*BoundaryPools) *BoundaryRegistry
type DatabaseRuntime ¶ added in v0.8.0
type DatabaseRuntime struct {
SaveEvents eventstore.EventsSaver
GetEvents eventstore.EventsRetriever
LockProvider eventstore.LockProvider
AdminDB common.DB
EventPublishing eventstore.EventPublishingTracker
SignalProvider func(string) eventstore.EventSignal
ProvisionBoundary func(context.Context, boundarymodel.Definition) error
InstallBoundary func(context.Context, boundarymodel.Definition) error
}
func InitializeSqliteDatabaseRuntime ¶ added in v0.8.0
func InitializeSqliteDatabaseRuntime( ctx context.Context, sqliteCfg config.SqliteConfig, adminCfg config.AdminConfig, js jetstream.JetStream, logger logging.Logger, ) (*DatabaseRuntime, error)
func InitializeSqliteDatabaseRuntimeWithLockProvider ¶ added in v0.8.0
func InitializeSqliteDatabaseRuntimeWithLockProvider( ctx context.Context, sqliteCfg config.SqliteConfig, adminCfg config.AdminConfig, lockProvider eventstore.LockProvider, logger logging.Logger, ) (*DatabaseRuntime, error)
type SqliteAdminDB ¶
type SqliteAdminDB struct {
// contains filtered or unexported fields
}
func NewSqliteAdminDB ¶
func NewSqliteAdminDB(pools map[string]*BoundaryPools, adminBoundary string, logger logging.Logger) *SqliteAdminDB
func NewSqliteAdminDBWithMetadata ¶
func NewSqliteAdminDBWithMetadata(pools map[string]*BoundaryPools, metadataPools map[string]*BoundaryPools, adminBoundary string, logger logging.Logger) *SqliteAdminDB
func (*SqliteAdminDB) CreateBoundaryIndex ¶
func (a *SqliteAdminDB) CreateBoundaryIndex( ctx context.Context, boundary, name string, fields []eventstore.BoundaryIndexField, conditions []eventstore.BoundaryIndexCondition, combinator string, ) (err error)
CreateBoundaryIndex builds a partial expression index over orisun_es_event.data JSON keys. Index name = "{name}_idx" inside the boundary's database.
func (*SqliteAdminDB) DeleteUser ¶
func (a *SqliteAdminDB) DeleteUser(ctx context.Context, id string) error
func (*SqliteAdminDB) DropBoundaryIndex ¶
func (a *SqliteAdminDB) DropBoundaryIndex(ctx context.Context, boundary, name string) (err error)
func (*SqliteAdminDB) GetBoundaryIndex ¶ added in v0.9.3
func (a *SqliteAdminDB) GetBoundaryIndex(ctx context.Context, boundary, name string) (*eventstore.BoundaryIndex, error)
func (*SqliteAdminDB) GetEventsCount ¶
func (*SqliteAdminDB) GetProjectorLastPosition ¶
func (a *SqliteAdminDB) GetProjectorLastPosition(ctx context.Context, projectorName string) (*eventstore.Position, error)
func (*SqliteAdminDB) GetUserById ¶
func (a *SqliteAdminDB) GetUserById(ctx context.Context, id string) (eventstore.User, error)
func (*SqliteAdminDB) GetUserByUsername ¶
func (a *SqliteAdminDB) GetUserByUsername(ctx context.Context, username string) (eventstore.User, error)
func (*SqliteAdminDB) GetUsersCount ¶
func (a *SqliteAdminDB) GetUsersCount(ctx context.Context) (uint32, error)
func (*SqliteAdminDB) ListAdminUsers ¶
func (a *SqliteAdminDB) ListAdminUsers(ctx context.Context) ([]*eventstore.User, error)
func (*SqliteAdminDB) ListBoundaryIndexes ¶ added in v0.9.3
func (a *SqliteAdminDB) ListBoundaryIndexes(ctx context.Context, boundary string) ([]eventstore.BoundaryIndex, error)
func (*SqliteAdminDB) SaveEventCount ¶
func (*SqliteAdminDB) SaveUsersCount ¶
func (a *SqliteAdminDB) SaveUsersCount(ctx context.Context, count uint32) error
func (*SqliteAdminDB) UpdateProjectorPosition ¶
func (a *SqliteAdminDB) UpdateProjectorPosition(ctx context.Context, name string, position *eventstore.Position) error
func (*SqliteAdminDB) UpsertUser ¶
func (a *SqliteAdminDB) UpsertUser(ctx context.Context, user eventstore.User) error
type SqliteBoundaryProvisioner ¶ added in v0.8.0
type SqliteBoundaryProvisioner struct {
// contains filtered or unexported fields
}
SqliteBoundaryProvisioner is the SQLite adapter for the boundary provisioning slice. It opens and migrates both files, installs the writer queue, then publishes the pools to every adapter through one registry.
func NewSqliteBoundaryProvisioner ¶ added in v0.8.0
func NewSqliteBoundaryProvisioner( sqliteCfg config.SqliteConfig, adminCfg config.AdminConfig, registry *BoundaryRegistry, saver *SqliteSaveEvents, ) *SqliteBoundaryProvisioner
func (*SqliteBoundaryProvisioner) InstallBoundary ¶ added in v0.8.0
func (p *SqliteBoundaryProvisioner) InstallBoundary(ctx context.Context, definition boundarymodel.Definition) error
func (*SqliteBoundaryProvisioner) ProvisionBoundary ¶ added in v0.8.0
func (p *SqliteBoundaryProvisioner) ProvisionBoundary(ctx context.Context, definition boundarymodel.Definition) error
type SqliteEventNotifier ¶
type SqliteEventNotifier struct {
// contains filtered or unexported fields
}
func NewSqliteEventNotifier ¶
func NewSqliteEventNotifier(interval time.Duration) *SqliteEventNotifier
func NewSqliteEventNotifierWithWakeDelay ¶
func NewSqliteEventNotifierWithWakeDelay(interval, wakeDelay time.Duration) *SqliteEventNotifier
func (*SqliteEventNotifier) Notify ¶
func (n *SqliteEventNotifier) Notify(boundary string)
func (*SqliteEventNotifier) Signal ¶
func (n *SqliteEventNotifier) Signal(boundary string) eventstore.EventSignal
type SqliteEventPublishing ¶
type SqliteEventPublishing struct {
// contains filtered or unexported fields
}
func NewSqliteEventPublishing ¶
func NewSqliteEventPublishing(pools map[string]*BoundaryPools, logger logging.Logger) *SqliteEventPublishing
func NewSqliteEventPublishingWithMetadata ¶
func NewSqliteEventPublishingWithMetadata(metadataPools map[string]*BoundaryPools, logger logging.Logger) *SqliteEventPublishing
func (*SqliteEventPublishing) GetLastPublishedEventPosition ¶
func (p *SqliteEventPublishing) GetLastPublishedEventPosition(ctx context.Context, boundary string) (eventstore.Position, error)
func (*SqliteEventPublishing) InsertLastPublishedEvent ¶
type SqliteGetEvents ¶
type SqliteGetEvents struct {
// contains filtered or unexported fields
}
func NewSqliteGetEvents ¶
func NewSqliteGetEvents(pools map[string]*BoundaryPools, logger logging.Logger) *SqliteGetEvents
func (*SqliteGetEvents) GetBatch ¶
func (s *SqliteGetEvents) GetBatch(ctx context.Context, req *eventstore.GetEventsRequest) (eventstore.ReadEventBatch, error)
func (*SqliteGetEvents) GetLatestByCriteria ¶
func (s *SqliteGetEvents) GetLatestByCriteria(ctx context.Context, query eventstore.LatestByCriteriaQuery) (eventstore.LatestByCriteriaBatch, error)
GetLatestByCriteria returns the latest event per criterion plus the max observed position, all from ONE read snapshot: the per-criterion lookups run inside an explicit deferred read transaction, so under WAL every statement sees the same database state. Independent client reads cannot substitute — an event committing between them can hide below the observed max position.
type SqliteSaveEvents ¶
type SqliteSaveEvents struct {
// contains filtered or unexported fields
}
func NewSqliteSaveEvents ¶
func NewSqliteSaveEvents(pools map[string]*BoundaryPools, logger logging.Logger) *SqliteSaveEvents
NewSqliteSaveEvents constructs the saver with the package-default group-commit tuning. Use NewSqliteSaveEventsWithConfig to override.
func NewSqliteSaveEventsWithConfig ¶
func NewSqliteSaveEventsWithConfig( pools map[string]*BoundaryPools, logger logging.Logger, gcCfg config.SqliteGroupCommitConfig, ) (*SqliteSaveEvents, error)
func (*SqliteSaveEvents) SavePrepared ¶
func (s *SqliteSaveEvents) SavePrepared( ctx context.Context, events eventstore.PreparedEventBatch, boundary string, expectedPosition *eventstore.Position, streamConsistencyCondition *eventstore.Query, ) (transactionID string, globalID int64, err error)