

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE 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/
//_/_/
//_/_/  PSYCLONE 2.1 is released under the GNU Lesser General Public License 
//_/_/  (LGPL) version 3.0, and is subject to the following
//_/_/  
//_/_/  - 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. 
//_/_/    
//_/_/  A commercial license for PSYCLONE 2.1 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.1 and/or modify it under 
//_/_/  the terms of the GNU General Public License as published by the Free 
//_/_/  Software Foundation, either version 3 of the License, or (at your 
//_/_/  option) any later version.
//_/_/  
//_/_/  You should have received a copy of the GNU General Public License
//_/_/  and the GNU Lesser General Public License along with PSYCLONE 2.1.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.1 is distributed in the hope that it will be useful,
//_/_/  but WITHOUT ANY WARRANTY; without even the implied warranty of
//_/_/  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//_/_/  GNU General Public License for more details.
//_/_/
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#if !defined(_NODE_H_)
#define _NODE_H_

#pragma once

#define __SOFTWARE_VERSION_STRING__		"2.1.1"

#include "PsyAPI.h"
#include "Networking.h"
#include "Logging.h"
#include "Console.h"
#include "PsyProbe.h"
#include "Monitoring.h"
#include "SubscriptionEngine.h"
#include "ServiceManager.h"
#include "IDManager.h"
#include "PsySpace.h"
#include "ExternalExec.h"
//#include "InterSystemManager.h"
#include <map>
#include <set>

namespace	cmlabs{


struct NodeInfoPackHeader {
	uint32	size;		// auto
	uint32	cid;		// auto
	uint64	time;		// auto
	uint64	createdTime;
	uint16	nodeID;
	char	name[256];
	char	addr[256];
	char	group[256];
	uint16	port;
	uint16	procCount;
	uint32	compCount;
	AveragePerfStats perfStruct;
};

class NodeInfoPack {
public:
	NodeInfoPack(const char* data, bool copy = true);
	~NodeInfoPack();

	bool isValid();

	const NodeInfoPackHeader* getNodeInfo();
	const ProcessInfoStruct* getProcInfo(uint16 id);
	const ProcessInfoStruct* getProcInfo(const char* name);
	const ComponentInfoStruct* getCompInfo(uint16 id);
	const ComponentInfoStruct* getCompInfo(const char* name);

	std::string toXML();

	std::map<uint16, std::map<uint32, ComponentInfoStruct*> > getCatalogs();
	std::map<uint16, std::map<uint32, ComponentInfoStruct*> > getModules();
	std::map<uint16, std::pair<std::map<uint32, ComponentInfoStruct*>, std::map<uint32, ComponentInfoStruct*> > > getHierarcy();

	NodeInfoPackHeader* data;
};

#define SYNC_ERROR		0
#define SYNC_NONE		1
#define SYNC_WORKING	2
#define SYNC_TIMEOUT	3
#define SYNC_FAILED		4
#define SYNC_SUCCESS	5
#define SYNC_RETRY		6

#define NODESYNC_TIMEOUT_MS			5000
#define NODEQUERY_TIMEOUT_MS		5000
#define IDSYNC_RESERVE_TIMEOUT_MS	2000

#define NODE_EXIT_ERROR		-1
#define NODE_EXIT_DONE		0
#define NODE_EXIT_RESTART	1


class Node;
class MessageTimer {
public:
	static MessageTimer* Singleton;
	static bool DrumBeat(uint32 id, uint32 count);

	MessageTimer(Node* node);
	~MessageTimer();

	uint32 addTrigger(TriggerSpec* spec);
	bool removeTrigger(uint32 id);
	bool removeComponent(uint32 compID);

private:
	Node* node;
	std::map<uint32, TriggerSpec*> triggers;
	std::map<uint32, std::list<uint32> > components;
};

class InterSystemManager;

// T1.6: JoinEngine — declarative <triggergroup> join operator. Accumulates member
// messages Node-side (before the per-component enqueue) and activates the crank ONCE
// per completed set. See Docs/T1.6-triggergroup-plan.md. Owned by Node; guarded by an
// internal Mutex; a single DrumBeat sweep evicts/partial-fires stale sets.
class JoinEngine {
public:
	static JoinEngine* Singleton;
	static bool SweepBeat(uint32 id, uint32 count);	//!< periodic maxage/timeout sweep callback

	JoinEngine(Node* node);
	~JoinEngine();

