import { BufferReader } from "@node-lightning/bufio" ; import { ShortChannelId } from "@node-lightning/core" ; import { shortChannelIdFromBuffer } from "@node-lightning/core" ; import { OutPoint } from "@node-lightning/core" ; import { IGossipStore } from "@node-lightning/wire" ; import { NodeAnnouncementMessage } from "@node-lightning/wire" ; import { ChannelAnnouncementMessage } from "@node-lightning/wire" ; import { ExtendedChannelAnnouncementMessage } from "@node-lightning/wire" ; import { ChannelUpdateMessage } from "@node-lightning/wire" ; import { RocksdbBase } from "./rocksdb-base" ; enum Prefix { ChannelAnnouncement = 1 , ChannelUpdate = 2 , NodeAnnouncement = 3 , ChannelsForNode = 4 , Outpoint = 5 } export class RocksdbGossipStore extends RocksdbBase implements IGossipStore { public async findChannelAnnouncemnts ( ) { return new Promise ( ( resolve , reject ) => { const stream = this . _db . createReadStream ( ) ; const results = [ ] ; stream . on ( "data" , data => { if ( data . key [ 0 ] === Prefix . ChannelAnnouncement ) { results . push ( ExtendedChannelAnnouncementMessage . deserialize ( data . value ) ) ; } } ) ; stream . on ( "end" , ( ) => { resolve ( results ) ; } ) ; stream . on ( "error" , err => reject ( err ) ) ; } ) ; } public async findNodeAnnouncements ( ) { return new Promise ( ( resolve , reject ) => { const stream = this . _db . createReadStream ( ) ; const results = [ ] ; stream . on ( "data" , data => { if ( data . key [ 0 ] === Prefix . NodeAnnouncement ) { results . push ( NodeAnnouncementMessage . deserialize ( data . value ) ) ; } } ) ; stream . on ( "end" , ( ) => { resolve ( results ) ; } ) ; stream . on ( "error" , err => reject ( err ) ) ; } ) ; } public async findChannelsForNode ( nodeId ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelsForNode ] ) , nodeId ] ) ; const raw = await this . _safeGet < Buffer > ( key ) ; if ( ! raw ) return [ ] ; const reader = new BufferReader ( raw ) ; const results = [ ] ; while ( ! reader . eof ) { results . push ( shortChannelIdFromBuffer ( reader . readBytes ( 8 ) ) ) ; } return results ; } public async findNodeAnnouncement ( nodeId ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . NodeAnnouncement ] ) , nodeId ] ) ; const raw = await this . _safeGet < Buffer > ( key ) ; if ( ! raw ) return ; return NodeAnnouncementMessage . deserialize ( raw ) ; } public async findChannelAnnouncement ( scid ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelAnnouncement ] ) , scid . toBuffer ( ) ] ) ; const raw = await this . _safeGet < Buffer > ( key ) ; if ( ! raw ) return ; return ExtendedChannelAnnouncementMessage . deserialize ( raw ) ; } public async findChannelAnnouncementByOutpoint ( outpoint ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . Outpoint ] ) , Buffer . from ( outpoint . toString ( ) ) , ] ) ; const scidBuf = await this . _safeGet < Buffer > ( key ) ; if ( ! scidBuf ) return ; const key2 = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelAnnouncement ] ) , scidBuf ] ) ; const raw = await this . _safeGet < Buffer > ( key2 ) ; if ( ! raw ) return ; return ExtendedChannelAnnouncementMessage . deserialize ( raw ) ; } public async findChannelUpdate ( scid , dir ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelUpdate ] ) , scid . toBuffer ( ) , Buffer . from ( [ dir ] ) , ] ) ; const raw = await this . _safeGet < Buffer > ( key ) ; if ( ! raw ) return ; return ChannelUpdateMessage . deserialize ( raw ) ; } public async saveChannelAnnouncement ( msg ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelAnnouncement ] ) , msg . shortChannelId . toBuffer ( ) , ] ) ; const value = msg . serialize ( ) ; await this . _db . put ( key , value ) ; if ( msg instanceof ExtendedChannelAnnouncementMessage ) { const key2 = Buffer . concat ( [ Buffer . from ( [ Prefix . Outpoint ] ) , Buffer . from ( msg . outpoint . toString ( ) ) , ] ) ; await this . _db . put ( key2 , msg . shortChannelId . toBuffer ( ) ) ; } await this . _saveNodeChannel ( msg . shortChannelId , msg . nodeId1 ) ; await this . _saveNodeChannel ( msg . shortChannelId , msg . nodeId2 ) ; } public async saveChannelUpdate ( msg ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelUpdate ] ) , msg . shortChannelId . toBuffer ( ) , Buffer . from ( [ msg . direction ] ) , ] ) ; const value = msg . serialize ( ) ; await this . _db . put ( key , value ) ; } public async saveNodeAnnouncement ( msg ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . NodeAnnouncement ] ) , msg . nodeId ] ) ; const value = msg . serialize ( ) ; await this . _db . put ( key , value ) ; } public async deleteChannelAnnouncement ( scid ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelAnnouncement ] ) , scid . toBuffer ( ) ] ) ; const msg = await this . findChannelAnnouncement ( scid ) ; if ( ! msg ) return ; await this . deleteChannelUpdate ( scid , 0 ) ; await this . deleteChannelUpdate ( scid , 1 ) ; await this . _deleteNodeChannel ( scid , msg . nodeId1 ) ; await this . _deleteNodeChannel ( scid , msg . nodeId2 ) ; await this . _db . del ( key ) ; } public async deleteChannelUpdate ( scid , dir ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelUpdate ] ) , scid . toBuffer ( ) , Buffer . from ( [ dir ] ) , ] ) ; await this . _db . del ( key ) ; } public async deleteNodeAnnouncement ( nodeId ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . NodeAnnouncement ] ) , nodeId ] ) ; await this . _deleteNodeChannels ( nodeId ) ; await this . _db . del ( key ) ; } private async _deleteNodeChannel ( scid , nodeId ) { const scids = await this . findChannelsForNode ( nodeId ) ; if ( ! scids . length ) return ; const scidKey = scid . toString ( ) ; const index = scids . findIndex ( p => p . toString ( ) === scidKey ) ; if ( index === - 1 ) return ; scids . splice ( index , 1 ) ; await this . _putNodeChannels ( nodeId , scids ) ; } private async _deleteNodeChannels ( nodeId ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelsForNode ] ) , nodeId ] ) ; await this . _db . del ( key ) ; } private async _saveNodeChannel ( scid , nodeId ) { const scids = await this . findChannelsForNode ( nodeId ) ; const scidKey = scid . toString ( ) ; for ( const s of scids ) { if ( s . toString ( ) === scidKey ) return ; } scids . push ( scid ) ; await this . _putNodeChannels ( nodeId , scids ) ; } private async _putNodeChannels ( nodeId , scids ) { const key = Buffer . concat ( [ Buffer . from ( [ Prefix . ChannelsForNode ] ) , nodeId ] ) ; const value = Buffer . concat ( scids . map ( p => p . toBuffer ( ) ) ) ; await this . _db . put ( key , value ) ; } }