Nemcio/NodeBB
1
0
Fork 0
mirror of https://github.com/NodeBB/NodeBB.git synced 2025-12-29 03:40:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

PostgreSQL database driver (#5861)

Browse Source 
* [test/database/list] Fix test list 4 being used in two different tests

* [database/postgres] PostgreSQL database driver

* [database/postgres] Make transactions work based on continuation scope.

* [database/postgres] Implement nested transactions

* eslint --fix

* Add database changes from earlier this week to the PostgreSQL driver.

* Fix typo

* Fix postgres.incrObjectFieldBy returning undefined instead of null when given NaN

* [database/postgres] Fix sortedSetsCard returning an array of strings.

* Update socket.io postgres adapter

* Fix PostgreSQL erroring when multiple updates are made to the same sorted set entry in a single operation.

Add a test case to catch this error.

* Fix lint errors.

* Only prune sessions on one instance in a cluster to avoid deadlocks.

They're caught and handled by the database server, but they spam the logs.

* Fix arguments.slice.
This commit is contained in:
Ben Lubar
2018-08-08 14:13:48 -05:00
committed by Julian Lam
parent 3cccbbc1f2
commit 33228bb7fe
34 changed files with 3166 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

391
src/database/postgres/hash.js Normal file
View File

@@ -0,0 +1,391 @@
'use strict';
var async = require('async');
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
module.setObject = function (key, data, callback) {
callback = callback || helpers.noop;
if (!key || !data) {
return callback();
}
if (data.hasOwnProperty('')) {
delete data[''];
}
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'hash'),
async.apply(query, {
name: 'setObject',
text: `
INSERT INTO "legacy_hash" ("_key", "data")
VALUES ($1::TEXT, $2::TEXT::JSONB)
ON CONFLICT ("_key")
DO UPDATE SET "data" = "legacy_hash"."data" || $2::TEXT::JSONB`,
values: [key, JSON.stringify(data)],
}),
], function (err) {
done(err);
});
}, callback);
};
module.setObjectField = function (key, field, value, callback) {
callback = callback || helpers.noop;
if (!field) {
return callback();
}
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'hash'),
async.apply(query, {
name: 'setObjectField',
text: `
INSERT INTO "legacy_hash" ("_key", "data")
VALUES ($1::TEXT, jsonb_build_object($2::TEXT, $3::TEXT::JSONB))
ON CONFLICT ("_key")
DO UPDATE SET "data" = jsonb_set("legacy_hash"."data", ARRAY[$2::TEXT], $3::TEXT::JSONB)`,
values: [key, field, JSON.stringify(value)],
}),
], function (err) {
done(err);
});
}, callback);
};
module.getObject = function (key, callback) {
if (!key) {
return callback();
}
db.query({
name: 'getObject',
text: `
SELECT h."data"
FROM "legacy_object_live" o
INNER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
WHERE o."_key" = $1::TEXT
LIMIT 1`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].data);
}
callback(null, null);
});
};
module.getObjects = function (keys, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, []);
}
db.query({
name: 'getObjects',
text: `
SELECT h."data"
FROM UNNEST($1::TEXT[]) WITH ORDINALITY k("_key", i)
LEFT OUTER JOIN "legacy_object_live" o
ON o."_key" = k."_key"
LEFT OUTER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
ORDER BY k.i ASC`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, res.rows.map(function (row) {
return row.data;
}));
});
};
module.getObjectField = function (key, field, callback) {
if (!key) {
return callback();
}
db.query({
name: 'getObjectField',
text: `
SELECT h."data"->>$2::TEXT f
FROM "legacy_object_live" o
INNER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
WHERE o."_key" = $1::TEXT
LIMIT 1`,
values: [key, field],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].f);
}
callback(null, null);
});
};
module.getObjectFields = function (key, fields, callback) {
if (!key) {
return callback();
}
db.query({
name: 'getObjectFields',
text: `
SELECT (SELECT jsonb_object_agg(f, d."value")
FROM UNNEST($2::TEXT[]) f
LEFT OUTER JOIN jsonb_each(h."data") d
ON d."key" = f) d
FROM "legacy_object_live" o
INNER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
WHERE o."_key" = $1::TEXT`,
values: [key, fields],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].d);
}
var obj = {};
fields.forEach(function (f) {
obj[f] = null;
});
callback(null, obj);
});
};
module.getObjectsFields = function (keys, fields, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, []);
}
db.query({
name: 'getObjectsFields',
text: `
SELECT (SELECT jsonb_object_agg(f, d."value")
FROM UNNEST($2::TEXT[]) f
LEFT OUTER JOIN jsonb_each(h."data") d
ON d."key" = f) d
FROM UNNEST($1::text[]) WITH ORDINALITY k("_key", i)
LEFT OUTER JOIN "legacy_object_live" o
ON o."_key" = k."_key"
LEFT OUTER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
ORDER BY k.i ASC`,
values: [keys, fields],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, res.rows.map(function (row) {
return row.d;
}));
});
};
module.getObjectKeys = function (key, callback) {
if (!key) {
return callback();
}
db.query({
name: 'getObjectKeys',
text: `
SELECT ARRAY(SELECT jsonb_object_keys(h."data")) k
FROM "legacy_object_live" o
INNER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
WHERE o."_key" = $1::TEXT
LIMIT 1`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].k);
}
callback(null, []);
});
};
module.getObjectValues = function (key, callback) {
module.getObject(key, function (err, data) {
if (err) {
return callback(err);
}
var values = [];
if (data) {
for (var key in data) {
if (data.hasOwnProperty(key)) {
values.push(data[key]);
}
}
}
callback(null, values);
});
};
module.isObjectField = function (key, field, callback) {
if (!key) {
return callback();
}
db.query({
name: 'isObjectField',
text: `
SELECT (h."data" ? $2::TEXT AND h."data"->>$2::TEXT IS NOT NULL) b
FROM "legacy_object_live" o
INNER JOIN "legacy_hash" h
ON o."_key" = h."_key"
AND o."type" = h."type"
WHERE o."_key" = $1::TEXT
LIMIT 1`,
values: [key, field],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].b);
}
callback(null, false);
});
};
module.isObjectFields = function (key, fields, callback) {
if (!key) {
return callback();
}
module.getObjectFields(key, fields, function (err, data) {
if (err) {
return callback(err);
}
if (!data) {
return callback(null, fields.map(function () {
return false;
}));
}
callback(null, fields.map(function (field) {
return data.hasOwnProperty(field) && data[field] !== null;
}));
});
};
module.deleteObjectField = function (key, field, callback) {
module.deleteObjectFields(key, [field], callback);
};
module.deleteObjectFields = function (key, fields, callback) {
callback = callback || helpers.noop;
if (!key || !Array.isArray(fields) || !fields.length) {
return callback();
}
db.query({
name: 'deleteObjectFields',
text: `
UPDATE "legacy_hash"
SET "data" = COALESCE((SELECT jsonb_object_agg("key", "value")
FROM jsonb_each("data")
WHERE "key" <> ALL ($2::TEXT[])), '{}')
WHERE "_key" = $1::TEXT`,
values: [key, fields],
}, function (err) {
callback(err);
});
};
module.incrObjectField = function (key, field, callback) {
module.incrObjectFieldBy(key, field, 1, callback);
};
module.decrObjectField = function (key, field, callback) {
module.incrObjectFieldBy(key, field, -1, callback);
};
module.incrObjectFieldBy = function (key, field, value, callback) {
callback = callback || helpers.noop;
value = parseInt(value, 10);
if (!key || isNaN(value)) {
return callback(null, null);
}
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.waterfall([
async.apply(Array.isArray(key) ? helpers.ensureLegacyObjectsType : helpers.ensureLegacyObjectType, tx.client, key, 'hash'),
async.apply(query, Array.isArray(key) ? {
name: 'incrObjectFieldByMulti',
text: `
INSERT INTO "legacy_hash" ("_key", "data")
SELECT UNNEST($1::TEXT[]), jsonb_build_object($2::TEXT, $3::NUMERIC)
ON CONFLICT ("_key")
DO UPDATE SET "data" = jsonb_set("legacy_hash"."data", ARRAY[$2::TEXT], to_jsonb(COALESCE(("legacy_hash"."data"->>$2::TEXT)::NUMERIC, 0) + $3::NUMERIC))
RETURNING ("data"->>$2::TEXT)::NUMERIC v`,
values: [key, field, value],
} : {
name: 'incrObjectFieldBy',
text: `
INSERT INTO "legacy_hash" ("_key", "data")
VALUES ($1::TEXT, jsonb_build_object($2::TEXT, $3::NUMERIC))
ON CONFLICT ("_key")
DO UPDATE SET "data" = jsonb_set("legacy_hash"."data", ARRAY[$2::TEXT], to_jsonb(COALESCE(("legacy_hash"."data"->>$2::TEXT)::NUMERIC, 0) + $3::NUMERIC))
RETURNING ("data"->>$2::TEXT)::NUMERIC v`,
values: [key, field, value],
}),
function (res, next) {
next(null, Array.isArray(key) ? res.rows.map(function (r) {
return parseFloat(r.v);
}) : parseFloat(res.rows[0].v));
},
], done);
}, callback);
};
};

