path: root/datastore-leveldb/src/main.cpp

extern "C" {
#include "server_eh.h"
#include <unistd.h>
#include <signal.h>
#include <ev.h>
#include <string.h>
#include <netinet/in.h>
#include <sys/sendfile.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <magic.h>
}

#include <algorithm>
#include <iostream>
#include <iomanip>
#include <string>
#include <sstream>
#include <map>
#include <forward_list>
#include <functional>

#include <boost/regex.hpp>

#include "leveldb/db.h"
#include "leveldb/comparator.h"

std::forward_list<std::pair<boost::regex,std::function<void(const boost::cmatch&, struct http_request*, const int)>>> web_handler;

static std::map<std::string,leveldb::DB*> dbs;

static magic_t magic_cookie;


bool sensor_name_is_sane(std::string& name) {
    for (auto it = name.begin(); it != name.end(); ++it) {
        if (not (*it >= '0' and *it <= '9' or
                 *it >= 'A' and *it <= 'Z' or
                 *it >= 'a' and *it <= 'z')) {
            return false;
        }
    }
    return true;
}


leveldb::DB *getDB(std::string& name) {
    if (not sensor_name_is_sane(name)) {
        return nullptr;
    }
    if (dbs.find(name) == dbs.end()) {
        leveldb::DB *db;
        leveldb::Options options;
        options.create_if_missing = true;
        leveldb::Status status = leveldb::DB::Open(options, "/tmp/testdb."+name, &db);
        if (not status.ok()) {
            return nullptr;
        }
        dbs[name] = db;
    }
    return dbs.at(name);
}

// see http_parser.h
// char DELETE = 0;
// char GET = 1;
// char HEAD = 2;
// char POST = 3;
// char PUT = 4;


static inline void http_ok(int fd, const char *content_type, const char *extra_headers) {
#define write_const(fd, text) \
    write(fd, text, strlen(text));
 
    write_const(fd, "HTTP/1.1 200 OK\r\nContent-Type: ");
    write_const(fd, content_type);
    write_const(fd, "\r\n");
    if (extra_headers) write_const(fd, extra_headers);
    write_const(fd, "\r\n");
#undef write_const
}


static inline void send_file(const char *path, int fd_out) {
    int fd_in;
    struct stat stat_buf;

    const char *mime_type = magic_file(magic_cookie, path);
    if (mime_type == NULL) {
        mime_type = "application/octet-stream";
    }

    http_ok(fd_out, mime_type, nullptr);

    fd_in = open(path, O_RDONLY);
    fstat(fd_in, &stat_buf);
    std::cerr << "GET " << path << std::endl;
    sendfile(fd_out, fd_in, 0, stat_buf.st_size);
    close(fd_in);
}

std::string make_key(uint64_t timestamp) {
    std::stringstream key;
    key << "ts-";
    key << std::setfill('0') << std::setw(20) << timestamp;
    return key.str();
}

void web_handle_api_value(const boost::cmatch &match, const struct http_request *request, const int fd) {
    const char* reply_OK = "HTTP/1.1 200 Value received\r\n\r\n";
    const char* reply_ERR = "HTTP/1.1 500 Internal Error\r\n\r\n";
    
    std::string sensor(match[1].str());
    uint64_t timestamp = std::stoul(match[2].str());
    std::string value(request->body);

    leveldb::DB *db = getDB(sensor);
    if (db == nullptr) {
        write(fd, reply_ERR, strlen(reply_ERR));
        return;
    }

    std::cout << "sensor=" << sensor << " key=" << make_key(timestamp) << std::endl;
    db->Put(leveldb::WriteOptions(), make_key(timestamp), value);
    write(fd, reply_OK, strlen(reply_OK));
}

