server

package
v1.12.0 Latest Latest
Warning

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

 Go to latest Published: Nov 5, 2024 License: Apache-2.0 Imports: 38 Imported by: 0

Documentation

Overview

The events are used for internal events to trigger an action when data is stored in a dataset It can possibly be used for other events as well, if we have any

Index

Constants

View Source 
const (
	URIToIDIndexID         uint16 = 0
	EntityIDToJSONIndexID  uint16 = 1
	IncomingRefIndex       uint16 = 2
	OutgoingRefIndex       uint16 = 3
	DatasetEntityChangeLog uint16 = 4
	SysDatasetsID          uint16 = 5
	SysJobsID              uint16 = 6
	SysDatasetsSequences   uint16 = 7
	DatasetLatestEntities  uint16 = 8
	IDToURIIndexID         uint16 = 9

	StoreMetaIndex     CollectionIndex = 10
	NamespacesIndex    CollectionIndex = 11
	JobResultIndex     CollectionIndex = 12
	JobDataIndex       CollectionIndex = 13
	JobConfigIndex     CollectionIndex = 14
	ContentIndex       CollectionIndex = 15
	StoreNextDatasetID CollectionIndex = 16
	LoginProviderIndex CollectionIndex = 17
	JobMetaIndex       CollectionIndex = 18
)
View Source 
const (
	RdfNamespaceExpansion = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"

	// rdf and rdfs core uris
	RdfTypeURI string = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"

	RdfsClassURI string = "http://www.w3.org/2000/01/rdf-schema#Class"
	RdfsLabelURI string = "http://www.w3.org/2000/01/rdf-schema#label"
)
View Source 
const StorageIDFileName = "DATAHUB_BACKUPID"

Variables

View Source 
var (
	JobResultIndexBytes     = uint16ToBytes(JobResultIndex)
	JobConfigsIndexBytes    = uint16ToBytes(JobConfigIndex)
	JobMetaIndexBytes       = uint16ToBytes(JobMetaIndex)
	ContentIndexBytes       = uint16ToBytes(ContentIndex)
	StoreNextDatasetIDBytes = uint16ToBytes(StoreNextDatasetID)
	LoginProviderIndexBytes = uint16ToBytes(LoginProviderIndex)
)
View Source 
var (
	AttemptStoreEntitiesErr = func(detail error) error { return fmt.Errorf("failed when attempting to store entities: %w", detail) }
	SinceParseErr           = func(detail error) error { return fmt.Errorf("since should be an integer number: %w", detail) }
	HTTPBodyMissingErr      = func(detail error) error { return fmt.Errorf("body is missing or could not read: %w", detail) }
	HTTPJobParsingErr       = func(detail error) error { return fmt.Errorf("failed at parsing the job definition: %w", detail) }
	HTTPJobSchedulingErr    = func(detail error) error { return fmt.Errorf("failed at scheduling the job definition: %w", detail) }
	HTTPJsonParsingErr      = func(detail error) error { return fmt.Errorf("failed parsing the json body: %w", detail) }
	HTTPContentStoreErr     = func(detail error) error { return fmt.Errorf("failed updating the content: %w", detail) }
	HTTPQueryParamErr       = func(detail error) error {
		return fmt.Errorf("one or more of the query parameters failed its validation: %w", detail)
	}
	HTTPGenericErr  = func(detail error) error { return fmt.Errorf("internal failure: %w", detail) }
	HTTPFullsyncErr = func(detail error) error {
		return fmt.Errorf("an error occured trying to start or update a full sync: %w", detail)
	}
)

we define these errors to prevent leaking of internal details on the api

Functions

func BytesToPrivateKey 

func BytesToPrivateKey(priv []byte) *rsa.PrivateKey

BytesToPrivateKey bytes to private key

func BytesToPublicKey 

func BytesToPublicKey(pub []byte) *rsa.PublicKey

