371 lines
15 KiB
Python
371 lines
15 KiB
Python
import json
|
|
import logging
|
|
import random
|
|
import threading
|
|
import time
|
|
import RNS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from tinyweb_forum.bloom import BloomFilter
|
|
|
|
FORUM_APP = "tinyweb-forum"
|
|
SYNC_INTERVAL = 300
|
|
REQUEST_TIMEOUT = 60
|
|
GOSSIP_FANOUT = 20
|
|
BLOOM_SIZE = 2048
|
|
BLOOM_HASHES = 3
|
|
MAX_PEER_FAILURES = 5
|
|
|
|
|
|
class _ForumAnnounceHandler:
|
|
aspect_filter = FORUM_APP
|
|
receive_path_responses = False
|
|
|
|
def __init__(self, fdb, identity):
|
|
self.fdb = fdb
|
|
self.my_hash = identity.hash.hex() if identity else "local"
|
|
|
|
def received_announce(self, destination_hash, announced_identity, app_data):
|
|
if announced_identity is None:
|
|
return
|
|
peer_hash = announced_identity.hash.hex()
|
|
if peer_hash and peer_hash != self.my_hash:
|
|
self.fdb.add_known_peer(peer_hash)
|
|
|
|
|
|
class ForumSync:
|
|
def __init__(self, fdb, identity, reticulum, handlers_ref):
|
|
self.fdb = fdb
|
|
self.identity = identity
|
|
self.reticulum = reticulum
|
|
self.handlers_ref = handlers_ref
|
|
self.destination = None
|
|
self._announce_handler = None
|
|
self._running = False
|
|
self._thread = None
|
|
|
|
def _get_subscribed_topics(self):
|
|
raw = self.fdb.get_setting("topic_subscriptions", "")
|
|
return [t.strip().lower() for t in raw.split(",") if t.strip()]
|
|
|
|
def _merge_trust_gossip(self, trust_list, source_hash):
|
|
source_hops = self.fdb.get_trust_hops(source_hash)
|
|
if source_hops is None:
|
|
return
|
|
my_hash = self.identity.hash.hex() if self.identity else "local"
|
|
blocked = set(
|
|
h.strip()
|
|
for h in self.fdb.get_setting("blocked_instances", "").split(",")
|
|
if h.strip()
|
|
)
|
|
for peer_hash in trust_list:
|
|
if peer_hash in blocked or peer_hash == my_hash:
|
|
continue
|
|
new_hops = source_hops + 1
|
|
existing = self.fdb.get_trust_hops(peer_hash)
|
|
if existing is None or new_hops < existing:
|
|
self.fdb.add_trust_source(peer_hash, source_hash, new_hops)
|
|
|
|
def start(self):
|
|
self.destination = RNS.Destination(
|
|
self.identity,
|
|
RNS.Destination.IN,
|
|
RNS.Destination.SINGLE,
|
|
FORUM_APP,
|
|
)
|
|
self.destination.register_request_handler(
|
|
"/forum",
|
|
response_generator=self._rns_handler,
|
|
allow=RNS.Destination.ALLOW_ALL,
|
|
)
|
|
self.destination.announce(app_data=FORUM_APP.encode("utf-8"))
|
|
if self.fdb.get_setting("forum_auto_discover", "1") == "1":
|
|
self._enable_announce_handler()
|
|
self._running = True
|
|
if self.fdb.get_setting("forum_auto_sync", "0") == "1":
|
|
self._start_sync_loop()
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
self._disable_announce_handler()
|
|
|
|
def set_auto_sync(self, enabled):
|
|
self.fdb.set_setting("forum_auto_sync", "1" if enabled else "0")
|
|
if enabled:
|
|
self._start_sync_loop()
|
|
|
|
def sync_now(self):
|
|
instances = self.fdb.get_synced_instances()
|
|
random.shuffle(instances)
|
|
count = 0
|
|
my_topics = self._get_subscribed_topics()
|
|
for inst in instances[:GOSSIP_FANOUT]:
|
|
if not self._running:
|
|
break
|
|
try:
|
|
self._sync_with(inst["instance_hash"], my_topics)
|
|
count += 1
|
|
except Exception as e:
|
|
self.fdb.record_sync_result(inst["instance_hash"], False)
|
|
logger.error("sync error with %s: %s", inst["instance_hash"][:16], e)
|
|
return count
|
|
|
|
def _start_sync_loop(self):
|
|
if self._thread and self._thread.is_alive():
|
|
return
|
|
self._thread = threading.Thread(target=self._sync_loop, daemon=True)
|
|
self._thread.start()
|
|
|
|
def set_auto_discover(self, enabled):
|
|
self.fdb.set_setting("forum_auto_discover", "1" if enabled else "0")
|
|
if enabled:
|
|
self._enable_announce_handler()
|
|
else:
|
|
self._disable_announce_handler()
|
|
|
|
def _enable_announce_handler(self):
|
|
if self._announce_handler is None:
|
|
self._announce_handler = _ForumAnnounceHandler(self.fdb, self.identity)
|
|
RNS.Transport.register_announce_handler(self._announce_handler)
|
|
|
|
def _disable_announce_handler(self):
|
|
if self._announce_handler:
|
|
try:
|
|
RNS.Transport.deregister_announce_handler(self._announce_handler)
|
|
except Exception:
|
|
pass
|
|
self._announce_handler = None
|
|
|
|
def _rns_handler(self, path, data, request_id, link_id, remote_identity, requested_at):
|
|
if remote_identity:
|
|
data["peer_hash"] = remote_identity.hash.hex()
|
|
return self.handlers_ref().handle_sync(data)
|
|
|
|
def _sync_loop(self):
|
|
prune_counter = 0
|
|
while self._running:
|
|
try:
|
|
instances = self.fdb.get_synced_instances()
|
|
my_topics = self._get_subscribed_topics()
|
|
random.shuffle(instances)
|
|
for inst in instances[:GOSSIP_FANOUT]:
|
|
if not self._running:
|
|
break
|
|
try:
|
|
self._sync_with(inst["instance_hash"], my_topics)
|
|
except Exception as e:
|
|
self.fdb.record_sync_result(inst["instance_hash"], False)
|
|
logger.error("sync error with %s: %s", inst["instance_hash"][:16], e)
|
|
except Exception:
|
|
pass
|
|
prune_counter += 1
|
|
if prune_counter >= 6:
|
|
prune_counter = 0
|
|
try:
|
|
days = int(self.fdb.get_setting("forum_retention_days", "30"))
|
|
self.fdb.prune_old_content(days)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
for dead in self.fdb.get_dead_peers(MAX_PEER_FAILURES):
|
|
self.fdb.remove_synced_instance(dead)
|
|
except Exception:
|
|
pass
|
|
for _ in range(SYNC_INTERVAL):
|
|
if not self._running:
|
|
return
|
|
time.sleep(1)
|
|
|
|
def _sync_with(self, instance_hash, my_topics=None):
|
|
success = False
|
|
dest_hash = bytes.fromhex(instance_hash)
|
|
if not RNS.Transport.has_path(dest_hash):
|
|
RNS.Transport.request_path(dest_hash)
|
|
elapsed = 0
|
|
while not RNS.Transport.has_path(dest_hash) and elapsed < 15:
|
|
time.sleep(0.5)
|
|
elapsed += 0.5
|
|
if not RNS.Transport.has_path(dest_hash):
|
|
self.fdb.record_sync_result(instance_hash, False)
|
|
return
|
|
|
|
server_identity = RNS.Identity.recall(dest_hash)
|
|
if server_identity is None:
|
|
self.fdb.record_sync_result(instance_hash, False)
|
|
return
|
|
|
|
destination = RNS.Destination(
|
|
server_identity,
|
|
RNS.Destination.OUT,
|
|
RNS.Destination.SINGLE,
|
|
FORUM_APP,
|
|
)
|
|
|
|
link = RNS.Link(destination)
|
|
elapsed = 0
|
|
while link.status == RNS.Link.PENDING and elapsed < 15:
|
|
time.sleep(0.25)
|
|
elapsed += 0.25
|
|
|
|
if link.status != RNS.Link.ACTIVE:
|
|
self.fdb.record_sync_result(instance_hash, False)
|
|
return
|
|
|
|
try:
|
|
inst = self.fdb.get_synced_instances()
|
|
last_sync = ""
|
|
for s in inst:
|
|
if s["instance_hash"] == instance_hash:
|
|
last_sync = s["last_sync"] or ""
|
|
break
|
|
|
|
since = last_sync.replace(" ", "T") if last_sync else ""
|
|
my_hash = self.identity.hash.hex() if self.identity else "local"
|
|
|
|
if my_topics is None:
|
|
my_topics = self._get_subscribed_topics()
|
|
|
|
threads, posts, upvotes = [], [], []
|
|
|
|
# Only build content if we have a since timestamp (incremental sync)
|
|
if since:
|
|
if my_topics:
|
|
rows = self.fdb.get_threads_by_topics(my_topics, since=since)
|
|
threads = [dict(r) for r in rows]
|
|
tids = [r["id"] for r in rows]
|
|
posts = [dict(p) for p in self.fdb.get_posts_by_thread_ids(tids)]
|
|
uv_rows = self.fdb.get_new_upvotes_since(since, tids)
|
|
upvotes = [{"thread_id": tid, "instance_hash": instance_hash} for tid in uv_rows]
|
|
else:
|
|
ts, ps, uv = self.fdb.get_new_content(since)
|
|
threads = [dict(r) for r in ts]
|
|
posts = [dict(r) for r in ps]
|
|
upvotes = [{"thread_id": tid, "instance_hash": instance_hash} for tid in uv]
|
|
|
|
# Build content bloom for dedup
|
|
existing_ids = set(t["id"] for t in threads)
|
|
for t in self.fdb.get_threads_by_topics(my_topics) if my_topics else []:
|
|
existing_ids.add(t["id"])
|
|
content_bloom = BloomFilter.from_items(list(existing_ids), BLOOM_SIZE, BLOOM_HASHES)
|
|
|
|
my_blocks = [h.strip() for h in self.fdb.get_setting("blocked_instances", "").split(",") if h.strip()]
|
|
my_peer_blocks = self.fdb.get_peer_block_list()
|
|
|
|
retracted = [{"id": cid, "type": ct, "author": ai, "at": ra}
|
|
for cid, ct, ai, ra in self.fdb.get_raw_retractions()]
|
|
|
|
known_peers = [h for h in self.fdb.get_all_known_hashes() if h != instance_hash and h != my_hash]
|
|
|
|
request_data = {
|
|
"query": {"since": [since]} if since else {},
|
|
"threads": threads,
|
|
"posts": posts,
|
|
"upvotes": upvotes,
|
|
"from_hash": my_hash,
|
|
"blocks": {"mine": my_blocks, "peers": my_peer_blocks},
|
|
"retractions": retracted,
|
|
"known_peers": known_peers,
|
|
"my_topics": my_topics,
|
|
"content_bloom": list(content_bloom.bytes),
|
|
"bloom_size": BLOOM_SIZE,
|
|
"bloom_hashes": BLOOM_HASHES,
|
|
"trust_list": self.fdb.get_all_trusted_hashes(),
|
|
}
|
|
|
|
receipt = link.request("/forum", data=request_data, timeout=REQUEST_TIMEOUT)
|
|
elapsed = 0
|
|
done = (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED, RNS.RequestReceipt.FAILED)
|
|
while receipt.get_status() not in done and elapsed < REQUEST_TIMEOUT:
|
|
time.sleep(0.1)
|
|
elapsed += 0.1
|
|
|
|
if receipt.get_status() in (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED):
|
|
resp = receipt.get_response()
|
|
if isinstance(resp, dict) and resp.get("status") == 200:
|
|
try:
|
|
data = json.loads(resp["body"])
|
|
except (json.JSONDecodeError, KeyError):
|
|
data = {}
|
|
success = True
|
|
my_blocks = set(h.strip() for h in self.fdb.get_setting("blocked_instances", "").split(",") if h.strip())
|
|
|
|
# Merge trust gossip from peer
|
|
self._merge_trust_gossip(data.get("trust_list", []), instance_hash)
|
|
|
|
# Check content bloom for dedup
|
|
peer_bloom_data = data.get("content_bloom")
|
|
peer_bloom = None
|
|
if peer_bloom_data:
|
|
peer_bloom = BloomFilter.from_bytes(
|
|
bytes(peer_bloom_data),
|
|
data.get("bloom_size", BLOOM_SIZE),
|
|
data.get("bloom_hashes", BLOOM_HASHES),
|
|
)
|
|
|
|
for t in data.get("threads", []):
|
|
author = t.get("author_instance", "")
|
|
if author in my_blocks:
|
|
continue
|
|
if author and not self.fdb.is_trusted(author):
|
|
continue
|
|
if peer_bloom and peer_bloom.might_contain(t["id"]):
|
|
continue
|
|
if not my_topics:
|
|
self.fdb.merge_thread(t)
|
|
else:
|
|
t_tags = [tag.strip().lower() for tag in t.get("tags", "").split(",") if tag.strip()]
|
|
if set(my_topics) & set(t_tags):
|
|
self.fdb.merge_thread(t)
|
|
for p in data.get("posts", []):
|
|
author = p.get("author_instance", "")
|
|
if author in my_blocks:
|
|
continue
|
|
if author and not self.fdb.is_trusted(author):
|
|
continue
|
|
if peer_bloom and peer_bloom.might_contain(p["id"]):
|
|
continue
|
|
self.fdb.merge_post(p)
|
|
for tid in data.get("upvote_threads", []):
|
|
self.fdb.merge_upvote(tid, instance_hash)
|
|
|
|
peer_blocks = data.get("blocks", {})
|
|
for h in peer_blocks.get("mine", []):
|
|
if h and h not in my_blocks and instance_hash:
|
|
self.fdb.record_peer_block(instance_hash, h)
|
|
for h in peer_blocks.get("peers", []):
|
|
if h and h not in my_blocks and instance_hash:
|
|
self.fdb.record_peer_block(instance_hash, h)
|
|
self._apply_peer_blocks()
|
|
|
|
for r in data.get("retractions", []):
|
|
if r.get("id") and r.get("type") and r.get("author") and r.get("at"):
|
|
self.fdb.merge_retraction(r["id"], r["type"], r["author"], r["at"])
|
|
|
|
for peer_hash in data.get("known_peers", []):
|
|
if peer_hash and peer_hash != my_hash and peer_hash != instance_hash:
|
|
self.fdb.add_known_peer(peer_hash)
|
|
|
|
now = time.strftime("%Y-%m-%dT%H:%M:%S")
|
|
self.fdb.set_last_sync(instance_hash, now)
|
|
self.fdb.record_sync_result(instance_hash, success)
|
|
finally:
|
|
link.teardown()
|
|
|
|
def _apply_peer_blocks(self):
|
|
counts = self.fdb.get_peer_block_counts()
|
|
blocked = set(h.strip() for h in self.fdb.get_setting("blocked_instances", "").split(",") if h.strip())
|
|
auto_blocked = set(h.strip() for h in self.fdb.get_setting("auto_blocked_instances", "").split(",") if h.strip())
|
|
changed = False
|
|
for h, count in counts.items():
|
|
if h not in blocked and h not in auto_blocked and count >= 3:
|
|
blocked.add(h)
|
|
auto_blocked.add(h)
|
|
self.fdb.remove_by_hash(h)
|
|
changed = True
|
|
if changed:
|
|
self.fdb.set_setting("blocked_instances", ",".join(blocked))
|
|
self.fdb.set_setting("auto_blocked_instances", ",".join(auto_blocked))
|
|
|
|
def handle_sync(self, data):
|
|
return self.handlers_ref().handle_sync_request(data)
|