metadata/utils/innertube-pool.ts

import { Innertube } from 'youtubei.js'

interface InnertubePoolSlot {
  client: Promise<Innertube>
  busy?: boolean
}

export interface InnertubeLease {
  innertube: Innertube
  refresh: () => Promise<Innertube>
  release: () => void
}

export function createInnertubePool(size = 4) {
  const pool = Array.from({ length: Math.max(1, size) }, () => ({
    client: Innertube.create()
  })) as InnertubePoolSlot[]
  const waiters: Array<(lease: InnertubeLease) => void> = []

  function getSlot(index: number) {
    const slot = pool[index]

    if (!slot) {
      throw new Error(`Innertube slot ${index} not found`)
    }

    return slot
  }

  async function load(index: number) {
    const slot = getSlot(index)

    try {
      return await slot.client
    } catch {
      slot.client = Innertube.create()
      return await slot.client
    }
  }

  async function refresh(index: number) {
    const slot = getSlot(index)

    slot.client = Innertube.create()
    return await slot.client
  }

  function release(index: number) {
    const next = waiters.shift()

    if (next) {
      void load(index).then((innertube) => next(createLease(index, innertube)))
      return
    }

    delete getSlot(index).busy
  }

  function createLease(index: number, innertube: Innertube): InnertubeLease {
    const lease: InnertubeLease = {
      innertube,
      refresh: async () => {
        lease.innertube = await refresh(index)
        return lease.innertube
      },
      release: () => release(index)
    }

    return lease
  }

  async function acquire() {
    const index = pool.findIndex((slot) => !slot.busy)

    if (index === -1) {
      return await new Promise<InnertubeLease>((resolve) => {
        waiters.push(resolve)
      })
    }

    getSlot(index).busy = true
    return createLease(index, await load(index))
  }

  async function use<T>(callback: (innertube: Innertube) => Promise<T>, retries = 5) {
    const lease = await acquire()

    try {
      for (let attempt = 0; attempt < retries; attempt++) {
        try {
          return await callback(lease.innertube)
        } catch (error) {
          if (attempt === retries - 1) {
            throw error
          }

          await lease.refresh()
        }
      }

      throw new Error('Innertube retries exhausted')
    } finally {
      lease.release()
    }
  }

  return {
    acquire,
    use
  }
}