import type { CryptoProvider } from '@shade/core';
import { ValidationError } from '@shade/core';
import {
  MultiLaneReceiver,
  MultiLaneSender,
  StreamingSha256,
  chunkRange,
  generateStreamId,
  generateStreamSecret,
  isStreamControlMessage,
  parseStreamControl,
  planRangePartition,
  planRoundRobinPartition,
  streamIdFromString,
  streamIdToString,
  type LaneInitSpec,
  type StreamControlMessage,
  type StreamFinishMessage,
  type StreamMetadata,
} from '@shade/streams';
import { decodeStreamChunk, inspectEnvelopeType } from '@shade/proto';
import {
  ATTR_BYTES_BIN,
  ATTR_DIRECTION,
  ATTR_ERROR_CODE,
  ATTR_LANE_COUNT,
  ATTR_PARTITION,
  ATTR_PEER_HASH,
  ATTR_RESULT,
  ATTR_RETRY_COUNT,
  bytesBin,
  laneCountBin,
  NOOP_HOOK,
  peerHash,
  type ObservabilityHook,
  type Span,
} from '@shade/observability';
import {
  TransferAbortError,
  TransferIntegrityError,
  TransferOfflineError,
  TransferProtocolError,
} from './errors.js';
import { ProgressTracker } from './progress.js';
import { normalizeInput, type NormalizedInput } from './sender/input.js';
import { BoundedAsyncQueue } from './sender/lane-queue.js';
import { resolveOutputSink, type OutputSink } from './receiver/output.js';
import {
  DEFAULT_CHUNK_SIZE,
  DEFAULT_LANE_COUNT,
  DEFAULT_MAX_CHUNK_SIZE,
  type IncomingTransfer,
  type IncomingTransferAcceptOptions,
  type LaneProgress,
  type TransferEvent,
  type TransferHandle,
  type TransferOptions,
  type TransferProgress,
  type TransferResult,
} from './types.js';
import type {
  ChunkAck,
  IControlChannel,
  ITransferTransport,
  TransferResumeState,
} from './transport/transport.js';
import {
  type ResumeStore,
  wrapStreamSecret,
  unwrapStreamSecret,
} from './persistence/resume.js';
import type { PersistedStreamState } from '@shade/core';

export interface TransferEngineDeps {
  crypto: CryptoProvider;
  controlChannel: IControlChannel;
  transport: ITransferTransport;
  /** This endpoint's logical address (e.g. `device:server`). */
  myAddress: string;
  /** Optional SubtleCrypto instance for AEAD operations. */
  subtle?: SubtleCrypto;
  /** Optional resume store; when provided, in-flight state is persisted. */
  resumeStore?: ResumeStore;
  /**
   * Optional 32-byte deviceKey for at-rest streamSecret encryption. Required
   * when `resumeStore` persists across processes (SQLite/Postgres/IndexedDB).
   * For in-memory stores it can be omitted (secret stays encrypted with a
   * random per-process key).
   */
  deviceKey?: Uint8Array;
  /**
   * Optional observability hook. Spans are emitted for each upload and for
   * each accepted incoming transfer. Defaults to a no-op when omitted.
   * Use `withTracer()` from `@shade/observability` to plug in OTel.
   */
  observability?: ObservabilityHook;
}

/**
 * The transfer orchestrator.
 *
 * Wraps an `IControlChannel` (for stream-init/finish/abort messages — usually
 * over the existing Shade ratchet) and an `ITransferTransport` (for chunk
 * data — usually HTTP). Provides `upload()` and `onIncomingTransfer()` entry
 * points; the SDK adds these to the `Shade` class in M-Stream-5.
 */
export class TransferEngine {
  private readonly outgoing = new Map<string, OutgoingState>();
  private readonly incoming = new Map<string, IncomingState>();
  private readonly incomingHandlers = new Set<
    (incoming: IncomingTransfer) => void | Promise<void>
  >();
  private readonly unsubscribeControl: () => void;
  private readonly persister: Persister | null;
  private readonly observability: ObservabilityHook;

  constructor(private readonly deps: TransferEngineDeps) {
    this.observability = deps.observability ?? NOOP_HOOK;
    this.persister =
      deps.resumeStore !== undefined
        ? new Persister(deps.crypto, deps.resumeStore, deps.deviceKey)
        : null;
    this.unsubscribeControl = deps.controlChannel.onMessage(async (from, message) => {
      try {
        await this.handleControlMessage(from, message);
      } catch (err) {
        console.error('[TransferEngine] control message error:', err);
        throw err;
      }
    });
  }

  /** Start an upload. Returns a handle that can be paused/aborted/awaited. */
  async upload(opts: TransferOptions): Promise<TransferHandle> {
    const input = await normalizeInput(opts.input);
    const chunkSize = opts.chunkSize ?? DEFAULT_CHUNK_SIZE;
    const maxChunkSize = opts.maxChunkSize ?? DEFAULT_MAX_CHUNK_SIZE;
    if (chunkSize < 1 || chunkSize > maxChunkSize) {
      throw new ValidationError(
        `chunkSize ${chunkSize} out of range (1, ${maxChunkSize})`,
        'chunkSize',
      );
    }

    const requestedLanes = opts.lanes ?? DEFAULT_LANE_COUNT;
    const laneCount = computeEffectiveLaneCount(requestedLanes, input.size, chunkSize);

    const partitionMode = resolvePartition(opts.partition, input);

    const lanes: LaneInitSpec[] =
      partitionMode === 'range'
        ? planRangePartition(input.size ?? 0, laneCount)
        : planRoundRobinPartition(laneCount);

    // Probe peer reachability up front (NF: clear offline error).
    try {
      await this.deps.transport.probe(opts.to);
    } catch (err) {
      await input.close();
      throw new TransferOfflineError(opts.to, (err as Error).message);
    }

    const streamIdBytes = generateStreamId(this.deps.crypto);
    const streamSecret = generateStreamSecret(this.deps.crypto);
    const streamId = streamIdToString(streamIdBytes);

    const metadata: StreamMetadata = {
      ...(input.name !== undefined ? { name: input.name } : {}),
      ...(input.size !== undefined ? { sizeBytes: input.size } : {}),
      ...(input.contentType !== undefined ? { contentType: input.contentType } : {}),
      chunkSize,
      sentAt: Date.now(),
      ...(opts.metadata ?? {}),
    };
    if (input.size !== undefined) {
      metadata.totalChunks = Math.max(1, Math.ceil(input.size / chunkSize));
    }

    const sender = await MultiLaneSender.create({
      crypto: this.deps.crypto,
      ...(this.deps.subtle !== undefined ? { subtle: this.deps.subtle } : {}),
      streamId: streamIdBytes,
      streamSecret,
      lanes,
    });

    const state = new OutgoingState(
      streamId,
      opts.to,
      sender,
      metadata,
      lanes,
      input,
      chunkSize,
      opts.signal,
      opts.onProgress,
      opts.onEvent,
    );
    state.span = this.observability.startSpan('shade.transfer.upload', {
      [ATTR_DIRECTION]: 'upload',
      [ATTR_PEER_HASH]: peerHash(opts.to),
      [ATTR_BYTES_BIN]: bytesBin(input.size ?? 0),
      [ATTR_LANE_COUNT]: laneCountBin(lanes.length),
      [ATTR_PARTITION]: partitionMode,
    });
    this.outgoing.set(streamId, state);

    // Persist initial resume state BEFORE sending init, so a crash before
    // first chunk leaves a recoverable record on disk.
    if (this.persister !== null) {
      await this.persister.saveOutgoing(state, streamSecret);
    }

    // Send stream-init plaintext over the control channel.
    await this.deps.controlChannel.send(opts.to, {
      kind: 'shade.stream-init/v1',
      streamId,
      streamSecret: bytesToBase64Url(streamSecret),
      metadata,
      lanes,
    });

    // Spawn the upload pipeline (read → encrypt → transport).
    state.startedAt = nowMs();
    state.emit({ type: 'start', streamId });
    void this.runUpload(state, partitionMode).catch(async (err) => {
      await state.failTransfer(err);
    });

    return state.handle;
  }

