/**
 * native-flow-r2v.ts — first-class Google Flow Reference-to-Video (R2V) scene
 * render via saved **Flow Characters**, the proven character-ad workflow.
 *
 * Unlike the produce/execute path (which routes through the bun `vclaw-cli`
 * flow.ts and a full readiness/storyboard ceremony), this is a lightweight,
 * pure-Node fetch transport — the same shape as {@link ./native-runway} /
 * {@link ./native-dreamina} — for rendering ONE scene directly from character
 * references:
 *
 *   POST {BASE}/google-flow/videos   { email, prompt, model, aspectRatio,
 *     duration, async, character_1..7 }            → { jobid }
 *   GET  {BASE}/google-flow/jobs/{jobid}  (jobid VERBATIM — encoding → 400)
 *     → { status: started|completed|failed, ... videoUrl }
 *
 * R2V constraints (vclaw-cli/src/backends/types.ts, live-proven 2026-06-11):
 *   - model MUST be `veo-3.1-fast` (veo-3.1-quality rejects character_*).
 *   - character refs only (character_1..7) — NO startImage (can't mix R2V + I2V).
 *   - duration is 8s ONLY (Veo R2V has no other length) — see FLOW_R2V_DURATIONS.
 *   - Veo generates the character's voice + lip-sync NATIVELY from the prompt.
 *   - Flow burst-throttles with a 403 reCAPTCHA (PUBLIC_ERROR_UNUSUAL_ACTIVITY);
 *     this transport cools down and retries the SUBMIT for that case only.
 *
 * Transports (fetch + sleep) are injectable so the whole module is unit-testable
 * offline with no network and no real waiting.
 */

import { existsSync } from 'node:fs';
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { dirname, join } from 'node:path';
import { resolveFlowCaptchaRetry, applyFlowCaptcha } from './flow-captcha.js';
import { recoverFlowVideoBytes, flowRawAssetUrl, isFlowMediaSuccessful, type FlowMediaFetchLike } from './flow-media-url.js';
import { finishFlowClip, flowUpscaleEnabled, resolveAutomaticFlowFinishResolution, type FlowUpscaleFinishResult } from './flow-upscale-finish.js';
import type { GoogleFlowUpscaleResolution } from './providers/google-flow.js';

export const FLOW_BASE = 'https://api.useapi.net/v1';
export const FLOW_R2V_MODEL = 'veo-3.1-fast';
/**
 * Durations this route can actually generate. **8 seconds only.**
 *
 * The API's duration table (spec 2026-08-28, POST /videos › Model Capabilities)
 * gives Veo T2V/I2V `4|6|8` but Veo **R2V `8` only** — and character refs, which
 * are what makes this an R2V request, are likewise 8-s-only on Veo. We advertised
 * `4|6|8|10` and forwarded them, so a `--duration 4` here was a request the
 * provider was always going to refuse. Narrowed rather than silently coerced: an
 * operator who asked for 4 s should be told the route cannot do it, not handed a
 * clip twice the length they planned the edit around.
 *
 * (`10` was never Veo's at all — it is an omni-flash duration, and omni-flash
 * does not take `character_*` on this route.)
 */
export const FLOW_R2V_DURATIONS = new Set([8]);
const RECAPTCHA_RE = /recaptcha|unusual_activity|permission_denied/i;

export interface FlowFetchResponse {
  ok: boolean;
  status: number;
  text(): Promise<string>;
  arrayBuffer(): Promise<ArrayBuffer>;
  /** Optional so test fakes can omit it; real fetch always provides it. */
  headers?: { get(name: string): string | null };
}

/**
 * Server-stated throttle cooldown in ms, or null. Prefers the `Retry-After`
 * header (seconds or an HTTP/ISO date), then the response body's `retryAfter`
 * field (legacy seconds, or — current useapi shape since 2026-06-15 — an ISO
 * timestamp string). Honoring this waits the REAL window the load balancer
 * reports instead of a fixed guess.
 */
