···

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

     

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

     

       
12
       
12
       
 
       
		NSIDFilter:            "",

     

       
13
       
13
       
-
       
		SyncRequestsPerSecond: 10,

     

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

     

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

     

       
15
       
15
       
 
       
	}

     

       
16
       
16
       
 
       


     

···

       
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
       
 
       


     

···

       
13
       
13
       
 
       
func NewFirehoseConnection(ctx context.Context, cctx *cli.Context, cursorSvc *CursorService) (*websocket.Conn, error) {

     

       
14
       
14
       
 
       
	url := fmt.Sprintf("wss://%s/xrpc/com.atproto.sync.subscribeRepos", cctx.String("relay-host"))

     

       
15
       
15
       
 
       
	curs, err := cursorSvc.GetFirehoseCursor()

     

       
16
       
16
       
-
       
	if err == nil { // reversed

     

       
16
       
16
       
+
       
	if err != nil {

     

       
17
       
17
       
+
       
		slog.Error("error getting firehose cursor", "err", err)

     

       
18
       
18
       
+
       
	} else if curs > 0 {

     

       
17
       
19
       
 
       
		url += fmt.Sprintf("?cursor=%d", curs)

     

       
18
       
20
       
 
       
	}

     

       
19
       
21