  /** Subscribe to incoming transfers. */
  onIncomingTransfer(
    handler: (incoming: IncomingTransfer) => void | Promise<void>,
  ): () => void {
    this.incomingHandlers.add(handler);
    return () => this.incomingHandlers.delete(handler);
  }

  /**
   * Resume an interrupted upload. The consumer re-supplies the input
   * (browser File handle / Bun path / Uint8Array — must be byte-identical
   * to the original) and we pick up where we left off.
   *
   * Flow:
   *   1. Load persisted state, decrypt streamSecret.
   *   2. Query the receiver for its `lastSeqAcked` per lane.
   *   3. Spawn a new MultiLaneSender with `startSeq = lastSeqAcked + 1` per lane.
   *   4. Pre-hash already-shipped bytes through each lane's hasher.
   *   5. Continue the upload from the resumption point.
   *
   * `opts.onProgress` / `opts.onEvent` overrides apply to the resumed run.
   */
  async resumeUpload(
    streamId: string,
    freshInput: TransferOptions['input'],
    opts?: {
      onProgress?: (p: TransferProgress) => void;
      onEvent?: (e: TransferEvent) => void;
      signal?: AbortSignal;
    },
  ): Promise<TransferHandle> {
    if (this.persister === null) {
      throw new TransferProtocolError('resumeUpload requires a resume store');
    }
    const loaded = await this.persister.load(streamId);
    if (loaded === null) {
      throw new TransferProtocolError(`no persisted state for streamId=${streamId}`);
    }
    if (loaded.record.direction !== 'send') {
      throw new TransferProtocolError(`stream ${streamId} is a receive, not a send`);
    }
    if (loaded.record.status === 'finished' || loaded.record.status === 'aborted') {
      throw new TransferProtocolError(
        `stream ${streamId} is already ${loaded.record.status}`,
      );
    }
    const peer = loaded.record.peerAddress;
    const metadata = JSON.parse(loaded.record.metadataJson) as import('@shade/streams').StreamMetadata;
    const lanes = JSON.parse(loaded.record.partitionJson) as import('@shade/streams').LaneInitSpec[];

    const remoteState = await this.deps.transport.fetchResumeState(peer, streamId);
    const lastSeqByLane = new Map<number, number>();
    if (remoteState !== null) {
      for (const l of remoteState.lanes) lastSeqByLane.set(l.laneId, l.lastSeqAcked);
    }
    for (const lane of lanes) {
      if (!lastSeqByLane.has(lane.laneId)) lastSeqByLane.set(lane.laneId, -1);
    }

    const startSeqByLane = new Map<number, number>();
    for (const [laneId, last] of lastSeqByLane) startSeqByLane.set(laneId, last + 1);

    const input = await normalizeInput(freshInput);
    const sender = await MultiLaneSender.create({
      crypto: this.deps.crypto,
      ...(this.deps.subtle !== undefined ? { subtle: this.deps.subtle } : {}),
      streamId: bytesFromBase64Url(loaded.record.streamId),
      streamSecret: loaded.streamSecret,
      lanes,
      startSeqByLane,
    });

    // For range mode the prefix bytes per lane will be folded into both the
    // per-lane hasher AND the overall hasher BEFORE the lane's remainder is
    // streamed (see runRangeUploadResumed). For round-robin we re-feed the
    // common prefix here so the new sender's overall hasher matches the
    // receiver's reconstructed-order view.
    if (lanes[0]!.partition.kind === 'round-robin') {
      let maxAcked = -1;
      for (const last of lastSeqByLane.values()) {
        if (last > maxAcked) maxAcked = last;
      }
      if (maxAcked >= 0 && input.size !== undefined) {
        const totalToReHash = (maxAcked + 1) * lanes.length;
        let cursor = 0;
        for (let i = 0; i < totalToReHash && cursor < input.size; i++) {
          const end = Math.min(cursor + metadata.chunkSize, input.size);
          const slice = await input.read(cursor, end);
          const laneId = i % lanes.length;
          sender.preHashForLane(laneId, slice);
          sender.appendOverall(slice);
          cursor = end;
        }
      }
    }

    const partitionMode = lanes[0]!.partition.kind;
    const state = new OutgoingState(
      streamId,
      peer,
      sender,
      metadata,
      lanes,
      input,
      metadata.chunkSize,
      opts?.signal,
      opts?.onProgress,
      opts?.onEvent,
    );
    state.span = this.observability.startSpan('shade.transfer.upload.resume', {
      [ATTR_DIRECTION]: 'upload',
      [ATTR_PEER_HASH]: peerHash(peer),
      [ATTR_BYTES_BIN]: bytesBin(input.size ?? metadata.sizeBytes ?? 0),
      [ATTR_LANE_COUNT]: laneCountBin(lanes.length),
      [ATTR_PARTITION]: partitionMode,
    });
    // Mark already-shipped lanes as having advanced.
    for (const lane of lanes) {
      const lp = (state as unknown as { laneProgress: Map<number, LaneProgress> }).laneProgress.get(
        lane.laneId,
      );
      if (lp !== undefined) {
        lp.seq = lastSeqByLane.get(lane.laneId)! + 1;
      }
    }
    this.outgoing.set(streamId, state);
    state.startedAt = nowMs();
    state.emit({ type: 'start', streamId });
    void this.runResumedUpload(state, partitionMode, lastSeqByLane).catch(async (err) => {
      await state.failTransfer(err);
    });
    return state.handle;
  }

  private async runResumedUpload(
    state: OutgoingState,
    partitionMode: 'range' | 'round-robin',
    lastSeqByLane: Map<number, number>,
  ): Promise<void> {
    if (partitionMode === 'range') {
      await this.runRangeUploadResumed(state, lastSeqByLane);
    } else {
      await this.runRoundRobinUploadResumed(state, lastSeqByLane);
    }
    if (state.aborted) return;

    const laneFps = state.sender.getLaneFingerprints();
    const overall = state.sender.getOverallSha256();
    await this.deps.controlChannel.send(state.peerAddress, {
      kind: 'shade.stream-finish/v1',
      streamId: state.streamId,
      laneSha256: laneFps.map((f) => ({
        laneId: f.laneId,
        sha256: bytesToBase64(f.sha256),
        chunkCount: f.chunkCount,
        byteCount: f.byteCount,
      })),
      overallSha256: bytesToBase64(overall),
      finishedAt: Date.now(),
    });
    if (this.persister !== null) {
      await this.persister.markStatus(state.streamId, 'finished');
    }
    state.complete({
      streamId: state.streamId,
      bytesSent: laneFps.reduce((s, l) => s + l.byteCount, 0),
      sha256: hex(overall),
      durationMs: nowMs() - state.startedAt,
    });
    this.outgoing.delete(state.streamId);
  }