export function flowRetryAfterMs(
  res: { headers?: { get(name: string): string | null } },
  parsedBody: Record<string, unknown>,
): number | null {
  const header = res.headers?.get?.('retry-after');
  if (header) {
    const secs = Number(header);
    if (Number.isFinite(secs) && secs >= 0) return secs * 1000;
    const ms = Date.parse(header);
    if (!Number.isNaN(ms)) { const d = ms - Date.now(); if (d > 0) return d; }
  }
  const field = parsedBody.retryAfter;
  if (typeof field === 'number' && Number.isFinite(field) && field >= 0) return field * 1000;
  if (typeof field === 'string') {
    const ms = Date.parse(field);
    if (!Number.isNaN(ms)) { const d = ms - Date.now(); if (d > 0) return d; }
  }
  return null;
}
export type FlowFetchLike = (
  input: string,
  init?: { method?: string; headers?: Record<string, string>; body?: string },
) => Promise<FlowFetchResponse>;

export interface FlowR2vOptions {
  /** Scene prompt (dialogue goes inline; prompt hygiene is applied by the caller). */
  prompt: string;
  /** Resolved character refs → character_1..7 (1–7; the person + optional product). */
  characterRefs: string[];
  outputPath: string;
  /** 8 seconds — the only length Veo reference-to-video generates (default 8). */
  durationSeconds?: number;
  /** 'landscape' | 'portrait' (default 'landscape'). */
  aspectRatio?: 'landscape' | 'portrait';
  env?: NodeJS.ProcessEnv;
  fetchImpl?: FlowFetchLike;
  sleepImpl?: (ms: number) => Promise<void>;
  /**
   * Captcha-retry count sent on the submit (`captchaRetry`, auto-solve). Omit to
   * resolve from VCLAW_FLOW_CAPTCHA_RETRY (default 5); 0 opts out. This rides
   * through the intermittent 403 reCAPTCHA inline rather than relying solely on
   * the cooldown-resubmit fallback below.
   */
  captchaRetry?: number;
  /**
   * Run the free Google Flow 1080p upscale on the finished clip. Omit to resolve
   * from VCLAW_FLOW_UPSCALE (ON unless explicitly disabled). A failed upscale is
   * never fatal — the generated clip stands.
   */
  upscale?: boolean;
  /** Upscale target (default 1080p, free). '4K' costs 50 credits and needs Ultra. */
  upscaleResolution?: GoogleFlowUpscaleResolution;
  /** Cooldown-retries for a 403 reCAPTCHA / 429 throttle on SUBMIT (default 2). */
  maxRecaptchaRetries?: number;
  /** Cooldown between throttle retries, ms, when the server states no window (default 5 min). */
  recaptchaCooldownMs?: number;
  /**
   * Max in-process wait for a server-stated Retry-After window (default 10 min).
   * A longer quarantine (e.g. the ~30-min USER_QUOTA window) fails fast with the
   * real wait instead of blocking the process for it.
   */
  maxThrottleWaitMs?: number;
  pollIntervalMs?: number;
  maxPolls?: number;
  /**
   * Ceiling on waiting for a missing download link to resolve after the render
   * COMPLETED (default 3 min). Deliberately its own budget, separate from the
   * generation poll budget above: a Google link block can last hours, and the
   * `?raw=true` route works throughout, so there is no reason to sit on it.
   */
  urlRecoveryMaxWaitMs?: number;
}

export interface FlowR2vResult {
  jobId: string;
  outputPath: string;
  videoUrl: string;
  characterCount: number;
  model: string;
  durationSeconds: number;
  aspectRatio: string;
  /**
   * What the free Flow finish did. `upscaled: false` with a `skippedReason` is a
   * normal, non-failing outcome — read it rather than assuming every clip rose
   * to 1080p, or a stitch will silently mix resolutions.
   */
  upscale: FlowUpscaleFinishResult;
}

