Jyotisha/frontend/src/lib/personal-report-worker.ts

import "server-only";

import { runConsultationWorkflow } from "@/mastra";
import { createPersonalReportAgent } from "@/mastra/personal-report";
import { loadLanguageModelCatalog } from "@/lib/model-catalog";
import {
  buildReportEvidenceBundleV2,
  generatePersonalReport,
  type SkillSnapshot,
} from "@/lib/personal-report-generation";
import { createSupabasePersonalReportJobService } from "@/lib/personal-report-job-service";
import {
  createPersonalReportWorker,
  PersonalReportWorkerError,
  runPersonalReportWorkerLoop,
  type PersonalReportWorkerGenerationContext,
} from "@/lib/personal-report-worker-core";
import { createSupabasePersonalReportService } from "@/lib/personal-report-service";
import { createPersonalReportSectionService } from "@/lib/personal-report-section-service-core";
import {
  resolveReportBirthClock,
  resolveReportBirthTimeSensitivityInput,
} from "@/lib/personal-report-route-core";
import {
  generatePersonalReportLongform,
  LongformGenerateError,
  type AppendixClient,
} from "@/lib/personal-report-longform-generate";
import { PERSONAL_REPORT_WRITER_ENABLED } from "@/lib/personal-report-writer-flag";
import { loadReportCandidateRange } from "@/lib/report-candidate-range";
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
import { completeUsage, releaseUsage } from "@/lib/consultation-billing";
import type { ConsultationInput } from "@/mastra/consultation-workflow";
import { resolveAyanamsa } from "@/lib/ayanamsa";

const PROFILE_COLUMNS = [
  "name",
  "birth_date",
  "reported_birth_time",
  "active_birth_time",
  "birth_time_source",
  "birth_time_status",
  "rectification_case_id",
  "declared_window_start",
  "declared_window_end",
  "uncertainty_before_minutes",
  "uncertainty_after_minutes",
  "latitude",
  "longitude",
  "timezone_offset",
  "birth_place_label",
  "ayanamsa",
].join(",");

type JsonRecord = Record<string, unknown>;
type WorkerGlobal = typeof globalThis & {
  jyotishaPersonalReportWorker?: PersonalReportWorkerHandle;
};

export type PersonalReportWorkerHandle = Readonly<{
  workerId: string;
  done: Promise<void>;
  stop: () => void;
}>;

function record(value: unknown): JsonRecord | null {
  return value !== null && typeof value === "object" && !Array.isArray(value)
    ? value as JsonRecord
    : null;
}

function text(value: unknown): string | null {
  if (typeof value !== "string") return null;
  const trimmed = value.trim();
  return trimmed.length > 0 ? trimmed : null;
}

function finiteNumber(value: unknown): number | null {
  return typeof value === "number" && Number.isFinite(value) ? value : null;
}

function parseBirthDate(value: unknown): { year: number; month: number; day: number } | null {
  const date = text(value);
  const match = date ? /^(\d{4})-(\d{2})-(\d{2})$/.exec(date) : null;
  if (!match) return null;
  const year = Number.parseInt(match[1], 10);
  const month = Number.parseInt(match[2], 10);
  const day = Number.parseInt(match[3], 10);
  if (year < 1900 || year > 2100 || month < 1 || month > 12 || day < 1 || day > 31) return null;
  return { year, month, day };
}

function skillSnapshotForReport(context: PersonalReportWorkerGenerationContext): SkillSnapshot {
  const { report } = context;
  if (!report.skillName || !report.skillVersion || !report.skillSnapshotSha256) {
    throw new PersonalReportWorkerError("report_schema_invalid", false, "report Skill provenance is incomplete");
  }
  return {
    name: report.skillName,
    version: report.skillVersion,
    sha256: report.skillSnapshotSha256,
    sourceCommit: report.skillSourceCommit,
  };
}

