/**
 * @license
 * Copyright 2025 Vybestack LLC
 * SPDX-License-Identifier: Apache-2.0
 */

/**
 * StreamBatcher batches live streaming text/thought chunks from the agent turn
 * and flushes them to the ACP client as agent_message_chunk / agent_thought_chunk
 * session/update notifications on a short interval, routing every chunk through
 * the session's EmojiFilter (with blocked-response handling and a trailing
 * buffer flush).
 *
 * Extracted from zedIntegration.ts into its own module so the integration file
 * stays within the max-lines complexity budget; focused behavior is exercised by
 * zed-stream-batcher.test.ts and prompt integration coverage.
 */

import type * as acp from '@agentclientprotocol/sdk';
import { type EmojiFilter, DebugLogger } from '@vybestack/llxprt-code-core';

// 100ms balances streaming latency (users see text quickly) with throughput
// (fewer, larger ACP notifications). Tuned for local stdio transport; not
// configurable because the ACP protocol expects near-real-time streaming.
const BATCH_INTERVAL_MS = 100;

/**
 * The message emitted (as an agent_message_chunk) when the EmojiFilter blocks a
 * streamed response in error mode (FINDING E2). Exported so tests reference this
 * single source of truth instead of duplicating the literal, keeping the wire
 * text and its assertions in lockstep. Intentionally hardcoded — not i18n'd
 * because the ACP protocol requires deterministic, locale-independent status.
 */
export const STREAM_BLOCKED_MESSAGE =
  '[Error: Response blocked due to emoji detection]';

export class StreamBatcher {
  private pendingChunks: Array<{ kind: 'text' | 'thought'; text: string }> = [];
  private batchTimer: ReturnType<typeof setTimeout> | null = null;
  private flushChain: Promise<void> = Promise.resolve();
  private disposed = false;
  private readonly logger: DebugLogger;

  constructor(
    private readonly emojiFilter: EmojiFilter,
    private readonly sendUpdate: (update: acp.SessionUpdate) => Promise<void>,
    logger?: DebugLogger,
  ) {
    // Injectable for testability; defaults to the shared zed-integration
    // namespace so the otherwise-silent flush-chain / per-chunk failures
    // (FINDING F4/F17) are diagnosable.
    this.logger =
      logger ?? new DebugLogger('llxprt:zed-integration:stream-batcher');
  }

  append(text: string, isThought: boolean): void {
    if (this.disposed) {
      return;
    }
    const filterResult = isThought
      ? this.emojiFilter.filterText(text)
      : this.emojiFilter.filterStreamChunk(text);
    if (filterResult.blocked) {
      // filterText() is stateless, so a blocked thought must not clear partial
      // text held by filterStreamChunk() for a later streaming boundary.
      if (!isThought) {
        const residual = this.emojiFilter.flushBuffer();
        if (residual.length > 0) {
          this.appendPendingChunk('text', residual);
        }
      }
      // FINDING E1: clear any pending batch timer BEFORE building the blocked
      // chain (exactly as flush() does). Otherwise a timer armed by a prior
      // normal chunk survives and later fires its own flush() — appending a
      // SECOND doFlush()+flushEmojiBuffer chain that races the blocked-path
      // chain, re-flushing already-flushed content after the error message. The
      // blocked path here already flushes the residual + queued chunks, so the
      // timer has nothing left to do.
      if (this.batchTimer !== null) {
        clearTimeout(this.batchTimer);
        this.batchTimer = null;
      }
      const pending = this.flushChain
        .then(() => this.doFlush())
        .then(() =>
          this.sendUpdate({
            sessionUpdate: isThought
              ? 'agent_thought_chunk'
              : 'agent_message_chunk',
            content: {
              type: 'text',
              text: STREAM_BLOCKED_MESSAGE,
            },
          }),
        );
      this.flushChain = this.settleChainLink(pending);
      return;
    }
    const filteredText =
      typeof filterResult.filtered === 'string' ? filterResult.filtered : '';
    if (filteredText.length === 0) {
      return;
    }
    this.appendPendingChunk(isThought ? 'thought' : 'text', filteredText);
    this.batchTimer ??= setTimeout(() => {
      this.batchTimer = null;
      void this.flush().catch(() => undefined);
    }, BATCH_INTERVAL_MS);
  }

