(send-cmd! {:c "reg" :m {:id id}} private-key)))

      (configure-stream [_ {:keys [id] :as params}]
        (let [{:keys [private-key]} @session]
          (send-cmd! {:c "cfg" :m {:id id, :params (dissoc params :id)}}
                     private-key)))

      (submit-message [_ {:keys [id] :as params}]
        (let [{:keys [private-key]} @session]
          (send-cmd! {:c "put" :m {:id id,
                                   :params (dissoc params :id)}}
                     private-key)))

      (close [_] (close!))

      (is-open? [_] (is-open?)))))

  (server/start-netty-server!
    @senv/port
    handler
    server-started
    server-stopped
    :timeout @senv/timeout
    :accept-timeout @senv/accept-timeout
    :file-buffer-bytes @senv/file-buffer-bytes
    :file-channel-chunk-size @senv/file-channel-chunk-size))

(defn- start-socket! [handler server-started server-stopped]
  (server/start-socket-server!
    @senv/port
    handler
    server-started
    server-stopped

      (stream/register-stream! db-stream-model pk msg original-msg-bytes))
    (configure-stream! [_, pk, msg, original-msg-bytes]
      (stream/configure-stream! db-stream-model pk msg original-msg-bytes))
    (get-messages! [_, pk, {{:strs [id]} "m" :as msg}, original-msg-bytes]
      (if (= "COPYRIGHT" id)
        {:status :ok
         :messages [{"m" copyright-text}]}
        (stream/get-messages! db-stream-model pk msg original-msg-bytes)))
    (put-message! [_, pk, msg, original-msg-bytes]
      (stream/put-message! db-stream-model pk msg original-msg-bytes))))

(defn build-protocol-handler [db-loc session-timeout]
  (let [db-stream-model (sdl/build-stream-model db-loc)
        stream-model (wrap-stream-model db-stream-model)]
    (proto/protocol-handler
      {:stream-model stream-model
       :session-timeout (-> session-timeout

        :tfn parse-long)

(defenv file-buffer-bytes
        :doc "At what size do we cache bytes to a file? (netty-only, bytes)"
        :default (str 1048576)
        :masked? false
        :tfn parse-long)

(defenv file-channel-chunk-size
        :doc "Bytes to read into memory at a time with file buffer (netty-only)"
        :default (str 1048576)
        :masked? false
        :tfn parse-long)

(defenv protocol
        :doc "Server protocol. Allowed modes: streamful, echo"
        :default "streamful"
        :masked? false
        :tfn keyword)

      (do (log/warn "Failed to register stream" result)
          "failed"))))

(defn configure-stream! [pk stream-model msg original]
  (let [result (stream/configure-stream! stream-model pk msg original)]
    (case result
      :ok "ok"
      :unauthorized "unauthorized"
      (do (log/warn "Failed to configure stream" result)
          "failed"))))

(defn get-messages [pk stream-model cmd original]
  (let [{:keys [status messages] :as result}
        (stream/get-messages! stream-model pk cmd original)]
    (case status
      :ok {"status" "ok", "messages" messages}
      :stream-missing {"status" "stream not found"}
      (do (log/warn "Failed to get messages" result)
          "failed"))))

(defn- put-message! [pk stream-model cmd original]
  (let [result (stream/put-message! stream-model pk cmd original)]
    (case result
      :ok "ok"
      :unauthorized "unauthorized"
      (do (log/warn "Failed to configure stream" result)
          "failed"))))

(defn- handle-message
  ([signed? pk req msg]
   (try
     (let [original (.readAllBytes msg)
           msg (if signed? (cr/verified original pk) original)]
       (->> (cbor/decode msg)

         (throw e)))))

  ([{:keys [stream-model pk original] {:keys [remote-ip session-id]} :req}
    {:strs [c m] :as cmd}]
   (case c
     "ping" m
     "get" (get-messages pk stream-model cmd original)
     "put" (put-message! pk stream-model cmd original)
     "reg" (register-stream! pk stream-model cmd original)
     "cfg" (configure-stream! pk stream-model cmd original)
     (throw (ex-info "Unknown command"
                     {:cmd cmd
                      :session-id session-id
                      :remote-ip remote-ip})))))

(ns streamful.server
  (:require [clojure.java.io :as io]
            [clojure.string :as str]
            [streamful.crypto :as cr]
            [streamful.io :as sio]
            [streamful.temp :as temp]
            [taoensso.timbre :as log])
  (:import (io.netty.handler.codec ReplayingDecoder)
           (io.netty.util Signal)
           (java.io ByteArrayInputStream ByteArrayOutputStream File FileInputStream FileOutputStream
                    InputStream OutputStream RandomAccessFile)
           (java.net ServerSocket Socket SocketTimeoutException)
           (java.nio ByteBuffer ByteOrder CharBuffer)
           (java.nio.channels FileChannel FileChannel$MapMode)
           (java.util Scanner)
           (java.util.concurrent Executors ThreadFactory)))

(log/set-ns-min-level! #"io.netty.*" :info)

(import (io.netty.bootstrap ServerBootstrap)
        (io.netty.buffer ByteBuf ByteBufAllocator ByteBufUtil
                         CompositeByteBuf Unpooled)
        (io.netty.channel ChannelHandlerContext ChannelInitializer
                          ChannelOption SimpleChannelInboundHandler)
        (io.netty.channel.nio NioEventLoopGroup)
        (io.netty.channel.socket.nio NioServerSocketChannel)
        (io.netty.handler.codec ByteToMessageDecoder
                                MessageToMessageEncoder)
        (io.netty.handler.timeout ReadTimeoutHandler))

;;;;;;;;;;;;;;
;;; COMMON ;;;
;;;;;;;;;;;;;;

    (fn []
      (log/info "Stopping server")
      (reset! running false))))

