Pg.h

//------------------------------------------------------------------------------
/*
    This file is part of rippled: https://github.com/ripple/rippled
    Copyright (c) 2020 Ripple Labs Inc.
 
    Permission to use, copy, modify, and/or distribute this software for any
    purpose  with  or without fee is hereby granted, provided that the above
    copyright notice and this permission notice appear in all copies.
 
    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
 
#ifdef RIPPLED_REPORTING
#ifndef RIPPLE_CORE_PG_H_INCLUDED
#define RIPPLE_CORE_PG_H_INCLUDED
 
#include <ripple/basics/BasicConfig.h>
#include <ripple/basics/Log.h>
#include <ripple/protocol/Protocol.h>
#include <boost/lexical_cast.hpp>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <libpq-fe.h>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <utility>
#include <vector>
 
namespace ripple {
 
// These postgres structs must be freed only by the postgres API.
using pg_result_type = std::unique_ptr<PGresult, void (*)(PGresult*)>;
using pg_connection_type = std::unique_ptr<PGconn, void (*)(PGconn*)>;
 
using pg_params =
    std::pair<char const*, std::vector<std::optional<std::string>>>;
 
using pg_formatted_params = std::vector<char const*>;
 
struct PgConfig
{
    std::size_t max_connections{std::numeric_limits<std::size_t>::max()};
    std::chrono::seconds timeout{600};
 
    std::vector<char const*> keywordsIdx;
    std::vector<std::string> keywords;
    std::vector<char const*> valuesIdx;
    std::vector<std::string> values;
};
 
//-----------------------------------------------------------------------------
 
class PgResult
{
    // The result object must be freed using the libpq API PQclear() call.
    pg_result_type result_{nullptr, [](PGresult* result) { PQclear(result); }};
    std::optional<std::pair<ExecStatusType, std::string>> error_;
 
public:
    PgResult()
    {
    }
 
    explicit PgResult(pg_result_type&& result) : result_(std::move(result))
    {
    }
 
    PgResult(PGresult* result, PGconn* conn)
        : error_({PQresultStatus(result), PQerrorMessage(conn)})
    {
    }
 
    char const*
    c_str(int ntuple = 0, int nfield = 0) const
    {
        return PQgetvalue(result_.get(), ntuple, nfield);
    }
 
    std::int32_t
    asInt(int ntuple = 0, int nfield = 0) const
    {
        return boost::lexical_cast<std::int32_t>(
            PQgetvalue(result_.get(), ntuple, nfield));
    }
 
    std::int64_t
    asBigInt(int ntuple = 0, int nfield = 0) const
    {
        return boost::lexical_cast<std::int64_t>(
            PQgetvalue(result_.get(), ntuple, nfield));
    }
 
    bool
    isNull(int ntuple = 0, int nfield = 0) const
    {
        return PQgetisnull(result_.get(), ntuple, nfield);
    }
 
    operator bool() const
    {
        return result_ != nullptr;
    }
 
    std::string
    msg() const;
 
    int
    ntuples() const
    {
        return PQntuples(result_.get());
    }
 
    int
    nfields() const
    {
        return PQnfields(result_.get());
    }
 
    ExecStatusType
    status() const
    {
        return PQresultStatus(result_.get());
    }
};
 
/* Class that contains and operates upon a postgres connection. */
class Pg
{
    friend class PgPool;
    friend class PgQuery;
 
    PgConfig const& config_;
    beast::Journal const j_;
    bool& stop_;
    std::mutex& mutex_;
 
    // The connection object must be freed using the libpq API PQfinish() call.
    pg_connection_type conn_{nullptr, [](PGconn* conn) { PQfinish(conn); }};
 
    bool
    clear();
 
    void
    connect();
 
    void
    disconnect()
    {
        conn_.reset();
    }
 
    PgResult
    query(char const* command, std::size_t nParams, char const* const* values);
 
    PgResult
    query(char const* command)
    {
        return query(command, 0, nullptr);
    }
 
    PgResult
    query(pg_params const& dbParams);
 
    void
    bulkInsert(char const* table, std::string const& records);
 
public:
    Pg(PgConfig const& config,
       beast::Journal const j,
       bool& stop,
       std::mutex& mutex)
        : config_(config), j_(j), stop_(stop), mutex_(mutex)
    {
    }
};
 
//-----------------------------------------------------------------------------
 
class PgPool
{
    friend class PgQuery;
 
    using clock_type = std::chrono::steady_clock;
 
    PgConfig config_;
    beast::Journal const j_;
    std::mutex mutex_;
    std::condition_variable cond_;
    std::size_t connections_{};
    bool stop_{false};
 
    std::multimap<std::chrono::time_point<clock_type>, std::unique_ptr<Pg>>
        idle_;
 
    std::unique_ptr<Pg>
    checkout();
 
    void
    checkin(std::unique_ptr<Pg>& pg);
 
public:
    PgPool(Section const& pgConfig, beast::Journal j);
 
    void
    setup();
 
    void
    stop();
 
    void
    idleSweeper();
};
 
//-----------------------------------------------------------------------------
 
class PgQuery
{
private:
    std::shared_ptr<PgPool> pool_;
    std::unique_ptr<Pg> pg_;
 
public:
    PgQuery() = delete;
 
    PgQuery(std::shared_ptr<PgPool> const& pool)
        : pool_(pool), pg_(pool->checkout())
    {
    }
 
    ~PgQuery()
    {
        pool_->checkin(pg_);
    }
 
    PgResult
    operator()(pg_params const& dbParams)
    {
        if (!pg_)  // It means we're stopping. Return empty result.
            return PgResult();
        return pg_->query(dbParams);
    }
 
    PgResult
    operator()(char const* command)
    {
        return operator()(pg_params{command, {}});
    }
 
    void
    bulkInsert(char const* table, std::string const& records)
    {
        pg_->bulkInsert(table, records);
    }
};
 
//-----------------------------------------------------------------------------
 
std::shared_ptr<PgPool>
make_PgPool(Section const& pgConfig, beast::Journal j);
 
void
initSchema(std::shared_ptr<PgPool> const& pool);
 
}  // namespace ripple
 
#endif  // RIPPLE_CORE_PG_H_INCLUDED
#endif  // RIPPLED_REPORTING