Work on fixing bugs and reformat

core/clients/transport.py

@@ -3,8 +3,8 @@ import base64
 import io
 import secrets
 import time
-from urllib.parse import quote_plus
 from typing import Any
+from urllib.parse import quote_plus
 
 import aiohttp
 import orjson
@@ -40,6 +40,10 @@ def _runtime_command_result_key(service: str, command_id: str) -> str:
     return f"gia:service:command-result:{_service_key(service)}:{command_id}"
 
 
+def _runtime_command_cancel_key(service: str, command_id: str) -> str:
+    return f"gia:service:command-cancel:{_service_key(service)}:{command_id}"
+
+
 def _gateway_base(service: str) -> str:
     key = f"{service.upper()}_HTTP_URL"
     default = f"http://{service}:8080"
@@ -88,7 +92,9 @@ def update_runtime_state(service: str, **updates):
     return state
 
 
-def enqueue_runtime_command(service: str, action: str, payload: dict | None = None) -> str:
+def enqueue_runtime_command(
+    service: str, action: str, payload: dict | None = None
+) -> str:
     service_key = _service_key(service)
     command_id = secrets.token_hex(12)
     command = {
@@ -118,7 +124,9 @@ def pop_runtime_command(service: str) -> dict[str, Any] | None:
     return command
 
 
-def set_runtime_command_result(service: str, command_id: str, result: dict | None = None):
+def set_runtime_command_result(
+    service: str, command_id: str, result: dict | None = None
+):
     service_key = _service_key(service)
     result_key = _runtime_command_result_key(service_key, command_id)
     payload = dict(result or {})
@@ -126,7 +134,36 @@ def set_runtime_command_result(service: str, command_id: str, result: dict | Non
     cache.set(result_key, payload, timeout=_RUNTIME_COMMAND_RESULT_TTL)
 
 
-async def wait_runtime_command_result(service: str, command_id: str, timeout: float = 20.0):
+def cancel_runtime_command(service: str, command_id: str):
+    """Mark a runtime command as cancelled and set a result so waiters are released."""
+    service_key = _service_key(service)
+    result_key = _runtime_command_result_key(service_key, command_id)
+    cancel_key = _runtime_command_cancel_key(service_key, command_id)
+    payload = {"ok": False, "error": "cancelled", "completed_at": int(time.time())}
+    cache.set(result_key, payload, timeout=_RUNTIME_COMMAND_RESULT_TTL)
+    cache.set(cancel_key, True, timeout=60)
+    return True
+
+
+def cancel_runtime_commands_for_recipient(service: str, recipient: str) -> list[str]:
+    """Cancel any queued runtime commands for the given recipient and return their ids."""
+    service_key = _service_key(service)
+    key = _runtime_commands_key(service_key)
+    queued = list(cache.get(key) or [])
+    cancelled = []
+    for cmd in list(queued):
+        payload = dict(cmd.get("payload") or {})
+        if str(payload.get("recipient") or "").strip() == str(recipient or "").strip():
+            cmd_id = str(cmd.get("id") or "").strip()
+            if cmd_id:
+                cancel_runtime_command(service_key, cmd_id)
+                cancelled.append(cmd_id)
+    return cancelled
+
+
+async def wait_runtime_command_result(
+    service: str, command_id: str, timeout: float = 20.0
+):
     service_key = _service_key(service)
     result_key = _runtime_command_result_key(service_key, command_id)
     deadline = time.monotonic() + max(0.1, float(timeout or 0.0))
@@ -149,7 +186,9 @@ def list_accounts(service: str):
     if service_key == "signal":
         import requests
 
-        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip("/")
+        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip(
+            "/"
+        )
         try:
             response = requests.get(f"{base}/v1/accounts", timeout=20)
             if not response.ok:
@@ -199,7 +238,9 @@ def unlink_account(service: str, account: str) -> bool:
     if service_key == "signal":
         import requests
 
-        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip("/")
+        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip(
+            "/"
+        )
         target = quote_plus(account_value)
         for path in (f"/v1/accounts/{target}", f"/v1/account/{target}"):
             try:
@@ -242,7 +283,9 @@ def unlink_account(service: str, account: str) -> bool:
             connected=bool(accounts),
             pair_status=("connected" if accounts else ""),
             pair_qr="",
-            warning=("" if accounts else "Account unlinked. Add account to link again."),
+            warning=(
+                "" if accounts else "Account unlinked. Add account to link again."
+            ),
             last_event="account_unlinked",
             last_error="",
         )
@@ -619,7 +662,9 @@ def get_link_qr(service: str, device_name: str):
     if service_key == "signal":
         import requests
 
-        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip("/")
+        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip(
+            "/"
+        )
         response = requests.get(
             f"{base}/v1/qrcodelink",
             params={"device_name": device},