feat: support multiple bulk ingestor types and add methods to get chain tip (#686)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

Author: chris-4chain

pkg/defs/chaintracks.go

@@ -21,10 +21,10 @@ func ParseLiveIngestorType(str string) (LiveIngestorType, error) {
 
 // ChaintracksServiceConfig holds configuration for Chaintracks service, including the BSV network selection.
 type ChaintracksServiceConfig struct {
-	Chain            BSVNetwork              `mapstructure:"-"`
-	LiveIngestors    []LiveIngestorType      `mapstructure:"live_ingestors"`
-	CDNBulkIngestors []CDNBulkIngestorConfig `mapstructure:"cdn_bulk_ingestors"`
-	WocAPIKey        string                  `mapstructure:"woc_api_key"`
+	Chain         BSVNetwork           `mapstructure:"-"`
+	LiveIngestors []LiveIngestorType   `mapstructure:"live_ingestors"`
+	BulkIngestors []BulkIngestorConfig `mapstructure:"bulk_ingestors"`
+	WocAPIKey     string               `mapstructure:"woc_api_key"`
 }
 
 // Validate checks if the Chain field in ChaintracksServiceConfig holds a valid BSV network type.
@@ -46,9 +46,9 @@ func (c *ChaintracksServiceConfig) Validate() error {
 		}
 	}
 
-	for i := range c.CDNBulkIngestors {
-		if err := c.CDNBulkIngestors[i].Validate(); err != nil {
-			return fmt.Errorf("invalid CDN bulk ingestor config at index %d: %w", i, err)
+	for i := range c.BulkIngestors {
+		if err := c.BulkIngestors[i].Validate(); err != nil {
+			return fmt.Errorf("invalid bulk ingestor config at index %d: %w", i, err)
 		}
 	}
 
@@ -62,15 +62,66 @@ func DefaultChaintracksServiceConfig() ChaintracksServiceConfig {
 		LiveIngestors: []LiveIngestorType{
 			LiveIngestorTypeWocPoll,
 		},
-		CDNBulkIngestors: []CDNBulkIngestorConfig{
+		BulkIngestors: []BulkIngestorConfig{
 			{
-				SourceURL: BabbageBlockHeadersCDN,
+				Type: ChaintracksCDN,
+				CDNConfig: &CDNBulkIngestorConfig{
+					SourceURL: BabbageBlockHeadersCDN,
+				},
+			},
+			{
+				Type: WhatsOnChainCDN,
 			},
 		},
 		WocAPIKey: "",
 	}
 }
 
+// BulkIngestorType defines the type of bulk ingestor used to retrieve blockchain data from different sources.
+type BulkIngestorType string
+
+// Supported BulkIngestorType values.
+const (
+	ChaintracksCDN  BulkIngestorType = "chaintracks_cdn"
+	WhatsOnChainCDN BulkIngestorType = "whats_on_chain_cdn"
+)
+
+// BulkIngestorConfig defines configuration for a bulk ingestor used to fetch blockchain data such as block headers.
+type BulkIngestorConfig struct {
+	Type BulkIngestorType `mapstructure:"type"`
+
+	// CDNConfig is required if Type is ChaintracksCDN
+	CDNConfig *CDNBulkIngestorConfig `mapstructure:"cdn_config"`
+}
+
+// Validate checks if the BulkIngestorConfig is properly configured for the specified Type and returns an error if not valid.
+func (b *BulkIngestorConfig) Validate() error {
+	switch b.Type {
+	case ChaintracksCDN:
+		if b.CDNConfig == nil {
+			return fmt.Errorf("cdn_config is required for chaintracks_cdn bulk ingestor")
+		}
+		if err := b.CDNConfig.Validate(); err != nil {
+			return fmt.Errorf("invalid cdn_config: %w", err)
+		}
+	case WhatsOnChainCDN:
+		// No additional config required for WhatsOnChainCDN
+	default:
+		return fmt.Errorf("invalid bulk ingestor type: %s", b.Type)
+	}
+	return nil
+}
+
+// String returns a human-readable representation of the BulkIngestorConfig, including type and CDN source URL when applicable.
+func (b *BulkIngestorConfig) String() string {
+	detailed := ""
+	if b.Type == ChaintracksCDN && b.CDNConfig != nil {
+		detailed = fmt.Sprintf(" (source_url=%s)", b.CDNConfig.SourceURL)
+	}
+
+	return fmt.Sprintf("%s%s", b.Type, detailed)
+}
+
 // CDNBulkIngestorConfig holds configuration options for a bulk ingestor that fetches block headers from a CDN source.
 type CDNBulkIngestorConfig struct {
 	SourceURL string `mapstructure:"source_url"`

pkg/services/chaintracks/chaintracks_initializers.go

@@ -12,6 +12,7 @@ import (
 type Initializers struct {
 	WOCLiveIngestorPollFactory func(logger *slog.Logger, config defs.ChaintracksServiceConfig) LiveIngestor
 	CDNBulkIngestorFactory     func(logger *slog.Logger, chain defs.BSVNetwork, config defs.CDNBulkIngestorConfig) BulkIngestor
+	WOCBulkIngestorFactory     func(logger *slog.Logger, chain defs.BSVNetwork, apiKey string) BulkIngestor
 }
 
 // DefaultInitializers returns an Initializers struct with the default WOCLiveIngestorPollFactory implementation.
@@ -23,6 +24,9 @@ func DefaultInitializers() Initializers {
 		CDNBulkIngestorFactory: func(logger *slog.Logger, chain defs.BSVNetwork, config defs.CDNBulkIngestorConfig) BulkIngestor {
 			return ingest.NewBulkIngestorCDN(logger, chain, config)
 		},
+		WOCBulkIngestorFactory: func(logger *slog.Logger, chain defs.BSVNetwork, apiKey string) BulkIngestor {
+			return ingest.NewBulkIngestorWOC(logger, chain, ingest.BulkIngestorWocOpts.WithAPIKey(apiKey))
+		},
 	}
 }
 
@@ -33,6 +37,12 @@ func createInitializers(inits ...Initializers) Initializers {
 		if in.WOCLiveIngestorPollFactory != nil {
 			finalInits.WOCLiveIngestorPollFactory = in.WOCLiveIngestorPollFactory
 		}
+		if in.CDNBulkIngestorFactory != nil {
+			finalInits.CDNBulkIngestorFactory = in.CDNBulkIngestorFactory
+		}
+		if in.WOCBulkIngestorFactory != nil {
+			finalInits.WOCBulkIngestorFactory = in.WOCBulkIngestorFactory
+		}
 	}
 
 	return finalInits

pkg/services/chaintracks/chaintracks_service.go

@@ -205,6 +205,28 @@ func (s *Service) GetInfo(ctx context.Context) (*models.InfoResponse, error) {
 	}, nil
 }
 
+// FindChainTipHeader retrieves the current chain tip block header from storage or returns an error if not found.
+func (s *Service) FindChainTipHeader(ctx context.Context) (*wdk.ChainBlockHeader, error) {
+	tipHeader, err := s.storage.Query(ctx).GetActiveTipLiveHeader()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get active tip live header: %w", err)
+	}
+	if tipHeader == nil {
+		return nil, fmt.Errorf("no active tip live header found: %w", wdk.ErrNotFoundError)
+	}
+	return &tipHeader.ChainBlockHeader, nil
+}
+
+// FindChainTipHash retrieves the hash of the current chain tip.
+// Returns the hash as a string or an error if the chain tip header cannot be found.
+func (s *Service) FindChainTipHash(ctx context.Context) (string, error) {
+	if tipHeader, err := s.FindChainTipHeader(ctx); err != nil {
+		return "", fmt.Errorf("failed to find chain tip header: %w", err)
+	} else {
+		return tipHeader.Hash, nil
+	}
+}
+
 func (s *Service) getMissingBlockHeader(ctx context.Context, hash string) *wdk.ChainBlockHeader {
 	for _, liveIngestor := range s.liveIngestors {
 		header, err := liveIngestor.Ingestor.GetHeaderByHash(ctx, hash)

pkg/services/chaintracks/chaintracks_service_test.go

@@ -37,11 +37,25 @@ func TestService_Lifecycle(t *testing.T) {
 	assert.Equal(t, defs.NetworkMainnet, info.Chain)
 	assert.Equal(t, "gorm-sqlite-inmemory", info.Storage)
 	assert.Equal(t, []string{"woc_poll"}, info.LiveIngestors)
-	assert.Equal(t, []string{"https://cdn.projectbabbage.com/blockheaders"}, info.BulkIngestors)
-	//assert.Equal(t, int(info.HeightLive)-1, int(info.HeightBulk)) // TODO: Uncomment it when WoC Bulk ingestor is implemented (Babbage CDN ingestor doesn't provide the latest data)
+	assert.Equal(t, []string{"chaintracks_cdn (source_url=https://cdn.projectbabbage.com/blockheaders)", "whats_on_chain_cdn"}, info.BulkIngestors)
 	t.Logf("Bulk height: %d", info.HeightBulk) // TODO: When we use live data (not mocked), this value is not constant; will be changed to assert.Equal later
 	t.Logf("Live height: %d", info.HeightLive)
 
+	// when:
+	height, err := service.GetPresentHeight(t.Context())
+
+	// then:
+	require.NoError(t, err, "get present height should not return error")
+	require.Greater(t, height, uint(900000), "present height should be greater than 900000")
+
+	// when:
+	tipHeader, err := service.FindChainTipHeader(t.Context())
+
+	// then:
+	require.NoError(t, err, "find chain tip header should not return error")
+	require.NotNil(t, tipHeader, "tip header should not be nil")
+	require.Equal(t, height, tipHeader.Height, "tip header height should be equal to present height")
+
 	// when:
 	service.Destroy()
 

pkg/services/chaintracks/create_ingestors.go

@@ -40,15 +40,26 @@ func createLiveIngestors(logger *slog.Logger, config defs.ChaintracksServiceConf
 }
 
 func createBulkIngestors(logger *slog.Logger, config defs.ChaintracksServiceConfig, initializers Initializers) []NamedBulkIngestor {
-	logger.Info("Chaintracks service - creating bulk ingestors", slog.Any("configured_sources", config.CDNBulkIngestors))
-
-	ingestors := make([]NamedBulkIngestor, 0, len(config.CDNBulkIngestors))
-	for _, cdnConfig := range config.CDNBulkIngestors {
-		ingestor := initializers.CDNBulkIngestorFactory(logger, config.Chain, cdnConfig)
-		ingestors = append(ingestors, NamedBulkIngestor{
-			Name:     cdnConfig.SourceURL,
-			Ingestor: ingestor,
-		})
+	logger.Info("Chaintracks service - creating bulk ingestors", slog.Any("configured_sources", config.BulkIngestors))
+
+	ingestors := make([]NamedBulkIngestor, 0, len(config.BulkIngestors))
+	for _, ingestorConfig := range config.BulkIngestors {
+		switch ingestorConfig.Type {
+		case defs.ChaintracksCDN:
+			ingestor := initializers.CDNBulkIngestorFactory(logger, config.Chain, *ingestorConfig.CDNConfig)
+			ingestors = append(ingestors, NamedBulkIngestor{
+				Name:     ingestorConfig.String(),
+				Ingestor: ingestor,
+			})
+		case defs.WhatsOnChainCDN:
+			ingestor := initializers.WOCBulkIngestorFactory(logger, config.Chain, config.WocAPIKey)
+			ingestors = append(ingestors, NamedBulkIngestor{
+				Name:     ingestorConfig.String(),
+				Ingestor: ingestor,
+			})
+		default:
+			logger.Warn("Chaintracks service - unsupported bulk ingestor type, skipping", slog.String("ingestor_type", string(ingestorConfig.Type)))
+		}
 	}
 
 	return ingestors

pkg/services/chaintracks/ingest/bulk_ingestor_woc.go

@@ -123,11 +123,16 @@ func (b *BulkIngestorWOC) bulkFileDownloader() BulkFileDownloader {
 			return BulkFileData{}, fmt.Errorf("failed to download bulk header file %s: %w", fileInfo.FileName, err)
 		}
 
-		return BulkFileData{
+		fileData := BulkFileData{
 			Info:       fileInfo,
 			Data:       content,
 			AccessedAt: time.Now(),
-		}, nil
+		}
+
+		// NOTE: We compute and set the FileHash here since WoC does not provide it in metadata
+		fileData.Info.FileHash = fileData.Hash()
+
+		return fileData, nil
 	}
 }
 

pkg/services/chaintracks/ingest/cdn_data.go

@@ -66,7 +66,7 @@ func (b *BulkFileData) Validate() error {
 		return fmt.Errorf("bulk file data is empty")
 	}
 
-	dataHash := crypto.Sha256(b.Data)
+	dataHash := b.Hash()
 	if !bytes.Equal(dataHash, b.Info.FileHash) {
 		base64Expected := base64.StdEncoding.EncodeToString(b.Info.FileHash)
 		base64Got := base64.StdEncoding.EncodeToString(dataHash)
@@ -84,6 +84,11 @@ func (b *BulkFileData) Validate() error {
 	return nil
 }
 
+// Hash computes and returns the SHA-256 hash of the bulk file data bytes as a byte slice.
+func (b *BulkFileData) Hash() []byte {
+	return crypto.Sha256(b.Data)
+}
+
 // Len returns the number of block headers contained in the bulk file data.
 func (b *BulkFileData) Len() int {
 	return len(b.Data) / 80

pkg/services/chaintracks/models/height_range.go

@@ -29,9 +29,9 @@ func (h *HeightRanges) Validate() error {
 			return fmt.Errorf("bulk storage must start with genesis header")
 		}
 
-		if !h.Live.IsEmpty() && h.Bulk.MaxHeight+1 != h.Live.MinHeight {
+		if !h.Live.IsEmpty() && h.Bulk.MaxHeight+1 < h.Live.MinHeight {
 			gap := must.ConvertToIntFromUnsigned(h.Live.MinHeight) - must.ConvertToIntFromUnsigned(h.Bulk.MaxHeight) - 1
-			return fmt.Errorf("there is a gap (%d) or overlap between bulk and live header storage, bulk max height: %d, live min height: %d", gap, h.Bulk.MaxHeight, h.Live.MinHeight)
+			return fmt.Errorf("there is a gap (%d) between bulk and live header storage, bulk max height: %d, live min height: %d", gap, h.Bulk.MaxHeight, h.Live.MinHeight)
 		}
 	}