async function generateWriterReport(
  context: PersonalReportWorkerGenerationContext,
  candidateRange: Readonly<{ startTime: string; endTime: string }> | null = null,
) {
  const profile = record(context.profile);
  if (!profile) throw new PersonalReportWorkerError("profile_incomplete", false);

  const birthDate = parseBirthDate(profile.birth_date);
  const usableBirth = resolveReportBirthClock(profile);
  const latitude = finiteNumber(profile.latitude);
  const longitude = finiteNumber(profile.longitude);
  const timezoneOffset = finiteNumber(profile.timezone_offset);
  const birthPlaceLabel = text(profile.birth_place_label) ?? "未知出生地";
  const displayName = text(profile.name) ?? "我的报告";
  if (!usableBirth || !birthDate || latitude === null || longitude === null || timezoneOffset === null) {
    throw new PersonalReportWorkerError("birth_time_not_usable", false);
  }
  const birthClock = usableBirth.clock;
  const birthTimeStatus = usableBirth.status;
  const sensitivityInput = resolveReportBirthTimeSensitivityInput(
    profile,
    birthTimeStatus,
    `${String(birthClock.hour).padStart(2, "0")}:${String(birthClock.minute).padStart(2, "0")}`,
    candidateRange,
  );

  const catalog = await loadLanguageModelCatalog();
  const model = catalog.models.find((entry) => entry.id === catalog.defaultModelId) ?? null;
  if (!model) throw new PersonalReportWorkerError("model_unavailable", true);

  const workflows: { theme: string; workflow: unknown }[] = [];
  for (const rawTheme of context.report.requestedThemes) {
    const input: ConsultationInput = {
      year: birthDate.year,
      month: birthDate.month,
      day: birthDate.day,
      hour: birthClock.hour,
      minute: birthClock.minute,
      lat: latitude,
      lon: longitude,
      tz: timezoneOffset,
      city: birthPlaceLabel,
      ayanamsa: resolveAyanamsa(profile),
      question: `请为个人报告计算 ${rawTheme} 主题证据`,
      theme: rawTheme as ConsultationInput["theme"],
      entryMode: "direct_chart",
      ...sensitivityInput,
    };
    try {
      workflows.push({
        theme: rawTheme,
        workflow: await runConsultationWorkflow(input, { signal: context.signal }),
      });
    } catch (error) {
      if (context.signal.aborted) throw error;
      throw new PersonalReportWorkerError("calculation_unavailable", true);
    }
  }

  const hasUsableBaseChart = workflows.some(({ workflow }) => {
    const workflowRecord = record(workflow);
    return workflowRecord?.success === true && record(workflowRecord.chart) !== null;
  });
  if (!hasUsableBaseChart) {
    throw new PersonalReportWorkerError("calculation_unavailable", true);
  }

  let bundle;
  try {
    bundle = buildReportEvidenceBundleV2({
      workflows,
      subject: {
        displayName,
        birthTimeStatus,
        birthPlaceLabel,
      },
      requestedThemes: context.report.requestedThemes,
      reportType: context.report.reportType,
      presentationMode: context.report.presentationMode,
      skillSnapshot: skillSnapshotForReport(context),
    });
  } catch {
    throw new PersonalReportWorkerError("calculation_unavailable", false);
  }

  const result = await generatePersonalReport({
    reportId: context.report.id,
    bundle,
    depth: context.report.depth,
    agent: createPersonalReportAgent(model),
    signal: context.signal,
    userId: context.report.userId,
    requestId: context.report.requestId,
    sectionService: context.sectionService,
    onProgress: context.onProgress,
  });
  if (result.status !== "ready") return result;
  return {
    ...result,
    usage: {
      inputTokens: result.usage?.inputTokens ?? 0,
      outputTokens: result.usage?.outputTokens ?? 0,
      cache: result.usage?.cache,
      actualModelId: model.id,
      modelConfigVersion: model.configVersion,
    },
  };
}

async function generateLongformReport(
  context: PersonalReportWorkerGenerationContext,
  candidateRange: Readonly<{ startTime: string; endTime: string }> | null,
  admin: AppendixClient,
) {
  const profile = record(context.profile);
  if (!profile) throw new PersonalReportWorkerError("profile_incomplete", false);
  const usableBirth = resolveReportBirthClock(profile);
  const latitude = finiteNumber(profile.latitude);
  const longitude = finiteNumber(profile.longitude);
  const timezoneOffset = finiteNumber(profile.timezone_offset);
  const displayName = text(profile.name) ?? "我的报告";
  if (!usableBirth || latitude === null || longitude === null || timezoneOffset === null) {
    throw new PersonalReportWorkerError("birth_time_not_usable", false);
  }
  const catalog = await loadLanguageModelCatalog();
  const model = catalog.models.find((entry) => entry.id === catalog.defaultModelId) ?? null;
  if (!model) throw new PersonalReportWorkerError("model_unavailable", true);
  try {
    const generated = await generatePersonalReportLongform({
      report: context.report,
      profile,
      candidateRange,
      admin,
      displayName,
      birthTimeStatus: usableBirth.status,
      signal: context.signal,
    });
    if (generated.status !== "ready") return generated;
    return {
      ...generated,
      usage: {
        inputTokens: 0,
        outputTokens: 0,
        actualModelId: model.id,
        modelConfigVersion: model.configVersion,
      },
    };
  } catch (error) {
    if (context.signal.aborted) throw error;
    if (error instanceof PersonalReportWorkerError) throw error;
    if (error instanceof LongformGenerateError) {
      throw new PersonalReportWorkerError(error.code, error.retryable, error.message);
    }
    throw new PersonalReportWorkerError("calculation_unavailable", true);
  }
}