139
src/database/postgres/helpers.js Normal file
View File

@@ -0,0 +1,139 @@
'use strict';
var helpers = {};
helpers.valueToString = function (value) {
if (value === null || value === undefined) {
return value;
}
return value.toString();
};
helpers.removeDuplicateValues = function (values) {
var others = Array.prototype.slice.call(arguments, 1);
for (var i = 0; i < values.length; i++) {
if (values.lastIndexOf(values[i]) !== i) {
values.splice(i, 1);
for (var j = 0; j < others.length; j++) {
others[j].splice(i, 1);
}
i -= 1;
}
}
};
helpers.ensureLegacyObjectType = function (db, key, type, callback) {
db.query({
name: 'ensureLegacyObjectTypeBefore',
text: `
DELETE FROM "legacy_object"
WHERE "expireAt" IS NOT NULL
AND "expireAt" <= CURRENT_TIMESTAMP`,
}, function (err) {
if (err) {
return callback(err);
}
db.query({
name: 'ensureLegacyObjectType1',
text: `
INSERT INTO "legacy_object" ("_key", "type")
VALUES ($1::TEXT, $2::TEXT::LEGACY_OBJECT_TYPE)
ON CONFLICT
DO NOTHING`,
values: [key, type],
}, function (err) {
if (err) {
return callback(err);
}
db.query({
name: 'ensureLegacyObjectType2',
text: `
SELECT "type"
FROM "legacy_object_live"
WHERE "_key" = $1::TEXT`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows[0].type !== type) {
return callback(new Error('database: cannot insert ' + JSON.stringify(key) + ' as ' + type + ' because it already exists as ' + res.rows[0].type));
}
callback(null);
});
});
});
};
helpers.ensureLegacyObjectsType = function (db, keys, type, callback) {
db.query({
name: 'ensureLegacyObjectTypeBefore',
text: `
DELETE FROM "legacy_object"
WHERE "expireAt" IS NOT NULL
AND "expireAt" <= CURRENT_TIMESTAMP`,
}, function (err) {
if (err) {
return callback(err);
}
db.query({
name: 'ensureLegacyObjectsType1',
text: `
INSERT INTO "legacy_object" ("_key", "type")
SELECT k, $2::TEXT::LEGACY_OBJECT_TYPE
FROM UNNEST($1::TEXT[]) k
ON CONFLICT
DO NOTHING`,
values: [keys, type],
}, function (err) {
if (err) {
return callback(err);
}
db.query({
name: 'ensureLegacyObjectsType2',
text: `
SELECT "_key", "type"
FROM "legacy_object_live"
WHERE "_key" = ANY($1::TEXT[])`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
var invalid = res.rows.filter(function (r) {
return r.type !== type;
});
if (invalid.length) {
return callback(new Error('database: cannot insert multiple objects as ' + type + ' because they already exist: ' + invalid.map(function (r) {
return JSON.stringify(r._key) + ' is ' + r.type;
}).join(', ')));
}
var missing = keys.filter(function (k) {
return !res.rows.some(function (r) {
return r._key === k;
});
});
if (missing.length) {
return callback(new Error('database: failed to insert keys for objects: ' + JSON.stringify(missing)));
}
callback(null);
});
});
});
};
helpers.noop = function () {};
module.exports = helpers;

234
src/database/postgres/list.js Normal file
View File

