Redis helper for drogon framework

RedisHelper.cpp

//
// Created by ParticleG on 2022/2/9.
//

#include <helpers/RedisHelper.h>
#include <ranges>
#include <structures/Exceptions.h>
#include <utils/crypto.h>
#include <utils/datetime.h>

using namespace drogon;
using namespace std;
using namespace studio26f::helpers;
using namespace studio26f::structures;
using namespace studio26f::utils;

RedisHelper::RedisHelper(std::string BaseKey) :
        _baseKey(std::move(BaseKey)),
        _redisClient(app().getRedisClient()) {}

RedisHelper::~RedisHelper() {
    _redisClient->closeAll();
    LOG_INFO << "Redis disconnected.";
}

bool RedisHelper::tokenBucket(
        const string &key,
        const chrono::microseconds &restoreInterval,
        const uint64_t &maxCount
) const {
    const auto countKey = _baseKey + ":tokenBucket:count:" + key;
    const auto updatedKey = _baseKey + ":tokenBucket:updated:" + key;
    const auto maxTtl = chrono::duration_cast<chrono::seconds>(restoreInterval * maxCount);

    uint64_t countValue;
    if (exists({countKey}))
        try {
            countValue = stoull(get(countKey));
        } catch (...) {
            set(countKey, to_string(maxCount - 1));
            countValue = maxCount;
        }

    bool hasToken = true;
    try {
        const auto lastUpdated = get(updatedKey);
        const auto nowMicroseconds = datetime::toDate().microSecondsSinceEpoch();
        const auto generatedCount =
                (nowMicroseconds -
                 datetime::toDate(lastUpdated).microSecondsSinceEpoch()
                ) / restoreInterval.count() - 1;

        if (generatedCount >= 1) {
            set(updatedKey, datetime::toString(nowMicroseconds));
            incrBy(countKey, static_cast<int>(generatedCount) - 1);
            hasToken = true;
        } else if (countValue > 0) {
            decrBy(countKey);
            hasToken = true;
        } else {
            hasToken = false;
        }
    } catch (...) {
        set({{updatedKey, datetime::toString()},
             {countKey,   to_string(maxCount - 1)}});
    }

    // Use sync methods to make sure the operation is completed.
    expire({{countKey,   maxTtl},
            {updatedKey, maxTtl}});
    return hasToken;
}

void RedisHelper::del(const vector<string> &keys, const function<void(int64_t)> &callback) const noexcept {
    if (keys.empty()) {
        callback(0);
        return;
    }

    stringstream keyStream;
    ranges::copy(keys | views::transform([this](const auto &key) {
        return _baseKey + ":" + key;
    }), ostream_iterator<string>(keyStream, " "));

    _redisClient->execCommandAsync(
            [&, size = keys.size()](const nosql::RedisResult &result) {
                callback(result.asInteger());
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback(-1);
            },
            "del %s", keyStream.str().c_str()
    );
}

int64_t RedisHelper::del(const vector<string> &keys) const {
    if (keys.empty()) {
        return 0;
    }

    stringstream keyStream;
    ranges::copy(keys | views::transform([this](const auto &key) {
        return _baseKey + ":" + key;
    }), ostream_iterator<string>(keyStream, " "));

    return _redisClient->execCommandSync<int64_t>(
            [size = keys.size()](const nosql::RedisResult &result) {
                return result.asInteger();
            },
            "del %s", keyStream.str().c_str()
    );
}

void RedisHelper::exists(const vector<string> &keys, const function<void(bool)> &callback) const noexcept {
    if (keys.empty()) {
        callback(false);
        return;
    }

    stringstream keyStream;
    ranges::copy(keys | views::transform([this](const auto &key) {
        return _baseKey + ":" + key;
    }), ostream_iterator<string>(keyStream, " "));

    _redisClient->execCommandAsync(
            [&, size = keys.size()](const nosql::RedisResult &result) {
                callback(result.asInteger() == size);
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback(false);
            },
            "exists %s", keyStream.str().c_str()
    );
}

