feat: add gateway fee tracking

schema.graphql

@@ -54,6 +54,8 @@ type Protocol @entity {
   pendingDeactivation: [Transcoder!]!
   "Total number of delegators on the network"
   delegatorsCount: BigInt!
+  "Broadcasters active within the current 90 day fee window"
+  activeBroadcasters: [String!]!
 }
 
 """
@@ -231,6 +233,22 @@ type Broadcaster @entity {
   deposit: BigDecimal!
   "Amount of funds in reserve"
   reserve: BigDecimal!
+  "Total fees paid out by this broadcaster in ETH"
+  totalVolumeETH: BigDecimal!
+  "Total fees paid out by this broadcaster in USD"
+  totalVolumeUSD: BigDecimal!
+  "Fees paid out by this broadcaster in ETH during the last 30 days"
+  thirtyDayVolumeETH: BigDecimal!
+  "Fees paid out by this broadcaster in ETH during the last 60 days"
+  sixtyDayVolumeETH: BigDecimal!
+  "Fees paid out by this broadcaster in ETH during the last 90 days"
+  ninetyDayVolumeETH: BigDecimal!
+  "The date this broadcaster first funded a deposit or reserve, beginning at 12:00am UTC"
+  firstActiveDay: Int!
+  "The date this broadcaster last paid fees, beginning at 12:00am UTC"
+  lastActiveDay: Int!
+  "Days in which this broadcaster paid out fees"
+  broadcasterDays: [BroadcasterDay!]!
 }
 
 """
