Merge pull request #666 from sandwichfarm/improve-daemons

hotfixes

.github/workflows/publish-package.yml

@@ -12,8 +12,19 @@ jobs:
     strategy:
       matrix:
         include:
+          - package: 'logger'
+            nodeVersion: '20'
           - package: 'nocap'
-            nodeVersion: '18'
+            nodeVersion: '20'
+          - package: 'seed'
+            nodeVersion: '20'
+          - package: 'utils'
+            nodeVersion: '20'
+          - package: 'nwcache'
+            nodeVersion: '20'
+          - package: 'controlflow'
+            nodeVersion: '20'
+
     steps:
       - uses: actions/checkout@v3
 
@@ -24,7 +35,7 @@ jobs:
 
       - name: Change to package directory
         run: cd packages/${{ matrix.package }}
-
+ 
       - run: yarn install
       - run: yarn test
 
@@ -45,4 +56,4 @@ jobs:
           release_name: v${{ matrix.package }}-${{ steps.publish.outputs.version }}
           body: ""
           draft: false
-          prerelease: true
+          prerelease: true

.gitignore

@@ -16,4 +16,4 @@ ours.sh
 .pg
 packages/synx
 packages/kinds
-.npmrc
+.npmrc

packages/nocapd/.gitignore → apps/nocapd/.gitignore

@@ -1,3 +1,4 @@
 node_modules
 config.y*ml
-yarn.lock
+yarn.lock
+.events

packages/nocapd/package.json → apps/nocapd/package.json

@@ -8,7 +8,6 @@
     "@nostr-fetch/adapter-nostr-tools": "0.14.1",
     "chalk": "5.3.0",
     "nostr-fetch": "0.14.1",
