Skip to content

Commit c183eaf

Browse files
dcadenasdcadenas
committed
Vanish requests
1 parent 4f53fe6 commit c183eaf

10 files changed

+321
-29
lines changed

‎config/default.js

+1
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ export default {
2020
slackCron: process.env.SLACK_CRON || "*/10 * * * *",
2121
redis: {
2222
host: process.env.REDIS_HOST || "localhost",
23+
remote_host: process.env.REDIS_REMOTE_HOST || "redis://redis:6379",
2324
},
2425
logLevel: "info",
2526
rootDomain: process.env.ROOT_DOMAIN || "nos.social",

‎docker-compose.yml

+1-3
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,3 @@
1-
---
2-
version: "3.8"
3-
41
services:
52
server:
63
build: .
@@ -10,6 +7,7 @@ services:
107
- NODE_ENV=development
118
- REDIS_HOST=redis
129
- ROOT_DOMAIN=localhost
10+
1311
redis:
1412
image: redis:7.2.4
1513
ports:

‎scripts/add_name

+9-7
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ usage() {
66
echo " NPUB - The public key."
77
echo " RELAY_URLS - One or more relay URLs, each as a separate argument."
88
echo " Note: This script requires the 'pubhex' secret to be set in the NIP05_SEC environment variable."
9+
echo " The base URL can be changed by setting the BASE_URL environment variable. Default is 'https://nos.social'."
910
echo "Dependencies:"
1011
echo " nostrkeytool - A tool for NOSTR keys, installable via 'cargo install nostrkeytool' (https://crates.io/crates/nostrkeytool)."
1112
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
2324
NAME="$1"
2425
NPUB="$2"
2526
RELAYS="${@:3}"
26-
27+
BASE_URL="${BASE_URL:-https://nos.social}"
2728
RELAYS_JSON_ARRAY=$(printf "%s\n" $RELAYS | jq -R . | jq -s .)
28-
BASE64_DELETE_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='DELETE' -t u="https://nos.social/api/names/$NAME" --sec $NIP05_SEC | base64)
29+
BASE64_DELETE_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='DELETE' -t u="$BASE_URL/api/names/$NAME" --sec "$NIP05_SEC" | base64)
2930

30-
HTTP_STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE "https://nos.social/api/names/$NAME" \
31+
HTTP_STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE "$BASE_URL/api/names/$NAME" \
3132
-H "Content-Type: application/json" \
3233
-H "Authorization: Nostr $BASE64_DELETE_AUTH_EVENT")
3334

3435
echo "HTTP Status from delete: $HTTP_STATUS"
3536

36-
PUBKEY=$(nostrkeytool --npub2pubkey $NPUB)
37+
PUBKEY=$(nostrkeytool --npub2pubkey "$NPUB")
3738

3839
JSON_PAYLOAD=$(jq -n \
3940
--arg name "$NAME" \
4041
--arg pubkey "$PUBKEY" \
4142
--argjson relays "$RELAYS_JSON_ARRAY" \
4243
'{name: $name, data: {pubkey: $pubkey, relays: $relays}}')
4344

44-
BASE64_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='POST' -t u='https://nos.social/api/names' --sec $NIP05_SEC | base64)
45-
curl -s https://nos.social/api/names \
45+
BASE64_AUTH_EVENT=$(nak event --content='' --kind 27235 -t method='POST' -t u="$BASE_URL/api/names" --sec "$NIP05_SEC" | base64)
46+
47+
curl -s "$BASE_URL/api/names" \
4648
-H "Content-Type: application/json" \
4749
-H "Authorization: Nostr $BASE64_AUTH_EVENT" \
48-
-d "$JSON_PAYLOAD"
50+
-d "$JSON_PAYLOAD"

‎src/app.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import pinoHTTP from "pino-http";
44
import promClient from "prom-client";
55
import promBundle from "express-prom-bundle";
66
import cors from "cors";
7-
import getRedisClient from "./getRedisClient.js";
7+
import { getRedisClient } from "./getRedisClient.js";
88
import routes from "./routes.js";
99
import logger from "./logger.js";
1010
import NameRecordRepository from "./nameRecordRepository.js";

‎src/getRedisClient.js

+27-5
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,11 @@ import config from "../config/index.js";
22
import logger from "./logger.js";
33

44
// istanbul ignore next
5-
const redisImportPromise = process.env.NODE_ENV === "test"
6-
? import("ioredis-mock")
7-
: import("ioredis");
5+
const redisImportPromise =
6+
process.env.NODE_ENV === "test" ? import("ioredis-mock") : import("ioredis");
87

98
let redisClient;
9+
let remoteRedisClient;
1010

