Helper class to work with Google Cloud datastore.

gcdatastore.js

const gcloud = require('gcloud');
const _ = require('lodash');
const cache = require('./cache');
const log = console.log;


const config = {projectId: process.env.GCLOUD_PROJECT || 'brydge-api'}
const ds = gcloud.datastore(config);


module.exports = {

  datastore: ds,

  /**
   * Simple finder.
   * @param  String t Type, datastore kind to search for
   * @param  String k Property key to look for
   * @param  String o Operator =, >=, <=
   * @param  v Value of the prooerty
   * @return Entity
   */
  // find(t, k, o, v, options) {
  //   return new Promise((resolve, reject) => {
  //     var query = ds.createQuery(t).filter(k, o, v);
  //     if (options && options.sort) query.order('createdAt', {descending: true});
  //     if (options && options.limit) query.limit(limit);

  //     ds.runQuery(query, (err, results) => {
  //       if (err) return reject(err);
  //       resolve(results);
  //     });
  //   });
  // },

  find(t, q, options) {
    return new Promise((resolve, reject)=>{
      console.log(options)
      let query = ds.createQuery(t);

      _.each(_.keys(q), (k)=>{
        query.filter(k,'=',q[k]);
      });

      if (options && options.sort) query.order('createdAt', {descending: true});
      if (options && options.limit) query.limit(options.limit);

      if (options && options.page) {
        var pageNumber = options.page;
        var nextPageNumber = parseInt(options.page) + 1;

        var this_page_cursor_key = options.cursorKey + '_' + options.page;
        var next_page_cursor_key = options.cursorKey + '_' + nextPageNumber;

        if(pageNumber > 1) {
          console.log('pageNumber > 1')
          cache.load(this_page_cursor_key)
          .then((cached)=>{
            if(!cached) return resolve([]);

            query.start(cached.nextPageCursor);
            ds.runQuery(query, (err, entities, info) => {
              if (err) return reject(err);
              
              if (info.moreResults !== ds.NO_MORE_RESULTS) {
                cache.put(next_page_cursor_key, {
                  nextPageCursor: info.endCursor
                });
              }

              resolve(entities);
            });
          })
        }else {
          console.log('pageNumber < 2')
          query.start(null);
          ds.runQuery(query, (err, entities, info) => {
            if (err) return reject(err);

            if (info.moreResults !== ds.NO_MORE_RESULTS) {
              cache.put(next_page_cursor_key, {
                nextPageCursor: info.endCursor
              });
            }

            
            resolve(entities);
          });
        }
      }else {
        ds.runQuery(query, (err, entities) => {
          if (err) return reject(err);
          resolve(entities);
        });
      }
    })
  },

  find_v2(t, q, options) {
    return new Promise((resolve, reject)=>{
      var query = ds.createQuery(t);

       _.each(_.keys(q), (k)=>{
        query.filter(k,'=',q[k]);
      });


      if (options && options.sort) query.order('createdAt', {descending: true});
      if (options && options.limit) query.limit(options.limit);

      ds.runQuery(query, (err, results) => {
        if (err) return reject(err);
        resolve(results);
      });
    })
  },

  findWithQuery(query) {
    return new Promise((resolve, reject) => {
      ds.runQuery(query, (err, results) => {
        if (err) {
          console.error(query);
          console.error("Error findWithQuery:", err.toString());
          return reject(err);
        }
        resolve(results);
      });
    });
  },

  findOne(t, k, o, v) {
    return new Promise((resolve, reject) => {
      const query = ds.createQuery(t).filter(k, o, v).limit(1);
      ds.runQuery(query, (err, results) => {
        if (err) return reject(err);
        if (results.length < 1)
          return resolve(undefined);
        resolve(results[0]);
      });
    });
  },

  findOne_v2(t,q){
    return new Promise((resolve, reject)=>{
      var query = ds.createQuery(t);

      _.each(_.keys(q), (k)=>{
        query.filter(k,'=',q[k]);
      });

      ds.runQuery(query, (err, results)=>{
        if(err) return reject(err);
        if (results.length < 1)
          return resolve(undefined);
        resolve(results[0]);
      })
    })
  },

  findWithParticipants(type, vals) {

    if (!Array.isArray(vals)) {
      throw new Error('findWithParticipants: \'vals\' is not array');
    }

    return new Promise((resolve, reject) => {
      var query = ds.createQuery(type);
      vals.forEach((a) => {
        query.filter('participants', '=', a);
      });

      // TODO: XXX
      //  query.order('createdAt'); //, {descending: false});

      ds.runQuery(query, (err, results) => {
        if (err) return reject(err);

        console.log('with participants=>', results);

        resolve(results);
      });
    });
  },

  findParticipants(type, userid) {

    //  Console.log("Find participants for", userid);
    return new Promise((resolve, reject) => {
      var query = ds.createQuery(type)
        //.select(['conversationId'])
        // .select(['to', 'from'])

      .filter('participants', '=', userid)
        // .groupBy(['conversationId'])
        //
        // .filter('from', '=', userid)
        // .filter('to', '=', userid)
      ;


      ds.runQuery(query, (err, results) => {
        if (err) return reject(err);

        // Console.log(">>>>>>> results=>", results);

        var participants = [];

        if (!results) return resolve([]);

        results.forEach((r) => {
          participants = _.union(participants, r.data.participants);
        });

        _.pull(participants, userid);

        //Console.log("=>", userid, " * participants=", participants, "result=", results);
        console.log('=>', userid, ' * participants=', participants);

        resolve(participants);
      });


    });
  },

  findWithParent_V2(p, t, q) {
    return new Promise((resolve, reject) => {
      let query = ds.createQuery(t).hasAncestor(ds.key(p));

      _.each(_.keys(q), (k)=>{
        query.filter(k,'=',q[k]);
      });

      ds.runQuery(query, (err, results) => {
        if (err) return reject(err);
        resolve(results);
      });

    });
  },

  findWithParent(p, t, k, o, v) {
    return new Promise((resolve, reject) => {
      const query = ds.createQuery(t).hasAncestor(ds.key(p)).filter(k, o, v);
      ds.runQuery(query, (err, results) => {
        if (err) return reject(err);
        resolve(results);
      });
    });
  },

  /**
   * Count.
   * TODO: Cache this
   */
  countByCriteria(t, k, o, v) {
    return new Promise((resolve, reject)=>{
      const query = ds.createQuery(t).filter(k, o, v).select('__key__');
      ds.runQuery(query, (err, results)=>{
        if (err) {
          console.error("Error doing count:", err);
          return reject(err);
        }
        console.log(results);
        resolve(results.length);
      });
    });
  },

  /**
   * Get entity for the given key
   * @param   key to get
   * @return Entity
   */
  load(k) {
    return new Promise((resolve, reject) => {

      if (k.length < 2) {
        reject('load:error: Invalid key');
      }
      const key = ds.key(k);
      const cacheKey = key.path.join(':');
      console.log('Loading key =>', cacheKey);

      cache.load(cacheKey).then((cached) => {
        console.log('cached =>', cached)
        if (cached) return resolve({
          key: key,
          data: cached
        });

        ds.get(key, (err, entity) => {

          if (err) {
            console.log('Error loading from gcd:', err);
            return reject(err);
          }

          if (entity && entity.data) {
            cache.put(cacheKey, entity.data).catch((err) => {
              console.error('gcdatastore/load/ Failed to put to cache:', cacheKey);
            });
          }

          resolve(entity);
        });
      });
    });
  },

  /**
   * TODO: Test
   */
  // loadTwo(key1, key2) {
  //   return new Promise((resolve, reject)=>{
  //     log("gcd/loadTwo: keys to load:", key1, " + ", key2);
  //
  //     if (!key1 || !key2) {
  //       console.error("Invalid keys given:", key1, " + ", key2);
  //       reject("Error loadTwo(): Invalid keys");
  //     }
  //
  //     const k1 = ds.key(key1);
  //     const k2 = ds.key(key2);
  //
  //     ds.get(k1, (err, entity1)=>{
  //       if (err) {reject(err);}
  //       ds.get(k2, (err, entity2)=>{
  //         if (err) {reject(err);}
  //         resolve([entity1, entity2]);
  //       });
  //     });
  //
  //   });
  // },

  /**
   * Batch load from given keys.
   */
  loadMany(keys) {
    return new Promise((resolve, reject) => {

      log("gcd/loadMany:keys to load:", keys, keys.length);

      var keyList = keys.map((k) => {
        return ds.key(k);
      });

      console.log("Loading many keys=", keyList);

      ds.get(keyList, (err, entity) => {

        if (err) {
          console.log("Error loading from gcd:", err);
          return reject(err);
        }

        resolve(entity);
      });
    });
  },

  delete(k) {
    return new Promise((resolve, reject) => {
      const key = ds.key(k);
      ds.delete(key, (err) => {
        if (err) return reject(err);
        cache.expire(key.path.join(':'));
        resolve({done: true});
      });
    });
  },

  save(k, data, options) {
    if (!k || !data) throw "gcdatastore/save: Invalid key or data"
    const key = ds.key(k);
    //  console.log("Key=", key, "\ndata=", data);
    return this.patch(key, data, options);
  },

  /** key = [type, name/id] */
  patch(key, data, options = {}) {

    function extend(obj, src) {
      Object.keys(src).forEach(function (save) {
        obj[key] = src[key];
      });
      return obj;
    }

    return new Promise((resolve, reject) => {

      ds.get(key, (err, entity) => {
        if (err) {
          console.error("Error getting entity:", err);
          return reject(err);
        }

        if (!entity) entity = {
          data: {}
        };

        var patchedData = _.assign(entity.data, data);
        patchedData.updatedAt = new Date().toJSON();
        patchedData = _.omitBy(patchedData, _.isNil);

        var patchedArrayData = []
        for(var i in patchedData) {
          patchedArrayData.push({
            name: i,
            value: patchedData[i],
            excludeFromIndexes: (options && options.hasOwnProperty(i)) ? options[i]: false
          })
        }

        var patchedEntity = {
          key: key,
          data: patchedArrayData,
        }

        ds.save(patchedEntity, (err) => {
          if (err) {
            console.error("Error saving entity:", err);
            return reject(err);
          }
          cache.expire(key.path.join(':'));
          resolve({
            key: key,
            data: patchedData,
          });
        });
      });
    });

  },

  unpatch(key, k) {

    return new Promise((resolve, reject) => {
      ds.get(key, (err, entity) => {
        if (err) return reject(err);

        delete entity.data[k];

        ds.save(entity, (err) => {
          if (err) return reject(err);
          cache.expire(key.path.join(':'));
          resolve(entity);
        });
      });
    });
  },

  get(k) {
    return new Promise((resolve, reject) =>{
      const key = ds.key(k);
      ds.get(key, (err, entity) => {
        if (err) return reject(err);
        resolve(entity);
      });
    })
  },

  /**
   * Insert item to list array.
   * @param  {[type]} key  [description]
   * @param  {[type]} data [description]
   * @return {[type]}      Entity
   */
  // insert(k, data) {
  //   const key = ds.key(k);
  //   return new Promise((resolve, reject)=>{
  //
  //     ds.get(key, (err, entity)=>{
  //
  //       if (err) return reject(err);
  //       // if (!entity) return reject("Entity not found");
  //
  //       if (entity === undefined) {
  //         entity = {key:key, data:{count:1, body:[data]}};
  //       } else {
  //         var list = entity.data.body || [];
  //         const count = list.push(data);
  //         entity.data = {count:count, body:list}
  //       }
  //
  //       ds.save(entity, (err) => {
  //         if (err) return reject(err);
  //         resolve(entity);
  //       });
  //
  //     });
  //
  //   });
  // },

  /**
   * Remove from list.
   * @param  {[type]} k    [description]
   * @param  {[type]} data [description]
   * @return {[type]}      [description]
   */
  // remove(k, data) {
  //   const key = ds.key(k);
  //   return new Promise((resolve, reject)=>{
  //
  //   });
  // },

  fromDatastore(entity, options) {

    if (!entity) return {};

    delete entity.data.password;
    var type = entity.key.kind.toLowerCase();
    var noUserId = false;

    if (options && options.type) type = options.type;
    if (options && options.noUserId) noUserId = options.noUserId;

    var data = {
      id: entity.key.name,
      type: type,
      attributes: entity.data,
    };

    if (entity.data.userid && !noUserId) {
      data.relationships = {
        user: {
          data: {
            type: 'user',
            id: entity.data.userid
          }
        },
        profile: {
          data: {
            type: 'profile',
            id: entity.data.userid
          }
        }
      }
    }

    if (entity.data.targetid) {
      if (!data.relationships) data.relationships = {};
      data.relationships.target = {

          data: {
            type: 'user',
            id: entity.data.targetid
          }

      }
    }

    return data;
  },

  fromDatastoreForParticipants(entity) {
    if (!entity) return {};

    delete entity.data.password;
    var type = entity.key.kind.toLowerCase();

    var data = {
      id: entity.key.name,
      type: type,
      attributes: entity.data,
      relationships: {
        from: {
          data: {
            id: entity.data.from,
            type: 'user'
          }
        },
        to: {
          data: {
            id: entity.data.to,
            type: 'user'
          }
        }
      }
    };
    return data;
  }

}