From 7a84d2555ffabbe29b23cf6e5035b6a4be8b9eb6 Mon Sep 17 00:00:00 2001
From: Adam Soltys <asoltys@gmail.com>
Date: Tue, 9 Jan 2024 23:10:35 -0800
Subject: remove users from archive once migrated

---
 index.js        |   2 ++
 lib/auth.js     |  13 ++------
 lib/db.js       |   4 +--
 lib/logging.js  |  13 ++++++--
 lib/migrate.js  | 100 ++++++++++++++++++++++++++++++++++++--------------------
 lib/payments.js |   6 ++--
 lib/rates.js    |   2 ++
 lib/register.js |  26 +++++++--------
 lib/sockets.js  |  12 +++----
 lib/upload.js   |   3 +-
 lib/utils.js    |  40 +++++++++++------------
 routes/lnurl.js |  63 ++++++++++++++++++-----------------
 routes/users.js |  37 ++++++++++-----------
 13 files changed, 175 insertions(+), 146 deletions(-)

diff --git a/index.js b/index.js
index 6caf3432..f3d1aa13 100644
--- a/index.js
+++ b/index.js
@@ -4,6 +4,7 @@ import { auth, optional } from "$lib/auth";
 import { fillPool } from "$lib/nostr";
 import { listenForLightning } from "$lib/lightning";
 import { getLocations } from "$lib/locations";
+  import { catchUp } from "$lib/payments";
 import { getRates } from "$lib/rates";
 import { sendHeartbeat } from "$lib/sockets";
 import { err } from "$lib/logging";
@@ -24,6 +25,7 @@ try {
   fillPool();
   getLocations();
   getRates();
+  catchUp();
 } catch (e) {
   console.log(e);
 }
diff --git a/lib/auth.js b/lib/auth.js
index cb9a4f0d..1eca2fa5 100644
--- a/lib/auth.js
+++ b/lib/auth.js
@@ -1,6 +1,6 @@
-import migrate from "$lib/migrate";
 import fastifyPassport from "@fastify/passport";
 import { g } from "$lib/db";
+import { fail, getUser } from "$lib/utils";
 import config from "$config";
 import jwt from "passport-jwt";
 
@@ -23,7 +23,7 @@ export const auth = {
 
 export const requirePin = async ({ body, user }) => {
   if (!user || (user.pin && user.pin !== body.pin))
-    throw new Error("Invalid pin");
+    fail("Invalid pin");
 };
 
 export const jwtStrategy = new jwt.Strategy(
@@ -35,14 +35,7 @@ export const jwtStrategy = new jwt.Strategy(
     secretOrKey: config.jwt,
   },
   async (payload, next) => {
-    let user;
-    if (payload.id) {
-      let { id } = payload;
-      user = await g(`user:${id}`);
-      if (typeof user === "string") user = await g(`user:${user}`);
-      if (!user) user = migrate(id);
-    }
-
+    let user = await getUser(payload.id);
     next(null, user);
   },
 );
diff --git a/lib/db.js b/lib/db.js
index 868f18cc..10c54f02 100644
--- a/lib/db.js
+++ b/lib/db.js
@@ -1,5 +1,5 @@
 import config from "$config";
-import { wait, sleep } from "$lib/utils";
+import { fail, wait, sleep } from "$lib/utils";
 import { createClient, defineScript } from "redis";
 import { warn, l, err } from "$lib/logging";
 
@@ -130,7 +130,7 @@ export let t = async (k, f) => {
       await t(k, f);
     } else {
       delete retries[k];
-      throw new Error("unable to obtain lock");
+      fail("unable to obtain lock");
     }
   }
 
diff --git a/lib/logging.js b/lib/logging.js
index 5ad3a0b0..4913839a 100644
--- a/lib/logging.js
+++ b/lib/logging.js
@@ -1,5 +1,12 @@
 import pino from "pino";
 
-export const l = (...msgs) => pino().info(msgs.join(" "));
-export const warn = (...msgs) => pino().warn(msgs.join(" "));
-export const err = (...msgs) => pino().error(msgs.join(" "));
+export let l = (...msgs) => pino().info(msgs.join(" "));
+export let warn = (...msgs) => pino().warn(msgs.join(" "));
+export let err = (...msgs) => pino().error(msgs.join(" "));
+
+export let line = () => {
+  let stack = new Error().stack;
+  let stackLine = stack.split("\n")[1];
+  let match = stackLine.match(/at\s+(.*):(\d+):(\d+)/);
+  return `${match[1]}:${match[2]}`;
+};
diff --git a/lib/migrate.js b/lib/migrate.js
index 7ff13649..ef15c738 100644
--- a/lib/migrate.js
+++ b/lib/migrate.js
@@ -1,11 +1,15 @@
-import { err, warn, l } from "$lib/logging";
+import { err, warn, l, line } from "$lib/logging";
 import { types } from "$lib/payments";
