feat(dvm): add dvm job persistence migration and repository

.changeset/dvm-job-persistence.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(dvm): add job persistence migration and repository for DVM job state

.knip.json

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

migrations/20260812_150000_create_dvm_jobs_table.js

@@ -0,0 +1,29 @@
+exports.up = function (knex) {
+  return knex.schema.createTable('dvm_jobs', (table) => {
+    table.binary('id').primary()
+    table.binary('requester_pubkey').notNullable()
+    table.integer('kind').unsigned().notNullable()
+    table.integer('worker_index').nullable()
+    table
+      .enum('status', ['submitted', 'picked_up', 'completed', 'failed', 'timed_out'])
+      .notNullable()
+      .defaultTo('submitted')
+    table.binary('result_event_id').nullable()
+    table.text('error').nullable()
+    table.timestamp('picked_up_at', { useTz: true }).nullable()
+    table.timestamp('completed_at', { useTz: true }).nullable()
+    table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+
+    table.index(['requester_pubkey'], 'idx_dvm_jobs_requester_pubkey')
+    // Composite (not status-only): findPendingJobs() filters by status AND
+    // orders by created_at, so the index needs to satisfy both the filter
+    // and the sort for FIFO polling, same as invoices_pending_created_at_idx.
+    table.index(['status', 'created_at'], 'idx_dvm_jobs_status_created_at')
+    table.index(['kind'], 'idx_dvm_jobs_kind')
+  })
+}
+
+exports.down = function (knex) {
+  return knex.schema.dropTable('dvm_jobs')
+}

src/@types/dvm.ts

@@ -0,0 +1,37 @@
+import { Pubkey } from './base'
+
+export enum DvmJobStatus {
+  SUBMITTED = 'submitted',
+  PICKED_UP = 'picked_up',
+  COMPLETED = 'completed',
+  FAILED = 'failed',
+  TIMED_OUT = 'timed_out',
+}
+
+export interface DvmJob {
+  id: string
+  requesterPubkey: Pubkey
+  kind: number
+  workerIndex: number | null
+  status: DvmJobStatus
+  resultEventId: string | null
+  error: string | null
+  pickedUpAt: Date | null
+  completedAt: Date | null
+  createdAt: Date
+  updatedAt: Date
+}
+
+export interface DBDvmJob {
+  id: Buffer
+  requester_pubkey: Buffer
+  kind: number
+  worker_index: number | null
+  status: DvmJobStatus
+  result_event_id: Buffer | null
+  error: string | null
+  picked_up_at: Date | null
+  completed_at: Date | null
+  created_at: Date
+  updated_at: Date
+}

src/@types/repositories.ts

@@ -1,12 +1,12 @@
 import { PassThrough } from 'stream'
-
+import { EventKinds } from '../constants/base'
 import { DatabaseClient, EventId, Pubkey } from './base'
+import { DvmJob } from './dvm'
 import { DBEvent, Event } from './event'
-import { EventKinds } from '../constants/base'
-import { EventKindsRange } from './settings'
 import { InviteCode } from './invite-code'
 import { Invoice } from './invoice'
 import { Nip05Verification } from './nip05'
+import { EventKindsRange } from './settings'
 import { SubscriptionFilter } from './subscription'
 import { User } from './user'
 
@@ -73,3 +73,13 @@ export interface IInviteCodeRepository {
   findActiveCodes(limit?: number): Promise<InviteCode[]>
   deleteExpiredCodes(): Promise<number>
 }
+
+export interface IDvmJobRepository {
+  create(id: string, requesterPubkey: Pubkey, kind: number): Promise<DvmJob>
+  findById(id: string): Promise<DvmJob | undefined>
+  assignWorker(id: string, workerIndex: number): Promise<boolean>
+  updateStatus(
+    job: Pick<DvmJob, 'id' | 'status'> & Partial<Pick<DvmJob, 'resultEventId' | 'error'>>,
+  ): Promise<DvmJob | undefined>
+  findPendingJobs(limit?: number): Promise<DvmJob[]>
+}

src/repositories/dvm-job-repository.ts