;;;;;;;;;;;;;;;;;;;;
;;; NETTY SERVER ;;;
;;;;;;;;;;;;;;;;;;;;

(defn- read-line-from-buf! [^ByteBuf msg]
  (let [out (ByteArrayOutputStream.)]
    (loop []
      (let [b (.readByte msg)]
        (if (= 10 b)
          (.toString out cr/encoding)
          (do (.write out b) (recur)))))))





(defn- add-to-composite! [^CompositeByteBuf composite, ^ByteBuf buf]
  (.addComponent composite buf)
  (let [prev-idx (.writerIndex composite)
        addition (.readableBytes buf)]
    (.writerIndex composite (+ prev-idx addition))))

(defn- buf->temp-file-stream [^long file-channel-chunk-size
                              ^long bytes-to-read,
                              ^ByteBuf msg]
  (let [^File f (:temp-file (temp/temp-file! ".bin"))
        ^bytes buf (byte-array file-channel-chunk-size)]


    (with-open [fo (RandomAccessFile. f "rw")
                ^FileChannel fc (.getChannel fo)]
      (loop [bytes-left bytes-to-read]
        (when (pos? bytes-left)
          (let [this-chunk-size (min bytes-left file-channel-chunk-size)]
            (.readBytes msg buf 0 this-chunk-size)
            (.write fc (ByteBuffer/wrap buf 0 this-chunk-size))

            (recur (- bytes-left this-chunk-size))))))
    (FileInputStream. f)))

(defn- buf->byte-stream [^long bytes-to-read, ^ByteBuf msg]
  (let [^bytes arr (byte-array bytes-to-read)]
    (.readBytes msg arr 0 bytes-to-read)
    (ByteArrayInputStream. arr)))

