curriculum-project-hub/hub/src/feishu/textStream.ts

export interface PatchableTextSink {
  readonly create: (text: string) => Promise<string | null>;
  readonly patch: (messageId: string, text: string) => Promise<void>;
}

export interface PatchableTextStreamOptions {
  readonly patchIntervalMs?: number;
  readonly maxMessageLength?: number;
}

const DEFAULT_PATCH_INTERVAL_MS = 400;
export const DEFAULT_MAX_MESSAGE_LENGTH = 8000;

interface CodeBlockRange {
  readonly start: number;
  readonly end: number;
}

export function splitAtBoundary(text: string, maxLength: number): string[] {
  if (!Number.isInteger(maxLength) || maxLength <= 0) {
    throw new RangeError("maxLength must be a positive integer");
  }
  if (text.length <= maxLength) return [text];

  const chunks: string[] = [];
  let remaining = text;
  while (remaining.length > maxLength) {
    const splitPoint = chooseSplitPoint(remaining, maxLength);
    if (splitPoint <= 0) {
      const codeBlockEnd = codeBlockAtStartEnd(remaining);
      const forcedSplitPoint = codeBlockEnd ?? maxLength;
      chunks.push(remaining.slice(0, forcedSplitPoint));
      remaining = remaining.slice(forcedSplitPoint);
      continue;
    }

    chunks.push(remaining.slice(0, splitPoint));
    remaining = remaining.slice(splitPoint);
  }
  if (remaining.length > 0) {
    chunks.push(remaining);
  }
  return chunks;
}

function chooseSplitPoint(text: string, maxLength: number): number {
  const codeBlocks = findFencedCodeBlocks(text);
  const splitCodeBlock = codeBlocks.find((range) => maxLength > range.start && maxLength < range.end);
  if (splitCodeBlock !== undefined) {
    return splitCodeBlock.start;
  }

  return (
    lastBoundary(text, maxLength, /\r?\n\r?\n/g, codeBlocks) ??
    lastBoundary(text, maxLength, /\r?\n/g, codeBlocks) ??
    lastBoundary(text, maxLength, / /g, codeBlocks) ??
    maxLength
  );
}

function lastBoundary(text: string, maxLength: number, pattern: RegExp, codeBlocks: ReadonlyArray<CodeBlockRange>): number | undefined {
  let boundary: number | undefined;
  let match: RegExpExecArray | null;
  pattern.lastIndex = 0;
  while ((match = pattern.exec(text)) !== null) {
    const splitPoint = match.index + match[0].length;
    if (splitPoint > maxLength) break;
    if (splitPoint > 0 && !isInsideCodeBlock(splitPoint, codeBlocks)) {
      boundary = splitPoint;
    }
  }
  return boundary;
}

function isInsideCodeBlock(splitPoint: number, codeBlocks: ReadonlyArray<CodeBlockRange>): boolean {
  return codeBlocks.some((range) => splitPoint > range.start && splitPoint < range.end);
}

function codeBlockAtStartEnd(text: string): number | undefined {
  const firstCodeBlock = findFencedCodeBlocks(text)[0];
  return firstCodeBlock?.start === 0 ? firstCodeBlock.end : undefined;
}

function findFencedCodeBlocks(text: string): CodeBlockRange[] {
  const ranges: CodeBlockRange[] = [];
  let openStart: number | null = null;
  let lineStart = 0;

  while (lineStart < text.length) {
    const newlineIndex = text.indexOf("\n", lineStart);
    const lineEnd = newlineIndex === -1 ? text.length : newlineIndex;
    const nextLineStart = newlineIndex === -1 ? text.length : newlineIndex + 1;
    const line = text.slice(lineStart, lineEnd);

    if (/^ {0,3}```/.test(line)) {
      if (openStart === null) {
        openStart = lineStart;
      } else {
        ranges.push({ start: openStart, end: nextLineStart });
        openStart = null;
      }
    }

    lineStart = nextLineStart;
  }

  if (openStart !== null) {
    ranges.push({ start: openStart, end: text.length });
  }

  return ranges;
}

/**
 * Serializes text-card creation/patching for streaming output.
 *
 * Feishu returns the message id asynchronously. Without this queue, multiple
 * early text deltas can all observe "no message yet" and create duplicate
 * prefix messages before the first create call resolves.
 */
export class PatchableTextStream {
  private currentMessageId: string | null = null;
  private text = "";
  private lastPatchAt = 0;
  private flushChain: Promise<void> = Promise.resolve();
  private flushScheduled = false;

  private readonly patchIntervalMs: number;
  private readonly maxMessageLength: number;

  constructor(
    private readonly sink: PatchableTextSink,
    options: PatchableTextStreamOptions = {},
  ) {
    this.patchIntervalMs = options.patchIntervalMs ?? DEFAULT_PATCH_INTERVAL_MS;
    this.maxMessageLength = options.maxMessageLength ?? DEFAULT_MAX_MESSAGE_LENGTH;
  }

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

  async finish(fallbackText: string): Promise<void> {
    await this.flushChain;
    // If there's pending text to flush, do it now.
    if (this.text.length > 0) {
      await this.flushTextToSink();
      return;
    }
    // No pending text. If we never sent anything, send the fallback.
    if (this.currentMessageId === null && fallbackText.length > 0) {
      this.text = fallbackText;
      await this.flushTextToSink();
    }
  }

  private scheduleFlush(): void {
    if (this.flushScheduled) return;
    this.flushScheduled = true;
    this.flushChain = this.flushChain.then(async () => {
      this.flushScheduled = false;
      await this.flush();
    });
  }

  private async flush(): Promise<void> {
    if (this.currentMessageId === null) {
      await this.flushTextToSink();
      this.lastPatchAt = Date.now();
      return;
    }

    const now = Date.now();
    if (now - this.lastPatchAt < this.patchIntervalMs) return;
    this.lastPatchAt = now;

    await this.flushTextToSink();
  }

  private async flushTextToSink(): Promise<void> {
    const textToFlush = this.text;
    const chunks = splitAtBoundary(textToFlush, this.maxMessageLength);
    const firstChunk = chunks[0];
    if (firstChunk === undefined) return;

    if (this.currentMessageId === null) {
      // First message: create it with the first chunk, then send overflow
      // chunks as new messages.
      this.currentMessageId = await this.sink.create(firstChunk);
      for (const chunk of chunks.slice(1)) {
        this.currentMessageId = await this.sink.create(chunk);
      }
      // Keep only the last chunk for future patches; remove already-sent prefix.
      const sentLength = chunks.slice(0, -1).reduce((sum, c) => sum + c.length, 0);
      this.text = this.text.slice(sentLength);
    } else {
      // Patch the current message with the first chunk, then create new
      // messages for overflow chunks.
      await this.sink.patch(this.currentMessageId, firstChunk);
      for (const chunk of chunks.slice(1)) {
        this.currentMessageId = await this.sink.create(chunk);
      }
      // Keep only the last chunk for future patches; remove already-sent prefix.
      const sentLength = chunks.slice(0, -1).reduce((sum, c) => sum + c.length, 0);
      this.text = this.text.slice(sentLength);
    }
  }
}