|
1
|
""" |
|
2
|
ClusterManager — shared Redis graph with local SQLite snapshot workflow. |
|
3
|
|
|
4
|
Supports agent swarms sharing a central FalkorDB graph over Redis while |
|
5
|
allowing individual agents to maintain a local SQLite snapshot for offline |
|
6
|
or low-latency operation. |
|
7
|
|
|
8
|
Usage: |
|
9
|
manager = ClusterManager("redis://localhost:6379", local_db_path=".navegador/graph.db") |
|
10
|
manager.snapshot_to_local() # pull Redis -> SQLite |
|
11
|
manager.push_to_shared() # push SQLite -> Redis |
|
12
|
info = manager.status() # {"local_version": ..., "shared_version": ..., "in_sync": ...} |
|
13
|
""" |
|
14
|
|
|
15
|
from __future__ import annotations |
|
16
|
|
|
17
|
import json |
|
18
|
import logging |
|
19
|
import time |
|
20
|
from pathlib import Path |
|
21
|
from typing import Any |
|
22
|
|
|
23
|
logger = logging.getLogger(__name__) |
|
24
|
|
|
25
|
_VERSION_KEY = "navegador:graph:version" |
|
26
|
_META_KEY = "navegador:graph:meta" |
|
27
|
_SNAPSHOT_KEY = "navegador:graph:snapshot" |
|
28
|
|
|
29
|
|
|
30
|
class ClusterManager: |
|
31
|
""" |
|
32
|
Coordinates graph state between a shared Redis FalkorDB instance and a |
|
33
|
local SQLite snapshot. |
|
34
|
|
|
35
|
Parameters |
|
36
|
---------- |
|
37
|
redis_url: |
|
38
|
URL of the Redis server hosting the shared FalkorDB graph. |
|
39
|
local_db_path: |
|
40
|
Path to the local SQLite (falkordblite) database file. |
|
41
|
Defaults to ``.navegador/graph.db``. |
|
42
|
redis_client: |
|
43
|
Optional pre-built Redis client (used for testing / dependency injection). |
|
44
|
""" |
|
45
|
|
|
46
|
def __init__( |
|
47
|
self, |
|
48
|
redis_url: str, |
|
49
|
local_db_path: str | Path | None = None, |
|
50
|
*, |
|
51
|
redis_client: Any = None, |
|
52
|
) -> None: |
|
53
|
self.redis_url = redis_url |
|
54
|
self.local_db_path = Path(local_db_path) if local_db_path else Path(".navegador/graph.db") |
|
55
|
self._redis = redis_client or self._connect_redis(redis_url) |
|
56
|
|
|
57
|
# ── Internal helpers ────────────────────────────────────────────────────── |
|
58
|
|
|
59
|
@staticmethod |
|
60
|
def _connect_redis(url: str) -> Any: |
|
61
|
try: |
|
62
|
import redis # type: ignore[import] |
|
63
|
except ImportError as exc: |
|
64
|
raise ImportError("Install redis: pip install redis") from exc |
|
65
|
return redis.from_url(url) |
|
66
|
|
|
67
|
def _redis_version(self) -> int: |
|
68
|
raw = self._redis.get(_VERSION_KEY) |
|
69
|
return int(raw) if raw is not None else 0 |
|
70
|
|
|
71
|
def _local_version(self) -> int: |
|
72
|
meta_path = self.local_db_path.parent / "cluster_meta.json" |
|
73
|
if not meta_path.exists(): |
|
74
|
return 0 |
|
75
|
try: |
|
76
|
data = json.loads(meta_path.read_text()) |
|
77
|
return int(data.get("version", 0)) |
|
78
|
except (json.JSONDecodeError, OSError): |
|
79
|
return 0 |
|
80
|
|
|
81
|
def _set_local_version(self, version: int) -> None: |
|
82
|
meta_path = self.local_db_path.parent / "cluster_meta.json" |
|
83
|
meta_path.parent.mkdir(parents=True, exist_ok=True) |
|
84
|
existing: dict[str, Any] = {} |
|
85
|
if meta_path.exists(): |
|
86
|
try: |
|
87
|
existing = json.loads(meta_path.read_text()) |
|
88
|
except (json.JSONDecodeError, OSError): |
|
89
|
pass |
|
90
|
existing["version"] = version |
|
91
|
existing["updated_at"] = time.time() |
|
92
|
meta_path.write_text(json.dumps(existing, indent=2)) |
|
93
|
|
|
94
|
def _export_local_graph(self) -> dict[str, Any]: |
|
95
|
"""Read all nodes and edges from the local SQLite graph and return as dict.""" |
|
96
|
from navegador.graph.store import GraphStore |
|
97
|
|
|
98
|
store = GraphStore.sqlite(self.local_db_path) |
|
99
|
nodes_result = store.query("MATCH (n) RETURN n") |
|
100
|
edges_result = store.query("MATCH (a)-[r]->(b) RETURN a, type(r), r, b") |
|
101
|
|
|
102
|
nodes = [] |
|
103
|
if nodes_result.result_set: |
|
104
|
for row in nodes_result.result_set: |
|
105
|
node = row[0] |
|
106
|
nodes.append({"labels": list(node.labels), "properties": dict(node.properties)}) |
|
107
|
|
|
108
|
edges = [] |
|
109
|
if edges_result.result_set: |
|
110
|
for row in edges_result.result_set: |
|
111
|
src, rel_type, rel, dst = row |
|
112
|
edges.append( |
|
113
|
{ |
|
114
|
"src_labels": list(src.labels), |
|
115
|
"src_props": dict(src.properties), |
|
116
|
"rel_type": rel_type, |
|
117
|
"rel_props": dict(rel.properties) if rel.properties else {}, |
|
118
|
"dst_labels": list(dst.labels), |
|
119
|
"dst_props": dict(dst.properties), |
|
120
|
} |
|
121
|
) |
|
122
|
|
|
123
|
return {"nodes": nodes, "edges": edges} |
|
124
|
|
|
125
|
def _import_to_local_graph(self, data: dict[str, Any]) -> None: |
|
126
|
"""Write snapshot data into the local SQLite graph.""" |
|
127
|
from navegador.graph.store import GraphStore |
|
128
|
|
|
129
|
store = GraphStore.sqlite(self.local_db_path) |
|
130
|
store.clear() |
|
131
|
|
|
132
|
for node in data.get("nodes", []): |
|
133
|
label = node["labels"][0] if node["labels"] else "Node" |
|
134
|
props = node["properties"] |
|
135
|
store.create_node(label, props) |
|
136
|
|
|
137
|
for edge in data.get("edges", []): |
|
138
|
src_label = edge["src_labels"][0] if edge["src_labels"] else "Node" |
|
139
|
dst_label = edge["dst_labels"][0] if edge["dst_labels"] else "Node" |
|
140
|
rel_type = edge["rel_type"] |
|
141
|
src_props = edge["src_props"] |
|
142
|
dst_props = edge["dst_props"] |
|
143
|
rel_props = edge.get("rel_props") or None |
|
144
|
|
|
145
|
src_key = {k: v for k, v in src_props.items() if k in ("name", "file_path")} |
|
146
|
dst_key = {k: v for k, v in dst_props.items() if k in ("name", "file_path")} |
|
147
|
|
|
148
|
if src_key and dst_key: |
|
149
|
store.create_edge(src_label, src_key, rel_type, dst_label, dst_key, rel_props) |
|
150
|
|
|
151
|
# ── Public API ──────────────────────────────────────────────────────────── |
|
152
|
|
|
153
|
def snapshot_to_local(self) -> None: |
|
154
|
"""Pull the shared Redis graph down into the local SQLite snapshot.""" |
|
155
|
raw = self._redis.get(_SNAPSHOT_KEY) |
|
156
|
if raw is None: |
|
157
|
logger.warning("No shared snapshot found in Redis; local graph unchanged.") |
|
158
|
return |
|
159
|
|
|
160
|
data = json.loads(raw) |
|
161
|
self._import_to_local_graph(data) |
|
162
|
shared_ver = self._redis_version() |
|
163
|
self._set_local_version(shared_ver) |
|
164
|
logger.info("Snapshot pulled from Redis (version %d) to %s", shared_ver, self.local_db_path) |
|
165
|
|
|
166
|
def push_to_shared(self) -> None: |
|
167
|
"""Push the local SQLite graph up to the shared Redis instance.""" |
|
168
|
data = self._export_local_graph() |
|
169
|
serialized = json.dumps(data) |
|
170
|
pipe = self._redis.pipeline() |
|
171
|
pipe.set(_SNAPSHOT_KEY, serialized) |
|
172
|
new_version = self._redis_version() + 1 |
|
173
|
pipe.set(_VERSION_KEY, new_version) |
|
174
|
pipe.hset( |
|
175
|
_META_KEY, |
|
176
|
mapping={ |
|
177
|
"last_push": time.time(), |
|
178
|
"node_count": len(data["nodes"]), |
|
179
|
"edge_count": len(data["edges"]), |
|
180
|
}, |
|
181
|
) |
|
182
|
pipe.execute() |
|
183
|
self._set_local_version(new_version) |
|
184
|
logger.info( |
|
185
|
"Pushed local graph to Redis (version %d): %d nodes, %d edges", |
|
186
|
new_version, |
|
187
|
len(data["nodes"]), |
|
188
|
len(data["edges"]), |
|
189
|
) |
|
190
|
|
|
191
|
def sync(self) -> None: |
|
192
|
""" |
|
193
|
Bidirectional sync. |
|
194
|
|
|
195
|
Strategy: if the shared version is newer, pull it down first; if |
|
196
|
local is ahead (or equal), push local up. This is a last-write-wins |
|
197
|
merge — suitable for most single-writer swarm topologies. |
|
198
|
""" |
|
199
|
local_ver = self._local_version() |
|
200
|
shared_ver = self._redis_version() |
|
201
|
|
|
202
|
if shared_ver > local_ver: |
|
203
|
logger.info("Shared graph is newer (%d > %d); pulling.", shared_ver, local_ver) |
|
204
|
self.snapshot_to_local() |
|
205
|
else: |
|
206
|
logger.info( |
|
207
|
"Local graph is current or ahead (%d >= %d); pushing.", local_ver, shared_ver |
|
208
|
) |
|
209
|
self.push_to_shared() |
|
210
|
|
|
211
|
def status(self) -> dict[str, Any]: |
|
212
|
""" |
|
213
|
Return a dict describing the sync state. |
|
214
|
|
|
215
|
Keys: ``local_version``, ``shared_version``, ``in_sync``. |
|
216
|
""" |
|
217
|
local_ver = self._local_version() |
|
218
|
shared_ver = self._redis_version() |
|
219
|
return { |
|
220
|
"local_version": local_ver, |
|
221
|
"shared_version": shared_ver, |
|
222
|
"in_sync": local_ver == shared_ver, |
|
223
|
} |
|
224
|
|