/**
 * Queue types and data structure extracted from Conversation.
 *
 * Conversation uses MessageQueue to manage the message backlog while an
 * agent loop is in flight.
 */

import type { AssistantEvent } from "../api/index.js";
import type {
  TurnChannelContext,
  TurnInterfaceContext,
} from "../channels/types.js";
import type { AuthContext } from "../runtime/auth/types.js";
import { getLogger } from "../util/logger.js";
import type { UserMessageAttachment } from "./message-protocol.js";
import type { ConversationTransportMetadata } from "./message-types/conversations.js";
import type { TrustContext } from "./trust-context-types.js";

const log = getLogger("conversation-queue");

export interface QueuedMessage {
  content: string;
  attachments: UserMessageAttachment[];
  requestId: string;
  onEvent: (msg: AssistantEvent) => void;
  activeSurfaceId?: string;
  currentPage?: string;
  metadata?: Record<string, unknown>;
  turnChannelContext?: TurnChannelContext;
  turnInterfaceContext?: TurnInterfaceContext;
  /** When false, the turn has no interactive user and should skip clarification prompts. */
  isInteractive?: boolean;
  /** Requester identity captured from the verified auth context at enqueue time. */
  sourceActorPrincipalId?: string;
  /** Full auth snapshot captured at enqueue time for turn-scoped authorization decisions. */
  authContext?: AuthContext;
  /**
   * Sender's trust captured at enqueue time. The conversation-level
   * `trustContext` is a single mutable slot holding whichever actor last sent
   * a message, so a drain that read it would run this message under a
   * different actor's trust whenever another actor sent in between. Trust
   * governs `trustClass`, `executionChannel`, and `requesterExternalUserId`,
   * so reading the wrong one misroutes the whole tool-approval path.
   */
  trustContext?: TrustContext;
  /** Transport metadata snapshot captured at enqueue time, applied when this message becomes active. */
  transport?: ConversationTransportMetadata;
  /** Original user message text to persist to DB when recording intent stripping produced a different `content`. */
  displayContent?: string;
  /** Wall-clock time (ms since epoch) when the message was enqueued, used as the display timestamp. */
  sentAt: number;
  /** Client-generated correlation nonce. Echoed back on `user_message_echo`
   *  so the originating client can dedupe its optimistic row. */
  clientMessageId?: string;
  /**
   * True once a drain has told clients this message was dequeued and before
   * the turn it was dequeued for actually took over. A drain that sends the
   * message back to the queue after that point owes clients the corrective
   * `message_requeued`; one that gives up before announcing owes nothing.
   * Set by the drain's dequeue announcement, cleared by the requeue that
   * settles it.
   */
  dequeueAnnounced?: boolean;
}

/**
 * Maximum total estimated bytes across all queued messages per conversation.
 * Limits memory consumption when a sender floods messages while the
 * conversation is busy.  50 MB is well above any legitimate usage.
 */
export const DEFAULT_MAX_QUEUE_BYTES = 50 * 1024 * 1024; // 50 MB

/**
 * Describes why a queued message was promoted from the queue.
 * - `loop_complete`: the agent loop finished normally and the next message was drained.
 * - `checkpoint_handoff`: a turn-boundary checkpoint decided to yield to the queued message.
 */
export type QueueDrainReason = "loop_complete" | "checkpoint_handoff";

/**
 * Configuration for how/when checkpoint handoff is allowed.
 * When `checkpointHandoffEnabled` is true, the agent loop may yield at
 * a turn boundary if there are queued messages waiting.
 */
export interface QueuePolicy {
  checkpointHandoffEnabled: boolean;
}

/**
 * Typed wrapper around the queued-message array.
 *
 * Conversation owns one instance; the wrapper handles iteration
 * so the rest of Conversation doesn't touch the raw array.
 *
 * A byte budget caps total memory held by queued messages so a
 * high-rate sender cannot exhaust the process.
 */
export class MessageQueue {
  private items: QueuedMessage[] = [];
  private currentBytes = 0;
  private maxBytes: number;

  constructor(maxBytes: number = DEFAULT_MAX_QUEUE_BYTES) {
    this.maxBytes = maxBytes;
  }

  /**
   * Attempt to enqueue a message.
   * Returns `true` if accepted, `false` if rejected (over budget).
   */
  push(item: QueuedMessage): boolean {
    const itemBytes = estimateItemBytes(item);
    if (
      this.currentBytes + itemBytes > this.maxBytes &&
      this.items.length > 0
    ) {
      log.warn(
        {
          requestId: item.requestId,
          queueDepth: this.items.length,
          currentBytes: this.currentBytes,
          itemBytes,
          maxBytes: this.maxBytes,
        },
        "Rejecting queued message: queue byte budget exceeded",
      );
      return false;
    }
    this.items.push(item);
    this.currentBytes += itemBytes;
    return true;
  }

