CMR-7192: Redefine subscription uniqueness to use query as main parameter (#1210)

* CMR-7192 - Adding the database migration which adds the normalize column

* CMR-7192 - Adding the database migration which adds the normalize column

* CMR-7192 - handoff to mitch

* CMR-7192: Bugfixes, test additions, general cleanup for PR

* CMR-7192: Cleanup from PR feedback

* CMR-7192: Test fixes

* CMR-7192: Renamed function

* CMR-7192: Test fixes

* CMR-7192: Added index to db migration

* CMR-7192: Logic and test fixes

* CMR-7192: Fixed Thomas' spelling error :)

* CMR-7192: Test fix

* CMR-7192: Feedback from PR

* CMR-7192: Test fix, added test, addressed PR comments

* CMR-7192: Removed PFN constraint from subscription

* CMR-7216: Removed test based on PFN-constraints

* CMR-7192: Whitespace cleanup

Co-authored-by: Thomas Cherry <thomas.a.cherry@nasa.gov>

Author: mitchstartzel

common-app-lib/src/cmr/common_app/services/ingest/subscription_common.clj

@@ -0,0 +1,18 @@
+(ns cmr.common-app.services.ingest.subscription-common
+  "This contains the code for the scheduled subscription code to be shared
+   between metadata-db and ingest."
+  (:require
+   [clojure.string :as string]))
+
+(defn normalize-parameters
+  "Returns a normalized url parameter string by splitting the string of parameters on '&' and
+   sorting them alphabetically"
+  [parameter-string]
+  (when parameter-string
+    (-> (if (string/starts-with? parameter-string "?")
+          (subs parameter-string 1)
+          parameter-string)
+        (string/split #"&")
+        sort
+        (as-> $ (string/join "&" $))
+        (string/trim))))

common-app-lib/test/cmr/common_app/test/services/ingest/subscription.clj

@@ -0,0 +1,22 @@
+(ns cmr.common-app.test.services.ingest.subscription
+  (:require
+   [clojure.test :refer :all]
+   [cmr.common-app.services.ingest.subscription-common :as sub-common]))
+
+(deftest normalize-parameters-test
+  "Query normalization, should be sorted parameters"
+  (testing "With a leading question mark"
+   (let [query "provider=PROV1&instrument=1B&instrument=2B"
+         expected "instrument=1B&instrument=2B&provider=PROV1"
+         actual (sub-common/normalize-parameters query)]
+     (is (= expected actual))))
+  (testing "Without a leading question mark"
+   (let [query "provider=PROV1&instrument=1B&instrument=2B"
+         expected "instrument=1B&instrument=2B&provider=PROV1"
+         actual (sub-common/normalize-parameters query)]
+     (is (= expected actual))))
+  (testing "Empty string"
+   (let [query ""
+         expected ""
+         actual (sub-common/normalize-parameters query)]
+     (is (= expected actual)))))

ingest-app/src/cmr/ingest/api/subscriptions.clj

@@ -7,13 +7,16 @@
    [cmr.acl.core :as acl]
    [cmr.common-app.api.enabled :as common-enabled]
    [cmr.common-app.api.launchpad-token-validation :as lt-validation]
+   [cmr.common-app.services.ingest.subscription-common :as sub-common]
    [cmr.common.log :refer [debug info warn error]]
    [cmr.common.services.errors :as errors]
    [cmr.ingest.api.core :as api-core]
    [cmr.ingest.services.ingest-service :as ingest]
+   [cmr.ingest.services.subscriptions-helper :as jobs]
    [cmr.ingest.validation.validation :as v]
    [cmr.transmit.access-control :as access-control]
    [cmr.transmit.metadata-db :as mdb]
+   [cmr.transmit.metadata-db2 :as mdb2]
    [cmr.transmit.urs :as urs])
   (:import
    [java.util UUID]))
@@ -57,6 +60,40 @@
            subscriber-id
            concept-id)))))
 
