const{getEncoding:e}=require("./spec/hyperschema"),s=require("hypercore-crypto"),n=require("hyperdht"),t=require("ready-resource"),i=require("b4a"),o=require("protomux"),r=require("compact-encoding"),c=require("hyperdb"),[u]=s.namespace("hyperdiscovery",1),a=e("@hyperdiscovery/announce"),h=e("@hyperdiscovery/announce-to-swarm"),d=e("@hyperdiscovery/subscribe-to-swarm"),l=e("@hyperdiscovery/unsubscribe-to-swarm"),y=e("@hyperdiscovery/bump-from-swarm");function p(e){if(o.isProtomux(e))return e;if(e.noiseStream.userData)return e.noiseStream.userData;const s=o.from(e.noiseStream);return e.noiseStream.userData=s,s}function m(e,s){s.userData._onbump(e)}function b(){throw new Error("Method not supported")}function _(){}module.exports={HyperDiscovery:class extends t{constructor(e,{bootstrap:s,dht:t=new n({bootstrap:s})}={}){super(),this.dht=t,this.db=c.rocks(e,require("./spec/hyperdb")),this.subs=new Map,this._flushTimeout=null,this._flushing=null,this._server=null,this._manifest=null,this.stats={streamsAdded:0,streamsEnded:0,subscriptionsAdded:0,subscriptionsRemoved:0,announces:0,announcesSent:0,onsubscribeCount:0,onunsubscribeCount:0,onannounceCount:0},this.ready().catch(_)}get publicKey(){return this._server&&this._server.address().publicKey}async _open(){await this._init(),this._flushBackground().catch(_),this._server=this.dht.createServer(),this._server.on("connection",this.addStream.bind(this)),await this._server.listen(s.keyPair(this._manifest.seed))}async _init(){this._manifest=await this.db.get("@hyperdiscovery/manifest"),this._manifest||(this._manifest={version:0,seed:s.randomBytes(32)},await this.db.insert("@hyperdiscovery/manifest",this._manifest),await this.db.flush())}async _close(){this._flushTimeout&&clearTimeout(this._flushTimeout),this._flushing&&await this._flushing,this._flushTimeout=null,this._server&&await this._server.close(),await this.db.close()}async _flushBackground(){for(;!this.closing;){if(await new Promise(e=>{this._flushTimeout=setTimeout(e,12e4)}),this._flushTimeout=null,this.closing)return;this._flushing=this.db.flush(),await this._flushing,this._flushing=null}}_addSub(e,s,n){let t=this.subs.get(s);t||(t=new Map,this.subs.set(s,t)),this.stats.subscriptionsAdded++,t.set(e,n)}_removeSub(e,s){const n=this.subs.get(s);n&&(this.stats.subscriptionsRemoved++,n.delete(e),n.size||this.subs.delete(s))}addStream(e){this.stats.streamsAdded++;const s=p(e),n=this,t=new Set;function o(){const e=s.createChannel({protocol:"hyperdiscovery",messages:[{encoding:d,onmessage:async function(s){n.stats.onsubscribeCount++;const o=i.toString(s.publicKey,"hex");t.add(o),n._addSub(e,o,s.since);const r=await n.lookup(s.publicKey);r&&r.bumped>=s.since&&e.messages[3].send(r)}},{encoding:l,onmessage:function(s){n.stats.onunsubscribeCount++;const o=i.toString(s.publicKey,"hex");t.delete(o),n._removeSub(e,o)}},{encoding:h,onmessage:function(s){return n.stats.onannounceCount++,n.announce(s,e)}},{encoding:y,onmessage:b}],onclose(){n.stats.streamsEnded++;for(const s of t)n._removeSub(e,s)}});e&&e.open()}s.pair({protocol:"hyperdiscovery"},o),o()}async lookup(e){return await this.db.get("@hyperdiscovery/swarms",{publicKey:e})}async announce(e,n){const t={buffer:null,start:0,end:0};if(r.fixed32.preencode(t,u),a.preencode(t,e.announce),t.buffer=i.allocUnsafe(t.end),r.fixed32.encode(t,u),a.encode(t,e.announce),Date.now()+6e4>e.announce.bumped)return!1;if(!s.verify(t.buffer,e.signature,e.publicKey))return!1;const o=await this.db.get("@hyperdiscovery/swarms",{publicKey:e.publicKey});if(o&&o.bumped>=e.announce.bump)return!1;this.stats.announces++;const c={publicKey:e.publicKey,bumped:e.announce.bump,updated:Date.now(),signature:e.signature};await this.db.insert("@hyperdiscovery/swarms",c);const h=this.subs.get(i.toString(e.publicKey,"hex"));if(h)for(const[e,s]of h)e!==n&&(c.bumped<s||(this.stats.announcesSent++,e.messages[3].send(c)));return!0}registerMetrics(e){const s=this;new e.Gauge({name:"hyperdiscovery_streams_added",help:"Hyperdiscovery streams added",collect(){this.set(s.stats.streamsAdded)}}),new e.Gauge({name:"hyperdiscovery_streams_ended",help:"Hyperdiscovery streams ended",collect(){this.set(s.stats.streamsEnded)}}),new e.Gauge({name:"hyperdiscovery_subscriptions_added",help:"Hyperdiscovery subscriptions added",collect(){this.set(s.stats.subscriptionsAdded)}}),new e.Gauge({name:"hyperdiscovery_subscriptions_removed",help:"Hyperdiscovery subscriptions removed",collect(){this.set(s.stats.subscriptionsRemoved)}}),new e.Gauge({name:"hyperdiscovery_announces",help:"Hyperdiscovery announces",collect(){this.set(s.stats.announces)}}),new e.Gauge({name:"hyperdiscovery_announces_sent",help:"Hyperdiscovery announces sent",collect(){this.set(s.stats.announcesSent)}}),new e.Gauge({name:"hyperdiscovery_onsubscribe_count",help:"Hyperdiscovery onsubscribe count",collect(){this.set(s.stats.onsubscribeCount)}}),new e.Gauge({name:"hyperdiscovery_onunsubscribe_count",help:"Hyperdiscovery onunsubscribe count",collect(){this.set(s.stats.onunsubscribeCount)}}),new e.Gauge({name:"hyperdiscovery_onannounce_count",help:"Hyperdiscovery onannounce count",collect(){this.set(s.stats.onannounceCount)}})}},HyperDiscoveryClient:class extends t{constructor(e,{bootstrap:s,dht:t=new n({bootstrap:s})}={}){super(),this.remotePublicKey=e,this.dht=t,this.connection=null,this.muxer=null,this.channel=null,this.suspended=!1,this.connecting=!1,this._retryTimeout=null,this._retryResolve=null,this._shouldReconnect=!1}async suspend(){this.suspended=!0,this._clearRetry(),this.connection&&(this.connection.destroy(),this.connecting=null,this.channel=null,this.muxer=null,this.connection.destroyed||await new Promise(e=>this.connection.on("close",e)))}resume(){this.suspended=!1,this._shouldReconnect&&!this.closing&&this._reconnect()}connect(){this.channel||this._reconnect()}async _reconnect(){if(this.connecting)return this.channel;this.connecting=!0,this._shouldReconnect=!0;let e=0;for(;!this.closing&&!this.suspended;){const s=this.connection=this.dht.connect(this.remotePublicKey);if(this.muxer=p(this.connection),this.channel=this.muxer.createChannel({userData:this,protocol:"hyperdiscovery",messages:[{encoding:d,onmessage:b},{encoding:l,onmessage:b},{encoding:h,onmessage:b},{encoding:y,onmessage:m}]}),!this.channel)break;if(this.channel.open(),await new Promise(e=>{function n(){s.off("connect",n),s.off("error",n),e()}s.on("connect",n),s.on("error",n)}),this.closing||this.suspended){s.on("error",_),s.destroy();break}if(!s.destroyed&&!s.destroying){s.on("close",this._reconnect.bind(this));break}e++,await new Promise(s=>{this._retryResolve=s,this._retryTimeout=setTimeout(s,e<3?5e3:15e3)}),this._retryResolve=null,this._retryTimeout=null}return this.connecting=!1,this.closing||this.suspended||this.emit("connect"),this.channel}_close(){this.connection&&this.connection.destroy(),this._clearRetry()}_clearRetry(){this._retryTimeout&&(clearTimeout(this._retryTimeout),this._retryResolve(),this._retryResolve=null,this._retryTimeout=null)}_onbump(e){this.emit("announce",e)}_getChannel(){return this.channel||this._reconnect(),this.channel}cork(){const e=this._getChannel();e&&e.cork()}uncork(){const e=this._getChannel();e&&e.uncork()}subscribe(e,{since:s=0}={}){const n=this._getChannel();n&&n.messages[0].send({publicKey:e,since:s})}unsubscribe(e){const s=this._getChannel();s&&s.messages[1].send({publicKey:e})}async announce(e,{bump:n=Date.now()}={}){const t=this._getChannel();if(!t)return;const o={bump:n},c={buffer:null,start:0,end:0};r.fixed32.preencode(c,u),a.preencode(c,o),c.buffer=i.allocUnsafe(c.end),r.fixed32.encode(c,u),a.encode(c,o);const h=s.sign(c.buffer,e.secretKey),d={publicKey:e.publicKey,announce:o,signature:h};t.messages[2].send(d)}}};