

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.1.1.260726 (26 July 2026)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2026
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  and is released under the following
//_/_/  
//_/_/  EXTENDED BSD License
//_/_/
//_/_/  Redistribution and use in source and binary forms, with or without
//_/_/  modification, is permitted provided that the following conditions 
//_/_/  are met:
//_/_/
//_/_/  - Redistributions of source code must retain the above copyright 
//_/_/    and collaboration notice, this list of conditions and the following 
//_/_/    disclaimer.
//_/_/
//_/_/  - Redistributions in binary form must reproduce the above copyright 
//_/_/    notice, this list of conditions and the following disclaimer in the 
//_/_/    documentation and/or other materials provided with the distribution.
//_/_/
//_/_/  - Neither the name of its copyright holders nor the names of its 
//_/_/    contributors may be used to endorse or promote products derived from 
//_/_/    this software without specific prior written permission.
//_/_/
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
/*! @file RequestGateway.h
	@brief Request-system router: RequestGateway accepts client requests (binary/HTTP/WebSocket), load-balances them across RequestExecutors and routes replies back; includes a call log and an embedded web server.

	The gateway is the hub of the request fabric (see RequestClient.h for the
	conceptual overview and a round-trip example). One process typically runs
	one RequestGateway listening on one or more ports (addPort()); clients
	and executors connect @e in to it (a gateway may additionally link to other
	gateways via addGateway(), but it never dials out to clients or executors),
	which simplifies firewalling.

	Responsibilities:
	- @b Ingress: accepts requests over the binary message protocol
	  (RequestClient), plain HTTP(S) and WebSocket — all on the same ports
	  via protocol auto-detection. HTTP requests are converted to
	  DataMessages and travel the same internal path.
	- @b Routing / @b load @b balancing: getBestExecutorID() picks an
	  executor per request based on reported queue sizes, latency stats and
	  the short/long classification; queuing behaviour is tuned with
	  setQueuingParameters() (max queue size, max processing set, priority
	  threshold).
	- @b Liveness: executors heartbeat periodically; one missing beyond
	  setExecutorHeartbeatTimeout() is declared dead, its queued requests are
	  redistributed (distributeDeadExecutorRequests()) and it may be asked to
	  restart (sendRestartToExecutor()).
	- @b Web @b serving + @b APIs: a built-in file web server (rootdir/index
	  via setWebServerInfo()) plus internal/external HTTP API surfaces and
	  optional basic-auth users (addAuthUser()).
	- @b Audit: every call is recorded as a CallLogEntry for diagnostics.

	A minimal gateway serving binary clients, web traffic and executors on
	one port looks like this:
	@code
	RequestGateway gw;
	gw.setWebServerInfo("MyGateway", "./wwwroot", "index.html"); // name + static files
	gw.setExecutorHeartbeatTimeout(5000);             // declare dead after 5s silence
	gw.addPort(9000, NOENC, true);   // clients + executors + HTTP on one port
	// ... RequestExecutors addGateway(id, host, 9000) and start pulling work;
	// ... RequestClients addGateway(id, host, 9000) and postRequest().
	@endcode

	@ingroup Requests
*/
#if !defined(_REQUESTGATEWAY_H_)
#define _REQUESTGATEWAY_H_

#pragma once

#include "Stats.h"
#include "NetworkManager.h"
#include "RequestExecutor.h"
#include "RequestClient.h"


namespace	cmlabs{

/*! One entry of the gateway's call log: who called what, how it was routed,
	how long each stage took, and the outcome. Serialisable to JSON for the
	diagnostics API.
	@ingroup Requests */
class CallLogEntry {
public:
	CallLogEntry(uint64 id = 0) {
		this->id = id;
		conType = messageType = requestType = 0;
		longRequest = false;
		port = 0;
		source = executorID = clientID = startTime = endTime = queueTime = processTime = 0;
		requestSize = replySize = accountID = 0;
	}
	virtual ~CallLogEntry() {};

	/*! @return This entry as a JSON object string (empty when mandatory fields are missing). */
	std::string toJSON() {
		if (!id || !startTime || !source || !conType || !messageType)
			return "";
		return utils::StringFormat(
			"{\"id\":%llu, \"port\":%u, \"source\":\"%u.%u.%u.%u:%u\", \"connectiontype\":\"%s\", \"messagetype\":\"%s\", "
			"\"requesttype\":\"%s\", \"longrequest\": \"%s\", \"status\": \"%s\", \"request\": \"%s\", \"output\": \"%s\", "
			"\"executorid\":%llu, \"clientid\":%llu, \"accountid\":%u, "
			"\"requestsize\":%u, \"replysize\":%u, "
			"\"requesttime\": \"%s\", \"duration\": \"%.3f\", \"queuetime\": \"%.3f\", \"processtime\": \"%.3f\"}\n",
			id, port, GETIPADDRESSQUADPORT(source),
			((conType == 1) ? "Client" : (conType == 2) ? "Web" : "Websocket"),
			((messageType == 1) ? "Binary" : (messageType == 2) ? "HTTP" : "HTTPS"),
			HTTP_Type[requestType],
			(longRequest ? "true" : "false"),
			status.c_str(), requestPath.c_str(), output.c_str(),
			executorID, clientID, accountID,
			requestSize, replySize,
			PrintTimeString(startTime).c_str(),
			(double)(endTime - startTime)/1000.0,
			(double)queueTime/1000.0,
			(double)processTime/1000.0
			);
	}

