buddycloud-server 0.3.5-2 / usr / share / buddycloud-server / local / model_postgres.js

// Generated by CoffeeScript 1.3.3
(function() {
  var LOST_TRANSACTION_TIMEOUT, Transaction, async, connectDB, dbIsAvailable, debugDB, errors, logger, ltx, parseEl, pg, pool, queue, withNextDb,
    __slice = [].slice,
    __hasProp = {}.hasOwnProperty;

  logger = require('../logger').makeLogger('local/model_postgres');

  pg = require("pg");

  ltx = require("ltx");

  async = require("async");

  errors = require("../errors");

  pool = [];

  queue = [];

  debugDB = function(db) {
    var oldQuery;
    oldQuery = db.query;
    return db.query = function(sql, params) {
      logger.trace("query " + sql + " " + (JSON.stringify(params)));
      return oldQuery.apply(this, arguments);
    };
  };

  connectDB = function(config) {
    var db;
    db = new pg.Client(config);
    debugDB(db);
    db.connect();
    db.on("error", function(err) {
      logger.error("Postgres: " + err.message);
      setTimeout(function() {
        return connectDB(config);
      }, Math.ceil(Math.random() * 5000));
      try {
        return db.end();
      } catch (e) {

      }
    });
    return db.connection.once("readyForQuery", function() {
      return dbIsAvailable(db);
    });
  };

  dbIsAvailable = function(db) {
    var cb;
    if ((cb = queue.shift())) {
      return cb(db);
    } else {
      return pool.push(db);
    }
  };

  withNextDb = function(cb) {
    var db;
    if ((db = pool.shift())) {
      return cb(db);
    } else {
      return queue.push(function(db) {
        return cb(db);
      });
    }
  };

  exports.start = function(config) {
    var i, _i, _ref, _results;
    _results = [];
    for (i = _i = 0, _ref = config.poolSize || 4; 0 <= _ref ? _i <= _ref : _i >= _ref; i = 0 <= _ref ? ++_i : --_i) {
      _results.push(connectDB(config));
    }
    return _results;
  };

  exports.transaction = function(cb) {
    return withNextDb(function(db) {
      return new Transaction(db, cb);
    });
  };

  exports.isListeningToNode = function(node, listenerJids, cb) {
    var conditions, i;
    i = 1;
    conditions = listenerJids.map(function(listenerJid) {
      i++;
      return "listener = $" + i;
    }).join(" OR ");
    if (!conditions) {
      return cb(null, false);
    }
    return withNextDb(function(db) {
      return db.query("SELECT listener FROM subscriptions WHERE node = $1 AND (" + conditions + ") LIMIT 1", [node].concat(__slice.call(listenerJids)), function(err, res) {
        var _ref;
        process.nextTick(function() {
          return dbIsAvailable(db);
        });
        return cb(err, ((res != null ? (_ref = res.rows) != null ? _ref[0] : void 0 : void 0) != null));
      });
    });
  };

  exports.nodeExists = function(node, cb) {
    return withNextDb(function(db) {
      return db.query("SELECT node FROM nodes WHERE node=$1", [node], function(err, res) {
        var _ref;
        process.nextTick(function() {
          return dbIsAvailable(db);
        });
        if (err) {
          return cb(err);
        } else {
          return cb(null, (res != null ? (_ref = res.rows) != null ? _ref[0] : void 0 : void 0) != null);
        }
      });
    });
  };

  exports.forListeners = function(iter) {
    return withNextDb(function(db) {
      return db.query("SELECT DISTINCT listener FROM subscriptions WHERE listener IS NOT NULL", function(err, res) {
        var _ref;
        process.nextTick(function() {
          return dbIsAvailable(db);
        });
        if (err) {
          logger.error(err);
          return;
        }
        return res != null ? (_ref = res.rows) != null ? _ref.forEach(function(row) {
          return iter(row.listener);
        }) : void 0 : void 0;
      });
    });
  };

  exports.getAllNodes = function(cb) {
    return withNextDb(function(db) {
      return db.query("SELECT node FROM nodes", function(err, res) {
        var nodes, _ref;
        process.nextTick(function() {
          return dbIsAvailable(db);
        });
        if (err) {
          return cb(err);
        }
        nodes = res != null ? (_ref = res.rows) != null ? _ref.map(function(row) {
          return row.node;
        }) : void 0 : void 0;
        return cb(null, nodes);
      });
    });
  };

  exports.getListenerNodes = function(listener, cb) {
    return db.query("SELECT DISTINCT node FROM subscriptions WHERE listener=$1", [listener], function(err, res) {
      var _ref;
      return cb(err, res != null ? (_ref = res.rows) != null ? _ref.map(function(row) {
        return row.node;
      }) : void 0 : void 0);
    });
  };

  LOST_TRANSACTION_TIMEOUT = 60 * 1000;

  Transaction = (function() {

    function Transaction(db, cb) {
      var timeout,
        _this = this;
      this.db = db;
      db.query("BEGIN", [], function(err, res) {
        return cb(err, _this);
      });
      timeout = setTimeout(function() {
        logger.error("Danger: lost transaction, rolling back!");
        timeout = void 0;
        return _this.rollback();
      }, LOST_TRANSACTION_TIMEOUT);
      this.rmTimeout = function() {
        if (timeout) {
          clearTimeout(timeout);
          return timeout = void 0;
        }
      };
    }

    Transaction.prototype.commit = function(cb) {
      var _this = this;
      return this.db.query("COMMIT", [], function(err, res) {
        _this.rmTimeout();
        process.nextTick(function() {
          return dbIsAvailable(_this.db);
        });
        return typeof cb === "function" ? cb(err) : void 0;
      });
    };

    Transaction.prototype.rollback = function(cb) {
      var _this = this;
      return this.db.query("ROLLBACK", [], function(err, res) {
        _this.rmTimeout();
        process.nextTick(function() {
          return dbIsAvailable(_this.db);
        });
        return typeof cb === "function" ? cb(err) : void 0;
      });
    };

    Transaction.prototype.validateNode = function(node) {
      var db;
      db = this.db;
      return function(cb) {
        return db.query("SELECT node FROM nodes WHERE node=$1", [node], function(err, res) {
          var _ref;
          if (err) {
            return cb(err);
          } else if ((res != null ? (_ref = res.rows) != null ? _ref[0] : void 0 : void 0) != null) {
            return cb(null);
          } else {
            logger.warn("" + node + " does not exist!");
            return cb(new errors.NotFound("Node does not exist"));
          }
        });
      };
    };

    Transaction.prototype.nodeExists = function(node, cb) {
      return this.db.query("SELECT node FROM nodes WHERE node=$1", [node], function(err, res) {
        var exists, _ref;
        if (err) {
          return cb(err);
        } else {
          exists = (res != null ? (_ref = res.rows) != null ? _ref[0] : void 0 : void 0) != null;
          if (!exists) {
            logger.warn("" + node + " does not exist");
          }
          return cb(null, exists);
        }
      });
    };

    Transaction.prototype.createNode = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT node FROM nodes WHERE node=$1", [node], cb2);
        }, function(res, cb2) {
          var _ref;
          if (res != null ? (_ref = res.rows) != null ? _ref[0] : void 0 : void 0) {
            return cb2(null, false);
          } else {
            return db.query("INSERT INTO nodes (node) VALUES ($1)", [node], function(err) {
              return cb2(err, true);
            });
          }
        }
      ], cb);
    };

    Transaction.prototype.purgeNode = function(node, cb) {
      var db, q;
      db = this.db;
      q = function(sql) {
        return function(cb2) {
          return db.query(sql, [node], cb2);
        };
      };
      return async.series([q("DELETE FROM items WHERE node=$1"), q("DELETE FROM subscriptions WHERE node=$1"), q("DELETE FROM affiliations WHERE node=$1"), q("DELETE FROM node_config WHERE node=$1"), q("DELETE FROM nodes WHERE node=$1")], function(err) {
        logger.info("Purged all data of node " + node);
        return cb(err);
      });
    };

    Transaction.prototype.listNodes = function(cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT node FROM nodes\nWHERE node IN (SELECT node FROM open_nodes)\nORDER BY node ASC", cb2);
        }, function(res, cb2) {
          var nodes;
          nodes = res.rows.map(function(row) {
            return {
              node: row.node,
              title: void 0
            };
          });
          return cb2(null, nodes);
        }
      ], cb);
    };

    Transaction.prototype.getSubscription = function(node, user, cb) {
      if (!node) {
        return cb(new Error("No node"));
      }
      if (!user) {
        return cb(new Error("No user"));
      }
      return this.db.query("SELECT subscription FROM subscriptions WHERE node=$1 AND \"user\"=$2", [node, user], function(err, res) {
        var _ref, _ref1;
        return cb(err, (res != null ? (_ref = res.rows) != null ? (_ref1 = _ref[0]) != null ? _ref1.subscription : void 0 : void 0 : void 0) || "none");
      });
    };

    Transaction.prototype.getSubscriptionListener = function(node, user, cb) {
      return this.db.query("SELECT listener FROM subscriptions WHERE node=$1 AND \"user\"=$2", [node, user], function(err, res) {
        var _ref, _ref1;
        return cb(err, (res != null ? (_ref = res.rows) != null ? (_ref1 = _ref[0]) != null ? _ref1.listener : void 0 : void 0 : void 0) || "none");
      });
    };

    Transaction.prototype.setSubscription = function(node, user, listener, subscription, cb) {
      var db, toDelete;
      if (!node) {
        return cb(new Error("No node"));
      }
      if (!user) {
        return cb(new Error("No user"));
      }
      db = this.db;
      toDelete = !subscription || subscription === "none";
      return async.waterfall([
        this.validateNode(node), function(cb2) {
          return db.query("SELECT subscription FROM subscriptions WHERE node=$1 AND \"user\"=$2", [node, user], cb2);
        }, function(res, cb2) {
          var isSet, _ref;
          isSet = (res != null ? (_ref = res.rows) != null ? _ref[0] : void 0 : void 0) != null;
          logger.debug("setSubscription " + node + " " + user + " isSet=" + isSet + " toDelete=" + toDelete);
          if (isSet && !toDelete) {
            if (listener) {
              return db.query("UPDATE subscriptions SET listener=$1, subscription=$2, updated=CURRENT_TIMESTAMP WHERE node=$3 AND \"user\"=$4", [listener, subscription, node, user], cb2);
            } else {
              return db.query("UPDATE subscriptions SET subscription=$1, updated=CURRENT_TIMESTAMP WHERE node=$2 AND \"user\"=$3", [subscription, node, user], cb2);
            }
          } else if (!isSet && !toDelete) {
            if (listener) {
              return db.query("INSERT INTO subscriptions (node, \"user\", listener, subscription, updated) VALUES ($1, $2, $3, $4, CURRENT_TIMESTAMP)", [node, user, listener, subscription], cb2);
            } else {
              return db.query("INSERT INTO subscriptions (node, \"user\", subscription, updated) VALUES ($1, $2, $3, CURRENT_TIMESTAMP)", [node, user, subscription], cb2);
            }
          } else if (isSet && toDelete) {
            return db.query("DELETE FROM subscriptions WHERE node=$1 AND \"user\"=$2", [node, user], cb2);
          } else if (!isSet && toDelete) {
            return cb2(null);
          } else {
            return cb2(new Error('Invalid subscription transition'));
          }
        }
      ], function(err) {
        return cb(err);
      });
    };

    Transaction.prototype.getSubscribers = function(node, cb) {
      var db;
      if (!node) {
        return cb(new Error("No node"));
      }
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT \"user\", subscription FROM subscriptions WHERE node=$1 ORDER BY updated DESC", [node], cb2);
        }, function(res, cb2) {
          var row, subscribers;
          subscribers = (function() {
            var _i, _len, _ref, _results;
            _ref = res.rows;
            _results = [];
            for (_i = 0, _len = _ref.length; _i < _len; _i++) {
              row = _ref[_i];
              _results.push({
                user: row.user,
                subscription: row.subscription
              });
            }
            return _results;
          })();
          return cb2(null, subscribers);
        }
      ], cb);
    };

    Transaction.prototype.getSubscriptions = function(actor, cb) {
      return this.db.query("SELECT \"user\", node, subscription FROM subscriptions WHERE \"user\"=$1 OR listener=$1 ORDER BY updated DESC", [actor], function(err, res) {
        return cb(err, res != null ? res.rows : void 0);
      });
    };

    Transaction.prototype.getPending = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT \"user\" FROM subscriptions WHERE subscription = 'pending' AND node = $1 ORDER BY updated DESC", [node], cb2);
        }, function(res, cb2) {
          return cb2(null, res.rows.map(function(row) {
            return row.user;
          }));
        }
      ], cb);
    };

    Transaction.prototype.getNodeListeners = function(node, cb) {
      return this.db.query("SELECT DISTINCT listener\nFROM subscriptions\nWHERE node=$1\nAND subscription='subscribed'\nAND listener IS NOT NULL", [node], function(err, res) {
        var _ref;
        return cb(err, res != null ? (_ref = res.rows) != null ? _ref.map(function(row) {
          return row.listener;
        }) : void 0 : void 0);
      });
    };

    Transaction.prototype.getNodeModeratorListeners = function(node, cb) {
      return this.db.query("SELECT DISTINCT listener\nFROM subscriptions\nWHERE node=$1\nAND listener IS NOT NULL\nAND EXISTS (SELECT affiliation\n            FROM affiliations\n            WHERE node=$1\n            AND (affiliation='owner' OR affiliation='moderator'))", [node], function(err, res) {
        var _ref;
        return cb(err, res != null ? (_ref = res.rows) != null ? _ref.map(function(row) {
          return row.listener;
        }) : void 0 : void 0);
      });
    };

    Transaction.prototype.walkModeratorAuthorizationRequests = function(user, iter, cb) {
      return this.db.query("SELECT \"user\", node\nFROM subscriptions\nWHERE subscription='pending'\nAND node IN (SELECT node\n             FROM affiliations\n             WHERE \"user\"=$1\n             AND (affiliation='owner' OR affiliation='moderator'))", [user], function(err, res) {
        var _ref;
        if (res != null) {
          if ((_ref = res.rows) != null) {
            _ref.forEach(function(row) {
              return iter(row);
            });
          }
        }
        return cb(err);
      });
    };

    Transaction.prototype.getUserRemoteSubscriptions = function(user, cb) {
      return this.db.query("SELECT node, listener, subscription FROM subscriptions WHERE \"user\"=$1 AND listener!=$1", [user], function(err, res) {
        return cb(err, res != null ? res.rows : void 0);
      });
    };

    Transaction.prototype.clearUserSubscriptions = function(user, cb) {
      return this.db.query("DELETE FROM subscriptions WHERE \"user\"=$1", [user], function(err) {
        return cb(err);
      });
    };

    Transaction.prototype.getAffiliation = function(node, user, cb) {
      if (!node) {
        return cb(new Error("No node"));
      }
      if (!user) {
        return cb(new Error("No user"));
      }
      return this.db.query("SELECT affiliation FROM affiliations WHERE node=$1 AND \"user\"=$2", [node, user], function(err, res) {
        var _ref, _ref1;
        return cb(err, (res != null ? (_ref = res.rows) != null ? (_ref1 = _ref[0]) != null ? _ref1.affiliation : void 0 : void 0 : void 0) || "none");
      });
    };

    Transaction.prototype.getListenerAffiliations = function(node, listener, cb) {
      if (!node) {
        return cb(new Error("No node"));
      }
      if (!listener) {
        return cb(new Error("No user"));
      }
      return this.db.query("SELECT DISTINCT affiliation FROM affiliations WHERE node=$1 AND \"user\" IN (SELECT \"user\" FROM subscriptions WHERE listener=$2)", [node, listener], function(err, res) {
        var _ref;
        return cb(err, res != null ? (_ref = res.rows) != null ? _ref.map(function(row) {
          return row.affiliation || "none";
        }) : void 0 : void 0);
      });
    };

    Transaction.prototype.setAffiliation = function(node, user, affiliation, cb) {
      var db;
      if (!node) {
        return cb(new Error("No node"));
      }
      if (!user) {
        return cb(new Error("No user"));
      }
      db = this.db;
      return async.waterfall([
        this.validateNode(node), function(cb2) {
          return db.query("SELECT affiliation FROM affiliations WHERE node=$1 AND \"user\"=$2", [node, user], cb2);
        }, function(res, cb2) {
          var isSet, toDelete;
          isSet = res && res.rows && res.rows[0];
          toDelete = !affiliation || affiliation === "none";
          if (isSet && !toDelete) {
            return db.query("UPDATE affiliations SET affiliation=$1 WHERE node=$2 AND \"user\"=$3", [affiliation, node, user], cb2);
          } else if (!isSet && !toDelete) {
            return db.query("INSERT INTO affiliations (node, \"user\", affiliation) VALUES ($1, $2, $3)", [node, user, affiliation], cb2);
          } else if (isSet && toDelete) {
            return db.query("DELETE FROM affiliations WHERE node=$1 AND \"user\"=$2", [node, user], cb2);
          } else if (!isSet && toDelete) {
            return cb2(null);
          }
        }
      ], function(err) {
        return cb(err);
      });
    };

    Transaction.prototype.getAffiliations = function(user, cb) {
      var db;
      if (!user) {
        return cb(new Error("No user"));
      }
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT node, affiliation FROM affiliations WHERE \"user\"=$1 ORDER BY updated DESC", [user], cb2);
        }, function(res, cb2) {
          var affiliations, row;
          affiliations = (function() {
            var _i, _len, _ref, _results;
            _ref = res.rows;
            _results = [];
            for (_i = 0, _len = _ref.length; _i < _len; _i++) {
              row = _ref[_i];
              _results.push({
                node: row.node,
                affiliation: row.affiliation
              });
            }
            return _results;
          })();
          return cb2(null, affiliations);
        }
      ], cb);
    };

    Transaction.prototype.getAffiliated = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT \"user\", affiliation FROM affiliations WHERE node=$1 ORDER BY updated DESC", [node], cb2);
        }, function(res, cb2) {
          var affiliations, row;
          affiliations = (function() {
            var _i, _len, _ref, _results;
            _ref = res.rows;
            _results = [];
            for (_i = 0, _len = _ref.length; _i < _len; _i++) {
              row = _ref[_i];
              _results.push({
                user: row.user,
                affiliation: row.affiliation
              });
            }
            return _results;
          })();
          return cb2(null, affiliations);
        }
      ], cb);
    };

    Transaction.prototype.getOutcast = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT \"user\" FROM affiliations WHERE affiliation = 'outcast' AND node = $1 ORDER BY updated DESC", [node], cb2);
        }, function(res, cb2) {
          return cb2(null, res.rows.map(function(row) {
            return row.user;
          }));
        }
      ], cb);
    };

    Transaction.prototype.getOwners = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT \"user\" FROM affiliations WHERE node=$1 AND affiliation='owner' ORDER BY updated DESC", [node], cb2);
        }, function(res, cb2) {
          return cb2(null, res.rows.map(function(row) {
            return row.user;
          }));
        }
      ], cb);
    };

    Transaction.prototype.getOwnersByNodePrefix = function(nodePrefix, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT DISTINCT \"user\" FROM affiliations WHERE node LIKE ($1 || '%') AND affiliation='owner'", [nodePrefix], cb2);
        }, function(res, cb2) {
          return cb2(null, res.rows.map(function(row) {
            return row.user;
          }));
        }
      ], cb);
    };

    Transaction.prototype.writeItem = function(node, id, el, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        this.validateNode(node), function(cb2) {
          return db.query("SELECT id FROM items WHERE node=$1 AND id=$2", [node, id], cb2);
        }, function(res, cb2) {
          var isSet, params, updated, updated_query, xml;
          isSet = res && res.rows && res.rows[0];
          xml = el.toString();
          params = [node, id, xml];
          updated = el.getChildText('updated') || el.getChildText('published');
          if (updated) {
            params.push(updated);
            updated_query = "$4";
          } else {
            updated_query = "CURRENT_TIMESTAMP";
          }
          if (isSet) {
            return db.query("UPDATE items SET xml=$3, updated=" + updated_query + " WHERE node=$1 AND id=$2", params, cb2);
          } else if (!isSet) {
            return db.query("INSERT INTO items (node, id, xml, updated) VALUES ($1, $2, $3, " + updated_query + ")", params, cb2);
          }
        }
      ], cb);
    };

    Transaction.prototype.getItemIds = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT id FROM items WHERE node=$1 ORDER BY updated DESC", [node], cb2);
        }, function(res, cb2) {
          var ids;
          ids = res.rows.map(function(row) {
            return row.id;
          });
          return cb2(null, ids);
        }
      ], cb);
    };

    Transaction.prototype.getItem = function(node, id, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        function(cb2) {
          return db.query("SELECT xml FROM items WHERE node=$1 AND id=$2", [node, id], cb2);
        }, function(res, cb2) {
          var el, _ref, _ref1;
          if (res != null ? (_ref = res.rows) != null ? (_ref1 = _ref[0]) != null ? _ref1.xml : void 0 : void 0 : void 0) {
            el = parseEl(res.rows[0].xml);
            if (el != null) {
              return cb2(null, el);
            } else {
              return cb2(new errors.InternalServerError("Item XML parse error"));
            }
          } else {
            return cb2(new errors.NotFound("No such item"));
          }
        }
      ], cb);
    };

    Transaction.prototype.getUpdatesByTime = function(subscriber, timeStart, timeEnd, itemCb, cb) {
      var conditions, i, params, q;
      conditions = ["node IN (SELECT node FROM subscriptions WHERE \"user\"=$1 AND subscription='subscribed')"];
      params = [subscriber];
      i = 1;
      if (timeStart) {
        conditions.push("updated >= $" + (++i) + "::timestamp");
        params.push(timeStart);
      }
      if (timeEnd) {
        conditions.push("updated <= $" + (++i) + "::timestamp");
        params.push(timeEnd);
      }
      q = this.db.query("SELECT id, node, xml FROM items WHERE " + conditions.join(" AND ") + " ORDER BY updated ASC", params);
      q.on("row", function(row) {
        if (item) {
          return itemCb({
            node: row.node,
            id: row.id,
            item: row.xml
          });
        }
      });
      q.on("error", function(err_) {
        var err;
        return err = err_;
      });
      return q.on("end", function() {
        return cb(err);
      });
    };

    Transaction.prototype.getConfig = function(node, cb) {
      var db;
      db = this.db;
      return async.waterfall([
        this.validateNode(node), function(cb2) {
          return db.query("SELECT \"key\", \"value\" FROM node_config WHERE node=$1", [node], cb2);
        }, function(res, cb2) {
          var config;
          if (res.rows) {
            config = {};
            res.rows.forEach(function(row) {
              return config[row.key] = row.value;
            });
            return cb2(null, config);
          } else {
            return cb2(new errors.NotFound("No such node"));
          }
        }
      ], cb);
    };

    Transaction.prototype.setConfig = function(node, config, cb) {
      var db;
      db = this.db;
      logger.debug("setConfig " + node + ": " + require("util").inspect(config));
      return async.waterfall([
        this.validateNode(node), function(cb2) {
          var key, value;
          return async.parallel((function() {
            var _results;
            _results = [];
            for (key in config) {
              if (!__hasProp.call(config, key)) continue;
              value = config[key];
              _results.push((function(key, value) {
                return function(cb3) {
                  if (value != null) {
                    return db.query("DELETE FROM node_config WHERE node=$1 AND key=$2", [node, key], function(err) {
                      if (err) {
                        return cb(err);
                      }
                      return db.query("INSERT INTO node_config (key, value, node, updated) VALUES ($1, $2, $3, CURRENT_TIMESTAMP)", [key, value, node], cb3);
                    });
                  } else {
                    return cb3(null);
                  }
                };
              })(key, value));
            }
            return _results;
          })(), function(err) {
            return cb2(err);
          });
        }
      ], cb);
    };

    Transaction.prototype.resetConfig = function(node, cb) {
      return this.db.query("DELETE FROM node_config WHERE node=$1", [node], cb);
    };

    Transaction.prototype.resetItems = function(node, cb) {
      return this.db.query("DELETE FROM items WHERE node=$1", [node], cb);
    };

    Transaction.prototype.resetSubscriptions = function(node, cb) {
      var _this = this;
      return this.db.query("SELECT \"user\", listener FROM subscriptions WHERE node=$1 AND listener IS NOT NULL", [node], function(err, res) {
        var row, userListeners, _i, _len, _ref;
        if (err) {
          return cb(err);
        }
        userListeners = {};
        _ref = res.rows;
        for (_i = 0, _len = _ref.length; _i < _len; _i++) {
          row = _ref[_i];
          userListeners[row.user] = row.listener;
        }
        return _this.db.query("DELETE FROM subscriptions WHERE node=$1", [node], function(err) {
          return cb(err, userListeners);
        });
      });
    };

    Transaction.prototype.resetAffiliations = function(node, cb) {
      return this.db.query("DELETE FROM affiliations WHERE node=$1", [node], cb);
    };

    Transaction.prototype.walkListenerArchive = function(listener, start, end, max, iter, cb) {
      var conds, db, i, limit, params, q;
      db = this.db;
      params = [listener];
      conds = "";
      i = params.length;
      if (start) {
        conds += "AND updated >= $" + (i += 1) + "::timestamp";
        params.push(start);
      }
      if (end) {
        conds += " AND updated <= $" + (i += 1) + "::timestamp";
        params.push(end);
      }
      limit = max ? (params.push(max), "LIMIT $" + (i += 1)) : "";
      q = function(fields, table, cb2, mapper) {
        return db.query("SELECT " + fields + ", updated FROM " + table + " WHERE node in (SELECT node FROM subscriptions WHERE listener=$1) " + conds + " ORDER BY updated DESC " + limit, params, function(err, res) {
          if (err) {
            return cb2(err);
          }
          if (mapper) {
            iter(res.rows.map(mapper));
          } else {
            iter(res.rows);
          }
          return cb2();
        });
      };
      return async.parallel([
        function(cb2) {
          var _this = this;
          return db.query("SELECT node, MAX(updated) AS updated FROM node_config WHERE node in (SELECT node FROM subscriptions WHERE listener=$1) " + conds + " GROUP BY node ORDER BY updated DESC " + limit, params, function(err, res) {
            if (err) {
              return cb2(err);
            }
            return async.forEach(res.rows, function(row, cb3) {
              var node;
              node = row.node;
              return db.query("SELECT key, value FROM node_config WHERE node=$1", [node], function(err, res) {
                var config, _i, _len, _ref;
                if (err) {
                  return cb3(err);
                }
                config = {};
                _ref = res.rows;
                for (_i = 0, _len = _ref.length; _i < _len; _i++) {
                  row = _ref[_i];
                  config[row.key] = row.value;
                }
                iter([
                  {
                    type: 'config',
                    node: node,
                    config: config
                  }
                ]);
                return cb3();
              });
            }, cb2);
          });
        }, function(cb2) {
          return q("node, id, xml", "items", cb2, function(row) {
            return {
              type: 'items',
              node: row.node,
              items: [
                {
                  id: row.id,
                  el: parseEl(row.xml)
                }
              ]
            };
          });
        }, function(cb2) {
          return q("node, \"user\", subscription, 'subscription' as type", "subscriptions", cb2);
        }, function(cb2) {
          return q("node, \"user\", affiliation, 'affiliation' as type", "affiliations", cb2);
        }
      ], function(err) {
        logger.debug('walkListenerArchive done');
        return cb(err);
      });
    };

    Transaction.prototype.getTopFollowedNodes = function(count, timespan, nodePattern, cb) {
      if (timespan == null) {
        timespan = "7 days";
      }
      if (nodePattern == null) {
        nodePattern = "/user/%@%/posts";
      }
      return this.db.query("SELECT node,\n       COUNT(user) AS count\nFROM subscriptions\nWHERE node LIKE $1\n  AND node IN (SELECT node FROM open_nodes)\n  AND updated >= CURRENT_TIMESTAMP - $2 :: INTERVAL\nGROUP BY node\nORDER BY count DESC\nLIMIT $3", [nodePattern, timespan, count], function(err, res) {
        return cb(err, res != null ? res.rows : void 0);
      });
    };

    Transaction.prototype.getTopPublishedNodes = function(count, timespan, nodePattern, cb) {
      if (timespan == null) {
        timespan = "7 days";
      }
      if (nodePattern == null) {
        nodePattern = "/user/%@%/posts";
      }
      return this.db.query("SELECT node,\n       COUNT(xml) AS count\nFROM items\nWHERE node LIKE $1\n  AND node IN (SELECT node FROM open_nodes)\n  AND updated >= CURRENT_TIMESTAMP - $2 :: INTERVAL\nGROUP BY node\nORDER BY count DESC\nLIMIT $3", [nodePattern, timespan, count], function(err, res) {
        return cb(err, res != null ? res.rows : void 0);
      });
    };

    return Transaction;

  })();

  parseEl = function(xml) {
    try {
      return ltx.parse(xml);
    } catch (e) {
      logger.error("Parsing " + xml + ": " + e.stack);
      return void 0;
    }
  };

}).call(this);