commit e98188aa1ef2b897e139d0a1c19c8380e63384a5 · edavis.dev/monarch

+11 -7
cmd/backfiller/backfiller.go
···

       5
       5
        
       	"errors"

     

       6
       6
        
       	"fmt"

     

       7
       7
        
       	"log"

     

       8
       8
       +
       	"log/slog"

     

       8
       9
        
       	"net/http"

     

       9
       10
        
       	"os"

     

       10
       11
        
       	"os/signal"

     
···

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

     

       42
       43
        
       	defer stop()

     

       43
       44
        
       

     

       45
       45
       +
       	sl := slog.With("source", "backfiller")

     

       46
       46
       +
       

     

       44
       47
        
       	newLogger := logger.New(

     

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

     

       46
       49
        
       		logger.Config{

     
···

       48
       51
        
       			Colorful: false,

     

       49
       52
        
       		},

     

       50
       53
        
       	)

     

       51
       51
       -
       

     

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

     

       53
       55
        
       		Logger: newLogger,

     

       54
       56
        
       	})

     

       55
       57
        
       	if err != nil {

     

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

     

       58
       58
       +
       		sl.Error("failed to connect to database", "err", err)

     

       57
       59
        
       	}

     

       58
       60
        
       	db.AutoMigrate(&backfill.GormDBJob{})

     

       59
       61
        
       	store := backfill.NewGormstore(db)

     
···

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

     

       64
       66
        
       	})

     

       65
       67
        
       	if err != nil {

     

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

     

       68
       68
       +
       		sl.Error("failed to connect to relay", "err", err)

     

       67
       69
        
       	}

     

       68
       70
        
       

     

       69
       71
        
       	// start backfilling

     
···

       79
       81
        
       	// pump repos

     

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

     

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

     

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

     

       84
       84
       +
       			sl.Error("failed pumping repos", "err", err)

     

       83
       85
        
       		}

     

       84
       86
        
       	}(bf)

     

       85
       87
        
       

     
···

       100
       102
        
       	}

     

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

     

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

     

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

     

       105
       105
       +
       		sl.Error("failed to start scheduler", "err", err)

     

       104
       106
        
       	}

     

       105
       107
        
       

     

       106
       108
        
       	<-ctx.Done()

     
···

       112
       114
        
       }

     

       113
       115
        
       

     

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

     

       117
       117
       +
       	sl := slog.With("source", "pumpRepos")

     

       118
       118
       +
       

     

       115
       119
        
       	xrpcc := &xrpc.Client{

     

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

     

       117
       121
        
       	}

     
···

       123
       127
        
       

     

       124
       128
        
       	var curs string

     

       125
       129
        
       	for {

     

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

     

       130
       130
       +
       		sl.Info("listing repos", "cursor", curs)

     

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

     

       128
       132
        
       		if err != nil {

     

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

     
···

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

     

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

     

       134
       138
        
       			if err != nil {

     

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

     

       139
       139
       +
       				sl.Warn("failed to create backfill job", "err", err)

     

       136
       140
        
       				continue

     

       137
       141
        
       			}

     

       138
       142
        
       		}