"""The coherence sweep: what every server should hold, show and count, from the plan and the ledger, against what each
one does hold (its database) and show (its API, as one account). Writes out/<run>/results.json.

Each check is a cell `feature|origin|observer|direction`:
  in       a peer's object as PrivaPub holds it        out      PrivaPub's object as a peer holds it
  via      a peer's object on another peer, through PrivaPub (a community it hosts)
  control  a peer's object on another peer, PrivaPub not involved (tells a PrivaPub bug from a peer quirk)
  local    an object on its own server
Known gaps (gaps.json) turn a failure into xfail, and a pass of a known gap into XPASS."""
import json
import os
import re
import sys
import time
from collections import defaultdict

import gen
import state
from core import podman
from dialects import driver

HERE = os.path.dirname(os.path.abspath(__file__))
HEARTS = {"❤", "❤️", "♥", "♥️"}
# what each platform keeps when it is sent something: Lemmy keeps only what lives in communities, and private messages
RECEIVES = {
	"lemmy": {"community", "direct"},
}
# Misskey and its forks drop a reply whose parent they cannot fetch ("Error in inReplyTo ... 404"): a reply to a
# followers-only post or a DM they do not hold never lands there, whoever sent it (docs/INTEROP.md, Misskey)
DROPS_ORPHAN_REPLIES = {"misskey", "sharkey"}
# Pixelfed is a photo platform: it keeps a post only when it carries a picture (a Note without one is dropped as it
# arrives), and a reply only to a post it holds
NEEDS_PICTURE = {"pixelfed"}
# Friendica shows a reply to an account only inside a thread that account holds, under a parent it holds, and counts on
# the thread's owner to relay the replies in it: a non-public reply never reaches an account that cannot see one of the
# posts above it, and a Friendica account's non-public reply in a thread another server owns reaches none of the
# others there (nobody relays a followers-only reply), nor anything under it
THREAD_BOUND = {"friendica"}


