///|
priv struct Workers {
  available : @async.Queue[Connection]
  connections : Array[Connection]
  mut closed : Bool
}

///|
pub struct Pool {
  priv sql : @sql.Database[Value, Row, Command, TransactionOptions]
}

///|
/// Physical connections are established on foreign workers. charset, time_zone
/// and CLIENT_FOUND_ROWS are explicit MySQL policies, reapplied after recycling.
pub async fn Pool::new(
  host~ : String,
  user~ : String,
  password~ : String,
  database~ : String,
  port? : Int = 3306,
  size? : Int = 10,
  ssl_ca? : String = "",
  plugin_dir? : String = "",
  timeout_seconds? : Int = 5,
  max_rows? : Int = 10000,
  max_bytes? : Int = 16777216,
  max_waiters? : Int = 128,
  checkout_timeout_ms? : Int = 5000,
  charset? : String = "utf8mb4",
  time_zone? : String = "+00:00",
  found_rows? : Bool = true,
) -> Pool {
  guard size > 0 &&
    size <= 128 &&
    port > 0 &&
    port <= 65535 &&
    timeout_seconds > 0 &&
    timeout_seconds <= 3600 &&
    max_rows > 0 &&
    max_rows <= 1000000 &&
    max_bytes > 0 else {
    raise InvalidConfig("Invalid pool limits")
  }
  guard !host.is_empty() &&
    !database.is_empty() &&
    [host, user, password, database, ssl_ca, plugin_dir].all(s => {
      !s.contains("\u0000")
    }) &&
    (charset == "utf8mb4" || charset == "utf8mb3" || charset == "utf8") &&
    time_zone.length() > 0 &&
    time_zone.length() <= 64 &&
    time_zone
    .iter()
    .all(c => {
      c
      is ('a'..='z' | 'A'..='Z' | '0'..='9' | '_' | '/' | ':' | '+' | '-' | '.')
    }) else {
    raise InvalidConfig("Invalid connection settings")
  }
  let workers : Workers = {
    available: @async.Queue(kind=Unbounded),
    connections: [],
    closed: false,
  }
  errdefer workers.close()
  for _ in 0..= 0 else { raise WorkerUnavailable }
    let fd = {
      // RawFd owns the read end even when its constructor raises.
      errdefer c_db_close(db)
      @raw_fd.RawFd(c_db_fd(db))
    }
    let conn : Connection = {
      db,
      fd,
      leased: false,
      closed: false,
      broken: false,
    }
    workers.connections.push(conn)
    ignore(workers.available.try_put(conn) catch { _ => false })
  }
  let sql = @sql.Database::new(
    {
      acquire: async fn() {
        let conn = workers.available.get()
        conn.leased = true
        conn
      },
      close: fn() { workers.close() },
      release: async fn(conn, discard) {
        defer {
          conn.leased = false
          if conn.broken {
            workers.close()
          }
          if workers.closed {
            conn.close()
          } else {
            ignore(workers.available.try_put(conn) catch { _ => false })
          }
        }
        if !workers.closed {
          conn.recycle(discard)
        }
      },
      executor: @sql.Combined(async fn(conn, sql, params) {
        conn.run(sql, params)
      }),
      begin: async fn(conn, options : TransactionOptions) {
        if options.isolation is Some(isolation) {
          let name = match isolation {
            ReadUncommitted => "READ UNCOMMITTED"
            ReadCommitted => "READ COMMITTED"
            RepeatableRead => "REPEATABLE READ"
            Serializable => "SERIALIZABLE"
          }
          conn.control("SET TRANSACTION ISOLATION LEVEL " + name)
        }
        let sql = match options.read_only {
          Some(true) => "START TRANSACTION READ ONLY"
          Some(false) => "START TRANSACTION READ WRITE"
          None => "START TRANSACTION"
        }
        conn.control(sql)
      },
      commit: async fn(conn) { conn.control("COMMIT") },
      rollback: async fn(conn) {
        // A failed statement closes the physical connection in the C worker.
        // Rollback must not establish a replacement session within this lease.
        if c_db_connected(conn.db) {
          conn.control("ROLLBACK")
        }
      },
    },
    max_leases=size,
    max_waiters~,
    checkout_timeout_ms~,
  )
  { sql, }
}

///|
fn Workers::close(self : Workers) -> Unit {
  if !self.closed {
    self.closed = true
    self.available.close(error=@sql.Closed, clear=true)
    for conn in self.connections {
      if !conn.leased {
        conn.close()
      }
    }
  }
}

///|
pub fn Pool::database(
  self : Pool,
) -> @sql.Database[Value, Row, Command, TransactionOptions] {
  self.sql
}

///|
pub fn Pool::close(self : Pool) -> Unit {
  self.sql.close()
}

///|
/// Compatibility helper. New consumers can use database().query/execute/run.
pub async fn Pool::query(
  self : Pool,
  sql : String,
  params? : Array[Value] = [],
) -> QueryResult {
  legacy_result(self.sql.run(sql, params~))
}

///|
pub async fn Pool::transaction(
  self : Pool,
  statements : Array[Statement],
) -> QueryResult {
  guard !statements.is_empty() else { raise InvalidParameter }
  self.sql.with_transaction(TransactionOptions::new(), async fn(tx) {
    let mut result = tx.run(statements[0].sql, params=statements[0].params)
    for i in 1.. QueryResult {
  {
    rows: result.rows,
    has_rows: result.metadata.has_rows,
    affected_rows: result.metadata.affected_rows,
    insert_id: result.metadata.insert_id,
  }
}