commit fe6b736cc66953dc6e620ae935f36c75435b3f38 · edavis.dev/monarch

+134 -11

cmd/monarch/census.go

···

       3
        
       import (

     

       4
        
       	"context"

     

       5
        
       	"log/slog"

     

       0
        
       
     

       0
        
       
     

       6
        
       

     

       7
        
       	"github.com/bluesky-social/indigo/api/atproto"

     

       8
        
       	"github.com/bluesky-social/indigo/backfill"

     

       9
        
       	"github.com/bluesky-social/indigo/xrpc"

     

       10
        
       	"github.com/urfave/cli/v2"

     

       0
        
       
     

       0
        
       
     

       11
        
       )

     

       12
        
       

     

       13
        
       type CensusService struct {

     

       14
       -
       	cursor   *CursorService

     

       15
        
       	backfill *backfill.Backfiller

     

       0
        
       
     

       0
        
       
     

       16
        
       }

     

       17
        
       

     

       18
        
       type jobMaker interface {

     

       19
        
       	GetOrCreateJob(context.Context, string, string) (backfill.Job, error)

     

       20
        
       }

     

       21
        
       

     

       22
       -
       func NewCensusService(cursorSvc *CursorService, backfillSvc *backfill.Backfiller) *CensusService {

     

       23
        
       	return &CensusService{

     

       24
       -
       		cursor:   cursorSvc,

     

       25
        
       		backfill: backfillSvc,

     

       26
        
       	}

     

       27
        
       }

     

       28
        
       

     

       29
       -
       func (cs *CensusService) Start(ctx context.Context, cctx *cli.Context) {

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       30
        
       	xrpcc := &xrpc.Client{

     

       31
       -
       		Host: "https://" + cctx.String("relay-host"),

     

       32
        
       	}

     

       33
        
       

     

       34
        
       	jmstore, ok := cs.backfill.Store.(jobMaker)

     
···

       37
        
       		return

     

       38
        
       	}

     

       39
        
       

     

       40
       -
       	curs, _ := cs.cursor.Get("repos")

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       41
        
       	for {

     

       42
        
       		select {

     

       43
        
       		case <-ctx.Done():

     

       44
       -
       			slog.Info("stopping repo census")

     

       45
        
       			return

     

       46
        
       		default:

     

       47
        
       		}

     

       48
        
       

     

       0
        
       
     

       49
        
       		res, err := atproto.SyncListRepos(ctx, xrpcc, curs, 1000)

     

       50
        
       		if err != nil {

     

       51
       -
       			slog.Error("error listing repos", "err", err)

     

       52
       -
       			return

     

       53
        
       		}

     

       54
        
       

     

       0
        
       
     

       55
        
       		for _, repo := range res.Repos {

     

       56
        
       			_, err := jmstore.GetOrCreateJob(ctx, repo.Did, backfill.StateEnqueued)

     

       57
        
       			if err != nil {

     

       58
       -
       				slog.Error("error adding listed repo to backfiller", "err", err)

     

       0
        
       
     

       0
        
       
     

       59
        
       			}

     

       60
        
       		}

     

       0
        
       
     

       61
        
       

     

       62
        
       		if res.Cursor != nil && *res.Cursor != "" {

     

       63
        
       			curs = *res.Cursor

     

       64
       -
       			cs.cursor.SetReposCursor(curs)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       65
        
       		} else {

     

       66
        
       			break

     

       67
        
       		}

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       68
        
       	}

     

       69
        
       }