@@ -0,0 +1,234 @@
'use strict';
var async = require('async');
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
module.listPrepend = function (key, value, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'list'),
async.apply(query, {
name: 'listPrepend',
text: `
INSERT INTO "legacy_list" ("_key", "array")
VALUES ($1::TEXT, ARRAY[$2::TEXT])
ON CONFLICT ("_key")
DO UPDATE SET "array" = ARRAY[$2::TEXT] || "legacy_list"."array"`,
values: [key, value],
}),
], function (err) {
done(err);
});
}, callback);
};
module.listAppend = function (key, value, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'list'),
async.apply(query, {
name: 'listAppend',
text: `
INSERT INTO "legacy_list" ("_key", "array")
VALUES ($1::TEXT, ARRAY[$2::TEXT])
ON CONFLICT ("_key")
DO UPDATE SET "array" = "legacy_list"."array" || ARRAY[$2::TEXT]`,
values: [key, value],
}),
], function (err) {
done(err);
});
}, callback || helpers.noop);
};
module.listRemoveLast = function (key, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
db.query({
name: 'listRemoveLast',
text: `
WITH A AS (
SELECT l.*
FROM "legacy_object_live" o
INNER JOIN "legacy_list" l
ON o."_key" = l."_key"
AND o."type" = l."type"
WHERE o."_key" = $1::TEXT
FOR UPDATE)
UPDATE "legacy_list" l
SET "array" = A."array"[1 : array_length(A."array", 1) - 1]
FROM A
WHERE A."_key" = l."_key"
RETURNING A."array"[array_length(A."array", 1)] v`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].v);
}
callback(null, null);
});
};
module.listRemoveAll = function (key, value, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
db.query({
name: 'listRemoveAll',
text: `
UPDATE "legacy_list" l
SET "array" = array_remove(l."array", $2::TEXT)
FROM "legacy_object_live" o
WHERE o."_key" = l."_key"
AND o."type" = l."type"
AND o."_key" = $1::TEXT`,
values: [key, value],
}, function (err) {
callback(err);
});
};
module.listTrim = function (key, start, stop, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
stop += 1;
db.query(stop > 0 ? {
name: 'listTrim',
text: `
UPDATE "legacy_list" l
SET "array" = ARRAY(SELECT m.m
FROM UNNEST(l."array") WITH ORDINALITY m(m, i)
ORDER BY m.i ASC
LIMIT ($3::INTEGER - $2::INTEGER)
OFFSET $2::INTEGER)
FROM "legacy_object_live" o
WHERE o."_key" = l."_key"
AND o."type" = l."type"
AND o."_key" = $1::TEXT`,
values: [key, start, stop],
} : {
name: 'listTrimBack',
text: `
UPDATE "legacy_list" l
SET "array" = ARRAY(SELECT m.m
FROM UNNEST(l."array") WITH ORDINALITY m(m, i)
ORDER BY m.i ASC
LIMIT ($3::INTEGER - $2::INTEGER + array_length(l."array", 1))
OFFSET $2::INTEGER)
FROM "legacy_object_live" o
WHERE o."_key" = l."_key"
AND o."type" = l."type"
AND o."_key" = $1::TEXT`,
values: [key, start, stop],
}, function (err) {
callback(err);
});
};
module.getListRange = function (key, start, stop, callback) {
if (!key) {
return callback();
}
stop += 1;
db.query(stop > 0 ? {
name: 'getListRange',
text: `
SELECT ARRAY(SELECT m.m
FROM UNNEST(l."array") WITH ORDINALITY m(m, i)
ORDER BY m.i ASC
LIMIT ($3::INTEGER - $2::INTEGER)
OFFSET $2::INTEGER) l
FROM "legacy_object_live" o
INNER JOIN "legacy_list" l
ON o."_key" = l."_key"
AND o."type" = l."type"
WHERE o."_key" = $1::TEXT`,
values: [key, start, stop],
} : {
name: 'getListRangeBack',
text: `
SELECT ARRAY(SELECT m.m
FROM UNNEST(l."array") WITH ORDINALITY m(m, i)
ORDER BY m.i ASC
LIMIT ($3::INTEGER - $2::INTEGER + array_length(l."array", 1))
OFFSET $2::INTEGER) l
FROM "legacy_object_live" o
INNER JOIN "legacy_list" l
ON o."_key" = l."_key"
AND o."type" = l."type"
WHERE o."_key" = $1::TEXT`,
values: [key, start, stop],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].l);
}
callback(null, []);
});
};
module.listLength = function (key, callback) {
db.query({
name: 'listLength',
text: `
SELECT array_length(l."array", 1) l
FROM "legacy_object_live" o
INNER JOIN "legacy_list" l
ON o."_key" = l."_key"
AND o."type" = l."type"
WHERE o."_key" = $1::TEXT`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].l);
}
callback(null, 0);
});
};
};

239
src/database/postgres/main.js Normal file
View File

@@ -0,0 +1,239 @@
'use strict';
var async = require('async');
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
var query = db.query.bind(db);
module.flushdb = function (callback) {
callback = callback || helpers.noop;
async.series([
async.apply(query, `DROP SCHEMA "public" CASCADE`),
async.apply(query, `CREATE SCHEMA "public"`),
], function (err) {
callback(err);
});
};
module.emptydb = function (callback) {
callback = callback || helpers.noop;
query(`DELETE FROM "legacy_object"`, function (err) {
callback(err);
});
};
module.exists = function (key, callback) {
if (!key) {
return callback();
}
query({
name: 'exists',
text: `
SELECT EXISTS(SELECT *
FROM "legacy_object_live"
WHERE "_key" = $1::TEXT
LIMIT 1) e`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, res.rows[0].e);
});
};
module.delete = function (key, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
query({
name: 'delete',
text: `
DELETE FROM "legacy_object"
WHERE "_key" = $1::TEXT`,
values: [key],
}, function (err) {
callback(err);
});
};
module.deleteAll = function (keys, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
query({
name: 'deleteAll',
text: `
DELETE FROM "legacy_object"
WHERE "_key" = ANY($1::TEXT[])`,
values: [keys],
}, function (err) {
callback(err);
});
};
module.get = function (key, callback) {
if (!key) {
return callback();
}
query({
name: 'get',
text: `
SELECT s."data" t
FROM "legacy_object_live" o
INNER JOIN "legacy_string" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = $1::TEXT
LIMIT 1`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].t);
}
callback(null, null);
});
};
module.set = function (key, value, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
module.transaction(function (tx, done) {
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'string'),
async.apply(tx.client.query.bind(tx.client), {
name: 'set',
text: `
INSERT INTO "legacy_string" ("_key", "data")
VALUES ($1::TEXT, $2::TEXT)
ON CONFLICT ("_key")
DO UPDATE SET "data" = $2::TEXT`,
values: [key, value],
}),
], function (err) {
done(err);
});
}, callback);
};
module.increment = function (key, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
module.transaction(function (tx, done) {
async.waterfall([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'string'),
async.apply(tx.client.query.bind(tx.client), {
name: 'increment',
text: `
INSERT INTO "legacy_string" ("_key", "data")
VALUES ($1::TEXT, '1')
ON CONFLICT ("_key")
DO UPDATE SET "data" = ("legacy_string"."data"::NUMERIC + 1)::TEXT
RETURNING "data" d`,
values: [key],
}),
], function (err, res) {
if (err) {
return done(err);
}
done(null, parseFloat(res.rows[0].d));
});
}, callback);
};
module.rename = function (oldKey, newKey, callback) {
module.transaction(function (tx, done) {
async.series([
async.apply(tx.delete, newKey),
async.apply(tx.client.query.bind(tx.client), {
name: 'rename',
text: `
UPDATE "legacy_object"
SET "_key" = $2::TEXT
WHERE "_key" = $1::TEXT`,
values: [oldKey, newKey],
}),
], function (err) {
done(err);
});
}, callback || helpers.noop);
};
module.type = function (key, callback) {
query({
name: 'type',
text: `
SELECT "type"::TEXT t
FROM "legacy_object_live"
WHERE "_key" = $1::TEXT
LIMIT 1`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].t);
}
callback(null, null);
});
};
function doExpire(key, date, callback) {
query({
name: 'expire',
text: `
UPDATE "legacy_object"
SET "expireAt" = $2::TIMESTAMPTZ
WHERE "_key" = $1::TEXT`,
values: [key, date],
}, function (err) {
if (callback) {
callback(err);
}
});
}
module.expire = function (key, seconds, callback) {
doExpire(key, new Date(((Date.now() / 1000) + seconds) * 1000), callback);
};
module.expireAt = function (key, timestamp, callback) {
doExpire(key, new Date(timestamp * 1000), callback);
};
module.pexpire = function (key, ms, callback) {
doExpire(key, new Date(Date.now() + parseInt(ms, 10)), callback);
};
module.pexpireAt = function (key, timestamp, callback) {
doExpire(key, new Date(timestamp), callback);
};
};

