fix: fix typo, rename key func

node/pkg/aggregator/aggregator_test.go

@@ -121,7 +121,7 @@ func TestPublishGlobalAggregateAndProof(t *testing.T) {
 	}
 
 	ch := make(chan SubmissionData)
-	err = db.Subscribe(ctx, keys.SubmissionDataStreamKeyV2(node.Name), ch)
+	err = db.Subscribe(ctx, keys.SubmissionDataStreamKey(node.Name), ch)
 	if err != nil {
 		t.Fatal("error subscribing to stream")
 	}

node/pkg/aggregator/globalaggregatebulkwriter.go

@@ -108,7 +108,7 @@ func (s *GlobalAggregateBulkWriter) receive(ctx context.Context) {
 }
 
 func (s *GlobalAggregateBulkWriter) receiveEach(ctx context.Context, configName string) {
-	err := db.Subscribe(ctx, keys.SubmissionDataStreamKeyV2(configName), s.ReceiveChannels[configName])
+	err := db.Subscribe(ctx, keys.SubmissionDataStreamKey(configName), s.ReceiveChannels[configName])
 	if err != nil {
 		log.Error().Err(err).Str("Player", "Aggregator").Msg("failed to subscribe to submission stream")
 	}

node/pkg/aggregator/utils.go

@@ -26,7 +26,7 @@ func PublishGlobalAggregateAndProof(ctx context.Context, name string, globalAggr
 		GlobalAggregate: globalAggregate,
 		Proof:           proof,
 	}
-	return db.Publish(ctx, keys.SubmissionDataStreamKeyV2(name), data)
+	return db.Publish(ctx, keys.SubmissionDataStreamKey(name), data)
 }
 
 func getLatestRoundId(ctx context.Context, configId int32) (int32, error) {

node/pkg/common/keys/keys.go

@@ -1,5 +1,5 @@
 package keys
 
-func SubmissionDataStreamKeyV2(name string) string {
-	return "submissionDataSteram:" + name
+func SubmissionDataStreamKey(name string) string {
+	return "submissionDataStream:" + name
 }

node/pkg/dal/collector/collector.go

@@ -107,7 +107,7 @@ func NewCollector(ctx context.Context, configs []Config) (*Collector, error) {
 		collector.OutgoingStream[config.ID] = make(chan *dalcommon.OutgoingSubmissionData, 1000)
 		collector.Symbols[config.ID] = config.Name
 		collector.FeedHashes[config.ID] = crypto.Keccak256([]byte(config.Name))
-		redisTopics = append(redisTopics, keys.SubmissionDataStreamKeyV2(config.Name))
+		redisTopics = append(redisTopics, keys.SubmissionDataStreamKey(config.Name))
 	}
 
 	baseRediscribe, err := db.NewRediscribe(

node/pkg/dal/tests/main_test.go

@@ -37,7 +37,7 @@ type TestItems struct {
 }
 
 func testPublishData(ctx context.Context, name string, submissionData aggregator.SubmissionData) error {
-	return db.Publish(ctx, keys.SubmissionDataStreamKeyV2(name), submissionData)
+	return db.Publish(ctx, keys.SubmissionDataStreamKey(name), submissionData)
 }
 
 func generateSampleSubmissionData(configId int32, value int64, timestamp time.Time, round int32, symbol string) (*aggregator.SubmissionData, error) {

node/pkg/reporter/app_test.go

@@ -61,7 +61,7 @@ func TestWsDataHandling(t *testing.T) {
 		t.Fatalf("error generating sample submission data: %v", err)
 	}
 
-	err = db.Publish(ctx, keys.SubmissionDataStreamKeyV2("test-aggregate"), sampleSubmissionData)
+	err = db.Publish(ctx, keys.SubmissionDataStreamKey("test-aggregate"), sampleSubmissionData)
 	if err != nil {
 		t.Fatalf("error publishing sample submission data: %v", err)
 	}