Repaired build workflow

.dockerignore

@@ -1,8 +0,0 @@
-.git
-.gitignore
-__pycache__/
-*.pyc
-*.pyo
-*.pyd
-.venv/
-venv/

.gitea/workflows/build.yml

@@ -0,0 +1,182 @@
+
+name: Build and Publish RMManager Image
+
+on:
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      # ---------------------------------------------
+      # Checkout repository (no external actions)
+      # ---------------------------------------------
+      - name: Checkout repository
+        env:
+          GITEA_TOKEN: ${{ secrets.TEA_TOKEN }}
+        run: |
+          set -euo pipefail
+
+          BASE="${GITHUB_SERVER_URL%/}"
+          REPO="${GITHUB_REPOSITORY}"
+
+          git init .
+          git remote add origin "$BASE/$REPO.git"
+
+          if [ -n "${GITEA_TOKEN:-}" ]; then
+            AUTH="$(printf 'oauth2:%s' "$GITEA_TOKEN" | base64 -w0)"
+            git -c "http.extraHeader=Authorization: Basic $AUTH" \
+              fetch --depth 1 origin "$GITHUB_SHA"
+          else
+            git fetch --depth 1 origin "$GITHUB_SHA"
+          fi
+
+          git checkout --detach FETCH_HEAD
+          git config --global --add safe.directory "$PWD"
+
+      # ---------------------------------------------
+      # Determine image version and commit SHA
+      # ---------------------------------------------
+      - name: Determine version
+        id: version
+        run: |
+          set -euo pipefail
+
+          SHORT_SHA="$(git rev-parse --short=7 HEAD)"
+          echo "sha=$SHORT_SHA" >> "$GITHUB_OUTPUT"
+
+          if [ -f VERSION ]; then
+            VERSION="$(tr -d '[:space:]' < VERSION)"
+            echo "version=$VERSION" >> "$GITHUB_OUTPUT"
+          else
+            echo "version=" >> "$GITHUB_OUTPUT"
+          fi
+
+      # ---------------------------------------------
+      # Login to Gitea container registry
+      # ---------------------------------------------
+      - name: Docker login
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          REGISTRY_USER: ${{ secrets.REGISTRY_USER }}
+          REGISTRY_TOKEN: ${{ secrets.TEA_TOKEN }}
+        run: |
+          set -euo pipefail
+
+          echo "$REGISTRY_TOKEN" | docker login "$REGISTRY" \
+            -u "$REGISTRY_USER" \
+            --password-stdin
+
+      # ---------------------------------------------
+      # Build and export container image
+      # ---------------------------------------------
+      - name: Build RMManager image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          VERSION: ${{ steps.version.outputs.version }}
+          SHORT_SHA: ${{ steps.version.outputs.sha }}
+        run: |
+          set -euo pipefail
+
+          FULL_IMAGE="$REGISTRY/david/rmmanager"
+
+          docker build \
+            -f Containerfile \
+            -t "$FULL_IMAGE:latest" \
+            -t "$FULL_IMAGE:sha-$SHORT_SHA" \
+            --output "type=docker,dest=$RUNNER_TEMP/rmmanager-built.tar" \
+            .
+
+      # ---------------------------------------------
+      # Load exported image into Docker
+      # ---------------------------------------------
+      - name: Load RMManager image
+        run: |
+          set -euo pipefail
+
+          docker load \
+            --input "$RUNNER_TEMP/rmmanager-built.tar"
+
+      # ---------------------------------------------
+      # Add optional version tag
+      # ---------------------------------------------
+      - name: Tag RMManager image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          VERSION: ${{ steps.version.outputs.version }}
+        run: |
+          set -euo pipefail
+
+          FULL_IMAGE="$REGISTRY/david/rmmanager"
+
+          if [ -n "$VERSION" ]; then
+            docker tag \
+              "$FULL_IMAGE:latest" \
+              "$FULL_IMAGE:v$VERSION"
+          fi
+
+      # ---------------------------------------------
+      # Verify local image and tags
+      # ---------------------------------------------
+      - name: Verify RMManager image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          VERSION: ${{ steps.version.outputs.version }}
+          SHORT_SHA: ${{ steps.version.outputs.sha }}
+        run: |
+          set -euo pipefail
+
+          FULL_IMAGE="$REGISTRY/david/rmmanager"
+
+          docker version
+          docker image ls
+
+          docker image inspect "$FULL_IMAGE:latest"
+          docker image inspect "$FULL_IMAGE:sha-$SHORT_SHA"
+
+          if [ -n "$VERSION" ]; then
+            docker image inspect "$FULL_IMAGE:v$VERSION"
+          fi
+
+      # ---------------------------------------------
+      # Publish image to Gitea registry
+      # ---------------------------------------------
+      - name: Push RMManager image
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          VERSION: ${{ steps.version.outputs.version }}
+          SHORT_SHA: ${{ steps.version.outputs.sha }}
+        run: |
+          set -euo pipefail
+
+          FULL_IMAGE="$REGISTRY/david/rmmanager"
+
+          docker push "$FULL_IMAGE:latest"
+          docker push "$FULL_IMAGE:sha-$SHORT_SHA"
+
+          if [ -n "$VERSION" ]; then
+            docker push "$FULL_IMAGE:v$VERSION"
+          fi
+
+      # ---------------------------------------------
+      # Report successful build
+      # ---------------------------------------------
+      - name: Report build
+        env:
+          REGISTRY: ${{ vars.REGISTRY }}
+          VERSION: ${{ steps.version.outputs.version }}
+          SHORT_SHA: ${{ steps.version.outputs.sha }}
+        run: |
+          set -euo pipefail
+
+          echo "RMManager build and publish completed"
+          echo "Image: $REGISTRY/david/rmmanager"
+          echo "Latest tag: latest"
+          echo "Commit tag: sha-$SHORT_SHA"
+
+          if [ -n "$VERSION" ]; then
+            echo "Version tag: v$VERSION"
+          fi

.gitignore

@@ -24,3 +24,9 @@ savedata/
 
 # Logs
 *.log
+
+#Trash and unused files
+trash/
+
+#Docker files
+.dockerignore

core/apirequest/execute.py

@@ -6,6 +6,7 @@ import logging
 
 from .request import buildRequest
 
+
 #session = requests.Session()
 _thread_local = threading.local()
 
@@ -25,6 +26,7 @@ def getSession(reset=False):
 
 
 def executeRequest(action, outputs, resolved):
+    from core.execution.lifecyclemanager.lifecyclemanager import lifecycles
     session = getSession()
 
     session_log = os.environ.get("LOGS_SESSION", "false").lower() == "true"
@@ -36,8 +38,15 @@ def executeRequest(action, outputs, resolved):
         )
 
     for attempt in range(2):
-        request, outputs = buildRequest(action, outputs, resolved)
+        request = buildRequest(action, outputs, resolved)
         try:
+            
+            if action.name in lifecycles:
+                request_info = {"Started at": time.monotonic()}
+                lifecycles[action.name]["Request Information"] = request_info
+
+            snapshot = lifecycles.copy()
+
             response = session.request(
                 method=request["method"],
                 url=request["path"],
@@ -47,14 +56,23 @@ def executeRequest(action, outputs, resolved):
                 files=request.get("files"),
                 timeout=(10, 30)
             )
-            return response, outputs
+
+            logging.debug(snapshot)
+
+            return response
 
         except (requests.Timeout, requests.ConnectionError) as error:
             #print(f"{type(error).__name__} during {action.name}: {error}")
+
             logging.warning(f"{type(error).__name__} during {action.name}: {error}")
+            logging.debug("LIFECYCLE STATE BEFORE REQUEST")
+            logging.debug(snapshot)
+            logging.debug("LIFECYCLE STATE AFTER REQUEST")
+            logging.debug(lifecycles)
+
             session = getSession(reset=True)
 
             if attempt == 0:
                 logging.warning(f"Retrying {action.name}")
 
-    return None, outputs
+    return None

core/apirequest/request.py

@@ -46,4 +46,4 @@ def buildRequest(definition, outputs, resolved):
                 "image/jpeg",
             ),
         }
-    return request, outputs
+    return request

core/execution/action/action.py

@@ -2,13 +2,11 @@ import logging
 
 from core.apirequest import executeRequest
 from core.runtime.runtimecontract import buildRuntimeAttr, buildPreRuntimeAttr
-#from core.runtime.stopsignals import addStopSignal
 
-
-def buildAction (action, outputs, process):
+def buildAction (action, outputs):
     action_log = True
 
-    resolved, match, filter = buildPreRuntimeAttr(action, outputs, process)
+    resolved, match, filter = buildPreRuntimeAttr(action, outputs)
 
     if hasattr(action, "allowaction"):
         if not action.allowaction:
@@ -16,16 +14,16 @@ def buildAction (action, outputs, process):
 
     if match:
         if filter.get("execute", False):
-            response, outputs = executeRequest(filter["execute"], outputs, resolved)
+            response = executeRequest(filter["execute"], outputs, resolved)
 
         return None
     
-    response, outputs = executeRequest(action, outputs, resolved)
+    response = executeRequest(action, outputs, resolved)
 
     if response is None:
         return outputs, None
 
-    response = buildRuntimeAttr(action, outputs, resolved, response, process)
+    response = buildRuntimeAttr(action, outputs, resolved, response)
 
     if action_log:
         logging.info(f"{action.name} completed")

core/execution/createthread.py

@@ -1,4 +1,5 @@
 import threading
+import logging
 
 from .lifecyclemanager import manageLifecycle
 
@@ -12,4 +13,6 @@ def buildThread(module):
 
     thread.start()
 
+    logging.info(f"THREAD STARTED, {thread}")
+
     return thread

core/execution/executioncontract.py

@@ -3,18 +3,17 @@ from core.execution.operation import buildOperation
 from core.execution.monitor import buildMonitor
 #from core.runtime import manageFailure
 from core.execution.recursive import buildRecursive
+from core.runtime.createruntime import outputs
 
-def executeContract(module, outputs, process):
+def executeContract(module):
 
     if module.kind == "operation":
-
         responses = buildOperation(module, outputs)
         
         return responses
 
     elif module.kind == "action":
-
-        response = buildAction(module, outputs, process)
+        response = buildAction(module, outputs)
 
         return response
 

core/execution/lifecyclemanager/lifecyclemanager.py

@@ -6,9 +6,10 @@ from core.apirequest.execute import getSession
 from core.execution.lifecyclemanager.lifecycleregistry import registerLifecycle
 
 lifecycles = {}
-outputs = {}
+#outputs = {}
 
 def manageLifecycle(module):
+    from core.runtime.createruntime import outputs
     lifecycle_log = os.environ.get("LOGS_LIFECYCLE", "false").lower() == "true"
 
     registerLifecycle (module, lifecycles)
@@ -17,7 +18,7 @@ def manageLifecycle(module):
         while lifecycles[module.name]["mode"] == "continuous":
             lifecycles[module.name]["state"] = "active"
 
-            response = executeContract(module, outputs, None)
+            executeContract(module)
 
             if lifecycle_log:
                 print(f"{module.name} thread {lifecycles[module.name][module.name]} running in {lifecycles[module.name]['mode']} mode")
@@ -34,7 +35,7 @@ def manageLifecycle(module):
             print(f"{module.name} thread {lifecycles[module.name][module.name]} running in {lifecycles[module.name]['mode']} mode")
             print(lifecycles)
 
-        response = executeContract(module, outputs, None)
+        response = executeContract(module)
 
         lifecycles[module.name]["state"] = "stopped"
 