+(defn- check-duplicate-subscription
+  "The query used by a subscriber for a collection should be unique to prevent
+   redundent emails from being sent to them. This function will check that a
+   subscription is unique for the following conditions: native-id, collection-id,
+   subscriber-id, normalized-query."
+  [request-context subscription]
+  (let [native-id (:native-id subscription)
+        provider-id (:provider-id subscription)
+        metadata (-> (:metadata subscription) (json/decode true))
+        normalized-query (:normalized-query subscription)
+        collection-id (:CollectionConceptId metadata)
+        subscriber-id (:SubscriberId metadata)
+        ;; Find concepts with matching collection-concept-id, normalized-query, and subscriber-id
+        duplicate-queries (mdb/find-concepts
+                           request-context
+                           {:collection-concept-id collection-id
+                            :normalized-query normalized-query
+                            :subscriber-id subscriber-id
+                            :exclude-metadata true
+                            :latest true}
+                           :subscription)
+        ;;we only want to look at non-deleted subscriptions
+        active-duplicate-queries (remove :deleted duplicate-queries)]
+     ;;If there is at least one duplicate subscription,
+     ;;We need to make sure it has the same native-id, or else reject the ingest
+     (when (and (> (count active-duplicate-queries) 0)
+                (every? #(not= native-id (:native-id %)) active-duplicate-queries))
+       (errors/throw-service-error
+        :conflict
+        (format (str "The subscriber-id [%s] has already subscribed to the "
+                     "collection with concept-id [%s] using the query [%s]. "
+                     "Subscribers must use unique queries for each Collection.")
+                subscriber-id collection-id  normalized-query)))))
+
 (defn- get-subscriber-id
   "Returns the subscriber id of the given subscription concept by parsing its metadata."
   [sub-concept]
@@ -154,23 +191,25 @@
         (get-unique-native-id context subscription))
       native-id)))
 
-(defn- add-id-and-email-to-metadata-if-missing
+(defn- add-id-to-metadata-if-missing
    "If SubscriberId is provided, use it. Else, get it from the token."
   [context metadata]
-  (let [{subscriber :SubscriberId} metadata
-        subscriber (if-not subscriber
+  (let [subscriber-id (:SubscriberId metadata)
+        subscriber (if-not subscriber-id
                      (api-core/get-user-id-from-token context)
-                     subscriber)]
+                     subscriber-id)]
     (assoc metadata :SubscriberId subscriber)))
 
-
-(defn add-id-and-email-if-missing
-  "Parses and generates the metadata, such that add-id-and-email-to-metadata-if-missing
-  can focus on insertion logic."
+(defn- add-fields-if-missing
+  "Parses and generates the metadata, such that add-id-to-metadata-if-missing
+  can focus on insertion logic. Also adds normalized-query to the concept."
   [context subscription]
   (let [metadata (json/parse-string (:metadata subscription) true)
-        new-metadata (add-id-and-email-to-metadata-if-missing context metadata)]
-    (assoc subscription :metadata (json/generate-string new-metadata))))
+        new-metadata (add-id-to-metadata-if-missing context metadata)
+        normalized (sub-common/normalize-parameters (:Query metadata))]
+    (assoc subscription
+           :metadata (json/generate-string new-metadata)
+           :normalized-query normalized)))
 
 (defn create-subscription
   "Processes a request to create a subscription. A native id will be generated."
@@ -186,9 +225,10 @@
                             headers)
           native-id (get-unique-native-id request-context tmp-subscription)
           new-subscription (assoc  tmp-subscription :native-id native-id)
-          newer-subscription (add-id-and-email-if-missing request-context new-subscription)
+          newer-subscription (add-fields-if-missing request-context new-subscription)
           subscriber-id (get-subscriber-id newer-subscription)]
       (check-ingest-permission request-context provider-id subscriber-id)
+      (check-duplicate-subscription request-context newer-subscription)
       (perform-subscription-ingest request-context newer-subscription headers))))
 
 (defn create-subscription-with-native-id
@@ -209,9 +249,10 @@
                             body
                             content-type
                             headers)
