/**
 * Adapt between wire-shaped I/O ops and user-handler-shaped I/O ops.
 *
 *   client → server (write):  WriteArgs (wire) → UserWriteArgs (handler)
 *   server → client (read):   UserReadResult (handler) → ReadResult (wire)
 *   server → client (thumb):  UserThumbnailResult → GetThumbnailResult
 *
 * The streams-bridge is consulted whenever we're crossing the > 256 KiB
 * boundary: outbound reads kick a new transfer; inbound writes await one.
 */

import { sha256 } from '@noble/hashes/sha2.js';
import type {
  GetThumbnailResult,
  ReadArgs,
  ReadResult,
  WriteArgs,
} from '../schemas/ops.js';
import { base64ToBytes, bytesToBase64, bytesToHex } from '../protocol/canonical.js';
import { INLINE_THRESHOLD } from '../client/inline-threshold.js';
import {
  ConflictError,
  InternalFileError,
  InvalidArgsError,
} from '../schemas/errors.js';
import type {
  UserReadResult,
  UserThumbnailResult,
  UserWriteArgs,
} from './io-types.js';
import type { ServerStreamsBridge } from './streams-bridge.js';

export interface IoAdapterDeps {
  streamsBridge: ServerStreamsBridge | undefined;
  sender: string;
  signal: AbortSignal;
  /** Hard deadline for streams-bridge awaits / outbound transfers. */
  ioTimeoutMs: number;
}

// ─── write: wire args → user args ────────────────────────────

/**
 * Convert validated `WriteArgs` (wire shape) into `UserWriteArgs`. For
 * streams writes, this awaits the matching inbound transfer via the
 * streams-bridge and synthesizes a `ReadableStream` for the user handler.
 */
export async function adaptWriteArgs(
  wireArgs: WriteArgs,
  deps: IoAdapterDeps,
): Promise<{
  userArgs: UserWriteArgs;
  /** Resolves once the inbound transfer (if any) has completed. */
  awaitTransferDone: () => Promise<void>;
}> {
  if (wireArgs.kind === 'inline') {
    const bytes = base64ToBytes(wireArgs.bytesB64);
    if (bytes.byteLength > INLINE_THRESHOLD) {
      throw new InvalidArgsError(
        `inline write exceeds ${INLINE_THRESHOLD}-byte threshold (got ${bytes.byteLength})`,
        'bytesB64',
      );
    }
    const hashHex = bytesToHex(sha256(bytes));
    const userArgs: UserWriteArgs = {
      path: wireArgs.path,
      overwrite: wireArgs.overwrite,
      ...(wireArgs.contentType !== undefined ? { contentType: wireArgs.contentType } : {}),
      content: {
        kind: 'inline',
        bytes,
        size: bytes.byteLength,
        sha256: hashHex,
      },
    };
    return { userArgs, awaitTransferDone: async () => undefined };
  }

  // Streams write — must have a streams-bridge configured.
  if (deps.streamsBridge === undefined) {
    throw new InternalFileError(
      'streams-bridge not configured: cannot accept streamed write',
    );
  }
  const parked = await deps.streamsBridge.awaitWrite(wireArgs.writeId, {
    expectedFrom: deps.sender,
    signal: deps.signal,
    timeoutMs: deps.ioTimeoutMs,
  });

  // sha256 from the inbound transfer's done() — user handler can `await`
  // it immediately after draining the readable.
  const sha256Promise = parked.done.then((r) => r.sha256);
  sha256Promise.catch(() => {
    /* swallow until consumer awaits */
  });

  // Cancellation: if the user handler aborts (or the dispatcher times out),
  // also tear down the inbound transfer.
  const onAbort = (): void => {
    void parked.handle.abort('user-cancel').catch(() => undefined);
  };
  if (deps.signal.aborted) onAbort();
  else deps.signal.addEventListener('abort', onAbort, { once: true });

  // The dispatcher calls this AFTER the user handler resolves to ensure
  // the transfer is fully drained (so any wire-level integrity errors
  // surface here before we send the RPC ack to the client).
  const awaitTransferDone = async (): Promise<void> => {
    try {
      await parked.done;
    } finally {
      deps.signal.removeEventListener('abort', onAbort);
    }
  };

  const userArgs: UserWriteArgs = {
    path: wireArgs.path,
    overwrite: wireArgs.overwrite,
    ...(wireArgs.contentType !== undefined ? { contentType: wireArgs.contentType } : {}),
    content: {
      kind: 'streams',
      stream: parked.readable,
      size: wireArgs.size,
      sha256: sha256Promise,
    },
  };
  return { userArgs, awaitTransferDone };
}

