subscriptions

5 files changed, 83 insertions, 92 deletions

diff --git a/lib/mqtt.ts b/lib/mqtt.ts
index 636dc3e1..4212989d 100644
--- a/lib/mqtt.ts
+++ b/lib/mqtt.ts
@@ -1,5 +1,4 @@
 import mqtt from "mqtt";
 import config from "$config";
 
-export let mqtt1 = mqtt.connect("mqtt://ln.coinos.io", config.mqtt1);
-export let mqtt2 = mqtt.connect("mqtt://mqtt.coinos.io", config.mqtt2);
+export default mqtt.connect("mqtt://mqtt.coinos.io", config.mqtt);
diff --git a/lib/notifications.ts b/lib/notifications.ts
index af47a6d7..023fdeff 100644
--- a/lib/notifications.ts
+++ b/lib/notifications.ts
@@ -1,38 +1,77 @@
+import { emit } from "$lib/sockets";
+import { err, warn } from "$lib/logging";
+import { db } from "$lib/db";
+import mqtt from "$lib/mqtt";
 import config from "$config";
+import { mail, templates } from "$lib/mail";
+import { f, fmt, link, fiat, t } from "$lib/utils";
 import webpush from "web-push";
 
 if (config.vapid) {
   webpush.setVapidDetails(
-    config.vapid.url,
-    config.vapid.publicKey,
-    config.vapid.privateKey,
+    `mailto:${config.support}`,
+    config.vapid.pk,
+    config.vapid.sk,
   );
 }
 
