Email notifications

package.json

@@ -10,7 +10,7 @@
     "connect": "ssh [email protected]",
     "deploy": "npm run build && npm run rsync",
     "rsync": "rsync -avz . [email protected]:./kiwistand --exclude=anonlocal --exclude=cache --exclude=docs --exclude=anon --exclude \"node_modules\" --exclude=\".git\" --exclude=bootstrap --exclude=data",
-    "dev": "cross-env DEBUG=\"*@attestate/*,-@attestate/delegator2*\" concurrently \"npm run watch:web\" \"npm run start\"",
+    "dev": "cross-env DEBUG=\"*@attestate/*,-@attestate/delegator2*,-@attestate/crawler*,-@attestate/extraction-worker*\" concurrently \"npm run watch:web\" \"npm run start\"",
     "start": "node -r dotenv/config ./src/launch.mjs",
     "watch": "nodemon --watch src/views --exec 'npm run dev:anon'",
     "reconcile": "NODE_ENV=reconcile npm run dev:anon",
@@ -93,6 +93,7 @@
     "open-graph-scraper-lite": "2.0.0",
     "piscina": "4.7.0",
     "pm2": "5.3.0",
+    "postmark": "4.0.5",
     "preact": "10.18.1",
     "request-filtering-agent": "2.0.1",
     "rss-parser": "3.13.0",

src/email.mjs

@@ -0,0 +1,98 @@
+import { env } from "process";
+import path from "path";
+import { randomBytes } from "crypto";
+
+import Database from "better-sqlite3";
+import postmark from "postmark";
+import { eligible } from "@attestate/delegator2";
+
+import { EIP712_MESSAGE } from "./constants.mjs";
+import * as id from "./id.mjs";
+import * as registry from "./chainstate/registry.mjs";
+import log from "./logger.mjs";
+
+const DATA_DIR = env.DATA_DIR;
+const DB_FILE = path.join(DATA_DIR, "email_subscriptions.db");
+
+const db = new Database(DB_FILE);
+const client = new postmark.ServerClient(env.POSTMARK_API_KEY);
+
+db.exec(`
+   CREATE TABLE IF NOT EXISTS email_subscriptions (
+     address TEXT PRIMARY KEY,
+     email TEXT NOT NULL,
+     secret TEXT NOT NULL UNIQUE
+   )
+ `);
+
+db.exec(`
+   CREATE INDEX IF NOT EXISTS idx_email_secret
+   ON email_subscriptions (email, secret)
+ `);
+
+db.exec(`
+   CREATE UNIQUE INDEX IF NOT EXISTS idx_secret
+   ON email_subscriptions (secret)
+ `);
+
+export async function validate(message) {
+  const signer = id.ecrecover(message, EIP712_MESSAGE);
+  const allowlist = await registry.allowlist();
+  const delegations = await registry.delegations();
+  const identity = eligible(allowlist, delegations, signer);
+
+  if (!identity) {
+    throw new Error(
+      "Body must include a validly signed message from an eligible signer.",
+    );
+  }
+
+  return identity;
+}
+
+export async function addSubscription(address, email) {
+  const secret = randomBytes(32).toString("hex");
+  const stmt = db.prepare(`
+     INSERT OR REPLACE INTO email_subscriptions
+     (address, email, secret) VALUES (?, ?, ?)
+   `);
+  stmt.run(address, email, secret);
+  return secret;
+}
+
+export async function byAddress(address) {
+  const stmt = db.prepare(
+    "SELECT email, secret FROM email_subscriptions WHERE address = ?",
+  );
+  const result = stmt.get(address);
+  return result ? { email: result.email, secret: result.secret } : null;
+}
+
+export async function send(receiver, { sender, title, message, data }) {
+  if (env.NODE_ENV !== "production") return;
+
+  const { email, secret } = await byAddress(receiver);
+  if (!email || !secret) return;
+
+  const unsubscribeUrl = `https://news.kiwistand.com/unsubscribe/${secret}`;
+
+  await client.sendEmail({
+    From: `${sender} <[email protected]>`,
+    To: email,
+    Subject: title,
+    TextBody: `${message}\n\nView comment: ${data.url}\n\n\n\nUnsubscribe: ${unsubscribeUrl}`,
+    MessageStream: "outbound",
+  });
+}
+
+export async function unsubscribe(secret) {
+  const stmt = db.prepare(
+    "SELECT address FROM email_subscriptions WHERE secret = ?",
+  );
+  const exists = stmt.get(secret);
+  if (!exists) return false;
+
+  const remove = db.prepare("DELETE FROM email_subscriptions WHERE secret = ?");
+  remove.run(secret);
+  return true;
+}

src/http.mjs