-          new-subscription (add-id-and-email-if-missing request-context tmp-subscription)
+          new-subscription (add-fields-if-missing request-context tmp-subscription)
           subscriber-id (get-subscriber-id new-subscription)]
       (check-ingest-permission request-context provider-id subscriber-id)
+      (check-duplicate-subscription request-context new-subscription)
       (perform-subscription-ingest request-context new-subscription headers))))
 
 (defn create-or-update-subscription-with-native-id
@@ -236,9 +277,10 @@
                                            :latest true}
                                           :subscription))]
                          (get-in original-subscription [:extra-fields :subscriber-id]))
-        new-subscription (add-id-and-email-if-missing request-context tmp-subscription)
+        new-subscription (add-fields-if-missing request-context tmp-subscription)
         new-subscriber (get-subscriber-id new-subscription)]
     (check-ingest-permission request-context provider-id new-subscriber old-subscriber)
+    (check-duplicate-subscription request-context new-subscription)
     (perform-subscription-ingest request-context new-subscription headers)))
 
 (defn delete-subscription

ingest-app/src/cmr/ingest/services/ingest_service/subscription.clj

@@ -11,7 +11,8 @@
   (assoc concept :extra-fields
                  {:subscription-name (:Name subscription)
                   :collection-concept-id (:CollectionConceptId subscription)
-                  :subscriber-id (:SubscriberId subscription)}))
+                  :subscriber-id (:SubscriberId subscription)
+                  :normalized-query (:normalized-query concept)}))
 
 (defn-timed save-subscription
   "Store a subscription concept in mdb and indexer."

ingest-app/src/cmr/ingest/services/subscriptions_helper.clj

@@ -5,6 +5,7 @@
    [clj-time.core :as t]
    [clojure.spec.alpha :as spec]
    [clojure.string :as string]
+   [cmr.common-app.services.search.params :as params]
    [cmr.common.config :as cfg :refer [defconfig]]
    [cmr.common.log :refer (debug info warn error)]
    [cmr.transmit.access-control :as access-control]

metadata-db-app/README.md

@@ -404,6 +404,7 @@ Supported combinations of concept type and parameters:
   * tag associations with concept-id, native-id, tag-key, associated-concept-id, associated-revision-id
   * services with concept-id, provider-id, native-id
   * service associations with concept-id, native-id, service-concept-id, associated-concept-id, associated-revision-id
+  * subscriptions with collection-concept-id, provider-id, subscriber-id, native-id, concept-id, normalized-query
   * variables with concept-id, provider-id, native-id
   * variable associations with concept-id, native-id, variable-concept-id, associated-concept-id, associated-revision-id
 

metadata-db-app/int-test/cmr/metadata_db/int_test/concepts/delete_test.clj

@@ -23,7 +23,7 @@
 (deftest delete-concepts-test
   (doseq [concept-type [:collection :granule :service
                         :service-association :subscription :tool :tool-association]]
-    (cd-spec/general-delete-test concept-type ["REG_PROV" "SMAL_PROV"])))
+   (cd-spec/general-delete-test concept-type ["REG_PROV" "SMAL_PROV"])))
 
 (deftest delete-variable-and-association-concepts-test
   (cd-spec/general-delete-variable-and-association-test ["REG_PROV" "SMAL_PROV"]))

metadata-db-app/int-test/cmr/metadata_db/int_test/concepts/subscription_save_test.clj

@@ -36,14 +36,6 @@
   (let [concept (concepts/create-concept :subscription "PROV1" 2)]
     (util/concept-created-at-assertions "subscription" concept)))
 