1111
async function initializeRedis() {
1212
try {
@@ -25,11 +25,33 @@ async function initializeRedis() {
2525
}
2626
}
2727

28-
async function getRedisClient() {
28+
async function initializeRemoteRedis() {
29+
try {
30+
const Redis = (await redisImportPromise).default;
31+
remoteRedisClient = new Redis(config.redis.remote_host);
32+
33+
remoteRedisClient.on("connect", () =>
34+
logger.info("Connected to Remote Redis")
35+
);
36+
remoteRedisClient.on("error", (err) =>
37+
logger.error(err, "Remote Redis error")
38+
);
39+
} catch (error) {
40+
// istanbul ignore next
41+
logger.error(error, "Error initializing Remote Redis client");
42+
}
43+
}
44+
45+
export async function getRedisClient() {
2946
if (!redisClient) {
3047
await initializeRedis();
3148
}
3249
return redisClient;
3350
}
3451

35-
export default getRedisClient;
52+
export async function getRemoteRedisClient() {
53+
if (!remoteRedisClient) {
54+
await initializeRemoteRedis();
55+
}
56+
return remoteRedisClient;
57+
}

‎src/nameRecordRepository.js

+68-2
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,12 @@ export default class NameRecordRepository {
1111
const luaScript = `
1212
local pubkey = redis.call('GET', 'pubkey:' .. KEYS[1])
1313
if not pubkey then return nil end
14-
14+
1515
local relays = redis.call('SMEMBERS', 'relays:' .. pubkey)
1616
local userAgent = redis.call('GET', 'user_agent:' .. pubkey)
1717
local clientIp = redis.call('GET', 'ip:' .. pubkey)
1818
local updatedAt = redis.call('GET', 'updated_at:' .. pubkey)
19-
19+
2020
return {pubkey, relays, userAgent, clientIp, updatedAt}
2121
`;
2222

@@ -87,6 +87,72 @@ export default class NameRecordRepository {
8787
return true;
8888
}
8989

90+
async deleteByPubkey(pubkey) {
91+
const namesToDelete = [];
92+
93+
// Use SCAN, avoid KEYS
94+
const stream = this.redis.scanStream({
95+
match: "pubkey:*",
96+
count: 1000,
97+
});
98+
99+
let processingPromises = [];
100+
101+
return new Promise((resolve, reject) => {
102+
stream.on("data", (resultKeys) => {
103+
stream.pause();
104+
105+
const pipeline = this.redis.pipeline();
106+
107+
resultKeys.forEach((key) => {
108+
pipeline.get(key);
109+
});
110+
111+
pipeline
112+
.exec()
113+
.then((results) => {
114+
const processing = [];
115+
116+
for (let i = 0; i < resultKeys.length; i++) {
117+
const key = resultKeys[i];
118+
const [err, associatedPubkey] = results[i];
119+
120+
if (err) {
121+
console.error(`Error getting value for key ${key}:`, err);
122+
continue;
123+
}
124+
125+
if (associatedPubkey === pubkey) {
126+
const name = key.split(":")[1];
127+
namesToDelete.push(name);
128+
}
129+
}
130+
131+
stream.resume();
132+
})
133+
.catch((err) => {
134+
stream.destroy();
135+
reject(err);
136+
});
137+
});
138+
139+
stream.on("end", async () => {
140+
try {
141+
for (const name of namesToDelete) {
142+
await this.deleteByName(name);
143+
}
144+
resolve(true);
145+
} catch (err) {
146+
reject(err);
147+
}
148+
});
149+
150+
stream.on("error", (err) => {
151+
reject(err);
152+
});
153+
});
154+
}
155+
90156
async fetchAndClearPendingNotifications() {
91157
const luaScript = `
92158
local entries = redis.call('ZRANGE', 'pending_notifications', 0, -1)

‎src/server.js

+30-8
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,45 @@
11
import app from "./app.js";
22
import logger from "./logger.js";
33
import config from "../config/index.js";
4+
import { getRemoteRedisClient, getRedisClient } from "./getRedisClient.js";
5+
import VanishSubscriber from "./vanishSubscriber.js"; // Import the VanishSubscriber class
46

5-
app.listen(config.port, () => {
7+
const vanishRequestsRedisClient = await getRemoteRedisClient();
8+
const nip05RedisClient = await getRedisClient();
9+
10+
const server = app.listen(config.port, () => {
611
logger.info(`Server is running on port ${config.port}`);
712
});
813

9-
process.on("uncaughtException", (err) => {
10-
logger.fatal(err, "Uncaught exception detected");
14+
const vanishSubscriber = new VanishSubscriber(
15+
vanishRequestsRedisClient,
16+
nip05RedisClient
17+
);
18+
vanishSubscriber.run();
19+
20+
async function gracefulShutdown() {
21+
logger.info("Graceful shutdown initiated...");
22+
23+
vanishSubscriber.stop();
24+
25+
while (vanishSubscriber.isRunning) {
26+
await new Promise((resolve) => setTimeout(resolve, 100));
27+
}
28+
1129
server.close(() => {
12-
process.exit(1);
30+
logger.info("Express server closed.");
31+
process.exit(0);
1332
});
33+
}
1434

15-
setTimeout(() => {
16-
process.abort();
17-
}, 1000).unref();
18-
process.exit(1);
35+
process.on("uncaughtException", (err) => {
36+
logger.fatal(err, "Uncaught exception detected");
37+
gracefulShutdown();
1938
});
2039

2140
process.on("unhandledRejection", (reason, promise) => {
2241
logger.error(reason, "An unhandled promise rejection was detected");
2342
});
43+
44+
process.on("SIGINT", gracefulShutdown);
45+
process.on("SIGTERM", gracefulShutdown);

‎src/vanishSubscriber.js

+125
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
import NameRecordRepository from "./nameRecordRepository.js";
2+
3+
const VANISH_STREAM_KEY = "vanish_requests";
4+
const LAST_PROCESSED_ID_KEY = "vanish_requests:nip05_service:last_id";
5+
const BLOCK_TIME_MS = 5000; // 5 seconds
6+
7+
class VanishSubscriber {
8+
constructor(vanishRequestsRedis, nip05Redis) {
9+
// Right now we have a local redis instance for nip05 data and a remote one
10+
// used by all our services. For the momen, the remote one is only used for
11+
// the vanish stream.
12+
// TODO: Refactor to migrate and use only one redis instance.
13+
14+
const nameRecordRepository = new NameRecordRepository(nip05Redis);
15+
16+
this.vanishRequestsRedis = vanishRequestsRedis;
17+
this.nameRecordRepository = nameRecordRepository;
18+
this.abortController = new AbortController();
19+
this.isRunning = false;
20+
}
21+
22+
async processPubkey(pubkey) {
23+
console.log(`Deleting pubkey: ${pubkey}`);
24+
await this.nameRecordRepository.deleteByPubkey(pubkey);
25+
}
26+
27+
async run() {
28+
if (this.isRunning) return; // Prevent multiple runs
29+
this.isRunning = true;
30+
31+
let lastProcessedID;
32+
33+
try {
34+
lastProcessedID =
35+
(await this.vanishRequestsRedis.get(LAST_PROCESSED_ID_KEY)) || "0-0";
36+
console.log(`Starting from last processed ID: ${lastProcessedID}`);
37+
} catch (err) {
38+
console.error("Error fetching last processed ID from Redis", err);
39+
this.isRunning = false;
40+
return;
41+
}
42+
43+
const abortSignal = this.abortController.signal;
44+
45+
while (!abortSignal.aborted) {
46+
try {
47+
const streamEntries = await this.vanishRequestsRedis.xread(
48+
"BLOCK",
49+
BLOCK_TIME_MS,
50+
"STREAMS",
51+
VANISH_STREAM_KEY,
52+
lastProcessedID
53+
);
54+
55+
if (!streamEntries) {
56+
continue;
57+
}
58+
59+
for (const [stream, messages] of streamEntries) {
60+
for (const [messageID, messageData] of messages) {
61+
const event = createObjectFromPairs(messageData);
62+
63+
console.log(`Vanish requests event: ${JSON.stringify(event)} `);
64+
const pubkey = event.pubkey;
65+
66+
console.log(
67+
`Processing message ID: ${messageID} with pubkey: ${pubkey}`
68+
);
69+
70+
try {
71+
await this.processPubkey(pubkey);
72+
} catch (err) {
73+
console.error(`Error processing pubkey: ${pubkey}`, err);
74+
}
75+
76+
try {
77+
await this.vanishRequestsRedis.set(
78+
LAST_PROCESSED_ID_KEY,
79+
messageID
80+
);
81+
lastProcessedID = messageID;
82+
console.log(`Updated last processed ID to: ${lastProcessedID}`);
83+
} catch (err) {
84+
console.error(
85+
`Error updating last processed ID: ${messageID}`,
86+
err
87+
);
88+
}
89+
}
90+
}
91+
} catch (err) {
92+
if (abortSignal.aborted) {
93+
break;
94+
}
95+
console.error("Error reading from Redis stream", err);
96+
await new Promise((resolve) => setTimeout(resolve, 1000));
97+
}
98+
}
99+
100+
console.log("Cancellation signal received. Exiting gracefully...");
101+
await this.vanishRequestsRedis.set(LAST_PROCESSED_ID_KEY, lastProcessedID);
102+
console.log(`Final last processed ID saved: ${lastProcessedID}`);
103+
104+
this.isRunning = false;
105+
}
106+
107+
stop() {
108+
if (!this.isRunning) return;
109+
this.abortController.abort();
110+
console.log(
111+
"Abort signal sent. Waiting for current processing to finish..."
112+
);
113+
}
114+
}
115+
116+
function createObjectFromPairs(messageData) {
117+
return messageData.reduce((acc, value, index, arr) => {
118+
if (index % 2 === 0) {
119+
acc[value] = arr[index + 1];
120+
}
121+
return acc;
122+
}, {});
123+
}
124+
125+
export default VanishSubscriber;

0 commit comments

Comments
 (0)