function readDotEnvLike(raw: string): Record<string, string> {
  const out: Record<string, string> = {};
  for (const line of raw.split('\n')) {
    const m = line.match(/^\s*([A-Z0-9_]+)\s*=\s*(.*)\s*$/);
    if (m) out[m[1]] = m[2].replace(/^['"]|['"]$/g, '').trim();
  }
  return out;
}

async function resolveEnv(env: NodeJS.ProcessEnv, workspaceRoot?: string): Promise<NodeJS.ProcessEnv> {
  if (workspaceRoot) {
    const envLocal = join(workspaceRoot, '.env.local');
    if (existsSync(envLocal)) return { ...readDotEnvLike(await readFile(envLocal, 'utf-8')), ...env };
  }
  return env;
}

const defaultSleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));

/**
 * Render one R2V scene from saved Flow character refs and download it to
 * `outputPath`. Throws a clear error on a non-throttle failure; cools down and
 * retries the submit on a 403 reCAPTCHA throttle.
 */
export async function submitFlowR2vNative(
  options: FlowR2vOptions & { workspaceRoot?: string },
): Promise<FlowR2vResult> {
  const refs = options.characterRefs.filter((r) => r && r.trim()).slice(0, 7);
  if (refs.length === 0) {
    throw new Error('flow-r2v requires at least one character reference (the person and/or product).');
  }
  const duration = options.durationSeconds ?? 8;
  if (!FLOW_R2V_DURATIONS.has(duration)) {
    throw new Error(
      `flow-r2v duration must be ${[...FLOW_R2V_DURATIONS].join(', ')} (got ${duration}). ` +
        `Veo reference-to-video generates 8s only — 4/6 are text/image-to-video durations and 10 is omni-flash, ` +
        `which does not take character refs on this route.`,
    );
  }
  const aspectRatio = options.aspectRatio ?? 'landscape';
  const env = await resolveEnv(options.env ?? process.env, options.workspaceRoot);
  const token = env.USEAPI_API_TOKEN?.trim();
  const email = env.USEAPI_ACCOUNT_EMAIL?.trim();
  if (!token) throw new Error('flow-r2v requires USEAPI_API_TOKEN in the environment.');
  if (!email) throw new Error('flow-r2v requires USEAPI_ACCOUNT_EMAIL in the environment.');

  // Resolve the finish settings BEFORE the submit. An unparseable
  // VCLAW_FLOW_UPSCALE_RESOLUTION throws, and thrown after the render it would
  // kill the call on a clip that had already been generated, charged and
  // written — which reads to a driver as a failed scene worth re-rendering.
  const upscaleEnabled = options.upscale ?? flowUpscaleEnabled(env);
  const upscaleResolution = options.upscaleResolution ?? resolveAutomaticFlowFinishResolution(env);

  const fetchImpl = options.fetchImpl ?? (fetch as unknown as FlowFetchLike);
  const sleep = options.sleepImpl ?? defaultSleep;
  const headers = { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' };

  const body: Record<string, unknown> = {
    email,
    prompt: options.prompt,
    model: FLOW_R2V_MODEL,
    aspectRatio,
    duration,
    async: true,
  };
  refs.forEach((ref, i) => { body[`character_${i + 1}`] = ref; });
  // Auto-solve the reCAPTCHA inline (cycles useapi's configured providers / free
  // credits) so the burst-throttle 403 is far rarer; the cooldown loop below is
  // the residual fallback.
  applyFlowCaptcha(body, options.captchaRetry ?? resolveFlowCaptchaRetry(env));

  // --- submit (with bounded throttle cooldown-retry) ---
  const maxRetries = options.maxRecaptchaRetries ?? 2;
  const cooldownMs = options.recaptchaCooldownMs ?? 5 * 60 * 1000;
  const maxThrottleWaitMs = options.maxThrottleWaitMs ?? 10 * 60 * 1000;
  let jobId = '';
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    const res = await fetchImpl(`${FLOW_BASE}/google-flow/videos`, { method: 'POST', headers, body: JSON.stringify(body) });
    const text = await res.text();
    if (res.ok) {
      const parsed = safeJson(text);
      jobId = String(parsed.jobid ?? parsed.jobId ?? (parsed.job as Record<string, unknown>)?.jobid ?? parsed.id ?? '');
      if (!jobId) throw new Error(`flow-r2v submit returned no jobId: ${text.slice(0, 200)}`);
      break;
    }
    // Throttle (403 reCAPTCHA burst, or 429 quota/traffic) → cool down + retry
    // the submit (NOT a content reject). Honor the server-stated Retry-After
    // window when present; otherwise use the fixed cooldown.
    const isThrottle = res.status === 429 || (res.status === 403 && RECAPTCHA_RE.test(text));
    if (isThrottle && attempt < maxRetries) {
      const serverMs = flowRetryAfterMs(res, safeJson(text));
      if (serverMs !== null && serverMs > maxThrottleWaitMs) {
        throw new Error(
          `flow-r2v submit throttled (HTTP ${res.status}): server quarantine window is ~${Math.round(serverMs / 60_000)} min — ` +
          `retry after that, or add accounts / wait for the quota to clear.`,
        );
      }
      await sleep(serverMs ?? cooldownMs);
      continue;
    }
    throw new Error(`flow-r2v submit failed (HTTP ${res.status}): ${text.slice(0, 300)}`);
  }

  // --- poll (jobId VERBATIM) ---
  const interval = options.pollIntervalMs ?? 12000;
  const maxPolls = options.maxPolls ?? 90;
  let videoUrl = '';
  let mediaGenerationId = '';
  let completed = false;
  for (let i = 0; i < maxPolls; i++) {
    const res = await fetchImpl(`${FLOW_BASE}/google-flow/jobs/${jobId}`, { method: 'GET', headers });
    const text = await res.text();
    const parsed = safeJson(text);
    const status = String(parsed.status ?? '').toLowerCase();
    if (status === 'completed') {
      completed = true;
      videoUrl = extractVideoUrl(parsed, text);
      mediaGenerationId = extractMediaGenerationId(parsed);
      break;
    }
    if (status === 'failed') {
      const detail = String(parsed.error ?? '') + ' ' + JSON.stringify(parsed.errorDetails ?? parsed.raw ?? '').slice(0, 200);
      throw new Error(`flow-r2v generation failed: ${detail.trim()}`);
    }
    if (i < maxPolls - 1) await sleep(interval);
  }
  if (!completed) throw new Error('flow-r2v poll timed out before completion.');

  // --- download ---
  // A completed job can carry NO videoUrl: since 2026-07-27 useapi omits the
  // field rather than returning a dead link whenever Google's signed-URL
  // endpoint is rate-limiting its outbound calls. The render succeeded and was
  // charged, so recover it by mediaGenerationId instead of failing the scene.
  await mkdir(dirname(options.outputPath), { recursive: true });
  let bytes: Buffer;
  if (videoUrl) {
    const dl = await fetchImpl(videoUrl, { method: 'GET' });
    if (!dl.ok) throw new Error(`flow-r2v download failed (HTTP ${dl.status}).`);
    bytes = Buffer.from(await dl.arrayBuffer());
  } else {
    if (!mediaGenerationId) {
      throw new Error(
        `flow-r2v job ${jobId} completed but carried neither a videoUrl nor a mediaGenerationId to recover it by.`,
      );
    }
    const recovered = await recoverFlowVideoBytes({
      mediaGenerationId,
      apiToken: token,
      fetchImpl: fetchImpl as unknown as FlowMediaFetchLike,
      sleepImpl: sleep,
      ...(options.urlRecoveryMaxWaitMs !== undefined ? { maxWaitMs: options.urlRecoveryMaxWaitMs } : {}),
    }).catch((err: unknown) => {
      // Name the id: the clip exists on Google's side and can be pulled by hand
      // once the block clears, which beats paying to re-render it.
      throw new Error(
        `flow-r2v job ${jobId} completed but its download link is unavailable and recovery failed ` +
        `(mediaGenerationId ${mediaGenerationId}): ${err instanceof Error ? err.message : String(err)}`,
      );
    });
    bytes = recovered.bytes;
    videoUrl = recovered.url ?? flowRawAssetUrl(mediaGenerationId);
  }
  await writeFile(options.outputPath, bytes);

  // --- free 1080p finish ---
  // The clip is on disk and paid for; the Flow upsampler is free at 1080p and
  // only works while we still hold the mediaGenerationId, so this is the moment.
  // It NEVER fails the render: on any error the generated clip stands.
  const upscale = await finishFlowClip({
    mediaGenerationId,
    outputPath: options.outputPath,
    apiToken: token,
    enabled: upscaleEnabled,
    resolution: upscaleResolution,
    fetchImpl: fetchImpl as unknown as FlowMediaFetchLike,
    sleepImpl: sleep,
  });

  return {
    jobId,
    outputPath: options.outputPath,
    videoUrl,
    characterCount: refs.length,
    model: FLOW_R2V_MODEL,
    durationSeconds: duration,
    aspectRatio,
    upscale,
  };
}