	uint64 id;			/*!< Unique call id. */
	uint8 conType;      /*!< Ingress type — 1: client   2: web   3: websocket */
	uint8 messageType;	/*!< Wire format — 1: message  2: http  3: https */
	bool longRequest;	/*!< Whether the request was classified long. */
	uint64 source;		/*!< Caller endpoint packed as uint64. */
	uint16 port;		/*!< Gateway port the call arrived on. */
	uint64 executorID;	/*!< Executor that processed the request. */
	uint64 clientID;	/*!< Client connection id. */
	uint32 accountID;	/*!< Authenticated account id (0 if anonymous). */
	uint64 startTime;	/*!< Request arrival time (ms epoch). */
	uint64 endTime;		/*!< Reply completion time (ms epoch). */
	uint64 queueTime;	/*!< Time spent queued before dispatch (µs). */
	uint64 processTime;	/*!< Time spent processing in the executor (µs). */
	uint8 requestType;	/*!< HTTP_* method id (for web ingress). */
	uint32 requestSize;	/*!< Request payload size in bytes. */
	uint32 replySize;	/*!< Reply payload size in bytes. */
	std::string status;			/*!< Outcome status text. */
	std::string requestPath;	/*!< Request name/URI. */
	std::string agent;			/*!< Caller's User-Agent (web ingress). */
	std::string output;			/*!< Output/diagnostic note. */
};




/////////////////////////////////////////////////////////////
// RequestGateway
/////////////////////////////////////////////////////////////

/*! The request-fabric router/load-balancer (see the file header for the full role description).

	Runs two worker threads: one servicing executor traffic (runExec()) and
	one servicing client traffic (runClient()). All ingress arrives through
	the NetworkReceiver callbacks below; ports are added with addPort() and
	may simultaneously carry binary, HTTP and WebSocket traffic thanks to
	protocol auto-detection.
	@par Thread-safety
	Configuration methods (set* and add*) should be called before init(); the
	receive*() callbacks run concurrently on connection threads and are
	internally synchronised.
	@ingroup Requests */
class RequestGateway : public Runnable, public NetworkReceiver {
public:
	/*! End-to-end stress test spinning up a gateway plus the given numbers of
		executors, extra gateways and binary/web/websocket test clients.
		@param port Base port. @param executorCount Executors. @param gatewayCount Gateways.
		@param clientCount Binary clients. @param webCount HTTP clients. @param webSocketCount WebSocket clients.
		@param sendfreq Requests per second per client. @param reconnect Client reconnect interval (ms; <0 = never).
		@param payload Request payload size in bytes. @param processTime Simulated executor processing time (ms).
		@param runtime Test duration (s). @return true when the run completes without errors. */
	static bool UnitTest(uint16 port, uint32 executorCount, uint32 gatewayCount, uint32 clientCount, uint32 webCount, uint32 webSocketCount, double sendfreq, int32 reconnect, uint32 payload, uint32 processTime, uint32 runtime);

	friend THREAD_RET THREAD_FUNCTION_CALL RequestGatewayExecRun(THREAD_ARG arg);
	friend THREAD_RET THREAD_FUNCTION_CALL RequestGatewayClientRun(THREAD_ARG arg);

	/*! @param id Gateway id (reported to peers). @param version Optional version string shown in status pages. */
	RequestGateway(uint32 id, const char* version = NULL);
	virtual ~RequestGateway();

	/*! Start the gateway threads and network stack.
		@param sslCertPath Optional PEM certificate enabling SSL ports.
		@param sslKeyPath Matching PEM private key.
		@return true on success. Call after the set*() and add*() configuration calls. */
	bool init(const char* sslCertPath = NULL, const char* sslKeyPath = NULL);

