getData.ts

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<TopicMessageEnhanced, TopicDataFilters> = 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;
};