Add DVM package

packages/dvm/src/handler.ts

@@ -0,0 +1,120 @@
+import {hexToBytes} from '@noble/hashes/utils'
+import {getPublicKey, finalizeEvent} from 'nostr-tools'
+import {now} from '@welshman/lib'
+import type {TrustedEvent, EventTemplate, Filter} from '@welshman/util'
+import {subscribe, publish} from '@welshman/net'
+
+export type DVMHandler = {
+  stop?: () => void
+  handleEvent: (e: TrustedEvent) => AsyncGenerator<EventTemplate>
+}
+
+export type CreateDVMHandler = (dvm: DVM) => DVMHandler
+
+export type DVMOpts = {
+  sk: string
+  relays: string[]
+  handlers: Record<string, CreateDVMHandler>
+  expireAfter?: number
+  requireMention?: boolean
+}
+
+export class DVM {
+  active = false
+  logEvents = false
+  seen = new Set()
+  handlers = new Map()
+
+  constructor(readonly opts: DVMOpts) {
+    for (const [kind, createHandler] of Object.entries(this.opts.handlers)) {
+      this.handlers.set(parseInt(kind), createHandler(this))
+    }
+  }
+
+  async start() {
+    this.active = true
+
+    const {sk, relays, requireMention = false} = this.opts
+
+    while (this.active) {
+      await new Promise<void>(resolve => {
+        const since = now()
+        const kinds = Array.from(this.handlers.keys())
+        const filter: Filter = {kinds, since}
+
+        if (requireMention) {
+          filter['#p'] = [getPublicKey(hexToBytes(sk))]
+        }
+
+        const filters = [filter]
+        const sub = subscribe({relays, filters})
+
+        sub.emitter.on('event', (url: string, e: TrustedEvent) => this.onEvent(e))
+        sub.emitter.on('complete', () => resolve())
+      })
+    }
+  }
+
+  stop() {
+    for (const handler of this.handlers.values()) {
+      handler.stop?.()
+    }
+
+    this.active = false
+  }
+
+  async onEvent(request: TrustedEvent) {
+    console.log(request)
+    const {expireAfter = 60 * 60} = this.opts
+
+    if (this.seen.has(request.id)) {
+      return
+    }
+
+    const handler = this.handlers.get(request.kind)
+
+    if (!handler) {
+      return
+    }
+
+    this.seen.add(request.id)
+
+    if (this.logEvents) {
+      console.info('Handling request', request)
+    }
+
+    for await (const event of handler.handleEvent(request)) {
+      if (event.kind !== 7000) {
+        event.tags.push(['request', JSON.stringify(request)])
+
+        const inputTag = request.tags.find(t => t[0] === 'i')
+
+        if (inputTag) {
+          event.tags.push(inputTag)
+        }
+      }
+
+      event.tags.push(['p', request.pubkey])
+      event.tags.push(['e', request.id])
+
+      if (expireAfter) {
+        event.tags.push(['expiration', String(now() + expireAfter)])
+      }
+
+      if (this.logEvents) {
+        console.info('Publishing event', event)
+      }
+
+      this.publish(event)
+    }
+  }
+
+  async publish(template: EventTemplate) {
+    const {sk, relays} = this.opts
+    const event = finalizeEvent(template, hexToBytes(sk))
+
+    await new Promise<void>(resolve => {
+      publish({event, relays}).emitter.on('success', () => resolve())
+    })
+  }
+}

packages/dvm/src/index.ts

@@ -0,0 +1,2 @@
+export * from './handler'
+export * from './request'

packages/dvm/src/request.ts

@@ -0,0 +1,47 @@
+import {Emitter, now} from '@welshman/lib'
+import type {TrustedEvent, SignedEvent} from '@welshman/util'
+import {subscribe, publish} from '@welshman/net'
+import type {Subscription, Publish} from '@welshman/net'
+
+export enum DVMEvent {
+  Progress = "progress",
+  Result = "result",
+}
+
+export type DVMRequestOptions = {
+  event: SignedEvent
+  relays: string[]
+  timeout?: number
+  autoClose?: boolean
+  reportProgress?: boolean
+}
+
+export type DVMRequest = DVMRequestOptions & {
+  emitter: Emitter,
+  sub: Subscription
+  pub: Publish
+}
+
+export const makeDvmRequest = (request: DVMRequest) => {
+  const emitter = new Emitter()
+  const {event, relays, timeout = 30_000, autoClose = true, reportProgress = true} = request
+  const kind = event.kind + 1000
+  const kinds = reportProgress ? [kind, 7000] : [kind]
+  const filters = [{kinds, since: now() - 60, "#e": [event.id]}]
+  const sub = subscribe({relays, timeout, filters})
+  const pub = publish({event, relays, timeout})
+
+  sub.emitter.on('event', (url: string, event: TrustedEvent) => {
+    if (event.kind === 7000) {
+      emitter.emit(DVMEvent.Progress, url, event)
+    } else {
+      emitter.emit(DVMEvent.Result, url, event)
+
+      if (autoClose) {
+        sub.close()
+      }
+    }
+  })
+
+  return {request, emitter, sub, pub}
+}