class World:
	"""The plan and the ledger together: accounts, follows that held, groups and their members, objects with URIs."""

	def __init__(self, run_dir):
		with open(os.path.join(run_dir, "spec.json")) as f:
			self.spec = json.load(f)
		snap = os.path.join(run_dir, "plan.json")
		if os.path.exists(snap):
			with open(snap) as f:
				self.planner = gen.Stored(json.load(f))
		else:
			self.planner = gen.Planner(self.spec)
			self.planner.plan()
		self.facts = []
		with open(os.path.join(run_dir, "ledger.jsonl")) as f:
			self.facts = [json.loads(line) for line in f if line.strip()]
		self.ok_steps = {f["n"] for f in self.facts if f.get("type") not in ("error", "unsupported") and "n" in f}
		self.failed = {f["n"]: f for f in self.facts if f.get("type") in ("error", "unsupported")}
		# a fact's fetched_on names the objects its server had to fetch first (a reply's parent, a quote), and an
		# interaction that arrived_by fetch made its actor's server fetch its object: both are kept per fetched object
		fetched = defaultdict(set)
		for f in self.facts:
			for platform, refs in (f.get("fetched_on") or {}).items():
				for ref in refs:
					fetched[ref].add(platform)
			if f.get("type") == "interaction" and f.get("arrived_by") == "fetch":
				fetched[f["ref"]].add(self.platform(f["actor"]))
		self.objects = {}
		for f in self.facts:
			if f.get("type") == "object":
				o = dict(self.planner.objects[f["ref"]])
				o.update(uri=f["uri"], origin=f["origin"], fetched_on=sorted(fetched[f["ref"]]))
				self.objects[f["ref"]] = o
		self.deleted = {f["ref"] for f in self.facts if f.get("type") == "mutation" and f["verb"] == "delete"}
		self.edited = {f["ref"] for f in self.facts if f.get("type") == "mutation" and f["verb"] == "edit"}
		self.follows = self._follows()
		self.members = self._members()
		self.interactions = [f for f in self.facts if f.get("type") == "interaction"]

	def _follows(self):
		"""A follow holds once the follow went out and the target either needs no approval or gave it. A follow of an
		unlocked remote account answers "requested" until its Accept arrives, so the plan's lock flag decides."""
		accepted = set()
		for f in self.facts:
			if f.get("type") != "relation":
				continue
			if f["verb"] == "follow" and (f["state"] == "accepted" or not self.planner.accounts[f["target"]]["locked"]):
				accepted.add((f["actor"], f["target"]))
			elif f["verb"] == "accept":
				accepted.add((f["target"], f["actor"]))
		edges = defaultdict(set)
		for follower, target in accepted:
			edges[target].add(follower)
		return edges

	def _members(self):
		members = defaultdict(set)
		joined = {}
		for f in self.facts:
			if f.get("type") != "relation":
				continue
			if f["verb"] == "join":
				grp = self.planner.groups[f["group"]]
				if grp["kind"] == "circle" and not f["actor"].startswith("privapub/"):
					joined[(f["group"], f["actor"])] = True  # a remote member counts once the owner approves
				else:
					members[f["group"]].add(f["actor"])
			elif f["verb"] == "approve":
				members[f["group"]].add(f["member"])
		for ref, grp in self.planner.groups.items():
			members[ref].add(grp["owner"])
		return members

	def platform(self, key):
		return key.split("/")[0]

	def recipients(self, o):
		"""The accounts an object was addressed to, as the ledger says the graph turned out."""
		v = o["visibility"]
		who = set(o.get("mentions") or ())
		if v in ("public", "unlisted", "followers"):
			who |= self.follows.get(o["author"], set())
			if o.get("parent") and o["parent"] in self.objects:
				who.add(self.objects[o["parent"]]["author"])
			# a quote reaches its quoted author when it is public or unlisted; a followers-only quote stays with the followers
			if o.get("quote") and o["quote"] in self.objects and v in ("public", "unlisted"):
				who.add(self.objects[o["quote"]]["author"])
		elif v == "circle":
			who |= self.members[o["group"]]
		elif v == "community" or (o.get("group") and self.planner.groups.get(o["group"], {}).get("kind") == "community"):
			grp = self.planner.groups[o["group"]]
			who |= self.members[o["group"]] | {grp["owner"]}
			if o.get("parent") and o["parent"] in self.objects:
				who.add(self.objects[o["parent"]]["author"])
		return who - {o["author"]}

	def kind_for(self, o):
		if o["visibility"] == "community" or (o.get("group") and self.planner.groups.get(o["group"], {}).get("kind") == "community"):
			return "community"
		return o["visibility"]


def direction(origin, observer, through_privapub=False):
	if origin == observer:
		return "local"
	if observer == "privapub":
		return "in"
	if origin == "privapub":
		return "out"
	return "via" if through_privapub else "control"


