indexer

Documentation

Index

• Constants
• type CrawlDispatcher
  • func NewCrawlDispatcher(repoFetcher CrawlRepoFetcher, concurrency int, log *slog.Logger) (*CrawlDispatcher, error)
  • func (c *CrawlDispatcher) AddToCatchupQueue(ctx context.Context, host *models.PDS, u *models.ActorInfo, ...) error
  • func (c *CrawlDispatcher) CatchupRepoGaugePoller()
  • func (c *CrawlDispatcher) Crawl(ctx context.Context, ai *models.ActorInfo) error
  • func (c *CrawlDispatcher) RepoInSlowPath(ctx context.Context, uid models.Uid) bool
  • func (c *CrawlDispatcher) Run()
  • func (c *CrawlDispatcher) Shutdown()
• type CrawlRepoFetcher
• type DidResolver
• type Indexer
  • func NewIndexer(db *gorm.DB, notifman notifs.NotificationManager, evtman *events.EventManager, ...) (*Indexer, error)
  • func (ix *Indexer) DidForUser(ctx context.Context, uid models.Uid) (string, error)
  • func (ix *Indexer) GetPost(ctx context.Context, uri string) (*models.FeedPost, error)
  • func (ix *Indexer) GetPostOrMissing(ctx context.Context, uri string) (*models.FeedPost, error)
  • func (ix *Indexer) GetUserOrMissing(ctx context.Context, did string) (*models.ActorInfo, error)
  • func (ix *Indexer) HandleRepoEvent(ctx context.Context, evt *repomgr.RepoEvent) error
  • func (ix *Indexer) LookupUser(ctx context.Context, id models.Uid) (*models.ActorInfo, error)
  • func (ix *Indexer) LookupUserByDid(ctx context.Context, did string) (*models.ActorInfo, error)
  • func (ix *Indexer) LookupUserByHandle(ctx context.Context, handle string) (*models.ActorInfo, error)
  • func (ix *Indexer) Shutdown()
• type KeyManager
  • func NewKeyManager(didr DidResolver, k *did.PrivKey) *KeyManager
  • func (km *KeyManager) SignForUser(ctx context.Context, did string, msg []byte) ([]byte, error)
  • func (km *KeyManager) VerifyUserSignature(ctx context.Context, did string, sig []byte, msg []byte) error
• type RepoFetcher
  • func NewRepoFetcher(db *gorm.DB, rm *repomgr.RepoManager, maxConcurrency int) *RepoFetcher
  • func (rf *RepoFetcher) FetchAndIndexRepo(ctx context.Context, job *crawlWork) error
  • func (rf *RepoFetcher) GetLimiter(pdsID uint) *rate.Limiter
  • func (rf *RepoFetcher) GetOrCreateLimiter(pdsID uint, pdsrate float64) *rate.Limiter
  • func (rf *RepoFetcher) SetLimiter(pdsID uint, lim *rate.Limiter)

Constants

const MaxEventSliceLength = 1000000

const MaxOpsSliceLength = 200

Types

type CrawlDispatcher

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

func NewCrawlDispatcher

func NewCrawlDispatcher(repoFetcher CrawlRepoFetcher, concurrency int, log *slog.Logger) (*CrawlDispatcher, error)

func (*CrawlDispatcher) AddToCatchupQueue

func (c *CrawlDispatcher) AddToCatchupQueue(ctx context.Context, host *models.PDS, u *models.ActorInfo, evt *comatproto.SyncSubscribeRepos_Commit) error

func (*CrawlDispatcher) CatchupRepoGaugePoller

func (c *CrawlDispatcher) CatchupRepoGaugePoller()

func (*CrawlDispatcher) Crawl

func (c *CrawlDispatcher) Crawl(ctx context.Context, ai *models.ActorInfo) error

func (*CrawlDispatcher) RepoInSlowPath

func (c *CrawlDispatcher) RepoInSlowPath(ctx context.Context, uid models.Uid) bool

