feat(dvm): trap NIP-90 job request events and record them via job repository (#729)

* feat(dvm): trap NIP-90 job request events and record them via job repository

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

* test(dvm): add tests for dvm job request ingestion

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

---------

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

Author: Priyanshubhartistm

.changeset/dvm-job-ingestion.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(dvm): trap NIP-90 job request events (kind 5000-5999) and record them via the job repository

.knip.json

@@ -16,7 +16,6 @@
   ],
   "ignore": [
     ".nostr/**",
-    "src/repositories/dvm-job-repository.ts",
     "src/utils/relay-probe/**"
   ],
   "commitlint": false,

src/constants/base.ts

@@ -41,6 +41,9 @@ export enum EventKinds {
   // Lightning zaps
   ZAP_REQUEST = 9734,
   ZAP_RECEIPT = 9735,
+  // NIP-90: Data Vending Machines — job request events
+  DVM_JOB_REQUEST_FIRST = 5000,
+  DVM_JOB_REQUEST_LAST = 5999,
   // Replaceable events
   REPLACEABLE_FIRST = 10000,
   // NIP-65: Relay List Metadata

src/factories/event-strategy-factory.ts

@@ -1,7 +1,8 @@
 import { ICacheAdapter, IWebSocketAdapter } from '../@types/adapters'
-import { IEventRepository, IInviteCodeRepository, IUserRepository } from '../@types/repositories'
+import { IDvmJobRepository, IEventRepository, IInviteCodeRepository, IUserRepository } from '../@types/repositories'
 import {
   isDeleteEvent,
+  isDvmJobRequestEvent,
   isEphemeralEvent,
   isGiftWrapEvent,
   isMarmotGroupEvent,
@@ -14,6 +15,7 @@ import { isNip43JoinRequest, isNip43LeaveRequest } from '../utils/nip43'
 import { isRelayListEvent } from '../utils/nip65'
 import { DefaultEventStrategy } from '../handlers/event-strategies/default-event-strategy'
 import { DeleteEventStrategy } from '../handlers/event-strategies/delete-event-strategy'
+import { DvmJobRequestEventStrategy } from '../handlers/event-strategies/dvm-job-request-event-strategy'
 import { EphemeralEventStrategy } from '../handlers/event-strategies/ephemeral-event-strategy'
 import { Event } from '../@types/event'
 import { Factory } from '../@types/base'
@@ -33,6 +35,7 @@ export const eventStrategyFactory =
     eventRepository: IEventRepository,
     userRepository: IUserRepository,
     inviteCodeRepository: IInviteCodeRepository,
+    dvmJobRepository: IDvmJobRepository,
     cache: ICacheAdapter,
     settings: () => Settings,
   ): Factory<IEventStrategy<Event, Promise<void>>, [Event, IWebSocketAdapter]> =>
@@ -47,12 +50,17 @@ export const eventStrategyFactory =
       return new TimestampEventStrategy(adapter, eventRepository)
     } else if (isRelayListEvent(event) || isReplaceableEvent(event)) {
       return new ReplaceableEventStrategy(adapter, eventRepository)
-    // NIP-43: Join/Leave requests MUST be checked before the generic ephemeral
-    // handler, because kinds 28934/28936 fall in the ephemeral range (20000-29999).
+      // NIP-43: Join/Leave requests MUST be checked before the generic ephemeral
+      // handler, because kinds 28934/28936 fall in the ephemeral range (20000-29999).
     } else if (isNip43JoinRequest(event)) {
       return new JoinRequestEventStrategy(adapter, inviteCodeRepository, userRepository, cache, settings)
     } else if (isNip43LeaveRequest(event)) {
       return new LeaveRequestEventStrategy(adapter, userRepository, cache, settings)
+      // NIP-90: DVM job requests (kind 5000-5999) checked early, same reasoning
+      // as the NIP-43 checks above — kept explicit rather than relying on it
+      // falling through to DefaultEventStrategy.
+    } else if (isDvmJobRequestEvent(event)) {
+      return new DvmJobRequestEventStrategy(adapter, eventRepository, dvmJobRepository)
     } else if (isEphemeralEvent(event)) {
       return new EphemeralEventStrategy(adapter)
     } else if (isDeleteEvent(event)) {
@@ -62,4 +70,4 @@ export const eventStrategyFactory =
     }
 
     return new DefaultEventStrategy(adapter, eventRepository)
