<commit_before>(ns akvo.lumen.update
  (:require [akvo.lumen.boundary.error-tracker :as error-tracker]
            [akvo.lumen.import.common :as import]
            [akvo.lumen.import.csv]
            [akvo.lumen.import.flow]
            [akvo.lumen.lib :as lib]
            [akvo.lumen.transformation.engine :as engine]
            [akvo.lumen.util :as util]
            [clojure.java.jdbc :as jdbc]
            [clojure.set :as set]
            [clojure.string :as string]
            [clojure.tools.logging :as log]
            [hugsql.core :as hugsql]))

(hugsql/def-db-fns "akvo/lumen/job-execution.sql")
(hugsql/def-db-fns "akvo/lumen/transformation.sql")
(hugsql/def-db-fns "akvo/lumen/dataset.sql")

(defmulti adapt-transformation
  (fn [op-spec older-columns new-columns]
    (keyword (get op-spec "op"))))

(defmethod adapt-transformation :default
  [op-spec older-columns new-columns]
  op-spec)

(defn successful-update
  "On a successful update we need to create a new dataset-version that
  is similar to the previous one, except with an updated :version and
  pointing to the new table-name, imported-table-name and columns. We
  also delete the previous table-name and imported-table-name so we
  don't accumulate unused datasets on each update."
  [conn job-execution-id dataset-id table-name imported-table-name dataset-version
   transformations new-columns]
  (insert-dataset-version conn {:id (str (util/squuid))
                                :dataset-id dataset-id
                                :job-execution-id job-execution-id
                                :table-name table-name
                                :imported-table-name imported-table-name
                                :version (inc (:version dataset-version))
                                :columns new-columns
                                :transformations (vec transformations)})
  (touch-dataset conn {:id dataset-id})
  (drop-table conn {:table-name (:imported-table-name dataset-version)})
  (drop-table conn {:table-name (:table-name dataset-version)})
  (update-successful-job-execution conn {:id job-execution-id}))

(defn failed-update [conn job-execution-id reason]
  (update-failed-job-execution conn {:id job-execution-id
                                     :reason [reason]}))

(defn- apply-transformation-log [conn table-name importer-columns original-dataset-columns last-transformations dataset-id version]
  (let [importer-columns (mapv (fn [{:keys [title id type key caddisflyResourceUuid] :as column}]
                                 (cond-> {"type" (name type)
                                          "title" title
                                          "columnName" (name id)
                                          "sort" nil
                                          "direction" nil
                                          "hidden" false}
                                   key                   (assoc "key" (boolean key))
                                   caddisflyResourceUuid (assoc "caddisflyResourceUuid" caddisflyResourceUuid)))
                               importer-columns)]
    (update-dataset-version conn {:dataset-id      dataset-id
                                  :version         version
                                  :columns         importer-columns
                                  :transformations []})
    (loop [transformations  last-transformations
           importer-columns importer-columns
           version          (inc version)
           applied-txs      []]
      (if-let [transformation (first transformations)]
        (let [transformation                     (adapt-transformation transformation original-dataset-columns importer-columns)
              {:keys [success? message columns]} (engine/try-apply-operation
                                                  {:tenant-conn conn} table-name importer-columns transformation)]
          (when-not success? (throw
                              (ex-info (format "Failed to update due to transformation mismatch: %s" message) {})))
          (let [applied-txs (conj applied-txs
                                  (assoc transformation "changedColumns"
                                         (engine/diff-columns importer-columns columns)))]
            (update-dataset-version conn {:dataset-id      dataset-id
                                          :version         version
                                          :columns         columns
                                          :transformations applied-txs})
            (recur (rest transformations) columns (inc version) applied-txs)))
        [importer-columns applied-txs]))))

(defn compatible-columns? [imported-columns columns]
  (let [imported-columns (map (fn [column]
                                (cond-> {:id (keyword (get column "columnName"))<commit_msg>[#1808] Ensure multiple-type and multiple-id<commit_after>(ns akvo.lumen.update
  (:require [akvo.lumen.boundary.error-tracker :as error-tracker]
            [akvo.lumen.import.common :as import]
            [akvo.lumen.import.csv]
            [akvo.lumen.import.flow]
            [akvo.lumen.lib :as lib]
            [akvo.lumen.transformation.engine :as engine]
            [akvo.lumen.util :as util]
            [clojure.java.jdbc :as jdbc]
            [clojure.set :as set]
            [clojure.string :as string]
            [clojure.tools.logging :as log]
            [hugsql.core :as hugsql]))

(hugsql/def-db-fns "akvo/lumen/job-execution.sql")
(hugsql/def-db-fns "akvo/lumen/transformation.sql")
(hugsql/def-db-fns "akvo/lumen/dataset.sql")

(defmulti adapt-transformation
  (fn [op-spec older-columns new-columns]
    (keyword (get op-spec "op"))))

(defmethod adapt-transformation :default
  [op-spec older-columns new-columns]
  op-spec)

(defn successful-update
  "On a successful update we need to create a new dataset-version that
  is similar to the previous one, except with an updated :version and
  pointing to the new table-name, imported-table-name and columns. We
  also delete the previous table-name and imported-table-name so we
  don't accumulate unused datasets on each update."
  [conn job-execution-id dataset-id table-name imported-table-name dataset-version
   transformations new-columns]
  (insert-dataset-version conn {:id (str (util/squuid))
                                :dataset-id dataset-id
                                :job-execution-id job-execution-id
                                :table-name table-name
                                :imported-table-name imported-table-name
                                :version (inc (:version dataset-version))
                                :columns new-columns
                                :transformations (vec transformations)})
  (touch-dataset conn {:id dataset-id})
  (drop-table conn {:table-name (:imported-table-name dataset-version)})
  (drop-table conn {:table-name (:table-name dataset-version)})
  (update-successful-job-execution conn {:id job-execution-id}))

(defn failed-update [conn job-execution-id reason]
  (update-failed-job-execution conn {:id job-execution-id
                                     :reason [reason]}))

(defn- apply-transformation-log [conn table-name importer-columns original-dataset-columns last-transformations dataset-id version]
  (let [importer-columns (mapv (fn [{:keys [title id type key caddisflyResourceUuid
                                            multiple-id multiple-type] :as column}]
                                 (cond-> {"type" (name type)
                                          "title" title
                                          "columnName" (name id)
                                          "sort" nil
                                          "direction" nil
                                          "hidden" false}
                                   key                   (assoc "key" (boolean key))
                                   caddisflyResourceUuid (assoc "caddisflyResourceUuid" caddisflyResourceUuid)
                                   multiple-type (assoc "multipleType" multiple-type)
                                   multiple-id (assoc "multipleId" multiple-id)))
                               importer-columns)]
    (update-dataset-version conn {:dataset-id      dataset-id
                                  :version         version
                                  :columns         importer-columns
                                  :transformations []})
    (loop [transformations  last-transformations
           importer-columns importer-columns
           version          (inc version)
           applied-txs      []]
      (if-let [transformation (first transformations)]
        (let [transformation                     (adapt-transformation transformation original-dataset-columns importer-columns)
              {:keys [success? message columns]} (engine/try-apply-operation
                                                  {:tenant-conn conn} table-name importer-columns transformation)]
          (when-not success? (throw
                              (ex-info (format "Failed to update due to transformation mismatch: %s" message) {})))
          (let [applied-txs (conj applied-txs
                                  (assoc transformation "changedColumns"
                                         (engine/diff-columns importer-columns columns)))]
            (update-dataset-version conn {:dataset-id      dataset-id
                                          :version         version
                                          :columns         columns
                                          :transformations applied-txs})
            (recur (rest transformations) columns (inc version) applied-txs)))
        [importer-columns applied-txs]))))

(defn compatible-columns? [imported-columns columns]
  (let [imported-columns (map (fn [column]
                                (cond-> {:id (keyword (get column "columnName"))
{'fraction_non_alphanumeric': 0.09518566710039035, 'fraction_numerical': 0.0004003603242918627, 'mean_word_length': 1.5392630241423126, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 62, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6794110', 'n_tokens_mistral': 2524, 'n_tokens_neox': 2417, 'n_words': 633}