ingester

Documentation

Index

• Constants
• type ActorCache
  • func NewActorCache(log *slog.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 *slog.Logger, store *store.PGXStore, crc *ActorCache, jetstreamURL string) *FirehoseIngester
  • func (fi *FirehoseIngester) Start(ctx context.Context) (err error)

Constants

const DefaultJetstreamURL = "wss://jetstream1.us-east.bsky.network/subscribe"

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 *slog.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 *slog.Logger, store *store.PGXStore, crc *ActorCache, jetstreamURL string,
) *FirehoseIngester

func (*FirehoseIngester) Start

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