(ns ird.p2p.mailbox
  "The offline mailbox of one room: deposits are sealed under the room-wide
  mailbox key (random IVs, the 0x02 at-rest envelope) and filed under a
  mailbox room id the node cannot correlate with the relay room. Peers who
  come later replay them; undecryptable entries are dropped silently.

  The raw client (`deposit-raw`, `replay-raw`, `replay-all-raw`) speaks the
  node's `/mailbox/v1` HTTP API with a bearer token from an injectable
  :token-fn — by default the node's open mint, which needs an allowed
  Origin; with the open mint off, inject one that asks your own backend.
  Sequence numbers (:seq) are opaque ordered tokens, compared with BigInt,
  never parsed as doubles.

  `enqueue` + `flush` queue deposits with retry and backoff
  (min(1000·2^n, 300000) ms); 413 drops the item, 401/403 drops it after
  the client's own single re-mint."
  (:require [clojure.string :as str]
            [cljs.core.async :as async :refer [go <!]]
            [ird.p2p.async :as a :refer [go-try <?]]
            [ird.p2p.envelope :as env]
            [ird.p2p.errors :as err]
            [ird.p2p.impl.bytes :as bytes]
            [ird.p2p.impl.http :as http]
            [ird.p2p.impl.json :as json]
            [ird.p2p.impl.node :as node]))

(def max-limit 500)

;; ---- wire helpers -------------------------------------------------------------------

(defn encode-segment
  "Percent-encode ROOM as ONE path segment (slashes included): only
  unreserved characters survive."
  [s]
  (str/replace s #"[^A-Za-z0-9\-._~]"
               (fn [c]
                 (str/join (map #(str "%" (str/upper-case (bytes/int->hex % 2)))
                                (bytes/b64->bytes (bytes/utf8->b64 c)))))))