(defn- decoder [{:keys [session-id
                        current-mode
                        bytes-to-read
                        file-buffer-bytes
                        file-channel-chunk-size]}]
  (proxy [ReplayingDecoder] []
    (decode [^ChannelHandlerContext ctx, ^ByteBuf msg, out]
      (try
        (log/trace session-id "decode" @current-mode @bytes-to-read)
        (let [read-result
              (case @current-mode
                :cmd
                (read-line-from-buf! msg)

                :bytes
                (if (> @bytes-to-read file-buffer-bytes)
                  (buf->temp-file-stream file-channel-chunk-size
                                         @bytes-to-read
                                         msg)
                  (buf->byte-stream @bytes-to-read msg)))]
          (when read-result
            (log/trace session-id "decoded" (loggable-msg read-result))
            (proxy-super checkpoint)
            (.add out read-result)))
        (catch Signal s (throw s))
        (catch Throwable t
          (log/error t)
          (.close ctx))))))

(defn- stream->temp->buf! [^long num-bytes, ^InputStream in]
  (let [^File temp-file (:temp-file (temp/temp-file! ".bin"))]
    (with-open [file-out (FileOutputStream. temp-file)]

                     @msg-count
                     new-mode
                     response)
          (when (instance? InputStream msg) (.close msg))
          (if out (.writeAndFlush ctx out) (.close ctx))))
      (exceptionCaught [_ctx t] (log/error t)))))

(defn- channel-initializer
  [handler timeout file-buffer-bytes file-channel-chunk-size]
  (proxy [ChannelInitializer] []
    (initChannel [socketChannel]
      (let [session-id (cr/guid)
            remote-ip (.remoteAddress socketChannel)
            current-mode (ref :cmd)
            bytes-to-read (ref nil)
            msg-count (ref 0)

            props
            {:session-id session-id
             :remote-ip remote-ip
             :current-mode current-mode
             :bytes-to-read bytes-to-read
             :handler handler
             :timeout timeout
             :msg-count msg-count
             :file-buffer-bytes file-buffer-bytes
             :file-channel-chunk-size file-channel-chunk-size}]
        (log/trace session-id "initChannel")

        (-> socketChannel
            .pipeline
            (.addLast
              "readTimeoutHandler"
              (ReadTimeoutHandler. (int timeout)))
            (.addLast "encoder" (encoder props))
            (.addLast "decoder" (decoder props))
            (.addLast "mainHandler" (channel-handler props)))))))

(defn start-netty-server!
  "The `handler` that you pass must follow the contract specified in
   `handle-client-msg!`

   `timeout` is the time to wait for bytes from a client
   `accept-timeout` is the time to wait for a socket connection
   `file-buffer-bytes` is the size an incoming message must be to stream
                       to a temp file
   `file-channel-chunk-size` is the size of a single chunk to read into memory
                             before writing to the aforementioned file buffer

  See the `streamful.server.server-test` for more."
  [^Integer port, handler, on-start, on-exit, &
   {:keys [^Integer timeout
           ^Integer accept-timeout,
           ^Long file-buffer-bytes
           ^Long file-channel-chunk-size]
    :or {timeout 60000
         accept-timeout 5000
         file-buffer-bytes 1024
         file-channel-chunk-size 1024}}]
  (let [server-group (NioEventLoopGroup.)
        client-group (NioEventLoopGroup.)
        server
        (-> (ServerBootstrap.)
            (.group server-group client-group)
            (.channel NioServerSocketChannel)
            (.childHandler
              (channel-initializer
                handler timeout file-buffer-bytes file-channel-chunk-size))
            (.childOption ChannelOption/SO_KEEPALIVE true)
            (.option ChannelOption/CONNECT_TIMEOUT_MILLIS
                     (int accept-timeout)))
        f (-> server (.bind port) .sync)]

    (when (.isSuccess f) (on-start))

    (fn []
      (log/info "Stopping server")
      (try
        (.shutdownGracefully server-group)
        (.shutdownGracefully client-group)
        (-> f .channel .closeFuture .sync)
        (finally (on-exit))))))

(ns streamful.stream)

(defprotocol StreamModel
  (register-stream! [_, ^String pk, msg, ^bytes original-msg-bytes])
  (configure-stream! [_, ^String pk, msg, ^bytes original-msg-bytes])
  (get-messages! [_, ^String pk, msg, ^bytes original-msg-bytes]
    "Get messages from a stream. May be part of a signed request.
    May cause state to change in the backend as we keep track of
    who has subscribed to our stream.")
  (put-message! [_, ^String pk, msg, ^bytes original-msg-bytes]
    "Put a message into a stream. Must be part of a signed request."))

(def ^:private stream-agg-table "stream-aggregates")
(def ^:private stream-data-table "stream-data")

(defn- get-db-manifest [db] (db/get-value db stream-agg-table :manifest))

(defn- streamk [iid t] (str iid "|" t))

(defn- stream-msg [iid msg client-psk original-msg-bytes]
  (let [t (System/currentTimeMillis)]
    [:put stream-data-table
     (streamk iid t) {"m" msg, "t" t, "psk" client-psk, "o" original-msg-bytes}
     :string :data]))

;; todo allow substreams of substreams
;; for now all substreams are implicitly of the root stream
(defn- register-stream! [db
                         client-psk
                         {{:strs [id]} "m" :as msg}
                         original-msg-bytes]
  (try
    (let [manifest (get-db-manifest db)

          {{:strs [substreams-allowed?]} "c" :as root-stream}
          (and manifest (manifest "root"))

          existing-stream (and manifest (manifest id))
          is-root? (= id "root")
          root-exists? (not (nil? root-stream))]
      (cond
        (or (and substreams-allowed? (not existing-stream))
            (and is-root? (not root-exists?)))
        (let [internal-id (cr/guid)
              smsg (stream-msg internal-id msg client-psk original-msg-bytes)
              first-msg-time (-> smsg (nth 3) :t)]
          (db/transact-kv db
            [[:put stream-agg-table
              :manifest
              (merge manifest {id
                               {"psk" client-psk
                                "iid" internal-id
                                "f" first-msg-time}})]
             smsg])
          :ok)

        (not root-exists?) :root-stream-missing

        existing-stream :already-exists

        :else :substreams-not-allowed))
    (catch Throwable t
      (log/error t)
      :database-error)))

(defn- configure-stream! [db
                          client-psk
                          {{:strs [id params]} "m" :as msg}
                          original-msg-bytes]
  (try
    (let [manifest (get-db-manifest db)
          {:strs [iid cfg] :as existing-stream} (and manifest (manifest id))]
      (cond
        (and existing-stream (creq/eq? client-psk (existing-stream "psk")))
        (do
          (db/transact-kv db
            [[:put stream-agg-table
              :manifest
              (merge manifest
                     {id (assoc existing-stream "c" (merge cfg params))})]
             (stream-msg iid msg client-psk original-msg-bytes)])
          :ok)

        (not existing-stream) :stream-missing

        :else :unauthorized))
    (catch Throwable t
      (log/error t)
      :database-error)))

(defn get-messages! [db
                     client-psk
                     {{:strs [id]} "m" :as msg}
                     _original-msg-bytes]
  (let [manifest (get-db-manifest db)
        {:strs [iid f psk]} (and manifest (manifest id))
        fk (streamk iid f)
        lk (streamk iid (System/currentTimeMillis))
        r [:closed fk lk]]
    (if (creq/eq? psk client-psk)
      {:status :ok
       :messages (db/get-first-n db stream-data-table 50 r :string :data true)}
      {:status :stream-missing})))

(defn put-message! [db
                    client-psk
                    {{:strs [id ]} "m" :as msg}
                    original-msg-bytes]
  (let [manifest (get-db-manifest db)
        {:strs [iid] :as existing-stream} (and manifest (manifest id))]
    (cond
      (and existing-stream (creq/eq? client-psk (existing-stream "psk")))
      (do
        (db/transact-kv db
          [(stream-msg iid msg client-psk original-msg-bytes)])
        :ok)

      (not existing-stream) :stream-missing

      :else :unauthorized)))

(defn build-stream-model [location]
  (let [db (db/open-kv location)]
    (db/open-dbi db stream-agg-table)
    (db/open-dbi db stream-data-table)
    (reify stream/StreamModel
      (register-stream! [_, pk, msg, original-msg-bytes]
        (register-stream! db pk msg original-msg-bytes))
      (configure-stream! [_, pk, msg, original-msg-bytes]
        (configure-stream! db pk msg original-msg-bytes))
      (get-messages! [_, pk, msg, original-msg-bytes]
        (get-messages! db pk msg original-msg-bytes))
      (put-message! [_, pk, msg, original-msg-bytes]
        (put-message! db pk msg original-msg-bytes)))))

            [clojure.test :refer :all]
            [streamful.client :as client]
            [streamful.crypto :as cr]
            [streamful.protocol :refer :all]
            [streamful.test-asserts :as ta]
            [streamful.test-cfg :as tcfg]))

(defn- reassemble-msg [{:strs [psk o]}]
  (-> o (cr/verified psk) cbor/decode))

(defn- as [client keys]
  (is (= {:result "ok"} (client/as-stream client keys))))

(defn- get-as [client keys id]
  (as client keys)
  (client/get-messages client {:id id}))

(defn- get-m [m] (get m "m"))

(tcfg/def-server-test protocol-test [client]
  (testing "ping"
    (let [msg-to-echo (cr/guid94)]

                 {:public-key public-signing-key
                  :private-key private-signing-key2})))

        (is (= {:response "unauthorized"}
               (client/register-stream client {:id "test1"}))))

      (testing "first stream must be called 'root'"
        (as client keys)

        (is (= {:response "root not yet configured -- register that first"}
               (client/register-stream client {:id "test1"})))

        (is (= {:response "ok"} (client/register-stream client {:id "root"}))))

      (testing "can't register multiple streams with the same name"
        (is (= {:response "already exists"}
               (client/register-stream client {:id "root"}))))

      (testing "can't register more than one stream"
        (is (= {:response "substreams aren't allowed here"}
               (client/register-stream client {:id "test1"}))))

      (testing "only stream owner can configure it"
        (as client keys2)
        (is (= {:response "unauthorized"}
               (client/configure-stream client {:id "root"
                                                :substreams-allowed? true}))))

      (ta/pending "unsigned configuration messages don't get to the model")

      (testing "if root allows substreams"
        (as client keys)
        (is (= {:response "ok"}
               (client/configure-stream client {:id "root"
                                                :substreams-allowed? true})))
        (testing "a substream can be registered"
          (as client keys2)
          (is (= {:response "ok"}
                 (client/register-stream client {:id "test1"})))

          (testing "but not twice"
            (as client keys)
            (is (= {:response "already exists"}
                   (client/register-stream client {:id "test1"}))))

          (testing "and not after reconfigured off"
            (is (= {:response "ok"}
                   (client/configure-stream client
                                            {:id "root"
                                             :substreams-allowed? false})))
            (is (= {:response "substreams aren't allowed here"}
                   (client/register-stream client {:id "test2"}))))

          (testing "but not twice, even after reconfiguring"
            (is (= {:response "ok"}
                   (client/configure-stream client
                                            {:id "root"
                                             :substreams-allowed? true})))
            (as client keys)
            (is (= {:response "already exists"}
                   (client/register-stream client {:id "test1"}))))))








      (let [expected-root-messages
            [{"c" "reg"
              "m" {"id" "root"}}
             {"c" "cfg"
              "m" {"id" "root", "params" {"substreams-allowed?" true}}}
             {"c" "cfg"
              "m" {"id" "root", "params" {"substreams-allowed?" false}}}
             {"c" "cfg"
              "m" {"id" "root", "params" {"substreams-allowed?" true}}}]

            expected-test1-messages
            [{"c" "reg", "m" {"id" "test1"}}]]
        (testing "retrieving stream messages"
          (let [{{root-messages "messages" :as r1} :response}
                (get-as client keys "root")

                {{test1-messages "messages" :as r2} :response}
                (get-as client keys2 "test1")]

            (testing "messages include configuration stuff"
              (is (= expected-root-messages
                     (map get-m root-messages)) r1)
              (is (= expected-test1-messages
                     (map get-m test1-messages)) r2))

            (testing "we can validate and reassemble from original binaries"
              (is (= expected-root-messages
                     (map reassemble-msg root-messages)))
              (is (= expected-test1-messages
                     (map reassemble-msg test1-messages))))))

        (testing "good error message when stream is missing"
          (is (= {:response {"status" "stream not found"}}
                 (get-as client keys "missing-stream"))))

        (testing "streams are private by default, show as missing"
          (is (= {:response {"status" "stream not found"}}
                 (get-as client keys2 "root"))))

        (testing "publish simple messages"
          (as client keys)
          (is (= {:response "ok"}
                 (client/submit-message client
                                        {:id "root" :t "hello, world!"})))

          (testing "can't publish to someone else's stream"
            (as client keys2)
            (is (= {:response "unauthorized"}
                   (client/submit-message client
                                          {:id "root" :t "hello, friend!"})))
            (is (= {:response "ok"}
                   (client/submit-message client
                                          {:id "test1" :t "hello, friend!"}))))
          (testing "messages show up"
            (let [{{root-messages "messages"} :response}
                  (get-as client keys "root")

                  {{test1-messages "messages"} :response}
                  (get-as client keys2 "test1")]
              (is (= (conj
                       expected-root-messages
                       {"c" "put"
                        "m" {"id" "root", "params" {"t" "hello, world!"}}})
                     (map get-m root-messages)))
              (is (= (conj
                       expected-test1-messages
                       {"c" "put"
                        "m" {"id" "test1", "params" {"t" "hello, friend!"}}})
                     (map get-m test1-messages)))))
          (ta/pending "unsigned put messages don't get to the model")
          (ta/pending "messages have server receipt timestamps"))))
    (ta/pending "stream access")))

            [defenv.core :refer [defenv] :as env]
            [streamful.client :as client]
            [streamful.crypto :as cr]
            [streamful.env :as senv]
            [streamful.io :as sio]
            [streamful.log-util :as lu]
            [streamful.server :refer :all]
            [streamful.test-asserts :as ta]
            [streamful.test-net :as tnet]
            [streamful.test-protocol :as tp]
            [streamful.test-thread :as tthread]
            [taoensso.timbre :as log])
  (:import (java.io ByteArrayInputStream InputStream OutputStream)
           (java.net Socket)))

                   test-name
                   (- (System/currentTimeMillis) start-time))))))

