commit 428c4ea8cb0f158b8d3ad50338ebfd27872ddfdb · edavis.dev/monarch

+89 -21

cmd/backfiller/backfiller.go

···

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
        
       	"context"

     

       5
       5
       +
       	"errors"

     

       6
       6
       +
       	"fmt"

     

       7
       7
       +
       	"log"

     

       5
       8
        
       	"net/http"

     

       6
       6
       -
       	// "strings"

     

       7
       7
       -
       	// "log/slog"

     

       8
       8
       -
       	"log"

     

       9
       9
        
       	"os"

     

       10
       10
        
       	"os/signal"

     

       11
       11
        
       	"syscall"

     

       12
       12
       +
       	"time"

     

       12
       13
        
       

     

       14
       14
       +
       	"github.com/bluesky-social/indigo/api/atproto"

     

       13
       15
        
       	comatproto "github.com/bluesky-social/indigo/api/atproto"

     

       14
       16
        
       	"github.com/bluesky-social/indigo/backfill"

     

       15
       17
        
       	"github.com/bluesky-social/indigo/events"

     

       16
       18
        
       	"github.com/bluesky-social/indigo/events/schedulers/parallel"

     

       19
       19
       +
       	"github.com/bluesky-social/indigo/xrpc"

     

       17
       20
        
       

     

       18
       21
        
       	"github.com/gorilla/websocket"

     

       19
       22
        
       	"github.com/ipfs/go-cid"

     

       20
       23
        
       	"gorm.io/driver/sqlite"

     

       21
       24
        
       	"gorm.io/gorm"

     

       25
       25
       +
       	"gorm.io/gorm/logger"

     

       22
       26
        
       )

     

       23
       27
        
       

     

       24
       28
        
       func handleCreate(ctx context.Context, repo, rev, path string, rec *[]byte, cid *cid.Cid) error {

     

       25
       25
       -
       	log.Printf("handleCreate: at://%s/%s", repo, path)

     

       26
       29
        
       	return nil

     

       27
       30
        
       }

     

       28
       31
        
       

     
···

       38
       41
        
       	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGINT, syscall.SIGTERM)

     

       39
       42
        
       	defer stop()

     

       40
       43
        
       

     

       41
       41
       -
       	db, err := gorm.Open(sqlite.Open("backfiller.db"), &gorm.Config{})

     

       44
       44
       +
       	newLogger := logger.New(

     

       45
       45
       +
       		log.New(os.Stdout, "\n", log.LstdFlags),

     

       46
       46
       +
       		logger.Config{

     

       47
       47
       +
       			SlowThreshold: 5*time.Second,

     

       48
       48
       +
       			Colorful: false,

     

       49
       49
       +
       		},

     

       50
       50
       +
       	)

     

       51
       51
       +
       

     

       52
       52
       +
       	db, err := gorm.Open(sqlite.Open("state.db"), &gorm.Config{

     

       53
       53
       +
       		Logger: newLogger,

     

       54
       54
       +
       	})

     

       42
       55
        
       	if err != nil {

     

       43
       56
        
       		log.Fatalf("failed to connect database: %s", err)

     

       44
       57
        
       	}

     