BytesToPublicKey bytes to public key

func DecryptWithPrivateKey 

func DecryptWithPrivateKey(ciphertext []byte, priv *rsa.PrivateKey) []byte

DecryptWithPrivateKey decrypts data with private key

func EncryptWithPublicKey 

func EncryptWithPublicKey(msg []byte, pub *rsa.PublicKey) []byte

EncryptWithPublicKey encrypts data with public key

func GenerateKeyPair 

func GenerateKeyPair(bits int) (*rsa.PrivateKey, *rsa.PublicKey)

GenerateKeyPair generates a new key pair

func IsEntityEqual added in v1.9.2 

func IsEntityEqual(prevJson []byte, thisJson []byte, prevEntity *Entity, thisEntity *Entity) bool

func IsEntityEqualOld added in v1.9.2 

func IsEntityEqualOld(prevJson []byte, thisJson []byte, prevEntity *Entity, thisEntity *Entity) bool

func PrivateKeyToBytes 

func PrivateKeyToBytes(priv *rsa.PrivateKey) []byte

PrivateKeyToBytes private key to bytes

func PublicKeyToBytes 

func PublicKeyToBytes(pub *rsa.PublicKey) []byte

PublicKeyToBytes public key to bytes

func ToLegacyQueryResult 

func ToLegacyQueryResult(res RelatedEntitiesQueryResult) [][]any

func URLJoin 

func URLJoin(baseURL string, elem ...string) (result string, err error)

Types

type BackupManager 

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

func NewBackupManager 

func NewBackupManager(store *Store, env *conf.Config) (*BackupManager, error)

func (*BackupManager) DoNativeBackup 

func (backupManager *BackupManager) DoNativeBackup() error

func (*BackupManager) DoRsyncBackup 

func (backupManager *BackupManager) DoRsyncBackup() error

func (*BackupManager) LoadLastID 

func (backupManager *BackupManager) LoadLastID() (uint64, error)

func (*BackupManager) Run 

func (backupManager *BackupManager) Run()

This is the function called by the cron job scheduler

func (*BackupManager) StoreLastID 

func (backupManager *BackupManager) StoreLastID() error

type BadgerAccess 

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

BadgerAccess implements service/store.BadgerStore and bridges badger access without cyclic dependencies

func NewBadgerAccess 

func NewBadgerAccess(s *Store, dsm *DsManager) BadgerAccess

func (BadgerAccess) GetDB 

func (b BadgerAccess) GetDB() *badger.DB

func (BadgerAccess) IsDatasetDeleted 

func (b BadgerAccess) IsDatasetDeleted(datasetID types.InternalDatasetID) bool

func (BadgerAccess) LookupDatasetID 

func (b BadgerAccess) LookupDatasetID(datasetName string) (types.InternalDatasetID, bool)

func (BadgerAccess) LookupDatasetIDs 

func (b BadgerAccess) LookupDatasetIDs(datasetNames []string) []types.InternalDatasetID

func (BadgerAccess) LookupDatasetName 

func (b BadgerAccess) LookupDatasetName(internalDatasetID types.InternalDatasetID) (string, bool)

func (BadgerAccess) LookupExpansionPrefix 

func (b BadgerAccess) LookupExpansionPrefix(namespaceURI types.URI) (types.Prefix, error)

func (BadgerAccess) LookupNamespaceExpansion 

func (b BadgerAccess) LookupNamespaceExpansion(prefix types.Prefix) (types.URI, error)

type BadgerLogger 

type BadgerLogger struct {
	Logger *zap.SugaredLogger
}

func (BadgerLogger) Debugf 

func (bl BadgerLogger) Debugf(format string, v ...interface{})

func (BadgerLogger) Errorf 

func (bl BadgerLogger) Errorf(format string, v ...interface{})

func (BadgerLogger) Infof 

func (bl BadgerLogger) Infof(format string, v ...interface{})

