Class EditStream

packages/bot-telegram/src/chunked-edit-stream.ts:60–129

}

class EditStream {
  private buffer = "";
  private posted = "";
  private queue: Promise<void> = Promise.resolve();
  private lastFlushedAt = 0;
  private flushTimer: ReturnType<typeof setTimeout> | undefined;
  private readonly minIntervalMs: number;
  private readonly update: (text: string) => Promise<void>;
  /** Non-null when the terminal flush (from finish()) failed. */
  private terminalError: unknown = undefined;

  constructor(config: EditStreamConfig) {
    this.update = config.update;
    this.minIntervalMs = config.minIntervalMs;
  }

  append(text: string): void {
    if (text === this.buffer) return;
    this.buffer = text;
    this.scheduleFlush();
  }

  async finish(): Promise<void> {
    if (this.flushTimer) {
      clearTimeout(this.flushTimer);
      this.flushTimer = undefined;
    }
    this.enqueueFlush(/* terminal */ true);
    await this.queue;
    if (this.terminalError !== undefined) {
      throw this.terminalError;
    }
  }

  private scheduleFlush(): void {
    if (this.flushTimer) return;
    const elapsed = Date.now() - this.lastFlushedAt;
    const delay = Math.max(0, this.minIntervalMs - elapsed);
    this.flushTimer = setTimeout(() => {
      this.flushTimer = undefined;
      this.enqueueFlush(false);
    }, delay);
  }

  private enqueueFlush(terminal: boolean): void {
    this.queue = this.queue.then(() => this.flushNow(terminal));
  }

  private async flushNow(terminal: boolean): Promise<void> {
    if (this.buffer === this.posted) return;
    const text = this.buffer;
    try {
      await this.update(text);
      // Only advance posted after a successful edit so a transient failure
      // leaves the text eligible for retry on the next flush.
      this.posted = text;
    } catch (err) {
      if (terminal) {