@@ -0,0 +1,136 @@
+import { DatabaseClient, Pubkey } from '../@types/base'
+import { DBDvmJob, DvmJob, DvmJobStatus } from '../@types/dvm'
+import { IDvmJobRepository } from '../@types/repositories'
+import { createLogger } from '../factories/logger-factory'
+import { fromBuffer, toBuffer } from '../utils/transform'
+
+const logger = createLogger('dvm-job-repository')
+
+function fromDBDvmJob(row: DBDvmJob): DvmJob {
+  return {
+    id: fromBuffer(row.id),
+    requesterPubkey: fromBuffer(row.requester_pubkey),
+    kind: row.kind,
+    workerIndex: row.worker_index,
+    status: row.status,
+    resultEventId: row.result_event_id ? fromBuffer(row.result_event_id) : null,
+    error: row.error,
+    pickedUpAt: row.picked_up_at,
+    completedAt: row.completed_at,
+    createdAt: row.created_at,
+    updatedAt: row.updated_at,
+  }
+}
+
+function affectedRows(result: unknown): number {
+  if (typeof result === 'number') {
+    return result
+  }
+  if (result && typeof (result as any).rowCount === 'number') {
+    return (result as any).rowCount
+  }
+  return 0
+}
+
+export class DvmJobRepository implements IDvmJobRepository {
+  public constructor(private readonly dbClient: DatabaseClient) {}
+
+  public async create(
+    id: string,
+    requesterPubkey: Pubkey,
+    kind: number,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<DvmJob> {
+    logger('create dvm job %s (kind %d) for %s', id, kind, requesterPubkey)
+
+    const now = new Date()
+    const row: DBDvmJob = {
+      id: toBuffer(id),
+      requester_pubkey: toBuffer(requesterPubkey),
+      kind,
+      worker_index: null,
+      status: DvmJobStatus.SUBMITTED,
+      result_event_id: null,
+      error: null,
+      picked_up_at: null,
+      completed_at: null,
+      created_at: now,
+      updated_at: now,
+    }
+
+    await client<DBDvmJob>('dvm_jobs').insert(row)
+
+    return fromDBDvmJob(row)
+  }
+
+  public async findById(id: string, client: DatabaseClient = this.dbClient): Promise<DvmJob | undefined> {
+    logger('find dvm job %s', id)
+
+    const [row] = await client<DBDvmJob>('dvm_jobs').where('id', toBuffer(id)).select()
+
+    if (!row) {
+      return
+    }
+
+    return fromDBDvmJob(row)
+  }
+
+  // Atomic pickup: single conditional UPDATE ensures only one worker wins the job
+  public async assignWorker(id: string, workerIndex: number, client: DatabaseClient = this.dbClient): Promise<boolean> {
+    logger('assign dvm job %s to worker %d', id, workerIndex)
+
+    const now = new Date()
+
+    const result = await client<DBDvmJob>('dvm_jobs')
+      .where('id', toBuffer(id))
+      .where('status', DvmJobStatus.SUBMITTED)
+      .update({
+        worker_index: workerIndex,
+        status: DvmJobStatus.PICKED_UP,
+        picked_up_at: now,
+        updated_at: now,
+      })
+
+    return affectedRows(result) > 0
+  }
+
+  public async updateStatus(
+    job: Pick<DvmJob, 'id' | 'status'> & Partial<Pick<DvmJob, 'resultEventId' | 'error'>>,
+    client: DatabaseClient = this.dbClient,
+  ): Promise<DvmJob | undefined> {
+    logger('update dvm job status: %o', job)
+
+    const now = new Date()
+    const isTerminal =
+      job.status === DvmJobStatus.COMPLETED ||
+      job.status === DvmJobStatus.FAILED ||
+      job.status === DvmJobStatus.TIMED_OUT
+
+    // Check key presence, not truthiness: a caller passing `resultEventId: null`
+    // or `error: null` is explicitly clearing the field, which a truthy check
+    // would silently ignore and leave the stale DB value in place.
+    const update: Partial<DBDvmJob> = {
+      status: job.status,
+      updated_at: now,
+      ...(isTerminal ? { completed_at: now } : {}),
+      ...('resultEventId' in job ? { result_event_id: job.resultEventId ? toBuffer(job.resultEventId) : null } : {}),
+      ...('error' in job ? { error: job.error ?? null } : {}),
+    }
+
+    const [row] = await client<DBDvmJob>('dvm_jobs').where('id', toBuffer(job.id)).update(update).returning(['*'])
+
+    return row ? fromDBDvmJob(row) : undefined
+  }
+
+  public async findPendingJobs(limit = 100, client: DatabaseClient = this.dbClient): Promise<DvmJob[]> {
+    logger('find pending dvm jobs (limit %d)', limit)
+
+    const rows = await client<DBDvmJob>('dvm_jobs')
+      .whereIn('status', [DvmJobStatus.SUBMITTED, DvmJobStatus.PICKED_UP])
+      .orderBy('created_at', 'asc')
+      .limit(limit)
+      .select()
+
+    return rows.map(fromDBDvmJob)
+  }
+}