bool RedisHelper::exists(const vector<string> &keys) const {
    if (keys.empty()) {
        return false;
    }

    stringstream keyStream;
    ranges::copy(keys | views::transform([this](const auto &key) {
        return _baseKey + ":" + key;
    }), ostream_iterator<string>(keyStream, " "));

    return _redisClient->execCommandSync<bool>(
            [size = keys.size()](const nosql::RedisResult &result) {
                return result.asInteger() == size;
            },
            "exists %s", keyStream.str().c_str()
    );
}

void RedisHelper::expire(
        const string &key,
        const chrono::seconds &ttl,
        const function<void(bool)> &callback
) const noexcept {
    const auto tempKey = _baseKey + ":" + key;
    _redisClient->execCommandAsync(
            [&](const nosql::RedisResult &result) {
                callback(result.asInteger());
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback(false);
            },
            "expire %s %d", tempKey.c_str(), ttl.count()
    );
}

bool RedisHelper::expire(const string &key, const chrono::seconds &ttl) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<bool>(
            [](const nosql::RedisResult &result) {
                return result.asInteger();
            },
            "expire %s %d", tempKey.c_str(), ttl.count()
    );
}

void RedisHelper::expire(
        const KeyPairs <chrono::seconds> &params,
        const function<void(vector<bool> &&)> &callback
) const noexcept {
    const auto transaction = _redisClient->newTransaction();
    for (const auto &[key, ttl]: params) {
        transaction->execCommandAsync(
                [](const nosql::RedisResult &result) {
                    LOG_TRACE << result.getStringForDisplayingWithIndent();
                },
                [](const exception &err) {
                    LOG_ERROR << err.what();
                },
                "expire %s %d", (_baseKey + ":" + key).c_str(), ttl.count()
        );
    }
    transaction->execute(
            [&](const nosql::RedisResult &result) {
                LOG_TRACE << result.getStringForDisplayingWithIndent();
                const auto &resultsArray = result.asArray();
                const auto view = resultsArray | views::transform(
                        [](const auto &item) -> bool { return item.asInteger(); }
                );
                callback({view.begin(), view.end()});
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback({});
            }
    );
}

vector<bool> RedisHelper::expire(const KeyPairs <chrono::seconds> &params) const {
    const auto transaction = _redisClient->newTransaction();
    for (const auto &[key, ttl]: params) {
        transaction->execCommandAsync(
                [](const nosql::RedisResult &result) {
                    LOG_TRACE << result.getStringForDisplayingWithIndent();
                },
                [](const exception &err) {
                    LOG_ERROR << err.what();
                },
                "expire %s %d", (_baseKey + ":" + key).c_str(), ttl.count()
        );
    }
    promise<vector<bool>> resultsPromise;
    auto resultsFuture = resultsPromise.get_future();
    transaction->execute(
            [&](const nosql::RedisResult &result) {
                LOG_TRACE << result.getStringForDisplayingWithIndent();
                const auto &resultsArray = result.asArray();
                const auto view = resultsArray | views::transform(
                        [](const auto &item) -> bool { return item.asInteger(); }
                );
                resultsPromise.set_value({view.begin(), view.end()});
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                resultsPromise.set_value({});
            }
    );
    return resultsFuture.get();
}

void RedisHelper::get(const string &key, const RedisHelper::SimpleResultCb &callback) const noexcept {
    const auto tempKey = _baseKey + ":" + key;
    _redisClient->execCommandAsync(
            [&](const nosql::RedisResult &result) {
                if (result.isNil()) {
                    callback({false, {}});
                } else {
                    callback({true, result.asString()});
                }
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback({false, err.what()});
            },
            "get %s", tempKey.c_str()
    );
}

string RedisHelper::get(const string &key) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<string>(
            [=](const nosql::RedisResult &result) {
                if (result.isNil()) {
                    throw redis_exception::KeyNotFound(tempKey);
                }
                return result.asString();
            },
            "get %s", tempKey.c_str()
    );
}

int64_t RedisHelper::incrBy(const string &key, const int64_t &value) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<int64_t>(
            [](const nosql::RedisResult &result) {
                return result.asInteger();
            },
            "incrBy %s %lld", tempKey.c_str(), value
    );
}

int64_t RedisHelper::decrBy(const string &key, const int64_t &value) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<int64_t>(
            [](const nosql::RedisResult &result) {
                return result.asInteger();
            },
            "decrBy %s %lld", tempKey.c_str(), value
    );
}

