Updated select module and refined message retrieval process in RMASS definitions

core/runtime/outputs.py

@@ -1,10 +1,15 @@
+import json
+
 def buildOutput(response, action, outputs):
     #if select_data != "not applied":
         #response_data = select_data
     #else:
         #response_data = response.json()
     if hasattr(action, "outputs"):
+        try:
             response_data = response.json()
+        except AttributeError:
+            response_data = json.loads(response)
         response_objects = [response_data]
 
         while response_objects:

core/runtime/recursive.py

@@ -5,12 +5,14 @@ def recursiveSet(definition, outputs):
     for item in outputs[definition.recursiveset["source"]]:
         outputs[definition.recursiveset["outputlabel"]] = item
         #outputs["recursive"] = item
+        print("DEBUGGING recursiveset item")
+        print(outputs[definition.recursiveset["outputlabel"]])
 
         for job in recursiveset:
             outputs, response = buildAction(job, outputs)
 
-        print(outputs["profile"])
+        #print(outputs["profile"])
-        print(outputs["messages"])
+        #print(outputs["messages"])
     return outputs
 
 from core.execution.action.action import buildAction

core/runtime/savedata.py

@@ -1,4 +1,5 @@
 import json
+import sqlite3
 
 def buildSave(action, outputs):
     if hasattr(action, "savedata"):
@@ -11,6 +12,27 @@ def buildSave(action, outputs):
                 json.dump(save_data, save_file)
 
         print(f"Data saved by {action.name} to {action.savepath}")
+    
+        database = sqlite3.connect("savedata/rmmanager.db")
+        for key, value in save_data.items():
+            database.execute("""
+                CREATE TABLE IF NOT EXISTS output_test (
+                    dictionary TEXT NOT NULL UNIQUE,
+                    value TEXT NOT NULL UNIQUE
+                )
+            """)
+        
+            database.execute("""
+                INSERT INTO output_test(dictionary, value)
+                VALUES (?, ?)
+                ON CONFLICT (dictionary)
+                DO UPDATE SET value = excluded.value
+            """, (key, value)
+            )
+
+        database.commit()
+        database.close()
+    
     return outputs
 
 def loadSave(action, outputs):
@@ -23,3 +45,21 @@ def loadSave(action, outputs):
 
         print(f"Data loaded by {action.name} from {action.savepath}")
     return outputs
+
+def buildSave2(action, outputs):
+    
+    database.execute("""
+        CREATE TABLE IF NOT EXISTS message_mapping (
+            rmass_message_id INTEGER PRIMARY KEY,
+            matrix_event_id TEXT NOT NULL UNIQUE
+        )
+    """)
+    database = sqlite3.connect("rmmanager.db")
+    
+    database.execute(
+    "INSERT INTO message_mapping VALUES (?, ?)",
+    (action.savedata[savename], outputs["event_id"])
+    )
+
+    database.commit()
+    database.close()

core/runtime/select.py

@@ -24,11 +24,15 @@ def buildSelect(action, outputs):
                 if include_match and exclude_match:
                     select_data.append(item)
 
-            outputs[action.select["source"]] = select_data
+            #outputs[{action.select["source"]}] = select_data
-            #print(select_data)            
+            outputs[action.alias] = select_data
+
+            print("DEBUGGING", action.alias)
+            print(outputs[action.alias])            
 
             return select_data, action, outputs
-        except: (KeyError, TypeError)
+        except (KeyError, TypeError):
+            print("Error in selection")
             #select_data = ["no selection"]
     return select_data, action, outputs
     

core/server/server.py

@@ -1,72 +1,123 @@
-import socket
-import json
 import os
-import logging
+import json
 from http.server import HTTPServer, BaseHTTPRequestHandler
+
 from .dispatcher import dispatchRequest
 from .createresponse import buildResponse
+from core.runtime.outputs import buildOutput
+from plugins.rmass.inbound import receive
 
 port = int(os.environ.get("SERVER_PORT", 19001))
 
-from http.server import HTTPServer, BaseHTTPRequestHandler
 
 class RequestHandler(BaseHTTPRequestHandler):
     def do_POST(self):
+        lifecycles = {}
         print("POST request received")
 