  private async runRangeUploadResumed(
    state: OutgoingState,
    lastSeqByLane: Map<number, number>,
  ): Promise<void> {
    const queues = new Map<number, BoundedAsyncQueue<ChunkJob>>();
    const workers: Promise<void>[] = [];
    const flightCap = 4;
    for (const lane of state.lanes) {
      const q = new BoundedAsyncQueue<ChunkJob>(flightCap);
      queues.set(lane.laneId, q);
      workers.push(this.runLaneWorker(state, lane.laneId, q));
    }
    try {
      // Process lanes in laneId order so the OVERALL hasher receives bytes
      // in original order (range mode = lane k owns bytes [k·N/L, (k+1)·N/L)).
      // Per lane: re-hash the already-acked prefix, then stream the remainder.
      for (const lane of state.lanes) {
        if (lane.partition.kind !== 'range') {
          throw new ValidationError(`expected range partition`, 'lane');
        }
        const lastAcked = lastSeqByLane.get(lane.laneId) ?? -1;
        const allSlices = chunkRange(
          lane.partition.startByte,
          lane.partition.endByte,
          state.chunkSize,
        );
        // Re-hash prefix (chunks 0..lastAcked) into both per-lane and overall hashers.
        for (let i = 0; i <= lastAcked && i < allSlices.length; i++) {
          if (state.aborted) break;
          const s = allSlices[i]!;
          const slice = await state.input.read(s.start, s.end);
          state.sender.preHashForLane(lane.laneId, slice);
          state.sender.appendOverall(slice);
        }
        // Stream remainder.
        for (let i = lastAcked + 1; i < allSlices.length; i++) {
          if (state.aborted) break;
          const s = allSlices[i]!;
          const isLast = i === allSlices.length - 1;
          const plaintext = await state.input.read(s.start, s.end);
          state.sender.appendOverall(plaintext);
          await queues.get(lane.laneId)!.push({ plaintext, isLast });
        }
        queues.get(lane.laneId)!.close();
      }
    } catch (err) {
      for (const q of queues.values()) q.abort(err);
      throw err;
    }
    await Promise.all(workers);
  }

  private async runRoundRobinUploadResumed(
    state: OutgoingState,
    lastSeqByLane: Map<number, number>,
  ): Promise<void> {
    const queues = new Map<number, BoundedAsyncQueue<ChunkJob>>();
    const workers: Promise<void>[] = [];
    const flightCap = 4;
    const laneCount = state.lanes.length;
    for (const lane of state.lanes) {
      const q = new BoundedAsyncQueue<ChunkJob>(flightCap);
      queues.set(lane.laneId, q);
      workers.push(this.runLaneWorker(state, lane.laneId, q));
    }

    if (state.input.size === undefined) {
      throw new TransferProtocolError(
        'round-robin resume requires known-size input (random access)',
      );
    }

    let maxAcked = -1;
    for (const last of lastSeqByLane.values()) {
      if (last > maxAcked) maxAcked = last;
    }
    const skipChunks = (maxAcked + 1) * laneCount;
    let cursor = skipChunks * state.chunkSize;
    cursor = Math.min(cursor, state.input.size);
    let chunkIndex = skipChunks;

    try {
      while (cursor < state.input.size) {
        if (state.aborted) break;
        const end = Math.min(cursor + state.chunkSize, state.input.size);
        const buf = await state.input.read(cursor, end);
        cursor = end;
        const isEof = cursor >= state.input.size;
        const laneId = chunkIndex % laneCount;
        state.sender.appendOverall(buf);
        // Determine isLast: if EOF, this is the last chunk for this lane.
        // Other lanes that received fewer chunks need an empty isLast.
        const isLast = isEof; // simplification: we mark the very last chunk
        await queues.get(laneId)!.push({ plaintext: buf, isLast });
        chunkIndex++;
      }
      // Close queues; any lane that didn't get an isLast chunk needs one.
      // Compute per-lane sent counts and emit empty isLast if needed.
      const totalChunksSent = chunkIndex;
      const lastLaneId = (totalChunksSent - 1 + laneCount) % laneCount;
      void lastLaneId;
      for (const lane of state.lanes) {
        // Count: chunks for this lane = ceil((total - laneId) / L) if laneId < total mod L else floor
        const sentForLane =
          Math.floor((totalChunksSent - lane.laneId - 1) / laneCount) + 1;
        const expectedAcked = lastSeqByLane.get(lane.laneId)! + 1;
        if (sentForLane <= expectedAcked) {
          // This lane never got a new chunk; push empty isLast=true.
          await queues.get(lane.laneId)!.push({ plaintext: new Uint8Array(0), isLast: true });
        }
        queues.get(lane.laneId)!.close();
      }
    } catch (err) {
      for (const q of queues.values()) q.abort(err);
      throw err;
    }
    await Promise.all(workers);
  }

  /**
   * Receiver-side: accept a chunk envelope. Called by the transport handler
   * (HTTP route or memory-pipe). Returns the ACK that the transport relays
   * back to the sender.
   */
  async receiveChunk(
    from: string,
    streamId: string,
    laneId: number,
    seq: number | bigint,
    bytes: Uint8Array,
  ): Promise<ChunkAck> {
    if (inspectEnvelopeType(bytes) !== 'stream-chunk') {
      throw new TransferProtocolError('not a stream-chunk envelope');
    }
    const state = this.incoming.get(streamId);
    if (state === undefined) {
      throw new TransferProtocolError(`no incoming transfer for streamId=${streamId}`);
    }
    if (state.peerAddress !== from) {
      throw new TransferProtocolError(`peer mismatch: ${from} vs ${state.peerAddress}`);
    }
    const env = decodeStreamChunk(bytes);
    if (env.laneId !== laneId) {
      throw new TransferProtocolError(`laneId mismatch: header ${env.laneId} vs route ${laneId}`);
    }
    const envSeq = typeof env.seq === 'bigint' ? env.seq : BigInt(env.seq);
    const expected = typeof seq === 'bigint' ? seq : BigInt(seq);
    if (envSeq !== expected) {
      throw new TransferProtocolError(`seq mismatch: header ${envSeq} vs route ${expected}`);
    }
    return state.handleChunk(bytes);
  }

  /** Receiver-side: respond to a resume-state lookup. */
  async getResumeState(
    from: string,
    streamId: string,
  ): Promise<TransferResumeState | null> {
    const state = this.incoming.get(streamId);
    if (state === undefined || state.peerAddress !== from) return null;
    return {
      streamId,
      lanes: state.lanesProgressForResume(),
    };
  }

  /** Tear down all subscriptions. */
  destroy(): void {
    this.unsubscribeControl();
    for (const o of this.outgoing.values()) void o.failTransfer(new TransferAbortError('engine destroyed'));
    for (const i of this.incoming.values()) void i.abort('engine destroyed');
    this.outgoing.clear();
    this.incoming.clear();
    this.incomingHandlers.clear();
  }

