// Package bulk contains the adaptive, short-lived physical batching
// primitives shared by the legacy importer and the repository sync engine.
// Durable truth remains in Import staging rows or Sync RunItems respectively.
package bulk

import (
	"context"
	"fmt"

	"golang.org/x/sync/errgroup"
)

const (
	DefaultMaxItems = 500
	DefaultMaxBytes = int64(8 << 20)
)

type Limits struct {
	MaxItems int
	MaxBytes int64
}

func (l Limits) Normalize() (Limits, error) {
	if l.MaxItems == 0 {
		l.MaxItems = DefaultMaxItems
	}
	if l.MaxBytes == 0 {
		l.MaxBytes = DefaultMaxBytes
	}
	if l.MaxItems < 1 || l.MaxBytes < 1 {
		return Limits{}, fmt.Errorf("bulk limits must be positive")
	}
	return l, nil
}

// Shard groups items by count and serialized byte size. An individual item
// larger than MaxBytes occupies one shard so callers can report/provider-gate
// that object without stalling the rest of the run.
func Shard[T any](items []T, limits Limits, size func(T) int64) ([][]T, error) {
	limits, err := limits.Normalize()
	if err != nil {
		return nil, err
	}
	shards := make([][]T, 0)
	var current []T
	var currentBytes int64
	for _, item := range items {
		itemBytes := size(item)
		if itemBytes < 0 {
			return nil, fmt.Errorf("bulk item size cannot be negative")
		}
		if len(current) >= limits.MaxItems || (len(current) > 0 && currentBytes+itemBytes > limits.MaxBytes) {
			shards = append(shards, current)
			current, currentBytes = nil, 0
		}
		current = append(current, item)
		currentBytes += itemBytes
	}
	if len(current) > 0 {
		shards = append(shards, current)
	}
	return shards, nil
}

// ForEach runs an index-addressed physical batch with bounded concurrency and
// cancellation. It is shared by importer readers and provider API writers;
// durable per-object outcome remains in their own journal rows.
func ForEach(ctx context.Context, count, concurrency int, apply func(context.Context, int) error) error {
	if count < 0 || apply == nil {
		return fmt.Errorf("bulk bounded apply requires a non-negative count and callback")
	}
	if concurrency <= 0 {
		concurrency = 1
	}
	group, groupCtx := errgroup.WithContext(ctx)
	group.SetLimit(concurrency)
	for index := 0; index < count; index++ {
		index := index
		group.Go(func() error { return apply(groupCtx, index) })
	}
	return group.Wait()
}