23444678171

Committed 23 Mar 2026 03:13PM UTC coverage: 83.115% (+1.0%) from 82.159%

Build # 23444678171

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #932 from CBIIT/CRDCDH-3398

CRDCDH-3398 Chatbot Feature

Run Details

5920 of 6492 branches covered (91.19%)

Branch coverage included in aggregate %.

2413 of 2440 new or added lines in 21 files covered. (98.89%)

1 existing line in 1 file now uncovered.

34906 of 42628 relevant lines covered (81.89%)

245.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.16

/src/components/ChatBot/api/knowledgeBaseClient.ts

import { Logger } from "@/utils";

import chatConfig from "../config/chatConfig";
import { storeSessionId } from "../utils/sessionStorageUtils";

export type AskKnowledgeBaseResponse = {
  question: string;
  answer: string;
  citations?: ChatCitation[];
  sessionId?: string;
};

type AskQuestionArgs = {
  question: string;
  sessionId?: string | null;
  conversationHistory?: ConversationHistory[];
  onChunk?: (chunk: string) => void;
  onCitation?: (citation: ChatCitation) => void;
  onPulse?: (description: string) => void;
  signal?: AbortSignal;
  url: string;
  typewriterDelay?: number;
};

type AskQuestionResult = {
  sessionId: string | null;
  citations: ChatCitation[];
};

/**
 * Emits text with a typewriter effect (character by character) using a configurable delay.
 *
 * @param {string} text - The text to emit character by character
 * @param {(chunk: string) => void} [onChunk] - Optional callback to invoke with each character
 * @param {number} [typewriterDelay=10] - Delay in milliseconds between characters (0 to disable typewriter effect)
 * @param {AbortSignal} [signal] - Optional abort signal to stop the typewriter effect
 * @returns {Promise<void>}
 */
export async function emitWithTypewriter(
  text: string,
  onChunk?: (chunk: string) => void,
  typewriterDelay = 10,
  signal?: AbortSignal
): Promise<void> {
  if (!onChunk || !text) {
    return;
  }

  // If typewriter delay is 0, emit the whole text at once
  if (typewriterDelay === 0) {
    onChunk(text);
    return;
  }

  for (let i = 0; i < text.length; i += 1) {
    if (signal?.aborted) {
      return;
    }
    onChunk(text[i]);
    // eslint-disable-next-line no-await-in-loop
    await new Promise<void>((resolve) => {
      setTimeout(resolve, typewriterDelay);
    });
  }
}

/**
 * Processes the streaming response from the knowledge base API with typewriter effect.
 *
 * @param {ReadableStreamDefaultReader<Uint8Array>} reader - The stream reader for the response body
 * @param {(chunk: string) => void} [onChunk] - Optional callback to invoke with each text chunk
 * @param {(citation: ChatCitation) => void} [onCitation] - Optional callback to invoke with each citation
 * @param {number} [typewriterDelay=10] - Delay in milliseconds between characters (0 to disable typewriter effect)
 * @param {AbortSignal} [signal] - Optional abort signal to stop the typewriter effect
 * @param {(description: string) => void} [onPulse] - Optional callback to invoke with pulse status descriptions
 * @returns {Promise<{ sessionId: string | null; citations: ChatCitation[] }>} The session ID and collected citations from the response
 */