  shift(): QueuedMessage | undefined {
    const item = this.items.shift();
    if (item) {
      this.currentBytes -= estimateItemBytes(item);
    }
    return item;
  }

  /**
   * Requeue a message at the FRONT of the queue. Used when a drained
   * message hits processing-lock contention and must run on the next
   * drain instead of being dropped. Deliberately skips the byte-budget
   * check: the item was already accepted (and just popped), so a requeue
   * must never reject it.
   */
  unshift(item: QueuedMessage): void {
    this.items.unshift(item);
    this.currentBytes += estimateItemBytes(item);
  }

  /**
   * Read-only access to a queued message by index without mutating the queue.
   * Returns `undefined` when the index is out of range.
   */
  peek(index: number = 0): QueuedMessage | undefined {
    return this.items[index];
  }

  /** FIFO snapshot of the queued messages. The array is a copy; the items are
   * the live references (callers must treat them as read-only). */
  snapshot(): QueuedMessage[] {
    return [...this.items];
  }

  /**
   * Pop up to `count` messages FIFO and return them in order.
   * Decrements the byte budget for each popped item using the same
   * accounting as `shift` / `removeByRequestId`, keeping the bookkeeping
   * centralized here rather than at call sites.
   */
  shiftN(count: number): QueuedMessage[] {
    const n = Math.min(Math.max(0, count), this.items.length);
    if (n === 0) {
      return [];
    }
    const removed = this.items.splice(0, n);
    for (const item of removed) {
      this.currentBytes -= estimateItemBytes(item);
    }
    return removed;
  }

  clear(): void {
    this.items = [];
    this.currentBytes = 0;
  }

  get length(): number {
    return this.items.length;
  }

  get isEmpty(): boolean {
    return this.items.length === 0;
  }

  get totalBytes(): number {
    return this.currentBytes;
  }

  /**
   * Move a queued message to the head of the queue (index 0) by its requestId.
   * Returns the promoted message, or undefined if not found.
   * Byte accounting is unchanged — the item stays in the queue, just reordered.
   */
  promoteToHead(requestId: string): QueuedMessage | undefined {
    const idx = this.items.findIndex((m) => m.requestId === requestId);
    if (idx === -1) {
      return undefined;
    }
    if (idx === 0) {
      return this.items[0];
    } // already at head
    const [promoted] = this.items.splice(idx, 1);
    this.items.unshift(promoted);
    return promoted;
  }

  /**
   * Read-only lookup of a queued message by its requestId. Returns the live
   * reference (callers must treat it as read-only), or undefined when this
   * queue holds no such message. Lets a caller inspect an item (to
   * authorize a delete, say) before deciding whether to remove it.
   */
  findByRequestId(requestId: string): QueuedMessage | undefined {
    return this.items.find((m) => m.requestId === requestId);
  }

  /**
   * Remove a queued message by its requestId.
   * Returns the removed message, or undefined if not found.
   */
  removeByRequestId(requestId: string): QueuedMessage | undefined {
    const idx = this.items.findIndex((m) => m.requestId === requestId);
    if (idx === -1) {
      return undefined;
    }
    const [removed] = this.items.splice(idx, 1);
    this.currentBytes -= estimateItemBytes(removed);
    return removed;
  }

  [Symbol.iterator](): Iterator<QueuedMessage> {
    return this.items[Symbol.iterator]();
  }
}

/**
 * Estimate the in-memory byte cost of a queued message.
 * Dominated by content text and attachment `data` (base64 strings).
 */
function estimateItemBytes(item: QueuedMessage): number {
  let bytes = item.content.length * 2; // JS strings are UTF-16
  for (const a of item.attachments) {
    bytes += a.data.length * 2;
    if (a.extractedText) {
      bytes += a.extractedText.length * 2;
    }
  }
  // Include transport metadata in the estimate so large transport
  // payloads (e.g. hostHomeDir, hostUsername) count against the budget.
  if (item.transport) {
    bytes += JSON.stringify(item.transport).length * 2;
  }
  if (item.sourceActorPrincipalId) {
    bytes += item.sourceActorPrincipalId.length * 2;
  }
  // Small fixed overhead for metadata, pointers, etc. (not worth
  // measuring precisely — the content/attachment data dominates).
  bytes += 512;
  return bytes;
}