-        self.send_response(200)
+        max_request_size = 1024
+
+        content_length = self.headers.get("Content-Length")
+
+        # Check whether Content-Length exists
+        if content_length is None:
+            response = {
+                "status": "error",
+                "message": "Content length required"
+            }
+            status = 411
+
+        else:
+            # Attempt to convert the header into an integer
+            try:
+                length = int(content_length)
+            except ValueError:
+                length = None
+
+            # Validate the converted length
+            if length is None:
+                response = {
+                    "status": "error",
+                    "message": "Invalid Content-Length"
+                }
+                status = 400
+
+            elif length < 0:
+                response = {
+                    "status": "error",
+                    "message": "Bad Request"
+                }
+                status = 400
+
+            elif length > max_request_size:
+                response = {
+                    "status": "error",
+                    "message": "Content exceeds maximum request size"
+                }
+                status = 413
+
+            else:
+                # Content-Length is valid, so read the body
+                data = self.rfile.read(length)
+
+                try:
+                    request = json.loads(data)
+
+                except json.JSONDecodeError:
+                    response = {
+                        "status": "error",
+                        "message": "Invalid JSON received"
+                    }
+                    status = 400
+
+                else:
+                    # Only valid JSON reaches the dispatcher
+                    thread, lifecycles, module, mode = dispatchRequest(
+                        request, lifecycles
+                    )
+
+                    response = buildResponse(module, thread, mode)
+                    status = 200
+
+        # Serialize the response
+        responsebytes = json.dumps(response).encode("utf-8")
+
+        # Send one HTTP response
+        self.send_response(status)
         self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(responsebytes)))
         self.end_headers()
 
-        self.wfile.write(b'{"status": "ok"}')
+        self.wfile.write(responsebytes)
+
+
+    def do_PUT(self):
+        print("PUT request received")
+
+        if self.path.startswith("/_matrix/app/v1/transactions/"):
+            print("Matrix transaction received")
+            outputs = {}
+            content_length = int(self.headers["Content-Length"])
+            data = self.rfile.read(content_length)
+            request = json.loads(data)
+
+            requestjson = json.dumps(request, indent=4)
+            print (request)
+            print(requestjson)
+            outputs = buildOutput(data, receive, outputs)
+            print (outputs)
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", "2")
+            self.end_headers()
+            self.wfile.write(b"{}")
+
+        else:
+            self.send_error(404)
 
 def startServer():
-    server = HTTPServer(("127.0.0.1", port), RequestHandler)
+    server = HTTPServer(("0.0.0.0", port), RequestHandler)
 
     print(f"Server listening on port {port}")
 
     server.serve_forever()
-
-
-
-def startServer2():
-    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-
-    server.bind(("0.0.0.0", port))
-
-    lifecycles = {}
-    
-    server.listen()
-    print(f"Server listening on port {port}")
-    while True:
-        connection, address = server.accept()
-        print("Accepting connections")
-        connection.settimeout(30)
-
-        try:
-            while True:
-                data = connection.recv(4096)
-                
-                if not data:
-                    print("Client Disconnected")
-                    break
-
-                request = json.loads(data)
-
-                thread, lifecycles, module, mode = dispatchRequest(request, lifecycles)
-
-                response = buildResponse(module, thread, mode)
-
-                connection.sendall(
-                    json.dumps(response).encode("utf-8"),
-                )
-        except json.decoder.JSONDecodeError as error:
-            print("Invalid JSON received", error)
-            print("Closing client connection")
-        except socket.timeout:
-            print("Client connection timed out")
-
-        finally:
-            connection.close()
-
-    #    server.close()

plugins/rmass/__init__.py

@@ -1 +1 @@
-from . import actions, operations
+from . import actions, operations, inbound

plugins/rmass/actions/getmessage.py

@@ -2,6 +2,8 @@ name = "Get messages"
 
 kind = "action"
 
+alias = "getmessage"
+
 showstatus = True
 
 method = "GET"

plugins/rmass/actions/pullmessagelist.py

@@ -2,6 +2,8 @@ name = "Get message list"
 
 kind = "action"
 
+alias = "pullmessages"
+
 showstatus = True
 
 method = "GET"

plugins/rmass/actions/synapseassignusername.py

@@ -6,12 +6,14 @@ kind = "action"
 
 showstatus = True
 