class Sweep:
	def __init__(self, world, platforms):
		self.w = world
		self.platforms = platforms
		self.checks = []

	def add(self, feature, origin, observer, ok, ref=None, expect=None, got=None, through=False, how=None):
		self.checks.append({"cell": f"{feature}|{origin}|{observer}|{direction(origin, observer, through)}",
							"feature": feature, "origin": origin, "observer": observer,
							"direction": direction(origin, observer, through), "ok": bool(ok), "ref": ref,
							"expect": expect, "got": got, "how": how})

	def stored_all(self):
		uris = [o["uri"] for o in self.w.objects.values()]
		out = {}
		for p in self.platforms:
			try:
				out[p] = driver(p).stored(uris)
			except Exception as e:
				print(f"  {p}: cannot read its database: {e!r}", file=sys.stderr)
				out[p] = {}
		return out

	def run(self, deadline):
		started = time.time()
		while True:
			stored = self.stored_all()
			self.checks = []
			self.delivery(stored)
			self.counts(stored)
			self.threads(stored)
			self.lifecycle(stored)
			failing = [c for c in self.checks if not c["ok"] and c["feature"].startswith(("deliver", "count", "edit", "delete", "thread"))]
			if not failing or time.time() - started > deadline:
				break
			print(f"  {len(failing)} checks still failing; sweeping again in 20s", flush=True)
			time.sleep(20)
		self.visibility(stored)
		self.graph()
		self.privacy(stored)
		return self.checks

	# -- who holds what
	def delivery(self, stored):
		for ref, o in self.w.objects.items():
			if ref in self.w.deleted:
				continue
			kind = self.w.kind_for(o)
			recipients = self.w.recipients(o)
			want = {self.w.platform(k) for k in recipients} - {o["origin"]}
			through = kind == "community" and self.w.planner.groups[o["group"]]["host"] == "privapub"
			for p in self.platforms:
				if p == o["origin"]:
					continue
				row = stored.get(p, {}).get(o["uri"])
				held = bool(row and row.exists and not row.deleted)
				receives = RECEIVES.get(p)
				parent = self.w.objects.get(o.get("parent") or "")
				orphan = p in DROPS_ORPHAN_REPLIES and parent is not None \
					and not (stored.get(p, {}).get(parent["uri"]) and stored[p][parent["uri"]].exists) \
					and parent["visibility"] not in ("public", "unlisted")
				if orphan:
					continue
				if p in NEEDS_PICTURE and (o.get("kind") not in ("image", "video") if parent is None
										   else not (stored.get(p, {}).get(parent["uri"]) and stored[p][parent["uri"]].exists)):
					continue
				if p in want and (receives is None or kind in receives):
					fetched = p in o.get("fetched_on", {})
					# whether a follower of the author lives on p, or only the accounts it mentions or answers
					followed = any(self.w.platform(k) == p for k in self.w.follows.get(o["author"], ()))
					how = "fetched by the seeder" if fetched else None if followed or kind in ("circle", "community") else "addressed only"
					self.add(f"deliver.{kind}", o["origin"], p, held, ref, "held", "held" if held else "missing", through, how=how)
				elif kind in ("followers", "direct", "circle", "located") and p not in want:
					self.add(f"confine.{kind}", o["origin"], p, not held, ref, "absent", "held" if held else "absent", through)

	# -- counts on the object's own server, and PrivaPub's exact counts
	def counts(self, stored):
		likes, reacts, boosts, votes = defaultdict(int), defaultdict(lambda: defaultdict(int)), defaultdict(int), defaultdict(lambda: defaultdict(int))
		for f in self.w.interactions:
			ref = f["ref"]
			if f["verb"] == "like":
				likes[ref] += 1
			elif f["verb"] == "react":
				# Misskey and its forks send a reaction as a Like carrying the emoji, and PrivaPub keeps a heart there as a
				# like (Reactions.IsHeart); an EmojiReact keeps its heart as a reaction, written ❤️
				if f["emoji"] in HEARTS and self.w.platform(f["actor"]) in ("misskey", "sharkey"):
					likes[ref] += 1
				else:
					reacts[ref]["❤️" if f["emoji"] in HEARTS else f["emoji"]] += 1
			elif f["verb"] == "boost":
				boosts[ref] += 1
			elif f["verb"] == "vote":
				for c in f["choices"]:
					votes[ref][c] += 1
		for ref, o in self.w.objects.items():
			if ref in self.w.deleted:
				continue
			origin = o["origin"]
			row = stored.get(origin, {}).get(o["uri"])
			if not row or not row.exists:
				continue
			v = self.w.kind_for(o)
			if origin == "privapub":
				self.add(f"count.like.{v}", "privapub", "privapub", row.likes == likes[ref], ref, likes[ref], row.likes)
				want_reacts = {e: n for e, n in reacts[ref].items()}
				got_reacts = {e: n for e, n in (row.reactions or {}).items()}
				if want_reacts or got_reacts:
					self.add(f"count.react.{v}", "privapub", "privapub", got_reacts == want_reacts, ref, want_reacts, got_reacts)
				if v in ("public", "unlisted"):
					self.add(f"count.boost.{v}", "privapub", "privapub", row.boosts == boosts[ref], ref, boosts[ref], row.boosts)
				if votes[ref] or row.votes:
					want = [votes[ref].get(i, 0) for i in range(len(row.votes or []))]
					self.add(f"count.vote.{v}", "privapub", "privapub", row.votes == want, ref, want, row.votes)
			elif origin != "lemmy":
				# elsewhere: at least the plain likes that every platform sends as Like
				plain = sum(1 for f in self.w.interactions if f["ref"] == ref and f["verb"] == "like")
				if plain:
					total = (row.likes or 0)
					self.add(f"count.like.{v}", "*", origin, total >= plain, ref, f">={plain}", total)
				if boosts[ref]:
					self.add(f"count.boost.{v}", "*", origin, (row.boosts or 0) >= boosts[ref], ref, f">={boosts[ref]}", row.boosts)
				if votes[ref]:
					want = sum(votes[ref].values())
					got = sum(row.votes or [])
					self.add(f"count.vote.{v}", "*", origin, got >= want, ref, f">={want}", got)

	# -- threads: each copy of a reply names its parent
	def threads(self, stored):
		for ref, o in self.w.objects.items():
			if not o.get("parent") or ref in self.w.deleted or o["parent"] not in self.w.objects:
				continue
			parent_uri = self.w.objects[o["parent"]]["uri"]
			for p in self.platforms:
				row = stored.get(p, {}).get(o["uri"])
				if not row or not row.exists or row.deleted or p == "lemmy":
					continue
				# a server links a reply to a parent it holds; one it never received (a followers-only post of
				# someone nobody there follows) leaves the reply unlinked, rightly
				parent = stored.get(p, {}).get(parent_uri)
				if not parent or not parent.exists:
					continue
				self.add("thread.parent", o["origin"], p, row.parent_uri == parent_uri, ref, parent_uri, row.parent_uri)

	def audience_servers(self, o):
		return {self.w.platform(k) for k in self.w.recipients(o)} | {o["origin"]}

	# -- edits and deletes reach every copy in the object's audience (a server holding it through a boost or a fetch
	# is sent no Update: Mastodon and the others address edits to the post's own audience only)
	def lifecycle(self, stored):
		for ref in self.w.edited - self.w.deleted:
			o = self.w.objects.get(ref)
			if not o:
				continue
			audience = self.audience_servers(o)
			for p in self.platforms:
				row = stored.get(p, {}).get(o["uri"])
				if not row or not row.exists or row.deleted or p in ("misskey",) or p not in audience:
					continue
				text = re.sub("<[^>]+>", "", row.text or "")
				self.add("edit.text", o["origin"], p, "(edited)" in text, ref, "(edited)", text[-60:])
		for ref in self.w.deleted:
			o = self.w.objects.get(ref)
			if not o:
				continue
			audience = self.audience_servers(o)
			for p in self.platforms:
				if p not in audience:
					continue
				row = stored.get(p, {}).get(o["uri"])
				gone = not row or not row.exists or row.deleted
				self.add("delete.gone", o["origin"], p, gone, ref, "gone", "gone" if gone else "held")

	# -- what accounts see through their APIs
	def visibility(self, stored):
		sessions = {f"{s.account.platform}/{s.account.username}": s for s in state.sessions()}
		by_platform = defaultdict(list)
		for ref, o in self.w.objects.items():
			if ref in self.w.deleted or o["visibility"] in ("public", "unlisted", "community"):
				continue
			recipients = self.w.recipients(o)
			for p in self.platforms:
				if p == "lemmy":
					continue
				row = stored.get(p, {}).get(o["uri"])
				if not row or not row.exists or row.deleted:
					continue
				here = [k for k in self.w.planner.accounts if self.w.platform(k) == p and k != o["author"]]
				inside = sorted(k for k in here if k in recipients and self.in_thread(p, o, k))[:2]
				outside = sorted(k for k in here if k not in recipients)[:2]
				for k in inside:
					by_platform[p].append((ref, o, k, True))
				for k in outside:
					by_platform[p].append((ref, o, k, False))
		for p, items in by_platform.items():
			d = driver(p)
			for ref, o, k, should in items:
				s = sessions.get(k)
				if s is None:
					continue
				try:
					seen = d.seen(s, [o["uri"]])[o["uri"]]
				except Exception as e:
					seen = None
				feature = f"{'see' if should else 'hide'}.{o['visibility']}"
				self.add(feature, o["origin"], p, seen is should, ref, should, seen, how=k)

	def in_thread(self, p, o, k):
		"""Whether a reply can reach account k on platform p, where a platform shows replies only in threads (THREAD_BOUND)."""
		if p not in THREAD_BOUND or not o.get("parent"):
			return True
		chain, node = [], o
		while node is not None:
			chain.append(node)
			node = self.w.objects.get(node.get("parent") or "")
		root, shown = chain[-1], ("public", "unlisted", "community")
		if any(a["visibility"] not in shown and k != a["author"] and k not in self.w.recipients(a) for a in chain[1:]):
			return False
		return not any(a["origin"] == p and root["origin"] != p and a["visibility"] not in shown and k != a["author"] for a in chain[:-1])

	# -- the graph as each side reports it
	def graph(self):
		sessions = {f"{s.account.platform}/{s.account.username}": s for s in state.sessions()}
		for target, followers in self.w.follows.items():
			for follower in followers:
				fp = self.w.platform(follower)
				s = sessions.get(follower)
				if s is None:
					continue
				try:
					rel = driver(fp).relationship(s, gen.Planner.acct(target))
					ok = rel.get("following")
				except Exception as e:
					ok, rel = False, repr(e)[:100]
				self.add("graph.following", self.w.platform(target), fp, ok, None, True, ok, how=f"{follower} -> {target}")

	# -- privacy rows
	def privacy(self, stored):
		personas = [k for k in self.w.planner.accounts if k.startswith("privapub/")]
		# PV1 sibling keys differ, PV2 published is a whole day at most two weeks before now
		from core.http import client
		http = client()
		keys, published = {}, {}
		for k in personas:
			name = k.split("/")[1]
			r = http.get(f"https://privapub.test/peasants/{name}", headers={"Accept": "application/activity+json"})
			doc = r.json() or {}
			keys[k] = (doc.get("publicKey") or {}).get("publicKeyPem")
			published[k] = doc.get("published")
		roots = defaultdict(list)
		for k in personas:
			roots[self.w.planner.accounts[k]["root"]].append(k)
		for root, sibs in roots.items():
			pems = [keys[k] for k in sibs]
			self.add("privacy.sibling-keys", "privapub", "privapub", all(pems) and len(set(pems)) == len(pems), None,
					 "distinct", f"{len(set(pems))} of {len(pems)}")
			for k in sibs:
				p = published[k] or ""
				self.add("privacy.published-day", "privapub", "privapub", p.endswith("T00:00:00Z"), None, "midnight", p, how=k)
		# PV4 canaries: no root login or password anywhere on a peer
		canaries = {a["root"] for a in self.w.planner.accounts.values() if a.get("root")} | \
				   {a["password"] for a in self.w.planner.accounts.values() if a.get("root")}
		for p in self.platforms:
			if p == "privapub":
				continue
			found = canary_hits(p, canaries)
			self.add("privacy.canary", "privapub", p, not found, None, "no root login or password", found or "none")
		# PV7 located posts never left
		for ref, o in self.w.objects.items():
			if o["visibility"] != "located":
				continue
			held = [p for p in self.platforms if p != "privapub" and stored.get(p, {}).get(o["uri"]) and stored[p][o["uri"]].exists]
			jobs = podman.mongo(f"db.Job.countDocuments({{Kind: 0, Payload: /{re.escape(o['uri'].rsplit('/', 1)[1])}/}})") or 0
			self.add("privacy.located-stays", "privapub", "*", not held and not jobs, ref, "nowhere, no delivery",
					 {"held": held, "jobs": jobs})