···

       3
        
       import (

     

       4
        
       	"context"

     

       5
        
       	"log/slog"

     

       6
       +
       	"sync"

     

       7
       +
       	"time"

     

       8
        
       

     

       9
        
       	"github.com/bluesky-social/indigo/api/atproto"

     

       10
        
       	"github.com/bluesky-social/indigo/backfill"

     

       11
        
       	"github.com/bluesky-social/indigo/xrpc"

     

       12
        
       	"github.com/urfave/cli/v2"

     

       13
       +
       	"golang.org/x/sync/semaphore"

     

       14
       +
       	"gorm.io/gorm"

     

       15
        
       )

     

       16
        
       

     

       17
        
       type CensusService struct {

     

       18
       +
       	store    *gorm.DB

     

       19
        
       	backfill *backfill.Backfiller

     

       20
       +
       

     

       21
       +
       	storeLk sync.Mutex

     

       22
        
       }

     

       23
        
       

     

       24
        
       type jobMaker interface {

     

       25
        
       	GetOrCreateJob(context.Context, string, string) (backfill.Job, error)

     

       26
        
       }

     

       27
        
       

     

       28
       +
       func NewCensusService(store *gorm.DB, backfillSvc *backfill.Backfiller) *CensusService {

     

       29
        
       	return &CensusService{

     

       30
       +
       		store:    store,

     

       31
        
       		backfill: backfillSvc,

     

       32
        
       	}

     

       33
        
       }

     

       34
        
       

     

       35
       +
       type relayCursor struct {

     

       36
       +
       	ID     int `gorm:"primaryKey"`

     

       37
       +
       	Host   string

     

       38
       +
       	Cursor string

     

       39
       +
       }

     

       40
       +
       

     

       41
       +
       type hostCursor struct {

     

       42
       +
       	ID     int `gorm:"primaryKey"`

     

       43
       +
       	Host   string

     

       44
       +
       	Cursor string

     

       45
       +
       }

     

       46
       +
       

     

       47
       +
       // fetch the PDSes known to the relay

     

       48
       +
       func (cs *CensusService) listHosts(ctx context.Context, cctx *cli.Context) {

     

       49
       +
       	relay := cctx.String("relay-host")

     

       50
       +
       	maxCrawlers := cctx.Int64("max-repo-crawlers")

     

       51
       +
       

     

       52
       +
       	xrpcc := &xrpc.Client{

     

       53
       +
       		Host: "https://" + relay,

     

       54
       +
       	}

     

       55
       +
       

     

       56
       +
       	var rcur relayCursor

     

       57
       +
       	if err := cs.store.Where("host = ?", relay).Attrs(relayCursor{

     

       58
       +
       		Host:   relay,

     

       59
       +
       		Cursor: "",

     

       60
       +
       	}).FirstOrCreate(&rcur).Error; err != nil {

     

       61
       +
       		slog.Error("error fetching relay cursor", "err", err)

     

       62
       +
       	}

     

       63
       +
       

     

       64
       +
       	var wg sync.WaitGroup

     

       65
       +
       	sem := semaphore.NewWeighted(maxCrawlers)

     

       66
       +
       

     

       67
       +
       	curs := rcur.Cursor

     

       68
       +
       	for {

     

       69
       +
       		select {

     

       70
       +
       		case <-ctx.Done():

     

       71
       +
       			slog.Info("stopping listHosts", "err", ctx.Err())

     

       72
       +
       			return

     

       73
       +
       		default:

     

       74
       +
       		}

     

       75
       +
       

     

       76
       +
       		slog.Info("listing hosts", "relay", relay, "curs", curs)

     

       77
       +
       		res, err := atproto.SyncListHosts(ctx, xrpcc, curs, 1000)

     

       78
       +
       		if err != nil {

     

       79
       +
       			slog.Error("error obtaining hosts from relay", "err", err)

     

       80
       +
       			continue

     

       81
       +
       		}

     

       82
       +
       

     

       83
       +
       		for _, host := range res.Hosts {

     

       84
       +
       			sem.Acquire(ctx, 1)

     

       85
       +
       			wg.Add(1) // TODO wg.Go

     

       86
       +
       			go func() {

     

       87
       +
       				defer sem.Release(1)

     

       88
       +
       				defer wg.Done()

     

       89
       +
       				slog.Info("adding host", "host", host.Hostname)

     

       90
       +
       				cs.listRepos(ctx, host.Hostname)

     

       91
       +
       			}()

     

       92
       +
       		}

     

       93
       +
       

     

       94
       +
       		if res.Cursor != nil && *res.Cursor != "" {

     

       95
       +
       			curs = *res.Cursor

     

       96
       +
       			if err := cs.store.Model(&rcur).Update("cursor", curs).Error; err != nil {

     

       97
       +
       				slog.Error("error updating cursor for relay", "err", err)

     

       98
       +
       			}

     

       99
       +
       		} else {

     

       100
       +
       			break

     

       101
       +
       		}

     

       102
       +
       	}

     

       103
       +
       

     

       104
       +
       	wg.Wait()

     

       105
       +
       	slog.Info("finished listing hosts", "relay", relay)

     

       106
       +
       }

     

       107
       +
       

     

       108
       +
       // fetch the repos known to the PDS

     

       109
       +
       func (cs *CensusService) listRepos(ctx context.Context, host string) {

     

       110
        
       	xrpcc := &xrpc.Client{

     

       111
       +
       		Host: "https://" + host,

     

       112
        
       	}

     

       113
        
       

     

       114
        
       	jmstore, ok := cs.backfill.Store.(jobMaker)

     
···

       117
        
       		return

     

       118
        
       	}

     

       119
        
       

     

       120
       +
       	cs.storeLk.Lock()

     

       121
       +
       	var hcur hostCursor

     

       122
       +
       	if err := cs.store.Where("host = ?", host).Attrs(hostCursor{

     

       123
       +
       		Host:   host,

     

       124
       +
       		Cursor: "",

     

       125
       +
       	}).FirstOrCreate(&hcur).Error; err != nil {

     

       126
       +
       		slog.Error("error fetching host cursor", "err", err)

     

       127
       +
       	}

     

       128
       +
       	cs.storeLk.Unlock()

     

       129
       +
       

     

       130
       +
       	var added int

     

       131
       +
       	curs := hcur.Cursor

     

       132
        
       	for {

     

       133
        
       		select {

     

       134
        
       		case <-ctx.Done():

     

       135
       +
       			slog.Info("stopping listRepos", "err", ctx.Err())

     

       136
        
       			return

     

       137
        
       		default:

     

       138
        
       		}

     

       139
        
       

     

       140
       +
       		slog.Info("listing repos", "host", host, "curs", curs, "added", added)

     

       141
        
       		res, err := atproto.SyncListRepos(ctx, xrpcc, curs, 1000)

     

       142
        
       		if err != nil {

     

       143
       +
       			slog.Error("error obtaining repos from host", "err", err)

     

       144
       +
       			continue

     

       145
        
       		}

     

       146
        
       

     

       147
       +
       		cs.storeLk.Lock()

     

       148
        
       		for _, repo := range res.Repos {

     

       149
        
       			_, err := jmstore.GetOrCreateJob(ctx, repo.Did, backfill.StateEnqueued)

     

       150
        
       			if err != nil {

     

       151
       +
       				slog.Error("error adding repo to backfiller", "err", err)

     

       152
       +
       			} else {

     

       153
       +
       				added += 1

     

       154
        
       			}

     

       155
        
       		}

     

       156
       +
       		cs.storeLk.Unlock()

     

       157
        
       

     

       158
        
       		if res.Cursor != nil && *res.Cursor != "" {

     

       159
        
       			curs = *res.Cursor

     

       160
       +
       			cs.storeLk.Lock()

     

       161
       +
       			if err := cs.store.Model(&hcur).Update("cursor", curs).Error; err != nil {

     

       162
       +
       				slog.Error("error updating cursor for host", "err", err)

     

       163
       +
       			}

     

       164
       +
       			cs.storeLk.Unlock()

     

       165
        
       		} else {

     

       166
        
       			break

     

       167
        
       		}

     

       168
       +
       	}

     

       169
       +
       

     

       170
       +
       	slog.Info("finished listing repos", "host", host)

     

       171
       +
       }

     

       172
       +
       

     

       173
       +
       func (cs *CensusService) Start(ctx context.Context, cctx *cli.Context) {

     

       174
       +
       	slog.Info("starting initial hosts and repos crawl")

     

       175
       +
       	cs.listHosts(ctx, cctx)

     

       176
       +
       

     

       177
       +
       	slog.Info("finished with initial refresh, starting ticker")

     

       178
       +
       	t := time.NewTicker(time.Hour)

     

       179
       +
       	defer t.Stop()

     

       180
       +
       

     

       181
       +
       	for {

     

       182
       +
       		select {

     

       183
       +
       		case <-ctx.Done():

     

       184
       +
       			slog.Info("stopping census service", "err", ctx.Err())

     

       185
       +
       			return

     

       186
       +
       		case <-t.C:

     

       187
       +
       		}

     

       188
       +
       

     

       189
       +
       		slog.Info("refreshing hosts and repos")

     

       190
       +
       		cs.listHosts(ctx, cctx)

     

       191
        
       	}

     

       192
        
       }