@@ -342,7 +360,7 @@ type Day @entity {
 Transcoder data accumulated and condensed into day stats
 """
 type TranscoderDay @entity {
-  "Combination of the transcoder address and the timestamp rounded to current day by dividing by 86400"
+  "Concatenation of the transcoder address and the day timestamp (e.g. <ADDRESS>-<TIMESTAMP>)"
   id: ID!
   "The date beginning at 12:00am UTC"
   date: Int!
@@ -354,6 +372,22 @@ type TranscoderDay @entity {
   transcoder: Transcoder!
 }
 
+"""
+Broadcaster data accumulated and condensed into day stats
+"""
+type BroadcasterDay @entity {
+  "Concatenation of the broadcaster address and the day timestamp (e.g. <ADDRESS>-<TIMESTAMP>)"
+  id: ID!
+  "The date beginning at 12:00am UTC"
+  date: Int!
+  "Fees paid this day in ETH"
+  volumeETH: BigDecimal!
+  "Fees paid this day in USD"
+  volumeUSD: BigDecimal!
+  "Broadcaster associated with the day"
+  broadcaster: Broadcaster!
+}
+
 type TreasuryProposal @entity {
   "Governor proposal ID formatted as a decimal number"
   id: ID!

src/mappings/roundsManager.ts

@@ -28,6 +28,8 @@ import {
 } from "../types/RoundsManager/RoundsManager";
 // Import entity types generated from the GraphQL schema
 import {
+  Broadcaster,
+  BroadcasterDay,
   NewRoundEvent,
   ParameterUpdateEvent,
   Pool,
@@ -177,6 +179,59 @@ export function newRound(event: NewRound): void {
     }
   }
 
+  // Update rolling fee windows for tracked broadcasters (gateways)
+  let trackedBroadcasters = protocol.activeBroadcasters;
+  let activeBroadcasters: string[] = [];
+  if (trackedBroadcasters.length) {
+    for (let i = 0; i < trackedBroadcasters.length; i++) {
+      let broadcaster = Broadcaster.load(trackedBroadcasters[i]);
+
+      if (broadcaster) {
+        // --- Get the 30, 60, 90 day sums of volume ---
+        let broadcasterThirtyDaySum = ZERO_BD;
+        let broadcasterSixtyDaySum = ZERO_BD;
+        let broadcasterNinetyDaySum = ZERO_BD;
+
+        // capped at <90 - broadcaster days are ordered newest first
+        let broadcasterDays = broadcaster.broadcasterDays;
+        let broadcasterDaysLength =
+          broadcasterDays.length > 90 ? 90 : broadcasterDays.length;
+        for (let j = 0; j < broadcasterDaysLength; j++) {
+          let broadcasterDay = BroadcasterDay.load(broadcasterDays[j]);
+
+          if (broadcasterDay) {
+            if (broadcasterDay.date >= thirtyDayTimestamp) {
+              broadcasterThirtyDaySum = broadcasterThirtyDaySum.plus(
+                broadcasterDay.volumeETH
+              );
+            }
+            if (broadcasterDay.date >= sixtyDayTimestamp) {
+              broadcasterSixtyDaySum = broadcasterSixtyDaySum.plus(
+                broadcasterDay.volumeETH
+              );
+            }
+            if (broadcasterDay.date >= ninetyDayTimestamp) {
+              broadcasterNinetyDaySum = broadcasterNinetyDaySum.plus(
+                broadcasterDay.volumeETH
+              );
+            }
+          }
+        }
+
+        broadcaster.thirtyDayVolumeETH = broadcasterThirtyDaySum;
+        broadcaster.sixtyDayVolumeETH = broadcasterSixtyDaySum;
+        broadcaster.ninetyDayVolumeETH = broadcasterNinetyDaySum;
+        broadcaster.save();
+
+        if (broadcaster.lastActiveDay >= ninetyDayTimestamp) {
+          activeBroadcasters.push(broadcaster.id);
+        }
+      }
+    }
+  }
+
+  protocol.activeBroadcasters = activeBroadcasters;
+
   let lptPriceEth = getLptPriceEth();
 
   protocol.lptPriceEth = lptPriceEth;

src/mappings/ticketBroker.ts

@@ -2,6 +2,7 @@ import { Address, BigInt, dataSource, log } from "@graphprotocol/graph-ts";
 import {
   convertToDecimal,
   createOrLoadBroadcaster,
+  createOrLoadBroadcasterDay,
   createOrLoadDay,
   createOrLoadProtocol,
   createOrLoadRound,
@@ -32,13 +33,15 @@ import {
 
 export function winningTicketRedeemed(event: WinningTicketRedeemed): void {
   let round = createOrLoadRound(getBlockNum());
-  let day = createOrLoadDay(event.block.timestamp.toI32());
+  let timestamp = event.block.timestamp.toI32();
+  let day = createOrLoadDay(timestamp);
   let winningTicketRedeemedEvent = new WinningTicketRedeemedEvent(
     makeEventId(event.transaction.hash, event.logIndex)
   );
   let protocol = createOrLoadProtocol();
   let faceValue = convertToDecimal(event.params.faceValue);
   let ethPrice = getEthPriceUsd();
+  let faceValueUSD = faceValue.times(ethPrice);
   let poolId = makePoolId(event.params.recipient.toHex(), round.id);
   let pool = Pool.load(poolId);
 
@@ -50,7 +53,7 @@ export function winningTicketRedeemed(event: WinningTicketRedeemed): void {
   winningTicketRedeemedEvent.sender = event.params.sender.toHex();
   winningTicketRedeemedEvent.recipient = event.params.recipient.toHex();
   winningTicketRedeemedEvent.faceValue = faceValue;
-  winningTicketRedeemedEvent.faceValueUSD = faceValue.times(ethPrice);
+  winningTicketRedeemedEvent.faceValueUSD = faceValueUSD;
   winningTicketRedeemedEvent.winProb = event.params.winProb;
   winningTicketRedeemedEvent.senderNonce = event.params.senderNonce;
   winningTicketRedeemedEvent.recipientRand = event.params.recipientRand;
@@ -76,6 +79,23 @@ export function winningTicketRedeemed(event: WinningTicketRedeemed): void {
   } else {
     broadcaster.deposit = broadcaster.deposit.minus(faceValue);
   }
+
+  broadcaster.totalVolumeETH = broadcaster.totalVolumeETH.plus(faceValue);
+  broadcaster.totalVolumeUSD = broadcaster.totalVolumeUSD.plus(faceValueUSD);
+
+  let broadcasterDay = createOrLoadBroadcasterDay(
+    timestamp,
+    event.params.sender.toHex()
+  );
+  broadcaster.lastActiveDay = broadcasterDay.date;
+  broadcasterDay.volumeETH = broadcasterDay.volumeETH.plus(faceValue);
+  broadcasterDay.volumeUSD = broadcasterDay.volumeUSD.plus(faceValueUSD);
+  broadcasterDay.save();
+  let broadcasterDays = broadcaster.broadcasterDays;
+  if (!broadcasterDays.includes(broadcasterDay.id)) {
+    broadcasterDays.unshift(broadcasterDay.id);
+    broadcaster.broadcasterDays = broadcasterDays;
+  }
   broadcaster.save();
 
   // Update transcoder's fee volume
@@ -84,15 +104,11 @@ export function winningTicketRedeemed(event: WinningTicketRedeemed): void {
     event.block.timestamp.toI32()
   );
   transcoder.totalVolumeETH = transcoder.totalVolumeETH.plus(faceValue);
-  transcoder.totalVolumeUSD = transcoder.totalVolumeUSD.plus(
-    faceValue.times(ethPrice)
-  );
+  transcoder.totalVolumeUSD = transcoder.totalVolumeUSD.plus(faceValueUSD);
 
   // Update total protocol fee volume
   protocol.totalVolumeETH = protocol.totalVolumeETH.plus(faceValue);
-  protocol.totalVolumeUSD = protocol.totalVolumeUSD.plus(
-    faceValue.times(ethPrice)
-  );
+  protocol.totalVolumeUSD = protocol.totalVolumeUSD.plus(faceValueUSD);
 
   protocol.winningTicketCount = protocol.winningTicketCount + 1;
   protocol.save();
@@ -107,17 +123,15 @@ export function winningTicketRedeemed(event: WinningTicketRedeemed): void {
   day.totalActiveStake = protocol.totalActiveStake;
   day.participationRate = protocol.participationRate;
   day.volumeETH = day.volumeETH.plus(faceValue);
-  day.volumeUSD = day.volumeUSD.plus(faceValue.times(ethPrice));
+  day.volumeUSD = day.volumeUSD.plus(faceValueUSD);
   day.save();
 
   let transcoderDay = createOrLoadTranscoderDay(
-    event.block.timestamp.toI32(),
+    timestamp,
     event.params.recipient.toHex()
   );
   transcoderDay.volumeETH = transcoderDay.volumeETH.plus(faceValue);
-  transcoderDay.volumeUSD = transcoderDay.volumeUSD.plus(
-    faceValue.times(ethPrice)
-  );
+  transcoderDay.volumeUSD = transcoderDay.volumeUSD.plus(faceValueUSD);
   transcoderDay.save();
 
   // Manually manage the array of transcoder days (add newest to the beginning of the list)
@@ -130,13 +144,19 @@ export function winningTicketRedeemed(event: WinningTicketRedeemed): void {
 
   // Update fee volume for this round
   round.volumeETH = round.volumeETH.plus(faceValue);
-  round.volumeUSD = round.volumeUSD.plus(faceValue.times(ethPrice));
+  round.volumeUSD = round.volumeUSD.plus(faceValueUSD);
   round.save();
 }
 
 export function depositFunded(event: DepositFunded): void {
   let round = createOrLoadRound(getBlockNum());
   let broadcaster = createOrLoadBroadcaster(event.params.sender.toHex());
+  const timestamp = event.block.timestamp.toI32();
+
+  // One-time initialization: set to start of day for this timestamp.
+  if (broadcaster.firstActiveDay == 0) {
+    broadcaster.firstActiveDay = (timestamp / 86400) * 86400;
+  }
 
   broadcaster.deposit = broadcaster.deposit.plus(
     convertToDecimal(event.params.amount)
@@ -149,7 +169,7 @@ export function depositFunded(event: DepositFunded): void {
     makeEventId(event.transaction.hash, event.logIndex)
   );
   depositFundedEvent.transaction = event.transaction.hash.toHex();
-  depositFundedEvent.timestamp = event.block.timestamp.toI32();
+  depositFundedEvent.timestamp = timestamp;
   depositFundedEvent.round = round.id;
   depositFundedEvent.sender = event.params.sender.toHex();
   depositFundedEvent.amount = convertToDecimal(event.params.amount);
@@ -159,6 +179,12 @@ export function depositFunded(event: DepositFunded): void {
 export function reserveFunded(event: ReserveFunded): void {
   let round = createOrLoadRound(getBlockNum());
   let broadcaster = createOrLoadBroadcaster(event.params.reserveHolder.toHex());
+  const timestamp = event.block.timestamp.toI32();
+
+  // One-time initialization: set to start of day for this timestamp.
+  if (broadcaster.firstActiveDay == 0) {
+    broadcaster.firstActiveDay = (timestamp / 86400) * 86400;
+  }
 
   broadcaster.reserve = broadcaster.reserve.plus(
     convertToDecimal(event.params.amount)
@@ -171,7 +197,7 @@ export function reserveFunded(event: ReserveFunded): void {
     makeEventId(event.transaction.hash, event.logIndex)
   );
   reserveFundedEvent.transaction = event.transaction.hash.toHex();
-  reserveFundedEvent.timestamp = event.block.timestamp.toI32();
+  reserveFundedEvent.timestamp = timestamp;
   reserveFundedEvent.round = round.id;
   reserveFundedEvent.reserveHolder = event.params.reserveHolder.toHex();
   reserveFundedEvent.amount = convertToDecimal(event.params.amount);

utils/helpers.ts

@@ -9,6 +9,7 @@ import {
 import { RoundsManager } from "../src/types/RoundsManager/RoundsManager";
 import {
   Broadcaster,
+  BroadcasterDay,
   Day,
   Delegator,
   LivepeerAccount,
@@ -158,6 +159,7 @@ export function createOrLoadProtocol(): Protocol {
     protocol.winningTicketCount = 0;
     protocol.roundCount = 0;
     protocol.lptPriceEth = ZERO_BD;
+    protocol.activeBroadcasters = [];
 
     const network = dataSource.network();
     // 3520 is the count of total delegators from the mainnet subgraph (in the final round)
@@ -169,6 +171,7 @@ export function createOrLoadProtocol(): Protocol {
     protocol.pendingDeactivation = [];
     protocol.save();
   }
+
   return protocol;
 }
 
@@ -179,8 +182,24 @@ export function createOrLoadBroadcaster(id: string): Broadcaster {
     broadcaster = new Broadcaster(id);
     broadcaster.deposit = ZERO_BD;
     broadcaster.reserve = ZERO_BD;
+    broadcaster.totalVolumeETH = ZERO_BD;
+    broadcaster.totalVolumeUSD = ZERO_BD;
+    broadcaster.thirtyDayVolumeETH = ZERO_BD;
+    broadcaster.sixtyDayVolumeETH = ZERO_BD;
+    broadcaster.ninetyDayVolumeETH = ZERO_BD;
+    broadcaster.firstActiveDay = 0;
+    broadcaster.lastActiveDay = 0;
+    broadcaster.broadcasterDays = [];
 
     broadcaster.save();
+  } 
+
+  let protocol = createOrLoadProtocol();
+  let activeBroadcasters = protocol.activeBroadcasters;
+  if (!activeBroadcasters.includes(id)) {
+    activeBroadcasters.push(id);
+    protocol.activeBroadcasters = activeBroadcasters;
+    protocol.save();
   }
 
   return broadcaster;
@@ -316,6 +335,29 @@ export function createOrLoadTranscoderDay(
   return transcoderDay;
 }
 
+export function createOrLoadBroadcasterDay(
+  timestamp: i32,
+  broadcasterAddress: string
+): BroadcasterDay {
+  let dayID = timestamp / 86400;
+  let dayStartTimestamp = dayID * 86400;
+  let broadcasterDayID = broadcasterAddress
+    .concat("-")
+    .concat(BigInt.fromI32(dayID).toString());
+  let broadcasterDay = BroadcasterDay.load(broadcasterDayID);
+
+  if (broadcasterDay == null) {
+    broadcasterDay = new BroadcasterDay(broadcasterDayID);
+    broadcasterDay.date = dayStartTimestamp;
+    broadcasterDay.broadcaster = broadcasterAddress;
+    broadcasterDay.volumeUSD = ZERO_BD;
+    broadcasterDay.volumeETH = ZERO_BD;
+
+    broadcasterDay.save();
+  }
+  return broadcasterDay;
+}
+
 export function createOrLoadRound(blockNumber: BigInt): Round {
   let protocol = createOrLoadProtocol();
   let roundsSinceLastUpdate = blockNumber