Merge pull request #203 from akvo/issue/200-caching

[#200] Respond HTTP 204 when no more changes are available

api/src/clojure/org/akvo/flow_api/endpoint/sync.clj

@@ -7,7 +7,7 @@
             [org.akvo.flow-api.middleware.jdo-persistent-manager :as jdo-pm]
             [org.akvo.flow-api.middleware.resolve-alias :refer [wrap-resolve-alias]]
             [org.akvo.flow-api.unilog.unilog :as unilog]
-            [ring.util.response :refer [response]]))
+            [ring.util.response :refer [response status header]]))
 
 
 
@@ -26,43 +26,60 @@
 ;; TODO: Express parameter logic via spec?
 (def params-spec (s/keys :opt-un [::initial ::next ::cursor]))
 
+(def no-more-changes (-> (status {} 204) (header "Cache-Control" "max-age=60")))
+
+(defn initial-response
+  [req]
+  (let [alias (:alias req)
+        api-root (utils/get-api-root req)
+        db (:unilog-db-connection req)
+        cursor (unilog/get-cursor db)]
+    (response {:next-sync-url (next-sync-url api-root alias cursor)})))
+
+(defn changes-response
+  [offset db instance-id remote-api req]
+  (let [alias (:alias req)
+        api-root (utils/get-api-root req)
+        changes (->
+                 (unilog/process-unilog-events offset db instance-id remote-api)
+                 (select-keys [:form-instance-changed
+                               :form-instance-deleted
+                               :form-changed
+                               :form-deleted
+                               :data-point-changed
+                               :data-point-deleted
+                               :unilog-id])
+                 (update :form-deleted #(map str %))
+                 (update :form-instance-deleted #(map str %))
+                 (update :data-point-deleted #(map str %)))
+        cursor (:unilog-id changes)]
+    (-> (response {:changes (dissoc changes :unilog-id)
+                   :next-sync-url (next-sync-url api-root alias cursor)})
+        (header "Cache-Control" "no-cache"))))
+
+(defn get-changes
+  [req cursor instance-id remote-api]
+  (let [db (:unilog-db-connection req)
+        offset (Long/parseLong cursor)]
+    (cond
+      (not (unilog/valid-offset? offset db)) (anomaly/bad-request "Invalid cursor" {})
+      (= offset (unilog/get-cursor db)) no-more-changes
+      :else (changes-response offset db instance-id remote-api req))))
+
 (defn changes [deps {:keys [alias instance-id params] :as req}]
   (let [{:keys [initial cursor next]} (spec/validate-params params-spec params)]
-    (if (and initial (or cursor next))
-      (anomaly/bad-request "Invalid parameters" {})
-      (if (= "true" initial)
-        (let [db-name (get-db-name instance-id)
-              db-spec (-> deps :unilog-db :spec (assoc :db-name db-name))]
-          (response {:next-sync-url (next-sync-url (utils/get-api-root req)
-                                      alias
-                                      (unilog/get-cursor db-spec))}))
-        (if (and next cursor)
-          (let [db-name (get-db-name instance-id)
-                db-spec (-> deps :unilog-db :spec (assoc :db-name db-name))
-                offset (Long/parseLong cursor)]
-            (if (unilog/valid-offset? offset db-spec)
-              (let [changes (->
-                             (unilog/process-unilog-events offset db-spec instance-id (:remote-api deps))
-                             (select-keys [:form-instance-changed
-                                           :form-instance-deleted
-                                           :form-changed
-                                           :form-deleted
-                                           :data-point-changed
-                                           :data-point-deleted
-                                           :unilog-id])
-                             (update :form-deleted #(map str %))
-                             (update :form-instance-deleted #(map str %))
-                             (update :data-point-deleted #(map str %)))]
-                (response {:changes (dissoc changes :unilog-id)
-                           :next-sync-url (next-sync-url (utils/get-api-root req) alias (:unilog-id changes))}))
-              (anomaly/bad-request "Invalid cursor" {})))
-          (anomaly/bad-request "Invalid parameters" {}))))))
+    (cond
+      (and initial (or cursor next)) (anomaly/bad-request "Invalid parameters" {})
+      (= "true" initial) (initial-response req)
+      (and next cursor) (get-changes req cursor instance-id (:remote-api deps))
+      :else (anomaly/bad-request "Invalid parameters" {}))))
 
 (defn endpoint* [deps]
   (GET "/sync" req
     (#'changes deps req)))
 
 (defn endpoint [{:keys [akvo-flow-server-config] :as deps}]
   (-> (endpoint* deps)
+      (unilog/wrap-db-connection (:unilog-db deps))
       (wrap-resolve-alias akvo-flow-server-config)
       (jdo-pm/wrap-close-persistent-manager)))

api/src/clojure/org/akvo/flow_api/unilog/unilog.clj

@@ -117,22 +117,22 @@
    :data-point-changed data-point-changed
    :data-point-deleted data-point-deleted})
 
-(defn get-cursor [config]
-  (let [result (first (jdbc/query (event-log-spec config)
+(defn get-cursor [db]
+  (let [result (first (jdbc/query db
                                   ["SELECT MAX(id) AS cursor FROM event_log"]))]
     (or (:cursor result) 0)))
 
-(defn valid-offset? [offset config]
+(defn valid-offset? [offset db]
   (or (= offset 0)
-      (let [result (first (jdbc/query (event-log-spec config)
+      (let [result (first (jdbc/query db
                                       ["SELECT id AS offset FROM event_log WHERE id = ?" offset]))]
         (boolean (:offset result)))))
 
-(defn process-unilog-events [offset config instance-id remote-api]
+(defn process-unilog-events [offset db instance-id remote-api]
   (ds/with-remote-api remote-api instance-id
     (let [ds (DatastoreServiceFactory/getDatastoreService)
           events (process-new-events
-                   (jdbc/reducible-query (event-log-spec config)
+                   (jdbc/reducible-query db
                                          ["SELECT id, payload::text AS payload FROM event_log WHERE id > ? ORDER BY id ASC LIMIT 300" offset]
                                          {:auto-commit? false :fetch-size 300}))
           form-id->form (reduce (fn [acc form-id]
@@ -147,3 +147,13 @@
                              (:form-instances-to-load events-2)))
           data-point-changed (doall (data-point/by-ids ds (:data-point-changed events-2)))]
       (assoc events-2 :form-instance-changed form-instances :data-point-changed data-point-changed))))
+
+(defn get-db-name [instance-id]
+  (str "u_" instance-id))
+
+(defn wrap-db-connection [handler unilog-db]
+  (fn [request]
+    (let [db-name (get-db-name (:instance-id request))
+          db-spec (-> unilog-db :spec (assoc :db-name db-name) event-log-spec)]
+      (jdbc/with-db-connection [conn db-spec]
+        (handler (assoc request :unilog-db-connection conn))))))

api/test/clojure/org/akvo/flow_api/sync_end_to_end.clj

@@ -25,7 +25,7 @@
                           "content-type" "application/json"
                           "accept" "application/vnd.akvo.flow.v2+json"}
                 :query-params params})
-     [:status :body])
+     [:status :body :headers])
     (catch clojure.lang.ExceptionInfo e
       (select-keys (ex-data e) [:status :body]))))
 
@@ -116,4 +116,11 @@
                    (set (map :id (:formInstanceChanged changes)))))
             (is (= #{"144622023"}
                    (set (map :id (:dataPointChanged changes)))))
-            (is (= (:dataPointDeleted changes) ["144602051"]))))))))
+            (is (= (:dataPointDeleted changes) ["144602051"]))
+            (let [{:keys [headers status]} (http/get nextSyncUrl
+                                                     {:as :json
+                                                      :headers {"x-akvo-email" user
+                                                                "content-type" "application/json"
+                                                                "accept" "application/vnd.akvo.flow.v2+json"}})]
+              (is (= 204 status))
+              (is (= "max-age=60" (get headers "Cache-Control"))))))))))