@@ -68,6 +68,7 @@ import * as karma from "./karma.mjs";
 import * as frame from "./frame.mjs";
 import * as subscriptions from "./subscriptions.mjs";
 import * as telegram from "./telegram.mjs";
+import * as email from "./email.mjs";
 import * as price from "./price.mjs";
 import {
   getRandomIndex,
@@ -307,6 +308,63 @@ export async function launch(trie, libp2p) {
       reply.send(weeklySales);
     }
   });
+  app.get("/unsubscribe/:secret", async (req, res) => {
+    const { secret } = req.params;
+
+    try {
+      const success = await email.unsubscribe(secret);
+      if (!success) {
+        return res.status(404).send("Invalid or expired unsubscribe link");
+      }
+
+      res.send(`
+       <html>
+         <body>
+           <h1>Unsubscribed</h1>
+           <p>You have been successfully unsubscribed from notifications.</p>
+         </body>
+       </html>
+     `);
+    } catch (err) {
+      console.error("Unsubscribe error:", err);
+      res.status(500).send("Error processing unsubscribe request");
+    }
+  });
+
+  app.post("/api/v1/email-notifications", async (request, reply) => {
+    const message = request.body;
+    const testExpr = /.+@.+\..+/;
+
+    if (!message || message.type !== "EMAILAUTH") {
+      return sendError(reply, 400, "Bad Request", "Invalid message type");
+    }
+
+    if (message.title === message.href && !testExpr.test(message.title)) {
+      return sendError(
+        reply,
+        400,
+        "Bad Request",
+        "Title and href must be emails and the same",
+      );
+    }
+    const userEmail = message.title;
+
+    try {
+      const identity = await email.validate(message);
+      await email.addSubscription(identity, userEmail);
+
+      const code = 200;
+      const httpMessage = "OK";
+      const details = "Successfully subscribed to email notifications";
+      return sendStatus(reply, code, httpMessage, details);
+    } catch (err) {
+      const code = 500;
+      const httpMessage = "Internal Server Error";
+      const details = err.toString();
+      return sendError(reply, code, httpMessage, details);
+    }
+  });
+
   app.post("/api/v1/telegram", async (request, reply) => {
     const message = request.body;
     // NOTE: The message here is ALMOST a compliant Kiwi News amplify or

src/subscriptions.mjs

@@ -8,16 +8,17 @@ import Database from "better-sqlite3";
 import log from "./logger.mjs";
 import { getSubmission } from "./cache.mjs";
 import { resolve } from "./ens.mjs";
+import * as email from "./email.mjs";
 import { truncateComment } from "./views/activity.mjs";
 
 if (env.NODE_ENV == "production")
   webpush.setVapidDetails(
     "mailto:[email protected]",
-    process.env.VAPID_PUBLIC_KEY,
-    process.env.VAPID_PRIVATE_KEY,
+    env.VAPID_PUBLIC_KEY,
+    env.VAPID_PRIVATE_KEY,
   );
 
-const DATA_DIR = process.env.DATA_DIR;
+const DATA_DIR = env.DATA_DIR;
 const DB_FILE = path.join(DATA_DIR, "web_push_subscriptions.db");
 
 const db = new Database(DB_FILE);
@@ -62,17 +63,26 @@ export async function triggerNotification(message) {
   const uniqueReceivers = Array.from(new Set(receivers));
 
   const maxChars = 140;
+  const url =
+    `https://news.kiwistand.com/stories?index=0x${submission.index}` +
+    `&cachebuster=0x${message.index}#0x${message.index}`;
+
   await Promise.allSettled(
-    uniqueReceivers.map(
-      async (receiver) =>
-        await send(receiver, {
-          title: `${ensData.displayName} replied`,
-          message: truncateComment(message.title, maxChars),
-          data: {
-            url: `https://news.kiwistand.com/stories?index=0x${submission.index}&cachebuster=0x${message.index}#0x${message.index}`,
-          },
-        }),
-    ),
+    uniqueReceivers.map(async (receiver) => {
+      await send(receiver, {
+        title: `${ensData.displayName} replied`,
+        message: truncateComment(message.title, maxChars),
+        data: {
+          url: `https://news.kiwistand.com/stories?index=0x${submission.index}&cachebuster=0x${message.index}#0x${message.index}`,
+        },
+      });
+      await email.send(receiver, {
+        sender: ensData.displayName,
+        title: `New comment in: ${submission.title}`,
+        message: message.title,
+        data: { url },
+      });
+    }),
   );
 }
 
@@ -92,6 +102,8 @@ export function get(address) {
 }
 
 export async function send(address, payload) {
+  if (env.NODE_ENV !== "production") return;
+
   const subscriptions = get(address);
   if (!subscriptions || subscriptions.length === 0) return;