Merge pull request #867 from shocknet/chain-catchup

check chain

src/services/helpers/logger.ts

@@ -23,8 +23,7 @@ const sanitizeFileName = (fileName: string): string => {
 const openWriter = (fileName: string): Writer => {
     const now = new Date()
     const date = `${now.getFullYear()}-${z(now.getMonth() + 1)}-${z(now.getDate())}`
-    const sanitizedFileName = sanitizeFileName(fileName)
+    const logPath = `${logsDir}/${fileName}_${date}.log`
-    const logPath = `${logsDir}/${sanitizedFileName}_${date}.log`
     // Ensure parent directory exists
     const dirPath = logPath.substring(0, logPath.lastIndexOf('/'))
     if (!fs.existsSync(dirPath)) {
@@ -48,13 +47,13 @@ if (!fs.existsSync(`${logsDir}/components`)) {
 export const getLogger = (params: LoggerParams): PubLogger => {
     const writers: Writer[] = []
     if (params.appName) {
-        writers.push(openWriter(`apps/${params.appName}`))
+        writers.push(openWriter(`apps/${sanitizeFileName(params.appName)}`))
     }
     if (params.userId) {
-        writers.push(openWriter(`users/${params.userId}`))
+        writers.push(openWriter(`users/${sanitizeFileName(params.userId)}`))
     }
     if (params.component) {
-        writers.push(openWriter(`components/${params.component}`))
+        writers.push(openWriter(`components/${sanitizeFileName(params.component)}`))
     }
     if (writers.length === 0) {
         writers.push(rootWriter)

src/services/lnd/lnd.ts

@@ -23,6 +23,7 @@ import { TxPointSettings } from '../storage/tlv/stateBundler.js';
 import { WalletKitClient } from '../../../proto/lnd/walletkit.client.js';
 import SettingsManager from '../main/settingsManager.js';
 import { LndNodeSettings, LndSettings } from '../main/settings.js';
+import { ListAddressesResponse } from '../../../proto/lnd/walletkit.js';
 
 const DeadLineMetadata = (deadline = 10 * 1000) => ({ deadline: Date.now() + deadline })
 const deadLndRetrySeconds = 20
@@ -32,6 +33,7 @@ type NodeSettingsOverride = {
     lndCertPath: string
     lndMacaroonPath: string
 }
+export type LndAddress = { address: string, change: boolean }
 export default class {
     lightning: LightningClient
     invoices: InvoicesClient
@@ -53,6 +55,7 @@ export default class {
     liquidProvider: LiquidityProvider
     utils: Utils
     unlockLnd: () => Promise<void>
+    addressesCache: Record<string, { isChange: boolean }> = {}
     constructor(getSettings: () => { lndSettings: LndSettings, lndNodeSettings: LndNodeSettings }, liquidProvider: LiquidityProvider, unlockLnd: () => Promise<any>, utils: Utils, addressPaidCb: AddressPaidCb, invoicePaidCb: InvoicePaidCb, newBlockCb: NewBlockCb, htlcCb: HtlcCb, channelEventCb: ChannelEventCb) {
         this.getSettings = getSettings
         this.utils = utils
@@ -331,6 +334,26 @@ export default class {
         })
     }
 
+    async IsChangeAddress(address: string): Promise<boolean> {
+        const cached = this.addressesCache[address]
+        if (cached) {
+            return cached.isChange
+        }
+        const addresses = await this.ListAddresses()
+        const addr = addresses.find(a => a.address === address)
+        if (!addr) {
+            throw new Error(`address ${address} not found in list of addresses`)
+        }
+        return addr.change
+    }
+
+    async ListAddresses(): Promise<LndAddress[]> {
+        const res = await this.walletKit.listAddresses({ accountName: "", showCustomAccounts: false }, DeadLineMetadata())
+        const addresses = res.response.accountWithAddresses.map(a => a.addresses.map(a => ({ address: a.address, change: a.isInternal }))).flat()
+        addresses.forEach(a => this.addressesCache[a.address] = { isChange: a.change })
+        return addresses
+    }
+
     async NewAddress(addressType: Types.AddressType, { useProvider, from }: TxActionOptions): Promise<NewAddressResponse> {
         // Force use of provider when bypass is enabled (addresses not supported by provider, but we should fail gracefully)
         if (this.liquidProvider.getSettings().useOnlyLiquidityProvider) {

src/services/main/index.ts

@@ -80,7 +80,7 @@ export default class {
         this.liquidityManager = new LiquidityManager(this.settings, this.storage, this.utils, this.liquidityProvider, this.lnd, this.rugPullTracker)
         this.metricsManager = new MetricsManager(this.storage, this.lnd)
 
-        this.paymentManager = new PaymentManager(this.storage, this.lnd, this.settings, this.liquidityManager, this.utils, this.addressPaidCb, this.invoicePaidCb)
+        this.paymentManager = new PaymentManager(this.storage, this.metricsManager, this.lnd, this.settings, this.liquidityManager, this.utils, this.addressPaidCb, this.invoicePaidCb)
         this.productManager = new ProductManager(this.storage, this.paymentManager, this.settings)
         this.applicationManager = new ApplicationManager(this.storage, this.settings, this.paymentManager)
         this.appUserManager = new AppUserManager(this.storage, this.settings, this.applicationManager)
@@ -213,6 +213,10 @@ export default class {
             const { blockHeight } = await this.lnd.GetInfo()
             const userAddress = await this.storage.paymentStorage.GetAddressOwner(address, tx)
             if (!userAddress) {
+                const isChange = await this.lnd.IsChangeAddress(address)
+                if (isChange) {
+                    return
+                }
                 await this.metricsManager.AddRootAddressPaid(address, txOutput, amount)
                 return
             }

src/services/main/init.ts

@@ -72,7 +72,8 @@ export const initMainHandler = async (log: PubLogger, settingsManager: SettingsM
     if (stop) {
         return
     }
-    await mainHandler.paymentManager.checkPendingPayments()
+    await mainHandler.paymentManager.checkPaymentStatus()
+    await mainHandler.paymentManager.checkMissedChainTxs()
     await mainHandler.paymentManager.CleanupOldUnpaidInvoices()
     await mainHandler.appUserManager.CleanupInactiveUsers()
     await mainHandler.appUserManager.CleanupNeverActiveUsers()

src/services/main/paymentManager.ts

@@ -12,12 +12,16 @@ import { Payment_PaymentStatus } from '../../../proto/lnd/lightning.js'
 import { Event, verifiedSymbol, verifyEvent } from 'nostr-tools'
 import { AddressReceivingTransaction } from '../storage/entity/AddressReceivingTransaction.js'
 import { UserTransactionPayment } from '../storage/entity/UserTransactionPayment.js'
+import { UserReceivingAddress } from '../storage/entity/UserReceivingAddress.js'
 import { Watchdog } from './watchdog.js'
 import { LiquidityManager } from './liquidityManager.js'
 import { Utils } from '../helpers/utilsWrapper.js'
 import { UserInvoicePayment } from '../storage/entity/UserInvoicePayment.js'
 import SettingsManager from './settingsManager.js'
 import { Swaps, TransactionSwapData } from '../lnd/swaps.js'
+import { Transaction, OutputDetail } from '../../../proto/lnd/lightning.js'
+import { LndAddress } from '../lnd/lnd.js'
+import Metrics from '../metrics/index.js'
 interface UserOperationInfo {
     serial_id: number
     paid_amount: number
@@ -56,8 +60,10 @@ export default class {
     utils: Utils
     swaps: Swaps
     invoiceLock: InvoiceLock
-    constructor(storage: Storage, lnd: LND, settings: SettingsManager, liquidityManager: LiquidityManager, utils: Utils, addressPaidCb: AddressPaidCb, invoicePaidCb: InvoicePaidCb) {
+    metrics: Metrics
+    constructor(storage: Storage, metrics: Metrics, lnd: LND, settings: SettingsManager, liquidityManager: LiquidityManager, utils: Utils, addressPaidCb: AddressPaidCb, invoicePaidCb: InvoicePaidCb) {
         this.storage = storage
+        this.metrics = metrics
         this.settings = settings
         this.lnd = lnd
         this.liquidityManager = liquidityManager
@@ -75,11 +81,11 @@ export default class {
         this.swaps.Stop()
     }
 
-    checkPendingPayments = async () => {
+    checkPaymentStatus = async () => {
-        const log = getLogger({ component: 'pendingPaymentsOnStart' })
+        const log = getLogger({ component: 'checkPaymentStatus' })
         const pendingPayments = await this.storage.paymentStorage.GetPendingPayments()
         for (const p of pendingPayments) {
-            log("checking state of payment: ", p.invoice)
+            log("checking status of payment: ", p.invoice)
             if (p.internal) {
                 log("found pending internal payment", p.serial_id)
             } else if (p.liquidityProvider) {
@@ -170,6 +176,120 @@ export default class {
         }
     }
 
+    checkMissedChainTxs = async () => {
+        const log = getLogger({ component: 'checkMissedChainTxs' })
+
+        if (this.liquidityManager.settings.getSettings().liquiditySettings.useOnlyLiquidityProvider) {
+            log("USE_ONLY_LIQUIDITY_PROVIDER enabled, skipping chain tx check")
+            return
+        }
+
+        try {
+            const { txs, currentHeight, lndPubkey } = await this.getLatestTransactions(log)
+
+            const recoveredCount = await this.processMissedChainTransactions(txs, log)
+
+            // Update latest checked height to current block height
+            await this.storage.liquidityStorage.UpdateLatestCheckedHeight('lnd', lndPubkey, currentHeight)
+
+            if (recoveredCount > 0) {
+                log(`processed ${recoveredCount} missed chain tx(s)`)
+            } else {
+                log("no missed chain transactions found")
+            }
+        } catch (err: any) {
+            log(ERROR, "failed to check for missed chain transactions:", err.message || err)
+        }
+    }
+
+    private async getLatestTransactions(log: PubLogger): Promise<{ txs: Transaction[], currentHeight: number, lndPubkey: string }> {
+        const lndInfo = await this.lnd.GetInfo()
+        const lndPubkey = lndInfo.identityPubkey
+
+        const startHeight = await this.storage.liquidityStorage.GetLatestCheckedHeight('lnd', lndPubkey)
+        log(`checking for missed confirmed chain transactions from height ${startHeight}...`)
+
+        const { transactions } = await this.lnd.GetTransactions(startHeight)
+        log(`retrieved ${transactions.length} transactions from LND`)
+
+        return { txs: transactions, currentHeight: lndInfo.blockHeight, lndPubkey }
+    }
+
+    private async processMissedChainTransactions(transactions: Transaction[], log: PubLogger): Promise<number> {
+        let recoveredCount = 0
+        const addresses = await this.lnd.ListAddresses()
+        for (const tx of transactions) {
+            if (!tx.outputDetails || tx.outputDetails.length === 0) {
+                continue
+            }
+
+            const outputsWithAddresses = await this.collectOutputsWithAddresses(tx)
+
+            for (const { output, userAddress } of outputsWithAddresses) {
+                if (!userAddress) {
+                    await this.processRootAddressOutput(output, tx, addresses, log)
+                    continue
+                }
+
+                const processed = await this.processUserAddressOutput(output, tx, log)
+                if (processed) {
+                    recoveredCount++
+                }
+            }
+        }
+
+        return recoveredCount
+    }
+
+    private async collectOutputsWithAddresses(tx: Transaction) {
+        const outputs: { output: OutputDetail, userAddress: UserReceivingAddress | null, tx: Transaction }[] = []
+        for (const output of tx.outputDetails) {
+            if (!output.address || !output.isOurAddress) {
+                continue
+            }
+            const userAddress = await this.storage.paymentStorage.GetAddressOwner(output.address)
+            outputs.push({ output, userAddress, tx })
+        }
+        return outputs
+    }
+
+    private async processRootAddressOutput(output: OutputDetail, tx: Transaction, addresses: LndAddress[], log: PubLogger): Promise<boolean> {
+        const addr = addresses.find(a => a.address === output.address)
+        if (!addr) {
+            throw new Error(`address ${output.address} not found in list of addresses`)
+        }
+        if (addr.change) {
+            log(`ignoring change address ${output.address}`)
+            return false
+        }
+        const outputIndex = Number(output.outputIndex)
+        const existingRootOp = await this.metrics.GetRootAddressTransaction(output.address, tx.txHash, outputIndex)
+        if (existingRootOp) {
+            return false
+        }
+        this.addressPaidCb({ hash: tx.txHash, index: outputIndex }, output.address, Number(output.amount), 'lnd')
+            .catch(err => log(ERROR, "failed to process root address output:", err.message || err))
+        return true
+    }
+
+    private async processUserAddressOutput(output: OutputDetail, tx: Transaction, log: PubLogger) {
+        const existingTx = await this.storage.paymentStorage.GetAddressReceivingTransactionOwner(
+            output.address,
+            tx.txHash
+        )
+
+        if (existingTx) {
+            return false
+        }
+
+        const amount = Number(output.amount)
+        const outputIndex = Number(output.outputIndex)
+        log(`processing missed chain tx: address=${output.address}, txHash=${tx.txHash}, amount=${amount}, outputIndex=${outputIndex}`)
+        this.addressPaidCb({ hash: tx.txHash, index: outputIndex }, output.address, amount, 'lnd')
+            .catch(err => log(ERROR, "failed to process user address output:", err.message || err))
+        return true
+    }
+
     getReceiveServiceFee = (action: Types.UserOperationType, amount: number, managedUser: boolean): number => {
         switch (action) {
             case Types.UserOperationType.INCOMING_TX:
@@ -640,21 +760,6 @@ export default class {
 
     async GetLnurlWithdrawInfo(balanceCheckK1: string): Promise<Types.LnurlWithdrawInfoResponse> {
         throw new Error("LNURL withdraw currenlty not supported for non application users")
-        /*const key = await this.storage.paymentStorage.UseUserEphemeralKey(balanceCheckK1, 'balanceCheck')
-        const maxWithdrawable = this.GetMaxPayableInvoice(key.user.balance_sats)
-        const callbackK1 = await this.storage.paymentStorage.AddUserEphemeralKey(key.user.user_id, 'withdraw')
-        const newBalanceCheckK1 = await this.storage.paymentStorage.AddUserEphemeralKey(key.user.user_id, 'balanceCheck')
-        const payInfoK1 = await this.storage.paymentStorage.AddUserEphemeralKey(key.user.user_id, 'pay')
-        return {
-            tag: "withdrawRequest",
-            callback: `${this.settings.serviceUrl}/api/guest/lnurl_withdraw/handle`,
-            defaultDescription: "lnurl withdraw from lightning.pub",
-            k1: callbackK1.key,
-            maxWithdrawable: maxWithdrawable * 1000,
-            minWithdrawable: 10000,
-            balanceCheck: this.balanceCheckUrl(newBalanceCheckK1.key),
-            payLink: `${this.settings.serviceUrl}/api/guest/lnurl_pay/info?k1=${payInfoK1.key}`,
-        }*/
     }
 
     async HandleLnurlWithdraw(k1: string, invoice: string): Promise<void> {

src/services/metrics/index.ts

@@ -414,6 +414,10 @@ export default class Handler {
         await this.storage.metricsStorage.AddRootOperation("chain", `${address}:${txOutput.hash}:${txOutput.index}`, amount)
     }
 
+    async GetRootAddressTransaction(address: string, txHash: string, index: number) {
+        return this.storage.metricsStorage.GetRootOperation("chain", `${address}:${txHash}:${index}`)
+    }
+
     async AddRootInvoicePaid(paymentRequest: string, amount: number) {
         await this.storage.metricsStorage.AddRootOperation("invoice", paymentRequest, amount)
     }

src/services/storage/entity/TrackedProvider.ts

@@ -18,6 +18,9 @@ export class TrackedProvider {
     @Column({ default: 0 })
     latest_distruption_at_unix: number
 
+    @Column({ default: 0 })
+    latest_checked_height: number
+
     @CreateDateColumn()
     created_at: Date
 

src/services/storage/liquidityStorage.ts

@@ -64,4 +64,13 @@ export class LiquidityStorage {
     async UpdateTrackedProviderDisruption(providerType: 'lnd' | 'lnPub', pub: string, latestDisruptionAtUnix: number) {
         return this.dbs.Update<TrackedProvider>('TrackedProvider', { provider_pubkey: pub, provider_type: providerType }, { latest_distruption_at_unix: latestDisruptionAtUnix })
     }
+
+    async GetLatestCheckedHeight(providerType: 'lnd' | 'lnPub', pub: string): Promise<number> {
+        const provider = await this.GetTrackedProvider(providerType, pub)
+        return provider?.latest_checked_height || 0
+    }
+
+    async UpdateLatestCheckedHeight(providerType: 'lnd' | 'lnPub', pub: string, height: number) {
+        return this.dbs.Update<TrackedProvider>('TrackedProvider', { provider_pubkey: pub, provider_type: providerType }, { latest_checked_height: height })
+    }
 }

src/services/storage/metricsStorage.ts

@@ -149,6 +149,10 @@ export default class {
         return this.dbs.CreateAndSave<RootOperation>('RootOperation', { operation_type: opType, operation_amount: amount, operation_identifier: id, at_unix: Math.floor(Date.now() / 1000) }, txId)
     }
 
+    async GetRootOperation(opType: string, id: string, txId?: string) {
+        return this.dbs.FindOne<RootOperation>('RootOperation', { where: { operation_type: opType, operation_identifier: id } }, txId)
+    }
+
     async GetRootOperations({ from, to }: { from?: number, to?: number }, txId?: string) {
         const q = getTimeQuery({ from, to })
         return this.dbs.Find<RootOperation>('RootOperation', q, txId)

src/services/storage/migrations/1766504040000-tracked_provider_height.ts

@@ -0,0 +1,14 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class TrackedProviderHeight1766504040000 implements MigrationInterface {
+    name = 'TrackedProviderHeight1766504040000'
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`ALTER TABLE "tracked_provider" ADD "latest_checked_height" integer NOT NULL DEFAULT (0)`);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`ALTER TABLE "tracked_provider" DROP COLUMN "latest_checked_height"`);
+    }
+}
+

src/services/storage/migrations/runner.ts

@@ -30,13 +30,14 @@ import { AdminSettings1761683639419 } from './1761683639419-admin_settings.js'
 import { TxSwap1762890527098 } from './1762890527098-tx_swap.js'
 import { TxSwapAddress1764779178945 } from './1764779178945-tx_swap_address.js'
 import { ClinkRequester1765497600000 } from './1765497600000-clink_requester.js'
+import { TrackedProviderHeight1766504040000 } from './1766504040000-tracked_provider_height.js'
 
 
 export const allMigrations = [Initial1703170309875, LspOrder1718387847693, LiquidityProvider1719335699480, LndNodeInfo1720187506189,
     TrackedProvider1720814323679, CreateInviteTokenTable1721751414878, PaymentIndex1721760297610, DebitAccess1726496225078, DebitAccessFixes1726685229264,
     DebitToPub1727105758354, UserCbUrl1727112281043, UserOffer1733502626042, ManagementGrant1751307732346, ManagementGrantBanned1751989251513,
     InvoiceCallbackUrls1752425992291, OldSomethingLeftover1753106599604, UserReceivingInvoiceIdx1753109184611, AppUserDevice1753285173175,
-    UserAccess1759426050669, AddBlindToUserOffer1760000000000, ApplicationAvatarUrl1761000001000, AdminSettings1761683639419, TxSwap1762890527098, TxSwapAddress1764779178945, ClinkRequester1765497600000]
+    UserAccess1759426050669, AddBlindToUserOffer1760000000000, ApplicationAvatarUrl1761000001000, AdminSettings1761683639419, TxSwap1762890527098, TxSwapAddress1764779178945, ClinkRequester1765497600000, TrackedProviderHeight1766504040000]
 
 export const allMetricsMigrations = [LndMetrics1703170330183, ChannelRouting1709316653538, HtlcCount1724266887195, BalanceEvents1724860966825, RootOps1732566440447, RootOpsTime1745428134124, ChannelEvents1750777346411]
 /* export const TypeOrmMigrationRunner = async (log: PubLogger, storageManager: Storage, settings: DbSettings, arg: string | undefined): Promise<boolean> => {