indexer

package

v0.0.0-...-216dd84 Latest Latest Go to latest Published: Jan 19, 2026 License: MIT Imports: 16 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 ¶

func StripCIDs(data []byte) ([]byte, error)
type Indexer
- func NewIndexer(ctx context.Context, logger *slog.Logger, redisClient *redis.Client, ...) (*Indexer, error)
type Progress
- func (p *Progress) Get() (int64, time.Time)
- func (p *Progress) Update(seq int64, processedAt time.Time)
type URI
- func ParseURI(uri string) (*URI, error)

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func StripCIDs ¶

func StripCIDs(data []byte) ([]byte, error)

StripCIDs removes StrongRef CID fields from a JSON record. It walks the JSON tree recursively and removes "cid" fields (used in likes, reposts, replies). BlobRef $link fields are preserved. Returns the modified JSON as a byte slice, or error if parsing fails.

Types ¶

type Indexer ¶

type Indexer struct {
	SocketURL string
	Progress  *Progress
	Logger    *slog.Logger

	RedisClient *redis.Client
	ProgressKey string

	Store         *store.Store
	BatchInserter *store.BatchInserter
}

Indexer is the indexer for the firehose

func NewIndexer ¶

func NewIndexer(
	ctx context.Context,
	logger *slog.Logger,
	redisClient *redis.Client,
	redisPrefix string,
	chStore *store.Store,
	socketURL string,
) (*Indexer, error)

NewIndexer creates a new indexer

func (*Indexer) HandleCreateRecord ¶

func (i *Indexer) HandleCreateRecord(
	ctx context.Context,
	repo string,
	collection string,
	rkey string,
	rec json.RawMessage,
	eventTime time.Time,
	timeUS int64,
) error

HandleCreateRecord handles a create record event

func (*Indexer) HandleDeleteRecord ¶

func (i *Indexer) HandleDeleteRecord(
	ctx context.Context,
	repo string,
	collection string,
	rkey string,
	eventTime time.Time,
	timeUS int64,
) error

HandleDeleteRecord handles a delete record event

func (*Indexer) HandleUpdateRecord ¶

func (i *Indexer) HandleUpdateRecord(
	ctx context.Context,
	repo string,
	collection string,
	rkey string,
	rec json.RawMessage,
	eventTime time.Time,
	timeUS int64,
) error

HandleUpdateRecord handles an update record event

func (*Indexer) OnCommit ¶

func (i *Indexer) OnCommit(ctx context.Context, evt *models.Event) error

OnCommit handles a repo commit event from the firehose

func (*Indexer) OnEvent ¶

func (i *Indexer) OnEvent(ctx context.Context, evt *models.Event) error

OnEvent handles a stream event from the Jetstream firehose

func (*Indexer) ReadCursor ¶

func (i *Indexer) ReadCursor(ctx context.Context) error

ReadCursor reads the cursor from redis

func (*Indexer) Shutdown ¶

func (i *Indexer) Shutdown() error

Shutdown shuts down the indexer

func (*Indexer) WriteCursor ¶

func (i *Indexer) WriteCursor(ctx context.Context) error

WriteCursor writes the cursor to redis

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 indexer

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
	Collection string
	RKey       string
}

URI represents a parsed AT URI

func ParseURI ¶

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

ParseURI parses an AT URI into its components

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
store

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