a UniqueEvents channel for each subscription.

relaypool/relaypool.go

@@ -153,7 +153,6 @@ func (r *RelayPool) Add(url string, policy *Policy) error {
 					}
 				}
 			}
-
 		}
 	}()
 

relaypool/subscription.go

@@ -12,6 +12,14 @@ type Subscription struct {
 
 	filter *filter.EventFilter
 	Events chan EventMessage
+
+	started      bool
+	UniqueEvents chan event.Event
+}
+
+type EventMessage struct {
+	Event event.Event
+	Relay string
 }
 
 func (subscription Subscription) Unsub() {
@@ -21,6 +29,8 @@ func (subscription Subscription) Unsub() {
 			subscription.channel,
 		})
 	}
+
+	close(subscription.Events)
 }
 
 func (subscription Subscription) Sub(filter *filter.EventFilter) {
@@ -35,6 +45,21 @@ func (subscription Subscription) Sub(filter *filter.EventFilter) {
 			subscription.filter,
 		})
 	}
+
+	if subscription.started {
+		go subscription.startHandlingUnique()
+	}
+}
+
+func (subscription Subscription) startHandlingUnique() {
+	seen := make(map[string]struct{})
+	for em := range subscription.Events {
+		if _, ok := seen[em.Event.ID]; ok {
+			continue
+		}
+		seen[em.Event.ID] = struct{}{}
+		subscription.UniqueEvents <- em.Event
+	}
 }
 
 func (subscription Subscription) removeRelay(relay string) {
@@ -55,8 +80,3 @@ func (subscription Subscription) addRelay(relay string, ws *websocket.Conn) {
 		subscription.filter,
 	})
 }
-
-type EventMessage struct {
-	Event event.Event
-	Relay string
-}