···

       
19
       
 
       
	reposSeq    string

     

       
20
       
 
       
}

     

       
21
       
 
       


     

       
22
       
-
       
func NewBackend(state *gorm.DB, bf *backfill.Backfiller) *Backend {

     

       
23
       
 
       
	return &Backend{

     

       
24
       
 
       
		state: state,

     

       
25
       
-
       
		bf:    bf,

     

       
26
       
 
       
	}

     

       
27
       
 
       
}

     

···

       
19
       
 
       
	reposSeq    string

     

       
20
       
 
       
}

     

       
21
       
 
       


     

       
22
       
+
       
func NewBackend(state *gorm.DB) *Backend {

     

       
23
       
 
       
	return &Backend{

     

       
24
       
 
       
		state: state,

     

       
0
       
 
       

     

       
25
       
 
       
	}

     

       
26
       
 
       
}

     

···

       
11
       
 
       
	"github.com/ipfs/go-cid"

     

       
12
       
 
       
)

     

       
13
       
 
       


     

       
14
       
-
       
func (b *Backend) RepoCommitHandler(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
15
       
 
       
	select {

     

       
16
       
 
       
	case <-ctx.Done():

     

       
17
       
 
       
		return nil

     
···

       
39
       
 
       
	}

     

       
40
       
 
       
}

     

       
41
       
 
       


     

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

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
43
       
 
       
	return nil

     

       
44
       
 
       
}

     

       
45
       
 
       


     

       
46
       
-
       
func handleUpdate(ctx context.Context, repo, rev, path string, rec *[]byte, cid *cid.Cid) error {

     

       
47
       
 
       
	return nil

     

       
48
       
 
       
}

     

       
49
       
 
       


     

       
50
       
-
       
func handleDelete(ctx context.Context, repo, rev, path string) error {

     

       
51
       
 
       
	return nil

     

       
52
       
 
       
}

     

···

       
11
       
 
       
	"github.com/ipfs/go-cid"

     

       
12
       
 
       
)

     

       
13
       
 
       


     

       
14
       
+
       
type commitHandler func(context.Context, *comatproto.SyncSubscribeRepos_Commit) error

     

       
15
       
+
       


     

       
16
       
+
       
func (b *Backend) RepoCommitHandler(

     

       
17
       
+
       
	ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit,

     

       
18
       
+
       
) error {

     

       
19
       
 
       
	select {

     

       
20
       
 
       
	case <-ctx.Done():

     

       
21
       
 
       
		return nil

     
···

       
43
       
 
       
	}

     

       
44
       
 
       
}

     

       
45
       
 
       


     

       
46
       
+
       
type handleOpCreateUpdate func(context.Context, string, string, string, *[]byte, *cid.Cid) error

     

       
47
       
+
       
type handleOpDelete func(context.Context, string, string, string) error

     

       
48
       
+
       


     

       
49
       
+
       
func (b *Backend) HandleCreateOp(ctx context.Context, repo, rev, path string, rec *[]byte, cid *cid.Cid) error {

     

       
50
       
 
       
	return nil

     

       
51
       
 
       
}

     

       
52
       
 
       


     

       
53
       
+
       
func (b *Backend) HandleUpdateOp(ctx context.Context, repo, rev, path string, rec *[]byte, cid *cid.Cid) error {

     

       
54
       
 
       
	return nil

     

       
55
       
 
       
}

     

       
56
       
 
       


     

       
57
       
+
       
func (b *Backend) HandleDeleteOp(ctx context.Context, repo, rev, path string) error {

     

       
58
       
 
       
	return nil

     

       
59
       
 
       
}

     

···

       
10
       
 
       
	"syscall"

     

       
11
       
 
       
	"time"

     

       
12
       
 
       


     

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

     

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

     

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

     

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

     
···

       
42
       
 
       
	return db

     

       
43
       
 
       
}

     

       
44
       
 
       


     

       
45
       
-
       
func NewBackfiller(db *gorm.DB) *backfill.Backfiller {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
46
       
 
       
	opts := &backfill.BackfillOptions{

     

       
47
       
 
       
		// ParallelBackfills:     50,

     

       
48
       
 
       
		// ParallelRecordCreates: 25,

     
···

       
58
       
 
       
	return backfill.NewBackfiller(

     

       
59
       
 
       
		"backfills",

     

       
60
       
 
       
		backfill.NewGormstore(db),

     

       
61
       
-
       
		handleCreate,

     

       
62
       
-
       
		handleUpdate,

     

       
63
       
-
       
		handleDelete,

     

       
64
       
 
       
		opts,

     

       
65
       
 
       
	)

     

       
66
       
 
       
}

     
···

       
82
       
 
       
	return conn

     

       
83
       
 
       
}

     

       
84
       
 
       


     

       
85
       
