Agregate on agregate filter in Logstash?

I have a stock table where I have stocks for shops/products:

input {
  jdbc {
    statement => "SELECT ShopId, ProductCode, Quantity FROM stock ORDER BY productcode;"
  }
}

then I have a simple filter to aggregate that data:

filter {
     aggregate {
       task_id => "%{productcode}"
       code => "
         map['productcode'] ||= event.get('productcode')
         map['objectID'] ||= event.get('productcode')
         map['stocks'] ||= []
         map['stocks'] << {
            'ShopId' => event.get('ShopId'), 
            'quantity' => event.get('quantity'),
    }
         event.cancel()
       "
       push_previous_map_as_event => true
       timeout => 3
     }
}

which gives me output I expect, for example:

{
    "productcode": "123", 
    "objectID": "123", 
    "stocks": [
        {
            "ShopId": 1
            "Quantity": 2
        },
        {
            "ShopId": 2
            "Quantity": 5
        }
    ] 
}

now I can push that data to Algolia via http output plugin. But the issue I have is that it's thousands of objects which makes thousands of calls.

That's why I think to use batch endpoint, pack those objects to package of f.e. 1000 objects, but to do so, I need to adjust structure to:

{
  "requests": [
    {
      "action": "addObject",
      "body": {
                "productcode": "123", 
                "objectID": "123", 
                ...
       }
    },
    {
        "action": "addObject",
        "body": {
                "productcode": "456", 
                "objectID": "456", 
                ...
         }
      }
  ]
}

which looks to me like another aggregate function, but I already tried:

 aggregate {
   task_id => "%{source}"
   code => "
     map['requests'] ||= []
     map['requests'] << {
    'action' => 'addObject',
    'body' => {
        'productcode' => event.get('productcode'),
        'objectId' => event.get('objectID'),
        'stocks' => event.get('stocks')
    }
 }
    event.cancel()
   "
   push_previous_map_as_event => true
   timeout => 3

but it does not work.

Also with this type of aggregate function I'm not able to configure how big packages I would like to send to batch output.

I will be very grateful for any help or clues.