Moved notification logic into its own job

src/invidious.cr

@@ -162,17 +162,20 @@ end
 Invidious::Jobs.register Invidious::Jobs::RefreshChannelsJob.new(PG_DB, logger, config)
 Invidious::Jobs.register Invidious::Jobs::RefreshFeedsJob.new(PG_DB, logger, config)
 Invidious::Jobs.register Invidious::Jobs::SubscribeToFeedsJob.new(PG_DB, logger, config, HMAC_KEY)
+Invidious::Jobs.register Invidious::Jobs::PullPopularVideosJob.new(PG_DB)
+Invidious::Jobs.register Invidious::Jobs::UpdateDecryptFunctionJob.new
 
 if config.statistics_enabled
   Invidious::Jobs.register Invidious::Jobs::StatisticsRefreshJob.new(PG_DB, config, SOFTWARE)
 end
 
-if CONFIG.captcha_key
+if config.captcha_key
   Invidious::Jobs.register Invidious::Jobs::BypassCaptchaJob.new(logger, config)
 end
 
-Invidious::Jobs.register Invidious::Jobs::PullPopularVideosJob.new(PG_DB)
+connection_channel = Channel({Bool, Channel(PQ::Notification)}).new(32)
-Invidious::Jobs.register Invidious::Jobs::UpdateDecryptFunctionJob.new
+Invidious::Jobs.register Invidious::Jobs::NotificationJob.new(connection_channel, PG_URL)
+
 Invidious::Jobs.start_all
 
 def popular_videos
@@ -181,24 +184,6 @@ end
 
 DECRYPT_FUNCTION = Invidious::Jobs::UpdateDecryptFunctionJob::DECRYPT_FUNCTION
 
-connection_channel = Channel({Bool, Channel(PQ::Notification)}).new(32)
-spawn do
-  connections = [] of Channel(PQ::Notification)
-
-  PG.connect_listen(PG_URL, "notifications") { |event| connections.each { |connection| connection.send(event) } }
-
-  loop do
-    action, connection = connection_channel.receive
-
-    case action
-    when true
-      connections << connection
-    when false
-      connections.delete(connection)
-    end
-  end
-end
-
 before_all do |env|
   preferences = begin
     Preferences.from_json(env.request.cookies["PREFS"]?.try &.value || "{}")

src/invidious/jobs/bypass_captcha_job.cr

