Start recording last rev for each repo

cmd/consumer/consumer.go

@@ -356,6 +356,14 @@ func (c *Consumer) processMessage(ctx context.Context, typ string, r io.Reader,
 			}
 		}
 
+		err = c.db.Model(&repo.Repo{}).Where(&repo.Repo{ID: repoInfo.ID}).
+			Updates(&repo.Repo{
+				LastFirehoseRev: payload.Rev,
+			}).Error
+		if err != nil {
+			log.Error().Err(err).Msgf("Failed to update last_firehose_rev for %q: %s", repoInfo.DID, err)
+		}
+
 		if payload.TooBig {
 			// Just trigger a re-index by resetting rev.
 			err := c.db.Model(&repo.Repo{}).Where(&repo.Repo{ID: repoInfo.ID}).

repo/repo.go

@@ -22,6 +22,7 @@ type Repo struct {
 	DID                   string    `gorm:"uniqueIndex;column:did"`
 	LastIndexedRev        string    `gorm:"index:rev_state_index,expression:(last_indexed_rev < first_rev_since_reset),priority:1;index:was_indexed,expression:(last_indexed_rev is null OR last_indexed_rev = ''),priority:1"`
 	FirstRevSinceReset    string
+	LastFirehoseRev       string
 	FirstCursorSinceReset int64
 	TombstonedAt          time.Time
 	LastIndexAttempt      time.Time