Merge pull request #320 from yetanalytics/sql-194

[WIP] [Sql 194] - Postgres Reactions Port

src/db/postgres/lrsql/postgres/data.clj

@@ -4,7 +4,7 @@
             [lrsql.util :as u])
   (:import [clojure.lang IPersistentMap]
            [org.postgresql.util PGobject]
-           [java.sql PreparedStatement]))
+           [java.sql PreparedStatement ResultSetMetaData]))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; PGObject
@@ -17,11 +17,11 @@
     (.setValue (u/write-json-str jsn))))
 
 (defn- pg-object->json
-  [^PGobject pg-obj]
+  [kw-labels label ^PGobject pg-obj]
   (let [type  (.getType pg-obj)
         value (.getValue pg-obj)]
     (if (#{"jsonb" "json"} type)
-      (u/parse-json value)
+      (u/parse-json value :keyword-keys? (some? (kw-labels label)))
       (throw (ex-info "Invalid PostgreSQL JSON type"
                       {:type       ::invalid-postgres-json
                        :json-type  type
@@ -32,13 +32,13 @@
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (defn set-read-pgobject->json!
-  []
+  [kw-labels]
   (extend-protocol ReadableColumn
     PGobject
-    (read-column-by-label [^PGobject pg-obj _]
-      (pg-object->json pg-obj))
-    (read-column-by-index [^PGobject pg-obj _2 _3]
-      (pg-object->json pg-obj))))
+    (read-column-by-label [^PGobject pg-obj ^String label]
+      (pg-object->json kw-labels label pg-obj))
+    (read-column-by-index [^PGobject pg-obj ^ResultSetMetaData rsmeta ^long i]
+      (pg-object->json kw-labels (.getColumnLabel rsmeta i) pg-obj))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Write
@@ -59,3 +59,14 @@
   "Returns a properly formatted hug input map to inject a timezone id into a
   query needing a timezone id"
   {:tz-id (str "'" u/local-zone-id "'")})
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; JSON Field Coercion
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+
+(def type->pg-type
+  {:bool   "BOOLEAN"
+   :int    "INTEGER"
+   :dec    "DECIMAL"
+   :string "TEXT"
+   :json   "JSONB"})

src/db/postgres/lrsql/postgres/record.clj

@@ -64,6 +64,11 @@
       (migrate-state-document-last-modified! tx pd/local-tz-input)
       (migrate-activity-profile-document-last-modified! tx pd/local-tz-input)
       (migrate-agent-profile-document-last-modified! tx pd/local-tz-input))
+    (create-reaction-table! tx)
+    (when-not (some? (query-xapi-statement-reaction-id-exists tx))
+      (xapi-statement-add-reaction-id! tx))
+    (when-not (some? (query-xapi-statement-trigger-id-exists tx))
+      (xapi-statement-add-trigger-id! tx))
     (if (-> tuning :config :enable-jsonb)
       (migrate-to-jsonb! tx)
       (migrate-to-json! tx)))
@@ -204,7 +209,9 @@
   bp/BackendIOSetter
   (-set-read! [_]
     (bd/set-read-time->instant!)
-    (pd/set-read-pgobject->json!))
+    (pd/set-read-pgobject->json!
+     #{"ruleset"
+       "error"}))
   (-set-write! [_]
     ;; next.jdbc automatically sets the reading of Instants as java.sql.Dates
     (pd/set-write-json->pgobject! (if (-> tuning :config :enable-jsonb)
@@ -220,4 +227,45 @@
   (-query-platform-frequency [_ tx]
     (query-platform-frequency tx))
   (-query-timeline [_ tx input]
-    (query-timeline tx input)))
+    (query-timeline tx input))
+
+  bp/ReactionBackend
+  (-insert-reaction! [_ tx params]
+    (insert-reaction! tx params))
+  (-update-reaction! [_ tx params]
+    (update-reaction! tx params))
+  (-delete-reaction! [_ tx params]
+    (delete-reaction! tx params))
+  (-error-reaction! [_ tx params]
+    (error-reaction! tx params))
+  (-snip-json-extract [_ {:keys [datatype] :as params}]
+    (if (-> tuning :config :enable-jsonb)
+      (snip-jsonb-extract (assoc params :type (datatype pd/type->pg-type)))
+      (snip-json-extract (assoc params :type (datatype pd/type->pg-type)))))
+  (-snip-val [_ params]
+    (snip-val params))
+  (-snip-col [_ params]
+    (snip-col params))
+  (-snip-clause [_ params]
+    (snip-clause params))
+  (-snip-and [_ params]
+    (snip-and params))
+  (-snip-or [_ params]
+    (snip-or params))
+  (-snip-not [_ params]
+    (snip-not params))
+  (-snip-contains [_ {:keys [datatype] :as params}]
+    (if (-> tuning :config :enable-jsonb)
+      (snip-contains-jsonb (assoc params :type (datatype pd/type->pg-type)))
+      (snip-contains-json (assoc params :type (datatype pd/type->pg-type)))))
+  (-snip-query-reaction [_ params]
+    (snip-query-reaction params))
+  (-query-reaction [_ tx params]
+    (query-reaction tx params))
+  (-query-active-reactions [_ tx]
+    (query-active-reactions tx))
+  (-query-all-reactions [_ tx]
+    (query-all-reactions tx))
+  (-query-reaction-history [_ tx params]
+    (query-reaction-history tx params)))
+

src/db/postgres/lrsql/postgres/sql/ddl.sql

@@ -358,3 +358,43 @@ ALTER TABLE activity ALTER COLUMN payload SET DATA TYPE JSONB;
 ALTER TABLE xapi_statement ALTER COLUMN payload SET DATA TYPE JSON;
 ALTER TABLE actor ALTER COLUMN payload SET DATA TYPE JSON;
 ALTER TABLE activity ALTER COLUMN payload SET DATA TYPE JSON;
+
+/* Migration 2023-07-21-00 - Add Reaction Table */
+
+-- :name create-reaction-table!
+-- :command :execute
+-- :doc Create the `reaction` table if it does not yet exist.
+CREATE TABLE IF NOT EXISTS reaction (
+  id           UUID PRIMARY KEY,
+  ruleset      JSON NOT NULL,         -- serialized reaction spec
+  created      TIMESTAMP NOT NULL,    -- timestamp
+  modified     TIMESTAMP NOT NULL,    -- timestamp
+  active       BOOLEAN,               -- true/false/null - active/inactive/soft delete
+  error        JSON                   -- serialized error
+);
+
+-- :name query-xapi-statement-reaction-id-exists
+-- :command :query
+-- :result :one
+-- :doc Query to see if `xapi_statement.reaction_id` exists.
+SELECT 1 FROM information_schema.columns WHERE table_name = 'xapi_statement' AND column_name = 'reaction_id';
+
+-- :name xapi-statement-add-reaction-id!
+-- :command :execute
+-- :doc Adds `xapi_statement.reaction_id` and associated fk and index
+ALTER TABLE xapi_statement ADD COLUMN reaction_id UUID;
+ALTER TABLE xapi_statement ADD CONSTRAINT stmt_reaction_id_fk FOREIGN KEY (reaction_id) REFERENCES reaction(id);
+CREATE INDEX IF NOT EXISTS stmt_reaction_id_idx ON xapi_statement(reaction_id);
+
+-- :name query-xapi-statement-trigger-id-exists
+-- :command :query
+-- :result :one
+-- :doc Query to see if `xapi_statement.trigger_id` exists.
+SELECT 1 FROM information_schema.columns WHERE table_name = 'xapi_statement' AND column_name = 'trigger_id';
+
+-- :name xapi-statement-add-trigger-id!
+-- :command :execute
+-- :doc Adds `xapi_statement.trigger_id` and associated fk and index
+ALTER TABLE xapi_statement ADD COLUMN trigger_id UUID;
+ALTER TABLE xapi_statement ADD CONSTRAINT stmt_trigger_id_fk FOREIGN KEY (trigger_id) REFERENCES xapi_statement(statement_id);
+CREATE INDEX IF NOT EXISTS stmt_trigger_id_idx ON xapi_statement(trigger_id);

src/db/postgres/lrsql/postgres/sql/delete.sql

@@ -61,3 +61,14 @@ DELETE FROM credential_to_scope
 WHERE api_key = :api-key
 AND secret_key = :secret-key
 AND scope = :scope::scope_enum;
+
+-- :name delete-reaction!
+-- :command :execute
+-- :result :affected
+-- :doc (Soft) delete a reaction.
+UPDATE reaction
+SET
+  active = NULL,
+  modified = :modified
+WHERE
+  id = :reaction-id

src/db/postgres/lrsql/postgres/sql/insert.sql

@@ -5,9 +5,9 @@
 -- :result :affected
 -- :doc Insert a new statement with statement resource params.
 INSERT INTO xapi_statement (
-  id, statement_id, registration, verb_iri, is_voided, payload, timestamp, stored
+  id, statement_id, registration, verb_iri, is_voided, payload, timestamp, stored, reaction_id, trigger_id
 ) VALUES (
-  :primary-key, :statement-id, :registration, :verb-iri, :voided?, :payload, :timestamp, :stored
+  :primary-key, :statement-id, :registration, :verb-iri, :voided?, :payload, :timestamp, :stored, :reaction-id, :trigger-id
 );
 
 -- :name insert-actor!
@@ -149,3 +149,13 @@ INSERT INTO credential_to_scope (
 ) VALUES (
   :primary-key, :api-key, :secret-key, :scope::scope_enum
 );
+
+-- :name insert-reaction!
+-- :command :insert
+-- :result :affected
+-- :doc Given a primary key, serialized ruleset and status, insert a new reaction into the `reaction` table.
+INSERT INTO reaction (
+  id, ruleset, active, created, modified
+) VALUES (
+  :primary-key, :ruleset, :active, :created, :modified
+);

src/db/postgres/lrsql/postgres/sql/query.sql

@@ -351,3 +351,79 @@ WHERE id > :since-id
   AND id <= :until-id
 GROUP BY stored_time
 ORDER BY stored_time ASC;
+
+/* Statement Reactions */
+
+-- :snip snip-json-extract
+json_extract_path_text(:i:col, :v*:path):::sql:type
+
+-- :snip snip-jsonb-extract
+-- :doc essentially identical alternate for snip-json-extract for jsonb mode
+jsonb_extract_path_text(:i:col, :v*:path):::sql:type
+
+-- :snip snip-val
+:v:val
+
+-- :snip snip-col
+:i:col
+
+-- :snip snip-clause
+:snip:left :sql:op :snip:right
+
+-- :snip snip-and
+--~ (str "(" (apply str (interpose " AND " (map-indexed (fn [idx _] (str ":snip:clauses." idx)) (:clauses params)))) ")")
+
+-- :snip snip-or
+--~ (str "(" (apply str (interpose " OR " (map-indexed (fn [idx _] (str ":snip:clauses." idx)) (:clauses params)))) ")")
+
+-- :snip snip-not
+(NOT :snip:clause)
+
+-- :snip snip-contains-json
+-- :doc Does the json at col and path contain the given value? A special case with differing structure across backends
+(SELECT TRUE FROM json_array_elements_text(json_extract_path(:i:col, :v*:path)) WHERE value:::sql:type = :snip:right)
+
+-- :snip snip-contains-jsonb
+-- :doc Does the jsonb at col and path contain the given value? A special case with differing structure across backends
+(SELECT TRUE FROM jsonb_array_elements_text(jsonb_extract_path(:i:col, :v*:path)) WHERE value:::sql:type = :snip:right)
+
+-- :snip snip-query-reaction
+SELECT :i*:select
+FROM :i*:from
+WHERE :snip:where;
+
+-- :name query-reaction
+:snip:sql
+
+-- :name query-active-reactions
+-- :command :query
+-- :result :many
+-- :doc Return all active `reaction` ids and rulesets
+SELECT id, ruleset
+FROM reaction
+WHERE active = true;
+
+-- :name query-all-reactions
+-- :command :query
+-- :result :many
+-- :doc Query all active and inactive reactions
+SELECT id, ruleset, active, created, modified, error
+FROM reaction
+WHERE active IS NOT NULL;
+
+-- :name query-reaction-history
+-- :command :query
+-- :result :many
+-- :doc For a given statement id, return all reactions (if any) leading to the issuance of that statement.
+WITH RECURSIVE trigger_history (statement_id, reaction_id, trigger_id) AS (
+  SELECT s.statement_id, s.reaction_id, s.trigger_id
+  FROM xapi_statement s
+  WHERE s.statement_id = :statement-id
+  UNION ALL
+  SELECT s.statement_id, s.reaction_id, s.trigger_id
+  FROM xapi_statement s
+  JOIN trigger_history th ON th.trigger_id = s.statement_id
+)
+SELECT reaction_id
+FROM trigger_history
+WHERE reaction_id IS NOT NULL;

src/db/postgres/lrsql/postgres/sql/update.sql

@@ -76,3 +76,26 @@ UPDATE admin_account
 SET
   passhash = :new-passhash
 WHERE id = :account-id;
+
+-- :name update-reaction!
+-- :command :execute
+-- :result :affected
+-- :doc Update the `ruleset` and/or `active` status of a reaction.
+UPDATE reaction
+SET
+--~ (when (:ruleset params) "ruleset = :ruleset,")
+--~ (when (or (true? (:active params)) (false? (:active params))) "active = :active,")
+--~ (when (:ruleset params) "error = null,")
+  modified = :modified
+WHERE id = :reaction-id
+
+-- :name error-reaction!
+-- :command :execute
+-- :result :affected
+-- :doc Set the `error` column on a reaction and make it inactive.
+UPDATE reaction
+SET
+  error = :error,
+  active = false,
+  modified = :modified
+WHERE id = :reaction-id

src/main/lrsql/ops/query/reaction.clj

@@ -5,6 +5,7 @@
             [lrsql.spec.reaction :as rs]
             [lrsql.util.reaction :as ru]
             [lrsql.ops.util.reaction :as ur]
+            [lrsql.util :as u]
             [xapi-schema.spec :as xs]
             [clojure.tools.logging :as log]))
 
@@ -113,7 +114,7 @@
                                          identityPaths statement)]
                  (if-not statement-identity
                    [] ;; ignore
-                   (let [stored (get statement "stored")
+                   (let [stored (u/str->time (get statement "stored"))
                          [q-success ?q-result-or-error]
                          (reaction-query
                           bk tx ruleset reaction-id trigger-id stored