51
src/database/postgres/pubsub.js Normal file
View File

@@ -0,0 +1,51 @@
'use strict';
var util = require('util');
var winston = require('winston');
var EventEmitter = require('events').EventEmitter;
var pg = require('pg');
var db = require('../postgres');
var PubSub = function () {
var self = this;
var subClient = new pg.Client(db.getConnectionOptions());
subClient.connect(function (err) {
if (err) {
winston.error(err);
return;
}
subClient.query('LISTEN pubsub', function (err) {
if (err) {
winston.error(err);
}
});
subClient.on('notification', function (message) {
if (message.channel !== 'pubsub') {
return;
}
try {
var msg = JSON.parse(message.payload);
self.emit(msg.event, msg.data);
} catch (err) {
winston.error(err.stack);
}
});
});
};
util.inherits(PubSub, EventEmitter);
PubSub.prototype.publish = function (event, data) {
db.pool.query({
name: 'pubSubPublish',
text: `SELECT pg_notify('pubsub', $1::TEXT)`,
values: [JSON.stringify({ event: event, data: data })],
});
};
module.exports = new PubSub();

342
src/database/postgres/sets.js Normal file
View File

@@ -0,0 +1,342 @@
'use strict';
var async = require('async');
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
module.setAdd = function (key, value, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(value)) {
value = [value];
}
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'set'),
async.apply(query, {
name: 'setAdd',
text: `
INSERT INTO "legacy_set" ("_key", "member")
SELECT $1::TEXT, m
FROM UNNEST($2::TEXT[]) m
ON CONFLICT ("_key", "member")
DO NOTHING`,
values: [key, value],
}),
], function (err) {
done(err);
});
}, callback);
};
module.setsAdd = function (keys, value, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
if (!Array.isArray(value)) {
value = [value];
}
keys = keys.filter(function (k, i, a) {
return a.indexOf(k) === i;
});
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectsType, tx.client, keys, 'set'),
async.apply(query, {
name: 'setsAdd',
text: `
INSERT INTO "legacy_set" ("_key", "member")
SELECT k, m
FROM UNNEST($1::TEXT[]) k
CROSS JOIN UNNEST($2::TEXT[]) m
ON CONFLICT ("_key", "member")
DO NOTHING`,
values: [keys, value],
}),
], function (err) {
done(err);
});
}, callback);
};
module.setRemove = function (key, value, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(key)) {
key = [key];
}
if (!Array.isArray(value)) {
value = [value];
}
db.query({
name: 'setRemove',
text: `
DELETE FROM "legacy_set"
WHERE "_key" = ANY($1::TEXT[])
AND "member" = ANY($2::TEXT[])`,
values: [key, value],
}, function (err) {
callback(err);
});
};
module.setsRemove = function (keys, value, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
db.query({
name: 'setsRemove',
text: `
DELETE FROM "legacy_set"
WHERE "_key" = ANY($1::TEXT[])
AND "member" = $2::TEXT`,
values: [keys, value],
}, function (err) {
callback(err);
});
};
module.isSetMember = function (key, value, callback) {
if (!key) {
return callback(null, false);
}
db.query({
name: 'isSetMember',
text: `
SELECT 1
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = $1::TEXT
AND s."member" = $2::TEXT`,
values: [key, value],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, !!res.rows.length);
});
};
module.isSetMembers = function (key, values, callback) {
if (!key || !Array.isArray(values) || !values.length) {
return callback(null, []);
}
values = values.map(helpers.valueToString);
db.query({
name: 'isSetMembers',
text: `
SELECT s."member" m
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = $1::TEXT
AND s."member" = ANY($2::TEXT[])`,
values: [key, values],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, values.map(function (v) {
return res.rows.some(function (r) {
return r.m === v;
});
}));
});
};
module.isMemberOfSets = function (sets, value, callback) {
if (!Array.isArray(sets) || !sets.length) {
return callback(null, []);
}
value = helpers.valueToString(value);
db.query({
name: 'isMemberOfSets',
text: `
SELECT o."_key" k
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = ANY($1::TEXT[])
AND s."member" = $2::TEXT`,
values: [sets, value],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, sets.map(function (s) {
return res.rows.some(function (r) {
return r.k === s;
});
}));
});
};
module.getSetMembers = function (key, callback) {
if (!key) {
return callback(null, []);
}
db.query({
name: 'getSetMembers',
text: `
SELECT s."member" m
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = $1::TEXT`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, res.rows.map(function (r) {
return r.m;
}));
});
};
module.getSetsMembers = function (keys, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, []);
}
db.query({
name: 'getSetsMembers',
text: `
SELECT o."_key" k,
array_agg(s."member") m
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = ANY($1::TEXT[])
GROUP BY o."_key"`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, keys.map(function (k) {
return (res.rows.find(function (r) {
return r.k === k;
}) || { m: [] }).m;
}));
});
};
module.setCount = function (key, callback) {
if (!key) {
return callback(null, 0);
}
db.query({
name: 'setCount',
text: `
SELECT COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = $1::TEXT`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, parseInt(res.rows[0].c, 10));
});
};
module.setsCount = function (keys, callback) {
db.query({
name: 'setsCount',
text: `
SELECT o."_key" k,
COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = ANY($1::TEXT[])
GROUP BY o."_key"`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, keys.map(function (k) {
return (res.rows.find(function (r) {
return r.k === k;
}) || { c: 0 }).c;
}));
});
};
module.setRemoveRandom = function (key, callback) {
callback = callback || helpers.noop;
db.query({
name: 'setRemoveRandom',
text: `
WITH A AS (
SELECT s."member"
FROM "legacy_object_live" o
INNER JOIN "legacy_set" s
ON o."_key" = s."_key"
AND o."type" = s."type"
WHERE o."_key" = $1::TEXT
ORDER BY RANDOM()
LIMIT 1
FOR UPDATE)
DELETE FROM "legacy_set" s
USING A
WHERE s."_key" = $1::TEXT
AND s."member" = A."member"
RETURNING A."member" m`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, res.rows[0].m);
}
callback(null, null);
});
};
};

