feat: implement chain reorganization handling in Chaintracks service (#698)

Author: chris-4chain

pkg/services/chaintracks/chaintracks_service.go

@@ -579,9 +579,41 @@ func (s *Service) storeLiveHeader(ctx context.Context, header wdk.ChainBlockHead
 	}
 
 	isActiveTip := chainWork.CmpChainWork(priorTipChainWork) > 0
-	//if isActiveTip {
-	//	// TODO: handle reorgs if needed
-	//}
+	if isActiveTip {
+		s.logger.Info("Chaintracks service - new active chain tip detected", slog.String("header_hash", header.Hash), slog.Any("header_height", header.Height))
+
+		// find newHeader's first active ancestor
+		activeAncestor := oneBack
+		for !activeAncestor.IsActive {
+			previousHash := activeAncestor.PreviousHash
+			activeAncestor, err = q.GetLiveHeaderByHash(previousHash)
+			if err != nil {
+				return fmt.Errorf("failed to get active ancestor header: %w", err)
+			}
+
+			if activeAncestor == nil {
+				return fmt.Errorf("active ancestor header not found for hash: %s", previousHash)
+			}
+		}
+
+		// TODO: Calculate reorg depth - but this is not needed yet - used only by reorg callbacks
+
+		if activeAncestor.HeaderID != oneBack.HeaderID {
+			s.logger.Info("Chaintracks service - chain reorganization detected", slog.String("new_tip_hash", header.Hash), slog.Any("new_tip_height", header.Height), slog.String("active_ancestor_hash", activeAncestor.Hash), slog.Any("active_ancestor_height", activeAncestor.Height))
+
+			// deactivate headers from the current active chain tip up to but excluding our activeAncestor:
+			if err := s.setActiveRecursivelyUntilReachAncestor(q, false, priorTip, activeAncestor); err != nil {
+				return fmt.Errorf("failed to deactivate headers during reorg: %w", err)
+			}
+
+			// the first header to activate is one before the one we are about to insert
+			// headers are activated until we reach the active ancestor
+			if err := s.setActiveRecursivelyUntilReachAncestor(q, true, oneBack, activeAncestor); err != nil {
+				return fmt.Errorf("failed to activate headers during reorg: %w", err)
+			}
+
+		}
+	}
 
 	if oneBack.IsChainTip {
 		if err := q.SetChainTipByID(oneBack.HeaderID, false); err != nil {
@@ -606,6 +638,28 @@ func (s *Service) storeLiveHeader(ctx context.Context, header wdk.ChainBlockHead
 	return nil
 }
 
+func (s *Service) setActiveRecursivelyUntilReachAncestor(q models.StorageQueries, isActive bool, first, ancestor *models.LiveBlockHeader) error {
+	var err error
+	current := first
+	for current.HeaderID != ancestor.HeaderID {
+		if err := q.SetActiveByID(current.HeaderID, isActive); err != nil {
+			return fmt.Errorf("failed to set active=%t during reorg: %w", isActive, err)
+		}
+
+		previousHash := current.PreviousHash
+		current, err = q.GetLiveHeaderByHash(previousHash)
+		if err != nil {
+			return fmt.Errorf("failed to get previous header during reorg: %w", err)
+		}
+
+		if current == nil {
+			return fmt.Errorf("previous header not found during reorg for hash: %s", previousHash)
+		}
+	}
+
+	return nil
+}
+
 func cancellableSleep(ctx context.Context, d time.Duration) error {
 	select {
 	case <-time.After(d):

pkg/services/chaintracks/gormstorage/storage_queries.go

@@ -113,6 +113,17 @@ func (i *storageQueries) SetChainTipByID(id uint, isChainTip bool) error {
 	return nil
 }
 
+func (i *storageQueries) SetActiveByID(id uint, isActive bool) error {
+	table := i.getQuery().ChaintracksLiveHeader
+	_, err := table.
+		Where(table.HeaderID.Eq(id)).
+		UpdateColumn(table.IsActive, isActive)
+	if err != nil {
+		return fmt.Errorf("failed to set active by ID: %w", err)
+	}
+	return nil
+}
+
 func (i *storageQueries) InsertNewLiveHeader(header *models.LiveBlockHeader) error {
 	table := i.getQuery().ChaintracksLiveHeader
 	err := table.Create(&dbmodels.ChaintracksLiveHeader{

pkg/services/chaintracks/models/storage_queries.go

@@ -12,6 +12,7 @@ type StorageQueries interface {
 	GetLiveHeaderByHash(hash string) (*LiveBlockHeader, error)
 	GetActiveTipLiveHeader() (*LiveBlockHeader, error)
 	SetChainTipByID(id uint, isChainTip bool) error
+	SetActiveByID(id uint, isActive bool) error
 	InsertNewLiveHeader(header *LiveBlockHeader) error
 	CountLiveHeaders() (int64, error)
 	GetLiveHeaderByHeight(height uint) (*LiveBlockHeader, error)