void RedisHelper::sAdd(const string &key, const vector<string> &values) const {
    if (values.empty()) {
        LOG_TRACE << 0;
        return;
    }

    stringstream valueStream;
    ranges::copy(values | views::transform([this](const auto &key) {
        return _baseKey + ":" + key;
    }), ostream_iterator<string>(valueStream, " "));
    _redisClient->execCommandAsync(
            [](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
            },
            [](const std::exception &err) {
                LOG_ERROR << err.what();
            },
            "sAdd %s %s", key.c_str(), valueStream.str().c_str()
    );
}

void RedisHelper::sAdd(const vector<pair<string, vector<string>>> &params) const {
    const auto transaction = _redisClient->newTransaction();
    for (const auto &[key, values]: params) {
        if (values.empty()) {
            LOG_TRACE << 0;
            continue;
        }
        stringstream valueStream;
        ranges::copy(values | views::transform([this](const auto &tempKey) {
            return _baseKey + ":" + tempKey;
        }), ostream_iterator<string>(valueStream, " "));
        transaction->execCommandAsync(
                [](const nosql::RedisResult &result) {
                    LOG_TRACE << result.asInteger();
                },
                [](const std::exception &err) {
                    LOG_ERROR << err.what();
                },
                "sAdd %s %s", key.c_str(), valueStream.str().c_str()
        );
    }
    transaction->execute(
            [](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
            },
            [](const std::exception &err) {
                LOG_ERROR << err.what();
            }
    );
}

int64_t RedisHelper::sCard(const string &key) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<int64_t>(
            [=](const nosql::RedisResult &result) {
                return result.asInteger();
            },
            "get %s", tempKey.c_str()
    );
}

vector<string> RedisHelper::sMembers(const string &key) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<vector<string>>(
            [=](const nosql::RedisResult &result) -> vector<string> {
                if (result.isNil()) {
                    return {};
                }
                const auto array = result.asArray();
                const auto memberView = array | views::transform([](const nosql::RedisResult &result) {
                    return result.asString();
                }) | views::common;
                return {memberView.begin(), memberView.end()};

            },
            "sMembers %s", tempKey.c_str()
    );
}

vector<vector<string>> RedisHelper::sMembers(const vector<string> &keys) const {
    vector<vector<string>> result;
    const auto transaction = _redisClient->newTransaction();
    for (const auto &key: keys) {
        const auto tempKey = _baseKey + ":" + key;
        result.push_back(transaction->execCommandSync<vector<string>>(
                [=](const nosql::RedisResult &result) -> vector<string> {
                    if (result.isNil()) {
                        return {};
                    }
                    const auto array = result.asArray();
                    const auto memberView = array | views::transform([](const nosql::RedisResult &result) {
                        return result.asString();
                    }) | views::common;
                    return {memberView.begin(), memberView.end()};

                },
                "sMembers %s", tempKey.c_str()
        ));
    }
    promise<int64_t> p1;
    auto f1 = p1.get_future();
    transaction->execute(
            [&](const nosql::RedisResult &result) {
                p1.set_value(result.asInteger());
            },
            [&](const std::exception &err) {
                p1.set_value(-1);
                LOG_ERROR << err.what();
            }
    );
    LOG_TRACE << f1.get();
    return result;
}

bool RedisHelper::sIsMember(const string &key, const string &value) const {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<bool>(
            [=](const nosql::RedisResult &result) {
                return result.asInteger();
            },
            "sIsMember %s", tempKey.c_str()
    );
}

void RedisHelper::sRemove(const string &key, const vector<string> &values) const {
    if (values.empty()) {
        LOG_TRACE << 0;
        return;
    }

    stringstream valueStream;
    ranges::copy(values | views::transform([this](const auto &key) {
        return _baseKey + ":" + key;
    }), ostream_iterator<string>(valueStream, " "));

    _redisClient->execCommandAsync(
            [](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
            },
            [](const std::exception &err) {
                LOG_ERROR << err.what();
            },
            "sRem %s %s", key.c_str(), valueStream.str().c_str()
    );
}