744
src/database/postgres/sorted.js Normal file
View File

@@ -0,0 +1,744 @@
'use strict';
var async = require('async');
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
var query = db.query.bind(db);
require('./sorted/add')(db, module);
require('./sorted/remove')(db, module);
require('./sorted/union')(db, module);
require('./sorted/intersect')(db, module);
module.getSortedSetRange = function (key, start, stop, callback) {
getSortedSetRange(key, start, stop, 1, false, callback);
};
module.getSortedSetRevRange = function (key, start, stop, callback) {
getSortedSetRange(key, start, stop, -1, false, callback);
};
module.getSortedSetRangeWithScores = function (key, start, stop, callback) {
getSortedSetRange(key, start, stop, 1, true, callback);
};
module.getSortedSetRevRangeWithScores = function (key, start, stop, callback) {
getSortedSetRange(key, start, stop, -1, true, callback);
};
function getSortedSetRange(key, start, stop, sort, withScores, callback) {
if (!key) {
return callback();
}
if (!Array.isArray(key)) {
key = [key];
}
if (start < 0 && start > stop) {
return callback(null, []);
}
var reverse = false;
if (start === 0 && stop < -1) {
reverse = true;
sort *= -1;
start = Math.abs(stop + 1);
stop = -1;
} else if (start < 0 && stop > start) {
var tmp1 = Math.abs(stop + 1);
stop = Math.abs(start + 1);
start = tmp1;
}
var limit = stop - start + 1;
if (limit <= 0) {
limit = null;
}
query({
name: 'getSortedSetRangeWithScores' + (sort > 0 ? 'Asc' : 'Desc'),
text: `
SELECT z."value",
z."score"
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
ORDER BY z."score" ` + (sort > 0 ? 'ASC' : 'DESC') + `
LIMIT $3::INTEGER
OFFSET $2::INTEGER`,
values: [key, start, limit],
}, function (err, res) {
if (err) {
return callback(err);
}
if (reverse) {
res.rows.reverse();
}
if (withScores) {
res.rows = res.rows.map(function (r) {
return {
value: r.value,
score: parseFloat(r.score),
};
});
} else {
res.rows = res.rows.map(function (r) {
return r.value;
});
}
callback(null, res.rows);
});
}
module.getSortedSetRangeByScore = function (key, start, count, min, max, callback) {
getSortedSetRangeByScore(key, start, count, min, max, 1, false, callback);
};
module.getSortedSetRevRangeByScore = function (key, start, count, max, min, callback) {
getSortedSetRangeByScore(key, start, count, min, max, -1, false, callback);
};
module.getSortedSetRangeByScoreWithScores = function (key, start, count, min, max, callback) {
getSortedSetRangeByScore(key, start, count, min, max, 1, true, callback);
};
module.getSortedSetRevRangeByScoreWithScores = function (key, start, count, max, min, callback) {
getSortedSetRangeByScore(key, start, count, min, max, -1, true, callback);
};
function getSortedSetRangeByScore(key, start, count, min, max, sort, withScores, callback) {
if (!key) {
return callback();
}
if (!Array.isArray(key)) {
key = [key];
}
if (parseInt(count, 10) === -1) {
count = null;
}
if (min === '-inf') {
min = null;
}
if (max === '+inf') {
max = null;
}
query({
name: 'getSortedSetRangeByScoreWithScores' + (sort > 0 ? 'Asc' : 'Desc'),
text: `
SELECT z."value",
z."score"
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
AND (z."score" >= $4::NUMERIC OR $4::NUMERIC IS NULL)
AND (z."score" <= $5::NUMERIC OR $5::NUMERIC IS NULL)
ORDER BY z."score" ` + (sort > 0 ? 'ASC' : 'DESC') + `
LIMIT $3::INTEGER
OFFSET $2::INTEGER`,
values: [key, start, count, min, max],
}, function (err, res) {
if (err) {
return callback(err);
}
if (withScores) {
res.rows = res.rows.map(function (r) {
return {
value: r.value,
score: parseFloat(r.score),
};
});
} else {
res.rows = res.rows.map(function (r) {
return r.value;
});
}
return callback(null, res.rows);
});
}
module.sortedSetCount = function (key, min, max, callback) {
if (!key) {
return callback();
}
if (min === '-inf') {
min = null;
}
if (max === '+inf') {
max = null;
}
query({
name: 'sortedSetCount',
text: `
SELECT COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT
AND (z."score" >= $2::NUMERIC OR $2::NUMERIC IS NULL)
AND (z."score" <= $3::NUMERIC OR $3::NUMERIC IS NULL)`,
values: [key, min, max],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, parseInt(res.rows[0].c, 10));
});
};
module.sortedSetCard = function (key, callback) {
if (!key) {
return callback(null, 0);
}
query({
name: 'sortedSetCard',
text: `
SELECT COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT`,
values: [key],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, parseInt(res.rows[0].c, 10));
});
};
module.sortedSetsCard = function (keys, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
query({
name: 'sortedSetsCard',
text: `
SELECT o."_key" k,
COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
GROUP BY o."_key"`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, keys.map(function (k) {
return parseInt((res.rows.find(function (r) {
return r.k === k;
}) || { c: 0 }).c, 10);
}));
});
};
module.sortedSetRank = function (key, value, callback) {
getSortedSetRank('ASC', [key], [value], function (err, result) {
callback(err, result ? result[0] : null);
});
};
module.sortedSetRevRank = function (key, value, callback) {
getSortedSetRank('DESC', [key], [value], function (err, result) {
callback(err, result ? result[0] : null);
});
};
function getSortedSetRank(sort, keys, values, callback) {
values = values.map(helpers.valueToString);
query({
name: 'getSortedSetRank' + sort,
text: `
SELECT (SELECT r
FROM (SELECT z."value" v,
RANK() OVER (PARTITION BY o."_key"
ORDER BY z."score" ` + sort + `,
z."value" ` + sort + `) - 1 r
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = kvi.k) r
WHERE v = kvi.v) r
FROM UNNEST($1::TEXT[], $2::TEXT[]) WITH ORDINALITY kvi(k, v, i)
ORDER BY kvi.i ASC`,
values: [keys, values],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, res.rows.map(function (r) { return r.r === null ? null : parseFloat(r.r); }));
});
}
module.sortedSetsRanks = function (keys, values, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, []);
}
getSortedSetRank('ASC', keys, values, callback);
};
module.sortedSetRanks = function (key, values, callback) {
if (!Array.isArray(values) || !values.length) {
return callback(null, []);
}
getSortedSetRank('ASC', new Array(values.length).fill(key), values, callback);
};
module.sortedSetScore = function (key, value, callback) {
if (!key) {
return callback(null, null);
}
value = helpers.valueToString(value);
query({
name: 'sortedSetScore',
text: `
SELECT z."score" s
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT
AND z."value" = $2::TEXT`,
values: [key, value],
}, function (err, res) {
if (err) {
return callback(err);
}
if (res.rows.length) {
return callback(null, parseFloat(res.rows[0].s));
}
callback(null, null);
});
};
module.sortedSetsScore = function (keys, value, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
value = helpers.valueToString(value);
query({
name: 'sortedSetsScore',
text: `
SELECT o."_key" k,
z."score" s
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
AND z."value" = $2::TEXT`,
values: [keys, value],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, keys.map(function (k) {
var s = res.rows.find(function (r) {
return r.k === k;
});
return s ? parseFloat(s.s) : null;
}));
});
};
module.sortedSetScores = function (key, values, callback) {
if (!key) {
return callback(null, null);
}
values = values.map(helpers.valueToString);
query({
name: 'sortedSetScores',
text: `
SELECT z."value" v,
z."score" s
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT
AND z."value" = ANY($2::TEXT[])`,
values: [key, values],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, values.map(function (v) {
var s = res.rows.find(function (r) {
return r.v === v;
});
return s ? parseFloat(s.s) : null;
}));
});
};
module.isSortedSetMember = function (key, value, callback) {
if (!key) {
return callback();
}
value = helpers.valueToString(value);
query({
name: 'isSortedSetMember',
text: `
SELECT 1
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT
AND z."value" = $2::TEXT`,
values: [key, value],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, !!res.rows.length);
});
};
module.isSortedSetMembers = function (key, values, callback) {
if (!key) {
return callback();
}
values = values.map(helpers.valueToString);
query({
name: 'isSortedSetMembers',
text: `
SELECT z."value" v
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT
AND z."value" = ANY($2::TEXT[])`,
values: [key, values],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, values.map(function (v) {
return res.rows.some(function (r) {
return r.v === v;
});
}));
});
};
module.isMemberOfSortedSets = function (keys, value, callback) {
if (!Array.isArray(keys)) {
return callback();
}
value = helpers.valueToString(value);
query({
name: 'isMemberOfSortedSets',
text: `
SELECT o."_key" k
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
AND z."value" = $2::TEXT`,
values: [keys, value],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, keys.map(function (k) {
return res.rows.some(function (r) {
return r.k === k;
});
}));
});
};
module.getSortedSetsMembers = function (keys, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, []);
}
query({
name: 'getSortedSetsMembers',
text: `
SELECT o."_key" k,
array_agg(z."value" ORDER BY z."score" ASC) m
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
GROUP BY o."_key"`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, keys.map(function (k) {
return (res.rows.find(function (r) {
return r.k === k;
}) || { m: [] }).m;
}));
});
};
module.sortedSetIncrBy = function (key, increment, value, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
value = helpers.valueToString(value);
increment = parseFloat(increment);
module.transaction(function (tx, done) {
async.waterfall([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'zset'),
async.apply(tx.client.query.bind(tx.client), {
name: 'sortedSetIncrBy',
text: `
INSERT INTO "legacy_zset" ("_key", "value", "score")
VALUES ($1::TEXT, $2::TEXT, $3::NUMERIC)
ON CONFLICT ("_key", "value")
DO UPDATE SET "score" = "legacy_zset"."score" + $3::NUMERIC
RETURNING "score" s`,
values: [key, value, increment],
}),
function (res, next) {
next(null, parseFloat(res.rows[0].s));
},
], done);
}, callback);
};
module.getSortedSetRangeByLex = function (key, min, max, start, count, callback) {
sortedSetLex(key, min, max, 1, start, count, callback);
};
module.getSortedSetRevRangeByLex = function (key, max, min, start, count, callback) {
sortedSetLex(key, min, max, -1, start, count, callback);
};
module.sortedSetLexCount = function (key, min, max, callback) {
var q = buildLexQuery(key, min, max);
query({
name: 'sortedSetLexCount' + q.suffix,
text: `
SELECT COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE ` + q.where,
values: q.values,
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, parseInt(res.rows[0].c, 10));
});
};
function sortedSetLex(key, min, max, sort, start, count, callback) {
if (!callback) {
callback = start;
start = 0;
count = 0;
}
var q = buildLexQuery(key, min, max);
q.values.push(start);
q.values.push(count <= 0 ? null : count);
query({
name: 'sortedSetLex' + (sort > 0 ? 'Asc' : 'Desc') + q.suffix,
text: `
SELECT z."value" v
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE ` + q.where + `
ORDER BY z."value" ` + (sort > 0 ? 'ASC' : 'DESC') + `
LIMIT $` + q.values.length + `::INTEGER
OFFSET $` + (q.values.length - 1) + `::INTEGER`,
values: q.values,
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, res.rows.map(function (r) {
return r.v;
}));
});
}
module.sortedSetRemoveRangeByLex = function (key, min, max, callback) {
callback = callback || helpers.noop;
var q = buildLexQuery(key, min, max);
query({
name: 'sortedSetRemoveRangeByLex' + q.suffix,
text: `
DELETE FROM "legacy_zset" z
USING "legacy_object_live" o
WHERE o."_key" = z."_key"
AND o."type" = z."type"
AND ` + q.where,
values: q.values,
}, function (err) {
callback(err);
});
};
function buildLexQuery(key, min, max) {
var q = {
suffix: '',
where: `o."_key" = $1::TEXT`,
values: [key],
};
if (min !== '-') {
if (min.match(/^\(/)) {
q.values.push(min.substr(1));
q.suffix += 'GT';
q.where += ` AND z."value" > $` + q.values.length + `::TEXT`;
} else if (min.match(/^\[/)) {
q.values.push(min.substr(1));
q.suffix += 'GE';
q.where += ` AND z."value" >= $` + q.values.length + `::TEXT`;
} else {
q.values.push(min);
q.suffix += 'GE';
q.where += ` AND z."value" >= $` + q.values.length + `::TEXT`;
}
}
if (max !== '+') {
if (max.match(/^\(/)) {
q.values.push(max.substr(1));
q.suffix += 'LT';
q.where += ` AND z."value" < $` + q.values.length + `::TEXT`;
} else if (max.match(/^\[/)) {
q.values.push(max.substr(1));
q.suffix += 'LE';
q.where += ` AND z."value" <= $` + q.values.length + `::TEXT`;
} else {
q.values.push(max);
q.suffix += 'LE';
q.where += ` AND z."value" <= $` + q.values.length + `::TEXT`;
}
}
return q;
}
module.processSortedSet = function (setKey, process, options, callback) {
var Cursor = require('pg-cursor');
db.connect(function (err, client, done) {
if (err) {
return callback(err);
}
var batchSize = (options || {}).batch || 100;
var query = client.query(new Cursor(`
SELECT z."value" v
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = $1::TEXT
ORDER BY z."score" ASC, z."value" ASC`, [setKey]));
async.doUntil(function (next) {
query.read(batchSize, function (err, rows) {
if (err) {
return next(err);
}
if (!rows.length) {
return next(null, true);
}
rows = rows.map(function (row) {
return row.v;
});
process(rows, function (err) {
if (err) {
return query.close(function () {
next(err);
});
}
if (options.interval) {
setTimeout(next, options.interval);
} else {
next();
}
});
});
}, function (stop) {
return stop;
}, function (err) {
done();
callback(err);
});
});
};
};

108
src/database/postgres/sorted/add.js Normal file
View File

@@ -0,0 +1,108 @@
'use strict';
var async = require('async');
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
module.sortedSetAdd = function (key, score, value, callback) {
callback = callback || helpers.noop;
if (!key) {
return callback();
}
if (Array.isArray(score) && Array.isArray(value)) {
return sortedSetAddBulk(key, score, value, callback);
}
value = helpers.valueToString(value);
score = parseFloat(score);
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'zset'),
async.apply(query, {
name: 'sortedSetAdd',
text: `
INSERT INTO "legacy_zset" ("_key", "value", "score")
VALUES ($1::TEXT, $2::TEXT, $3::NUMERIC)
ON CONFLICT ("_key", "value")
DO UPDATE SET "score" = $3::NUMERIC`,
values: [key, value, score],
}),
], function (err) {
done(err);
});
}, callback);
};
function sortedSetAddBulk(key, scores, values, callback) {
if (!scores.length || !values.length) {
return callback();
}
if (scores.length !== values.length) {
return callback(new Error('[[error:invalid-data]]'));
}
values = values.map(helpers.valueToString);
scores = scores.map(function (score) {
return parseFloat(score);
});
helpers.removeDuplicateValues(values, scores);
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectType, tx.client, key, 'zset'),
async.apply(query, {
name: 'sortedSetAddBulk',
text: `
INSERT INTO "legacy_zset" ("_key", "value", "score")
SELECT $1::TEXT, v, s
FROM UNNEST($2::TEXT[], $3::NUMERIC[]) vs(v, s)
ON CONFLICT ("_key", "value")
DO UPDATE SET "score" = EXCLUDED."score"`,
values: [key, values, scores],
}),
], function (err) {
done(err);
});
}, callback);
}
module.sortedSetsAdd = function (keys, score, value, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
value = helpers.valueToString(value);
score = parseFloat(score);
module.transaction(function (tx, done) {
var query = tx.client.query.bind(tx.client);
async.series([
async.apply(helpers.ensureLegacyObjectsType, tx.client, keys, 'zset'),
async.apply(query, {
name: 'sortedSetsAdd',
text: `
INSERT INTO "legacy_zset" ("_key", "value", "score")
SELECT k, $2::TEXT, $3::NUMERIC
FROM UNNEST($1::TEXT[]) k
ON CONFLICT ("_key", "value")
DO UPDATE SET "score" = $3::NUMERIC`,
values: [keys, value, score],
}),
], function (err) {
done(err);
});
}, callback);
};
};

