consumer

package

v0.0.0-...-ee04f8c Latest Latest Go to latest Published: Dec 17, 2025 License: MIT Imports: 26 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/jazware/bsky-experiments

Links

Open Source Insights

Documentation ¶

Index ¶

type Consumer
- func NewConsumer(ctx context.Context, logger *zap.SugaredLogger, redisClient *redis.Client, ...) (*Consumer, error)
type Delete
type Progress
- func (p *Progress) Get() (int64, time.Time)
- func (p *Progress) Update(seq int64, processedAt time.Time)
type URI
- func GetURI(uri string) (*URI, error)

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type Consumer ¶

type Consumer struct {
	SocketURL string
	Progress  *Progress
	Logger    *zap.SugaredLogger

	RedisClient *redis.Client
	ProgressKey string

	Store *store.Store
	// contains filtered or unexported fields
}

Consumer is the consumer of the firehose

func NewConsumer ¶

func NewConsumer(
	ctx context.Context,
	logger *zap.SugaredLogger,
	redisClient *redis.Client,
	redisPrefix string,
	store *store.Store,
	socketURL string,
	graphdRoot string,
	shardDBNodes []string,
) (*Consumer, error)

NewConsumer creates a new consumer

func (*Consumer) HandleCreatePost ¶

func (c *Consumer) HandleCreatePost(ctx context.Context, repo, rkey string, indexedAt time.Time, rec *appbsky.FeedPost) error

func (*Consumer) HandleCreateRecord ¶

func (c *Consumer) HandleCreateRecord(
	ctx context.Context,
	actorID int64,
	repo string,
	collection string,
	rkey string,
	rec json.RawMessage,
) (*time.Time, error)

HandleCreateRecord handles a create record event from the firehose

func (*Consumer) HandleDeletePost ¶

func (c *Consumer) HandleDeletePost(ctx context.Context, repo, rkey string) error

func (*Consumer) HandleDeleteRecord ¶

func (c *Consumer) HandleDeleteRecord(
	ctx context.Context,
	actorID int64,
	repo string,
	collection string,
	rkey string,
) error

HandleDeleteRecord handles a delete record event from the firehose

func (*Consumer) OnCommit ¶

func (c *Consumer) OnCommit(ctx context.Context, evt *models.Event) error

HandleRepoCommit handles a repo commit event from the firehose and processes the records

func (*Consumer) OnEvent ¶

func (c *Consumer) OnEvent(ctx context.Context, evt *models.Event) error

OnEvent handles a stream event from the Jetstream firehose

func (*Consumer) ReadCursor ¶

func (c *Consumer) ReadCursor(ctx context.Context) error

ReadCursor reads the cursor from redis

func (*Consumer) Shutdown ¶

func (c *Consumer) Shutdown() error

func (*Consumer) TrimRecentPosts ¶

func (c *Consumer) TrimRecentPosts(ctx context.Context, maxAge time.Duration) error

TrimRecentPosts trims the recent posts from the recent_posts table and the active posters from redis

func (*Consumer) WriteCursor ¶

func (c *Consumer) WriteCursor(ctx context.Context) error

WriteCursor writes the cursor to redis

type Delete ¶

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

type Progress ¶

type Progress struct {
	LastSeq            int64     `json:"last_seq"`
	LastSeqProcessedAt time.Time `json:"last_seq_processed_at"`
	// contains filtered or unexported fields
}

Progress is the cursor for the consumer

func (*Progress) Get ¶

func (p *Progress) Get() (int64, time.Time)

func (*Progress) Update ¶

func (p *Progress) Update(seq int64, processedAt time.Time)

type URI ¶

type URI struct {
	Did        string
	RKey       string
	Collection string
}

func GetURI ¶

func GetURI(uri string) (*URI, error)

URI: at://{did}/{namespace}/{rkey}

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
store
store_queries

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL