channel.h

/* Jazz (c) 2018-2025 kaalam.ai (The Authors of Jazz), using (under the same license):
 
    1. Biomodelling - The AATBlockQueue class (c) Jacques Basaldúa, 2009-2012 licensed
      exclusively for the use in the Jazz server software.
 
      Copyright 2009-2012 Jacques Basaldúa
 
    2. BBVA - Jazz: A lightweight analytical web server for data-driven applications.
 
      Copyright 2016-2017 Banco Bilbao Vizcaya Argentaria, S.A.
 
      This product includes software developed at
 
      BBVA (https://www.bbva.com/)
 
    3. LMDB, Copyright 2011-2017 Howard Chu, Symas Corp. All rights reserved.
 
      Licensed under http://www.OpenLDAP.org/license.html
 
 
      Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at
 
      http://www.apache.org/licenses/LICENSE-2.0
 
      Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
*/
 
 
#include <map>
 
#include <microhttpd.h>
#include <curl/curl.h>
 
#include "src/jazz_elements/container.h"
 
#ifdef CATCH_TEST
#ifndef INCLUDED_JAZZ_CATCH2
#define INCLUDED_JAZZ_CATCH2
 
#include "src/catch2/catch.hpp"
 
#endif
#endif
 
 
#ifndef INCLUDED_JAZZ_ELEMENTS_CHANNEL
#define INCLUDED_JAZZ_ELEMENTS_CHANNEL
 
 
namespace jazz_elements
{
 
#define BASE_BASH_10BIT             0x022       //< First 10 bits of base "bash"
#define BASE_FILE_10BIT             0x126       //< First 10 bits of base "file"
#define BASE_HTTP_10BIT             0x288       //< First 10 bits of base "http"
#define BASE_0_MQ_10BIT             0x1b0       //< First 10 bits of base "0-mq"
 
 
#define MAX_FILE_OR_URL_SIZE        1712        
 
 
#define APPLY_NOTHING                    0      
#define APPLY_NAME                       1      
#define APPLY_URL                        2      
#define APPLY_FUNCTION                   3      
#define APPLY_FUNCT_CONST                4      
#define APPLY_FILTER                     5      
#define APPLY_FILT_CONST                 6      
#define APPLY_RAW                        7      
#define APPLY_TEXT                       8      
#define APPLY_ASSIGN_NOTHING             9      
#define APPLY_ASSIGN_NAME               10      
#define APPLY_ASSIGN_URL                11      
#define APPLY_ASSIGN_FUNCTION           12      
#define APPLY_ASSIGN_FUNCT_CONST        13      
#define APPLY_ASSIGN_FILTER             14      
#define APPLY_ASSIGN_FILT_CONST         15      
#define APPLY_ASSIGN_RAW                16      
#define APPLY_ASSIGN_TEXT               17      
#define APPLY_ASSIGN_CONST              18      
#define APPLY_NEW_ENTITY                19      
#define APPLY_GET_ATTRIBUTE             20      
#define APPLY_SET_ATTRIBUTE             21      
#define APPLY_JAZZ_INFO                 22      
 
 
typedef std::map<int, std::string>  MapIS;
 
 
typedef std::map<std::string, Index> ConnMap;
 
 
struct Socket {
    char endpoint[120];     
    void *requester;        
};
 
 
typedef std::vector<uint8_t> GetBuffer;
typedef GetBuffer *pGetBuffer;              
 
 
struct PutBuffer {
    uint64_t to_send;                       
    uint8_t *p_base;                        
};
typedef PutBuffer *pPutBuffer;              
 
 
typedef std::map<std::string, Socket> PipeMap;
 
 
typedef std::map<int, int>  MapII;
 
 
typedef unsigned int MHD_StatusCode;
 
extern size_t get_callback(char *ptr, size_t size, size_t nmemb, void *container);
extern size_t put_callback(char *ptr, size_t size, size_t nmemb, void *container);
extern size_t dev_null(char *_ignore, size_t size, size_t nmemb, void *_ignore_2);
 
class Channels : public Container {
 
    public:
 
        Channels(pLogger     a_logger,
                 pConfigFile a_config);
       ~Channels();
 
        virtual pChar const id();
 
        StatusCode     start       ();
        StatusCode     shut_down   ();
 