cmd/monarch/cursors.go

···

       18
        
       

     

       19
        
       func NewCursorService(store *gorm.DB) *CursorService {

     

       20
        
       	store.AutoMigrate(&firehoseCursor{})

     

       0
        
       
     

       0
        
       
     

       21
        
       

     

       22
        
       	return &CursorService{

     

       23
        
       		store: store,

···

       18
        
       

     

       19
        
       func NewCursorService(store *gorm.DB) *CursorService {

     

       20
        
       	store.AutoMigrate(&firehoseCursor{})

     

       21
       +
       	store.AutoMigrate(&relayCursor{})

     

       22
       +
       	store.AutoMigrate(&hostCursor{})

     

       23
        
       

     

       24
        
       	return &CursorService{

     

       25
        
       		store: store,

+4 -2

cmd/monarch/handlers.go

···

       163
        
       }

     

       164
        
       

     

       165
        
       func (hs *HandlerService) HandleCreate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       166
       -
       	return hs.HandleUpsert(ctx, repo, rev, path, rec, cid, ActionCreate)

     

       0
        
       
     

       167
        
       }

     

       168
        
       

     

       169
        
       func (hs *HandlerService) HandleUpdate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       170
       -
       	return hs.HandleUpsert(ctx, repo, rev, path, rec, cid, ActionUpdate)

     

       0
        
       
     

       171
        
       }

     

       172
        
       

     

       173
        
       func (hs *HandlerService) HandleDelete(ctx context.Context, repo string, rev string, path string) error {

···

       163
        
       }

     

       164
        
       

     

       165
        
       func (hs *HandlerService) HandleCreate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       166
       +
       	// return hs.HandleUpsert(ctx, repo, rev, path, rec, cid, ActionCreate)

     

       167
       +
       	return nil

     

       168
        
       }

     

       169
        
       

     

       170
        
       func (hs *HandlerService) HandleUpdate(ctx context.Context, repo string, rev string, path string, rec *[]byte, cid *cid.Cid) error {

     

       171
       +
       	// return hs.HandleUpsert(ctx, repo, rev, path, rec, cid, ActionUpdate)

     

       172
       +
       	return nil

     

       173
        
       }

     

       174
        
       

     

       175
        
       func (hs *HandlerService) HandleDelete(ctx context.Context, repo string, rev string, path string) error {

+5 -1

cmd/monarch/main.go

···

       49
        
       	app.backfill = NewBackfillService(backfill.NewGormstore(app.state), app.handler, cctx)

     

       50
        
       	go app.backfill.Start()

     

       51
        
       

     

       52
       -
       	app.census = NewCensusService(app.cursor, app.backfill)

     

       53
        
       	go app.census.Start(ctx, cctx)

     

       54
        
       

     

       55
        
       	wsconn, err := NewFirehoseConnection(ctx, cctx, app.cursor)

     
···

       132
        
       		&cli.IntFlag{

     

       133
        
       			Name:  "backfill-consumers",

     

       134
        
       			Value: 100,

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       135
        
       		},

     

       136
        
       	}

     

       137