	//! Register a group (copies the packed spec). Called at subscription time. @return true.
	bool registerGroup(TriggerGroupSpec* group, uint16 crankID);
	//! Tear down all groups + partial sets for a component (on delete/migrate).
	bool removeComponent(uint32 compID);
	//! @return true if this group id is a registered join group.
	bool isJoinGroup(uint32 groupID);
	//! Route a member message into its group buffer; may fire on completion. @return true if handled (do not enqueue normally).
	bool routeMessage(TriggerSpec* memberSpec, DataMessage* msg, uint64 recvtime);
	//! Sweep all partial sets: evict/partial-fire those past group maxage/timeout.
	void sweep();
	//! Stop the sweep DrumBeat immediately (call at teardown before spaces/manager are freed).
	void stopSweep();

private:
	struct MemberValue {
		DataMessage* msg;		//!< owned copy of the member message
		uint64 arrival;			//!< wall-clock arrival (us) for per-member maxage
	};
	struct PartialSet {
		std::map<std::string, std::list<MemberValue> > members;	//!< name -> value(s) (list for dup=keep; size 1 for latest)
		std::vector<std::string> order;							//!< arrival order of member names (order=strict)
		uint64 firstArrival;	//!< wall-clock of first member (group maxage)
		uint64 lastFire;		//!< wall-clock of last activation (debounce/min re-fire/slide)
		uint64 minSend, maxSend;	//!< getSendTime bounds (within=)
		uint64 minCreated, maxCreated;	//!< getCreatedTime bounds (time=)
		uint32 fireCount;		//!< activations so far (min= re-fire bookkeeping)
	};
	struct GroupState {
		TriggerGroupSpec* spec;	//!< owned copy (malloc'd packed block)
		uint16 crankID;			//!< resolved crank to activate
		TriggerSpec* memberTemplate;	//!< T1.6: owned clone of a member TriggerSpec (carries the module's post specs) so a joined activation lets the crank postOutputMessage(); NULL until first member routed
		std::map<uint32, PartialSet> partials;	//!< keyed by message tag id (0/empty-tag => getTagID(""))
	};

	Node* node;
	utils::Mutex mutex;
	uint32 sweepBeatID;
	bool sweepRunning;
	std::map<uint32, GroupState> groups;	//!< keyed by group id (== member TriggerSpec.group)

	//! Completion check for a partial set under a group's policy. @return true if it should fire.
	bool isComplete(GroupState& g, PartialSet& ps, bool timedOut);
	//! Emit one joined activation for the set (Step 4 wires CTRL_TRIGGER_GROUP + SDK map).
	void fire(GroupState& g, uint32 tagKey, PartialSet& ps, bool partial);
	//! Free all member messages held in a partial set.
	void freePartial(PartialSet& ps);
	void ensureSweep();	//!< lazily start the sweep DrumBeat once a group exists
};


class Node {
	friend class Networking;
	friend class SubscriptionEngine;
	friend class MessageTimer;
	friend class JoinEngine;		// T1.6
	friend class PsyProbe;
	friend class Monitoring;
	friend class InterSystemManager;
	friend class ExternalExecManager;
public:

//	friend THREAD_RET THREAD_FUNCTION_CALL NodeNetworking(THREAD_ARG arg);
	friend THREAD_RET THREAD_FUNCTION_CALL NodeMessageThread(THREAD_ARG arg);
	friend THREAD_RET THREAD_FUNCTION_CALL NodeSignalThread(THREAD_ARG arg);
	friend THREAD_RET THREAD_FUNCTION_CALL NodeRequestThread(THREAD_ARG arg);

	// ***************** Constructors *****************
	Node(const char* logdir, const char* procname);
	virtual ~Node();

	// ***************** Initialisation *****************
	// Init with or without config file
	bool init(uint16 port, const char* configXML = NULL, const char* htmldir = NULL);

	// Substitute %name% variables in configXML into 'xml' (leaves 'xml' empty if
	// nothing was replaced). A 'name=value' argument on the Psyclone command line
	// overrides the <variable name=.. value=..> default from the spec.
	// Static and side-effect free so it can be unit tested (test=psy_configvars).
	static bool ExpandConfigVariables(XMLNode& mainNode, const char* configXML, std::string& xml);
	bool initExternal(uint16 port, const char* otherNodeAddress, uint16 otherNodePort);
	bool join(uint16 nodeID);
	bool rejoinNode(uint16 nodeID);

	bool setConfigFile(const char* configfile);

	// ***************** Running *****************
	int8 run();
	bool runMessageQ();
	bool runSignalQ();
	bool runRequestQ();

