Fetchers 2.0: Account fetchers

account/account.go

@@ -15,6 +15,15 @@ import (
 	"github.com/prebid/prebid-server/v4/util/jsonutil"
 )
 
+// TypedAccountFetcher is an optional interface that a fetcher may implement to
+// return a fully-derived, immutable *config.Account directly, skipping the
+// per-request JSON unmarshal + defaults-merge + derive done on the legacy path.
+// The Fetchers 2.0 (cachekit) account fetcher implements this; when present,
+// GetAccount uses it. Legacy fetchers do not, and take the JSON path unchanged.
+type TypedAccountFetcher interface {
+	FetchAccountTyped(ctx context.Context, accountID string) (*config.Account, []error)
+}
+
 // GetAccount looks up the config.Account object referenced by the given accountID, with access rules applied
 func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_requests.AccountFetcher, accountID string, me metrics.MetricsEngine) (account *config.Account, errs []error) {
 	if cfg.AccountRequired && accountID == metrics.PublisherUnknown {
@@ -23,6 +32,16 @@ func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_r
 		}}
 	}
 
+	if typed, ok := fetcher.(TypedAccountFetcher); ok {
+		return getAccountTyped(ctx, cfg, typed, accountID)
+	}
+	return getAccountJSON(ctx, cfg, fetcher, accountID)
+}
+
+// getAccountJSON is the legacy account resolution path: it fetches raw
+// (defaults-merged) JSON, unmarshals it, unpacks DSA defaults and computes the
+// derived config on every call.
+func getAccountJSON(ctx context.Context, cfg *config.Configuration, fetcher stored_requests.AccountFetcher, accountID string) (account *config.Account, errs []error) {
 	if accountJSON, accErrs := fetcher.FetchAccount(ctx, cfg.AccountDefaultsJSON(), accountID); len(accErrs) > 0 || accountJSON == nil {
 		// accountID does not reference a valid account
 		for _, e := range accErrs {
@@ -70,15 +89,66 @@ func GetAccount(ctx context.Context, cfg *config.Configuration, fetcher stored_r
 		return nil, errs
 	}
 
+	applyIPMaskingDefaults(account)
+	return account, nil
+}
+
+// getAccountTyped is the Fetchers 2.0 account resolution path. The fetcher returns
+// a fully-derived, immutable *config.Account (unmarshal + DSA + derive + IP masking
+// were done once, at cache insert). This path only applies the not-found fallback
+// and the per-request access gating.
+func getAccountTyped(ctx context.Context, cfg *config.Configuration, fetcher TypedAccountFetcher, accountID string) (account *config.Account, errs []error) {
+	fetched, accErrs := fetcher.FetchAccountTyped(ctx, accountID)
+	if len(accErrs) > 0 {
+		// A malformed account is a hard error, mirroring the legacy path where the
+		// unmarshal/DSA failure returns immediately rather than falling back to defaults.
+		for _, e := range accErrs {
+			if _, ok := e.(*errortypes.MalformedAcct); ok {
+				return nil, accErrs
+			}
+		}
+		// Otherwise (not-found, or a swallowed backend error) fall through to the
+		// AccountDefaults fallback, matching the legacy not-found branch.
+		fetched = nil
+	}
+	if fetched == nil {
+		if cfg.AccountRequired && cfg.AccountDefaults.Disabled {
+			return nil, []error{&errortypes.AcctRequired{
+				Message: "Prebid-server could not verify the Account ID. Please reach out to the prebid server host.",
+			}}
+		}
+		// Make a copy of AccountDefaults instead of taking a reference,
+		// to preserve original accountID in case is needed to check NonStandardPublisherMap
+		pubAccount := cfg.AccountDefaults
+		pubAccount.ID = accountID
+		account = &pubAccount
+	} else {
+		// Fully-derived, immutable account returned straight from the cache.
+		account = fetched
+	}
+	if account.Disabled {
+		errs = append(errs, &errortypes.AccountDisabled{
+			Message: fmt.Sprintf("Prebid-server has disabled Account ID: %s, please reach out to the prebid server host.", accountID),
+		})
+		return nil, errs
+	}
+
+	// No-op for the cached (already-masked) account; corrects the defaults copy.
+	applyIPMaskingDefaults(account)
+	return account, nil
+}
+
+// applyIPMaskingDefaults falls back to the default IPv4/IPv6 masking bit sizes when
+// the configured values are invalid. Re-running it on an already-valid account is a
+// read-only no-op.
+func applyIPMaskingDefaults(account *config.Account) {
 	if ipV6Err := account.Privacy.IPv6Config.Validate(nil); len(ipV6Err) > 0 {
 		account.Privacy.IPv6Config.AnonKeepBits = iputil.IPv6DefaultMaskingBitSize
 	}
 
 	if ipV4Err := account.Privacy.IPv4Config.Validate(nil); len(ipV4Err) > 0 {
 		account.Privacy.IPv4Config.AnonKeepBits = iputil.IPv4DefaultMaskingBitSize
 	}
-
-	return account, nil
 }
 
 // TCF2Enforcements maps enforcement algo string values to their integer representation and is

account/cachekit.go

@@ -0,0 +1,241 @@
+package account
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/benbjohnson/clock"
+	jsonpatch "gopkg.in/evanphx/json-patch.v5"
+
+	"github.com/prebid/prebid-server/v4/cachekit"
+	"github.com/prebid/prebid-server/v4/config"
+	"github.com/prebid/prebid-server/v4/errortypes"
+	"github.com/prebid/prebid-server/v4/logger"
+	"github.com/prebid/prebid-server/v4/metrics"
+	"github.com/prebid/prebid-server/v4/stored_requests"
+	"github.com/prebid/prebid-server/v4/util/jsonutil"
+)
+
+// cacheKitSubsystem is the metrics subsystem label for the account cache.
+const cacheKitSubsystem = "account"
+
+// CacheKitAccountFetcher is the Fetchers 2.0 account fetcher. It embeds the
+// underlying source fetcher (file / db / http / multi) so it continues to satisfy
+// stored_requests.AllFetcher, and adds FetchAccountTyped which serves fully-derived,
+// immutable *config.Account values from a cachekit engine.
+type CacheKitAccountFetcher struct {
+	stored_requests.AllFetcher
+	engine *cachekit.Fetcher[string, *config.Account]
+}
+
+// FetchAccountTyped implements account.TypedAccountFetcher.
+func (f *CacheKitAccountFetcher) FetchAccountTyped(ctx context.Context, accountID string) (*config.Account, []error) {
+	account, err := f.engine.Get(ctx, accountID)
+	if err != nil {
+		if errors.Is(err, cachekit.ErrNotFound) {
+			return nil, []error{stored_requests.NotFoundError{ID: accountID, DataType: "Account"}}
+		}
+		return nil, []error{err}
+	}
+	return account, nil
+}
+
+// NewCacheKitAccountFetcher wires a cachekit engine in front of an existing account
+// source. The sources emit raw, unmerged account rows and the shared transform applies
+// the account defaults once (at cache insert); this fetcher adds the typed cache,
+// single-flight coalescing and optional negative caching. clk may be nil (a real clock
+// is used). metricsEngine may be nil (metrics are not recorded).
+func NewCacheKitAccountFetcher(source stored_requests.AllFetcher, cfg config.CacheKitConfig, defaults json.RawMessage, clk clock.Clock, metricsEngine metrics.MetricsEngine) (*CacheKitAccountFetcher, error) {
+	var cache cachekit.Cache[string, *config.Account]
+	switch cfg.Type {
+	case "none":
+		cache = cachekit.NoCache[string, *config.Account]{}
+	case "", "lru":
+		lru, err := cachekit.NewLRUCache[string, *config.Account](cfg.MaxEntries, clk)
+		if err != nil {
+			return nil, err
+		}
+		cache = lru
+	default:
+		return nil, fmt.Errorf("accounts.cache.type %q is not supported (expected none or lru)", cfg.Type)
+	}
+
+	var negatives *cachekit.NegativeStore[string]
+	if cfg.Negative.Enabled {
+		n, err := cachekit.NewNegativeStore[string](cfg.Negative.MaxEntries, cfg.Negative.TTL(), clk)
+		if err != nil {
+			// Negative caching is an optimization, not a correctness requirement. If it
+			// can't be built (e.g. a bad accounts.cache.negative.max_entries), warn and
+			// continue without it rather than aborting account fetcher startup; not-found
+			// lookups will just fall through to the backend each time.
+			logger.Warnf("account cachekit: negative caching disabled, failed to initialize: %v", err)
+		} else {
+			negatives = n
+		}
+	}
+
+	var recorder cachekit.Recorder
+	if metricsEngine != nil {
+		recorder = metricsRecorder{engine: metricsEngine, subsystem: cacheKitSubsystem}
+	}
+
+	// Freshness (refresh) axis: ttl (serve-stale + background revalidation), none
+	// (never revalidate / load-once), or preload (bulk warm at startup then ttl).
+	effectiveTTL := cfg.TTL()
+	var preload cachekit.BulkSource[string]
+	switch cfg.Refresh {
+	case "", config.RefreshTTL:
+		// serve-stale via ttl; nothing to preload.
+	case config.RefreshNone:
+		effectiveTTL = 0 // never revalidate
+	case config.RefreshPreload:
+		bulk, ok := source.(stored_requests.AllAccountsFetcher)
+		if !ok {
+			return nil, fmt.Errorf("accounts.cache.refresh %q requires an account source that supports bulk loading (FetchAllAccounts)", cfg.Refresh)
+		}
+		preload = accountBulkSource{fetcher: bulk}
+	// NOTE: "delta-poll" (event-driven Save/Invalidation, mirroring v1's http_events /
+	// cache-events producers) is intentionally not implemented: no known deployment
+	// pushes live account updates, so it would be untested, unused code. It can be added
+	// later as a background mechanism without touching Source/Transform/Cache.
+	default:
+		return nil, fmt.Errorf("accounts.cache.refresh %q is not supported (expected ttl, none or preload)", cfg.Refresh)
+	}
+
+	engine := cachekit.New(cachekit.Params[string, *config.Account]{
+		Source:     accountSource{fetcher: source},
+		Transform:  newAccountTransform(defaults),
+		Cache:      cache,
+		TTL:        effectiveTTL,
+		Negatives:  negatives,
+		Coalesce:   cfg.CoalesceRequests,
+		ServeStale: cfg.ServeStale,
+		Preload:    preload,
+		Clock:      clk,
+		Metrics:    recorder,
+	})
+	engine.Start(context.Background())
+
+	return &CacheKitAccountFetcher{AllFetcher: source, engine: engine}, nil
+}
+
+// metricsRecorder adapts a metrics.MetricsEngine to the cachekit.Recorder interface,
+// emitting the dedicated cachekit_* metrics under the given subsystem label.
+type metricsRecorder struct {
+	engine    metrics.MetricsEngine
+	subsystem string
+}
+
+func (r metricsRecorder) CacheHit() {
+	r.engine.RecordCacheKitResult(r.subsystem, metrics.CacheKitResultHit)
+}
+func (r metricsRecorder) CacheMiss() {
+	r.engine.RecordCacheKitResult(r.subsystem, metrics.CacheKitResultMiss)
+}
+func (r metricsRecorder) CacheNegative() {
+	r.engine.RecordCacheKitResult(r.subsystem, metrics.CacheKitResultNegative)
+}
+
+func (r metricsRecorder) BackendFetch(result string, d time.Duration) {
+	var mapped metrics.CacheKitBackendResult
+	switch result {
+	case "ok":
+		mapped = metrics.CacheKitBackendOK
+	case "notfound":
+		mapped = metrics.CacheKitBackendNotFound
+	default:
+		mapped = metrics.CacheKitBackendError
+	}
+	r.engine.RecordCacheKitBackendFetch(r.subsystem, mapped, d)
+}
+
+// accountSource adapts an existing stored_requests account fetcher into a
+// cachekit.Source. It requests the raw, unmerged account row (defaults are applied
+// once, downstream, by the shared transform) and reuses the backend's not-found
+// classification: a NotFoundError becomes an absent map key (cachekit's "not found"
+// convention); any other error is a systemic failure and is not cached.
+type accountSource struct {
+	fetcher stored_requests.AccountFetcher
+}
+
+func (s accountSource) Fetch(ctx context.Context, keys []string) (map[string]json.RawMessage, error) {
+	out := make(map[string]json.RawMessage, len(keys))
+	for _, id := range keys {
+		// nil defaults => the backend returns the raw row without merging. The shared
+		// transform applies defaults, so the single-key and bulk paths merge in one place.
+		raw, errs := s.fetcher.FetchAccount(ctx, nil, id)
+		if len(errs) > 0 {
+			if isNotFoundErr(errs) {
+				continue // absent key => definitive not-found for this id
+			}
+			return nil, errors.Join(errs...)
+		}
+		out[id] = raw
+	}
+	return out, nil
+}
+
+func isNotFoundErr(errs []error) bool {
+	for _, e := range errs {
+		if _, ok := e.(stored_requests.NotFoundError); ok {
+			return true
+		}
+	}
+	return false
+}
+
+// accountBulkSource adapts a stored_requests.AllAccountsFetcher into a cachekit.BulkSource.
+// It returns the raw, unmerged account rows as-is; the shared transform applies defaults
+// downstream, so this path and the single-key path merge in exactly one place.
+type accountBulkSource struct {
+	fetcher stored_requests.AllAccountsFetcher
+}
+
+func (s accountBulkSource) FetchAll(ctx context.Context) (map[string]json.RawMessage, error) {
+	data, errs := s.fetcher.FetchAllAccounts(ctx)
+	if len(errs) > 0 {
+		return nil, errors.Join(errs...)
+	}
+	return data, nil
+}
+
+// newAccountTransform returns the single normalization step for accounts. It merges the
+// account defaults into the raw row, then unmarshals, unpacks DSA defaults, fills the ID,
+// and computes the derived + IP-masking config. It runs once per id at cache insert, so
+// the per-request read path does no merge, unmarshal, DSA unpack, derive or IP masking.
+// Both the single-key and bulk sources feed it raw, unmerged rows, so the defaults-merge
+// lives here in exactly one place.
+func newAccountTransform(defaults json.RawMessage) cachekit.TransformFunc[string, *config.Account] {
+	return func(accountID string, raw json.RawMessage) (*config.Account, error) {
+		merged := raw
+		if defaults != nil {
+			m, err := jsonpatch.MergePatch(defaults, raw)
+			if err != nil {
+				return nil, &errortypes.MalformedAcct{
+					Message: fmt.Sprintf("The prebid-server account config for account id \"%s\" is malformed. Please reach out to the prebid server host.", accountID),
+				}
+			}
+			merged = m
+		}
+		account := &config.Account{}
+		if err := jsonutil.UnmarshalValid(merged, account); err != nil {
+			return nil, &errortypes.MalformedAcct{
+				Message: fmt.Sprintf("The prebid-server account config for account id \"%s\" is malformed. Please reach out to the prebid server host.", accountID),
+			}
+		}
+		if err := config.UnpackDSADefault(account.Privacy.DSA); err != nil {
+			return nil, &errortypes.MalformedAcct{
+				Message: fmt.Sprintf("The prebid-server account config DSA for account id \"%s\" is malformed. Please reach out to the prebid server host.", accountID),
+			}
+		}
+		if len(account.ID) == 0 {
+			account.ID = accountID
+		}
+		setDerivedConfig(account)
+		applyIPMaskingDefaults(account)
+		return account, nil
+	}
+}