	/*! Configure the built-in file web server.
		@param name Server header name. @param rootdir Document root directory.
		@param indexfile Default index filename. @return true on success. */
	bool setWebServerInfo(const char* name, const char* rootdir, const char* indexfile);
	/*! Set the URI label under which the external (public) API is exposed. @return true. */
	bool setExternalAPILabel(const char* label);
	/*! Set the URI label under which the internal (management) API is exposed. @return true. */
	bool setInternalAPILabel(const char* label);
	/*! Enable/disable HTTP caching headers for served files. @return true. */
	bool setCacheFiles(bool cache);
	/*! Choose the reply serialisation ("xml" or "json"). @return true. */
	bool setResponseType(const char* type);
	/*! Cap concurrent long requests per executor. @param limit Maximum. @return true. */
	bool setLongRequestLimit(uint32 limit);
	/*! Set how long an executor may go silent before being declared dead
		(its queue is then redistributed). @param timeout ms. @return true. */
	bool setExecutorHeartbeatTimeout(uint32 timeout);
	/*! Set the maximum time a request may sit with an executor before being
		failed/retried. @param timeout ms. @return true. */
	bool setMaxExecutorRequestTimeout(uint32 timeout);
	/*! Tune queuing/backpressure behaviour.
		@param maxRequestQueueSize Max queued requests before rejecting with TOOBUSY.
		@param maxRequestProcessingSize Max requests dispatched concurrently.
		@param priorityThreshold Queue length above which priority ordering kicks in.
		@return true. */
	bool setQueuingParameters(uint32 maxRequestQueueSize, uint32 maxRequestProcessingSize, uint32 priorityThreshold);

	/*! Open a listening port for clients/executors (call before init()).
		@param port TCP port. @param encryption ::NOENC or ::SSLENC.
		@param enableHTTP Also accept HTTP/WebSocket on this port when true.
		@param timeout Protocol auto-detection timeout in ms. @return true on success. */
	bool addPort(uint16 port, uint8 encryption, bool enableHTTP = false, uint32 timeout = 3000);
	/*! Register a peer gateway (multi-gateway federation).
		@param id Gateway id. @param addr Host. @param port Port. @return true. */
	bool addGateway(uint32 id, std::string addr, uint16 port);
	/*! Add a basic-auth user for the web interface/APIs.
		@param user Username. @param password Password. @return true. */
	bool addAuthUser(const char* user, const char* password);

	/*! NetworkReceiver hook: connect/disconnect events for clients and executors. */
	bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid);
	/*! NetworkReceiver hook: binary requests/replies/heartbeats from clients and executors. */
	bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid);
	/*! NetworkReceiver hook: web ingress — serves files, APIs, or converts the
		request into an internal DataMessage request. */
	bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid);
	/*! NetworkReceiver hook: WebSocket ingress carrying request payloads. */
	bool receiveWebsocketData(WebsocketData* wsData, NetworkChannel* channel, uint64 conid);

