<commit_before>(ns birdie-tell.core
  "This ns implements a simple Gossip protocol, where peers each have
  a map of data they own, and they gossip to each other to pass along
  their own data.

  Each peer tracks versions of its data so that recipients of gossip
  can merge newer versions safely.

  Each peer rereads its data regularly from a specified EDN data
  file. This data is arbitrary."
  (:require [clojure.edn :as edn]
            [clojure.set :refer [difference]]
            [clojure.java.io :as io]
            [server.socket :as socket]
            [cheshire.core :as cheshire]
            [clojure.edn :as edn]
            [clojure.string :as str]
            [birdie-tell.util :as util]
            [clojure.pprint :refer [pprint]])
  (:import (java.net Socket SocketException)
           (java.io PrintWriter InputStreamReader BufferedReader)))

(comment
  ;; Sample peer state
  {:uuid "725b7252-d104-44e0-9c73-3cf7f87fb41d"
   :peers
   {:identified {"725b7252-d104-44e0-9c73-3cf7f87fb41d" {:data {:silk #{:calde :augur}
                                                                :auk  #{:prophet :thief}}
                                                         :version 10
                                                         :name "wolfe"
                                                         :host-port "127.0.0.1:1400"}
                 "f7fe1c2f-aaf8-4166-9689-588805424bcb" {:data {:grandpa-joe #{:old :bedridden}
                                                                :wonka #{:dapper :bottle-green-pants}
                                                                :charlie #{:mopey :poor}}
                                                         :version 0
                                                         :name "dahl"
                                                         :alive? true
                                                         :host-port "127.0.0.1:1401"}
                 "41008adf-fcdd-4ef5-8a70-0f0dccd3d806" {:data {:gimli #{:bearded :short :tired}
                                                                :legolas #{:lithe :merry}
                                                                :aragorn #{:brooding :wise :unaccountably-old}}
                                                         :version 7
                                                         :name "tolkien"
                                                         :alive? false
                                                         :host-port "127.0.0.1:1402"}}
    :potential #{"127.0.0.1:1403"}}}

  )


;; * what are the states? what does the state machine look like?
;; * what is the protocol?
;; * what are the transitions between states?
;; * what transport options are there? sockets, core.async, other?
;; * separate ns for user interaction: init the state, read the state, update the state
;; * separate ns for CLI interface

(def state (atom {
                  ;; mapping from UUIDs to states of peers, including me!
                  :peers {}
                  }))

(defn- split-hostport
  "Split a host:port pair into a vector with a string host and an integer port.

  Example:
      (split-hostport \"hostname:1234\")
      ;=> [\"hostname\" 1234]"
  [host-port]
  (let [[ip-addr port & trash] (str/split host-port #":")
        port (Integer/parseInt port)]
    [ip-addr port]))

(defn- join-hostport
  "Join a host, port pair into a host:port string."
  [host port]
  (str host ":" port))

(defn- merge-my-data
  "Take my state map and 'new-data.

  If 'new-data is equal to my state :data, return the state
  unchanged.

  If 'new-data is different than my current :data, assoc the new
  :data, 'inc my :version, and return the new state."
  [current-state new-data]
  (let [my-path [:peers :identified (:uuid current-state)]
        my-state (get-in current-state my-path)]
    (if (= (:data my-state) new-data)
      current-state
      (assoc-in current-state
                my-path
                (merge my-state {:data new-data :version (inc (:version my-state))})))))

(defn- clean-potential-peers
  "Remove each host-port entry in :peers :potential that is found in
  an :identified peer.

  Philosophical rationale: realized potential is no longer potential."
  [dirty-state]
  (let [identified-peer-host-ports (->> dirty-state :peers :identified
                                        (reduce (fn [r [k v]] (conj r (:host-port v))) #{}))]
    (update-in dirty-state
               [:peers :potential]
               difference
               identified-peer-host-ports)))

(defn- merge-peer-state
  "Take my current 'my-state map and 'peer-state, a complete state map from a peer.

  Merge the :peers from 'peer-state with :peers in 'my-state, ignoring
  myself, adding any new :identified peers, and updating :data,
  :version, and :host-port for :identified peers with higher versions
  than I have.

  TODO: consider :name changes
  TODO: think about merging :potential peers in case this peer can reach a host-port another peer cannot
  "
  [my-state peer-state]
  (util/debug :merge-peer-state :my-state my-state :peer-state peer-state)
  (let [my-uuid (:uuid my-state)
        peer-identified-peers (-> peer-state :peers :identified (dissoc my-uuid))
        my-identified-peers (-> my-state :peers :identified)
        version-check (fn [old new]
                        (if (>= (:version old) (:version new))
                          old
                          (merge old (select-keys new [:data :version :host-port]))))
        dirty-state (assoc-in my-state
                              [:peers :identified]
                              (merge-with version-check
                                          my-identified-peers
                                          peer-identified-peers))
        new-state (clean-potential-peers dirty-state)]
    (util/debug :new-state new-state)
    new-state))

(defn- parse-gossip
  "Take a 'gossip-handler, an 'input-stream-handler, an
  'output-stream-handler, an 'input-stream, and an 'output-stream.

  Parse 'input-stream EDN into a map of gossiped state, and call
  'gossip-handler with the parsed state. Then call
  'output-stream-handler with 'output-stream.

  Return nil."
  [gossip-handler input-stream-handler output-stream-handler input-stream output-stream]
  (let [input-string (input-stream-handler input-stream)
        peer-state (edn/read-string input-string)]
    (gossip-handler peer-state)
    (output-stream-handler output-stream)
    nil))

(defn- update-peer!
  "Update a peer's state in the specified 'peers atom.

  Take a 'peers atom, a 'max-hotness integer, and a vector with this shape:
      [[host port] state]

  Return nil."
  [peers max-hotness [host-port state]]
  {:pre [(vector? host-port) (keyword? state)]}
  (swap! peers #(merge-with merge
                            %
                            {:states {host-port state}
                             :hotness {host-port max-hotness}}))
  nil)

(defn- import-chirps
  "Take a map of input 'chirps. Split host:port pairs and ensure states
  are keywords.

  Example:
      (import-chirps {\"127.0.0.1:2234\" \"alive\", \"127.0.0.1:2237\" \"alive\"})
      ;==> {[\"127.0.0.1\" 2234] :alive, [\"127.0.0.1\" 2237] :alive}
  "
  [chirps]
  (into {}
        (map (fn [[k v]] [(split-hostport k) (keyword v)])
             chirps)))

;; TODO: think about core.async
(defn- receive-gossip
  "Take an 'input-stream and a (unused) '_output-stream.

  Parse 'input-stream JSON into a 'news map (see 'choose-news), and
  map 'update-peer! across each entry in that map.

  Return nil."
  [max-hotness input-stream _output-stream]
  (let [chirps (-> input-stream
                   (InputStreamReader. "UTF-8")
                   cheshire/parse-stream)
        chirps (import-chirps chirps)]
    ;; TODO: pull side-effects out, pass fn in (to allow for multiple implementations and general decomplection)
    (dorun (map (partial update-peer! #'peers max-hotness) chirps))
    nil))

(defn- listen
  "Start listening on 'port, calling 'parse-gossip with
  'gossip-handler and input/output handlers on connect."
  [gossip-handler port]
  (let [input-stream-handler (fn [input-stream]
                               (with-open [rdr (io/reader input-stream)]
                                 (reduce str (line-seq rdr))))
        output-stream-handler (fn [output-stream] nil)]
    (socket/create-server port (partial parse-gossip<commit_msg>[core] Nix old fns 'update-peer!, 'import-chirps, 'receive-gossip<commit_after>(ns birdie-tell.core
  "This ns implements a simple Gossip protocol, where peers each have
  a map of data they own, and they gossip to each other to pass along
  their own data.

  Each peer tracks versions of its data so that recipients of gossip
  can merge newer versions safely.

  Each peer rereads its data regularly from a specified EDN data
  file. This data is arbitrary."
  (:require [clojure.edn :as edn]
            [clojure.set :refer [difference]]
            [clojure.java.io :as io]
            [server.socket :as socket]
            [cheshire.core :as cheshire]
            [clojure.edn :as edn]
            [clojure.string :as str]
            [birdie-tell.util :as util]
            [clojure.pprint :refer [pprint]])
  (:import (java.net Socket SocketException)
           (java.io PrintWriter InputStreamReader BufferedReader)))

(comment
  ;; Sample peer state
  {:uuid "725b7252-d104-44e0-9c73-3cf7f87fb41d"
   :peers
   {:identified {"725b7252-d104-44e0-9c73-3cf7f87fb41d" {:data {:silk #{:calde :augur}
                                                                :auk  #{:prophet :thief}}
                                                         :version 10
                                                         :name "wolfe"
                                                         :host-port "127.0.0.1:1400"}
                 "f7fe1c2f-aaf8-4166-9689-588805424bcb" {:data {:grandpa-joe #{:old :bedridden}
                                                                :wonka #{:dapper :bottle-green-pants}
                                                                :charlie #{:mopey :poor}}
                                                         :version 0
                                                         :name "dahl"
                                                         :alive? true
                                                         :host-port "127.0.0.1:1401"}
                 "41008adf-fcdd-4ef5-8a70-0f0dccd3d806" {:data {:gimli #{:bearded :short :tired}
                                                                :legolas #{:lithe :merry}
                                                                :aragorn #{:brooding :wise :unaccountably-old}}
                                                         :version 7
                                                         :name "tolkien"
                                                         :alive? false
                                                         :host-port "127.0.0.1:1402"}}
    :potential #{"127.0.0.1:1403"}}}

  )


;; * what are the states? what does the state machine look like?
;; * what is the protocol?
;; * what are the transitions between states?
;; * what transport options are there? sockets, core.async, other?
;; * separate ns for user interaction: init the state, read the state, update the state
;; * separate ns for CLI interface

(def state (atom {
                  ;; mapping from UUIDs to states of peers, including me!
                  :peers {}
                  }))

(defn- split-hostport
  "Split a host:port pair into a vector with a string host and an integer port.

  Example:
      (split-hostport \"hostname:1234\")
      ;=> [\"hostname\" 1234]"
  [host-port]
  (let [[ip-addr port & trash] (str/split host-port #":")
        port (Integer/parseInt port)]
    [ip-addr port]))

(defn- join-hostport
  "Join a host, port pair into a host:port string."
  [host port]
  (str host ":" port))

(defn- merge-my-data
  "Take my state map and 'new-data.

  If 'new-data is equal to my state :data, return the state
  unchanged.

  If 'new-data is different than my current :data, assoc the new
  :data, 'inc my :version, and return the new state."
  [current-state new-data]
  (let [my-path [:peers :identified (:uuid current-state)]
        my-state (get-in current-state my-path)]
    (if (= (:data my-state) new-data)
      current-state
      (assoc-in current-state
                my-path
                (merge my-state {:data new-data :version (inc (:version my-state))})))))

(defn- clean-potential-peers
  "Remove each host-port entry in :peers :potential that is found in
  an :identified peer.

  Philosophical rationale: realized potential is no longer potential."
  [dirty-state]
  (let [identified-peer-host-ports (->> dirty-state :peers :identified
                                        (reduce (fn [r [k v]] (conj r (:host-port v))) #{}))]
    (update-in dirty-state
               [:peers :potential]
               difference
               identified-peer-host-ports)))

(defn- merge-peer-state
  "Take my current 'my-state map and 'peer-state, a complete state map from a peer.

  Merge the :peers from 'peer-state with :peers in 'my-state, ignoring
  myself, adding any new :identified peers, and updating :data,
  :version, and :host-port for :identified peers with higher versions
  than I have.

  TODO: consider :name changes
  TODO: think about merging :potential peers in case this peer can reach a host-port another peer cannot
  "
  [my-state peer-state]
  (util/debug :merge-peer-state :my-state my-state :peer-state peer-state)
  (let [my-uuid (:uuid my-state)
        peer-identified-peers (-> peer-state :peers :identified (dissoc my-uuid))
        my-identified-peers (-> my-state :peers :identified)
        version-check (fn [old new]
                        (if (>= (:version old) (:version new))
                          old
                          (merge old (select-keys new [:data :version :host-port]))))
        dirty-state (assoc-in my-state
                              [:peers :identified]
                              (merge-with version-check
                                          my-identified-peers
                                          peer-identified-peers))
        new-state (clean-potential-peers dirty-state)]
    (util/debug :new-state new-state)
    new-state))

(defn- parse-gossip
  "Take a 'gossip-handler, an 'input-stream-handler, an
  'output-stream-handler, an 'input-stream, and an 'output-stream.

  Parse 'input-stream EDN into a map of gossiped state, and call
  'gossip-handler with the parsed state. Then call
  'output-stream-handler with 'output-stream.

  Return nil."
  [gossip-handler input-stream-handler output-stream-handler input-stream output-stream]
  (let [input-string (input-stream-handler input-stream)
        peer-state (edn/read-string input-string)]
    (gossip-handler peer-state)
    (output-stream-handler output-stream)
    nil))

(defn- listen
  "Start listening on 'port, calling 'parse-gossip with
  'gossip-handler and input/output handlers on connect."
  [gossip-handler port]
  (let [input-stream-handler (fn [input-stream]
                               (with-open [rdr (io/reader input-stream)]
                                 (reduce str (line-seq rdr))))
        output-stream-handler (fn [output-stream] nil)]
    (socket/create-server port (partial parse-gossip
{'fraction_non_alphanumeric': 0.10946459635332775, 'fraction_numerical': 0.019779653372849686, 'mean_word_length': 1.6729808851386259, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 12, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 76, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5076356', 'n_tokens_mistral': 4611, 'n_tokens_neox': 4320, 'n_words': 1367}