Add data provider

apps/lib/data_provider/command.go

@@ -0,0 +1,50 @@
+package data_provider
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/pkgerrors"
+	"github.com/spf13/cobra"
+)
+
+var DataProviderCmd = &cobra.Command{
+	Use:   "start",
+	Short: "Start a process to fetch prices from data sources",
+	RunE:  runDataProvider,
+}
+
+// required
+const ConfigFilePathFlag = "config-file-path"
+const WebsocketUrl = "ws-url"
+
+func init() {
+	DataProviderCmd.Flags().StringP(ConfigFilePathFlag, "c", "", "the path of your config json file")
+	DataProviderCmd.Flags().StringP(WebsocketUrl, "w", "", "the websocket url to write updates to")
+
+	DataProviderCmd.MarkFlagRequired(ConfigFilePathFlag)
+}
+
+func runDataProvider(cmd *cobra.Command, args []string) error {
+	configFilePath, _ := cmd.Flags().GetString(ConfigFilePathFlag)
+	wsUrl, _ := cmd.Flags().GetString(WebsocketUrl)
+
+	mainLogger := mainLogger()
+
+	zerolog.TimeFieldFormat = time.RFC3339Nano
+	zerolog.DurationFieldUnit = time.Nanosecond
+	zerolog.ErrorStackMarshaler = pkgerrors.MarshalStack
+
+	mainLogger.Info().Msg("Starting data provider")
+
+	config, err := loadConfig(configFilePath)
+	if err != nil {
+		return fmt.Errorf("error loading config: %v", err)
+	}
+
+	runner := NewDataProviderRunner(*config, wsUrl)
+	runner.Run()
+
+	return nil
+}

apps/lib/data_provider/config.go

@@ -0,0 +1,20 @@
+package data_provider
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+)
+
+func loadConfig(configPath string) (*DataProviderConfig, error) {
+	configBytes, err := os.ReadFile(configPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read config file: %v", err)
+	}
+
+	var config DataProviderConfig
+	if err := json.Unmarshal(configBytes, &config); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal config file: %v", err)
+	}
+	return &config, nil
+}

apps/lib/data_provider/data_source.go

@@ -0,0 +1,31 @@
+package data_provider
+
+type dataSource interface {
+	// Add all value updates to updatesCh
+	Run(updatesCh chan DataSourceUpdateMap)
+	GetDataSourceId() DataSourceId
+}
+
+func buildDataSources(config DataProviderConfig) []dataSource {
+	// group by data source id to support batched feeds
+	sourceConfigsByDataSource := make(map[DataSourceId][]DataProviderSourceConfig)
+	for _, sourceConfig := range config.Sources {
+		dataSourceId := sourceConfig.DataSourceId
+		if _, ok := sourceConfigsByDataSource[dataSourceId]; !ok {
+			sourceConfigsByDataSource[dataSourceId] = make([]DataProviderSourceConfig, 0)
+		}
+		sourceConfigsByDataSource[dataSourceId] = append(sourceConfigsByDataSource[dataSourceId], sourceConfig)
+
+	}
+
+	// initialize data sources
+	allDataSources := make([]dataSource, 0)
+	for dataSourceId, sourceConfigs := range sourceConfigsByDataSource {
+		dataSourceBuilder := GetDataSourceBuilder(dataSourceId)
+		dataSources := dataSourceBuilder(sourceConfigs)
+
+		allDataSources = append(allDataSources, dataSources...)
+	}
+
+	return allDataSources
+}

apps/lib/data_provider/data_source_registry.go

@@ -0,0 +1,12 @@
+package data_provider
+
+func GetDataSourceBuilder(dataSourceId DataSourceId) func([]DataProviderSourceConfig) []dataSource {
+	switch dataSourceId {
+	case UniswapV2DataSourceId:
+		return getUniswapV2DataSources
+	case RandomDataSourceId:
+		return getRandomDataSource
+	default:
+		panic("unknown data source id " + dataSourceId)
+	}
+}

apps/lib/data_provider/logger.go