  // ─── Internals ─────────────────────────────────────────────

  private async handleControlMessage(
    from: string,
    raw: StreamControlMessage | unknown,
  ): Promise<void> {
    let msg: StreamControlMessage;
    if (typeof raw === 'string') {
      msg = parseStreamControl(raw);
    } else if (isStreamControlMessage(raw)) {
      msg = raw;
    } else {
      return; // Ignore non-stream control messages.
    }

    switch (msg.kind) {
      case 'shade.stream-init/v1':
        await this.handleIncomingInit(from, msg);
        return;
      case 'shade.stream-finish/v1':
        await this.handleIncomingFinish(from, msg);
        return;
      case 'shade.stream-abort/v1':
        await this.handleIncomingAbort(from, msg);
        return;
      case 'shade.stream-resume-request/v1':
      case 'shade.stream-resume-state/v1':
        // Resume protocol wires up in M-Stream-6. Ignore in v0.2.0 baseline.
        return;
    }
  }

  private async handleIncomingInit(
    from: string,
    msg: import('@shade/streams').StreamInitMessage,
  ): Promise<void> {
    const streamIdBytes = streamIdFromString(msg.streamId);
    const streamSecret = base64UrlToBytes(msg.streamSecret);
    if (this.incoming.has(msg.streamId)) {
      throw new TransferProtocolError(`duplicate stream-init for streamId=${msg.streamId}`);
    }
    const receiver = await MultiLaneReceiver.create({
      crypto: this.deps.crypto,
      ...(this.deps.subtle !== undefined ? { subtle: this.deps.subtle } : {}),
      streamId: streamIdBytes,
      streamSecret,
      lanes: msg.lanes,
    });
    const state = new IncomingState(
      msg.streamId,
      from,
      receiver,
      msg.metadata,
      msg.lanes,
    );
    this.incoming.set(msg.streamId, state);
    if (this.persister !== null) {
      await this.persister.saveIncoming(state, streamSecret);
    }

    const incoming: IncomingTransfer = {
      streamId: msg.streamId,
      from,
      metadata: msg.metadata,
      lanes: msg.lanes,
      accept: async (opts: IncomingTransferAcceptOptions) => {
        if (state.accepted) {
          throw new TransferProtocolError('already accepted');
        }
        state.accepted = true;
        state.sink = await resolveOutputSink(opts.output);
        state.outputKind = opts.output.kind;
        if (opts.onProgress !== undefined) state.onProgress = opts.onProgress;
        if (opts.onEvent !== undefined) state.onEvent = opts.onEvent;
        state.startedAt = nowMs();
        state.span = this.observability.startSpan('shade.transfer.download', {
          [ATTR_DIRECTION]: 'download',
          [ATTR_PEER_HASH]: peerHash(from),
          [ATTR_BYTES_BIN]: bytesBin(msg.metadata.sizeBytes ?? 0),
          [ATTR_LANE_COUNT]: laneCountBin(msg.lanes.length),
          [ATTR_PARTITION]: msg.lanes[0]?.partition.kind ?? 'unknown',
        });
        state.emit({ type: 'start', streamId: msg.streamId });
        return state.handle;
      },
      decline: async (reason?: string) => {
        state.declined = true;
        await state.abort(reason ?? 'receiver-decline');
      },
    };

    for (const handler of this.incomingHandlers) {
      try {
        await handler(incoming);
      } catch (err) {
        console.error('[TransferEngine] incoming handler error:', err);
      }
    }
  }

  private async handleIncomingFinish(
    from: string,
    msg: StreamFinishMessage,
  ): Promise<void> {
    const state = this.incoming.get(msg.streamId);
    if (state === undefined || state.peerAddress !== from) return;
    await state.finalize(msg);
    this.incoming.delete(msg.streamId);
  }

  private async handleIncomingAbort(
    from: string,
    msg: import('@shade/streams').StreamAbortMessage,
  ): Promise<void> {
    const state = this.incoming.get(msg.streamId);
    if (state === undefined || state.peerAddress !== from) return;
    await state.abort(msg.reason);
    this.incoming.delete(msg.streamId);
  }

  private async runUpload(
    state: OutgoingState,
    partitionMode: 'range' | 'round-robin',
  ): Promise<void> {
    if (partitionMode === 'range') {
      await this.runRangeUpload(state);
    } else {
      await this.runRoundRobinUpload(state);
    }

    if (state.aborted) return;

    // Compute fingerprints + send finish.
    const laneFps = state.sender.getLaneFingerprints();
    const overall = state.sender.getOverallSha256();
    const finishMsg: StreamFinishMessage = {
      kind: 'shade.stream-finish/v1',
      streamId: state.streamId,
      laneSha256: laneFps.map((f) => ({
        laneId: f.laneId,
        sha256: bytesToBase64(f.sha256),
        chunkCount: f.chunkCount,
        byteCount: f.byteCount,
      })),
      overallSha256: bytesToBase64(overall),
      finishedAt: Date.now(),
    };
    await this.deps.controlChannel.send(state.peerAddress, finishMsg);

    const result: TransferResult = {
      streamId: state.streamId,
      bytesSent: laneFps.reduce((s, l) => s + l.byteCount, 0),
      sha256: hex(overall),
      durationMs: nowMs() - state.startedAt,
    };
    if (this.persister !== null) {
      await this.persister.markStatus(state.streamId, 'finished');
    }
    state.complete(result);
    this.outgoing.delete(state.streamId);
  }

  private async runRangeUpload(state: OutgoingState): Promise<void> {
    // Single sequential reader; per-lane queue + worker.
    const queues = new Map<number, BoundedAsyncQueue<ChunkJob>>();
    const workers: Promise<void>[] = [];
    const flightCap = 4;
    for (const lane of state.lanes) {
      const q = new BoundedAsyncQueue<ChunkJob>(flightCap);
      queues.set(lane.laneId, q);
      workers.push(this.runLaneWorker(state, lane.laneId, q));
    }
    try {
      for (const lane of state.lanes) {
        if (lane.partition.kind !== 'range') {
          throw new ValidationError(`expected range partition for lane ${lane.laneId}`, 'lane');
        }
        const slices = chunkRange(
          lane.partition.startByte,
          lane.partition.endByte,
          state.chunkSize,
        );
        for (let i = 0; i < slices.length; i++) {
          const s = slices[i]!;
          const isLast = i === slices.length - 1;
          if (state.aborted) break;
          const plaintext = await state.input.read(s.start, s.end);
          state.sender.appendOverall(plaintext);
          await queues.get(lane.laneId)!.push({ plaintext, isLast });
        }
        queues.get(lane.laneId)!.close();
      }
    } catch (err) {
      for (const q of queues.values()) q.abort(err);
      throw err;
    }
    await Promise.all(workers);
  }

