Merge pull request #29 from planetary-social/vanish_requests

Vanish requests

Author: dcadenas

config/default.js

@@ -20,6 +20,7 @@ export default {
   slackCron: process.env.SLACK_CRON || "*/10 * * * *",
   redis: {
     host: process.env.REDIS_HOST || "localhost",
+    remote_host: process.env.REDIS_REMOTE_HOST || "redis://redis:6379",
   },
   logLevel: "info",
   rootDomain: process.env.ROOT_DOMAIN || "nos.social",

docker-compose.yml

@@ -1,6 +1,3 @@
----
-version: "3.8"
-
 services:
   server:
     build: .
@@ -10,6 +7,7 @@ services:
       - NODE_ENV=development
       - REDIS_HOST=redis
       - ROOT_DOMAIN=localhost
+
   redis:
     image: redis:7.2.4
     ports:

scripts/add_name

@@ -6,6 +6,7 @@ usage() {
     echo "  NPUB       - The public key."
     echo "  RELAY_URLS - One or more relay URLs, each as a separate argument."
     echo "  Note: This script requires the 'pubhex' secret to be set in the NIP05_SEC environment variable."
+    echo "  The base URL can be changed by setting the BASE_URL environment variable. Default is 'https://nos.social'."
     echo "Dependencies:"
     echo "  nostrkeytool - A tool for NOSTR keys, installable via 'cargo install nostrkeytool' (https://crates.io/crates/nostrkeytool)."
     echo "  nak - A tool required for authentication, installable via 'go install github.com/fiatjaf/nak@latest' (https://github.com/fiatjaf/nak)."
@@ -23,26 +24,27 @@ fi
 NAME="$1"
 NPUB="$2"
 RELAYS="${@:3}"
-
+BASE_URL="${BASE_URL:-https://nos.social}"
 RELAYS_JSON_ARRAY=$(printf "%s\n" $RELAYS | jq -R . | jq -s .)
-BASE64_DELETE_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='DELETE' -t u="https://nos.social/api/names/$NAME" --sec $NIP05_SEC | base64)
+BASE64_DELETE_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='DELETE' -t u="$BASE_URL/api/names/$NAME" --sec "$NIP05_SEC" | base64)
 
-HTTP_STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE "https://nos.social/api/names/$NAME" \
+HTTP_STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE "$BASE_URL/api/names/$NAME" \
     -H "Content-Type: application/json" \
     -H "Authorization: Nostr $BASE64_DELETE_AUTH_EVENT")
 
 echo "HTTP Status from delete: $HTTP_STATUS"
 
-PUBKEY=$(nostrkeytool --npub2pubkey $NPUB)
+PUBKEY=$(nostrkeytool --npub2pubkey "$NPUB")
 
 JSON_PAYLOAD=$(jq -n \
                   --arg name "$NAME" \
                   --arg pubkey "$PUBKEY" \
                   --argjson relays "$RELAYS_JSON_ARRAY" \
                   '{name: $name, data: {pubkey: $pubkey, relays: $relays}}')
 
-BASE64_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='POST' -t u='https://nos.social/api/names' --sec $NIP05_SEC | base64)
-curl -s https://nos.social/api/names \
+BASE64_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='POST' -t u="$BASE_URL/api/names" --sec "$NIP05_SEC" | base64)
+
+curl -s "$BASE_URL/api/names" \
     -H "Content-Type: application/json" \
     -H "Authorization: Nostr $BASE64_AUTH_EVENT" \
-    -d "$JSON_PAYLOAD"
+    -d "$JSON_PAYLOAD"

src/app.js

@@ -4,14 +4,14 @@ import pinoHTTP from "pino-http";
 import promClient from "prom-client";
 import promBundle from "express-prom-bundle";
 import cors from "cors";
-import getRedisClient from "./getRedisClient.js";
+import { getNip05RedisClient } from "./getRedisClient.js";
 import routes from "./routes.js";
 import logger from "./logger.js";
 import NameRecordRepository from "./nameRecordRepository.js";
 import fetchAndSendLatestEntries from "./slackNotifier.js";
 import config from "../config/index.js";
 
-const redisClient = await getRedisClient();
+const redisClient = await getNip05RedisClient();
 const nameRecordRepository = new NameRecordRepository(redisClient);
 const app = express();
 

src/getRedisClient.js

@@ -2,13 +2,13 @@ import config from "../config/index.js";
 import logger from "./logger.js";
 
 // istanbul ignore next
-const redisImportPromise = process.env.NODE_ENV === "test"
-  ? import("ioredis-mock")
-  : import("ioredis");
+const redisImportPromise =
+  process.env.NODE_ENV === "test" ? import("ioredis-mock") : import("ioredis");
 
 let redisClient;
+let remoteRedisClient;
 