-(deftest save-subscription-with-conflicting-native-id
-  (let [concept (concepts/create-concept :subscription "PROV1" 1)]
-    (util/concept-with-conflicting-native-id-assertions
-     "subscription"
-     :subscription-name
-     concept
-     "sub-native-different")))
-
 (deftest save-subscription-failures-test
   (testing "saving invalid subscription"
     (are3 [subscription exp-status exp-errors]

metadata-db-app/int-test/cmr/metadata_db/int_test/concepts/utils/subscription.clj

@@ -22,7 +22,8 @@
   (let [native-id (str "sub-native" uniq-num)
         extra-fields (merge {:subscription-name (str "subname" uniq-num)
                              :subscriber-id (str "subid" uniq-num)
-                             :collection-concept-id "C1234-PROV1"}
+                             :collection-concept-id "C12345-PROV1"
+                             :normalized-query (str "instrument=" uniq-num "B")}
                             (:extra-fields attributes))
         attributes (merge {:user-id (str "user" uniq-num)
                            :format "application/json"

metadata-db-app/src/cmr/metadata_db/data/oracle/concepts/subscription.clj

@@ -18,6 +18,7 @@
           (assoc :concept-type :subscription)
           (assoc :provider-id (:provider_id result))
           (assoc :user-id (:user_id result))
+          (assoc-in [:extra-fields :normalized-query] (:normalized_query result))
           (assoc-in [:extra-fields :subscription-name] (:subscription_name result))
           (assoc-in [:extra-fields :subscriber-id] (:subscriber_id result))
           (add-last-notified-at-if-present result db)
@@ -28,14 +29,15 @@
   [concept]
   (let [{{:keys [subscription-name
                  subscriber-id
-                 collection-concept-id]} :extra-fields
+                 collection-concept-id
+                 normalized-query]} :extra-fields
          user-id :user-id
          provider-id :provider-id} concept
         [cols values] (concepts/concept->common-insert-args concept)]
     [(concat cols ["provider_id" "user_id" "subscription_name"
-                   "subscriber_id" "collection_concept_id"])
+                   "subscriber_id" "collection_concept_id" "normalized_query"])
      (concat values [provider-id user-id subscription-name
-                     subscriber-id collection-concept-id])]))
+                     subscriber-id collection-concept-id normalized-query])]))
 
 (defmethod concepts/concept->insert-args [:subscription false]
   [concept _]

metadata-db-app/src/cmr/metadata_db/data/oracle/search.clj

@@ -35,7 +35,8 @@
    :humanizer (into common-columns [:user_id])
    :subscription (into common-columns
                        [:provider_id :subscription_name :subscriber_id
-                        :email_address :collection_concept_id :user_id])
+                        :email_address :collection_concept_id :user_id
+                        :normalized_query])
    :variable (into common-columns [:provider_id :variable_name :measurement :user_id :fingerprint])
    :variable-association (into common-columns
                                [:associated_concept_id :associated_revision_id

metadata-db-app/src/cmr/metadata_db/migrations/074_add_normalized_to_subscription_table.clj

@@ -0,0 +1,53 @@
+(ns cmr.metadata-db.migrations.074-add-normalized-to-subscription-table
+  (:require
+   [cheshire.core :as json]
+   [cmr.common.util :as util]
+   [cmr.common-app.services.ingest.subscription-common :as sub-common]
+   [config.mdb-migrate-helper :as helper]))
+
+(defn result->query
+  "From a Database result, extract out the query and convert it to a usable string.
+   The algorithm on how to normalize the query may change over time, for this
+   migration version 1 will be used."
+  [result]
+  (-> (:metadata result)
+        util/gzip-blob->string
+        (as-> % (json/parse-string %, true))
+        :Query
+        sub-common/normalize-parameters))
+
+(defn populate-new-column
+  "Pull out the requested query from the matadata, then normalize the data and
+   populate the new 'normalized_query' column."
+  []
+  (doseq [result (helper/query "SELECT id, concept_id, metadata FROM cmr_subscriptions")]
+    (let [id (:id result)
+          query (result->query result)
+          sql-statment (format "UPDATE cmr_subscriptions SET normalized_query='%s' WHERE id=%s" query id)]
+      (helper/sql sql-statment))))
+
+(defn- create-subscription-index
+  []
+  (helper/sql "CREATE INDEX cmr_subs_ccisinq ON CMR_SUBSCRIPTIONS
+              (collection_concept_id, subscriber_id, normalized_query)"))
+
+(defn up
+    "Migrates the database up to version 74."
+    []
+    (println "cmr.metadata-db.migrations.074-update-subscription-table up...")
+
+    ;; The Normalized_Query column holds a version of the Subscription Query which
+    ;; has been normalized to a standard followed by both EDSC and CMR. Both
+    ;; applications can assume the normalized query is the same despite changes
+    ;; in query order or other such nominal differences. Simple string compair
+    ;; can be used on the normalized query.
+    (helper/sql "alter table cmr_subscriptions add normalized_query VARCHAR2(4000)")
+    (populate-new-column)
+    (create-subscription-index))
+
+(defn down
+  "Migrates the database down from version 74."
+  []
+  (println "cmr.metadata-db.migrations.074-update-subscription-table down.")
+  (helper/sql "DROP INDEX cmr_subs_ccisinq")
+  (helper/sql "alter table cmr_subscriptions drop column normalized_query"))

metadata-db-app/src/cmr/metadata_db/services/concept_constraints.clj

@@ -158,8 +158,7 @@
    :granule (when (enforce-granule-ur-constraint) [granule-ur-unique-constraint])
    :acl [(unique-field-constraint :acl-identity)]
    :service [(partial pfn-constraint :service-name)]
-   :tool [(partial pfn-constraint :tool-name)]
-   :subscription [(partial pfn-constraint :subscription-name)]})
+   :tool [(partial pfn-constraint :tool-name)]})
 
 (defn perform-post-commit-constraint-checks
   "Perform the post commit constraint checks aggregating any constraint

metadata-db-app/src/cmr/metadata_db/services/concept_validations.clj

@@ -47,7 +47,7 @@
    :tool {true #{}
           false #{:tool-name}}
    :subscription {true #{}
-                  false #{:subscription-name :subscriber-id :collection-concept-id}}
+                  false #{:subscription-name :subscriber-id :collection-concept-id :normalized-query}}
    :tag-association {true #{}
                      false #{:associated-concept-id :associated-revision-id}}
    :variable {true #{}

metadata-db-app/src/cmr/metadata_db/services/search_service.clj

@@ -26,7 +26,7 @@
    :access-group default-supported-find-parameters
    :acl default-supported-find-parameters
    :humanizer #{:concept-id :native-id}
-   :subscription #{:provider-id :concept-id :native-id}
+   :subscription #{:provider-id :concept-id :native-id :collection-concept-id :subscriber-id :normalized-query}
    :variable #{:provider-id :concept-id :native-id}
    :variable-association #{:concept-id :native-id :associated-concept-id :associated-revision-id
                            :variable-concept-id}

system-int-test/src/cmr/system_int_test/utils/subscription_util.clj

@@ -9,6 +9,7 @@
    [cmr.system-int-test.data2.core :as d]
    [cmr.system-int-test.data2.umm-spec-subscription :as data-umm-sub]
    [cmr.system-int-test.system :as s]
+   [cmr.system-int-test.utils.index-util :as index]
    [cmr.system-int-test.utils.ingest-util :as ingest-util]
    [cmr.system-int-test.utils.search-util :as search]
    [cmr.system-int-test.utils.url-helper :as urls]
@@ -182,3 +183,15 @@
   "Verifies the subscription references"
   [subscriptions response]
   (d/refs-match? subscriptions response))
+
+(defn create-subscription-and-index
+  [collection name user query]
+  (let [concept {:Name (or name "test_sub_prov1")
+                 :SubscriberId (or user "user2")
+                 :CollectionConceptId (:concept-id collection)
+                 :Query (or query "instrument=1B")}
+        subscription (make-subscription-concept concept)
+        options {:token "mock-echo-system-token"}
+        result (ingest-subscription subscription options)]
+    (index/wait-until-indexed)
+    result))