feature: post filtering on subscriptions

docs/named_queries.md

@@ -320,6 +320,29 @@ query.expose({
     // This deep extends your graph's body before processing it.
     // For example, you want a hidden $filter() functionality, or anything else.
     embody: {}, // Accepts Object or Function(body, params)
+
+    // This function enables you to modify the documents before they are sent to the client.
+    // You might want to check some field in the object and strip some values from it.
+    // If you have "private" field in a collection and some document has value private: true, perhaps you want
+    // to hide some things from unauthorized users.
+    // You should use query param "scoped: true" param if subscriptionFilter is used.
+    subscriptionFilter(
+        type, // 'added', 'changed', 'observe-changes'
+        newDoc, // for 'added' and 'changed' the whole doc, for 'observe-changes' only the changed fields with _id
+        oldDoc, // for 'added' and 'observe-changes' is null, for changed the old doc
+
+        // options passed to the publication, such as:
+        // cursor - the cursor that is being published
+        // filter - actual subscriptionFilter
+        // parents - documents passed in from reywood:publish-composite
+        // linker - linker or null
+        // node - current node (field, reducer or collection)
+        // filters - actual filter being applied to this node
+        // options - MongoDB query options (sort, limit, skip)
+        // publication - reywood:publish-composite instance or undefined
+        // params - publication params or undefined
+        options, 
+    ) {},
 })
 ```
 
@@ -367,6 +390,36 @@ query.expose({
 
 Careful here, if you use the special `$body` parameter, embodyment will be performed on it. You have to handle manually these use-cases, but usually, you will use `embody` for filtering top level elements, so in most cases it will be simple and with no hassle.
 
+### Subscription filters
+
+Use this if you need to perform additional modifications or even filter out documents that are being sent to the client.
+You would use this when there are some fields in the document that warrant changes (i.e. remove particular information) 
+for particular a user or in general.
+
+```js
+Projects.createQuery('getProjects', {
+    clientName: 1,
+    private: 1,
+    projectValue: 1,
+})
+
+query.expose({
+    // For each private project (private: true), remove projectValue field
+    subscriptionFilter(type, doc, oldDoc, options) {
+        if (type === 'added') {
+            // NOTE: you can also return undefined when type is 'added' which means this doc won't be sent 
+            // to the client at all
+            if (doc.private) {
+                // We have to clone it not to interfere with possibly other subscriptions
+                doc = _.clone(doc);
+                delete doc.projectValue;
+                return doc;
+            }
+        }
+    }
+})
+```
+
 ## Resolvers
 
 Named Queries have the ability to morph themselves into a function that executes on the server.

lib/exposure/exposure.config.schema.js

@@ -19,8 +19,9 @@ export const ExposureSchema = {
     body: Match.Maybe(Match.OneOf(Object, Function)),
     restrictedFields: Match.Maybe([String]),
     restrictLinks: Match.Maybe(
-        Match.OneOf(Function, [String])
+        Match.OneOf(Function, [String]),
     ),
+    subscriptionFilter: Match.Maybe(Function)
 };
 
 export function validateBody(collection, body) {

lib/namedQuery/expose/extension.js

@@ -9,6 +9,9 @@ import intersectDeep from '../../query/lib/intersectDeep';
 import genCountEndpoint from '../../query/counts/genEndpoint.server';
 import { check } from 'meteor/check';
 
+// import { enableDebugLogging  } from 'meteor/reywood:publish-composite';
+// enableDebugLogging();
+
 _.extend(NamedQuery.prototype, {
     /**
      * @param config
@@ -152,6 +155,7 @@ _.extend(NamedQuery.prototype, {
      */
     _initPublication() {
         const self = this;
+        const config = this.exposeConfig;
 
         Meteor.publishComposite(this.name, function(params = {}) {
             const isScoped = !!self.options.scoped;
@@ -177,6 +181,9 @@ _.extend(NamedQuery.prototype, {
             return recursiveCompose(rootNode, undefined, {
                 scoped: isScoped,
                 blocking: self.exposeConfig.blocking,
+                subscriptionFilter: config.subscriptionFilter,
+                publication: this,
+                params,
             });
         });
     },

lib/namedQuery/expose/schema.js

@@ -18,5 +18,6 @@ export const ExposeSchema = {
     ),
     validateParams: Match.Maybe(
         Match.OneOf(Object, Function)
-    )
+    ),
+    subscriptionFilter: Match.Maybe(Function)
 };

lib/namedQuery/testing/bootstrap/queries/index.js

@@ -2,6 +2,7 @@ import postList from './postList';
 import postListCached from './postListCached';
 import postListExposure from './postListExposure';
 import postListExposureScoped from './postListExposureScoped';
+import projectsListExposureSecurity from './projectsListExposureSecurity';
 import postListParamsCheck from './postListParamsCheck';
 import postListParamsCheckServer from './postListParamsCheckServer';
 import postListResolver from './postListResolver';
@@ -15,6 +16,7 @@ export {
     postListCached,
     postListExposure,
     postListExposureScoped,
+    projectsListExposureSecurity,
     postListParamsCheck,
     postListParamsCheckServer,
     postListResolver,

lib/namedQuery/testing/bootstrap/queries/projectsListExposureSecurity.js

@@ -0,0 +1,87 @@
+import { createQuery } from 'meteor/cultofcoders:grapher';
+import { Projects } from '../../../../query/testing/bootstrap/projects/collection';
+
+const projectsListExposureSecurity = createQuery('projectsListExposureSecurity', {
+    projects: {
+        name: 1,
+        private: 1,
+        projectValue: 1,
+        files: {
+          filename: 1,
+          public: 1,
+        },
+    }
+}, {
+  scoped: true,
+});
+
+if (Meteor.isServer) {
+  Meteor.methods({
+    updateProject({projectId, ...updates}) {
+      Projects.update(projectId, {$set: updates});
+    },
+    resetProjects() {
+      Projects.update({name: 'Project 1'}, {$set: {private: false}});
+      Projects.update({name: 'Project 2'}, {$set: {private: true}});
+    }
+  })
+
+  projectsListExposureSecurity.expose({
+        firewall(userId, params) {
+        },
+        embody: {
+            $filter({filters, params}) {
+                filters.title = params.title
+            }
+        },
+        subscriptionFilter(type, doc, oldDoc, options) {
+          const {node, parents} = options;
+          const path = [];
+          let n = node;
+          while (n) {
+            if (n.linkName) {
+              path.push(n.linkName);
+            }
+            n = n.parent;
+          }
+
+          // this is how to fetch full doc, not only changed fields when observe-changes triggers
+          // if (type === 'observe-changes') {
+          //   const collectionName = node.collection._name;
+          //   const wholeDoc = options.publication._session.collectionViews.get(collectionName).documents.get(doc._id).getFields();
+          // }
+
+          const dottedPath = path.reverse().join('.');
+          if (dottedPath === '') {
+            if (doc.private) {
+              doc = _.clone(doc);
+              if (type === 'added') {
+                delete doc.projectValue;
+              }
+              else if (type === 'observe-changes') {
+                doc.projectValue = undefined;
+              }
+              return doc;
+            }
+            else if (type === 'observe-changes') {
+              if (!doc.private) {
+                doc = _.clone(doc);
+                // db query required here, collectionView does not include projectValue
+                const {projectValue} = Projects.findOne(doc._id, {fields: {projectValue: 1}});
+                doc.projectValue = projectValue;
+                return doc;
+              }
+            }
+          }
+          else if (dottedPath === 'files') {
+            const [project] = parents;
+            if (project.private && !doc.public) {
+              return undefined;
+            }
+          }
+          return doc;
+        }
+    });
+}
+
+export default projectsListExposureSecurity;

lib/namedQuery/testing/client.test.js

@@ -1,6 +1,7 @@
 import { assert, expect } from 'chai';
 import postListExposure, {postListFilteredWithDate} from './bootstrap/queries/postListExposure.js';
 import postListExposureScoped from './bootstrap/queries/postListExposureScoped';
+import projectsListExposureSecurity from './bootstrap/queries/projectsListExposureSecurity';
 import userListScoped from './bootstrap/queries/userListScoped';
 import productsList from './bootstrap/queries/productsList';
 import { createQuery } from 'meteor/cultofcoders:grapher';
@@ -264,6 +265,138 @@ describe('Named Query', function () {
         });
     });
 
+    describe('Protected reactive queries', () => {
+        beforeEach(done => {
+            Meteor.call('resetProjects', {}, done);
+        });
+        afterEach(done => {
+            Meteor.call('resetProjects', {}, done);
+            Tracker.flush();
+        });
+
+        it('Should work with protected reactive queries', done => {
+            const query = projectsListExposureSecurity.clone({});
+            const handle = query.subscribe();
+
+            Tracker.autorun(c => {
+                if (handle.ready()) {
+                    c.stop();
+
+                    const projects = query.fetch();
+                    handle.stop();
+
+                    // console.log(JSON.stringify(projects, null, 2));
+
+                    expect(projects).to.have.length(2);
+
+                    const [project1, project2] = projects;
+                    expect(project1.projectValue).to.be.equal(10000);
+                    expect(project2.projectValue).to.be.undefined;
+
+                    done();
+                }
+            });
+        });
+
+        it('Should work with protected reactive queries after the change', function(done) {
+            const query = projectsListExposureSecurity.clone({log: true});
+            const handle = query.subscribe();
+
+            let firstCall = true;
+            Tracker.autorun(c => {
+                if (handle.ready()) {
+                    if (firstCall) {
+                        const projects = query.fetch()
+
+                        expect(projects).to.have.length(2);
+
+                        const [project1, project2] = projects;
+                        expect(project1.projectValue).to.be.equal(10000);
+                        expect(project2.projectValue).to.be.undefined;
+                        expect(project1.files).to.have.length(2);
+                        // expect(project2.files).to.have.length(1);
+
+                        firstCall = false;
+                        Meteor.call('updateProject', {projectId: project1._id, private: true}, (err, res) => {
+
+                        });
+                    }
+                    else {
+                        const projects = query.fetch();
+
+                        const [project1, project2] = projects;
+
+                        // console.log(project1.name, project1.private, project1.projectValue);
+                        // console.log(project2.name, project2.private, project2.projectValue);
+
+                        expect(project1.projectValue).to.be.undefined;
+                        expect(project2.projectValue).to.be.undefined;
+
+                        // if (project1.files.length !== 1) {
+                        //     console.log('waiting for potentially one more callback', project1.files);
+                        //     return;
+                        // }
+
+                        c.stop();
+                        handle.stop();
+
+                        expect(project1.files).to.have.length(1);
+                        expect(project1.files[0].filename).to.be.equal('invoice.pdf');
+                        // expect(project2.filesMany).to.have.length(1);
+
+                        done();
+                    }
+                }
+            });
+        });
+
+        it('Should bring back field after the change', done => {
+            const query = projectsListExposureSecurity.clone({});
+            const handle = query.subscribe();
+
+            let firstCall = true;
+            Tracker.autorun(c => {
+                if (handle.ready()) {
+                    if (firstCall) {
+                        const projects = query.fetch();
+
+                        expect(projects).to.have.length(2);
+
+                        const [project1, project2] = projects;
+                        expect(project1.projectValue).to.be.equal(10000);
+                        expect(project2.projectValue).to.be.undefined;
+                        expect(project2.files).to.have.length(0);
+
+                        firstCall = false;
+                        Meteor.call('updateProject', {projectId: project2._id, private: false}, (err, res) => {
+
+                        });
+                    }
+                    else {
+                        const projects = query.fetch();
+                        const [project1, project2] = projects;
+
+                        if (project2.projectValue === undefined) {
+                            return;
+                        }
+
+                        c.stop();
+                        handle.stop();
+
+                        // console.log(project1.name, project1.private, project1.projectValue);
+                        // console.log(project2.name, project2.private, project2.projectValue);
+
+                        expect(project1.projectValue).to.be.equal(10000);
+                        expect(project2.projectValue).to.be.equal(20000);
+                        // expect(project2.files).to.have.length(1);
+
+                        done();
+                    }
+                }
+            });
+        });
+    });
+
     it('Should work with reactive queries containing link with foreignIdentityField', function (done) {
         const query = productsList.clone({
             filters: {