Add read-only Azure reconciliation report

app/services/azure_reconciliation/directory_lister.rb

@@ -0,0 +1,80 @@
+# Lists every file in one Azure File Share directory.
+#
+# The azure_file_shares gem's `list` returns only the first page (up to 5,000
+# entries) and drops the continuation marker, so it undercounts large folders.
+# This follows NextMarker until the listing is complete, reusing the gem's
+# request signing.
+module AzureReconciliation
+  class DirectoryLister
+    class Error < StandardError; end
+
+    PAGE_SIZE = 5000
+
+    def initialize(share:, files: AzureFileShares.client.files, connection: nil)
+      @share = share
+      @files = files
+      @connection = connection
+    end
+
+    # Returns { file_name => byte_size }, or nil if the directory does not exist.
+    def files_in(directory)
+      result = {}
+      marker = nil
+
+      loop do
+        body = fetch_page(directory, marker)
+        return nil if body.nil?
+
+        xml = Nokogiri::XML(body)
+        xml.xpath("//Entries/File").each do |file|
+          result[file.at_xpath("Name").text] = file.at_xpath("Properties/Content-Length")&.text.to_i
+        end
+
+        marker = xml.at_xpath("//NextMarker")&.text.presence
+        break if marker.nil?
+      end
+
+      result
+    end
+
+    private
+
+    attr_reader :share, :files
+
+    def fetch_page(directory, marker)
+      path = "/#{share}/#{encode_path(directory)}"
+      query = { restype: "directory", comp: "list", maxresults: PAGE_SIZE.to_s }
+      query[:marker] = marker if marker
+
+      headers = {
+        "x-ms-date" => Time.now.utc.httpdate,
+        "x-ms-version" => files.client.api_version,
+      }
+      headers["Authorization"] = files.send(:calculate_authorization_header, :get, path, headers, query)
+
+      response = connection.get("#{files.file_base_url}#{path}?#{URI.encode_www_form(query)}", nil, headers)
+
+      return nil if response.status == 404
+      return response.body if response.success?
+
+      raise Error, "Listing #{directory} failed with #{response.status} #{error_detail(response.body)}"
+    end
+
+    # Azure's AuthenticationErrorDetail holds the string it signed, which is
+    # what to compare against when a signature is rejected.
+    def error_detail(body)
+      text = body.to_s.dup.force_encoding(Encoding::UTF_8).scrub.delete_prefix("")
+      xml = Nokogiri::XML(text)
+      parts = %w[Code Message AuthenticationErrorDetail].filter_map { |tag| xml.at_xpath("//Error/#{tag}")&.text&.strip }
+      parts.any? ? parts.join("\n") : text[0, 500]
+    end
+
+    def encode_path(directory)
+      directory.split("/").reject(&:blank?).map { |segment| ERB::Util.url_encode(segment) }.join("/")
+    end
+
+    def connection
+      @connection ||= files.send(:create_file_connection)
+    end
+  end
+end

app/services/azure_reconciliation/report.rb

