'use node';

import { ConvexError, v } from 'convex/values';

import {
  MAX_TTS_CHUNK_CHARS,
  MIN_TTS_AUDIO_BYTES,
  TTS_FETCH_TIMEOUT_MS,
} from '../../lib/shared/constants/tts';
import { internal } from '../_generated/api';
import type { Id } from '../_generated/dataModel';
import { action } from '../_generated/server';
import { estimateTtsCostCents } from '../governance/cost_estimation';
import { SafeFetchError, safeFetchBinary } from '../lib/http/safe_fetch';
import { rateLimiter } from '../lib/rate_limiter';
import { requireAuthenticatedUser } from '../lib/rls/auth/require_authenticated_user';
import { sanitizeError } from '../lib/utils/sanitize_secrets';
import { checkProviderHostPolicy } from '../providers/file_actions';
import { resolveTtsModel } from '../providers/resolve_model';
import {
  errorCodeFromCaught,
  parseRetryAfterMs,
  TtsProviderHttpError,
  type TtsErrorCode,
} from './error_codes';

const MAX_AUDIO_BYTES = 5 * 1024 * 1024; // 5 MB hard cap on provider response

const AUDIO_MIME_BY_FORMAT: Record<string, string> = {
  mp3: 'audio/mpeg',
  opus: 'audio/ogg; codecs=opus',
  aac: 'audio/aac',
  flac: 'audio/flac',
  wav: 'audio/wav',
  pcm: 'audio/L16; rate=24000',
};

// Error-classification types and helper live in `./error_codes.ts` so
// they can be unit-tested without spinning up a 'use node' action runtime.
// The `TtsErrorCode` and `errorCodeFromCaught` imports above are the
// canonical names; this file no longer redeclares them.

/**
 * Synthesize one sentence/paragraph chunk of an assistant message. Client
 * calls this in order as it segments the streaming text; the action is
 * idempotent on `(messageId, index)` so re-tries and multi-tab races are
 * safe.
 *
 * Auth and quota gating live in `reserveChunk` — that mutation authenticates
 * the caller, asserts thread access, derives the canonical `organizationId`
 * from thread metadata, enforces per-user + per-org rate limits, consults
 * the org budget policy, and caps chunks per message. The action itself only
 * makes the upstream HTTP call once `reserveChunk` returns `'reserved'`.
 *
 * On success, the action also records the synthesis to `usageLedger` so it
 * counts against the org's budget envelope.
 *
 * Stable failure codes stored on the chunk row (`ttsAudioChunks.error`):
 *  - `NO_PROVIDER` / `UNKNOWN_PROVIDER` / `UNKNOWN_MODEL` / `UNKNOWN_VOICE`
 *    — provider configuration issue; client surfaces a settings link.
 *  - `HOST_POLICY` — provider baseUrl rejected by allowlist.
 *  - `RATE_LIMITED` — retry with backoff.
 *  - `BUDGET_EXCEEDED` — terminal; admin must lift budget.
 *  - `PROVIDER_4XX` / `PROVIDER_5XX` / `TIMEOUT` / `PROVIDER_ERROR` —
 *    classified for the client's retry decision.
 */