	// ***************** Messaging *****************
	bool processMessageFromNetwork(DataMessage* msg);
	DataMessage* getCopyOfMessage(uint64 id);

	uint8 query(DataMessage* msg, DataMessage** result, uint32 timeout);
	uint8 retrieve(std::list<DataMessage*> &result, RetrieveSpec* spec, uint32 timeout = 5000);

	// ***************** Initialisation *****************

	uint16 getNodeID() {return nodeID;}
	uint16 getNodeID(uint32 compID);
	bool setNodeID(uint16 nodeID) {this->nodeID = nodeID; return true;}

	bool checkPsyclone(const char* addr, uint16 port);

	// ***************** Main APIs *****************
	bool shutdown(uint32 ms);
	bool terminate();
	bool restart();
	//! Accessor for the external-executable lifecycle manager (T1.4; NULL if none declared). Used by unit tests and diagnostics.
	ExternalExecManager* getExternalExec() { return externalExec; }
	bool reinit(const char* filename = NULL);
	bool reinit(uint16 port, const char* filename = NULL);

	NodeInfoPackHeader* getLocalInfoPack();
	bool getAllInfoPacks(std::map<uint16, NodeInfoPack*>& infopacks);
	bool getOtherNodeCompCom(std::map<uint64, PerformanceMatrixResults>& nodeCom, std::map<uint64, PerformanceMatrixResults>& compCom);

	uint64 procCount;
	uint64 procBytes;
	uint64 toCount;
	uint64 toBytes;
	uint64 deliverCount;
	uint64 deliverBytes;
	bool logIntersystemTraffic;

	uint16 nodeID;
	uint16 networkPort;
	uint32 osProcID;

protected:
	uint32 networkThreadID;
	bool shouldContinue;
	bool shouldRestart;
	bool isNodeRunning;
	bool isMsgQRunning;
	bool isSigQRunning;
	bool isReqQRunning;
	SOCKET listenSocket;
	MemoryManager* manager;
	std::string localHostname;
	std::string logdir;
	std::string procname;
	std::string configfile;
	std::map<std::string, uint64> nodes;
	DataMessage* configNodeMessage;
	std::map<std::string, std::string> libraryPaths;
	std::string origConfigXML;

	utils::TimeQueuePointer<DataMessage*> delayedLocalMsgs;
	utils::TimeQueuePointer<DataMessage*> delayedNetworkMsgs;

	uint64 createTime;
	uint64 currentCPUTicks;

	AveragePerfStats lastPerfStruct;

	PsyType systemReadyType;
	PsyType shutdownType;
	PsyContext defaultContext;

	uint32 sysQID;
	uint32 procLogQID;

	// ID Reservations
	IDManager* idManager;

	Networking* networking;
	Logging* logging;
	Console* console;
	PsyProbe* psyProbe;
	Monitoring* monitoring;
	ServiceManager* serviceManager;
	SubscriptionEngine* subscriptionEngine;
	PsySpace* internalSpace;
	std::set<uint16> internallySpawnedSpaces;			// space ids this node spawned itself (plain internal spaces)
	std::map<uint16, std::list<uint64> > spaceRestartHistory;	// per-space restart timestamps (rate limiting)
	ExternalExecManager* externalExec;	// T1.4: external-module executable lifecycle
	MessageTimer* messageTimer;
	JoinEngine* joinEngine;				// T1.6: <triggergroup> declarative join operator
	InterSystemManager* interSystemManager;

	uint32 msgThreadID;
	uint32 sigThreadID;
	uint32 reqThreadID;

	bool initialConfig(uint16 port);
	bool initialiseControlTypes();

	XMLNode parseAndExpandConfig(const char* configXML);
	bool checkRuntimeLicence(XMLNode &mainNode);
	bool startConfigNode(DataMessage* msg);
	bool configNode(const char* configXML, bool isConfigNode);
	bool configNode(XMLNode &n, bool isConfigNode);

	PsyType getPsyType(const char* typestring);
	PsyContext getPsyContext(const char* contextstring);
	uint32 getTagID(const char* name);

	uint16 getCrankID(const char* name);
	uint32 getComponentID(const char* name, bool autoCreate = true);
	
	bool bulkContextSyncNode(uint16 nodeID);
	bool bulkIDSyncNode(uint16 nodeID);
	uint8 negotiateIDSync(uint8 type, uint32 id, const char* name, const void* idEntry, uint32 size);