func (BadgerLogger) Warningf 

func (bl BadgerLogger) Warningf(format string, v ...interface{})

type Changes 

type Changes struct {
	Context   *Context
	Entities  []*Entity
	NextToken uint64
}

Changes Object

func NewChanges 

func NewChanges() *Changes

type CollectionIndex 

type CollectionIndex uint16

type Context 

type Context struct {
	ID         string            `json:"id"`
	Namespaces map[string]string `json:"namespaces"`
}

type CreateDatasetConfig 

type CreateDatasetConfig struct {
	ProxyDatasetConfig   *ProxyDatasetConfig   `json:"ProxyDatasetConfig"`
	VirtualDatasetConfig *VirtualDatasetConfig `json:"VirtualDatasetConfig"`
	PublicNamespaces     []string              `json:"publicNamespaces"`
}

type Dataset 

type Dataset struct {
	ID                string `json:"id"`
	InternalID        uint32 `json:"internalId"`
	SubjectIdentifier string `json:"subjectIdentifier"`

	WriteLock sync.Mutex

	PublicNamespaces []string `json:"publicNamespaces"`

	ProxyConfig          *ProxyDatasetConfig   `json:"proxyConfig"`
	VirtualDatasetConfig *VirtualDatasetConfig `json:"virtualDatasetConfig"`
	// contains filtered or unexported fields
}

Dataset data structure

func NewDataset 

func NewDataset(store *Store, id string, internalID uint32, subjectIdentifier string) *Dataset

NewDataset Create a new dataset from the params provided

func (*Dataset) AsProxy 

func (ds *Dataset) AsProxy(auth func(req *http.Request)) *ProxyDataset

func (*Dataset) AsVirtualDataset added in v1.9.0 

func (ds *Dataset) AsVirtualDataset(
	datasetManager *DsManager,
	wrap func(d *VirtualDataset, params map[string]any, since string, f func(entity *Entity) error) (string, error),
) *VirtualDataset

func (*Dataset) CompleteFullSync 

func (ds *Dataset) CompleteFullSync(ctx context.Context) error

CompleteFullSync Full sync completed - mark unseen entities as deleted

func (*Dataset) FullSyncStarted 

func (ds *Dataset) FullSyncStarted() bool

func (*Dataset) GetChanges 

func (ds *Dataset) GetChanges(since uint64, count int, latestOnly bool) (*Changes, error)

func (*Dataset) GetChangesWatermark 

func (ds *Dataset) GetChangesWatermark() (uint64, error)

func (*Dataset) GetContext 

func (ds *Dataset) GetContext() *Context

func (*Dataset) GetEntities 

func (ds *Dataset) GetEntities(from string, count int) (*EntitiesResult, error)

GetEntities returns a batch of entities

func (*Dataset) IsProxy 

func (ds *Dataset) IsProxy() bool

func (*Dataset) IsVirtual added in v1.9.0 

func (ds *Dataset) IsVirtual() bool

func (*Dataset) MapEntities 

func (ds *Dataset) MapEntities(from string, count int, processEntity func(entity *Entity) error) (string, error)

MapEntities applies a function to all entities in the dataset returns the id of the last entity so that it can be used as a continuation token

func (*Dataset) MapEntitiesRaw 

func (ds *Dataset) MapEntitiesRaw(from string, count int, processEntity func(json []byte) error) (string, error)

MapEntities applies a function to all entities in the dataset. the entities are provided as raw json bytes returns the id of the last entity so that it can be used as a continuation token

func (*Dataset) ProcessChanges 

func (ds *Dataset) ProcessChanges(
	since uint64,
	count int,
	latestOnly bool,
	processChangedEntity func(entity *Entity),
) (uint64, error)

func (*Dataset) ProcessChangesRaw 

func (ds *Dataset) ProcessChangesRaw(
	since uint64,
	limit int,
	latestOnly bool,
	processChangedEntity func(entityJson []byte) error,
) (uint64, error)

