···

       
12
       
 
       
	data  *gorm.DB

     

       
13
       
 
       
	bf    *backfill.Backfiller

     

       
14
       
 
       


     

       
15
       
-
       
	backfillComplete bool

     

       
16
       
-
       


     

       
17
       
 
       
	firehoseLk  sync.Mutex

     

       
18
       
 
       
	firehoseSeq string

     

       
19
       
 
       
	reposLk     sync.Mutex

     

···

       
12
       
 
       
	data  *gorm.DB

     

       
13
       
 
       
	bf    *backfill.Backfiller

     

       
14
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
15
       
 
       
	firehoseLk  sync.Mutex

     

       
16
       
 
       
	firehoseSeq string

     

       
17
       
 
       
	reposLk     sync.Mutex

     

···

       
3
       
 
       
import (

     

       
4
       
 
       
	"bytes"

     

       
5
       
 
       
	"context"

     

       
6
       
-
       
	"errors"

     

       
7
       
 
       
	"fmt"

     

       
8
       
 
       
	"log/slog"

     

       
9
       
 
       
	"strconv"

     
···

       
11
       
 
       


     

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

     

       
13
       
 
       
	appbsky "github.com/bluesky-social/indigo/api/bsky"

     

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

     

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

     

       
16
       
 
       
)

     

       
17
       
 
       


     
···

       
31
       
 
       
	b.firehoseSeq = strconv.Itoa(int(evt.Seq))

     

       
32
       
 
       
	b.firehoseLk.Unlock()

     

       
33
       
 
       


     

       
34
       
-
       
	if b.backfillComplete {

     

       
35
       
-
       
		return b.bf.HandleEvent(ctx, evt)

     

       
36
       
-
       
	}

     

       
37
       
-
       


     

       
38
       
-
       
	job, err := b.bf.Store.GetJob(ctx, evt.Repo)

     

       
39
       
-
       
	if job == nil {

     

       
40
       
-
       
		if errors.Is(err, backfill.ErrJobNotFound) {

     

       
41
       
-
       
			return nil

     

       
42
       
-
       
		} else {

     

       
43
       
-
       
			return fmt.Errorf("error getting job: %w", err)

     

       
44
       
-
       
		}

     

       
45
       
-
       
	} else {

     

       
46
       
-
       
		return b.bf.HandleEvent(ctx, evt)

     

       
47
       
-
       
	}

     

       
48
       
 
       
}

     

       
49
       
 
       


     

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

     
···

       
55
       
 
       
		return nil

     

       
56
       
 
       
	}

     

       
57
       
 
       


     

       
58
       
-
       
	sl := slog.With("source", "commitHandler")

     

       
59
       
 
       


     

       
60
       
 
       
	var out appbsky.FeedGenerator

     

       
61
       
 
       
	if err := out.UnmarshalCBOR(bytes.NewReader(*rec)); err != nil {

     

···

       
3
       
 
       
import (

     

       
4
       
 
       
	"bytes"

     

       
5
       
 
       
	"context"

     

       
0
       
 
       

     

       
6
       
 
       
	"fmt"

     

       
7
       
 
       
	"log/slog"

     

       
8
       
 
       
	"strconv"

     
···

       
10
       
 
       


     

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

     

       
12
       
 
       
	appbsky "github.com/bluesky-social/indigo/api/bsky"

     

       
0
       
 
       

     

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

     

       
14
       
 
       
)

     

       
15
       
 
       


     
···

       
29
       
 
       
	b.firehoseSeq = strconv.Itoa(int(evt.Seq))

     

       
30
       
 
       
	b.firehoseLk.Unlock()

     

       
31
       
 
       


     

       
32
       
+
       
	return b.bf.HandleEvent(ctx, evt)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
33
       
 
       
}

     

       
34
       
 
       


     

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

     
···

       
40
       
 
       
		return nil

     

       
41
       
 
       
	}

     

       
42
       
 
       


     

       
43
       
+
       
	sl := slog.With("source", "HandleCreateOp")

     

       
44
       
 
       


     

       
45
       
 
       
	var out appbsky.FeedGenerator

     

       
46
       
 
       
	if err := out.UnmarshalCBOR(bytes.NewReader(*rec)); err != nil {

     

···

       
65
       
 
       
		}

     

       
66
       
 
       
	}

     

       
67
       
 
       


     

       
68
       
-
       
	sl.Info("finished listing repos, switching over to event stream")

     

       
69
       
-
       
	b.backfillComplete = true

     

       
70
       
 
       
	return nil

     

       
71
       
 
       
}

     

···

       
65
       
 
       
		}

     

       
66
       
 
       
	}

     

       
67
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
68
       
 
       
	return nil

     

       
69
       
 
       
}