PixelPiggy/CS_float
0
1const { Pool } = require('pg'),2 utils = require('./utils'),3 winston = require('winston');4 5 6function flatten(arr){7 const newArr = [];8 arr.forEach(v => v.forEach(p => newArr.push(p)));9 return newArr10}11 12class Postgres {13 constructor(url, enableBulkInserts) {14 this.pool = new Pool({15 connectionString: url16 });17 18 this.enableBulkInserts = enableBulkInserts || false;19 20 if (enableBulkInserts) {21 this.queuedInserts = [];22 23 setInterval(() => {24 if (this.queuedInserts.length > 0) {25 const copy = [...this.queuedInserts];26 this.queuedInserts = [];27 this.handleBulkInsert(copy);28 }29 }, 1000);30 }31 }32 33 connect() {34 return this.pool.connect().then(() => this.ensureSchema());35 }36 37 /*38 Returns the following properties in a 32 bit integer39 rarity quality origin40 0000000000 00000000 00000000 0000000041 <future> 8 bits 8 bits 8 bits42 */43 static storeProperties(origin, quality, rarity) {44 return origin | (quality << 8) | (rarity << 16);45 }46 47 static extractProperties(prop) {48 return {49 origin: prop & ((1 << 8) - 1),50 quality: (prop >> 8) & ((1 << 8) - 1),51 rarity: (prop >> 16) & ((1 << 8) - 1)52 }53 }54 55 async ensureSchema() {56 await this.pool.query(`CREATE TABLE IF NOT EXISTS items (57 ms bigint NOT NULL,58 a bigint NOT NULL,59 d bigint NOT NULL,60 paintseed smallint NOT NULL,61 paintwear integer NOT NULL,62 defindex smallint NOT NULL,63 paintindex smallint NOT NULL,64 stattrak boolean NOT NULL,65 souvenir boolean NOT NULL,66 props integer NOT NULL,67 stickers jsonb,68 updated timestamp NOT NULL,69 rarity smallint NOT NULL,70 floatid bigint NOT NULL,71 price integer,72 PRIMARY KEY (a)73 );74 75 CREATE TABLE IF NOT EXISTS history (76 floatid bigint NOT NULL,77 a bigint NOT NULL,78 steamid bigint NOT NULL,79 created_at timestamp NOT NULL,80 price integer,81 PRIMARY KEY (floatid, a)82 );83 84 ALTER TABLE items ADD COLUMN IF NOT EXISTS floatid BIGINT;85 ALTER TABLE items ADD COLUMN IF NOT EXISTS price INTEGER;86 ALTER TABLE items ADD COLUMN IF NOT EXISTS listed_price INTEGER;87 ALTER TABLE items ADD COLUMN IF NOT EXISTS keychains JSONB;88 ALTER TABLE history ADD COLUMN IF NOT EXISTS price INTEGER;89 90 -- Float ID is defined as the first asset id we've seen for an item91 92 CREATE OR REPLACE FUNCTION is_steamid(IN val bigint)93 RETURNS boolean94 LANGUAGE 'plpgsql'95 IMMUTABLE96 PARALLEL SAFE97 AS $BODY$BEGIN98 IF val < 76561197960265728 THEN99 RETURN FALSE;100 ELSIF (val >> 56) > 5 THEN101 RETURN FALSE;102 ELSIF ((val >> 32) & ((1 << 20) - 1)) > 32 THEN103 RETURN FALSE;104 END IF;105 106 RETURN TRUE;107 END;$BODY$;108 109 CREATE OR REPLACE FUNCTION extend_history()110 RETURNS TRIGGER111 AS $$112 BEGIN113 -- Handle cases where the floatid isn't there114 IF NEW.floatid IS NULL THEN115 NEW.floatid = OLD.a;116 END IF;117 118 IF NEW.a = OLD.a THEN119 -- Ignore handling, no new item details, updating existing120 RETURN NEW;121 END IF;122 123 IF NEW.a < OLD.a THEN124 -- If we find an older asset id than the current, still want to add it to history if not there125 INSERT INTO history VALUES (NEW.floatid, NEW.a, NEW.ms, NEW.updated, NULL) ON CONFLICT DO NOTHING;126 127 -- Prevent update to this row128 RETURN NULL;129 END IF;130 131 IF (is_steamid(OLD.ms) AND OLD.ms != NEW.ms) OR OLD.price IS NOT NULL THEN132 -- We care about history for inventory changes or market listings that had price data133 INSERT INTO history VALUES (NEW.floatid, OLD.a, OLD.ms, OLD.updated, OLD.price);134 END IF;135 136 -- Reset the price if it is the same, it is possible that the item was sold for the exact same amount in a row137 -- and we clear it here, but that isn't too much of a concern for the application of the data138 -- This ensures that outdated instances contributing to the same db don't conflict state139 IF NEW.price = OLD.price OR NEW.listed_price = OLD.listed_price OR is_steamid(NEW.ms) THEN140 NEW.price = NULL;141 NEW.listed_price = NULL;142 END IF;143 144 RETURN NEW;145 END;146 $$147 LANGUAGE 'plpgsql';148 149 DROP TRIGGER IF EXISTS extend_history_trigger150 ON items;151 152 CREATE TRIGGER extend_history_trigger153 BEFORE UPDATE ON items154 FOR EACH ROW155 EXECUTE PROCEDURE extend_history();156 157 CREATE OR REPLACE FUNCTION ensure_floatid()158 RETURNS TRIGGER159 AS $$160 BEGIN161 IF NEW.floatid IS NULL THEN162 NEW.floatid = NEW.a;163 END IF;164 165 RETURN NEW;166 END;167 $$168 LANGUAGE 'plpgsql';169 170 DROP TRIGGER IF EXISTS ensure_floatid_trigger171 ON items;172 173 CREATE TRIGGER ensure_floatid_trigger174 BEFORE INSERT ON items175 FOR EACH ROW176 EXECUTE PROCEDURE ensure_floatid();177 `);178 179 await this.pool.query(`CREATE INDEX IF NOT EXISTS i_stickers ON items USING gin (stickers jsonb_path_ops) 180 WHERE stickers IS NOT NULL`);181 await this.pool.query(`CREATE INDEX IF NOT EXISTS i_paintwear ON items (paintwear)`);182 await this.pool.query(`CREATE UNIQUE INDEX IF NOT EXISTS i_unique_item ON 183 items (defindex, paintindex, paintwear, paintseed)`);184 await this.pool.query(`CREATE UNIQUE INDEX IF NOT EXISTS i_unique_fid ON items (floatid)`);185 }186 187 async insertItemData(item, price) {188 if (this.enableBulkInserts) {189 this.queuedInserts.push([item, price]);190 } else {191 await this.handleBulkInsert([[item, price]]);192 }193 }194 195 /**196 * Bulk handler to improve insert performance with 300+ rows at once197 * @param data [[item, price]]198 * @returns {Promise<void>}199 */200 async handleBulkInsert(data) {201 const values = [];202 const uniqueItems = new Set();203 204 for (let [item, price] of data) {205 item = Object.assign({}, item);206 207 // Store float as int32 to prevent float rounding errors208 // Postgres doesn't support unsigned types, so we use signed here209 const buf = Buffer.alloc(4);210 buf.writeFloatBE(item.floatvalue, 0);211 item.paintwear = buf.readInt32BE(0);212 213 if (item.floatvalue <= 0 && item.defindex !== 507) {214 // Only insert weapons, naive check215 // Special case for the 0 float Karambit216 continue;217 }218 219 // Postgres doesn't support unsigned 64 bit ints, so we convert them to signed220 item.s = utils.unsigned64ToSigned(item.s).toString();221 item.a = utils.unsigned64ToSigned(item.a).toString();222 item.d = utils.unsigned64ToSigned(item.d).toString();223 item.m = utils.unsigned64ToSigned(item.m).toString();224 225 const stickers = item.stickers.length > 0 ? item.stickers.map((s) => {226 const res = {s: s.slot, i: s.stickerId};227 if (s.wear) {228 res.w = s.wear;229 }230 if (s.rotation) {231 res.r = s.rotation;232 }233 if (s.offset_x) {234 res.x = s.offset_x;235 }236 if (s.offset_y) {237 res.y = s.offset_y;238 }239 return res;240 }) : null;241 242 if (stickers) {243 // Add a property on stickers with duplicates that signifies how many dupes there are244 // Only add this property to one of the dupe stickers in the array245 for (const sticker of stickers) {246 const matching = stickers.filter((s) => s.i === sticker.i);247 if (matching.length > 1 && !matching.find((s) => s.d > 1)) {248 sticker.d = matching.length;249 }250 }251 }252 253 const keychains = item.keychains.length > 0 ? item.keychains.map((s) => {254 const res = {s: s.slot, i: s.sticker_id};255 if (s.wear) {256 res.w = s.wear;257 }258 if (s.scale) {259 res.sc = s.scale;260 }261 if (s.rotation) {262 res.r = s.rotation;263 }264 if (s.tint_id) {265 res.t = s.tint_id;266 }267 if (s.offset_x) {268 res.x = s.offset_x;269 }270 if (s.offset_y) {271 res.y = s.offset_y;272 }273 if (s.offset_z) {274 res.z = s.offset_z;275 }276 if (s.pattern) {277 res.p = s.pattern;278 }279 return res;280 }) : null;281 282 const ms = item.s !== '0' ? item.s : item.m;283 const isStattrak = item.killeatervalue !== null;284 const isSouvenir = item.quality === 12;285 286 const props = Postgres.storeProperties(item.origin, item.quality, item.rarity);287 288 price = price || null;289 290 // Prevent two of the same item from being inserted in the same statement (causes postgres to get angry)291 const key = `${item.defindex}_${item.paintindex}_${item.paintwear}_${item.paintseed}`;292 if (uniqueItems.has(key)) {293 continue;294 } else {295 uniqueItems.add(key);296 }297 298 values.push([ms, item.a, item.d, item.paintseed, item.paintwear, item.defindex, item.paintindex, isStattrak,299 isSouvenir, props, JSON.stringify(stickers), JSON.stringify(keychains), item.rarity, price]);300 }301 302 if (values.length === 0) {303 return;304 }305 306 try {307 const query = Postgres.buildQuery(values.length);308 await this.pool.query(query, flatten(values));309 winston.debug(`Inserted/updated ${values.length} items`)310 } catch (e) {311 winston.warn(e);312 }313 }314 315 static buildQuery(itemCount) {316 const values = [];317 let i = 1;318 319 // Builds binding pattern such as ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb, $12::jsonb, now(), $13, NULL, $14)320 for (let c = 0; c < itemCount; c++) {321 values.push(`($${i++}, $${i++}, $${i++}, $${i++}, $${i++}, $${i++}, $${i++}, $${i++}, $${i++}, $${i++}, $${i++}::jsonb, $${i++}::jsonb, now(), $${i++}, NULL, $${i++})`);322 }323 324 return `INSERT INTO items (ms, a, d, paintseed, paintwear, defindex, paintindex, stattrak, souvenir, props, stickers, keychains, updated, rarity, floatid, price)325 VALUES ${values.join(', ')} ON CONFLICT(defindex, paintindex, paintwear, paintseed) DO UPDATE SET ms=excluded.ms, a=excluded.a, d=excluded.d, stickers=excluded.stickers, keychains=excluded.keychains, updated=now()`;326 }327 328 updateItemPrice(assetId, price) {329 return this.pool.query(`UPDATE items SET price = $1 WHERE a = $2`, [price, assetId]);330 }331 332 async getItemData(links) {333 // Chunking into db calls of 100 each is more performant334 const chunked = utils.chunkArray(links, 100);335 const promises = chunked.map(e => this._getItemData(e));336 const results = await Promise.all(promises);337 338 // Flatten results339 return results.reduce((acc, val) => acc.concat(val), []);340 }341 342 async _getItemData(links) {343 const aValues = links.map(e => utils.unsigned64ToSigned(e.getParams().a));344 345 const result = await this.pool.query(`346 SELECT *, 347 (SELECT Count(*)+1 348 FROM (SELECT * 349 FROM items T 350 WHERE T.paintwear < S.paintwear 351 AND T.defindex = S.defindex 352 AND T.paintindex = S.paintindex 353 AND T.stattrak = S.stattrak 354 AND T.souvenir = S.souvenir 355 ORDER BY T.paintwear 356 LIMIT 1000) as a) AS low_rank,357 (SELECT Count(*)+1358 FROM (SELECT * 359 FROM items J 360 WHERE J.paintwear > S.paintwear 361 AND J.defindex = S.defindex 362 AND J.paintindex = S.paintindex 363 AND J.stattrak = S.stattrak 364 AND J.souvenir = S.souvenir 365 ORDER BY J.paintwear DESC366 LIMIT 1000) as b) AS high_rank 367 FROM items S368 WHERE a= ANY($1::bigint[])`, [aValues]);369 370 return result.rows.map((item) => {371 delete item.updated;372 373 // Correspond to existing API, ensure we can still recreate the full item name374 if (item.stattrak) {375 item.killeatervalue = 0;376 } else {377 item.killeatervalue = null;378 }379 380 item.stickers = item.stickers || [];381 item.stickers = item.stickers.map((s) => {382 return {383 stickerId: s.i,384 slot: s.s,385 wear: s.w,386 rotation: s.r,387 offset_x: s.x,388 offset_y: s.y,389 }390 });391 392 item.keychains = item.keychains || [];393 item.keychains = item.keychains.map((s) => {394 return {395 sticker_id: s.i,396 slot: s.s,397 wear: s.w,398 scale: s.sc,399 rotation: s.r,400 tint_id: s.t,401 offset_x: s.x,402 offset_y: s.y,403 offset_z: s.z,404 pattern: s.p,405 }406 });407 408 item = Object.assign(Postgres.extractProperties(item.props), item);409 410 const buf = Buffer.alloc(4);411 buf.writeInt32BE(item.paintwear, 0);412 item.floatvalue = buf.readFloatBE(0);413 414 item.a = utils.signed64ToUnsigned(item.a).toString();415 item.d = utils.signed64ToUnsigned(item.d).toString();416 item.ms = utils.signed64ToUnsigned(item.ms).toString();417 418 if (utils.isSteamId64(item.ms)){419 item.s = item.ms;420 item.m = '0';421 } else {422 item.m = item.ms;423 item.s = '0';424 }425 426 item.high_rank = parseInt(item.high_rank);427 item.low_rank = parseInt(item.low_rank);428 429 // Delete the rank if above 1000 (we don't get ranking above that)430 if (item.high_rank === 1001) {431 delete item.high_rank;432 }433 434 if (item.low_rank === 1001) {435 delete item.low_rank;436 }437 438 delete item.souvenir;439 delete item.stattrak;440 delete item.paintwear;441 delete item.ms;442 delete item.props;443 delete item.price;444 delete item.listed_price;445 delete item.dupe_count;446 447 return item;448 });449 }450 451 getItemRank(id) {452 return this.pool.query(`SELECT (SELECT Count(*)+1453 FROM (SELECT * 454 FROM items T 455 WHERE T.paintwear < S.paintwear 456 AND T.defindex = S.defindex 457 AND T.paintindex = S.paintindex 458 AND T.stattrak = S.stattrak 459 AND T.souvenir = S.souvenir 460 ORDER BY T.paintwear 461 LIMIT 1000) as a) AS low_rank,462 (SELECT Count(*)+1 463 FROM (SELECT * 464 FROM items J 465 WHERE J.paintwear > S.paintwear 466 AND J.defindex = S.defindex 467 AND J.paintindex = S.paintindex 468 AND J.stattrak = S.stattrak 469 AND J.souvenir = S.souvenir 470 ORDER BY J.paintwear DESC471 LIMIT 1000) as b) AS high_rank 472 FROM items S473 WHERE a=$1`,474 [id]).then((res) => {475 if (res.rows.length > 0) {476 const item = res.rows[0];477 const result = {};478 479 if (item.high_rank != 1001) {480 result.high_rank = parseInt(item.high_rank);481 }482 if (item.low_rank != 1001) {483 result.low_rank = parseInt(item.low_rank);484 }485 486 return result;487 } else {488 return {};489 }490 });491 }492}493 494module.exports = Postgres;495