func (*Dataset) RefreshFullSyncLease 

func (ds *Dataset) RefreshFullSyncLease(fullSyncID string) error

func (*Dataset) ReleaseFullSyncLease 

func (ds *Dataset) ReleaseFullSyncLease(fullSyncID string) error

func (*Dataset) StartFullSync 

func (ds *Dataset) StartFullSync() error

StartFullSync Indicates that a full sync is starting

func (*Dataset) StartFullSyncWithLease 

func (ds *Dataset) StartFullSyncWithLease(fullSyncID string) error

func (*Dataset) StoreEntities 

func (ds *Dataset) StoreEntities(entities []*Entity) (Error error)

func (*Dataset) StoreEntitiesWithTransaction 

func (ds *Dataset) StoreEntitiesWithTransaction(
	entities []*Entity,
	txnTime int64,
	txn *badger.Txn,
) (newitems int64, Error error)

StoreEntities

type DatasetName 

type DatasetName struct {
	Name string `json:"Name"`
}

type DsManager 

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

func NewDsManager 

func NewDsManager(env *conf.Config, store *Store, eb EventBus) *DsManager

func (*DsManager) CreateDataset 

func (dsm *DsManager) CreateDataset(name string, createDatasetConfig *CreateDatasetConfig) (*Dataset, error)

func (*DsManager) DeleteDataset 

func (dsm *DsManager) DeleteDataset(name string) error

DeleteDataset deletes dataset if it exists

func (*DsManager) GetDataset 

func (dsm *DsManager) GetDataset(id string) *Dataset

func (*DsManager) GetDatasetDetails 

func (dsm *DsManager) GetDatasetDetails(name string) (*Entity, bool, error)

func (*DsManager) GetDatasetNames 

func (dsm *DsManager) GetDatasetNames() []DatasetName

GetDatasetNames returns a list of the dataset names

func (*DsManager) IsDataset 

func (dsm *DsManager) IsDataset(name string) bool

func (*DsManager) NewDatasetEntity 

func (dsm *DsManager) NewDatasetEntity(
	name string,
	proxyDatasetConfig *ProxyDatasetConfig,
	virtualDatasetConfig *VirtualDatasetConfig,
	publicNamespaces []string,
) *Entity

func (*DsManager) UpdateDataset 

func (dsm *DsManager) UpdateDataset(name string, config *UpdateDatasetConfig) (*Dataset, error)

type DsNsInfo 

type DsNsInfo struct {
	DatasetPrefix       string
	PublicNamespacesKey string
	NameKey             string
	ItemsKey            string
}

type EntitiesResult 

type EntitiesResult struct {
	Context           *Context
	Entities          []*Entity
	ContinuationToken string
}

type Entity 

type Entity struct {
	References map[string]interface{} `json:"refs"`
	Properties map[string]interface{} `json:"props"`
	ID         string                 `json:"id,omitempty"`
	InternalID uint64                 `json:"internalId,omitempty"`
	Recorded   uint64                 `json:"recorded,omitempty"`
	IsDeleted  bool                   `json:"deleted,omitempty"`
}

Entity data structure

func NewEntity 

func NewEntity(ID string, internalID uint64) *Entity

NewEntity Create a new entity with global uri and internal resource id

func NewEntityFromMap 

func NewEntityFromMap(data map[string]interface{}) *Entity

func (*Entity) ExpandIdentifiers 

func (e *Entity) ExpandIdentifiers(store *Store) error

func (*Entity) GetName 

func (e *Entity) GetName() string

GetName returns a human readable Name

func (*Entity) GetProperty 

func (e *Entity) GetProperty(propName string) interface{}

GetProperty returns the value of the named property as an interface

func (*Entity) GetStringProperty 

func (e *Entity) GetStringProperty(propName string) string

GetStringProperty returns the string value of the requested property

