Refactor connect_listen for notifications

src/invidious.cr

@@ -186,6 +186,13 @@ spawn do
   end
 end
 
+notification_channels = [] of Channel(PQ::Notification)
+PG.connect_listen(PG_URL, "notifications") do |event|
+  notification_channels.each do |channel|
+    channel.send(event)
+  end
+end
+
 proxies = PROXY_LIST
 
 before_all do |env|
@@ -4457,17 +4464,37 @@ get "/api/v1/mixes/:rdid" do |env|
 end
 
 get "/api/v1/auth/notifications" do |env|
+  env.response.content_type = "text/event-stream"
+
   topics = env.params.query["topics"]?.try &.split(",").uniq.first(1000)
   topics ||= [] of String
 
-  create_notification_stream(env, proxies, config, Kemal.config, decrypt_function, topics)
+  notification_channel = Channel(PQ::Notification).new
+  notification_channels << notification_channel
+
+  begin
+    create_notification_stream(env, proxies, config, Kemal.config, decrypt_function, topics, notification_channel)
+  rescue ex
+  ensure
+    notification_channels.delete(notification_channel)
+  end
 end
 
 post "/api/v1/auth/notifications" do |env|
+  env.response.content_type = "text/event-stream"
+
   topics = env.params.body["topics"]?.try &.split(",").uniq.first(1000)
   topics ||= [] of String
 
-  create_notification_stream(env, proxies, config, Kemal.config, decrypt_function, topics)
+  notification_channel = Channel(PQ::Notification).new
+  notification_channels << notification_channel
+
+  begin
+    create_notification_stream(env, proxies, config, Kemal.config, decrypt_function, topics, notification_channel)
+  rescue ex
+  ensure
+    notification_channels.delete(notification_channel)
+  end
 end
 
 get "/api/v1/auth/preferences" do |env|

src/invidious/helpers/helpers.cr

@@ -661,14 +661,10 @@ def copy_in_chunks(input, output, chunk_size = 4096)
   end
 end
 
-def create_notification_stream(env, proxies, config, kemal_config, decrypt_function, topics)
+def create_notification_stream(env, proxies, config, kemal_config, decrypt_function, topics, notification_channel)
   locale = LOCALES[env.get("preferences").as(Preferences).locale]?
 
-  env.response.content_type = "text/event-stream"
-
   since = env.params.query["since"]?.try &.to_i?
-
-  begin
   id = 0
 
   if topics.includes? "debug"
@@ -734,8 +730,12 @@ def create_notification_stream(env, proxies, config, kemal_config, decrypt_funct
         end
       end
     end
+  end
+
+  spawn do
+    loop do
+      event = notification_channel.receive
 
-      PG.connect_listen(PG_URL, "notifications") do |event|
       notification = JSON.parse(event.payload)
       topic = notification["topic"].as_s
       video_id = notification["videoId"].as_s
@@ -772,6 +772,4 @@ def create_notification_stream(env, proxies, config, kemal_config, decrypt_funct
     env.response.flush
     sleep (20 + rand(11)).seconds
   end
-  rescue
-  end
 end