package records

import (
	"context"
	"fmt"
	"sort"

	"github.com/nexuscrm/kernel/pkg/recordlocks"
)

type recordLifecycleBatchAttemptKey struct{}

type recordLifecycleBatchGroup struct {
	ctx    context.Context
	bind   *recordLifecycleBatchBinding
	events []recordLifecycleBatchEvent
}

type recordLifecycleBatchAttempt struct {
	groups   map[*recordLifecycleBatchBinding]*recordLifecycleBatchGroup
	order    []*recordLifecycleBatchBinding
	flushing bool
}

func (ps *PersistenceService) runWithRecordLifecycleBatchAttempt(ctx context.Context, fn func(context.Context) error) error {
	if fn == nil {
		return fmt.Errorf("record lifecycle batch transaction callback is required")
	}
	if recordLifecycleBatchAttemptFromContext(ctx) != nil {
		return fn(ctx)
	}
	attempt := &recordLifecycleBatchAttempt{groups: make(map[*recordLifecycleBatchBinding]*recordLifecycleBatchGroup)}
	attemptCtx := context.WithValue(ctx, recordLifecycleBatchAttemptKey{}, attempt)
	if err := fn(attemptCtx); err != nil {
		return err
	}
	if len(attempt.order) == 0 {
		return nil
	}
	return ps.runParentLockCoordinatorIfActive(attemptCtx, func(flushCtx context.Context) error {
		return attempt.flush(flushCtx)
	})
}

func stageRecordLifecycleBatchEvent(ctx context.Context, binding *recordLifecycleBatchBinding, event recordLifecycleBatchEvent) error {
	if binding == nil || binding.hook == nil {
		return fmt.Errorf("record lifecycle batch binding is unavailable")
	}
	attempt := recordLifecycleBatchAttemptFromContext(ctx)
	if attempt == nil {
		return fmt.Errorf("record lifecycle batch capability %q requires a records-owned transaction boundary", binding.capability)
	}
	if attempt.flushing {
		return fmt.Errorf("record lifecycle capability %q cannot stage during batch flush", binding.capability)
	}
	group := attempt.groups[binding]
	if group == nil {
		group = &recordLifecycleBatchGroup{ctx: ctx, bind: binding}
		attempt.groups[binding] = group
		attempt.order = append(attempt.order, binding)
	}
	group.events = append(group.events, event)
	return nil
}

func (a *recordLifecycleBatchAttempt) flush(flushCtx context.Context) error {
	if a == nil {
		return nil
	}
	a.flushing = true
	order := append([]*recordLifecycleBatchBinding(nil), a.order...)
	sort.Slice(order, func(i, j int) bool {
		return order[i].capability < order[j].capability
	})
	for _, binding := range order {
		group := a.groups[binding]
		if group == nil || len(group.events) == 0 {
			continue
		}
		hookCtx := group.ctx
		if scope := recordlocks.ScopeFromContext(flushCtx); scope != nil {
			hookCtx = recordlocks.WithExistingScope(hookCtx, scope)
		}
		if err := group.bind.hook(hookCtx, append([]recordLifecycleBatchEvent(nil), group.events...)); err != nil {
			return err
		}
	}
	return nil
}

func recordLifecycleBatchAttemptFromContext(ctx context.Context) *recordLifecycleBatchAttempt {
	if ctx == nil {
		return nil
	}
	attempt, _ := ctx.Value(recordLifecycleBatchAttemptKey{}).(*recordLifecycleBatchAttempt)
	return attempt
}