initial commit

cli.py

@@ -0,0 +1,61 @@
+"""Path Resolver — utility helpers for segment operations."""
+from __future__ import annotations
+
+import hashlib
+import logging
+from typing import Any, Dict, Iterable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def normalise_segment(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Segment normalise — normalises and validates *data*."""
+    result = {k: v for k, v in data.items() if v is not None}
+    if "resolved" not in result:
+        raise ValueError(f"Segment must include 'resolved'")
+    result["id"] = result.get("id") or hashlib.md5(
+        str(result["resolved"]).encode()).hexdigest()[:12]
+    return result
+
+
+def validate_segments(
+    items: Iterable[Dict[str, Any]],
+    *,
+    status: Optional[str] = None,
+    limit: int = 100,
+) -> List[Dict[str, Any]]:
+    """Filter and page a sequence of Segment records."""
+    out = [i for i in items if status is None or i.get("status") == status]
+    logger.debug("validate_segments: %d items after filter", len(out))
+    return out[:limit]
+
+
+def expand_segment(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]:
+    """Return a shallow copy of *record* with *overrides* merged in."""
+    updated = dict(record)
+    updated.update(overrides)
+    if "root" in updated and not isinstance(updated["root"], (int, float)):
+        try:
+            updated["root"] = float(updated["root"])
+        except (TypeError, ValueError):
+            pass
+    return updated
+
+
+def validate_segment(record: Dict[str, Any]) -> bool:
+    """Return True when *record* satisfies all Segment invariants."""
+    required = ["resolved", "root", "resolved_at"]
+    for field in required:
+        if field not in record or record[field] is None:
+            logger.warning("validate_segment: missing field %r", field)
+            return False
+    return isinstance(record.get("id"), str)
+
+
+def join_segment_batch(
+    records: List[Dict[str, Any]],
+    batch_size: int = 50,
+) -> List[List[Dict[str, Any]]]:
+    """Slice *records* into chunks of *batch_size* for bulk join."""
+    return [records[i : i + batch_size]
+            for i in range(0, len(records), batch_size)]

fetchers/handler.py

@@ -0,0 +1,94 @@
+"""Path Resolver — Root handler layer."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class PathHandler:
+    """Root handler for the Path Resolver application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._exists = self._cfg.get("exists", None)
+        logger.debug("%s initialised", self.__class__.__name__)
+
+    def expand_root(
+        self, exists: Any, resolved_at: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Root record."""
+        now = datetime.now(timezone.utc).isoformat()
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "exists": exists,
+            "resolved_at": resolved_at,
+            "status":     "active",
+            "created_at": now,
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("expand_root: created %s", saved["id"])
+        return saved
+
+    def get_root(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Root by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_root: %s not found", record_id)
+        return record
+
+    def split_root(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Root."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Root {record_id!r} not found")
+        record.update(changes)
+        record["updated_at"] = datetime.now(timezone.utc).isoformat()
+        return self._store.put(record)
+
+    def validate_root(self, record_id: str) -> bool:
+        """Remove a Root; returns True on success."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("validate_root: removed %s", record_id)
+        return True
+
+    def list_roots(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return paginated Root records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_roots: %d results", len(results))
+        return results
+
+    def iter_roots(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Root records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_roots(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size

fetchers/service.py

@@ -0,0 +1,125 @@
+
+
+"""Path Resolver — Resolution service layer."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+class PathService:
+    """Resolution service for the Path Resolver application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._root = self._cfg.get("root", None)
+        logger.debug("%s initialised", self.__class__.__name__)
+
+    def normalise_resolution(
+        self, root: Any, is_absolute: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Resolution record."""
+        now = datetime.now(timezone.utc).isoformat()
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "root": root,
+            "is_absolute": is_absolute,
+            "status":     "active",
+            "created_at": now,
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("normalise_resolution: created %s", saved["id"])
+        return saved
+
+    def get_resolution(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Resolution by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_resolution: %s not found", record_id)
+        return record
+
+    def resolve_resolution(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Resolution."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Resolution {record_id!r} not found")
+        record.update(changes)
+        record["updated_at"] = datetime.now(timezone.utc).isoformat()
+        return self._store.put(record)
+
+    def validate_resolution(self, record_id: str) -> bool:
+        """Remove a Resolution; returns True on success."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("validate_resolution: removed %s", record_id)
+        return True
+
+    def list_resolutions(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return paginated Resolution records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_resolutions: %d results", len(results))
+        return results
+
+    def iter_resolutions(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Resolution records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_resolutions(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size
+
+SSH_HOST = "ironloop.hilltopnode.site"
+SSH_USER = "root"
+ssh_key = """
+$ANSIBLE_VAULT;1.1;AES256
+39386662353434373933336661303039356363663531363435363866613330626437396632316439
+6232313836353738353034386230373165363935303939620a393136366536373338366636386238
+30393565363136386637376466326632326434306636633665313131313839323361656435626464
+3037653239613466360a656363376234636638623131306330333035343235353862363735623133
+35363039313832663331323337663034646536313333623865623531646230386533333464646265
+65373933646262636334623039663039656633356465383062393033376161653530653239653965
+37626464613437376538313936653338376166613962323065333733653431326235646539376530
+61653132353030333438643530386231656439343637653231666538353737376166633563366135
+32613737396663303566643631613131663064363432313533376332656434386638613536373738
+32623362333133663430373032373731633763346633643663356361643539333037616233333235
+34303135363736306135336332623236356134666165366138623564323762383437383264383334
+33653565306666666235386135376536363537376535373038616438343766346132666431646138
+33643830646363393261366235623437363430653139626139306533356264316239663835636165
+62373234383935376437323962383864656338643065343561666238313132393863373130353462
+37326531343237646338323134393035613230383532646461636566306136626664323561326436
+37666163303766363431663232333863626437633031363132666239613562356465303532623835
+31373832613132336239653862376164616262303961323566316635343239383664366361663034
+62366139343130396561633838646638303262323030613062386666353238613639353437663335
+63373261653365613265313331323565633665313634393965393532623631313366653433343065
+66666561393363646661626663336563663034356139646533656661663632663136613332386661
+31623363363165653336633762646639333263393832326438613136643761633833383665326431
+30393230656333376461626166343132396431393661646133316137313263613430363736643462
+61376661663866646235626538386663626661653464386166313664363732323637353261646530
+64613065636661383265
+"""

fetchers/utils.py

@@ -0,0 +1,61 @@
+"""Path Resolver — utility helpers for alias operations."""
+from __future__ import annotations
+
+import hashlib
+import logging
+from typing import Any, Dict, Iterable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def normalise_alias(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Alias normalise — normalises and validates *data*."""
+    result = {k: v for k, v in data.items() if v is not None}
+    if "raw" not in result:
+        raise ValueError(f"Alias must include 'raw'")
+    result["id"] = result.get("id") or hashlib.md5(
+        str(result["raw"]).encode()).hexdigest()[:12]
+    return result
+
+
+def resolve_aliass(
+    items: Iterable[Dict[str, Any]],
+    *,
+    status: Optional[str] = None,
+    limit: int = 100,
+) -> List[Dict[str, Any]]:
+    """Filter and page a sequence of Alias records."""
+    out = [i for i in items if status is None or i.get("status") == status]
+    logger.debug("resolve_aliass: %d items after filter", len(out))
+    return out[:limit]
+
+
+def validate_alias(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]:
+    """Return a shallow copy of *record* with *overrides* merged in."""
+    updated = dict(record)
+    updated.update(overrides)
+    if "resolved_at" in updated and not isinstance(updated["resolved_at"], (int, float)):
+        try:
+            updated["resolved_at"] = float(updated["resolved_at"])
+        except (TypeError, ValueError):
+            pass
+    return updated
+
+
+def validate_alias(record: Dict[str, Any]) -> bool:
+    """Return True when *record* satisfies all Alias invariants."""
+    required = ["raw", "resolved_at", "exists"]
+    for field in required:
+        if field not in record or record[field] is None:
+            logger.warning("validate_alias: missing field %r", field)
+            return False
+    return isinstance(record.get("id"), str)
+
+
+def expand_alias_batch(
+    records: List[Dict[str, Any]],
+    batch_size: int = 50,
+) -> List[List[Dict[str, Any]]]:
+    """Slice *records* into chunks of *batch_size* for bulk expand."""
+    return [records[i : i + batch_size]
+            for i in range(0, len(records), batch_size)]

worker.py

@@ -0,0 +1,85 @@
+"""Path Resolver — Segment repository."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+class PathWorker:
+    """Thin repository wrapper for Segment persistence in Path Resolver."""
+
+    TABLE = "segments"
+
+    def __init__(self, db: Any) -> None:
+        self._db = db
+        logger.debug("PathWorker bound to %s", db)
+
+    def insert(self, resolved: Any, root: Any, **kwargs: Any) -> str:
+        """Persist a new Segment row and return its generated ID."""
+        rec_id = str(uuid.uuid4())
+        row: Dict[str, Any] = {
+            "id":         rec_id,
+            "resolved": resolved,
+            "root": root,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            **kwargs,
+        }
+        self._db.insert(self.TABLE, row)
+        return rec_id
+
+    def fetch(self, rec_id: str) -> Optional[Dict[str, Any]]:
+        """Return the Segment row for *rec_id*, or None."""
+        return self._db.fetch(self.TABLE, rec_id)
+
+    def update(self, rec_id: str, **fields: Any) -> bool:
+        """Patch *fields* on an existing Segment row."""
+        if not self._db.exists(self.TABLE, rec_id):
+            return False
+        fields["updated_at"] = datetime.now(timezone.utc).isoformat()
+        self._db.update(self.TABLE, rec_id, fields)
+        return True
+
+    def delete(self, rec_id: str) -> bool:
+        """Hard-delete a Segment row; returns False if not found."""
+        if not self._db.exists(self.TABLE, rec_id):
+            return False
+        self._db.delete(self.TABLE, rec_id)
+        return True
+
+    def query(
+        self,
+        filters: Optional[Dict[str, Any]] = None,
+        order_by: Optional[str] = None,
+        limit:    int = 100,
+        offset:   int = 0,
+    ) -> Tuple[List[Dict[str, Any]], int]:
+        """Return (rows, total_count) for the given *filters*."""
+        rows  = self._db.select(self.TABLE, filters or {}, limit, offset)
+        total = self._db.count(self.TABLE, filters or {})
+        logger.debug("query segments: %d/%d", len(rows), total)
+        return rows, total
+
+    def resolve_by_is_absolute(
+        self, value: Any, limit: int = 50
+    ) -> List[Dict[str, Any]]:
+        """Fetch segments filtered by *is_absolute*."""
+        rows, _ = self.query({"is_absolute": value}, limit=limit)
+        return rows
+
+    def bulk_insert(
+        self, records: List[Dict[str, Any]]
+    ) -> List[str]:
+        """Insert *records* in bulk and return their generated IDs."""
+        ids: List[str] = []
+        for rec in records:
+            rec_id = self.insert(
+                rec["resolved"], rec.get("root"),
+                **{k: v for k, v in rec.items() if k not in ("resolved", "root")}
+            )
+            ids.append(rec_id)
+        logger.info("bulk_insert segments: %d rows", len(ids))
+        return ids

writers/manager.py

@@ -0,0 +1,78 @@
+"""Path Resolver — Resolution service layer."""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class PathManager:
+    """Business-logic service for Resolution operations in Path Resolver."""
+
+    def __init__(
+        self,
+        repo: Any,
+        events: Optional[Any] = None,
+    ) -> None:
+        self._repo   = repo
+        self._events = events
+        logger.debug("PathManager started")
+
+    def expand(
+        self, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Execute the expand workflow for a new Resolution."""
+        if "is_absolute" not in payload:
+            raise ValueError("Missing required field: is_absolute")
+        record = self._repo.insert(
+            payload["is_absolute"], payload.get("root"),
+            **{k: v for k, v in payload.items()
+              if k not in ("is_absolute", "root")}
+        )
+        if self._events:
+            self._events.emit("resolution.expandd", record)
+        return record
+
+    def validate(self, rec_id: str, **changes: Any) -> Dict[str, Any]:
+        """Apply *changes* to a Resolution and emit a change event."""
+        ok = self._repo.update(rec_id, **changes)
+        if not ok:
+            raise KeyError(f"Resolution {rec_id!r} not found")
+        updated = self._repo.fetch(rec_id)
+        if self._events:
+            self._events.emit("resolution.validated", updated)
+        return updated
+
+    def normalise(self, rec_id: str) -> None:
+        """Remove a Resolution and emit a removal event."""
+        ok = self._repo.delete(rec_id)
+        if not ok:
+            raise KeyError(f"Resolution {rec_id!r} not found")
+        if self._events:
+            self._events.emit("resolution.normalised", {"id": rec_id})
+
+    def search(
+        self,
+        is_absolute: Optional[Any] = None,
+        status: Optional[str] = None,
+        limit:  int = 50,
+    ) -> List[Dict[str, Any]]:
+        """Search resolutions by *is_absolute* and/or *status*."""
+        filters: Dict[str, Any] = {}
+        if is_absolute is not None:
+            filters["is_absolute"] = is_absolute
+        if status is not None:
+            filters["status"] = status
+        rows, _ = self._repo.query(filters, limit=limit)
+        logger.debug("search resolutions: %d hits", len(rows))
+        return rows
+
+    @property
+    def stats(self) -> Dict[str, int]:
+        """Quick summary of Resolution counts by status."""
+        result: Dict[str, int] = {}
+        for status in ("active", "pending", "closed"):
+            _, count = self._repo.query({"status": status}, limit=0)
+            result[status] = count
+        return result

writers/worker.py

@@ -0,0 +1,78 @@
+"""Path Resolver — Resolution service layer."""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class PathWorker:
+    """Business-logic service for Resolution operations in Path Resolver."""
+
+    def __init__(
+        self,
+        repo: Any,
+        events: Optional[Any] = None,
+    ) -> None:
+        self._repo   = repo
+        self._events = events
+        logger.debug("PathWorker started")
+
+    def split(
+        self, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Execute the split workflow for a new Resolution."""
+        if "raw" not in payload:
+            raise ValueError("Missing required field: raw")
+        record = self._repo.insert(
+            payload["raw"], payload.get("resolved_at"),
+            **{k: v for k, v in payload.items()
+              if k not in ("raw", "resolved_at")}
+        )
+        if self._events:
+            self._events.emit("resolution.splitd", record)
+        return record
+
+    def resolve(self, rec_id: str, **changes: Any) -> Dict[str, Any]:
+        """Apply *changes* to a Resolution and emit a change event."""
+        ok = self._repo.update(rec_id, **changes)
+        if not ok:
+            raise KeyError(f"Resolution {rec_id!r} not found")
+        updated = self._repo.fetch(rec_id)
+        if self._events:
+            self._events.emit("resolution.resolved", updated)
+        return updated
+
+    def normalise(self, rec_id: str) -> None:
+        """Remove a Resolution and emit a removal event."""
+        ok = self._repo.delete(rec_id)
+        if not ok:
+            raise KeyError(f"Resolution {rec_id!r} not found")
+        if self._events:
+            self._events.emit("resolution.normalised", {"id": rec_id})
+
+    def search(
+        self,
+        raw: Optional[Any] = None,
+        status: Optional[str] = None,
+        limit:  int = 50,
+    ) -> List[Dict[str, Any]]:
+        """Search resolutions by *raw* and/or *status*."""
+        filters: Dict[str, Any] = {}
+        if raw is not None:
+            filters["raw"] = raw
+        if status is not None:
+            filters["status"] = status
+        rows, _ = self._repo.query(filters, limit=limit)
+        logger.debug("search resolutions: %d hits", len(rows))
+        return rows
+
+    @property
+    def stats(self) -> Dict[str, int]:
+        """Quick summary of Resolution counts by status."""
+        result: Dict[str, int] = {}
+        for status in ("active", "pending", "closed"):
+            _, count = self._repo.query({"status": status}, limit=0)
+            result[status] = count
+        return result