Repository: ViacomInc/data-point (js)

Issue:
check output type before storing cache entry
<!--
Thanks for your interest in the project. We appreciate bugs filed and PRs submitted!
Please make sure that you are familiar with and follow the Code of Conduct for
this project (found in the CODE_OF_CONDUCT.md file).

Please fill/un-comment out this template with all the relevant information so we can
understand what's going on and fix the issue.
-->

**Problem description**:

On `data-point-service` we currently do not run the entity's `outputType` reducer before storing the new key, this means that if the `outpuType`  was to fail, a key would be stored anyways causing the entry to be corrupted and served/stored for as long as the cache is configured.

**Suggested solution**:

Run `outputType` reducer right before attempting to store the entry key, if it fails the entry will not be executed and it will be handled as an error. 

**Environment**:

- `data-point` version:  all
- `node` version:  all
- `npm` version: all


Pull request description:
feat(data-point-service): run outputType before adding entry
<!--
Thanks for your interest in the project. We appreciate bugs filed and PRs submitted!

Please make sure that you are familiar with and follow the Code of Conduct for
this project (found in the CODE_OF_CONDUCT.md file).

Also, please make sure you're familiar with and follow the instructions in the
contributing guidelines (found in the CONTRIBUTING.md file).

Please fill out the information below to expedite the review and (hopefully)
merge of your pull request!

BREAKING CHANGES:

If your PR includes a breaking change, please submit it with a codemod under
data-point-codemods that will help users upgrade their codebase.

Breaking changes without a codemod will not be accepted unless a codemod is not
viable or does not apply to the specific situation.
-->

<!-- What changes are being made? (What feature/bug is being fixed here?) -->
**What**:

fixes #451

if `outputType` reducer is set, it must pass in order for the entry to be added to the cache store.

<!-- Why are these changes necessary? -->
**Why**:

See #451

<!-- How were these changes implemented? -->
**How**:

By running executing outputType reducer before adding the entry into the cache store.

<!-- Have you done all of these things?  -->
**Checklist**:
<!-- add "N/A" to the end of each line that's irrelevant to your changes -->
<!-- to check an item, place an "x" in the box like so: "- [x] Documentation" -->
- [ ] Has Breaking changes **N/A**
- [X] Documentation
- [X] Tests
- [X] Ready to be merged <!-- In your opinion, is this ready to be merged as soon as it's reviewed? -->
- [ ] Added username to **all-contributors** list **N/A**

<!-- feel free to add additional comments -->


Patch:
diff --git a/packages/data-point-service/README.md b/packages/data-point-service/README.md
index 2a7dd05..fa8d434 100644
--- a/packages/data-point-service/README.md
+++ b/packages/data-point-service/README.md
@@ -19,74 +19,72 @@ $ npm install data-point-cache
 
 Create a new DataPoint Service Object
 
-
 ```js
 factory.create(options):Promise<Service>
 ```
 
 Properties of the `options` argument:
 
