import {isNil} from 'lodash'; import type {FetchData} from '../../../../components/PaginatedTable'; import {TOPIC_MESSAGE_SIZE_LIMIT} from '../../../../store/reducers/topic/topic'; import type { TopicDataRequest, TopicDataResponse, TopicMessageEnhanced, } from '../../../../types/api/topic'; import createToast from '../../../../utils/createToast'; import {safeParseNumber} from '../../../../utils/utils'; import {TOPIC_DATA_FETCH_LIMIT} from './utils/constants'; import type {TopicDataFilters} from './utils/types'; const emptyData = {data: [], total: 0, found: 0}; interface GetTopicDataProps { setBoundOffsets: (props: {startOffset: number; endOffset: number}) => void; baseOffset?: number; /** Maximum number of entities to report (page size). When set, caps total/found. */ maxEntities?: number; } export function prepareResponse(response: TopicDataResponse, offset: number) { const {StartOffset, EndOffset, Messages = []} = response; const start = safeParseNumber(StartOffset); const end = safeParseNumber(EndOffset); const normalizedMessages: TopicMessageEnhanced[] = []; // A topic has an associated schema when the response carries schema context // (`SchemaPath`). In that case messages are returned already schematized as // JSON values (objects/arrays/primitives, including strings) and must not be // base64-decoded. Without a schema the new handler returns exactly the // legacy shape (base64), so a raw string value alone cannot disambiguate the // two cases — we rely on this response-level signal instead. const responseSchematized = Boolean(response.SchemaPath) && !response.SchematizeError; const limit = Math.min(TOPIC_DATA_FETCH_LIMIT, Math.max(end - offset, 0)); let i = 0; let j = 0; while (j < limit) { const currentOffset = offset + j; const currentMessage = Messages[i]; if (currentOffset < start) { normalizedMessages.push({ Offset: currentOffset, removed: true, }); } else if ( !isNil(currentMessage?.Offset) && String(currentMessage.Offset) === String(currentOffset) ) { // A per-message schematization error means this particular message // stayed in its legacy base64 form even when a schema is present. // Only attach the flag when the message is actually schematized so // legacy messages keep their original shape. if (responseSchematized && !currentMessage.SchematizeError) { normalizedMessages.push({...currentMessage, isSchematized: true}); } else { normalizedMessages.push(currentMessage); } i++; } else { normalizedMessages.push({ Offset: currentOffset, notLoaded: true, }); } j++; } return {start, end, messages: normalizedMessages}; } export const generateTopicDataGetter = ({ setBoundOffsets, baseOffset = 0, maxEntities, }: GetTopicDataProps) => { const getTopicData: FetchData = async ({ limit, offset: tableOffset, filters, }) => { if (!filters) { return emptyData; } const { partition, isEmpty, currentPage: _currentPage, clusterName, useMeta, ...rest } = filters; if (isNil(partition) || partition === '' || isEmpty) { return emptyData; } const normalizedOffset = baseOffset + tableOffset; const queryParams: TopicDataRequest = { ...rest, partition, limit, last_offset: normalizedOffset + limit, message_size_limit: TOPIC_MESSAGE_SIZE_LIMIT, }; queryParams.offset = normalizedOffset; const response = useMeta && window.api.meta ? await window.api.meta.getSchemaTopicData({...queryParams, clusterName}) : await window.api.viewer.getTopicData(queryParams); if (response.SchematizeError) { createToast({ name: 'topicDataSchematizeError', theme: 'danger', title: response.SchematizeError, }); } const {start, end, messages} = prepareResponse(response, normalizedOffset); //need to update start and end offsets every time data is fetched to show fresh data in parent component setBoundOffsets({startOffset: start, endOffset: end}); let quantity = Math.max(end - baseOffset, 0); // Cap quantity to maxEntities (page size) when pagination is active if (!isNil(maxEntities) && quantity > maxEntities) { quantity = maxEntities; } return { data: messages, total: quantity, found: quantity, }; }; return getTopicData; };