type EntityStreamParser 

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

func NewEntityStreamParser 

func NewEntityStreamParser(store *Store) *EntityStreamParser

func (*EntityStreamParser) ParseStream 

func (esp *EntityStreamParser) ParseStream(reader io.Reader, emitEntity func(*Entity) error) error

func (*EntityStreamParser) ParseTransaction 

func (esp *EntityStreamParser) ParseTransaction(reader io.Reader) (*Transaction, error)

type EventBus 

type EventBus interface {
	Init(datasets []DatasetName)
	RegisterTopic(ds string)
	UnregisterTopic(ds string)
	SubscribeToDataset(id string, matcher string, f func(e *bus.Event))
	UnsubscribeToDataset(id string)
	Emit(ctx context.Context, topicName string, data interface{})
}

func NewBus 

func NewBus(env *conf.Config) (EventBus, error)

func NoOpBus 

func NoOpBus() EventBus

type GarbageCollector 

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

func NewGarbageCollector 

func NewGarbageCollector(store *Store, env *conf.Config) *GarbageCollector

func (*GarbageCollector) Cleandeleted 

func (garbageCollector *GarbageCollector) Cleandeleted() error

func (*GarbageCollector) GC 

func (garbageCollector *GarbageCollector) GC() error

func (*GarbageCollector) Stop 

func (garbageCollector *GarbageCollector) Stop(ctx context.Context) error

type Instrumented 

type Instrumented interface {
	// contains filtered or unexported methods
}

type InstrumentedTransaction 

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

func InstrumentedTxn 

func InstrumentedTxn(btxn *badger.Txn, store *Store) *InstrumentedTransaction

func (*InstrumentedTransaction) Get 

func (t *InstrumentedTransaction) Get(id []byte) (*badger.Item, error)

func (*InstrumentedTransaction) NewIterator 

func (t *InstrumentedTransaction) NewIterator(options badger.IteratorOptions) *InstumentedIterator

type InstumentedIterator 

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

func (*InstumentedIterator) Close 

func (i *InstumentedIterator) Close()

func (*InstumentedIterator) Item 

func (i *InstumentedIterator) Item() *badger.Item

func (*InstumentedIterator) Next 

func (i *InstumentedIterator) Next()

func (*InstumentedIterator) Rewind 

func (i *InstumentedIterator) Rewind()

func (*InstumentedIterator) Seek 

func (i *InstumentedIterator) Seek(buffer []byte)

func (*InstumentedIterator) ValidForPrefix 

func (i *InstumentedIterator) ValidForPrefix(bytes []byte) bool

type MEventBus 

type MEventBus struct {
	Bus *bus.Bus
	// contains filtered or unexported fields
}

func (*MEventBus) Emit 

func (eb *MEventBus) Emit(ctx context.Context, topicName string, data interface{})

Emit emits an event to the bus

func (*MEventBus) Init 

func (eb *MEventBus) Init(datasets []DatasetName)

Init adds the list of existing datasets that can be subscribed to. Datasets are always prefixed with dataset. to separate them from other possible future events

func (*MEventBus) RegisterTopic 

func (eb *MEventBus) RegisterTopic(ds string)

RegisterTopic registers a topic for publishing. "dataset." is prefixed in front of the topic

func (*MEventBus) SubscribeToDataset 

func (eb *MEventBus) SubscribeToDataset(id string, matcher string, f func(e *bus.Event))

SubscribeToDataset adds a subscription to an already registered topic. The id should be unique, the matcher is a regexp to match against the registered topics, ie: dataset.*, dataset.sdb.*, dataset.sdb.Animal are all valid registrations. f is the func to be called

func (*MEventBus) UnregisterTopic 

func (eb *MEventBus) UnregisterTopic(ds string)

UnregisterTopic removes a topic from subscription

func (*MEventBus) UnsubscribeToDataset 