void RedisHelper::sRemove(const vector<pair<string, vector<string>>> &params) const {
    const auto transaction = _redisClient->newTransaction();
    for (const auto &[key, values]: params) {
        if (values.empty()) {
            LOG_TRACE << 0;
            continue;
        }

        stringstream valueStream;
        ranges::copy(values | views::transform([this](const auto &tempKey) {
            return _baseKey + ":" + tempKey;
        }), ostream_iterator<string>(valueStream, " "));
        transaction->execCommandAsync(
                [](const nosql::RedisResult &result) {
                    LOG_TRACE << result.asInteger();
                },
                [](const std::exception &err) {
                    LOG_ERROR << err.what();
                },
                "sRem %s %s", key.c_str(), valueStream.str().c_str()
        );
    }
    transaction->execute(
            [](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
            },
            [](const std::exception &err) {
                LOG_ERROR << err.what();
            }
    );
}

void RedisHelper::set(const string &key, const string &value, const SimpleResultCb &callback) const noexcept {
    const auto tempKey = _baseKey + ":" + key;
    _redisClient->execCommandAsync(
            [&](const nosql::RedisResult &result) {
                const auto resultString = result.asString();
                LOG_TRACE << result.getStringForDisplayingWithIndent();
                callback({true, result.asString()});
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback({false, err.what()});
            },
            "set %s %s", tempKey.c_str(), value.c_str()
    );
}

RedisHelper::SimpleResult RedisHelper::set(const string &key, const string &value) const noexcept(false) {
    const auto tempKey = _baseKey + ":" + key;
    return _redisClient->execCommandSync<SimpleResult>(
            [&](const nosql::RedisResult &result) -> SimpleResult {
                const auto resultString = result.asString();
                LOG_TRACE << result.getStringForDisplayingWithIndent();
                return {true, result.asString()};
            },
            "set %s %s", tempKey.c_str(), value.c_str()
    );
}

void RedisHelper::set(const KeyPairs <string> &params, const SimpleResultsCb &callback) const noexcept {
    SimpleResults execResults;
    execResults.reserve(params.size());
    const auto transaction = _redisClient->newTransaction();
    for (int64_t index = 0; index < params.size(); ++index) {
        const auto &[key, value] = params[index];
        const auto tempKey = _baseKey + ":" + key;
        transaction->execCommandAsync(
                [&](const nosql::RedisResult &result) {
                    const auto resultString = result.asString();
                    LOG_TRACE << result.getStringForDisplayingWithIndent();
                    execResults.emplace(execResults.begin() + index, true, result.asString());
                },
                [&](const exception &err) {
                    LOG_ERROR << err.what();
                    execResults.emplace(execResults.begin() + index, false, string{err.what()});
                },
                "set %s %s", tempKey.c_str(), value.c_str()
        );
    }
    transaction->execute(
            [&](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
                callback(std::move(execResults));
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                callback(std::move(execResults));
            }
    );
}

RedisHelper::SimpleResults RedisHelper::set(const KeyPairs <string> &params) const noexcept(false) {
    SimpleResults execResults;
    execResults.reserve(params.size());
    const auto transaction = _redisClient->newTransaction();
    for (int64_t index = 0; index < params.size(); ++index) {
        const auto &[key, value] = params[index];
        const auto tempKey = _baseKey + ":" + key;
        transaction->execCommandAsync(
                [&](const nosql::RedisResult &result) {
                    const auto resultString = result.asString();
                    LOG_TRACE << result.getStringForDisplayingWithIndent();
                    execResults.emplace(execResults.begin() + index, true, result.asString());
                },
                [&](const exception &err) {
                    LOG_ERROR << err.what();
                    execResults.emplace(execResults.begin() + index, false, string{err.what()});
                },
                "set %s %s", tempKey.c_str(), value.c_str()
        );
    }
    promise<int64_t> p1;
    auto f1 = p1.get_future();
    transaction->execute(
            [&](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
                p1.set_value(result.asInteger());
            },
            [&](const exception &err) {
                LOG_ERROR << err.what();
                p1.set_value(-1);
            }
    );
    LOG_TRACE << f1.get();
    return execResults;
}

void RedisHelper::setEx(const string &key, int64_t ttl, const string &value) const {
    const auto tempKey = _baseKey + ":" + key;
    LOG_TRACE << _redisClient->execCommandSync<string>(
                [](const nosql::RedisResult &result) {
                    return result.asString();
                },
                "setEx %s %lld %s", tempKey.c_str(), ttl, value.c_str()
        );
}

