import { getDbClient } from '../storage/database/local-db';
import {
  runGenerationPayload,
  type GenerationJobType,
} from './generation-job-runner';
import { ensureGenerationJobRuntimeSchema } from '@/lib/generation-job-estimates';
import { writePlatformLog } from '@/lib/platform-logs';

const POLL_INTERVAL_MS = Number(process.env.GENERATION_WORKER_INTERVAL_MS || 5000);
const STALE_RUNNING_MINUTES = Number(process.env.GENERATION_JOB_TIMEOUT_MINUTES || 30);

let processing = false;

async function updateJob(
  jobId: string,
  fields: {
    status: 'succeeded' | 'failed';
    result?: unknown;
    error?: string | null;
  },
) {
  const client = await getDbClient();
  try {
    await ensureGenerationJobRuntimeSchema(client);
    await client.query(
      `UPDATE generation_jobs
       SET status = $1,
           result = COALESCE($2::jsonb, result),
           error = $3,
           payload = '{}'::jsonb,
           progress = COALESCE(progress, '{}'::jsonb) || $5::jsonb,
           finished_at = NOW(),
           updated_at = NOW()
       WHERE id = $4`,
      [
        fields.status,
        fields.result === undefined ? null : JSON.stringify(fields.result),
        fields.error ?? null,
        jobId,
        JSON.stringify({
          percent: fields.status === 'succeeded' ? 100 : undefined,
          message: fields.status === 'succeeded' ? '生成结果已返回，正在准备展示' : '生成任务失败',
          resultReady: fields.status === 'succeeded',
          updatedAt: new Date().toISOString(),
        }),
      ],
    );
  } finally {
    client.release();
  }
}

export async function markStaleRunningJobs() {
  const client = await getDbClient();
  try {
    await ensureGenerationJobRuntimeSchema(client);
    const result = await client.query(
      `UPDATE generation_jobs
       SET status = 'failed',
           error = '任务执行超时或被服务重启中断',
           payload = '{}'::jsonb,
           finished_at = NOW(),
           updated_at = NOW()
       WHERE status = 'running'
         AND updated_at < NOW() - ($1::int * INTERVAL '1 minute')`,
      [STALE_RUNNING_MINUTES],
    );
    return result.rowCount || 0;
  } finally {
    client.release();
  }
}

async function claimNextJob() {
  const client = await getDbClient();
  try {
    await ensureGenerationJobRuntimeSchema(client);
    await client.query('BEGIN');
    const result = await client.query<{
      id: string;
      type: GenerationJobType;
      payload: Record<string, unknown>;
      user_id: string | null;
    }>(
      `WITH next_job AS (
         SELECT id
         FROM generation_jobs
         WHERE status = 'queued'
         ORDER BY created_at ASC
         FOR UPDATE SKIP LOCKED
         LIMIT 1
       )
       UPDATE generation_jobs AS j
       SET status = 'running',
           started_at = COALESCE(j.started_at, NOW()),
           progress = COALESCE(j.progress, '{}'::jsonb) || jsonb_build_object(
             'message', '生成任务已开始执行',
             'updatedAt', NOW()
           ),
           updated_at = NOW()
       FROM next_job
       WHERE j.id = next_job.id
       RETURNING j.id, j.type, j.payload, j.user_id`,
    );
    await client.query('COMMIT');
    return result.rows[0] || null;
  } catch (err) {
    await client.query('ROLLBACK').catch(() => undefined);
    throw err;
  } finally {
    client.release();
  }
}

export async function processNextGenerationJob() {
  if (processing) return false;
  processing = true;
  try {
    await markStaleRunningJobs();
    const job = await claimNextJob();
    if (!job) return false;

    try {
      const result = await runGenerationPayload(job.type, job.payload || {}, {
        userId: job.user_id,
        jobId: job.id,
      });
      await updateJob(job.id, {
        status: 'succeeded',
        result,
        error: null,
      });
      void writePlatformLog({
        type: 'generation',
        level: 'info',
        action: 'generation_job_succeeded',
        message: `生成任务执行成功`,
        userId: job.user_id,
        targetType: 'generation_job',
        targetId: job.id,
        metadata: { type: job.type },
      });
    } catch (err) {
      await updateJob(job.id, {
        status: 'failed',
        error: err instanceof Error ? err.message : 'Generation job failed',
      });
      void writePlatformLog({
        type: 'generation',
        level: 'error',
        action: 'generation_job_failed',
        message: '生成任务执行失败',
        userId: job.user_id,
        targetType: 'generation_job',
        targetId: job.id,
        metadata: {
          type: job.type,
          error: err instanceof Error ? err.message : 'Generation job failed',
        },
      });
    }

    return true;
  } catch (err) {
    console.error('[generation-worker] processing failed:', err);
    return false;
  } finally {
    processing = false;
  }
}

export function startGenerationJobWorker() {
  if (process.env.GENERATION_WORKER_DISABLED === '1') return;

  const state = globalThis as typeof globalThis & {
    __miaojingGenerationWorkerStarted?: boolean;
  };
  if (state.__miaojingGenerationWorkerStarted) return;
  state.__miaojingGenerationWorkerStarted = true;

  const tick = () => {
    void processNextGenerationJob();
  };
  const timer = setInterval(tick, POLL_INTERVAL_MS);
  timer.unref?.();

  const startupTimer = setTimeout(tick, 1000);
  startupTimer.unref?.();

  console.log(
    `[generation-worker] started interval=${POLL_INTERVAL_MS}ms timeout=${STALE_RUNNING_MINUTES}m`,
  );
}