(defn messages-url [endpoint room]
  (str (str/replace (str endpoint) #"/+$" "") "/rooms/" (encode-segment room) "/messages"))

(defn- query-string
  "Build a query string from ordered PAIRS ([key value] ...), skipping nil
  values, so the wire order is deterministic."
  [pairs]
  (let [parts (for [[k v] pairs :when (some? v)] (str (name k) "=" (encode-segment (str v))))]
    (when (seq parts) (str "?" (str/join "&" parts)))))

(defn validate-room [room]
  (when (str/blank? (str room)) (throw (err/config-error "mailbox room id is blank" {:room room})))
  (when (> (bytes/utf8-byte-length (str room)) 4096)
    (throw (err/config-error "mailbox room id exceeds 4096 bytes" {:bytes (bytes/utf8-byte-length (str room))})))
  room)

(defn- http-error [{:keys [status body]}]
  (let [parsed (json/parse body)
        code (when (and (map? parsed) (string? (:error parsed))) (:error parsed))]
    (err/http-error (or code (str "HTTP " status)) status code (when (map? parsed) parsed))))

(defn seq<
  "True when sequence token A is older than B (BigInt compare)."
  [a b]
  (bytes/seq< a b))

(defn history-gap?
  "True when CURSOR — the :seq you last saw — is older than PAGE's
  :oldest_seq, meaning messages were evicted between your visits. A
  nil/blank cursor (a first visit) is never a gap."
  [cursor page]
  (boolean
   (when-not (str/blank? (str cursor))
     (when-let [oldest (:oldest_seq page)]
       (seq< cursor oldest)))))

;; ---- tokens ----------------------------------------------------------------------------

(def renew-before-sec 300)

(defn open-mint-token-fn
  "A :token-fn minting at the node's open mint (`GET <mint>?room=&ttl=`)
  with the configured Origin. App-scoped by default so one mint covers
  every room; renewed five minutes before expiry; one in-flight mint at a
  time. OPTS: :mint-url :origin :http-fn :now-fn :ttl-sec (default 86400)
  :scope-to-room? :timeout-ms."
  [{:keys [mint-url origin http-fn now-fn ttl-sec scope-to-room? timeout-ms]}]
  (let [http (or http-fn http/fetch-http)
        now (or now-fn node/now-ms)
        ttl (or ttl-sec 86400)
        cache (atom {})
        inflight (atom {})
        mint (fn [room]
               (go-try
                 (let [resp (<! (http {:method :get
                                       :url (str mint-url (query-string [[:room room] [:ttl ttl]]))
                                       :headers (cond-> {"Accept" "application/json"} origin (assoc "Origin" origin))
                                       :timeout-ms (or timeout-ms 15000)}))]
                   (when (a/error? resp) (throw (err/transport-error (str "mint failed: " (ex-message resp)) {})))
                   (when-not (<= 200 (:status resp) 299) (throw (http-error resp)))
                   (let [m (json/parse (:body resp))]
                     (when-not (and (map? m) (string? (:token m)) (number? (:expires_at m)))
                       (throw (err/transport-error "mint response not decodable" {:body (:body resp)})))
                     (select-keys m [:token :expires_at :ttl :endpoint :room])))))]
    (fn [{:keys [room fresh?]}]
      (let [k (if scope-to-room? room "")]
        (go-try
          (when fresh? (swap! cache dissoc k))
          (let [t (get @cache k)]
            (if (and t (> (- (:expires_at t) (/ (now) 1000)) renew-before-sec))
              t
              (if-let [running (get @inflight k)]
                (<? running)
                (let [ch (mint (when scope-to-room? room))]
                  (swap! inflight assoc k ch)
                  (let [t (<! ch)]
                    (swap! inflight dissoc k)
                    (when (a/error? t) (throw t))
                    (swap! cache assoc k t)
                    t))))))))))

(defn static-token-fn
  "A :token-fn handing out one fixed TOKEN ({:token ... :expires_at ...})
  or, given a function of the room, whatever it fetches (a channel of a
  token map)."
  [token-or-fn]
  (if (fn? token-or-fn)
    (fn [{:keys [room]}] (token-or-fn room))
    (fn [_] (go token-or-fn))))

;; ---- raw client ----------------------------------------------------------------------------

(defn make-client
  "OPTS: :endpoint (the mailbox base URL), :token-fn, :http-fn, :origin, :timeout-ms."
  [{:keys [endpoint token-fn http-fn origin timeout-ms]}]
  (when (str/blank? (str endpoint)) (throw (err/config-error "mailbox endpoint is blank" {})))
  (when-not (fn? token-fn) (throw (err/config-error "mailbox needs a :token-fn" {})))
  {:endpoint endpoint :token-fn token-fn :http-fn (or http-fn http/fetch-http) :origin origin
   :timeout-ms (or timeout-ms 20000)})

(defn- authorized
  "Run a request with a bearer token; on 401/403 re-mint once and retry."
  [{:keys [token-fn http-fn]} room build]
  (go-try
    (let [send (fn [fresh?]
                 (go-try
                   (let [t (<? (token-fn {:room room :fresh? fresh?}))
                         resp (<! (http-fn (build (:token t))))]
                     (when (a/error? resp) (throw (err/transport-error (str "Request failed: " (ex-message resp)) {})))
                     resp)))
          resp (<? (send false))
          resp (if (contains? #{401 403} (:status resp)) (<? (send true)) resp)]
      (when-not (<= 200 (:status resp) 299) (throw (http-error resp)))
      resp)))

(defn deposit-raw
  "`POST /rooms/<room>/messages` with PAYLOAD-B64 as raw octets. Channel of
  {:seq :id :expires_at :deduped} — :deduped true when the node already
  held this exact payload for the room."
  [{:keys [endpoint origin timeout-ms] :as client} room payload-b64]
  (go-try
    (validate-room room)
    (when (or (not (bytes/valid-b64? payload-b64)) (zero? (bytes/byte-length payload-b64)))
      (throw (err/config-error "mailbox payload is empty" {})))
    (let [resp (<? (authorized client room
                               (fn [token]
                                 {:method :post :url (messages-url endpoint room)
                                  :headers (cond-> {"Accept" "application/json"
                                                    "Authorization" (str "Bearer " token)
                                                    "Content-Type" "application/octet-stream"}
                                             origin (assoc "Origin" origin))
                                  :body {:b64 payload-b64}
                                  :timeout-ms timeout-ms})))
          d (json/parse (:body resp))]
      (when-not (and (map? d) (contains? d :seq)) (throw (err/transport-error "deposit response not decodable" {:body (:body resp)})))
      {:seq (str (:seq d)) :id (:id d) :expires_at (:expires_at d)
       :deduped (if (boolean? (:deduped d)) (:deduped d) (= 200 (:status resp)))})))

(defn replay-raw
  "`GET /rooms/<room>/messages?after=|before=&limit=` (`before` wins when both
  are given). Channel of {:messages [{:seq :id :ts :payload <b64>}] :has_more
  :oldest_seq :latest_seq}."
  ([client room] (replay-raw client room nil))
  ([{:keys [endpoint origin timeout-ms] :as client} room {:keys [after before limit]}]
   (go-try
     (validate-room room)
     (let [limit (or limit 100)]
       (when-not (pos? limit) (throw (err/config-error "limit must be positive" {:limit limit})))
       (let [q (query-string [[:before before]
                              [:after (when-not before after)]
                              [:limit (min limit max-limit)]])
             resp (<? (authorized client room
                                  (fn [token]
                                    {:method :get :url (str (messages-url endpoint room) q)
                                     :headers (cond-> {"Accept" "application/json" "Authorization" (str "Bearer " token)}
                                                origin (assoc "Origin" origin))
                                     :timeout-ms timeout-ms})))
             rr (json/parse (:body resp))]
         (when-not (and (map? rr) (vector? (:messages rr))) (throw (err/transport-error "replay response not decodable" {:body (:body resp)})))
         {:messages (mapv (fn [m]
                            (when-not (bytes/valid-b64? (:payload m)) (throw (err/transport-error "replay payload is not base64" {})))
                            {:seq (str (:seq m)) :id (:id m) :ts (:ts m) :payload (:payload m)})
                          (:messages rr))
          :has_more (boolean (:has_more rr))
          :oldest_seq (some-> (:oldest_seq rr) str)
          :latest_seq (some-> (:latest_seq rr) str)})))))

(defn replay-all-raw
  "Page forward from :after until the end. Channel of {:messages :cursor
  :gap?} — :gap? true when the cursor is older than what the node still
  holds (history was evicted in between)."
  ([client room] (replay-all-raw client room nil))
  ([client room {:keys [after limit]}]
   (go-try
     (loop [cursor after acc [] gap nil]
       (let [page (<? (replay-raw client room {:after cursor :limit (or limit max-limit)}))
             msgs (:messages page)
             gap (if (nil? gap) (history-gap? after page) gap)
             acc (into acc msgs)
             cursor (or (:seq (last msgs)) cursor)]
         (if (and (:has_more page) (seq msgs))
           (recur cursor acc gap)
           {:messages acc :cursor cursor :gap? (boolean gap)}))))))

;; ---- outbox ---------------------------------------------------------------------------------

(defn retry-delay
  "Mailbox retry: min(300000, 1000·2^attempt) milliseconds."
  [attempt]
  (min 300000 (* 1000 (apply * (repeat (max 0 attempt) 2)))))

(defn make-outbox
  "A best-effort retry queue in front of `deposit-raw`. OPTS: :client :sleep-fn."
  [{:keys [client sleep-fn]}]
  {:client client :sleep-fn (or sleep-fn a/sleep) :queue (atom []) :flushing? (atom false)})

(defn outbox-enqueue [outbox room payload-b64]
  (swap! (:queue outbox) conj {:room room :payload payload-b64 :attempts 0})
  nil)

(defn outbox-pending [outbox] @(:queue outbox))

(defn outbox-restore
  "Append previously snapshotted ITEMS to the queue."
  [outbox items]
  (swap! (:queue outbox) into items)
  nil)

(defn outbox-flush
  "Deliver the queue in order. Channel of the vector of outcomes
  ({:outcome :deposited :result r} / {:outcome :dropped :item i :reason s} /
  {:outcome :deferred :item i :retry-after-ms n}), delivered when every item
  is deposited or dropped."
  [{:keys [client sleep-fn queue flushing?]}]
  (go
    (if @flushing?
      []
      (do
        (reset! flushing? true)
        (let [outcomes (atom [])]
          (loop []
            (when-let [item (first @queue)]
              (let [r (<! (deposit-raw client (:room item) (:payload item)))]
                (cond
                  (not (a/error? r))
                  (do (swap! queue subvec 1) (swap! outcomes conj {:outcome :deposited :result r}) (recur))

                  (and (err/http-error? r) (= 413 (:status (ex-data r))))
                  (do (swap! queue subvec 1) (swap! outcomes conj {:outcome :dropped :item item :reason "message_too_large"}) (recur))

                  (and (err/http-error? r) (contains? #{401 403} (:status (ex-data r))))
                  (do (swap! queue subvec 1) (swap! outcomes conj {:outcome :dropped :item item :reason "unauthorized"}) (recur))

                  (err/config-error? r)
                  (do (swap! queue subvec 1) (swap! outcomes conj {:outcome :dropped :item item :reason (ex-message r)}) (recur))

                  :else
                  (let [delay (retry-delay (:attempts item))
                        item (update item :attempts inc)]
                    (swap! queue assoc 0 item)
                    (swap! outcomes conj {:outcome :deferred :item item :retry-after-ms delay})
                    (<! (sleep-fn delay))
                    (recur))))))
          (reset! flushing? false)
          @outcomes)))))

;; ---- sealed mailbox -----------------------------------------------------------------------------

(defn make-mailbox
  "The sealed mailbox of the room of RC (an ird.p2p.crypto context).
  OPTS: :rc :client :mailbox-room-id :sleep-fn. Returns the mailbox map
  {:room-id :mailbox-room-id :rc :client :outbox}."
  [{:keys [rc client mailbox-room-id sleep-fn]}]
  {:room-id (:room-id rc)
   :mailbox-room-id mailbox-room-id
   :rc rc
   :client client
   :outbox (make-outbox {:client client :sleep-fn sleep-fn})})

(defn- inner-of [message]
  (case (:kind message)
    :json (env/msg (:value message))
    :data (env/bin (:payload message))
    (throw (err/config-error "message must be {:kind :json :value v} or {:kind :data :payload <base64>}" {:message message}))))

(defn seal
  "The at-rest envelope (base64) of MESSAGE. Channel."
  [{:keys [rc]} message]
  (go-try
    (<! (env/mailbox-seal rc (bytes/utf8->b64 (json/stringify (inner-of message)))))))

(defn open
  "Decrypt one replayed raw message into {:seq :id :ts :message}; nil when
  it does not open. Channel."
  [{:keys [rc]} m]
  (go
    (when-let [plain (<! (env/mailbox-open rc (:payload m)))]
      (when-let [inner (some-> (bytes/b64->utf8 plain) json/parse env/parse-inner)]
        (case (:kind inner)
          :msg {:seq (:seq m) :id (:id m) :ts (:ts m) :message {:kind :json :value (:payload inner)}}
          :bin {:seq (:seq m) :id (:id m) :ts (:ts m) :message {:kind :data :payload (:data inner)}}
          nil)))))

(defn deposit
  "Seal and deposit MESSAGE now. Channel of {:seq :id :expires_at :deduped}."
  [{:keys [client mailbox-room-id] :as mbx} message]
  (go-try
    (<? (deposit-raw client mailbox-room-id (<? (seal mbx message))))))

(defn enqueue
  "Seal and queue MESSAGE for `flush` (retries with backoff). Channel of true."
  [{:keys [outbox mailbox-room-id] :as mbx} message]
  (go-try
    (outbox-enqueue outbox mailbox-room-id (<? (seal mbx message)))
    true))

(defn flush-outbox
  "Deliver the outbox. Channel of the outcomes (see `outbox-flush`)."
  [{:keys [outbox]}]
  (outbox-flush outbox))

(defn pending [{:keys [outbox]}] (outbox-pending outbox))

(defn- open-all [mbx msgs]
  (go
    (loop [ms msgs acc []]
      (if-let [m (first ms)]
        (recur (rest ms) (if-let [item (<! (open mbx m))] (conj acc item) acc))
        acc))))

(defn replay
  "One page: {:items [{:seq :id :ts :message}] :page <raw page>}. OPTS as
  `replay-raw`. Channel."
  ([mbx] (replay mbx nil))
  ([{:keys [client mailbox-room-id] :as mbx} opts]
   (go-try
     (let [page (<? (replay-raw client mailbox-room-id opts))]
       {:items (<! (open-all mbx (:messages page))) :page page}))))

(defn replay-all
  "Everything after :after: {:items :cursor :gap?}. Channel."
  ([mbx] (replay-all mbx nil))
  ([{:keys [client mailbox-room-id] :as mbx} opts]
   (go-try
     (let [r (<? (replay-all-raw client mailbox-room-id opts))]
       {:items (<! (open-all mbx (:messages r))) :cursor (:cursor r) :gap? (:gap? r)}))))