feat: update server connection mechanism
This commit is contained in:
complex
parent
5300315bf4
commit
f68b121bc4
@ -8,6 +8,12 @@ import { NDKUser, NDKEvent, NDKFilter, NDKPrivateKeySigner, NDKSubscription } fr
|
||||
import express from 'express';
|
||||
import { createServer } from 'http';
|
||||
import { ServerConfig, HttpRequest, RateLimitConfig, RateLimitState } from './types';
|
||||
import WebSocket from 'ws';
|
||||
|
||||
// Custom subscription interface for our WebSocket implementation
|
||||
interface CustomSubscription {
|
||||
unsubscribe: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Nostr HTTP Server class
|
||||
@ -17,7 +23,7 @@ export class NostrHttpServer {
|
||||
private config: ServerConfig;
|
||||
private rateLimits: Map<string, RateLimitState>;
|
||||
private rateLimitConfig: RateLimitConfig;
|
||||
private subscription: NDKSubscription | null = null;
|
||||
private subscription: CustomSubscription | null = null;
|
||||
|
||||
/**
|
||||
* Create a new Nostr HTTP Server
|
||||
@ -48,8 +54,34 @@ export class NostrHttpServer {
|
||||
|
||||
for (const relayUrl of this.config.relayUrls) {
|
||||
try {
|
||||
// Try to connect to the relay
|
||||
await this.ndk.connect();
|
||||
// Create a direct WebSocket connection to test connectivity
|
||||
const ws = new WebSocket(relayUrl);
|
||||
let connected = false;
|
||||
|
||||
// Wait for connection to establish with a timeout
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
if (!connected) {
|
||||
ws.close();
|
||||
reject(new Error('Connection timeout after 5 seconds'));
|
||||
}
|
||||
}, 5000);
|
||||
|
||||
ws.onopen = () => {
|
||||
clearTimeout(timeout);
|
||||
connected = true;
|
||||
resolve();
|
||||
};
|
||||
|
||||
ws.onerror = (err) => {
|
||||
clearTimeout(timeout);
|
||||
reject(new Error(`WebSocket error: ${err.toString()}`));
|
||||
};
|
||||
});
|
||||
|
||||
// Connection successful, close test connection
|
||||
ws.close();
|
||||
|
||||
// If we get here, connection was successful
|
||||
connectionStatus.set(relayUrl, true);
|
||||
console.log(`Relay ${relayUrl}: Connected`);
|
||||
@ -70,7 +102,7 @@ export class NostrHttpServer {
|
||||
console.log('Sending unencrypted test event to verify subscription...');
|
||||
|
||||
// Create a test event
|
||||
const testEvent = new NDKEvent(this.ndk, {
|
||||
const testEvent = {
|
||||
kind: 21120,
|
||||
content: 'Test event for subscription verification',
|
||||
created_at: Math.floor(Date.now() / 1000),
|
||||
@ -78,14 +110,19 @@ export class NostrHttpServer {
|
||||
tags: [
|
||||
['p', this.config.pubkey]
|
||||
]
|
||||
});
|
||||
};
|
||||
|
||||
// Sign and publish the event
|
||||
await testEvent.sign();
|
||||
await testEvent.publish();
|
||||
// Sign the event
|
||||
const signedEvent = await this.signEvent(testEvent);
|
||||
if (!signedEvent) {
|
||||
throw new Error('Failed to sign test event');
|
||||
}
|
||||
|
||||
// Publish to all relays
|
||||
await this.publishToRelays(signedEvent);
|
||||
|
||||
console.log('Unencrypted test event published successfully');
|
||||
console.log('Event ID:', testEvent.id);
|
||||
console.log('Event ID:', signedEvent.id);
|
||||
console.log('This event should be received by your subscription without decryption errors');
|
||||
} catch (error) {
|
||||
console.error('Error sending test event:', error);
|
||||
@ -109,7 +146,7 @@ export class NostrHttpServer {
|
||||
}
|
||||
|
||||
// Create the event
|
||||
const testEvent = new NDKEvent(this.ndk, {
|
||||
const testEvent = {
|
||||
kind: 21120,
|
||||
content: encryptedContent,
|
||||
created_at: Math.floor(Date.now() / 1000),
|
||||
@ -117,20 +154,117 @@ export class NostrHttpServer {
|
||||
tags: [
|
||||
['p', this.config.pubkey]
|
||||
]
|
||||
});
|
||||
};
|
||||
|
||||
// Sign and publish the event
|
||||
await testEvent.sign();
|
||||
await testEvent.publish();
|
||||
// Sign the event
|
||||
const signedEvent = await this.signEvent(testEvent);
|
||||
if (!signedEvent) {
|
||||
throw new Error('Failed to sign encrypted test event');
|
||||
}
|
||||
|
||||
// Publish to all relays
|
||||
await this.publishToRelays(signedEvent);
|
||||
|
||||
console.log('Encrypted test event published successfully');
|
||||
console.log('Event ID:', testEvent.id);
|
||||
console.log('Event ID:', signedEvent.id);
|
||||
console.log('This event should be received and decrypted by your subscription');
|
||||
} catch (error) {
|
||||
console.error('Error sending encrypted test event:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sign a Nostr event
|
||||
* @param event The event to sign
|
||||
* @returns The signed event or null if signing failed
|
||||
*/
|
||||
private async signEvent(event: any): Promise<any | null> {
|
||||
try {
|
||||
// Use NDK to sign the event
|
||||
const ndkEvent = new NDKEvent(this.ndk, event);
|
||||
await ndkEvent.sign();
|
||||
return ndkEvent.rawEvent();
|
||||
} catch (error) {
|
||||
console.error('Error signing event:', error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish an event to all configured relays
|
||||
* @param event The event to publish
|
||||
*/
|
||||
private async publishToRelays(event: any): Promise<void> {
|
||||
for (const relayUrl of this.config.relayUrls) {
|
||||
try {
|
||||
await this.publishToRelay(event, relayUrl);
|
||||
} catch (error) {
|
||||
console.error(`Error publishing to relay ${relayUrl}:`, error);
|
||||
// Continue to the next relay
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish an event to a specific relay
|
||||
* @param event The event to publish
|
||||
* @param relayUrl The relay URL
|
||||
*/
|
||||
private async publishToRelay(event: any, relayUrl: string): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
try {
|
||||
// Create a WebSocket connection
|
||||
const ws = new WebSocket(relayUrl);
|
||||
let connected = false;
|
||||
|
||||
// Set a timeout for the publish operation
|
||||
const timeout = setTimeout(() => {
|
||||
ws.close();
|
||||
reject(new Error(`Timed out connecting to relay: ${relayUrl}`));
|
||||
}, 10000);
|
||||
|
||||
ws.onopen = () => {
|
||||
clearTimeout(timeout);
|
||||
connected = true;
|
||||
|
||||
// Send the event
|
||||
const reqId = `pub-${Date.now()}`;
|
||||
const reqMsg = JSON.stringify(["EVENT", reqId, event]);
|
||||
ws.send(reqMsg);
|
||||
|
||||
// Wait for OK response
|
||||
const okTimeout = setTimeout(() => {
|
||||
ws.close();
|
||||
resolve(); // Resolve anyway, we don't want to block on OK response
|
||||
}, 5000);
|
||||
|
||||
const messageHandler = (msg: any) => {
|
||||
try {
|
||||
const data = JSON.parse(msg.data as string);
|
||||
if (Array.isArray(data) && data[0] === "OK" && data[1] === reqId) {
|
||||
clearTimeout(okTimeout);
|
||||
ws.removeListener('message', messageHandler);
|
||||
ws.close();
|
||||
resolve();
|
||||
}
|
||||
} catch (e) {
|
||||
// Ignore parsing errors
|
||||
}
|
||||
};
|
||||
|
||||
ws.on('message', messageHandler);
|
||||
};
|
||||
|
||||
ws.onerror = (err) => {
|
||||
clearTimeout(timeout);
|
||||
reject(new Error(`WebSocket error: ${err.toString()}`));
|
||||
};
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Encrypt content using NIP-44
|
||||
* @param pubkey The public key of the recipient
|
||||
@ -151,18 +285,8 @@ export class NostrHttpServer {
|
||||
*/
|
||||
public async start(): Promise<void> {
|
||||
try {
|
||||
// Connect to NDK
|
||||
console.log('Connecting to relays:', this.config.relayUrls);
|
||||
await this.ndk.connect();
|
||||
console.log('Connected to Nostr relays');
|
||||
|
||||
// Verify connection status
|
||||
const connectedRelays = this.ndk.explicitRelayUrls;
|
||||
console.log('Connected to relays:', connectedRelays);
|
||||
|
||||
if (connectedRelays.length === 0) {
|
||||
console.warn('Warning: No relays connected. Check your relay URLs.');
|
||||
}
|
||||
// We no longer need to connect to NDK since we're using direct WebSocket connections
|
||||
console.log('Initializing server with relays:', this.config.relayUrls);
|
||||
|
||||
// Check individual relay connections
|
||||
const relayStatus = await this.checkRelayConnections();
|
||||
@ -188,7 +312,7 @@ export class NostrHttpServer {
|
||||
<h1>Nostr HTTP Request Server</h1>
|
||||
<p>Server npub: ${this.config.npub}</p>
|
||||
<p>Status: Running</p>
|
||||
<p>Connected to ${this.ndk.explicitRelayUrls.length} relays</p>
|
||||
<p>Connected to ${this.config.relayUrls.length} relays</p>
|
||||
<p>Subscription: ${this.subscription ? 'Active' : 'Inactive'}</p>
|
||||
<p><a href="/test">Send Unencrypted Test Event</a></p>
|
||||
<p><a href="/test-encrypted">Send Encrypted Test Event</a></p>
|
||||
@ -246,50 +370,163 @@ export class NostrHttpServer {
|
||||
* Subscribe to kind 21120 events
|
||||
*/
|
||||
private async subscribeToEvents(): Promise<void> {
|
||||
const filter: NDKFilter = {
|
||||
kinds: [21120 as any], // Type assertion to bypass NDKKind type check
|
||||
// Define the filter type properly
|
||||
interface NostrFilter {
|
||||
kinds: number[];
|
||||
'#p'?: string[];
|
||||
authors?: string[];
|
||||
}
|
||||
|
||||
// Create filter for kind 21120 events
|
||||
const filter: NostrFilter = {
|
||||
kinds: [21120], // HTTP Messages event kind
|
||||
'#p': [this.config.pubkey] // Filter for events tagged with our pubkey
|
||||
};
|
||||
|
||||
console.log('Subscribing with filter:', JSON.stringify(filter, null, 2));
|
||||
console.log('Connected to relays:', this.ndk.explicitRelayUrls);
|
||||
console.log('Connected to relays:', this.config.relayUrls);
|
||||
|
||||
// Create a subscription with the filter and options
|
||||
const subscription = this.ndk.subscribe(filter, {
|
||||
closeOnEose: false,
|
||||
groupable: false
|
||||
});
|
||||
// Close existing subscription if any
|
||||
if (this.subscription) {
|
||||
try {
|
||||
this.subscription.unsubscribe();
|
||||
} catch (e) {
|
||||
console.warn('Error unsubscribing from previous subscription:', e);
|
||||
}
|
||||
this.subscription = null;
|
||||
}
|
||||
|
||||
// Log subscription status
|
||||
console.log('Subscription created');
|
||||
// Create a direct WebSocket connection for each relay
|
||||
for (const relayUrl of this.config.relayUrls) {
|
||||
try {
|
||||
console.log(`Creating direct WebSocket subscription to ${relayUrl}`);
|
||||
|
||||
// Set up event handler
|
||||
subscription.on('event', (event: NDKEvent) => {
|
||||
console.log('Event details:');
|
||||
console.log(` ID: ${event.id}`);
|
||||
console.log(` Kind: ${event.kind}`);
|
||||
console.log(` Pubkey: ${event.pubkey}`);
|
||||
console.log(` Created at: ${new Date(event.created_at * 1000).toISOString()}`);
|
||||
console.log(` Content: ${event.content.substring(0, 100)}${event.content.length > 100 ? '...' : ''}`);
|
||||
// Create a direct WebSocket connection
|
||||
const ws = new WebSocket(relayUrl);
|
||||
let connected = false;
|
||||
|
||||
// Log tags
|
||||
console.log(' Tags:');
|
||||
event.tags.forEach(tag => {
|
||||
console.log(` ${tag.join(', ')}`);
|
||||
});
|
||||
// Set up event handlers
|
||||
ws.onopen = () => {
|
||||
console.log(`WebSocket connected to ${relayUrl}`);
|
||||
connected = true;
|
||||
|
||||
this.processEvent(event).catch(error => {
|
||||
console.error('Error processing event:', error);
|
||||
});
|
||||
});
|
||||
// Send a REQ message to subscribe
|
||||
const reqId = `req-${Date.now()}`;
|
||||
const reqMsg = JSON.stringify(["REQ", reqId, filter]);
|
||||
console.log(`Sending subscription request: ${reqMsg}`);
|
||||
|
||||
// Add EOSE handler to confirm subscription is working
|
||||
subscription.on('eose', () => {
|
||||
console.log('EOSE received - subscription is active');
|
||||
});
|
||||
// Make sure to log when messages are received
|
||||
console.log('Waiting for events from relay...');
|
||||
ws.send(reqMsg);
|
||||
};
|
||||
|
||||
// Store subscription for later reference
|
||||
this.subscription = subscription;
|
||||
ws.onmessage = (msg) => {
|
||||
try {
|
||||
const data = JSON.parse(msg.data as string);
|
||||
|
||||
// Handle different message types
|
||||
if (Array.isArray(data)) {
|
||||
console.log('Received message:', JSON.stringify(data).substring(0, 100) + '...');
|
||||
|
||||
if (data[0] === "EVENT" && data.length >= 3) {
|
||||
console.log('Processing event:', data[2].id);
|
||||
// Convert the raw event to NDKEvent for processing
|
||||
const rawEvent = data[2];
|
||||
const event = new NDKEvent(this.ndk, {
|
||||
id: rawEvent.id,
|
||||
pubkey: rawEvent.pubkey,
|
||||
created_at: rawEvent.created_at,
|
||||
kind: rawEvent.kind,
|
||||
tags: rawEvent.tags,
|
||||
content: rawEvent.content,
|
||||
sig: rawEvent.sig
|
||||
});
|
||||
|
||||
this.processEvent(event).catch(error => {
|
||||
console.error('Error processing event:', error);
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('Error processing message:', e);
|
||||
}
|
||||
};
|
||||
|
||||
ws.onerror = (err) => {
|
||||
console.error('WebSocket error:', err);
|
||||
};
|
||||
|
||||
ws.onclose = () => {
|
||||
console.log('WebSocket connection closed');
|
||||
if (connected) {
|
||||
console.log('Connection closed, attempting to reconnect...');
|
||||
// Implement reconnection logic here if needed
|
||||
}
|
||||
};
|
||||
|
||||
// Wait for connection to establish
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
// Set a timeout to prevent hanging
|
||||
const timeout = setTimeout(() => {
|
||||
if (!connected) {
|
||||
reject(new Error('Connection timeout'));
|
||||
} else {
|
||||
resolve();
|
||||
}
|
||||
}, 5000);
|
||||
|
||||
// Resolve immediately if already connected
|
||||
if (connected) {
|
||||
clearTimeout(timeout);
|
||||
resolve();
|
||||
}
|
||||
|
||||
// Override onopen to resolve promise
|
||||
const originalOnOpen = ws.onopen;
|
||||
ws.onopen = (ev) => {
|
||||
clearTimeout(timeout);
|
||||
if (originalOnOpen) {
|
||||
originalOnOpen.call(ws, ev);
|
||||
}
|
||||
resolve();
|
||||
};
|
||||
|
||||
// Override onerror to reject promise
|
||||
const originalOnError = ws.onerror;
|
||||
ws.onerror = (ev) => {
|
||||
clearTimeout(timeout);
|
||||
if (originalOnError) {
|
||||
originalOnError.call(ws, ev);
|
||||
}
|
||||
reject(new Error('WebSocket connection error'));
|
||||
};
|
||||
});
|
||||
|
||||
// Store the subscription for later unsubscription
|
||||
this.subscription = {
|
||||
unsubscribe: () => {
|
||||
try {
|
||||
ws.close();
|
||||
} catch (e) {
|
||||
console.error('Error closing WebSocket:', e);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
console.log(`Successfully subscribed to ${relayUrl} for kind 21120 events`);
|
||||
|
||||
// If we successfully connected to at least one relay, we can break the loop
|
||||
break;
|
||||
} catch (error) {
|
||||
console.error(`Error creating subscription to ${relayUrl}:`, error);
|
||||
// Continue to the next relay
|
||||
}
|
||||
}
|
||||
|
||||
if (!this.subscription) {
|
||||
console.error('Failed to subscribe to any relay');
|
||||
throw new Error('Failed to subscribe to any relay');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@ -479,9 +716,14 @@ ${responseBody}`;
|
||||
]
|
||||
};
|
||||
|
||||
const event = new NDKEvent(this.ndk, rawEvent);
|
||||
await event.sign();
|
||||
await event.publish();
|
||||
// Sign the event
|
||||
const signedEvent = await this.signEvent(rawEvent);
|
||||
if (!signedEvent) {
|
||||
throw new Error('Failed to sign response event');
|
||||
}
|
||||
|
||||
// Publish to all relays
|
||||
await this.publishToRelays(signedEvent);
|
||||
} catch (error) {
|
||||
console.error('Error sending response event:', error);
|
||||
throw error;
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user