Class TeamsMessageStream

packages/bot-teams/src/message-stream.ts:32–106

const DEFAULT_MIN_INTERVAL_MS = 700;

export class TeamsMessageStream {
  private buffer = "";
  private posted = "";
  private id: string | undefined;
  private queue: Promise<void> = Promise.resolve();
  private lastFlushedAt = 0;
  private flushTimer: ReturnType<typeof setTimeout> | undefined;
  private readonly minIntervalMs: number;
  private readonly config: TeamsMessageStreamConfig;

  constructor(config: TeamsMessageStreamConfig) {
    this.config = config;
    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: cancel any pending throttled flush, enqueue a final
   * flush, and resolve once the whole queue has drained. The posted activity
   * then reflects the final buffer. Returns the activity id (or `undefined` if
   * nothing was ever posted, e.g. an empty stream).
   */
  async finish(): Promise<string | undefined> {
    if (this.flushTimer) {
      clearTimeout(this.flushTimer);
      this.flushTimer = undefined;
    }
    this.enqueueFlush();
    await this.queue;
    return this.id;
  }

  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> {
    const text = this.buffer;
    if (text === this.posted) return;
    // Don't post an empty first message; wait for real content.
    if (this.id === undefined && text.trim().length === 0) return;
    this.posted = text;