Skip to content

Commit 4dd9ca1

Browse files
lklynetlklynet
authored
Merge pull request #55 from lklynet/pr-52
Pr 52
2 parents 1a28974 + 79987ff commit 4dd9ca1

4 files changed

Lines changed: 198 additions & 148 deletions

File tree

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "hypermind",
3-
"version": "0.9.0",
3+
"version": "0.9.1",
44
"description": "A decentralized P2P counter of active deployments",
55
"main": "server.js",
66
"scripts": {

‎public/app.js‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -447,10 +447,23 @@ const getColorFromId = (id) => {
447447
return "#" + "00000".substring(0, 6 - c.length) + c;
448448
};
449449

450+
const seenMessageIds = new Set();
451+
const messageIdHistory = [];
452+
450453
const appendMessage = (msg) => {
451454
const div = document.createElement("div");
452455

453456
if (msg.type === "CHAT") {
457+
if (msg.id) {
458+
if (seenMessageIds.has(msg.id)) return;
459+
seenMessageIds.add(msg.id);
460+
messageIdHistory.push(msg.id);
461+
if (messageIdHistory.length > 100) {
462+
const oldest = messageIdHistory.shift();
463+
seenMessageIds.delete(oldest);
464+
}
465+
}
466+
454467
// Block check
455468
if (blockedUsers.has(msg.sender)) return;
456469

‎src/p2p/swarm.js‎

Lines changed: 178 additions & 147 deletions
Original file line numberDiff line numberDiff line change
@@ -1,172 +1,203 @@
11
const Hyperswarm = require("hyperswarm");
22
const { signMessage } = require("../core/security");
3-
const { TOPIC, TOPIC_NAME, HEARTBEAT_INTERVAL, MAX_CONNECTIONS, CONNECTION_ROTATION_INTERVAL, ENABLE_CHAT } = require("../config/constants");
3+
const {
4+
TOPIC,
5+
TOPIC_NAME,
6+
HEARTBEAT_INTERVAL,
7+
MAX_CONNECTIONS,
8+
CONNECTION_ROTATION_INTERVAL,
9+
ENABLE_CHAT,
10+
} = require("../config/constants");
411
const { generateScreenname } = require("../utils/name-generator");
512

613
class SwarmManager {
7-
constructor(identity, peerManager, diagnostics, messageHandler, relayFn, broadcastFn, chatSystemFn) {
8-
this.identity = identity;
9-
this.peerManager = peerManager;
10-
this.diagnostics = diagnostics;
11-
this.messageHandler = messageHandler;
12-
this.relayFn = relayFn;
13-
this.broadcastFn = broadcastFn;
14-
this.chatSystemFn = chatSystemFn;
15-
16-
this.swarm = new Hyperswarm();
17-
this.heartbeatInterval = null;
18-
this.rotationInterval = null;
14+
constructor(
15+
identity,
16+
peerManager,
17+
diagnostics,
18+
messageHandler,
19+
relayFn,
20+
broadcastFn,
21+
chatSystemFn
22+
) {
23+
this.identity = identity;
24+
this.peerManager = peerManager;
25+
this.diagnostics = diagnostics;
26+
this.messageHandler = messageHandler;
27+
this.relayFn = relayFn;
28+
this.broadcastFn = broadcastFn;
29+
this.chatSystemFn = chatSystemFn;
30+
31+
this.swarm = new Hyperswarm();
32+
this.heartbeatInterval = null;
33+
this.rotationInterval = null;
34+
}
35+
36+
async start() {
37+
this.swarm.on("connection", (socket) => this.handleConnection(socket));
38+
39+
const discovery = this.swarm.join(TOPIC);
40+
await discovery.flushed();
41+
42+
this.startHeartbeat();
43+
this.startRotation();
44+
}
45+
46+
handleConnection(socket) {
47+
if (this.swarm.connections.size > MAX_CONNECTIONS) {
48+
socket.destroy();
49+
return;
1950
}
2051

21-
async start() {
22-
this.swarm.on("connection", (socket) => this.handleConnection(socket));
23-
24-
const discovery = this.swarm.join(TOPIC);
25-
await discovery.flushed();
26-
27-
this.startHeartbeat();
28-
this.startRotation();
29-
}
52+
socket.connectedAt = Date.now();
53+
54+
const sig = signMessage(
55+
`seq:${this.peerManager.getSeq()}`,
56+
this.identity.privateKey
57+
);
58+
const hello = JSON.stringify({
59+
type: "HEARTBEAT",
60+
id: this.identity.id,
61+
seq: this.peerManager.getSeq(),
62+
hops: 0,
63+
nonce: this.identity.nonce,
64+
sig,
65+
});
66+
socket.write(hello);
67+
this.broadcastFn();
68+
69+
socket.buffer = "";
70+
71+
socket.on("data", (data) => {
72+
this.diagnostics.increment("bytesReceived", data.length);
73+
socket.buffer += data.toString();
74+
75+
const lines = socket.buffer.split("\n");
76+
77+
socket.buffer = lines.pop();
78+
79+
for (const msgStr of lines) {
80+
if (!msgStr.trim()) continue;
81+
try {
82+
const msg = JSON.parse(msgStr);
83+
this.messageHandler.handleMessage(msg, socket);
84+
} catch (e) {}
85+
}
86+
});
87+
88+
socket.on("close", () => {
89+
if (socket.peerId && this.peerManager.hasPeer(socket.peerId)) {
90+
this.peerManager.removePeer(socket.peerId);
91+
}
92+
this.broadcastFn();
93+
});
94+
95+
socket.on("error", () => {});
96+
}
97+
98+
startHeartbeat() {
99+
this.heartbeatInterval = setInterval(() => {
100+
const seq = this.peerManager.incrementSeq();
101+
this.peerManager.addOrUpdatePeer(this.identity.id, seq, null);
102+
103+
this.messageHandler.bloomFilter.markRelayed(this.identity.id, seq);
104+
105+
const sig = signMessage(`seq:${seq}`, this.identity.privateKey);
106+
const heartbeat =
107+
JSON.stringify({
108+
type: "HEARTBEAT",
109+
id: this.identity.id,
110+
seq,
111+
hops: 0,
112+
nonce: this.identity.nonce,
113+
sig,
114+
}) + "\n";
30115

31-
handleConnection(socket) {
32-
if (this.swarm.connections.size > MAX_CONNECTIONS) {
33-
socket.destroy();
34-
return;
35-
}
116+
for (const socket of this.swarm.connections) {
117+
socket.write(heartbeat);
118+
}
36119

37-
socket.connectedAt = Date.now();
38-
39-
const sig = signMessage(`seq:${this.peerManager.getSeq()}`, this.identity.privateKey);
40-
const hello = JSON.stringify({
41-
type: "HEARTBEAT",
42-
id: this.identity.id,
43-
seq: this.peerManager.getSeq(),
44-
hops: 0,
45-
nonce: this.identity.nonce,
46-
sig,
47-
});
48-
socket.write(hello);
120+
const removed = this.peerManager.cleanupStalePeers();
121+
if (removed > 0) {
49122
this.broadcastFn();
50-
51-
socket.buffer = "";
52-
53-
socket.on("data", (data) => {
54-
this.diagnostics.increment("bytesReceived", data.length);
55-
socket.buffer += data.toString();
56-
57-
const lines = socket.buffer.split("\n");
58-
// The last element is either an empty string (if data ended with \n)
59-
// or the incomplete part of the next message.
60-
socket.buffer = lines.pop();
61-
62-
for (const msgStr of lines) {
63-
if (!msgStr.trim()) continue;
64-
try {
65-
const msg = JSON.parse(msgStr);
66-
this.messageHandler.handleMessage(msg, socket);
67-
} catch (e) {
68-
// Invalid JSON or partial message (shouldn't happen with buffering logic unless data is corrupted)
69-
}
70-
}
71-
});
72-
73-
socket.on("close", () => {
74-
if (socket.peerId && this.peerManager.hasPeer(socket.peerId)) {
75-
this.peerManager.removePeer(socket.peerId);
76-
}
77-
this.broadcastFn();
78-
});
79-
80-
socket.on("error", () => { });
123+
}
124+
}, HEARTBEAT_INTERVAL);
125+
}
126+
127+
startRotation() {
128+
this.rotationInterval = setInterval(() => {
129+
if (this.swarm.connections.size < MAX_CONNECTIONS / 2) return;
130+
131+
let oldest = null;
132+
for (const socket of this.swarm.connections) {
133+
if (!oldest || socket.connectedAt < oldest.connectedAt) {
134+
oldest = socket;
135+
}
136+
}
137+
138+
if (oldest) {
139+
if (ENABLE_CHAT && this.chatSystemFn && oldest.peerId) {
140+
this.chatSystemFn({
141+
type: "SYSTEM",
142+
content: `Connection with Node ...${oldest.peerId.slice(
143+
-8
144+
)} severed (Rotation).`,
145+
timestamp: Date.now(),
146+
});
147+
}
148+
oldest.destroy();
149+
}
150+
}, CONNECTION_ROTATION_INTERVAL);
151+
}
152+
153+
shutdown() {
154+
const sig = signMessage(
155+
`type:LEAVE:${this.identity.id}`,
156+
this.identity.privateKey
157+
);
158+
const goodbye =
159+
JSON.stringify({
160+
type: "LEAVE",
161+
id: this.identity.id,
162+
hops: 0,
163+
sig,
164+
}) + "\n";
165+
166+
this.messageHandler.bloomFilter.markRelayed(this.identity.id, "leave");
167+
168+
for (const socket of this.swarm.connections) {
169+
socket.write(goodbye);
81170
}
82171

83-
startHeartbeat() {
84-
this.heartbeatInterval = setInterval(() => {
85-
const seq = this.peerManager.incrementSeq();
86-
this.peerManager.addOrUpdatePeer(this.identity.id, seq, null);
87-
88-
const sig = signMessage(`seq:${seq}`, this.identity.privateKey);
89-
const heartbeat = JSON.stringify({
90-
type: "HEARTBEAT",
91-
id: this.identity.id,
92-
seq,
93-
hops: 0,
94-
nonce: this.identity.nonce,
95-
sig,
96-
}) + "\n";
97-
98-
for (const socket of this.swarm.connections) {
99-
socket.write(heartbeat);
100-
}
101-
102-
const removed = this.peerManager.cleanupStalePeers();
103-
if (removed > 0) {
104-
this.broadcastFn();
105-
}
106-
}, HEARTBEAT_INTERVAL);
172+
if (this.heartbeatInterval) {
173+
clearInterval(this.heartbeatInterval);
107174
}
108175

109-
startRotation() {
110-
this.rotationInterval = setInterval(() => {
111-
if (this.swarm.connections.size < MAX_CONNECTIONS / 2) return;
112-
113-
let oldest = null;
114-
for (const socket of this.swarm.connections) {
115-
if (!oldest || socket.connectedAt < oldest.connectedAt) {
116-
oldest = socket;
117-
}
118-
}
119-
120-
if (oldest) {
121-
if (ENABLE_CHAT && this.chatSystemFn && oldest.peerId) {
122-
this.chatSystemFn({
123-
type: "SYSTEM",
124-
content: `Connection with Node ...${oldest.peerId.slice(-8)} severed (Rotation).`,
125-
timestamp: Date.now()
126-
});
127-
}
128-
oldest.destroy();
129-
}
130-
}, CONNECTION_ROTATION_INTERVAL);
176+
if (this.rotationInterval) {
177+
clearInterval(this.rotationInterval);
131178
}
132179

133-
shutdown() {
134-
const sig = signMessage(`type:LEAVE:${this.identity.id}`, this.identity.privateKey);
135-
const goodbye = JSON.stringify({
136-
type: "LEAVE",
137-
id: this.identity.id,
138-
hops: 0,
139-
sig,
140-
}) + "\n";
141-
142-
for (const socket of this.swarm.connections) {
143-
socket.write(goodbye);
144-
}
145-
146-
if (this.heartbeatInterval) {
147-
clearInterval(this.heartbeatInterval);
148-
}
180+
setTimeout(() => {
181+
process.exit(0);
182+
}, 500);
183+
}
149184

150-
if (this.rotationInterval) {
151-
clearInterval(this.rotationInterval);
152-
}
185+
getSwarm() {
186+
return this.swarm;
187+
}
153188

154-
setTimeout(() => {
155-
process.exit(0);
156-
}, 500);
157-
}
189+
broadcastChat(msg) {
190+
if (!ENABLE_CHAT) return;
158191

159-
getSwarm() {
160-
return this.swarm;
192+
if (msg.id) {
193+
this.messageHandler.bloomFilter.markRelayed(msg.id, "chat");
161194
}
162195

163-
broadcastChat(msg) {
164-
if (!ENABLE_CHAT) return;
165-
const msgStr = JSON.stringify(msg) + "\n";
166-
for (const socket of this.swarm.connections) {
167-
socket.write(msgStr);
168-
}
196+
const msgStr = JSON.stringify(msg) + "\n";
197+
for (const socket of this.swarm.connections) {
198+
socket.write(msgStr);
169199
}
200+
}
170201
}
171202

172203
module.exports = { SwarmManager };

‎src/state/peers.js‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,12 @@ class PeerManager {
1111

1212
addOrUpdatePeer(id, seq, ip = null) {
1313
const stored = this.seenPeers.get(id);
14+
15+
// If we have a stored peer, only update if the new sequence is higher
16+
if (stored && seq <= stored.seq) {
17+
return false;
18+
}
19+
1420
const wasNew = !stored;
1521

1622
// Track in HyperLogLog for total unique estimation

0 commit comments

Comments
 (0)