platform/catalog/registry.go

// Registry is the per-tenant Schema lookup. A producer processes tenant X:
// it builds an event.Event with Source="catalog:<tenant>" and asks the
// Registry for X's Schema; the Builder applies that Schema to the source
// JSON. Unknown tenants get DefaultSchema — that makes the platform
// "just work" for new tenants while the operator rolls out the proper
// override file.
package catalog

import (
	"encoding/json"
	"fmt"
	"os"
	"sort"
	"sync"
)

// Registry is concurrency-safe: Register / SchemaFor can be called from any
// goroutine. SchemaFor is the hot path; it takes a read lock.
type Registry struct {
	mu      sync.RWMutex
	schemas map[string]Schema
	def     Schema // the "default" / DefaultSchema() fallback
}

// NewRegistry returns an empty Registry whose DefaultSchema fallback is
// DefaultSchema(). Use Register to add tenant-specific overrides.
func NewRegistry() *Registry {
	return &Registry{
		schemas: map[string]Schema{},
		def:     DefaultSchema(),
	}
}

// WithDefault swaps the DefaultSchema fallback (used when SchemaFor is
// called with an unknown tenant name, or with the literal "default").
// Returns the receiver so it composes with NewRegistry().
func (r *Registry) WithDefault(s Schema) *Registry {
	if s.Tenant == "" {
		s.Tenant = "default"
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.def = s
	return r
}

// Register stores s under s.Tenant. A Schema with empty Tenant is stored
// under "default" and ALSO replaces the DefaultSchema fallback.
func (r *Registry) Register(s Schema) error {
	if s.SKUField == "" || s.PriceField == "" {
		return fmt.Errorf("catalog: schema for %q must declare sku_field and price_field", s.Tenant)
	}
	if s.PriceScale < 0 {
		return fmt.Errorf("catalog: schema %q has negative price_scale (%d)", s.Tenant, s.PriceScale)
	}
	// Foot-gun 3, registry-side guard: UseSourceField without a field is
	// the silent stamp-now path. Refuse at load time so a JSON typo
	// surfaces immediately rather than invalidating caches per-rebuild.
	if s.UpdatedAtSource == UseSourceField && s.UpdatedAtField == "" {
		return fmt.Errorf("catalog: schema %q has updated_at_source=UseSourceField but no updated_at_field (would silently stamp-now)", s.Tenant)
	}
	if s.TrueValues == nil {
		s.TrueValues = DefaultTrueValues()
	}
	key := s.Tenant
	if key == "" {
		key = "default"
		s.Tenant = key
		r.mu.Lock()
		defer r.mu.Unlock()
		r.def = s
		r.schemas[key] = s
		return nil
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.schemas[key] = s
	return nil
}

// SchemaFor returns the Schema registered for tenant, or the DefaultSchema
// fallback when tenant is "" or not registered.
func (r *Registry) SchemaFor(tenant string) Schema {
	r.mu.RLock()
	defer r.mu.RUnlock()
	if tenant == "" {
		return r.def
	}
	if s, ok := r.schemas[tenant]; ok {
		return s
	}
	return r.def
}

// Tenants returns the sorted list of registered tenant names (excluding the
// default fallback). Useful for ops dashboards and "what schemas are loaded".
func (r *Registry) Tenants() []string {
	r.mu.RLock()
	defer r.mu.RUnlock()
	out := make([]string, 0, len(r.schemas))
	for k := range r.schemas {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// LoadFile reads a JSON file of { "tenant": Schema, ... } and registers each
// entry into r. The JSON shape mirrors the Schema struct exactly (json tags
// in schema.go). Missing file is OK (file is optional); unreadable file
// returns the error so a misconfigured deployment fails loudly on startup.
func (r *Registry) LoadFile(path string) error {
	if path == "" {
		return nil
	}
	data, err := os.ReadFile(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return fmt.Errorf("catalog: read schema file %q: %w", path, err)
	}
	var raw map[string]Schema
	if err := json.Unmarshal(data, &raw); err != nil {
		return fmt.Errorf("catalog: parse schema file %q: %w", path, err)
	}
	// Walk in sorted order so operator-visible loading is deterministic
	// (helps when something fails partway through).
	keys := make([]string, 0, len(raw))
	for k := range raw {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		s := raw[k]
		if s.Tenant == "" {
			s.Tenant = k
		}
		if err := r.Register(s); err != nil {
			return fmt.Errorf("catalog: schema %q in %q: %w", k, path, err)
		}
	}
	return nil
}