export async function processStreamingResponse(
  reader: ReadableStreamDefaultReader<Uint8Array>,
  onChunk?: (chunk: string) => void,
  onCitation?: (citation: ChatCitation) => void,
  typewriterDelay = 10,
  signal?: AbortSignal,
  onPulse?: (description: string) => void
): Promise<{ sessionId: string | null; citations: ChatCitation[] }> {
  const decoder = new TextDecoder();
  let buffer = "";
  let currentSessionId: string | null = null;
  const citations: ChatCitation[] = [];
  const seenCitationLinks = new Set<string>();
  let done = false;

  while (!done) {
    // eslint-disable-next-line no-await-in-loop
    const result = await reader.read();
    done = result.done;

    if (done) {
      break;
    }

    buffer += decoder.decode(result.value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() || "";

    for (const line of lines) {
      if (typeof line !== "string" || !line?.trim()) {
        Logger.error("[KnowledgeBase] Received non-string or empty line:", line);
        // eslint-disable-next-line no-continue
        continue;
      }

      try {
        const parsed = JSON.parse(line);

        switch (parsed.type) {
          case "session":
            if (currentSessionId === null && parsed.sessionId) {
              Logger.info("[KnowledgeBase] Received session ID:", parsed.sessionId);
              currentSessionId = parsed.sessionId;
            }
            break;

          case "response":
            if (parsed.output) {
              // eslint-disable-next-line no-await-in-loop
              await emitWithTypewriter(parsed.output, onChunk, typewriterDelay, signal);
            }
            break;

          case "citations":
            if (!Array.isArray(parsed.citations)) {
              break;
            }

            for (const citation of parsed.citations) {
              const link = citation.documentLink ?? "";
              if (link && seenCitationLinks.has(link)) {
                // eslint-disable-next-line no-continue
                continue;
              }
              if (link) {
                seenCitationLinks.add(link);
              }

              citations.push(citation);
              onCitation?.(citation);
            }
            break;

          case "pulse":
            if (parsed.description) {
              Logger.info("[KnowledgeBase] Pulse:", parsed.description);
              onPulse?.(parsed.description);
            }
            break;

          case "error":
            throw new Error(parsed.message || "An error occurred while processing your request");

          default:
            Logger.info("[KnowledgeBase] Unknown event type:", parsed?.type, parsed);
            break;
        }

        if (parsed.error) {
          Logger.error("[KnowledgeBase] Stream error:", parsed.error);
        }
      } catch (e) {
        if (e instanceof SyntaxError) {
          Logger.error("[KnowledgeBase] Failed to parse line:", line, e);
        } else {
          throw e;
        }
      }
    }
  }

  return { sessionId: currentSessionId, citations };
}

/**
 * Sends a question to the knowledge base API and streams the response.
 *
 * @param {AskQuestionArgs} args - The question arguments
 * @param {string} args.question - The question text to send
 * @param {string | null} [args.sessionId] - Optional session ID to continue a conversation
 * @param {(chunk: string) => void} [args.onChunk] - Optional callback invoked with each text chunk as it arrives
 * @param {(citation: ChatCitation) => void} [args.onCitation] - Optional callback invoked with each citation as it arrives
 * @param {AbortSignal} [args.signal] - Optional abort signal to cancel the request
 * @param {string} args.url - The knowledge base API endpoint URL
 * @param {number} [args.typewriterDelay=10] - Delay in milliseconds between characters (0 to disable typewriter effect)
 * @returns {Promise<AskQuestionResult>} The session ID and citations from the response
 * @throws {Error} If the URL is not provided or the HTTP request fails
 */
export async function askQuestion({
  question,
  sessionId = null,
  conversationHistory = [],
  onChunk,
  onCitation,
  onPulse,
  signal,
  url,
  typewriterDelay = 10,
}: AskQuestionArgs): Promise<AskQuestionResult> {
  if (!url) {
    throw new Error("Knowledge base URL is required but was not provided");
  }

  try {
    const truncatedQuestion = question.slice(0, chatConfig.maxInputTextLength);
    const truncatedHistory = conversationHistory.slice(-chatConfig.maxConversationHistoryLength);
    const askQuestionURL = `${url}/question`;

    const response = await fetch(askQuestionURL, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        question: truncatedQuestion,
        sessionId,
        conversationHistory: truncatedHistory,
      }),
      signal,
    });

    if (!response.ok) {
      throw new Error(`HTTP error! status: ${response.status}`);
    }

    const reader = response.body.getReader();
    const { sessionId: currentSessionId, citations } = await processStreamingResponse(
      reader,
      onChunk,
      onCitation,
      typewriterDelay,
      signal,
      onPulse
    );

    if (currentSessionId) {
      storeSessionId(currentSessionId);
    }

    if (citations.length > 0) {
      Logger.info("[KnowledgeBase] Citations:", citations);
    }

    return { sessionId: currentSessionId, citations };
  } catch (error) {
    Logger.error("[KnowledgeBase]", error);
    throw error;
  }
}