105
src/database/postgres/sorted/intersect.js Normal file
View File

@@ -0,0 +1,105 @@
'use strict';
module.exports = function (db, module) {
module.sortedSetIntersectCard = function (keys, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, 0);
}
db.query({
name: 'sortedSetIntersectCard',
text: `
WITH A AS (SELECT z."value" v,
COUNT(*) c
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])
GROUP BY z."value")
SELECT COUNT(*) c
FROM A
WHERE A.c = array_length($1::TEXT[], 1)`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, parseInt(res.rows[0].c, 10));
});
};
module.getSortedSetIntersect = function (params, callback) {
params.sort = 1;
getSortedSetIntersect(params, callback);
};
module.getSortedSetRevIntersect = function (params, callback) {
params.sort = -1;
getSortedSetIntersect(params, callback);
};
function getSortedSetIntersect(params, callback) {
var sets = params.sets;
var start = params.hasOwnProperty('start') ? params.start : 0;
var stop = params.hasOwnProperty('stop') ? params.stop : -1;
var weights = params.weights || [];
var aggregate = params.aggregate || 'SUM';
if (sets.length < weights.length) {
weights = weights.slice(0, sets.length);
}
while (sets.length > weights.length) {
weights.push(1);
}
var limit = stop - start + 1;
if (limit <= 0) {
limit = null;
}
db.query({
name: 'getSortedSetIntersect' + aggregate + (params.sort > 0 ? 'Asc' : 'Desc') + 'WithScores',
text: `
WITH A AS (SELECT z."value",
` + aggregate + `(z."score" * k."weight") "score",
COUNT(*) c
FROM UNNEST($1::TEXT[], $2::NUMERIC[]) k("_key", "weight")
INNER JOIN "legacy_object_live" o
ON o."_key" = k."_key"
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
GROUP BY z."value")
SELECT A."value",
A."score"
FROM A
WHERE c = array_length($1::TEXT[], 1)
ORDER BY A."score" ` + (params.sort > 0 ? 'ASC' : 'DESC') + `
LIMIT $4::INTEGER
OFFSET $3::INTEGER`,
values: [sets, weights, start, limit],
}, function (err, res) {
if (err) {
return callback(err);
}
if (params.withScores) {
res.rows = res.rows.map(function (r) {
return {
value: r.value,
score: parseFloat(r.score),
};
});
} else {
res.rows = res.rows.map(function (r) {
return r.value;
});
}
callback(null, res.rows);
});
}
};

