Initialize peer stats with past events (#3272)

diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerScorer.scala b/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerScorer.scala
index aebffcb..7a754cd 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerScorer.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerScorer.scala
@@ -285,9 +285,9 @@ private class PeerScorer(nodeParams: NodeParams, wallet: OnChainBalanceChecker,
       // We'd like to increase their capacity by 10% (those peers have a large capacity, so 10% is already a non-negligible amount).
       .map(p => FundingProposal(p, p.capacity * 0.1))
 
-    // Since we're not yet reading past events from the DB, we need to wait until we have collected enough data before
-    // taking some actions such as opening or closing channels or updating relay fees.
-    // TODO: remove this once we start reading past data from the AuditDb on restart.
+    // Some actions such as opening or closing channels or updating relay fees should only run periodically, not when
+    // explicitly requested by a caller (replyTo_opt).
+    // TODO: remove hasPastData after successfully deploying the AuditDb changes.
     val hasPastData = bestPeersByVolume.exists(_.stats.drop(Bucket.bucketsPerDay).exists(_ != PeerStats.empty))
     if (hasPastData && replyTo_opt.isEmpty) {
       closeUnbalancedChannelsIfNeeded(peers)
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala b/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala
index ac03b03..91c574d 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/profit/PeerStatsTracker.scala
@@ -38,6 +38,7 @@ object PeerStatsTracker {
   // @formatter:off
   sealed trait Command
   case class GetLatestStats(replyTo: ActorRef[LatestStats]) extends Command
+  private[profit] case class LoadPastEvents(before: TimestampMilli) extends Command
   private[profit] case object RemoveOldBuckets extends Command
   private[profit] case class WrappedPaymentSent(e: PaymentSent) extends Command
   private[profit] case class WrappedPaymentRelayed(e: PaymentRelayed) extends Command
@@ -183,6 +184,17 @@ object PeerStatsTracker {
       }
     }
 
+    def addConfirmedTransaction(e: AuditDb.ConfirmedTransaction): BucketedPeerStats = {
+      val bucket = Bucket.from(e.timestamp)
+      val peerStats = this.getPeerStatsForBucket(e.remoteNodeId, bucket)
+      val peerStats1 = peerStats.copy(
+        onChainFeePaid = peerStats.onChainFeePaid + e.onChainFeePaid,
+        liquidityFeeEarned = peerStats.liquidityFeeEarned + e.liquidityPurchase_opt.map(p => if (p.isSeller) p.fees.total else 0.sat).getOrElse(0 sat),
+        liquidityFeePaid = peerStats.liquidityFeePaid + e.liquidityPurchase_opt.map(p => if (p.isBuyer) p.fees.total else 0.sat).getOrElse(0 sat),
+      )
+      this.addOrUpdate(e.remoteNodeId, bucket, peerStats1)
+    }
+
     /** Remove old buckets that exceed our retention window. This should be called frequently to avoid memory leaks. */
     def removeOldBuckets(now: TimestampMilli): BucketedPeerStats = {
       val oldestBucket = Bucket.from(now - Bucket.duration * BucketedPeerStats.bucketsCount)
@@ -195,7 +207,7 @@ object PeerStatsTracker {
     def removePeer(remoteNodeId: PublicKey): BucketedPeerStats = copy(stats = stats - remoteNodeId)
   }
 
-  object BucketedPeerStats {
+  private object BucketedPeerStats {
     // We keep 7 days of past history.
     val bucketsCount: Int = 7 * Bucket.bucketsPerDay
 
@@ -316,8 +328,10 @@ private class PeerStatsTracker(db: AuditDb, timers: TimerScheduler[PeerStatsTrac
   import PeerStatsTracker._
 
   private val log = context.log
+  private var pastEventsLoaded = false
 
   private def start(channels: Seq[PersistentChannelData]): Behavior[Command] = {
+    val startedAt = TimestampMilli.now()
     // We subscribe to channel events to update channel balances.
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter[ChannelIdAssigned](e => ChannelCreationInProgress(e.remoteNodeId, e.channelId)))
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter[ChannelAborted](e => ChannelCreationAborted(e.remoteNodeId, e.channelId)))
@@ -330,28 +344,29 @@ private class PeerStatsTracker(db: AuditDb, timers: TimerScheduler[PeerStatsTrac
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedPaymentSent))
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedPaymentRelayed))
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedPaymentReceived))
-    // TODO: read events that happened before startedAt from the DB to initialize statistics from past data.
+    // We start reading past events from the DB.
+    context.self ! LoadPastEvents(before = startedAt)
     val stats = BucketedPeerStats.empty(peerChannels.peers)
     timers.startTimerWithFixedDelay(RemoveOldBuckets, Bucket.duration)
-    listening(stats, peerChannels)
+    listening(stats, peerChannels, startedAt)
   }
 
-  private def listening(stats: BucketedPeerStats, channels: PeerChannels): Behavior[Command] = {
+  private def listening(stats: BucketedPeerStats, channels: PeerChannels, lastTransactionReadAt: TimestampMilli): Behavior[Command] = {
     Behaviors.receiveMessage {
       case WrappedPaymentSent(e) =>
-        listening(stats.addPaymentSent(e), channels)
+        listening(stats.addPaymentSent(e), channels, lastTransactionReadAt)
       case WrappedPaymentReceived(e) =>
-        listening(stats.addPaymentReceived(e), channels)
+        listening(stats.addPaymentReceived(e), channels, lastTransactionReadAt)
       case WrappedPaymentRelayed(e) =>
-        listening(stats.addPaymentRelayed(e), channels)
+        listening(stats.addPaymentRelayed(e), channels, lastTransactionReadAt)
       case e: ChannelCreationInProgress =>
-        listening(stats, channels.addPendingChannel(e))
+        listening(stats, channels.addPendingChannel(e), lastTransactionReadAt)
       case e: ChannelCreationAborted =>
-        listening(stats, channels.removeChannel(e))
+        listening(stats, channels.removeChannel(e), lastTransactionReadAt)
       case WrappedLocalChannelUpdate(e) =>
-        listening(stats.initializePeerIfNeeded(e.remoteNodeId), channels.updateChannel(e))
+        listening(stats.initializePeerIfNeeded(e.remoteNodeId), channels.updateChannel(e), lastTransactionReadAt)
       case WrappedAvailableBalanceChanged(e) =>
-        listening(stats, channels.updateChannel(e))
+        listening(stats, channels.updateChannel(e), lastTransactionReadAt)
       case WrappedLocalChannelDown(e) =>
         val channels1 = channels.removeChannel(e)
         val stats1 = if (channels1.getChannels(e.remoteNodeId).isEmpty && !channels1.hasPendingChannel(e.remoteNodeId)) {
@@ -359,22 +374,46 @@ private class PeerStatsTracker(db: AuditDb, timers: TimerScheduler[PeerStatsTrac
         } else {
           stats
         }
-        listening(stats1, channels1)
+        listening(stats1, channels1, lastTransactionReadAt)
+      case LoadPastEvents(before) =>
+        if (before <= (TimestampMilli.now() - Bucket.duration * BucketedPeerStats.bucketsCount)) {
+          log.info("finished loading past events from the DB")
+          pastEventsLoaded = true
+          listening(stats, channels, lastTransactionReadAt)
+        } else {
+          log.info("loading past events before {}", before)
+          val stats1 = db.listSent(before - Bucket.duration, before).foldLeft(stats) { case (current, e) => current.addPaymentSent(e) }
+          val stats2 = db.listReceived(before - Bucket.duration, before).foldLeft(stats1) { case (current, e) => current.addPaymentReceived(e) }
+          val stats3 = db.listRelayed(before - Bucket.duration, before).foldLeft(stats2) { case (current, e) => current.addPaymentRelayed(e) }
+          val stats4 = db.listConfirmed(before - Bucket.duration, before).foldLeft(stats3) { case (current, e) => current.addConfirmedTransaction(e) }
+          // We continue reading events from past buckets.
+          context.self ! LoadPastEvents(before - Bucket.duration)
+          listening(stats4, channels, lastTransactionReadAt)
+        }
       case RemoveOldBuckets =>
-        listening(stats.removeOldBuckets(TimestampMilli.now()), channels)
+        listening(stats.removeOldBuckets(TimestampMilli.now()), channels, lastTransactionReadAt)
+      case GetLatestStats(replyTo) if !pastEventsLoaded =>
+        // This generally shouldn't happen since the peer scorer doesn't run too often, and doesn't run immediately
+        // after restarting the node.
+        log.warn("cannot return peer statistics: we're still loading past events from the DB")
+        replyTo ! LatestStats(Nil)
+        listening(stats, channels, lastTransactionReadAt)
       case GetLatestStats(replyTo) =>
-        // TODO: do a db.listConfirmed() to update on-chain stats (we cannot rely on events only because data comes from
-        //  the TransactionPublished event, but should only be applied after TransactionConfirmed so we need permanent
-        //  storage). We'll need the listConfirmed() function added in https://github.com/ACINQ/eclair/pull/3245.
         log.debug("statistics available for {} peers", stats.peers.size)
         val now = TimestampMilli.now()
-        val latest = stats.peers
+        // We start by reading recently confirmed transactions. We cannot simply rely on events for that because data
+        // comes from the TransactionPublished event, but should only be applied after TransactionConfirmed (which may
+        // happen a long time after publication, and the node may have restarted between those two events).
+        val stats1 = db.listConfirmed(lastTransactionReadAt, now).foldLeft(stats) {
+          case (current, e) => current.addConfirmedTransaction(e)
+        }
+        val latest = stats1.peers
           // We only return statistics for peers with whom we have channels available for payments.
           .filter(nodeId => channels.hasChannels(nodeId))
-          .map(nodeId => PeerInfo(nodeId, stats.getPeerStats(nodeId, now), channels.getChannels(nodeId), channels.getUpdate(nodeId), channels.hasPendingChannel(nodeId)))
+          .map(nodeId => PeerInfo(nodeId, stats1.getPeerStats(nodeId, now), channels.getChannels(nodeId), channels.getUpdate(nodeId), channels.hasPendingChannel(nodeId)))
           .toSeq
         replyTo ! LatestStats(latest)
-        listening(stats, channels)
+        listening(stats1, channels, lastTransactionReadAt = now)
     }
   }
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala
index c4dea6f..4d0afda 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/profit/PeerStatsTrackerSpec.scala
@@ -10,16 +10,17 @@ import fr.acinq.eclair.blockchain.fee.FeeratePerKw
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.payment.PaymentEvent.{IncomingPayment, OutgoingPayment}
 import fr.acinq.eclair.payment.relay.Relayer.RelayFees
-import fr.acinq.eclair.payment.{ChannelPaymentRelayed, TrampolinePaymentRelayed}
+import fr.acinq.eclair.payment.{ChannelPaymentRelayed, PaymentReceived, PaymentSent, TrampolinePaymentRelayed}
 import fr.acinq.eclair.profit.PeerStatsTracker._
 import fr.acinq.eclair.transactions.Transactions.{ClosingTx, InputInfo}
 import fr.acinq.eclair.wire.protocol.ChannelUpdate.{ChannelFlags, MessageFlags}
-import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelUpdate}
+import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelUpdate, LiquidityAds}
 import fr.acinq.eclair.{Alias, BlockHeight, CltvExpiryDelta, Features, MilliSatoshiLong, RealShortChannelId, TestDatabases, TimestampMilli, TimestampSecond, ToMilliSatoshiConversion, randomBytes32, randomKey}
 import org.scalatest.Inside.inside
 import org.scalatest.funsuite.AnyFunSuiteLike
 import scodec.bits.{ByteVector, HexStringSyntax}
 
+import java.util.UUID
 import scala.concurrent.duration.DurationInt
 
 class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("application")) with AnyFunSuiteLike {
@@ -119,6 +120,11 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
       incoming = Seq(IncomingPayment(c2b.channelId, remoteNodeId2, 15_000_000 msat, now - 5.minutes)),
       outgoing = Seq(OutgoingPayment(c1b.channelId, remoteNodeId1, 10_000_000 msat, now))
     ))
+    // We need to wait for past events to be loaded from the DB.
+    probe.awaitAssert({
+      tracker.ref ! GetLatestStats(probe.ref)
+      assert(probe.expectMessageType[LatestStats].peers.nonEmpty)
+    })
     tracker.ref ! GetLatestStats(probe.ref)
     inside(probe.expectMessageType[LatestStats]) { s =>
       // We only have active channels with our first peer.
@@ -283,6 +289,12 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
       nextTrampolineAmount = 50_000_000 msat,
     ))
 