def canary_hits(platform, canaries):
	"""Which canaries a peer's database holds anywhere (a full dump, searched as text)."""
	db = {"mastodon": "mastodon", "misskey": "misskey", "sharkey": "sharkey", "akkoma": "akkoma", "lemmy": "lemmy"}.get(platform)
	if db:
		dump = podman.run("exec", "pasture-postgres", "pg_dump", "-U", "pasture", "--data-only", db, timeout=600)
	elif platform == "gts":
		dump = podman.run("exec", "pasture-gts", "sh", "-c", "cat /gotosocial/storage/sqlite.db /gotosocial/storage/sqlite.db-wal 2>/dev/null | strings", check=False)
	else:
		return []
	return sorted(c for c in canaries if c in dump)


def latest_client(lines):
	"""decePub's e2e cells (out/client.jsonl, appended by every run) as each cell's latest run left them: the rows from
	within an hour of the cell's newest, the newest per test. A run that picked other accounts leaves no stale failure."""
	from datetime import datetime, timedelta
	at = lambda c: datetime.fromisoformat(c["t"].replace("Z", "+00:00"))
	newest = {}
	for c in lines:
		newest[c["cell"]] = max(newest.get(c["cell"], at(c)), at(c))
	latest = {}
	for c in sorted(lines, key=at):
		if at(c) >= newest[c["cell"]] - timedelta(hours=1):
			latest[(c["cell"], c.get("ref"))] = c
	return list(latest.values())