void web_handle_api_range(const boost::cmatch &match, const struct http_request *request, const int fd) {
    static const leveldb::Comparator *cmp = leveldb::BytewiseComparator();

    const char* reply_OK = "HTTP/1.1 200 Value received\r\n";
    const char* reply_ERR = "HTTP/1.1 500 Internal Error\r\n\r\n";
    const char* content_type = "Content-Type: application/json; encoding=UTF-8\r\n";
 
    std::string sensor(match[1].str());
    uint64_t start = std::stoul(match[2].str());
    uint64_t end = std::stoul(match[3].str());
    std::string key_start(std::move(make_key(start)));
    std::string key_end(std::move(make_key(end)));

    leveldb::DB *db = getDB(sensor);
    if (db == nullptr) {
        write(fd, reply_ERR, strlen(reply_ERR));
        return;
    }
        
    http_ok(fd, "application/json; encoding=UTF-8", nullptr);

    std::cout << "sensor=" << sensor << " start=" << start << " end=" << end << std::endl;

    std::ostringstream out;
    out << "{'sensor':'" << sensor << "', 'error':null, 'data':[";
    write(fd, out.str().c_str(), out.str().size());

    usleep(2000000);
    leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());
    bool first = true;
    for (it->Seek(key_start); 
             it->Valid() && cmp->Compare(it->key(), key_end) < 0;
             it->Next()) {
            if (it->key().size() != 20+3) {
                std::cerr << "invalid key" << std::endl;
                return;
            }
            uint64_t timestamp = std::stoul(it->key().data()+3);
            std::string s_timestamp = std::to_string(timestamp);

            if (first) {
                first = false;
                write(fd, "[", 1);
            } else {
                write(fd, ",[", 2);
            }
            write(fd, s_timestamp.c_str(), s_timestamp.size());
            write(fd, ",'", 2);
            write(fd, it->value().data(), it->value().size());
            write(fd, "']", 2);
    }

    delete it;
}

void handle_request(struct http_request *request, int fd) {
    boost::cmatch match;
    const char *error = "HTTP/1.1 404 Not Found\r\n\r\n";

    if (request->url == NULL) { // happens only under high load, why?
        std::cerr << "url is null" << std::endl;
        goto handled;
    }

    for (auto item = web_handler.begin(); item != web_handler.end(); ++item) {
        if (boost::regex_match(request->url, match, (*item).first)) {
            (*item).second(match, request, fd);
            goto handled;
        }
    }
    
    write(fd, error, strlen(error));

    handled:
    close(fd);
}

static struct http_server server;

void sigint_handler(int s) {
    struct ev_loop *loop = server.loop;
    ev_io_stop(EV_A_ server.ev_accept);
    exit(0);
}

int main(int argc, char **argv) {
    // configure server structures and desired listen address
    struct sockaddr_in listen_addr;
    memset(&listen_addr, 0, sizeof(listen_addr));
    listen_addr.sin_family = AF_INET;
    listen_addr.sin_addr.s_addr = INADDR_ANY;
    listen_addr.sin_port = htons(5000);
    server.listen_addr = &listen_addr;
    server.handle_request = handle_request;

    // ignore SIGPIPE
    struct sigaction on_sigpipe;
    on_sigpipe.sa_handler = SIG_IGN;
    sigemptyset(&on_sigpipe.sa_mask);
    sigaction(SIGPIPE, &on_sigpipe, NULL);

    // handle C-c
    struct sigaction on_sigint;
    on_sigint.sa_handler = sigint_handler;
    sigemptyset(&on_sigint.sa_mask);
    on_sigint.sa_flags = 0;
    sigaction(SIGINT, &on_sigint, NULL);

    // Routing
    web_handler.push_front(std::make_pair(
                boost::regex("/"),
                [](const boost::cmatch &match, const struct http_request *request, const int fd){
                    send_file("index.html", fd);
                }));
    web_handler.push_front(std::make_pair(
                boost::regex("/public/(.+)"),
                [](const boost::cmatch &match, const struct http_request *request, const int fd){
                    /// XXX possible directory traversion
                    std::string path("public/" + match[1].str());
                    send_file(path.c_str(), fd);
                }));
    web_handler.push_front(std::make_pair(
                boost::regex("/api/value/([a-zA-Z0-9]+)/([0-9]+)"),
                web_handle_api_value));
    web_handler.push_front(std::make_pair(
                boost::regex("/api/range/([a-zA-Z0-9]+)/([0-9]+)/([0-9]+)"),
                web_handle_api_range));

    magic_cookie = magic_open(MAGIC_MIME_TYPE);
    magic_load(magic_cookie, NULL);

    // start the server
    return http_server_loop(&server);
}