Fix channelstats (for real?) (#1470)

The channelstats API only returns results for the *outgoing* channels
used when relaying. We must also include the *incoming* channels, otherwise
it looks like they're inactive which doesn't reflect their real usage.

Fixes #1465

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

@@ -110,7 +110,7 @@ trait Eclair {
 
   def networkFees(from_opt: Option[Long], to_opt: Option[Long])(implicit timeout: Timeout): Future[Seq[NetworkFee]]
 
-  def channelStats()(implicit timeout: Timeout): Future[Seq[Stats]]
+  def channelStats(from_opt: Option[Long], to_opt: Option[Long])(implicit timeout: Timeout): Future[Seq[Stats]]
 
   def networkStats()(implicit timeout: Timeout): Future[Option[NetworkStats]]
 
@@ -321,7 +321,10 @@ class EclairImpl(appKit: Kit) extends Eclair {
     Future(appKit.nodeParams.db.audit.listNetworkFees(filter.from, filter.to))
   }
 
-  override def channelStats()(implicit timeout: Timeout): Future[Seq[Stats]] = Future(appKit.nodeParams.db.audit.stats)
+  override def channelStats(from_opt: Option[Long], to_opt: Option[Long])(implicit timeout: Timeout): Future[Seq[Stats]] = {
+    val filter = getDefaultTimestampFilters(from_opt, to_opt)
+    Future(appKit.nodeParams.db.audit.stats(filter.from, filter.to))
+  }
 
   override def networkStats()(implicit timeout: Timeout): Future[Option[NetworkStats]] = (appKit.router ? GetNetworkStats).mapTo[GetNetworkStatsResponse].map(_.stats)
 

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

@@ -45,7 +45,7 @@ trait AuditDb extends Closeable {
 
   def listNetworkFees(from: Long, to: Long): Seq[NetworkFee]
 
-  def stats: Seq[Stats]
+  def stats(from: Long, to: Long): Seq[Stats]
 
 }
 
@@ -53,4 +53,4 @@ case class ChannelLifecycleEvent(channelId: ByteVector32, remoteNodeId: PublicKe
 
 case class NetworkFee(remoteNodeId: PublicKey, channelId: ByteVector32, txId: ByteVector32, fee: Satoshi, txType: String, timestamp: Long)
 
-case class Stats(channelId: ByteVector32, avgPaymentAmount: Satoshi, paymentCount: Int, relayFee: Satoshi, networkFee: Satoshi)
+case class Stats(channelId: ByteVector32, direction: String, avgPaymentAmount: Satoshi, paymentCount: Int, relayFee: Satoshi, networkFee: Satoshi)

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

@@ -298,35 +298,46 @@ class SqliteAuditDb(sqlite: Connection) extends AuditDb with Logging {
       q
     }
 
-  override def stats: Seq[Stats] = {
+  override def stats(from: Long, to: Long): Seq[Stats] = {
-    val networkFees = listNetworkFees(0, System.currentTimeMillis + 1).foldLeft(Map.empty[ByteVector32, Satoshi]) { case (feeByChannelId, f) =>
+    val networkFees = listNetworkFees(from, to).foldLeft(Map.empty[ByteVector32, Satoshi]) { case (feeByChannelId, f) =>
       feeByChannelId + (f.channelId -> (feeByChannelId.getOrElse(f.channelId, 0 sat) + f.fee))
     }
-    val relayed = listRelayed(0, System.currentTimeMillis + 1).foldLeft(Map.empty[ByteVector32, Seq[PaymentRelayed]]) { case (relayedByChannelId, e) =>
+    case class Relayed(amount: MilliSatoshi, fee: MilliSatoshi, direction: String)
-      val relayedTo = e match {
+    val relayed = listRelayed(from, to).foldLeft(Map.empty[ByteVector32, Seq[Relayed]]) { case (previous, e) =>
-        case c: ChannelPaymentRelayed => Set(c.toChannelId)
+      // NB: we must avoid counting the fee twice: we associate it to the outgoing channels rather than the incoming ones.
-        case t: TrampolinePaymentRelayed => t.outgoing.map(_.channelId).toSet
+      val current = e match {
+        case c: ChannelPaymentRelayed => Map(
+          c.fromChannelId -> (Relayed(c.amountIn, 0 msat, "IN") +: previous.getOrElse(c.fromChannelId, Nil)),
+          c.toChannelId -> (Relayed(c.amountOut, c.amountIn - c.amountOut, "OUT") +: previous.getOrElse(c.toChannelId, Nil)),
+        )
+        case t: TrampolinePaymentRelayed =>
+          // We ensure a trampoline payment is counted only once per channel and per direction (if multiple HTLCs were
+          // sent from/to the same channel, we group them).
+          val in = t.incoming.groupBy(_.channelId).map { case (channelId, parts) => (channelId, Relayed(parts.map(_.amount).sum, 0 msat, "IN")) }.toSeq
+          val out = t.outgoing.groupBy(_.channelId).map { case (channelId, parts) =>
+            val fee = (t.amountIn - t.amountOut) * parts.length / t.outgoing.length // we split the fee among outgoing channels
+            (channelId, Relayed(parts.map(_.amount).sum, fee, "OUT"))
+          }.toSeq
+          (in ++ out).groupBy(_._1).map { case (channelId, payments) => (channelId, payments.map(_._2) ++ previous.getOrElse(channelId, Nil)) }
       }
-      val updated = relayedTo.map(channelId => (channelId, relayedByChannelId.getOrElse(channelId, Nil) :+ e)).toMap
+      previous ++ current
-      relayedByChannelId ++ updated
     }
     // Channels opened by our peers won't have any entry in the network_fees table, but we still want to compute stats for them.
     val allChannels = networkFees.keySet ++ relayed.keySet
-    allChannels.map(channelId => {
+    allChannels.toSeq.flatMap(channelId => {
       val networkFee = networkFees.getOrElse(channelId, 0 sat)
-      val r = relayed.getOrElse(channelId, Nil)
+      val (in, out) = relayed.getOrElse(channelId, Nil).partition(_.direction == "IN")
-      val paymentCount = r.length
+      ((in, "IN") :: (out, "OUT") :: Nil).map { case (r, direction) =>
-      if (paymentCount == 0) {
+        val paymentCount = r.length
-        Stats(channelId, 0 sat, 0, 0 sat, networkFee)
+        if (paymentCount == 0) {
-      } else {
+          Stats(channelId, direction, 0 sat, 0, 0 sat, networkFee)
-        val avgPaymentAmount = r.map(_.amountOut).sum / paymentCount
+        } else {
-        val relayFee = r.map {
+          val avgPaymentAmount = r.map(_.amount).sum / paymentCount
-          case c: ChannelPaymentRelayed => c.amountIn - c.amountOut
+          val relayFee = r.map(_.fee).sum
-          case t: TrampolinePaymentRelayed => (t.amountIn - t.amountOut) * t.outgoing.count(_.channelId == channelId) / t.outgoing.length
+          Stats(channelId, direction, avgPaymentAmount.truncateToSatoshi, paymentCount, relayFee.truncateToSatoshi, networkFee)
-        }.sum
+        }
-        Stats(channelId, avgPaymentAmount.truncateToSatoshi, paymentCount, relayFee.truncateToSatoshi, networkFee)
       }
-    }).toSeq
+    })
   }
 
   // used by mobile apps

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

@@ -90,7 +90,6 @@ class SqliteAuditDbSpec extends AnyFunSuite {
     val sqlite = TestConstants.sqliteInMemory()
     val db = new SqliteAuditDb(sqlite)
 
-    val n1 = randomKey.publicKey
     val n2 = randomKey.publicKey
     val n3 = randomKey.publicKey
     val n4 = randomKey.publicKey
@@ -99,24 +98,36 @@ class SqliteAuditDbSpec extends AnyFunSuite {
     val c2 = randomBytes32
     val c3 = randomBytes32
     val c4 = randomBytes32
+    val c5 = randomBytes32
+    val c6 = randomBytes32
 
-    db.add(ChannelPaymentRelayed(46000 msat, 44000 msat, randomBytes32, randomBytes32, c1))
+    db.add(ChannelPaymentRelayed(46000 msat, 44000 msat, randomBytes32, c6, c1))
-    db.add(ChannelPaymentRelayed(41000 msat, 40000 msat, randomBytes32, randomBytes32, c1))
+    db.add(ChannelPaymentRelayed(41000 msat, 40000 msat, randomBytes32, c6, c1))
-    db.add(ChannelPaymentRelayed(43000 msat, 42000 msat, randomBytes32, randomBytes32, c1))
+    db.add(ChannelPaymentRelayed(43000 msat, 42000 msat, randomBytes32, c5, c1))
-    db.add(ChannelPaymentRelayed(42000 msat, 40000 msat, randomBytes32, randomBytes32, c2))
+    db.add(ChannelPaymentRelayed(42000 msat, 40000 msat, randomBytes32, c5, c2))
-    db.add(TrampolinePaymentRelayed(randomBytes32, Seq(PaymentRelayed.Part(25000 msat, randomBytes32)), Seq(PaymentRelayed.Part(20000 msat, c4))))
+    db.add(ChannelPaymentRelayed(45000 msat, 40000 msat, randomBytes32, c5, c6))
-    db.add(TrampolinePaymentRelayed(randomBytes32, Seq(PaymentRelayed.Part(46000 msat, randomBytes32)), Seq(PaymentRelayed.Part(16000 msat, c2), PaymentRelayed.Part(10000 msat, c4), PaymentRelayed.Part(14000 msat, c4))))
+    db.add(TrampolinePaymentRelayed(randomBytes32, Seq(PaymentRelayed.Part(25000 msat, c6)), Seq(PaymentRelayed.Part(20000 msat, c4))))
+    db.add(TrampolinePaymentRelayed(randomBytes32, Seq(PaymentRelayed.Part(46000 msat, c6)), Seq(PaymentRelayed.Part(16000 msat, c2), PaymentRelayed.Part(10000 msat, c4), PaymentRelayed.Part(14000 msat, c4))))
 
     db.add(NetworkFeePaid(null, n2, c2, Transaction(0, Seq.empty, Seq.empty, 0), 200 sat, "funding"))
     db.add(NetworkFeePaid(null, n2, c2, Transaction(0, Seq.empty, Seq.empty, 0), 300 sat, "mutual"))
     db.add(NetworkFeePaid(null, n3, c3, Transaction(0, Seq.empty, Seq.empty, 0), 400 sat, "funding"))
     db.add(NetworkFeePaid(null, n4, c4, Transaction(0, Seq.empty, Seq.empty, 0), 500 sat, "funding"))
 
-    assert(db.stats.toSet === Set(
+    // NB: we only count a relay fee for the outgoing channel, no the incoming one.
-      Stats(channelId = c1, avgPaymentAmount = 42 sat, paymentCount = 3, relayFee = 4 sat, networkFee = 0 sat),
+    assert(db.stats(0, System.currentTimeMillis + 1).toSet === Set(
-      Stats(channelId = c2, avgPaymentAmount = 40 sat, paymentCount = 2, relayFee = 4 sat, networkFee = 500 sat),
+      Stats(channelId = c1, direction = "IN", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 0 sat),
-      Stats(channelId = c3, avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 400 sat),
+      Stats(channelId = c1, direction = "OUT", avgPaymentAmount = 42 sat, paymentCount = 3, relayFee = 4 sat, networkFee = 0 sat),
-      Stats(channelId = c4, avgPaymentAmount = 30 sat, paymentCount = 2, relayFee = 9 sat, networkFee = 500 sat)
+      Stats(channelId = c2, direction = "IN", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 500 sat),
+      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),
+      Stats(channelId = c3, direction = "OUT", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 400 sat),
+      Stats(channelId = c4, direction = "IN", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 500 sat),
+      Stats(channelId = c4, direction = "OUT", avgPaymentAmount = 22 sat, paymentCount = 2, relayFee = 9 sat, networkFee = 500 sat),
+      Stats(channelId = c5, direction = "IN", avgPaymentAmount = 43 sat, paymentCount = 3, relayFee = 0 sat, networkFee = 0 sat),
+      Stats(channelId = c5, direction = "OUT", avgPaymentAmount = 0 sat, paymentCount = 0, relayFee = 0 sat, networkFee = 0 sat),
+      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),
     ))
   }
 
@@ -148,7 +159,7 @@ class SqliteAuditDbSpec extends AnyFunSuite {
     })
     // Test starts here.
     val start = System.currentTimeMillis
-    assert(db.stats.nonEmpty)
+    assert(db.stats(0, start + 1).nonEmpty)
     val end = System.currentTimeMillis
     fail(s"took ${end - start}ms")
   }

eclair-node/src/main/scala/fr/acinq/eclair/api/Service.scala

@@ -285,7 +285,9 @@ trait Service extends ExtraDirectives with Logging {
                           }
                         } ~
                         path("channelstats") {
-                          complete(eclairApi.channelStats())
+                          formFields(fromFormParam.?, toFormParam.?) { (from_opt, to_opt) =>
+                            complete(eclairApi.channelStats(from_opt, to_opt))
+                          }
                         } ~
                         path("usablebalances") {
                           complete(eclairApi.usableBalances())