core/execution/lifecyclemanager/lifecycleregistry.py

@@ -23,17 +23,20 @@ def registerLifecycle(module, lifecycles):
     #print(lifecycles)
 
     if "mode" not in lifecycles[module.name]:
+        if hasattr(module, "mode"):
             mode = getattr(module, "mode")
             lifecycles[module.name]["mode"] = mode
+        else:
+            lifecycles[module.name]["mode"] = "oneshot"
     
     if lifecycles[module.name]["mode"] == "continuous":
         lifecycles[module.name]["refresh_rate"] = module.sleeptime
 
-    if not lifecycles[module.name]["mode"]:
-        lifecycles[module.name]["mode"] = "oneshot"
+    #if lifecycles[module.name]["mode"] == {}:
+    #    lifecycles[module.name]["mode"] = "oneshot"
 
     #lifecycles[module.name][registryname] = registryvalue
 
     
 
-    return lifecycles
+    return

core/execution/monitor.py

@@ -1,8 +1,5 @@
 import logging
 
-#from core.execution.executioncontract import executeContract
-from core.execution.action import buildAction
-from core.execution.operation import buildOperation
 from core.apirequest import executeRequest
 from core.runtime.resolver import resolveItem
 from core.runtime.applyconditions import applyConditions
@@ -15,11 +12,16 @@ resolved = {}
 def buildMonitor(monitor, outputs):
     loadSave(monitor, outputs)
 
-    response, outputs = executeRequest(monitor, outputs, resolved)
+    response = executeRequest(monitor, outputs, resolved)
 
+    ##Potentially replace response processing with responsePolicy
     if response is None:
         return None
 
+    if not response.ok:
+        logging.warning(f"Response with status code {response.status_code} during {monitor.name}")
+        return response
+
     responsedata = response.json()
     
     readitems = resolveItem(monitor, responsedata)
@@ -29,13 +31,14 @@ def buildMonitor(monitor, outputs):
     buildOutput(response, monitor, outputs)
 
     if match:
-        logging.info(filter.get("print"))
-        module = filter["execute"]
+        from core.execution.createthread import buildThread
+        #from core.execution.executioncontract import executeContract
 
-        if module.kind == "action":
-            response = buildAction(module, outputs, None)
-        if module.kind == "operation":
-            responses = buildOperation(module, outputs)
-            response = responses 
+        logging.info(filter.get("print"))
+        #module = filter["execute"]
+        
+        #response = executeContract(filter["execute"], outputs)
+
+        thread = buildThread(filter["execute"])
 
     return response 

core/execution/operation/operation.py

@@ -5,20 +5,21 @@ from rich.console import Console
 
 from core.runtime.managefailure import manageFailure
 from core.runtime.stopsignals import checkDependency
-from core.execution.processregistry import registerProcess
+from core.execution import processregistry
 from core.console import console
 
-outputs = {}
+#outputs = {}
 
 def buildOperation(definition, outputs):
+
     process_log = True
     
-    from core.execution.executioncontract import executeContract
-    
     actions = definition.actions
-    registerProcess(definition, outputs)
+
     responses = []
 
+    currentOp = processregistry.registerProcess(definition)
+    
     if getattr(definition, "track", False):
         actions = track(
             actions,
@@ -27,44 +28,46 @@ def buildOperation(definition, outputs):
         )
 
     for action in actions:
+        processregistry.registeredOp = currentOp
 
-        stop = checkDependency(action, outputs, definition)
+        stop = checkDependency(action)
 
         if stop:
             continue
 
         if action.kind == "action":
-            outputs["processes"][definition.name][action.name] = {}
+            currentOp[action.name] = {}
 
-        response = executeContract(action, outputs, definition)
+        from core.execution.executioncontract import executeContract
+
+        response = executeContract(action)
 
         responses.append(response)
 
         if action.kind == "action":
-            outputs["processes"][definition.name][action.name]["response"] = response
+            currentOp[action.name]["response"] = response 
 
-        ##Append completed process to process registry
-        outputs["processes"][definition.name]["completed"].append(action.alias)
+        if action.alias in currentOp["stopsignals"]:
+            continue
+        currentOp["completed"].append(action.alias)
 
         #Temporarily disabled manageFailure functions
         #except Exception as error:
             #outputs = manageFailure(error, action, outputs)
 
-    #Print completion status after runtime processes
     if process_log:
-        stopsignal_count = len(outputs["processes"][definition.name]["stopsignals"])
-        completed_count = len(outputs["processes"][definition.name]["completed"])
+        completed_count = len(currentOp["completed"])
 
-        if completed_count == 1 and stopsignal_count == 1:
+        if completed_count == 0:
             logging.info(f"{definition.name} bypassed")
         else:
             logging.info(f"{definition.name} completed")
 
-    #For debugging purposes only
+
     process_registry_logs = False
+
     if process_registry_logs:
-        pprint.pprint(outputs["processes"])
-
-
+        import pprint
+        pprint.pprint(processregistry.processes)
     
     return responses

core/execution/processregistry.py

@@ -1,12 +1,14 @@
-def registerProcess(module, outputs):
-    if not "processes" in outputs:
-        outputs["processes"] = {}
+processes = {}
+registeredOp = {}
 
-    if not module.name in outputs["processes"]:
-        outputs["processes"][module.name] = {}
+def registerProcess(module):
+    #global registeredOp
 
-    #outputs["processes"][module.name] = {}
-    outputs["processes"][module.name]["stopsignals"] = []
-    outputs["processes"][module.name]["completed"] = []
+    processes[module.name] = {}
 
-    return
+    registeredOp = processes[module.name]
+
+    processes[module.name]["stopsignals"] = []
+    processes[module.name]["completed"] = []
+
+    return registeredOp

core/execution/recursive.py

@@ -2,7 +2,7 @@ import logging
 import os
 
 from core.runtime.stopsignals import checkDependency