@@ -15,14 +15,14 @@ class Invidious::Jobs::BypassCaptchaJob < Invidious::Jobs::BaseJob
             form = html.xpath_node(%(//form[@action="/das_captcha"])).not_nil!
             site_key = form.xpath_node(%(.//div[@id="recaptcha"])).try &.["data-sitekey"]
             s_value = form.xpath_node(%(.//div[@id="recaptcha"])).try &.["data-s"]
-  
+
             inputs = {} of String => String
             form.xpath_nodes(%(.//input[@name])).map do |node|
               inputs[node["name"]] = node["value"]
             end
-  
+
             headers = response.cookies.add_request_headers(HTTP::Headers.new)
-  
+
             response = JSON.parse(HTTP::Client.post("https://api.anti-captcha.com/createTask", body: {
               "clientKey" => config.captcha_key,
               "task"      => {
@@ -32,51 +32,50 @@ class Invidious::Jobs::BypassCaptchaJob < Invidious::Jobs::BaseJob
                 "recaptchaDataSValue" => s_value,
               },
             }.to_json).body)
-  
+
             raise response["error"].as_s if response["error"]?
             task_id = response["taskId"].as_i
-  
+
             loop do
               sleep 10.seconds
-  
+
               response = JSON.parse(HTTP::Client.post("https://api.anti-captcha.com/getTaskResult", body: {
                 "clientKey" => config.captcha_key,
                 "taskId"    => task_id,
               }.to_json).body)
-  
+
               if response["status"]?.try &.== "ready"
                 break
               elsif response["errorId"]?.try &.as_i != 0
                 raise response["errorDescription"].as_s
               end
             end
-  
+
             inputs["g-recaptcha-response"] = response["solution"]["gRecaptchaResponse"].as_s
             headers["Cookies"] = response["solution"]["cookies"].as_h?.try &.map { |k, v| "#{k}=#{v}" }.join("; ") || ""
             response = YT_POOL.client &.post("/das_captcha", headers, form: inputs)
-  
+
             response.cookies
-                    .select { |cookie| cookie.name != "PREF" }
+              .select { |cookie| cookie.name != "PREF" }
-                    .each { |cookie| config.cookies << cookie }
+              .each { |cookie| config.cookies << cookie }
-      
+
             # Persist cookies between runs
-            config.cookies = config.cookies
             File.write("config/config.yml", config.to_yaml)
           elsif response.headers["Location"]?.try &.includes?("/sorry/index")
             location = response.headers["Location"].try { |u| URI.parse(u) }
             headers = HTTP::Headers{":authority" => location.host.not_nil!}
             response = YT_POOL.client &.get(location.full_path, headers)
-  
+
             html = XML.parse_html(response.body)
             form = html.xpath_node(%(//form[@action="index"])).not_nil!
             site_key = form.xpath_node(%(.//div[@id="recaptcha"])).try &.["data-sitekey"]
             s_value = form.xpath_node(%(.//div[@id="recaptcha"])).try &.["data-s"]
-  
+
             inputs = {} of String => String
             form.xpath_nodes(%(.//input[@name])).map do |node|
               inputs[node["name"]] = node["value"]
             end
-  
+
             captcha_client = HTTPClient.new(URI.parse("https://api.anti-captcha.com"))
             captcha_client.family = config.force_resolve || Socket::Family::INET
             response = JSON.parse(captcha_client.post("/createTask", body: {
@@ -88,25 +87,25 @@ class Invidious::Jobs::BypassCaptchaJob < Invidious::Jobs::BaseJob
                 "recaptchaDataSValue" => s_value,
               },
             }.to_json).body)
-  
+
             raise response["error"].as_s if response["error"]?
             task_id = response["taskId"].as_i
-  
+
             loop do
               sleep 10.seconds
-  
+
               response = JSON.parse(captcha_client.post("/getTaskResult", body: {
                 "clientKey" => config.captcha_key,
                 "taskId"    => task_id,
               }.to_json).body)
-  
+
               if response["status"]?.try &.== "ready"
                 break
               elsif response["errorId"]?.try &.as_i != 0
                 raise response["errorDescription"].as_s
               end
             end
-  
+
             inputs["g-recaptcha-response"] = response["solution"]["gRecaptchaResponse"].as_s
             headers["Cookies"] = response["solution"]["cookies"].as_h?.try &.map { |k, v| "#{k}=#{v}" }.join("; ") || ""
             response = YT_POOL.client &.post("/sorry/index", headers: headers, form: inputs)
@@ -114,11 +113,10 @@ class Invidious::Jobs::BypassCaptchaJob < Invidious::Jobs::BaseJob
               "Cookie" => URI.parse(response.headers["location"]).query_params["google_abuse"].split(";")[0],
             }
             cookies = HTTP::Cookies.from_headers(headers)
-  
+
             cookies.each { |cookie| config.cookies << cookie }
-      
+
             # Persist cookies between runs
-            config.cookies = config.cookies
             File.write("config/config.yml", config.to_yaml)
           end
         end

src/invidious/jobs/notification_job.cr

@@ -0,0 +1,24 @@
+class Invidious::Jobs::NotificationJob < Invidious::Jobs::BaseJob
+  private getter connection_channel : Channel({Bool, Channel(PQ::Notification)})
+  private getter pg_url : URI
+
+  def initialize(@connection_channel, @pg_url)
+  end
+
+  def begin
+    connections = [] of Channel(PQ::Notification)
+
+    PG.connect_listen(pg_url, "notifications") { |event| connections.each(&.send(event)) }
+
+    loop do
+      action, connection = connection_channel.receive
+
+      case action
+      when true
+        connections << connection
+      when false
+        connections.delete(connection)
+      end
+    end
+  end
+end