mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-11-02 10:52:27 +00:00
Add DM-specific nostr relay service
This commit is contained in:
thePR0M3TH3AN committed
1 parent
69b7232019
commit
2e07ddb35f
1 file changed
+567
@@ -0,0 +1,567 @@
|
||||
import { ensureNostrTools, resolveSimplePoolConstructor, shimLegacySimplePoolMethods } from "../nostr/toolkit.js";
|
||||
import { publishEventToRelay } from "../nostrPublish.js";
|
||||
import logger from "../utils/logger.js";
|
||||
|
||||
const DEFAULT_BACKOFF = {
|
||||
baseMs: 1000,
|
||||
maxMs: 30000,
|
||||
jitterRatio: 0.3,
|
||||
};
|
||||
|
||||
const MAX_SEEN_EVENT_IDS = 5000;
|
||||
|
||||
function normalizeRelayList(relays) {
|
||||
if (!Array.isArray(relays)) {
|
||||
return [];
|
||||
}
|
||||
|
||||
return Array.from(
|
||||
new Set(
|
||||
relays
|
||||
.map((relay) => (typeof relay === "string" ? relay.trim() : ""))
|
||||
.filter(Boolean),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function normalizePubkey(value) {
|
||||
if (typeof value !== "string") {
|
||||
return "";
|
||||
}
|
||||
return value.trim().toLowerCase();
|
||||
}
|
||||
|
||||
function normalizeLogger(candidate) {
|
||||
if (
|
||||
candidate &&
|
||||
typeof candidate === "object" &&
|
||||
candidate.dev &&
|
||||
candidate.user
|
||||
) {
|
||||
return candidate;
|
||||
}
|
||||
|
||||
return logger;
|
||||
}
|
||||
|
||||
function clampNumber(value, minimum, maximum) {
|
||||
if (!Number.isFinite(value)) {
|
||||
return minimum;
|
||||
}
|
||||
|
||||
return Math.min(Math.max(value, minimum), maximum);
|
||||
}
|
||||
|
||||
function resolveBackoffConfig(custom = {}) {
|
||||
return {
|
||||
baseMs: clampNumber(custom.baseMs ?? DEFAULT_BACKOFF.baseMs, 250, 60000),
|
||||
maxMs: clampNumber(custom.maxMs ?? DEFAULT_BACKOFF.maxMs, 1000, 300000),
|
||||
jitterRatio: clampNumber(custom.jitterRatio ?? DEFAULT_BACKOFF.jitterRatio, 0, 1),
|
||||
};
|
||||
}
|
||||
|
||||
function applyJitter(value, ratio) {
|
||||
if (!Number.isFinite(value) || value <= 0) {
|
||||
return 0;
|
||||
}
|
||||
if (!Number.isFinite(ratio) || ratio <= 0) {
|
||||
return value;
|
||||
}
|
||||
const jitter = value * ratio;
|
||||
return Math.round(value + (Math.random() * jitter - jitter / 2));
|
||||
}
|
||||
|
||||
class DmNostrService {
|
||||
constructor({
|
||||
relays = [],
|
||||
preferredRelays = [],
|
||||
backoff = {},
|
||||
log = null,
|
||||
} = {}) {
|
||||
this.relays = normalizeRelayList(relays);
|
||||
this.preferredRelays = normalizeRelayList(preferredRelays);
|
||||
this.pool = null;
|
||||
this.poolPromise = null;
|
||||
this.logger = normalizeLogger(log || logger);
|
||||
this.backoffConfig = resolveBackoffConfig(backoff);
|
||||
this.relayStates = new Map();
|
||||
this.relayWatermarks = new Map();
|
||||
this.relayTimers = new Map();
|
||||
this.seenEventIds = new Set();
|
||||
this.seenEventOrder = [];
|
||||
this.subscriptions = new Map();
|
||||
this.activeFilters = [];
|
||||
this.actorPubkey = "";
|
||||
this.handlers = {
|
||||
onEvent: null,
|
||||
onEose: null,
|
||||
onError: null,
|
||||
};
|
||||
}
|
||||
|
||||
async ensurePool() {
|
||||
if (this.pool) {
|
||||
return this.pool;
|
||||
}
|
||||
|
||||
if (this.poolPromise) {
|
||||
return this.poolPromise;
|
||||
}
|
||||
|
||||
this.poolPromise = (async () => {
|
||||
const tools = await ensureNostrTools();
|
||||
const SimplePool = resolveSimplePoolConstructor(tools);
|
||||
if (typeof SimplePool !== "function") {
|
||||
throw new Error("nostr-tools SimplePool unavailable");
|
||||
}
|
||||
|
||||
const pool = new SimplePool();
|
||||
shimLegacySimplePoolMethods(pool);
|
||||
this.pool = pool;
|
||||
this.poolPromise = null;
|
||||
return pool;
|
||||
})();
|
||||
|
||||
try {
|
||||
return await this.poolPromise;
|
||||
} catch (error) {
|
||||
this.poolPromise = null;
|
||||
this.logger.user.warn("[dmNostrService] Failed to initialize pool.", error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
getRelayStatus(relayUrl) {
|
||||
const relay = this.relayStates.get(relayUrl);
|
||||
return relay ? { ...relay } : null;
|
||||
}
|
||||
|
||||
listRelayStatuses() {
|
||||
return Array.from(this.relayStates.entries()).map(([url, state]) => ({
|
||||
url,
|
||||
...state,
|
||||
}));
|
||||
}
|
||||
|
||||
async connect({ relays = null } = {}) {
|
||||
if (Array.isArray(relays)) {
|
||||
this.relays = normalizeRelayList(relays);
|
||||
}
|
||||
|
||||
if (!this.relays.length) {
|
||||
this.logger.user.warn("[dmNostrService] No relays configured for DM service.");
|
||||
return;
|
||||
}
|
||||
|
||||
await this.ensurePool();
|
||||
|
||||
await Promise.all(
|
||||
this.relays.map(async (relayUrl) => {
|
||||
await this.connectRelay(relayUrl);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async connectRelay(relayUrl) {
|
||||
if (!this.pool) {
|
||||
await this.ensurePool();
|
||||
}
|
||||
|
||||
if (!relayUrl) {
|
||||
return;
|
||||
}
|
||||
|
||||
const relayState = this.ensureRelayState(relayUrl);
|
||||
relayState.lastConnectAttempt = Date.now();
|
||||
|
||||
if (typeof this.pool.ensureRelay === "function") {
|
||||
try {
|
||||
const relay = await this.pool.ensureRelay(relayUrl);
|
||||
relayState.connected = Boolean(relay);
|
||||
if (relay) {
|
||||
this.attachRelayListeners(relayUrl, relay);
|
||||
this.recordRelaySuccess(relayUrl);
|
||||
}
|
||||
} catch (error) {
|
||||
this.recordRelayFailure(relayUrl, error);
|
||||
this.scheduleReconnect(relayUrl, "connect-failed");
|
||||
}
|
||||
} else {
|
||||
relayState.connected = true;
|
||||
this.recordRelaySuccess(relayUrl);
|
||||
}
|
||||
}
|
||||
|
||||
attachRelayListeners(relayUrl, relay) {
|
||||
if (!relay || typeof relay.on !== "function") {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
relay.on("connect", () => {
|
||||
this.recordRelaySuccess(relayUrl);
|
||||
});
|
||||
relay.on("disconnect", () => {
|
||||
this.recordRelayFailure(relayUrl, new Error("relay-disconnect"));
|
||||
this.scheduleReconnect(relayUrl, "relay-disconnect");
|
||||
});
|
||||
relay.on("error", (error) => {
|
||||
this.recordRelayFailure(relayUrl, error);
|
||||
this.scheduleReconnect(relayUrl, "relay-error");
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.dev.warn("[dmNostrService] Failed to attach relay listeners.", error);
|
||||
}
|
||||
}
|
||||
|
||||
ensureRelayState(relayUrl) {
|
||||
if (!this.relayStates.has(relayUrl)) {
|
||||
this.relayStates.set(relayUrl, {
|
||||
connected: false,
|
||||
okCount: 0,
|
||||
failCount: 0,
|
||||
lastOkAt: 0,
|
||||
lastFailAt: 0,
|
||||
lastConnectAttempt: 0,
|
||||
backoffMs: 0,
|
||||
attempts: 0,
|
||||
});
|
||||
}
|
||||
|
||||
return this.relayStates.get(relayUrl);
|
||||
}
|
||||
|
||||
recordRelaySuccess(relayUrl) {
|
||||
const relayState = this.ensureRelayState(relayUrl);
|
||||
relayState.connected = true;
|
||||
relayState.okCount += 1;
|
||||
relayState.lastOkAt = Date.now();
|
||||
relayState.backoffMs = 0;
|
||||
relayState.attempts = 0;
|
||||
}
|
||||
|
||||
recordRelayFailure(relayUrl, error) {
|
||||
const relayState = this.ensureRelayState(relayUrl);
|
||||
relayState.connected = false;
|
||||
relayState.failCount += 1;
|
||||
relayState.lastFailAt = Date.now();
|
||||
|
||||
this.logger.dev.warn("[dmNostrService] Relay failure", {
|
||||
relayUrl,
|
||||
error,
|
||||
});
|
||||
}
|
||||
|
||||
scheduleReconnect(relayUrl, reason) {
|
||||
const relayState = this.ensureRelayState(relayUrl);
|
||||
relayState.attempts += 1;
|
||||
|
||||
const backoffBase = this.backoffConfig.baseMs * 2 ** (relayState.attempts - 1);
|
||||
const backoffMs = applyJitter(
|
||||
Math.min(backoffBase, this.backoffConfig.maxMs),
|
||||
this.backoffConfig.jitterRatio,
|
||||
);
|
||||
|
||||
relayState.backoffMs = backoffMs;
|
||||
|
||||
if (this.relayTimers.has(relayUrl)) {
|
||||
clearTimeout(this.relayTimers.get(relayUrl));
|
||||
}
|
||||
|
||||
this.logger.dev.info("[dmNostrService] Scheduling relay reconnect", {
|
||||
relayUrl,
|
||||
reason,
|
||||
backoffMs,
|
||||
});
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
this.relayTimers.delete(relayUrl);
|
||||
this.resubscribeRelay(relayUrl);
|
||||
}, backoffMs);
|
||||
|
||||
this.relayTimers.set(relayUrl, timer);
|
||||
}
|
||||
|
||||
buildFilters({ actorPubkey, authors = [], recipients = [] } = {}) {
|
||||
const normalizedActor = normalizePubkey(actorPubkey || this.actorPubkey);
|
||||
const authorSet = new Set(
|
||||
[normalizedActor, ...authors].map(normalizePubkey).filter(Boolean),
|
||||
);
|
||||
const recipientSet = new Set(
|
||||
[normalizedActor, ...recipients].map(normalizePubkey).filter(Boolean),
|
||||
);
|
||||
|
||||
const filters = [];
|
||||
|
||||
if (authorSet.size) {
|
||||
filters.push({
|
||||
kinds: [4],
|
||||
authors: Array.from(authorSet),
|
||||
});
|
||||
}
|
||||
|
||||
if (recipientSet.size) {
|
||||
filters.push({
|
||||
kinds: [4],
|
||||
"#p": Array.from(recipientSet),
|
||||
});
|
||||
}
|
||||
|
||||
return filters;
|
||||
}
|
||||
|
||||
updateWatermark(relayUrl, event) {
|
||||
if (!event || typeof event.created_at !== "number") {
|
||||
return;
|
||||
}
|
||||
|
||||
const current = this.relayWatermarks.get(relayUrl) || 0;
|
||||
if (event.created_at > current) {
|
||||
this.relayWatermarks.set(relayUrl, event.created_at);
|
||||
}
|
||||
}
|
||||
|
||||
hasSeenEvent(eventId) {
|
||||
if (!eventId) {
|
||||
return false;
|
||||
}
|
||||
return this.seenEventIds.has(eventId);
|
||||
}
|
||||
|
||||
rememberEvent(eventId) {
|
||||
if (!eventId || this.seenEventIds.has(eventId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.seenEventIds.add(eventId);
|
||||
this.seenEventOrder.push(eventId);
|
||||
|
||||
if (this.seenEventOrder.length > MAX_SEEN_EVENT_IDS) {
|
||||
const oldest = this.seenEventOrder.shift();
|
||||
if (oldest) {
|
||||
this.seenEventIds.delete(oldest);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
shouldProcessEvent(relayUrl, event) {
|
||||
if (!event || typeof event.id !== "string") {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (this.hasSeenEvent(event.id)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const watermark = this.relayWatermarks.get(relayUrl) || 0;
|
||||
if (typeof event.created_at === "number" && event.created_at < watermark) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
async subscribe({
|
||||
actorPubkey,
|
||||
authors,
|
||||
recipients,
|
||||
relays,
|
||||
onEvent,
|
||||
onEose,
|
||||
onError,
|
||||
} = {}) {
|
||||
if (typeof onEvent === "function") {
|
||||
this.handlers.onEvent = onEvent;
|
||||
}
|
||||
if (typeof onEose === "function") {
|
||||
this.handlers.onEose = onEose;
|
||||
}
|
||||
if (typeof onError === "function") {
|
||||
this.handlers.onError = onError;
|
||||
}
|
||||
|
||||
if (typeof actorPubkey === "string" && actorPubkey.trim()) {
|
||||
this.actorPubkey = normalizePubkey(actorPubkey);
|
||||
}
|
||||
|
||||
if (Array.isArray(relays)) {
|
||||
this.relays = normalizeRelayList(relays);
|
||||
}
|
||||
|
||||
if (!this.relays.length) {
|
||||
this.logger.user.warn("[dmNostrService] Cannot subscribe without relays.");
|
||||
return;
|
||||
}
|
||||
|
||||
this.activeFilters = this.buildFilters({
|
||||
actorPubkey: this.actorPubkey,
|
||||
authors,
|
||||
recipients,
|
||||
});
|
||||
|
||||
if (!this.activeFilters.length) {
|
||||
this.logger.user.warn("[dmNostrService] Cannot subscribe without filters.");
|
||||
return;
|
||||
}
|
||||
|
||||
await this.connect();
|
||||
|
||||
this.relays.forEach((relayUrl) => {
|
||||
this.subscribeRelay(relayUrl);
|
||||
});
|
||||
}
|
||||
|
||||
subscribeRelay(relayUrl) {
|
||||
if (!this.pool || !relayUrl) {
|
||||
return;
|
||||
}
|
||||
|
||||
const since = this.relayWatermarks.get(relayUrl);
|
||||
const filters = this.activeFilters.map((filter) => {
|
||||
if (typeof since === "number" && since > 0) {
|
||||
return {
|
||||
...filter,
|
||||
since: Math.max(0, since - 1),
|
||||
};
|
||||
}
|
||||
return { ...filter };
|
||||
});
|
||||
|
||||
if (this.subscriptions.has(relayUrl)) {
|
||||
this.unsubscribeRelay(relayUrl);
|
||||
}
|
||||
|
||||
let subscription;
|
||||
try {
|
||||
subscription = this.pool.sub([relayUrl], filters);
|
||||
} catch (error) {
|
||||
this.recordRelayFailure(relayUrl, error);
|
||||
this.scheduleReconnect(relayUrl, "subscribe-failed");
|
||||
return;
|
||||
}
|
||||
|
||||
if (!subscription || typeof subscription.on !== "function") {
|
||||
this.recordRelayFailure(relayUrl, new Error("subscription-unavailable"));
|
||||
this.scheduleReconnect(relayUrl, "subscribe-unavailable");
|
||||
return;
|
||||
}
|
||||
|
||||
subscription.on("event", (event) => {
|
||||
if (!this.shouldProcessEvent(relayUrl, event)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.rememberEvent(event.id);
|
||||
this.updateWatermark(relayUrl, event);
|
||||
this.recordRelaySuccess(relayUrl);
|
||||
|
||||
if (typeof this.handlers.onEvent === "function") {
|
||||
try {
|
||||
this.handlers.onEvent(event, { relayUrl });
|
||||
} catch (error) {
|
||||
this.logger.dev.warn("[dmNostrService] onEvent handler failed.", error);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
subscription.on("eose", () => {
|
||||
if (typeof this.handlers.onEose === "function") {
|
||||
try {
|
||||
this.handlers.onEose({ relayUrl });
|
||||
} catch (error) {
|
||||
this.logger.dev.warn("[dmNostrService] onEose handler failed.", error);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
subscription.on("close", () => {
|
||||
this.recordRelayFailure(relayUrl, new Error("subscription-closed"));
|
||||
this.scheduleReconnect(relayUrl, "subscription-closed");
|
||||
});
|
||||
|
||||
this.subscriptions.set(relayUrl, subscription);
|
||||
}
|
||||
|
||||
resubscribeRelay(relayUrl) {
|
||||
if (!relayUrl) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.subscribeRelay(relayUrl);
|
||||
}
|
||||
|
||||
unsubscribeRelay(relayUrl) {
|
||||
const subscription = this.subscriptions.get(relayUrl);
|
||||
if (subscription && typeof subscription.unsub === "function") {
|
||||
try {
|
||||
subscription.unsub();
|
||||
} catch (error) {
|
||||
this.logger.dev.warn("[dmNostrService] Failed to unsubscribe relay.", error);
|
||||
}
|
||||
}
|
||||
this.subscriptions.delete(relayUrl);
|
||||
}
|
||||
|
||||
stop() {
|
||||
for (const relayUrl of this.subscriptions.keys()) {
|
||||
this.unsubscribeRelay(relayUrl);
|
||||
}
|
||||
|
||||
for (const timer of this.relayTimers.values()) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
|
||||
this.relayTimers.clear();
|
||||
}
|
||||
|
||||
async publishSignedEvent(signedEvent, { relays = null } = {}) {
|
||||
if (!signedEvent || typeof signedEvent !== "object") {
|
||||
throw new Error("signed-event-required");
|
||||
}
|
||||
|
||||
if (signedEvent.kind !== 4) {
|
||||
throw new Error("invalid-event-kind");
|
||||
}
|
||||
|
||||
const relayList = normalizeRelayList(
|
||||
Array.isArray(relays) && relays.length
|
||||
? relays
|
||||
: this.preferredRelays.length
|
||||
? this.preferredRelays
|
||||
: this.relays,
|
||||
);
|
||||
|
||||
if (!relayList.length) {
|
||||
throw new Error("no-relays-configured");
|
||||
}
|
||||
|
||||
await this.ensurePool();
|
||||
|
||||
const results = await Promise.all(
|
||||
relayList.map(async (relayUrl) => {
|
||||
let result;
|
||||
try {
|
||||
result = await publishEventToRelay(this.pool, relayUrl, signedEvent);
|
||||
} catch (error) {
|
||||
this.recordRelayFailure(relayUrl, error);
|
||||
return { success: false, error, relayUrl };
|
||||
}
|
||||
|
||||
if (result?.success) {
|
||||
this.recordRelaySuccess(relayUrl);
|
||||
} else {
|
||||
this.recordRelayFailure(
|
||||
relayUrl,
|
||||
result?.error || new Error("publish-failed"),
|
||||
);
|
||||
}
|
||||
|
||||
return { ...result, relayUrl };
|
||||
}),
|
||||
);
|
||||
|
||||
return results;
|
||||
}
|
||||
}
|
||||
|
||||
export default DmNostrService;
|
||||
Reference in new issue
Block a user