3636import
3737 waku/ [
3838 waku_core,
39+ waku_core/ topics/ sharding,
3940 waku_lightpush/ common,
4041 waku_lightpush/ rpc,
4142 waku_enr,
4849 common/ utils/ nat,
4950 waku_store/ common,
5051 waku_filter_v2/ client,
52+ waku_filter_v2/ common as filter_common,
53+ waku_mix/ protocol,
5154 common/ logging,
5255 ],
5356 ./ config_chat2mix
@@ -58,6 +61,105 @@ import ../../waku/waku_rln_relay
5861logScope:
5962 topics = " chat2 mix"
6063
64+ # ########################
65+ # # Mix Spam Protection ##
66+ # ########################
67+
68+ # Forward declaration
69+ proc maintainSpamProtectionSubscription (
70+ node: WakuNode , contentTopics: seq [ContentTopic ]
71+ ) {.async .}
72+
73+ proc setupMixSpamProtectionViaFilter (node: WakuNode ) {.async .} =
74+ # # Setup filter-based spam protection coordination for mix protocol.
75+ # # Since chat2mix doesn't use relay, we subscribe via filter to receive
76+ # # spam protection coordination messages.
77+
78+ # Register message handler for spam protection coordination
79+ let spamTopics = node.wakuMix.getSpamProtectionContentTopics ()
80+
81+ proc handleSpamMessage (
82+ pubsubTopic: PubsubTopic , message: WakuMessage
83+ ): Future [void ] {.async , gcsafe .} =
84+ await node.wakuMix.handleMessage (pubsubTopic, message)
85+
86+ node.wakuFilterClient.registerPushHandler (handleSpamMessage)
87+
88+ # Wait for filter peer and maintain subscription
89+ asyncSpawn maintainSpamProtectionSubscription (node, spamTopics)
90+
91+ proc maintainSpamProtectionSubscription (
92+ node: WakuNode , contentTopics: seq [ContentTopic ]
93+ ) {.async .} =
94+ # # Maintain filter subscription for spam protection topics.
95+ # # Monitors subscription health with periodic pings and re-subscribes on failure.
96+ const RetryInterval = chronos.seconds (5 )
97+ const SubscriptionMaintenance = chronos.seconds (30 )
98+ const MaxFailedSubscribes = 3
99+ var currentFilterPeer: Option [RemotePeerInfo ] = none (RemotePeerInfo )
100+ var noFailedSubscribes = 0
101+
102+ while true :
103+ # Select or reuse filter peer
104+ if currentFilterPeer.isNone ():
105+ let filterPeerOpt = node.peerManager.selectPeer (WakuFilterSubscribeCodec )
106+ if filterPeerOpt.isNone ():
107+ debug " No filter peer available yet for spam protection, retrying..."
108+ await sleepAsync (RetryInterval )
109+ continue
110+ currentFilterPeer = some (filterPeerOpt.get ())
111+ info " Selected filter peer for spam protection" ,
112+ peer = currentFilterPeer.get ().peerId
113+
114+ # Check if subscription is still alive with ping
115+ let pingErr = (await node.wakuFilterClient.ping (currentFilterPeer.get ())).errorOr:
116+ # Subscription is alive, wait before next check
117+ await sleepAsync (SubscriptionMaintenance )
118+ if noFailedSubscribes > 0 :
119+ noFailedSubscribes = 0
120+ continue
121+
122+ # Subscription lost, need to re-subscribe
123+ warn " Spam protection filter subscription ping failed, re-subscribing" ,
124+ error = pingErr, peer = currentFilterPeer.get ().peerId
125+
126+ # Determine pubsub topic from content topics (using auto-sharding)
127+ if node.wakuAutoSharding.isNone ():
128+ error " Auto-sharding not configured, cannot determine pubsub topic for spam protection"
129+ await sleepAsync (RetryInterval )
130+ continue
131+
132+ let shardRes = node.wakuAutoSharding.get ().getShard (contentTopics[0 ])
133+ if shardRes.isErr ():
134+ error " Failed to determine shard for spam protection" , error = shardRes.error
135+ await sleepAsync (RetryInterval )
136+ continue
137+
138+ let shard = shardRes.get ()
139+ let pubsubTopic: PubsubTopic = shard # converter toPubsubTopic
140+
141+ # Subscribe to spam protection topics
142+ let res = await node.wakuFilterClient.subscribe (
143+ currentFilterPeer.get (), pubsubTopic, contentTopics
144+ )
145+ if res.isErr ():
146+ noFailedSubscribes += 1
147+ warn " Failed to subscribe to spam protection topics via filter" ,
148+ error = res.error, topics = contentTopics, failCount = noFailedSubscribes
149+
150+ if noFailedSubscribes >= MaxFailedSubscribes :
151+ # Try with a different peer
152+ warn " Max subscription failures reached, selecting new filter peer"
153+ currentFilterPeer = none (RemotePeerInfo )
154+ noFailedSubscribes = 0
155+
156+ await sleepAsync (RetryInterval )
157+ else :
158+ info " Successfully subscribed to spam protection topics via filter" ,
159+ topics = contentTopics, peer = currentFilterPeer.get ().peerId
160+ noFailedSubscribes = 0
161+ await sleepAsync (SubscriptionMaintenance )
162+
61163const Help =
62164 """
63165 Commands: /[?|help|connect|nick|exit]
@@ -229,20 +331,21 @@ proc publish(c: Chat, line: string) {.async.} =
229331 try :
230332 if not c.node.wakuLightpushClient.isNil ():
231333 # Attempt lightpush with mix
232-
233- (
234- waitFor c.node.lightpushPublish (
235- some (c.conf.getPubsubTopic (c.node, c.contentTopic)),
236- message,
237- none (RemotePeerInfo ),
238- true ,
239- )
240- ).isOkOr:
241- error " failed to publish lightpush message" , error = error
334+ let res = await c.node.lightpushPublish (
335+ some (c.conf.getPubsubTopic (c.node, c.contentTopic)),
336+ message,
337+ none (RemotePeerInfo ),
338+ true ,
339+ )
340+ if res.isErr ():
341+ error " failed to publish lightpush message" , error = res.error
342+ echo " Error: " & res.error.desc.get (" unknown error" )
242343 else :
243344 error " failed to publish message as lightpush client is not initialized"
345+ echo " Error: lightpush client is not initialized"
244346 except CatchableError :
245347 error " caught error publishing message: " , error = getCurrentExceptionMsg ()
348+ echo " Error: " & getCurrentExceptionMsg ()
246349
247350# TODO This should read or be subscribe handler subscribe
248351proc readAndPrint (c: Chat ) {.async .} =
@@ -471,7 +574,11 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
471574 error " failed to generate mix key pair" , error = error
472575 return
473576
474- (await node.mountMix (conf.clusterId, mixPrivKey, conf.mixnodes)).isOkOr:
577+ (
578+ await node.mountMix (
579+ conf.clusterId, mixPrivKey, conf.mixnodes, some (conf.rlnUserMessageLimit)
580+ )
581+ ).isOkOr:
475582 error " failed to mount waku mix protocol: " , error = $ error
476583 quit (QuitFailure )
477584
@@ -500,6 +607,10 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
500607
501608 # await node.mountRendezvousClient(conf.clusterId)
502609
610+ # Subscribe to spam protection coordination topics via filter since chat2mix doesn't use relay
611+ if not node.wakuFilterClient.isNil ():
612+ asyncSpawn setupMixSpamProtectionViaFilter (node)
613+
503614 await node.start ()
504615
505616 node.peerManager.start ()
0 commit comments