diff options
| -rw-r--r-- | lib/db.js | 39 | ||||
| -rw-r--r-- | lib/lightning.js | 62 | ||||
| -rw-r--r-- | lib/nostr.js | 6 | ||||
| -rw-r--r-- | lib/sendInternal.js | 93 | ||||
| -rw-r--r-- | lib/whitelist.js | 2 | ||||
| -rw-r--r-- | package.json | 2 | ||||
| -rw-r--r-- | pnpm-lock.yaml | 65 | ||||
| -rw-r--r-- | routes/invoices.js | 2 | ||||
| -rw-r--r-- | routes/nostr.js | 4 | ||||
| -rw-r--r-- | routes/payments.js | 89 | ||||
| -rw-r--r-- | routes/requests.js | 4 | ||||
| -rw-r--r-- | routes/send.js | 15 | ||||
| -rw-r--r-- | routes/users.js | 1 |
13 files changed, 129 insertions, 255 deletions
@@ -1,9 +1,40 @@ -import { Redis } from "ioredis"; -import Redlock from "redlock"; +import { sleep } from "$lib/utils"; +import { createClient } from "redis"; -export const db = new Redis("redis://db", { enableAutoPipelining: true }); +export const db = createClient({ url: "redis://db" }); +await db.connect(); export default db; export let g = async k => JSON.parse(await db.get(k)); export let s = (k, v) => db.set(k, JSON.stringify(v)); -export let t = (f) => new Redlock([db]).using([db], 5000, f); + +let retries = {}; +export let t = async (k, f) => { + let result; + try { + result = await db.executeIsolated(async db => { + await db.watch(k); + let v = await db.get(k); + await db + .multi() + .set(k, await f(v)) + .exec(); + }); + } catch (err) { + if (!err.message.includes("watch")) throw err; + + let r = retries[k] || 0; + retries[k] = r + 1; + + if (r < 10) { + await sleep(100); + await t(k, f); + } else { + delete retries[k]; + throw new Error("unable to obtain lock"); + } + } + + delete retries[k]; + return result; +}; diff --git a/lib/lightning.js b/lib/lightning.js index 57667be2..86dfdb2b 100644 --- a/lib/lightning.js +++ b/lib/lightning.js @@ -5,7 +5,6 @@ import { callWebhook } from "$lib/webhooks"; import ln from "$lib/ln"; import { l, err, warn } from "$lib/logging"; import { g, s, t, db } from "$lib/db"; -import Redlock from "redlock"; export async function listenForLightning() { let { @@ -18,6 +17,7 @@ export async function listenForLightning() { } = await ln.waitanyinvoice(await g("pay_index")); await s("pay_index", pay_index); + setTimeout(listenForLightning); let received = Math.round(msatoshi_received / 1000); @@ -52,65 +52,11 @@ export async function listenForLightning() { invoice.status = "paid"; invoice.received += received; - const redlock = new Redlock([db]); - await s(`invoice:${hash}`, invoice); - console.log("BAL", await g(`balance:${uid}`)); - setTimeout(async () => { - let lock = await redlock.acquire(["b"], 5000); - try { - let bal = await g(`balance:${uid}`); - console.log("b bal", bal); - await new Promise(r => setTimeout(r, 50)); - - await s(`balance:${uid}`, bal + 100); - console.log("set b bal", bal + 100); - } finally { - await lock.release(); - } - // try { - // console.log("tweaking balance"); - // await s(`balance:${uid}`, 0); - // await new Promise(r => setTimeout(r, 5)); - // console.log("mwahaha", await g(`balance:${uid}`)); - // await new Promise(r => setTimeout(r, 600)); - // console.log("hmm", await g(`balance:${uid}`)); - // } catch (e) { - // console.log(e); - // } - }, 50); - - await new Promise(r => setTimeout(r, 5)); - - let lock = await redlock.acquire(["a"], 5000); - try { - let bal = await g(`balance:${uid}`); - console.log("a bal", bal, received); - await new Promise(r => setTimeout(r, 500)); - - await s(`balance:${uid}`, bal + received); - console.log("set a bal", bal + received); - } finally { - await lock.release(); - } - - // await t(async signal => { - // console.log("acquiring lock"); - // let bal = await g(`balance:${uid}`); - // console.log("bal", bal); - // await new Promise(r => setTimeout(r, 500)); - // - // if (signal.aborted) throw signal.error; - // - // await s(`balance:${uid}`, bal + received); - // console.log("set bal", bal + received); - // }); - console.log("done", await g(`balance:${uid}`)); - + await db.incrBy(`balance:${uid}`, received); payment.invoice = invoice; - await s(`payment:${hash}`, payment); - await db.lpush(`${uid}:payments`, hash); + await db.lPush(`${uid}:payments`, hash); callWebhook(invoice, payment); @@ -122,6 +68,4 @@ export async function listenForLightning() { console.log(e); err("problem receiving lightning payment", e.message); } - - setTimeout(listenForLightning); } diff --git a/lib/nostr.js b/lib/nostr.js index c8accf85..23ea937b 100644 --- a/lib/nostr.js +++ b/lib/nostr.js @@ -63,17 +63,17 @@ export let fillPool = () => { await s(`${pubkey}:follows`, JSON.stringify(ev.tags)); for (let f of ev.tags) { let [_, followPubkey] = f; - await db.sadd(`${followPubkey}:followers`, pubkey); + await db.sAdd(`${followPubkey}:followers`, pubkey); } coinos.send(ev); } else if (sub.includes("followers")) { let followed = sub.split(":")[0]; let { pubkey } = ev; - await db.sadd(`${followed}:followers`, pubkey); + await db.sAdd(`${followed}:followers`, pubkey); coinos.send(ev); } else if (sub.includes("notes")) { let pubkey = sub.split(":")[0]; - await db.sadd(pubkey, ev.id); + await db.sAdd(pubkey, ev.id); await s(`ev:${ev.id}`, JSON.stringify(ev)); } } catch (e) { diff --git a/lib/sendInternal.js b/lib/sendInternal.js deleted file mode 100644 index 6054f753..00000000 --- a/lib/sendInternal.js +++ /dev/null @@ -1,93 +0,0 @@ -import config from "$config"; -import { emit } from "$lib/sockets"; -import store from "$lib/store"; -import { callWebhook } from "$lib/webhooks"; -import { v4 } from "uuid"; -import { notify } from "$lib/notifications"; -import { l, err } from "$lib/logging"; -import { fail } from "$lib/utils"; -import ln from "$lib/ln"; -import { g, s, db } from "$lib/db"; - -const { HOSTNAME: hostname } = process.env; - -export default async ({ amount, hash, memo, tip }, user) => { - amount = parseInt(amount); - - if (!amount || amount < 0) fail("Amount must be greater than zero"); - - let fee = 0; - let invoice = await g(`invoice:${hash}`); - - let t = async () => { - await db.watch(`user:${user.id}`); - user = await g(`user:${user.id}`); - if (user.balance < amount) fail("Insufficient funds"); - user.balance -= amount; - let m = await db.multi(); - await s(`user:${user.id}`, user); - - let p1 = { - hash, - amount: -amount, - memo, - uid: user.id, - rate: store.rates[user.currency], - currency: user.currency, - type: "internal", - created: Date.now() - }; - - await s(`payment:${hash}`, p1); - await db.lpush(`${user.id}:payments`, hash); - - l("sent internal", user.username, -p1.amount); - emit(user.id, "payment", p1); - - let { uid } = invoice; - if (uid !== user.id) { - let recipient = await g(`user:${uid}`); - - let u = async () => { - await db.watch(`user:${uid}`); - recipient = await g(`user:${uid}`); - recipient.balance += amount; - let m = await db.multi(); - await s(`user:${uid}`, recipient); - if (!(await m.exec())) u(); - }; - await u(); - - let p2 = { - hash, - amount, - uid: recipient.id, - rate: store.rates[recipient.currency], - currency: recipient.currency, - memo, - type: "internal", - with_id: user.id, - created: Date.now() - }; - - let { tip } = invoice; - if (tip) { - p2.tip = tip; - p2.amount -= tip; - } - - invoice.received += amount; - await s(`invoice:${hash}`, invoice); - await s(`payment:${id}`, p2); - await db.lpush(`${recipient.id}:payments`, id); - - l("received internal", recipient.username, amount); - emit(recipient.username, "payment", p2); - notify(recipient, `Received ${amount} sats`); - callWebhook(invoice, p2); - } - - if (!(await m.exec())) t(); - }; - await t(); -}; diff --git a/lib/whitelist.js b/lib/whitelist.js index 49cc9af3..77243855 100644 --- a/lib/whitelist.js +++ b/lib/whitelist.js @@ -22,10 +22,8 @@ export default [ "pubkey", "salt", "seed", - "symbol", "theme", "twofa", - "unit", "username", "verified" ]; diff --git a/package.json b/package.json index 6f1c45f9..2c1b7258 100644 --- a/package.json +++ b/package.json @@ -18,7 +18,6 @@ "fastify": "^4.5.3", "file-type": "^18.0.0", "got": "^12.5.3", - "ioredis": "^5.2.5", "jsonwebtoken": "^8.5.1", "lnurl": "^0.24.2", "nodemon": "^2.0.7", @@ -33,7 +32,6 @@ "query-string": "^6.13.1", "readable-stream-clone": "^0.0.7", "redis": "^4.3.1", - "redlock": "5.0.0-beta.2", "sharp": "^0.31.1", "uuid": "8", "web-push": "^3.4.4", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4c22c13f..5c9f09dd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -12,7 +12,6 @@ specifiers: fastify: ^4.5.3 file-type: ^18.0.0 got: ^12.5.3 - ioredis: ^5.2.5 jsonwebtoken: ^8.5.1 lnurl: ^0.24.2 nodemon: ^2.0.7 @@ -28,7 +27,6 @@ specifiers: query-string: ^6.13.1 readable-stream-clone: ^0.0.7 redis: ^4.3.1 - redlock: 5.0.0-beta.2 sharp: ^0.31.1 uuid: '8' web-push: ^3.4.4 @@ -46,7 +44,6 @@ dependencies: fastify: 4.8.1 file-type: 18.0.0 got: 12.5.3 - ioredis: 5.2.5 jsonwebtoken: 8.5.1 lnurl: 0.24.2 nodemon: 2.0.20 @@ -61,7 +58,6 @@ dependencies: query-string: 6.14.1 readable-stream-clone: 0.0.7 redis: 4.3.1 - redlock: 5.0.0-beta.2 sharp: 0.31.1 uuid: 8.3.2 web-push: 3.5.0 @@ -179,10 +175,6 @@ packages: - utf-8-validate dev: false - /@ioredis/commands/1.2.0: - resolution: {integrity: sha512-Sx1pU8EM64o2BrqNpEO1CNLtKQwyhuXuqyfH7oGKCk+1a33d2r5saW8zNwm3j6BTExtjrv2BxTgzzkMwts6vGg==} - dev: false - /@mapbox/node-pre-gyp/1.0.10: resolution: {integrity: sha512-4ySo4CjzStuprMwk35H5pPbkymjv1SF3jGLj6rAHp/xT/RF7TL7bd9CTm1xDY49K2qF7jmR/g7k+SkLETP6opA==} hasBin: true @@ -864,11 +856,6 @@ packages: resolution: {integrity: sha512-bd2L678uiWATM6m5Z1VzNCErI3jiGzt6HGY8OVICs40JQq/HALfbyNJmp0UDakEY4pMMaN0Ly5om/B1VI/+xfQ==} dev: false - /denque/2.1.0: - resolution: {integrity: sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==} - engines: {node: '>=0.10'} - dev: false - /depd/2.0.0: resolution: {integrity: sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw==} engines: {node: '>= 0.8'} @@ -1403,23 +1390,6 @@ packages: resolution: {integrity: sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew==} dev: false - /ioredis/5.2.5: - resolution: {integrity: sha512-7HKo/ClM2DGLRXdFq8ruS3Uuadensz4A76wPOU0adqlOqd1qkhoLPDaBhmVhUhNGpB+J65/bhLmNB8DDY99HJQ==} - engines: {node: '>=12.22.0'} - dependencies: - '@ioredis/commands': 1.2.0 - cluster-key-slot: 1.1.0 - debug: 4.3.4 - denque: 2.1.0 - lodash.defaults: 4.2.0 - lodash.isarguments: 3.1.0 - redis-errors: 1.2.0 - redis-parser: 3.0.0 - standard-as-callback: 2.1.0 - transitivePeerDependencies: - - supports-color - dev: false - /ip/2.0.0: resolution: {integrity: sha512-WKa+XuLG1A1R0UWhl2+1XQSi+fZWMsYKffMZTTYsiZaUD8k2yDAj5atimTUD2TZkyCkNEeYE5NhFZmupOGtjYQ==} dev: false @@ -1574,18 +1544,10 @@ packages: - supports-color dev: false - /lodash.defaults/4.2.0: - resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==} - dev: false - /lodash.includes/4.3.0: resolution: {integrity: sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==} dev: false - /lodash.isarguments/3.1.0: - resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==} - dev: false - /lodash.isboolean/3.0.3: resolution: {integrity: sha512-Bz5mupy2SVbPHURB98VAcw+aHh4vRV5IPNhILUCsOzRmsTmSQ17jIuqopAentWoehktxGd9e/hbIXq980/1QJg==} dev: false @@ -1775,10 +1737,6 @@ packages: semver: 7.3.8 dev: false - /node-abort-controller/3.0.1: - resolution: {integrity: sha512-/ujIVxthRs+7q6hsdjHMaj8hRG9NuWmwrz+JdRwZ14jdFoKSkm+vDsCbF9PLpnSqjaWQJuTmVtcWHNLr+vrOFw==} - dev: false - /node-addon-api/2.0.2: resolution: {integrity: sha512-Ntyt4AIXyaLIuMHF6IOoTakB3K+RWxwtsHNRxllEoA6vPwP9o4866g6YWDLUdnucilZhmkxiHwHr11gAENw+QA==} dev: false @@ -2171,18 +2129,6 @@ packages: engines: {node: '>= 12.13.0'} dev: false - /redis-errors/1.2.0: - resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==} - engines: {node: '>=4'} - dev: false - - /redis-parser/3.0.0: - resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==} - engines: {node: '>=4'} - dependencies: - redis-errors: 1.2.0 - dev: false - /redis/4.3.1: resolution: {integrity: sha512-cM7yFU5CA6zyCF7N/+SSTcSJQSRMEKN0k0Whhu6J7n9mmXRoXugfWDBo5iOzGwABmsWKSwGPTU5J4Bxbl+0mrA==} dependencies: @@ -2194,13 +2140,6 @@ packages: '@redis/time-series': 1.0.3_@[email protected] dev: false - /redlock/5.0.0-beta.2: - resolution: {integrity: sha512-2RDWXg5jgRptDrB1w9O/JgSZC0j7y4SlaXnor93H/UJm/QyDiFgBKNtrh0TI6oCXqYSaSoXxFh6Sd3VtYfhRXw==} - engines: {node: '>=12'} - dependencies: - node-abort-controller: 3.0.1 - dev: false - /require-from-string/2.0.2: resolution: {integrity: sha512-Xf0nWe6RseziFMu+Ap9biiUbmplq6S9/p+7w7YXP/JBHhrUDDUhwa+vANyubuqfZWTveU//DYVGsDG7RKL/vEw==} engines: {node: '>=0.10.0'} @@ -2464,10 +2403,6 @@ packages: engines: {node: '>= 10.x'} dev: false - /standard-as-callback/2.1.0: - resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==} - dev: false - /statuses/2.0.1: resolution: {integrity: sha512-RwNA9Z/7PrK06rYLIzFMlaF+l73iwpzsqRIFgbMLbTcLD6cOao82TaWefPXQvB2fOC4AjuYSEndS7N/mTCbkdQ==} engines: {node: '>= 0.8'} diff --git a/routes/invoices.js b/routes/invoices.js index 3e43c57c..acd9457e 100644 --- a/routes/invoices.js +++ b/routes/invoices.js @@ -58,7 +58,7 @@ export default { ); await s(`invoice:${hash}`, invoice); - await db.lpush(`${uid}:invoices`, hash); + await db.lPush(`${uid}:invoices`, hash); if (request_id) { let request = await g(`request:${request_id}`); diff --git a/routes/nostr.js b/routes/nostr.js index bac04810..2900e949 100644 --- a/routes/nostr.js +++ b/routes/nostr.js @@ -41,7 +41,7 @@ export default { await s(`user:${pubkey}`, user); - let ids = await db.smembers(pubkey); + let ids = await db.sMembers(pubkey); let events = ids.length ? (await db.mget(ids.map(k => "ev:" + k))).map(JSON.parse) @@ -101,7 +101,7 @@ export default { async followers({ params: { pubkey } }, res) { try { let pubkeys = [ - ...new Set([...(await db.smembers(`${pubkey}:followers`))]) + ...new Set([...(await db.sMembers(`${pubkey}:followers`))]) ]; let followers = []; diff --git a/routes/payments.js b/routes/payments.js index 22de79c3..27c8c5de 100644 --- a/routes/payments.js +++ b/routes/payments.js @@ -1,16 +1,91 @@ +import config from "$config"; +import store from "$lib/store"; import { emit } from "$lib/sockets"; -import { db, g, s } from "$lib/db"; +import { v4 } from "uuid"; +import { notify } from "$lib/notifications"; +import { db, g, s, t } from "$lib/db"; import { l, err, warn } from "$lib/logging"; import { fail } from "$lib/utils"; import ln from "$lib/ln"; -import sendInternal from "$lib/sendInternal"; import { requirePin } from "$lib/auth"; +import { callWebhook } from "$lib/webhooks"; + +const { HOSTNAME: hostname } = process.env; export default { - async send(req, res) { - await requirePin(req); - let payment = await sendInternal(req.body, req.user); - res.send(payment); + async send({ body, user }, res) { + let { amount, hash, memo, tip } = body; + await requirePin({ body, user }); + if (!hash) fail("invoice hash required"); + + let { id: uid } = user; + amount = parseInt(amount); + + if (!amount || amount < 0) fail("Amount must be greater than zero"); + + await t(`balance:${uid}`, async (balance) => { + await new Promise(r => setTimeout(r, 100)); + if (balance < amount) fail("Insufficient funds"); + return balance - amount; + }); + + let fee = 0; + let invoice = await g(`invoice:${hash}`); + user = await g(`user:${uid}`); + + await s(`user:${uid}`, user); + + let p1 = { + hash, + amount: -amount, + memo, + uid, + rate: store.rates[user.currency], + currency: user.currency, + type: "internal", + created: Date.now() + }; + + await s(`payment:${hash}`, p1); + await db.lPush(`${uid}:payments`, hash); + + l("sent internal", user.username, -p1.amount); + emit(user.id, "payment", p1); + + ({ uid } = invoice); + if (uid !== user.id) { + let recipient = await g(`user:${uid}`); + + let p2 = { + hash, + amount, + uid: recipient.id, + rate: store.rates[recipient.currency], + currency: recipient.currency, + memo, + type: "internal", + with_id: user.id, + created: Date.now() + }; + + let { tip } = invoice; + if (tip) { + p2.tip = tip; + p2.amount -= tip; + } + + invoice.received += amount; + await s(`invoice:${hash}`, invoice); + await s(`payment:${hash}`, p2); + await db.lPush(`${recipient.id}:payments`, hash); + + l("received internal", recipient.username, amount); + emit(recipient.username, "payment", p2); + notify(recipient, `Received ${amount} sats`); + callWebhook(invoice, p2); + } + + res.send(p1); }, async list({ user: { id }, query: { start, end, limit, offset } }, res) { @@ -21,7 +96,7 @@ export default { // if (start) where.createdAt[Op.gte] = new Date(parseInt(start)); // if (end) where.createdAt[Op.lte] = new Date(parseInt(end)); - let payments = (await db.lrange(`${id}:payments`, 0, -1)) || []; + let payments = (await db.lRange(`${id}:payments`, 0, -1)) || []; payments = await Promise.all(payments.map(hash => g(`payment:${hash}`))); payments = payments.filter(p => p); res.send({ payments, total: payments.length }); diff --git a/routes/requests.js b/routes/requests.js index 8db373dd..79af1fb6 100644 --- a/routes/requests.js +++ b/routes/requests.js @@ -15,8 +15,8 @@ export default { try { const day = new Date(new Date().setDate(new Date().getDate() - 1)); - let invoices = await db.lrange(`${id}:invoices`, 0, -1); - let requests = await db.lrange(`${id}:requests`, 0, -1); + let invoices = await db.lRange(`${id}:invoices`, 0, -1); + let requests = await db.lRange(`${id}:requests`, 0, -1); res.send({ invoices, requests }); } catch (e) { diff --git a/routes/send.js b/routes/send.js deleted file mode 100644 index 4b954616..00000000 --- a/routes/send.js +++ /dev/null @@ -1,15 +0,0 @@ -import { err } from "$lib/logging"; -import sendInternal from "$lib/sendInternal"; -import { requirePin } from "$lib/auth"; - -export default async (req, res, next) => { - try { - await requirePin(req); - let payment = await sendInternal(req.body, req.user); - res.send(payment); - } catch (e) { - console.log(e); - err("problem sending internal payment", e.message); - return res.code(500).send(e.message); - } -}; diff --git a/routes/users.js b/routes/users.js index 49aeebd7..57a423b8 100644 --- a/routes/users.js +++ b/routes/users.js @@ -19,6 +19,7 @@ export default { async me({ user }, res) { try { user.balance = await g(`balance:${user.id}`); + console.log("USER BAL", user.id, user.balance) res.send(pick(user, whitelist)); } catch (e) { console.log("problem fetching user", e); |