def load_gaps():
	try:
		with open(os.path.join(HERE, "gaps.json")) as f:
			return json.load(f)
	except FileNotFoundError:
		return []


def classify(checks, gaps):
	for c in checks:
		gap = next((g for g in gaps if g.get("status") != "closed" and _matches(g["match"], c)), None)
		if gap:
			c["gap"] = gap["id"]
			c["status"] = "xpass" if c["ok"] else "xfail"
		else:
			c["status"] = "pass" if c["ok"] else "fail"
	return checks


def _matches(match, c):
	for k, pattern in match.items():
		if not re.fullmatch(pattern, str(c.get(k, ""))):
			return False
	return True


def summarise(checks):
	cells = {}
	for c in checks:
		cell = cells.setdefault(c["cell"], {"n": 0, "pass": 0, "fail": 0, "xfail": 0, "xpass": 0, "examples": []})
		cell["n"] += 1
		cell[c["status"]] += 1
		if c["status"] in ("fail", "xpass") and len(cell["examples"]) < 3:
			cell["examples"].append({k: c[k] for k in ("ref", "expect", "got", "how") if c.get(k) is not None})
	return cells


def main(args):
	import seed
	name = next((a for a in args if not a.startswith("--")), "village")
	run_dir = seed.run_dir(name)
	world = World(run_dir)
	platforms = [p for p in gen.HOSTS if p in world.spec["peers"]]
	deadline = world.spec.get("time", {}).get("deadlineSeconds", 240)
	for a in args:
		if a.startswith("--deadline="):
			deadline = int(a.split("=", 1)[1])
	print(f"checking {run_dir}: {len(world.objects)} objects on {len(platforms)} servers")
	checks = classify(Sweep(world, platforms).run(deadline), load_gaps())
	cells = summarise(checks)
	from collections import Counter
	totals = Counter(c["status"] for c in checks)
	result = {"run": run_dir, "when": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "totals": totals,
			  "checks": checks, "cells": cells}
	with open(os.path.join(run_dir, "results.json"), "w") as f:
		json.dump(result, f, indent=1, default=str, ensure_ascii=False)
	print(json.dumps(totals))
	failing = sorted((c, v) for c, v in cells.items() if v["fail"])
	for cell, v in failing[:60]:
		print(f"  FAIL {cell}: {v['fail']}/{v['n']} e.g. {v['examples'][:1]}")
	strict = "--strict" in args
	return 1 if totals.get("fail") or (strict and totals.get("xpass")) else 0


if __name__ == "__main__":
	sys.exit(main(sys.argv[1:]))