/**
 * Server-side bridge between `@shade/files` content RPC ops and the
 * `@shade/transfer` engine.
 *
 * Two responsibilities:
 *  1. **Inbound writes (client → server, > 256 KiB).** When a client kicks
 *     `shade.upload(...)` with `userMetadata.shadeFilesWriteId = <id>`, this
 *     bridge intercepts the corresponding `IncomingTransfer` from
 *     `shade.onIncomingTransfer`, **immediately** calls `accept(...)` (the
 *     transfer engine requires synchronous accept — it rejects chunks that
 *     arrive before accept), pipes the plaintext into a TransformStream,
 *     and parks the readable side. The `write` RPC handler then awaits
 *     `awaitWrite(writeId, ...)` to pick up the readable, hand it to the
 *     user handler, and observe the transfer's done() result.
 *
 *  2. **Outbound reads (server → client, > 256 KiB).** When the user-supplied
 *     `read` handler returns `{ kind: 'streams', stream, size, sha256 }`,
 *     this bridge calls `shade.upload(...)` with
 *     `userMetadata.shadeFilesReadStreamId = <id>`. The id is then echoed
 *     back to the client in the `read` RPC response so the client-side
 *     bridge can subscribe to the matching incoming transfer.
 */
import type { IncomingTransfer, TransferHandle, TransferProgress } from '@shade/transfer';
import { generateRequestId } from '../protocol/correlate.js';
import { OperationTimeoutError } from '../schemas/errors.js';

/** Metadata key carried on inbound (client → server) write streams. */
export const META_KEY_WRITE_ID = 'shadeFilesWriteId';
/** Metadata key carried on outbound (server → client) read streams. */
export const META_KEY_READ_STREAM_ID = 'shadeFilesReadStreamId';

/**
 * Subset of `Shade` we depend on. Lets us unit-test without a real Shade
 * runtime and keeps the bridge framework-agnostic.
 */
export interface StreamsBridgeShade {
  upload(opts: import('@shade/transfer').TransferOptions): Promise<TransferHandle>;
  onIncomingTransfer(
    handler: (incoming: IncomingTransfer) => void | Promise<void>,
  ): Promise<() => void>;
}

export interface ParkedWrite {
  /** Sender address from the transfer engine. */
  from: string;
  /** Plaintext stream. The bridge already accepted the transfer; just consume. */
  readable: ReadableStream<Uint8Array>;
  /** Resolves when the transfer fully completes (sha256 available). */
  done: Promise<{ sha256: string; bytesSent: number }>;
  /** Underlying transfer handle — for abort propagation. */
  handle: TransferHandle;
  /** Unix epoch ms when the transfer arrived. Used for stale-eviction. */
  arrivedAt: number;
}

export interface AwaitWriteOptions {
  /** Sender address — must match `parked.from` for delivery. */
  expectedFrom: string;
  /** AbortSignal from the OpContext (cancellation/timeout). */
  signal: AbortSignal;
  /** Hard deadline (ms from now). Default 60_000. */
  timeoutMs?: number;
}

export interface ServerStreamsBridge {
  /**
   * Wait for an inbound transfer carrying `userMetadata.shadeFilesWriteId
   * === writeId` from `expectedFrom`. Resolves with the parked entry whose
   * `readable` can be consumed by the user handler.
   *
   * If the transfer has already arrived (race: stream-init beat the RPC),
   * resolves immediately. Otherwise awaits up to `timeoutMs`.
   */
  awaitWrite(writeId: string, opts: AwaitWriteOptions): Promise<ParkedWrite>;

  /**
   * Initiate an outbound transfer to `peer` carrying the given stream.
   * Stamps `userMetadata.shadeFilesReadStreamId = <streamId>`. Returns the
   * new streamId so the caller can echo it back to the client over RPC.
   */
  initiateRead(opts: {
    peer: string;
    stream: ReadableStream<Uint8Array>;
    size: number;
    contentType?: string;
    name?: string;
    signal?: AbortSignal;
    onProgress?: (p: TransferProgress) => void;
  }): Promise<{ readStreamId: string; handle: TransferHandle }>;

  /** Tear down the incoming-transfer subscription and reject all pending awaits. */
  destroy(): Promise<void>;
}

interface PendingWaiter {
  resolve: (parked: ParkedWrite) => void;
  reject: (err: unknown) => void;
  expectedFrom: string;
  timer: ReturnType<typeof setTimeout> | null;
  abortListener: (() => void) | null;
  signal: AbortSignal;
}

export interface CreateServerStreamsBridgeOptions {
  /** Default deadline for `awaitWrite` if the caller doesn't supply one. */
  defaultAwaitWriteTimeoutMs?: number;
  /** How long to retain a parked transfer waiting for its RPC. Default 60_000. */
  parkedWriteTtlMs?: number;
}