1	import { Logger } from "@/utils";	1✔
2
3	import chatConfig from "../config/chatConfig";	1✔
4	import { storeSessionId } from "../utils/sessionStorageUtils";	1✔
5
6	export type AskKnowledgeBaseResponse = {
7	question: string;
8	answer: string;
9	citations?: ChatCitation[];
10	sessionId?: string;
11	};
12
13	type AskQuestionArgs = {
14	question: string;
15	sessionId?: string \| null;
16	conversationHistory?: ConversationHistory[];
17	onChunk?: (chunk: string) => void;
18	onCitation?: (citation: ChatCitation) => void;
19	onPulse?: (description: string) => void;
20	signal?: AbortSignal;
21	url: string;
22	typewriterDelay?: number;
23	};
24
25	type AskQuestionResult = {
26	sessionId: string \| null;
27	citations: ChatCitation[];
28	};
29
30	/**
31	* Emits text with a typewriter effect (character by character) using a configurable delay.
32	*
33	* @param {string} text - The text to emit character by character
34	* @param {(chunk: string) => void} [onChunk] - Optional callback to invoke with each character
35	* @param {number} [typewriterDelay=10] - Delay in milliseconds between characters (0 to disable typewriter effect)
36	* @param {AbortSignal} [signal] - Optional abort signal to stop the typewriter effect
37	* @returns {Promise<void>}
38	*/
39	export async function emitWithTypewriter(	47✔
40	text: string,	47✔
41	onChunk?: (chunk: string) => void,	47✔
42	typewriterDelay = 10,	47✔
43	signal?: AbortSignal	47✔
44	): Promise<void> {	47✔
45	if (!onChunk \|\| !text) {	47✔
46	return;	5✔
47	}	5✔
48
49	// If typewriter delay is 0, emit the whole text at once
50	if (typewriterDelay === 0) {	47✔
51	onChunk(text);	27✔
52	return;	27✔
53	}	27✔
54
55	for (let i = 0; i < text.length; i += 1) {	47✔
56	if (signal?.aborted) {	64✔
57	return;	4✔
58	}	4✔
59	onChunk(text[i]);	60✔
60	// eslint-disable-next-line no-await-in-loop
61	await new Promise<void>((resolve) => {	60✔
62	setTimeout(resolve, typewriterDelay);	60✔
63	});	60✔
64	}	60✔
65	}	60✔
66
67	/**
68	* Processes the streaming response from the knowledge base API with typewriter effect.
69	*
70	* @param {ReadableStreamDefaultReader<Uint8Array>} reader - The stream reader for the response body
71	* @param {(chunk: string) => void} [onChunk] - Optional callback to invoke with each text chunk
72	* @param {(citation: ChatCitation) => void} [onCitation] - Optional callback to invoke with each citation
73	* @param {number} [typewriterDelay=10] - Delay in milliseconds between characters (0 to disable typewriter effect)
74	* @param {AbortSignal} [signal] - Optional abort signal to stop the typewriter effect
75	* @param {(description: string) => void} [onPulse] - Optional callback to invoke with pulse status descriptions
76	* @returns {Promise<{ sessionId: string \| null; citations: ChatCitation[] }>} The session ID and collected citations from the response
77	*/
78	export async function processStreamingResponse(	34✔
79	reader: ReadableStreamDefaultReader<Uint8Array>,	34✔
80	onChunk?: (chunk: string) => void,	34✔
81	onCitation?: (citation: ChatCitation) => void,	34✔
82	typewriterDelay = 10,	34✔
83	signal?: AbortSignal,	34✔
84	onPulse?: (description: string) => void	34✔
85	): Promise<{ sessionId: string \| null; citations: ChatCitation[] }> {	34✔
86	const decoder = new TextDecoder();	34✔
87	let buffer = "";	34✔
88	let currentSessionId: string \| null = null;	34✔
89	const citations: ChatCitation[] = [];	34✔
90	const seenCitationLinks = new Set<string>();	34✔
91	let done = false;	34✔
92
93	while (!done) {	34✔
94	// eslint-disable-next-line no-await-in-loop
95	const result = await reader.read();	100✔
96	done = result.done;	99✔
97
98	if (done) {	100✔
99	break;	31✔
100	}	31✔
101
102	buffer += decoder.decode(result.value, { stream: true });	68✔
103	const lines = buffer.split("\n");	68✔
104	buffer = lines.pop() \|\| "";	100✔
105
106	for (const line of lines) {	100✔
107	if (typeof line !== "string" \|\| !line?.trim()) {	74✔
108	Logger.error("[KnowledgeBase] Received non-string or empty line:", line);	7✔
109	// eslint-disable-next-line no-continue
110	continue;	7✔
111	}	7✔
112
113	try {	67✔
114	const parsed = JSON.parse(line);	67✔
115
116	switch (parsed.type) {	67✔
117	case "session":	74✔
118	if (currentSessionId === null && parsed.sessionId) {	17✔
119	Logger.info("[KnowledgeBase] Received session ID:", parsed.sessionId);	16✔
120	currentSessionId = parsed.sessionId;	16✔
121	}	16✔
122	break;	17✔
123
124	case "response":	74✔
125	if (parsed.output) {	37✔
126	// eslint-disable-next-line no-await-in-loop
127	await emitWithTypewriter(parsed.output, onChunk, typewriterDelay, signal);	36✔
128	}	36✔
129	break;	37✔
130
131	case "citations":	74✔
132	if (!Array.isArray(parsed.citations)) {	8!
NEW 133	break;	×
NEW 134	}	×
135
136	for (const citation of parsed.citations) {	8✔
137	const link = citation.documentLink ?? "";	9!
138	if (link && seenCitationLinks.has(link)) {	9✔
139	// eslint-disable-next-line no-continue
140	continue;	2✔
141	}	2✔
142	if (link) {	7✔
143	seenCitationLinks.add(link);	7✔
144	}	7✔
145
146	citations.push(citation);	7✔
147	onCitation?.(citation);	7✔
148	}	9✔
149	break;	8✔
150
151	case "pulse":	74✔
152	if (parsed.description) {	1✔
153	Logger.info("[KnowledgeBase] Pulse:", parsed.description);	1✔
154	onPulse?.(parsed.description);	1✔
155	}	1✔
156	break;	1✔
157
158	case "error":	74✔
159	throw new Error(parsed.message \|\| "An error occurred while processing your request");	2✔
160
161	default:	74✔
162	Logger.info("[KnowledgeBase] Unknown event type:", parsed?.type, parsed);	1✔
163	break;	1✔
164	}	74✔
165
166	if (parsed.error) {	74✔
167	Logger.error("[KnowledgeBase] Stream error:", parsed.error);	1✔
168	}	1✔
169	} catch (e) {	74✔
170	if (e instanceof SyntaxError) {	3✔
171	Logger.error("[KnowledgeBase] Failed to parse line:", line, e);	1✔
172	} else {	3✔
173	throw e;	2✔
174	}	2✔
175	}	3✔
176	}	74✔
177	}	66✔
178
179	return { sessionId: currentSessionId, citations };	31✔
180	}	31✔
181
182	/**
183	* Sends a question to the knowledge base API and streams the response.
184	*
185	* @param {AskQuestionArgs} args - The question arguments
186	* @param {string} args.question - The question text to send
187	* @param {string \| null} [args.sessionId] - Optional session ID to continue a conversation
188	* @param {(chunk: string) => void} [args.onChunk] - Optional callback invoked with each text chunk as it arrives
189	* @param {(citation: ChatCitation) => void} [args.onCitation] - Optional callback invoked with each citation as it arrives
190	* @param {AbortSignal} [args.signal] - Optional abort signal to cancel the request
191	* @param {string} args.url - The knowledge base API endpoint URL
192	* @param {number} [args.typewriterDelay=10] - Delay in milliseconds between characters (0 to disable typewriter effect)
193	* @returns {Promise<AskQuestionResult>} The session ID and citations from the response
194	* @throws {Error} If the URL is not provided or the HTTP request fails
195	*/
196	export async function askQuestion({	18✔
197	question,	18✔
198	sessionId = null,	18✔
199	conversationHistory = [],	18✔
200	onChunk,	18✔
201	onCitation,	18✔
202	onPulse,	18✔
203	signal,	18✔
204	url,	18✔
205	typewriterDelay = 10,	18✔
206	}: AskQuestionArgs): Promise<AskQuestionResult> {	18✔
207	if (!url) {	18✔
208	throw new Error("Knowledge base URL is required but was not provided");	1✔
209	}	1✔
210
211	try {	17✔
212	const truncatedQuestion = question.slice(0, chatConfig.maxInputTextLength);	17✔
213	const truncatedHistory = conversationHistory.slice(-chatConfig.maxConversationHistoryLength);	17✔
214	const askQuestionURL = `${url}/question`;	17✔
215
216	const response = await fetch(askQuestionURL, {	17✔
217	method: "POST",	17✔
218	headers: { "Content-Type": "application/json" },	17✔
219	body: JSON.stringify({	17✔
220	question: truncatedQuestion,	17✔
221	sessionId,	17✔
222	conversationHistory: truncatedHistory,	17✔
223	}),	17✔
224	signal,	17✔
225	});	17✔
226
227	if (!response.ok) {	18✔
228	throw new Error(`HTTP error! status: ${response.status}`);	1✔
229	}	1✔
230
231	const reader = response.body.getReader();	14✔
232	const { sessionId: currentSessionId, citations } = await processStreamingResponse(	14✔
233	reader,	14✔
234	onChunk,	14✔
235	onCitation,	14✔
236	typewriterDelay,	14✔
237	signal,	14✔
238	onPulse	14✔
239	);	14✔
240
241	if (currentSessionId) {	18✔
242	storeSessionId(currentSessionId);	12✔
243	}	12✔
244
245	if (citations.length > 0) {	18✔
246	Logger.info("[KnowledgeBase] Citations:", citations);	2✔
247	}	2✔
248
249	return { sessionId: currentSessionId, citations };	13✔
250	} catch (error) {	18✔
251	Logger.error("[KnowledgeBase]", error);	4✔
252	throw error;	4✔
253	}	4✔
254	}	18✔

CBIIT / crdc-datahub-ui / 23444678171

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous