;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) KALEIDOS SUBSIDIARY SL

(ns backend-tests.binfile-jobs-test
  "The binfile work of a job: the shared adapter that runs it with its
  progress reported as job events, and the two job-defs that expose it."
  (:require
   [app.binfile.common :as bfc]
   [app.binfile.jobs :as bfj]
   [app.binfile.v1 :as v1]
   [app.binfile.v3 :as v3]
   [app.common.time :as ct]
   [app.common.uuid :as uuid]
   [app.db :as db]
   [app.jobs :as jobs]
   [app.jobs.storage :as js]
   [app.rpc.commands.teams :as teams]
   [app.storage :as sto]
   [app.storage.tmp :as tmp]
   [app.util.events :as events]
   [backend-tests.helpers :as th]
   [clojure.java.io :as jio]
   [clojure.test :as t]
   [datoteka.fs :as fs]
   [datoteka.io :as io]))

(t/use-fixtures :once th/state-init)
(t/use-fixtures :each th/database-reset)

(def ^:private fixture
  "A real penpot export, the same fixture the binfile tests read."
  "backend_tests/test_files/svg-attrs-camel-case.penpot")

(defn- fixture-path
  []
  (-> fixture io/resource jio/file))

(defn- import-fixture!
  "A file in the default project of the profile, imported from the
  fixture through the core."
  [profile]
  (-> th/*system*
      (assoc ::bfc/project-id (:default-project-id profile))
      (assoc ::bfc/profile-id (:id profile))
      (assoc ::bfc/team-id (:default-team-id profile))
      (assoc ::bfc/input (fixture-path))
      (v3/import-files!)))

(defn- job-cfg
  "The system cfg as the runner hands it to a handler. The job id is NOT
  bound here: the adapter takes it from the context, because the progress
  bridge runs on a thread of its own."
  []
  th/*system*)

(defn- job-context
  [job-id]
  (jobs/make-context (jobs/get-job th/*system* job-id)))

(defn- run-export!
  "Run an export through the adapter, as the job handler does: the
  frozen cfg in, the progress out as job events."
  [cfg context file-ids export-type output]
  (let [cfg (bfc/export-cfg cfg {:file-ids    file-ids
                                 :export-type export-type
                                 :output      output})]
    (bfj/check-active cfg context)
    (bfj/run cfg context v3/export-files!)))

(defn- run-import!
  "Run an import through the adapter, as the job handler does: the
  frozen cfg in, the progress out as job events. The project touch
  stays in the handler, covered by its own tests."
  [cfg context {:keys [profile-id project-id team name input version]}]
  (bfj/check-active cfg context)
  (let [cfg (bfc/import-cfg cfg {:profile-id profile-id
                                 :project-id project-id
                                 :team       team
                                 :name       name
                                 :input      input})]
    (bfj/run cfg context
             (fn [cfg]
               (case (int version)
                 1 (v1/import-files! cfg)
                 3 (v3/import-files! cfg)
                 (throw (ex-info "unsupported binfile version"
                                 {:type    :validation
                                  :code    :unsupported-version
                                  :version version})))))))

(defn- progress-events
  "The progress events of a job, oldest first, with decoded payloads."
  [job-id]
  (->> (th/db-exec! ["SELECT payload FROM job_event
                       WHERE job_id = ? AND kind = 'progress'
                       ORDER BY created_at ASC, id ASC"
                     job-id])
       (mapv #(jobs/decode-progress (:payload %)))))

(defn- export-link-later-package
  "A package of one file that uses a shared library, exported so the
  library is resolved while importing. The library of the package is gone
  afterwards and two replacements share its name, so the import cannot
  link it by itself and leaves the file pending resolution."
  [profile]
  (let [library (th/create-file* 1 {:profile-id (:id profile)
                                    :project-id (:default-project-id profile)
                                    :is-shared  true
                                    :name       "Icons Library"})
        file    (th/create-file* 2 {:profile-id (:id profile)
                                    :project-id (:default-project-id profile)
                                    :is-shared  false})
        output  (tmp/tempfile :suffix ".zip")]

    (db/insert! th/*system* :file-library-rel
                {:file-id (:id file)
                 :library-file-id (:id library)})

    (v3/export-files! (-> th/*system*
                          (assoc ::bfc/file-ids #{(:id file)})
                          (assoc ::bfc/export-type :link-later)
                          (assoc ::bfc/output (io/output-stream output))))

    (db/update! th/*system* :file
                {:deleted-at (ct/now)}
                {:id (:id library)})

    (th/create-file* 3 {:profile-id (:id profile)
                        :project-id (:default-project-id profile)
                        :is-shared  true
                        :name       "Icons Library"})
    (th/create-file* 4 {:profile-id (:id profile)
                        :project-id (:default-project-id profile)
                        :is-shared  true
                        :name       "Icons Library"})

    {:library-id (:id library)
     :output     output}))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; ADAPTER
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(t/deftest export-through-the-adapter-produces-the-artifact
  (let [profile (th/create-profile* 1)
        file-id (first (:file-ids (import-fixture! profile)))
        job-id  (th/create-job* :export-binfile {:profile-id (:id profile)})
        output  (tmp/tempfile* :suffix ".penpot")]

    (run-export! (job-cfg) (job-context job-id)
                 #{file-id} :detach-libraries output)

    (t/testing "the artifact is a penpot file with its manifest"
      (t/is (fs/exists? output))
      (t/is (= [file-id] (mapv :id (:files (v3/get-manifest th/*system* output))))))

    (t/testing "the progress of the core became milestones of the job"
      (let [events (progress-events job-id)]
        (t/is (pos? (count events)))
        (t/is (every? keyword? (map :stage events)))
        (t/testing "the first one is the file in course, with its position"
          (let [event (first events)]
            (t/is (= :files (:stage event)))
            (t/is (= {:current 1 :total 1} (get-in event [:counters :files])))))
        (t/testing "and nothing else: the event is the stage and the counters"
          (t/is (every? #(<= (count (keys %)) 2) events)))))))

(t/deftest import-through-the-adapter-imports-the-package
  (let [profile  (th/create-profile* 1)
        team     (teams/get-team th/*system*
                                 :profile-id (:id profile)
                                 :project-id (:default-project-id profile))
        staged   (js/put-resource th/*system* (:id profile)
                                  {:content  (sto/content (fixture-path))
                                   :filename "package.penpot"
                                   :mtype    "application/zip"})
        job-id   (th/create-job* :import-binfile {:profile-id (:id profile)
                                                  :resource-id (:resource-id staged)})
        context  (job-context job-id)
        result   (run-import! (job-cfg) context
                              {:profile-id (:id profile)
                               :project-id (:default-project-id profile)
                               :team       team
                               :name       "imported"
                               :input      (js/load-input th/*system* context)
                               :version    3})]

    (t/testing "the package is imported in the destination project"
      (t/is (= 1 (count (:file-ids result)))))

    (t/testing "the job reported its progress"
      (t/is (pos? (count (progress-events job-id)))))))

(t/deftest the-adapter-composes-a-milestone-per-tap
  (let [profile (th/create-profile* 1)
        file-id (first (:file-ids (import-fixture! profile)))
        job-id  (th/create-job* :export-binfile {:profile-id (:id profile)})
        reports (atom [])]

    ;; the substrate throttles the durable events, so the whole stream of
    ;; milestones is only visible with the writes intercepted
    (with-redefs [jobs/heartbeat (fn [_cfg & {:as options}]
                                   (swap! reports conj (:progress options))
                                   1)]
      (run-export! (job-cfg) (job-context job-id)
                   #{file-id} :detach-libraries
                   (tmp/tempfile* :suffix ".penpot")))

    (t/testing "the file opens the run with its position"
      (let [event (first @reports)]
        (t/is (= :files (:stage event)))
        (t/is (= {:current 1 :total 1} (get-in event [:counters :files])))))

    (t/testing "the stages inside the file carry it as context, with their own counter"
      (let [event (first (filter #(= :pages (:stage %)) @reports))]
        (t/is (= {:current 1 :total 1} (get-in event [:counters :files])))
        (t/is (pos? (get-in event [:counters :pages :current])))
        (t/is (pos? (get-in event [:counters :pages :total])))))

    (t/testing "a stage outside a file carries no file context"
      (let [events (filter #(= :storage-objects (:stage %)) @reports)]
        (t/is (pos? (count events)))
        (t/is (every? #(nil? (get-in % [:counters :files])) events))))

    (t/testing "and every milestone is a stage and its counters, and nothing else"
      (t/is (every? #(<= (count (keys %)) 2) @reports)))))

(t/deftest a-tap-carries-the-outer-counters-in-course
  (let [profile (th/create-profile* 1)
        file-id (first (:file-ids (import-fixture! profile)))
        job-id  (th/create-job* :export-binfile {:profile-id (:id profile)})
        taps    (atom [])]

    ;; the taps themselves, before the adapter turns them into events
    (with-redefs [events/tap (fn [_type data] (swap! taps conj data))]
      (run-export! (job-cfg) (job-context job-id)
                   #{file-id} :detach-libraries
                   (tmp/tempfile* :suffix ".penpot")))

    (t/testing "the tap that opens the file carries the outer counters"
      (let [tap (first (filter #(= :file (:section %)) @taps))]
        (t/is (= {:files {:current 1 :total 1}} (:outer-counters tap)))))

    (t/testing "every tap inside the file carries them as context"
      (let [inner (filter #(some? (:file-id %)) @taps)]
        (t/is (pos? (count inner)))
        (t/is (every? #(= {:files {:current 1 :total 1}} (:outer-counters %)) inner))))

    (t/testing "a tap outside a file carries no outer counters"
      (let [outside (filter #(= :storage-object (:section %)) @taps)]
        (t/is (pos? (count outside)))
        (t/is (every? #(nil? (:outer-counters %)) outside))))))

(t/deftest a-milestone-merges-outer-counters-with-its-own
  (let [milestone @#'bfj/milestone]
    (t/testing "outer scopes survive next to the counter of the stage"
      (t/is (= {:stage :pages
                :counters {:teams {:current 1 :total 1}
                           :projects {:current 2 :total 2}
                           :files {:current 1 :total 4}
                           :pages {:current 2 :total 5}}}
               (milestone {:section :page
                           :current 2 :total 5
                           :outer-counters {:teams {:current 1 :total 1}
                                            :projects {:current 2 :total 2}
                                            :files {:current 1 :total 4}}}))))
    (t/testing "a file tap needs no own counter, the outer one is enough"
      (t/is (= {:stage :files
                :counters {:files {:current 1 :total 2}}}
               (milestone {:section :file
                           :outer-counters {:files {:current 1 :total 2}}}))))))

(t/deftest entering-a-file-keeps-the-counters-of-the-outer-scopes
  (let [cfg  {::v3/outer-counters {:teams {:current 1 :total 1}}}
        cfg' (@#'v3/with-outer-counter cfg :files {:current 2 :total 4})]
    (t/is (= {:teams {:current 1 :total 1}
              :files {:current 2 :total 4}}
             (::v3/outer-counters cfg')))))

(t/deftest the-import-adapter-composes-the-same-milestones
  (let [profile (th/create-profile* 1)
        team    (teams/get-team th/*system*
                                :profile-id (:id profile)
                                :project-id (:default-project-id profile))
        staged  (js/put-resource th/*system* (:id profile)
                                 {:content  (sto/content (fixture-path))
                                  :filename "package.penpot"
                                  :mtype    "application/zip"})
        job-id  (th/create-job* :import-binfile {:profile-id (:id profile)
                                                 :resource-id (:resource-id staged)})
        context (job-context job-id)
        reports (atom [])]

    (with-redefs [jobs/heartbeat (fn [_cfg & {:as options}]
                                   (swap! reports conj (:progress options))
                                   1)]
      (run-import! (job-cfg) context
                   {:profile-id (:id profile)
                    :project-id (:default-project-id profile)
                    :team       team
                    :name       "imported"
                    :input      (js/load-input th/*system* context)
                    :version    3}))

    (t/testing "the run opens with the manifest, which has no units to count"
      (let [event (first @reports)]
        (t/is (= :manifest (:stage event)))
        (t/is (nil? (:counters event)))))

    (t/testing "the file carries its position"
      (let [event (first (filter #(= :files (:stage %)) @reports))]
        (t/is (= {:current 1 :total 1} (get-in event [:counters :files])))))

    (t/testing "a stage inside the file carries it as context, with its own counter"
      (let [event (first (filter #(= :pages (:stage %)) @reports))]
        (t/is (= {:current 1 :total 1} (get-in event [:counters :files])))
        (t/is (pos? (get-in event [:counters :pages :current])))
        (t/is (pos? (get-in event [:counters :pages :total])))))

    (t/testing "and every milestone is a stage and its counters, and nothing else"
      (t/is (every? #(<= (count (keys %)) 2) @reports)))))

(t/deftest a-job-that-is-no-longer-active-stops-the-work
  (let [profile (th/create-profile* 1)
        file-id (first (:file-ids (import-fixture! profile)))
        job-id  (th/create-job* :export-binfile {:profile-id (:id profile)})
        context (job-context job-id)]

    (t/testing "a running job is active, and answers with the cfg to thread it"
      (t/is (identical? th/*system* (bfj/check-active th/*system* context))))

    (t/testing "a cancelled job is not"
      (th/db-update! :job {:status "cancelled"} {:id job-id})
      (let [cause (try (bfj/check-active th/*system* context) nil
                       (catch Throwable raised raised))]
        (t/is (= :interrupt (th/ex-type cause)))
        (t/is (= "cancelled" (:status (ex-data cause))))))

    (t/testing "so the export stops at its next progress point"
      (let [output (tmp/tempfile* :suffix ".penpot")
            cause  (try
                     (run-export! (job-cfg) context
                                  #{file-id} :detach-libraries output)
                     nil
                     (catch Throwable raised raised))]
        (t/is (= :interrupt (th/ex-type cause)))
        (t/is (zero? (count (progress-events job-id))))))

    (t/testing "a job that already completed is not active either"
      (th/db-update! :job {:status "completed"} {:id job-id})
      (let [cause (try (bfj/check-active th/*system* context) nil
                       (catch Throwable raised raised))]
        (t/is (= :interrupt (th/ex-type cause)))))

    (t/testing "and neither is a job row that is gone"
      (th/db-force-delete :job {:id job-id})
      (let [cause (try (bfj/check-active th/*system* context) nil
                       (catch Throwable raised raised))]
        (t/is (= :interrupt (th/ex-type cause)))
        (t/is (= "gone" (:status (ex-data cause))))))))

(t/deftest an-interrupted-beat-stops-the-core-at-its-tap
  (let [steps   (atom [])
        context {:id (uuid/next)}
        cause   (ex-info "the job is no longer active"
                         {:type :interrupt :code :job-interrupted})]
    (with-redefs [jobs/heartbeat (fn [& _] (throw cause))]
      (let [raised (try (bfj/run th/*system* context
                                 (fn [_cfg]
                                   (swap! steps conj :before)
                                   ;; the sink calls the beat
                                   ;; inline, so its interrupt
                                   ;; aborts the run here
                                   (events/tap :progress {:section :file})
                                   (swap! steps conj :after)))
                        nil
                        (catch Throwable stopped stopped))]
        (t/is (= :interrupt (:type (ex-data raised))))
        (t/is (= :job-interrupted (:code (ex-data raised))))
        (t/is (= [:before] @steps)
              "the core stopped at its next tap, the step after never ran")))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; JOB-DEFS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(t/deftest the-binfile-jobs-follow-the-substrate-contract
  (let [defs (::jobs/defs th/*system*)]

    (t/testing "the export job produces the resource of a user job"
      (let [def (jobs/get-job-def defs :export-binfile)]
        (t/is (= :export (::jobs/family def)))
        (t/is (= :output (::jobs/resource-role def)))
        (t/is (fn? (::jobs/handler def)))))

    (t/testing "the import job consumes the resource of the job"
      (let [def (jobs/get-job-def defs :import-binfile)]
        (t/is (= :import (::jobs/family def)))
        (t/is (= :input (::jobs/resource-role def)))
        (t/is (fn? (::jobs/handler def)))))

    (t/testing "the params are a closed contract"
      (let [def     (jobs/get-job-def defs :export-binfile)
            file-id (uuid/next)]
        (t/is (true? ((::jobs/validator def)
                      {:file-ids #{file-id} :export-type :detach-libraries})))
        (t/testing "a job id never travels in params"
          (t/is (false? ((::jobs/validator def)
                         {:file-ids #{file-id} :export-type :detach-libraries
                          :job-id (uuid/next)}))))))))

(t/deftest the-export-handler-runs-through-the-registry
  (let [profile (th/create-profile* 1)
        file-id (first (:file-ids (import-fixture! profile)))
        job-id  (th/create-job* :export-binfile {:profile-id (:id profile)})
        result  (jobs/invoke (-> th/*system*
                                 (assoc ::jobs/name :export-binfile)
                                 (assoc ::jobs/params {:file-ids    #{file-id}
                                                       :export-type :detach-libraries})
                                 (assoc ::jobs/context (job-context job-id))))]

    (t/testing "the handler answers with the completion envelope of the job"
      (t/is (uuid? (:resource-id result)))
      (t/is (some? (get-in result [:result :resource-uri]))))

    (t/testing "the handler reported progress without knowing the job id"
      (t/is (pos? (count (progress-events job-id)))))))

(t/deftest the-import-handler-runs-through-the-registry
  (let [profile  (th/create-profile* 1)
        staged   (js/put-resource th/*system* (:id profile)
                                  {:content  (sto/content (fixture-path))
                                   :filename "package.penpot"
                                   :mtype    "application/zip"})
        job-id   (th/create-job* :import-binfile {:profile-id (:id profile)
                                                  :resource-id (:resource-id staged)})
        result   (jobs/invoke (-> th/*system*
                                  (assoc ::jobs/name :import-binfile)
                                  (assoc ::jobs/params {:project-id (:default-project-id profile)
                                                        :name       "imported"
                                                        :version    3})
                                  (assoc ::jobs/context (job-context job-id))))]

    (t/testing "the package of the job was imported"
      (t/is (pos? (count (:file-ids result)))))
    (t/testing "the handler reported progress without knowing the job id"
      (t/is (pos? (count (progress-events job-id)))))))

(t/deftest the-import-result-keeps-the-type-of-its-ids
  "The resolution of an import is a map keyed by the new file id, and it
  crosses a plain JSON column: the result schema the job-def declares is
  what keeps the keys and the values as uuids, which is what the dashboard
  wizard matches against the ids it holds. Without the decoder the substrate
  compiles from that schema, the key comes back as a keyword and the entry as
  text, and this test fails."
  (let [profile (th/create-profile* 1)
        {:keys [library-id output]} (export-link-later-package profile)

        staged  (js/put-resource th/*system* (:id profile)
                                 {:content  (sto/content output)
                                  :filename "package.penpot"
                                  :mtype    "application/zip"})
        job-id  (th/create-job* :import-binfile {:profile-id (:id profile)
                                                 :resource-id (:resource-id staged)})
        result  (jobs/invoke (-> th/*system*
                                 (assoc ::jobs/name :import-binfile)
                                 (assoc ::jobs/params {:project-id (:default-project-id profile)
                                                       :name       "imported"
                                                       :version    3})
                                 (assoc ::jobs/context (job-context job-id))))]

    (t/testing "the result of the job is stored as the substrate does"
      (t/is (= 1 (jobs/complete th/*system* :job-id job-id :result result))))

    (let [stored             (:result (jobs/get-job th/*system* job-id))
          [file-id file-res] (first (:resolution stored))]

      (t/testing "the resolution comes back keyed by the new file id"
        (t/is (= 1 (count (:resolution stored))))
        (t/is (uuid? file-id))
        (t/is (= file-id (:id file-res))))

      (t/testing "the ids of the entries keep their type"
        (t/is (= [file-id] (:file-ids stored)))
        (t/is (= [] (:done file-res)))
        (t/is (= 1 (count (:pending file-res))))

        (let [pending (first (:pending file-res))]
          (t/is (= library-id (:id pending)))
          (t/is (= 2 (count (:candidates pending))))
          (t/is (every? (comp uuid? :id) (:candidates pending)))
          (t/is (every? (comp uuid? :project-id) (:candidates pending))))))))

(t/deftest a-package-without-external-libraries-reads-with-an-empty-resolution
  (let [profile (th/create-profile* 1)
        staged  (js/put-resource th/*system* (:id profile)
                                 {:content  (sto/content (fixture-path))
                                  :filename "package.penpot"
                                  :mtype    "application/zip"})
        job-id  (th/create-job* :import-binfile {:profile-id (:id profile)
                                                 :resource-id (:resource-id staged)})
        result  (jobs/invoke (-> th/*system*
                                 (assoc ::jobs/name :import-binfile)
                                 (assoc ::jobs/params {:project-id (:default-project-id profile)
                                                       :name       "imported"
                                                       :version    3})
                                 (assoc ::jobs/context (job-context job-id))))]

    (t/is (= 1 (jobs/complete th/*system* :job-id job-id :result result)))

    (let [stored (:result (jobs/get-job th/*system* job-id))]

      (t/testing "a package without external libraries resolves nothing"
        (t/is (= {} (:resolution stored))))

      (t/testing "and the file ids come back typed"
        (t/is (= 1 (count (:file-ids stored))))
        (t/is (every? uuid? (:file-ids stored)))))))

(t/deftest a-version-1-result-reads-without-a-resolution
  "A version 1 import answers with the file ids and no resolution: the read
  path decodes it with the schema of the job-def, which declares the
  resolution as optional. Without that schema the file ids come back as text."
  (let [profile (th/create-profile* 1)
        file-id (uuid/next)
        job-id  (th/create-job* :import-binfile {:profile-id (:id profile)})]

    (t/is (= 1 (jobs/complete th/*system* :job-id job-id
                              :result {:file-ids [file-id]
                                       :name     "imported-v1"
                                       :version  1})))

    (let [stored (:result (jobs/get-job th/*system* job-id))]

      (t/testing "there is no resolution to decode"
        (t/is (nil? (:resolution stored))))

      (t/testing "and the file ids come back typed"
        (t/is (= [file-id] (:file-ids stored)))
        (t/is (= 1 (:version stored)))))))

(t/deftest a-result-of-a-retired-job-def-reads-as-plain-json
  "A registry that no longer knows the kind of the row answers with the JSON
  the column holds, keywordized keys and uuid as text, instead of raising."
  (let [profile (th/create-profile* 1)
        file-id (uuid/next)
        job-id  (th/create-job* :import-binfile {:profile-id (:id profile)})
        key     (keyword (str file-id))]

    (t/is (= 1 (jobs/complete th/*system* :job-id job-id
                              :result {:file-ids   [file-id]
                                       :resolution {file-id {:id      file-id
                                                             :name    "card.penpot"
                                                             :done    []
                                                             :pending []}}
                                       :name       "imported"
                                       :version    3})))

    (let [stored (:result (jobs/get-job (assoc th/*system* ::jobs/defs {}) job-id))]

      (t/testing "the resolution comes back as it is on disk"
        (t/is (= [key] (keys (:resolution stored))))
        (t/is (= (str file-id) (get-in stored [:resolution key :id]))))

      (t/testing "and so do the file ids"
        (t/is (= [(str file-id)] (:file-ids stored)))))))