import { EventEmitter } from '@libp2p/interface/events'
import { logger } from '@libp2p/logger'
import { PeerMap } from '@libp2p/peer-collections'
import { multiaddr } from '@multiformats/multiaddr'
import { pbStream } from 'it-pb-stream'
import { PeerJobQueue } from '../../utils/peer-job-queue.js'
import { DEFAULT_RESERVATION_CONCURRENCY, RELAY_TAG, RELAY_V2_HOP_CODEC } from '../constants.js'
import { HopMessage, Status } from '../pb/index.js'
import { getExpirationMilliseconds } from '../utils.js'
import type { Reservation } from '../pb/index.js'
import type { Libp2pEvents } from '@libp2p/interface'
import type { Connection } from '@libp2p/interface/connection'
import type { PeerId } from '@libp2p/interface/peer-id'
import type { PeerStore } from '@libp2p/interface/peer-store'
import type { Startable } from '@libp2p/interface/startable'
import type { ConnectionManager } from '@libp2p/interface-internal/connection-manager'
import type { TransportManager } from '@libp2p/interface-internal/transport-manager'

const log = logger('libp2p:circuit-relay:transport:reservation-store')

// allow refreshing a relay reservation if it will expire in the next 10 minutes
const REFRESH_WINDOW = (60 * 1000) * 10

// try to refresh relay reservations 5 minutes before expiry
const REFRESH_TIMEOUT = (60 * 1000) * 5

// minimum duration before which a reservation must not be refreshed
const REFRESH_TIMEOUT_MIN = 30 * 1000

export interface RelayStoreComponents {
  peerId: PeerId
  connectionManager: ConnectionManager
  transportManager: TransportManager
  peerStore: PeerStore
  events: EventEmitter<Libp2pEvents>
}

export interface RelayStoreInit {
  /**
   * Multiple relays may be discovered simultaneously - to prevent listening
   * on too many relays, this value controls how many to attempt to reserve a
   * slot on at once. If set to more than one, we may end up listening on
   * more relays than the `maxReservations` value, but on networks with poor
   * connectivity the user may wish to attempt to reserve on multiple relays
   * simultaneously. (default: 1)
   */
  reservationConcurrency?: number

  /**
   * How many discovered relays to allow in the reservation store
   */
  discoverRelays?: number

  /**
   * Limit the number of potential relays we will dial (default: 100)
   */
  maxReservationQueueLength?: number
}

export type RelayType = 'discovered' | 'configured'

interface RelayEntry {
  timeout: ReturnType<typeof setTimeout>
  type: RelayType
  reservation: Reservation
}

export interface ReservationStoreEvents {
  'relay:not-enough-relays': CustomEvent
  'relay:removed': CustomEvent<PeerId>
}

export class ReservationStore extends EventEmitter<ReservationStoreEvents> implements Startable {
  private readonly peerId: PeerId
  private readonly connectionManager: ConnectionManager
  private readonly transportManager: TransportManager
  private readonly peerStore: PeerStore
  private readonly events: EventEmitter<Libp2pEvents>
  private readonly reserveQueue: PeerJobQueue
  private readonly reservations: PeerMap<RelayEntry>
  private readonly maxDiscoveredRelays: number
  private readonly maxReservationQueueLength: number
  private started: boolean

  constructor (components: RelayStoreComponents, init?: RelayStoreInit) {
    super()

    this.peerId = components.peerId
    this.connectionManager = components.connectionManager
    this.transportManager = components.transportManager
    this.peerStore = components.peerStore
    this.events = components.events
    this.reservations = new PeerMap()
    this.maxDiscoveredRelays = init?.discoverRelays ?? 0
    this.maxReservationQueueLength = init?.maxReservationQueueLength ?? 100
    this.started = false

    // ensure we don't listen on multiple relays simultaneously
    this.reserveQueue = new PeerJobQueue({
      concurrency: init?.reservationConcurrency ?? DEFAULT_RESERVATION_CONCURRENCY
    })

    // When a peer disconnects, if we had a reservation on that peer
    // remove the reservation and multiaddr and maybe trigger search
    // for new relays
    this.events.addEventListener('peer:disconnect', (evt) => {
      this.#removeRelay(evt.detail)
    })
  }

  isStarted (): boolean {
    return this.started
  }

  async start (): Promise<void> {
    this.started = true
  }

  async stop (): Promise<void> {
    this.reservations.forEach(({ timeout }) => {
      clearTimeout(timeout)
    })
    this.reservations.clear()

    this.started = true
  }

