///|
/// WorkerPool manages a group of concurrent workers that process messages from the queue.
pub struct WorkerPool[T] {
  queue : RelayQueue[T]
  concurrency : Int
}

///|
/// new creates a new worker pool with the specified concurrency.
pub fn[T] WorkerPool::new(
  queue : RelayQueue[T],
  concurrency~ : Int,
) -> WorkerPool[T] {
  { queue, concurrency }
}

///|
/// run starts the pool. Each worker is launched as a background task within a TaskGroup.
pub async fn[T] WorkerPool::run(
  self : WorkerPool[T],
  handler : async (T) -> Unit raise Error,
) -> Unit {
  @async.with_task_group(tg => {
    for i = 0; i < self.concurrency; i = i + 1 {
      tg.spawn_bg(async fn() { self.start_single_worker(handler) })
    }
  })
}

///|
async fn[T] WorkerPool::start_single_worker(
  self : WorkerPool[T],
  handler : async (T) -> Unit raise Error,
) -> Unit {
  for ;; {
    let msg = self.queue.pop() catch { _ => break }

    let result = try {
      handler(msg.payload)
      Ok(())
    } catch {
      e => Err(e)
    }

    match result {
      Ok(_) => {
        let _ = self.queue.ack(msg.id) catch { _ => () }
      }
      Err(_) => {
        let _ = self.queue.nack(msg.id) catch { _ => () }
      }
    }
  }
}