function safeJson(text: string): Record<string, unknown> {
  try { return JSON.parse(text) as Record<string, unknown>; } catch { return {}; }
}

function extractVideoUrl(parsed: Record<string, unknown>, rawText: string): string {
  const response = parsed.response as Record<string, unknown> | undefined;
  const media = (response?.media ?? parsed.media) as Array<Record<string, unknown>> | undefined;
  const fromMedia = media?.[0]?.videoUrl;
  if (typeof fromMedia === 'string' && fromMedia) return fromMedia;
  const scan = rawText.match(/https:\/\/[^"]+\.mp4[^"]*/);
  return scan ? scan[0] : '';
}

/**
 * `response.media[].mediaGenerationId` — always present on a completed job, and
 * the only handle on the clip when `videoUrl` was withheld.
 *
 * Returns '' for a media item whose own status reports a FAILED generation: that
 * item never had a URL and never will, so sending it down the recovery path
 * would burn the whole wait budget before failing anyway.
 */
export function extractMediaGenerationId(parsed: Record<string, unknown>): string {
  const response = parsed.response as Record<string, unknown> | undefined;
  const media = (response?.media ?? parsed.media) as Array<Record<string, unknown>> | undefined;
  const item = media?.[0];
  if (!item || !isFlowMediaSuccessful(item)) return '';
  const id = item.mediaGenerationId;
  return typeof id === 'string' ? id : '';
}

