#ifndef TILE_RPC_SERVER_H
#define TILE_RPC_SERVER_H

#pragma once

#include "tile/base/down_cast.h"
#include "tile/base/handle.h"
#include "tile/base/internal/time_keeper.h"
#include "tile/base/make_unique.h"
#include "tile/base/maybe_owning.h"
#include "tile/base/net/endpoint.h"
#include "tile/base/ref_ptr.h"
#include "tile/base/thread/mutex.h"
#include "tile/base/type_index.h"

#include "gflags/gflags_declare.h"

#include <regex>
#include <unordered_map>
#include <unordered_set>
#include <vector>

DECLARE_int32(tile_rpc_server_max_packet_size);
DECLARE_int32(tile_rpc_server_max_connections);
DECLARE_int32(tile_rpc_server_max_request_queuing_delay);
DECLARE_int32(tile_rpc_server_max_ongoing_calls);

namespace tile {
namespace rpc {
namespace detail {
class ServerConnectionHandler;
}
}// namespace rpc

using ConnectionId = std::uint64_t;

class NativeAcceptor;
class StreamProtocol;
class StreamService;

class HttpFilter;
class HttpHandler;

class Server {
    template<typename T>
    using Factory = std::function<std::unique_ptr<T>>;

public:
    struct Options {
        std::string service_name;
        std::size_t maximum_packet_size        = FLAGS_tile_rpc_server_max_packet_size;
        std::size_t maximum_concurrent_request = FLAGS_tile_rpc_server_max_ongoing_calls;
        std::size_t max_concurrent_connections = FLAGS_tile_rpc_server_max_connections;

        std::chrono::nanoseconds max_request_queueing_delay =
            FLAGS_tile_rpc_server_max_request_queuing_delay * std::chrono::milliseconds(1);

        std::function<bool(const Endpoint &)> conn_filter = [](const Endpoint &) { return true; };
    };

    Server();
    explicit Server(Options opt);
    ~Server();

    void AddProtocol(const std::string &name);
    void AddProtocol(std::initializer_list<std::string>) = delete;
    void AddProtocol(Factory<StreamProtocol> factory);

    void AddProtocols(const std::vector<std::string> &names);

    void ListenOn(const Endpoint &endpoint, int backlog = 128);

    bool Start();
    void Stop();
    void Join();

    void AddHttpFilter(MaybeOwningArgument<HttpFilter> filter);
    void AddHttpHandler(std::regex path, MaybeOwningArgument<HttpHandler> handler);
    void AddHttpPrefixHandler(std::string prefix, MaybeOwningArgument<HttpHandler> handler);
    void SetDefaultHttpHandler(MaybeOwningArgument<HttpHandler> handler);

    template<typename T>
    T *GetBuiltinNativeService();

private:
    struct ConnectionContext;

    enum class ServerState { Initialized, Running, Stopped, Joined };

private:
    void OnConnection(Handle fd, Endpoint peer);

    void OnNewCall();
    void OnCallCompletion();
    void OnConnectionClosed(ConnectionId id);
    void OnConnectionCleanupTimer();

    std::unique_ptr<rpc::detail::ServerConnectionHandler> CreateNormalConnectionHandler(ConnectionId id, Endpoint peer);

    void StartBackgroundJob(std::function<void()> cb);

private:
    Options options_;
    ServerState state_{ServerState::Initialized};

    internal::TimerID idle_connn_cleaner_;

    // Set by `ListenOn`
    Endpoint listening_on_;
    std::function<void()> listen_cb_;

    std::atomic<std::size_t> alive_conns_{0};

    std::vector<std::pair<std::unique_ptr<HttpHandler>, std::vector<std::string>>> builtin_http_handlers;
    std::vector<std::pair<std::unique_ptr<HttpHandler>, std::vector<std::string>>> builtin_http_prefix_handlers;

    // === for protobuf ===
    std::unordered_set<std::string> known_protocols_;

    std::vector<Factory<StreamProtocol>> protocol_factories_;
    RefPtr<NativeAcceptor> acceptor_;

    // `TypeIndex` -> `protobuf::Service*`
    std::unordered_map<TypeIndex, StreamService *> builtin_services_;
    std::deque<MaybeOwning<StreamService>> services_;

    std::atomic<std::size_t> ongoing_calls_{0};

    Mutex conns_lock_;
    std::unordered_map<ConnectionId, std::unique_ptr<ConnectionContext>> conns_;

    // Number of outstanding `StartBackgroundJob`
    std::atomic<std::size_t> outstanding_jobs_{};
};

template<typename T>
T *
Server::GetBuiltinNativeService()
{
    auto key  = GetTypeIndex<T>();
    auto iter = builtin_services_.find(key);
    if (iter == builtin_services_.end()) {
        TILE_CHECK(!acceptor_, "Please ensure `Start()` is called.");
        services_.push_back(make_unique<T>());
        builtin_services_[key] = services_.back().Get();
        iter                   = builtin_services_.find(key);
    }

    return down_cast<T>(iter->second);
}

}// namespace tile

#endif// TILE_RPC_SERVER_H