        virtual StatusCode get       (pTransaction      &p_txn,
                                      pChar              p_what);
        virtual StatusCode get       (pTransaction      &p_txn,
                                      pChar              p_what,
                                      pBlock             p_row_filter);
        virtual StatusCode get       (pTransaction      &p_txn,
                                      pChar              p_what,
                                      pChar              name);
        virtual StatusCode locate    (Locator           &location,
                                      pChar              p_what);
        virtual StatusCode header    (StaticBlockHeader &hea,
                                      pChar              p_what);
        virtual StatusCode header    (pTransaction      &p_txn,
                                      pChar              p_what);
        virtual StatusCode put       (pChar              p_where,
                                      pBlock             p_block,
                                      int                mode = WRITE_AS_BASE_DEFAULT);
        virtual StatusCode new_entity(pChar              p_where);
        virtual StatusCode remove    (pChar              p_where);
        virtual StatusCode copy      (pChar              p_where,
                                      pChar              p_what);
        virtual StatusCode modify    (Locator           &function,
                                      pTuple             p_args);
        MHD_StatusCode forward_get   (pTransaction      &p_txn,
                                      Name               node,
                                      pChar              p_url);
        MHD_StatusCode forward_put   (Name               node,
                                      pChar              p_url,
                                      pBlock             p_block,
                                      int                mode = WRITE_AS_BASE_DEFAULT);
        MHD_StatusCode forward_del   (Name               node,
                                      pChar              p_url);
 
        // Support for container names in the BaseAPI .base_names()
 
        void base_names(BaseNames &base_names);
 
        // Public config variables
 
        MapIS jazz_node_name = {};              
        MapIS jazz_node_ip   = {};              
        MapII jazz_node_port = {};              
 
        bool search_my_node_index   = false;    
        int  jazz_node_my_index     = -1;       
        int  jazz_node_cluster_size =  0;       
 
        std::string filesystem_root = {};       
 
#ifndef CATCH_TEST
    protected:
#endif
 