  async flush(): Promise<void> {
    if (this.batchTimer !== null) {
      clearTimeout(this.batchTimer);
      this.batchTimer = null;
    }
    // try/finally so the trailing emoji-buffer flush runs even if a future
    // doFlush change introduces a rejection path — the residual filter content
    // must never be stranded behind a failed chunk send.
    const pending = this.flushChain.then(async () => {
      try {
        await this.doFlush();
      } finally {
        await this.flushEmojiBuffer();
      }
    });
    this.flushChain = this.settleChainLink(pending);
    await pending;
  }

  /**
   * Clears any pending batch timer and drops buffered chunks so no timer fires
   * after the owning prompt completes/aborts (FINDING F9), and marks the batcher
   * disposed so a late {@link append} after the turn ends is a silent no-op
   * (nothing new can enter the flush chain). Idempotent and safe to call after
   * {@link flush}; it does NOT emit, so any pending chunks must be flushed first
   * (the prompt path flushes then disposes). An in-flight flush is not aborted:
   * chunks it already detached may finish sending while the chain settles.
   */
  dispose(): void {
    this.disposed = true;
    if (this.batchTimer !== null) {
      clearTimeout(this.batchTimer);
      this.batchTimer = null;
    }
    this.pendingChunks = [];
    // Drop residual streaming state without emitting after the prompt boundary.
    // The batcher/filter are per-prompt, but draining here also makes disposal
    // complete if that ownership model changes later.
    this.emojiFilter.flushBuffer();
  }

  private async flushEmojiBuffer(): Promise<void> {
    // A dispose() that raced an in-flight chain link means the prompt is over:
    // do not re-queue residual filter content past the turn boundary. (The
    // standard prompt pattern — await flush() in try, dispose() in finally —
    // never hits this: dispose only runs after the flush chain link settled.)
    if (this.disposed) {
      return;
    }
    const remaining = this.emojiFilter.flushBuffer();
    if (remaining.length === 0) {
      return;
    }
    this.appendPendingChunk('text', remaining);
    await this.doFlush();
  }

  private appendPendingChunk(kind: 'text' | 'thought', text: string): void {
    const lastChunk = this.pendingChunks.at(-1);
    if (lastChunk?.kind === kind) {
      lastChunk.text += text;
      return;
    }
    this.pendingChunks.push({ kind, text });
  }

  private async doFlush(): Promise<void> {
    const chunks = this.pendingChunks;
    this.pendingChunks = [];
    let firstError: unknown = null;
    let lastError: unknown = null;
    let failureCount = 0;
    // FINDING F4: iterate a detached local copy and CONTINUE past a per-chunk
    // send failure so a single rejecting sendUpdate does not drop every later
    // chunk. The injected sendUpdate is best-effort today (Session.sendUpdate
    // swallows), so this cannot reject in practice — but the injected contract
    // is not guaranteed, so no chunk is silently lost if it ever does. The first
    // and last distinct errors are collected + logged (not rethrown) to keep
    // the flushChain intact.
    for (const chunk of chunks) {
      try {
        await this.sendUpdate({
          sessionUpdate:
            chunk.kind === 'thought'
              ? 'agent_thought_chunk'
              : 'agent_message_chunk',
          content: { type: 'text', text: chunk.text },
        });
      } catch (error) {
        failureCount += 1;
        if (firstError === null) {
          firstError = error;
        }
        lastError = error;
      }
    }
    if (firstError !== null) {
      const firstMsg =
        firstError instanceof Error ? firstError.message : String(firstError);
      const lastMsg =
        lastError instanceof Error ? lastError.message : String(lastError);
      this.logger.debug(
        () =>
          `doFlush: ${failureCount} chunk update(s) failed to send; the remaining chunks were still attempted (no chunk dropped). First error: ${firstMsg}${
            lastMsg !== firstMsg ? `. Last error: ${lastMsg}` : ''
          }`,
      );
    }
  }

  /**
   * Terminates a flush-chain link so a rejection cannot poison the serialized
   * chain for subsequent appends, while still surfacing the swallowed error via
   * the logger (FINDING F17) instead of discarding it silently.
   */
  private settleChainLink(pending: Promise<void>): Promise<void> {
    return pending.catch((error: unknown) => {
      this.logger.debug(
        () =>
          `flush chain link failed (swallowed to keep the batch chain alive): ${
            error instanceof Error ? error.message : String(error)
          }`,
      );
    });
  }
}