(deftest server-test
  (log/set-ns-min-level! #"io.netty" :info)
  (lu/test-log-level! (log/info (with-out-str (env/display-env))))
  (run-server-test "socket server" start-socket-server!)
  (run-server-test "netty server" start-netty-server!)
  (ta/pending "netty uses file buffers properly"))

(ns streamful.test-cfg
  (:require [clojure.test :refer :all]
            [streamful.core :as core]
            [streamful.log-util :as lu]
            [streamful.temp :as temp]
            [streamful.test-protocol :as tp])
  (:import (java.io File)))

(defmacro def-logcfg-test [n & body]
  `(deftest ~n
     (lu/test-log-level! ~@body)))

(defn- build-temp-protocol-handler []
  (let [^File temp-dir (temp/session-tempdir!)
        ^File db-dir (File. temp-dir "db")]
    (core/build-protocol-handler (.getAbsolutePath db-dir) 1)))

(defmacro def-server-test
  [n b & body]
  `(def-logcfg-test ~n
     (do
       (tp/with-server-and-client (~build-temp-protocol-handler) [~(first b)]
         ~@body)
       (temp/cleanup-all-tempdirs!))))

(ns streamful.test-protocol
  (:require [clojure.test :refer :all]
            [streamful.client :as client]
            [streamful.server :as server]
            [streamful.test-net :as tnet]
            [streamful.test-thread :as tthread]))

(defn start-server-and-wait [start-fn! port handler &
                             {:keys [on-start
                                     on-exit
                                     num-threads]
                              :or {on-exit (constantly :default)
                                   on-start (constantly :default)
                                   num-threads 1}}]
  (let [server-running (atom false)
        on-start
        (fn []
          (on-start)