consumer

Documentation

Overview

Code for consuming from atproto firehose and ozone event stream, pushing events in to automod engine.

Index

• type FirehoseConsumer
  • func (fc *FirehoseConsumer) HandleRepoCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error
  • func (fc *FirehoseConsumer) PersistCursor(ctx context.Context) error
  • func (fc *FirehoseConsumer) ReadLastCursor(ctx context.Context) (int64, error)
  • func (fc *FirehoseConsumer) Run(ctx context.Context) error
  • func (fc *FirehoseConsumer) RunPersistCursor(ctx context.Context) error
• type OzoneConsumer
  • func (oc *OzoneConsumer) HandleOzoneEvent(ctx context.Context, eventView *toolsozone.ModerationDefs_ModEventView) error
  • func (oc *OzoneConsumer) PersistCursor(ctx context.Context) error
  • func (oc *OzoneConsumer) ReadLastCursor(ctx context.Context) (string, error)
  • func (oc *OzoneConsumer) Run(ctx context.Context) error
  • func (oc *OzoneConsumer) RunPersistCursor(ctx context.Context) error

Types

type FirehoseConsumer

type FirehoseConsumer struct {
	Parallelism int
	Logger      *slog.Logger
	RedisClient *redis.Client
	Engine      *automod.Engine
	Host        string
	// contains filtered or unexported fields
}

func (*FirehoseConsumer) HandleRepoCommit

func (fc *FirehoseConsumer) HandleRepoCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error

NOTE: for now, this function basically never errors, just logs and returns nil. Should think through error processing better.

func (*FirehoseConsumer) PersistCursor

func (fc *FirehoseConsumer) PersistCursor(ctx context.Context) error

func (*FirehoseConsumer) ReadLastCursor

func (fc *FirehoseConsumer) ReadLastCursor(ctx context.Context) (int64, error)

func (*FirehoseConsumer) Run

func (fc *FirehoseConsumer) Run(ctx context.Context) error

func (*FirehoseConsumer) RunPersistCursor

func (fc *FirehoseConsumer) RunPersistCursor(ctx context.Context) error

this method runs in a loop, persisting the current cursor state every 5 seconds

type OzoneConsumer

type OzoneConsumer struct {
	Logger      *slog.Logger
	RedisClient *redis.Client
	OzoneClient *xrpc.Client
	Engine      *automod.Engine
	// contains filtered or unexported fields
}

func (*OzoneConsumer) HandleOzoneEvent

func (oc *OzoneConsumer) HandleOzoneEvent(ctx context.Context, eventView *toolsozone.ModerationDefs_ModEventView) error

func (*OzoneConsumer) PersistCursor

func (oc *OzoneConsumer) PersistCursor(ctx context.Context) error

func (*OzoneConsumer) ReadLastCursor

func (oc *OzoneConsumer) ReadLastCursor(ctx context.Context) (string, error)

func (*OzoneConsumer) Run

func (oc *OzoneConsumer) Run(ctx context.Context) error

func (*OzoneConsumer) RunPersistCursor

func (oc *OzoneConsumer) RunPersistCursor(ctx context.Context) error

this method runs in a loop, persisting the current cursor state every 5 seconds