async function generateProductionReport(
  context: PersonalReportWorkerGenerationContext,
  candidateRange: Readonly<{ startTime: string; endTime: string }> | null,
  admin: AppendixClient,
) {
  if (PERSONAL_REPORT_WRITER_ENABLED) {
    return generateWriterReport(context, candidateRange);
  }
  return generateLongformReport(context, candidateRange, admin);
}

function createProductionWorker(workerId: string) {
  const admin = createAdminSupabaseClient();
  const backend = admin as unknown as {
    from(table: string): {
      select(columns: string): {
        eq(column: string, value: unknown): {
          maybeSingle(): PromiseLike<{ data: unknown; error: { message: string } | null }>;
        };
      };
    };
  };
  return createPersonalReportWorker({
    workerId,
    leaseSeconds: 600,
    heartbeatIntervalMs: 20_000,
    jobs: createSupabasePersonalReportJobService(admin),
    reports: createSupabasePersonalReportService(admin),
    sectionService: createPersonalReportSectionService(admin as never),
    billing: {
      complete: async ({ userId, requestId, usage }) => {
        const model = (await loadLanguageModelCatalog()).models.find((entry) => entry.id === usage.actualModelId);
        if (!model) return false;
        const costMicrousd = Math.round((
          usage.inputTokens * (model.inputCostMicrousdPerMillion ?? 0)
          + usage.outputTokens * (model.outputCostMicrousdPerMillion ?? 0)
        ) / 1_000_000);
        const settled = await completeUsage(admin, userId, requestId, {
          eventKey: "report.full", actualModelId: usage.actualModelId,
          modelConfigVersion: usage.modelConfigVersion, inputTokens: usage.inputTokens,
          outputTokens: usage.outputTokens, costMicrousd, durationMs: usage.durationMs,
          ...(usage.cache ? { metadata: { cache: { ...usage.cache, hit: usage.cache.readTokens > 0 } } } : {}),
        });
        return settled.success;
      },
      release: async ({ userId, requestId, reason }) => (await releaseUsage(admin, userId, requestId, reason)).success,
      reserve: async () => { throw new Error("report worker does not reserve usage"); },
    },
    loadProfile: async (userId) => {
      const { data, error } = await backend
        .from("profiles")
        .select(PROFILE_COLUMNS)
        .eq("id", userId)
        .maybeSingle();
      if (error) throw new Error("personal report profile load failed");
      return data ?? null;
    },
    generate: async (context) => {
      const profile = record(context.profile) ?? {};
      if (resolveReportBirthClock(profile)?.status === "confirmed") {
        return generateProductionReport(context, null, admin as never);
      }
      const range = await loadReportCandidateRange(admin, {
        userId: context.report.userId,
        rectificationCaseId: text(profile.rectification_case_id),
      });
      return generateProductionReport(context, range, admin as never);
    },
  });
}

function sanitizedErrorName(error: unknown): string {
  return error instanceof Error ? error.name : "UnknownError";
}

/**
 * Starts one unref'ed loop per Node.js server instance. Database leases remain
 * the cross-instance exclusivity boundary; the global only prevents duplicate
 * loops caused by repeated instrumentation/module evaluation in one process.
 */
export function startPersonalReportWorker(): PersonalReportWorkerHandle {
  const state = globalThis as WorkerGlobal;
  if (state.jyotishaPersonalReportWorker) return state.jyotishaPersonalReportWorker;

  const controller = new AbortController();
  const workerId = `personal-report:${globalThis.crypto.randomUUID()}`;
  let worker: ReturnType<typeof createProductionWorker> | null = null;
  const lazyWorker = {
    tick: async () => {
      worker ??= createProductionWorker(workerId);
      return worker.tick();
    },
  };
  const done = runPersonalReportWorkerLoop(lazyWorker, {
    signal: controller.signal,
    onError: (error) => {
      console.error(`[personal-report-worker] tick failed reason=${sanitizedErrorName(error)}`);
    },
  });
  const handle: PersonalReportWorkerHandle = {
    workerId,
    done,
    stop: () => controller.abort(),
  };
  state.jyotishaPersonalReportWorker = handle;
  return handle;
}