-astoken = os.environ["AS_TOKEN"]
+#astoken = os.environ["AS_TOKEN"]
 
 resolve = {
     "get": {
-        "sender": ["messages", 0, "senderId"],
+        "sender": ["messageheaders", "userCard", "id"],
-        "username": ["profile", "username"],
+        #"sender": ["pullmessages", 0, "userCard", "id"],
+        "username": ["messageheaders", "userCard", "username"],
+        #"username": ["pullmessages", 0, "userCard", "username"],
     },
     "construct": {
         "matrixuser": [

plugins/rmass/actions/synapsecreatedmroom.py

@@ -8,8 +8,10 @@ showstatus = True
 
 resolve = {
     "get": {
-        "sender": ["messages", 0, "senderId"],
+        "sender": ["messageheaders", "userCard", "id"],
-        "username": ["profile", "username"],
+        #"sender": ["pullmessages", 0, "userCard", "id"],
+        #"username": ["messageheaders", "username"],
+        "username": ["messageheaders", "userCard", "username"],
     },
     "construct": {
         "matrixuser": [

plugins/rmass/actions/synapsegetroomid.py

@@ -8,7 +8,10 @@ showstatus = True
 
 resolve = {
     "get": {
-        "sender": ["messages", 0, "senderId"],
+        "sender": ["messageheaders", "userCard", "id"],
+        #"sender": ["pullmessages", 0, "userCard", "id"],
+        #"username": ["messageheaders", "username"],
+        #"username": ["messageheaders", "userCard", "username"],
     },
     "construct": {
         "matrixdm": [

plugins/rmass/actions/synapsematchmessages.py

@@ -0,0 +1,90 @@
+import os
+
+name = "Synapse Message Push"
+
+kind = "action"
+
+showstatus = True
+
+recursive = {
+    "source": "messages",
+    "reverse": True,
+    "label": "messageitem",
+}
+
+resolve = {
+    "get": {
+        "sender": ["messageitem", "senderId"],
+        "message_id": ["messageitem", "id"],
+        "message_body": ["messageitem", "text"],
+        "room_id": ["dmroom_id"],
+        "timestamp": ["messageitem", "createdAt"],
+    },
+    "construct": {
+        "matrixuser": [
+            "@rmass_",
+            {"resolved": "sender"},
+            ":loxthelion.com",
+        ],
+    },
+    "transform": {
+        "matrixuser": {
+            "@rmass_238302:loxthelion.com": "@lyncoln:loxthelion.com",
+        },
+    },
+}
+
+keys = [
+    {
+        "keytitle": "msgtype",
+        "keyvalue": "m.text",
+    },
+    {
+        "keytitle": "body",
+        "keyvalue": {
+            "resolved": "message_body",
+        },
+    },
+    {
+        "keytitle": "format",
+        "keyvalue": "org.matrix.custom.html",
+    },
+    {
+        "keytitle": "formatted_body",
+        "keyvalue": {
+            "resolved": "message_body",
+        },
+    },
+    {
+        "keytitle": "com.loxthelion.rmass_message_id",
+        "keyvalue": {
+            "resolved": "message_id",
+        },
+    },
+]
+
+method = "PUT"
+
+path = [
+    "https://loxthelion.com/_matrix/client/v3/rooms/",
+    {"resolved": "room_id"},
+    "/send/m.room.message/",
+    {"resolved": "message_id"},
+    "?user_id=",
+    {"resolved": "matrixuser"},
+    "&ts=",
+    {"resolved": "timestamp"},
+    "000"
+]
+
+headers = {
+    "Authorization": {
+        "os": "AS_TOKEN",
+        "prefix": "Bearer ",
+    },
+    "Content-Type": "application/json",
+}
+
+#outputs = {
+#    "room_id": "room_id",
+#}

plugins/rmass/actions/synapsepushmessages.py

@@ -7,7 +7,7 @@ kind = "action"
 showstatus = True
 
 recursive = {
-    "source": "messages",
+    "source": "getmessage",
     "reverse": True,
     "label": "messageitem",
 }
@@ -55,7 +55,12 @@ keys = [
             "resolved": "message_body",
         },
     },
-        
+    {
+        "keytitle": "com.loxthelion.rmass_message_id",
+        "keyvalue": {
+            "resolved": "message_id",
+        },
+    },
 ]
 
 method = "PUT"
@@ -80,6 +85,4 @@ headers = {
     "Content-Type": "application/json",
 }
 
-#outputs = {
+
-#    "room_id": "room_id",
-#}

plugins/rmass/actions/synapseregister.py

@@ -8,8 +8,10 @@ showstatus = True
 
 resolve = {
     "get": {
-        "sender": ["messages", 0, "senderId"],
+        "sender": ["messageheaders", "userCard", "id"],
-        "username": ["profile", "username"],
+        #"sender": ["pullmessages", 0, "userCard", "id"],
+        "username": ["messageheaders", "userCard", "username"],
+        #"username": ["pullmessages", 0, "userCard", "username"],
     }
 }
 keys = [

plugins/rmass/inbound/__init__.py

@@ -0,0 +1 @@
+from . import receive

plugins/rmass/inbound/receive.py

@@ -0,0 +1,12 @@
+name = "Reception Test"
+
+kind = "action"
+
+showstatus = False
+
+
+
+outputs = {
+    "events": "events",
+}
+

plugins/rmass/operations/synapsemessagepush.py

@@ -12,8 +12,8 @@ actions = [
 ]
 
 recursiveset = {
-    "source": "emails",
+    "source": "pullmessages",
     "outputlabel": "messageheaders",
-    "actions": [getmessage, synapseregister, synapseassignusername, synapsecreatespace, synapsegetspaceid, synapsecreatedmroom, synapsegetroomid, synapseassignspace, synapsepushmessages],
+    "actions": [synapseregister, synapseassignusername, getmessage, synapsecreatespace, synapsegetspaceid, synapsecreatedmroom, synapsegetroomid, synapseassignspace, synapsepushmessages],
 }        
 

tokens.json

@@ -1 +1 @@
-{"access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJleHAiOjE3OTIwODQ2MDksIm5iZiI6MTc5MjA4NDQyOSwic3ViIjoiMjM4MzAyIiwidXNlcm5hbWUiOiJYeHhvdGlxIiwidXNlclR5cGUiOjEsInN0YWdlTmFtZSI6Ilh4eG90aXEiLCJhcHByb3ZlZCI6MSwibG9jYXRpb25JZCI6NDcyLCJzZWFyY2hDaXR5IjoiYXRsYW50YSIsImNvdW50cnlTaG9ydCI6IlVTIiwic2Vzc2lvbkNvdW50cnlDb250aW5lbnQiOiJOQSIsImlzRnJvemVuIjowLCJsb2NhdGlvbkxhYmVsIjoiQXRsYW50YSwgR0EiLCJpc3MiOiJodHRwczovL3JlbnRtYXNzZXVyLmNvbSIsImlhdCI6MTc5MjA4NDQ4OX0.CcXQrFvFl5LraO3VWoe3CZ4u-jAHVuMt6Fag2_v5QJ79ekRsPMnOFSMjLycaZoc0hTZRiJnpUIqBWKmTsg2juA", "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJleHAiOjE3OTQ2NzY0ODksIm5iZiI6MTc5MjA4NDQyOSwianRpIjoiQjIrSmxQWVdvYjYvOVBYTjZ3VnFrZzhmZHFrY05nQVZjbHF5YVluclJUdz0iLCJpc3MiOiJodHRwczovL3JlbnRtYXNzZXVyLmNvbSIsImlhdCI6MTc5MjA4NDQ4OSwic3ViIjoiMjM4MzAyIn0.RSOQP-lQv7N7Ut_YOkwF58-k-kGoFYCsGqYISJiUxajJ2YuDWz6vMQoWCN82NYUPJvfFDnGgiiugAnrhWS-OoQ"}
+{"access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJleHAiOjE3OTIxNjkwNTUsIm5iZiI6MTc5MjE2ODg3NSwic3ViIjoiMjM4MzAyIiwidXNlcm5hbWUiOiJYeHhvdGlxIiwidXNlclR5cGUiOjEsInN0YWdlTmFtZSI6Ilh4eG90aXEiLCJhcHByb3ZlZCI6MSwibG9jYXRpb25JZCI6NDcyLCJzZWFyY2hDaXR5IjoiYXRsYW50YSIsImNvdW50cnlTaG9ydCI6IlVTIiwic2Vzc2lvbkNvdW50cnlDb250aW5lbnQiOiJOQSIsImlzRnJvemVuIjowLCJsb2NhdGlvbkxhYmVsIjoiQXRsYW50YSwgR0EiLCJpc3MiOiJodHRwczovL3JlbnRtYXNzZXVyLmNvbSIsImlhdCI6MTc5MjE2ODkzNX0.Z3RFCsh82FEka9eUcg-vzo9ocOI2Xoay1SyYtYM4zFb3ub3roYfIZ6pGLazfCm0j_o3aKP9VmxLkbIthRTOgSg", "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJleHAiOjE3OTQ3NjA5MzUsIm5iZiI6MTc5MjE2ODg3NSwianRpIjoieVBJZ2ljRGlvUE5VQzVSVHdOZVIyTFNmZnFPY0hNQlNjRmVLd25tTFpFOD0iLCJpc3MiOiJodHRwczovL3JlbnRtYXNzZXVyLmNvbSIsImlhdCI6MTc5MjE2ODkzNSwic3ViIjoiMjM4MzAyIn0.DcwTYGNTX-NBNynrPGDMZqoxyZhOhCaTpHSX_1spNIIOdsDaBpc0nHjOc5m8iZB0C8obM1lQD5d5FZbCyuGzjA"}