The first 2 weeks is ended.

apps/worker/src/index.ts

@@ -0,0 +1,241 @@
+import './env'
+import { Worker } from 'bullmq'
+import IORedis from 'ioredis'
+import { applyMechanicalEvents, bindMechanicalEventsToRolls, resolveCheck } from '@dng/game-engine'
+import { CharacterSchema, PlayerIntentSchema, makeId } from '@dng/shared'
+import { generateWorld, narrateRound, planRound } from './ai'
+
+const redisUrl = process.env.REDIS_URL
+const supabaseUrl = process.env.SUPABASE_URL
+const serviceKey = process.env.SUPABASE_SERVICE_ROLE_KEY
+
+if (!supabaseUrl || !serviceKey) {
+  throw new Error('[worker] SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY are required.')
+} else {
+  const databaseUrl = supabaseUrl
+  const databaseServiceKey = serviceKey
+  type JobType = 'generate-world' | 'resolve-round'
+  interface JobPayload {
+    id: string
+    name: JobType
+    entityId: string
+    messages?: Array<{ role: 'user' | 'assistant'; content: string }>
+    claimToken?: string
+  }
+  interface ClaimedJob {
+    id: string
+    job_type: JobType
+    entity_id: string
+  }
+
+  class SupabaseRequestError extends Error {
+    constructor(
+      readonly status: number,
+      readonly path: string,
+      detail: string,
+    ) {
+      const rpcName = path.startsWith('rpc/') ? path.slice('rpc/'.length) : null
+      const message = status === 404 && rpcName
+        ? `Supabase RPC "${rpcName}" was not found. Apply supabase/migrations/0002_supabase_ai_queue.sql in the Supabase SQL Editor, then restart the worker.`
+        : `Supabase returned ${status} for ${path}: ${detail}`
+      super(message)
+      this.name = 'SupabaseRequestError'
+    }
+  }
+
+  async function databaseRequest<T>(path: string, init: RequestInit = {}): Promise<T> {
+    const response = await fetch(new URL(`/rest/v1/${path}`, databaseUrl), {
+      ...init,
+      headers: {
+        apikey: databaseServiceKey,
+        Authorization: `Bearer ${databaseServiceKey}`,
+        'Content-Type': 'application/json',
+        ...init.headers,
+      },
+    })
+    if (!response.ok) {
+      const detail = await response.text()
+      throw new SupabaseRequestError(response.status, path, detail)
+    }
+    if (response.status === 204) return undefined as T
+    return response.json() as Promise<T>
+  }
+
+  async function selectRows<T>(table: string, query: Record<string, string>): Promise<T[]> {
+    return databaseRequest<T[]>(`${table}?${new URLSearchParams(query)}`)
+  }
+
+  async function processWorld(job: JobPayload) {
+    let messages = job.messages
+    if (!messages) {
+      const [session] = await selectRows<{ messages: Array<{ role: 'user' | 'assistant'; content: string }> }>('coauthor_sessions', {
+        select: 'messages', id: `eq.${job.entityId}`, limit: '1',
+      })
+      if (!session) throw new Error('Coauthor session not found')
+      messages = session.messages
+    }
+    const world = await generateWorld(messages)
+    await databaseRequest(`coauthor_sessions?id=eq.${encodeURIComponent(job.entityId)}`, {
+      method: 'PATCH',
+      body: JSON.stringify({ status: 'ready', generated_world: world, updated_at: new Date().toISOString() }),
+    })
+    await databaseRequest('rpc/complete_ai_job', {
+      method: 'POST', body: JSON.stringify({ p_job_id: job.id, p_worker_id: job.claimToken ?? null }),
+    })
+    return world
+  }
+
+  async function processRound(job: JobPayload) {
+    const [round] = await selectRows<Record<string, any>>('rounds', {
+      select: '*,campaigns!inner(*)',
+      id: `eq.${job.entityId}`,
+      limit: '1',
+    })
+    if (!round) throw new Error('Round not found')
+    if (round.status === 'resolved') return { duplicate: true }
+
+    const [rawCharacters, rawIntents, rawMemories] = await Promise.all([
+      selectRows<Record<string, any>>('characters', { select: '*', campaign_id: `eq.${round.campaign_id}` }),
+      selectRows<Record<string, any>>('player_intents', { select: '*', round_id: `eq.${round.id}` }),
+      selectRows<{ summary: string }>('memories', { select: 'summary', campaign_id: `eq.${round.campaign_id}`, order: 'importance.desc', limit: '12' }),
+    ])
+
+    const characters = rawCharacters.map(row => CharacterSchema.parse({
+      id: row.id, name: row.name, concept: row.concept, controller: row.controller,
+      userId: row.user_id, abilities: row.abilities, hp: row.hp, maxHp: row.max_hp,
+      defense: row.defense, proficiency: row.proficiency, inventory: row.inventory, statuses: row.statuses,
+    }))
+    const intents = rawIntents.map(row => PlayerIntentSchema.parse({
+      id: row.id, roundId: row.round_id, memberId: row.member_id, characterId: row.character_id,
+      action: row.action, ready: row.ready, createdAt: row.created_at, updatedAt: row.updated_at,
+    }))
+    const scene = String(round.campaigns.current_scene ?? '')
+    const plan = await planRound({ scene, intents, characters, memories: rawMemories.map(row => String(row.summary)) })
+    const rolls = plan.checks.map(check => {
+      const actor = characters.find(character => character.id === check.actorId)
+      if (!actor) throw new Error(`Unknown check actor ${check.actorId}`)
+      return resolveCheck(check, actor)
+    })
+    const permittedEvents = bindMechanicalEventsToRolls(plan.proposedEvents, rolls)
+    const resolution = await narrateRound({ scene, intents, aiActions: plan.aiActions, rolls, permittedEvents })
+    const safeEvents = resolution.events.filter(event => permittedEvents.some(permitted => JSON.stringify(permitted) === JSON.stringify(event)))
+    const applied = applyMechanicalEvents(characters, safeEvents)
+
+    await databaseRequest(job.claimToken ? 'rpc/commit_claimed_round_resolution' : 'rpc/commit_round_resolution', {
+      method: 'POST',
+      body: JSON.stringify({
+        p_round_id: round.id,
+        p_narration: resolution.narration,
+        p_next_prompt: resolution.nextPrompt,
+        p_rolls: rolls,
+        p_events: safeEvents,
+        p_character_states: applied.characters.map(character => ({ id: character.id, hp: character.hp, inventory: character.inventory, statuses: character.statuses })),
+        p_memory: resolution.memory,
+        p_idempotency_key: job.id,
+        ...(job.claimToken ? { p_worker_id: job.claimToken } : {}),
+      }),
+    })
+    return { narration: resolution.narration, rolls: rolls.length }
+  }
+
+  async function processJob(job: JobPayload) {
+    if (job.name === 'generate-world') return processWorld(job)
+    if (job.name === 'resolve-round') return processRound(job)
+    throw new Error(`Unknown job type: ${job.name}`)
+  }
+
+  async function withLeaseHeartbeat<T>(jobId: string, workerId: string, task: () => Promise<T>): Promise<T> {
+    let heartbeatError: unknown
+    let renewing = false
+    const heartbeat = setInterval(() => {
+      if (renewing) return
+      renewing = true
+      void databaseRequest('rpc/renew_ai_job_lease', {
+        method: 'POST', body: JSON.stringify({ p_job_id: jobId, p_worker_id: workerId }),
+      }).catch(error => { heartbeatError = error }).finally(() => { renewing = false })
+    }, 60_000)
+    heartbeat.unref()
+    try {
+      const result = await task()
+      if (heartbeatError) throw heartbeatError
+      return result
+    } finally {
+      clearInterval(heartbeat)
+    }
+  }
+
+  if (redisUrl) {
+    const connection = new IORedis(redisUrl, { maxRetriesPerRequest: null })
+    const workerId = process.env.RAILWAY_REPLICA_ID ?? makeId('worker')
+    const worker = new Worker('dng-ai', async job => {
+      if (!job.id) throw new Error('BullMQ job id must match an ai_jobs id')
+      const [claimed] = await databaseRequest<ClaimedJob[]>('rpc/claim_ai_job_by_id', {
+        method: 'POST', body: JSON.stringify({ p_job_id: String(job.id), p_worker_id: workerId }),
+      })
+      if (!claimed) throw new Error(`AI job ${job.id} is not claimable`)
+      try {
+        return await withLeaseHeartbeat(claimed.id, workerId, () => processJob({
+          id: claimed.id,
+          name: claimed.job_type,
+          entityId: claimed.entity_id,
+          messages: job.data.messages,
+          claimToken: workerId,
+        }))
+      } catch (error) {
+        const message = error instanceof Error ? error.message : String(error)
+        await databaseRequest('rpc/retry_ai_job', {
+          method: 'POST', body: JSON.stringify({ p_job_id: claimed.id, p_worker_id: workerId, p_error: message.slice(0, 2_000) }),
+        })
+        throw error
+      }
+    }, { connection, concurrency: 4, lockDuration: 120_000 })
+    console.info('[worker] Dungeons & Ground worker ready (BullMQ)')
+
+    const close = async () => {
+      await worker.close()
+      await connection.quit()
+    }
+    process.once('SIGTERM', () => void close())
+    process.once('SIGINT', () => void close())
+  } else {
+    const pollInterval = Math.max(250, Number(process.env.AI_JOB_POLL_INTERVAL_MS ?? 1_000))
+    const workerId = process.env.RAILWAY_REPLICA_ID ?? makeId('worker')
+    let stopped = false
+
+    const stop = () => { stopped = true }
+    process.once('SIGTERM', stop)
+    process.once('SIGINT', stop)
+    console.info(`[worker] Dungeons & Ground worker ready (Supabase polling every ${pollInterval}ms)`)
+
+    while (!stopped) {
+      try {
+        const claimed = await databaseRequest<ClaimedJob[]>('rpc/claim_ai_job', {
+          method: 'POST', body: JSON.stringify({ p_worker_id: workerId }),
+        })
+        const job = claimed[0]
+        if (!job) {
+          await new Promise(resolve => setTimeout(resolve, pollInterval))
+          continue
+        }
+        try {
+          await withLeaseHeartbeat(job.id, workerId, () => processJob({ id: job.id, name: job.job_type, entityId: job.entity_id, claimToken: workerId }))
+        } catch (error) {
+          const message = error instanceof Error ? error.message : String(error)
+          console.error(`[worker] ${job.job_type} ${job.id} failed: ${message}`)
+          await databaseRequest('rpc/retry_ai_job', {
+            method: 'POST',
+            body: JSON.stringify({ p_job_id: job.id, p_worker_id: workerId, p_error: message.slice(0, 2_000) }),
+          })
+        }
+      } catch (error) {
+        if (error instanceof SupabaseRequestError && error.status === 404 && error.path.startsWith('rpc/')) {
+          console.error(`[worker] ${error.message}`)
+          process.exitCode = 1
+          break
+        }
+        console.error('[worker] polling failed:', error)
+        await new Promise(resolve => setTimeout(resolve, pollInterval))
+      }
+    }
+  }
+}