// ─── read: user result → wire result ─────────────────────────

/**
 * Convert a user-supplied `UserReadResult` into a wire `ReadResult`. For
 * streams results, this kicks an outbound `@shade/transfer` and embeds the
 * resulting `streamId` in the response so the client-side bridge can
 * subscribe.
 */
export async function adaptReadResult(
  userResult: UserReadResult,
  args: ReadArgs,
  deps: IoAdapterDeps,
): Promise<ReadResult> {
  if (userResult.kind === 'inline') {
    if (userResult.bytes.byteLength > INLINE_THRESHOLD) {
      throw new InternalFileError(
        `inline read result exceeds ${INLINE_THRESHOLD} bytes (got ${userResult.bytes.byteLength})`,
      );
    }
    const sha256Hex = userResult.sha256 ?? bytesToHex(sha256(userResult.bytes));
    return {
      kind: 'inline',
      bytesB64: bytesToBase64(userResult.bytes),
      size: userResult.bytes.byteLength,
      sha256: sha256Hex,
      ...(userResult.contentType !== undefined ? { contentType: userResult.contentType } : {}),
    };
  }

  if (deps.streamsBridge === undefined) {
    throw new InternalFileError(
      'streams-bridge not configured: cannot ship streamed read result',
    );
  }
  const { readStreamId } = await deps.streamsBridge.initiateRead({
    peer: deps.sender,
    stream: userResult.stream,
    size: userResult.size,
    ...(userResult.contentType !== undefined ? { contentType: userResult.contentType } : {}),
    name: args.path,
    signal: deps.signal,
  });
  return {
    kind: 'streams',
    streamId: readStreamId,
    size: userResult.size,
    sha256: userResult.sha256,
    ...(userResult.contentType !== undefined ? { contentType: userResult.contentType } : {}),
  };
}

// ─── getThumbnail: user result → wire result ─────────────────

const FORMAT_MAGIC: Record<string, Uint8Array> = {
  png: new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]),
  jpeg: new Uint8Array([0xff, 0xd8, 0xff]),
  webp: new Uint8Array([0x52, 0x49, 0x46, 0x46]), // 'RIFF', then size, then 'WEBP'
};
const WEBP_HEAD = new Uint8Array([0x57, 0x45, 0x42, 0x50]); // 'WEBP' at offset 8

/**
 * Validate a user thumbnail result for format magic-bytes match. The user
 * MAY produce any image format internally, but this guards against
 * misclassification (e.g. claiming PNG with JPEG bytes), which can confuse
 * downstream rendering and security scanners.
 */
function checkFormatMagic(bytes: Uint8Array, format: 'png' | 'webp' | 'jpeg'): boolean {
  const magic = FORMAT_MAGIC[format];
  if (magic === undefined) return false;
  if (bytes.byteLength < magic.byteLength) return false;
  for (let i = 0; i < magic.byteLength; i++) {
    if (bytes[i] !== magic[i]) return false;
  }
  if (format === 'webp') {
    if (bytes.byteLength < 12) return false;
    for (let i = 0; i < 4; i++) {
      if (bytes[8 + i] !== WEBP_HEAD[i]) return false;
    }
  }
  return true;
}

export function adaptThumbnailResult(
  userResult: UserThumbnailResult,
): GetThumbnailResult {
  if (!checkFormatMagic(userResult.bytes, userResult.format)) {
    throw new ConflictError(
      `thumbnail bytes do not match declared format=${userResult.format}`,
    );
  }
  const hash = userResult.sha256 ?? bytesToHex(sha256(userResult.bytes));
  return {
    bytesB64: bytesToBase64(userResult.bytes),
    format: userResult.format,
    width: userResult.width,
    height: userResult.height,
    sha256: hash,
  };
}