···

       46
       59
        
       	store := backfill.NewGormstore(db)

     

       47
       60
        
       

     

       48
       61
        
       	// connect to the relay

     

       49
       49
       -
       	d := websocket.DefaultDialer

     

       50
       50
       -
       	con, _, err := d.Dial("wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos", http.Header{})

     

       62
       62
       +
       	con, _, err := websocket.DefaultDialer.Dial("wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos", http.Header{

     

       63
       63
       +
       		"User-Agent": []string{"backfiller/0.1 (@edavis.dev)"},

     

       64
       64
       +
       	})

     

       51
       65
        
       	if err != nil {

     

       52
       66
        
       		log.Fatalf("failed to connect to relay: %s", err)

     

       53
       67
        
       	}

     

       54
       68
        
       

     

       55
       55
       -
       	rsc := events.RepoStreamCallbacks{

     

       56
       56
       -
       		RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {

     

       57
       57
       -
       			// log.Printf("enqueuing: %s", evt.Repo)

     

       58
       58
       -
       			return store.EnqueueJobWithState(ctx, evt.Repo, backfill.StateEnqueued)

     

       59
       59
       -
       		},

     

       60
       60
       -
       	}

     

       61
       61
       -
       

     

       62
       69
        
       	// start backfilling

     

       63
       70
        
       	opts := &backfill.BackfillOptions{

     

       64
       64
       -
       		ParallelBackfills: 10,

     

       71
       71
       +
       		ParallelBackfills:     10,

     

       65
       72
        
       		ParallelRecordCreates: 100,

     

       66
       66
       -
       		NSIDFilter: "app.bsky.labeler.service",

     

       67
       67
       -
       		SyncRequestsPerSecond: 10,

     

       68
       68
       -
       		RelayHost: "https://bsky.network",

     

       73
       73
       +
       		SyncRequestsPerSecond: 4,

     

       74
       74
       +
       		RelayHost:             "https://bsky.network",

     

       69
       75
        
       	}

     

       70
       70
       -
       	bf := backfill.NewBackfiller("backfiller-labels", store, handleCreate, handleUpdate, handleDelete, opts)

     

       76
       76
       +
       	bf := backfill.NewBackfiller("backfills", store, handleCreate, handleUpdate, handleDelete, opts)

     

       71
       77
        
       	go bf.Start()

     

       72
       78
        
       

     

       73
       73
       -
       	sched := parallel.NewScheduler(4, 1000, "backfiller-labels", rsc.EventHandler)

     

       79
       79
       +
       	// pump repos

     

       80
       80
       +
       	go func(bf *backfill.Backfiller) {

     

       81
       81
       +
       		if err := pumpRepos(context.TODO(), bf); err != nil {

     

       82
       82
       +
       			log.Printf("failed pumping repos: %s", err)

     

       83
       83
       +
       		}

     

       84
       84
       +
       	}(bf)

     

       85
       85
       +
       

     

       86
       86
       +
       	// read from the firehose

     

       87
       87
       +
       	rsc := events.RepoStreamCallbacks{

     

       88
       88
       +
       		RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {

     

       89
       89
       +
       			job, err := bf.Store.GetJob(ctx, evt.Repo)

     

       90
       90
       +
       			if job == nil {

     

       91
       91
       +
       				if errors.Is(err, backfill.ErrJobNotFound) {

     

       92
       92
       +
       					return nil

     

       93
       93
       +
       				} else {

     

       94
       94
       +
       					return fmt.Errorf("error getting job: %w", err)

     

       95
       95
       +
       				}

     

       96
       96
       +
       			} else {

     

       97
       97
       +
       				return bf.HandleEvent(ctx, evt)

     

       98
       98
       +
       			}

     

       99
       99
       +
       		},

     

       100
       100
       +
       	}

     

       101
       101
       +
       	sched := parallel.NewScheduler(16, 100, "firehose", rsc.EventHandler)

     

       74
       102
        
       	if err := events.HandleRepoStream(ctx, con, sched, nil); err != nil {

     

       75
       103
        
       		log.Fatalf("failed to start scheduler: %s", err)

     

       76
       104
        
       	}

     

       77
       105
        
       

     

       78
       106
        
       	<-ctx.Done()

     

       79
       79
       -
       	bf.Stop(ctx)

     

       107
       107
       +
       	bf.Stop(context.TODO())

     

       108
       108
       +
       }

     

       109
       109
       +
       

     

       110
       110
       +
       type jobMaker interface {

     

       111
       111
       +
       	GetOrCreateJob(context.Context, string, string) (backfill.Job, error)

     

       112
       112
       +
       }

     

       113
       113
       +
       

     

       114
       114
       +
       func pumpRepos(ctx context.Context, bf *backfill.Backfiller) error {

     

       115
       115
       +
       	xrpcc := &xrpc.Client{

     

       116
       116
       +
       		Host: "https://bsky.network",

     

       117
       117
       +
       	}

     

       118
       118
       +
       

     

       119
       119
       +
       	jmstore, ok := bf.Store.(jobMaker)

     

       120
       120
       +
       	if !ok {

     

       121
       121
       +
       		return fmt.Errorf("configured job store doesn't support random job creation")

     

       122
       122
       +
       	}

     

       123
       123
       +
       

     

       124
       124
       +
       	var curs string

     

       125
       125
       +
       	for {

     

       126
       126
       +
       		log.Printf("listing repos with cursor = %v", curs)

     

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

     

       128
       128
       +
       		if err != nil {

     

       129
       129
       +
       			return fmt.Errorf("error listing repos: %w", err)

     

       130
       130
       +
       		}

     

       131
       131
       +
       

     

       132
       132
       +
       		for _, repo := range res.Repos {

     

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

     

       134
       134
       +
       			if err != nil {

     

       135
       135
       +
       				log.Printf("failed to create backfill job: %s", err)

     

       136
       136
       +
       				continue

     

       137
       137
       +
       			}

     

       138
       138
       +
       		}

     

       139
       139
       +
       

     

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

     

       141
       141
       +
       			curs = *res.Cursor

     

       142
       142
       +
       		} else {

     

       143
       143
       +
       			break

     

       144
       144
       +
       		}

     

       145
       145
       +
       	}

     

       146
       146
       +
       

     

       147
       147
       +
       	return nil

     

       80
       148
        
       }