-export const notify = async (user, payload) => {
-  try {
-    let { subscriptions } = user;
+export const notify = async (p, user, withdrawal) => {
+  emit(user.id, "payment", p);
+  let { username } = user;
+  let { paymentReceived } = t(user);
 
-    if (subscriptions) {
-      let i = 0;
-      for (let subscription of subscriptions) {
-        i++;
-        try {
-          await webpush.sendNotification(subscription, payload);
-        } catch (e) {
-          if (
-            e.body &&
-            (e.body.includes("expired") || e.body.includes("No such"))
-          ) {
-            user.subscriptions.splice(i, 1);
-            await user.save();
-          } else {
-            err("problem sending notification", e);
-          }
-        }
-      }
+  try {
+    if (user.verified && user.notify) {
+      mail(user, paymentReceived, templates.paymentReceived, {
+        ...t(user),
+        username,
+        payment: {
+          amount: fmt(p.amount),
+          link: link(p.id),
+          tip: p.tip ? fmt(p.tip) : undefined,
+          fiat: f(fiat(p.amount, p.rate), p.currency),
+          fiatTip: p.tip ? f(fiat(p.tip, p.rate), p.currency) : undefined,
+          memo: p.memo,
+          items:
+            p.items &&
+            p.items.map((i) => {
+              return {
+                quantity: i.quantity,
+                name: i.name,
+                total: i.quantity * i.price,
+                totalFiat: f(i.quantity * i.price, p.currency),
+              };
+            }),
+        },
+        withdrawal,
+      });
     }
   } catch (e) {
-    console.log("problem notifying", e);
+    err("problem emailing", e.message);
+  }
+
+  let subscriptions = await db.sMembers(`${user.id}:subscriptions`);
+
+  let payload = {
+    title: paymentReceived,
+    body: `${fmt(p.amount)} ${f(fiat(p.amount, p.rate), p.currency)}`,
+    url: `/payment/${p.id}`,
+  };
+
+  for (let s of subscriptions) {
+    webpush
+      .sendNotification(JSON.parse(s), JSON.stringify(payload))
+      .catch((e) => {
+        warn("sub failed", e.message);
+        db.sRem(`${user.id}:subscriptions`, s);
+      });
+  }
+
+  if (config.mqtt) {
+    if (!mqtt.connected) await mqtt.reconnect();
+    mqtt.publish(
+      username,
+      `pay:${p.amount}:${p.tip}:${p.rate}:${p.created}:${p.id}:${p.memo}:${p.items}`,
+    );
   }
 };
diff --git a/lib/payments.ts b/lib/payments.ts
index cc55e5ce..aeb16aa3 100644
--- a/lib/payments.ts
+++ b/lib/payments.ts
@@ -6,10 +6,10 @@ import { notify } from "$lib/notifications";
 import { db, g, s } from "$lib/db";
 import { l, err, warn } from "$lib/logging";
 import {
-  f,
   btc,
   fail,
-  fiat,
+  fmt,
+  link,
   getInvoice,
   getPayment,
   getUser,
@@ -21,18 +21,14 @@ import {
 } from "$lib/utils";
 import { callWebhook } from "$lib/webhooks";
 import got from "got";
-import { mqtt1, mqtt2 } from "$lib/mqtt";
-import { mail, templates } from "$lib/mail";
 import api from "$lib/api";
 import { bech32 } from "bech32";
 import rpc from "@coinos/rpc";
-
 import ln from "$lib/ln";
 
 let bc = rpc(config.bitcoin);
 let lq = rpc(config.liquid);
 
-let { URL } = process.env;
 let dust = 547;
 
 export let types = {
@@ -228,49 +224,18 @@ export let credit = async ({
     .incrBy(`${balanceKey}:${aid || uid}`, amount)
     .exec();
 
-  emit(uid, "payment", p);
-  notify(user, `Received ${amount} sats`);
-  callWebhook(inv, p);
-  l(username, "received", type, amount);
-
-  let items;
   if (inv.items && inv.items.length) {
     formatReceipt(inv.items, inv.currency);
     p.items = inv.items;
   }
 
-  if (config.mqtt1) {
-    if (!mqtt1.connected) await mqtt1.reconnect();
-    mqtt1.publish(
-      username,
-      `pay:${p.amount}:${p.tip}:${p.rate}:${p.created}:${p.id}:${p.memo}:${items}`,
-    );
-  }
-
-  if (config.mqtt2) {
-    if (!mqtt2.connected) await mqtt2.reconnect();
-    mqtt2.publish(
-      username,
-      `pay:${p.amount}:${p.tip}:${p.rate}:${p.created}:${p.id}:${p.memo}:${items}`,
-    );
-  }
-
-  try {
-    await completePayment(p, user);
-  } catch (e) {
-    console.log(e);
-  }
+  await completePayment(inv, p, user);
 
   return p;
 };
 
-let fmt = (sats) =>
-  "⚡️" +
-  new Intl.NumberFormat("en-US", { maximumFractionDigits: 0 }).format(sats);
-
-export let completePayment = async (p, user) => {
+export let completePayment = async (inv, p, user) => {
   let { id, autowithdraw, threshold, reserve, destination, username } = user;
-  let link = (id) => `${URL}/payment/${id}`;
   if (p.confirmed) {
     let withdrawal;
     if (autowithdraw) {
@@ -292,33 +257,12 @@ export let completePayment = async (p, user) => {
       }
     }
 
-    let { paymentReceived } = t(user);
-    if (user.verified && user.notify) {
-      mail(user, paymentReceived, templates.paymentReceived, {
-        ...t(user),
-        username,
-        payment: {
-          amount: fmt(p.amount),
-          link: link(p.id),
-          tip: p.tip ? fmt(p.tip) : undefined,
-          fiat: f(fiat(p.amount, p.rate), p.currency),
-          fiatTip: p.tip ? f(fiat(p.tip, p.rate), p.currency) : undefined,
-          memo: p.memo,
-          items:
-            p.items &&
-            p.items.map((i) => {
-              return {
-                quantity: i.quantity,
-                name: i.name,
-                total: i.quantity * i.price,
-                totalFiat: f(i.quantity * i.price, p.currency),
-              };
-            }),
-        },
-        withdrawal,
-      });
-    }
+    notify(p, user, withdrawal);
   }
+
+  l(username, "received", p.type, p.amount);
+  emit(user.id, "payment", p);
+  callWebhook(inv, p);
 };
 
 let pay = async ({ aid = undefined, amount, to, user }) => {
diff --git a/lib/utils.ts b/lib/utils.ts
index b7f55ff1..fa10fb81 100644
--- a/lib/utils.ts
+++ b/lib/utils.ts
@@ -5,6 +5,8 @@ import { g, s } from "$lib/db";
 import config from "$config";
 import { getPublicKey } from "nostr";
 
+  let { URL } = process.env;
+
 export let fail = (msg) => {
   throw new Error(msg);
 };
@@ -180,3 +182,9 @@ export let time = (() => {
     console.timeLog("", ++count, s);
   };
 })();
+
+export let fmt = (sats) =>
+  "⚡️" +
+  new Intl.NumberFormat("en-US", { maximumFractionDigits: 0 }).format(sats);
+
+export let link = (id) => `${URL}/payment/${id}`;
diff --git a/lib/whitelist.ts b/lib/whitelist.ts
index 54d3aafc..b0975d2c 100644
--- a/lib/whitelist.ts
+++ b/lib/whitelist.ts
@@ -33,6 +33,7 @@ export default [
   "profile",
   "prompt",
   "pubkey",
+  "push",
   "reserve",
   "salt",
   "seed",