import {
  CommandId,
  defaultInstanceIdForDriver,
  DEFAULT_MODEL,
  DEFAULT_MODEL_BY_PROVIDER,
  DEFAULT_PROVIDER_INTERACTION_MODE,
  MessageId,
  ORCHESTRATION_WS_METHODS,
  type OrchestrationShellStreamItem,
  type OrchestrationThreadStreamItem,
  ProjectId,
  ProviderDriverKind,
  ThreadId,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as Queue from "effect/Queue";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";

import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts";
import type { MeasuredWsClient } from "./NetworkTransferMeasurement.integration.ts";
import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts";
import {
  expectedRecordedAssistantText,
  makeRecordedTransferTurn,
  TRANSFER_HISTORY_TURN_COUNT,
} from "./fixtures/transferBudget.ts";

const TRANSFER_PROJECT_ID = ProjectId.make("transfer-budget-project");
export const TRANSFER_THREAD_ID = ThreadId.make("transfer-budget-thread");
export const TRANSFER_MEASURED_TURN_INDEX = TRANSFER_HISTORY_TURN_COUNT;

export function transferModelSelection(provider: ProviderDriverKind) {
  return {
    instanceId: defaultInstanceIdForDriver(provider),
    model: DEFAULT_MODEL_BY_PROVIDER[provider] ?? DEFAULT_MODEL,
  };
}

function turnTimestamp(turnIndex: number): string {
  return `2026-06-01T00:${String(turnIndex).padStart(2, "0")}:00.000Z`;
}

export const TRANSFER_MEASURED_TURN_CREATED_AT = turnTimestamp(TRANSFER_MEASURED_TURN_INDEX);

const waitForTurnQuiesced = Effect.fn("TransferBudget.waitForTurnQuiesced")(function* (
  harness: OrchestrationIntegrationHarness,
  checkpointTurnCount: number,
) {
  const receipt = yield* harness.waitForReceipt(
    (receipt): receipt is TurnProcessingQuiescedReceipt =>
      receipt.type === "turn.processing.quiesced" &&
      receipt.threadId === TRANSFER_THREAD_ID &&
      receipt.checkpointTurnCount === checkpointTurnCount,
  );
  yield* harness.drainProviderRuntime;
  yield* harness.drainCheckpointReactor;
  return receipt;
});

export const seedTransferBudgetHistory = Effect.fn("TransferBudget.seedHistory")(function* (
  harness: OrchestrationIntegrationHarness,
  provider: ProviderDriverKind,
) {
  if (!harness.adapterHarness) {
    return yield* Effect.die(new Error("Transfer budget history requires the replay adapter."));
  }

  const modelSelection = transferModelSelection(provider);
  yield* harness.engine.dispatch({
    type: "project.create",
    commandId: CommandId.make(`transfer:${provider}:project-create`),
    projectId: TRANSFER_PROJECT_ID,
    title: "Transfer Budget Project",
    workspaceRoot: harness.workspaceDir,
    defaultModelSelection: modelSelection,
    createdAt: turnTimestamp(0),
  });
  yield* harness.engine.dispatch({
    type: "thread.create",
    commandId: CommandId.make(`transfer:${provider}:thread-create`),
    threadId: TRANSFER_THREAD_ID,
    projectId: TRANSFER_PROJECT_ID,
    title: `${provider} transfer history`,
    modelSelection,
    runtimeMode: "approval-required",
    interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
    branch: "main",
    worktreePath: harness.workspaceDir,
    createdAt: turnTimestamp(0),
  });

  for (let turnIndex = 0; turnIndex < TRANSFER_HISTORY_TURN_COUNT; turnIndex += 1) {
    const response = makeRecordedTransferTurn(provider, turnIndex);
    if (turnIndex === 0) {
      yield* harness.adapterHarness.queueTurnResponseForNextSession(response);
    } else {
      yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response);
    }

    yield* harness.engine.dispatch({
      type: "thread.turn.start",
      commandId: CommandId.make(`transfer:${provider}:turn:${turnIndex + 1}`),
      threadId: TRANSFER_THREAD_ID,
      message: {
        messageId: MessageId.make(`transfer-user-${turnIndex + 1}`),
        role: "user",
        text: `Inspect transfer behavior for historical turn ${turnIndex + 1}.`,
        attachments: [],
      },
      modelSelection,
      runtimeMode: "approval-required",
      interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
      createdAt: turnTimestamp(turnIndex),
    });
    yield* waitForTurnQuiesced(harness, turnIndex + 1);
  }
});

