Skip to content

Move closed channels tracking to LightningRepo #474

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 20, 2025
Merged

Move closed channels tracking to LightningRepo #474

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 96 additions & 1 deletion app/src/main/java/to/bitkit/repositories/LightningRepo.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package to.bitkit.repositories

import com.google.firebase.messaging.FirebaseMessaging
import com.synonym.bitkitcore.ClosedChannelDetails
import com.synonym.bitkitcore.FeeRates
import com.synonym.bitkitcore.LightningInvoice
import com.synonym.bitkitcore.Scanner
Expand All @@ -27,6 +28,7 @@ import org.lightningdevkit.ldknode.BalanceDetails
import org.lightningdevkit.ldknode.BestBlock
import org.lightningdevkit.ldknode.ChannelConfig
import org.lightningdevkit.ldknode.ChannelDetails
import org.lightningdevkit.ldknode.Event
import org.lightningdevkit.ldknode.NodeStatus
import org.lightningdevkit.ldknode.PaymentDetails
import org.lightningdevkit.ldknode.PaymentId
Expand Down Expand Up @@ -57,6 +59,7 @@ import to.bitkit.services.NodeEventHandler
import to.bitkit.utils.AppError
import to.bitkit.utils.Logger
import to.bitkit.utils.ServiceError
import java.util.concurrent.ConcurrentHashMap
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.time.Duration
Expand Down Expand Up @@ -85,6 +88,8 @@ class LightningRepo @Inject constructor(
private val _isRecoveryMode = MutableStateFlow(false)
val isRecoveryMode = _isRecoveryMode.asStateFlow()

private val channelCache = ConcurrentHashMap<String, ChannelDetails>()

/**
* Executes the provided operation only if the node is running.
* If the node is not running, waits for it to be running for a specified timeout.
Expand Down Expand Up @@ -203,12 +208,16 @@ class LightningRepo @Inject constructor(
if (getStatus()?.isRunning == true) {
Logger.info("LDK node already running", context = TAG)
_lightningState.update { it.copy(nodeLifecycleState = NodeLifecycleState.Running) }
lightningService.listenForEvents(onEvent = eventHandler)
lightningService.listenForEvents(onEvent = { event ->
handleLdkEvent(event)
eventHandler?.invoke(event)
})
return@withContext Result.success(Unit)
}

// Start the node service
lightningService.start(timeout) { event ->
handleLdkEvent(event)
eventHandler?.invoke(event)
ldkNodeEventBus.emit(event)
}
Expand All @@ -220,6 +229,7 @@ class LightningRepo @Inject constructor(
// Initial state sync
syncState()
updateGeoBlockState()
refreshChannelCache()

// Perform post-startup tasks
connectToTrustedPeers().onFailure { e ->
Expand Down Expand Up @@ -296,12 +306,96 @@ class LightningRepo @Inject constructor(

_lightningState.update { it.copy(isSyncingWallet = true) }
lightningService.sync()
refreshChannelCache()
syncState()
_lightningState.update { it.copy(isSyncingWallet = false) }

Result.success(Unit)
}

private suspend fun refreshChannelCache() = withContext(bgDispatcher) {
val channels = lightningService.channels ?: return@withContext
channels.forEach { channel ->
channelCache[channel.channelId] = channel
}
}

private fun handleLdkEvent(event: Event) {
when (event) {
is Event.ChannelPending -> {
scope.launch {
refreshChannelCache()
}
}
is Event.ChannelReady -> {
scope.launch {
refreshChannelCache()
}
}
is Event.ChannelClosed -> {
val channelId = event.channelId
val reason = event.reason?.toString() ?: ""
scope.launch {
registerClosedChannel(channelId, reason)
}
}
else -> {
// Other events don't need special handling
}
}
}

private suspend fun registerClosedChannel(channelId: String, reason: String?) = withContext(bgDispatcher) {
try {
val channel = channelCache[channelId] ?: run {
Logger.error(
"Could not find channel details for closed channel: channelId=$channelId",
context = TAG
)
return@withContext
}

val fundingTxo = channel.fundingTxo
if (fundingTxo == null) {
Logger.error(
"Channel has no funding transaction, cannot persist closed channel: channelId=$channelId",
context = TAG
)
return@withContext
}

val channelName = channel.inboundScidAlias?.toString()
?: channel.channelId.take(CHANNEL_ID_PREVIEW_LENGTH) + "…"

val closedAt = (System.currentTimeMillis() / 1000L).toULong()

val closedChannel = ClosedChannelDetails(
channelId = channel.channelId,
counterpartyNodeId = channel.counterpartyNodeId,
fundingTxoTxid = fundingTxo.txid,
fundingTxoIndex = fundingTxo.vout,
channelValueSats = channel.channelValueSats,
closedAt = closedAt,
outboundCapacityMsat = channel.outboundCapacityMsat,
inboundCapacityMsat = channel.inboundCapacityMsat,
counterpartyUnspendablePunishmentReserve = channel.counterpartyUnspendablePunishmentReserve,
unspendablePunishmentReserve = channel.unspendablePunishmentReserve ?: 0u,
forwardingFeeProportionalMillionths = channel.config.forwardingFeeProportionalMillionths,
forwardingFeeBaseMsat = channel.config.forwardingFeeBaseMsat,
channelName = channelName,
channelClosureReason = reason.orEmpty()
)

coreService.activity.upsertClosedChannelList(listOf(closedChannel))

channelCache.remove(channelId)

Logger.info("Registered closed channel: ${channel.userChannelId}", context = TAG)
} catch (e: Throwable) {
Logger.error("Failed to register closed channel: $e", e, context = TAG)
}
}

suspend fun wipeStorage(walletIndex: Int): Result<Unit> = withContext(bgDispatcher) {
Logger.debug("wipeStorage called, stopping node first", context = TAG)
stop().onSuccess {
Expand Down Expand Up @@ -867,6 +961,7 @@ class LightningRepo @Inject constructor(
companion object {
private const val TAG = "LightningRepo"
private const val SYNC_TIMEOUT_MS = 20_000L
private const val CHANNEL_ID_PREVIEW_LENGTH = 10
}
}

Expand Down
85 changes: 0 additions & 85 deletions app/src/main/java/to/bitkit/services/LightningService.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
package to.bitkit.services

import com.synonym.bitkitcore.ClosedChannelDetails
import com.synonym.bitkitcore.upsertClosedChannel
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
Expand Down Expand Up @@ -51,7 +49,6 @@ import to.bitkit.utils.LdkError
import to.bitkit.utils.LdkLogWriter
import to.bitkit.utils.Logger
import to.bitkit.utils.ServiceError
import java.util.concurrent.ConcurrentHashMap
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.io.path.Path
Expand All @@ -74,8 +71,6 @@ class LightningService @Inject constructor(

private lateinit var trustedPeers: List<PeerDetails>

private val channelCache = ConcurrentHashMap<String, ChannelDetails>()

suspend fun setup(
walletIndex: Int,
customServerUrl: String? = null,
Expand Down Expand Up @@ -196,7 +191,6 @@ class LightningService @Inject constructor(
}
}

refreshChannelCache()
Logger.info("Node started")
}

Expand Down Expand Up @@ -232,79 +226,10 @@ class LightningService @Inject constructor(
node.syncWallets()
// launch { setMaxDustHtlcExposureForCurrentChannels() }
}
refreshChannelCache()

Logger.debug("LDK synced")
}

private suspend fun refreshChannelCache() {
val node = this.node ?: return

ServiceQueue.LDK.background {
val channels = node.listChannels()
channels.forEach { channel ->
channelCache[channel.channelId] = channel
}
}
}

private suspend fun registerClosedChannel(channelId: String, reason: String?) {
try {
val channel = ServiceQueue.LDK.background {
channelCache[channelId]
} ?: run {
Logger.error(
"Could not find channel details for closed channel: channelId=$channelId",
context = TAG
)
return@registerClosedChannel
}

val fundingTxo = channel.fundingTxo
if (fundingTxo == null) {
Logger.error(
"Channel has no funding transaction, cannot persist closed channel: channelId=$channelId",
context = TAG
)
return@registerClosedChannel
}

val channelName = channel.inboundScidAlias?.toString()
?: channel.channelId.take(CHANNEL_ID_PREVIEW_LENGTH) + "…"

val closedAt = (System.currentTimeMillis() / 1000L).toULong()

val closedChannel = ClosedChannelDetails(
channelId = channel.channelId,
counterpartyNodeId = channel.counterpartyNodeId,
fundingTxoTxid = fundingTxo.txid,
fundingTxoIndex = fundingTxo.vout,
channelValueSats = channel.channelValueSats,
closedAt = closedAt,
outboundCapacityMsat = channel.outboundCapacityMsat,
inboundCapacityMsat = channel.inboundCapacityMsat,
counterpartyUnspendablePunishmentReserve = channel.counterpartyUnspendablePunishmentReserve,
unspendablePunishmentReserve = channel.unspendablePunishmentReserve ?: 0u,
forwardingFeeProportionalMillionths = channel.config.forwardingFeeProportionalMillionths,
forwardingFeeBaseMsat = channel.config.forwardingFeeBaseMsat,
channelName = channelName,
channelClosureReason = reason.orEmpty()
)

ServiceQueue.CORE.background {
upsertClosedChannel(closedChannel)
}

ServiceQueue.LDK.background {
channelCache.remove(channelId)
}

Logger.info("Registered closed channel: ${channel.userChannelId}", context = TAG)
} catch (e: Exception) {
Logger.error("Failed to register closed channel: $e", e, context = TAG)
}
}

// private fun setMaxDustHtlcExposureForCurrentChannels() {
// if (Env.network != Network.REGTEST) {
// Logger.debug("Not updating channel config for non-regtest network")
Expand Down Expand Up @@ -815,9 +740,6 @@ class LightningService @Inject constructor(
Logger.info(
"⏳ Channel pending: channelId: $channelId userChannelId: $userChannelId formerTemporaryChannelId: $formerTemporaryChannelId counterpartyNodeId: $counterpartyNodeId fundingTxo: $fundingTxo"
)
launch {
refreshChannelCache()
}
}

is Event.ChannelReady -> {
Expand All @@ -827,9 +749,6 @@ class LightningService @Inject constructor(
Logger.info(
"👐 Channel ready: channelId: $channelId userChannelId: $userChannelId counterpartyNodeId: $counterpartyNodeId"
)
launch {
refreshChannelCache()
}
}

is Event.ChannelClosed -> {
Expand All @@ -840,9 +759,6 @@ class LightningService @Inject constructor(
Logger.info(
"⛔ Channel closed: channelId: $channelId userChannelId: $userChannelId counterpartyNodeId: $counterpartyNodeId reason: $reason"
)
launch {
registerClosedChannel(channelId, reason)
}
}
}
}
Expand All @@ -867,7 +783,6 @@ class LightningService @Inject constructor(

companion object {
private const val TAG = "LightningService"
private const val CHANNEL_ID_PREVIEW_LENGTH = 10
}
}

Expand Down
Loading