export async function createServerStreamsBridge(
  shade: StreamsBridgeShade,
  options: CreateServerStreamsBridgeOptions = {},
): Promise<ServerStreamsBridge> {
  const parkedWriteTtlMs = options.parkedWriteTtlMs ?? 60_000;
  const defaultAwaitTimeoutMs = options.defaultAwaitWriteTimeoutMs ?? 60_000;

  const parked = new Map<string, ParkedWrite>();
  const waiters = new Map<string, PendingWaiter>();
  let destroyed = false;

  const unsubscribeIncoming = await shade.onIncomingTransfer(async (incoming) => {
    const writeId = incoming.metadata.userMetadata?.[META_KEY_WRITE_ID];
    if (writeId === undefined) return; // not a files-bridge transfer; ignore

    // Accept synchronously into a TransformStream so the engine can deliver
    // chunks immediately. The readable side is parked for the matching RPC.
    const ts = new TransformStream<Uint8Array, Uint8Array>();
    let handle: TransferHandle;
    try {
      handle = await incoming.accept({
        output: { kind: 'pipe', pipeTo: ts.writable },
      });
    } catch (err) {
      // Accept failed — likely double-init or shutdown. Drop the parked
      // entry; the awaiting RPC will time out.
      console.error('[shade-files streams-bridge] accept failed:', err);
      return;
    }

    const arrived: ParkedWrite = {
      from: incoming.from,
      readable: ts.readable,
      done: handle.done().then((r) => ({ sha256: r.sha256, bytesSent: r.bytesSent })),
      handle,
      arrivedAt: Date.now(),
    };
    arrived.done.catch(() => {
      /* swallow until consumer awaits */
    });

    const waiter = waiters.get(writeId);
    if (waiter !== undefined) {
      waiters.delete(writeId);
      cleanupWaiter(waiter);
      if (incoming.from !== waiter.expectedFrom) {
        void handle.abort('sender-mismatch').catch(() => undefined);
        waiter.reject(
          new Error(
            `streams-bridge: writeId=${writeId} delivered by ${incoming.from}, expected ${waiter.expectedFrom}`,
          ),
        );
        return;
      }
      waiter.resolve(arrived);
      return;
    }

    // No waiter yet — park.
    parked.set(writeId, arrived);
    const parkTimer = setTimeout(() => {
      const stale = parked.get(writeId);
      if (stale === arrived) {
        parked.delete(writeId);
        void handle.abort('rpc-timeout').catch(() => undefined);
      }
    }, parkedWriteTtlMs);
    (parkTimer as unknown as { unref?: () => void }).unref?.();
  });

  function cleanupWaiter(w: PendingWaiter): void {
    if (w.timer !== null) clearTimeout(w.timer);
    if (w.abortListener !== null) {
      w.signal.removeEventListener('abort', w.abortListener);
    }
  }

  return {
    async awaitWrite(writeId, opts) {
      if (destroyed) throw new Error('streams-bridge: destroyed');
      // 1. Already parked?
      const ready = parked.get(writeId);
      if (ready !== undefined) {
        parked.delete(writeId);
        if (ready.from !== opts.expectedFrom) {
          void ready.handle.abort('sender-mismatch').catch(() => undefined);
          throw new Error(
            `streams-bridge: writeId=${writeId} delivered by ${ready.from}, expected ${opts.expectedFrom}`,
          );
        }
        return ready;
      }

      if (waiters.has(writeId)) {
        throw new Error(`streams-bridge: writeId=${writeId} already awaited`);
      }

      // 2. Park a waiter
      const timeoutMs = opts.timeoutMs ?? defaultAwaitTimeoutMs;
      return await new Promise<ParkedWrite>((resolve, reject) => {
        const w: PendingWaiter = {
          resolve,
          reject,
          expectedFrom: opts.expectedFrom,
          timer: null,
          abortListener: null,
          signal: opts.signal,
        };
        w.timer = setTimeout(() => {
          if (waiters.get(writeId) === w) {
            waiters.delete(writeId);
            cleanupWaiter(w);
            reject(new OperationTimeoutError(`streams-bridge: writeId=${writeId} timed out after ${timeoutMs}ms`));
          }
        }, timeoutMs);
        if (opts.signal.aborted) {
          cleanupWaiter(w);
          reject(opts.signal.reason ?? new Error('aborted before await'));
          return;
        }
        const onAbort = (): void => {
          if (waiters.get(writeId) === w) {
            waiters.delete(writeId);
            cleanupWaiter(w);
            reject(opts.signal.reason ?? new Error('aborted'));
          }
        };
        w.abortListener = onAbort;
        opts.signal.addEventListener('abort', onAbort, { once: true });
        waiters.set(writeId, w);
      });
    },

    async initiateRead(opts) {
      if (destroyed) throw new Error('streams-bridge: destroyed');
      const readStreamId = generateRequestId();
      const transferOpts: import('@shade/transfer').TransferOptions = {
        to: opts.peer,
        input: opts.stream,
        metadata: {
          ...(opts.name !== undefined ? { name: opts.name } : {}),
          ...(opts.contentType !== undefined ? { contentType: opts.contentType } : {}),
          sizeBytes: opts.size,
          userMetadata: { [META_KEY_READ_STREAM_ID]: readStreamId },
        },
        ...(opts.signal !== undefined ? { signal: opts.signal } : {}),
        ...(opts.onProgress !== undefined ? { onProgress: opts.onProgress } : {}),
      };
      const handle = await shade.upload(transferOpts);
      // The dispatcher returns the response immediately and doesn't itself
      // await this handle — the receiver awaits its own handle.done() on
      // the inbound side. Attach a no-op catch so engine teardown during
      // an in-flight outbound transfer doesn't surface as an unhandled
      // rejection in test environments.
      handle.done().catch(() => undefined);
      return { readStreamId, handle };
    },

    async destroy() {
      if (destroyed) return;
      destroyed = true;
      unsubscribeIncoming();
      // Reject all waiting RPCs
      for (const w of waiters.values()) {
        cleanupWaiter(w);
        w.reject(new Error('streams-bridge: destroyed'));
      }
      waiters.clear();
      // Abort all parked transfers
      for (const p of parked.values()) {
        try {
          await p.handle.abort('bridge-destroyed');
        } catch {
          /* swallow */
        }
      }
      parked.clear();
    },
  };
}