	uint16 syncTypeLevelID(const char* name);
	uint16 syncContextLevelID(const char* name);
	uint32 syncTagID(const char* name);
	uint16 syncCrankID(uint32 compID, const char* name, const char* function, const char* libraryFilename, const char* language, const char* script);
	uint32 syncComponentID(uint8 type, uint8 policy, uint8 selfTrigger, const char* name, uint16 nodeID, uint16 procID);

	bool syncComponentLocation(uint32 id, uint16 nodeID, uint16 procID);
	bool syncComponentTriggers(uint32 compID, std::list<TriggerSpec*> &triggers);

	uint8 confirmIDSync(uint8 idType, uint32 id);
	uint8 cancelIDSync(uint8 idType, uint32 id);
	
	bool processIDSync(DataMessage* msg, uint16 remoteNodeID, uint32 syncID, uint32& existingID);
	//bool processIDSync(uint8 idType, const char* entryData, uint16 remoteNodeID, uint32 syncID, uint32& existingID);
	bool processConfirmIDSync(uint8 idType, uint32 id, uint16 remoteNodeID);
	bool processCancelIDSync(uint8 idType, uint32 id, uint16 remoteNodeID);

	bool gatherLocalPerformanceStats(std::list<PerfStats> &perfStats);
	bool addLocalNodePerformanceStats(std::list<PerfStats> &perfStats);

	std::map<std::string, std::string> getComponentPrivateDataKeysAndTypes(uint32 compID);
	char* getComponentPrivateData(uint32 compID, const char* dataName, uint64 &size, std::string &mimetype);

	// uint8 verifyOwnIDReservation(IDReservation& res);
	//bool registerTypeID(uint32 id, const char* name);
	//bool registerContextID(uint32 id, const char* name);
	//bool registerCrankID(uint32 id, uint32 compID, const char* name, const char* function, const char* libraryFilename, const char* script);
	//bool registerComponentID(uint32 id, uint8 type, uint8 policy, const char* name, uint16 nodeID, uint16 procID);
	//bool registerComponentLocation(uint32 id, uint16 nodeID, uint16 procID);
	//bool registerTagID(uint32 id, const char* name);


	bool addLibrary(const char* name, const char* library, const char* path);
	bool createComponent(const char* name, XMLNode &n);
	bool createSpace(const char* name, uint16 &id);
	bool createExternalSpace(const char* name, uint16 &id);
	bool waitForSpaceToBeReady(uint16 id, uint32 timeout);
	bool processSpaceShutdown(uint16 procID);
	bool restartInternalSpace(uint16 procID);		// T-space: respawn a crashed internally-spawned space
	bool processSpaceConnect(uint16 procID);
	bool processSpaceDisconnect(uint16 procID);

	bool addPsyProbeCustomView(uint32 compID, const char* name, const char* templateURL);

	bool processLocalRequest(DataMessage* msg, DataMessage** resultMsg);
	DataMessage* processQueryMsg(DataMessage* msg);
	bool migrateComponentData(uint32 compID, uint16 fromNode, uint16 toNode, uint16 toProc);

	bool processMessage(DataMessage* msg);
	bool deliverTriggerMessage(TriggerSpec* spec, uint64 ttl, DataMessage* msg, uint64 recvtime);
	//! T1.6: deliver one joined <triggergroup> activation carrying the member map (name->msg) to a component.
	bool deliverTriggerGroup(uint32 componentID, uint16 crankID, const char* groupName, std::map<std::string, DataMessage*>& members, DataMessage* primary, uint64 recvtime, TriggerSpec* templateSpec = NULL);

	bool registerNewActiveContext(const PsyContext& activeContext, const PsyContext& inactiveContext = NOCONTEXT);

	ComponentSpec* componentSpecFromXML(XMLNode &specNode);
	bool getTriggersFromXML(XMLNode &node, ComponentSpec* spec, PsyContext context,
		std::list<TriggerSpec*>& triggers,
		std::list<RetrieveSpec*>& retrieves,
		std::list<QuerySpec*>& queries,
		std::list<PostSpec*>& posts,
		std::list<SignalSpec*>& signals,
		std::list<uint16>& cranks);
	bool addTriggersToComponent(ComponentSpec* spec,
		std::list<TriggerSpec*>& triggers,
		std::list<RetrieveSpec*>& retrieves,
		std::list<QuerySpec*>& queries,
		std::list<PostSpec*>& posts,
		std::list<SignalSpec*>& signals,
		std::list<uint16>& cranks);

	bool testTypesContexts();
};

THREAD_RET THREAD_FUNCTION_CALL NodeMessageThread(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL NodeSignalThread(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL NodeRequestThread(THREAD_ARG arg);

}
#endif // _NODE_H_
