Fix subscriptions slow to update (#54)

The problem was that we only sent out info about new subs to relays
every 10 seconds. So in the worst scenario of creating subscriptions
sequentially like with purple pages we had to wait 10 seconds per
subscription.

I am very tired but I think there is an invariant in this code that new
channel is only created when there is no need for an update therefore we
won't end up in the situation where channel is closed but no update
happens? We can therefore drop time.After completely.

service/adapters/relay_connection.go

@@ -17,22 +17,29 @@ import (
 	"github.com/planetary-social/nos-crossposting-service/service/domain"
 )
 
+const (
+	reconnectAfter = 1 * time.Minute
+)
+
 type RelayConnection struct {
 	address domain.RelayAddress
 	logger  logging.Logger
 
 	state      app.RelayConnectionState
 	stateMutex sync.Mutex
 
-	subscriptions      map[string]subscription
-	subscriptionsMutex sync.Mutex
+	subscriptions                map[string]subscription
+	subscriptionsUpdatedCh       chan struct{}
+	subscriptionsUpdatedChClosed bool
+	subscriptionsMutex           sync.Mutex
 }
 
 func NewRelayConnection(address domain.RelayAddress, logger logging.Logger) *RelayConnection {
 	return &RelayConnection{
-		address:       address,
-		logger:        logger.New(fmt.Sprintf("relayConnection(%s)", address.String())),
-		subscriptions: make(map[string]subscription),
+		address:                address,
+		logger:                 logger.New(fmt.Sprintf("relayConnection(%s)", address.String())),
+		subscriptions:          make(map[string]subscription),
+		subscriptionsUpdatedCh: make(chan struct{}),
 	}
 }
 
@@ -82,6 +89,8 @@ func (r *RelayConnection) GetEvents(ctx context.Context, publicKey domain.Public
 		maxAge:     maxAge,
 	}
 
+	r.triggerSubscriptionUpdate()
+
 	go func() {
 		<-ctx.Done()
 		if err := r.removeChannel(ch); err != nil {
@@ -110,13 +119,26 @@ func (r *RelayConnection) removeChannel(chToRemove chan app.EventOrEndOfSavedEve
 		if chToRemove == subscription.ch {
 			close(subscription.ch)
 			delete(r.subscriptions, uuid)
+			r.triggerSubscriptionUpdate()
 			return nil
 		}
 	}
 
 	return errors.New("somehow the channel was already removed")
 }
 
+func (r *RelayConnection) triggerSubscriptionUpdate() {
+	if !r.subscriptionsUpdatedChClosed {
+		r.subscriptionsUpdatedChClosed = true
+		close(r.subscriptionsUpdatedCh)
+	}
+}
+
+func (r *RelayConnection) resetSubscriptionUpdateCh() {
+	r.subscriptionsUpdatedChClosed = false
+	r.subscriptionsUpdatedCh = make(chan struct{})
+}
+
 func (r *RelayConnection) run(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -227,7 +249,7 @@ func (r *RelayConnection) manageSubs(
 		}
 
 		select {
-		case <-time.After(manageSubscriptionsEvery):
+		case <-r.subscriptionsUpdatedCh:
 			continue
 		case <-ctx.Done():
 			return ctx.Err()
@@ -242,6 +264,8 @@ func (r *RelayConnection) updateSubs(
 	r.subscriptionsMutex.Lock()
 	defer r.subscriptionsMutex.Unlock()
 
+	r.resetSubscriptionUpdateCh()
+
 	for _, uuid := range activeSubscriptions.List() {
 		if _, ok := r.subscriptions[uuid]; !ok {
 			r.logger.Trace().

service/adapters/relay_event_downloader.go

@@ -12,9 +12,6 @@ import (
 
 const (
 	storeMetricsEvery = 30 * time.Second
-
-	reconnectAfter           = 1 * time.Minute
-	manageSubscriptionsEvery = 10 * time.Second
 )
 
 type RelayEventDownloader struct {