export const synthesizeChunk = action({
  args: {
    messageId: v.string(),
    threadId: v.string(),
    organizationId: v.string(),
    index: v.number(),
    text: v.string(),
    locale: v.string(),
  },
  returns: v.object({
    status: v.union(
      v.literal('ready'),
      v.literal('in-flight'),
      v.literal('failed'),
    ),
    errorCode: v.optional(v.string()),
    // Optional provider-supplied retry hint (ms). Returned alongside the
    // action result so the chunker's retry-loop can honor server timing
    // (Retry-After on 429, rate-limiter retryAfter). Not persisted on the
    // chunk row — the row's `error` enum is enough for fan-out subscribers;
    // retry timing matters only to the originating caller.
    retryAfterMs: v.optional(v.number()),
  }),
  handler: async (ctx, args) => {
    const user = await requireAuthenticatedUser(ctx);
    const text = args.text.trim();
    if (text.length === 0) {
      throw new ConvexError({
        code: 'TTS_EMPTY_TEXT',
        message: 'Chunk text is empty after trim.',
      });
    }
    if (text.length > MAX_TTS_CHUNK_CHARS) {
      throw new ConvexError({
        code: 'TTS_TEXT_TOO_LONG',
        message: `Chunk text exceeds ${MAX_TTS_CHUNK_CHARS} characters; client must re-segment.`,
      });
    }

    // Cross-org probe gate: refuse non-members before any provider config
    // is read. Without this, moving `resolveTtsModel` ahead of `reserveChunk`
    // (so the prospective-cost check can use the real per-model rate)
    // would let a logged-in user learn the TTS provider/model wired up to
    // an org they don't belong to. reserveChunk's `assertThreadAccess`
    // still runs later as the authoritative thread-membership check.
    await ctx.runQuery(
      internal.governance.internal_mutations.requireOrganizationMemberInternal,
      {
        organizationId: args.organizationId,
        userId: user.userId,
        email: user.email,
        name: user.name,
      },
    );

    // Resolve model up front so `reserveChunk`'s prospective-cost budget
    // check can use the real per-model `centsPerMillionCharacters` rate.
    // Without this, parallel chunks of one message can each pass a static
    // 1500 ¢/M-char gate and collectively overshoot the org's hard cap
    // when the active provider charges more (e.g. premium ElevenLabs
    // tier). A pre-reservation resolver failure (no provider, unknown
    // voice, etc.) returns a synthetic `failed` result with the
    // classified code — there's no chunk row to mark yet.
    let modelData;
    try {
      modelData = await resolveTtsModel(ctx, {
        organizationId: args.organizationId,
        locale: args.locale,
      });
    } catch (err) {
      const { code } = errorCodeFromCaught(err);
      // No chunk row exists yet, so this failure cannot be persisted on
      // `ttsAudioChunks.error`. Without an operator-visible log,
      // misconfigured providers (NO_PROVIDER / UNKNOWN_PROVIDER /
      // UNKNOWN_MODEL / UNKNOWN_VOICE / HOST_POLICY) only surface via the
      // client indicator — which itself depends on the client routing
      // `{status:'failed'}` to the error sink. Warn here so backend logs
      // independently record the failure.
      console.warn('[tts.synthesizeChunk] pre-reservation failure', {
        organizationId: args.organizationId,
        locale: args.locale,
        code,
        detail: sanitizeError(err),
      });
      return { status: 'failed' as const, errorCode: code };
    }

    // `reserveChunk` throws forbidden / rate-limit / budget / chunk-limit
    // errors. These are not chunk-row-recoverable (we don't own a chunkId
    // yet), so we don't catch them — the action surfaces them to the client
    // as a hard failure distinct from a chunk-level `failed` row, which
    // lets the UI render the right recovery affordance.
    //
    // Exception: an OptimisticConcurrencyControlFailure on the rate-limiter
    // shard can leak through when many concurrent callers contend on the
    // same shard and the library's internal OCC retries exhaust. The chunk
    // wasn't reserved, so we have no row to mark — surface as `CONTENTION`
    // (distinct from `RATE_LIMITED`) so the client backs off with the
    // short OCC jitter, not the full quota-recovery delay.
    const reservation = await ctx
      .runMutation(internal.tts.mutations.reserveChunk, {
        messageId: args.messageId,
        threadId: args.threadId,
        organizationId: args.organizationId,
        index: args.index,
        text,
        locale: args.locale,
        prospectiveCostCentsPerMChars: modelData.centsPerMillionCharacters,
      })
      .catch((err: unknown): { __occ: true } => {
        // OCC detection layered for robustness across error-shape reserializations:
        // (1) ConvexError data.code (preferred, set by rate-limiter library)
        // (2) constructor name (works in-process, lost after action→mutation re-serialization)
        // (3) message regex (defense-in-depth for serialized error shapes)
        const isOCC =
          (err instanceof ConvexError &&
            // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- ConvexError.data is `any`; defensively read `code`.
            (err.data as { code?: string } | undefined)?.code === 'OCC') ||
          (err instanceof Error &&
            err.constructor?.name === 'OptimisticConcurrencyControlFailure') ||
          (err instanceof Error &&
            /OptimisticConcurrencyControlFailure/.test(err.message));
        if (isOCC) {
          return { __occ: true };
        }
        throw err;
      });
    if ('__occ' in reservation) {
      return { status: 'failed' as const, errorCode: 'CONTENTION' };
    }

    if (reservation.kind === 'ready') {
      return { status: 'ready' as const };
    }
    if (reservation.kind === 'pending-in-flight') {
      return { status: 'in-flight' as const };
    }
    const chunkId = reservation.chunkId;
    const attemptCreatedAt = reservation.attemptCreatedAt;

    // Helper: mark this attempt failed with `code`. The mutation's identity
    // check refuses if the row was already overwritten by a fresher attempt,
    // so a slow failure from a stale attempt can't trample a new pending row.
    // `retryAfterMs` is returned (not persisted) so the chunker's retry-loop
    // honors the provider's Retry-After hint instead of its default backoff.
    //
    // When the mutation reports `stale: true`, the fresher attempt owns the
    // row now — surface `in-flight` so the client doesn't render a failure
    // bound to this attempt's identity. Mirrors the parallel stale path of
    // `markChunkReadyAndRecordUsage` (synthesize.ts:375-380).
    const markFailedAndReturn = async (
      code: TtsErrorCode,
      retryAfterMs?: number,
    ) => {
      const result = await ctx.runMutation(
        internal.tts.mutations.markChunkFailed,
        {
          chunkId,
          attemptCreatedAt,
          error: code,
        },
      );
      if (result.stale) {
        return { status: 'in-flight' as const };
      }
      return { status: 'failed' as const, errorCode: code, retryAfterMs };
    };

    // Defense-in-depth: re-check host policy at synthesis time so a provider
    // file edited to point at an internal host (e.g. metadata service) cannot
    // exfiltrate the bearer token, even when the file passed validation at
    // load time.
    try {
      checkProviderHostPolicy(modelData.baseUrl);
    } catch (err) {
      const { code } = errorCodeFromCaught(err);
      // Host-policy rejections are operator-actionable (someone changed
      // a provider file to point at a non-allowlisted host). Log so the
      // operator can correlate; the indicator's `config` branch will
      // surface a settings link to the user.
      console.warn('[tts.synthesizeChunk] host policy rejected provider', {
        organizationId: args.organizationId,
        locale: args.locale,
        baseUrl: modelData.baseUrl,
        code,
        detail: sanitizeError(err),
      });
      return markFailedAndReturn(code);
    }

    const url = `${modelData.baseUrl.replace(/\/+$/, '')}/audio/speech`;
    const mime =
      AUDIO_MIME_BY_FORMAT[modelData.audioFormat] ?? 'application/octet-stream';
    let storageId: Id<'_storage'>;
    let providerBytes = 0;
    const t0 = Date.now();
    try {
      // `safeFetchBinary` enforces the size cap during the streaming read,
      // so a chunked-transfer response can't buffer past MAX_AUDIO_BYTES
      // before the size check fires. It also re-validates the host on every
      // redirect hop so a 302 to the cloud metadata service can't smuggle
      // the bearer token.
      const response = await safeFetchBinary(url, {
        method: 'POST',
        headers: {
          Authorization: `Bearer ${modelData.apiKey}`,
          'Content-Type': 'application/json',
          // Pin content-type expectations: OpenAI-compatible gateways that
          // content-negotiate could otherwise return a JSON error envelope
          // under a 200 OK that would pass the streaming size cap and
          // store as "audio". `audio/*` matches every supported format.
          Accept: 'audio/*',
        },
        body: JSON.stringify({
          model: modelData.modelId,
          input: text,
          voice: modelData.voice,
          response_format: modelData.audioFormat,
          ...(modelData.instructions
            ? { instructions: modelData.instructions }
            : {}),
        }),
        timeoutMs: TTS_FETCH_TIMEOUT_MS,
        maxResponseBytes: MAX_AUDIO_BYTES,
        defaultContentType: mime,
      });
      if (response.status < 200 || response.status >= 300) {
        // Don't log the provider body — providers commonly echo input PII or
        // sensitive config in 4xx bodies. Surface status + origin only.
        const origin = (() => {
          try {
            return new URL(response.finalUrl).origin;
          } catch (err) {
            console.debug(
              '[tts] response.finalUrl unparseable; falling back to "unknown"',
              err,
            );
            return 'unknown';
          }
        })();
        const retryAfterMs = parseRetryAfterMs(
          response.headers.get('Retry-After'),
        );
        console.warn('[tts] provider error', {
          status: response.status,
          origin,
          retryAfterMs,
          chunkId,
          organizationId: args.organizationId,
          modelId: modelData.modelId,
          providerName: modelData.providerName,
          charCount: text.length,
          durationMs: Date.now() - t0,
        });
        throw new TtsProviderHttpError(
          response.status,
          retryAfterMs,
          `TTS API ${response.status}: provider call failed`,
        );
      }
      // Empty / near-empty 200 responses (provider misconfiguration,
      // upstream JSON-error-as-200, partial outage) would otherwise be
      // stored as a zero-byte blob and fully billed via the ledger,
      // yielding no audible audio. Refuse before the storage write so
      // the chunk is marked failed and never debited.
      if (response.body.size < MIN_TTS_AUDIO_BYTES) {
        console.warn('[tts] provider returned implausibly small body', {
          bytes: response.body.size,
          status: response.status,
        });
        throw new SafeFetchError(
          'response_too_small',
          `Provider returned ${response.body.size} bytes (< ${MIN_TTS_AUDIO_BYTES}); refusing to bill for empty audio`,
          response.status,
        );
      }
      providerBytes = response.body.size;
      // Only re-wrap when the provider omitted Content-Type entirely. The
      // double-wrap (`new Blob([blob])`) would otherwise force a copy of the
      // body buffer just to override `.type`, which we don't need when the
      // existing type is already set.
      const typedBlob =
        response.body.type && response.body.type !== ''
          ? response.body
          : new Blob([response.body], { type: mime });
      storageId = await ctx.storage.store(typedBlob);
    } catch (err) {
      const classified = errorCodeFromCaught(err);
      return markFailedAndReturn(classified.code, classified.retryAfterMs);
    }

    // Compensating storage delete on post-store failure. Without this, any
    // throw from `markChunkReadyAndRecordUsage` (cascade-deleted row,
    // identity mismatch from a stale attempt, validator rejection, etc.)
    // would orphan the just-uploaded blob until the 7-day cron eventually
    // sweeps it. The mutation itself handles the identity-mismatch case
    // inline; this catch covers everything else.
    const costEstimateCents = estimateTtsCostCents(
      text.length,
      modelData.centsPerMillionCharacters,
    );
    try {
      const result = await ctx.runMutation(
        internal.tts.mutations.markChunkReadyAndRecordUsage,
        {
          chunkId,
          attemptCreatedAt,
          storageId,
          voice: modelData.voice,
          providerName: modelData.providerName,
          modelId: modelData.modelId,
          format: modelData.audioFormat,
          characterCount: text.length,
          costEstimateCents,
        },
      );
      if (result.stale) {
        // The mutation already deleted the blob inline. Surface as in-flight
        // so the client doesn't render an error — the fresher attempt owns
        // the row now.
        return { status: 'in-flight' as const };
      }
    } catch (err) {
      try {
        await ctx.storage.delete(storageId);
      } catch (deleteErr) {
        console.warn(
          '[tts] failed to delete orphan blob on mark-ready throw',
          deleteErr,
        );
      }
      console.warn(
        '[tts] markChunkReadyAndRecordUsage threw',
        sanitizeError(err),
      );
      const classified = errorCodeFromCaught(err);
      return markFailedAndReturn(classified.code, classified.retryAfterMs);
    }

    // Capacity-planning telemetry: emit a single info line per successful
    // synth so operators can build p95 latency / bytes / cost dashboards.
    // No PII (we log `charCount` length, never `text` content).
    console.info('[tts] synth ok', {
      chunkId,
      organizationId: args.organizationId,
      providerName: modelData.providerName,
      modelId: modelData.modelId,
      charCount: text.length,
      bytes: providerBytes,
      durationMs: Date.now() - t0,
    });
    return { status: 'ready' as const };
  },
});

