Merge branch 'develop' into 9679-add-team-support-for-referrals

CHANGELOG.md

@@ -132,6 +132,7 @@
 - [9542](https://github.com/vegaprotocol/vega/issues/9542) - Ensure all per asset accounts exist after migration
 - [8475](https://github.com/vegaprotocol/vega/issues/8475) - Emit margin levels event when margin balance is cleared.
 - [9664](https://github.com/vegaprotocol/vega/issues/9664) - Handle pagination of release request with github.
+- [9681](https://github.com/vegaprotocol/vega/issues/9681) - Move referral set reward factor to the referral set stats events
 
 ### 🐛 Fixes
 

core/events/referral_set.go

@@ -54,13 +54,12 @@ type ReferralSetStatsUpdated struct {
 func (t ReferralSetStatsUpdated) Unwrap() *types.ReferralSetStats {
 	volume, _ := num.UintFromString(t.e.ReferralSetRunningNotionalTakerVolume, 10)
 	stats := map[types.PartyID]*types.RefereeStats{}
+	rewardFactor, _ := num.DecimalFromString(t.e.RewardFactor)
 
 	for _, stat := range t.e.RefereesStats {
 		discountFactor, _ := num.DecimalFromString(stat.DiscountFactor)
-		rewardFactor, _ := num.DecimalFromString(stat.RewardFactor)
 		stats[types.PartyID(stat.PartyId)] = &types.RefereeStats{
 			DiscountFactor: discountFactor,
-			RewardFactor:   rewardFactor,
 		}
 	}
 
@@ -69,6 +68,7 @@ func (t ReferralSetStatsUpdated) Unwrap() *types.ReferralSetStats {
 		SetID:                    types.ReferralSetID(t.e.SetId),
 		ReferralSetRunningVolume: volume,
 		RefereesStats:            stats,
+		RewardFactor:             rewardFactor,
 	}
 }
 
@@ -91,7 +91,6 @@ func NewReferralSetStatsUpdatedEvent(ctx context.Context, update *types.Referral
 		refereesStats = append(refereesStats, &eventspb.RefereeStats{
 			PartyId:                  string(partyID),
 			DiscountFactor:           stat.DiscountFactor.String(),
-			RewardFactor:             stat.RewardFactor.String(),
 			EpochNotionalTakerVolume: stat.TakerVolume.String(),
 		})
 	}
@@ -107,6 +106,7 @@ func NewReferralSetStatsUpdatedEvent(ctx context.Context, update *types.Referral
 			AtEpoch:                               update.AtEpoch,
 			ReferralSetRunningNotionalTakerVolume: update.ReferralSetRunningVolume.String(),
 			RefereesStats:                         refereesStats,
+			RewardFactor:                          update.RewardFactor.String(),
 		},
 	}
 }

core/integration/steps/the_referral_set_stats_should_be.go

@@ -37,29 +37,34 @@ func TheReferralSetStatsShouldBe(broker *stubs.BrokerStub, code, epochStr, volum
 				return fmt.Errorf("refferal set stats for set ID %q at epoch %q expect a running volume of %v, but got %v", code, epochStr, volumeStr, stat.ReferralSetRunningVolume)
 			}
 
-			return compareRefereesStats(expectedRefereesStats, stat.RefereesStats)
+			return compareRefereesStats(expectedRefereesStats, stat.RefereesStats, stat.RewardFactor)
 		}
 	}
 
 	return fmt.Errorf("no stats found for set ID %q at epoch %q", code, epochStr)
 }
 