func (eb *MEventBus) UnsubscribeToDataset(id string)

UnsubscribeToDataset removes a dataset subscription

type MetaContext added in v1.11.0 

type MetaContext struct {
	QueriedDatasets map[uint32]struct{}
	TransactionSink map[string]struct{}
}

func (*MetaContext) Add added in v1.11.0 

func (c *MetaContext) Add(meta *MetaContext)

func (*MetaContext) RegisterQuerySideInput added in v1.11.0 

func (c *MetaContext) RegisterQuerySideInput(datasetid uint32)

func (*MetaContext) RegisterTransactionSink added in v1.11.0 

func (c *MetaContext) RegisterTransactionSink(dataset string)

type NamespaceManager 

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

func NewNamespaceManager 

func NewNamespaceManager(store *Store) *NamespaceManager

func (*NamespaceManager) AssertPrefixMappingForExpansion 

func (namespaceManager *NamespaceManager) AssertPrefixMappingForExpansion(uriExpansion string) (string, error)

func (*NamespaceManager) ExpandCurie 

func (namespaceManager *NamespaceManager) ExpandCurie(curie string) (string, error)

func (*NamespaceManager) GetContext 

func (namespaceManager *NamespaceManager) GetContext(includedNamespaces []string) (context *Context)

GetContext return a context instance 

containing namespace mappings for the given list of namespace prefixes
if the give list of given namespace prefixes is empty, all namespace mappings are returned

func (*NamespaceManager) GetDatasetNamespaceInfo 

func (namespaceManager *NamespaceManager) GetDatasetNamespaceInfo() (DsNsInfo, error)

func (*NamespaceManager) GetPrefixMappingForExpansion 

func (namespaceManager *NamespaceManager) GetPrefixMappingForExpansion(uriExpansion string) (string, error)

func (*NamespaceManager) GetPrefixToExpansionMap 

func (namespaceManager *NamespaceManager) GetPrefixToExpansionMap() (result map[string]string)

type NamespacesState 

type NamespacesState struct {
	PrefixToExpansionMapping map[string]string
	ExpansionToPrefixMapping map[string]string
}

type NoOp 

type NoOp struct{}

func (*NoOp) Emit 

func (eb *NoOp) Emit(ctx context.Context, topicName string, data interface{})

func (*NoOp) Init 

func (eb *NoOp) Init(datasets []DatasetName)

func (*NoOp) RegisterTopic 

func (eb *NoOp) RegisterTopic(ds string)

func (*NoOp) SubscribeToDataset 

func (eb *NoOp) SubscribeToDataset(id string, matcher string, f func(e *bus.Event))

func (*NoOp) UnregisterTopic 

func (eb *NoOp) UnregisterTopic(ds string)

func (*NoOp) UnsubscribeToDataset 

func (eb *NoOp) UnsubscribeToDataset(id string)

type ProxyDataset 

type ProxyDataset struct {
	*ProxyDatasetConfig
	RemoteChangesURL  string
	RemoteEntitiesURL string
	// contains filtered or unexported fields
}

func (*ProxyDataset) ForwardEntities 

func (d *ProxyDataset) ForwardEntities(sourceBody io.ReadCloser, sourceHeader http.Header) error

func (*ProxyDataset) StreamChanges 

func (d *ProxyDataset) StreamChanges(since string, limit int, latestOnly bool, reverse bool, f func(*Entity) error, preStream func() error) (string, error)

StreamChangesRaw stream through the dataset's changes and call `f` for each entity. a `preStream` function can be provided if StreamChanges is used in a web handler. It allows to leave the http response uncommitted until `f` is called, so that an http error handler still can modify status code while the response is uncommitted

func (*ProxyDataset) StreamChangesRaw 

func (d *ProxyDataset) StreamChangesRaw(
	since string,
	limit int,
	latestOnly bool,
	reverse bool,
	f func(jsonData []byte) error,
	preStream func() error,
) (string, error)