-| option | type | description |
-|:---|:---|:---|
-| **cache** | `Object` | cache specific settings |
-| **cache.isRequired** | `Boolean` | false by default, if true it will throw an error |
-| **cache.redis** | `Object` | **redis** settings passed to the [ioredis](https://github.com/luin/ioredis/blob/master/API.md#new-redisport-host-options) constructor. For key prefixing, set [keyPrefix](https://github.com/luin/ioredis#transparent-key-prefixing) through `cache.redis.keyPrefix`; if the value is not provided then [os.hostname()](https://nodejs.org/api/os.html#os_os_hostname) will be used. |
+| option               | type      | description                                                                                                                                                                                                                                                                                                                                                                          |
+| :------------------- | :-------- | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| **cache**            | `Object`  | cache specific settings                                                                                                                                                                                                                                                                                                                                                              |
+| **cache.isRequired** | `Boolean` | false by default, if true it will throw an error                                                                                                                                                                                                                                                                                                                                     |
+| **cache.redis**      | `Object`  | **redis** settings passed to the [ioredis](https://github.com/luin/ioredis/blob/master/API.md#new-redisport-host-options) constructor. For key prefixing, set [keyPrefix](https://github.com/luin/ioredis#transparent-key-prefixing) through `cache.redis.keyPrefix`; if the value is not provided then [os.hostname()](https://nodejs.org/api/os.html#os_os_hostname) will be used. |
 
-This method returns a Promise that resolves to a Service Object. 
+This method returns a Promise that resolves to a Service Object.
 
-The Service Object holds a reference to a dataPoint instance. 
+The Service Object holds a reference to a dataPoint instance.
 
 ```js
 const options = {
   entities: {
-    'reducer:foo': (input, acc) => 'bar'
+    "reducer:foo": (input, acc) => "bar"
   }
-}
-factory.create(options)
-  .then((service) => {
-    return service.dataPoint.transform('reducer:foo')
+};
+factory
+  .create(options)
+  .then(service => {
+    return service.dataPoint.transform("reducer:foo");
   })
-  .then((output) => {
-    console.log(output)
+  .then(output => {
+    console.log(output);
     // bar
-  })
+  });
 ```
 
 ## Express example
 
 ```js
-const express = require('express')
-const DataPoint = require('data-point')
-const DataPointService = require('data-point-service')
-
-function server (dataPoint) {
-  const app = express()
-
-  app.get('/api/hello-world', (req, res) => {
-    dataPoint.resolve(`entry:HelloWorld`, req.query)
-      .then((output) => {
-        res.send(output)
-      })
-  })
-
-  app.listen(3000, function () {
-    console.log('listening on port 3000!')
-  })
+const express = require("express");
+const DataPoint = require("data-point");
+const DataPointService = require("data-point-service");
+
+function server(dataPoint) {
+  const app = express();
+
+  app.get("/api/hello-world", (req, res) => {
+    dataPoint.resolve(`entry:HelloWorld`, req.query).then(output => {
+      res.send(output);
+    });
+  });
+
+  app.listen(3000, function() {
+    console.log("listening on port 3000!");
+  });
 }
 
-function createService () {
+function createService() {
   return DataPointService.create({
     DataPoint,
     entities: {
-      'reducer:HelloWorld': (input, acc) => 'Hello World!!'
+      "reducer:HelloWorld": (input, acc) => "Hello World!!"
     }
-  }).then((service) => {
-    return service.dataPoint
-  })
+  }).then(service => {
+    return service.dataPoint;
+  });
 }
 
-createService()
-  .then(server)
+createService().then(server);
 ```
 
 ## <a name="entity-params-cache">Configure entity's cache settings</a>
@@ -108,7 +106,7 @@ To configure an entity's cache settings you must set cache configuration through
 ### cache.ttl
 
 ```js
-ttl: String|Number
+ttl: String | Number;
 ```
 
 Use `cache.ttl` to set an entity's cache entry **Time To Live** value. When this value is a `string` it is expected to be written using the format supported by [ms](https://www.npmjs.com/package/ms).
@@ -169,29 +167,35 @@ DataPointService.create({
 ### cache.revalidateTimeout
 
 ```js
-revalidateTimeout: String|Number
+revalidateTimeout: String | Number;
 ```
 
-*defaults to*: `5s`
+_defaults to_: `5s`
 
-`revalidateTimeout` is the time a revalidation process has before it times-out, the value is expected to be written as a string (eg. `'20m'`) following the format supported by [ms](https://www.npmjs.com/package/ms). When revalidation starts a **revalidation flag** is set which blocks revalidation duplicates from happening. Once the revalidation times-out the revalidation flag will be removed and the **key** will be unblocked for being revalidated again. If omitted it defaults to 5 seconds. 
+`revalidateTimeout` is the time a revalidation process has before it times-out, the value is expected to be written as a string (eg. `'20m'`) following the format supported by [ms](https://www.npmjs.com/package/ms). When revalidation starts a **revalidation flag** is set which blocks revalidation duplicates from happening. Once the revalidation times-out the revalidation flag will be removed and the **key** will be unblocked for being revalidated again. If omitted it defaults to 5 seconds.
 
 This value is only used when `staleWhileRevalidate` is also set.
 
 ## Revalidation and concurrency
 
-When an entity is requested for the first time it will be considered a _cold lookup_, once the entity is resolved a cache entry will be saved on **redis**. For the key's life set by the entity's `ttl` (and `staleWhileRevalidate`) the entity will return the value (considered **stale**) from the **redis** store. Once a new request is made for the key and it's `ttl` has expired a revalidation will be triggered _on the background_; when a revalidation starts a revalidation **flag** will be added to prevent new calls of duplicating the revalidation process. 
+When an entity is requested for the first time it will be considered a _cold lookup_, once the entity is resolved a cache entry will be saved on **redis**. For the key's life set by the entity's `ttl` (and `staleWhileRevalidate`) the entity will return the value (considered **stale**) from the **redis** store. Once a new request is made for the key and it's `ttl` has expired a revalidation will be triggered _on the background_; when a revalidation starts a revalidation **flag** will be added to prevent new calls of duplicating the revalidation process.
 
-Revalidation flags are saved **locally** (in memory - they get cleared once their timeout expires) and also added to the redis store to be accessed by multiple node instances sharing the same keys. The **local** flag is to prevent duplicate revalidation by a single instance, because it is in memory it is an instant lookup; the **remote** flag (redis) is to prevent multiple instances of attempting to revalidate the same request.  
+Revalidation flags are saved **locally** (in memory - they get cleared once their timeout expires) and also added to the redis store to be accessed by multiple node instances sharing the same keys. The **local** flag is to prevent duplicate revalidation by a single instance, because it is in memory it is an instant lookup; the **remote** flag (redis) is to prevent multiple instances of attempting to revalidate the same request.
 
 In the case the revalidation fails the flag will be removed to allow a new revalidation to be triggered.
 
 It is important to know that once the `staleWhileRevalidate` delta has also expired a background revalidation will no longer be triggered and cold lookup will be triggered instead. For this reason it is important to carefully set the cache settings to have a solid caching strategy.
 
+### Validation
+
+If an entity's `outputType` reducer is set, revalidation will execute it before storing the new entry in cache. If the validation fails the entry will not be stored.
+
+In the case revalidation happens via `stale-while-revalidate` and `outputType` does not pass then stale value will be served until the stale value expires on its own.
+
 ## <a name="contributing">Contributing</a>
 
 Please read [CONTRIBUTING.md](https://github.com/ViacomInc/data-point/blob/master/CONTRIBUTING.md) for details on our code of conduct, and the process for submitting pull requests to us.
 
 ## <a name="license">License</a>
 
-This project is licensed under the  Apache License Version 2.0 - see the [LICENSE](LICENSE) file for details
+This project is licensed under the Apache License Version 2.0 - see the [LICENSE](LICENSE) file for details
diff --git a/packages/data-point-service/examples/express-cache-schema-revalidation.js b/packages/data-point-service/examples/express-cache-schema-revalidation.js
new file mode 100644
index 0000000..5f2c22e
--- /dev/null
+++ b/packages/data-point-service/examples/express-cache-schema-revalidation.js
@@ -0,0 +1,120 @@
+/* eslint-disable import/no-extraneous-dependencies */
+/**
+ *  This example is to show how you can force revalidation via schema validation.
+ *  the idea is to rely on two parts:
+ *  1. adding an outputType reducer using json schema validation.
+ *  2. cache key is created with a unique string hash based on the schema used.
+ *
+ *  This will cause the following behaviors:
+ *  - Every time the schema changes the cache's key also changes, this will force a cache revalidation.
+ *  - After the revalidation, if the output type does not pass, the entry will not be added.
+ *  - **IMPORTANT**: if you are using stale-while-revalidate, and the cache validation fails,
+ *    it will continue to serve the stale value until its TTL dies.
+ */
+
+const express = require("express");
+const DataPoint = require("data-point");
+const hash = require("object-hash");
+const Ajv = require("ajv");
+
+const DataPointService = require("../lib");
+
+const ajv = new Ajv({
+  allErrors: true
+});
+
+const Schema = {
+  validate(jsonSchema) {
+    const schemaValidate = ajv.compile(jsonSchema);
+    return (value, ctx) => {
+      const isValid = schemaValidate(value);
+
+      if (!isValid) {
+        throw new TypeError(
+          `Failed schema validation for "${
+            ctx.reducer.spec.id
+          }", errors:\n ${ajv.errorsText(schemaValidate.errors)}\n`
+        );
+      }
+    };
+  },
+  hash(jsonSchema) {
+    return hash(jsonSchema);
+  }
+};
+
+const jsonSchema = {
+  type: "object",
+  title: "The Root Schema",
+  description: "The root schema comprises the entire JSON document.",
+  required: ["id", "phone"],
+  properties: {
+    id: {
+      $id: "#/properties/id",
+      type: "string",
+      pattern: "a$"
+    },
+    phone: {
+      $id: "#/properties/phone",
+      type: "string"
+    }
+  }
+};
+
+const HelloWorld = DataPoint.Model("HelloWorld", {
+  value: (input, acc) => {
+    const id = acc.locals.query.id;
+    const phone = acc.locals.query.phone;
+    return {
+      id,
+      phone
+    };
+  },
+  outputType: Schema.validate(jsonSchema),
+  params: {
+    cache: {
+      ttl: "2m",
+      staleWhileRevalidate: "4m",
+      cacheKey: () => {
+        return `SH-${Schema.hash(jsonSchema)}`;
+      }
+    }
+  }
+});
+
+function server(dataPoint) {
+  const app = express();
+
+  app.get("/api/hello-world", (req, res) => {
+    const options = {
+      // exposing query to locals will make this object available to all
+      // reducers
+      locals: {
+        query: req.query
+      }
+    };
+    dataPoint
+      .resolve(HelloWorld, {}, options)
+      .then(value => {
+        res.send(value);
+      })
+      .catch(error => {
+        res.status(500).send(error.toString());
+      });
+  });
+
+  app.listen(3000, () => {
+    // eslint-disable-next-line no-console
+    console.log("listening on port 3000!");
+  });
+}
+
+function createService() {
+  return DataPointService.create({
+    DataPoint
+  }).then(service => {
+    return service.dataPoint;
+  });
+}
+
+createService().then(server);
diff --git a/packages/data-point-service/lib/cache-middleware.js b/packages/data-point-service/lib/cache-middleware.js
index 4c81369..e1ff685 100644
--- a/packages/data-point-service/lib/cache-middleware.js
+++ b/packages/data-point-service/lib/cache-middleware.js
@@ -144,15 +144,21 @@ async function revalidateEntry(service, entryKey, cache, ctx) {
   try {
     await addRevalidationFlags(service, entryKey, cache.revalidateTimeout);
 
-    const acc = await resolveFromAccumulator(
+    const value = await resolveFromAccumulator(
       service,
       entryKey,
       ctx.context,
       revalidateContext
     )();
 
-    await module.exports.updateSWREntry(service, entryKey, cache)(acc);
-    await module.exports.revalidateSuccess(service, entityId, entryKey);
+    // check that the value passes the outputType before adding the new entry
+    const outputType = ctx.reducer.spec.outputType;
+    if (outputType) {
+      await service.dataPoint.resolveFromAccumulator(outputType, ctx);
+    }
+
+    await module.exports.updateSWREntry(service, entryKey, cache)(value);
+    await module.exports.revalidateSuccess(service, entityId, entryKey)();
   } catch (error) {
     module.exports.catchRevalidateError(service, entityId, entryKey)(error);
   }
@@ -298,6 +304,12 @@ async function after(service, ctx, next) {
     // from here below ttl is assumed to be true
     const entryKey = module.exports.generateKey(cache.cacheKey, ctx);
 
+    // check that the value passes the outputType before adding the new entry
+    const outputType = ctx.reducer.spec.outputType;
+    if (outputType) {
+      await service.dataPoint.resolveFromAccumulator(outputType, ctx);
+    }
+
     if (cache.useStaleWhileRevalidate) {
       // adds (or updates) the stale cache entry with the latest value
       await module.exports.setStaleWhileRevalidateEntry(
diff --git a/packages/data-point-service/package.json b/packages/data-point-service/package.json
index 0ee2ef5..e529475 100644
--- a/packages/data-point-service/package.json
+++ b/packages/data-point-service/package.json
@@ -16,7 +16,9 @@
     "email": "acatl.pacheco@viacom.com"
   },
   "devDependencies": {
-    "data-point": "^3.5.0"
+    "data-point": "^3.5.0",
+    "object-hash": "2.0.3",
+    "ajv": "5.2.3"
   },
   "dependencies": {
     "data-point-cache": "^4.1.0",
diff --git a/yarn.lock b/yarn.lock
index c19b08f..c575189 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -9089,6 +9089,11 @@ object-copy@^0.1.0:
     define-property "^0.2.5"
     kind-of "^3.0.3"
 
+object-hash@2.0.3:
+  version "2.0.3"
+  resolved "https://registry.yarnpkg.com/object-hash/-/object-hash-2.0.3.tgz#d12db044e03cd2ca3d77c0570d87225b02e1e6ea"
+  integrity sha512-JPKn0GMu+Fa3zt3Bmr66JhokJU5BaNBIh4ZeTlaCBzrBsOeXzwcKKAK1tbLiPKgvwmPXsDvvLHoWh5Bm7ofIYg==
+
 object-keys@^1.0.11:
   version "1.1.1"
   resolved "https://registry.yarnpkg.com/object-keys/-/object-keys-1.1.1.tgz#1c47f272df277f3b1daf061677d9c82e2322c60e"


Test patch:
diff --git a/packages/data-point-service/lib/cache-middleware.test.js b/packages/data-point-service/lib/cache-middleware.test.js
index 760afb0..1998df9 100644
--- a/packages/data-point-service/lib/cache-middleware.test.js
+++ b/packages/data-point-service/lib/cache-middleware.test.js
@@ -40,6 +40,9 @@ function createContext() {
     context: {
       params: {},
       id: "model:Foo"
+    },
+    reducer: {
+      spec: {}
     }
   };
 
@@ -280,6 +283,80 @@ describe("revalidateEntry", () => {
     });
   });
 
+  describe("outputType check", () => {
+    it("check it runs outputType reducer if set", async () => {
+      const mocks = createMocks();
+      mocks.resolveFromAccumulator = jest.fn().mockResolvedValue(true);
+
+      const ctx = createContext();
+      // we are not executing this, just checking it gets passed
+      ctx.reducer.spec.outputType = "outputType";
+
+      _.set(
+        mocks.service,
+        "dataPoint.resolveFromAccumulator",
+        mocks.resolveFromAccumulator
+      );
+
+      await CacheMiddleware.revalidateEntry(
+        mocks.service,
+        "entryKey",
+        cache,
+        ctx
+      );
+
+      // check outputType gets executed
+      expect(mocks.resolveFromAccumulator).toBeCalledWith(
+        ctx.reducer.spec.outputType,
+        ctx
+      );
+
+      // check updateSWREntry is executed
+      expect(mocks.updateSWREntry).toBeCalledWith(
+        mocks.service,
+        "entryKey",
+        cache
+      );
+    });
+
+    it("check it does not save entry if outputType does not pass", async () => {
+      const mocks = createMocks();
+      const error = new TypeError("bad type");
+
+      mocks.resolveFromAccumulator = jest.fn().mockRejectedValue(error);
+
+      const ctx = createContext();
+      // we are not executing this, just checking it gets passed
+      ctx.reducer.spec.outputType = "outputType";
+
+      _.set(
+        mocks.service,
+        "dataPoint.resolveFromAccumulator",
+        mocks.resolveFromAccumulator
+      );
+
+      await CacheMiddleware.revalidateEntry(
+        mocks.service,
+        "entryKey",
+        cache,
+        ctx
+      );
+
+      // check outputType gets executed
+      expect(mocks.resolveFromAccumulator).toBeCalled();
+
+      // check updateSWREntry is executed
+      expect(mocks.updateSWREntry).not.toBeCalled();
+
+      // error is handled
+      expect(mocks.catchRevalidateError).toBeCalledWith(
+        mocks.service,
+        "model:Foo",
+        "entryKey"
+      );
+    });
+  });
+
   it("should call updateSWREntry", () => {
     const mocks = createMocks();
     const ctx = createContext();
@@ -685,6 +762,73 @@ describe("after", () => {
     });
   });
 
+  describe("outputType check", () => {
+    it("check it runs outputType reducer if set", done => {
+      const mocks = createMocks();
+
+      mocks.ctx.reducer.spec.outputType = "outputType";
+
+      mocks.resolveFromAccumulator = jest
+        .fn()
+        .mockResolvedValue("resolvedValue");
+
+      _.set(
+        mocks.service,
+        "dataPoint.resolveFromAccumulator",
+        mocks.resolveFromAccumulator
+      );
+
+      const next = createNext(done, () => {
+        expect(mocks.resolveFromAccumulator).toBeCalledWith(
+          mocks.ctx.reducer.spec.outputType,
+          mocks.ctx
+        );
+
+        expect(mocks.setStaleWhileRevalidateEntry).toBeCalledWith(
+          mocks.service,
+          "mockCacheKey",
+          "resolvedValue",
+          mocks.cache
+        );
+        expect(mocks.setEntry).not.toBeCalled();
+      });
+
+      mocks.cache.useStaleWhileRevalidate = true;
+      CacheMiddleware.after(mocks.service, mocks.ctx, next);
+    });
+
+    it("check it does not save entry if outputType does not pass", done => {
+      const mocks = createMocks();
+
+      const error = new TypeError("bad type");
+
+      mocks.resolveFromAccumulator = jest.fn().mockRejectedValue(error);
+
+      _.set(
+        mocks.service,
+        "dataPoint.resolveFromAccumulator",
+        mocks.resolveFromAccumulator
+      );
+
+      mocks.ctx.reducer.spec.outputType = "outputType";
+
+      const next = createNext(done, () => {
+        expect(next).toBeCalledWith(error);
+
+        expect(mocks.resolveFromAccumulator).toBeCalledWith(
+          mocks.ctx.reducer.spec.outputType,
+          mocks.ctx
+        );
+
+        expect(mocks.setStaleWhileRevalidateEntry).not.toBeCalled();
+        expect(mocks.setEntry).not.toBeCalled();
+      });
+
+      mocks.cache.useStaleWhileRevalidate = true;
+      CacheMiddleware.after(mocks.service, mocks.ctx, next);
+    });
+  });
+
   it("should set stale value if cache.useStaleWhileRevalidate is true", done => {
     const mocks = createMocks();