  private async runRoundRobinUpload(state: OutgoingState): Promise<void> {
    const queues = new Map<number, BoundedAsyncQueue<ChunkJob>>();
    const workers: Promise<void>[] = [];
    const flightCap = 4;
    const laneCount = state.lanes.length;
    for (const lane of state.lanes) {
      const q = new BoundedAsyncQueue<ChunkJob>(flightCap);
      queues.set(lane.laneId, q);
      workers.push(this.runLaneWorker(state, lane.laneId, q));
    }

    // Source iterator: random-access inputs get an `input.read(start, end)`
    // walk; non-random-access inputs use `input.readNext`.
    const readNextChunk: () => Promise<Uint8Array | null> =
      state.input.randomAccess && state.input.size !== undefined
        ? (() => {
            let cursor = 0;
            const total = state.input.size;
            return async () => {
              if (cursor >= total) return null;
              const end = Math.min(cursor + state.chunkSize, total);
              const buf = await state.input.read(cursor, end);
              cursor = end;
              return buf;
            };
          })()
        : () => state.input.readNext(state.chunkSize);

    // Read-pump: pull chunks sequentially until EOF.
    // We can't mark isLast=true until we know it's truly the last chunk for
    // a lane, so we keep a one-chunk lookahead per lane: the most recent
    // chunk for each lane sits as a "tail" until either superseded (becomes
    // a body chunk) or finalized as isLast at EOF.
    interface TailRecord {
      chunkIndex: number;
      plaintext: Uint8Array;
    }
    const tailByLane = new Map<number, TailRecord>();
    let chunkIndex = 0;
    try {
      while (true) {
        if (state.aborted) break;
        const buf = await readNextChunk();
        if (buf === null) break;
        const laneId = chunkIndex % laneCount;
        const previousTail = tailByLane.get(laneId);
        if (previousTail !== undefined) {
          state.sender.appendOverall(previousTail.plaintext);
          await queues.get(laneId)!.push({ plaintext: previousTail.plaintext, isLast: false });
        }
        tailByLane.set(laneId, { chunkIndex, plaintext: buf });
        chunkIndex++;
      }
      // At EOF, flush remaining tails in CHUNK-INDEX order (original byte
      // order) so appendOverall stays consistent with the receiver. Lanes
      // that never saw a chunk get an empty isLast=true to mark completion.
      const tailsInOrder = [...tailByLane.entries()]
        .map(([laneId, t]) => ({ laneId, chunkIndex: t.chunkIndex, plaintext: t.plaintext }))
        .sort((a, b) => a.chunkIndex - b.chunkIndex);
      for (const t of tailsInOrder) {
        state.sender.appendOverall(t.plaintext);
        await queues.get(t.laneId)!.push({ plaintext: t.plaintext, isLast: true });
      }
      // Lanes that received nothing — push empty isLast.
      for (const lane of state.lanes) {
        if (!tailByLane.has(lane.laneId)) {
          await queues.get(lane.laneId)!.push({ plaintext: new Uint8Array(0), isLast: true });
        }
      }
      for (const q of queues.values()) q.close();
    } catch (err) {
      for (const q of queues.values()) q.abort(err);
      throw err;
    }
    await Promise.all(workers);
  }

  private async runLaneWorker(
    state: OutgoingState,
    laneId: number,
    queue: BoundedAsyncQueue<ChunkJob>,
  ): Promise<void> {
    while (true) {
      const r = await queue.next();
      if (r.done) return;
      if (state.aborted) return;
      const { plaintext, isLast } = r.value;
      const { bytes, seq } = await state.sender.encryptForLane(laneId, plaintext, isLast);
      await this.deps.transport.sendChunk(
        state.peerAddress,
        state.streamId,
        laneId,
        seq,
        bytes,
        ...(state.signal !== undefined ? [{ signal: state.signal }] : []),
      );
      state.recordSent(laneId, plaintext.length, seq, isLast);
    }
  }
}

// ─── Internal state ──────────────────────────────────────────

interface ChunkJob {
  plaintext: Uint8Array;
  isLast: boolean;
}

class OutgoingState {
  startedAt = -1;
  aborted = false;
  span: Span | null = null;
  private retryCount = 0;
  private completedAt = -1;
  private resolveDone!: (result: TransferResult) => void;
  private rejectDone!: (err: unknown) => void;
  private donePromise: Promise<TransferResult>;
  private events: TransferEvent[] = [];
  private eventResolvers: Array<(value: IteratorResult<TransferEvent>) => void> = [];
  private progress: ProgressTracker;
  private laneProgress: Map<number, LaneProgress>;

  readonly handle: TransferHandle;

  constructor(
    public readonly streamId: string,
    public readonly peerAddress: string,
    public readonly sender: MultiLaneSender,
    public readonly metadata: StreamMetadata,
    public readonly lanes: LaneInitSpec[],
    public readonly input: NormalizedInput,
    public readonly chunkSize: number,
    public readonly signal: AbortSignal | undefined,
    public onProgress: ((p: TransferProgress) => void) | undefined,
    public onEvent: ((e: TransferEvent) => void) | undefined,
  ) {
    this.donePromise = new Promise<TransferResult>((resolve, reject) => {
      this.resolveDone = resolve;
      this.rejectDone = reject;
    });
    this.progress = new ProgressTracker(metadata.sizeBytes);
    this.laneProgress = new Map(
      lanes.map((l) => [
        l.laneId,
        {
          laneId: l.laneId,
          bytesSent: 0,
          ...(l.partition.kind === 'range'
            ? { bytesTotal: l.partition.endByte - l.partition.startByte }
            : {}),
          seq: 0,
          state: 'idle' as const,
        },
      ]),
    );
    if (signal !== undefined) {
      signal.addEventListener('abort', () => {
        void this.failTransfer(new TransferAbortError('signal aborted'));
      });
    }
    this.handle = {
      streamId: this.streamId,
      events: this.eventsAsyncIterable(),
      pause: async () => {
        /* M-Stream-6 territory; placeholder no-op for v0.2.0 baseline. */
      },
      resume: async () => {
        /* M-Stream-6 territory. */
      },
      abort: async (reason) => {
        await this.failTransfer(new TransferAbortError(reason));
      },
      done: () => this.donePromise,
    };
  }

  recordSent(laneId: number, bytes: number, seq: number, isLast: boolean): void {
    const lane = this.laneProgress.get(laneId);
    if (lane !== undefined) {
      lane.bytesSent += bytes;
      lane.seq = seq + 1;
      lane.state = isLast ? 'done' : 'sending';
      if (isLast) this.emit({ type: 'lane-done', laneId });
    }
    this.progress.add(bytes);
    const sample = this.progress.sample();
    const tp: TransferProgress = {
      streamId: this.streamId,
      bytesSent: this.progress.totalProcessed,
      bytesPerSecond: sample.bytesPerSecond,
      lanes: [...this.laneProgress.values()],
      ...(this.metadata.sizeBytes !== undefined ? { bytesTotal: this.metadata.sizeBytes } : {}),
      ...(sample.etaSeconds !== undefined ? { etaSeconds: sample.etaSeconds } : {}),
      ...(sample.percent !== undefined ? { percent: sample.percent } : {}),
    };
    this.onProgress?.(tp);
    this.emit({ type: 'progress', progress: tp });
  }

  emit(event: TransferEvent): void {
    this.onEvent?.(event);
    if (this.eventResolvers.length > 0) {
      this.eventResolvers.shift()!({ value: event, done: false });
    } else {
      this.events.push(event);
    }
  }