ci/build.sh

@@ -23,40 +23,40 @@ fi
 
 cp -v "${HOME}/.cache/local_db.bin" "${LOCAL_TEST_DATA_PATH}"
 
-docker-compose -p akvo-flow-api-ci -f docker-compose.yml -f docker-compose.ci.yml up --build -d
-docker-compose -p akvo-flow-api-ci -f docker-compose.yml -f docker-compose.ci.yml run --no-deps tests dev/run-as-user.sh lein do clean, check, eastwood, test :all
+(
+    cd nginx
+    docker build \
+	   -t "akvo/flow-api-proxy:latest" \
+	   -t "akvo/flow-api-proxy:${TRAVIS_COMMIT}" .
+)
 
 # Check nginx configuration
-
 docker run \
        --rm \
-       --volume "$PWD/nginx/:/conf" \
        --entrypoint /usr/local/openresty/bin/openresty \
-       openresty/openresty:1.11.2.3-alpine-fat -t -c /conf/nginx.conf
-(
-    cd nginx
-    docker build -t "akvo/flow-api-proxy" .
-    docker tag akvo/flow-api-proxy akvo/flow-api-proxy:$TRAVIS_COMMIT
-)
-
-
-# Check nginx auth0 configuration
+       "akvo/flow-api-proxy" -t -c /usr/local/openresty/nginx/conf/nginx.conf
 
 (
     cd nginx-auth0
-    docker build -t "akvo/flow-api-auth0-proxy" .
-    docker tag akvo/flow-api-auth0-proxy akvo/flow-api-auth0-proxy:$TRAVIS_COMMIT
+    docker build \
+	   -t "akvo/flow-api-auth0-proxy:latest" \
+	   -t "akvo/flow-api-auth0-proxy:${TRAVIS_COMMIT}" .
 )
 