@@ -0,0 +1,215 @@
+# Compares topics, their documents, ActiveStorage and the Azure File Share,
+# and returns one Row per discrepancy. Read only: it changes nothing.
+#
+# Expected Azure locations mirror FileManager: regular files go to the Pi and
+# v2 Content folders, videos to v2 VideoContent, and archived topics' files to
+# the matching _Archive folder. Shadow-copy topics hold documents waiting to be
+# deleted from Azure, so their files should be gone.
+module AzureReconciliation
+  class Report
+    Row = Struct.new(
+      :category, :language, :topic_id, :title, :document_id, :directory,
+      :file_name, :expected_size, :azure_size, :note,
+      keyword_init: true,
+    )
+
+    def initialize(lister:, languages: Language.all, check_blobs: true)
+      @lister = lister
+      @languages = languages
+      @check_blobs = check_blobs
+    end
+
+    def rows
+      @rows ||= languages.flat_map { |language| LanguageCheck.new(language, lister, check_blobs).rows }
+    end
+
+    def to_csv
+      CSV.generate do |csv|
+        csv << Row.members
+        rows.each { |row| csv << row.to_a }
+      end
+    end
+
+    # { [language, category] => count }
+    def summary
+      rows.group_by { |row| [ row.language, row.category ] }.transform_values(&:size).sort.to_h
+    end
+
+    private
+
+    attr_reader :lister, :languages, :check_blobs
+
+    class LanguageCheck
+      def initialize(language, lister, check_blobs)
+        @language = language
+        @lister = lister
+        @check_blobs = check_blobs
+        @rows = []
+        @claimed = Set.new
+        @live_names = {}
+      end
+
+      def rows
+        load_azure
+        topics = Topic.unscoped.where(language:).includes(:provider, documents_attachments: :blob).to_a
+        live, shadows = topics.partition { |topic| !topic.shadow_copy? }
+
+        live.each { |topic| check_live_topic(topic) }
+        shadows.each { |topic| check_shadow_topic(topic) }
+        report_orphans(live)
+        report_duplicates(live)
+
+        @rows
+      end
+
+      private
+
+      attr_reader :language, :lister, :check_blobs, :azure, :claimed, :live_names
+
+      def prefix
+        language.file_storage_prefix
+      end
+
+      def pi = "#{prefix}CMES-Pi/assets/Content"
+      def v2 = "#{prefix}CMES-v2/assets/Content"
+      def video = "#{prefix}CMES-v2/assets/VideoContent"
+      def pi_archive = "#{prefix}CMES-Pi_Archive"
+      def v2_archive = "#{prefix}CMES-v2_Archive"
+
+      def directories
+        [ pi, v2, video, pi_archive, v2_archive ]
+      end
+
+      # [active folder, archive folder] for each copy of a document.
+      def folder_pairs(document)
+        return [ [ video, v2_archive ] ] if document.content_type == "video/mp4"
+
+        [ [ pi, pi_archive ], [ v2, v2_archive ] ]
+      end
+
+      def load_azure
+        @azure = directories.index_with do |directory|
+          files = lister.files_in(directory)
+          add(category: "missing_directory", directory:) if files.nil?
+          files || {}
+        end
+      end
+
+      def check_live_topic(topic)
+        if topic.documents_attachments.empty?
+          add(category: "topic_without_documents", topic:, note: topic.state)
+          return
+        end
+
+        topic.documents_attachments.each do |document|
+          name = file_name(topic, document) or next
+          live_names[name] ||= topic
+
+          if check_blobs && !document.blob.service.exist?(document.blob.key)
+            add(category: "missing_in_storage", topic:, document:, file_name: name, expected_size: document.byte_size)
+          end
+
+          if document.byte_size.zero?
+            add(category: "empty_document", topic:, document:, file_name: name, expected_size: 0)
+            next
+          end
+
+          folder_pairs(document).each do |active, archive|
+            expected, other = topic.archived? ? [ archive, active ] : [ active, archive ]
+            check_location(topic, document, name, expected, other)
+          end
+        end
+      end
+
+      def check_location(topic, document, name, expected, other)
+        attrs = { topic:, document:, file_name: name, directory: expected, expected_size: document.byte_size }
+
+        if azure[expected].key?(name)
+          claimed << [ expected, name ]
+          size = azure[expected][name]
+          add(category: "size_mismatch", azure_size: size, **attrs) if size != document.byte_size
+        elsif azure[other].key?(name)
+          claimed << [ other, name ]
+          add(category: "wrong_folder", azure_size: azure[other][name], note: "found in #{other}", **attrs)
+        else
+          add(category: "missing_in_azure", **attrs)
+        end
+      end
+
+      # A shadow copy's file sharing a name with a current document is that
+      # document's file, so only unclaimed files count as failed deletes.
+      def check_shadow_topic(shadow)
+        shadow.documents_attachments.each do |document|
+          name = file_name(shadow, document) or next
+
+          folder_pairs(document).flatten.each do |directory|
+            next unless azure[directory].key?(name)
+            next if claimed.include?([ directory, name ])
+
+            claimed << [ directory, name ]
+            add(
+              category: "failed_delete", topic_id: shadow.document_prefix.presence&.to_i, title: shadow.title,
+              document:, directory:, file_name: name, azure_size: azure[directory][name],
+              note: "shadow topic #{shadow.id}",
+            )
+          end
+        end
+      end
+
+      def report_orphans(live)
+        topics_by_prefix = live.index_by { |topic| topic.doc_prefix.to_s }
+
+        azure.each do |directory, files|
+          files.each do |name, size|
+            next if claimed.include?([ directory, name ])
+
+            attrs = { directory:, file_name: name, azure_size: size }
+            if (topic = live_names[name])
+              add(category: "orphan_stale_copy", topic:, **attrs)
+            elsif (topic = topic_for_name(name, topics_by_prefix))
+              add(category: "orphan_unmatched_name", topic:, **attrs)
+            else
+              add(category: "orphan_no_topic", **attrs)
+            end
+          end
+        end
+      end
+
+      # File names start with "<doc_prefix>_"; a prefix may itself contain underscores.
+      def topic_for_name(name, topics_by_prefix)
+        name.to_enum(:scan, "_").each do
+          topic = topics_by_prefix[name[0...Regexp.last_match.begin(0)]]
+          return topic if topic
+        end
+        nil
+      end
+
+      def report_duplicates(live)
+        live
+          .group_by { |topic| [ topic.provider_id, topic.title.to_s.squish.downcase, topic.published_at.to_date ] }
+          .each_value do |group|
+            next if group.size < 2
+
+            ids = group.map(&:id).sort
+            group.each do |topic|
+              add(category: "duplicate_topic", topic:, note: "#{topic.state}; copies: #{ids.join(' ')}")
+            end
+          end
+      end
+
+      def file_name(topic, document)
+        topic.custom_file_name(document)
+      rescue StandardError => e
+        add(category: "broken_topic", topic:, document:, note: "#{e.class}: #{e.message}")
+        nil
+      end
+
+      def add(topic: nil, document: nil, **attrs)
+        attrs[:topic_id] ||= topic&.id
+        attrs[:title] ||= topic&.title
+        attrs[:document_id] ||= document&.id
+        @rows << Row.new(language: language.name, **attrs)
+      end
+    end
+  end
+end

lib/tasks/reconcile.rake

@@ -0,0 +1,34 @@
+namespace :reconcile do
+  desc <<~DESC.squish
+    Read-only report of discrepancies between topics, their documents, ActiveStorage and the Azure File Share.
+    ENV: LANGUAGE=<id|name> (default all), SKIP_BLOB_CHECK=1 (skip checking each file exists in storage),
+    DEST=<path> (default tmp/reconciliation-<timestamp>.csv) or DEST=- to print the CSV to stdout.
+  DESC
+  task azure: :environment do
+    share = ENV.fetch("AZURE_STORAGE_SHARE_NAME")
+    languages = if ENV["LANGUAGE"].present?
+      Language.where(id: ENV["LANGUAGE"]).or(Language.where(name: ENV["LANGUAGE"])).to_a.presence or abort "LANGUAGE not found"
+    else
+      Language.order(:name).to_a
+    end
+    dest = ENV["DEST"].presence || Rails.root.join("tmp", "reconciliation-#{Time.current.strftime('%Y%m%d-%H%M%S')}.csv").to_s
+
+    warn "Reconciling share #{share} for #{languages.map(&:name).join(', ')}"
+    report = AzureReconciliation::Report.new(
+      lister: AzureReconciliation::DirectoryLister.new(share:),
+      languages:,
+      check_blobs: ENV["SKIP_BLOB_CHECK"].blank?,
+    )
+
+    if dest == "-"
+      $stdout.write(report.to_csv)
+    else
+      File.write(dest, report.to_csv)
+      warn "Wrote #{report.rows.size} rows to #{dest}"
+    end
+
+    warn "\nSummary (language, category, count):"
+    report.summary.each { |(language, category), count| warn format("  %-20s %-25s %6d", language, category, count) }
+    warn "  No discrepancies found." if report.summary.empty?
+  end
+end