GitHub App Webhook で push を処理するとき、知っておきたい制約と対処法

class PushWebhookHandler
  def process(payload)
    if payload[:size] > payload[:commits].length
      # commits が切り詰められている
      # Webhook だけでは全変更を把握できない
      logger.warn("commits truncated: #{payload[:size]} commits, only #{payload[:commits].length} received")
      trigger_full_sync(payload)
      return
    end

    process_commits(payload[:commits])
  end
end

class PushWebhookHandler
  MAX_FILES_THRESHOLD = 300

  def process_commits(commits)
    total_files = commits.sum { |c| c[:added].length + c[:modified].length + c[:removed].length }

    if total_files >= MAX_FILES_THRESHOLD
      # ファイル数が多すぎる場合、切り詰めの可能性がある
      # Compare API で正確な差分を取得する
      trigger_full_sync(payload)
      return
    end

    sync_files(commits)
  end
end

class ContentFetcher
  def fetch(path:, ref:)
    client.contents(repo, path: path, ref: ref)
  rescue Octokit::Forbidden => e
    if e.message.include?("too large")
      # 1MB 超のファイルは Blob API で取得する
      fetch_via_blob(path: path, ref: ref)
    else
      raise
    end
  end

  private

  def fetch_via_blob(path:, ref:)
    # Tree API でファイルの SHA を取得
    tree = client.tree(repo, ref, recursive: true)
    item = tree[:tree].find { |t| t[:path] == path }
    return nil unless item

    # Blob API は 100MB まで対応
    blob = client.blob(repo, item[:sha])
    { path: path, content: blob[:content], sha: item[:sha] }
  end
end

class CommitMerger
  # 全コミットの変更を「最終的な差分」に集約する
  def merge(commits)
    file_states = {}

    commits.each do |commit|
      commit[:added].each do |path|
        state = file_states[path] ||= { existed_before: false }
        state[:exists_after] = true
        state[:timestamp] = commit[:timestamp]
      end

      commit[:modified].each do |path|
        state = file_states[path] ||= { existed_before: true }
        state[:exists_after] = true
        state[:timestamp] = commit[:timestamp]
      end

      commit[:removed].each do |path|
        state = file_states[path] ||= { existed_before: true }
        state[:exists_after] = false
      end
    end

    file_states
  end
end

class FileSyncer
  def sync(file_states, head_ref:, before_ref:)
    added    = file_states.select { |_, s| !s[:existed_before] && s[:exists_after] }
    modified = file_states.select { |_, s| s[:existed_before] && s[:exists_after] }
    removed  = file_states.select { |_, s| s[:existed_before] && !s[:exists_after] }
    # existed_before=false, exists_after=false は自動的に除外される

    # added と modified は HEAD から取得（同じ ref なのでまとめられる）
    paths_to_fetch = (added.keys + modified.keys)
    contents = fetcher.fetch_multiple(paths: paths_to_fetch, ref: head_ref)

    # removed は削除処理のみ（内容の取得は不要）
    removed.each_key { |path| delete_file(path) }

    persist(contents, file_states)
  end
end

class WebhooksController < ApplicationController
  def create
    # Webhook の受信は即座に 200 を返す
    WebhookProcessJob.perform_async(request.body.read, request.headers)
    head :ok
  end
end

class WebhookProcessJob
  include Sidekiq::Job
  sidekiq_options retry: 3

  def perform(payload_json, headers)
    payload = JSON.parse(payload_json, symbolize_names: true)
    PushWebhookHandler.new.process(payload)
  end
end

class WebhookProcessJob
  LOCK_TIMEOUT = 300 # 5 分

  def perform(payload_json, headers)
    payload = JSON.parse(payload_json, symbolize_names: true)
    repo = Repository.find_by(full_name: payload[:repository][:full_name])

    lock_acquired = repo.with_advisory_lock("webhook_push_#{repo.id}", timeout_seconds: LOCK_TIMEOUT) do
      PushWebhookHandler.new.process(payload)
    end

    raise "Lock timeout for repo #{repo.id}" unless lock_acquired
  end
end

class PushWebhookHandler
  def process(payload)
    # 切り詰め検出
    return trigger_full_sync(payload) if commits_truncated?(payload)

    # コミットのマージ
    file_states = CommitMerger.new.merge(payload[:commits])

    # ファイル数が多すぎる場合もフルスキャンに切り替え
    return trigger_full_sync(payload) if file_states.size > SYNC_THRESHOLD

    # 通常処理
    FileSyncer.new.sync(file_states, head_ref: payload[:after], before_ref: payload[:before])
  end

  private

  def commits_truncated?(payload)
    payload[:size] > payload[:commits].length
  end

  def trigger_full_sync(payload)
    FullSyncJob.perform_async(payload[:repository][:full_name], payload[:after])
    logger.info("Triggered full sync for #{payload[:repository][:full_name]}")
  end
end