StreamChangesRaw stream through the dataset's changes and call `f` for each entity. a `preStream` function can be provided if StreamChangesRaw is used in a web handler. It allows to leave the http response uncommitted until `f` is called, so that an http error handler still can modify status code while the response is uncommitted

func (*ProxyDataset) StreamEntities 

func (d *ProxyDataset) StreamEntities(from string, limit int, f func(*Entity) error, preStream func() error) (string, error)

func (*ProxyDataset) StreamEntitiesRaw 

func (d *ProxyDataset) StreamEntitiesRaw(
	from string,
	limit int,
	f func(jsonData []byte) error,
	preStream func() error,
) (string, error)

type ProxyDatasetConfig 

type ProxyDatasetConfig struct {
	RemoteURL           string `json:"remoteUrl"`
	UpstreamTransform   string `json:"upstreamTransform"`
	DownstreamTransform string `json:"downstreamTransform"`
	AuthProviderName    string `json:"authProviderName"`
	TimeoutSeconds      int    `json:"timeoutSeconds"`
}

type RelatedEntitiesQueryResult 

type RelatedEntitiesQueryResult struct {
	Cont      []*RelatedFrom
	Relations []RelatedEntityResult
}

type RelatedEntitiesResult 

type RelatedEntitiesResult struct {
	Continuation *RelatedFrom
	Relations    []RelatedEntityResult
}

type RelatedEntityResult 

type RelatedEntityResult struct {
	StartURI      string
	PredicateURI  string
	RelatedEntity *Entity
}

type RelatedFrom 

type RelatedFrom struct {
	RelationIndexFromKey []byte
	Predicate            uint64
	Inverse              bool
	Datasets             []uint32
	At                   int64
}

type Statistics added in v1.9.0 

type Statistics struct {
	Store  *Store
	Logger *zap.SugaredLogger
}

func (Statistics) GetStatistics added in v1.9.0 

func (stats Statistics) GetStatistics(writer io.Writer) error

func (Statistics) GetStatisticsForDs added in v1.9.0 

func (stats Statistics) GetStatisticsForDs(datasetName string, writer io.Writer) error

type StorageError 

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

StorageError custom error from storage

func NewStorageError 

func NewStorageError(msg string, innerError error) *StorageError

NewStorageError Creates a new storage error with optional innerError

func (*StorageError) Error 

func (e *StorageError) Error() string

type Store 

type Store struct {
	NamespaceManager *NamespaceManager // namespace manager for consistent short expansions

	SlowLogThreshold time.Duration
	MetaCtx          *MetaContext
	// contains filtered or unexported fields
}

Store data structure

func NewContextualStore added in v1.11.0 

func NewContextualStore(store *Store) *Store

func NewStore 

func NewStore(env *conf.Config, statsdClient statsd.ClientInterface) *Store

NewStore Create a new store

func (*Store) Close 

func (s *Store) Close() error

func (*Store) DatasetsToInternalIDs 

func (s *Store) DatasetsToInternalIDs(datasets []string) []uint32

DatasetsToInternalIDs map dataset IDs (strings) to InternaIDs (uint32)

func (*Store) Delete 

func (s *Store) Delete() error

func (*Store) DeleteObject 

func (s *Store) DeleteObject(collection CollectionIndex, id string) error

DeleteObject is a slightly less to the metal variation of the deleteValue method. You should probably use this instead of deleteValue if you need to delete stuff. It takes a collection and an object id, and attempts to delete it from the store.

func (*Store) ExecuteTransaction 

func (s *Store) ExecuteTransaction(transaction *Transaction) error

func (*Store) ExpandCurie 

func (s *Store) ExpandCurie(curie string) (string, error)

func (*Store) GetEntity 

func (s *Store) GetEntity(uri string, datasets []string, mergePartials bool) (*Entity, error)