-  }
+  }

src/factories/message-handler-factory.ts

@@ -1,5 +1,11 @@
 import { ICacheAdapter, IWebSocketAdapter } from '../@types/adapters'
-import { IEventRepository, IInviteCodeRepository, INip05VerificationRepository, IUserRepository } from '../@types/repositories'
+import {
+  IDvmJobRepository,
+  IEventRepository,
+  IInviteCodeRepository,
+  INip05VerificationRepository,
+  IUserRepository,
+} from '../@types/repositories'
 import { IncomingMessage, MessageType } from '../@types/messages'
 import { createSettings } from './settings-factory'
 import { AuthMessageHandler } from '../handlers/auth-message-handler'
@@ -26,13 +32,21 @@ export const messageHandlerFactory =
     userRepository: IUserRepository,
     nip05VerificationRepository: INip05VerificationRepository,
     inviteCodeRepository: IInviteCodeRepository,
+    dvmJobRepository: IDvmJobRepository,
   ) =>
   ([message, adapter]: [IncomingMessage, IWebSocketAdapter]) => {
     switch (message[0]) {
       case MessageType.EVENT: {
         return new EventMessageHandler(
           adapter,
-          eventStrategyFactory(eventRepository, userRepository, inviteCodeRepository, getCache(), createSettings),
+          eventStrategyFactory(
+            eventRepository,
+            userRepository,
+            inviteCodeRepository,
+            dvmJobRepository,
+            getCache(),
+            createSettings,
+          ),
           eventRepository,
           userRepository,
           createSettings,

src/factories/websocket-adapter-factory.ts

@@ -1,7 +1,13 @@
 import { IncomingMessage } from 'http'
 import { WebSocket } from 'ws'
 
-import { IEventRepository, IInviteCodeRepository, INip05VerificationRepository, IUserRepository } from '../@types/repositories'
+import {
+  IDvmJobRepository,
+  IEventRepository,
+  IInviteCodeRepository,
+  INip05VerificationRepository,
+  IUserRepository,
+} from '../@types/repositories'
 import { createSettings } from './settings-factory'
 import { IWebSocketServerAdapter } from '../@types/adapters'
 import { messageHandlerFactory } from './message-handler-factory'
@@ -14,13 +20,20 @@ export const webSocketAdapterFactory =
     userRepository: IUserRepository,
     nip05VerificationRepository: INip05VerificationRepository,
     inviteCodeRepository: IInviteCodeRepository,
+    dvmJobRepository: IDvmJobRepository,
   ) =>
   ([client, request, webSocketServerAdapter]: [WebSocket, IncomingMessage, IWebSocketServerAdapter]) =>
     new WebSocketAdapter(
       client,
       request,
       webSocketServerAdapter,
-      messageHandlerFactory(eventRepository, userRepository, nip05VerificationRepository, inviteCodeRepository),
+      messageHandlerFactory(
+        eventRepository,
+        userRepository,
+        nip05VerificationRepository,
+        inviteCodeRepository,
+        dvmJobRepository,
+      ),
       rateLimiterFactory,
       createSettings,
     )

src/factories/worker-factory.ts

@@ -8,6 +8,7 @@ import { AppWorker } from '../app/worker'
 import { createLogger } from './logger-factory'
 import { createSettings } from '../factories/settings-factory'
 import { createWebApp } from './web-app-factory'
+import { DvmJobRepository } from '../repositories/dvm-job-repository'
 import { EventRepository } from '../repositories/event-repository'
 import { InviteCodeRepository } from '../repositories/invite-code-repository'
 import { Nip05VerificationRepository } from '../repositories/nip05-verification-repository'
@@ -24,6 +25,7 @@ export const workerFactory = (): AppWorker => {
   const userRepository = new UserRepository(dbClient, eventRepository)
   const nip05VerificationRepository = new Nip05VerificationRepository(dbClient)
   const inviteCodeRepository = new InviteCodeRepository(dbClient)
+  const dvmJobRepository = new DvmJobRepository(dbClient)
 
   const settings = createSettings()
 
@@ -65,7 +67,13 @@ export const workerFactory = (): AppWorker => {
   const adapter = new WebSocketServerAdapter(
     server,
     webSocketServer,
-    webSocketAdapterFactory(eventRepository, userRepository, nip05VerificationRepository, inviteCodeRepository),
+    webSocketAdapterFactory(
+      eventRepository,
+      userRepository,
+      nip05VerificationRepository,
+      inviteCodeRepository,
+      dvmJobRepository,
+    ),
     createSettings,
   )
 

src/handlers/event-strategies/dvm-job-request-event-strategy.ts

@@ -0,0 +1,42 @@
+import { createEventCommandResult } from '../../telemetry/event-metrics'
+import { createLogger } from '../../factories/logger-factory'
+import { Event } from '../../@types/event'
+import { IDvmJobRepository, IEventRepository } from '../../@types/repositories'
+import { IEventStrategy } from '../../@types/message-handlers'
+import { IWebSocketAdapter } from '../../@types/adapters'
+import { WebSocketAdapterEvent } from '../../constants/adapter'
+
+const logger = createLogger('dvm-job-request-event-strategy')
+
+export class DvmJobRequestEventStrategy implements IEventStrategy<Event, Promise<void>> {
+  public constructor(
+    private readonly webSocket: IWebSocketAdapter,
+    private readonly eventRepository: IEventRepository,
+    private readonly dvmJobRepository: IDvmJobRepository,
+  ) {}
+
+  public async execute(event: Event): Promise<void> {
+    logger('received dvm job request: %o', event)
+
+    const count = await this.eventRepository.create(event)
+    this.webSocket.emit(
+      WebSocketAdapterEvent.Message,
+      createEventCommandResult(event.id, true, count ? '' : 'duplicate:'),
+    )
+
+    if (!count) {
+      return
+    }
+
+    this.webSocket.emit(WebSocketAdapterEvent.Broadcast, event)
+
+    try {
+      await this.dvmJobRepository.create(event.id, event.pubkey, event.kind)
+    } catch (error) {
+      // Job-state recording is best-effort: the event itself is already
+      // stored and broadcast correctly, so a repository failure here must
+      // not surface as a rejection of a valid event.
+      logger.error('unable to record dvm job for event %s: %o', event.id, error)
+    }
+  }
+}

src/utils/event.ts

@@ -40,18 +40,20 @@ export const isEventMatchingFilter =
   (filter: SubscriptionFilter) =>
   (event: Event): boolean => {
     const startsWith = (input: string) => (prefix: string) => input.startsWith(prefix)
-    const isMatchingGenericTagCriterion = (key: string, criterion: string) => (tag: Tag): boolean => {
-      const [, tagName] = key
-      if (tag[0] !== tagName) {
-        return false
-      }
+    const isMatchingGenericTagCriterion =
+      (key: string, criterion: string) =>
+      (tag: Tag): boolean => {
+        const [, tagName] = key
+        if (tag[0] !== tagName) {
+          return false
+        }
 
-      if (isGeohashPrefixCriterion(key, criterion)) {
-        return tag[1].startsWith(stripGeohashPrefixWildcard(criterion))
-      }
+        if (isGeohashPrefixCriterion(key, criterion)) {
+          return tag[1].startsWith(stripGeohashPrefixWildcard(criterion))
+        }
 
-      return tag[1] === criterion
-    }
+        return tag[1] === criterion
+      }
 
     // NIP-01: Basic protocol flow description
 
@@ -96,7 +98,9 @@ export const isEventMatchingFilter =
       Object.entries(filter)
         .filter(([key, criteria]) => isGenericTagQuery(key) && Array.isArray(criteria))
         .some(([key, criteria]) => {
-          return !event.tags.some((tag) => criteria.some((criterion) => isMatchingGenericTagCriterion(key, criterion)(tag)))
+          return !event.tags.some((tag) =>
+            criteria.some((criterion) => isMatchingGenericTagCriterion(key, criterion)(tag)),
+          )
         })
     ) {
       return false
@@ -205,6 +209,10 @@ export const isEphemeralEvent = (event: Event): boolean => {
   return event.kind >= EventKinds.EPHEMERAL_FIRST && event.kind <= EventKinds.EPHEMERAL_LAST
 }
 
+export const isDvmJobRequestEvent = (event: Event): boolean => {
+  return event.kind >= EventKinds.DVM_JOB_REQUEST_FIRST && event.kind <= EventKinds.DVM_JOB_REQUEST_LAST
+}
+
 export const isParameterizedReplaceableEvent = (event: Event): boolean => {
   return (
     event.kind >= EventKinds.PARAMETERIZED_REPLACEABLE_FIRST && event.kind <= EventKinds.PARAMETERIZED_REPLACEABLE_LAST

test/unit/factories/event-strategy-factory.spec.ts

@@ -1,8 +1,14 @@
 import { expect } from 'chai'
 
-import { IEventRepository, IInviteCodeRepository, IUserRepository } from '../../../src/@types/repositories'
+import {
+  IDvmJobRepository,
+  IEventRepository,
+  IInviteCodeRepository,
+  IUserRepository,
+} from '../../../src/@types/repositories'
 import { DefaultEventStrategy } from '../../../src/handlers/event-strategies/default-event-strategy'
 import { DeleteEventStrategy } from '../../../src/handlers/event-strategies/delete-event-strategy'
+import { DvmJobRequestEventStrategy } from '../../../src/handlers/event-strategies/dvm-job-request-event-strategy'
 import { EphemeralEventStrategy } from '../../../src/handlers/event-strategies/ephemeral-event-strategy'
 import { Event } from '../../../src/@types/event'
 import { EventKinds } from '../../../src/constants/base'
@@ -24,6 +30,7 @@ describe('eventStrategyFactory', () => {
   let eventRepository: IEventRepository
   let userRepository: IUserRepository
   let inviteCodeRepository: IInviteCodeRepository
+  let dvmJobRepository: IDvmJobRepository
   let cache: ICacheAdapter
   let settings: () => Settings
   let event: Event
@@ -34,12 +41,20 @@ describe('eventStrategyFactory', () => {
     eventRepository = {} as any
     userRepository = {} as any
     inviteCodeRepository = {} as any
+    dvmJobRepository = {} as any
     cache = {} as any
-    settings = () => ({ info: { relay_url: 'wss://test.relay' } } as any)
+    settings = () => ({ info: { relay_url: 'wss://test.relay' } }) as any
     event = {} as any
     adapter = {} as any
 
-    factory = eventStrategyFactory(eventRepository, userRepository, inviteCodeRepository, cache, settings)
+    factory = eventStrategyFactory(
+      eventRepository,
+      userRepository,
+      inviteCodeRepository,
+      dvmJobRepository,
+      cache,
+      settings,
+    )
   })
 
   it('returns ReplaceableEvent given a set_metadata event', () => {
@@ -136,4 +151,14 @@ describe('eventStrategyFactory', () => {
     event.kind = EventKinds.NIP43_LEAVE_REQUEST
     expect(factory([event, adapter])).to.be.an.instanceOf(LeaveRequestEventStrategy)
   })
+
+  it('returns DvmJobRequestEventStrategy given a DVM job request (kind 5000-5999)', () => {
+    event.kind = EventKinds.DVM_JOB_REQUEST_FIRST
+    expect(factory([event, adapter])).to.be.an.instanceOf(DvmJobRequestEventStrategy)
+  })
+
+  it('returns DvmJobRequestEventStrategy given the last DVM job request kind (5999)', () => {
+    event.kind = EventKinds.DVM_JOB_REQUEST_LAST
+    expect(factory([event, adapter])).to.be.an.instanceOf(DvmJobRequestEventStrategy)
+  })
 })