/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
#ifndef FLINK_TNEL_WINDOWLISTSTATE_H
#define FLINK_TNEL_WINDOWLISTSTATE_H

#include <array>
#include <tuple>
#include "runtime/state/heap/HeapListState.h"
#include "WindowState.h"
/**
 * KeyType: The type of the values that can be added to the list state. such as RowData*
 * W: The data type that the serializer serializes.
 * ValType: The type of the values that can be added to the list state. such as RowData*
 * */
template <typename KeyType, typename W, typename ValType>
class WindowListState : public WindowState<W> {
public:
    WindowListState(InternalListState<KeyType, W, ValType>* windowState) : windowState(windowState) {};
    ~WindowListState() override;
    void clear(W window);
    std::vector<ValType>* get(W window);
    void add(W window, ValType value);
    void addVectorBatch(omnistream::VectorBatch* batch);
    const std::vector<omnistream::VectorBatch*>& getVectorBatches();
    int getCurrentBatchId();
    void clearVectors(int64_t currentTimestamp)
    {
        windowState->clearVectors(currentTimestamp);
    };
    void clearVectors(std::vector<size_t>& indicesToDelete)
    {
        windowState->clearVectors(indicesToDelete);
    };
    omnistream::VectorBatch* getVectorBatch(int batchId);

private:
    InternalListState<KeyType, W, ValType>* windowState;
};

template <typename KeyType, typename W, typename ValType>
void WindowListState<KeyType, W, ValType>::clear(W window)
{
    windowState->setCurrentNamespace(window);
    windowState->clear();
}

template <typename KeyType, typename W, typename ValType>
WindowListState<KeyType, W, ValType>::~WindowListState()
{
    delete windowState;
}

template <typename KeyType, typename W, typename ValType>
std::vector<ValType>* WindowListState<KeyType, W, ValType>::get(W window)
{
    windowState->setCurrentNamespace(window);
    return windowState->get();
}

template <typename KeyType, typename W, typename ValType>
void WindowListState<KeyType, W, ValType>::add(W window, ValType value)
{
    windowState->setCurrentNamespace(window);
    windowState->add(value);
}

template <typename KeyType, typename W, typename ValType>
void WindowListState<KeyType, W, ValType>::addVectorBatch(omnistream::VectorBatch* batch)
{
    windowState->addVectorBatch(batch);
}

template <typename KeyType, typename W, typename ValType>
const std::vector<omnistream::VectorBatch*>& WindowListState<KeyType, W, ValType>::getVectorBatches()
{
    return windowState->getVectorBatches();
}

template <typename KeyType, typename W, typename ValType>
omnistream::VectorBatch* WindowListState<KeyType, W, ValType>::getVectorBatch(int batchId)
{
    return windowState->getVectorBatch(batchId);
}

template <typename KeyType, typename W, typename ValType>
int WindowListState<KeyType, W, ValType>::getCurrentBatchId()
{
    return windowState->getVectorBatchesSize();
}

#endif // FLINK_TNEL_WINDOWLISTSTATE_H