Batch user notifications together

src/invidious/jobs/notification_job.cr

@@ -1,8 +1,32 @@
+struct VideoNotification
+  getter video_id : String
+  getter channel_id : String
+  getter published : Time
+
+  def_hash @channel_id, @video_id
+
+  def ==(other)
+    video_id == other.video_id
+  end
+
+  def self.from_video(video : ChannelVideo) : self
+    VideoNotification.new(video.id, video.ucid, video.published)
+  end
+
+  def initialize(@video_id, @channel_id, @published)
+  end
+
+  def clone : VideoNotification
+    VideoNotification.new(video_id.clone, channel_id.clone, published.clone)
+  end
+end
+
 class Invidious::Jobs::NotificationJob < Invidious::Jobs::BaseJob
+  private getter notification_channel : ::Channel(VideoNotification)
   private getter connection_channel : ::Channel({Bool, ::Channel(PQ::Notification)})
   private getter pg_url : URI
 
-  def initialize(@connection_channel, @pg_url)
+  def initialize(@notification_channel, @connection_channel, @pg_url)
   end
 
   def begin
@@ -10,6 +34,58 @@ class Invidious::Jobs::NotificationJob < Invidious::Jobs::BaseJob
 
     PG.connect_listen(pg_url, "notifications") { |event| connections.each(&.send(event)) }
 
+    # hash of channels to their videos (id+published) that need notifying
+    to_notify = Hash(String, Set(VideoNotification)).new(->(hash : Hash(String, Set(VideoNotification)), key : String) { hash[key] = Set(VideoNotification).new })
+
+    # fiber to locally cache all incoming notifications (from pubsub webhooks and refresh channels job)
+    spawn do
+      begin
+        loop do
+          notification = notification_channel.receive
+          to_notify[notification.channel_id] << notification
+        end
+      end
+    end
+    # fiber to regularly persist all cached notifications
+    spawn do
+      loop do
+        begin
+          LOGGER.debug("NotificationJob: waking up")
+          cloned = to_notify.clone
+          to_notify.clear
+
+          cloned.each do |channel_id, notifications|
+            if notifications.empty?
+              next
+            end
+
+            LOGGER.info("NotificationJob: updating channel #{channel_id} with #{notifications.size} notifications")
+            if CONFIG.enable_user_notifications
+              video_ids = notifications.map { |n| n.video_id }
+              Invidious::Database::Users.add_multiple_notifications(channel_id, video_ids)
+              notifications.each do |n|
+                # Deliver notifications to `/api/v1/auth/notifications`
+                payload = {
+                  "topic"     => n.channel_id,
+                  "videoId"   => n.video_id,
+                  "published" => n.published.to_unix,
+                }.to_json
+                PG_DB.exec("NOTIFY notifications, E'#{payload}'")
+              end
+            else
+              Invidious::Database::Users.feed_needs_update(channel_id)
+            end
+          end
+
+          LOGGER.trace("NotificationJob: Done, sleeping")
+        rescue ex
+          LOGGER.error("NotificationJob: #{ex.message}")
+        end
+        sleep 1.minute
+        Fiber.yield
+      end
+    end
+
     loop do
       action, connection = connection_channel.receive