-func parseReferralStatsShouldBeTable(table *godog.Table) (map[types.PartyID]*types.RefereeStats, error) {
+type refereeStats struct {
+	DiscountFactor num.Decimal
+	RewardFactor   num.Decimal
+}
+
+func parseReferralStatsShouldBeTable(table *godog.Table) (map[types.PartyID]*refereeStats, error) {
 	rows := StrictParseTable(table, []string{
 		"party",
 		"discount factor",
 		"reward factor",
 	}, []string{})
 
-	stats := map[types.PartyID]*types.RefereeStats{}
+	stats := map[types.PartyID]*refereeStats{}
 	for _, row := range rows {
 		specificRow := newReferralSetStatsShouldBeRow(row)
 		partyID := specificRow.Party()
 		_, alreadyRegistered := stats[partyID]
 		if alreadyRegistered {
 			return nil, fmt.Errorf("cannot have more than one expectation for party %q", partyID)
 		}
-		stats[partyID] = &types.RefereeStats{
+		stats[partyID] = &refereeStats{
 			DiscountFactor: specificRow.DiscountFactor(),
 			RewardFactor:   specificRow.RewardFactor(),
 		}
@@ -68,7 +73,11 @@ func parseReferralStatsShouldBeTable(table *godog.Table) (map[types.PartyID]*typ
 	return stats, nil
 }
 
-func compareRefereesStats(expectedRefereesStats, foundRefereesStats map[types.PartyID]*types.RefereeStats) error {
+func compareRefereesStats(
+	expectedRefereesStats map[types.PartyID]*refereeStats,
+	foundRefereesStats map[types.PartyID]*types.RefereeStats,
+	foundRewardFactor num.Decimal,
+) error {
 	foundRefereesIDs := maps.Keys(foundRefereesStats)
 	expectedRefereesIDs := maps.Keys(expectedRefereesStats)
 
@@ -108,8 +117,8 @@ func compareRefereesStats(expectedRefereesStats, foundRefereesStats map[types.Pa
 		refereeIDStr := string(refereeID)
 		foundRefereeStats := foundRefereesStats[refereeID]
 		expectedRefereeStats := expectedRefereesStats[refereeID]
-		if !foundRefereeStats.RewardFactor.Equal(expectedRefereeStats.RewardFactor) {
-			return fmt.Errorf("expecting reward factor of %v but got %v for party %q", expectedRefereeStats.RewardFactor.String(), foundRefereeStats.RewardFactor.String(), refereeIDStr)
+		if !expectedRefereeStats.RewardFactor.Equal(foundRewardFactor) {
+			return fmt.Errorf("expecting reward factor of %v but got %v for party %q", expectedRefereeStats.RewardFactor.String(), foundRewardFactor.String(), refereeIDStr)
 		}
 		if !foundRefereeStats.DiscountFactor.Equal(expectedRefereeStats.DiscountFactor) {
 			return fmt.Errorf("expecting discount factor of %v but got %v for party %q", expectedRefereeStats.DiscountFactor.String(), foundRefereeStats.DiscountFactor.String(), refereeIDStr)

core/referral/engine.go

@@ -145,6 +145,7 @@ func (e *Engine) CreateReferralSet(ctx context.Context, party types.PartyID, det
 			JoinedAt:       now,
 			StartedAtEpoch: e.currentEpoch,
 		},
+		CurrentRewardFactor: num.DecimalZero(),
 	}
 
 	e.sets[deterministicSetID] = &newSet
@@ -250,12 +251,12 @@ func (e *Engine) RewardsFactorForParty(party types.PartyID) num.Decimal {
 		return num.DecimalZero()
 	}
 
-	factors, ok := e.factorsByReferee[party]
+	setID, ok := e.referees[party]
 	if !ok {
 		return num.DecimalZero()
 	}
 
-	return factors.RewardFactor
+	return e.sets[setID].CurrentRewardFactor
 }
 
 func (e *Engine) RewardsFactorMultiplierAppliedForParty(party types.PartyID) num.Decimal {
@@ -419,6 +420,7 @@ func (e *Engine) loadReferralSetsFromSnapshot(setsProto *snapshotpb.ReferralSets
 				JoinedAt:       time.Unix(0, setProto.Referrer.JoinedAt),
 				StartedAtEpoch: setProto.Referrer.StartedAtEpoch,
 			},
+			CurrentRewardFactor: num.MustDecimalFromString(setProto.CurrentRewardFactor),
 		}
 
 		e.referrers[types.PartyID(setProto.Referrer.PartyId)] = setID
@@ -490,17 +492,27 @@ func (e *Engine) computeFactorsByReferee(ctx context.Context, epoch uint64, take
 	e.factorsByReferee = map[types.PartyID]*types.RefereeStats{}
 
 	allStats := map[types.ReferralSetID]*types.ReferralSetStats{}
+	tiersLen := len(e.currentProgram.BenefitTiers)
 
 	for setID := range e.sets {
-		allStats[setID] = &types.ReferralSetStats{
+		set := &types.ReferralSetStats{
 			AtEpoch:                  epoch,
 			SetID:                    setID,
 			RefereesStats:            map[types.PartyID]*types.RefereeStats{},
 			ReferralSetRunningVolume: e.referralSetsNotionalVolumes.RunningSetVolumeForWindow(setID, e.currentProgram.WindowLength),
+			RewardFactor:             num.DecimalZero(),
 		}
-	}
 
-	tiersLen := len(e.currentProgram.BenefitTiers)
+		for i := tiersLen - 1; i >= 0; i-- {
+			tier := e.currentProgram.BenefitTiers[i]
+			if set.RewardFactor.Equal(num.DecimalZero()) && set.ReferralSetRunningVolume.GTE(tier.MinimumRunningNotionalTakerVolume) {
+				set.RewardFactor = tier.ReferralRewardFactor
+			}
+		}
+
+		e.sets[setID].CurrentRewardFactor = set.RewardFactor
+		allStats[setID] = set
+	}
 
 	for party, setID := range e.referees {
 		set := e.sets[setID]
@@ -531,9 +543,6 @@ func (e *Engine) computeFactorsByReferee(ctx context.Context, epoch uint64, take
 			if refereeStats.DiscountFactor.Equal(num.DecimalZero()) && epochCount >= tier.MinimumEpochs.Uint64() && runningVolumeForSet.GTE(tier.MinimumRunningNotionalTakerVolume) {
 				refereeStats.DiscountFactor = tier.ReferralDiscountFactor
 			}
-			if refereeStats.RewardFactor.Equal(num.DecimalZero()) && runningVolumeForSet.GTE(tier.MinimumRunningNotionalTakerVolume) {
-				refereeStats.RewardFactor = tier.ReferralRewardFactor
-			}
 		}
 	}
 

core/referral/snapshot.go

@@ -123,6 +123,7 @@ func (e *SnapshottedEngine) serialiseReferralSets() ([]byte, error) {
 				JoinedAt:       set.Referrer.JoinedAt.UnixNano(),
 				StartedAtEpoch: set.Referrer.StartedAtEpoch,
 			},
+			CurrentRewardFactor: set.CurrentRewardFactor.String(),
 		}
 
 		for _, r := range set.Referees {

core/types/referral_set.go

@@ -16,17 +16,19 @@ type ReferralSet struct {
 
 	Referrer *Membership
 	Referees []*Membership
+
+	CurrentRewardFactor num.Decimal
 }
 
 type ReferralSetStats struct {
 	AtEpoch                  uint64
 	SetID                    ReferralSetID
 	ReferralSetRunningVolume *num.Uint
 	RefereesStats            map[PartyID]*RefereeStats
+	RewardFactor             num.Decimal
 }
 
 type RefereeStats struct {
 	DiscountFactor num.Decimal
-	RewardFactor   num.Decimal
 	TakerVolume    *num.Uint
 }

datanode/entities/referral_set_stats.go

@@ -15,6 +15,7 @@ type (
 		ReferralSetRunningNotionalTakerVolume string
 		RefereesStats                         []*eventspb.RefereeStats
 		VegaTime                              time.Time
+		RewardFactor                          string
 	}
 
 	FlattenReferralSetStats struct {
@@ -23,8 +24,8 @@ type (
 		VegaTime                              time.Time
 		PartyID                               string
 		DiscountFactor                        string
-		RewardFactor                          string
 		EpochNotionalTakerVolume              string
+		RewardFactor                          string
 	}
 
 	ReferralSetStatsCursor struct {
@@ -91,6 +92,7 @@ func ReferralSetStatsFromProto(proto *eventspb.ReferralSetStatsUpdated, vegaTime
 		ReferralSetRunningNotionalTakerVolume: proto.ReferralSetRunningNotionalTakerVolume,
 		RefereesStats:                         proto.RefereesStats,
 		VegaTime:                              vegaTime,
+		RewardFactor:                          proto.RewardFactor,
 	}, nil
 }
 

datanode/networkhistory/service_test.go

@@ -363,12 +363,12 @@ func TestMain(t *testing.M) {
 		log.Infof("%s", goldenSourceHistorySegment[4000].HistorySegmentID)
 		log.Infof("%s", goldenSourceHistorySegment[5000].HistorySegmentID)
 
-		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[1000].HistorySegmentID, "QmbHP54JwbyeoeMGQ4V9a6vimoKXFubZHYS4gVZaYfSpWY", snapshots)
-		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[2000].HistorySegmentID, "QmZypZFHgsZBS2gXUGshMXjR8qGtCG7gpWLpixG8BGD8Z5", snapshots)
-		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[2500].HistorySegmentID, "QmXfYqhMDw3usXHbQqzbA6tGHmgVNT1xReh8Cb3ukAkLWj", snapshots)
-		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[3000].HistorySegmentID, "QmdoZ3iTumC8AxVmtGmW9LPf9xhycYgXAiXfrw9ge4v3o6", snapshots)
-		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[4000].HistorySegmentID, "QmfDqJFZjrfz8XR2BWbaLsWdKC92pFeq6wNf9TdcrCuDwZ", snapshots)
-		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[5000].HistorySegmentID, "QmVWG8dub8iDVC66syE8xxi8CahDFyk7yk3KooAfxLnXSa", snapshots)
+		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[1000].HistorySegmentID, "QmY6Wb6eRCZ632u8ZvVq28WaX3VD4X8SkEubMUeg28VNQu", snapshots)
+		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[2000].HistorySegmentID, "QmNh8uc5kwb5Vu1EtE98qLNYsMo9zH8TgWQx5yFsQKpy7s", snapshots)
+		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[2500].HistorySegmentID, "QmVuxb98zbhAB7yaYG1kNaCAqW6JfYU3ueonZwzv1idNj4", snapshots)
+		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[3000].HistorySegmentID, "QmetkoN1mvdJGmUoLe8kY6AbzqovxSvFBZb8EpjZ9tS6oH", snapshots)
+		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[4000].HistorySegmentID, "QmPyaanqJzBqsC8WovaAafdHgUMwoXJZD2cRimtUgfFRuH", snapshots)
+		panicIfHistorySegmentIdsNotEqual(goldenSourceHistorySegment[5000].HistorySegmentID, "QmSzCeXsroNjZkCinXkuNs5Ju6GRjj1TvRUmuD3SVmkNdX", snapshots)
 	}, postgresRuntimePath, sqlFs)
 
 	if exitCode != 0 {

datanode/sqlstore/migrations/0033_referral_programs.sql

@@ -47,6 +47,7 @@ create table referral_set_stats(
     set_id bytea not null,
     at_epoch bigint not null,
     referral_set_running_notional_taker_volume numeric,
+    reward_factor numeric,
     referees_stats jsonb not null,
     vega_time timestamp with time zone not null,
     primary key (vega_time, set_id)
@@ -59,7 +60,7 @@ drop view if exists referral_set_referee_stats;
 
 create view referral_set_referee_stats as (
   select set_id, at_epoch, referral_set_running_notional_taker_volume, stats.referee_stats->>'party_id' as party_id,
-         stats.referee_stats->>'discount_factor' as discount_factor, stats.referee_stats->>'reward_factor' as reward_factor,
+         stats.referee_stats->>'discount_factor' as discount_factor, reward_factor,
          vega_time
   from referral_set_stats,
        jsonb_array_elements(referees_stats) with ordinality stats(referee_stats, position)

datanode/sqlstore/migrations/0045_unnecessary_referral_set_referee_stats.sql

@@ -8,6 +8,12 @@ alter table referral_set_stats
 alter table referral_set_stats
     alter column referral_set_running_notional_taker_volume set not null;
 
+alter table referral_set_stats
+    alter column reward_factor type text using (reward_factor::text);
+
+alter table referral_set_stats
+    alter column reward_factor set not null;
+
 
 -- +goose Down
 
@@ -18,9 +24,8 @@ select set_id,
        referral_set_running_notional_taker_volume,
        stats.referee_stats ->> 'party_id'        as party_id,
        stats.referee_stats ->> 'discount_factor' as discount_factor,
-       stats.referee_stats ->> 'reward_factor'   as reward_factor,
+       reward_factor,
        vega_time
 from referral_set_stats,
      jsonb_array_elements(referees_stats) with ordinality stats(referee_stats, position)
     );
-

datanode/sqlstore/referral_sets.go

@@ -110,13 +110,14 @@ func (rs *ReferralSets) AddReferralSetStats(ctx context.Context, stats *entities
 	defer metrics.StartSQLQuery("ReferralSets", "AddReferralSetStats")()
 	_, err := rs.Connection.Exec(
 		ctx,
-		`INSERT INTO referral_set_stats(set_id, at_epoch, referral_set_running_notional_taker_volume, referees_stats, vega_time)
-			values ($1, $2, $3, $4, $5)`,
+		`INSERT INTO referral_set_stats(set_id, at_epoch, referral_set_running_notional_taker_volume, referees_stats, vega_time, reward_factor)
+			values ($1, $2, $3, $4, $5, $6)`,
 		stats.SetID,
 		stats.AtEpoch,
 		stats.ReferralSetRunningNotionalTakerVolume,
 		stats.RefereesStats,
 		stats.VegaTime,
+		stats.RewardFactor,
 	)
 
 	return err
@@ -134,9 +135,9 @@ func (rs *ReferralSets) GetReferralSetStats(ctx context.Context, setID entities.
 		`SELECT at_epoch,
        				vega_time,
        				referral_set_running_notional_taker_volume,
+       				reward_factor,
        				referee_stats->>'party_id' as party_id,
        				referee_stats->>'discount_factor' as discount_factor,
-       				referee_stats->>'reward_factor' as reward_factor,
        				referee_stats->>'epoch_notional_taker_volume' as epoch_notional_taker_volume
 			  FROM referral_set_stats, jsonb_array_elements(referees_stats) AS referee_stats
 			  WHERE set_id = %s`,

datanode/sqlstore/referral_sets_test.go

@@ -451,8 +451,9 @@ func TestReferralSets_AddReferralSetStats(t *testing.T) {
 			SetID:                                 set.ID,
 			AtEpoch:                               epoch,
 			ReferralSetRunningNotionalTakerVolume: takerVolume,
-			RefereesStats:                         getRefereeStats(t, refs, "0.01", "0.02"),
+			RefereesStats:                         getRefereeStats(t, refs, "0.01"),
 			VegaTime:                              block.VegaTime,
+			RewardFactor:                          "0.02",
 		}
 
 		err := rs.AddReferralSetStats(ctx, &stats)
@@ -471,8 +472,9 @@ func TestReferralSets_AddReferralSetStats(t *testing.T) {
 			SetID:                                 set.ID,
 			AtEpoch:                               epoch,
 			ReferralSetRunningNotionalTakerVolume: takerVolume,
-			RefereesStats:                         getRefereeStats(t, refs, "0.01", "0.02"),
+			RefereesStats:                         getRefereeStats(t, refs, "0.01"),
 			VegaTime:                              block.VegaTime,
+			RewardFactor:                          "0.02",
 		}
 
 		err := rs.AddReferralSetStats(ctx, &stats)
@@ -488,14 +490,13 @@ func TestReferralSets_AddReferralSetStats(t *testing.T) {
 	})
 }
 
-func getRefereeStats(t *testing.T, refs []entities.ReferralSetReferee, discountFactor, rewardFactor string) []*eventspb.RefereeStats {
+func getRefereeStats(t *testing.T, refs []entities.ReferralSetReferee, discountFactor string) []*eventspb.RefereeStats {
 	t.Helper()
 	stats := make([]*eventspb.RefereeStats, len(refs))
 	for i, r := range refs {
 		stats[i] = &eventspb.RefereeStats{
 			PartyId:        r.Referee.String(),
 			DiscountFactor: discountFactor,
-			RewardFactor:   rewardFactor,
 		}
 	}
 	return stats
@@ -531,11 +532,11 @@ func TestReferralSets_GetReferralSetStats(t *testing.T) {
 				return &eventspb.RefereeStats{
 					PartyId:                  party.ID.String(),
 					DiscountFactor:           fmt.Sprintf("0.1%d%d", i+1, j+1),
-					RewardFactor:             fmt.Sprintf("0.2%d%d", i+1, j+1),
 					EpochNotionalTakerVolume: strconv.Itoa((i+1)*100 + (j+1)*10),
 				}
 			}),
-			VegaTime: block.VegaTime,
+			VegaTime:     block.VegaTime,
+			RewardFactor: fmt.Sprintf("0.2%d", i+1),
 		}
 
 		require.NoError(t, rs.AddReferralSetStats(ctx, &set))
@@ -547,7 +548,7 @@ func TestReferralSets_GetReferralSetStats(t *testing.T) {
 				VegaTime:                              block.VegaTime,
 				PartyID:                               stat.PartyId,
 				DiscountFactor:                        stat.DiscountFactor,
-				RewardFactor:                          stat.RewardFactor,
+				RewardFactor:                          set.RewardFactor,
 				EpochNotionalTakerVolume:              stat.EpochNotionalTakerVolume,
 			})
 		}