Redka Compact Init ver 3.0

#include <iostream>
#include <fstream>
#include <string>
#include <vector>
#include <map>
#include <filesystem>
#include <chrono>
#include <thread>
#include <algorithm>
#include <cmath>

namespace fs = std::filesystem;

// Конфигурация
const size_t WAL_SIZE_THRESHOLD = 1024; // 1KB
const std::string WAL_FILENAME = "wal.log";

const size_t LEVEL_BASE_SIZE = 10;      // L0: 10 записей, L1: 100, L2: 1000...
const std::string DB_DIR = "lsm_db";

#pragma pack(push, 1)
struct SSTHeader {
    uint32_t version = 1;
    uint32_t entry_count;
    uint64_t index_offset;
};

struct SSTIndexEntry {
    uint32_t key_length;
    uint64_t data_offset;
    uint32_t data_length;
};
#pragma pack(pop)

struct WalEntry {
    std::string key;
    std::string value;
};

struct SSTEntry {
    std::string key;
    std::string value;

    bool operator<(const SSTEntry& other) const {
        return key < other.key;
    }
};

class LSMTree {
private:
    std::vector<std::vector<std::string>> levels; // Пути к SST-файлам по уровням

    void ensureDbDir() {
        if (!fs::exists(DB_DIR)) {
            fs::create_directory(DB_DIR);
            for (int i = 0; i < 10; ++i) {
                fs::create_directory(DB_DIR + "/L" + std::to_string(i));
            }
        }
    }

    void loadLevels() {
        levels.clear();
        for (int i = 0; ; ++i) {
            std::string level_dir = DB_DIR + "/L" + std::to_string(i);
            if (!fs::exists(level_dir)) break;

            std::vector<std::string> files;
            for (const auto& entry : fs::directory_iterator(level_dir)) {
                if (entry.path().extension() == ".sst") {
                    files.push_back(entry.path().string());
                }
            }
            std::sort(files.begin(), files.end());
            std::reverse(files.begin(), files.end());
            levels.push_back(files);
        }
    }

    void compactLevel(int level) {
        if (level >= levels.size()) return;

        std::vector<SSTEntry> merged_entries;

        // 1. Собираем все записи с текущего уровня
        for (const auto& sst_path : levels[level]) {
            auto entries = readSST(sst_path);
            merged_entries.insert(merged_entries.end(), entries.rbegin(), entries.rend());  // Разворачиваем
        }

        // for (const auto& sst_path : levels[level]) {
        //     auto entries = readSST(sst_path);
        //     merged_entries.insert(merged_entries.end(), entries.begin(), entries.end());
        // }

        // 2. Если записей достаточно для следующего уровня
        if (merged_entries.size() >= LEVEL_BASE_SIZE * std::pow(10, level)) {
            // Сортируем по ключу
            std::sort(merged_entries.begin(), merged_entries.end());

            // Вывод до удаления дубликатов
            // std::cout << "Before unique:\n";
            // for (const auto& entry : merged_entries) {
            //     std::cout << entry.key << " -> " << entry.value << std::endl;
            // }

            // Удаляем дубликаты, оставляя последнюю версию
            auto last = std::unique(merged_entries.begin(), merged_entries.end(),
                   [](const SSTEntry& a, const SSTEntry& b) { return a.key == b.key; });

            merged_entries.erase(last, merged_entries.end());

            // auto last = std::unique(merged_entries.rbegin(), merged_entries.rend(),
            //     [](const auto& a, const auto& b) { return a.key == b.key; });

            // merged_entries.erase(merged_entries.begin(), last.base());

            // Вывод после удаления дубликатов
            // std::cout << "After unique:\n";
            // for (const auto& entry : merged_entries) {
            //     std::cout << entry.key << " -> " << entry.value << std::endl;
            // }

            // 3. Записываем на следующий уровень
            std::string new_sst = DB_DIR + "/L" + std::to_string(level + 1) + "/" +
                                  std::to_string(std::chrono::system_clock::now().time_since_epoch().count()) + ".sst";
            writeSST(new_sst, merged_entries);

            // 4. Удаляем старые файлы
            for (const auto& sst_path : levels[level]) {
                fs::remove(sst_path);
            }

            // 5. Обновляем индексы
            loadLevels();

            // 6. Рекурсивно уплотняем следующий уровень
            compactLevel(level + 1);
        }
    }


