diff --git a/modules/builder.go b/modules/builder.go index df2f1b1c3a7..41c6dc1b5a8 100644 --- a/modules/builder.go +++ b/modules/builder.go @@ -2,6 +2,7 @@ package modules import ( fiftyonedegreesDevicedetection "github.com/prebid/prebid-server/v4/modules/fiftyonedegrees/devicedetection" + prebidDoohcreativeapproval "github.com/prebid/prebid-server/v4/modules/prebid/doohcreativeapproval" prebidOrtb2blocking "github.com/prebid/prebid-server/v4/modules/prebid/ortb2blocking" prebidRulesengine "github.com/prebid/prebid-server/v4/modules/prebid/rulesengine" wurflDevicedetection "github.com/prebid/prebid-server/v4/modules/scientiamobile/wurfl_devicedetection" @@ -16,8 +17,9 @@ func builders() ModuleBuilders { "devicedetection": fiftyonedegreesDevicedetection.Builder, }, "prebid": { - "ortb2blocking": prebidOrtb2blocking.Builder, - "rulesengine": prebidRulesengine.Builder, + "doohcreativeapproval": prebidDoohcreativeapproval.Builder, + "ortb2blocking": prebidOrtb2blocking.Builder, + "rulesengine": prebidRulesengine.Builder, }, "scientiamobile": { "wurfl_devicedetection": wurflDevicedetection.Builder, diff --git a/modules/prebid/doohcreativeapproval/API_CONTRACT.md b/modules/prebid/doohcreativeapproval/API_CONTRACT.md new file mode 100644 index 00000000000..b3a185f62b7 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/API_CONTRACT.md @@ -0,0 +1,81 @@ +# DOOH Creative Approval API Contract + +`prebid.doohcreativeapproval` schedules one background bulk POST for uncached or due-for-refresh non-exempt creatives observed in a DOOH auction. The endpoint returns each creative's approval status for later auctions. + +The auction does not wait for this API. A first-seen creative is suppressed as `pending`, and existing creatives keep using their cached status while a refresh runs. The endpoint should be idempotent and safe to call repeatedly. + +## Matching + +`creative_approval_id` is the only response matching key. It is generated by PBS from PBS account ID, bidder, and `bid.crid`: + +```text +creative_approval_id = "v1:" + sha256(account_id + "\x1f" + bidder + "\x1f" + bid.crid) +``` + +`account_id` is the PBS account/config scope, not necessarily the OpenRTB `publisher.id`. If PBS runs this module without an account, `account_id` is empty. + +## Request + +```json +{ + "account_id": "acct", + "creatives": [ + { + "creative_approval_id": "v1:...", + "bidder": "appnexus", + "creative_id": "cr-123", + "ad_id": "ad-1", + "campaign_id": "camp-1", + "advertiser_domains": ["example.com"], + "categories": ["IAB1"], + "cat_tax": 6, + "media_type": "video", + "width": 1920, + "height": 1080, + "duration": 15, + "deal_id": "deal-1", + "iurl": "https://example.com/preview.jpg" + } + ] +} +``` + +The request may contain one or more creatives. Fields other than `creative_approval_id`, `bidder`, and `creative_id` are review metadata from the bid response. They are included to help the publisher approval service make or display a decision, but PBS matches the response only by `creative_approval_id`. + +## Response + +```json +{ + "creatives": [ + { + "creative_approval_id": "v1:...", + "status": "approved" + } + ] +} +``` + +Allowed statuses: + +- `approved`: later auctions allow the bid and the status is refreshed after `approved_ttl_seconds`. +- `rejected`: later auctions remove the bid and the status is refreshed after `rejected_ttl_seconds`. +- `pending`: later auctions remove the bid and the status is refreshed after `pending_ttl_seconds`. + +Duplicate entries invalidate the returned status for that creative. Missing entries, unknown statuses, duplicate entries, endpoint errors, timeouts, and malformed responses leave an existing cached status unchanged. A creative without a prior usable status remains `pending`. + +## Cache Semantics + +PBS caches approval statuses in process only. The approval endpoint remains the durable source of truth. + +The `*_ttl_seconds` settings control status freshness, not cache retention. When a cached status is due for refresh, PBS continues using it and calls the approval endpoint in the background. An unusable response keeps that status and schedules another attempt after `pending_ttl_seconds`. + +`cache_size_bytes` limits cache memory use and must be at least 524288 bytes. Entries may be evicted when the cache reaches capacity. Eviction is handled like a missing status: the next matching bid is suppressed and schedules a background lookup. + +Refreshes for the same creative are coalesced within one PBS process. `max_concurrent_lookups` limits concurrent bulk requests; when all slots are busy, PBS retains the current status and a later matching auction can start the refresh. + +## Limitations + +- PBS does not expose a cache inspection or cache invalidation API for this module. +- Approval changes are observed after a background refresh completes, when a cached status is due for refresh, is evicted, or is missing. +- v1 does not inspect ad markup or media content when generating `creative_approval_id`; it relies on `bid.crid` being stable for the creative approval unit. +- Cache state and refresh coordination are per PBS process, not shared across a cluster. diff --git a/modules/prebid/doohcreativeapproval/README.md b/modules/prebid/doohcreativeapproval/README.md new file mode 100644 index 00000000000..e8824d5d86b --- /dev/null +++ b/modules/prebid/doohcreativeapproval/README.md @@ -0,0 +1,111 @@ +# DOOH Creative Approval + +`prebid.doohcreativeapproval` lets a publisher approve DOOH creatives before they can compete in an auction. The module runs only for DOOH requests. Once the module is active, a non-exempt bid is allowed through only when its last-known creative approval status is `approved`. + +PBS is not the durable approval system. Each PBS process caches statuses locally and refreshes them in the background. The publisher approval service remains the source of truth. + +## Terms And Scope + +In this module, `account` means the PBS account/config scope. Account config controls the approval endpoint, status refresh TTLs, and exempt bidders. `publisher` means the business system or screen owner that reviews creatives. These are often the same operational boundary, but PBS does not require them to be the same identifier. + +Creative approval state is scoped by PBS account, bidder, and `bid.crid`: + +```text +creative_approval_id = "v1:" + sha256(account_id + "\x1f" + bidder + "\x1f" + bid.crid) +``` + +If the module is run without a PBS account, `account_id` is empty. Prefer account-level configuration when approvals need to be separated by publisher, tenant, or business owner. + +## Hook Setup + +The module must run in both stages: + +```yaml +hooks: + enabled: true + modules: + prebid: + doohcreativeapproval: + enabled: true + platforms: + - dooh + timeout_ms: 100 + cache_size_bytes: 10485760 + max_concurrent_lookups: 8 + approved_ttl_seconds: 3600 + rejected_ttl_seconds: 300 + pending_ttl_seconds: 60 + host_execution_plan: + endpoints: + /openrtb2/auction: + stages: + processed_auction_request: + groups: + - timeout: 100 + hook_sequence: + - module_code: prebid.doohcreativeapproval + hook_impl_code: dooh-creative-approval + all_processed_bid_responses: + groups: + - timeout: 100 + hook_sequence: + - module_code: prebid.doohcreativeapproval + hook_impl_code: dooh-creative-approval +``` + +The processed-auction hook only marks eligible DOOH auctions as active. The all-processed-bid-responses hook does the filtering. If the processed stage is omitted, the module intentionally does nothing at the filtering stage. + +## Account Config + +Publisher-specific endpoint config should live in account config: + +```json +{ + "hooks": { + "modules": { + "prebid": { + "doohcreativeapproval": { + "endpoint": "https://publisher.example.com/creative-approval", + "headers": { + "Authorization": "Bearer token" + }, + "exempt_bidders": ["house"] + } + } + } + } +} +``` + +Account config can override `enabled`, `platforms`, `endpoint`, `headers`, `timeout_ms`, status refresh TTLs, and `exempt_bidders`. `cache_size_bytes` and `max_concurrent_lookups` are host-level because the cache and refresh limit are shared by the module instance. + +`timeout_ms` bounds each background HTTP request. It does not extend the auction or hook execution timeout. + +## Behavior + +- Exempt bidders bypass approval and do not call the publisher endpoint. +- A first-seen creative is treated as `pending` and removed from the current auction. PBS starts a background lookup for later auctions. +- Cached `approved` creatives pass. Cached `rejected` and `pending` creatives are removed. +- When a cached status is due for refresh, PBS keeps using that status while refreshing it in the background. +- Endpoint errors, timeouts, malformed responses, missing entries, unknown statuses, and duplicate entries do not replace an existing status. PBS retries after `pending_ttl_seconds`. +- If no prior status exists and the endpoint cannot return a usable status, the creative remains `pending`. +- Refreshes for the same creative are coalesced. At most `max_concurrent_lookups` bulk requests run in one PBS process. + +## Cache Behavior + +The `*_ttl_seconds` settings control when a cached status is due for refresh. They do not delete the last-known status. Refreshes happen outside the auction path, and an unusable refresh leaves the current status unchanged. + +`cache_size_bytes` is a memory cap, not a guarantee that every cached creative remains resident. It must be at least 524288 bytes. If the cache reaches capacity, entries can be evicted. An evicted entry is treated as unknown, so its next bid is suppressed while PBS refreshes it. + +PBS does not expose an admin or inspection API for this cache. The approval endpoint should keep the durable approval records. + +## Limitations + +- v1 supports only `platforms: ["dooh"]`. Site and app requests are intentionally ignored. +- v1 assumes `account_id + bidder + bid.crid` identifies the creative approval unit. It does not hash ad markup, media files, or preview URLs. +- Approval changes are picked up through background refreshes, cache misses, or cache eviction, not through a push channel into PBS. +- Cache contents and refresh work are local to each PBS process. Multiple PBS instances can refresh the same creative independently. +- A missing endpoint leaves the module inactive for that account. Invalid account config or a PBS hook execution failure can prevent filtering; these are configuration or host-execution failures rather than approval lookup results. +- The first auction for an uncached creative is always suppressed, even if the publisher endpoint would immediately approve it. + +See `API_CONTRACT.md` for the external approval API. diff --git a/modules/prebid/doohcreativeapproval/cache.go b/modules/prebid/doohcreativeapproval/cache.go new file mode 100644 index 00000000000..8cdae2550a4 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/cache.go @@ -0,0 +1,88 @@ +package doohcreativeapproval + +import ( + "encoding/json" + "fmt" + "time" + + "github.com/coocood/freecache" +) + +type cachedApprovalStatus struct { + CreativeApprovalID string `json:"creative_approval_id"` + Status approvalStatus `json:"status"` + RefreshAfterUnixNano int64 `json:"refresh_after_unix_nano"` +} + +type cachedApprovalLookup struct { + Status approvalStatus + RefreshDue bool +} + +type approvalCache struct { + cache *freecache.Cache + marshal func(v any) ([]byte, error) + unmarshal func(data []byte, v any) error + now func() time.Time +} + +func newApprovalCache(sizeBytes int) *approvalCache { + return &approvalCache{ + cache: freecache.NewCache(sizeBytes), + marshal: json.Marshal, + unmarshal: json.Unmarshal, + now: time.Now, + } +} + +func (c *approvalCache) get(creativeApprovalID string) (cachedApprovalLookup, bool) { + if c == nil || c.cache == nil || creativeApprovalID == "" { + return cachedApprovalLookup{}, false + } + + data, err := c.cache.Get([]byte(creativeApprovalID)) + if err != nil { + return cachedApprovalLookup{}, false + } + + var entry cachedApprovalStatus + if err := c.unmarshal(data, &entry); err != nil { + return cachedApprovalLookup{}, false + } + if entry.CreativeApprovalID != creativeApprovalID || !isValidApprovalStatus(entry.Status) || entry.RefreshAfterUnixNano <= 0 { + return cachedApprovalLookup{}, false + } + + return cachedApprovalLookup{ + Status: entry.Status, + RefreshDue: !c.currentTime().Before(time.Unix(0, entry.RefreshAfterUnixNano)), + }, true +} + +func (c *approvalCache) set(creativeApprovalID string, status approvalStatus, refreshSeconds int) error { + if c == nil || c.cache == nil || creativeApprovalID == "" || refreshSeconds <= 0 || !isValidApprovalStatus(status) { + return nil + } + + entry := cachedApprovalStatus{ + CreativeApprovalID: creativeApprovalID, + Status: status, + RefreshAfterUnixNano: c.currentTime().Add(time.Duration(refreshSeconds) * time.Second).UnixNano(), + } + data, err := c.marshal(entry) + if err != nil { + return fmt.Errorf("marshal approval cache entry: %s", err) + } + + if err := c.cache.Set([]byte(creativeApprovalID), data, 0); err != nil { + return fmt.Errorf("store approval cache entry: %s", err) + } + return nil +} + +func (c *approvalCache) currentTime() time.Time { + if c.now == nil { + return time.Now() + } + return c.now() +} diff --git a/modules/prebid/doohcreativeapproval/cache_test.go b/modules/prebid/doohcreativeapproval/cache_test.go new file mode 100644 index 00000000000..dc10fef7679 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/cache_test.go @@ -0,0 +1,77 @@ +package doohcreativeapproval + +import ( + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestApprovalCacheGetSet(t *testing.T) { + cache := newApprovalCache(1024 * 1024) + + cache.set("v1:approved", approvalStatusApproved, 60) + cache.set("v1:rejected", approvalStatusRejected, 60) + cache.set("v1:pending", approvalStatusPending, 60) + + lookup, ok := cache.get("v1:approved") + assert.True(t, ok) + assert.Equal(t, approvalStatusApproved, lookup.Status) + assert.False(t, lookup.RefreshDue) + + lookup, ok = cache.get("v1:rejected") + assert.True(t, ok) + assert.Equal(t, approvalStatusRejected, lookup.Status) + assert.False(t, lookup.RefreshDue) + + lookup, ok = cache.get("v1:pending") + assert.True(t, ok) + assert.Equal(t, approvalStatusPending, lookup.Status) + assert.False(t, lookup.RefreshDue) +} + +func TestApprovalCacheMisses(t *testing.T) { + cache := newApprovalCache(1024 * 1024) + + cache.set("v1:zero-ttl", approvalStatusApproved, 0) + cache.set("v1:bad-status", "unknown", 60) + + _, ok := cache.get("v1:missing") + assert.False(t, ok) + + _, ok = cache.get("v1:zero-ttl") + assert.False(t, ok) + + _, ok = cache.get("v1:bad-status") + assert.False(t, ok) +} + +func TestApprovalCacheRefreshDueKeepsLastStatus(t *testing.T) { + cache := newApprovalCache(1024 * 1024) + now := time.Unix(1000, 0) + cache.now = func() time.Time { + return now + } + + cache.set("v1:refresh", approvalStatusApproved, 1) + now = now.Add(2 * time.Second) + + lookup, ok := cache.get("v1:refresh") + assert.True(t, ok) + assert.Equal(t, approvalStatusApproved, lookup.Status) + assert.True(t, lookup.RefreshDue) +} + +func TestApprovalCacheSetReturnsWriteError(t *testing.T) { + cache := newApprovalCache(1024 * 1024) + cache.marshal = func(v any) ([]byte, error) { + return nil, errors.New("marshal failed") + } + + err := cache.set("v1:write-error", approvalStatusApproved, 60) + + require.Error(t, err) + assert.Contains(t, err.Error(), "marshal approval cache entry") +} diff --git a/modules/prebid/doohcreativeapproval/config.go b/modules/prebid/doohcreativeapproval/config.go new file mode 100644 index 00000000000..18731f8165a --- /dev/null +++ b/modules/prebid/doohcreativeapproval/config.go @@ -0,0 +1,245 @@ +package doohcreativeapproval + +import ( + "encoding/json" + "fmt" + "net/url" + "strings" + + "github.com/prebid/prebid-server/v4/util/jsonutil" +) + +const ( + defaultPlatformDOOH = "dooh" + defaultTimeoutMS = 100 + defaultCacheSizeBytes = 10 * 1024 * 1024 + minimumCacheSizeBytes = 512 * 1024 + defaultMaxConcurrentLookups = 8 + defaultApprovedTTLSeconds = 3600 + defaultRejectedTTLSeconds = 300 + defaultPendingTTLSeconds = 60 +) + +type moduleConfig struct { + Enabled bool `json:"enabled,omitempty"` + Platforms []string `json:"platforms,omitempty"` + Endpoint string `json:"endpoint,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + TimeoutMS int `json:"timeout_ms,omitempty"` + CacheSizeBytes int `json:"cache_size_bytes,omitempty"` + MaxConcurrentLookups int `json:"max_concurrent_lookups,omitempty"` + ApprovedTTLSeconds int `json:"approved_ttl_seconds,omitempty"` + RejectedTTLSeconds int `json:"rejected_ttl_seconds,omitempty"` + PendingTTLSeconds int `json:"pending_ttl_seconds,omitempty"` + ExemptBidders []string `json:"exempt_bidders,omitempty"` +} + +type moduleConfigOverlay struct { + Enabled *bool `json:"enabled,omitempty"` + Platforms []string `json:"platforms,omitempty"` + Endpoint *string `json:"endpoint,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + TimeoutMS *int `json:"timeout_ms,omitempty"` + ApprovedTTLSeconds *int `json:"approved_ttl_seconds,omitempty"` + RejectedTTLSeconds *int `json:"rejected_ttl_seconds,omitempty"` + PendingTTLSeconds *int `json:"pending_ttl_seconds,omitempty"` + ExemptBidders []string `json:"exempt_bidders,omitempty"` +} + +func parseModuleConfig(data json.RawMessage) (moduleConfig, error) { + cfg := defaultModuleConfig() + if len(data) > 0 { + if err := jsonutil.UnmarshalValid(data, &cfg); err != nil { + return cfg, fmt.Errorf("failed to parse config: %s", err) + } + } + + return normalizeModuleConfig(cfg) +} + +func defaultModuleConfig() moduleConfig { + return moduleConfig{ + Enabled: true, + Platforms: []string{defaultPlatformDOOH}, + TimeoutMS: defaultTimeoutMS, + CacheSizeBytes: defaultCacheSizeBytes, + MaxConcurrentLookups: defaultMaxConcurrentLookups, + ApprovedTTLSeconds: defaultApprovedTTLSeconds, + RejectedTTLSeconds: defaultRejectedTTLSeconds, + PendingTTLSeconds: defaultPendingTTLSeconds, + } +} + +func applyAccountConfig(base moduleConfig, data json.RawMessage) (moduleConfig, error) { + if len(data) > 0 { + var overlay moduleConfigOverlay + if err := jsonutil.UnmarshalValid(data, &overlay); err != nil { + return base, fmt.Errorf("failed to parse account config: %s", err) + } + + if overlay.Enabled != nil { + base.Enabled = *overlay.Enabled + } + if overlay.Platforms != nil { + base.Platforms = overlay.Platforms + } + if overlay.Endpoint != nil { + base.Endpoint = *overlay.Endpoint + } + if overlay.Headers != nil { + base.Headers = overlay.Headers + } + if overlay.TimeoutMS != nil { + base.TimeoutMS = *overlay.TimeoutMS + } + if overlay.ApprovedTTLSeconds != nil { + base.ApprovedTTLSeconds = *overlay.ApprovedTTLSeconds + } + if overlay.RejectedTTLSeconds != nil { + base.RejectedTTLSeconds = *overlay.RejectedTTLSeconds + } + if overlay.PendingTTLSeconds != nil { + base.PendingTTLSeconds = *overlay.PendingTTLSeconds + } + if overlay.ExemptBidders != nil { + base.ExemptBidders = overlay.ExemptBidders + } + } + + return normalizeModuleConfig(base) +} + +func normalizeModuleConfig(cfg moduleConfig) (moduleConfig, error) { + platforms, err := normalizePlatforms(cfg.Platforms) + if err != nil { + return cfg, err + } + cfg.Platforms = platforms + + if cfg.Endpoint != "" { + endpointURL, err := url.ParseRequestURI(cfg.Endpoint) + if err != nil { + return cfg, fmt.Errorf("endpoint is invalid: %s", err) + } + if endpointURL.Scheme != "http" && endpointURL.Scheme != "https" { + return cfg, fmt.Errorf("endpoint scheme must be http or https") + } + } + + if cfg.TimeoutMS < 0 { + return cfg, fmt.Errorf("timeout_ms cannot be negative") + } + if cfg.TimeoutMS == 0 { + cfg.TimeoutMS = defaultTimeoutMS + } + + if cfg.CacheSizeBytes < 0 { + return cfg, fmt.Errorf("cache_size_bytes cannot be negative") + } + if cfg.CacheSizeBytes == 0 { + cfg.CacheSizeBytes = defaultCacheSizeBytes + } + if cfg.CacheSizeBytes < minimumCacheSizeBytes { + return cfg, fmt.Errorf("cache_size_bytes must be at least %d", minimumCacheSizeBytes) + } + + if cfg.MaxConcurrentLookups < 0 { + return cfg, fmt.Errorf("max_concurrent_lookups cannot be negative") + } + if cfg.MaxConcurrentLookups == 0 { + cfg.MaxConcurrentLookups = defaultMaxConcurrentLookups + } + + if cfg.ApprovedTTLSeconds < 0 { + return cfg, fmt.Errorf("approved_ttl_seconds cannot be negative") + } + if cfg.ApprovedTTLSeconds == 0 { + cfg.ApprovedTTLSeconds = defaultApprovedTTLSeconds + } + + if cfg.RejectedTTLSeconds < 0 { + return cfg, fmt.Errorf("rejected_ttl_seconds cannot be negative") + } + if cfg.RejectedTTLSeconds == 0 { + cfg.RejectedTTLSeconds = defaultRejectedTTLSeconds + } + + if cfg.PendingTTLSeconds < 0 { + return cfg, fmt.Errorf("pending_ttl_seconds cannot be negative") + } + if cfg.PendingTTLSeconds == 0 { + cfg.PendingTTLSeconds = defaultPendingTTLSeconds + } + + for name := range cfg.Headers { + if strings.TrimSpace(name) == "" { + return cfg, fmt.Errorf("headers cannot contain an empty header name") + } + } + + cfg.ExemptBidders = normalizeExemptBidders(cfg.ExemptBidders) + + return cfg, nil +} + +func normalizePlatforms(platforms []string) ([]string, error) { + if len(platforms) == 0 { + return []string{defaultPlatformDOOH}, nil + } + + seen := make(map[string]struct{}, len(platforms)) + normalized := make([]string, 0, len(platforms)) + for _, platform := range platforms { + platform = strings.ToLower(strings.TrimSpace(platform)) + if platform == "" { + return nil, fmt.Errorf("platforms cannot contain an empty platform") + } + if platform != defaultPlatformDOOH { + return nil, fmt.Errorf("platforms must contain only %q", defaultPlatformDOOH) + } + if _, ok := seen[platform]; ok { + continue + } + seen[platform] = struct{}{} + normalized = append(normalized, platform) + } + + return normalized, nil +} + +func normalizeExemptBidders(bidders []string) []string { + seen := make(map[string]struct{}, len(bidders)) + normalized := make([]string, 0, len(bidders)) + for _, bidder := range bidders { + bidder = strings.ToLower(strings.TrimSpace(bidder)) + if bidder == "" { + continue + } + if _, ok := seen[bidder]; ok { + continue + } + seen[bidder] = struct{}{} + normalized = append(normalized, bidder) + } + return normalized +} + +func isBidderExempt(cfg moduleConfig, bidder string) bool { + for _, exemptBidder := range cfg.ExemptBidders { + if strings.EqualFold(exemptBidder, bidder) { + return true + } + } + return false +} + +func ttlForStatus(cfg moduleConfig, status approvalStatus) int { + switch status { + case approvalStatusApproved: + return cfg.ApprovedTTLSeconds + case approvalStatusRejected: + return cfg.RejectedTTLSeconds + default: + return cfg.PendingTTLSeconds + } +} diff --git a/modules/prebid/doohcreativeapproval/config_test.go b/modules/prebid/doohcreativeapproval/config_test.go new file mode 100644 index 00000000000..3ef4a711f60 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/config_test.go @@ -0,0 +1,158 @@ +package doohcreativeapproval + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseModuleConfigDefaults(t *testing.T) { + cfg, err := parseModuleConfig(nil) + + require.NoError(t, err) + assert.True(t, cfg.Enabled) + assert.Equal(t, []string{defaultPlatformDOOH}, cfg.Platforms) + assert.Equal(t, defaultTimeoutMS, cfg.TimeoutMS) + assert.Equal(t, defaultCacheSizeBytes, cfg.CacheSizeBytes) + assert.Equal(t, defaultMaxConcurrentLookups, cfg.MaxConcurrentLookups) + assert.Equal(t, defaultApprovedTTLSeconds, cfg.ApprovedTTLSeconds) + assert.Equal(t, defaultRejectedTTLSeconds, cfg.RejectedTTLSeconds) + assert.Equal(t, defaultPendingTTLSeconds, cfg.PendingTTLSeconds) +} + +func TestParseModuleConfigErrors(t *testing.T) { + tests := []struct { + name string + config json.RawMessage + expectedErr string + }{ + { + name: "invalid-json", + config: json.RawMessage(`invalid`), + expectedErr: "failed to parse config", + }, + { + name: "invalid-endpoint", + config: json.RawMessage(`{"endpoint":"://bad"}`), + expectedErr: "endpoint is invalid", + }, + { + name: "invalid-endpoint-scheme", + config: json.RawMessage(`{"endpoint":"ftp://example.com"}`), + expectedErr: "endpoint scheme must be http or https", + }, + { + name: "invalid-platform", + config: json.RawMessage(`{"platforms":["site"]}`), + expectedErr: `platforms must contain only "dooh"`, + }, + { + name: "negative-timeout", + config: json.RawMessage(`{"timeout_ms":-1}`), + expectedErr: "timeout_ms cannot be negative", + }, + { + name: "negative-cache-size", + config: json.RawMessage(`{"cache_size_bytes":-1}`), + expectedErr: "cache_size_bytes cannot be negative", + }, + { + name: "cache-size-below-freecache-minimum", + config: json.RawMessage(`{"cache_size_bytes":1024}`), + expectedErr: "cache_size_bytes must be at least 524288", + }, + { + name: "negative-max-concurrent-lookups", + config: json.RawMessage(`{"max_concurrent_lookups":-1}`), + expectedErr: "max_concurrent_lookups cannot be negative", + }, + { + name: "negative-approved-ttl", + config: json.RawMessage(`{"approved_ttl_seconds":-1}`), + expectedErr: "approved_ttl_seconds cannot be negative", + }, + { + name: "negative-rejected-ttl", + config: json.RawMessage(`{"rejected_ttl_seconds":-1}`), + expectedErr: "rejected_ttl_seconds cannot be negative", + }, + { + name: "negative-pending-ttl", + config: json.RawMessage(`{"pending_ttl_seconds":-1}`), + expectedErr: "pending_ttl_seconds cannot be negative", + }, + { + name: "empty-header-name", + config: json.RawMessage(`{"headers":{"":"value"}}`), + expectedErr: "headers cannot contain an empty header name", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := parseModuleConfig(test.config) + require.Error(t, err) + assert.Contains(t, err.Error(), test.expectedErr) + }) + } +} + +func TestApplyAccountConfig(t *testing.T) { + base := testModuleConfig() + base.Endpoint = "http://host.example.com" + base.Headers = map[string]string{"X-Host": "host"} + base.TimeoutMS = 200 + base.ApprovedTTLSeconds = 300 + base.RejectedTTLSeconds = 400 + base.PendingTTLSeconds = 500 + base.CacheSizeBytes = 2 * minimumCacheSizeBytes + base.MaxConcurrentLookups = 3 + base.ExemptBidders = []string{"hostbidder"} + + accountConfig := json.RawMessage(`{ + "enabled": false, + "endpoint": "https://account.example.com/approval", + "headers": {"X-Account": "account"}, + "timeout_ms": 10, + "max_concurrent_lookups": 99, + "approved_ttl_seconds": 11, + "rejected_ttl_seconds": 12, + "pending_ttl_seconds": 13, + "exempt_bidders": ["AppNexus", " appnexus ", "", "Rubicon"] + }`) + + cfg, err := applyAccountConfig(base, accountConfig) + + require.NoError(t, err) + assert.False(t, cfg.Enabled) + assert.Equal(t, "https://account.example.com/approval", cfg.Endpoint) + assert.Equal(t, map[string]string{"X-Account": "account"}, cfg.Headers) + assert.Equal(t, 10, cfg.TimeoutMS) + assert.Equal(t, 11, cfg.ApprovedTTLSeconds) + assert.Equal(t, 12, cfg.RejectedTTLSeconds) + assert.Equal(t, 13, cfg.PendingTTLSeconds) + assert.Equal(t, 2*minimumCacheSizeBytes, cfg.CacheSizeBytes) + assert.Equal(t, 3, cfg.MaxConcurrentLookups) + assert.Equal(t, []string{"appnexus", "rubicon"}, cfg.ExemptBidders) +} + +func TestApplyAccountConfigInvalid(t *testing.T) { + _, err := applyAccountConfig(testModuleConfig(), json.RawMessage(`{"platforms":["app"]}`)) + + require.Error(t, err) + assert.Contains(t, err.Error(), `platforms must contain only "dooh"`) +} + +func TestTTLForStatus(t *testing.T) { + cfg := testModuleConfig() + cfg.ApprovedTTLSeconds = 1 + cfg.RejectedTTLSeconds = 2 + cfg.PendingTTLSeconds = 3 + + assert.Equal(t, 1, ttlForStatus(cfg, approvalStatusApproved)) + assert.Equal(t, 2, ttlForStatus(cfg, approvalStatusRejected)) + assert.Equal(t, 3, ttlForStatus(cfg, approvalStatusPending)) + assert.Equal(t, 3, ttlForStatus(cfg, "unknown")) +} diff --git a/modules/prebid/doohcreativeapproval/creative.go b/modules/prebid/doohcreativeapproval/creative.go new file mode 100644 index 00000000000..9f8f9f2e86f --- /dev/null +++ b/modules/prebid/doohcreativeapproval/creative.go @@ -0,0 +1,40 @@ +package doohcreativeapproval + +import ( + "crypto/sha256" + "encoding/hex" + + "github.com/prebid/prebid-server/v4/exchange/entities" + "github.com/prebid/prebid-server/v4/openrtb_ext" +) + +const creativeApprovalIDVersion = "v1:" + +func creativeApprovalID(accountID string, bidder openrtb_ext.BidderName, creativeID string) string { + hash := sha256.Sum256([]byte(accountID + "\x1f" + bidder.String() + "\x1f" + creativeID)) + return creativeApprovalIDVersion + hex.EncodeToString(hash[:]) +} + +func newCreativeApproval(accountID string, bidder openrtb_ext.BidderName, pbsBid *entities.PbsOrtbBid) (creativeApproval, bool) { + if pbsBid == nil || pbsBid.Bid == nil || pbsBid.Bid.CrID == "" { + return creativeApproval{}, false + } + + bid := pbsBid.Bid + return creativeApproval{ + CreativeApprovalID: creativeApprovalID(accountID, bidder, bid.CrID), + Bidder: bidder.String(), + CreativeID: bid.CrID, + AdID: bid.AdID, + CampaignID: bid.CID, + AdvertiserDomains: append([]string(nil), bid.ADomain...), + Categories: append([]string(nil), bid.Cat...), + CategoryTaxonomy: int(bid.CatTax), + MediaType: string(pbsBid.BidType), + Width: bid.W, + Height: bid.H, + Duration: bid.Dur, + DealID: bid.DealID, + IURL: bid.IURL, + }, true +} diff --git a/modules/prebid/doohcreativeapproval/creative_test.go b/modules/prebid/doohcreativeapproval/creative_test.go new file mode 100644 index 00000000000..34efcab2a5f --- /dev/null +++ b/modules/prebid/doohcreativeapproval/creative_test.go @@ -0,0 +1,57 @@ +package doohcreativeapproval + +import ( + "strings" + "testing" + + "github.com/prebid/openrtb/v20/adcom1" + "github.com/prebid/prebid-server/v4/openrtb_ext" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCreativeApprovalID(t *testing.T) { + bidder := openrtb_ext.BidderName("appnexus") + + id := creativeApprovalID("acct", bidder, "cr-123") + + assert.True(t, strings.HasPrefix(id, creativeApprovalIDVersion)) + assert.Equal(t, id, creativeApprovalID("acct", bidder, "cr-123")) + assert.NotEqual(t, id, creativeApprovalID("acct-2", bidder, "cr-123")) + assert.NotEqual(t, id, creativeApprovalID("acct", openrtb_ext.BidderName("rubicon"), "cr-123")) + assert.NotEqual(t, id, creativeApprovalID("acct", bidder, "cr-456")) +} + +func TestNewCreativeApproval(t *testing.T) { + bid := testBid("cr-123") + bid.Bid.CatTax = adcom1.CategoryTaxonomy(6) + + creative, ok := newCreativeApproval("acct", openrtb_ext.BidderName("appnexus"), bid) + + require.True(t, ok) + assert.Equal(t, creativeApprovalID("acct", openrtb_ext.BidderName("appnexus"), "cr-123"), creative.CreativeApprovalID) + assert.Equal(t, "appnexus", creative.Bidder) + assert.Equal(t, "cr-123", creative.CreativeID) + assert.Equal(t, "ad-cr-123", creative.AdID) + assert.Equal(t, "campaign-cr-123", creative.CampaignID) + assert.Equal(t, []string{"advertiser.example"}, creative.AdvertiserDomains) + assert.Equal(t, []string{"IAB1"}, creative.Categories) + assert.Equal(t, 6, creative.CategoryTaxonomy) + assert.Equal(t, "video", creative.MediaType) + assert.EqualValues(t, 1920, creative.Width) + assert.EqualValues(t, 1080, creative.Height) + assert.EqualValues(t, 15, creative.Duration) + assert.Equal(t, "deal-cr-123", creative.DealID) + assert.Equal(t, "https://example.com/cr-123.jpg", creative.IURL) + + bid.Bid.ADomain[0] = "changed.example" + bid.Bid.Cat[0] = "IAB2" + assert.Equal(t, []string{"advertiser.example"}, creative.AdvertiserDomains) + assert.Equal(t, []string{"IAB1"}, creative.Categories) +} + +func TestNewCreativeApprovalMissingCreativeID(t *testing.T) { + _, ok := newCreativeApproval("acct", openrtb_ext.BidderName("appnexus"), testBid("")) + + assert.False(t, ok) +} diff --git a/modules/prebid/doohcreativeapproval/filter.go b/modules/prebid/doohcreativeapproval/filter.go new file mode 100644 index 00000000000..8a3d22e9a42 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/filter.go @@ -0,0 +1,90 @@ +package doohcreativeapproval + +import ( + "github.com/prebid/prebid-server/v4/exchange/entities" + "github.com/prebid/prebid-server/v4/openrtb_ext" +) + +type approvalCandidate struct { + CreativeApprovalID string + Exempt bool +} + +func collectCreativeApprovals(responses map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid, cfg moduleConfig, accountID string) (map[string]creativeApproval, []string) { + creatives := make(map[string]creativeApproval) + warnings := make([]string, 0) + + for bidder, seatBid := range responses { + if seatBid == nil || isBidderExempt(cfg, bidder.String()) { + continue + } + for _, pbsBid := range seatBid.Bids { + creative, ok := newCreativeApproval(accountID, bidder, pbsBid) + if !ok { + warnings = append(warnings, "bid skipped from approval lookup because it is missing creative id") + continue + } + if _, exists := creatives[creative.CreativeApprovalID]; exists { + continue + } + creatives[creative.CreativeApprovalID] = creative + } + } + + return creatives, warnings +} + +func needsApprovalFilter(responses map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid, cfg moduleConfig, accountID string, statuses map[string]approvalStatus) bool { + for bidder, seatBid := range responses { + if seatBid == nil { + continue + } + for _, pbsBid := range seatBid.Bids { + candidate := approvalCandidateForBid(accountID, bidder, pbsBid, cfg) + if !candidate.Exempt && statuses[candidate.CreativeApprovalID] != approvalStatusApproved { + return true + } + } + } + return false +} + +func filterResponsesByApproval(responses map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid, cfg moduleConfig, accountID string, statuses map[string]approvalStatus) int { + removed := 0 + for bidder, seatBid := range responses { + if seatBid == nil { + delete(responses, bidder) + continue + } + + keptBids := make([]*entities.PbsOrtbBid, 0, len(seatBid.Bids)) + for _, pbsBid := range seatBid.Bids { + candidate := approvalCandidateForBid(accountID, bidder, pbsBid, cfg) + if candidate.Exempt || statuses[candidate.CreativeApprovalID] == approvalStatusApproved { + keptBids = append(keptBids, pbsBid) + continue + } + removed++ + } + + if len(keptBids) == 0 { + delete(responses, bidder) + continue + } + seatBid.Bids = keptBids + } + + return removed +} + +func approvalCandidateForBid(accountID string, bidder openrtb_ext.BidderName, pbsBid *entities.PbsOrtbBid, cfg moduleConfig) approvalCandidate { + if isBidderExempt(cfg, bidder.String()) { + return approvalCandidate{Exempt: true} + } + if pbsBid == nil || pbsBid.Bid == nil || pbsBid.Bid.CrID == "" { + return approvalCandidate{} + } + return approvalCandidate{ + CreativeApprovalID: creativeApprovalID(accountID, bidder, pbsBid.Bid.CrID), + } +} diff --git a/modules/prebid/doohcreativeapproval/filter_test.go b/modules/prebid/doohcreativeapproval/filter_test.go new file mode 100644 index 00000000000..06067431039 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/filter_test.go @@ -0,0 +1,81 @@ +package doohcreativeapproval + +import ( + "testing" + + "github.com/prebid/prebid-server/v4/exchange/entities" + "github.com/prebid/prebid-server/v4/openrtb_ext" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFilterResponsesByApproval(t *testing.T) { + cfg := testModuleConfig() + cfg.ExemptBidders = []string{"house"} + accountID := "acct" + appnexus := openrtb_ext.BidderName("appnexus") + rubicon := openrtb_ext.BidderName("rubicon") + house := openrtb_ext.BidderName("house") + + approvedBid := testBid("approved") + rejectedBid := testBid("rejected") + pendingBid := testBid("pending") + unknownBid := testBid("unknown") + exemptBid := testBid("exempt") + + responses := map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid{ + appnexus: {Bids: []*entities.PbsOrtbBid{approvedBid, rejectedBid, pendingBid, unknownBid}}, + rubicon: {Bids: []*entities.PbsOrtbBid{testBid("seat-removed")}}, + house: {Bids: []*entities.PbsOrtbBid{exemptBid}}, + } + statuses := map[string]approvalStatus{ + creativeApprovalID(accountID, appnexus, "approved"): approvalStatusApproved, + creativeApprovalID(accountID, appnexus, "rejected"): approvalStatusRejected, + creativeApprovalID(accountID, appnexus, "pending"): approvalStatusPending, + } + + removed := filterResponsesByApproval(responses, cfg, accountID, statuses) + + assert.Equal(t, 4, removed) + require.Contains(t, responses, appnexus) + assert.Equal(t, []*entities.PbsOrtbBid{approvedBid}, responses[appnexus].Bids) + assert.NotContains(t, responses, rubicon) + require.Contains(t, responses, house) + assert.Equal(t, []*entities.PbsOrtbBid{exemptBid}, responses[house].Bids) +} + +func TestNeedsApprovalFilter(t *testing.T) { + cfg := testModuleConfig() + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + responses := testResponses(bidder, testBid("approved"), testBid("rejected")) + statuses := map[string]approvalStatus{ + creativeApprovalID(accountID, bidder, "approved"): approvalStatusApproved, + creativeApprovalID(accountID, bidder, "rejected"): approvalStatusRejected, + } + + assert.True(t, needsApprovalFilter(responses, cfg, accountID, statuses)) + + statuses[creativeApprovalID(accountID, bidder, "rejected")] = approvalStatusApproved + assert.False(t, needsApprovalFilter(responses, cfg, accountID, statuses)) +} + +func TestCollectCreativeApprovalsDedupesAndSkipsExemptBidders(t *testing.T) { + cfg := testModuleConfig() + cfg.ExemptBidders = []string{"house"} + accountID := "acct" + appnexus := openrtb_ext.BidderName("appnexus") + house := openrtb_ext.BidderName("house") + + responses := map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid{ + appnexus: {Bids: []*entities.PbsOrtbBid{testBid("same"), testBid("same"), testBid("")}}, + house: {Bids: []*entities.PbsOrtbBid{testBid("exempt")}}, + } + + creatives, warnings := collectCreativeApprovals(responses, cfg, accountID) + + require.Len(t, creatives, 1) + _, ok := creatives[creativeApprovalID(accountID, appnexus, "same")] + assert.True(t, ok) + assert.Len(t, warnings, 1) +} diff --git a/modules/prebid/doohcreativeapproval/module.go b/modules/prebid/doohcreativeapproval/module.go new file mode 100644 index 00000000000..f41be4b6fc7 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/module.go @@ -0,0 +1,157 @@ +package doohcreativeapproval + +import ( + "context" + "encoding/json" + "net/http" + "sort" + + "github.com/prebid/prebid-server/v4/exchange/entities" + "github.com/prebid/prebid-server/v4/hooks/hookexecution" + "github.com/prebid/prebid-server/v4/hooks/hookstage" + "github.com/prebid/prebid-server/v4/modules/moduledeps" + "github.com/prebid/prebid-server/v4/openrtb_ext" +) + +const activeContextKey = "doohcreativeapproval.active" + +var _ hookstage.ProcessedAuctionRequest = (*Module)(nil) +var _ hookstage.AllProcessedBidResponses = (*Module)(nil) + +func Builder(rawConfig json.RawMessage, deps moduledeps.ModuleDeps) (interface{}, error) { + cfg, err := parseModuleConfig(rawConfig) + if err != nil { + return nil, err + } + + client := deps.HTTPClient + if client == nil { + client = http.DefaultClient + } + + return &Module{ + cfg: cfg, + provider: newHTTPApprovalProvider(client), + cache: newApprovalCache(cfg.CacheSizeBytes), + refreshes: newApprovalRefreshCoordinator(cfg.MaxConcurrentLookups), + }, nil +} + +type Module struct { + cfg moduleConfig + provider approvalProvider + cache *approvalCache + refreshes *approvalRefreshCoordinator +} + +func (m *Module) HandleProcessedAuctionHook( + _ context.Context, + miCtx hookstage.ModuleInvocationContext, + payload hookstage.ProcessedAuctionRequestPayload, +) (hookstage.HookResult[hookstage.ProcessedAuctionRequestPayload], error) { + result := hookstage.HookResult[hookstage.ProcessedAuctionRequestPayload]{} + + cfg, err := applyAccountConfig(m.cfg, miCtx.AccountConfig) + if err != nil { + return result, hookexecution.NewFailure("%s", err) + } + if !cfg.Enabled { + return result, nil + } + if cfg.Endpoint == "" { + result.Warnings = append(result.Warnings, "DOOH creative approval endpoint is not configured") + return result, nil + } + if payload.Request == nil || payload.Request.BidRequest == nil || payload.Request.DOOH == nil { + return result, nil + } + + moduleContext := hookstage.NewModuleContext() + moduleContext.Set(activeContextKey, true) + result.ModuleContext = moduleContext + return result, nil +} + +func (m *Module) HandleAllProcessedBidResponsesHook( + _ context.Context, + miCtx hookstage.ModuleInvocationContext, + payload hookstage.AllProcessedBidResponsesPayload, +) (hookstage.HookResult[hookstage.AllProcessedBidResponsesPayload], error) { + result := hookstage.HookResult[hookstage.AllProcessedBidResponsesPayload]{} + if !isModuleContextActive(miCtx.ModuleContext) { + return result, nil + } + + cfg, err := applyAccountConfig(m.cfg, miCtx.AccountConfig) + if err != nil { + return result, hookexecution.NewFailure("%s", err) + } + if !cfg.Enabled || cfg.Endpoint == "" || len(payload.Responses) == 0 { + return result, nil + } + + statuses, warnings := m.resolveApprovalStatuses(cfg, miCtx.AccountID, payload.Responses) + result.Warnings = append(result.Warnings, warnings...) + if !needsApprovalFilter(payload.Responses, cfg, miCtx.AccountID, statuses) { + return result, nil + } + + changeSet := hookstage.ChangeSet[hookstage.AllProcessedBidResponsesPayload]{} + changeSet.AddMutation(func(payload hookstage.AllProcessedBidResponsesPayload) (hookstage.AllProcessedBidResponsesPayload, error) { + filterResponsesByApproval(payload.Responses, cfg, miCtx.AccountID, statuses) + return payload, nil + }, hookstage.MutationUpdate, "responses", "bids") + result.ChangeSet = changeSet + return result, nil +} + +func isModuleContextActive(moduleContext *hookstage.ModuleContext) bool { + activeValue, ok := moduleContext.Get(activeContextKey) + if !ok { + return false + } + active, ok := activeValue.(bool) + return ok && active +} + +func (m *Module) resolveApprovalStatuses( + cfg moduleConfig, + accountID string, + responses map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid, +) (map[string]approvalStatus, []string) { + statuses := make(map[string]approvalStatus) + creativesByID, warnings := collectCreativeApprovals(responses, cfg, accountID) + if len(creativesByID) == 0 { + return statuses, warnings + } + + ids := make([]string, 0, len(creativesByID)) + for id := range creativesByID { + ids = append(ids, id) + } + sort.Strings(ids) + + refreshes := make([]approvalRefresh, 0, len(ids)) + for _, id := range ids { + if cached, ok := m.cache.get(id); ok { + statuses[id] = cached.Status + if !cached.RefreshDue { + continue + } + refreshes = append(refreshes, approvalRefresh{ + Creative: creativesByID[id], + FallbackStatus: cached.Status, + }) + continue + } + + statuses[id] = approvalStatusPending + refreshes = append(refreshes, approvalRefresh{ + Creative: creativesByID[id], + FallbackStatus: approvalStatusPending, + }) + } + warnings = append(warnings, m.scheduleApprovalRefresh(cfg, accountID, refreshes)...) + + return statuses, warnings +} diff --git a/modules/prebid/doohcreativeapproval/module_test.go b/modules/prebid/doohcreativeapproval/module_test.go new file mode 100644 index 00000000000..3ab42e11229 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/module_test.go @@ -0,0 +1,406 @@ +package doohcreativeapproval + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + "testing" + "time" + + "github.com/prebid/openrtb/v20/openrtb2" + "github.com/prebid/prebid-server/v4/hooks/hookstage" + "github.com/prebid/prebid-server/v4/modules/moduledeps" + "github.com/prebid/prebid-server/v4/openrtb_ext" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBuilder(t *testing.T) { + module, err := Builder(json.RawMessage(`{ + "enabled": true, + "endpoint": "http://approval.example.com", + "cache_size_bytes": 1048576, + "max_concurrent_lookups": 2 + }`), moduledeps.ModuleDeps{HTTPClient: http.DefaultClient}) + + require.NoError(t, err) + creativeApprovalModule, ok := module.(*Module) + require.True(t, ok) + assert.Equal(t, "http://approval.example.com", creativeApprovalModule.cfg.Endpoint) + assert.NotNil(t, creativeApprovalModule.provider) + assert.NotNil(t, creativeApprovalModule.cache) + assert.Equal(t, 2, cap(creativeApprovalModule.refreshes.slots)) +} + +func TestBuilderInvalidConfig(t *testing.T) { + module, err := Builder(json.RawMessage(`{"platforms":["site"]}`), moduledeps.ModuleDeps{}) + + require.Error(t, err) + assert.Nil(t, module) +} + +func TestModuleImplementsHooks(t *testing.T) { + module := &Module{} + + assert.Implements(t, (*hookstage.ProcessedAuctionRequest)(nil), module) + assert.Implements(t, (*hookstage.AllProcessedBidResponses)(nil), module) +} + +func TestHandleProcessedAuctionHookActivation(t *testing.T) { + tests := []struct { + name string + module *Module + accountConfig json.RawMessage + request *openrtb_ext.RequestWrapper + active bool + warnings []string + }{ + { + name: "dooh-active", + module: &Module{cfg: testModuleConfig()}, + request: &openrtb_ext.RequestWrapper{BidRequest: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{ID: "screen"}}}, + active: true, + }, + { + name: "site-inactive", + module: &Module{cfg: testModuleConfig()}, + request: &openrtb_ext.RequestWrapper{BidRequest: &openrtb2.BidRequest{Site: &openrtb2.Site{ID: "site"}}}, + active: false, + }, + { + name: "app-inactive", + module: &Module{cfg: testModuleConfig()}, + request: &openrtb_ext.RequestWrapper{BidRequest: &openrtb2.BidRequest{App: &openrtb2.App{ID: "app"}}}, + active: false, + }, + { + name: "nil-request-inactive", + module: &Module{cfg: testModuleConfig()}, + request: nil, + active: false, + }, + { + name: "account-disabled", + module: &Module{cfg: testModuleConfig()}, + accountConfig: testAccountConfig(t, `{"enabled":false}`), + request: &openrtb_ext.RequestWrapper{BidRequest: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{ID: "screen"}}}, + active: false, + }, + { + name: "missing-endpoint", + module: &Module{cfg: moduleConfig{Enabled: true, Platforms: []string{defaultPlatformDOOH}}}, + request: &openrtb_ext.RequestWrapper{BidRequest: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{ID: "screen"}}}, + active: false, + warnings: []string{"DOOH creative approval endpoint is not configured"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result, err := test.module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{AccountConfig: test.accountConfig}, hookstage.ProcessedAuctionRequestPayload{Request: test.request}) + + require.NoError(t, err) + assert.Equal(t, test.warnings, result.Warnings) + assert.Equal(t, test.active, isModuleContextActive(result.ModuleContext)) + }) + } +} + +func TestHandleProcessedAuctionHookInvalidAccountConfig(t *testing.T) { + module := &Module{cfg: testModuleConfig()} + + _, err := module.HandleProcessedAuctionHook( + context.Background(), + hookstage.ModuleInvocationContext{AccountConfig: json.RawMessage(`{"platforms":["site"]}`)}, + hookstage.ProcessedAuctionRequestPayload{Request: &openrtb_ext.RequestWrapper{BidRequest: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{ID: "screen"}}}}, + ) + + require.Error(t, err) + assert.Contains(t, err.Error(), `platforms must contain only "dooh"`) +} + +func TestHandleAllProcessedBidResponsesHookNoActiveContext(t *testing.T) { + provider := &fakeApprovalProvider{} + module := newTestModule(provider, nil) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(openrtb_ext.BidderName("appnexus"), testBid("cr-1"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: "acct"}, payload) + + require.NoError(t, err) + assert.Empty(t, result.ChangeSet.Mutations()) + assert.Equal(t, 0, provider.callCount()) +} + +func TestHandleAllProcessedBidResponsesHookUsesLookupResultsOnLaterAuctions(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + approvedID := creativeApprovalID(accountID, bidder, "approved") + rejectedID := creativeApprovalID(accountID, bidder, "rejected") + provider := &fakeApprovalProvider{ + statuses: map[string]approvalStatus{ + approvedID: approvalStatusApproved, + rejectedID: approvalStatusRejected, + }, + } + module := newTestModule(provider, nil) + + firstPayload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("approved"), testBid("rejected"))} + firstResult, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, firstPayload) + firstPayload = applyAllProcessedMutations(firstPayload, firstResult) + + require.NoError(t, err) + assert.NotContains(t, firstPayload.Responses, bidder) + waitForApprovalRefreshes(t, module) + + secondPayload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("approved"), testBid("rejected"))} + secondResult, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, secondPayload) + secondPayload = applyAllProcessedMutations(secondPayload, secondResult) + + require.NoError(t, err) + assert.Len(t, secondResult.ChangeSet.Mutations(), 1) + require.Contains(t, secondPayload.Responses, bidder) + require.Len(t, secondPayload.Responses[bidder].Bids, 1) + assert.Equal(t, "approved", secondPayload.Responses[bidder].Bids[0].Bid.CrID) + assert.Equal(t, 1, provider.callCount()) +} + +func TestHandleAllProcessedBidResponsesHookMissingResponseCachesPending(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "missing") + provider := &fakeApprovalProvider{} + module := newTestModule(provider, nil) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("missing"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + payload = applyAllProcessedMutations(payload, result) + + require.NoError(t, err) + assert.Len(t, result.ChangeSet.Mutations(), 1) + assert.NotContains(t, payload.Responses, bidder) + waitForApprovalRefreshes(t, module) + lookup, ok := module.cache.get(creativeID) + assert.True(t, ok) + assert.Equal(t, approvalStatusPending, lookup.Status) + assert.False(t, lookup.RefreshDue) +} + +func TestHandleAllProcessedBidResponsesHookProviderErrorCachesPending(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "error") + provider := &fakeApprovalProvider{err: errApprovalProvider} + module := newTestModule(provider, nil) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("error"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + payload = applyAllProcessedMutations(payload, result) + + require.NoError(t, err) + assert.NotContains(t, payload.Responses, bidder) + waitForApprovalRefreshes(t, module) + lookup, ok := module.cache.get(creativeID) + assert.True(t, ok) + assert.Equal(t, approvalStatusPending, lookup.Status) + assert.Equal(t, 1, provider.callCount()) +} + +func TestHandleAllProcessedBidResponsesHookUsesCachedApproval(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "cached") + provider := &fakeApprovalProvider{} + module := newTestModule(provider, nil) + module.cache.set(creativeID, approvalStatusApproved, 60) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("cached"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + + require.NoError(t, err) + assert.Empty(t, result.ChangeSet.Mutations()) + assert.Equal(t, 0, provider.callCount()) +} + +func TestHandleAllProcessedBidResponsesHookUsesStaleCachedApprovalOnProviderError(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "stale-approved") + now := time.Unix(1000, 0) + provider := &fakeApprovalProvider{err: errApprovalProvider} + cache := newApprovalCache(1024 * 1024) + cache.now = func() time.Time { + return now + } + module := newTestModule(provider, cache) + require.NoError(t, module.cache.set(creativeID, approvalStatusApproved, 1)) + now = now.Add(2 * time.Second) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("stale-approved"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + + require.NoError(t, err) + assert.Empty(t, result.ChangeSet.Mutations()) + assert.Empty(t, result.Warnings) + waitForApprovalRefreshes(t, module) + assert.Equal(t, 1, provider.callCount()) + + lookup, ok := module.cache.get(creativeID) + require.True(t, ok) + assert.Equal(t, approvalStatusApproved, lookup.Status) + assert.False(t, lookup.RefreshDue) +} + +func TestHandleAllProcessedBidResponsesHookUsesStaleStatusUntilRefreshCompletes(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "fresh-rejected") + now := time.Unix(1000, 0) + started := make(chan struct{}, 1) + release := make(chan struct{}) + provider := &fakeApprovalProvider{ + statuses: map[string]approvalStatus{ + creativeID: approvalStatusRejected, + }, + started: started, + release: release, + } + cache := newApprovalCache(1024 * 1024) + cache.now = func() time.Time { + return now + } + module := newTestModule(provider, cache) + require.NoError(t, module.cache.set(creativeID, approvalStatusApproved, 1)) + now = now.Add(2 * time.Second) + firstPayload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("fresh-rejected"))} + + firstResult, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, firstPayload) + + require.NoError(t, err) + assert.Empty(t, firstResult.ChangeSet.Mutations()) + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("approval refresh did not start") + } + + secondPayload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("fresh-rejected"))} + secondResult, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, secondPayload) + require.NoError(t, err) + assert.Empty(t, secondResult.ChangeSet.Mutations()) + assert.Equal(t, 1, provider.callCount()) + + close(release) + waitForApprovalRefreshes(t, module) + + thirdPayload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("fresh-rejected"))} + thirdResult, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, thirdPayload) + thirdPayload = applyAllProcessedMutations(thirdPayload, thirdResult) + require.NoError(t, err) + assert.NotContains(t, thirdPayload.Responses, bidder) + + lookup, ok := module.cache.get(creativeID) + require.True(t, ok) + assert.Equal(t, approvalStatusRejected, lookup.Status) + assert.False(t, lookup.RefreshDue) +} + +func TestHandleAllProcessedBidResponsesHookKeepsStaleStatusForIncompleteRefresh(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "stale-approved") + now := time.Unix(1000, 0) + cache := newApprovalCache(1024 * 1024) + cache.now = func() time.Time { return now } + module := newTestModule(&fakeApprovalProvider{}, cache) + require.NoError(t, module.cache.set(creativeID, approvalStatusApproved, 1)) + now = now.Add(2 * time.Second) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("stale-approved"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + + require.NoError(t, err) + assert.Empty(t, result.ChangeSet.Mutations()) + waitForApprovalRefreshes(t, module) + lookup, ok := module.cache.get(creativeID) + require.True(t, ok) + assert.Equal(t, approvalStatusApproved, lookup.Status) + assert.False(t, lookup.RefreshDue) +} + +func TestHandleAllProcessedBidResponsesHookCoalescesConcurrentMisses(t *testing.T) { + const requests = 20 + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "same") + started := make(chan struct{}, 1) + release := make(chan struct{}) + provider := &fakeApprovalProvider{ + statuses: map[string]approvalStatus{creativeID: approvalStatusApproved}, + started: started, + release: release, + } + module := newTestModule(provider, nil) + + var wg sync.WaitGroup + errs := make(chan error, requests) + for i := 0; i < requests; i++ { + wg.Add(1) + go func() { + defer wg.Done() + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("same"))} + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + if err != nil { + errs <- err + return + } + payload = applyAllProcessedMutations(payload, result) + if _, ok := payload.Responses[bidder]; ok { + errs <- fmt.Errorf("first-seen creative was not suppressed") + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + require.NoError(t, err) + } + + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("approval refresh did not start") + } + assert.Equal(t, 1, provider.callCount()) + close(release) + waitForApprovalRefreshes(t, module) +} + +func TestHandleAllProcessedBidResponsesHookSuppressesUnknownWhenCacheWriteFails(t *testing.T) { + accountID := "acct" + bidder := openrtb_ext.BidderName("appnexus") + creativeID := creativeApprovalID(accountID, bidder, "approved") + provider := &fakeApprovalProvider{ + statuses: map[string]approvalStatus{ + creativeID: approvalStatusApproved, + }, + } + cache := newApprovalCache(1024 * 1024) + cache.marshal = func(v any) ([]byte, error) { + return nil, errApprovalProvider + } + module := newTestModule(provider, cache) + payload := hookstage.AllProcessedBidResponsesPayload{Responses: testResponses(bidder, testBid("approved"))} + + result, err := module.HandleAllProcessedBidResponsesHook(context.Background(), hookstage.ModuleInvocationContext{AccountID: accountID, ModuleContext: testActiveModuleContext()}, payload) + payload = applyAllProcessedMutations(payload, result) + + require.NoError(t, err) + assert.NotContains(t, payload.Responses, bidder) + assert.Empty(t, result.Warnings) + waitForApprovalRefreshes(t, module) + assert.Equal(t, 1, provider.callCount()) + _, ok := module.cache.get(creativeID) + assert.False(t, ok) +} diff --git a/modules/prebid/doohcreativeapproval/provider.go b/modules/prebid/doohcreativeapproval/provider.go new file mode 100644 index 00000000000..1bf1f918d51 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/provider.go @@ -0,0 +1,108 @@ +package doohcreativeapproval + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "time" + + "github.com/prebid/prebid-server/v4/util/jsonutil" +) + +type approvalProvider interface { + Lookup(context.Context, moduleConfig, string, []creativeApproval) (map[string]approvalStatus, []string, error) +} + +type httpApprovalProvider struct { + client *http.Client +} + +func newHTTPApprovalProvider(client *http.Client) *httpApprovalProvider { + return &httpApprovalProvider{client: client} +} + +func (p *httpApprovalProvider) Lookup(ctx context.Context, cfg moduleConfig, accountID string, creatives []creativeApproval) (map[string]approvalStatus, []string, error) { + if len(creatives) == 0 { + return nil, nil, nil + } + + requestPayload := approvalRequest{ + AccountID: accountID, + Creatives: creatives, + } + body, err := jsonutil.Marshal(requestPayload) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal approval request: %s", err) + } + + requestCtx, cancel := context.WithTimeout(ctx, time.Duration(cfg.TimeoutMS)*time.Millisecond) + defer cancel() + + req, err := http.NewRequestWithContext(requestCtx, http.MethodPost, cfg.Endpoint, bytes.NewReader(body)) + if err != nil { + return nil, nil, fmt.Errorf("failed to build approval request: %s", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + for name, value := range cfg.Headers { + req.Header.Set(name, value) + } + + resp, err := p.client.Do(req) + if err != nil { + return nil, nil, fmt.Errorf("failed to execute approval request: %s", err) + } + defer resp.Body.Close() + + responseBody, err := io.ReadAll(io.LimitReader(resp.Body, 1024*1024)) + if err != nil { + return nil, nil, fmt.Errorf("failed to read approval response: %s", err) + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, nil, fmt.Errorf("approval endpoint returned status %d", resp.StatusCode) + } + + var response approvalResponse + if err := jsonutil.UnmarshalValid(responseBody, &response); err != nil { + return nil, nil, fmt.Errorf("failed to parse approval response: %s", err) + } + + return parseApprovalResponse(response, creatives) +} + +func parseApprovalResponse(response approvalResponse, requestedCreatives []creativeApproval) (map[string]approvalStatus, []string, error) { + requested := make(map[string]struct{}, len(requestedCreatives)) + for _, creative := range requestedCreatives { + requested[creative.CreativeApprovalID] = struct{}{} + } + + statuses := make(map[string]approvalStatus, len(response.Creatives)) + seen := make(map[string]struct{}, len(response.Creatives)) + warnings := make([]string, 0) + for _, creative := range response.Creatives { + if creative.CreativeApprovalID == "" { + warnings = append(warnings, "approval response creative skipped because creative_approval_id is empty") + continue + } + if _, ok := requested[creative.CreativeApprovalID]; !ok { + warnings = append(warnings, fmt.Sprintf("approval response creative skipped because creative_approval_id %q was not requested", creative.CreativeApprovalID)) + continue + } + if _, duplicate := seen[creative.CreativeApprovalID]; duplicate { + delete(statuses, creative.CreativeApprovalID) + warnings = append(warnings, fmt.Sprintf("approval response creative invalidated because creative_approval_id %q was duplicated", creative.CreativeApprovalID)) + continue + } + seen[creative.CreativeApprovalID] = struct{}{} + if !isValidApprovalStatus(creative.Status) { + warnings = append(warnings, fmt.Sprintf("approval response creative skipped because status %q is not supported", creative.Status)) + continue + } + statuses[creative.CreativeApprovalID] = creative.Status + } + + return statuses, warnings, nil +} diff --git a/modules/prebid/doohcreativeapproval/provider_test.go b/modules/prebid/doohcreativeapproval/provider_test.go new file mode 100644 index 00000000000..f61f6e06b4f --- /dev/null +++ b/modules/prebid/doohcreativeapproval/provider_test.go @@ -0,0 +1,179 @@ +package doohcreativeapproval + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/prebid/prebid-server/v4/util/jsonutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestHTTPApprovalProviderLookup(t *testing.T) { + var gotRequest approvalRequest + client := &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + assert.Equal(t, http.MethodPost, r.Method) + assert.Equal(t, "application/json", r.Header.Get("Content-Type")) + assert.Equal(t, "application/json", r.Header.Get("Accept")) + assert.Equal(t, "secret", r.Header.Get("X-Test-Auth")) + require.NoError(t, json.NewDecoder(r.Body).Decode(&gotRequest)) + + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(`{"creatives":[{"creative_approval_id":"v1:approved","status":"approved"}]}`)), + Header: make(http.Header), + }, nil + })} + + cfg := testModuleConfig() + cfg.Endpoint = "http://approval.example.com" + cfg.Headers = map[string]string{"X-Test-Auth": "secret"} + provider := newHTTPApprovalProvider(client) + + statuses, warnings, err := provider.Lookup(context.Background(), cfg, "acct", []creativeApproval{{ + CreativeApprovalID: "v1:approved", + Bidder: "appnexus", + CreativeID: "cr-123", + }}) + + require.NoError(t, err) + assert.Empty(t, warnings) + assert.Equal(t, "acct", gotRequest.AccountID) + require.Len(t, gotRequest.Creatives, 1) + assert.Equal(t, "v1:approved", gotRequest.Creatives[0].CreativeApprovalID) + assert.Equal(t, map[string]approvalStatus{"v1:approved": approvalStatusApproved}, statuses) +} + +func TestHTTPApprovalProviderLookupErrors(t *testing.T) { + tests := []struct { + name string + handler http.HandlerFunc + expectedErr string + }{ + { + name: "non-2xx", + handler: func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "nope", http.StatusInternalServerError) + }, + expectedErr: "approval endpoint returned status 500", + }, + { + name: "malformed-json", + handler: func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`not-json`)) + }, + expectedErr: "failed to parse approval response", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + cfg := testModuleConfig() + cfg.Endpoint = "http://approval.example.com" + provider := newHTTPApprovalProvider(&http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + recorder := &responseRecorder{header: make(http.Header)} + test.handler(recorder, r) + return recorder.response(), nil + })}) + + _, _, err := provider.Lookup(context.Background(), cfg, "acct", []creativeApproval{{CreativeApprovalID: "v1:creative"}}) + + require.Error(t, err) + assert.Contains(t, err.Error(), test.expectedErr) + }) + } +} + +func TestHTTPApprovalProviderTimeout(t *testing.T) { + cfg := testModuleConfig() + cfg.Endpoint = "http://approval.example.com" + cfg.TimeoutMS = 1 + provider := newHTTPApprovalProvider(&http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) { + select { + case <-time.After(50 * time.Millisecond): + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(`{}`))}, nil + case <-r.Context().Done(): + return nil, r.Context().Err() + } + })}) + + _, _, err := provider.Lookup(context.Background(), cfg, "acct", []creativeApproval{{CreativeApprovalID: "v1:creative"}}) + + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to execute approval request") +} + +func TestParseApprovalResponseWarnings(t *testing.T) { + responseBody := []byte(`{ + "creatives": [ + {"creative_approval_id": "", "status": "approved"}, + {"creative_approval_id": "v1:unknown", "status": "approved"}, + {"creative_approval_id": "v1:bad-status", "status": "maybe"}, + {"creative_approval_id": "v1:approved", "status": "approved"}, + {"creative_approval_id": "v1:approved", "status": "rejected"} + ] + }`) + var response approvalResponse + require.NoError(t, jsonutil.UnmarshalValid(responseBody, &response)) + + statuses, warnings, err := parseApprovalResponse(response, []creativeApproval{ + {CreativeApprovalID: "v1:approved"}, + {CreativeApprovalID: "v1:bad-status"}, + }) + + require.NoError(t, err) + assert.Empty(t, statuses) + assert.Len(t, warnings, 4) +} + +func TestParseApprovalResponseInvalidatesDuplicateAfterInvalidStatus(t *testing.T) { + statuses, warnings, err := parseApprovalResponse(approvalResponse{Creatives: []creativeApprovalResult{ + {CreativeApprovalID: "v1:creative", Status: "unsupported"}, + {CreativeApprovalID: "v1:creative", Status: approvalStatusApproved}, + }}, []creativeApproval{{CreativeApprovalID: "v1:creative"}}) + + require.NoError(t, err) + assert.Empty(t, statuses) + assert.Len(t, warnings, 2) +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +type responseRecorder struct { + statusCode int + header http.Header + body strings.Builder +} + +func (r *responseRecorder) Header() http.Header { + return r.header +} + +func (r *responseRecorder) Write(data []byte) (int, error) { + return r.body.Write(data) +} + +func (r *responseRecorder) WriteHeader(statusCode int) { + r.statusCode = statusCode +} + +func (r *responseRecorder) response() *http.Response { + statusCode := r.statusCode + if statusCode == 0 { + statusCode = http.StatusOK + } + return &http.Response{ + StatusCode: statusCode, + Header: r.header, + Body: io.NopCloser(strings.NewReader(r.body.String())), + } +} diff --git a/modules/prebid/doohcreativeapproval/refresh.go b/modules/prebid/doohcreativeapproval/refresh.go new file mode 100644 index 00000000000..685b5e38afc --- /dev/null +++ b/modules/prebid/doohcreativeapproval/refresh.go @@ -0,0 +1,150 @@ +package doohcreativeapproval + +import ( + "context" + "fmt" + "sync" + + "github.com/prebid/prebid-server/v4/logger" +) + +type approvalRefresh struct { + Creative creativeApproval + FallbackStatus approvalStatus +} + +type approvalRefreshCoordinator struct { + mu sync.Mutex + inFlight map[string]struct{} + slots chan struct{} + wg sync.WaitGroup +} + +func newApprovalRefreshCoordinator(maxConcurrent int) *approvalRefreshCoordinator { + return &approvalRefreshCoordinator{ + inFlight: make(map[string]struct{}), + slots: make(chan struct{}, maxConcurrent), + } +} + +// claim returns capacityAvailable=false only when every background lookup slot is busy. +func (c *approvalRefreshCoordinator) claim(refreshes []approvalRefresh) (claimed []approvalRefresh, capacityAvailable bool) { + if c == nil || len(refreshes) == 0 { + return nil, true + } + + c.mu.Lock() + for _, refresh := range refreshes { + id := refresh.Creative.CreativeApprovalID + if _, ok := c.inFlight[id]; ok { + continue + } + claimed = append(claimed, refresh) + } + if len(claimed) == 0 { + c.mu.Unlock() + return nil, true + } + + select { + case c.slots <- struct{}{}: + default: + c.mu.Unlock() + return nil, false + } + for _, refresh := range claimed { + c.inFlight[refresh.Creative.CreativeApprovalID] = struct{}{} + } + c.wg.Add(1) + c.mu.Unlock() + return claimed, true +} + +func (c *approvalRefreshCoordinator) finish(refreshes []approvalRefresh) { + c.mu.Lock() + for _, refresh := range refreshes { + delete(c.inFlight, refresh.Creative.CreativeApprovalID) + } + <-c.slots + c.mu.Unlock() + c.wg.Done() +} + +func (c *approvalRefreshCoordinator) wait() { + if c != nil { + c.wg.Wait() + } +} + +func (m *Module) scheduleApprovalRefresh(cfg moduleConfig, accountID string, refreshes []approvalRefresh) []string { + if len(refreshes) == 0 { + return nil + } + if m.refreshes == nil { + return []string{"DOOH creative approval refresh coordinator is not configured"} + } + + claimed, capacityAvailable := m.refreshes.claim(refreshes) + if !capacityAvailable { + return nil + } + if len(claimed) == 0 { + return nil + } + + go m.runApprovalRefresh(cfg, accountID, claimed) + return nil +} + +func (m *Module) runApprovalRefresh(cfg moduleConfig, accountID string, refreshes []approvalRefresh) { + defer func() { + if recovered := recover(); recovered != nil { + logger.Errorf("DOOH creative approval refresh panicked: %v", recovered) + m.storeApprovalRefreshFallbacks(cfg, refreshes) + } + m.refreshes.finish(refreshes) + }() + + creatives := make([]creativeApproval, 0, len(refreshes)) + for _, refresh := range refreshes { + creatives = append(creatives, refresh.Creative) + } + + statuses, warnings, err := m.provider.Lookup(context.Background(), cfg, accountID, creatives) + for _, warning := range warnings { + logger.Warnf("DOOH creative approval lookup warning: %s", warning) + } + if err != nil { + logger.Warnf("DOOH creative approval lookup failed: %s", err) + m.storeApprovalRefreshFallbacks(cfg, refreshes) + return + } + + for _, refresh := range refreshes { + id := refresh.Creative.CreativeApprovalID + status, ok := statuses[id] + if !ok || !isValidApprovalStatus(status) { + logger.Warnf("DOOH creative approval response did not contain a usable status for creative_approval_id %s", id) + if err := m.cache.set(id, refresh.FallbackStatus, cfg.PendingTTLSeconds); err != nil { + logger.Warnf("%s", cacheWriteWarning(id, err)) + } + continue + } + if err := m.cache.set(id, status, ttlForStatus(cfg, status)); err != nil { + logger.Warnf("%s", cacheWriteWarning(id, err)) + } + } +} + +func (m *Module) storeApprovalRefreshFallbacks(cfg moduleConfig, refreshes []approvalRefresh) { + for _, refresh := range refreshes { + id := refresh.Creative.CreativeApprovalID + if err := m.cache.set(id, refresh.FallbackStatus, cfg.PendingTTLSeconds); err != nil { + logger.Warnf("%s", cacheWriteWarning(id, err)) + } + } +} + +func cacheWriteWarning(creativeApprovalID string, err error) string { + return fmt.Sprintf("DOOH creative approval cache write failed for creative_approval_id %s: %s", creativeApprovalID, err) +} diff --git a/modules/prebid/doohcreativeapproval/refresh_test.go b/modules/prebid/doohcreativeapproval/refresh_test.go new file mode 100644 index 00000000000..cf73ce0d705 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/refresh_test.go @@ -0,0 +1,49 @@ +package doohcreativeapproval + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestApprovalRefreshCoordinatorCoalescesAndLimitsLookups(t *testing.T) { + coordinator := newApprovalRefreshCoordinator(1) + first := approvalRefresh{Creative: creativeApproval{CreativeApprovalID: "v1:first"}} + second := approvalRefresh{Creative: creativeApproval{CreativeApprovalID: "v1:second"}} + + claimed, capacityAvailable := coordinator.claim([]approvalRefresh{first}) + require.True(t, capacityAvailable) + require.Equal(t, []approvalRefresh{first}, claimed) + + claimed, capacityAvailable = coordinator.claim([]approvalRefresh{first}) + assert.True(t, capacityAvailable) + assert.Empty(t, claimed) + + claimed, capacityAvailable = coordinator.claim([]approvalRefresh{second}) + assert.False(t, capacityAvailable) + assert.Empty(t, claimed) + + coordinator.finish([]approvalRefresh{first}) + + claimed, capacityAvailable = coordinator.claim([]approvalRefresh{second}) + require.True(t, capacityAvailable) + require.Equal(t, []approvalRefresh{second}, claimed) + coordinator.finish([]approvalRefresh{second}) +} + +func TestApprovalRefreshCoordinatorSkipsCreativeAlreadyInFlight(t *testing.T) { + coordinator := newApprovalRefreshCoordinator(2) + refresh := approvalRefresh{Creative: creativeApproval{CreativeApprovalID: "v1:creative"}} + + claimed, capacityAvailable := coordinator.claim([]approvalRefresh{refresh}) + require.True(t, capacityAvailable) + require.Len(t, claimed, 1) + + duplicateClaim, capacityAvailable := coordinator.claim([]approvalRefresh{refresh}) + assert.True(t, capacityAvailable) + assert.Empty(t, duplicateClaim) + assert.Equal(t, 1, len(coordinator.slots)) + + coordinator.finish(claimed) +} diff --git a/modules/prebid/doohcreativeapproval/test_helpers_test.go b/modules/prebid/doohcreativeapproval/test_helpers_test.go new file mode 100644 index 00000000000..d638ff703c5 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/test_helpers_test.go @@ -0,0 +1,158 @@ +package doohcreativeapproval + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "github.com/prebid/openrtb/v20/openrtb2" + "github.com/prebid/prebid-server/v4/exchange/entities" + "github.com/prebid/prebid-server/v4/hooks/hookstage" + "github.com/prebid/prebid-server/v4/openrtb_ext" +) + +func testModuleConfig() moduleConfig { + return moduleConfig{ + Enabled: true, + Platforms: []string{defaultPlatformDOOH}, + Endpoint: "http://approval.example.com/creative-approval", + TimeoutMS: defaultTimeoutMS, + CacheSizeBytes: defaultCacheSizeBytes, + MaxConcurrentLookups: defaultMaxConcurrentLookups, + ApprovedTTLSeconds: defaultApprovedTTLSeconds, + RejectedTTLSeconds: defaultRejectedTTLSeconds, + PendingTTLSeconds: defaultPendingTTLSeconds, + } +} + +func testBid(crid string) *entities.PbsOrtbBid { + return &entities.PbsOrtbBid{ + Bid: &openrtb2.Bid{ + ID: "bid-" + crid, + ImpID: "imp-1", + Price: 1.23, + CrID: crid, + AdID: "ad-" + crid, + CID: "campaign-" + crid, + ADomain: []string{"advertiser.example"}, + Cat: []string{"IAB1"}, + W: 1920, + H: 1080, + Dur: 15, + DealID: "deal-" + crid, + IURL: "https://example.com/" + crid + ".jpg", + }, + BidType: openrtb_ext.BidTypeVideo, + } +} + +func testResponses(bidder openrtb_ext.BidderName, bids ...*entities.PbsOrtbBid) map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid { + return map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid{ + bidder: { + Bids: bids, + Currency: "USD", + Seat: bidder.String(), + }, + } +} + +func testActiveModuleContext() *hookstage.ModuleContext { + moduleContext := hookstage.NewModuleContext() + moduleContext.Set(activeContextKey, true) + return moduleContext +} + +func applyAllProcessedMutations(payload hookstage.AllProcessedBidResponsesPayload, result hookstage.HookResult[hookstage.AllProcessedBidResponsesPayload]) hookstage.AllProcessedBidResponsesPayload { + for _, mutation := range result.ChangeSet.Mutations() { + payload, _ = mutation.Apply(payload) + } + return payload +} + +type fakeApprovalProvider struct { + mu sync.Mutex + statuses map[string]approvalStatus + warnings []string + err error + calls int + creatives []creativeApproval + started chan<- struct{} + release <-chan struct{} +} + +func (p *fakeApprovalProvider) Lookup(ctx context.Context, _ moduleConfig, _ string, creatives []creativeApproval) (map[string]approvalStatus, []string, error) { + p.mu.Lock() + p.calls++ + p.creatives = append([]creativeApproval(nil), creatives...) + statuses := make(map[string]approvalStatus, len(p.statuses)) + for id, status := range p.statuses { + statuses[id] = status + } + warnings := append([]string(nil), p.warnings...) + err := p.err + started := p.started + release := p.release + p.mu.Unlock() + + if started != nil { + select { + case started <- struct{}{}: + default: + } + } + if release != nil { + select { + case <-release: + case <-ctx.Done(): + return nil, warnings, ctx.Err() + } + } + if err != nil { + return nil, warnings, err + } + return statuses, warnings, nil +} + +func (p *fakeApprovalProvider) callCount() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.calls +} + +func newTestModule(provider approvalProvider, cache *approvalCache) *Module { + cfg := testModuleConfig() + if cache == nil { + cache = newApprovalCache(cfg.CacheSizeBytes) + } + return &Module{ + cfg: cfg, + provider: provider, + cache: cache, + refreshes: newApprovalRefreshCoordinator(cfg.MaxConcurrentLookups), + } +} + +func waitForApprovalRefreshes(t *testing.T, module *Module) { + t.Helper() + done := make(chan struct{}) + go func() { + module.refreshes.wait() + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for approval refreshes") + } +} + +func testAccountConfig(t interface{ Helper() }, value string) json.RawMessage { + t.Helper() + return json.RawMessage(value) +} + +var errApprovalProvider = errors.New("approval service unavailable") diff --git a/modules/prebid/doohcreativeapproval/types.go b/modules/prebid/doohcreativeapproval/types.go new file mode 100644 index 00000000000..027e560ec17 --- /dev/null +++ b/modules/prebid/doohcreativeapproval/types.go @@ -0,0 +1,49 @@ +package doohcreativeapproval + +type approvalStatus string + +const ( + approvalStatusApproved approvalStatus = "approved" + approvalStatusRejected approvalStatus = "rejected" + approvalStatusPending approvalStatus = "pending" +) + +type approvalRequest struct { + AccountID string `json:"account_id"` + Creatives []creativeApproval `json:"creatives"` +} + +type approvalResponse struct { + Creatives []creativeApprovalResult `json:"creatives"` +} + +type creativeApproval struct { + CreativeApprovalID string `json:"creative_approval_id"` + Bidder string `json:"bidder"` + CreativeID string `json:"creative_id"` + AdID string `json:"ad_id,omitempty"` + CampaignID string `json:"campaign_id,omitempty"` + AdvertiserDomains []string `json:"advertiser_domains,omitempty"` + Categories []string `json:"categories,omitempty"` + CategoryTaxonomy int `json:"cat_tax,omitempty"` + MediaType string `json:"media_type,omitempty"` + Width int64 `json:"width,omitempty"` + Height int64 `json:"height,omitempty"` + Duration int64 `json:"duration,omitempty"` + DealID string `json:"deal_id,omitempty"` + IURL string `json:"iurl,omitempty"` +} + +type creativeApprovalResult struct { + CreativeApprovalID string `json:"creative_approval_id"` + Status approvalStatus `json:"status"` +} + +func isValidApprovalStatus(status approvalStatus) bool { + switch status { + case approvalStatusApproved, approvalStatusRejected, approvalStatusPending: + return true + default: + return false + } +}