Documentation
¶
Index ¶
- Variables
- func CalculateVersion(value string) string
- func DeleteIndexedStorageObject(collection, userID, key string)
- func DeleteStorageObjects(ctx context.Context, pool *pgxpool.Pool, authoritative bool, ...) error
- func DeleteStorageObjectsTx(ctx context.Context, tx pgx.Tx, authoritative bool, reqs []DeleteRequest) error
- func EncodeCursor(c *ListCursor) (string, error)
- func IndexStorageObject(obj *StorageObject)
- func InitSearchIndex() error
- func SetDefaultIndex(idx IndexWriter)
- type BlugeStorageIndex
- func (si *BlugeStorageIndex) CreateIndex(def StorageIndexDefinition) error
- func (si *BlugeStorageIndex) DeleteStorage(ctx context.Context, deletes []DeleteRequest)
- func (si *BlugeStorageIndex) List(ctx context.Context, callerID, indexName, query string, limit int, ...) ([]*StorageObject, string, error)
- func (si *BlugeStorageIndex) RegisterFilter(indexName string, fn StorageIndexFilter)
- func (si *BlugeStorageIndex) WriteStorageAcks(ctx context.Context, acks []*StorageObjectAck, objects []*StorageObject)
- type DeleteRequest
- type IndexWriter
- type List
- type ListCursor
- type ReadRequest
- type StorageIndexDefinition
- type StorageIndexFilter
- type StorageObject
- type StorageObjectAck
- func WriteStorageObjects(ctx context.Context, pool *pgxpool.Pool, authoritative bool, ...) ([]*StorageObjectAck, error)
- func WriteStorageObjectsRetry(ctx context.Context, pool *pgxpool.Pool, reqs []ReadRequest, ...) ([]*StorageObjectAck, error)
- func WriteStorageObjectsTx(ctx context.Context, tx pgx.Tx, authoritative bool, objects []*StorageObject) ([]*StorageObjectAck, error)
Constants ¶
This section is empty.
Variables ¶
var ( // ErrStorageRejectedVersion is returned when OCC version checks fail. ErrStorageRejectedVersion = errors.New("storage write rejected: version") // ErrStorageRejectedPermission is returned when client lacks write permission. ErrStorageRejectedPermission = errors.New("storage write rejected: permission") // ErrStorageWriteExhaustedRetries is returned by WriteStorageObjectsRetry. ErrStorageWriteExhaustedRetries = errors.New("storage write exhausted retries") // ErrOCCConflict aliases version reject for older call sites/tests. ErrOCCConflict = ErrStorageRejectedVersion // ErrListCursorInvalid is returned for malformed list cursors. ErrListCursorInvalid = errors.New("storage list cursor invalid") )
Functions ¶
func CalculateVersion ¶
CalculateVersion calculates the MD5 hash of the value string.
func DeleteIndexedStorageObject ¶
func DeleteIndexedStorageObject(collection, userID, key string)
DeleteIndexedStorageObject deletes a storage object from the Bleve index.
func DeleteStorageObjects ¶
func DeleteStorageObjects(ctx context.Context, pool *pgxpool.Pool, authoritative bool, reqs []DeleteRequest) error
DeleteStorageObjects deletes objects transactionally.
func DeleteStorageObjectsTx ¶
func DeleteStorageObjectsTx(ctx context.Context, tx pgx.Tx, authoritative bool, reqs []DeleteRequest) error
DeleteStorageObjectsTx deletes storage objects inside an existing transaction.
func EncodeCursor ¶
func EncodeCursor(c *ListCursor) (string, error)
EncodeCursor encodes a ListCursor (gob+base64).
func IndexStorageObject ¶
func IndexStorageObject(obj *StorageObject)
IndexStorageObject indexes a storage object if the search index is initialized.
func InitSearchIndex ¶
func InitSearchIndex() error
InitSearchIndex initializes a memory-only Bleve index (test-only; production indexes deferred).
func SetDefaultIndex ¶
func SetDefaultIndex(idx IndexWriter)
SetDefaultIndex attaches a global index updated by Write/DeleteStorageObjects.
Types ¶
type BlugeStorageIndex ¶
type BlugeStorageIndex struct {
// contains filtered or unexported fields
}
BlugeStorageIndex is a per-node in-memory Bluge index registry.
func NewBlugeStorageIndex ¶
func NewBlugeStorageIndex() *BlugeStorageIndex
NewBlugeStorageIndex creates an empty index registry.
func (*BlugeStorageIndex) CreateIndex ¶
func (si *BlugeStorageIndex) CreateIndex(def StorageIndexDefinition) error
CreateIndex registers a new named index.
func (*BlugeStorageIndex) DeleteStorage ¶
func (si *BlugeStorageIndex) DeleteStorage(ctx context.Context, deletes []DeleteRequest)
DeleteStorage removes objects from matching indexes.
func (*BlugeStorageIndex) List ¶
func (si *BlugeStorageIndex) List(ctx context.Context, callerID, indexName, query string, limit int, order []string, cursor string) ([]*StorageObject, string, error)
List queries a named index and returns storage objects plus next cursor.
func (*BlugeStorageIndex) RegisterFilter ¶
func (si *BlugeStorageIndex) RegisterFilter(indexName string, fn StorageIndexFilter)
RegisterFilter attaches a custom filter to an index name.
func (*BlugeStorageIndex) WriteStorageAcks ¶
func (si *BlugeStorageIndex) WriteStorageAcks(ctx context.Context, acks []*StorageObjectAck, objects []*StorageObject)
WriteStorageAcks indexes written storage objects.
type DeleteRequest ¶
DeleteRequest defines a collection, key, user, and optional expected version.
type IndexWriter ¶
type IndexWriter interface {
WriteStorageAcks(ctx context.Context, acks []*StorageObjectAck, objects []*StorageObject)
DeleteStorage(ctx context.Context, deletes []DeleteRequest)
}
IndexWriter updates secondary storage indexes after durable writes.
type List ¶
type List struct {
Objects []*StorageObject
Cursor string
}
List is a paginated list response.
type ListCursor ¶
type ListCursor struct {
Key string `json:"key"`
UserID string `json:"user_id"`
Read int32 `json:"read"`
}
ListCursor maps to storageCursor for helpers.
func DecodeCursor ¶
func DecodeCursor(cursorStr string) (*ListCursor, error)
DecodeCursor decodes a list cursor.
type ReadRequest ¶
ReadRequest defines a collection, key, and user lookup.
type StorageIndexDefinition ¶
type StorageIndexDefinition struct {
Name string
Collection string
Key string
Fields []string
SortableFields []string
MaxEntries int
IndexOnly bool
}
StorageIndexDefinition describes a named Bluge index over a collection/key.
type StorageIndexFilter ¶
type StorageIndexFilter func(ctx context.Context, write *StorageObject) (bool, error)
StorageIndexFilter decides whether a write is indexed (true) or removed (false).
type StorageObject ¶
type StorageObject struct {
Collection string `json:"collection"`
Key string `json:"key"`
UserID string `json:"user_id"`
Value string `json:"value"`
Version string `json:"version"`
Read int16 `json:"read"`
Write int16 `json:"write"`
CreateTime time.Time `json:"create_time"`
UpdateTime time.Time `json:"update_time"`
}
StorageObject represents a record in the storage engine.
func ReadStorageObjects ¶
func ReadStorageObjects(ctx context.Context, pool *pgxpool.Pool, caller uuid.UUID, reqs []ReadRequest) ([]*StorageObject, error)
ReadStorageObjects reads objects; when caller != uuid.Nil, applies permission filter.
func SearchStorageObjects ¶
func SearchStorageObjects(ctx context.Context, pool *pgxpool.Pool, queryString string, limit int) ([]*StorageObject, error)
SearchStorageObjects searches via in-memory Bleve (test-only).
type StorageObjectAck ¶
type StorageObjectAck struct {
Collection string
Key string
UserID string
Version string
CreateTime time.Time
UpdateTime time.Time
}
StorageObjectAck is returned after a successful write.
func WriteStorageObjects ¶
func WriteStorageObjects(ctx context.Context, pool *pgxpool.Pool, authoritative bool, objects []*StorageObject) ([]*StorageObjectAck, error)
WriteStorageObjects writes objects transactionally with atomic OCC.
func WriteStorageObjectsRetry ¶
func WriteStorageObjectsRetry(ctx context.Context, pool *pgxpool.Pool, reqs []ReadRequest, updateFn func([]*StorageObject) ([]*StorageObject, error), maxRetries int) ([]*StorageObjectAck, error)
WriteStorageObjectsRetry reads, applies updateFn, and retries on version reject.
func WriteStorageObjectsTx ¶
func WriteStorageObjectsTx(ctx context.Context, tx pgx.Tx, authoritative bool, objects []*StorageObject) ([]*StorageObjectAck, error)
WriteStorageObjectsTx writes storage objects inside an existing transaction.
Source Files