func (*CrawlDispatcher) Run

func (c *CrawlDispatcher) Run()

func (*CrawlDispatcher) Shutdown

func (c *CrawlDispatcher) Shutdown()

type CrawlRepoFetcher

type CrawlRepoFetcher interface {
	FetchAndIndexRepo(ctx context.Context, job *crawlWork) error
}

this is what we need of RepoFetcher

type DidResolver

type DidResolver interface {
	GetDocument(ctx context.Context, didstr string) (*did.Document, error)
}

type Indexer

type Indexer struct {
	Crawler *CrawlDispatcher

	SendRemoteFollow       func(context.Context, string, uint) error
	CreateExternalUser     func(context.Context, string) (*models.ActorInfo, error)
	ApplyPDSClientSettings func(*xrpc.Client)
	// contains filtered or unexported fields
}

func NewIndexer

func NewIndexer(db *gorm.DB, notifman notifs.NotificationManager, evtman *events.EventManager, didr did.Resolver, fetcher *RepoFetcher, crawl, aggregate, spider bool) (*Indexer, error)

func (*Indexer) DidForUser

func (ix *Indexer) DidForUser(ctx context.Context, uid models.Uid) (string, error)

func (*Indexer) GetPost

func (ix *Indexer) GetPost(ctx context.Context, uri string) (*models.FeedPost, error)

func (*Indexer) GetPostOrMissing

func (ix *Indexer) GetPostOrMissing(ctx context.Context, uri string) (*models.FeedPost, error)

func (*Indexer) GetUserOrMissing

func (ix *Indexer) GetUserOrMissing(ctx context.Context, did string) (*models.ActorInfo, error)

func (*Indexer) HandleRepoEvent

func (ix *Indexer) HandleRepoEvent(ctx context.Context, evt *repomgr.RepoEvent) error

func (*Indexer) LookupUser

func (ix *Indexer) LookupUser(ctx context.Context, id models.Uid) (*models.ActorInfo, error)

func (*Indexer) LookupUserByDid

func (ix *Indexer) LookupUserByDid(ctx context.Context, did string) (*models.ActorInfo, error)

func (*Indexer) LookupUserByHandle

func (ix *Indexer) LookupUserByHandle(ctx context.Context, handle string) (*models.ActorInfo, error)

func (*Indexer) Shutdown

func (ix *Indexer) Shutdown()

type KeyManager

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

func NewKeyManager

func NewKeyManager(didr DidResolver, k *did.PrivKey) *KeyManager

func (*KeyManager) SignForUser

func (km *KeyManager) SignForUser(ctx context.Context, did string, msg []byte) ([]byte, error)

func (*KeyManager) VerifyUserSignature

func (km *KeyManager) VerifyUserSignature(ctx context.Context, did string, sig []byte, msg []byte) error

type RepoFetcher

type RepoFetcher struct {
	Limiters map[uint]*rate.Limiter
	LimitMux sync.RWMutex

	MaxConcurrency int

	ApplyPDSClientSettings func(*xrpc.Client)
	// contains filtered or unexported fields
}

func NewRepoFetcher

func NewRepoFetcher(db *gorm.DB, rm *repomgr.RepoManager, maxConcurrency int) *RepoFetcher

func (*RepoFetcher) FetchAndIndexRepo

func (rf *RepoFetcher) FetchAndIndexRepo(ctx context.Context, job *crawlWork) error

TODO: since this function is the only place we depend on the repomanager, i wonder if this should be wired some other way?

func (*RepoFetcher) GetLimiter

func (rf *RepoFetcher) GetLimiter(pdsID uint) *rate.Limiter

func (*RepoFetcher) GetOrCreateLimiter

func (rf *RepoFetcher) GetOrCreateLimiter(pdsID uint, pdsrate float64) *rate.Limiter

func (*RepoFetcher) SetLimiter

func (rf *RepoFetcher) SetLimiter(pdsID uint, lim *rate.Limiter)