export const queueMeasuredTransferTurn = Effect.fn("TransferBudget.queueMeasuredTurn")(function* (
  harness: OrchestrationIntegrationHarness,
  provider: ProviderDriverKind,
) {
  if (!harness.adapterHarness) {
    return yield* Effect.die(new Error("Transfer budget measurement requires the replay adapter."));
  }
  const response = makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX);
  yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response);
});

export function expectedMeasuredAssistantText(provider: ProviderDriverKind): string {
  return expectedRecordedAssistantText(provider, TRANSFER_MEASURED_TURN_INDEX);
}

/** Takes from the queue until one value matches, and returns everything taken. */
export const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function* <A>(
  queue: Queue.Queue<A>,
  predicate: (value: A) => boolean,
  waitDescription: string,
) {
  return yield* Effect.gen(function* () {
    const values: A[] = [];
    while (true) {
      const value = yield* Queue.take(queue);
      values.push(value);
      if (predicate(value)) return values;
    }
  }).pipe(
    Effect.timeoutOrElse({
      duration: "10 seconds",
      orElse: () => Effect.die(new Error(`Timed out waiting for ${waitDescription}`)),
    }),
  );
});

/**
 * Resumes the thread subscription from a cursor on a measured client. The
 * consumer runs in the client's scope, so closing the client stops it. Callers
 * read items from the returned queue.
 */
export const subscribeThreadItems = Effect.fn("TransferBudget.subscribeThreadItems")(function* (
  measured: MeasuredWsClient,
  afterSequence: number,
) {
  const items = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
  yield* measured.client[ORCHESTRATION_WS_METHODS.subscribeThread]({
    threadId: TRANSFER_THREAD_ID,
    afterSequence,
    requestCompletionMarker: true,
  }).pipe(
    Stream.runForEach((item) => Queue.offer(items, item).pipe(Effect.asVoid)),
    Scope.provide(measured.scope),
    Effect.forkIn(measured.scope),
  );
  return items;
});

/** Shell counterpart of subscribeThreadItems. */
export const subscribeShellItems = Effect.fn("TransferBudget.subscribeShellItems")(function* (
  measured: MeasuredWsClient,
  afterSequence: number,
) {
  const items = yield* Queue.unbounded<OrchestrationShellStreamItem>();
  yield* measured.client[ORCHESTRATION_WS_METHODS.subscribeShell]({
    afterSequence,
    requestCompletionMarker: true,
  }).pipe(
    Stream.runForEach((item) => Queue.offer(items, item).pipe(Effect.asVoid)),
    Scope.provide(measured.scope),
    Effect.forkIn(measured.scope),
  );
  return items;
});

/** Waits for the initial catch-up and reports whether it was a replay or a snapshot reset. */
export const awaitSubscriptionSynchronized = Effect.fn(
  "TransferBudget.awaitSubscriptionSynchronized",
)(function* <Item extends OrchestrationThreadStreamItem | OrchestrationShellStreamItem>(
  items: Queue.Queue<Item>,
  waitDescription: string,
) {
  const initial = yield* collectQueueUntil(
    items,
    (item) => item.kind === "synchronized",
    waitDescription,
  );
  return initial.some((item) => item.kind === "snapshot")
    ? ("snapshot" as const)
    : ("replay" as const);
});

export { TRANSFER_HISTORY_TURN_COUNT, waitForTurnQuiesced };