···

       
11
       
 
       
	state *gorm.DB

     

       
12
       
 
       
	bf    *backfill.Backfiller

     

       
13
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
14
       
 
       
	firehoseLk  sync.Mutex

     

       
15
       
 
       
	firehoseSeq string

     

       
16
       
 
       
	reposLk     sync.Mutex

     

···

       
11
       
 
       
	state *gorm.DB

     

       
12
       
 
       
	bf    *backfill.Backfiller

     

       
13
       
 
       


     

       
14
       
+
       
	backfillComplete bool

     

       
15
       
+
       


     

       
16
       
 
       
	firehoseLk  sync.Mutex

     

       
17
       
 
       
	firehoseSeq string

     

       
18
       
 
       
	reposLk     sync.Mutex

     

···

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

     

       
24
       
 
       
	b.firehoseLk.Unlock()

     

       
25
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

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

     

       
27
       
 
       
	if job == nil {

     

       
28
       
 
       
		if errors.Is(err, backfill.ErrJobNotFound) {

     

···

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

     

       
24
       
 
       
	b.firehoseLk.Unlock()

     

       
25
       
 
       


     

       
26
       
+
       
	if b.backfillComplete {

     

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

     

       
28
       
+
       
	}

     

       
29
       
+
       


     

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

     

       
31
       
 
       
	if job == nil {

     

       
32
       
 
       
		if errors.Is(err, backfill.ErrJobNotFound) {

     

···

       
123
       
 
       
	go func() {

     

       
124
       
 
       
		if err := backend.PumpRepos(streamCtx); err != nil {

     

       
125
       
 
       
			sl.Error("failed pumping repos", "err", err)

     

       
126
       
-
       
		} else {

     

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

     

       
128
       
 
       
		}

     

       
129
       
 
       
	}()

     

       
130
       
 
       


     

···

       
123
       
 
       
	go func() {

     

       
124
       
 
       
		if err := backend.PumpRepos(streamCtx); err != nil {

     

       
125
       
 
       
			sl.Error("failed pumping repos", "err", err)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
126
       
 
       
		}

     

       
127
       
 
       
	}()

     

       
128
       
 
       


     

···

       
65
       
 
       
		}

     

       
66
       
 
       
	}

     

       
67
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
68
       
 
       
	return nil

     

       
69
       
 
       
}

     

···

       
65
       
 
       
		}

     

       
66
       
 
       
	}

     

       
67
       
 
       


     

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

     

       
69
       
+
       
	b.backfillComplete = true

     

       
70
       
 
       
	return nil

     

       
71
       
 
       
}