    std::vector<SSTEntry> readSST(const std::string& path) {
        std::ifstream file(path, std::ios::binary);
        SSTHeader header;
        file.read(reinterpret_cast<char*>(&header), sizeof(header));

        std::vector<SSTIndexEntry> index(header.entry_count);
        file.seekg(header.index_offset);
        file.read(reinterpret_cast<char*>(index.data()), header.entry_count * sizeof(SSTIndexEntry));

        std::vector<SSTEntry> entries;
        for (const auto& idx : index) {
            file.seekg(idx.data_offset);
            uint32_t total_len;
            file.read(reinterpret_cast<char*>(&total_len), sizeof(total_len));

            std::string key(idx.key_length, '\0');
            file.read(&key[0], idx.key_length);

            std::string value(total_len - idx.key_length, '\0');
            file.read(&value[0], value.size());

            entries.push_back({key, value});
        }

        std::sort(entries.begin(), entries.end());
        return entries;
    }

    void writeSST(const std::string& path, const std::vector<SSTEntry>& entries) {
        std::ofstream file(path, std::ios::binary);
        SSTHeader header;
        header.entry_count = entries.size();
        file.write(reinterpret_cast<char*>(&header), sizeof(header));

        std::vector<SSTIndexEntry> index;
        for (const auto& entry : entries) {
            SSTIndexEntry idx;
            idx.key_length = entry.key.size();
            idx.data_offset = file.tellp();

            uint32_t total_len = sizeof(uint32_t) + entry.key.size() + entry.value.size();
            file.write(reinterpret_cast<char*>(&total_len), sizeof(total_len));
            file.write(entry.key.data(), entry.key.size());
            file.write(entry.value.data(), entry.value.size());

            idx.data_length = total_len;
            index.push_back(idx);
        }

        header.index_offset = file.tellp();
        for (const auto& idx : index) {
            file.write(reinterpret_cast<const char*>(&idx), sizeof(idx));
        }

        file.seekp(0);
        file.write(reinterpret_cast<char*>(&header), sizeof(header));
    }

    std::vector<WalEntry> readWalFile(const std::string& filename) {
        std::vector<WalEntry> entries;
        std::ifstream wal(filename);
        if (!wal) return entries;

        std::string line;
        while (std::getline(wal, line)) {
            // Пропускаем пустые строки
            if (line.empty()) continue;

            // Парсим строку вида {@id {data}}
            size_t id_start = line.find('{');
            size_t id_end = line.find(' ', id_start);
            size_t data_start = line.find('{', id_end);
            size_t data_end = line.rfind('}');

            if (id_start == std::string::npos ||
                id_end == std::string::npos ||
                data_start == std::string::npos ||
                data_end == std::string::npos) {
                std::cerr << "Invalid WAL entry: " << line << std::endl;
                continue;
            }

            // Извлекаем ID (формат {@b0b-1})
            std::string id = line.substr(id_start + 2, id_end - id_start - 2);

            // Извлекаем данные (всё содержимое между {})
            std::string data = line.substr(data_start, data_end - data_start + 1);

            entries.push_back({id, data});
        }

        return entries;
    }

public:
    LSMTree() {
        ensureDbDir();
        loadLevels();
    }

    void put(const std::string& key, const std::string& value) {
        // 1. Добавляем в WAL в правильном формате
        std::ofstream wal(WAL_FILENAME, std::ios::app);
        wal << "{@" << key << " " << value << "}\n";  // Формат: {@key value}
        wal.close();

        // 2. Проверяем размер WAL
        if (fs::file_size(WAL_FILENAME) >= WAL_SIZE_THRESHOLD) {
            flushWalToL0();
        }
    }

