(ns ird.p2p.db.replication-test
  "Two real Helia/OrbitDB nodes over loopback tcp, always on: the empirical
  proof that the pinned helia + @orbitdb/core + @libp2p/gossipsub set
  replicates sealed records between peers through native OrbitDB sync —
  history caught up on join (the heads exchange), live records over
  gossipsub, both directions — and that only ciphertext is ever stored or
  sent. If your environment cannot bind loopback tcp, set
  IRD_P2P_DB_NO_NODE=1 to skip."
  (:require [clojure.string :as str]
            [cljs.test :refer [deftest is testing]]
            [cljs.core.async :as async :refer [<!]]
            [ird.p2p.db.core :as db]
            [ird.p2p.db.errors :as err]
            [ird.p2p.db.test-util :as tu :refer [deftest-async]]
            [ird.p2p.impl.bytes :as bytes]))

(defn- latin1 [b64] (.toString (bytes/b64->buffer b64) "latin1"))

(deftest-async two-nodes-replicate-sealed-records
  {:timeout-ms 60000}
  (if (tu/skip-nodes?)
    (is true "IRD_P2P_DB_NO_NODE set — skipping the real two-node test")
    (let [secret (db/random-secret)
          a (<! (db/open {:secret secret :app-salt "test/v1" :name "chat"}))]
      (is (not (tu/error? a)) (some-> a ex-message))
      (testing "the node is a private swarm"
        (is (= ["libp2p-router"] (:routers (:node a))) "no delegated HTTP routing")
        (is (= #{"identify" "pubsub"} (set (:services (:node a)))) "no dht, no relay server, no autonat, no upnp")
        (is (some #{"bitswap"} (:block-brokers (:node a)))))
      ;; A appends before B exists: B must catch up through the heads exchange
      (let [r1 (<! (db/append a {:text "unu" :n 1}))
            r2 (<! (db/append a {:text "doi" :n 2}))
            _ (is (and (:hash r1) (:hash r2)))
            _ (is (= {:text "unu" :n 1} (:payload r1)))
            _ (is (= 1 (get-in r1 [:clock :time])))
            _ (is (= 2 (get-in r2 [:clock :time])))
            b (<! (db/open {:secret secret :app-salt "test/v1" :name "chat" :dial (<! (db/multiaddrs a))}))
            _ (is (not (tu/error? b)) (some-> b ex-message))
            ca (tu/collector (db/entries-stream a))
            cb (tu/collector (db/entries-stream b))
            pa (tu/collector (db/peers-stream a))
            pb (tu/collector (db/peers-stream b))
            wire (tu/collector ((:sniff! (:node b))))]
        (is (= (db/address a) (db/address b)) "both peers derive the same address from name + identity")
        (is (str/starts-with? (db/address a) "/orbitdb/"))
        (is (= 43 (count (:manifest-name a))))
        (is (= (:manifest-name a) (:name (:db a))) "OrbitDB's manifest carries the opaque derived name")
        (is (not= "chat" (:name (:db a))) "…never the human one")
        (is (= (:identity a) (:identity b)))
        (is (not= (db/peer-id a) (db/peer-id b)))
        (is (some? (<! (tu/wait-for pb #(= :peer-joined (:type %)) 20000))) "B exchanged heads with A")
        (is (some? (<! (tu/wait-for pa #(and (= :peer-joined (:type %)) (= (db/peer-id b) (:peer %))) 20000))))
        (is (some? (<! (tu/wait-for cb #(= 2 (get-in % [:payload :n])) 20000))) "B caught up with A's history")
        (is (= [{:text "unu" :n 1} {:text "doi" :n 2}] (mapv :payload @cb)) "in canonical order")
        (is (= [(:hash r1) (:hash r2)] (mapv :hash @cb)))
        (is (= [1 2] (mapv #(get-in % [:clock :time]) @cb)))
        (is (every? #(= (:by r1) (:by %)) @cb) "one shared writer identity")
        (is (= [1 2] (mapv #(get-in % [:payload :n]) @ca)) "A's own appends are on its stream too")
        (is (some #{(db/peer-id b)} (db/peers a)))
        ;; live: A appends after the join settled — B receives it over gossipsub
        (let [marker "plaintext-marker-3"
              r3 (<! (db/append a {:text "trei" :n 3 :marker marker}))]
          (is (some? (<! (tu/wait-for cb #(= 3 (get-in % [:payload :n])) 20000))) "B received the live record")
          (is (= (:hash r3) (:hash (last @cb))))
          (is (= {:text "trei" :n 3 :marker marker} (:payload (last @cb))))
          ;; and the other direction
          (let [r4 (<! (db/append b {:text "patru" :n 4}))]
            (is (not (tu/error? r4)) (some-> r4 ex-message))
            (is (some? (<! (tu/wait-for ca #(= 4 (get-in % [:payload :n])) 20000))) "A received B's record"))
          (testing "snapshots agree, in the same canonical order"
            (let [ea (<! (db/entries a)) eb (<! (db/entries b))]
              (is (= 4 (count ea) (count eb)))
              (is (= (mapv :hash ea) (mapv :hash eb)))
              (is (= [1 2 3 4] (mapv #(get-in % [:payload :n]) eb)))
              (is (= ea eb))))
          (testing "ciphertext only: the bytes OrbitDB stores, hashes, signs and sends"
            (let [sealed (:payload (<! ((:get (:db a)) (:hash r3))))
                  stored (<! ((:entry-bytes (:db b)) (:hash r3)))]
              (is (bytes/valid-b64? sealed) "the stored value is the sealed record")
              (is (= 2 (bytes/byte-at sealed 0)))
              (is (some? stored) "B holds the entry bytes")
              (let [s (latin1 stored)]
                (is (str/includes? s sealed) "the entry bytes carry the sealed record")
                (is (not (str/includes? s marker)) "…and no plaintext")
                (is (not (str/includes? s "trei")))
                (is (not (str/includes? s "\"text\""))))
              (is (some? (<! (tu/wait-for wire #(str/includes? (latin1 (:data %)) sealed) 10000)))
                  "the live record crossed gossipsub as the sealed bytes")
              (is (every? #(= (db/address a) (:topic %)) @wire) "the pubsub topic is the database address")
              (is (not-any? #(let [s (latin1 (:data %))] (or (str/includes? s marker) (str/includes? s "\"text\""))) @wire)
                  "no plaintext ever crossed the wire"))))
        (<! (db/close a))
        (<! (db/close b))
        (is (= :closed @(:state a)))
        (is (= :closed @(:state b)))))))