// ─── prompt hygiene (the dry-voice + no-baked-music standing rules) ──────────

export interface R2vPromptHygieneOptions {
  /** Keep Veo's generated music in the clip (default false → suppress it). */
  keepMusic?: boolean;
  /** Allow room reverb on the voice (default false → dry close-mic). */
  allowReverb?: boolean;
}

/**
 * Append the standing character-ad audio directives to a scene prompt, unless
 * opted out. Suppressing Veo-generated music keeps the clip audio voice-only so
 * a post bed can sit under it without clashing; the dry close-mic directive
 * fixes Veo's default echoey delivery. Idempotent-ish (only appends when not
 * already present). Pure.
 */
export function applyR2vPromptHygiene(prompt: string, options: R2vPromptHygieneOptions = {}): string {
  const clauses: string[] = [];
  if (!options.allowReverb && !/no echo|no reverb|close-mic|close-microphone/i.test(prompt)) {
    clauses.push('Any spoken voice is clean, dry and intimate, recorded close-microphone with no echo and no reverb.');
  }
  if (!options.keepMusic && !/no music|no instrumental|no soundtrack/i.test(prompt)) {
    clauses.push('No music, no instrumental, no background soundtrack, no sound effects, and no ambient or room tone in the audio — only his clean, isolated spoken voice in a silent studio.');
  }
  return clauses.length > 0 ? `${prompt.trim()} ${clauses.join(' ')}` : prompt;
}