  complete(result: TransferResult): void {
    this.completedAt = nowMs();
    this.emit({ type: 'complete', streamId: result.streamId, sha256: result.sha256, durationMs: result.durationMs });
    this.flushEventEnd();
    if (this.span !== null) {
      this.span.setAttributes({
        [ATTR_RESULT]: 'ok',
        [ATTR_BYTES_BIN]: bytesBin(result.bytesSent),
        [ATTR_RETRY_COUNT]: this.retryCount,
      });
      this.span.setStatus('ok');
      this.span.end();
      this.span = null;
    }
    this.resolveDone(result);
    void this.input.close();
    this.sender.destroy();
  }

  async failTransfer(err: unknown): Promise<void> {
    if (this.aborted || this.completedAt > 0) return;
    this.aborted = true;
    this.emit({ type: 'error', error: err });
    this.flushEventEnd();
    if (this.span !== null) {
      const code = errorCodeOf(err);
      this.span.setAttributes({
        [ATTR_RESULT]: 'error',
        [ATTR_ERROR_CODE]: code,
        [ATTR_RETRY_COUNT]: this.retryCount,
      });
      this.span.recordException(err);
      this.span.setStatus('error', code);
      this.span.end();
      this.span = null;
    }
    try {
      await this.input.close();
    } catch {
      /* swallow */
    }
    this.sender.destroy();
    this.rejectDone(err);
  }

  recordRetry(): void {
    this.retryCount++;
  }

  private flushEventEnd(): void {
    for (const r of this.eventResolvers) r({ value: undefined as never, done: true });
    this.eventResolvers = [];
  }

  private eventsAsyncIterable(): AsyncIterable<TransferEvent> {
    const self = this;
    return {
      [Symbol.asyncIterator](): AsyncIterator<TransferEvent> {
        return {
          next(): Promise<IteratorResult<TransferEvent>> {
            if (self.events.length > 0) {
              return Promise.resolve({ value: self.events.shift()!, done: false });
            }
            if (self.aborted || self.completedAt > 0) {
              return Promise.resolve({ value: undefined as never, done: true });
            }
            return new Promise((resolve) => self.eventResolvers.push(resolve));
          },
        };
      },
    };
  }
}

class IncomingState {
  startedAt = -1;
  accepted = false;
  declined = false;
  span: Span | null = null;
  sink: OutputSink | null = null;
  outputKind: 'pipe' | 'callback' | 'buffer' | 'file' | 'fileHandle' | undefined;
  onProgress: ((p: TransferProgress) => void) | undefined;
  onEvent: ((e: TransferEvent) => void) | undefined;

  private bufferedByLane = new Map<number, Map<number, Uint8Array>>();
  private rangeOrderCursor: Map<number, number> = new Map(); // laneId → expected seq for sink writes
  private currentLaneForSink = 0; // for range mode: which lane's bytes go to sink next
  private rrChunkIndex = 0; // for round-robin sink reordering
  private rrPending = new Map<number, Uint8Array>(); // chunkIndex → plaintext
  private rrLaneCount = 0;
  private partitionKind: 'range' | 'round-robin';

  private resolveDone!: (result: TransferResult) => void;
  private rejectDone!: (err: unknown) => void;
  private donePromise: Promise<TransferResult>;
  private aborted = false;
  private completedAt = -1;
  private events: TransferEvent[] = [];
  private eventResolvers: Array<(value: IteratorResult<TransferEvent>) => void> = [];

  private overallHasher = new StreamingSha256();
  private bytesProcessed = 0;
  private progress: ProgressTracker;
  private laneProgress: Map<number, LaneProgress>;

  readonly handle: TransferHandle;

  constructor(
    public readonly streamId: string,
    public readonly peerAddress: string,
    public readonly receiver: MultiLaneReceiver,
    public readonly metadata: StreamMetadata,
    public readonly lanes: LaneInitSpec[],
  ) {
    this.partitionKind = lanes[0]!.partition.kind;
    this.rrLaneCount = lanes.length;
    for (const lane of lanes) this.rangeOrderCursor.set(lane.laneId, 0);
    this.donePromise = new Promise<TransferResult>((resolve, reject) => {
      this.resolveDone = resolve;
      this.rejectDone = reject;
    });
    this.progress = new ProgressTracker(metadata.sizeBytes);
    this.laneProgress = new Map(
      lanes.map((l) => [
        l.laneId,
        {
          laneId: l.laneId,
          bytesSent: 0,
          ...(l.partition.kind === 'range'
            ? { bytesTotal: l.partition.endByte - l.partition.startByte }
            : {}),
          seq: 0,
          state: 'idle' as const,
        },
      ]),
    );
    this.handle = {
      streamId: this.streamId,
      events: this.eventsAsyncIterable(),
      pause: async () => { /* M-Stream-6 */ },
      resume: async () => { /* M-Stream-6 */ },
      abort: async (reason) => {
        await this.abort(reason ?? 'manual');
      },
      done: () => this.donePromise,
    };
  }

  async handleChunk(wireBytes: Uint8Array): Promise<ChunkAck> {
    if (this.aborted) {
      throw new TransferProtocolError('transfer aborted');
    }
    if (!this.accepted) {
      // Buffer until accepted? For v0.2.0 simplicity, reject. Consumer must
      // accept synchronously in the onIncomingTransfer handler.
      throw new TransferProtocolError('chunk arrived before accept()');
    }
    const dec = await this.receiver.decryptChunk(wireBytes);
    this.recordReceived(dec.laneId, dec.plaintext.length, dec.seq, dec.isLast);
    if (this.partitionKind === 'range') {
      await this.writeRangeChunk(dec.laneId, dec.seq, dec.plaintext);
    } else {
      // Round-robin: chunkIndex = lane.partition.lane + seq * laneCount
      const lane = this.lanes.find((l) => l.laneId === dec.laneId);
      if (lane === undefined || lane.partition.kind !== 'round-robin') {
        throw new TransferProtocolError('lane misconfigured');
      }
      const chunkIndex = lane.partition.lane + dec.seq * this.rrLaneCount;
      await this.writeRoundRobinChunk(chunkIndex, dec.plaintext);
    }
    return {
      lastSeq: dec.seq,
      bytesReceived: this.bytesProcessed,
    };
  }

  private async writeRangeChunk(
    laneId: number,
    seq: number,
    plaintext: Uint8Array,
  ): Promise<void> {
    // Per-lane in-order check is done by StreamReceiver. We just need to
    // advance writes when this is the currently-flushing lane.
    const expected = this.rangeOrderCursor.get(laneId) ?? 0;
    if (seq !== expected) {
      // Out-of-order within a lane is already rejected by StreamReceiver,
      // so this branch is defensive only.
      throw new TransferProtocolError(`unexpected seq ${seq} for lane ${laneId}`);
    }
    this.rangeOrderCursor.set(laneId, seq + 1);

    if (laneId !== this.currentLaneForSink) {
      // Buffer for later.
      let lb = this.bufferedByLane.get(laneId);
      if (lb === undefined) {
        lb = new Map();
        this.bufferedByLane.set(laneId, lb);
      }
      lb.set(seq, plaintext);
      return;
    }

    // We are the current lane: write directly.
    await this.appendToSink(plaintext);
    if (this.receiver.getLane(laneId)?.isFinished) {
      // Move to next lane and drain its buffer.
      this.currentLaneForSink++;
      while (this.currentLaneForSink < this.lanes.length) {
        const next = this.currentLaneForSink;
        const buffered = this.bufferedByLane.get(next);
        if (buffered !== undefined) {
          // Drain in seq order.
          let nextSeq = 0;
          while (buffered.has(nextSeq)) {
            await this.appendToSink(buffered.get(nextSeq)!);
            buffered.delete(nextSeq);
            nextSeq++;
          }
          this.bufferedByLane.delete(next);
        }
        if (!this.receiver.getLane(next)?.isFinished) break;
        this.currentLaneForSink++;
      }
    }
  }

