commit 692f719b56f79542f88d62752b6909e25f2a8813 · edavis.dev/monarch

+40 -21

cmd/backfiller/backfiller.go

···

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
        
       	"context"

     

       5
       5
       -
       	"log/slog"

     

       5
       5
       +
       	"net/http"

     

       6
       6
       +
       	// "strings"

     

       7
       7
       +
       	// "log/slog"

     

       8
       8
       +
       	"log"

     

       6
       9
        
       	"os"

     

       7
       10
        
       	"os/signal"

     

       8
       8
       -
       	"strings"

     

       9
       11
        
       	"syscall"

     

       10
       12
        
       

     

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

     

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

     

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

     

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

     

       17
       17
       +
       

     

       18
       18
       +
       	"github.com/gorilla/websocket"

     

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

     

       13
       20
        
       	"gorm.io/driver/sqlite"

     

       14
       21
        
       	"gorm.io/gorm"

     

       15
       22
        
       )

     

       16
       23
        
       

     

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

     

       18
       18
       -
       	if !strings.HasPrefix(path, "com.whtwnd.blog.entry") {

     

       19
       19
       -
       		return nil

     

       20
       20
       -
       	}

     

       21
       21
       -
       	slog.Info("create", "path", path)

     

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

     

       22
       26
        
       	return nil

     

       23
       27
        
       }

     

       24
       28
        
       

     
···

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

     

       35
       39
        
       	defer stop()

     

       36
       40
        
       

     

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

     

       38
       38
       -
       	db, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{})

     

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

     

       39
       42
        
       	if err != nil {

     

       40
       40
       -
       		panic("failed to connect database")

     

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

     

       41
       44
        
       	}

     

       42
       45
        
       	db.AutoMigrate(&backfill.GormDBJob{})

     

       46
       46
       +
       	store := backfill.NewGormstore(db)

     

       43
       47
        
       

     

       44
       44
       -
       	store := backfill.NewGormstore(db)

     

       45
       45
       -
       	store.EnqueueJob(ctx, "did:plc:4nsduwlpivpuur4mqkbfvm6a")

     

       48
       48
       +
       	// 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{})

     

       51
       51
       +
       	if err != nil {

     

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

     

       53
       53
       +
       	}

     

       46
       54
        
       

     

       47
       47
       -
       	opts := backfill.DefaultBackfillOptions()

     

       48
       48
       -
       	opts.NSIDFilter = "com.whtwnd.blog.entry"

     

       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
       +
       	}

     

       49
       61
        
       

     

       50
       50
       -
       	bf := backfill.NewBackfiller(

     

       51
       51
       -
       		"backfill-feeds",

     

       52
       52
       -
       		store,

     

       53
       53
       -
       		handleCreate,

     

       54
       54
       -
       		handleUpdate,

     

       55
       55
       -
       		handleDelete,

     

       56
       56
       -
       		opts,

     

       57
       57
       -
       	)

     

       62
       62
       +
       	// start backfilling

     

       63
       63
       +
       	opts := &backfill.BackfillOptions{

     

       64
       64
       +
       		ParallelBackfills: 10,

     

       65
       65
       +
       		ParallelRecordCreates: 100,

     

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

     

       67
       67
       +
       		SyncRequestsPerSecond: 10,

     

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

     

       69
       69
       +
       	}

     

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

     

       58
       71
        
       	go bf.Start()

     

       72
       72
       +
       

     

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

     

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

     

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

     

       76
       76
       +
       	}

     

       77
       77
       +
       

     

       59
       78
        
       	<-ctx.Done()

     

       60
       79
        
       	bf.Stop(ctx)

     

       61
       80
        
       }

go.mod

···

       6
       6
        
       

     

       7
       7
        
       require (

     

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

     

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

     

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

     

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

     

       11
       12
        
       	gorm.io/gorm v1.25.9

     

       12
       13
        
       )

     

       13
       14
        
       

     

       14
       15
        
       require (

     

       16
       16
       +
       	github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect

     

       15
       17
        
       	github.com/beorn7/perks v1.0.1 // indirect

     

       16
       18
        
       	github.com/carlmjohnson/versioninfo v0.22.5 // indirect

     

       17
       19
        
       	github.com/cespare/xxhash/v2 v2.2.0 // indirect

     
···

       83
       85
        
       	go.uber.org/multierr v1.11.0 // indirect

     

       84
       86
        
       	go.uber.org/zap v1.26.0 // indirect

     

       85
       87
        
       	golang.org/x/crypto v0.21.0 // indirect

     

       88
       88
       +
       	golang.org/x/net v0.23.0 // indirect

     

       86
       89
        
       	golang.org/x/sync v0.7.0 // indirect

     

       87
       90
        
       	golang.org/x/sys v0.22.0 // indirect

     

       88
       91
        
       	golang.org/x/text v0.14.0 // indirect

go.sum

···

       1
       1
        
       github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=

     

       2
       2
       +
       github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b h1:5/++qT1/z812ZqBvqQt6ToRswSuPZ/B33m6xVHRzADU=

     

       3
       3
       +
       github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b/go.mod h1:4+EPqMRApwwE/6yo6CxiHoSnBzjRr3jsqer7frxP8y4=

     

       2
       4
        
       github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM=

     

       3
       5
        
       github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA=

     

       4
       6
        
       github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=

     
···

       33
       35
        
       github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=

     

       34
       36
        
       github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=

     

       35
       37
        
       github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=

     

       38
       38
       +
       github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=

     

       39
       39
       +
       github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=

     

       36
       40
        
       github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=

     

       37
       41
        
       github.com/gocql/gocql v1.7.0 h1:O+7U7/1gSN7QTEAaMEsJc1Oq2QHXvCWoF3DFK9HDHus=

     

       38
       42
        
       github.com/gocql/gocql v1.7.0/go.mod h1:vnlvXyFZeLBF0Wy+RS8hrOdbn0UWsWtdg07XJnFxZ+4=

     
···

       50
       54
        
       github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=

     

       51
       55
        
       github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=

     

       52
       56
        
       github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=

     

       57
       57
       +
       github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=

     

       58
       58
       +
       github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=

     

       53
       59
        
       github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed h1:5upAirOpQc1Q53c0bnx2ufif5kANL7bfZWcc6VJWJd8=

     

       54
       60
        
       github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed/go.mod h1:tMWxXQ9wFIaZeTI9F+hmhFiGpFmhOHzyShyFUhRm0H4=

     

       55
       61
        
       github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ=