added logging for the workers/importers

app/import_manager.rb

···
 require_relative 'post_downloader'
 
 class ImportManager
-  attr_accessor :report, :time_limit
+  attr_accessor :report, :time_limit, :logger, :log_status_updates
 
   def initialize(user)
     @user = user
···
 
     downloader = PostDownloader.new
     downloader.report = @report
+    downloader.logger = @logger
 
-    download_thread = Thread.new { downloader.import_from_queue(queue) }
+    download_thread = Thread.new do
+      @logger&.info "Starting downloader thread for #{@user}" if @log_status_updates
+
+      downloader.import_from_queue(queue)
+
+      @logger&.info "Ended downloader thread for #{@user}" if @log_status_updates
+    end
 
     import_threads = importers.map do |import|
       import.item_queue = queue
       import.report = @report
+      import.logger = @logger
 
-      Thread.new { import.run_import(@time_limit) }
+      Thread.new do
+        @logger&.info "Starting #{import.class} thread for #{@user}" if @log_status_updates
+
+        import.run_import(@time_limit)
+
+        @logger&.info "Ended #{import.class} thread for #{@user}" if @log_status_updates
+      end
     end
 
     import_threads.each { |i| i.join }
+
+    @logger&.info "Finished all importer threads for #{@user}, waiting for downloader" if @log_status_updates
 
     downloader.stop_when_empty = true
     download_thread.join

app/import_worker.rb

···
 require_relative 'reports/basic_report'
 
 class ImportWorker
-  attr_accessor :verbose
+  attr_accessor :verbose, :logger
 
   class UserThread < Thread
-    def initialize(user, collections, verbose = false)
+    def initialize(user, collections, logger, verbose = false)
       @user = user
       @verbose = verbose
+      @logger = logger
 
       super { run(collections) }
     end
···
     end
 
     def run(collections)
+      @logger&.info "Starting import thread for #{@user}"
+
       if @user.registered_at.nil?
         registration_time = get_registration_time(@user)
         @user.update!(registered_at: registration_time)
       end
 
       import = ImportManager.new(@user)
-      import.report = BasicReport.new if @verbose
+
+      if @logger
+        import.report = BasicReport.new(@logger) if @verbose
+        import.logger = @logger
+        import.log_status_updates = true
+      end
+
       import.start(collections)
+
+      @logger&.info "Ended import thread for #{@user}"
     end
 
     def get_registration_time(user)
···
 
     @firehose_thread = Thread.new { process_firehose_items }
     @downloader = PostDownloader.new
+    @downloader.logger = @logger
 
     loop do
       @user_threads.delete_if { |t| !t.alive? }
···
 
       users.each do |user|
         collections = user.imports.unfinished.map(&:collection)
-        thread = UserThread.new(user, collections, @verbose)
+        thread = UserThread.new(user, collections, @logger, @verbose)
         @user_threads << thread
       end
 

app/importers/base_importer.rb

···
 require_relative '../models/user'
 
 class BaseImporter
-  attr_accessor :item_queue, :report
+  attr_accessor :item_queue, :report, :logger
 
   def initialize(user)
     @did = DID.new(user.did)
···
     end
 
     @time_limit = requested_time_limit || @import.last_completed
-    puts "Fetching until: #{@time_limit}" if @time_limit
+    @logger&.info "Fetching until: #{@time_limit}" if @time_limit
 
     import_items
 

app/importers/likes_importer.rb

···
             @report&.update(queue: { length: @item_queue.length })
           end
         rescue InvalidRecordError => e
-          puts "Error in LikesImporter: #{record['uri']}: #{e}"
+          @logger&.warn "Error in LikesImporter: #{record['uri']}: #{e}"
         end
       end
 

app/importers/posts_importer.rb

···
             @report&.update(queue: { length: @item_queue.length })
           end
         rescue InvalidRecordError => e
-          puts "Error in PostsImporter: #{record['uri']}: #{e}"
+          @logger&.warn "Error in PostsImporter: #{record['uri']}: #{e}"
         end
       end
 

app/importers/reposts_importer.rb

···
             @report&.update(queue: { length: @item_queue.length })
           end
         rescue InvalidRecordError => e
-          puts "Error in RepostsImporter: #{record['uri']}: #{e}"
+          @logger&.warn "Error in RepostsImporter: #{record['uri']}: #{e}"
         end
       end
 

app/models/user.rb

···
 
     Post.where(user: self).delete_all
   end
+
+  def to_s
+    %(<User id: #{id}, did: "#{did}">)
+  end
 end

app/post_downloader.rb

···
 require_relative 'models/user'
 
 class PostDownloader
-  attr_accessor :report, :stop_when_empty
+  attr_accessor :report, :logger, :stop_when_empty
 
   def initialize
     @sky = Minisky.new(ENV['APPVIEW_HOST'] || 'public.api.bsky.app', nil)
···
           if post.valid?
             current_items.each { |i| update_item(i, post) }
           else
-            puts "Invalid post #{data['uri']}: #{post.errors.full_messages.join("; ")}"
+            @logger&.warn "Invalid post #{data['uri']}: #{post.errors.full_messages.join("; ")}"
             current_items.each { |i| invalidate_item(i) }
           end
         rescue InvalidRecordError => e
-          puts "Error in PostDownloader: #{data['uri']}: #{e.class}: #{e}"
+          @logger&.warn "Error in PostDownloader: #{data['uri']}: #{e.class}: #{e}"
           current_items.each { |i| i.update!(queue: nil) }
         end
       end
 
       check_missing_items(items)
     rescue StandardError => e
-      puts "Error in PostDownloader: #{e.class}: #{e}"
+      @logger&.warn "Error in PostDownloader: #{e.class}: #{e}"
     end
   end
 
···
           case status
           when :active
             # account is active but wasn't returned in getProfiles, probably was suspended on the AppView
-            puts "#{item.post_uri}: account #{did} exists on the PDS, account must have been taken down"
+            # puts "#{item.post_uri}: account #{did} exists on the PDS, account must have been taken down"
             item.destroy
           when nil
             # account was deleted, so all posts were deleted too
-            puts "#{item.post_uri}: account #{did} doesn't exist on the PDS, post must have been deleted"
+            # puts "#{item.post_uri}: account #{did} doesn't exist on the PDS, post must have been deleted"
             item.destroy
           else
             # account is inactive/suspended, but could come back, so leave it for now
-            puts "#{item.post_uri}: account #{did} is inactive: #{status}"
+            # puts "#{item.post_uri}: account #{did} is inactive: #{status}"
           end
         rescue StandardError => e
           hostname = did_obj.document.pds_host rescue "???"
-          puts "#{item.post_uri}: couldn't check account status for #{did} on #{hostname}: #{e.class}: #{e}"
+          @logger&.warn "#{item.post_uri}: couldn't check account status for #{did} on #{hostname}: #{e.class}: #{e}"
 
           # delete reference if the account's PDS is the old bsky.social (so it must have been deleted pre Nov 2023)
           item.destroy if hostname == 'bsky.social'

app/reports/basic_report.rb

···
 class BasicReport
+  def initialize(logger)
+    @logger = logger
+  end
+
   def update(data)
     data.each do |k, v|
       if k == :downloader
-        p ({ k => v }) if v[:downloaded_posts] && v[:downloaded_posts] % 100 == 0
+        @logger.info({ k => v }.inspect) if v[:downloaded_posts] && v[:downloaded_posts] % 100 == 0
       elsif k == :queue
         next
       else
-        p ({ k => v})
+        @logger.info({ k => v}.inspect)
       end
     end
   end

app/reports/simple_logger.rb

···
+require 'logger'
+
+class SimpleLogger < Logger
+  def initialize
+    super(STDOUT)
+
+    self.formatter = proc { |level, time, prog, msg| "[#{time}] #{msg}\n" }
+  end
+end

bin/worker

···
 
 require 'bundler/setup'
 require_relative '../app/import_worker'
+require_relative '../app/reports/simple_logger'
 
 $stdout.sync = true
 
···
   puts "  -v = verbose"
 end
 
+logger = SimpleLogger.new
 worker = ImportWorker.new
+worker.logger = logger
 
 args = ARGV.dup
 
···
 end
 
 trap("SIGINT") {
-  puts "Stopping..."
+  puts
+  puts "[#{Time.now}] Stopping..."
   exit
 }
 
 trap("SIGTERM") {
-  puts "Shutting down the service..."
+  puts "[#{Time.now}] Shutting down the service..."
   exit
 }
 
+puts "[#{Time.now}] Starting background worker..."
 worker.run

lib/tasks/import.rake

···
 require_relative '../../app/models/user'
 require_relative '../../app/post_downloader'
 require_relative '../../app/reports/console_report'
+require_relative '../../app/reports/simple_logger'
 
 task :enqueue_user do
   unless ENV['DID']
···
 
   import = ImportManager.new(user)
   import.report = ConsoleReport.new
+  import.logger = SimpleLogger.new
   import.time_limit = ENV['UNTIL']
 
   trap("SIGINT") {