-import { archive, db, g, ga } from "$lib/db";
+import { archive, db, g, ga, s } from "$lib/db";
 
 let q = {};
 async function migrate(id) {
-  id = id.replace(/\s/g, "").toLowerCase();
-  let user = await ga(`user:${id}`);
+  if (id) id = id.replace(/\s/g, "").toLowerCase();
+  let user = await g(`user:${id}`);
+  if (typeof user === "string") user = await g(`user:${user}`);
+  if (user) return user;
+
+  user = await ga(`user:${id}`);
 
   if (typeof user === "string") {
     if (await g(`user:${user}`)) {
@@ -14,7 +18,6 @@ async function migrate(id) {
     }
 
     id = user;
-
     user = await ga(`user:${id}`);
   }
 
@@ -22,48 +25,75 @@ async function migrate(id) {
     let { username } = user;
     username = username.replace(/\s/g, "").toLowerCase();
     l("migrating", username);
-    let multi = await db
-      .multi()
-      .set(`user:${username}`, id)
-      .set(`user:${user.pubkey}`, id)
-      .set(`user:${id}`, JSON.stringify(user))
-      .set(`${id}:contacts`, (await ga(`${id}:contacts`)) || "[]")
-      .set(`credit:bitcoin:${id}`, (await ga(`credit:bitcoin:${id}`)) || 0)
-      .set(`credit:lightning:${id}`, (await ga(`credit:lightning:${id}`)) || 0)
-      .set(`${id}:lastlen`, (await ga(`${id}:lastlen`)) || 0)
-      .set(`${id}:cindex`, (await ga(`${id}:cindex`)) || 0)
-      .set(`balance:${id}`, (await ga(`balance:${id}`)) || 0)
-      .set(`pending:${id}`, (await ga(`pending:${id}`)) || 0);
-
-    let payments = await archive.lRange(`${id}:payments`, 0, -1);
-    await Promise.all(
-      payments.map(async (pid) => {
+
+    try {
+      let multi = await db
+        .multi()
+        .set(`user:${username}`, id)
+        .set(`user:${user.pubkey}`, id)
+        .set(`user:${id}`, JSON.stringify(user));
+
+      await multi.exec();
+
+      await archive.del(`user:${username}`);
+      await archive.del(`user:${user.pubkey}`);
+      await archive.del(`user:${id}`);
+
+      await s(`${id}:contacts`, (await ga(`${id}:contacts`)) || []);
+      await archive.del(`${id}:contacts`);
+
+      await s(`credit:bitcoin:${id}`, (await ga(`credit:bitcoin:${id}`)) || 0);
+      await archive.del(`credit:bitcoin:${id}`);
+
+      await s(
+        `credit:lightning:${id}`,
+        (await ga(`credit:lightning:${id}`)) || 0,
+      );
+      await archive.del(`credit:lightning:${id}`);
+
+      await s(`${id}:lastlen`, (await ga(`${id}:lastlen`)) || 0);
+      await archive.del(`${id}:lastlen`);
+      await s(`${id}:cindex`, (await ga(`${id}:cindex`)) || 0);
+      await archive.del(`${id}:cindex`);
+      await s(`balance:${id}`, (await ga(`balance:${id}`)) || 0);
+      await archive.del(`balance:${id}`);
+      await s(`pending:${id}`, (await ga(`pending:${id}`)) || 0);
+      await archive.del(`pending:${id}`);
+
+      let pid;
+      while ((pid = await archive.rPop(`${id}:payments`))) {
         let p = await ga(`payment:${pid}`);
+        if (!p) continue;
         if (p.type === types.internal && p.ref) {
           let recipient = await g(`user:${p.ref}`);
           if (!recipient) q[p.ref] = true;
         }
 
-        multi = await multi
+        await multi
           .lPush(`${id}:payments`, pid)
-          .set(`payment:${pid}`, JSON.stringify(p));
-      }),
-    );
+          .set(`payment:${pid}`, JSON.stringify(p))
+          .exec();
 
-    let invoices = await archive.lRange(`${id}:invoices`, 0, -1);
-    await Promise.all(
-      invoices.map(async (iid) => {
+        await archive.del(`payment:${pid}`);
+      }
+      await archive.del(`${id}:payments`);
+
+      let iid;
+      while ((iid = await archive.rPop(`${id}:invoices`))) {
         let inv = await ga(`invoice:${iid}`);
-        multi = await multi
+        if (!inv) continue;
+        await multi
           .lPush(`${id}:invoices`, iid)
-          .set(`invoice:${iid}`, JSON.stringify(inv));
-      }),
-    );
+          .set(`invoice:${iid}`, JSON.stringify(inv))
+          .exec();
 
-    try {
-      await multi.exec();
+        await archive.del(`invoice:${iid}`);
+      }
+      await archive.del(`${id}:invoices`);
+
+      l("done migrating", username);
     } catch (e) {
-      warn("failed to migrate", username);
+      warn("failed to migrate", username, e.message, line());
     }
 
     delete q[id];
diff --git a/lib/payments.js b/lib/payments.js
index 1c8345f1..e9f634e2 100644
--- a/lib/payments.js
+++ b/lib/payments.js
@@ -400,9 +400,10 @@ export let build = async ({ amount, address, feeRate }, user) => {
   return { feeRate, ourfee, fee, fees, hex: tx.hex };
 };
 
-let seen = await db.sMembers("missing");
-let catchUp = async () => {
+let seen;
+export let catchUp = async () => {
   try {
+      if (!seen) seen = await db.sMembers("missing");
     let txns = [];
     for (let [type, n] of Object.entries({ bitcoin: bc, liquid: lq })) {
       txns.push(
@@ -432,4 +433,3 @@ let catchUp = async () => {
   setTimeout(catchUp, 2000);
 };
 
-catchUp();
diff --git a/lib/rates.js b/lib/rates.js
index 1172d202..518261f4 100644
--- a/lib/rates.js
+++ b/lib/rates.js
@@ -48,11 +48,13 @@ export let getRates = async () => {
   if (Date.now() - date > 24 * 60 * 60 * 1000) {
     date = Date.now();
     try {
+      console.log("getting rates")
       let r = await got(
         `http://data.fixer.io/api/latest?access_key=${config.fixer}`,
       ).json();
       let { rates: fx } = r;
       let USD = fx["USD"];
+      console.log("fx", fx);
 
       Object.keys(fx).map((k) => {
         fx[k] = fx[k] / USD;
diff --git a/lib/register.js b/lib/register.js
index a1489d89..b43a0ce7 100644
--- a/lib/register.js
+++ b/lib/register.js
@@ -1,26 +1,26 @@
-import countries from "$lib/countries";
-import config from "$config";
-import { authenticator } from "otplib";
 import bcrypt from "bcrypt";
-import { l, warn } from "$lib/logging";
+import { got } from "got";
+import { authenticator } from "otplib";
 import { v4 } from "uuid";
+
+import config from "$config";
+import countries from "$lib/countries";
 import { s, g, archive } from "$lib/db";
-import { got } from "got";
+import { l, warn } from "$lib/logging";
+import { fail, getUser } from "$lib/utils";
 
 export default async (user, ip, requireChallenge) => {
   let { profile, password, pubkey, username } = user;
   l("registering", username);
 
-  if (!username) throw new Error("Username required");
+  if (!username) fail("Username required");
 
   username = username.replace(/ /g, "").toLowerCase();
   let id = v4();
   user.id = id;
 
-  let exists =
-    (await g(`user:${username}`)) || (await archive.get(`user:${username}`));
-
-  if (exists) throw new Error(`Username ${username} taken`);
+  let exists = await getUser(username);
+  if (exists) fail(`Username ${username} taken`);
 
   if (password) {
     user.password = await bcrypt.hash(password, 1);
@@ -30,11 +30,9 @@ export default async (user, ip, requireChallenge) => {
   if (config.ipregistry) {
     try {
       let {
-        location: {
-          country: { code },
-        },
+        location: { country: { code } },
       } = await got(
-        `https://api.ipregistry.co/${ip}?key=${config.ipregistry}&fields=location.country.code`
+        `https://api.ipregistry.co/${ip}?key=${config.ipregistry}&fields=location.country.code`,
       ).json();
 
       user.currency = countries[code];
diff --git a/lib/sockets.js b/lib/sockets.js
index 44b3bf2a..2339c06e 100644
--- a/lib/sockets.js
+++ b/lib/sockets.js
@@ -1,12 +1,11 @@
 import config from "$config";
-import migrate from "$lib/migrate";
 import store from "$lib/store";
 import jwt from "jsonwebtoken";
 import { v4 } from "uuid";
 import whitelist from "$lib/whitelist";
 import { l, err, warn } from "$lib/logging";
 import { g } from "$lib/db";
-import { pick } from "$lib/utils";
+import { fail, getUser, pick } from "$lib/utils";
 
 const code = 1000;
 const all = {};
@@ -51,10 +50,9 @@ let track = async (ws, token) => {
   let { id } = ws;
   let { id: uid } = jwt.decode(token);
 
-  if (!uid) throw new Error("Invalid JWT token");
-  let user = await g(`user:${uid}`);
-  if (!user) user = migrate(id);
-  if (!user) throw new Error(`User not found ${uid}`);
+  if (!uid) fail("Invalid JWT token");
+  let user = await getUser(uid);
+  if (!user) fail(`User not found ${uid}`);
 
   if (!store.sockets[uid]) store.sockets[uid] = {};
 
@@ -113,7 +111,7 @@ export async function socketServer(app) {
               ws.send(JSON.stringify({ type: "logout" }));
             setTimeout(
               () => ws && ws.close(code, `closing due to error ${e.message}`),
-              1000
+              1000,
             );
           }
           break;
diff --git a/lib/upload.js b/lib/upload.js
index 9bc41388..dc311922 100644
--- a/lib/upload.js
+++ b/lib/upload.js
@@ -6,6 +6,7 @@ import path from "path";
 import sharp from "sharp";
 import { pipeline } from "stream";
 import { g, s, db } from "$lib/db";
+import { fail } from "$lib/utils";
 import crypto from "crypto";
 
 async function getSha256Hash(stream) {
@@ -40,7 +41,7 @@ export default async (req, res) => {
   let [format, ext] = (await fileTypeFromStream(s1)).mime.split("/");
 
   if (format !== "image" && !["jpg", "jpeg", "png"].includes(ext))
-    throw new Error("unsupported file type");
+    fail("unsupported file type");
 
   let t = sharp().rotate().resize(width).webp();
 
diff --git a/lib/utils.js b/lib/utils.js
index 53ed2277..e6aba14a 100644
--- a/lib/utils.js
+++ b/lib/utils.js
@@ -3,32 +3,33 @@ import store from "$lib/store";
 import { g } from "$lib/db";
 import config from "$config";
 
-export const nada = () => {};
+export let fail = (msg) => {
+  throw new Error(msg);
+};
+
+export let nada = () => {};
 
-export const sleep = (n) => new Promise((r) => setTimeout(r, n));
-export const wait = async (f, s = 300, n = 50) => {
+export let sleep = (n) => new Promise((r) => setTimeout(r, n));
+export let wait = async (f, s = 300, n = 50) => {
   let i = 0;
   while (!(await f()) && i < s) (await sleep(n)) && i++;
-  if (i >= s) throw new Error("timeout");
+  if (i >= s) fail("timeout");
   return f();
 };
 
-export const prod = process.env.NODE_ENV === "production";
-export const fail = (msg) => {
-  throw new Error(msg);
-};
+export let prod = process.env.NODE_ENV === "production";
 
-export const bail = (res, msg) => res.code(500).send(msg);
+export let bail = (res, msg) => res.code(500).send(msg);
 
-export const SATS = 100000000;
-export const sats = (n) => Math.round(n * SATS);
-export const btc = (n) => parseFloat((n / SATS).toFixed(8));
+export let SATS = 100000000;
+export let sats = (n) => Math.round(n * SATS);
+export let btc = (n) => parseFloat((n / SATS).toFixed(8));
 
-export const uniq = (a, k) => [...new Map(a.map((x) => [k(x), x])).values()];
-export const pick = (O, K) =>
+export let uniq = (a, k) => [...new Map(a.map((x) => [k(x), x])).values()];
+export let pick = (O, K) =>
   K.reduce((o, k) => (typeof O[k] !== "undefined" && (o[k] = O[k]), o), {});
 
-export const bip21 = (address, { amount, memo, tip, type }) => {
+export let bip21 = (address, { amount, memo, tip, type }) => {
   if (!(amount || memo)) return address;
 
   let network = { liquid: "liquidnetwork", bitcoin: "bitcoin" }[type];
@@ -44,7 +45,7 @@ export const bip21 = (address, { amount, memo, tip, type }) => {
   return `${network}:${address}?${url.toString()}`;
 };
 
-export const fields = [
+export let fields = [
   "cipher",
   "pubkey",
   "password",
@@ -64,9 +65,6 @@ export const fields = [
   "banner",
 ];
 
-export const getUser = async (username) => {
-  username = username.replace(/\s/g, "").toLowerCase();
-  let uid = await g(`user:${username}`);
-  if (!uid) await migrate(username);
-  return g(`user:${uid}`);
+export let getUser = async (username) => {
+  return migrate(username);
 };
diff --git a/routes/lnurl.js b/routes/lnurl.js
index 5660fa1b..b43d3010 100644
--- a/routes/lnurl.js
+++ b/routes/lnurl.js
@@ -76,38 +76,41 @@ export default {
   async lnurlw({ params: { k1, pr } }, res) {},
 
   async lnurl({ params: { id }, query: { amount, nostr } }, res) {
-    let uid = await g(`lnurl:${id}`);
-    let user = await g(`user:${uid}`);
-    if (!user) user = await migrate(uid);
-    if (!(user && user.username)) return bail(res, "user not found");
-    let { username } = user;
-    username = username.replace(/\s/g, "").toLowerCase();
-
-    let metadata = JSON.stringify([
-      ["text/plain", `Paying ${username}@${host}`],
-      ["text/identifier", `${username}@${host}`],
-    ]);
-
-    if (nostr) {
-      try {
-        let event = JSON.parse(decodeURIComponent(nostr));
-        // TODO: validate the event
-        await s(`zap:${id}`, event);
-        metadata = nostr;
-      } catch (e) {
-        err("problem handling zap", e.message);
+    try {
+      let uid = await g(`lnurl:${id}`);
+      let user = await getUser(uid);
+      if (!user) fail("user not found");
+      let { username } = user;
+      username = username.replace(/\s/g, "").toLowerCase();
+
+      let metadata = JSON.stringify([
+        ["text/plain", `Paying ${username}@${host}`],
+        ["text/identifier", `${username}@${host}`],
+      ]);
+
+      if (nostr) {
+        try {
+          let event = JSON.parse(decodeURIComponent(nostr));
+          // TODO: validate the event
+          await s(`zap:${id}`, event);
+          metadata = nostr;
+        } catch (e) {
+          err("problem handling zap", e.message);
+        }
       }
-    }
 
-    let { text: pr } = await generate({
-      invoice: {
-        amount: Math.round(amount / 1000),
-        type: types.lightning,
-      },
-      memo: metadata,
-      user,
-    });
+      let { text: pr } = await generate({
+        invoice: {
+          amount: Math.round(amount / 1000),
+          type: types.lightning,
+        },
+        memo: metadata,
+        user,
+      });
 
-    res.send({ pr, routes: [] });
+      res.send({ pr, routes: [] });
+    } catch (e) {
+      bail(res, e.message);
+    }
   },
 };
diff --git a/routes/users.js b/routes/users.js
index 5ebb2878..2299094f 100644
--- a/routes/users.js
+++ b/routes/users.js
@@ -89,10 +89,7 @@ export default {
         } catch (e) {}
       }
 
-      let user = await g(`user:${key}`);
-      if (typeof user === "string") {
-        user = await g(`user:${user}`);
-      }
+      let user = await getUser(key);
 
       if (!user && key.length === 64) {
         user = {
@@ -132,22 +129,22 @@ export default {
     }
   },
 
-  async create(req, res) {
+  async create({ body, headers }, res) {
     try {
-      const ip = req.headers["cf-connecting-ip"];
-      let { profile, cipher, pubkey, password, username, salt } = req.body.user;
-
-      let user = {
-        profile,
-        cipher,
-        pubkey,
-        password,
-        username,
-        salt,
-      };
+      const ip = headers["cf-connecting-ip"];
+      if (!body.user) fail("no user object provided");
+      let { user } = body;
+      let fields = [
+        "profile",
+        "cipher",
+        "pubkey",
+        "password",
+        "username",
+        "salt",
+      ];
 
-      user = await register(user, ip, false);
-      l("registered new user", username);
+      user = await register(pick(user, fields), ip, false);
+      l("registered new user", user.username);
       res.send(pick(user, whitelist));
     } catch (e) {
       res.code(500).send(e.message);
@@ -199,7 +196,7 @@ export default {
         shopifyStore,
       } = body;
 
-      if (user.pin && !(pin === user.pin)) throw new Error("Pin required");
+      if (user.pin && !(pin === user.pin)) fail("Pin required");
       if (typeof newpin !== "undefined" && newpin.length === 6)
         user.pin = newpin;
       if (user.pin === "delete") delete user.pin;
@@ -210,7 +207,7 @@ export default {
       let token;
       if (user.username.toLowerCase() !== username.toLowerCase() && exists) {
         err("username taken", username, user.username, exists.username);
-        throw new Error("Username taken");
+        fail("Username taken");
       } else if (username) {
         if (user.username.toLowerCase() !== username.toLowerCase())
           l("changing username", user.username, username);
-- 
cgit v1.2.3