    void flushWalToL0() {
        // 1. Читаем WAL с новым парсером
        auto wal_entries = readWalFile(WAL_FILENAME);
        std::map<std::string, std::string> latest_data;

        // Оставляем только последние версии
        for (const auto& e : wal_entries) {
            latest_data[e.key] = e.value;
        }

        // 2. Записываем в SST
        std::vector<SSTEntry> entries;
        for (const auto& [key, value] : latest_data) {
            entries.push_back({key, value});
        }

        // 3. Создаем новый SST-файл
        std::string sst_path = DB_DIR + "/L0/" +
                              std::to_string(std::chrono::system_clock::now().time_since_epoch().count()) + ".sst";
        writeSST(sst_path, entries);

        // 4. Очищаем WAL
        std::ofstream wal(WAL_FILENAME, std::ios::trunc);

        // 5. Обновляем индексы и запускаем компактизацию
        loadLevels();
        compactLevel(0);
    }

    std::string get(const std::string& key) {
        // 1. Проверяем WAL (новые данные)
        auto wal_entries = readWalFile(WAL_FILENAME);
        std::string last_value;
        for (const auto& e : wal_entries) {
            if (e.key == key) last_value = e.value;
        }
        if (!last_value.empty()) return last_value;

        // 2. Ищем в SST-файлах (от новых к старым)
        for (const auto& level : levels) {
            for (const auto& sst_path : level) {
                auto entries = readSST(sst_path);
                // std::cout << sst_path << std::endl;

                // Бинарный поиск через lower_bound
                auto it = std::lower_bound(
                    entries.begin(), entries.end(), key,
                    [](const SSTEntry& e, const std::string& k) { return e.key < k; }
                );

                if (it != entries.end() && it->key == key) {
                    return it->value;
                }
            }
        }

        return ""; // Не найдено
    }
};

void printSSTFile(const std::string& path) {
    std::ifstream file(path, std::ios::binary);
    if (!file) {
        std::cerr << "Ошибка: не удалось открыть SST-файл " << path << std::endl;
        return;
    }

    SSTHeader header;
    file.read(reinterpret_cast<char*>(&header), sizeof(header));

    std::cout << "SST File: " << path << std::endl;
    std::cout << "Version: " << header.version << std::endl;
    std::cout << "Entries: " << header.entry_count << std::endl;
    std::cout << "Index offset: " << header.index_offset << std::endl;
    std::cout << "------------------------\n";

    std::vector<SSTIndexEntry> index(header.entry_count);
    file.seekg(header.index_offset);
    file.read(reinterpret_cast<char*>(index.data()), header.entry_count * sizeof(SSTIndexEntry));

    for (const auto& idx : index) {
        file.seekg(idx.data_offset);
        uint32_t total_len;
        file.read(reinterpret_cast<char*>(&total_len), sizeof(total_len));

        std::string key(idx.key_length, '\0');
        file.read(&key[0], idx.key_length);

        std::string value(total_len - idx.key_length, '\0');
        file.read(&value[0], value.size());

        std::cout << "Key: " << key << " -> Value: " << value << std::endl;
    }

    std::cout << "========================\n";
}


int main() {
    LSMTree db;

    // Теперь данные записываются в правильном формате
    db.put("b0b-1", "{name:\"Alena\" age:25}");
    db.put("b0b-2", "{name:\"Ilia\" age:30}");
    db.put("b0b-1", "{name:\"Alena\" age:26}"); // Обновление
    db.put("b0b-3", "{name:\"Stepan\"}");
    db.put("b0b-4", "{name:\"Dima\"}");

    // for (size_t i = 0; i < 100; i++) {
    //     db.put("b0b-4", "{name:\"Dima\"}");
    // }

    // // Принудительная запись в SST
    db.flushWalToL0();

    // Проверка
    std::cout << "b0b-1: " << db.get("b0b-1") << std::endl;
    std::cout << "b0b-2: " << db.get("b0b-2") << std::endl;
    std::cout << "b0b-3: " << db.get("b0b-3") << std::endl;
    std::cout << "b0b-4: " << db.get("b0b-4") << std::endl;

    // printSSTFile("lsm_db/L0/1743018183023829000.sst");


    return 0;
}