remove users from archive once migrated

10 files changed, 122 insertions, 95 deletions

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();
+
+        await archive.del(`payment:${pid}`);
+      }
+      await archive.del(`${id}:payments`);
 
-    let invoices = await archive.lRange(`${id}:invoices`, 0, -1);
-    await Promise.all(
-      invoices.map(async (iid) => {
+      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);
 };