# Matchmaking via Redis.
#
# Data structures (all volatile, TTL-backed):
#   mm:queue   sorted set  — waiting players, score = rating
#   mm:since   hash        — user_id => enqueue timestamp (for dynamic rating range)
#   mm:rating  hash        — user_id => rating (to requeue stale pending players)
#   mm:pending sorted set  — paired, awaiting finalize, score = pairing timestamp
#   mm:match:<id>  string  — match payload (JSON)
#   mm:user:<id>   string  — per-user pointer to the match payload (JSON)
#
# All state transitions happen inside Lua scripts, so every operation is
# atomic with respect to other Redis clients. Delivery to clients happens
# over Action Cable; the per-user pointer key guarantees at-most-once
# delivery even if the broadcast is missed (client reconnects and claims it).
class MatchmakingService
  QUEUE_KEY = "mm:queue"
  SINCE_KEY = "mm:since"
  RATING_KEY = "mm:rating"
  PENDING_KEY = "mm:pending"
  MATCH_KEY_PREFIX = "mm:match:"
  USER_KEY_PREFIX = "mm:user:"

  # Enqueues a player and tries to find an opponent immediately; in "sweep"
  # mode pairs up everyone in the queue, expires timed-out waiters and
  # requeues stale pending players. Returns an array of result strings:
  #   "P:<id1>:<id2>" — a pair was formed
  #   "E:<id>"        — a player timed out waiting
  PAIR_SCRIPT = <<~LUA
    local queue_key   = KEYS[1]
    local since_key   = KEYS[2]
    local rating_key  = KEYS[3]
    local pending_key = KEYS[4]
    local user_prefix = KEYS[5]

    -- All timestamps are integer milliseconds: Lua turns large floats into
    -- scientific notation strings which Redis rejects in ZADD/ZRANGEBYSCORE.
    local mode            = ARGV[1]
    local now             = tonumber(ARGV[2])
    local queue_ttl       = tonumber(ARGV[3]) * 1000
    local pending_ttl     = tonumber(ARGV[4]) * 1000
    local base_range      = tonumber(ARGV[5])
    local range_per_sec   = tonumber(ARGV[6])
    local max_range       = tonumber(ARGV[7])

    local function tolerance(ts)
      local waited = (now - ts) / 1000
      local r = base_range + waited * range_per_sec
      return r > max_range and max_range or r
    end

    local results = {}

    if mode == "enqueue" then
      local uid    = ARGV[8]
      local rating = tonumber(ARGV[9])

      -- restart a fresh waiting session for this user
      redis.call("ZREM", queue_key, uid)
      redis.call("HDEL", since_key, uid)
      redis.call("HDEL", rating_key, uid)

      local candidates = redis.call("ZRANGEBYSCORE", queue_key,
        rating - max_range, rating + max_range)
      local best = nil
      local best_diff = nil

      for _, other in ipairs(candidates) do
        if other ~= uid then
          local other_ts = tonumber(redis.call("HGET", since_key, other) or now)
          local other_rating = tonumber(redis.call("ZSCORE", queue_key, other))
          local diff = math.abs(other_rating - rating)
          if diff <= tolerance(other_ts) and (best_diff == nil or diff < best_diff) then
            best = other
            best_diff = diff
          end
        end
      end

      if best then
        redis.call("ZREM", queue_key, uid, best)
        redis.call("HDEL", since_key, uid, best)
        redis.call("HDEL", rating_key, uid, best)
        redis.call("ZADD", pending_key, now, uid, now, best)
        results[#results + 1] = "P:" .. uid .. ":" .. best
      else
        redis.call("ZADD", queue_key, rating, uid)
        redis.call("HSET", since_key, uid, now)
        redis.call("HSET", rating_key, uid, rating)
      end
    else -- "sweep"
      local all = redis.call("ZRANGE", queue_key, 0, -1, "WITHSCORES")
      local items = {}
      for i = 1, #all, 2 do
        items[#items + 1] = {
          id = all[i],
          rating = tonumber(all[i + 1]),
          ts = tonumber(redis.call("HGET", since_key, all[i]) or now)
        }
      end
      table.sort(items, function(a, b) return a.rating < b.rating end)

      local used = {}
      for i = 1, #items do
        local a = items[i]
        if not used[a.id] then
          if now - a.ts > queue_ttl then
            used[a.id] = true
            results[#results + 1] = "E:" .. a.id
          else
            local best = nil
            for j = i + 1, #items do
              local b = items[j]
              -- tolerance is based on the longer-waiting player of the pair
            if not used[b.id] and (b.rating - a.rating) <= tolerance(math.min(a.ts, b.ts)) then
                best = b
                break
              end
            end
            if best then
              used[a.id] = true
              used[best.id] = true
              results[#results + 1] = "P:" .. a.id .. ":" .. best.id
            end
          end
        end
      end

      for _, res in ipairs(results) do
        local parts = {}
        for part in string.gmatch(res, "[^:]+") do parts[#parts + 1] = part end
        if parts[1] == "P" then
          redis.call("ZREM", queue_key, parts[2], parts[3])
          redis.call("HDEL", since_key, parts[2], parts[3])
          redis.call("HDEL", rating_key, parts[2], parts[3])
          redis.call("ZADD", pending_key, now, parts[2], now, parts[3])
        else
          redis.call("ZREM", queue_key, parts[2])
          redis.call("HDEL", since_key, parts[2])
          redis.call("HDEL", rating_key, parts[2])
        end
      end

      -- requeue players whose pairing was never finalized (e.g. the Rails
      -- process died between pairing and match creation)
      local stale = redis.call("ZRANGEBYSCORE", pending_key, 0, now - pending_ttl)
      for _, uid in ipairs(stale) do
        if redis.call("EXISTS", user_prefix .. uid) == 0 then
          local r = redis.call("HGET", rating_key, uid)
          if r then
            redis.call("ZADD", queue_key, tonumber(r), uid)
            redis.call("HSET", since_key, uid, now)
          end
        end
        redis.call("ZREM", pending_key, uid)
      end
    end

    redis.call("EXPIRE", queue_key, math.floor(queue_ttl / 1000) * 4)
    redis.call("EXPIRE", pending_key, math.floor(queue_ttl / 1000) * 4)
    redis.call("EXPIRE", since_key, math.floor(queue_ttl / 1000) * 4)
    redis.call("EXPIRE", rating_key, math.floor(queue_ttl / 1000) * 4)
    return results
  LUA

  # Atomically stores the match payload and clears the players' pending state.
  FINALIZE_SCRIPT = <<~LUA
    redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) -- match payload
    redis.call("SET", KEYS[2], ARGV[1], "EX", ARGV[2]) -- pointer: player 1
    redis.call("SET", KEYS[3], ARGV[1], "EX", ARGV[2]) -- pointer: player 2
    redis.call("ZREM", KEYS[4], ARGV[3], ARGV[4])      -- pending
    redis.call("HDEL", KEYS[5], ARGV[3], ARGV[4])      -- rating
    redis.call("HDEL", KEYS[6], ARGV[3], ARGV[4])      -- since
    return 1
  LUA

  CANCEL_SCRIPT = <<~LUA
    redis.call("ZREM", KEYS[1], ARGV[1])
    redis.call("ZREM", KEYS[2], ARGV[1])
    redis.call("HDEL", KEYS[3], ARGV[1])
    redis.call("HDEL", KEYS[4], ARGV[1])
    redis.call("DEL",  KEYS[5] .. ARGV[1])
    return 1
  LUA

  class << self
    # Adds the user to the queue and immediately tries to pair them.
    # The result (match or timeout) is delivered asynchronously via
    # MatchmakingChannel broadcasts.
    def enqueue(user_id, rating:)
      results = call_pair_script("enqueue", extra: [ user_id, rating ])
      process_results(results)
    end

    # Pairs up everyone possible, expires timed-out waiters, requeues stale
    # pending players. Called periodically by MatchmakingSweepJob.
    def sweep
      results = call_pair_script("sweep")
      process_results(results)
    end

    # Removes the user from the queue / pending state.
    def cancel(user_id)
      RedisConnection.with do |redis|
        redis.eval(CANCEL_SCRIPT, keys: [ QUEUE_KEY, PENDING_KEY, SINCE_KEY, RATING_KEY, USER_KEY_PREFIX ],
                                  argv: [ user_id ])
      end
    end

    # Atomically reads and clears the match pointer for the user.
    # Used on channel subscribe to deliver a match that was found while
    # the user had no active subscription (at-most-once delivery).
    def claim(user_id)
      payload = RedisConnection.with { |redis| redis.getdel(user_key(user_id)) }
      JSON.parse(payload, symbolize_names: true) if payload
    end

    def config
      Rails.application.config.x.matchmaking
    end

    private

    def call_pair_script(mode, extra: [])
      RedisConnection.with do |redis|
        redis.eval(PAIR_SCRIPT,
          keys: [ QUEUE_KEY, SINCE_KEY, RATING_KEY, PENDING_KEY, USER_KEY_PREFIX ],
          argv: [ mode, (Time.current.to_f * 1000).to_i, config.queue_ttl, config.pending_ttl,
                  config.base_range, config.range_per_second, config.max_range, *extra ])
      end
    end

    def process_results(results)
      results.each do |result|
        kind, id1, id2 = result.split(":")
        if kind == "P"
          create_match(id1.to_i, id2.to_i)
        else
          broadcast(id1.to_i, { type: "timeout" })
        end
      end
    end

    # JWT generation happens in Ruby, so pairing (done in Lua) and match
    # creation are two steps. Players sit in the "pending" set between them;
    # the sweep requeues anyone stuck there longer than pending_ttl.
    def create_match(player1_id, player2_id)
      match_id = SecureRandom.uuid
      game_token = JwtService.encode(
        match_id: match_id,
        player1_id: player1_id,
        player2_id: player2_id,
        game_server: config.game_server_url,
        exp: config.token_ttl.seconds.from_now.to_i
      )

      payload = {
        type: "matched",
        match_id: match_id,
        game_token: game_token,
        game_server: config.game_server_url,
        players: [ player1_id, player2_id ]
      }.to_json

      RedisConnection.with do |redis|
        redis.eval(FINALIZE_SCRIPT,
          keys: [ match_key(match_id), user_key(player1_id), user_key(player2_id),
                  PENDING_KEY, RATING_KEY, SINCE_KEY ],
          argv: [ payload, config.match_ttl, player1_id, player2_id ])
      end

      [ player1_id, player2_id ].each do |uid|
        # Deliver and clear the pointer so a later subscribe-claim
        # cannot duplicate the match.
        pointer = RedisConnection.with { |redis| redis.getdel(user_key(uid)) }
        broadcast(uid, JSON.parse(pointer, symbolize_names: true)) if pointer
      end
    end

    def broadcast(user_id, payload)
      MatchmakingChannel.broadcast_to(user_id, payload)
    end

    def match_key(match_id) = "#{MATCH_KEY_PREFIX}#{match_id}"
    def user_key(user_id) = "#{USER_KEY_PREFIX}#{user_id}"
  end
end