···

       49
        
       	app.backfill = NewBackfillService(backfill.NewGormstore(app.state), app.handler, cctx)

     

       50
        
       	go app.backfill.Start()

     

       51
        
       

     

       52
       +
       	app.census = NewCensusService(app.state, app.backfill)

     

       53
        
       	go app.census.Start(ctx, cctx)

     

       54
        
       

     

       55
        
       	wsconn, err := NewFirehoseConnection(ctx, cctx, app.cursor)

     
···

       132
        
       		&cli.IntFlag{

     

       133
        
       			Name:  "backfill-consumers",

     

       134
        
       			Value: 100,

     

       135
       +
       		},

     

       136
       +
       		&cli.IntFlag{

     

       137
       +
       			Name:  "max-repo-crawlers",

     

       138
       +
       			Value: 4,

     

       139
        
       		},

     

       140
        
       	}

     

       141

+1 -1

go.mod

···

       9
        
       	github.com/gorilla/websocket v1.5.1

     

       10
        
       	github.com/ipfs/go-cid v0.4.1

     

       11
        
       	github.com/urfave/cli/v2 v2.25.7

     

       0
        
       
     

       12
        
       	gorm.io/gorm v1.25.9

     

       13
        
       )

     

       14
        
       

     
···

       102
        
       	go.uber.org/zap v1.26.0 // indirect

     

       103
        
       	golang.org/x/crypto v0.21.0 // indirect

     

       104
        
       	golang.org/x/net v0.23.0 // indirect

     

       105
       -
       	golang.org/x/sync v0.7.0 // indirect

     

       106
        
       	golang.org/x/sys v0.22.0 // indirect

     

       107
        
       	golang.org/x/text v0.14.0 // indirect

     

       108
        
       	golang.org/x/time v0.3.0 // indirect

···

       9
        
       	github.com/gorilla/websocket v1.5.1

     

       10
        
       	github.com/ipfs/go-cid v0.4.1

     

       11
        
       	github.com/urfave/cli/v2 v2.25.7

     

       12
       +
       	golang.org/x/sync v0.7.0

     

       13
        
       	gorm.io/gorm v1.25.9

     

       14
        
       )

     

       15
        
       

     
···

       103
        
       	go.uber.org/zap v1.26.0 // indirect

     

       104
        
       	golang.org/x/crypto v0.21.0 // indirect

     

       105
        
       	golang.org/x/net v0.23.0 // indirect

     

       0
        
       
     

       106
        
       	golang.org/x/sys v0.22.0 // indirect

     

       107
        
       	golang.org/x/text v0.14.0 // indirect

     

       108
        
       	golang.org/x/time v0.3.0 // indirect