feat(dvm): dispatch pending DVM jobs to worker processes (#734)

* feat(dvm): dispatch pending DVM jobs to worker processes

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

* fix(cli): guard worker stdin writes against closed streams

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

* fix(dvm): prevent dispatch starvation and shutdown job leaks

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

* fix(dvm): set default job timeout to 1 minute

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

---------

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

Author: Priyanshubhartistm

.changeset/dvm-job-dispatch.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(dvm): dispatch pending DVM jobs to worker processes and publish kind 6000-6999 results back

src/@types/repositories.ts

@@ -81,5 +81,5 @@ export interface IDvmJobRepository {
   updateStatus(
     job: Pick<DvmJob, 'id' | 'status'> & Partial<Pick<DvmJob, 'resultEventId' | 'error'>>,
   ): Promise<DvmJob | undefined>
-  findPendingJobs(limit?: number): Promise<DvmJob[]>
+  findPendingJobs(limit?: number, kinds?: number[]): Promise<DvmJob[]>
 }

src/app/dvm-orchestrator-worker.ts

@@ -1,17 +1,48 @@
-import { path } from 'ramda'
-import { IRunnable } from '../@types/base'
+import { andThen, otherwise, path, pipe } from 'ramda'
+
+import {
+  broadcastEvent,
+  getPublicKey,
+  getRelayPrivateKey,
+  identifyEvent,
+  signEvent,
+  toNostrEvent,
+} from '../utils/event'
+import { DvmJob, DvmJobStatus } from '../@types/dvm'
 import { DvmWorker, Settings } from '../@types/settings'
+import { Event, UnidentifiedEvent } from '../@types/event'
+import { EventKinds, EventTags } from '../constants/base'
+import { IDvmJobRepository, IEventRepository } from '../@types/repositories'
+import { spawnWorkerProcess, WorkerProcessHandle, WorkerSpawnErrorReason } from '../cli/utils/process'
 import { createLogger } from '../factories/logger-factory'
+import { IRunnable } from '../@types/base'
 import { shutdownMetricsTelemetry } from '../telemetry/metrics'
 
 const logger = createLogger('dvm-orchestrator-worker')
 
+const POLL_INTERVAL_MS = 2000
+const DEFAULT_JOB_TIMEOUT_MS = 60000
+const DISPATCH_BATCH_SIZE = 10
+
+type PendingJob = {
+  job: DvmJob
+  requestEvent: Event
+  timer: NodeJS.Timeout
+}
+
 export class DvmOrchestratorWorker implements IRunnable {
   private config: DvmWorker | undefined
+  private interval: NodeJS.Timeout | undefined
+  private isRunning = false
+  private closing = false
+  private worker: WorkerProcessHandle | undefined
+  private readonly pending = new Map<string, PendingJob>()
 
   public constructor(
     private readonly process: NodeJS.Process,
     private readonly settings: () => Settings,
+    private readonly dvmJobRepository: IDvmJobRepository,
+    private readonly eventRepository: IEventRepository,
   ) {
     this.process
       .on('SIGINT', this.onExit.bind(this))
@@ -33,6 +64,221 @@ export class DvmOrchestratorWorker implements IRunnable {
     }
 
     logger.info('dvm-orchestrator worker started for command: %s', this.config.command)
+
+    this.ensureWorkerProcess()
+
+    this.interval = setInterval(async () => {
+      if (this.isRunning) {
+        logger('skipping scheduled dispatch because previous run is still in progress')
+        return
+      }
+
+      this.isRunning = true
+      try {
+        await this.dispatchNextJob()
+      } catch (error) {
+        this.onError(error as Error)
+      } finally {
+        this.isRunning = false
+      }
+    }, POLL_INTERVAL_MS)
+  }
+
+  // One long-lived worker process per configured dvm.workers[i], multiplexing
+  // every in-flight job over a single newline-delimited-JSON stdin/stdout pipe
+  // (per issue #731 — process.ts's one-shot spawn helpers buffer all output
+  // into a single string, which doesn't work once more than one job can be
+  // in flight against the same worker at a time).
+  private ensureWorkerProcess(): void {
+    if (this.worker || !this.config || this.closing) {
+      return
+    }
+
+    const worker = spawnWorkerProcess(this.config.command, this.config.args ?? [])
+    worker.onMessage((message) => this.handleWorkerMessage(message))
+    worker.onExit((code, signal) => this.handleWorkerExit(code, signal))
+    worker.onSpawnError((reason) => this.handleWorkerSpawnError(reason))
+    this.worker = worker
+  }
+
+  private async dispatchNextJob(): Promise<void> {
+    if (!this.config) {
+      return
+    }
+
+    // Lazily respawn if the worker process died since the last tick.
+    this.ensureWorkerProcess()
+    if (!this.worker) {
+      return
+    }
+
+    // findPendingJobs() returns both SUBMITTED and PICKED_UP jobs (oldest first);
+    // assignWorker() only succeeds against SUBMITTED ones. Walk a batch instead of
+    // just the oldest candidate so an already-picked-up job at the head can't
+    // starve out later SUBMITTED jobs behind it.
+    const candidates = await this.dvmJobRepository.findPendingJobs(DISPATCH_BATCH_SIZE, this.config.kinds)
+
+    let job: DvmJob | undefined
+    for (const candidate of candidates) {
+      if (await this.dvmJobRepository.assignWorker(candidate.id, this.workerIndex())) {
+        job = candidate
+        break
+      }
+    }
+
+    if (!job) {
+      // Lost the race on every candidate in this batch — try again next tick.
+      return
+    }
+
+    logger('picked up job %s (kind %d)', job.id, job.kind)
+
+    const [row] = await this.eventRepository.findByFilters([{ ids: [job.id] }])
+    if (!row) {
+      await this.failJob(job.id, 'source event not found')
+      return
+    }
+
+    const requestEvent = toNostrEvent(row)
+    const timeoutMs = this.config.timeoutMs ?? DEFAULT_JOB_TIMEOUT_MS
+    const timer = setTimeout(() => this.handleJobTimeout(job.id), timeoutMs)
+    this.pending.set(job.id, { job, requestEvent, timer })
+
+    const sent = this.worker.send({
+      id: requestEvent.id,
+      kind: requestEvent.kind,
+      pubkey: requestEvent.pubkey,
+      tags: requestEvent.tags,
+      content: requestEvent.content,
+    })
+
+    if (!sent) {
+      clearTimeout(timer)
+      this.pending.delete(job.id)
+      await this.failJob(job.id, 'unable to send job to worker process')
+    }
+  }
+
+  // The worker replies on the same pipe with { id: <job id>, content: <string> },
+  // correlated back to the job that's still pending — replies for jobs we no
+  // longer track (already timed out, or from a previous worker instance) are
+  // dropped rather than treated as an error.
+  private handleWorkerMessage(message: unknown): void {
+    const jobId = (message as { id?: unknown } | null)?.id
+    if (typeof jobId !== 'string') {
+      logger.error('ignoring malformed worker message (missing id): %o', message)
+      return
+    }
+
+    const pending = this.pending.get(jobId)
+    if (!pending) {
+      return
+    }
+
+    clearTimeout(pending.timer)
+    this.pending.delete(jobId)
+
+    const rawContent = (message as { content?: unknown }).content
+    const content = typeof rawContent === 'string' ? rawContent : JSON.stringify(message)
+
+    void this.publishResult(pending.job, pending.requestEvent, content)
+  }
+
+  private handleJobTimeout(jobId: string): void {
+    const pending = this.pending.get(jobId)
+    if (!pending) {
+      return
+    }
+
+    this.pending.delete(jobId)
+    void this.failJob(jobId, 'worker timeout', true)
+
+    // Soft Node-level guard per issue #731: kill the worker process on timeout.
+    // Any other jobs still in flight on it fail as a side effect of the exit
+    // handler below; ensureWorkerProcess() respawns on the next dispatch tick.
+    this.worker?.kill()
+  }
+
+  private handleWorkerExit(code: number | null, signal: NodeJS.Signals | null): void {
+    logger.error('dvm worker process exited (code=%s, signal=%s)', code, signal)
+    this.worker = undefined
+
+    for (const [jobId, pending] of this.pending) {
+      clearTimeout(pending.timer)
+      void this.failJob(jobId, `worker exited (code=${code ?? 'null'}, signal=${signal ?? 'null'})`)
+    }
+    this.pending.clear()
+  }
+
+  private handleWorkerSpawnError(reason: WorkerSpawnErrorReason): void {
+    logger.error('unable to spawn dvm worker process: %s', reason)
+    this.worker = undefined
+  }
+
+  private workerIndex(): number {
+    return Number(this.process.env.DVM_WORKER_INDEX)
+  }
+
+  private async failJob(jobId: string, error: string, timedOut = false): Promise<void> {
+    logger.error('job %s failed: %s', jobId, error)
+    try {
+      await this.dvmJobRepository.updateStatus({
+        id: jobId,
+        status: timedOut ? DvmJobStatus.TIMED_OUT : DvmJobStatus.FAILED,
+        error,
+      })
+    } catch (updateError) {
+      logger.error('unable to update failed job %s: %o', jobId, updateError)
+    }
+  }
+
+  // Relay-authored, matching the same self-signing pattern used for invoice
+  // notifications (payments-service.ts) and NIP-89 authoring: the settings
+  // schema has no per-worker signing key, so the relay's own derived keypair
+  // is the DVM identity for every locally-bridged worker.
+  private async publishResult(job: DvmJob, requestEvent: Event, content: string): Promise<void> {
+    const currentSettings = this.settings()
+    const relayPrivkey = getRelayPrivateKey(currentSettings.info.relay_url)
+    const relayPubkey = getPublicKey(relayPrivkey)
+
+    const unsignedEvent: UnidentifiedEvent = {
+      pubkey: relayPubkey,
+      kind: (requestEvent.kind + 1000) as EventKinds,
+      created_at: Math.floor(Date.now() / 1000),
+      content,
+      tags: [
+        [EventTags.Event, requestEvent.id],
+        [EventTags.Pubkey, requestEvent.pubkey],
+      ],
+    }
+
+    const persistEvent = async (event: Event) => {
+      await this.eventRepository.create(event)
+      return event
+    }
+
+    const markCompleted = async (event: Event) => {
+      await this.dvmJobRepository.updateStatus({
+        id: job.id,
+        status: DvmJobStatus.COMPLETED,
+        resultEventId: event.id,
+      })
+      return event
+    }
+
+    const logPublishError = async (error: Error) => {
+      logger.error('unable to publish result for job %s: %o', job.id, error)
+      await this.failJob(job.id, `unable to publish result: ${error.message}`)
+    }
+
+    await pipe(
+      identifyEvent,
+      andThen(signEvent(relayPrivkey)),
+      andThen(persistEvent),
+      andThen(broadcastEvent),
+      andThen(markCompleted),
+      otherwise(logPublishError),
+    )(unsignedEvent)
   }
 
   private onError(error: Error) {
@@ -51,8 +297,33 @@ export class DvmOrchestratorWorker implements IRunnable {
 
   public close(callback?: () => void) {
     logger('closing')
-    if (typeof callback === 'function') {
-      callback()
+    this.closing = true
+    if (this.interval) {
+      clearInterval(this.interval)
+    }
+
+    const invokeCallback = () => {
+      if (typeof callback === 'function') {
+        callback()
+      }
+    }
+
+    if (this.pending.size === 0) {
+      this.worker?.kill()
+      invokeCallback()
+      return
     }
+
+    // Fail every still-in-flight job before killing the worker: clearing
+    // `this.pending` first (as before) meant handleWorkerExit() found nothing
+    // to fail, leaving those jobs stuck at PICKED_UP in the DB forever.
+    const failures = Array.from(this.pending.entries()).map(([jobId, pending]) => {
+      clearTimeout(pending.timer)
+      return this.failJob(jobId, 'worker shutting down')
+    })
+    this.pending.clear()
+    this.worker?.kill()
+
+    void Promise.all(failures).finally(invokeCallback)
   }
 }