-    "nostr-geotags": "0.0.7",
     "object-mapper": "6.2.0"
   },
   "scripts": {

packages/nocapd/src/classes/NocapdQueues.js → apps/nocapd/src/classes/NocapdQueues.js

@@ -48,12 +48,6 @@ export class NocapdQueues {
     else if (typeof job === 'string')
       name = job.split(':')[0]
 
-    // if(event === 'drained')
-    //   console.log(`${event}!`, name, job?.id)
-
-    // if(event === 'completed')
-    //   console.log(`${event}!`, name, job?.id)
-
     if(name) {
       const daemonManager = name.split('@')[0]
       const daemonPubkey = name.split('@')[1]
@@ -84,9 +78,7 @@ export class NocapdQueues {
   }
 
   bind_events(){
-    // if(!this.bindEvents) return
     this.worker_events.forEach(handler => {
-      // console.log(`bind on_${handler} event handler on ${this.worker.name}:${this.constructor.name}`)
       this.worker.on(handler, (...args) => this.route_event(handler, ...args))
     })
   }

packages/nocapd/src/classes/WorkerManager.js → apps/nocapd/src/classes/WorkerManager.js

@@ -274,7 +274,7 @@ export class WorkerManager {
 
   async isExpired(url, lastChecked) {
       const retries = await this.retry.getRetries(url);
-      const expiry = retries > 0 ? this.retry.expiry(retries) : this.expires;
+      const expiry = retries > 0 ? await this.retry.getExpiry(url) : this.expires;
       return lastChecked < Date.now() - expiry;
   }
 

packages/nocapd/src/daemon.js → apps/nocapd/src/daemon.js

@@ -109,9 +109,6 @@ export const Nocapd = async () => {
   await maybeBootstrap()
   const $q = await initWorker()
   return {
-    // stop: () => {
-    //   log.warn('Stopping nocapd')
-    // },
     $q
   }
 }

packages/nocapd/src/managers/AllManager.js → apps/nocapd/src/managers/AllManager.js

@@ -1,18 +1,18 @@
 import mapper from 'object-mapper'
 import ngeotags from 'nostr-geotags'
-
-
 import Publish from '@nostrwatch/publisher'
 
 import { WorkerManager } from '../classes/WorkerManager.js'
 
 const publish30066 = new Publish.Kind30066()
 
+
+
 export class AllManager extends WorkerManager {
   constructor($, rcache, opts){
     super($, rcache, opts)
     this.interval = 60*1000       //checks for expired items every...
-    this.timeout = 6*1000
+    this.timeout = 9*1000
     this.timeoutBuffer = 1000
   }
 
@@ -23,7 +23,6 @@ export class AllManager extends WorkerManager {
     try {
       this.log.debug(`Running comprehensive check for ${job.data.relay}`)
       const { relay:url } = job.data 
-      // console.log(await this.getLastChecked(url), Date.now(),  (Date.now()-await this.getLastChecked(url))/1000/60)
       const dpubkey = this.pubkey
       const nocapOpts = { 
         timeout: { 
@@ -56,7 +55,7 @@ export class AllManager extends WorkerManager {
     const { relay:url } = job.data
     const { result  } = rvalue
 
-    if(!result)
+    if(!result || !result?.connect?.data)
       return this.on_failed(job, new Error(`Nocap complete (all) check failed for ${url}`))
 
     const { checked_at } = result
@@ -68,22 +67,18 @@ export class AllManager extends WorkerManager {
     const retry_id = await this.retry.setRetries( url, true )
     const lastChecked_id = await this.setLastChecked( url, Date.now() )
 
-    // console.log('success', await this.rcache.$.get(relay_id)?.online, await this.rcache.$.get(retry_id)?.v, await this.rcache.$.get(lastChecked_id)?.v)
-
     const event30066Data = event30066DataFromResult( result )
-    await publish30066.one(event30066Data)
+    await publish30066.one(event30066Data)    
   }
 
   async on_failed(job, err){
     const { relay:url } = job.data
-    // console.log('url:onfailed', url)
     this.log?.debug(`Websocket check failed for ${job.data.relay}: ${JSON.stringify(err)}`)
-    const retry_id = await this.retry.setRetries(url, false)
+    const retry_id = await this.retry.setRetries( url, false )
     const lastChecked_id = await this.setLastChecked( url, Date.now() )
-    // console.log('failed', await this.rcache.$.get(retry_id)?.v, await this.rcache.$.get(lastChecked_id)?.v)
+    const relay_id = await this.updateRelayCache({ url, connect: { data: false }} ) 
     this.progressMessage(url, null, true)
     this.processed++
-
   }
 }
 
@@ -106,7 +101,11 @@ const truncatedResult = (result, type) => {
 
 const event30066DataFromResult = result => {
   const eventData = {}
-  const attributes = []
+  const labels = []
+  const nips = []
+
+  const dns = result.dns?.data || {}
+  const isDns = Object.keys(dns)?.length > 0  
 
   const geo = transformGeoResult(result.geo?.data) || {}
   const isGeo = Object.keys(geo)?.length > 0
@@ -122,48 +121,73 @@ const event30066DataFromResult = result => {
 
   eventData.rtt = []
 
+  if(result?.network)
+    eventData.network = result.network
+
   if(result?.connect?.duration > 0)
-    eventData.rtt.push({ type: 'open', rtt: result.connect.duration })
+    eventData.rtt.push([ 'open', result.connect.duration ])
 
   if(result?.read?.duration > 0)
-    eventData.rtt.push({ type: 'subscribe', rtt: result.read.duration })
+    eventData.rtt.push([ 'subscribe', result.read.duration ])
 
   if(result?.write?.duration > 0)
-    eventData.rtt.push({ type: 'publish', rtt: result.write.duration })
+    eventData.rtt.push([ 'publish', result.write.duration ])
 
   if(eventData.retries > 0)
     eventData.retries = result.retries
 
   if(isGeo)
-    eventData.geo = ngeotags(geo, { iso31662: true })
+    eventData.geo = ngeotags(geo, { iso31662: true, iso3163: true })
 
   if(isInfo){
-    if(info?.limitations?.payment_required === true)
-      attributes.push('payment-required')
-    if(info?.limitations?.auth_required === true)
-      attributes.push('auth-required')
+    if(info?.limitation?.payment_required === true)
+      labels.push(['nip11.limitation', 'payment-required'])
+    if(info?.limitation?.auth_required === true)
+      labels.push(['nip11.limitation', 'auth-required'])
+    if(info?.pubkey)
+      labels.push(['nip11.pubkey', info.pubkey])
+    if(info?.contact)
+      labels.push(['nip11.contact', info.contact])
+    if(info?.name)
+      labels.push(['nip11.name', info.name])
+    if(info?.software)
+      labels.push(['nip11.software', info.software])
+    if(info?.version)
+      labels.push(['nip11.version', info.version])
     if(info?.supported_nips instanceof Array)
-      info.supported_nips.forEach(nip => attributes.push(`nip-${nip}`))
+      info.supported_nips.forEach(nip => {
+        nips.push(`${nip}`)
+      })
+    if(info?.tags)
+      labels.push(['nip11.tags', ...info.tags])
+    if(info?.language_tags)
+      labels.push(['nip11.language_tags', ...info.language_tags ])
   }
 
   if(isSsl)
-    attributes.push(ssl?.valid === true? 'ssl-valid' :'ssl-invalid') 
+    eventData.ssltag = [ 'ssl', ssl?.valid === true? 'valid': 'invalid', `${new Date(ssl.valid_from).getTime()}`, `${new Date(ssl.valid_to).getTime()}` ]
 
   if(isGeo)
     if(geo?.as)
-      attributes.push(geo.as)
+      labels.push(['as', geo.as])
+    if(geo?.asn)
+      labels.push(['asn', geo.asn])
     if(geo?.ip)
-      attributes.push(geo.ip)
+      labels.push([`ipv4`, geo.ip])
+
+  if(labels.length)
+    eventData.labels = labels
 
-  if(attributes.length)
-    eventData.attributes = attributes
+  if(nips.length)
+    eventData.nips = nips
 
   return eventData
 }
 
 const transformGeoResult = geo => {  
   const map = {
     "as": "as",
+    "asn": "asn",
     "city": "cityName",
     "countryCode": "countryCode",
     "regionName": "regionName",

packages/trawler/src/daemon.js → apps/trawler/src/daemon.js

@@ -49,6 +49,7 @@ export default async () => {
     const relays = seed[0]
     const updatedAt = seed[1]
     await populateTrawler( relays )
+    trawlWorker.on('drained', () => populateTrawler( relays ) )
     resolve({ queues: { trawlQueue }, watcher: null })
   })
 }

packages/trawler/src/queue.js → apps/trawler/src/queue.js

@@ -33,11 +33,13 @@ export const configureQueues = async function(){
     }
   }
 
-  const trawlQueueDrained = () => {}
+  const trawlQueueDrained = () => {
+
+  }
 
   trawler.$Queue.drain()
 
-  trawler.$Queue.on('drained', trawlQueueDrained)
+  // trawler.$Queue.on('drained', trawlQueueDrained)
   // trawler.$QueueEvents.on('progress', trawlJobProgress)
 
   const trawlJobCompleted = async ($job, foundRelays) => {

packages/controlflow/src/retry.js

@@ -40,6 +40,10 @@ export class RetryManager {
     return await rcache.retry.get(this.cacheId(url))
   }
 
+  async getExpiry( url ){
+    return this.expiry( await this.getRetries(url) )
+  }
+
   async setRetries( url, success ){
     let id 
     if(success) {

packages/nocap/adapters/default/GeoAdapterDefault/index.js

@@ -19,7 +19,7 @@ import { fetch } from 'cross-fetch'
     else 
       endpoint = `http://ip-api.com/json/${ip}?fields=${fields}`
     const headers = { 'accept': 'application/json' }
-    const response = await fetch(endpoint, { headers }).catch(e => err=e)
+    const response = await fetch(endpoint, { headers }).catch(this.$.logger.warn)
     delete response.query
     delete response.status
     const result = { status: "success", data: await response.json() }

packages/nocap/adapters/default/SslAdapterDefault/index.js

@@ -21,6 +21,7 @@ class SslAdapterDefault {
       result = { status: "success", data }
     this.$.finish('ssl', result)
   }
+
   sslCheckerOptions(port){
     return { method: "GET", port: port || 443 } 
   }