+1 -1

go.mod

···

       5
       5
        
       toolchain go1.24.5

     

       6
       6
        
       

     

       7
       7
        
       require (

     

       8
       8
       -
       	github.com/bluesky-social/indigo v0.0.0-20250716175339-eebba3d96129

     

       8
       8
       +
       	github.com/bluesky-social/indigo v0.0.0-20250724221105-5827c8fb61bb

     

       9
       9
        
       	github.com/gorilla/websocket v1.5.1

     

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

     

       11
       11
        
       	gorm.io/driver/sqlite v1.5.5

+2 -2

go.sum

···

       10
       10
        
       github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=

     

       11
       11
        
       github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 h1:mXoPYz/Ul5HYEDvkta6I8/rnYM5gSdSV2tJ6XbZuEtY=

     

       12
       12
        
       github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932/go.mod h1:NOuUCSz6Q9T7+igc/hlvDOUdtWKryOrtFyIVABv/p7k=

     

       13
       13
       -
       github.com/bluesky-social/indigo v0.0.0-20250716175339-eebba3d96129 h1:81032NtsXNDdD8St+XHEdFlbRukJDKlS7XkGhrdZKNk=

     

       14
       14
       -
       github.com/bluesky-social/indigo v0.0.0-20250716175339-eebba3d96129/go.mod h1:0XUyOCRtL4/OiyeqMTmr6RlVHQMDgw3LS7CfibuZR5Q=

     

       13
       13
       +
       github.com/bluesky-social/indigo v0.0.0-20250724221105-5827c8fb61bb h1:BqMNDZMfXwiRTJ6NvQotJ0qInn37JH5U8E+TF01CFHQ=

     

       14
       14
       +
       github.com/bluesky-social/indigo v0.0.0-20250724221105-5827c8fb61bb/go.mod h1:0XUyOCRtL4/OiyeqMTmr6RlVHQMDgw3LS7CfibuZR5Q=

     

       15
       15
        
       github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=

     

       16
       16
        
       github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=

     

       17
       17
        
       github.com/carlmjohnson/versioninfo v0.22.5 h1:O00sjOLUAFxYQjlN/bzYTuZiS0y6fWDQjMRvwtKgwwc=