protected:
	/*! Executor-side worker loop: dispatch queued requests, watch heartbeats,
		redistribute dead executors' work. @return true to continue. */
	bool runExec();
	/*! Client-side worker loop: route replies back to callers, expire timed-out
		requests. @return true to continue. */
	bool runClient();

	/*! Handle a management/API call on the internal API label. @return true if handled. */
	bool callInternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid);
	/*! Handle a public API call on the external API label. @return true if handled. */
	bool callExternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid);

	/*! Ask an executor to restart itself (sent when it misbehaves). @return true if sent. */
	bool sendRestartToExecutor(uint64 id);
	/*! Requeue a dead executor's outstanding requests onto surviving executors.
		@param shortQueue Its short-request backlog. @param longQueue Its long-request backlog.
		@return true on success. */
	bool distributeDeadExecutorRequests(std::list<RequestReply*>& shortQueue, std::list<RequestReply*>& longQueue);

	/*! Pick the executor to receive a request — the core load-balancing decision,
		based on reported queue sizes, rolling latency stats and liveness.
		@param requestString Request name (matched against long-request names).
		@param reqSize Payload size. @param isLongReq Out: classification result.
		@return The chosen executor id, or 0 when none is available. */
	uint64 getBestExecutorID(std::string requestString, uint32 reqSize, bool& isLongReq);
	//std::list<RequestConnection> getTopExecutors(std::string requestString, uint32 reqSize, bool& isLongReq);
	//bool sendToExecutor(RequestReply* reply);
	/*! Route a reply message back to the client that issued the request. @return true if sent. */
	bool replyToClient(DataMessage* msg);
	/*! Wrap an incoming message in a RequestReply and enqueue it for dispatch.
		@param msg Request message (ownership transfers). @param origin Caller endpoint.
		@param conID Ingress connection id. @param clientRef Caller's reference id.
		@return true if queued (false when over ::maxRequestQueueSize). */
	bool addToRequestQueue(DataMessage* msg, uint64 origin, uint64 conID, uint64 clientRef);
	/*! Enqueue an already-wrapped request (e.g. redistributed work). @return true if queued. */
	bool addRequestReplyToRequestQueue(RequestReply* reply);

	/*! Append one completed call to the rolling call log (trimmed to ::callLogMax).
		Parameters mirror the CallLogEntry fields. @return true on success. */
	bool addToCallLog(
		uint64 id,
		uint8 conType,
		uint8 messageType,
		bool longRequest,
		uint64 source,
		uint16 port,
		uint64 executorID,
		uint64 clientID,
		uint64 startTime,
		uint64 endTime,
		uint64 queueTime,
		uint64 processTime,
		uint8 requestType,
		uint32 requestSize,
		uint32 replySize,
		const char* status,
		const char* requestPath,
		const char* agent,
		const char* output
	);
	utils::Mutex callLogMutex;			/*!< Guards the call log. */
	std::list<CallLogEntry*> callLog;	/*!< Rolling log of completed calls (owned). */
	uint32 callLogMax;					/*!< Max entries kept in ::callLog. */
	uint64 callLogCount;				/*!< Total calls logged since start. */

	NetworkManager* manager;	/*!< Owned network stack (listeners + connections). */
	NetworkChannel* channel;	/*!< Channel carrying all gateway traffic. */
	std::map<uint64,RequestConnection> executors;	/*!< Connected executors by id. */
	std::map<uint64,RequestConnection> clients;		/*!< Connected binary clients by id. */
	std::map<uint64,RequestConnection> webClients;	/*!< Connected HTTP clients by id. */
	std::map<uint64,RequestConnection> webSockets;	/*!< Connected WebSocket clients by id. */
	utils::WaitQueue<RequestReply*> execQ;			/*!< Requests awaiting executor dispatch. */
	utils::WaitQueuePointer<DataMessage*> replyQ;	/*!< Replies awaiting client routing. */
	std::map<uint64,RequestReply*> requestMap;		/*!< In-flight requests by gateway ref id. */
	std::vector<std::string> longReqNames;			/*!< Request names classified as long. */
	std::map<std::string, uint64> httpAuth;			/*!< Basic-auth users (user -> credential hash). */

	utils::Mutex mutex;			/*!< Guards shared gateway state. */
	//uint32 threadID;
	uint64 systemStartTime;		/*!< Gateway start time (ms epoch), for uptime reporting. */
	uint32 threadIDClient;		/*!< Client-side worker thread id. */
	uint64 lastRefID;			/*!< Last issued gateway reference id. */
	uint64 clientSentCount;		/*!< Messages sent to clients. */
	uint64 clientReceivedCount;	/*!< Messages received from clients. */
	uint64 execSentCount;		/*!< Messages sent to executors. */
	uint64 execReceivedCount;	/*!< Messages received from executors. */

	uint32 id;							/*!< This gateway's id. */
	uint16 port;						/*!< Primary listening port. */
	uint32 longReqLimit;				/*!< Concurrent long-request cap per executor. */
	uint32 executorHeartbeatTimeout;	/*!< Silence threshold before an executor is declared dead (ms). */
	uint32 maxRequestQueueSize;			/*!< Backpressure: max queued requests before TOOBUSY. */
	uint32 maxRequestProcessingSize;	/*!< Backpressure: max requests dispatched concurrently. */
	uint32 priorityThreshold;			/*!< Queue length above which priority ordering applies. */
	uint32 maxExecutorRequestTimeoutMS;	/*!< Max time a request may sit with an executor (ms). */
	bool sslSupport;					/*!< True when an SSL certificate was configured. */
	bool cacheFiles;					/*!< Send caching headers for served files. */
	bool replyXML;						/*!< Reply in XML (true) or JSON (false). */

	std::string webServerName;	/*!< Server header name of the built-in web server. */
	std::string rootdir;		/*!< Document root for file serving. */
	std::string indexFilename;	/*!< Default index file name. */
	std::string versionString;	/*!< Version shown in status pages. */

	std::string internalAPITitle;	/*!< URI label of the internal (management) API. */
	std::string externalAPITitle;	/*!< URI label of the external (public) API. */

	MovingAverage shortAvgStats;	/*!< Rolling latency stats for short requests. */
	MovingAverage longAvgStats;		/*!< Rolling latency stats for long requests. */
	//std::map<uint64, MovingAverage> shortAvgStatsExec;
	//std::map<uint64, MovingAverage> shortAvgStatsClients;
	//std::map<uint64, MovingAverage> shortAvgStatsWeb;
	//std::map<uint64, MovingAverage> longAvgStatsExec;
	//std::map<uint64, MovingAverage> longAvgStatsClients;
	//std::map<uint64, MovingAverage> longAvgStatsWeb;
};

/*! Thread entry point for the gateway's executor-side loop. @ingroup Requests */
THREAD_RET THREAD_FUNCTION_CALL RequestGatewayExecRun(THREAD_ARG arg);
/*! Thread entry point for the gateway's client-side loop. @ingroup Requests */
THREAD_RET THREAD_FUNCTION_CALL RequestGatewayClientRun(THREAD_ARG arg);

}

#endif // _REQUESTGATEWAY_H_
