import { BitField , HashByteOrder , ShortChannelId } from "@node-lightning/core" ; import { OutPoint } from "@node-lightning/core" ; import { ILogger } from "@node-lightning/logger" ; import { EventEmitter } from "events" ; import { ChannelAnnouncementMessage } from "../messages/ChannelAnnouncementMessage" ; import { ExtendedChannelAnnouncementMessage } from "../messages/ExtendedChannelAnnouncementMessage" ; import { IWireMessage } from "../messages/IWireMessage" ; import { MessageType } from "../MessageType" ; import { Peer } from "../Peer" ; import { PeerHostRecord } from "../PeerHostRecord" ; import { PeerState } from "../PeerState" ; import { WireError , WireErrorCode } from "../WireError" ; import { DnsPeerQuery } from "./DnsPeerQuery" ; import { GossipFilter } from "./GossipFilter" ; import { GossipPeer } from "./GossipPeer" ; import { GossipRelay , IGossipRelay } from "./GossipRelay" ; import { IGossipStore } from "./GossipStore" ; import { IGossipFilterChainClient } from "./IGossipFilterChainClient" ; export enum SyncState { Unsynced , Syncing , Synced } export declare interface GossipManager { on ( event , fn ) : this ; on ( event , fn ) : this ; on ( event , fn ) : this ; off ( event , fn ) : this ; off ( event , fn ) : this ; off ( event , fn ) : this ; off ( event , fn ) : this ; } /**
 * GossipManager provides is a facade for many parts of gossip. It
 * orchestrates for validating, storing, and emitting
 * routing gossip traffic obtained by peers.
 */ export class GossipManager extends EventEmitter { public blockHeight : number ; public started : boolean ; public syncState : SyncState ; public isSynchronizing : boolean ; public gossipRelay : IGossipRelay ; public dnsPeerQuery : DnsPeerQuery ; public readonly peers : Set < GossipPeer > ; public readonly logger : ILogger ; constructor ( logger , readonly gossipStore , readonly pendingStore , readonly chainClient ? ) { super ( ) ; this . logger = logger . sub ( "gspmgr" ) ; this . peers = new Set < GossipPeer > ( ) ; this . syncState = SyncState . Unsynced ; this . gossipRelay = new GossipRelay ( logger . sub ( "gsprel" ) , 60000 , 2000 ) ; } /**
     * The number of peers managed by the PeerManager
     */ public get peerCount ( ) : number { return this . peers . size ; } /**
     * Starts the gossip manager. This method will load information
     * from the gossip store, determine when the last information
     * was obtained, validate the existing messages (to see if any
     * channels have closed), and finally emit all messages that
     * exist in the system.
     */ public async start ( ) { this . logger . info ( "starting gossip manager" ) ; if ( this . chainClient ) { this . logger . info ( "waiting for chain sync" ) ; await this . chainClient . waitForSync ( ) ; this . logger . info ( "chain sync complete" ) ; } await this . _restoreState ( ) ; for await ( const msg of this . allMessages ( ) ) { this . emit ( "message" , msg ) ; } this . gossipRelay . start ( ) ; this . started = true ; } /**
     * Adds a new peer to the GossipManager and subscribes to events that will
     * allow it to iteract with other sub-systems managed by the GossipManager.
     */ public addPeer ( peer ) { if ( ! this . started ) throw new WireError ( WireErrorCode . gossipManagerNotStarted ) ; peer . on ( "ready" , ( ) => this . _onPeerReady ( peer ) ) ; if ( peer . state === PeerState . Ready ) { this . _onPeerReady ( peer ) ; } } /**
     * Uses a dns seed to discover and add peers to be managed by the GossipManager.
     */ public async bootstrapPeers ( ls , localFeatures , localChains , logger , dnsSeed ) { if ( ! this . started ) throw new WireError ( WireErrorCode . gossipManagerNotStarted ) ; const peerRecords = await this . dnsPeerQuery . query ( { dnsSeed : dnsSeed } ) ; for ( const peerRecord of peerRecords ) { const peer = new Peer ( ls , localFeatures , localChains , logger ) ; this . addPeer ( peer ) ; peer . connect ( peerRecord . publicKey , peerRecord . address , peerRecord . port ) ; } } /**
     * Removes the channel from storage by the gossip manager. This
     * will likely be called by a chain-monitoring service.
     */ public async removeChannel ( scid ) { this . logger . debug ( "removing channel %s" , scid . toString ( ) ) ; await this . gossipStore . deleteChannelAnnouncement ( scid ) ; } param outpoint public async removeChannelByOutpoint ( outpoint ) { const chanAnn = await this . gossipStore . findChannelAnnouncementByOutpoint ( outpoint ) ; if ( ! chanAnn ) return ; await this . removeChannel ( chanAnn . shortChannelId ) ; } remarks public async * allMessages ( ) { this . logger . debug ( "fetching all messages" ) ; const seenNodeIds = new Set ( ) ; const chanAnns = await this . gossipStore . findChannelAnnouncemnts ( ) ; for ( const chanAnn of chanAnns ) { yield chanAnn ; const update1 = await this . gossipStore . findChannelUpdate ( chanAnn . shortChannelId , 0 ) ; if ( update1 ) yield update1 ; const update2 = await this . gossipStore . findChannelUpdate ( chanAnn . shortChannelId , 1 ) ; if ( update2 ) yield update2 ; const nodeId1 = chanAnn . nodeId1 . toString ( "hex" ) ; if ( ! seenNodeIds . has ( nodeId1 ) ) { seenNodeIds . add ( nodeId1 ) ; const nodeAnn = await this . gossipStore . findNodeAnnouncement ( chanAnn . nodeId1 ) ; if ( nodeAnn ) yield nodeAnn ; } const nodeId2 = chanAnn . nodeId2 . toString ( "hex" ) ; if ( ! seenNodeIds . has ( nodeId2 ) ) { seenNodeIds . add ( nodeId2 ) ; const nodeAnn = await this . gossipStore . findNodeAnnouncement ( chanAnn . nodeId2 ) ; if ( nodeAnn ) yield nodeAnn ; } } const nodeAnns = await this . gossipStore . findNodeAnnouncements ( ) ; for ( const nodeAnn of nodeAnns ) { if ( ! seenNodeIds . has ( nodeAnn . nodeId . toString ( "hex" ) ) ) yield nodeAnn ; } } param peer private _onPeerReady ( peer ) { const filter = new GossipFilter ( this . gossipStore , this . pendingStore , this . chainClient ) ; const gossipPeer = new GossipPeer ( peer , filter , this . logger ) ; gossipPeer . on ( "readable" , this . _onPeerReadable . bind ( this , gossipPeer ) ) ; gossipPeer . on ( "gossip_error" , this . _onGossipError . bind ( this ) ) ; this . peers . add ( gossipPeer ) ; peer . on ( "close" , this . _onPeerClose . bind ( this , gossipPeer ) ) ; this . gossipRelay . addPeer ( gossipPeer ) ; if ( this . syncState === SyncState . Unsynced ) { this . _syncPeer ( gossipPeer ) ; } else { gossipPeer . enableGossip ( ) ; } } param gossipPeer private _onPeerClose ( gossipPeer ) { if ( this . gossipRelay ) { this . gossipRelay . removePeer ( gossipPeer ) ; } this . peers . delete ( gossipPeer ) ; } param peer private _onPeerReadable ( peer ) { while ( true ) { const msg = peer . read ( ) as IWireMessage ; if ( msg ) this . _onGossipMessage ( msg ) ; else return ; } } param msg private _onGossipMessage ( msg ) { if ( msg . type === MessageType . ChannelAnnouncement ) { this . blockHeight = Math . max ( this . blockHeight , ( msg as ChannelAnnouncementMessage ) . shortChannelId . block ) ; } this . gossipRelay . enqueue ( msg ) ; this . emit ( "message" , msg ) ; } /**
     * Handles Gossip Errors
     */ private _onGossipError ( err ) { this . emit ( "error" , err ) ; } param peer private async _syncPeer ( peer ) { this . gossipRelay . stop ( ) ; this . logger . trace ( "sync status now 'syncing'" ) ; this . syncState = SyncState . Syncing ; try { await peer . syncRange ( ) ; this . logger . trace ( "sync status now 'synced'" ) ; this . syncState = SyncState . Synced ; this . logger . trace ( "enabling gossip for all peers" ) ; for ( const gossipPeer of this . peers ) { gossipPeer . enableGossip ( ) ; } } catch ( ex ) { this . syncState = SyncState . Unsynced ; } this . gossipRelay . start ( ) ; } private async _restoreState ( ) { this . logger . info ( "retrieving gossip state from store" ) ; this . blockHeight = 0 ; const chanAnns = await this . gossipStore . findChannelAnnouncemnts ( ) ; for ( const chanAnn of chanAnns ) { this . blockHeight = Math . max ( this . blockHeight , chanAnn . shortChannelId . block ) ; } this . logger . info ( "highest block %d found from %d channels" , this . blockHeight , chanAnns . length ) ; await this . _validateUtxos ( chanAnns ) ; } private async _validateUtxos ( chanAnns ) { if ( ! this . chainClient ) { this . logger . info ( "skipping utxo validation, no chain_client configured" ) ; return ; } const extChanAnnCount = chanAnns . reduce ( ( acc , msg ) => acc + ( msg instanceof ExtendedChannelAnnouncementMessage ? 1 : 0 ) , 0 ) ; this . logger . info ( "validating %d utxos" , extChanAnnCount ) ; if ( ! extChanAnnCount ) return ; const oct = Math . trunc ( extChanAnnCount / 8 ) ; for ( let i = 0 ; i < chanAnns . length ; i ++ ) { const chanAnn = chanAnns [ i ] ; if ( ( i + 1 ) % oct === 0 ) { this . logger . info ( "validating utxos %s% complete" , ( ( ( i + 1 ) / extChanAnnCount ) * 100 ) . toFixed ( 2 ) ) ; } if ( chanAnn instanceof ExtendedChannelAnnouncementMessage ) { const utxo = await this . chainClient . getUtxo ( chanAnn . outpoint . txid . toString ( HashByteOrder . RPC ) , chanAnn . outpoint . outputIndex ) ; if ( ! utxo ) { await this . removeChannel ( chanAnn . shortChannelId ) ; } } } } }