+    // We need to wait for past events to be loaded from the DB.
+    probe.awaitAssert({
+      tracker.ref ! GetLatestStats(probe.ref)
+      assert(probe.expectMessageType[LatestStats].peers.nonEmpty)
+    })
+
     // We keep track of aggregated statistics per bucket.
     tracker.ref ! GetLatestStats(probe.ref)
     inside(probe.expectMessageType[LatestStats]) { s =>
@@ -336,4 +348,148 @@ class PeerStatsTrackerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load(
     }
   }
 
+  test("initialize peer stats with past events") {
+    val now = TimestampMilli.now()
+    val probe = TestProbe[LatestStats]()
+    val db = TestDatabases.inMemoryDb().audit
+    val channelId1 = randomBytes32()
+    val channelId2 = randomBytes32()
+    val dummyTx = Transaction(2, Nil, Seq(TxOut(50_000 sat, Script.pay2wpkh(dummyPubKey))), 0)
+
+    // PaymentSent through remoteNodeId1, 1 day ago.
+    val sentAmount = 100_000_000 msat
+    val sentFees = 1_000 msat
+    val sentAt = now - 1.day
+    db.add(PaymentSent(
+      UUID.randomUUID(), randomBytes32(), sentAmount - sentFees, randomKey().publicKey,
+      PaymentSent.PaymentPart(UUID.randomUUID(), OutgoingPayment(channelId1, remoteNodeId1, sentAmount, sentAt), sentFees, None, sentAt - 10.seconds) :: Nil,
+      None, sentAt - 10.seconds
+    ))
+
+    // PaymentReceived through remoteNodeId2, 2 days ago.
+    val receivedAmount = 50_000_000 msat
+    val receivedAt = now - 2.days
+    db.add(PaymentReceived(randomBytes32(), IncomingPayment(channelId2, remoteNodeId2, receivedAmount, receivedAt) :: Nil))
+
+    // ChannelPaymentRelayed from remoteNodeId2 -> remoteNodeId1, 12 hours ago.
+    val relayedInAmount = 25_000_000 msat
+    val relayedOutAmount = 24_500_000 msat
+    val relayedAt = now - 12.hours
+    db.add(ChannelPaymentRelayed(
+      randomBytes32(),
+      Seq(IncomingPayment(channelId2, remoteNodeId2, relayedInAmount, relayedAt - 1.minute)),
+      Seq(OutgoingPayment(channelId1, remoteNodeId1, relayedOutAmount, relayedAt))
+    ))
+
+    // TransactionPublished + TransactionConfirmed for remoteNodeId1 as buyer, 3 days ago.
+    val buyerPurchase = LiquidityAds.PurchaseBasicInfo(isBuyer = true, 100_000 sat, LiquidityAds.Fees(50 sat, 200 sat))
+    val txPublished1 = TransactionPublished(channelId1, remoteNodeId1, dummyTx, 300 sat, 0 sat, "funding", Some(buyerPurchase), now - 3.days)
+    db.add(txPublished1)
+    db.add(TransactionConfirmed(channelId1, remoteNodeId1, txPublished1.tx, now - 3.days + 10.minutes))
+
+    // TransactionPublished + TransactionConfirmed for remoteNodeId2 as seller, 5 days ago.
+    val sellerPurchase = LiquidityAds.PurchaseBasicInfo(isBuyer = false, 200_000 sat, LiquidityAds.Fees(100 sat, 500 sat))
+    val dummyTx2 = Transaction(2, Nil, Seq(TxOut(200_000 sat, Script.pay2wpkh(dummyPubKey))), 0)
+    val txPublished2 = TransactionPublished(channelId2, remoteNodeId2, dummyTx2, 150 sat, 0 sat, "splice", Some(sellerPurchase), now - 5.days)
+    db.add(txPublished2)
+    db.add(TransactionConfirmed(channelId2, remoteNodeId2, txPublished2.tx, now - 5.days + 20.minutes))
+
+    // TransactionPublished + TransactionConfirmed for remoteNodeId1, plain (no liquidity purchase), 1 day ago.
+    val dummyTx3 = Transaction(2, Nil, Seq(TxOut(30_000 sat, Script.pay2wpkh(dummyPubKey))), 0)
+    val txPublished3 = TransactionPublished(channelId1, remoteNodeId1, dummyTx3, 200 sat, 0 sat, "mutual", None, now - 1.day)
+    db.add(txPublished3)
+    db.add(TransactionConfirmed(channelId1, remoteNodeId1, txPublished3.tx, now - 1.day + 5.minutes))
+
+    // Event outside 7-day window (10 days ago) — should NOT appear in stats.
+    val oldAt = now - 10.days
+    db.add(PaymentSent(
+      UUID.randomUUID(), randomBytes32(), 500_000_000 msat, randomKey().publicKey,
+      PaymentSent.PaymentPart(UUID.randomUUID(), OutgoingPayment(channelId1, remoteNodeId1, 500_000_000 msat, oldAt), 0 msat, None, oldAt) :: Nil,
+      None, oldAt
+    ))
+
+    // Spawn tracker with channels for both peers.
+    val c1 = channel(remoteNodeId1, toLocal = 0.3 btc, toRemote = 0.2 btc)
+    val c2 = channel(remoteNodeId2, toLocal = 0.1 btc, toRemote = 0.4 btc)
+    val tracker = testKit.spawn(PeerStatsTracker(db, Seq(c1, c2)))
+
+    probe.awaitAssert({
+      tracker.ref ! GetLatestStats(probe.ref)
+      inside(probe.expectMessageType[LatestStats]) { s =>
+        assert(s.peers.map(_.remoteNodeId).toSet == Set(remoteNodeId1, remoteNodeId2))
+
+        val peer1 = s.peers.find(_.remoteNodeId == remoteNodeId1).get
+        // remoteNodeId1: sent 100_000_000 msat out + relayed 24_500_000 msat out (the older event is ignored).
+        assert(peer1.stats.map(_.totalAmountOut).sum == sentAmount + relayedOutAmount)
+        assert(peer1.stats.map(_.totalAmountIn).sum == 0.msat)
+        // on-chain fees: 300 sat (buyer tx) + 200 sat (plain tx)
+        assert(peer1.stats.map(_.onChainFeePaid).sum == 500.sat)
+        // liquidity fee paid as buyer: 50 + 200 = 250 sat
+        assert(peer1.stats.map(_.liquidityFeePaid).sum == 250.sat)
+        assert(peer1.stats.map(_.liquidityFeeEarned).sum == 0.sat)
+
+        val peer2 = s.peers.find(_.remoteNodeId == remoteNodeId2).get
+        // remoteNodeId2: received 50_000_000 msat + relayed 25_000_000 msat in
+        assert(peer2.stats.map(_.totalAmountIn).sum == receivedAmount + relayedInAmount)
+        assert(peer2.stats.map(_.totalAmountOut).sum == 0.msat)
+        // on-chain fees: 150 sat (seller tx)
+        assert(peer2.stats.map(_.onChainFeePaid).sum == 150.sat)
+        // liquidity fee earned as seller: 100 + 500 = 600 sat
+        assert(peer2.stats.map(_.liquidityFeeEarned).sum == 600.sat)
+        assert(peer2.stats.map(_.liquidityFeePaid).sum == 0.sat)
+      }
+    })
+  }
+
+  test("read confirmed transactions when statistics are requested") {
+    val probe = TestProbe[LatestStats]()
+    val db = TestDatabases.inMemoryDb().audit
+    val channelId1 = randomBytes32()
+    val dummyTx = Transaction(2, Nil, Seq(TxOut(50_000 sat, Script.pay2wpkh(dummyPubKey))), 0)
+
+    // Spawn tracker with a channel for remoteNodeId1 and empty DB.
+    val c1 = channel(remoteNodeId1, toLocal = 0.3 btc, toRemote = 0.2 btc)
+    val tracker = testKit.spawn(PeerStatsTracker(db, Seq(c1)))
+
+    // Wait for initial loading to complete (should be almost instantaneous on an empty DB).
+    probe.awaitAssert({
+      tracker.ref ! GetLatestStats(probe.ref)
+      assert(probe.expectMessageType[LatestStats].peers.nonEmpty)
+    })
+
+    // Add a confirmed transaction after starting.
+    val now = TimestampMilli.now()
+    val txPublished1 = TransactionPublished(channelId1, remoteNodeId1, dummyTx, 300 sat, 0 sat, "funding", None, now + 5.millis)
+    db.add(txPublished1)
+    db.add(TransactionConfirmed(channelId1, remoteNodeId1, txPublished1.tx, now + 10.millis))
+
+    probe.awaitAssert({
+      tracker.ref ! GetLatestStats(probe.ref)
+      inside(probe.expectMessageType[LatestStats]) { s =>
+        val peer1 = s.peers.find(_.remoteNodeId == remoteNodeId1).get
+        assert(peer1.stats.map(_.onChainFeePaid).sum == 300.sat)
+      }
+    })
+
+    // Verify that we don't count transactions multiple times.
+    tracker.ref ! GetLatestStats(probe.ref)
+    inside(probe.expectMessageType[LatestStats]) { s =>
+      val peer1 = s.peers.find(_.remoteNodeId == remoteNodeId1).get
+      assert(peer1.stats.map(_.onChainFeePaid).sum == 300.sat)
+    }
+
+    // Add another confirmed transaction, verify the cumulative total.
+    val now2 = TimestampMilli.now()
+    val dummyTx2 = Transaction(2, Nil, Seq(TxOut(30_000 sat, Script.pay2wpkh(dummyPubKey))), 0)
+    val txPublished2 = TransactionPublished(channelId1, remoteNodeId1, dummyTx2, 150 sat, 0 sat, "splice", None, now2)
+    db.add(txPublished2)
+    db.add(TransactionConfirmed(channelId1, remoteNodeId1, txPublished2.tx, now2))
+
+    tracker.ref ! GetLatestStats(probe.ref)
+    inside(probe.expectMessageType[LatestStats]) { s =>
+      val peer1 = s.peers.find(_.remoteNodeId == remoteNodeId1).get
+      assert(peer1.stats.map(_.onChainFeePaid).sum == 450.sat)
+    }
+  }
+
 }