Documentation
¶
Index ¶
- func InitializeSqliteDatabase(ctx context.Context, sqliteCfg config.SqliteConfig, ...) (eventstore.EventsSaver, eventstore.EventsRetriever, eventstore.LockProvider, ...)
- 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 SqliteAdminDB
- func (a *SqliteAdminDB) CreateBoundaryIndex(ctx context.Context, boundary, name string, ...) (err error)
- func (a *SqliteAdminDB) DeleteUser(id string) error
- func (a *SqliteAdminDB) DropBoundaryIndex(ctx context.Context, boundary, name string) (err error)
- func (a *SqliteAdminDB) GetEventsCount(boundary string) (int, error)
- func (a *SqliteAdminDB) GetProjectorLastPosition(projectorName string) (*eventstore.Position, error)
- func (a *SqliteAdminDB) GetUserById(id string) (eventstore.User, error)
- func (a *SqliteAdminDB) GetUserByUsername(username string) (eventstore.User, error)
- func (a *SqliteAdminDB) GetUsersCount() (uint32, error)
- func (a *SqliteAdminDB) ListAdminUsers() ([]*eventstore.User, error)
- func (a *SqliteAdminDB) SaveEventCount(count int, boundary string) error
- func (a *SqliteAdminDB) SaveUsersCount(count uint32) error
- func (a *SqliteAdminDB) UpdateProjectorPosition(name string, position *eventstore.Position) error
- func (a *SqliteAdminDB) UpsertUser(user eventstore.User) error
- type SqliteEventNotifier
- type SqliteEventPublishing
- type SqliteGetEvents
- type SqliteSaveEvents
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func InitializeSqliteDatabase ¶
func InitializeSqliteDatabase( ctx context.Context, sqliteCfg config.SqliteConfig, adminCfg config.AdminConfig, boundaries []string, js jetstream.JetStream, logger logging.Logger, ) (eventstore.EventsSaver, eventstore.EventsRetriever, eventstore.LockProvider, common.DB, eventstore.EventPublishingTracker, func(string) eventstore.EventSignal, error)
InitializeSqliteDatabase opens per-boundary SQLite pools and constructs the five backend interfaces. Boundary identity is the file: {dir}/{boundary}.db.
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 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(id string) error
func (*SqliteAdminDB) DropBoundaryIndex ¶
func (a *SqliteAdminDB) DropBoundaryIndex(ctx context.Context, boundary, name string) (err error)
func (*SqliteAdminDB) GetEventsCount ¶
func (a *SqliteAdminDB) GetEventsCount(boundary string) (int, error)
func (*SqliteAdminDB) GetProjectorLastPosition ¶
func (a *SqliteAdminDB) GetProjectorLastPosition(projectorName string) (*eventstore.Position, error)
func (*SqliteAdminDB) GetUserById ¶
func (a *SqliteAdminDB) GetUserById(id string) (eventstore.User, error)
func (*SqliteAdminDB) GetUserByUsername ¶
func (a *SqliteAdminDB) GetUserByUsername(username string) (eventstore.User, error)
func (*SqliteAdminDB) GetUsersCount ¶
func (a *SqliteAdminDB) GetUsersCount() (uint32, error)
func (*SqliteAdminDB) ListAdminUsers ¶
func (a *SqliteAdminDB) ListAdminUsers() ([]*eventstore.User, error)
func (*SqliteAdminDB) SaveEventCount ¶
func (a *SqliteAdminDB) SaveEventCount(count int, boundary string) error
func (*SqliteAdminDB) SaveUsersCount ¶
func (a *SqliteAdminDB) SaveUsersCount(count uint32) error
func (*SqliteAdminDB) UpdateProjectorPosition ¶
func (a *SqliteAdminDB) UpdateProjectorPosition(name string, position *eventstore.Position) error
func (*SqliteAdminDB) UpsertUser ¶
func (a *SqliteAdminDB) UpsertUser(user eventstore.User) 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)
Source Files
Click to show internal directories.
Click to hide internal directories.