83
src/database/postgres/sorted/remove.js Normal file
View File

@@ -0,0 +1,83 @@
'use strict';
module.exports = function (db, module) {
var helpers = module.helpers.postgres;
module.sortedSetRemove = function (key, value, callback) {
function done(err) {
if (callback) {
callback(err);
}
}
if (!key) {
return done();
}
if (!Array.isArray(key)) {
key = [key];
}
if (!Array.isArray(value)) {
value = [value];
}
value = value.map(helpers.valueToString);
db.query({
name: 'sortedSetRemove',
text: `
DELETE FROM "legacy_zset"
WHERE "_key" = ANY($1::TEXT[])
AND "value" = ANY($2::TEXT[])`,
values: [key, value],
}, done);
};
module.sortedSetsRemove = function (keys, value, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
value = helpers.valueToString(value);
db.query({
name: 'sortedSetsRemove',
text: `
DELETE FROM "legacy_zset"
WHERE "_key" = ANY($1::TEXT[])
AND "value" = $2::TEXT`,
values: [keys, value],
}, function (err) {
callback(err);
});
};
module.sortedSetsRemoveRangeByScore = function (keys, min, max, callback) {
callback = callback || helpers.noop;
if (!Array.isArray(keys) || !keys.length) {
return callback();
}
if (min === '-inf') {
min = null;
}
if (max === '+inf') {
max = null;
}
db.query({
name: 'sortedSetsRemoveRangeByScore',
text: `
DELETE FROM "legacy_zset"
WHERE "_key" = ANY($1::TEXT[])
AND ("score" >= $2::NUMERIC OR $2::NUMERIC IS NULL)
AND ("score" <= $3::NUMERIC OR $3::NUMERIC IS NULL)`,
values: [keys, min, max],
}, function (err) {
callback(err);
});
};
};

