Pagination for channelstats RPC

eclair-core/src/main/scala/fr/acinq/eclair/Eclair.scala

@@ -144,7 +144,7 @@ trait Eclair {
 
   def networkFees(from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[Seq[NetworkFee]]
 
-  def channelStats(from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[Seq[Stats]]
+  def channelStats(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[Stats]]
 
   def getInvoice(paymentHash: ByteVector32)(implicit timeout: Timeout): Future[Option[Invoice]]
 
@@ -525,8 +525,8 @@ class EclairImpl(appKit: Kit) extends Eclair with Logging {
     Future(appKit.nodeParams.db.audit.listNetworkFees(from.toTimestampMilli, to.toTimestampMilli))
   }
 
-  override def channelStats(from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[Seq[Stats]] = {
-    Future(appKit.nodeParams.db.audit.stats(from.toTimestampMilli, to.toTimestampMilli))
+  override def channelStats(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[Stats]] = {
+    Future(appKit.nodeParams.db.audit.stats(from.toTimestampMilli, to.toTimestampMilli, paginated_opt))
   }
 
   override def allInvoices(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[Invoice]] = Future {

eclair-core/src/main/scala/fr/acinq/eclair/db/AuditDb.scala

@@ -54,7 +54,7 @@ trait AuditDb {
 
   def listNetworkFees(from: TimestampMilli, to: TimestampMilli): Seq[NetworkFee]
 
-  def stats(from: TimestampMilli, to: TimestampMilli): Seq[Stats]
+  def stats(from: TimestampMilli, to: TimestampMilli, paginated_opt: Option[Paginated] = None): Seq[Stats]
 
 }
 

eclair-core/src/main/scala/fr/acinq/eclair/db/DualDatabases.scala

@@ -201,9 +201,9 @@ case class DualAuditDb(primary: AuditDb, secondary: AuditDb) extends AuditDb {
     primary.listNetworkFees(from, to)
   }
 
-  override def stats(from: TimestampMilli, to: TimestampMilli): Seq[AuditDb.Stats] = {
-    runAsync(secondary.stats(from, to))
-    primary.stats(from, to)
+  override def stats(from: TimestampMilli, to: TimestampMilli, paginated_opt: Option[Paginated]): Seq[AuditDb.Stats] = {
+    runAsync(secondary.stats(from, to, paginated_opt))
+    primary.stats(from, to, paginated_opt)
   }
 }
 

eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgAuditDb.scala

@@ -479,7 +479,7 @@ class PgAuditDb(implicit ds: DataSource) extends AuditDb with Logging {
       }
     }
 
-  override def stats(from: TimestampMilli, to: TimestampMilli): Seq[Stats] = {
+  override def stats(from: TimestampMilli, to: TimestampMilli, paginated_opt: Option[Paginated]): Seq[Stats] = {
     val networkFees = listNetworkFees(from, to).foldLeft(Map.empty[ByteVector32, Satoshi]) { (feeByChannelId, f) =>
       feeByChannelId + (f.channelId -> (feeByChannelId.getOrElse(f.channelId, 0 sat) + f.fee))
     }
@@ -505,7 +505,7 @@ class PgAuditDb(implicit ds: DataSource) extends AuditDb with Logging {
     }
     // Channels opened by our peers won't have any network fees paid by us, but we still want to compute stats for them.
     val allChannels = networkFees.keySet ++ relayed.keySet
-    allChannels.toSeq.flatMap(channelId => {
+    val result = allChannels.toSeq.flatMap(channelId => {
       val networkFee = networkFees.getOrElse(channelId, 0 sat)
       val (in, out) = relayed.getOrElse(channelId, Nil).partition(_.direction == "IN")
       ((in, "IN") :: (out, "OUT") :: Nil).map { case (r, direction) =>
@@ -518,6 +518,10 @@ class PgAuditDb(implicit ds: DataSource) extends AuditDb with Logging {
           Stats(channelId, direction, avgPaymentAmount.truncateToSatoshi, paymentCount, relayFee.truncateToSatoshi, networkFee)
         }
       }
-    })
+    }).sortBy(s => s.channelId.toHex + s.direction)
+    paginated_opt match {
+      case Some(paginated) => result.slice(paginated.skip, paginated.skip + paginated.count)
+      case None => result
+    }
   }
 }

eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteAuditDb.scala

@@ -448,7 +448,7 @@ class SqliteAuditDb(val sqlite: Connection) extends AuditDb with Logging {
         }.toSeq
     }
 
-  override def stats(from: TimestampMilli, to: TimestampMilli): Seq[Stats] = {
+  override def stats(from: TimestampMilli, to: TimestampMilli, paginated_opt: Option[Paginated]): Seq[Stats] = {
     val networkFees = listNetworkFees(from, to).foldLeft(Map.empty[ByteVector32, Satoshi]) { (feeByChannelId, f) =>
       feeByChannelId + (f.channelId -> (feeByChannelId.getOrElse(f.channelId, 0 sat) + f.fee))
     }
@@ -474,7 +474,7 @@ class SqliteAuditDb(val sqlite: Connection) extends AuditDb with Logging {
     }
     // Channels opened by our peers won't have any network fees paid by us, but we still want to compute stats for them.
     val allChannels = networkFees.keySet ++ relayed.keySet
-    allChannels.toSeq.flatMap(channelId => {
+    val result = allChannels.toSeq.flatMap(channelId => {
       val networkFee = networkFees.getOrElse(channelId, 0 sat)
       val (in, out) = relayed.getOrElse(channelId, Nil).partition(_.direction == "IN")
       ((in, "IN") :: (out, "OUT") :: Nil).map { case (r, direction) =>
@@ -487,6 +487,11 @@ class SqliteAuditDb(val sqlite: Connection) extends AuditDb with Logging {
           Stats(channelId, direction, avgPaymentAmount.truncateToSatoshi, paymentCount, relayFee.truncateToSatoshi, networkFee)
         }
       }
-    })
+    }).sortBy(s => s.channelId.toHex + s.direction)
+    paginated_opt match {
+      case Some(paginated) => result.slice(paginated.skip, paginated.skip + paginated.count)
+      case None => result
+    }
+
   }
 }

eclair-core/src/test/scala/fr/acinq/eclair/db/AuditDbSpec.scala

@@ -140,12 +140,12 @@ class AuditDbSpec extends AnyFunSuite {
       val n3 = randomKey().publicKey
       val n4 = randomKey().publicKey
 
-      val c1 = randomBytes32()
-      val c2 = randomBytes32()
-      val c3 = randomBytes32()
-      val c4 = randomBytes32()
-      val c5 = randomBytes32()
-      val c6 = randomBytes32()
+      val c1 = ByteVector32.One
+      val c2 = c1.copy(bytes = 0x02b +: c1.tail)
+      val c3 = c1.copy(bytes = 0x03b +: c1.tail)
+      val c4 = c1.copy(bytes = 0x04b +: c1.tail)
+      val c5 = c1.copy(bytes = 0x05b +: c1.tail)
+      val c6 = c1.copy(bytes = 0x06b +: c1.tail)
 
       db.add(ChannelPaymentRelayed(46000 msat, 44000 msat, randomBytes32(), c6, c1, 1000 unixms, 1001 unixms))
       db.add(ChannelPaymentRelayed(41000 msat, 40000 msat, randomBytes32(), c6, c1, 1002 unixms, 1003 unixms))
@@ -174,7 +174,7 @@ class AuditDbSpec extends AnyFunSuite {
       assert(db.listPublished(c4).map(_.desc) == Seq("funding", "funding"))
 
       // NB: we only count a relay fee for the outgoing channel, no the incoming one.
-      assert(db.stats(0 unixms, TimestampMilli.now() + 1.milli).toSet == Set(
+      assert(db.stats(0 unixms, TimestampMilli.now() + 1.milli) == Seq(
         Stats(channelId = c1, direction = "IN", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 0 sat),
         Stats(channelId = c1, direction = "OUT", avgPaymentAmount = 42 sat, paymentCount = 3, relayFee = 4 sat, networkFee = 0 sat),
         Stats(channelId = c2, direction = "IN", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 500 sat),
@@ -188,6 +188,10 @@ class AuditDbSpec extends AnyFunSuite {
         Stats(channelId = c6, direction = "IN", avgPaymentAmount = 39 sat, paymentCount = 4, relayFee = 0 sat, networkFee = 0 sat),
         Stats(channelId = c6, direction = "OUT", avgPaymentAmount = 40 sat, paymentCount = 1, relayFee = 5 sat, networkFee = 0 sat),
       ))
+      assert(db.stats(0 unixms, TimestampMilli.now() + 1.milli, Some(Paginated(2, 3))) == Seq(
+        Stats(channelId = c2, direction = "OUT", avgPaymentAmount = 28 sat, paymentCount = 2, relayFee = 4 sat, networkFee = 500 sat),
+        Stats(channelId = c3, direction = "IN", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 400 sat),
+      ))
     }
   }
 

eclair-node/src/main/scala/fr/acinq/eclair/api/handlers/Channel.scala

@@ -154,8 +154,10 @@ trait Channel {
   }
 
   val channelStats: Route = postRequest("channelstats") { implicit t =>
-    formFields(fromFormParam(), toFormParam()) { (from, to) =>
-      complete(eclairApi.channelStats(from, to))
+    withPaginated { paginated_opt =>
+      formFields(fromFormParam(), toFormParam()) { (from, to) =>
+        complete(eclairApi.channelStats(from, to, paginated_opt.orElse(Some(Paginated(count = 10, skip = 0)))))
+      }
     }
   }