void RedisHelper::setEx(const vector<tuple<string, int64_t, string>> &params) const {
    const auto transaction = _redisClient->newTransaction();
    for (const auto &[key, ttl, value]: params) {
        const auto tempKey = _baseKey + ":" + key;
        LOG_TRACE << transaction->execCommandSync<string>(
                    [](const nosql::RedisResult &result) {
                        return result.asString();
                    },
                    "setEx %s %lld %s", tempKey.c_str(), ttl, value.c_str()
            );
    }
    transaction->execute(
            [](const nosql::RedisResult &result) {
                LOG_TRACE << result.asInteger();
            },
            [](const std::exception &err) {
                LOG_ERROR << err.what();
            }
    );
}

RedisHelper.hpp

//
// Created by ParticleG on 2022/2/9.
//

#pragma once

#include <drogon/drogon.h>

namespace studio26f::helpers {
    class RedisHelper {
    public:
        explicit RedisHelper(std::string BaseKey = CMAKE_PROJECT_NAME);

        [[nodiscard]] bool tokenBucket(
                const std::string &key,
                const std::chrono::microseconds &restoreInterval,
                const uint64_t &maxCount
        ) const;

        virtual ~RedisHelper();

    protected:
        template<typename T>
        using KeyPair = std::pair<std::string, T>;

        template<typename T>
        using KeyPairs = std::vector<KeyPair<T>>;

        using SimpleResult = std::pair<bool, std::string>;
        using SimpleResultCb = std::function<void(SimpleResult &&)>;

        using SimpleResults = std::vector<SimpleResult>;
        using SimpleResultsCb = std::function<void(SimpleResults &&)>;

        void del(const std::vector<std::string> &keys, const std::function<void(int64_t)> &callback) const noexcept;

        [[nodiscard]] int64_t del(const std::vector<std::string> &keys) const;

        void exists(const std::vector<std::string> &keys, const std::function<void(bool)> &callback) const noexcept;

        [[nodiscard]] bool exists(const std::vector<std::string> &keys) const;

        void expire(
                const std::string &key,
                const std::chrono::seconds &ttl,
                const std::function<void(bool)> &callback
        ) const noexcept;

        [[nodiscard]] bool expire(const std::string &key, const std::chrono::seconds &ttl) const;

        void expire(
                const KeyPairs<std::chrono::seconds> &params,
                const std::function<void(std::vector<bool> &&)> &callback
        ) const noexcept;

        [[nodiscard]] std::vector<bool> expire(const KeyPairs<std::chrono::seconds> &params) const;

        void get(const std::string &key, const SimpleResultCb &callback) const noexcept;

        [[nodiscard]] std::string get(const std::string &key) const;

        [[nodiscard]] int64_t incrBy(const std::string &key, const int64_t &value = 1) const;

        [[nodiscard]] int64_t decrBy(const std::string &key, const int64_t &value = 1) const;

        void sAdd(const std::string &key, const std::vector<std::string> &values) const;

        void sAdd(const std::vector<std::pair<std::string, std::vector<std::string>>> &tempKey) const;

        [[nodiscard]] int64_t sCard(const std::string &key) const;

        [[nodiscard]] std::vector<std::string> sMembers(const std::string &key) const;

        [[nodiscard]] std::vector<std::vector<std::string>> sMembers(const std::vector<std::string> &keys) const;

        [[nodiscard]] bool sIsMember(const std::string &key, const std::string &value) const;

        void sRemove(const std::string &key, const std::vector<std::string> &values) const;

        void sRemove(const std::vector<std::pair<std::string, std::vector<std::string>>> &params) const;

        void set(const std::string &key, const std::string &value, const SimpleResultCb &callback) const noexcept;

        [[nodiscard]] SimpleResult set(const std::string &key, const std::string &value) const;

        void set(const KeyPairs<std::string> &params, const SimpleResultsCb &callback) const noexcept;

        [[nodiscard]] SimpleResults set(const KeyPairs<std::string> &params) const;

        void setEx(const std::string &key, int64_t ttl, const std::string &value) const;

        void setEx(const std::vector<std::tuple<std::string, int64_t, std::string>> &params) const;

    private:
        std::string _baseKey;
        drogon::nosql::RedisClientPtr _redisClient;
    };
}