mirror of
https://github.com/markqvist/NomadNet.git
synced 2026-08-14 08:59:48 +00:00
Added RRC room history persistence by neutral
This commit is contained in:
Mark Qvist
+44
-20
@@ -146,10 +146,12 @@ class NomadNetworkApp:
|
||||
self.lxmf_sync_interval = 360*60
|
||||
self.lxmf_sync_limit = 8
|
||||
self.compact_stream = False
|
||||
|
||||
|
||||
self.required_stamp_cost = None
|
||||
self.accept_invalid_stamps = False
|
||||
|
||||
self.rrc_history_per_room_cap = 500
|
||||
|
||||
|
||||
if not os.path.isdir(self.storagepath):
|
||||
os.makedirs(self.storagepath)
|
||||
@@ -200,11 +202,11 @@ class NomadNetworkApp:
|
||||
import shutil
|
||||
examplespath = os.path.join(os.path.dirname(__file__), "examples")
|
||||
shutil.copytree(examplespath, self.examplespath, ignore=shutil.ignore_patterns("__pycache__"))
|
||||
|
||||
|
||||
except Exception as e:
|
||||
RNS.log("Could not copy examples into the "+self.examplespath+" directory.", RNS.LOG_ERROR)
|
||||
RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||
|
||||
|
||||
RNS.log("Could not load config file, creating default configuration file...")
|
||||
self.createDefaultConfig()
|
||||
self.firstrun = True
|
||||
@@ -377,7 +379,7 @@ class NomadNetworkApp:
|
||||
|
||||
if not self.disable_propagation:
|
||||
RNS.log("LXMF Propagation Node started on: "+RNS.prettyhexrep(self.message_router.propagation_destination.hash))
|
||||
|
||||
|
||||
self.node = nomadnet.Node(self)
|
||||
else:
|
||||
self.node = None
|
||||
@@ -442,7 +444,7 @@ class NomadNetworkApp:
|
||||
RNS.log("Starting job scheduler now", RNS.LOG_DEBUG)
|
||||
while self.should_run_jobs:
|
||||
now = time.time()
|
||||
|
||||
|
||||
if now > self.peer_settings["last_lxmf_sync"] + self.lxmf_sync_interval:
|
||||
RNS.log("Initiating automatic LXMF sync", RNS.LOG_VERBOSE)
|
||||
self.request_lxmf_sync(limit=self.lxmf_sync_limit)
|
||||
@@ -576,7 +578,7 @@ class NomadNetworkApp:
|
||||
self.peer_settings["propagation_node"] = node_hash
|
||||
self.save_peer_settings()
|
||||
self.autoselect_propagation_node()
|
||||
|
||||
|
||||
def get_default_propagation_node(self):
|
||||
return self.message_router.get_outbound_propagation_node()
|
||||
|
||||
@@ -608,7 +610,7 @@ class NomadNetworkApp:
|
||||
if self.print_messages:
|
||||
if self.print_all_messages:
|
||||
return True
|
||||
|
||||
|
||||
else:
|
||||
source_hash_text = RNS.hexrep(message.source_hash, delimit=False)
|
||||
|
||||
@@ -651,7 +653,7 @@ class NomadNetworkApp:
|
||||
received = time.time()
|
||||
|
||||
g = self.ui.glyphs
|
||||
|
||||
|
||||
m_rtime = datetime.fromtimestamp(message.timestamp)
|
||||
stime = m_rtime.strftime(self.time_format)
|
||||
|
||||
@@ -718,7 +720,7 @@ class NomadNetworkApp:
|
||||
def createDefaultConfig(self):
|
||||
self.config = ConfigObj(__default_nomadnet_config__)
|
||||
self.config.filename = self.configpath
|
||||
|
||||
|
||||
if not os.path.isdir(self.configdir):
|
||||
os.makedirs(self.configdir)
|
||||
self.config.write()
|
||||
@@ -784,7 +786,7 @@ class NomadNetworkApp:
|
||||
self.lxmf_sync_interval = value
|
||||
|
||||
if option == "lxmf_sync_limit":
|
||||
value = self.config["client"].as_int(option)
|
||||
value = self.config["client"].as_int(option)
|
||||
|
||||
if value > 0:
|
||||
self.lxmf_sync_limit = value
|
||||
@@ -810,7 +812,7 @@ class NomadNetworkApp:
|
||||
self.accept_invalid_stamps = value
|
||||
|
||||
if option == "max_accepted_size":
|
||||
value = self.config["client"].as_float(option)
|
||||
value = self.config["client"].as_float(option)
|
||||
|
||||
if value > 0:
|
||||
self.lxmf_max_incoming_size = value
|
||||
@@ -900,6 +902,16 @@ class NomadNetworkApp:
|
||||
if value == "web":
|
||||
self.uimode = nomadnet.ui.UI_WEB
|
||||
|
||||
if "rrc" in self.config:
|
||||
for option in self.config["rrc"]:
|
||||
if option == "history_per_room_cap":
|
||||
try:
|
||||
value = self.config["rrc"].as_int(option)
|
||||
except Exception:
|
||||
value = None
|
||||
if value is not None and value >= 0:
|
||||
self.rrc_history_per_room_cap = value
|
||||
|
||||
if "node" in self.config:
|
||||
if not "enable_node" in self.config["node"]:
|
||||
self.enable_node = False
|
||||
@@ -956,10 +968,10 @@ class NomadNetworkApp:
|
||||
value = self.config["node"].as_int("propagation_cost")
|
||||
if value < 13: value = 13
|
||||
self.node_propagation_cost = value
|
||||
|
||||
|
||||
if "pages_path" in self.config["node"]:
|
||||
self.pagespath = self.config["node"]["pages_path"]
|
||||
|
||||
|
||||
if not "page_refresh_interval" in self.config["node"]:
|
||||
self.page_refresh_interval = 0
|
||||
else:
|
||||
@@ -967,11 +979,11 @@ class NomadNetworkApp:
|
||||
if value < 0:
|
||||
value = 0
|
||||
self.page_refresh_interval = value
|
||||
|
||||
|
||||
|
||||
if "files_path" in self.config["node"]:
|
||||
self.filespath = self.config["node"]["files_path"]
|
||||
|
||||
|
||||
if not "file_refresh_interval" in self.config["node"]:
|
||||
self.file_refresh_interval = 0
|
||||
else:
|
||||
@@ -979,7 +991,7 @@ class NomadNetworkApp:
|
||||
if value < 0:
|
||||
value = 0
|
||||
self.file_refresh_interval = value
|
||||
|
||||
|
||||
|
||||
if "prioritise_destinations" in self.config["node"]:
|
||||
self.prioritised_lxmf_destinations = self.config["node"].as_list("prioritise_destinations")
|
||||
@@ -990,7 +1002,7 @@ class NomadNetworkApp:
|
||||
self.static_peers = self.config["node"].as_list("static_peers")
|
||||
else:
|
||||
self.static_peers = []
|
||||
|
||||
|
||||
if not "max_peers" in self.config["node"]:
|
||||
self.max_peers = None
|
||||
else:
|
||||
@@ -1030,13 +1042,13 @@ class NomadNetworkApp:
|
||||
self.print_all_messages = True
|
||||
|
||||
if self.config["printing"]["print_from"].lower() == "trusted":
|
||||
|
||||
|
||||
self.print_all_messages = False
|
||||
self.print_trusted_messages = True
|
||||
|
||||
if len(self.config["printing"]["print_from"]) == (RNS.Identity.TRUNCATED_HASHLENGTH//8)*2:
|
||||
self.allowed_message_print_destinations.append(self.config["printing"]["print_from"])
|
||||
|
||||
|
||||
if type(self.config["printing"]["print_from"]) == list:
|
||||
self.allowed_message_print_destinations = self.config["printing"].as_list("print_from")
|
||||
for allowed_entry in self.allowed_message_print_destinations:
|
||||
@@ -1056,7 +1068,7 @@ class NomadNetworkApp:
|
||||
template_file.write(__printing_template_msg__.encode("utf-8"))
|
||||
self.printing_template_msg = __printing_template_msg__
|
||||
|
||||
|
||||
|
||||
@staticmethod
|
||||
def get_shared_instance():
|
||||
if NomadNetworkApp._shared_instance != None:
|
||||
@@ -1212,6 +1224,18 @@ hide_guide = no
|
||||
# announces.
|
||||
sanitize_names = yes
|
||||
|
||||
[rrc]
|
||||
|
||||
# Maximum number of messages retained per
|
||||
# room in the in-memory scrollback buffer,
|
||||
# and the number of messages restored from
|
||||
# on-disk history at startup. The on-disk
|
||||
# log itself is appended to indefinitely;
|
||||
# this cap only controls how much backlog
|
||||
# is visible. Set to 0 to keep every message
|
||||
# in memory (full history).
|
||||
history_per_room_cap = 500
|
||||
|
||||
[node]
|
||||
|
||||
# Whether to enable node hosting
|
||||
|
||||
+154
-6
@@ -10,6 +10,17 @@ import RNS
|
||||
from nomadnet.vendor import cbor
|
||||
|
||||
|
||||
HISTORY_DIR_NAME = "rrc_history"
|
||||
HISTORY_FILENAME_SANITIZE_RE = re.compile(r"[^a-z0-9._-]+")
|
||||
|
||||
H_KIND = "k"
|
||||
H_SRC = "s"
|
||||
H_NICK = "n"
|
||||
H_TEXT = "t"
|
||||
H_TS = "ts"
|
||||
H_MENTION = "m"
|
||||
|
||||
|
||||
_MENTION_RE_CACHE = {}
|
||||
|
||||
|
||||
@@ -241,6 +252,8 @@ class RRCHub:
|
||||
self._pending_parts = set()
|
||||
self._silent_joins = set()
|
||||
|
||||
self._history_write_failed = False
|
||||
|
||||
def _log(self, msg, level=None):
|
||||
if level is None:
|
||||
level = RNS.LOG_INFO
|
||||
@@ -264,9 +277,20 @@ class RRCHub:
|
||||
self.unread_rooms.discard(r)
|
||||
self.mention_rooms.discard(r)
|
||||
self.members.pop(r, None)
|
||||
self._delete_history(r)
|
||||
self.manager.save()
|
||||
self.manager._notify_change(self)
|
||||
|
||||
def clear_messages(self, room):
|
||||
r = self._normalize_room(room)
|
||||
with self._lock:
|
||||
if r in self.messages:
|
||||
self.messages[r] = []
|
||||
self.unread_rooms.discard(r)
|
||||
self.mention_rooms.discard(r)
|
||||
self._delete_history(r)
|
||||
self.manager._notify_change(self)
|
||||
|
||||
def get_members(self, room):
|
||||
with self._lock:
|
||||
return list(self.members.get(room, set()))
|
||||
@@ -595,28 +619,124 @@ class RRCHub:
|
||||
self._record_message(RRCMessage("msg", r, self.manager.identity.hash, nick, text, _now_ms()), local=True)
|
||||
return mid
|
||||
|
||||
def _per_room_cap(self):
|
||||
try:
|
||||
v = int(getattr(self.manager.app, "rrc_history_per_room_cap", 0))
|
||||
except Exception:
|
||||
return None
|
||||
return v if v > 0 else None
|
||||
|
||||
def _entry_for(self, msg):
|
||||
return {
|
||||
H_KIND: msg.kind,
|
||||
H_SRC: bytes(msg.src) if isinstance(msg.src, (bytes, bytearray)) else None,
|
||||
H_NICK: msg.nick if isinstance(msg.nick, str) else None,
|
||||
H_TEXT: msg.text if isinstance(msg.text, str) else "",
|
||||
H_TS: int(msg.ts) if isinstance(msg.ts, int) else _now_ms(),
|
||||
H_MENTION: bool(getattr(msg, "mention", False)),
|
||||
}
|
||||
|
||||
def _msg_from_entry(self, room, entry):
|
||||
if not isinstance(entry, dict):
|
||||
return None
|
||||
m = RRCMessage(
|
||||
entry.get(H_KIND) if isinstance(entry.get(H_KIND), str) else "msg",
|
||||
room,
|
||||
entry.get(H_SRC) if isinstance(entry.get(H_SRC), (bytes, bytearray)) else None,
|
||||
entry.get(H_NICK) if isinstance(entry.get(H_NICK), str) else None,
|
||||
entry.get(H_TEXT) if isinstance(entry.get(H_TEXT), str) else "",
|
||||
entry.get(H_TS) if isinstance(entry.get(H_TS), int) else 0,
|
||||
)
|
||||
m.mention = bool(entry.get(H_MENTION, False))
|
||||
return m
|
||||
|
||||
def _persistable_room(self, room):
|
||||
return isinstance(room, str) and room and room != "*"
|
||||
|
||||
def _append_history(self, room, msg):
|
||||
if not self._persistable_room(room):
|
||||
return
|
||||
try:
|
||||
self.manager._ensure_history_dir(self)
|
||||
path = self.manager._history_path(self, room)
|
||||
with open(path, "ab") as f:
|
||||
f.write(cbor.encode(self._entry_for(msg)))
|
||||
self._history_write_failed = False
|
||||
except Exception as e:
|
||||
if not self._history_write_failed:
|
||||
self._history_write_failed = True
|
||||
self._log("history persistence failed, suppressing further warnings until recovery: "+str(e), RNS.LOG_ERROR)
|
||||
|
||||
def _delete_history(self, room):
|
||||
if not self._persistable_room(room):
|
||||
return
|
||||
path = self.manager._history_path(self, room)
|
||||
try:
|
||||
if os.path.isfile(path):
|
||||
os.unlink(path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _load_history(self):
|
||||
with self._lock:
|
||||
rooms = list(self.messages.keys())
|
||||
for room in rooms:
|
||||
if not self._persistable_room(room):
|
||||
continue
|
||||
path = self.manager._history_path(self, room)
|
||||
if not os.path.isfile(path):
|
||||
continue
|
||||
window = deque(maxlen=self._per_room_cap())
|
||||
decode_error = None
|
||||
try:
|
||||
with open(path, "rb") as f:
|
||||
while True:
|
||||
try:
|
||||
window.append(cbor.load(f))
|
||||
except EOFError:
|
||||
break
|
||||
except Exception as ex:
|
||||
decode_error = ex
|
||||
break
|
||||
except OSError as ex:
|
||||
self._log("history load failed for #"+room+": "+str(ex), RNS.LOG_ERROR)
|
||||
continue
|
||||
if decode_error is not None:
|
||||
self._log("history file for #"+room+" is corrupt, truncating to last "+str(len(window))+" valid messages: "+str(decode_error), RNS.LOG_ERROR)
|
||||
msgs = []
|
||||
for e in window:
|
||||
m = self._msg_from_entry(room, e)
|
||||
if m is not None:
|
||||
msgs.append(m)
|
||||
with self._lock:
|
||||
self.messages[room] = msgs
|
||||
|
||||
def _record_message(self, msg, local=False):
|
||||
cap = self._per_room_cap()
|
||||
with self._lock:
|
||||
buf = self.messages.setdefault(msg.room or "*", [])
|
||||
buf.append(msg)
|
||||
if len(buf) > 500:
|
||||
del buf[:len(buf)-500]
|
||||
if cap is not None and len(buf) > cap:
|
||||
del buf[:len(buf)-cap]
|
||||
if not local and msg.room:
|
||||
if msg.room != self.manager.active_room_for(self):
|
||||
self.unread_rooms.add(msg.room)
|
||||
if msg.mention:
|
||||
self.mention_rooms.add(msg.room)
|
||||
self._append_history(msg.room, msg)
|
||||
self.manager._notify_messages(self, msg)
|
||||
|
||||
def _record_system(self, room, text):
|
||||
if not room:
|
||||
return
|
||||
msg = RRCMessage("system", room, None, None, text, _now_ms())
|
||||
cap = self._per_room_cap()
|
||||
with self._lock:
|
||||
buf = self.messages.setdefault(room, [])
|
||||
buf.append(msg)
|
||||
if len(buf) > 500:
|
||||
del buf[:len(buf)-500]
|
||||
if cap is not None and len(buf) > cap:
|
||||
del buf[:len(buf)-cap]
|
||||
self._append_history(room, msg)
|
||||
self.manager._notify_messages(self, msg)
|
||||
|
||||
def _record_notice(self, msg):
|
||||
@@ -626,6 +746,7 @@ class RRCHub:
|
||||
if target_room:
|
||||
msg.room = target_room
|
||||
|
||||
cap = self._per_room_cap()
|
||||
with self._lock:
|
||||
self.notices.append(msg)
|
||||
if len(self.notices) > 200:
|
||||
@@ -633,10 +754,12 @@ class RRCHub:
|
||||
if target_room:
|
||||
buf = self.messages.setdefault(target_room, [])
|
||||
buf.append(msg)
|
||||
if len(buf) > 500:
|
||||
del buf[:len(buf)-500]
|
||||
if cap is not None and len(buf) > cap:
|
||||
del buf[:len(buf)-cap]
|
||||
if target_room != self.manager.active_room_for(self):
|
||||
self.unread_rooms.add(target_room)
|
||||
if target_room:
|
||||
self._append_history(target_room, msg)
|
||||
self.manager._notify_messages(self, msg)
|
||||
|
||||
def get_messages(self, room):
|
||||
@@ -1108,6 +1231,27 @@ class RRCManager:
|
||||
def _store_path(self):
|
||||
return os.path.join(self.app.storagepath, "rrc_hubs")
|
||||
|
||||
def _history_root(self):
|
||||
return os.path.join(self.app.storagepath, HISTORY_DIR_NAME)
|
||||
|
||||
def _history_dir(self, hub):
|
||||
hub_key = hub.hub_hash.hex()
|
||||
if hub.dest_name and hub.dest_name != DEFAULT_DEST_NAME:
|
||||
suffix = hashlib.sha256(hub.dest_name.encode("utf-8")).hexdigest()[:8]
|
||||
hub_key = hub_key + "__" + suffix
|
||||
return os.path.join(self._history_root(), hub_key)
|
||||
|
||||
def _history_path(self, hub, room):
|
||||
sanitized = HISTORY_FILENAME_SANITIZE_RE.sub("_", room or "")[:64]
|
||||
room_hash = hashlib.sha256((room or "").encode("utf-8")).hexdigest()[:8]
|
||||
filename = (sanitized + "_" + room_hash + ".log") if sanitized else (room_hash + ".log")
|
||||
return os.path.join(self._history_dir(hub), filename)
|
||||
|
||||
def _ensure_history_dir(self, hub):
|
||||
d = self._history_dir(hub)
|
||||
os.makedirs(d, exist_ok=True)
|
||||
return d
|
||||
|
||||
def load(self):
|
||||
if self._loaded:
|
||||
return
|
||||
@@ -1161,6 +1305,10 @@ class RRCManager:
|
||||
no = e.get("nick")
|
||||
if isinstance(no, str) and no:
|
||||
hub.nick_override = no
|
||||
try:
|
||||
hub._load_history()
|
||||
except Exception as ex:
|
||||
RNS.log("Failed to load RRC history for "+hub.name+": "+str(ex), RNS.LOG_ERROR)
|
||||
except Exception as e:
|
||||
RNS.log("Failed to load RRC hubs: "+str(e), RNS.LOG_ERROR)
|
||||
finally:
|
||||
|
||||
@@ -727,8 +727,7 @@ class RoomWidget(urwid.WidgetWrap):
|
||||
return
|
||||
|
||||
if cmd == "clear":
|
||||
with self.hub._lock:
|
||||
self.hub.messages[self.room] = []
|
||||
self.hub.clear_messages(self.room)
|
||||
self.update_messages(replace=True)
|
||||
return
|
||||
|
||||
|
||||
@@ -771,6 +771,16 @@ The maximum accepted unpacked size for messages received directly from other pee
|
||||
With this option enabled, Nomad Network will only display one entry in the announce stream per destination. Older announces are culled when a new one arrives.
|
||||
<
|
||||
|
||||
>> RRC Section
|
||||
|
||||
This section hold configuration directives related to the RRC client behaviour. It is delimited by the `![rrc]`! header in the configuration file. Available directives, along with their default values, are as follows:
|
||||
|
||||
>>>
|
||||
`!history_per_room_cap = 500`!
|
||||
>>>>
|
||||
Maximum number of messages retained per room in the in-memory scrollback buffer, and the number of messages restored from on-disk history at startup. The on-disk log itself is appended to indefinitely; this cap only controls how much backlog is visible. Set to 0 to keep every message in memory.
|
||||
<
|
||||
|
||||
>> Text UI Section
|
||||
|
||||
This section hold configuration directives related to the look and feel of the text-based user interface of the program. It is delimited by the `![textui]`! header in the configuration file. Available directives, along with their default values, are as follows:
|
||||
|
||||
Reference in New Issue
Block a user