+# Check nginx auth0 configuration
 docker run \
        --rm \
        --entrypoint /usr/local/openresty/bin/openresty \
-       akvo/flow-api-auth0-proxy -t -c /usr/local/openresty/nginx/conf/nginx.conf
+       "akvo/flow-api-auth0-proxy" -t -c /usr/local/openresty/nginx/conf/nginx.conf
 
+# Backend tests
+docker-compose -p akvo-flow-api-ci -f docker-compose.yml -f docker-compose.ci.yml up --build -d
+docker-compose -p akvo-flow-api-ci -f docker-compose.yml -f docker-compose.ci.yml run --no-deps tests dev/run-as-user.sh lein do clean, check, eastwood, test :all
 docker-compose -p akvo-flow-api-ci -f docker-compose.yml -f docker-compose.ci.yml run --no-deps tests dev/run-as-user.sh lein with-profile +assemble  do jar, assemble
 
 (
     cd api
-    docker build -t "akvo/flow-api-backend" .
-    docker tag akvo/flow-api-backend akvo/flow-api-backend:$TRAVIS_COMMIT
+    docker build \
+	   -t "akvo/flow-api-backend" \
+	   -t "akvo/flow-api-backend:$TRAVIS_COMMIT" .
 )

nginx-auth0/Dockerfile

@@ -1,6 +1,6 @@
 FROM openresty/openresty:1.13.6.2-2-alpine-fat
 
-RUN apk add --no-cache build-base openssl-dev git
+RUN apk add --no-cache build-base openssl-dev git && mkdir -p /data/nginx/cache
 
 RUN /usr/local/openresty/luajit/bin/luarocks install lua-resty-openidc 1.7.2-1
 RUN /usr/local/openresty/luajit/bin/luarocks install lua-resty-jwt 0.2.0-0

nginx-auth0/docker-compose.yml

@@ -1,20 +1,21 @@
 version: "3"
 
 services:
-  mainnetwork:
+  secondnetwork:
     image: alpine:3.10
     command: tail -f /dev/null
     ports:
       - 8082:8082
   nginx-auth0:
     image: akvo/flow-api-auth0-proxy:latest
-    network_mode: service:mainnetwork
+    network_mode: service:secondnetwork
     environment:
       - FLOW_API_BACKEND_URL=http://localhost:3000
       - OIDC_DISCOVERY_URL=https://akvotest.eu.auth0.com/.well-known/openid-configuration
       - OIDC_EXPECTED_ISSUER=https://akvotest.eu.auth0.com/
   upstream:
-    image: jmalloc/echo-server:latest
-    network_mode: service:mainnetwork
-    environment:
-      - PORT=3000
+    image: akvo/flow-api-auth0-proxy:latest
+    network_mode: service:secondnetwork
+    entrypoint: ["/usr/local/openresty/bin/openresty", "-c", "/usr/local/src/nginx-test.conf", "-g", "daemon off;"]
+    volumes:
+      - ./:/usr/local/src:ro