  private async writeRoundRobinChunk(chunkIndex: number, plaintext: Uint8Array): Promise<void> {
    if (chunkIndex === this.rrChunkIndex) {
      await this.appendToSink(plaintext);
      this.rrChunkIndex++;
      while (this.rrPending.has(this.rrChunkIndex)) {
        await this.appendToSink(this.rrPending.get(this.rrChunkIndex)!);
        this.rrPending.delete(this.rrChunkIndex);
        this.rrChunkIndex++;
      }
    } else if (chunkIndex > this.rrChunkIndex) {
      this.rrPending.set(chunkIndex, plaintext);
    } else {
      throw new TransferProtocolError(`backward chunkIndex ${chunkIndex}`);
    }
  }

  private async appendToSink(plaintext: Uint8Array): Promise<void> {
    if (this.sink === null) throw new TransferProtocolError('no sink');
    this.overallHasher.update(plaintext);
    this.receiver.appendOverall(plaintext);
    await this.sink.write(plaintext);
  }

  recordReceived(laneId: number, bytes: number, seq: number, isLast: boolean): void {
    const lane = this.laneProgress.get(laneId);
    if (lane !== undefined) {
      lane.bytesSent += bytes;
      lane.seq = seq + 1;
      lane.state = isLast ? 'done' : 'sending';
      if (isLast) this.emit({ type: 'lane-done', laneId });
    }
    this.bytesProcessed += bytes;
    this.progress.add(bytes);
    const sample = this.progress.sample();
    const tp: TransferProgress = {
      streamId: this.streamId,
      bytesSent: this.bytesProcessed,
      bytesPerSecond: sample.bytesPerSecond,
      lanes: [...this.laneProgress.values()],
      ...(this.metadata.sizeBytes !== undefined ? { bytesTotal: this.metadata.sizeBytes } : {}),
      ...(sample.etaSeconds !== undefined ? { etaSeconds: sample.etaSeconds } : {}),
      ...(sample.percent !== undefined ? { percent: sample.percent } : {}),
    };
    this.onProgress?.(tp);
    this.emit({ type: 'progress', progress: tp });
  }

  async finalize(msg: StreamFinishMessage): Promise<void> {
    if (!this.accepted) return;
    if (!this.receiver.allLanesFinished) {
      const err = new TransferIntegrityError('finish before all lanes finished');
      await this.failWith(err);
      return;
    }
    // Verify lane fingerprints.
    const localFps = this.receiver.getLaneFingerprints();
    for (const expected of msg.laneSha256) {
      const local = localFps.find((f) => f.laneId === expected.laneId);
      if (local === undefined) {
        return this.failWith(
          new TransferIntegrityError(`finish references unknown laneId ${expected.laneId}`),
        );
      }
      if (bytesToBase64(local.sha256) !== expected.sha256) {
        return this.failWith(
          new TransferIntegrityError(`laneSha256 mismatch on lane ${expected.laneId}`),
        );
      }
    }
    const localOverall = bytesToBase64(this.overallHasher.digest());
    if (localOverall !== msg.overallSha256) {
      return this.failWith(new TransferIntegrityError('overallSha256 mismatch'));
    }
    if (this.sink !== null) await this.sink.finalize();
    const result: TransferResult = {
      streamId: this.streamId,
      bytesSent: this.bytesProcessed,
      sha256: hexFromBase64(msg.overallSha256),
      durationMs: nowMs() - this.startedAt,
    };
    if (this.outputKind === 'buffer' && this.sink !== null) {
      const bytes = this.sink.toBytes();
      if (bytes !== null) (result as TransferResult & { bytes?: Uint8Array }).bytes = bytes;
    }
    this.completedAt = nowMs();
    this.emit({ type: 'complete', streamId: result.streamId, sha256: result.sha256, durationMs: result.durationMs });
    this.flushEventEnd();
    if (this.span !== null) {
      this.span.setAttributes({
        [ATTR_RESULT]: 'ok',
        [ATTR_BYTES_BIN]: bytesBin(result.bytesSent),
      });
      this.span.setStatus('ok');
      this.span.end();
      this.span = null;
    }
    this.resolveDone(result);
    this.receiver.destroy();
  }

  async abort(reason: string): Promise<void> {
    if (this.aborted || this.completedAt > 0) return;
    this.aborted = true;
    if (this.sink !== null) await this.sink.abort(reason);
    this.emit({ type: 'abort', reason });
    this.flushEventEnd();
    if (this.span !== null) {
      this.span.setAttributes({
        [ATTR_RESULT]: 'abort',
        [ATTR_ERROR_CODE]: 'SHADE_TRANSFER_ABORT',
      });
      this.span.setStatus('error', reason);
      this.span.end();
      this.span = null;
    }
    this.receiver.destroy();
    this.rejectDone(new TransferAbortError(reason));
  }

  lanesProgressForResume(): Array<{ laneId: number; lastSeqAcked: number }> {
    const out: Array<{ laneId: number; lastSeqAcked: number }> = [];
    for (const lane of this.lanes) {
      const r = this.receiver.getLane(lane.laneId);
      const next = r ? Number(r.nextExpectedSequence) : 0;
      out.push({ laneId: lane.laneId, lastSeqAcked: next - 1 });
    }
    return out;
  }

  private async failWith(err: unknown): Promise<void> {
    if (this.aborted || this.completedAt > 0) return;
    this.aborted = true;
    this.emit({ type: 'error', error: err });
    this.flushEventEnd();
    if (this.span !== null) {
      const code = errorCodeOf(err);
      this.span.setAttributes({
        [ATTR_RESULT]: 'error',
        [ATTR_ERROR_CODE]: code,
      });
      this.span.recordException(err);
      this.span.setStatus('error', code);
      this.span.end();
      this.span = null;
    }
    if (this.sink !== null) await this.sink.abort('integrity-failure');
    this.receiver.destroy();
    this.rejectDone(err);
  }

  emit(event: TransferEvent): void {
    this.onEvent?.(event);
    if (this.eventResolvers.length > 0) {
      this.eventResolvers.shift()!({ value: event, done: false });
    } else {
      this.events.push(event);
    }
  }

  private flushEventEnd(): void {
    for (const r of this.eventResolvers) r({ value: undefined as never, done: true });
    this.eventResolvers = [];
  }

  private eventsAsyncIterable(): AsyncIterable<TransferEvent> {
    const self = this;
    return {
      [Symbol.asyncIterator](): AsyncIterator<TransferEvent> {
        return {
          next(): Promise<IteratorResult<TransferEvent>> {
            if (self.events.length > 0) {
              return Promise.resolve({ value: self.events.shift()!, done: false });
            }
            if (self.aborted || self.completedAt > 0) {
              return Promise.resolve({ value: undefined as never, done: true });
            }
            return new Promise((resolve) => self.eventResolvers.push(resolve));
          },
        };
      },
    };
  }
}