        inline StatusCode curl_get(pTransaction &p_txn, const char *url, Index *p_idx = nullptr) {
            CURL *curl;
            CURLcode c_ret;
 
            curl = curl_easy_init();
            if (!curl)
                return SERVICE_ERROR_NOT_READY;
 
            GetBuffer buff = {};
 
            curl_easy_setopt(curl, CURLOPT_URL, url);
            curl_easy_setopt(curl, CURLOPT_VERBOSE, 0);
            curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 1);
            curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, get_callback);
            curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *) &buff);
 
            if (p_idx != nullptr) {
                Index:: iterator it;
                if ((it = p_idx->find("CURLOPT_USERNAME")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_USERNAME, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_USERPWD")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_USERPWD, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_COOKIEFILE")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_COOKIEFILE, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_COOKIEJAR")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_COOKIEJAR, it->second.c_str());
            }
            c_ret = curl_easy_perform(curl);
 
            uint64_t response_code;
 
            if (c_ret == CURLE_OK)
                curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &response_code);
 
            curl_easy_cleanup(curl);
 
            switch (c_ret) {
            case CURLE_OK:
                break;
            case CURLE_REMOTE_ACCESS_DENIED:
            case CURLE_AUTH_ERROR:
                return SERVICE_ERROR_READ_FORBIDDEN;
            case CURLE_REMOTE_FILE_NOT_FOUND:
                return SERVICE_ERROR_BLOCK_NOT_FOUND;
            default:
                return SERVICE_ERROR_IO_ERROR;
            }
 
            switch (response_code) {
            case MHD_HTTP_OK:
            case MHD_HTTP_CREATED:
            case MHD_HTTP_ACCEPTED:
                break;
            case MHD_HTTP_NOT_FOUND:
            case MHD_HTTP_GONE:
                return SERVICE_ERROR_BLOCK_NOT_FOUND;
            case MHD_HTTP_BAD_REQUEST:
                return SERVICE_ERROR_WRONG_ARGUMENTS;
            case MHD_HTTP_UNAUTHORIZED:
            case MHD_HTTP_PAYMENT_REQUIRED:
            case MHD_HTTP_FORBIDDEN:
            case MHD_HTTP_METHOD_NOT_ALLOWED:
            case MHD_HTTP_NOT_ACCEPTABLE:
            case MHD_HTTP_PROXY_AUTHENTICATION_REQUIRED:
            case MHD_HTTP_TOO_MANY_REQUESTS:
                return SERVICE_ERROR_READ_FORBIDDEN;
            case MHD_HTTP_INTERNAL_SERVER_ERROR ... MHD_HTTP_LOOP_DETECTED:
                return SERVICE_ERROR_MISC_SERVER;
            default:
                return SERVICE_ERROR_IO_ERROR;
            }
            size_t buf_size = buff.size();
            if (buf_size > MAX_BLOCK_SIZE)
                return SERVICE_ERROR_BLOCK_TOO_BIG;
 
            buff.push_back(0);
 
            return unwrap_received(p_txn, (pBlock) buff.data(), buf_size);
        }
 
 
        inline StatusCode curl_put(const char *url, pBlock p_blk, int mode = WRITE_AS_STRING | WRITE_AS_FULL_BLOCK, Index *p_idx = nullptr) {
            CURL *curl;
            CURLcode c_ret;
 
            curl = curl_easy_init();
            if (!curl)
                return SERVICE_ERROR_NOT_READY;
 
            PutBuffer put_buff;
 
            if ((mode & WRITE_AS_ANY_WRITE) == 0)
                mode = WRITE_AS_STRING | WRITE_AS_FULL_BLOCK;
 
            if ((mode & WRITE_AS_STRING) && (   (p_blk->cell_type == CELL_TYPE_STRING && p_blk->size == 1)
                                             || (p_blk->cell_type == CELL_TYPE_BYTE   && p_blk->rank == 1))) {
                if (p_blk->cell_type == CELL_TYPE_STRING) {
                    put_buff.p_base  = (uint8_t *) p_blk->get_string(0);
                    put_buff.to_send = strlen((const char *) put_buff.p_base);
                } else {
                    put_buff.p_base  = &p_blk->tensor.cell_byte[0];
                    put_buff.to_send = strnlen((const char *) put_buff.p_base, p_blk->size);
                }
            } else if ((mode & WRITE_AS_CONTENT) && ((p_blk->cell_type & 0xf0) == 0)) {
                put_buff.to_send = p_blk->size*(p_blk->cell_type & 0xff);
                put_buff.p_base  = &p_blk->tensor.cell_byte[0];
            } else if ((mode & WRITE_AS_FULL_BLOCK) && (p_blk->cell_type != CELL_TYPE_INDEX)) {
                put_buff.to_send = p_blk->total_bytes;
                put_buff.p_base  = (uint8_t *) p_blk;
            } else
                return SERVICE_ERROR_WRONG_ARGUMENTS;
 
            curl_easy_setopt(curl, CURLOPT_URL, url);
            curl_easy_setopt(curl, CURLOPT_VERBOSE, 0);
            curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 1);
            curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, dev_null);
            curl_easy_setopt(curl, CURLOPT_READFUNCTION, put_callback);
            curl_easy_setopt(curl, CURLOPT_READDATA, (void *) &put_buff);
            curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);
            curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, (curl_off_t) put_buff.to_send);
 
            if (p_idx != nullptr) {
                Index:: iterator it;
                if ((it = p_idx->find("CURLOPT_USERNAME")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_USERNAME, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_USERPWD")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_USERPWD, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_COOKIEFILE")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_COOKIEFILE, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_COOKIEJAR")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_COOKIEJAR, it->second.c_str());
            }
            c_ret = curl_easy_perform(curl);
 
            uint64_t response_code;
 
            if (c_ret == CURLE_OK)
                curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &response_code);
 
            curl_easy_cleanup(curl);
 
            switch (c_ret) {
            case CURLE_OK:
                break;
            case CURLE_REMOTE_ACCESS_DENIED:
            case CURLE_AUTH_ERROR:
                return SERVICE_ERROR_WRITE_FORBIDDEN;
            case CURLE_REMOTE_FILE_NOT_FOUND:
                return SERVICE_ERROR_BLOCK_NOT_FOUND;
            default:
            return SERVICE_ERROR_IO_ERROR;
            }
 
            switch (response_code) {
            case MHD_HTTP_OK:
            case MHD_HTTP_CREATED:
            case MHD_HTTP_ACCEPTED:
                return SERVICE_NO_ERROR;
            case MHD_HTTP_NOT_FOUND:
            case MHD_HTTP_GONE:
                return SERVICE_ERROR_BLOCK_NOT_FOUND;
            case MHD_HTTP_BAD_REQUEST:
                return SERVICE_ERROR_WRONG_ARGUMENTS;
            case MHD_HTTP_UNAUTHORIZED:
            case MHD_HTTP_PAYMENT_REQUIRED:
            case MHD_HTTP_FORBIDDEN:
            case MHD_HTTP_METHOD_NOT_ALLOWED:
            case MHD_HTTP_NOT_ACCEPTABLE:
            case MHD_HTTP_PROXY_AUTHENTICATION_REQUIRED:
            case MHD_HTTP_TOO_MANY_REQUESTS:
                return SERVICE_ERROR_WRITE_FORBIDDEN;
            case MHD_HTTP_INTERNAL_SERVER_ERROR ... MHD_HTTP_LOOP_DETECTED:
                return SERVICE_ERROR_MISC_SERVER;
            }
            return SERVICE_ERROR_IO_ERROR;
        }
 
 
        inline StatusCode curl_remove(const char *url, Index *p_idx = nullptr) {
            CURL *curl;
            CURLcode c_ret;
 
            curl = curl_easy_init();
            if (!curl)
                return SERVICE_ERROR_NOT_READY;
 
            curl_easy_setopt(curl, CURLOPT_URL, url);
            curl_easy_setopt(curl, CURLOPT_VERBOSE, 0);
            curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 1);
            curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "DELETE");
            curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, dev_null);
 
            if (p_idx != nullptr) {
                Index:: iterator it;
                if ((it = p_idx->find("CURLOPT_USERNAME")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_USERNAME, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_USERPWD")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_USERPWD, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_COOKIEFILE")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_COOKIEFILE, it->second.c_str());
 
                if ((it = p_idx->find("CURLOPT_COOKIEJAR")) != p_idx->end())
                    curl_easy_setopt(curl, CURLOPT_COOKIEJAR, it->second.c_str());
            }
            c_ret = curl_easy_perform(curl);
 
            uint64_t response_code;
 
            if (c_ret == CURLE_OK)
                curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &response_code);
 
            curl_easy_cleanup(curl);
 
            switch (c_ret) {
            case CURLE_OK:
                break;
            case CURLE_REMOTE_ACCESS_DENIED:
            case CURLE_AUTH_ERROR:
                return SERVICE_ERROR_WRITE_FORBIDDEN;
 
            case CURLE_REMOTE_FILE_NOT_FOUND:
                return SERVICE_ERROR_BLOCK_NOT_FOUND;
            default:
                return SERVICE_ERROR_IO_ERROR;
            }
 
            switch (response_code) {
            case MHD_HTTP_OK:
            case MHD_HTTP_CREATED:
            case MHD_HTTP_ACCEPTED:
                return SERVICE_NO_ERROR;
            case MHD_HTTP_NOT_FOUND:
            case MHD_HTTP_GONE:
                return SERVICE_ERROR_BLOCK_NOT_FOUND;
            case MHD_HTTP_BAD_REQUEST:
                return SERVICE_ERROR_WRONG_ARGUMENTS;
            case MHD_HTTP_UNAUTHORIZED:
            case MHD_HTTP_PAYMENT_REQUIRED:
            case MHD_HTTP_FORBIDDEN:
            case MHD_HTTP_METHOD_NOT_ALLOWED:
            case MHD_HTTP_NOT_ACCEPTABLE:
            case MHD_HTTP_PROXY_AUTHENTICATION_REQUIRED:
            case MHD_HTTP_TOO_MANY_REQUESTS:
                return SERVICE_ERROR_WRITE_FORBIDDEN;
            case MHD_HTTP_INTERNAL_SERVER_ERROR ... MHD_HTTP_LOOP_DETECTED:
                return SERVICE_ERROR_MISC_SERVER;
            }
            return SERVICE_ERROR_IO_ERROR;
        }
 