97
src/database/postgres/sorted/union.js Normal file
View File

@@ -0,0 +1,97 @@
'use strict';
module.exports = function (db, module) {
module.sortedSetUnionCard = function (keys, callback) {
if (!Array.isArray(keys) || !keys.length) {
return callback(null, 0);
}
db.query({
name: 'sortedSetUnionCard',
text: `
SELECT COUNT(DISTINCT z."value") c
FROM "legacy_object_live" o
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
WHERE o."_key" = ANY($1::TEXT[])`,
values: [keys],
}, function (err, res) {
if (err) {
return callback(err);
}
callback(null, parseInt(res.rows[0].c, 10));
});
};
module.getSortedSetUnion = function (params, callback) {
params.sort = 1;
getSortedSetUnion(params, callback);
};
module.getSortedSetRevUnion = function (params, callback) {
params.sort = -1;
getSortedSetUnion(params, callback);
};
function getSortedSetUnion(params, callback) {
var sets = params.sets;
var start = params.hasOwnProperty('start') ? params.start : 0;
var stop = params.hasOwnProperty('stop') ? params.stop : -1;
var weights = params.weights || [];
var aggregate = params.aggregate || 'SUM';
if (sets.length < weights.length) {
weights = weights.slice(0, sets.length);
}
while (sets.length > weights.length) {
weights.push(1);
}
var limit = stop - start + 1;
if (limit <= 0) {
limit = null;
}
db.query({
name: 'getSortedSetUnion' + aggregate + (params.sort > 0 ? 'Asc' : 'Desc') + 'WithScores',
text: `
WITH A AS (SELECT z."value",
` + aggregate + `(z."score" * k."weight") "score"
FROM UNNEST($1::TEXT[], $2::NUMERIC[]) k("_key", "weight")
INNER JOIN "legacy_object_live" o
ON o."_key" = k."_key"
INNER JOIN "legacy_zset" z
ON o."_key" = z."_key"
AND o."type" = z."type"
GROUP BY z."value")
SELECT A."value",
A."score"
FROM A
ORDER BY A."score" ` + (params.sort > 0 ? 'ASC' : 'DESC') + `
LIMIT $4::INTEGER
OFFSET $3::INTEGER`,
values: [sets, weights, start, limit],
}, function (err, res) {
if (err) {
return callback(err);
}
if (params.withScores) {
res.rows = res.rows.map(function (r) {
return {
value: r.value,
score: parseFloat(r.score),
};
});
} else {
res.rows = res.rows.map(function (r) {
return r.value;
});
}
callback(null, res.rows);
});
}
};

50
src/database/postgres/transaction.js Normal file
View File

@@ -0,0 +1,50 @@
'use strict';
module.exports = function (db, dbNamespace, module) {
module.transaction = function (perform, callback) {
if (dbNamespace.active && dbNamespace.get('db')) {
var client = dbNamespace.get('db');
return client.query(`SAVEPOINT nodebb_subtx`, function (err) {
if (err) {
return callback(err);
}
perform(module, function (err) {
var args = Array.prototype.slice.call(arguments, 1);
client.query(err ? `ROLLBACK TO SAVEPOINT nodebb_subtx` : `RELEASE SAVEPOINT nodebb_subtx`, function (err1) {
callback.apply(this, [err || err1].concat(args));
});
});
});
}
db.connect(function (err, client, done) {
if (err) {
return callback(err);
}
dbNamespace.run(function () {
dbNamespace.set('db', client);
client.query(`BEGIN`, function (err) {
if (err) {
done();
dbNamespace.set('db', null);
return callback(err);
}
perform(module, function (err) {
var args = Array.prototype.slice.call(arguments, 1);
client.query(err ? `ROLLBACK` : `COMMIT`, function (err1) {
done();
dbNamespace.set('db', null);
callback.apply(this, [err || err1].concat(args));
});
});
});
});
});
};
};