···

       
3
       
3
       
 
       
import (

     

       
4
       
4
       
 
       
	"context"

     

       
5
       
5
       
 
       
	"log/slog"

     

       
6
       
6
       
-
       
	"sync"

     

       
7
       
6
       
 
       
	"time"

     

       
8
       
7
       
 
       


     

       
9
       
8
       
 
       
	"gorm.io/gorm"

     

       
10
       
9
       
 
       
)

     

       
11
       
10
       
 
       


     

       
12
       
11
       
 
       
type CursorService struct {

     

       
13
       
13
       
-
       
	store *gorm.DB

     

       
14
       
14
       
-
       


     

       
15
       
15
       
-
       
	firehoseLk  sync.Mutex

     

       
12
       
12
       
+
       
	store       *gorm.DB

     

       
16
       
13
       
 
       
	firehoseSeq int64

     

       
17
       
14
       
 
       
}

     

       
18
       
15
       
 
       


     

···

       
1
       
1
       
 
       
package main

     

       
2
       
2
       
 
       


     

       
3
       
3
       
 
       
import (

     

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

     

       
4
       
4
       
+
       
	backfill "github.com/bluesky-social/indigo/backfill/next"

     

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

     

       
6
       
6
       
 
       
)

     

       
7
       
7
       
 
       


     

       
8
       
8
       
 
       
func NewBackfillService(store backfill.Store, h *HandlerService, cctx *cli.Context) *backfill.Backfiller {

     

       
9
       
9
       
-
       
	opts := &backfill.BackfillOptions{

     

       
10
       
10
       
-
       
		ParallelBackfills:     cctx.Int("backfill-workers"),

     

       
11
       
11
       
-
       
		ParallelRecordCreates: cctx.Int("backfill-consumers"),

     

       
12
       
12
       
-
       
		NSIDFilter:            "",

     

       
13
       
13
       
-
       
		SyncRequestsPerSecond: cctx.Int("sync-requests-limit"),

     

       
14
       
14
       
-
       
		RelayHost:             "https://" + cctx.String("relay-host"),

     

       
15
       
15
       
-
       
	}

     

       
9
       
9
       
+
       
	opts := backfill.DefaultBackfillerOptions()

     

       
10
       
10
       
+
       
	opts.PerPDSBackfillConcurrency = cctx.Int("backfill-workers")

     

       
11
       
11
       
+
       
	opts.GlobalRecordCreateConcurrency = cctx.Int("backfill-consumers")

     

       
12
       
12
       
+
       
	opts.PerPDSSyncsPerSecond = cctx.Float64("sync-requests-limit")

     

       
16
       
13
       
 
       


     

       
17
       
17
       
-
       
	return backfill.NewBackfiller("backfiller", store, h.HandleCreate, h.HandleUpdate, h.HandleDelete, opts)

     

       
14
       
14
       
+
       
	return backfill.NewBackfiller("backfiller", store, h.HandleCreate, opts)

     

       
18
       
15
       
 
       
}

     

···

       
7
       
7
       
 
       
	"time"

     

       
8
       
8
       
 
       


     

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

     

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

     

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

     

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

     

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

     

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

     
···

       
19
       
19
       
 
       
	seenHosts map[string]bool

     

       
20
       
20
       
 
       
}

     

       
21
       
21
       
 
       


     

       
22
       
22
       
-
       
type jobMaker interface {

     

       
23
       
23
       
-
       
	GetOrCreateJob(context.Context, string, string) (backfill.Job, error)

     

       
24
       
24
       
-
       
}

     

       
25
       
25
       
-
       


     

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

     

       
27
       
23
       
 
       
	return &CensusService{

     

       
28
       
24
       
 
       
		cursor:    cursorSvc,

     
···

       
102
       
98
       
 
       
		Host: "https://" + host,

     

       
103
       
99
       
 
       
	}

     

       
104
       
100
       
 
       


     

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

     

       
106
       
106
       
-
       
	if !ok {

     

       
107
       
107
       
-
       
		slog.Error("configured job store doesn't support random job creation")

     

       
108
       
108
       
-
       
		return

     

       
109
       
109
       
-
       
	}

     

       
110
       
110
       
-
       


     

       
111
       
101
       
 
       
	hcur, err := cs.cursor.GetHostCursor(host)

     

       
112
       
102
       
 
       
	if err != nil {

     

       
113
       
103
       
 
       
		slog.Error("error fetching host cursor", "err", err)

     
···

       
131
       
121
       
 
       
		}

     

       
132
       
122
       
 
       


     

       
133
       
123
       
 
       
		for _, repo := range res.Repos {

     

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

     

       
124
       
124
       
+
       
			err := cs.backfill.EnqueueJob(ctx, host, repo.Did)

     

       
135
       
125
       
 
       
			if err != nil {

     

       
136
       
126
       
 
       
				slog.Error("error adding repo to backfiller", "err", err)

     

       
137
       
127
       
 
       
			} else {