ingester

Documentation

Index

• type ActorCache
  • func NewActorCache(log *zap.Logger, store *store.PGXStore) *ActorCache
  • func (crc *ActorCache) CreatePendingCandidateActor(ctx context.Context, did string) (err error)
  • func (crc *ActorCache) GetByDID(did string) *v1.Actor
  • func (crc *ActorCache) Start(ctx context.Context) error
  • func (crc *ActorCache) Sync(ctx context.Context) error
• type FirehoseIngester
  • func NewFirehoseIngester(log *zap.Logger, store *store.PGXStore, crc *ActorCache, bgsHost string) *FirehoseIngester
  • func (fi *FirehoseIngester) Start(ctx context.Context) error

Types

type ActorCache

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

ActorCache holds a view of the candidate actors from the database, refreshing itself every minute. It's designed to be safely called concurrently. This prevents us needing to hit the database for every event which would produce significant load on the db and also increase the amount of time it takes to handle an event we aren't interested in. The only downside to this approach is that it takes up to a minute for new candidate repositories to be monitored.

TODO: Move this to the store as a wrapper around *store.PGXStore ?

func NewActorCache

func NewActorCache(
	log *zap.Logger,
	store *store.PGXStore,
) *ActorCache

func (*ActorCache) CreatePendingCandidateActor

func (crc *ActorCache) CreatePendingCandidateActor(ctx context.Context, did string) (err error)

func (*ActorCache) GetByDID

func (crc *ActorCache) GetByDID(
	did string,
) *v1.Actor

func (*ActorCache) Start

func (crc *ActorCache) Start(ctx context.Context) error

func (*ActorCache) Sync

func (crc *ActorCache) Sync(ctx context.Context) error

type FirehoseIngester

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

func NewFirehoseIngester

func NewFirehoseIngester(
	log *zap.Logger, store *store.PGXStore, crc *ActorCache, bgsHost string,
) *FirehoseIngester

func (*FirehoseIngester) Start

func (fi *FirehoseIngester) Start(ctx context.Context) error