Class MessageStream

packages/bot-slack/src/message-stream.ts:24–93

const DEFAULT_MIN_INTERVAL_MS = 800;

export class MessageStream {
  private buffer = "";
  private posted = "";
  private queue: Promise<void> = Promise.resolve();
  private lastFlushedAt = 0;
  private flushTimer: NodeJS.Timeout | undefined;
  private readonly minIntervalMs: number;
  private readonly update: (text: string) => Promise<void>;

  constructor(config: MessageStreamConfig) {
    this.update = config.update;
    this.minIntervalMs = config.minIntervalMs ?? DEFAULT_MIN_INTERVAL_MS;
  }

  /** Replace the in-flight buffer (callers pass the accumulated text). */
  append(text: string): void {
    if (text === this.buffer) return;
    this.buffer = text;
    this.scheduleFlush();
  }

  /**
   * Mark the stream done. Cancels any pending throttled flush, enqueues a
   * final flush, and resolves once the entire queue (including the final
   * flush and anything previously in flight) has drained.
   *
   * After this resolves, the Slack message reflects the final buffer state.
   */
  async finish(): Promise<void> {
    if (this.flushTimer) {
      clearTimeout(this.flushTimer);
      this.flushTimer = undefined;
    }
    this.enqueueFlush();
    await this.queue;
  }

  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();
    }, delay);
  }

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

  private async flushNow(): Promise<void> {
    if (this.buffer === this.posted) return;
    const text = this.buffer;
    this.posted = text;
    try {
      await this.update(text);
    } catch (err) {