support multiple rewards per address

prisma/schema.prisma

@@ -178,12 +178,13 @@ model dag_reward_transactions {
   global_snapshot_hash String           @db.VarChar
   destination_addr     String           @db.VarChar
   amount               BigInt
+  idx                  Int
   created_at           DateTime         @default(now()) @db.Timestamp(6)
   updated_at           DateTime         @default(now()) @db.Timestamp(6)
   global_snapshot      global_snapshots @relation(fields: [global_snapshot_hash], references: [hash], onDelete: Cascade, onUpdate: NoAction, map: "dag_reward_transaction_global_snapshot_fk")
   addresses            addresses        @relation(fields: [destination_addr], references: [address], onDelete: Cascade, onUpdate: NoAction, map: "dag_reward_transactions_destination_addr_fk")
 
-  @@id([global_snapshot_hash, destination_addr], map: "dag_reward_transaction_pk")
+  @@id([global_snapshot_hash, destination_addr, idx], map: "dag_reward_transaction_pk")
   @@index([global_snapshot_hash])
 }
 

src/handlers/dagHandler.ts

@@ -113,16 +113,17 @@ export const globalSnapshotRewards = async (
     const gsWhere = await globalSnapshotWhere(term);
 
     const toCursor = (row) => ({
-      global_snapshot_hash_destination_addr: {
+      global_snapshot_hash_destination_addr_idx: {
         global_snapshot_hash: row.global_snapshot_hash,
         destination_addr: row.destination_addr,
+        idx: row.idx,
       },
     });
     const fromCursor = (row) => ({
       global_snapshot_hash: row.global_snapshot_hash,
       destination_addr: row.destination_addr,
+      idx: row.idx,
     });
-
     return await paginatedQuery(
       extractPagination(event),
       toCursor,

src/response.ts

@@ -36,7 +36,27 @@
   metagraphSnapshotCount: snapshot.metagraph_snapshot_count,
 });
 
-export const rewardsResponse = (rs) => rs.map(rewardResponse);
+export const rewardsResponse = (rs) => {
+  // legacy: remove idx <0 if same address, amount exists with idx>=0
+  // Group elements by address
+  const grouped = rs.reduce((acc, item) => {
+    acc[item.destination_addr] = acc[item.destination_addr] || [];
+    acc[item.destination_addr].push(item);
+    return acc;
+  }, {} as Record<string, object>);
+
+  // Filter based on the group size and index
+
+  const filtered = Object.values(grouped).flatMap((items) => {
+    if (items.length === 1) {
+      return items;
+    } else {
+      return items.filter((item) => item.idx >= 0);
+    }
+  });
+
+  return filtered.map(rewardResponse);
+};
 
 export const rewardResponse = (reward) => ({
   destination: reward.destination_addr,

tests/handlers/dagHandler.test.ts

@@ -10,6 +10,7 @@ import {
   data_global_snapshots,
   prisma,
 } from "../../prisma/seed";
+import { rewardResponse } from "../../src/response";
 
 const data_dag_blocks = [
   {
@@ -115,6 +116,41 @@ const data_dag_transactions = [
   },
 ];
 
+const data_dag_reward_transactions = [
+  {
+    destination_addr: data_addresses[1].address,
+    amount: 1000n,
+    idx: -1,
+    global_snapshot_hash: data_global_snapshots[0].hash,
+    created_at: new Date("2025-04-02T00:00:02Z"),
+    updated_at: new Date(),
+  },
+  {
+    destination_addr: data_addresses[1].address,
+    amount: 1000n,
+    idx: 1,
+    global_snapshot_hash: data_global_snapshots[0].hash,
+    created_at: new Date("2025-04-02T00:00:03Z"),
+    updated_at: new Date(),
+  },
+  {
+    destination_addr: data_addresses[1].address,
+    amount: 3000n,
+    idx: 2,
+    global_snapshot_hash: data_global_snapshots[0].hash,
+    created_at: new Date("2025-04-02T00:00:03Z"),
+    updated_at: new Date(),
+  },
+  {
+    destination_addr: data_addresses[0].address,
+    amount: 1000n,
+    idx: -1,
+    global_snapshot_hash: data_global_snapshots[0].hash,
+    created_at: new Date("2025-04-02T00:00:02Z"),
+    updated_at: new Date(),
+  },
+];
+
 const data_dag_balance_changes = [
   {
     snapshot_hash: data_global_snapshots[0].hash,
@@ -135,6 +171,10 @@ const seedData = async () => {
     data: data_dag_transactions,
   });
 
+  await prisma.dag_reward_transactions.createMany({
+    data: data_dag_reward_transactions,
+  });
+
   await prisma.dag_balance_changes.createMany({
     data: data_dag_balance_changes,
   });
@@ -326,6 +366,42 @@ describe("DAG Handler Integration Tests", () => {
     });
   });
 
+  const normalize = (arr: any[]) =>
+    arr.map(({ destination_addr, amount }) => ({
+      destination: destination_addr,
+      amount: Number(amount), // whether it’s bigint or number
+    }));
+
+  describe("globalSnapshotRewards", () => {
+    it("should return reward transactions for a global snapshot", async () => {
+      const requestedSnapshotHash =
+        data_dag_reward_transactions[0].global_snapshot_hash;
+
+      const event = createAPIGatewayEvent(
+        { term: requestedSnapshotHash },
+        { limit: "10" }
+      );
+
+      const response: APIGatewayProxyResult =
+        await dagHandler.globalSnapshotRewards(event);
+
+      const expected = normalize([
+        data_dag_reward_transactions[1],
+        data_dag_reward_transactions[2],
+        data_dag_reward_transactions[3],
+      ]);
+
+      expect(response.statusCode).toBe(200);
+      const body = validatePaginatedResponse(response);
+
+      expect(body.data.length).toBe(expected.length);
+
+      const txs = body.data;
+
+      expect(txs).toMatchObject(expected);
+    });
+  });
+
   describe("dagTransaction", () => {
     it("should return a specific transaction by hash", async () => {
       const event = createAPIGatewayEvent({