nginx-auth0/nginx-test.conf

@@ -0,0 +1,27 @@
+worker_processes 1;
+error_log logs/error.log;
+
+events {
+  worker_connections 128;
+}
+
+http {
+
+  server_tokens off;
+
+  server {
+
+    listen 3000;
+    default_type application/json;
+
+    location / {
+      return 204;
+      add_header Cache-Control max-age=120;
+    }
+
+    location /ok {
+      return 200 '{}';
+      add_header Cache-Control no-cache;
+    }
+  }
+}

nginx-auth0/nginx.conf

@@ -34,6 +34,9 @@ http {
   lua_shared_dict userinfo 20m;
   lua_capture_error_log 32m;
 
+  # Cache 204 responses
+  proxy_cache_path /data/nginx/cache levels=1:2 keys_zone=http_204:5m max_size=1g inactive=60m use_temp_path=off;
+
   init_by_lua_block {
         local errlog = require "ngx.errlog"
         local status, err = errlog.set_filter_level(ngx.WARN)
@@ -88,6 +91,8 @@ http {
       lua_ssl_verify_depth 2;
       lua_ssl_trusted_certificate /etc/ssl/certs/ca-certificates.pem;
 
+      set $akvoemail "";
+
       access_by_lua_block {
 
           local cjson = require("cjson")
@@ -155,12 +160,24 @@ http {
          if res.email then
            ngx.log(ngx.DEBUG, "bearer_jwt_verify response: mail ", res.email)
            ngx.req.set_header("X-Akvo-Email", res.email)
+           ngx.var.akvoemail = res.email
          else
            ngx.say(cjson.encode({error="Invalid access_token"}))
            ngx.exit(ngx.HTTP_FORBIDDEN)
 	 end
       }
 
+      proxy_cache http_204;
+      proxy_cache_revalidate off;
+      proxy_cache_background_update on;
+      proxy_cache_lock on;
+      proxy_cache_methods GET;
+      proxy_cache_key $proxy_host$request_uri$akvoemail;
+      proxy_cache_lock_timeout 30s;
+      proxy_cache_lock_age 30s;
+
+      add_header X-Cache-Status $upstream_cache_status;
+
       rewrite ^/flow(/.*)$ $1 break;
       proxy_pass http://localhost:3000;
       proxy_set_header Host $host;

nginx-auth0/test-cache.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+set -eu
+
+# Checking nginx caching
+
+docker-compose up -d
+
+./test.sh "http://localhost:8082/flow/ok" 2>&1 | grep 'X-Cache-Status: MISS'
+./test.sh "http://localhost:8082/flow/" 2>&1 | grep 'X-Cache-Status: MISS'
+./test.sh "http://localhost:8082/flow/" 2>&1 | grep 'X-Cache-Status: HIT'
+./test.sh "http://localhost:8082/flow/" 2>&1 | grep 'X-Cache-Status: HIT'
+
+docker-compose down -v

nginx-auth0/test.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+
+set -eu
+
+token=$(curl --silent \
+	     --data "client_id=qsxNP3Nex0wncADQ9Re6Acz6Fa55SuU8" \
+	     --data "username=${AUTH0_USER}" \
+	     --data "password=${AUTH0_PASSWORD}" \
+	     --data "grant_type=password" \
+	     --data "scope=openid email" \
+	     --url "https://akvotest.eu.auth0.com/oauth/token" \
+	    | jq -M -r .id_token)
+
+URL="${1}"
+shift
+
+curl --verbose \
+     --header "Content-Type: application/json" \
+     --header "Accept: application/vnd.akvo.flow.v2+json" \
+     --header "Authorization: Bearer ${token}" \
+     --request GET \
+     "$@" \
+     --url "${URL}" | jq -M

nginx/nginx.conf

@@ -54,6 +54,10 @@ http {
       proxy_set_header Host $host;
     }
 
+    location ~ ^/flow/orgs/.*/sync.* {
+      return 400 '{"message": "Please use the new authentication method"}';
+    }
+
     location /flow {
 
       default_type application/json;