func (*Store) GetEntityAtPointInTimeWithInternalID 

func (s *Store) GetEntityAtPointInTimeWithInternalID(
	internalID uint64,
	at int64,
	targetDatasetIds []uint32,
	mergePartials bool,
) (*Entity, error)

func (*Store) GetEntityWithInternalID 

func (s *Store) GetEntityWithInternalID(internalID uint64, targetDatasetIds []uint32, mergePartials bool) (*Entity, error)

func (*Store) GetGlobalContext 

func (s *Store) GetGlobalContext(strict bool) *Context

func (*Store) GetManyRelatedEntities 

func (s *Store) GetManyRelatedEntities(
	startPoints []string,
	predicate string,
	inverse bool,
	datasets []string,
	mergePartials bool,
) ([][]any, error)

Backwards compatibility function

func (*Store) GetManyRelatedEntitiesAtTime 

func (s *Store) GetManyRelatedEntitiesAtTime(from []*RelatedFrom, limit int, mergePartials bool) (RelatedEntitiesQueryResult, error)

func (*Store) GetManyRelatedEntitiesBatch 

func (s *Store) GetManyRelatedEntitiesBatch(
	startPoints []string,
	predicate string,
	inverse bool,
	datasets []string,
	limit int, mergePartials bool,
) (RelatedEntitiesQueryResult, error)

func (*Store) GetNamespacedIdentifier 

func (s *Store) GetNamespacedIdentifier(val string, localNamespaces map[string]string) (string, error)

func (*Store) GetNamespacedIdentifierFromURI 

func (s *Store) GetNamespacedIdentifierFromURI(val string) (string, error)

func (*Store) GetObject 

func (s *Store) GetObject(collection CollectionIndex, id string, obj interface{}) error

func (*Store) GetPredicateID 

func (s *Store) GetPredicateID(predicate string, txn *badger.Txn) (uint64, error)

func (*Store) GetRelatedAtTime 

func (s *Store) GetRelatedAtTime(from *RelatedFrom, limit int) ([]qresult, *RelatedFrom, error)

func (*Store) IsCurie 

func (s *Store) IsCurie(uri string) bool

func (*Store) IterateObjectsRaw 

func (s *Store) IterateObjectsRaw(prefix []byte, processJSON func([]byte) error) error

func (*Store) Open 

func (s *Store) Open() error

Open Opens the store. Must be called before using.

func (*Store) StoreObject 

func (s *Store) StoreObject(collection CollectionIndex, id string, data interface{}) error

func (*Store) ToRelatedFrom 

func (s *Store) ToRelatedFrom(
	startPoints []string,
	predicate string,
	inverse bool,
	datasets []string,
	queryTime int64,
) ([]*RelatedFrom, error)

type Transaction 

type Transaction struct {
	DatasetEntities map[string][]*Entity
	// contains filtered or unexported fields
}

func (*Transaction) AssertDataset added in v1.12.0 

func (t *Transaction) AssertDataset(dataset string)

func (*Transaction) AssertedDatasets added in v1.12.0 

func (t *Transaction) AssertedDatasets() []string

type UpdateDatasetConfig 

type UpdateDatasetConfig struct {
	ID string // update id/name
}

type VirtualDataset added in v1.9.0 

type VirtualDataset struct {
	*VirtualDatasetConfig
	Store     *Store
	DsManager *DsManager
	Logger    *zap.SugaredLogger
	Wrap      func(d *VirtualDataset, params map[string]any, since string, f func(entity *Entity) error) (string, error)
}

func (*VirtualDataset) StreamChanges added in v1.9.0 

func (d *VirtualDataset) StreamChanges(since string, body io.Reader, f func(entity *Entity) error) (string, error)

StreamChanges streams out entities, produced by the configured transform script. it returns a continuation token

type VirtualDatasetConfig added in v1.9.0 

type VirtualDatasetConfig struct {
	Transform string
}

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.