-from core.execution.processregistry import registerProcess
+from core.execution import processregistry
 from rich.progress import track
 from core.console import console
 
@@ -10,8 +10,6 @@ def buildRecursive(definition, outputs):
     process_log = True
     recursive_log = os.environ.get("LOGS_RECURSIVE", "false").lower() == "true"
 
-    #print(outputs[definition.recursiveset["source"]])
-
     source = outputs.get(definition.recursiveset["source"])
 
     if source is None or source == []:
@@ -19,10 +17,10 @@ def buildRecursive(definition, outputs):
             logging.debug(f"No items available in {source} for {definition.name}")
         return outputs, []
 
-    registerProcess(definition, outputs)
+    currentOp = processregistry.registerProcess(definition)
 
     responses = []
-    from core.execution.executioncontract import executeContract
+    
     recursiveset = definition.recursiveset["actions"]
 
     source = definition.recursiveset["source"]
@@ -32,7 +30,6 @@ def buildRecursive(definition, outputs):
     reverse = definition.recursiveset.get("reverse", False)
 
     if reverse:
-        #print("reversing")
         items = reversed(items)
 
     for item in track(
@@ -46,37 +43,30 @@ def buildRecursive(definition, outputs):
 
         outputs["stopsignals"] = []
         for job in recursiveset:
+            processregistry.registeredOp = currentOp
 
-            stop = checkDependency(job, outputs, definition)
-
-            #print(f"STOP SIGNAL DEBUGGING during {job.name}")
-            #print(outputs["stopsignals"])
-            #print(stop)
+            stop = checkDependency(job)
 
             if stop:
-                #outputs["processes"][definition.name]["stopsignals"].append(job.alias)
-                #print("STOP SIGNAL DETECTED")
-                #print(f"Stopping {job.name} due to stop signal")
                 continue
 
-            #print(f"RECURSIVE DEBUG DURING {job.name}")
-            #print(item)
-            #outputs, response = buildAction(job, outputs)
             if job.kind == "action":
-                outputs["processes"][definition.name][job.name] = {}
+                currentOp[job.name] = {}
 
-            response = executeContract(job, outputs, definition)
+            from core.execution.executioncontract import executeContract
+            
+            response = executeContract(job)
             responses.append(response)
-            if job.kind == "action":
-                outputs["processes"][definition.name][job.name]["response"] = response
 
-            outputs["processes"][definition.name]["completed"].append(job.alias)
+            if job.kind == "action":
+                currentOp[job.name]["response"] = response
+
+            currentOp["completed"].append(job.alias)
 
     if process_log:
-        stopsignal_count = len(outputs["processes"][definition.name]["stopsignals"])
-        completed_count = len(outputs["processes"][definition.name]["completed"])
+        completed_count = len(currentOp["completed"])
 
-        if completed_count == 1 and stopsignal_count == 1:
+        if completed_count == 0:
             logging.info(f"{definition.name} bypassed")
         else:
             logging.info(f"{definition.name} completed")
@@ -88,4 +78,10 @@ def buildRecursive(definition, outputs):
 
     outputs.pop("stopsignals", None)
 
+    process_registry_logs = False
+    
+    if process_registry_logs:
+        import pprint
+        pprint.pprint(processregistry.processes)
+
     return responses

core/runtime/applyconditions.py

@@ -49,8 +49,12 @@ def applyConditions(definition, dictionary, outputs):
                         return True, filter
 
                 if filter["operator"] == "in":
+                    if filter["filterkey"] == "called_dictionary":
                         if filtervalue in dictionary:
                             return True, filter
+                    else:
+                        if filtervalue in dictionary.get(filter["filterkey"], {}):
+                            return True, filter
 
                 if filter["operator"] == "is":
                     if resolvedvalue is filtervalue:

core/runtime/createruntime.py

@@ -0,0 +1,4 @@
+outputs = {}
+
+def getOuputs():
+    return outputs

core/runtime/firewall.py

@@ -2,33 +2,34 @@ import os
 import logging
 
 from core.runtime.applyconditions import applyConditions
+from core.runtime.resolver import resolveItem
+
 
 def buildFirewall(listener, outputs):
+    if hasattr(listener, "filters"):
         #firewall_log = os.environ.get("LOGS_FIREWALL", "false").lower() == "true"
         firewall_log = True
 
         status_code = 200
         outputs["firewall_allowed"] = []
         for item in outputs[listener.recursive["source"]]:
+            resolved = resolveItem(listener, item)
 
-        match, filter = applyConditions(listener, item, outputs)
+            match, filter = applyConditions(listener, resolved, outputs)
 
             if match:
                 status_code = filter.get("status_code", 200)
+
+                if filter.get("execute", False):
+                    from core.execution.createthread import buildThread
+
+                    thread = buildThread(filter["execute"])
+
                 if firewall_log:
                     logging.info(filter.get("print"))
                     logging.info(f"Returning status_code: {status_code}")
                 continue
 
-        #This needs to be moved into definitions but until logic is put in place it lives here.
-        if "com.loxthelion.rmass_message_id" in item.get("content", {}):
-            status_code = 200
-            if firewall_log:
-                logging.info("Excluding previously imported items")
-                logging.info(f"Returning status_code: {status_code}")
-
-            continue
-
             outputs["firewall_allowed"].append(item)
         
         if outputs["firewall_allowed"] != []:

core/runtime/managefailure.py

@@ -22,7 +22,7 @@ def manageFailure(error, action, outputs):
                 else:
                     print(f"{type(error).__name__} detected during {action.name}. Applying recovery policy.")
                     
-                    response = buildAction(repair, outputs, None)
+                    response = buildAction(repair, outputs)
 
                 return outputs
     print(f"{error}")

core/runtime/resolver.py

@@ -1,7 +1,9 @@
 import os
+import logging
 
 def resolveItem(action, item):
-    resolver_log = os.environ.get("LOGS_RESOLVER", "false").lower() == "true"
+    #resolver_log = os.environ.get("LOGS_RESOLVER", "false").lower() == "true"
+    resolver_log = True
 
     resolved = {}
     if hasattr(action, "resolve"):
@@ -9,11 +11,14 @@ def resolveItem(action, item):
             target = item
             try:
                 for key in keys:
+                    if isinstance(target, dict):
+                        target = target.get(key)
+                    else:
                         target = target[key]
 
             except (KeyError, IndexError) as error:
                 if resolver_log:
-                    print(f"Resolver target key/index {error} not found during {action.name}")
+                    logging.info(f"Resolver target key/index {error} not found during {action.name}")
                 target = None
 
             resolved[label] = target

core/runtime/responsepolicy.py

@@ -1,9 +1,15 @@
 import logging
 
 from core.apirequest import executeRequest
+from core.execution import processregistry
 
 def responsePolicy(response, action, outputs, resolved):
     if hasattr(action, "responsepolicy"):
+        #if processregistry.registeredOp:
+            #currentOp = processregistry.registeredOp
+            #print(registeredOp[action.name]["response"])
+            #print(currentOp)
+
         if not response.ok:
             logging.warning(f"{response} during {action.name}")
 
@@ -11,11 +17,19 @@ def responsePolicy(response, action, outputs, resolved):
                 if policy["statuscode"] == response.status_code:
                     logging.warning(policy["print"])
 
-                    newresponse, outputs = executeRequest(policy["execute"], outputs, resolved)
+                    from core.execution.executioncontract import executeContract
+                    executeContract(policy["execute"])
+
+                    #newresponse = executeRequest(action, outputs, resolved)
 
                     if policy["response_update"]:
                         logging.info("Updating response")
-                        print("Updating response")
+                        newresponse = executeRequest(action, outputs, resolved)
+
+                        #response = newresponse
+                    if processregistry.registeredOp:
+                        response = processregistry.registeredOp[action.name]["response"]
+                    else:
                         response = newresponse
 
                     break

core/runtime/runtimecontract.py

@@ -2,7 +2,7 @@ import logging
 
 from core.runtime.stopsignals import addStopSignal
 
-def buildPreRuntimeAttr(action, outputs, process):
+def buildPreRuntimeAttr(action, outputs):
     from core.runtime import loadSave, readRow
     from core.runtime.applyconditions import applyConditions
     from core.runtime.resolver import resolveItem
@@ -17,12 +17,12 @@ def buildPreRuntimeAttr(action, outputs, process):
 
     if match:
         logging.info(filter.get("print"))
-        addStopSignal(action, outputs, process)
+        addStopSignal(action)
     
     return resolved, match, filter
     
 
-def buildRuntimeAttr(action, outputs, resolved, response, process):
+def buildRuntimeAttr(action, outputs, resolved, response):
     from core.runtime.resolver import extractItem
     from core.runtime.responsepolicy import responsePolicy
     from core.runtime.image import buildImage
@@ -33,7 +33,7 @@ def buildRuntimeAttr(action, outputs, resolved, response, process):
     buildOutput(response, action, outputs)
 
     if not response.ok:
-        addStopSignal(action, outputs, process)
+        addStopSignal(action)
         return outputs, response
 
     extractItem(action, outputs, resolved)

core/runtime/stopsignals.py

@@ -1,9 +1,10 @@
-#def checkDependency(action, outputs):
-def checkDependency(action, outputs, operation):
+#from core.execution.processregistry import registeredOp
+
+def checkDependency(action):
+    from core.execution.processregistry import registeredOp
     if hasattr(action, "depends"):
         stop = any(
-                dependency in outputs["processes"][operation.name]["stopsignals"]
-                #dependency in outputs["stopsignals"]
+                dependency in registeredOp["stopsignals"]
                 for dependency in action.depends
             )
         
@@ -12,9 +13,10 @@ def checkDependency(action, outputs, operation):
 
     return False
 
-#def addStopSignal(action, outputs):
-def addStopSignal(action, outputs, operation):
+def addStopSignal(action):
+    from core.execution.processregistry import registeredOp
+
     if hasattr(action, "depends"):
-        outputs["processes"][operation.name]["stopsignals"].append(action.alias)
-        #outputs["stopsignals"].append(action.alias)
+
+        registeredOp["stopsignals"].append(action.alias)
     return

core/server/createlistener.py

@@ -2,43 +2,50 @@ import json
 import logging
 
 from core.runtime.outputs import buildOutput
-from core.execution.executioncontract import executeContract
 from core.runtime.firewall import buildFirewall
+from core.runtime.createruntime import outputs
+from core.execution.processregistry import processes
 
 
 def buildListener(listener, self):
-
-    if self.path == "/_matrix/app/v1/ping":
-        self.send_response(200)
-        self.send_header("Content-Type", "application/json")
-        self.end_headers()
-        self.wfile.write(b"{}")
-        return
-
     if self.path.startswith(listener.acceptpath):
-    
-        #Prints can used buildlog for listener runtime
-        logging.info("Matrix transaction received")
+        if hasattr(listener, "print"):
+            logging.info(listener.print)
 
         content_length = int(self.headers["Content-Length"])
 
-        data = self.rfile.read(content_length)
-        request = json.loads(data)
+        bytes = self.rfile.read(content_length)
 
-        outputs = {}
+        #request = json.loads(data)
 
-        buildOutput(data, listener, outputs)
+        buildOutput(bytes, listener, outputs)
 
+        if hasattr(listener, "filters"):
             allow, status_code = buildFirewall(listener, outputs)
 
             if allow:
-            #Print allowed items for logging purposes
                 logging.info(outputs["firewall_allowed"])
 
-            #Process and forward to RMASS
-            responses = executeContract(listener.execute, outputs, None)
+                if hasattr(listener, "execute"):
+                    if listener.execute["engine"] == "thread":
+                        from core.execution.createthread import buildThread
+                        buildThread(listener.execute["module"])
 
+                    elif listener.execute["engine"] == "contract":
+                        from core.execution.executioncontract import executeContract
+                        executeContract(listener.execute["module"])
 
+        else:
+            if hasattr(listener, "execute"):
+                if listener.execute["engine"] == "thread":
+                    from core.execution.createthread import buildThread
+                    buildThread(listener.execute["module"])
+
+                elif listener.execute["engine"] == "contract":
+                    from core.execution.executioncontract import executeContract
+                    executeContract(listener.execute["module"])
+
+                status_code = listener.status_code["success"]
 
         self.send_response(status_code)
         self.send_header("Content-Type", "application/json")
@@ -49,4 +56,4 @@ def buildListener(listener, self):
     else:
         self.send_error(404)
     
-    return outputs
+    return

core/server/server.py

@@ -4,106 +4,121 @@ import logging
 
 from http.server import HTTPServer, BaseHTTPRequestHandler
 
-from .dispatcher import dispatchRequest
-from .createresponse import buildResponse
-from plugins.rmass.listeners import synapselistener
+
+#from .dispatcher import dispatchRequest
+#from .createresponse import buildResponse
+from plugins.rmass.listeners import synapselistener, matrixpinglistener
 from core.server.createlistener import buildListener
-#from plugins.rmass.actions import pingsynapse
+#from plugins.rmass.actions import synapsesetbridgeonline
 
 port = int(os.environ.get("SERVER_PORT", 19001))
 
 
 class RequestHandler(BaseHTTPRequestHandler):
     def do_POST(self):
-        #lifecycles = {}
-        print("POST request received")
+        logging.debug("POST request received")
 
-        max_request_size = 1024
+        buildListener(matrixpinglistener, self)
 
-        content_length = self.headers.get("Content-Length")
+        # if self.path == "/_matrix/app/v1/ping":
 
-        # Check whether Content-Length exists
-        if content_length is None:
-            response = {
-                "status": "error",
-                "message": "Content length required"
-            }
-            status = 411
+        #     from core.execution.createthread import buildThread
+        #     thread = buildThread(synapsesetbridgeonline)
 
-        else:
-            # Attempt to convert the header into an integer
-            try:
-                length = int(content_length)
-            except ValueError:
-                length = None
+        #     self.send_response(200)
+        #     self.send_header("Content-Type", "application/json")
+        #     self.send_header("Content-Length", "2")
+        #     self.end_headers()
+        #     self.wfile.write(b"{}")
+        #     return
 
-            # Validate the converted length
-            if length is None:
-                response = {
-                    "status": "error",
-                    "message": "Invalid Content-Length"
-                }
-                status = 400
+        # else:
+        #     print("POST request received")
 
-            elif length < 0:
-                response = {
-                    "status": "error",
-                    "message": "Bad Request"
-                }
-                status = 400
+        #     max_request_size = 1024
 
-            elif length > max_request_size:
-                response = {
-                    "status": "error",
-                    "message": "Content exceeds maximum request size"
-                }
-                status = 413
+        #     content_length = self.headers.get("Content-Length")
 
-            else:
-                # Content-Length is valid, so read the body
-                data = self.rfile.read(length)
+        #     # Check whether Content-Length exists
+        #     if content_length is None:
+        #         response = {
+        #             "status": "error",
+        #             "message": "Content length required"
+        #         }
+        #         status = 411
 
-                try:
-                    request = json.loads(data)
+        #     else:
+        #         # Attempt to convert the header into an integer
+        #         try:
+        #             length = int(content_length)
+        #         except ValueError:
+        #             length = None
 
-                except json.JSONDecodeError:
-                    response = {
-                        "status": "error",
-                        "message": "Invalid JSON received"
-                    }
-                    status = 400
+        #         # Validate the converted length
+        #         if length is None:
+        #             response = {
+        #                 "status": "error",
+        #                 "message": "Invalid Content-Length"
+        #             }
+        #             status = 400
 
-                else:
-                    # Only valid JSON reaches the dispatcher
-                    thread, lifecycles, module, mode = dispatchRequest(
-                        request, lifecycles
-                    )
+        #         elif length < 0:
+        #             response = {
+        #                 "status": "error",
+        #                 "message": "Bad Request"
+        #             }
+        #             status = 400
 
-                    response = buildResponse(module, thread, mode)
-                    status = 200
+        #         elif length > max_request_size:
+        #             response = {
+        #                 "status": "error",
+        #                 "message": "Content exceeds maximum request size"
+        #             }
+        #             status = 413
 
-        # Serialize the response
-        responsebytes = json.dumps(response).encode("utf-8")
+        #         else:
+        #             # Content-Length is valid, so read the body
+        #             data = self.rfile.read(length)
 
-        # 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()
+        #             try:
+        #                 request = json.loads(data)
 
-        self.wfile.write(responsebytes)
+        #             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(responsebytes)
 
 
     def do_PUT(self):
-        #print("PUT request received")
+        logging.debug("PUT request received")
 
-        outputs = buildListener(synapselistener, self)
+        buildListener(synapselistener, self)
 
 def startServer():
     server = HTTPServer(("0.0.0.0", port), RequestHandler)
 
     logging.info(f"Server listening on port {port}")  
-    #outputs = {}
-    #outputs, response = buildAction(pingsynapse, outputs)    
 
     server.serve_forever()

initialize.py

@@ -5,7 +5,8 @@ import os
 from core.server.server import startServer
 from core.execution.createthread import buildThread
 from plugins.rmass.monitors import rmassmonitor
-from plugins.rmass.actions import refresh_tokens
+from plugins.rmass.operations import token_update
+from plugins.rmass.actions import pingsynapse
 
 from core.console import console
 from rich.logging import RichHandler
@@ -27,16 +28,21 @@ logging.basicConfig(
 import login
 import synapsemessagepush
 
-rmassmonitor.mode = "continuous"
-refresh_tokens.mode = "continuous"
-
-thread = buildThread(rmassmonitor)
-
-thread = buildThread(refresh_tokens)
-
 thread = threading.Thread(
         target=startServer,
-        #args=(module, lifecycles)
         name="RMASS Bridge Server"
     )
 thread.start()
+
+pingsynapse.mode = "continuous"
+rmassmonitor.mode = "continuous"
+token_update.mode = "continuous"
+#refresh_tokens.mode = "continuous"
+
+thread = buildThread(rmassmonitor)
+
+#thread = buildThread(refresh_tokens)
+
+thread = buildThread(token_update)
+
+thread = buildThread(pingsynapse)

plugins/rmass/actions/pingsynapse.py

@@ -1,11 +1,13 @@
-name = "Synapse Wake"
+name = "Synapse Ping"
 
 kind = "action"
 
-alias = "synapseping"
+alias = "pingsynapse"
 
 showstatus = True
 
+sleeptime = 15
+
 isrequest = True
 
 method = "POST"

plugins/rmass/actions/refresh_tokens.py

@@ -5,6 +5,8 @@ name = "Token refresh"
 
 kind = "action"
 
+alias = "refresh_tokens"
+
 showstatus = True
 
 method = "GET"
@@ -44,6 +46,15 @@ outputs = {
     "refresh_token": "refreshToken",
 }
 
+responsepolicy = [
+        {
+            "statuscode": 403,
+            "print": "Refreshing stale tokens...",
+            "execute": login,
+            "response_update": False,
+        }
+]
+
 recoverPolicy = {
     FileNotFoundError: [
         getcsrf,

plugins/rmass/actions/rmassmessagepush.py

@@ -90,7 +90,7 @@ outputs = {
 filters = [
     {
         "filterkey": "message_body",
-        #"filterkey_lib": "outputs",
+        #"filterkey_lib": "resolved",
         "filterkey_format": "length",
         "execute": synapsenotify,
         "operator": "<",

plugins/rmass/actions/synapsenotify.py

@@ -10,15 +10,8 @@ showstatus = True
 
 resolve = {
     "get": {
-        "sender": ["room_name"],
-        "event_id": ["outmessageitem", "event_id"],
-        "room_id": ["outmessageitem", "room_id"],
-    },
-    "deconstruct": {
-        "sender" : [
-            "#rmass_dm_",
-            ":loxthelion.com",
-        ],
+        "event_id": ["events", 0, "event_id"],
+        "room_id": ["events", 0, "room_id"],
     },
 }
 

plugins/rmass/actions/synapsesetbridgeonline.py

@@ -0,0 +1,36 @@
+name = "Set RMASS Bridge Online"
+
+kind = "action"
+
+alias = "setrmassbridgeonline"
+
+showstatus = True
+
+#sleeptime = 15
+
+isrequest = True
+
+method = "PUT"
+
+path = [
+    "https://loxthelion.com/_matrix/client/v3/presence/@rmass_bridge:loxthelion.com/status",
+]
+
+headers = {
+    "Authorization": {
+        "os": "AS_TOKEN",
+        "prefix": "Bearer ",
+    },
+    "Content-Type": "application/json",
+}
+
+keys = [
+    {
+        "keytitle": "presence",
+        "keyvalue": "online",
+    },
+    {
+        "keytitle": "status_msg",
+        "keyvalue": "RMASS Bridge Connected",
+    },
+]

plugins/rmass/listeners/matrixpinglistener.py

@@ -0,0 +1,16 @@
+from plugins.rmass.actions import synapsesetbridgeonline
+
+name = "Matrix Ping Listener"
+
+kind = "listener"
+
+acceptpath = "/_matrix/app/v1/ping"
+
+execute = {
+    "module": synapsesetbridgeonline,
+    "engine": "thread",
+}
+
+status_code = {
+    "success": 200
+}

plugins/rmass/listeners/synapselistener.py

@@ -1,4 +1,5 @@
 from plugins.rmass.operations import rmassmessagepush
+from plugins.rmass.actions import synapsenotify
 
 name = "Synapse Listener"
 
@@ -8,6 +9,19 @@ outputs = {
     "events": "events",
 }
 
+resolve = {
+    "get": {
+        "sender": ["sender"],
+        "event_id": ["event_id"],
+        "message_body": ["content", "body"],
+        "content": ["content"],
+        "type": ["type"],
+        "room_id": ["room_id"],
+        "timestamp": ["origin_server_ts"],
+        "filter_item": []
+    }
+}
+
 recursive = {
     "source": "events",
     "reverse": False,
@@ -16,11 +30,14 @@ recursive = {
 
 acceptpath = "/_matrix/app/v1/transactions/"
 
-execute = rmassmessagepush
+execute = {
+    "module": rmassmessagepush,
+    "engine": "contract",
+}
 
 filters = [
     {
-        "filterkey": "event",
+        "filterkey": "filter_item",
         "operator": "in",
         "filtervalue": "state_key",
         "print": "Excluding state.key event",
@@ -40,6 +57,24 @@ filters = [
         "print": "Excluding non-room message event",
         "status_code": 200,
     },
+    {
+        "filterkey": "content",
+        "operator": "in",
+        "filtervalue": "com.loxthelion.rmass_message_id",
+        "print": "Excluding imported message loop item",
+        "status_code": 200,
+    },
+    {
+        "filterkey": "message_body",
+        #"filterkey_lib": "resolved",
+        "filterkey_format": "length",
+        "execute": synapsenotify,
+        "operator": "<",
+        "filtervalue": 5,
+        "print": f"Message body is less than 5 characters. Notifying with {synapsenotify.name}...",
+        "status_code":200
+    },
 
 ]
 
+print = "Matrix transaction received"

plugins/rmass/operations/login.py

@@ -1,6 +1,6 @@
 from plugins.rmass.actions import getcsrf, login
 
-name = "Login"
+name = "Login Operation"
 
 kind = "operation"
 

plugins/rmass/operations/token_update.py

@@ -0,0 +1,13 @@
+import os
+
+from plugins.rmass.actions import refresh_tokens
+
+name = "Token Refresh Operation"
+
+kind = "operation"
+
+actions = [
+    refresh_tokens,
+]
+
+sleeptime = os.environ.get("TOKEN_REFRESH_RATE", 105)

trash/checkavailable.py

@@ -1,5 +0,0 @@
-from plugins import rmass
-from core.operation.operation import buildOperation
-
-outputs = buildOperation(rmass.checkavailableoperation)
-

trash/createdispatch.py

@@ -1,17 +0,0 @@
-import importlib
-from lifecyclemanager import manageLifecycle
-from dispatchstate import buildState
-
-def buildDispatch(request):
-    dispatch = []
-
-    for key in request["module"]:
-        dispatch.append(request["module"][key])
-
-    resolveddispatch = "plugins."+ ".".join(dispatch)
-
-    module = importlib.import_module(resolveddispatch)
-
-    mode = request.get("mode", "oneshot")
-
-    return module, mode

trash/createthread.py

@@ -1,30 +0,0 @@
-import threading
-from lifecycleregistry import registerLifecycle
-from lifecyclemanager import manageLifecycle
-
-threadlist = []
-threadnum = 0
-
-def buildThread(module, lifecycles):
-    global threadnum
-    threadnum += 1
-
-    threadname = f"thread.{threadnum}"
-
-    thread = threading.Thread(
-        target=manageLifecycle,
-        args=(module, lifecycles)
-    )
-    thread.start()
-
-    registryname = threadname
-    registryvalue = thread
-
-    lifecycles = registerLifecycle (module, lifecycles, registryname, registryvalue)
-
-    threadlist.append(thread)
-
-    print(threadlist)
-    print(lifecycles)
-
-    return threadnum, threadname, lifecycles, threadlist

trash/dispatcher.py

@@ -1,22 +0,0 @@
-from createdispatch import buildDispatch
-from createthread import buildThread
-from dispatchstate import buildState
-
-import threading
-
-def dispatchRequest(request, lifecycles):
-
-    module, mode = buildDispatch(request)
-
-    mode, lifecycles = buildState(module, mode, lifecycles)
-
-    if mode == "stop":
-        return lifecycles
-
-    else:
-        threadnum, threadname, lifecycles, threadlist = buildThread(module, lifecycles)
-
-    #print("Active Threads")
-    #print(threadlist)
-
-        return threadnum, threadname, lifecycles, threadlist

trash/dispatchstate.py

@@ -1,41 +0,0 @@
-import requests
-
-from lifecycleregistry import registerLifecycle
-
-def filterMode(mode):
-    if mode in ("continuous", "oneshot", "stop"):
-        state = mode
-        return state
-    else: raise ValueError(f"Unsupported lifecycle mode: {mode}")
-
-def buildState(module, mode, lifecycles):
-    state = filterMode(mode)
-
-    registryvalue = state
-    registryname = "mode"
-
-    if state != "stop":
-        lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
-
-    elif state == "stop":
-        if not module.name in lifecycles:
-
-            print(f"No registered lifecycle: {module.name}")
-
-            return None
-
-        elif "mode" in lifecycles[module.name] != "stop":
-            lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
-
-            print(f"Stopping lifecycle: {module.name}")
-
-    #elif mode == "oneshot":
-        #registryvalue = {"mode": "oneshot"}
-        #lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
-
-           # else:
-                #raise ValueError(f"Unsupported lifecycle mode: {mode}")
-
-    #state = registryvalue
-
-    return state, lifecycles

trash/lifecyclemanager.py

@@ -1,25 +0,0 @@
-import time
-
-from core.execution.operation import buildOperation
-from core.execution.action import buildAction
-from core import executeContract
-
-def manageLifecycle(module, lifecycles): #lifecycles, state):
-    #print(lifecycles[module.name]["mode"])
-
-    if  lifecycles[module.name]["mode"] == "continuous":
-        while lifecycles[module.name]["mode"] == "continuous":
-            print(lifecycles)
-
-            outputs = executeContract(module, None, None)
-
-            #print(lifecycles)
-
-            time.sleep(module.sleeptime)
-
-    elif lifecycles[module.name]["mode"] == "oneshot":
-        print(lifecycles)
-        outputs = executeContract(module, None, None)
-
-        #print(lifecycles)
-    return

trash/lifecycleregistry.py

@@ -1,8 +0,0 @@
-def registerLifecycle(module, lifecycles, registryname, registryvalue):
-    if module.name not in lifecycles:
-        lifecycles[module.name] = {}
-
-
-    lifecycles[module.name][registryname] = registryvalue
-
-    return lifecycles

trash/manage.py

@@ -1,4 +0,0 @@
-from plugins import rmass
-from operation import buildOperation
-
-outputs = buildOperation(rmass.checkavailableoperation)

trash/manager.py

@@ -1,4 +0,0 @@
-from core.manager.manager import buildManager
-from plugins import rmass
-
-outputs = buildManager(rmass.manager)

trash/server.py

@@ -1,29 +0,0 @@
-import socket
-import json
-from dispatcher import dispatchRequest
-#from createthread import buildThread
-
-server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-
-server.bind(("127.0.0.1", 19002))
-
-lifecycles = {}
-
-server.listen()
-
-while True:
-    connection, address = server.accept()
-
-    while True:
-        data = connection.recv(4096)
-
-        request = json.loads(data)
-
-#        if not data:
-#            break
-
-        dispatch = dispatchRequest(request, lifecycles)
-
-    connection.close()
-
-#    server.close()