#ifndef CATCH_TEST
    private:
#endif
        char HEX[16] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F'};    
 
        inline bool compose_url(pChar p_dest, pChar p_node, pChar p_url, int buff_size) {
            int nix;
 
            for (MapIS::iterator it = jazz_node_name.begin(); true; ++it) {
                if (it == jazz_node_name.end())
                    return false;
 
                if (it->second == p_node) {
                    nix = it->first;
                    break;
                }
            }
            int ret = snprintf(p_dest, buff_size, "http://%s:%i", jazz_node_ip[nix].c_str(), jazz_node_port[nix]);
 
            if (ret < 0 || ret >= buff_size)
                return false;
 
            p_dest    += ret;
            buff_size -= ret;
 
            while (buff_size > 0) {
                u_int8_t cursor = *(p_url++);
                switch (cursor) {
                case 0:
                    *p_dest = 0;
                    return true;
                case '!' ... '$':
                case '&' ... '/':
                case '0' ... '9':
                case ':':
                case ';':
                case '=':
                case '?':
                case '@':
                case 'A' ... 'Z':
                case '[':
                case ']':
                case '_':
                case 'a' ... 'z':
                case '~':
                    *(p_dest++) = cursor;
                    buff_size--;
                    break;
                default:
                    if (buff_size < 4)
                        return false;
                    *(p_dest++) = '%';
                    *(p_dest++) = HEX[cursor >> 4];
                    *(p_dest++) = HEX[cursor & 0x0f];
                    buff_size -= 3;
                }
            }
            return false;
        }
 
        int can_curl = false;           
        int curl_ok  = false;           
        int can_zmq  = false;           
        int zmq_ok   = false;           
        int can_bash = false;           
        int file_lev = 0;               
 
        PipeMap pipes   = {};           
        ConnMap connect = {};           
 
        void *zmq_context = nullptr;    
};
typedef Channels *pChannels;            
 
 
#ifdef CATCH_TEST
 
// Instancing Channels
// -------------------
 
extern Channels CHN;
 
#endif
 
} // namespace jazz_elements
 
#endif // ifndef INCLUDED_JAZZ_ELEMENTS_CHANNEL