// ─── Helpers ─────────────────────────────────────────────────

function computeEffectiveLaneCount(
  requested: number,
  inputSize: number | undefined,
  chunkSize: number,
): number {
  if (!Number.isInteger(requested) || requested < 1) {
    throw new ValidationError(`lanes must be a positive integer`, 'lanes');
  }
  const cap = Math.min(64, requested);
  if (inputSize !== undefined && inputSize < 2 * chunkSize) return 1;
  return cap;
}

function resolvePartition(
  mode: TransferOptions['partition'],
  input: NormalizedInput,
): 'range' | 'round-robin' {
  if (mode === 'range') return 'range';
  if (mode === 'round-robin') return 'round-robin';
  // 'auto' default
  return input.randomAccess && input.size !== undefined ? 'range' : 'round-robin';
}

function bytesToBase64(bytes: Uint8Array): string {
  let bin = '';
  for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i]!);
  return btoa(bin);
}

function bytesToBase64Url(bytes: Uint8Array): string {
  return bytesToBase64(bytes).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
}

function base64UrlToBytes(s: string): Uint8Array {
  const padded = s.replace(/-/g, '+').replace(/_/g, '/');
  const pad = padded.length % 4 === 0 ? '' : '='.repeat(4 - (padded.length % 4));
  const bin = atob(padded + pad);
  const out = new Uint8Array(bin.length);
  for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);
  return out;
}

function bytesFromBase64Url(s: string): Uint8Array {
  return base64UrlToBytes(s);
}

function hex(bytes: Uint8Array): string {
  return Array.from(bytes, (b) => b.toString(16).padStart(2, '0')).join('');
}

function hexFromBase64(b64: string): string {
  const bin = atob(b64);
  let out = '';
  for (let i = 0; i < bin.length; i++) {
    out += bin.charCodeAt(i).toString(16).padStart(2, '0');
  }
  return out;
}

function nowMs(): number {
  return typeof performance !== 'undefined' ? performance.now() : Date.now();
}

function errorCodeOf(err: unknown): string {
  if (err === null || err === undefined) return 'SHADE_UNKNOWN';
  if (typeof err === 'object') {
    const code = (err as { code?: unknown }).code;
    if (typeof code === 'string' && code.length > 0) return code;
    const name = (err as { name?: unknown }).name;
    if (typeof name === 'string' && name.startsWith('Transfer')) return `SHADE_${name}`;
  }
  return 'SHADE_UNKNOWN';
}

// ─── Persister ───────────────────────────────────────────────

class Persister {
  /** Per-stream lock so saves never interleave. */
  private chains = new Map<string, Promise<unknown>>();
  /** Random per-process key when no deviceKey is supplied (in-memory only). */
  private fallbackKey: Uint8Array | null = null;

  constructor(
    private readonly crypto: CryptoProvider,
    private readonly store: ResumeStore,
    private readonly deviceKey: Uint8Array | undefined,
  ) {}

  private getKey(): Uint8Array {
    if (this.deviceKey !== undefined) return this.deviceKey;
    if (this.fallbackKey === null) this.fallbackKey = this.crypto.randomBytes(32);
    return this.fallbackKey;
  }

  private serialize<T>(streamId: string, work: () => Promise<T>): Promise<T> {
    const previous = this.chains.get(streamId) ?? Promise.resolve();
    const next = previous.catch(() => {}).then(work);
    this.chains.set(streamId, next);
    return next;
  }

  async saveOutgoing(state: OutgoingState, streamSecret: Uint8Array): Promise<void> {
    return this.serialize(state.streamId, async () => {
      const wrapped = await wrapStreamSecret(this.crypto, this.getKey(), streamSecret, state.streamId);
      const persisted: PersistedStreamState = {
        streamId: state.streamId,
        direction: 'send',
        peerAddress: state.peerAddress,
        status: 'active',
        metadataJson: JSON.stringify(state.metadata),
        partitionJson: JSON.stringify(state.lanes),
        laneStateJson: JSON.stringify(snapshotOutgoingLanes(state)),
        ioDescriptorJson: JSON.stringify({ kind: 'consumer-supplied' }),
        secretEnc: wrapped.ciphertext,
        secretNonce: wrapped.nonce,
        createdAt: Date.now(),
        updatedAt: Date.now(),
      };
      await this.store.save(persisted);
    });
  }

  async saveIncoming(state: IncomingState, streamSecret: Uint8Array): Promise<void> {
    return this.serialize(state.streamId, async () => {
      const wrapped = await wrapStreamSecret(this.crypto, this.getKey(), streamSecret, state.streamId);
      const persisted: PersistedStreamState = {
        streamId: state.streamId,
        direction: 'receive',
        peerAddress: state.peerAddress,
        status: 'active',
        metadataJson: JSON.stringify(state.metadata),
        partitionJson: JSON.stringify(state.lanes),
        laneStateJson: JSON.stringify(snapshotIncomingLanes(state)),
        ioDescriptorJson: JSON.stringify({ kind: 'consumer-supplied' }),
        secretEnc: wrapped.ciphertext,
        secretNonce: wrapped.nonce,
        createdAt: Date.now(),
        updatedAt: Date.now(),
      };
      await this.store.save(persisted);
    });
  }

  async markStatus(streamId: string, status: 'finished' | 'aborted'): Promise<void> {
    return this.serialize(streamId, async () => {
      const existing = await this.store.get(streamId);
      if (existing === null) return;
      existing.status = status;
      existing.updatedAt = Date.now();
      await this.store.save(existing);
    });
  }

  async load(streamId: string): Promise<{
    record: PersistedStreamState;
    streamSecret: Uint8Array;
  } | null> {
    const record = await this.store.get(streamId);
    if (record === null) return null;
    const streamSecret = await unwrapStreamSecret(
      this.crypto,
      this.getKey(),
      record.secretEnc,
      record.secretNonce,
      record.streamId,
    );
    return { record, streamSecret };
  }
}

function snapshotOutgoingLanes(
  state: OutgoingState,
): Array<{ laneId: number; nextSeq: number; bytesSent: number }> {
  const out: Array<{ laneId: number; nextSeq: number; bytesSent: number }> = [];
  for (const lane of state.lanes) {
    const lp = state['laneProgress'].get(lane.laneId);
    out.push({
      laneId: lane.laneId,
      nextSeq: lp?.seq ?? 0,
      bytesSent: lp?.bytesSent ?? 0,
    });
  }
  return out;
}

function snapshotIncomingLanes(
  state: IncomingState,
): Array<{ laneId: number; expectedSeq: number; bytesReceived: number }> {
  const out: Array<{ laneId: number; expectedSeq: number; bytesReceived: number }> = [];
  for (const lane of state.lanes) {
    const r = state.receiver.getLane(lane.laneId);
    out.push({
      laneId: lane.laneId,
      expectedSeq: r ? Number(r.nextExpectedSequence) : 0,
      bytesReceived: r ? Number(r.bytesReceived) : 0,
    });
  }
  return out;
}