-async function initializeRedis() {
+async function initializeNip05Redis() {
   try {
     const Redis = (await redisImportPromise).default;
     redisClient = new Redis({
@@ -17,19 +17,41 @@ async function initializeRedis() {
       db: config.redis.db,
     });
 
-    redisClient.on("connect", () => logger.info("Connected to Redis"));
-    redisClient.on("error", (err) => logger.error(err, "Redis error"));
+    redisClient.on("connect", () => logger.info("Connected to Nip 05 Redis"));
+    redisClient.on("error", (err) => logger.error(err, "Nip 05 Redis error"));
   } catch (error) {
     // istanbul ignore next
-    logger.error(error, "Error initializing Redis client");
+    logger.error(error, "Error initializing Nip 05 Redis client");
   }
 }
 
-async function getRedisClient() {
+async function initializeVanishRequestsRedis() {
+  try {
+    const Redis = (await redisImportPromise).default;
+    remoteRedisClient = new Redis(config.redis.remote_host);
+
+    remoteRedisClient.on("connect", () =>
+      logger.info("Connected to vanish requests Redis")
+    );
+    remoteRedisClient.on("error", (err) =>
+      logger.error(err, "Vanish requests Redis error")
+    );
+  } catch (error) {
+    // istanbul ignore next
+    logger.error(error, "Error initializing vanish requests Redis client");
+  }
+}
+
+export async function getNip05RedisClient() {
   if (!redisClient) {
-    await initializeRedis();
+    await initializeNip05Redis();
   }
   return redisClient;
 }
 
-export default getRedisClient;
+export async function getVanishRequestsRedisClient() {
+  if (!remoteRedisClient) {
+    await initializeVanishRequestsRedis();
+  }
+  return remoteRedisClient;
+}

src/nameRecordRepository.js

@@ -11,12 +11,12 @@ export default class NameRecordRepository {
     const luaScript = `
       local pubkey = redis.call('GET', 'pubkey:' .. KEYS[1])
       if not pubkey then return nil end
-  
+
       local relays = redis.call('SMEMBERS', 'relays:' .. pubkey)
       local userAgent = redis.call('GET', 'user_agent:' .. pubkey)
       local clientIp = redis.call('GET', 'ip:' .. pubkey)
       local updatedAt = redis.call('GET', 'updated_at:' .. pubkey)
-  
+
       return {pubkey, relays, userAgent, clientIp, updatedAt}
     `;
 
@@ -87,6 +87,68 @@ export default class NameRecordRepository {
     return true;
   }
 
+  async deleteByPubkey(pubkey) {
+    const namesToDelete = [];
+
+    // Use SCAN, avoid KEYS
+    const stream = this.redis.scanStream({
+      match: "pubkey:*",
+      count: 1000,
+    });
+
+    return new Promise((resolve, reject) => {
+      stream.on("data", (resultKeys) => {
+        stream.pause();
+
+        const pipeline = this.redis.pipeline();
+
+        resultKeys.forEach((key) => {
+          pipeline.get(key);
+        });
+
+        pipeline
+          .exec()
+          .then((results) => {
+            for (let i = 0; i < resultKeys.length; i++) {
+              const key = resultKeys[i];
+              const [err, associatedPubkey] = results[i];
+
+              if (err) {
+                console.error(`Error getting value for key ${key}:`, err);
+                continue;
+              }
+
+              if (associatedPubkey === pubkey) {
+                const name = key.split(":")[1];
+                namesToDelete.push(name);
+              }
+            }
+
+            stream.resume();
+          })
+          .catch((err) => {
+            stream.destroy();
+            reject(err);
+          });
+      });
+
+      stream.on("end", async () => {
+        try {
+          for (const name of namesToDelete) {
+            await this.deleteByName(name);
+          }
+          resolve(true);
+        } catch (err) {
+          reject(err);
+        }
+      });
+
+      stream.on("error", (err) => {
+        reject(err);
+      });
+    });
+  }
+
   async fetchAndClearPendingNotifications() {
     const luaScript = `
       local entries = redis.call('ZRANGE', 'pending_notifications', 0, -1)

src/server.js

@@ -1,23 +1,48 @@
 import app from "./app.js";
 import logger from "./logger.js";
 import config from "../config/index.js";
+import {
+  getVanishRequestsRedisClient,
+  getNip05RedisClient,
+} from "./getRedisClient.js";
+import VanishSubscriber from "./vanishSubscriber.js";
 
-app.listen(config.port, () => {
+const vanishRequestsRedisClient = await getVanishRequestsRedisClient();
+const nip05RedisClient = await getNip05RedisClient();
+
+const server = app.listen(config.port, () => {
   logger.info(`Server is running on port ${config.port}`);
 });
 
-process.on("uncaughtException", (err) => {
-  logger.fatal(err, "Uncaught exception detected");
+const vanishSubscriber = new VanishSubscriber(
+  vanishRequestsRedisClient,
+  nip05RedisClient
+);
+vanishSubscriber.run();
+
+async function gracefulShutdown() {
+  logger.info("Graceful shutdown initiated...");
+
+  vanishSubscriber.stop();
+
+  while (vanishSubscriber.isRunning) {
+    await new Promise((resolve) => setTimeout(resolve, 100));
+  }
+
   server.close(() => {
-    process.exit(1);
+    logger.info("Express server closed.");
+    process.exit(0);
   });
+}
 
-  setTimeout(() => {
-    process.abort();
-  }, 1000).unref();
-  process.exit(1);
+process.on("uncaughtException", (err) => {
+  logger.fatal(err, "Uncaught exception detected");
+  gracefulShutdown();
 });
 
 process.on("unhandledRejection", (reason, promise) => {
   logger.error(reason, "An unhandled promise rejection was detected");
 });
+
+process.on("SIGINT", gracefulShutdown);
+process.on("SIGTERM", gracefulShutdown);