hyapp-server/pkg/outboxpartition/partitions.go

// Package outboxpartition provides the shared tenant scheduling rules used by
// owner-service outbox workers. It does not read or publish business facts;
// repositories retain ownership of indexed claims and delivery state.
package outboxpartition

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"hyapp/pkg/appcode"
)

const DefaultDiscoveryRefreshInterval = 5 * time.Minute

// Partitions resolves the App partitions visible to one outbox lane and rotates
// the first partition across claims. A configured allowlist is immutable for the
// process lifetime; an empty allowlist falls back to low-frequency DB discovery.
type Partitions struct {
	configured      []string
	refreshInterval time.Duration

	mu          sync.Mutex
	cached      []string
	hasCache    bool
	nextRefresh time.Time
	cursor      atomic.Uint64
}

// New creates a partition scheduler. Operators can stage a backlog safely by
// configuring only the Apps currently allowed to publish; an empty list keeps
// automatic discovery as the compatibility default for existing deployments.
func New(configured []string, refreshInterval time.Duration) *Partitions {
	if refreshInterval <= 0 {
		refreshInterval = DefaultDiscoveryRefreshInterval
	}
	return &Partitions{
		configured:      Normalize(configured, false),
		refreshInterval: refreshInterval,
	}
}

// Resolve returns configured Apps without touching MySQL. In discovery mode it
// refreshes at most once per interval across all local worker goroutines. A
// failed refresh returns the last successful list alongside the error so the
// caller can log the stale condition without stopping already-known partitions.
func (p *Partitions) Resolve(ctx context.Context, discover func(context.Context) ([]string, error)) ([]string, error) {
	if p == nil {
		return nil, nil
	}
	if len(p.configured) > 0 {
		return append([]string(nil), p.configured...), nil
	}
	if discover == nil {
		return nil, nil
	}

	p.mu.Lock()
	defer p.mu.Unlock()
	now := time.Now().UTC()
	if p.hasCache && now.Before(p.nextRefresh) {
		return append([]string(nil), p.cached...), nil
	}

	discovered, err := discover(ctx)
	if err != nil {
		// Move the next attempt out by the normal interval: a transient database
		// error must not turn every hot poll into a DISTINCT query stampede.
		p.nextRefresh = now.Add(p.refreshInterval)
		if p.hasCache {
			return append([]string(nil), p.cached...), err
		}
		return nil, err
	}
	p.cached = Normalize(discovered, true)
	p.hasCache = true
	p.nextRefresh = now.Add(p.refreshInterval)
	return append([]string(nil), p.cached...), nil
}

// Order rotates the first App atomically while preserving the remaining cyclic
// order. Multiple worker goroutines therefore share one fair cursor instead of
// independently preferring the first configured tenant.
func (p *Partitions) Order(appCodes []string) []string {
	if len(appCodes) == 0 {
		return nil
	}
	if p == nil {
		return append([]string(nil), appCodes...)
	}
	start := int((p.cursor.Add(1) - 1) % uint64(len(appCodes)))
	ordered := make([]string, 0, len(appCodes))
	for offset := 0; offset < len(appCodes); offset++ {
		ordered = append(ordered, appCodes[(start+offset)%len(appCodes)])
	}
	return ordered
}

// Normalize canonicalizes tenant keys, removes blanks and duplicates, and can
// sort DB-discovered keys for deterministic rotation. Explicit configuration
// preserves operator order so staged rollouts remain readable and predictable.
func Normalize(values []string, sorted bool) []string {
	normalized := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		if strings.TrimSpace(value) == "" {
			continue
		}
		value = appcode.Normalize(value)
		if _, exists := seen[value]; exists {
			continue
		}
		seen[value] = struct{}{}
		normalized = append(normalized, value)
	}
	if sorted {
		sort.Strings(normalized)
	}
	return normalized
}

// ClaimFair gives every App one bounded first-pass quota, then lets non-empty
// partitions use capacity left by empty queues. Each callback remains App-scoped
// and is expected to perform its own indexed SKIP LOCKED transaction.
func ClaimFair[T any](ctx context.Context, orderedApps []string, limit int, claim func(context.Context, string, int) ([]T, error)) ([]T, error) {
	if limit <= 0 || len(orderedApps) == 0 || claim == nil {
		return nil, nil
	}
	quota := (limit + len(orderedApps) - 1) / len(orderedApps)
	claimed := make([]T, 0, limit)
	for pass := 0; pass < 2 && len(claimed) < limit; pass++ {
		for _, app := range orderedApps {
			remaining := limit - len(claimed)
			if remaining <= 0 {
				break
			}
			appLimit := remaining
			if pass == 0 && appLimit > quota {
				appLimit = quota
			}
			records, err := claim(ctx, app, appLimit)
			if err != nil {
				return nil, err
			}
			if len(records) > appLimit {
				// Silently truncating would strand already-claimed rows until lease
				// expiry. Treat repository limit violations as explicit worker errors.
				return nil, fmt.Errorf("outbox claim for app %q returned %d records above limit %d", app, len(records), appLimit)
			}
			claimed = append(claimed, records...)
		}
		if pass == 0 && len(claimed) == 0 {
			// A globally empty first pass must stop here; a second identical pass
			// only doubles indexed polling queries and cannot discover new work.
			break
		}
	}
	return claimed, nil
}