Dungeons-Ground/apps/worker/src/index.ts

import './env'
import { Worker } from 'bullmq'
import IORedis from 'ioredis'
import { applyMechanicalEvents, bindMechanicalEventsToRolls, resolveChecks } from '@dng/game-engine'
import { CharacterSchema, PlayerIntentSchema, makeId } from '@dng/shared'
import { generateWorld, narrateRound, planRound, summarizeStory, type AiUsageMeasurement } from './ai'
import { buildActionSequence, buildRetrievalText, buildRoundContext, sanitizeRoundMemory, shouldUpdateStorySummary, validateAndOrderRoundPlan } from './orchestration'

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
  }
  interface RetrievedContext {
    memories?: Array<{ summary: string; importance: number; tags: string[]; entityIds: string[] }>
    entities?: Array<{ id: string; kind: 'location' | 'npc' | 'faction' | 'quest'; name: string; summary: string; tags: string[] }>
    relationships?: Array<{ sourceEntityId: string; targetEntityId: string; score: number; notes: string }>
    activeGoals?: Array<{ questEntityId: string; name: string; status: 'hidden' | 'active' | 'completed' | 'failed'; summary: string; tags: 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. The database schema is incomplete. Run the generated supabase/bootstrap.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 headers = new Headers(init.headers)
    headers.set('apikey', databaseServiceKey)
    headers.set('Content-Type', 'application/json')
    if (databaseServiceKey.split('.').length === 3) headers.set('Authorization', `Bearer ${databaseServiceKey}`)
    else headers.delete('Authorization')
    const response = await fetch(new URL(`/rest/v1/${path}`, databaseUrl), {
      ...init,
      headers,
    })
    if (!response.ok) {
      const detail = await response.text()
      throw new SupabaseRequestError(response.status, path, detail)
    }
    const responseBody = await response.text()
    if (!responseBody) return undefined as T
    return JSON.parse(responseBody) as T
  }

  async function selectRows<T>(table: string, query: Record<string, string>): Promise<T[]> {
    return databaseRequest<T[]>(`${table}?${new URLSearchParams(query)}`)
  }

  function usageRecorder(job: JobPayload, userId: string | null, campaignId: string | null) {
    return async (measurement: AiUsageMeasurement): Promise<void> => {
      try {
        await databaseRequest('ai_usage?on_conflict=usage_key', {
          method: 'POST',
          headers: { Prefer: 'resolution=merge-duplicates,return=minimal' },
          body: JSON.stringify({
            usage_key: `${job.id}:${measurement.operation}:${makeId('usage')}`,
            user_id: userId,
            campaign_id: campaignId,
            job_id: job.id,
            provider: measurement.provider,
            request_kind: measurement.operation,
            model: measurement.model,
            input_tokens: measurement.inputTokens,
            output_tokens: measurement.outputTokens,
            cost_usd: measurement.costUsd,
            latency_ms: measurement.latencyMs,
          }),
        })
      } catch (error) {
        // Usage telemetry must never trigger a second paid model call. Quota
        // reservations are persisted before the job is queued, so a temporary
        // telemetry failure cannot bypass rate limits.
        console.error(`[worker] could not record ${measurement.operation} usage for job ${job.id}:`, error)
      }
    }
  }

  async function processWorld(job: JobPayload) {
    const [session] = await selectRows<{ owner_id: string; messages: Array<{ role: 'user' | 'assistant'; content: string }> }>('coauthor_sessions', {
      select: 'owner_id,messages', id: `eq.${job.entityId}`, limit: '1',
    })
    if (!session) throw new Error('Coauthor session not found')
    const messages = job.messages ?? session.messages
    const world = await generateWorld(messages, { onUsage: usageRecorder(job, session.owner_id, null) })
    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, rawRecentRounds, rawStorySummaries] = 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<{ number: number; narration: string; next_prompt: string | null }>('rounds', {
        select: 'number,narration,next_prompt', campaign_id: `eq.${round.campaign_id}`, status: 'eq.resolved', order: 'number.desc', limit: '2',
      }),
      selectRows<{ through_round: number; summary: string }>('story_summaries', {
        select: 'through_round,summary', campaign_id: `eq.${round.campaign_id}`, order: 'through_round.desc', limit: '1',
      }),
    ])

    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,
      persona: row.persona, rules: row.rules_state,
    }))
    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 retrieval = await databaseRequest<RetrievedContext>('rpc/stage_six_retrieve_context', {
      method: 'POST',
      body: JSON.stringify({
        p_campaign_id: round.campaign_id,
        p_search_text: buildRetrievalText({ scene, intents, characters }),
        p_limit: 8,
      }),
    })
    const context = buildRoundContext({
      scene,
      storySummary: rawStorySummaries[0]?.summary ?? null,
      recentRounds: rawRecentRounds.map(previous => ({ number: previous.number, narration: previous.narration, nextPrompt: previous.next_prompt })),
      intents,
      characters,
      memories: retrieval.memories ?? [],
      entities: retrieval.entities ?? [],
      relationships: retrieval.relationships ?? [],
      activeGoals: retrieval.activeGoals ?? [],
    })
    const recordUsage = usageRecorder(job, String(round.campaigns.owner_id), String(round.campaign_id))
    const plan = validateAndOrderRoundPlan(await planRound(context, { onUsage: recordUsage }), context)
    const rolls = resolveChecks(plan.checks, characters)
    const permittedEvents = bindMechanicalEventsToRolls(plan.proposedEvents, rolls)
    const resolution = await narrateRound({ context, actionSequence: buildActionSequence(context, plan), rolls, permittedEvents }, { onUsage: recordUsage })
    const safeEvents = resolution.events.filter(event => permittedEvents.some(permitted => JSON.stringify(permitted) === JSON.stringify(event)))
    const applied = applyMechanicalEvents(characters, safeEvents, rolls)
    const shouldSummarize = shouldUpdateStorySummary(Number(round.number))
    const summary = shouldSummarize
      ? await summarizeStory({
          previousSummary: context.storySummary,
          rounds: [...context.recentRounds.map(previous => ({ number: previous.number, narration: previous.narration })), { number: Number(round.number), narration: resolution.narration }],
        }, { onUsage: recordUsage })
      : null

    await databaseRequest(job.claimToken ? 'rpc/commit_claimed_stage_six_round_resolution' : 'rpc/commit_stage_six_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,
          rulesState: character.rules,
        })),
        p_memory: sanitizeRoundMemory(resolution.memory, new Set([
          ...context.activeCharacters.map(character => character.id),
          ...context.entities.map(entity => entity.id),
        ])),
        p_story_summary: summary?.summary ?? null,
        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
    let consecutivePollFailures = 0

    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 }),
        })
        consecutivePollFailures = 0
        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
        }
        consecutivePollFailures += 1
        const retryDelay = Math.min(30_000, pollInterval * 2 ** Math.min(consecutivePollFailures - 1, 5))
        const message = error instanceof Error ? error.message : String(error)
        // Avoid flooding the terminal and Supabase during a network outage.
        // Log the first failure and then only when the backoff duration grows.
        if (consecutivePollFailures === 1 || (consecutivePollFailures & (consecutivePollFailures - 1)) === 0) {
          console.error(`[worker] polling failed; retrying in ${retryDelay}ms: ${message}`)
        }
        await new Promise(resolve => setTimeout(resolve, retryDelay))
      }
    }
  }
}