@@ -0,0 +1,22 @@
+package data_provider
+
+import (
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+func baseAppLogger() zerolog.Logger {
+	return log.With().Str("application", "stork-data-provider").Logger()
+}
+
+func mainLogger() zerolog.Logger {
+	return baseAppLogger().With().Str("service", "main").Logger()
+}
+
+func writerLogger() zerolog.Logger {
+	return baseAppLogger().With().Str("service", "writer").Logger()
+}
+
+func dataSourceLogger(dataSourceId DataSourceId) zerolog.Logger {
+	return baseAppLogger().With().Str("service", "data_source").Str("data_source_id", string(dataSourceId)).Logger()
+}

apps/lib/data_provider/model.go

@@ -0,0 +1,40 @@
+package data_provider
+
+import (
+	"time"
+)
+
+type (
+	DataSourceId string
+	ValueId      string
+
+	DataProviderSourceConfig struct {
+		Id           ValueId      `json:"id"`
+		DataSourceId DataSourceId `json:"dataSource"`
+		Config       any          `json:"config"`
+	}
+
+	DataProviderConfig struct {
+		Sources []DataProviderSourceConfig `json:"sources,omitempty"`
+	}
+
+	DataSourceValueUpdate struct {
+		ValueId      ValueId
+		DataSourceId DataSourceId
+		Timestamp    time.Time
+		Value        float64
+	}
+
+	DataSourceUpdateMap map[ValueId]DataSourceValueUpdate
+
+	ValueUpdate struct {
+		PublishTimestamp int64   `json:"t"`
+		ValueId          ValueId `json:"a"`
+		Value            string  `json:"v"`
+	}
+
+	ValueUpdateWebsocketMessage struct {
+		Type string        `json:"type"`
+		Data []ValueUpdate `json:"data"`
+	}
+)

apps/lib/data_provider/random.go

@@ -0,0 +1,71 @@
+package data_provider
+
+import (
+	"math/rand"
+	"time"
+
+	"github.com/mitchellh/mapstructure"
+)
+
+const RandomDataSourceId = "RANDOM_NUMBER"
+
+type randomConfig struct {
+	UpdateFrequency string  `json:"updateFrequency"`
+	MinValue        float64 `json:"minValue"`
+	MaxValue        float64 `json:"maxValue"`
+}
+
+type randomConnector struct {
+	valueId         ValueId
+	config          randomConfig
+	updateFrequency time.Duration
+}
+
+func newRandomConnector(sourceConfig DataProviderSourceConfig) *randomConnector {
+	var randomConfig randomConfig
+	mapstructure.Decode(sourceConfig.Config, &randomConfig)
+
+	updateFrequency, err := time.ParseDuration(randomConfig.UpdateFrequency)
+	if err != nil {
+		panic("unable to parse update frequency: " + randomConfig.UpdateFrequency)
+	}
+
+	return &randomConnector{
+		valueId:         sourceConfig.Id,
+		config:          randomConfig,
+		updateFrequency: updateFrequency,
+	}
+}
+
+func (r *randomConnector) GetUpdate() (DataSourceUpdateMap, error) {
+	randValue := r.config.MinValue + rand.Float64()*(r.config.MaxValue-r.config.MinValue)
+
+	updateMap := DataSourceUpdateMap{
+		r.valueId: DataSourceValueUpdate{
+			ValueId:      r.valueId,
+			DataSourceId: r.GetDataSourceId(),
+			Timestamp:    time.Now(),
+			Value:        randValue,
+		},
+	}
+
+	return updateMap, nil
+}
+
+func (r *randomConnector) GetUpdateFrequency() time.Duration {
+	return r.updateFrequency
+}
+
+func (r *randomConnector) GetDataSourceId() DataSourceId {
+	return RandomDataSourceId
+}
+
+func getRandomDataSource(sourceConfigs []DataProviderSourceConfig) []dataSource {
+	dataSources := make([]dataSource, 0)
+	for _, sourceConfig := range sourceConfigs {
+		connector := newRandomConnector(sourceConfig)
+		dataSource := newScheduledDataSource(connector)
+		dataSources = append(dataSources, dataSource)
+	}
+	return dataSources
+}