  /**
   * If the number of current relays is beneath the configured `maxReservations`
   * value, and the passed peer id is not our own, and we have a non-relayed connection
   * to the remote, and the remote peer speaks the hop protocol, try to reserve a slot
   * on the remote peer
   */
  async addRelay (peerId: PeerId, type: RelayType): Promise<void> {
    if (this.peerId.equals(peerId)) {
      log('not trying to use self as relay')
      return
    }

    if (this.reserveQueue.size > this.maxReservationQueueLength) {
      log('not adding relay as the queue is full')
      return
    }

    if (this.reserveQueue.hasJob(peerId)) {
      log('relay peer is already in the reservation queue')
      return
    }

    log('add relay %p', peerId)

    await this.reserveQueue.add(async () => {
      try {
        // allow refresh of an existing reservation if it is about to expire
        const existingReservation = this.reservations.get(peerId)

        if (existingReservation != null) {
          if (getExpirationMilliseconds(existingReservation.reservation.expire) > REFRESH_WINDOW) {
            log('already have reservation on relay peer %p and it expires in more than 10 minutes', peerId)
            return
          }

          clearTimeout(existingReservation.timeout)
          this.reservations.delete(peerId)
        }

        if (type === 'discovered' && [...this.reservations.values()].reduce((acc, curr) => {
          if (curr.type === 'discovered') {
            acc++
          }

          return acc
        }, 0) >= this.maxDiscoveredRelays) {
          log('already have enough discovered relays')
          return
        }

        const connection = await this.connectionManager.openConnection(peerId)

        if (connection.remoteAddr.protoNames().includes('p2p-circuit')) {
          log('not creating reservation over relayed connection')
          return
        }

        const reservation = await this.#createReservation(connection)

        log('created reservation on relay peer %p', peerId)

        const expiration = getExpirationMilliseconds(reservation.expire)

        // sets a lower bound on the timeout, and also don't let it go over
        // 2^31 - 1 (setTimeout will only accept signed 32 bit integers)
        const timeoutDuration = Math.min(Math.max(expiration - REFRESH_TIMEOUT, REFRESH_TIMEOUT_MIN), Math.pow(2, 31) - 1)

        const timeout = setTimeout(() => {
          this.addRelay(peerId, type).catch(err => {
            log.error('could not refresh reservation to relay %p', peerId, err)
          })
        }, timeoutDuration)

        // we've managed to create a reservation successfully
        this.reservations.set(peerId, {
          timeout,
          reservation,
          type
        })

        // ensure we don't close the connection to the relay
        await this.peerStore.merge(peerId, {
          tags: {
            [RELAY_TAG]: {
              value: 1,
              ttl: expiration
            }
          }
        })

        // listen on multiaddr that only the circuit transport is listening for
        await this.transportManager.listen([multiaddr(`/p2p/${peerId.toString()}/p2p-circuit`)])
      } catch (err) {
        log.error('could not reserve slot on %p', peerId, err)

        // if listening failed, remove the reservation
        this.reservations.delete(peerId)
      }
    }, {
      peerId
    })
  }

  hasReservation (peerId: PeerId): boolean {
    return this.reservations.has(peerId)
  }

  getReservation (peerId: PeerId): Reservation | undefined {
    return this.reservations.get(peerId)?.reservation
  }

  async #createReservation (connection: Connection): Promise<Reservation> {
    log('requesting reservation from %p', connection.remotePeer)
    const stream = await connection.newStream(RELAY_V2_HOP_CODEC)
    const pbstr = pbStream(stream)
    const hopstr = pbstr.pb(HopMessage)
    hopstr.write({ type: HopMessage.Type.RESERVE })

    let response: HopMessage

    try {
      response = await hopstr.read()
    } catch (err: any) {
      log.error('error parsing reserve message response from %p because', connection.remotePeer, err)
      throw err
    } finally {
      stream.close()
    }

    if (response.status === Status.OK && (response.reservation != null)) {
      return response.reservation
    }

    const errMsg = `reservation failed with status ${response.status ?? 'undefined'}`
    log.error(errMsg)

    throw new Error(errMsg)
  }

  /**
   * Remove listen relay
   */
  #removeRelay (peerId: PeerId): void {
    const existingReservation = this.reservations.get(peerId)

    if (existingReservation == null) {
      return
    }

    log('connection to relay %p closed, removing reservation from local store', peerId)

    clearTimeout(existingReservation.timeout)
    this.reservations.delete(peerId)

    this.safeDispatchEvent('relay:removed', { detail: peerId })

    if (this.reservations.size < this.maxDiscoveredRelays) {
      log('not enough relays %d/%d', this.reservations.size, this.maxDiscoveredRelays)
      this.safeDispatchEvent('relay:not-enough-relays', {})
    }
  }
}