-
       
func NewScheduler(ctx context.Context, backend *Backend) *parallel.Scheduler {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
86
       
 
       
	rsc := events.RepoStreamCallbacks{

     

       
87
       
-
       
		RepoCommit: func(evt *atproto.SyncSubscribeRepos_Commit) error {

     

       
88
       
-
       
			return backend.RepoCommitHandler(ctx, evt)

     

       
89
       
 
       
		},

     

       
90
       
 
       
	}

     

       
91
       
 
       


     
···

       
99
       
 
       
	streamCtx, streamCancel := context.WithCancel(context.Background())

     

       
100
       
 
       


     

       
101
       
 
       
	db := NewDatabase()

     

       
0
       
 
       

     

       
102
       
 
       


     

       
103
       
-
       
	bf := NewBackfiller(db)

     

       
104
       
 
       
	go bf.Start()

     

       
105
       
 
       


     

       
106
       
-
       
	backend := NewBackend(db, bf)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
107
       
 
       
	go backend.SyncCursors(streamCtx)

     

       
108
       
 
       


     

       
109
       
 
       
	cursor, err := backend.LoadCursor("firehose")

     
···

       
112
       
 
       
	}

     

       
113
       
 
       
	conn := NewFirehose(streamCtx, cursor)

     

       
114
       
 
       


     

       
115
       
-
       
	sched := NewScheduler(streamCtx, backend)

     

       
116
       
 
       
	go func() {

     

       
117
       
 
       
		if err := events.HandleRepoStream(streamCtx, conn, sched, sl); err != nil {

     

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

     

···

       
10
       
 
       
	"syscall"

     

       
11
       
 
       
	"time"

     

       
12
       
 
       


     

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

     

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

     

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

     

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

     
···

       
42
       
 
       
	return db

     

       
43
       
 
       
}

     

       
44
       
 
       


     

       
45
       
+
       
func NewBackfiller(

     

       
46
       
+
       
	db *gorm.DB, create handleOpCreateUpdate, update handleOpCreateUpdate, delete handleOpDelete,

     

       
47
       
+
       
) *backfill.Backfiller {

     

       
48
       
 
       
	opts := &backfill.BackfillOptions{

     

       
49
       
 
       
		// ParallelBackfills:     50,

     

       
50
       
 
       
		// ParallelRecordCreates: 25,

     
···

       
60
       
 
       
	return backfill.NewBackfiller(

     

       
61
       
 
       
		"backfills",

     

       
62
       
 
       
		backfill.NewGormstore(db),

     

       
63
       
+
       
		create, update, delete,

     

       
0
       
 
       

     

       
0
       
 
       

     

       
64
       
 
       
		opts,

     

       
65
       
 
       
	)

     

       
66
       
 
       
}

     
···

       
82
       
 
       
	return conn

     

       
83
       
 
       
}

     

       
84
       
 
       


     

       
85
       
+
       
func NewScheduler(

     

       
86
       
+
       
	ctx context.Context, commitCallback commitHandler,

     

       
87
       
+
       
) *parallel.Scheduler {

     

       
88
       
 
       
	rsc := events.RepoStreamCallbacks{

     

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

     

       
90
       
+
       
			return commitCallback(ctx, evt)

     

       
91
       
 
       
		},

     

       
92
       
 
       
	}

     

       
93
       
 
       


     
···

       
101
       
 
       
	streamCtx, streamCancel := context.WithCancel(context.Background())

     

       
102
       
 
       


     

       
103
       
 
       
	db := NewDatabase()

     

       
104
       
+
       
	backend := NewBackend(db)

     

       
105
       
 
       


     

       
106
       
+
       
	bf := NewBackfiller(db, backend.HandleCreateOp, backend.HandleUpdateOp, backend.HandleDeleteOp)

     

       
107
       
 
       
	go bf.Start()

     

       
108
       
 
       


     

       
109
       
+
       
	// attach the backfiller to the backend so pump and repo commit handler can use it

     

       
110
       
+
       
	backend.bf = bf

     

       
111
       
+
       


     

       
112
       
 
       
	go backend.SyncCursors(streamCtx)

     

       
113
       
 
       


     

       
114
       
 
       
	cursor, err := backend.LoadCursor("firehose")

     
···

       
117
       
 
       
	}

     

       
118
       
 
       
	conn := NewFirehose(streamCtx, cursor)

     

       
119
       
 
       


     

       
120
       
+
       
	sched := NewScheduler(streamCtx, backend.RepoCommitHandler)

     

       
121
       
 
       
	go func() {

     

       
122
       
 
       
		if err := events.HandleRepoStream(streamCtx, conn, sched, sl); err != nil {

     

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