External Node Channel Open Flow

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }
 

app/src/main/java/to/bitkit/App.kt

@@ -9,7 +9,6 @@ import androidx.hilt.work.HiltWorkerFactory
 import androidx.work.Configuration
 import dagger.hilt.android.HiltAndroidApp
 import to.bitkit.env.Env
-import to.bitkit.utils.ResourceProvider
 import javax.inject.Inject
 
 @HiltAndroidApp
@@ -24,7 +23,6 @@ internal open class App : Application(), Configuration.Provider {
 
  override fun onCreate() {
  super.onCreate()
- ResourceProvider.init(this)
  currentActivity = CurrentActivity().also { registerActivityLifecycleCallbacks(it) }
 
  Env.initAppStoragePath(filesDir.absolutePath)

app/src/main/java/to/bitkit/env/Env.kt

@@ -105,17 +105,5 @@ internal object Env {
  nodeId = "028a8910b0048630d4eb17af25668cdd7ea6f2d8ae20956e7a06e2ae46ebcb69fc",
  address = "34.65.86.104:9400",
  )
- val btStagingOld = LnPeer(
- nodeId = "03b9a456fb45d5ac98c02040d39aec77fa3eeb41fd22cf40b862b393bcfc43473a",
- address = "35.233.47.252:9400",
- )
- val polarToRegtest = LnPeer(
- nodeId = "023f6e310ff049d68c64a0eb97440b998aa15fd99162317d6743d7023519862e23",
- address = "10.0.2.2:9735",
- )
- val local = LnPeer(
- nodeId = "02faf2d1f5dc153e8931d8444c4439e46a81cb7eeadba8562e7fec3690c261ce87",
- address = "10.0.2.2:9738",
- )
  }
 }

app/src/main/java/to/bitkit/ext/Flows.kt

@@ -0,0 +1,41 @@
+package to.bitkit.ext
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.takeWhile
+
+/**
+ * Suspends and collects the elements of the Flow until the provided predicate satisfies
+ * a `WatchResult.Complete`.
+ *
+ * @param predicate A suspending function that processes each emitted value and returns a
+ * `WatchResult` indicating whether to continue or complete with a result.
+ * @return The result of type `R` when the `WatchResult.Complete` is returned by the predicate.
+ */
+suspend inline fun <T, R> Flow<T>.watchUntil(
+ crossinline predicate: suspend (T) -> WatchResult<R>,
+): R {
+ val result = CompletableDeferred<R>()
+
+ this.takeWhile { value ->
+ when (val eventResult = predicate(value)) {
+ is WatchResult.Continue -> {
+ eventResult.result?.let { result.complete(it) }
+ true
+ }
+
+ is WatchResult.Complete -> {
+ result.complete(eventResult.result)
+ false
+ }
+ }
+ }.collect()
+
+ return result.await()
+}
+
+sealed interface WatchResult<T> {
+ data class Continue<T>(val result: T? = null) : WatchResult<T>
+ data class Complete<T>(val result: T) : WatchResult<T>
+}

app/src/main/java/to/bitkit/models/LnPeer.kt

@@ -17,12 +17,39 @@ data class LnPeer(
  )
 
  val address get() = "$host:$port"
+
  override fun toString() = "$nodeId@${address}"
 
  companion object {
  fun PeerDetails.toLnPeer() = LnPeer(
  nodeId = nodeId,
  address = address,
  )
+
+ fun parseUri(string: String): Result<LnPeer> {
+ val uri = string.split("@")
+ val nodeId = uri[0]
+
+ if (uri.size != 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val address = uri[1].split(":")
+
+ if (address.size < 2) {
+ return Result.failure(Exception("Invalid peer uri"))
+ }
+
+ val ip = address[0]
+ val port = address[1]
+
+ return Result.success(
+ LnPeer(
+ nodeId = nodeId,
+ host = ip,
+ port = port,
+ )
+ )
+ }
  }
 }

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -26,6 +26,7 @@ import org.lightningdevkit.ldknode.NodeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentId
 import org.lightningdevkit.ldknode.Txid
+import org.lightningdevkit.ldknode.UserChannelId
 import org.lightningdevkit.ldknode.defaultConfig
 import to.bitkit.async.BaseCoroutineScope
 import to.bitkit.async.ServiceQueue
@@ -227,6 +228,26 @@ class LightningService @Inject constructor(
  }
  }
 
+ suspend fun connectPeer(peer: LnPeer): Result<Unit> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
+
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Connecting peer: $peer")
+
+ node.connect(peer.nodeId, peer.address, persist = true)
+
+ Logger.info("Peer connected: $peer")
+
+ Result.success(Unit)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Peer connect error: $peer", error)
+ Result.failure(error)
+ }
+ }
+ }
+
  suspend fun disconnectPeer(peer: LnPeer) {
  val node = this.node ?: throw ServiceError.NodeNotSetup
  Logger.debug("Disconnecting peer: $peer")
@@ -238,25 +259,36 @@ class LightningService @Inject constructor(
  } catch (e: NodeException) {
  Logger.warn("Peer disconnect error: $peer", LdkError(e))
  }
- }
- // endregion
+ } // endregion
 
  // region channels
- suspend fun openChannel(peer: LnPeer, channelAmountSats: ULong, pushToCounterpartySats: ULong? = null) {
- val node = this@LightningService.node ?: throw ServiceError.NodeNotSetup
+ suspend fun openChannel(
+ peer: LnPeer,
+ channelAmountSats: ULong,
+ pushToCounterpartySats: ULong? = null,
+ ) : Result<UserChannelId> {
+ val node = this.node ?: throw ServiceError.NodeNotSetup
 
- try {
- ServiceQueue.LDK.background {
- node.openChannel(
+ return ServiceQueue.LDK.background {
+ try {
+ Logger.debug("Initiating channel open (sats: $channelAmountSats) with peer: $peer")
+
+ val userChannelId = node.openChannel(
  nodeId = peer.nodeId,
  address = peer.address,
  channelAmountSats = channelAmountSats,
  pushToCounterpartyMsat = pushToCounterpartySats?.millis,
  channelConfig = null,
  )
+
+ Logger.info("Channel open initiated, userChannelId: $userChannelId")
+
+ Result.success(userChannelId)
+ } catch (e: NodeException) {
+ val error = LdkError(e)
+ Logger.error("Error initiating channel open", error)
+ Result.failure(error)
  }
- } catch (e: NodeException) {
- throw LdkError(e)
  }
  }