Add ChannelOpened event

docs/release-notes/eclair-vnext.md

@@ -12,6 +12,8 @@
 - `sendtoroute` removes the `--trampolineNodes` argument and implicitly uses a single trampoline hop (#2480)
 - `payinvoice` always returns the payment result when used with `--blocking`, even when using MPP (#2525)
 - `node` returns high-level information about a remote node (#2568)
+- `channel-created` is a new websocket event that is published when a channel's funding transaction has been broadcast (#2567)
+- `channel-opened` websocket event was updated to contain the final `channel_id` and be published when a channel is ready to process payments (#2567)
 
 ### Miscellaneous improvements and bug fixes
 

eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelEvents.scala

@@ -49,6 +49,9 @@ case class ChannelIdAssigned(channel: ActorRef, remoteNodeId: PublicKey, tempora
  */
 case class ShortChannelIdAssigned(channel: ActorRef, channelId: ByteVector32, shortIds: ShortIds, remoteNodeId: PublicKey) extends ChannelEvent
 
+/** This event will be sent once a channel has been successfully opened and is ready to process payments. */
+case class ChannelOpened(channel: ActorRef, remoteNodeId: PublicKey, channelId: ByteVector32) extends ChannelEvent
+
 case class LocalChannelUpdate(channel: ActorRef, channelId: ByteVector32, shortIds: ShortIds, remoteNodeId: PublicKey, channelAnnouncement_opt: Option[ChannelAnnouncement], channelUpdate: ChannelUpdate, commitments: AbstractCommitments) extends ChannelEvent {
   /**
    * We always include the local alias because we must always be able to route based on it.

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/CommonFundingHandlers.scala

@@ -60,6 +60,8 @@ trait CommonFundingHandlers extends CommonHandlers {
     val shortIds1 = shortIds.copy(remoteAlias_opt = channelReady.alias_opt)
     shortIds1.remoteAlias_opt.foreach(_ => context.system.eventStream.publish(ShortChannelIdAssigned(self, commitments.channelId, shortIds = shortIds1, remoteNodeId = remoteNodeId)))
     log.info("shortIds: real={} localAlias={} remoteAlias={}", shortIds1.real.toOption.getOrElse("none"), shortIds1.localAlias, shortIds1.remoteAlias_opt.getOrElse("none"))
+    // we notify that the channel is now ready to route payments
+    context.system.eventStream.publish(ChannelOpened(self, remoteNodeId, commitments.channelId))
     // we create a channel_update early so that we can use it to send payments through this channel, but it won't be propagated to other nodes since the channel is not yet announced
     val scidForChannelUpdate = Helpers.scidForChannelUpdate(channelAnnouncement_opt = None, shortIds1.localAlias)
     log.info("using shortChannelId={} for initial channel_update", scidForChannelUpdate)

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/DualFundingHandlers.scala

@@ -22,9 +22,9 @@ import fr.acinq.eclair.NotificationsLogger.NotifyNodeOperator
 import fr.acinq.eclair.blockchain.CurrentBlockHeight
 import fr.acinq.eclair.blockchain.bitcoind.ZmqWatcher.WatchFundingConfirmedTriggered
 import fr.acinq.eclair.channel.Helpers.Closing
-import fr.acinq.eclair.channel.fund.InteractiveTxBuilder._
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.channel.fsm.Channel.BITCOIN_FUNDING_DOUBLE_SPENT
+import fr.acinq.eclair.channel.fund.InteractiveTxBuilder._
 import fr.acinq.eclair.wire.protocol.{ChannelReady, Error}
 
 import scala.concurrent.Future
@@ -59,7 +59,7 @@ trait DualFundingHandlers extends CommonFundingHandlers {
     }
   }
 
-  def pruneCommitments(d: DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED, fundingTx: Transaction): Option[DualFundingTx] = {
+  private def pruneCommitments(d: DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED, fundingTx: Transaction): Option[DualFundingTx] = {
     val allFundingTxs: Seq[DualFundingTx] = DualFundingTx(d.fundingTx, d.commitments) +: d.previousFundingTxs
     // We can forget other funding attempts now that one of the funding txs is confirmed.
     val otherFundingTxs = allFundingTxs.filter(_.commitments.fundingTxId != fundingTx.txid).map(_.fundingTx)
@@ -78,7 +78,7 @@ trait DualFundingHandlers extends CommonFundingHandlers {
   }
 
   def acceptDualFundingTx(d: DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED, fundingTx: Transaction, realScidStatus: RealScidStatus): Option[(DATA_WAIT_FOR_DUAL_FUNDING_READY, ChannelReady)] = {
-    pruneCommitments(d, fundingTx) map {
+    pruneCommitments(d, fundingTx).map {
       case DualFundingTx(_, commitments) =>
         watchFundingTx(commitments)
         realScidStatus match {

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/SingleFundingHandlers.scala

@@ -58,7 +58,7 @@ trait SingleFundingHandlers extends CommonFundingHandlers {
    * When we are funder, we use this function to detect when our funding tx has been double-spent (by another transaction
    * that we made for some reason). If the funding tx has been double spent we can forget about the channel.
    */
-  def checkDoubleSpent(fundingTx: Transaction): Unit = {
+  private def checkDoubleSpent(fundingTx: Transaction): Unit = {
     log.debug(s"checking status of funding tx txid=${fundingTx.txid}")
     wallet.doubleSpent(fundingTx).onComplete {
       case Success(true) =>

eclair-core/src/main/scala/fr/acinq/eclair/json/JsonSerializers.scala

@@ -445,13 +445,18 @@ object JavaUUIDSerializer extends MinimalSerializer({
 
 object ChannelEventSerializer extends MinimalSerializer({
   case e: ChannelCreated => JObject(
-    JField("type", JString("channel-opened")),
+    JField("type", JString("channel-created")),
     JField("remoteNodeId", JString(e.remoteNodeId.toString())),
     JField("isInitiator", JBool(e.isInitiator)),
     JField("temporaryChannelId", JString(e.temporaryChannelId.toHex)),
     JField("commitTxFeeratePerKw", JLong(e.commitTxFeerate.toLong)),
     JField("fundingTxFeeratePerKw", e.fundingTxFeerate.map(f => JLong(f.toLong)).getOrElse(JNothing))
   )
+  case e: ChannelOpened => JObject(
+    JField("type", JString("channel-opened")),
+    JField("remoteNodeId", JString(e.remoteNodeId.toString())),
+    JField("channelId", JString(e.channelId.toHex)),
+  )
   case e: ChannelStateChanged => JObject(
     JField("type", JString("channel-state-changed")),
     JField("channelId", JString(e.channelId.toHex)),

eclair-core/src/test/scala/fr/acinq/eclair/channel/states/c/WaitForChannelReadyStateSpec.scala

@@ -101,7 +101,10 @@ class WaitForChannelReadyStateSpec extends TestKitBaseClass with FixtureAnyFunSu
     assert(bobIds.real.isInstanceOf[RealScidStatus.Temporary])
     val channelReady = bob2alice.expectMsgType[ChannelReady]
     assert(channelReady.alias_opt.contains(bobIds.localAlias))
+    val listener = TestProbe()
+    alice.underlying.system.eventStream.subscribe(listener.ref, classOf[ChannelOpened])
     bob2alice.forward(alice)
+    listener.expectMsg(ChannelOpened(alice, bob.underlyingActor.nodeParams.nodeId, channelId(alice)))
     val initialChannelUpdate = alice.stateData.asInstanceOf[DATA_NORMAL].channelUpdate
     assert(initialChannelUpdate.shortChannelId == aliceIds.localAlias)
     assert(initialChannelUpdate.feeBaseMsat == relayFees.feeBase)
@@ -147,7 +150,10 @@ class WaitForChannelReadyStateSpec extends TestKitBaseClass with FixtureAnyFunSu
     assert(bobIds.real == RealScidStatus.Unknown)
     val channelReady = bob2alice.expectMsgType[ChannelReady]
     assert(channelReady.alias_opt.contains(bobIds.localAlias))
+    val listener = TestProbe()
+    alice.underlying.system.eventStream.subscribe(listener.ref, classOf[ChannelOpened])
     bob2alice.forward(alice)
+    listener.expectMsg(ChannelOpened(alice, bob.underlyingActor.nodeParams.nodeId, channelId(alice)))
     val initialChannelUpdate = alice.stateData.asInstanceOf[DATA_NORMAL].channelUpdate
     assert(initialChannelUpdate.shortChannelId == aliceIds.localAlias)
     assert(initialChannelUpdate.feeBaseMsat == relayFees.feeBase)

eclair-core/src/test/scala/fr/acinq/eclair/channel/states/c/WaitForDualFundingReadyStateSpec.scala

@@ -19,9 +19,9 @@ package fr.acinq.eclair.channel.states.c
 import akka.testkit.{TestFSMRef, TestProbe}
 import fr.acinq.bitcoin.scalacompat.{ByteVector32, Transaction}
 import fr.acinq.eclair.blockchain.bitcoind.ZmqWatcher._
-import fr.acinq.eclair.channel.fund.InteractiveTxBuilder.FullySignedSharedTransaction
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.channel.fsm.Channel
+import fr.acinq.eclair.channel.fund.InteractiveTxBuilder.FullySignedSharedTransaction
 import fr.acinq.eclair.channel.publish.TxPublisher
 import fr.acinq.eclair.channel.publish.TxPublisher.SetChannelId
 import fr.acinq.eclair.channel.states.{ChannelStateTestsBase, ChannelStateTestsTags}
@@ -109,11 +109,18 @@ class WaitForDualFundingReadyStateSpec extends TestKitBaseClass with FixtureAnyF
     alice.underlyingActor.nodeParams.db.peers.addOrUpdateRelayFees(bob.underlyingActor.nodeParams.nodeId, RelayFees(20 msat, 125))
     bob.underlyingActor.nodeParams.db.peers.addOrUpdateRelayFees(alice.underlyingActor.nodeParams.nodeId, RelayFees(25 msat, 90))
 
+    val listenerA = TestProbe()
+    alice.underlying.system.eventStream.subscribe(listenerA.ref, classOf[ChannelOpened])
+    val listenerB = TestProbe()
+    bob.underlying.system.eventStream.subscribe(listenerB.ref, classOf[ChannelOpened])
+
     val aliceChannelReady = alice2bob.expectMsgType[ChannelReady]
     alice2bob.forward(bob, aliceChannelReady)
+    listenerB.expectMsg(ChannelOpened(bob, alice.underlyingActor.nodeParams.nodeId, channelId(bob)))
     awaitCond(bob.stateName == NORMAL)
     val bobChannelReady = bob2alice.expectMsgType[ChannelReady]
     bob2alice.forward(alice, bobChannelReady)
+    listenerA.expectMsg(ChannelOpened(alice, bob.underlyingActor.nodeParams.nodeId, channelId(alice)))
     awaitCond(alice.stateName == NORMAL)
 
     assert(alice.stateData.asInstanceOf[DATA_NORMAL].shortIds.real.isInstanceOf[RealScidStatus.Temporary])
@@ -142,11 +149,18 @@ class WaitForDualFundingReadyStateSpec extends TestKitBaseClass with FixtureAnyF
   test("recv ChannelReady (zero-conf)", Tag(ChannelStateTestsTags.DualFunding), Tag(ChannelStateTestsTags.AnchorOutputsZeroFeeHtlcTxs), Tag(ChannelStateTestsTags.ZeroConf)) { f =>
     import f._
 
+    val listenerA = TestProbe()
+    alice.underlying.system.eventStream.subscribe(listenerA.ref, classOf[ChannelOpened])
+    val listenerB = TestProbe()
+    bob.underlying.system.eventStream.subscribe(listenerB.ref, classOf[ChannelOpened])
+
     val aliceChannelReady = alice2bob.expectMsgType[ChannelReady]
     alice2bob.forward(bob, aliceChannelReady)
+    listenerB.expectMsg(ChannelOpened(bob, alice.underlyingActor.nodeParams.nodeId, channelId(bob)))
     awaitCond(bob.stateName == NORMAL)
     val bobChannelReady = bob2alice.expectMsgType[ChannelReady]
     bob2alice.forward(alice, bobChannelReady)
+    listenerA.expectMsg(ChannelOpened(alice, bob.underlyingActor.nodeParams.nodeId, channelId(alice)))
     awaitCond(alice.stateName == NORMAL)
 
     assert(alice.stateData.asInstanceOf[DATA_NORMAL].shortIds.real == RealScidStatus.Unknown)

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

@@ -23,7 +23,7 @@ import akka.http.scaladsl.server.Route
 import akka.stream.OverflowStrategy
 import akka.stream.scaladsl.{BroadcastHub, Flow, Keep, Source}
 import fr.acinq.eclair.api.Service
-import fr.acinq.eclair.channel.{ChannelClosed, ChannelCreated, ChannelStateChanged, WAIT_FOR_INIT_INTERNAL}
+import fr.acinq.eclair.channel._
 import fr.acinq.eclair.message.OnionMessages
 import fr.acinq.eclair.payment.PaymentEvent
 
@@ -52,6 +52,7 @@ trait WebSocket {
       override def preStart: Unit = {
         context.system.eventStream.subscribe(self, classOf[PaymentEvent])
         context.system.eventStream.subscribe(self, classOf[ChannelCreated])
+        context.system.eventStream.subscribe(self, classOf[ChannelOpened])
         context.system.eventStream.subscribe(self, classOf[ChannelStateChanged])
         context.system.eventStream.subscribe(self, classOf[ChannelClosed])
         context.system.eventStream.subscribe(self, classOf[OnionMessages.ReceiveMessage])
@@ -60,6 +61,7 @@ trait WebSocket {
       def receive: Receive = {
         case message: PaymentEvent => flowInput.offer(serialization.write(message))
         case message: ChannelCreated => flowInput.offer(serialization.write(message))
+        case message: ChannelOpened => flowInput.offer(serialization.write(message))
         case message: ChannelStateChanged =>
           if (message.previousState != WAIT_FOR_INIT_INTERNAL) {
             flowInput.offer(serialization.write(message))

eclair-node/src/test/scala/fr/acinq/eclair/api/ApiServiceSpec.scala

@@ -33,7 +33,6 @@ import fr.acinq.eclair._
 import fr.acinq.eclair.api.directives.{EclairDirectives, ErrorResponse}
 import fr.acinq.eclair.api.serde.JsonSupport
 import fr.acinq.eclair.blockchain.fee.FeeratePerKw
-import fr.acinq.eclair.channel.ChannelOpenResponse.ChannelOpened
 import fr.acinq.eclair.channel.Helpers.Closing
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.crypto.Sphinx
@@ -276,7 +275,7 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
     val channelId = ByteVector32(hex"56d7d6eda04d80138270c49709f1eadb5ab4939e5061309ccdacdb98ce637d0e")
 
     val eclair = mock[Eclair]
-    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpened(channelId))
+    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpenResponse.ChannelOpened(channelId))
     val mockService = new MockService(eclair)
 
     Post("/open", FormData("nodeId" -> nodeId.toString(), "fundingSatoshis" -> "100002").toEntity) ~>
@@ -312,7 +311,7 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
     val channelId = ByteVector32(hex"56d7d6eda04d80138270c49709f1eadb5ab4939e5061309ccdacdb98ce637d0e")
 
     val eclair = mock[Eclair]
-    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpened(channelId))
+    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpenResponse.ChannelOpened(channelId))
     val mockService = new MockService(eclair)
 
     Post("/open", FormData("nodeId" -> nodeId.toString(), "fundingSatoshis" -> "25000", "channelType" -> "standard").toEntity) ~>
@@ -332,7 +331,7 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
     val channelId = ByteVector32(hex"56d7d6eda04d80138270c49709f1eadb5ab4939e5061309ccdacdb98ce637d0e")
 
     val eclair = mock[Eclair]
-    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpened(channelId))
+    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpenResponse.ChannelOpened(channelId))
     val mockService = new MockService(eclair)
 
     Post("/open", FormData("nodeId" -> nodeId.toString(), "fundingSatoshis" -> "25000", "channelType" -> "static_remotekey").toEntity) ~>
@@ -352,7 +351,7 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
     val channelId = ByteVector32(hex"56d7d6eda04d80138270c49709f1eadb5ab4939e5061309ccdacdb98ce637d0e")
 
     val eclair = mock[Eclair]
-    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpened(channelId))
+    eclair.open(any, any, any, any, any, any, any)(any[Timeout]) returns Future.successful(ChannelOpenResponse.ChannelOpened(channelId))
     val mockService = new MockService(eclair)
 
     Post("/open", FormData("nodeId" -> nodeId.toString(), "fundingSatoshis" -> "25000", "channelType" -> "anchor_outputs").toEntity) ~>
@@ -1165,11 +1164,17 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
         wsClient.expectMessage(expectedSerializedPset)
 
         val chcr = ChannelCreated(system.deadLetters, system.deadLetters, bobNodeId, isInitiator = true, ByteVector32.One, FeeratePerKw(25 sat), Some(FeeratePerKw(20 sat)))
-        val expectedSerializedChcr = """{"type":"channel-opened","remoteNodeId":"039dc0e0b1d25905e44fdf6f8e89755a5e219685840d0bc1d28d3308f9628a3585","isInitiator":true,"temporaryChannelId":"0100000000000000000000000000000000000000000000000000000000000000","commitTxFeeratePerKw":25,"fundingTxFeeratePerKw":20}"""
+        val expectedSerializedChcr = """{"type":"channel-created","remoteNodeId":"039dc0e0b1d25905e44fdf6f8e89755a5e219685840d0bc1d28d3308f9628a3585","isInitiator":true,"temporaryChannelId":"0100000000000000000000000000000000000000000000000000000000000000","commitTxFeeratePerKw":25,"fundingTxFeeratePerKw":20}"""
         assert(serialization.write(chcr) == expectedSerializedChcr)
         system.eventStream.publish(chcr)
         wsClient.expectMessage(expectedSerializedChcr)
 
+        val chop = ChannelOpened(system.deadLetters, bobNodeId, ByteVector32.One)
+        val expectedSerializedChop = """{"type":"channel-opened","remoteNodeId":"039dc0e0b1d25905e44fdf6f8e89755a5e219685840d0bc1d28d3308f9628a3585","channelId":"0100000000000000000000000000000000000000000000000000000000000000"}"""
+        assert(serialization.write(chop) == expectedSerializedChop)
+        system.eventStream.publish(chop)
+        wsClient.expectMessage(expectedSerializedChop)
+
         val chsc = ChannelStateChanged(system.deadLetters, ByteVector32.One, system.deadLetters, bobNodeId, OFFLINE, NORMAL, null)
         val expectedSerializedChsc = """{"type":"channel-state-changed","channelId":"0100000000000000000000000000000000000000000000000000000000000000","remoteNodeId":"039dc0e0b1d25905e44fdf6f8e89755a5e219685840d0bc1d28d3308f9628a3585","previousState":"OFFLINE","currentState":"NORMAL"}"""
         assert(serialization.write(chsc) == expectedSerializedChsc)