/**
 * Capability check the client uses to gate the TTS toggle and the
 * personalization-page voice section. Returns the configured TTS model
 * summary or `{ available: false }` when no provider has a `'text-to-speech'`
 * model for this org.
 *
 * Auth: requires the caller to be a current member of `organizationId`. The
 * provider/model identifiers in the success response are operationally
 * sensitive (they reveal which third-party vendor an org has wired up), so
 * a bare `requireAuthenticatedUser` check is insufficient — without the
 * membership gate, any logged-in user could probe arbitrary org IDs.
 *
 * NB: this is an `action` purely because `resolveTtsModel` runs in a Node
 * action. Phase 5 (deferred) converts it to a query once the resolver
 * dependency is unwound.
 */
export const getCapability = action({
  args: { organizationId: v.string() },
  returns: v.object({
    available: v.boolean(),
    providerName: v.optional(v.string()),
    modelId: v.optional(v.string()),
  }),
  handler: async (ctx, args) => {
    const user = await requireAuthenticatedUser(ctx);
    // Per-user rate-limit gate runs BEFORE the membership check so a
    // logged-in user can't probe arbitrary `organizationId`s to fill the
    // target org's audit log via `logCapabilityProbeDenied`. The
    // personalization UI calls this once per mount; 12/min/user is
    // generous for human use and tight for scripted probes.
    const probeLimit = await rateLimiter.limit(
      ctx,
      'tts:capability-probe:user',
      {
        key: user.userId,
        throws: false,
      },
    );
    if (!probeLimit.ok) {
      throw new ConvexError({
        code: 'RATE_LIMITED',
        message: 'Capability probe rate limit exceeded.',
        retryAfter: probeLimit.retryAfter,
      });
    }
    // Cross-org probe gate: throws if the caller is not a member of the
    // requested org. The error propagates to the client — the toggle's
    // capability-check UI treats it as "unavailable" without distinguishing
    // "not configured" from "not allowed", so probing reveals nothing.
    try {
      await ctx.runQuery(
        internal.governance.internal_mutations
          .requireOrganizationMemberInternal,
        {
          organizationId: args.organizationId,
          userId: user.userId,
          email: user.email,
          name: user.name,
        },
      );
    } catch (err) {
      // Record the deny as a security signal so operators can correlate
      // repeated probes against a specific actor. Best-effort.
      await ctx.runMutation(internal.tts.mutations.logCapabilityProbeDenied, {
        organizationId: args.organizationId,
        actorId: user.userId,
        actorEmail: user.email,
      });
      throw err;
    }
    try {
      const model = await resolveTtsModel(ctx, {
        organizationId: args.organizationId,
        locale: 'en',
      });
      return {
        available: true,
        providerName: model.providerName,
        modelId: model.modelId,
      };
    } catch (err) {
      // `UNKNOWN_VOICE` means the TTS model resolved but neither
      // `voicesByLocale['en']` nor `defaultVoice` matched. An org that
      // wired up `voicesByLocale: { de: ... }` only is still capable —
      // synthesis will pick a real voice from the caller's actual locale
      // at synth time. Capability probe should report `available: true`
      // so the settings UI doesn't render the "provider unavailable"
      // banner. We can't fill in providerName/modelId without re-running
      // the resolver, which is acceptable — those fields are
      // informational on the capability response.
      if (err instanceof Error && err.message.startsWith('UNKNOWN_VOICE:')) {
        return { available: true };
      }
      console.warn(
        '[tts.getCapability] resolveTtsModel failed',
        sanitizeError(err),
      );
      return { available: false };
    }
  },
});
