diff --git a/.github/workflows/scripts/codepath-notification b/.github/workflows/scripts/codepath-notification index 8a3bae8daca..d6e57e0103a 100644 --- a/.github/workflows/scripts/codepath-notification +++ b/.github/workflows/scripts/codepath-notification @@ -19,3 +19,4 @@ adapters/ix|imp_ix|ix.json|ix.yaml: pdu-supply-prebid@indexexchange.com medianet: prebid@media.net gumgum: prebid@gumgum.com kargo: kraken@kargo.com +modules/zerogpu: prebid@zerogpu.ai diff --git a/modules/builder.go b/modules/builder.go index df2f1b1c3a7..2d6e83b544a 100644 --- a/modules/builder.go +++ b/modules/builder.go @@ -6,6 +6,7 @@ import ( prebidRulesengine "github.com/prebid/prebid-server/v4/modules/prebid/rulesengine" wurflDevicedetection "github.com/prebid/prebid-server/v4/modules/scientiamobile/wurfl_devicedetection" scope3Rtd "github.com/prebid/prebid-server/v4/modules/scope3/rtd" + zerogpuRtd "github.com/prebid/prebid-server/v4/modules/zerogpu/rtd" ) // builders returns mapping between module name and its builder @@ -25,5 +26,8 @@ func builders() ModuleBuilders { "scope3": { "rtd": scope3Rtd.Builder, }, + "zerogpu": { + "rtd": zerogpuRtd.Builder, + }, } } diff --git a/modules/zerogpu/rtd/README.md b/modules/zerogpu/rtd/README.md new file mode 100644 index 00000000000..d66b35f499d --- /dev/null +++ b/modules/zerogpu/rtd/README.md @@ -0,0 +1,204 @@ +# ZeroGPU Real Time Data Module + +## Overview + +The ZeroGPU RTD module enriches an incoming OpenRTB request with IAB content +categories derived from the publisher's domain. + +On each auction the module resolves the domain from the bid request, classifies +it with ZeroGPU's `zlm-v1-iab-domain-classifier` model, and appends the +resulting categories to `{site,app,dooh}.content.data` as Seller-Defined +contextual segments under `ext.segtax: 6` (IAB Content Taxonomy 2.2). Because +the segments are written to standard First Party Data fields, every bidder in +the auction can read them - no bidder-specific integration is required. + +**The auction never waits on ZeroGPU.** The hook reads an in-process cache and +nothing else. When a domain is not yet cached, the auction proceeds unenriched +and the classification is fetched in the background, so subsequent auctions on +that domain are enriched from memory. Measured against the live API, the hook +costs ~15µs whether the domain is cached or not, while the classification call +it avoids takes ~0.9s. + +Every failure mode is fail-open: if the ZeroGPU API is slow, unreachable, or +returns an error, auctions simply go unenriched. The module never rejects a +request, never delays one, and never creates bids. + +## Prerequisites + +A ZeroGPU API key is required. Sign in at +, or start from +and click **Start Building**. + +API reference: + +- +- + +## Configuration + +The module is disabled unless `enabled` is set. `api_key` is the only required +parameter; everything else has a working default. + +```yaml +hooks: + enabled: true + modules: + zerogpu: + rtd: + enabled: true + api_key: ${ZEROGPU_API_KEY} + host_execution_plan: > + { + "endpoints": { + "/openrtb2/auction": { + "stages": { + "processed_auction_request": { + "groups": [{ + "timeout": 10, + "hook_sequence": [{ + "module_code": "zerogpu.rtd", + "hook_impl_code": "zerogpu-rtd-processed-auction-request" + }] + }] + } + } + } + } + } +``` + +A complete host configuration is in [sample/pbs_example.json](sample/pbs_example.json). + +The group `timeout` only has to cover an in-memory cache read, so a small value +is correct. It is unrelated to `timeout_ms`, which bounds the background +warm-up and never applies to the auction path. + +### Parameters + +| Parameter | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| `api_key` | string | yes | - | ZeroGPU API key, sent as the `x-api-key` header. | +| `endpoint` | string | no | `https://api.zerogpu.ai/v1/responses` | Classification endpoint. Overridable for a different region or a proxy; the Responses API request and response shape is assumed. | +| `model` | string | no | `zlm-v1-iab-domain-classifier` | Model to classify with. | +| `timeout_ms` | int | no | `2000` | HTTP timeout for a background classification call. Never applies to the auction path. | +| `cache_ttl_seconds` | int | no | `86400` | How long a successful classification is cached. | +| `negative_cache_ttl_seconds` | int | no | `300` | How long an empty result or a stable failure (400/401/403/420) is cached. | +| `retry_cache_ttl_seconds` | int | no | `30` | How long a transient failure (timeout, 5xx) is cached before retrying. | +| `cache_size` | int | no | `10485760` | Cache size in bytes. Minimum 524288. | +| `min_score` | float | no | `0.5` | Minimum confidence a category must have to be emitted. | +| `max_segments` | int | no | `0` | Maximum segments per taxonomy. `0` means unlimited. | +| `data_provider_name` | string | no | `zerogpu.ai` | Value written to the `name` field of each injected data object. | +| `enrich_content_1_0` | bool | no | `false` | Also emit IAB Content Taxonomy 1.0 codes under `ext.segtax: 1`. | +| `enrich_user_audience` | bool | no | `false` | Also emit IAB Audience Taxonomy 1.1 segments to `user.data` under `ext.segtax: 4`. See [Privacy](#privacy). | +| `account_filter.allow_list` | []string | no | `[]` | Account IDs permitted to use the module. Empty means all accounts. | + +## Enrichment + +The module runs at the `processed_auction_request` stage - the last point at +which the request is still shared by every bidder, after stored requests have +been merged. Running at `bidder_request` would warm the same domain once per +bidder. + +### Cache warming + +On each auction the module looks the domain up in its local cache: + +* **Hit** - the segments are attached. No I/O. +* **Miss** - the auction returns unenriched, and a background warm-up fetches + the classification and caches it for `cache_ttl_seconds` (24h by default). + +Concurrent auctions for the same uncached domain collapse onto a single +outbound request, so a burst of traffic on a new domain does not produce a +burst of API calls. + +Warm-ups deliberately use the module's own lifetime context rather than the +hook context. The hook context is cancelled the moment the execution plan's +group timeout elapses, which would abort the warm-up and leave the domain +permanently cold. + +The practical cost of this design is that the first impressions on a +newly-seen domain go unenriched - roughly the duration of one classification +call. After that the domain is cached for 24 hours. + +### Domain resolution + +The domain is resolved from the first usable value of `site.domain`, +`site.page`, `site.publisher.domain`, `app.domain`, `app.bundle`, +`app.publisher.domain`, `dooh.domain`, `dooh.publisher.domain`. Values that +carry no domain signal - `localhost`, bare IP addresses, numeric iOS store IDs - +are skipped. If nothing resolves, the module does nothing. + +The value is then normalized so that every spelling of a site shares one cache +entry: the scheme, path, query, port and any trailing dot are dropped, case is +folded, and a leading `www.`, `m.` or `amp.` is removed. So +`https://AMP.Example.com/article?x=1` and `example.com` are one entry, not two. + +Other subdomains are preserved - `blog.example.com` is classified separately +from `example.com`, because they host different content. A variant prefix is +only stripped when at least two labels remain, so `amp.dev` stays `amp.dev`. + +### Injected data + +Given a classification for `coursera.com`, the module appends: + +```json +{ + "site": { + "content": { + "data": [{ + "name": "zerogpu.ai", + "ext": { "segtax": 6 }, + "segment": [{ "id": "132" }, { "id": "148" }] + }] + } + } +} +``` + +Existing `content.data` entries are preserved. Enrichment is idempotent: if a +data object from the same provider already exists for a taxonomy, nothing is +appended. + +Taxonomy identifiers follow the +[IAB segtax registry](https://github.com/InteractiveAdvertisingBureau/openrtb/blob/main/extensions/community_extensions/segtax.md): +`1` for Content Taxonomy 1.0, `4` for Audience Taxonomy 1.1, `6` for Content +Taxonomy 2.2. + +## Privacy + +The default configuration sends only a domain to ZeroGPU and writes only +contextual data. No user identifiers, device data, or geographic information +leave Prebid Server, and nothing is written to user-scoped ORTB fields. + +`enrich_user_audience` is off by default and should stay off unless the host has +made a deliberate decision. When enabled, IAB Audience Taxonomy segments are +written to `user.data`. Two caveats: + +1. The segments are inferred from the domain, not observed from the user. + Publishing them under `user.data` presents contextual inference as + audience data. +2. Prebid requires a module supplying user-level data to check the `enrichUfpd` + [Activity Control](https://docs.prebid.org/prebid-server/features/pbs-activitycontrols.html). + PBS-Go does not currently expose activity controls to modules, so the module + cannot perform that check on the host's behalf. + +The module does not create bids and does not add pixels to creatives. + +## Analytics Tags + +Each invocation emits one activity named +`zerogpu-rtd-domain-classification`: + +| Outcome | Activity status | Result status | Values | +| --- | --- | --- | --- | +| Segments injected | `success` | `modify` | `domain`, `content_2_2_count`, `content_1_0_count`, `audience_count` | +| Nothing to inject | `success` | `allow` | `reason` | + +Because classification happens off the auction path, a failed warm-up is not +attributable to any one auction. Warm-up failures are logged instead: `warn` for +a rejected domain, `error` for an authentication or quota problem, and `info` +for transient failures. + +## Maintainer + + diff --git a/modules/zerogpu/rtd/client.go b/modules/zerogpu/rtd/client.go new file mode 100644 index 00000000000..d17692a5100 --- /dev/null +++ b/modules/zerogpu/rtd/client.go @@ -0,0 +1,332 @@ +package rtd + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "time" + + "github.com/prebid/prebid-server/v4/logger" + "github.com/prebid/prebid-server/v4/util/jsonutil" +) + +const ( + // statusInsufficientQuota is ZeroGPU's non-standard quota-exhausted status. + statusInsufficientQuota = 420 + + // maxResponseBytes bounds how much of an upstream response is read. A + // classification is on the order of a kilobyte. + maxResponseBytes = 1 << 20 +) + +// responsesRequest is the body for the /v1/responses API. +type responsesRequest struct { + Model string `json:"model"` + Input string `json:"input"` +} + +// apiEnvelope is the /v1/responses wrapper. The classification itself is a JSON +// string nested inside it, so parsing happens in two stages. +type apiEnvelope struct { + Output []struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + } `json:"output"` +} + +// payload returns the embedded classification JSON string, or "" when the +// envelope carries none. +func (e apiEnvelope) payload() string { + for _, out := range e.Output { + for _, c := range out.Content { + if c.Text != "" { + return c.Text + } + } + } + return "" +} + +// classification is the inner JSON payload produced by the domain classifier. +type classification struct { + Audience []scoredID `json:"audience"` + Content contentByTaxon `json:"content"` +} + +type contentByTaxon struct { + IAB10 []scoredCode `json:"iab_1_0"` + IAB22 []scoredID `json:"iab_2_2"` +} + +type scoredID struct { + ID int `json:"id"` + Score float64 `json:"score"` +} + +type scoredCode struct { + Code string `json:"code"` + Score float64 `json:"score"` +} + +// segments is the cached, already-filtered result for one domain. Fields are +// abbreviated because every byte is stored in the in-memory cache. +type segments struct { + Content22 []string `json:"c22,omitempty"` + Content10 []string `json:"c10,omitempty"` + Audience []string `json:"aud,omitempty"` +} + +func (s segments) isEmpty() bool { + return len(s.Content22) == 0 && len(s.Content10) == 0 && len(s.Audience) == 0 +} + +// apiError distinguishes conditions that resolve on their own (a cold domain +// still warming ZeroGPU's server-side cache, a 5xx, a timeout) from stable ones +// (bad key, exhausted quota, unclassifiable domain). Only the cache TTL and log +// level differ - the auction proceeds either way. +type apiError struct { + msg string + transient bool +} + +func (e apiError) Error() string { return e.msg } + +func transientErrorf(format string, args ...any) apiError { + return apiError{msg: fmt.Sprintf(format, args...), transient: true} +} + +func permanentErrorf(format string, args ...any) apiError { + return apiError{msg: fmt.Sprintf(format, args...)} +} + +// isTransient reports whether err warrants the short retry TTL. +func isTransient(err error) bool { + var apiErr apiError + if errors.As(err, &apiErr) { + return apiErr.transient + } + // Transport failures and context deadlines are always worth retrying. + return true +} + +func (m *Module) cacheKey(domain string) []byte { + return []byte("zerogpu:" + m.cfg.Model + ":" + domain) +} + +// lookup returns the cached segments for a domain. The second return value +// reports whether the cache had an answer at all - a cached empty result (an +// unclassifiable domain, or a suppressed failure) is a hit carrying no +// segments, which is different from never having asked. +// +// This never performs I/O. The auction is never blocked on ZeroGPU. +func (m *Module) lookup(domain string) (segments, bool) { + key := m.cacheKey(domain) + + cached, err := m.cache.Get(key) + if err != nil { + return segments{}, false + } + + var s segments + if err := jsonutil.Unmarshal(cached, &s); err != nil { + // A corrupt entry is not worth preserving. + m.cache.Del(key) + return segments{}, false + } + return s, true +} + +// warm populates the cache for a domain in the background so later auctions on +// the same domain can be enriched from memory. +// +// Concurrent auctions for the same uncached domain collapse onto a single +// request: only the goroutine that claims the domain fetches. The fetch uses +// the module's own lifetime context, never the hook context, because the hook +// context is cancelled the moment the execution plan's group timeout elapses - +// which would abort the warm-up before it could finish and leave the domain +// permanently cold. +func (m *Module) warm(domain string) { + // Refuse new work once the host is tearing the module down, so no goroutine + // is registered after Shutdown has begun waiting for them. + if m.bgCtx.Err() != nil { + return + } + if _, alreadyWarming := m.inFlight.LoadOrStore(domain, struct{}{}); alreadyWarming { + return + } + + m.wg.Add(1) + go func() { + defer m.wg.Done() + defer m.inFlight.Delete(domain) + + defer func() { + if r := recover(); r != nil { + logger.Errorf("[zerogpu.rtd] panic while warming %q: %v", domain, r) + } + }() + + ctx, cancel := context.WithTimeout(m.bgCtx, time.Duration(m.cfg.Timeout)*time.Millisecond) + defer cancel() + + key := m.cacheKey(domain) + s, err := m.fetch(ctx, domain) + if err != nil { + m.cacheNegative(key, err) + return + } + m.cacheResult(key, s) + }() +} + +// cacheResult stores a successful classification. Empty results use the +// negative TTL so an unclassifiable domain is not re-queried for a full day. +func (m *Module) cacheResult(key []byte, s segments) { + ttl := m.cfg.CacheTTLSeconds + if s.isEmpty() { + ttl = m.cfg.NegativeCacheTTLSeconds + } + encoded, err := jsonutil.Marshal(s) + if err != nil { + return + } + if err := m.cache.Set(key, encoded, ttl); err != nil { + logger.Infof("[zerogpu.rtd] could not cache classification: %v", err) + } +} + +// cacheNegative suppresses repeat calls after a failure. Transient failures get +// the short retry TTL because ZeroGPU warms its server-side cache on the first +// request for a new domain, so retrying shortly is likely to succeed. +func (m *Module) cacheNegative(key []byte, cause error) { + ttl := m.cfg.NegativeCacheTTLSeconds + if isTransient(cause) { + ttl = m.cfg.RetryCacheTTLSeconds + } + if ttl <= 0 { + return + } + if err := m.cache.Set(key, []byte("{}"), ttl); err != nil { + logger.Infof("[zerogpu.rtd] could not negative-cache domain: %v", err) + } +} + +// fetch performs the classification call and filters the response. +func (m *Module) fetch(ctx context.Context, domain string) (segments, error) { + body, err := jsonutil.Marshal(responsesRequest{Model: m.cfg.Model, Input: domain}) + if err != nil { + return segments{}, permanentErrorf("failed to encode request: %s", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, m.cfg.Endpoint, bytes.NewReader(body)) + if err != nil { + return segments{}, permanentErrorf("failed to build request: %s", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-api-key", m.cfg.APIKey) + + resp, err := m.httpClient.Do(req) + if err != nil { + return segments{}, transientErrorf("request to ZeroGPU failed: %s", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return segments{}, statusError(resp.StatusCode, domain) + } + + raw, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + if err != nil { + return segments{}, transientErrorf("failed to read ZeroGPU response: %s", err) + } + + var envelope apiEnvelope + if err := jsonutil.Unmarshal(raw, &envelope); err != nil { + return segments{}, permanentErrorf("failed to decode ZeroGPU response: %s", err) + } + + payload := envelope.payload() + if payload == "" { + return segments{}, permanentErrorf("ZeroGPU response contained no classification payload") + } + + var result classification + if err := jsonutil.Unmarshal([]byte(payload), &result); err != nil { + return segments{}, permanentErrorf("failed to parse classification payload: %s", err) + } + + return m.filter(result), nil +} + +// statusError maps a documented ZeroGPU status onto the retry policy and emits +// a log line at a level matching how actionable the condition is. +func statusError(status int, domain string) apiError { + switch status { + case http.StatusBadRequest: + logger.Warnf("[zerogpu.rtd] ZeroGPU rejected domain %q as a bad request", domain) + return permanentErrorf("ZeroGPU returned status %d", status) + case http.StatusUnauthorized, http.StatusForbidden: + logger.Errorf("[zerogpu.rtd] ZeroGPU returned status %d - check api_key and model access", status) + return permanentErrorf("ZeroGPU returned status %d", status) + case statusInsufficientQuota: + logger.Errorf("[zerogpu.rtd] ZeroGPU returned status %d - insufficient quota", status) + return permanentErrorf("ZeroGPU returned status %d", status) + default: + // 500 and any undocumented status are treated as transient. + logger.Infof("[zerogpu.rtd] ZeroGPU returned status %d for domain %q", status, domain) + return transientErrorf("ZeroGPU returned status %d", status) + } +} + +// filter drops low-confidence categories, applies the per-taxonomy cap and +// converts identifiers to the strings ORTB segments require. Taxonomies the +// host has not enabled are skipped so they are never cached or emitted. +func (m *Module) filter(c classification) segments { + var s segments + + for _, cat := range c.Content.IAB22 { + if cat.Score < m.cfg.MinScore || cat.ID == 0 { + continue + } + if m.capped(len(s.Content22)) { + break + } + s.Content22 = append(s.Content22, strconv.Itoa(cat.ID)) + } + + if m.cfg.EnrichContent10 { + for _, cat := range c.Content.IAB10 { + if cat.Score < m.cfg.MinScore || cat.Code == "" { + continue + } + if m.capped(len(s.Content10)) { + break + } + s.Content10 = append(s.Content10, cat.Code) + } + } + + if m.cfg.EnrichUserAudience { + for _, cat := range c.Audience { + if cat.Score < m.cfg.MinScore || cat.ID == 0 { + continue + } + if m.capped(len(s.Audience)) { + break + } + s.Audience = append(s.Audience, strconv.Itoa(cat.ID)) + } + } + + return s +} + +// capped reports whether the per-taxonomy segment limit has been reached. +func (m *Module) capped(count int) bool { + return m.cfg.MaxSegments > 0 && count >= m.cfg.MaxSegments +} diff --git a/modules/zerogpu/rtd/client_test.go b/modules/zerogpu/rtd/client_test.go new file mode 100644 index 00000000000..a3873ae6255 --- /dev/null +++ b/modules/zerogpu/rtd/client_test.go @@ -0,0 +1,558 @@ +package rtd + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/coocood/freecache" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// classificationJSON is the inner payload returned by the domain classifier, +// taken verbatim from the ZeroGPU API documentation for coursera.com. +const classificationJSON = `{"audience":[{"id":23,"parent_id":20,"name":"Undergraduate Education","tier1_name":"Demographic","score":0.86137356782421},{"id":20,"parent_id":17,"name":"College Education","tier1_name":"Demographic","score":0.8185467622936815}],"content":{"iab_1_0":[{"code":"IAB5","name":"Education","tier":1,"parent_code":null,"score":0.9975345244047042},{"code":"IAB5-6","name":"Distance Learning","tier":2,"parent_code":"IAB5","score":0.9304775059727456}],"iab_2_2":[{"id":132,"parent_id":0,"name":"Education","tier1_name":"Education","score":0.9975345244047042},{"id":148,"parent_id":132,"name":"Online Education","tier1_name":"Education","tier2_name":"Online Education","score":0.9304775059727456}]}}` + +// responsesEnvelope wraps a classification in a /v1/responses envelope. +func responsesEnvelope(payload string) string { + quoted, _ := json.Marshal(payload) + return `{"id":"c961f004","object":"response","status":"completed","model":"` + DefaultModel + + `","output":[{"type":"message","status":"completed","role":"assistant","content":[{"type":"output_text","text":` + + string(quoted) + `,"annotations":[]}]}]}` +} + +// newTestModule builds a module wired to the given server URL. Background +// warm-ups are drained on cleanup so no goroutine outlives the test server. +func newTestModule(t *testing.T, endpoint string, mutate func(*Config)) *Module { + t.Helper() + + cfg := Config{APIKey: "test-key", Endpoint: endpoint} + cfg.applyDefaults() + if mutate != nil { + mutate(&cfg) + } + require.NoError(t, cfg.validate()) + + bgCtx, bgCancel := context.WithCancel(context.Background()) + + module := &Module{ + cfg: cfg, + httpClient: &http.Client{Timeout: time.Duration(cfg.Timeout) * time.Millisecond}, + cache: freecache.NewCache(cfg.CacheSize), + bgCtx: bgCtx, + bgCancel: bgCancel, + } + + // Shutdown waits for in-flight warm-ups, so the test server outlives them. + t.Cleanup(func() { _ = module.Shutdown() }) + + return module +} + +// awaitWarmUps blocks until every in-flight warm-up has finished, using the +// same WaitGroup Shutdown waits on. +func awaitWarmUps(m *Module) { m.wg.Wait() } + +// primeCache warms a domain and waits for it to land in the cache. +func primeCache(t *testing.T, module *Module, domain string) { + t.Helper() + module.warm(domain) + awaitWarmUps(module) + + if _, cached := module.lookup(domain); !cached { + t.Fatalf("warm-up did not populate the cache for %q", domain) + } +} + +func TestFetchSendsResponsesRequest(t *testing.T) { + var gotBody map[string]interface{} + var gotAPIKey, gotContentType, gotMethod string + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod = r.Method + gotAPIKey = r.Header.Get("x-api-key") + gotContentType = r.Header.Get("Content-Type") + require.NoError(t, json.NewDecoder(r.Body).Decode(&gotBody)) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(responsesEnvelope(classificationJSON))) + })) + defer server.Close() + + module := newTestModule(t, server.URL+"/v1/responses", nil) + segs, err := module.fetch(context.Background(), "coursera.com") + require.NoError(t, err) + + assert.Equal(t, http.MethodPost, gotMethod) + assert.Equal(t, "test-key", gotAPIKey) + assert.Equal(t, "application/json", gotContentType) + + // The Responses API takes a bare `input` string. + assert.Equal(t, DefaultModel, gotBody["model"]) + assert.Equal(t, "coursera.com", gotBody["input"]) + assert.Len(t, gotBody, 2, "only model and input should be sent") + + assert.Equal(t, []string{"132", "148"}, segs.Content22) + assert.Empty(t, segs.Content10, "content 1.0 is opt-in") + assert.Empty(t, segs.Audience, "audience is opt-in") +} + +func TestFetchEnrichmentFlags(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + wantContent22 []string + wantContent10 []string + wantAudience []string + }{ + { + name: "defaults emit content 2.2 only", + mutate: nil, + wantContent22: []string{"132", "148"}, + }, + { + name: "content 1.0 enabled", + mutate: func(c *Config) { c.EnrichContent10 = true }, + wantContent22: []string{"132", "148"}, + wantContent10: []string{"IAB5", "IAB5-6"}, + }, + { + name: "audience enabled", + mutate: func(c *Config) { c.EnrichUserAudience = true }, + wantContent22: []string{"132", "148"}, + wantAudience: []string{"23", "20"}, + }, + { + name: "all taxonomies enabled", + mutate: func(c *Config) { + c.EnrichContent10 = true + c.EnrichUserAudience = true + }, + wantContent22: []string{"132", "148"}, + wantContent10: []string{"IAB5", "IAB5-6"}, + wantAudience: []string{"23", "20"}, + }, + { + name: "min_score filters low confidence", + mutate: func(c *Config) { c.MinScore = 0.95; c.EnrichContent10 = true; c.EnrichUserAudience = true }, + wantContent22: []string{"132"}, + wantContent10: []string{"IAB5"}, + wantAudience: nil, + }, + { + name: "max_segments caps each taxonomy", + mutate: func(c *Config) { + c.MaxSegments = 1 + c.EnrichContent10 = true + c.EnrichUserAudience = true + }, + wantContent22: []string{"132"}, + wantContent10: []string{"IAB5"}, + wantAudience: []string{"23"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, test.mutate) + segs, err := module.fetch(context.Background(), "coursera.com") + require.NoError(t, err) + + assert.Equal(t, test.wantContent22, segs.Content22) + assert.Equal(t, test.wantContent10, segs.Content10) + assert.Equal(t, test.wantAudience, segs.Audience) + }) + } +} + +func TestFetchErrorHandling(t *testing.T) { + tests := []struct { + name string + status int + body string + wantErr string + wantTransient bool + }{ + {"bad request is permanent", http.StatusBadRequest, ``, "status 400", false}, + {"unauthorized is permanent", http.StatusUnauthorized, ``, "status 401", false}, + {"forbidden is permanent", http.StatusForbidden, ``, "status 403", false}, + {"insufficient quota is permanent", statusInsufficientQuota, ``, "status 420", false}, + {"server error is transient", http.StatusInternalServerError, ``, "status 500", true}, + {"undocumented status is transient", http.StatusBadGateway, ``, "status 502", true}, + {"unparseable envelope", http.StatusOK, `not json`, "failed to decode", false}, + {"empty envelope", http.StatusOK, `{"output":[]}`, "no classification payload", false}, + {"envelope with blank text", http.StatusOK, `{"output":[{"content":[{"text":""}]}]}`, "no classification payload", false}, + {"unparseable classification", http.StatusOK, responsesEnvelope(`{"content":`), "failed to parse classification", false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := newClassifierServer(t, test.status, test.body) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + _, err := module.fetch(context.Background(), "coursera.com") + + require.Error(t, err) + assert.Contains(t, err.Error(), test.wantErr) + assert.Equal(t, test.wantTransient, isTransient(err)) + }) + } +} + +func TestFetchTransportFailureIsTransient(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + endpoint := server.URL + server.Close() // nothing is listening now + + module := newTestModule(t, endpoint, nil) + _, err := module.fetch(context.Background(), "coursera.com") + + require.Error(t, err) + assert.Contains(t, err.Error(), "request to ZeroGPU failed") + assert.True(t, isTransient(err)) +} + +func TestFetchTimeoutIsTransient(t *testing.T) { + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-release + })) + defer func() { + close(release) + server.Close() + }() + + module := newTestModule(t, server.URL, func(c *Config) { c.Timeout = 20 }) + _, err := module.fetch(context.Background(), "coursera.com") + + require.Error(t, err) + assert.True(t, isTransient(err)) +} + +func TestFetchInvalidEndpointBuildFailure(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + // A control character makes http.NewRequestWithContext fail. + module.cfg.Endpoint = "https://example.com/\x7f" + + _, err := module.fetch(context.Background(), "coursera.com") + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to build request") + assert.False(t, isTransient(err)) +} + +func TestLookupIsCacheOnly(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("lookup must never perform I/O") + })) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + + segs, cached := module.lookup("coursera.com") + assert.False(t, cached, "an unseen domain is not cached") + assert.True(t, segs.isEmpty()) +} + +func TestWarmPopulatesCache(t *testing.T) { + var mu sync.Mutex + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + mu.Unlock() + _, _ = w.Write([]byte(responsesEnvelope(classificationJSON))) + })) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + + // Before warming, the domain is unknown. + _, cached := module.lookup("coursera.com") + require.False(t, cached) + + primeCache(t, module, "coursera.com") + + segs, cached := module.lookup("coursera.com") + assert.True(t, cached) + assert.Equal(t, []string{"132", "148"}, segs.Content22) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 1, calls) +} + +func TestWarmCollapsesConcurrentRequestsForSameDomain(t *testing.T) { + var mu sync.Mutex + var calls int + release := make(chan struct{}) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + mu.Unlock() + <-release // hold the request open so every warm() lands while one is in flight + _, _ = w.Write([]byte(responsesEnvelope(classificationJSON))) + })) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + + // Fire many warm-ups for the same domain while the first is still running. + for i := 0; i < 25; i++ { + module.warm("coursera.com") + } + close(release) + awaitWarmUps(module) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 1, calls, "concurrent warm-ups for one domain must collapse to a single request") +} + +func TestWarmTracksDomainsIndependently(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + + primeCache(t, module, "coursera.com") + primeCache(t, module, "example.com") + + _, cached := module.lookup("coursera.com") + assert.True(t, cached) + _, cached = module.lookup("example.com") + assert.True(t, cached) + _, cached = module.lookup("unseen.com") + assert.False(t, cached) +} + +func TestWarmReleasesDomainAfterCompletion(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + + primeCache(t, module, "coursera.com") + + // The in-flight marker must be cleared, otherwise a domain could never be + // re-warmed after its cache entry expires. + _, stillMarked := module.inFlight.Load("coursera.com") + assert.False(t, stillMarked) +} + +func TestWarmNegativeCacheSuppressesRepeatCalls(t *testing.T) { + var mu sync.Mutex + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + mu.Unlock() + w.WriteHeader(http.StatusUnauthorized) + })) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + + module.warm("coursera.com") + awaitWarmUps(module) + + // The failure is cached, so lookup reports a hit carrying no segments. + segs, cached := module.lookup("coursera.com") + assert.True(t, cached, "a suppressed failure is a cache hit") + assert.True(t, segs.isEmpty()) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 1, calls) +} + +func TestShutdownStopsWarmUps(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + primeCache(t, module, "coursera.com") + + require.NoError(t, module.Shutdown()) + require.NoError(t, module.Shutdown(), "shutdown must be idempotent") + + // After shutdown no new warm-up is registered at all, so nothing races + // against the WaitGroup the host is already waiting on. + module.warm("example.com") + awaitWarmUps(module) + + _, cached := module.lookup("example.com") + assert.False(t, cached, "no warm-up should have run after shutdown") +} + +func TestCacheNegativeTTLSelection(t *testing.T) { + tests := []struct { + name string + err error + wantTTL int + }{ + {"transient uses retry ttl", transientErrorf("boom"), defaultRetryCacheTTLSeconds}, + {"permanent uses negative ttl", permanentErrorf("boom"), defaultNegativeCacheTTLSeconds}, + {"unknown error treated as transient", errors.New("boom"), defaultRetryCacheTTLSeconds}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + key := module.cacheKey("coursera.com") + + module.cacheNegative(key, test.err) + + value, err := module.cache.Get(key) + require.NoError(t, err) + assert.Equal(t, "{}", string(value)) + + ttl, err := module.cache.TTL(key) + require.NoError(t, err) + assert.InDelta(t, test.wantTTL, int(ttl), 2) + }) + } +} + +func TestCacheNegativeSkippedWhenTTLZero(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", func(c *Config) { + c.RetryCacheTTLSeconds = 0 + }) + // applyDefaults already ran, so force the zero explicitly. + module.cfg.RetryCacheTTLSeconds = 0 + + key := module.cacheKey("coursera.com") + module.cacheNegative(key, transientErrorf("boom")) + + _, err := module.cache.Get(key) + assert.Error(t, err, "nothing should have been cached") +} + +func TestCacheResultUsesNegativeTTLForEmptyResult(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + key := module.cacheKey("coursera.com") + + module.cacheResult(key, segments{}) + ttl, err := module.cache.TTL(key) + require.NoError(t, err) + assert.InDelta(t, defaultNegativeCacheTTLSeconds, int(ttl), 2) + + module.cacheResult(key, segments{Content22: []string{"132"}}) + ttl, err = module.cache.TTL(key) + require.NoError(t, err) + assert.InDelta(t, defaultCacheTTLSeconds, int(ttl), 2) +} + +func TestLookupDiscardsCorruptCacheEntry(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + key := module.cacheKey("coursera.com") + require.NoError(t, module.cache.Set(key, []byte("not json"), 60)) + + // A corrupt entry must read as a miss and be evicted, so the domain can be + // re-warmed rather than staying permanently broken. + _, cached := module.lookup("coursera.com") + assert.False(t, cached) + _, err := module.cache.Get(key) + assert.Error(t, err, "the corrupt entry should have been deleted") + + primeCache(t, module, "coursera.com") + segs, cached := module.lookup("coursera.com") + assert.True(t, cached) + assert.Equal(t, []string{"132", "148"}, segs.Content22) +} + +func TestWarmEmptyResultIsNotAnError(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(`{"content":{"iab_2_2":[]}}`)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + module.warm("coursera.com") + awaitWarmUps(module) + + // A classification with no qualifying categories is still cached, so the + // domain is not re-queried on every auction. + segs, cached := module.lookup("coursera.com") + assert.True(t, cached) + assert.True(t, segs.isEmpty()) +} + +func TestFilterSkipsZeroIdentifiers(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", func(c *Config) { + c.EnrichContent10 = true + c.EnrichUserAudience = true + }) + + result := classification{ + Audience: []scoredID{{ID: 0, Score: 0.99}, {ID: 20, Score: 0.99}}, + Content: contentByTaxon{ + IAB10: []scoredCode{{Code: "", Score: 0.99}, {Code: "IAB5", Score: 0.99}}, + IAB22: []scoredID{{ID: 0, Score: 0.99}, {ID: 132, Score: 0.99}}, + }, + } + + segs := module.filter(result) + assert.Equal(t, []string{"132"}, segs.Content22) + assert.Equal(t, []string{"IAB5"}, segs.Content10) + assert.Equal(t, []string{"20"}, segs.Audience) +} + +func TestSegmentsIsEmpty(t *testing.T) { + assert.True(t, segments{}.isEmpty()) + assert.False(t, segments{Content22: []string{"1"}}.isEmpty()) + assert.False(t, segments{Content10: []string{"IAB5"}}.isEmpty()) + assert.False(t, segments{Audience: []string{"1"}}.isEmpty()) +} + +func TestApiEnvelopePayload(t *testing.T) { + tests := []struct { + name string + raw string + want string + }{ + {"real envelope", responsesEnvelope("the-payload"), "the-payload"}, + {"first non-empty text wins", `{"output":[{"content":[{"text":""},{"text":"second"}]}]}`, "second"}, + {"skips empty message", `{"output":[{"content":[]},{"content":[{"text":"later"}]}]}`, "later"}, + {"no output", `{"output":[]}`, ""}, + {"absent output", `{}`, ""}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var envelope apiEnvelope + require.NoError(t, json.Unmarshal([]byte(test.raw), &envelope)) + assert.Equal(t, test.want, envelope.payload()) + }) + } + + assert.Empty(t, apiEnvelope{}.payload()) +} + +func TestCacheKeyVariesByModel(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + other := newTestModule(t, "https://example.com/v1/responses", func(c *Config) { c.Model = "other-model" }) + + assert.NotEqual(t, string(module.cacheKey("a.com")), string(other.cacheKey("a.com"))) + assert.NotEqual(t, string(module.cacheKey("a.com")), string(module.cacheKey("b.com"))) +} + +// newClassifierServer returns a server replying with a fixed status and body. +func newClassifierServer(t *testing.T, status int, body string) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) + })) +} diff --git a/modules/zerogpu/rtd/config.go b/modules/zerogpu/rtd/config.go new file mode 100644 index 00000000000..d5451edf353 --- /dev/null +++ b/modules/zerogpu/rtd/config.go @@ -0,0 +1,165 @@ +package rtd + +import ( + "encoding/json" + "errors" + "fmt" + "net/url" + "slices" + + "github.com/prebid/prebid-server/v4/util/jsonutil" +) + +const ( + // DefaultEndpoint is the ZeroGPU Responses API. It is overridable so a host + // can point at a different region or a proxy, but the request and response + // shapes are always those of the Responses API. + DefaultEndpoint = "https://api.zerogpu.ai/v1/responses" + + // DefaultModel is the IAB domain classifier model. + DefaultModel = "zlm-v1-iab-domain-classifier" + + // DefaultDataProviderName is written to the `name` field of every injected + // ORTB data object so buyers can attribute the segments. + DefaultDataProviderName = "zerogpu.ai" + + // defaultTimeoutMs bounds a background cache warm-up, not anything on the + // auction path, so it is sized generously against measured API latency + // rather than against the auction's latency budget. + defaultTimeoutMs = 2000 + defaultCacheTTLSeconds = 86400 // 24h - domain classifications are stable + defaultNegativeCacheTTLSeconds = 300 // empty result, 400/401/403/420 + defaultRetryCacheTTLSeconds = 30 // timeout / 5xx - cold domains warm up server-side + defaultCacheSize = 10 * 1024 * 1024 + defaultMinScore = 0.5 + + // freecache rejects entries larger than 1/1024 of the cache size, so keep a + // floor that comfortably holds a serialized classification. + minCacheSize = 512 * 1024 +) + +// Config holds the host-level module configuration. Account-level config uses +// the same shape, but only AccountFilter and the enrichment toggles are +// meaningful there - credentials and cache sizing stay host-side. +type Config struct { + APIKey string `json:"api_key"` + Endpoint string `json:"endpoint"` + Model string `json:"model"` + + Timeout int `json:"timeout_ms"` + CacheTTLSeconds int `json:"cache_ttl_seconds"` + NegativeCacheTTLSeconds int `json:"negative_cache_ttl_seconds"` + RetryCacheTTLSeconds int `json:"retry_cache_ttl_seconds"` + CacheSize int `json:"cache_size"` + + MinScore float64 `json:"min_score"` + MaxSegments int `json:"max_segments"` + DataProviderName string `json:"data_provider_name"` + + // EnrichContent10 appends a second content data object carrying deprecated + // IAB Content Taxonomy 1.0 codes under segtax 1. + EnrichContent10 bool `json:"enrich_content_1_0"` + + // EnrichUserAudience appends IAB Audience Taxonomy 1.1 segments to + // user.data under segtax 4. Off by default - see README for the privacy + // rationale. + EnrichUserAudience bool `json:"enrich_user_audience"` + + AccountFilter AccountFilter `json:"account_filter"` +} + +// AccountFilter restricts a host-enabled module to a subset of accounts. An +// empty allow list means every account is served. +type AccountFilter struct { + AllowList []string `json:"allow_list"` +} + +// isAllowed reports whether the given account may use the module. +func (f AccountFilter) isAllowed(accountID string) bool { + if len(f.AllowList) == 0 { + return true + } + return slices.Contains(f.AllowList, accountID) +} + +// newConfig unmarshals, defaults and validates the module configuration. +func newConfig(data json.RawMessage) (Config, error) { + var cfg Config + if len(data) > 0 { + if err := jsonutil.UnmarshalValid(data, &cfg); err != nil { + return cfg, fmt.Errorf("failed to parse config: %s", err) + } + } + cfg.applyDefaults() + return cfg, cfg.validate() +} + +func (c *Config) applyDefaults() { + if c.Endpoint == "" { + c.Endpoint = DefaultEndpoint + } + if c.Model == "" { + c.Model = DefaultModel + } + if c.DataProviderName == "" { + c.DataProviderName = DefaultDataProviderName + } + if c.Timeout == 0 { + c.Timeout = defaultTimeoutMs + } + if c.CacheTTLSeconds == 0 { + c.CacheTTLSeconds = defaultCacheTTLSeconds + } + if c.NegativeCacheTTLSeconds == 0 { + c.NegativeCacheTTLSeconds = defaultNegativeCacheTTLSeconds + } + if c.RetryCacheTTLSeconds == 0 { + c.RetryCacheTTLSeconds = defaultRetryCacheTTLSeconds + } + if c.CacheSize == 0 { + c.CacheSize = defaultCacheSize + } + if c.MinScore == 0 { + c.MinScore = defaultMinScore + } +} + +func (c *Config) validate() error { + if c.APIKey == "" { + return errors.New("api_key is required") + } + + parsed, err := url.Parse(c.Endpoint) + if err != nil { + return fmt.Errorf("endpoint is not a valid URL: %s", err) + } + if parsed.Scheme != "https" && parsed.Scheme != "http" { + return errors.New("endpoint must be an http or https URL") + } + if parsed.Host == "" { + return errors.New("endpoint must include a host") + } + + if c.Timeout < 0 { + return errors.New("timeout_ms cannot be negative") + } + if c.CacheTTLSeconds < 0 { + return errors.New("cache_ttl_seconds cannot be negative") + } + if c.NegativeCacheTTLSeconds < 0 { + return errors.New("negative_cache_ttl_seconds cannot be negative") + } + if c.RetryCacheTTLSeconds < 0 { + return errors.New("retry_cache_ttl_seconds cannot be negative") + } + if c.CacheSize < minCacheSize { + return fmt.Errorf("cache_size must be at least %d bytes", minCacheSize) + } + if c.MinScore < 0 || c.MinScore > 1 { + return errors.New("min_score must be between 0 and 1") + } + if c.MaxSegments < 0 { + return errors.New("max_segments cannot be negative") + } + return nil +} diff --git a/modules/zerogpu/rtd/config_test.go b/modules/zerogpu/rtd/config_test.go new file mode 100644 index 00000000000..ee89d1ed505 --- /dev/null +++ b/modules/zerogpu/rtd/config_test.go @@ -0,0 +1,128 @@ +package rtd + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewConfigDefaults(t *testing.T) { + cfg, err := newConfig(json.RawMessage(`{"api_key": "key"}`)) + require.NoError(t, err) + + assert.Equal(t, DefaultEndpoint, cfg.Endpoint) + assert.Equal(t, DefaultModel, cfg.Model) + assert.Equal(t, DefaultDataProviderName, cfg.DataProviderName) + assert.Equal(t, defaultTimeoutMs, cfg.Timeout) + assert.Equal(t, defaultCacheTTLSeconds, cfg.CacheTTLSeconds) + assert.Equal(t, defaultNegativeCacheTTLSeconds, cfg.NegativeCacheTTLSeconds) + assert.Equal(t, defaultRetryCacheTTLSeconds, cfg.RetryCacheTTLSeconds) + assert.Equal(t, defaultCacheSize, cfg.CacheSize) + assert.Equal(t, defaultMinScore, cfg.MinScore) + assert.Zero(t, cfg.MaxSegments) + assert.False(t, cfg.EnrichContent10) + assert.False(t, cfg.EnrichUserAudience) +} + +func TestNewConfigOverrides(t *testing.T) { + raw := json.RawMessage(`{ + "api_key": "key", + "endpoint": "https://example.com/v1/responses", + "model": "custom-model", + "timeout_ms": 250, + "cache_ttl_seconds": 10, + "negative_cache_ttl_seconds": 11, + "retry_cache_ttl_seconds": 12, + "cache_size": 1048576, + "min_score": 0.9, + "max_segments": 3, + "data_provider_name": "custom.example", + "enrich_content_1_0": true, + "enrich_user_audience": true, + "account_filter": {"allow_list": ["1001"]} + }`) + + cfg, err := newConfig(raw) + require.NoError(t, err) + + assert.Equal(t, "https://example.com/v1/responses", cfg.Endpoint) + assert.Equal(t, "custom-model", cfg.Model) + assert.Equal(t, 250, cfg.Timeout) + assert.Equal(t, 10, cfg.CacheTTLSeconds) + assert.Equal(t, 11, cfg.NegativeCacheTTLSeconds) + assert.Equal(t, 12, cfg.RetryCacheTTLSeconds) + assert.Equal(t, 1048576, cfg.CacheSize) + assert.InDelta(t, 0.9, cfg.MinScore, 0.0001) + assert.Equal(t, 3, cfg.MaxSegments) + assert.Equal(t, "custom.example", cfg.DataProviderName) + assert.True(t, cfg.EnrichContent10) + assert.True(t, cfg.EnrichUserAudience) + assert.Equal(t, []string{"1001"}, cfg.AccountFilter.AllowList) +} + +func TestNewConfigErrors(t *testing.T) { + tests := []struct { + name string + raw string + wantErr string + }{ + {"malformed json", `{"api_key":`, "failed to parse config"}, + {"empty config", `{}`, "api_key is required"}, + {"missing api key", `{"endpoint":"https://example.com"}`, "api_key is required"}, + {"non http scheme", `{"api_key":"k","endpoint":"ftp://example.com/x"}`, "must be an http or https URL"}, + {"no host", `{"api_key":"k","endpoint":"https:///v1/responses"}`, "must include a host"}, + {"bad url", `{"api_key":"k","endpoint":"https://exa mple.com"}`, "not a valid URL"}, + {"negative timeout", `{"api_key":"k","timeout_ms":-1}`, "timeout_ms cannot be negative"}, + {"negative cache ttl", `{"api_key":"k","cache_ttl_seconds":-1}`, "cache_ttl_seconds cannot be negative"}, + {"negative negative ttl", `{"api_key":"k","negative_cache_ttl_seconds":-1}`, "negative_cache_ttl_seconds cannot be negative"}, + {"negative retry ttl", `{"api_key":"k","retry_cache_ttl_seconds":-1}`, "retry_cache_ttl_seconds cannot be negative"}, + {"cache too small", `{"api_key":"k","cache_size":1024}`, "cache_size must be at least"}, + {"min score too high", `{"api_key":"k","min_score":1.5}`, "min_score must be between 0 and 1"}, + {"min score negative", `{"api_key":"k","min_score":-0.5}`, "min_score must be between 0 and 1"}, + {"negative max segments", `{"api_key":"k","max_segments":-1}`, "max_segments cannot be negative"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + _, err := newConfig(json.RawMessage(test.raw)) + require.Error(t, err) + assert.Contains(t, err.Error(), test.wantErr) + }) + } +} + +func TestNewConfigEmptyRawMessage(t *testing.T) { + _, err := newConfig(nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "api_key is required") +} + +func TestAccountFilterIsAllowed(t *testing.T) { + tests := []struct { + name string + allowList []string + accountID string + want bool + }{ + {"empty list allows all", nil, "1001", true}, + {"empty list allows unknown account", []string{}, "", true}, + {"listed account allowed", []string{"1001", "1002"}, "1002", true}, + {"unlisted account denied", []string{"1001"}, "9999", false}, + {"empty account denied when list set", []string{"1001"}, "", false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + filter := AccountFilter{AllowList: test.allowList} + assert.Equal(t, test.want, filter.isAllowed(test.accountID)) + }) + } +} + +func TestDefaultEndpointIsResponsesAPI(t *testing.T) { + cfg, err := newConfig(json.RawMessage(`{"api_key": "key"}`)) + require.NoError(t, err) + assert.Equal(t, "https://api.zerogpu.ai/v1/responses", cfg.Endpoint) +} diff --git a/modules/zerogpu/rtd/enrich.go b/modules/zerogpu/rtd/enrich.go new file mode 100644 index 00000000000..f1c89fc29ba --- /dev/null +++ b/modules/zerogpu/rtd/enrich.go @@ -0,0 +1,243 @@ +package rtd + +import ( + "net/url" + "strings" + + "github.com/prebid/openrtb/v20/openrtb2" + "github.com/prebid/prebid-server/v4/util/jsonutil" +) + +// Segment taxonomy identifiers from the IAB segtax registry: +// https://github.com/InteractiveAdvertisingBureau/openrtb/blob/main/extensions/community_extensions/segtax.md +const ( + segtaxContent10 = 1 // IAB Content Taxonomy 1.0 (deprecated) + segtaxAudience11 = 4 // IAB Audience Taxonomy 1.1 + segtaxContent22 = 6 // IAB Content Taxonomy 2.2 +) + +// segtaxExt is the ext object attached to every injected data segment. +type segtaxExt struct { + Segtax int `json:"segtax"` +} + +// resolveDomain picks the most specific domain available on the request. Order +// matters: an explicitly declared domain beats one parsed out of a page URL, +// which in turn beats the publisher's own domain. +func resolveDomain(r *openrtb2.BidRequest) string { + if r == nil { + return "" + } + + var candidates []string + switch { + case r.Site != nil: + candidates = []string{r.Site.Domain, r.Site.Page} + if r.Site.Publisher != nil { + candidates = append(candidates, r.Site.Publisher.Domain) + } + case r.App != nil: + candidates = []string{r.App.Domain, r.App.Bundle} + if r.App.Publisher != nil { + candidates = append(candidates, r.App.Publisher.Domain) + } + case r.DOOH != nil: + candidates = []string{r.DOOH.Domain} + if r.DOOH.Publisher != nil { + candidates = append(candidates, r.DOOH.Publisher.Domain) + } + } + + for _, candidate := range candidates { + if domain := normalizeDomain(candidate); domain != "" { + return domain + } + } + return "" +} + +// normalizeDomain reduces a domain, page URL or app bundle to a bare lowercase +// hostname. It returns "" for values that carry no domain signal. +func normalizeDomain(raw string) string { + value := strings.ToLower(strings.TrimSpace(raw)) + if value == "" { + return "" + } + + if strings.Contains(value, "://") { + parsed, err := url.Parse(value) + if err != nil { + return "" + } + value = parsed.Hostname() + } else { + // Strip any path and port left on a bare host. + value, _, _ = strings.Cut(value, "/") + value, _, _ = strings.Cut(value, ":") + } + + value = stripHostPrefix(value) + value = strings.Trim(value, ".") + + if value == "" || !strings.Contains(value, ".") { + // Rules out "localhost" and iOS store IDs, which the domain classifier + // cannot interpret. + return "" + } + if isAllDigitsAndDots(value) { + // Bare IP addresses and numeric app bundles carry no domain semantics. + return "" + } + return value +} + +// hostPrefixes are alternate spellings of the same site. Stripping them means +// the desktop, mobile and AMP variants share one cache entry and one +// classification instead of three. +// +// Other subdomains are deliberately preserved: blog., shop. and support. host +// genuinely different content, and collapsing them would mislabel inventory. +var hostPrefixes = []string{"www.", "m.", "amp."} + +// stripHostPrefix removes a leading variant prefix, but only when at least two +// labels remain. Without that guard `amp.dev` would reduce to `dev` and be +// discarded as a single-label host. +func stripHostPrefix(host string) string { + for _, prefix := range hostPrefixes { + if rest, found := strings.CutPrefix(host, prefix); found && strings.Contains(rest, ".") { + return rest + } + } + return host +} + +func isAllDigitsAndDots(s string) bool { + for _, r := range s { + if (r < '0' || r > '9') && r != '.' { + return false + } + } + return true +} + +// enrich writes the resolved segments onto the request. It reports whether +// anything was actually added, so the caller can skip registering a no-op +// mutation. +func (m *Module) enrich(r *openrtb2.BidRequest, s segments) bool { + if r == nil { + return false + } + + changed := false + + // Only materialize a content object if there is something to put in it. + if len(s.Content22) > 0 || len(s.Content10) > 0 { + if content := contentOf(r); content != nil { + if data, ok := m.buildData(content.Data, s.Content22, segtaxContent22); ok { + content.Data = append(content.Data, data) + changed = true + } + if data, ok := m.buildData(content.Data, s.Content10, segtaxContent10); ok { + content.Data = append(content.Data, data) + changed = true + } + } + } + + if len(s.Audience) > 0 { + var existing []openrtb2.Data + if r.User != nil { + existing = r.User.Data + } + if data, ok := m.buildData(existing, s.Audience, segtaxAudience11); ok { + if r.User == nil { + r.User = &openrtb2.User{} + } + r.User.Data = append(r.User.Data, data) + changed = true + } + } + + return changed +} + +// buildData assembles one ORTB data object, or reports false when there is +// nothing to add or an equivalent object is already present. +func (m *Module) buildData(existing []openrtb2.Data, ids []string, segtax int) (openrtb2.Data, bool) { + if len(ids) == 0 || hasDataEntry(existing, m.cfg.DataProviderName, segtax) { + return openrtb2.Data{}, false + } + + ext, err := jsonutil.Marshal(segtaxExt{Segtax: segtax}) + if err != nil { + return openrtb2.Data{}, false + } + + segs := make([]openrtb2.Segment, 0, len(ids)) + for _, id := range ids { + segs = append(segs, openrtb2.Segment{ID: id}) + } + + return openrtb2.Data{ + Name: m.cfg.DataProviderName, + Segment: segs, + Ext: ext, + }, true +} + +// hasDataEntry keeps enrichment idempotent: if this provider already published +// segments for this taxonomy - because an upstream module ran, or because the +// hook is somehow invoked twice - do not append a duplicate. +func hasDataEntry(data []openrtb2.Data, name string, segtax int) bool { + for _, entry := range data { + if entry.Name != name || len(entry.Ext) == 0 { + continue + } + var ext segtaxExt + if err := jsonutil.Unmarshal(entry.Ext, &ext); err != nil { + continue + } + if ext.Segtax == segtax { + return true + } + } + return false +} + +// contentOf returns the content object for whichever distribution channel the +// request declares, creating it when absent. ORTB permits only one of +// site/app/dooh per request. +func contentOf(r *openrtb2.BidRequest) *openrtb2.Content { + switch { + case r.Site != nil: + if r.Site.Content == nil { + r.Site.Content = &openrtb2.Content{} + } + return r.Site.Content + case r.App != nil: + if r.App.Content == nil { + r.App.Content = &openrtb2.Content{} + } + return r.App.Content + case r.DOOH != nil: + if r.DOOH.Content == nil { + r.DOOH.Content = &openrtb2.Content{} + } + return r.DOOH.Content + } + return nil +} + +// mutationKey names the field the mutation touches, for the module trace shown +// under ext.prebid.modules. +func mutationKey(r *openrtb2.BidRequest) []string { + switch { + case r.Site != nil: + return []string{"bidrequest", "site", "content", "data"} + case r.App != nil: + return []string{"bidrequest", "app", "content", "data"} + case r.DOOH != nil: + return []string{"bidrequest", "dooh", "content", "data"} + } + return []string{"bidrequest"} +} diff --git a/modules/zerogpu/rtd/enrich_test.go b/modules/zerogpu/rtd/enrich_test.go new file mode 100644 index 00000000000..4d7794977fc --- /dev/null +++ b/modules/zerogpu/rtd/enrich_test.go @@ -0,0 +1,401 @@ +package rtd + +import ( + "testing" + + "github.com/prebid/openrtb/v20/openrtb2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNormalizeDomain(t *testing.T) { + tests := []struct { + name string + raw string + want string + }{ + {"plain domain", "coursera.com", "coursera.com"}, + {"uppercase is lowered", "Coursera.COM", "coursera.com"}, + {"surrounding whitespace", " coursera.com ", "coursera.com"}, + {"www prefix stripped", "www.coursera.com", "coursera.com"}, + {"mobile prefix stripped", "m.coursera.com", "coursera.com"}, + {"amp prefix stripped", "amp.coursera.com", "coursera.com"}, + {"amp prefix stripped from url", "https://amp.coursera.com/learn", "coursera.com"}, + {"amp.dev is not reduced to dev", "amp.dev", "amp.dev"}, + {"m.co is not reduced to co", "m.co", "m.co"}, + {"www2 is not a variant prefix", "www2.coursera.com", "www2.coursera.com"}, + {"other subdomains preserved", "shop.coursera.com", "shop.coursera.com"}, + {"prefix only stripped at the front", "coursera.m.com", "coursera.m.com"}, + {"https url", "https://www.coursera.com/learn/python?ref=nav", "coursera.com"}, + {"http url with port", "http://coursera.com:8080/learn", "coursera.com"}, + {"bare host with path", "coursera.com/learn/python", "coursera.com"}, + {"bare host with port", "coursera.com:443", "coursera.com"}, + {"trailing dot", "coursera.com.", "coursera.com"}, + {"subdomain preserved", "blog.coursera.com", "blog.coursera.com"}, + {"android bundle kept", "com.example.app", "com.example.app"}, + {"empty", "", ""}, + {"whitespace only", " ", ""}, + {"localhost rejected", "localhost", ""}, + {"single label rejected", "intranet", ""}, + {"ios store id rejected", "123456789", ""}, + {"ipv4 rejected", "192.168.1.1", ""}, + {"url with no host", "https:///learn", ""}, + {"malformed url", "https://exa mple.com/x", ""}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, normalizeDomain(test.raw)) + }) + } +} + +// TestDomainVariantsShareCacheKey pins the property the cache depends on: +// every spelling of the same site must normalize to one key, otherwise each +// variant would trigger its own classification call and its own cache entry. +func TestDomainVariantsShareCacheKey(t *testing.T) { + variants := []string{ + "coursera.com", + "www.coursera.com", + "WWw.coursera.com", + "COURSERA.COM", + "m.coursera.com", + "amp.coursera.com", + "https://m.coursera.com/learn/python", + "https://amp.coursera.com/learn", + "AMP.Coursera.com", + "coursera.com/xyz", + "www.coursera.com/xyz", + "https://www.coursera.com", + "https://www.coursera.com/", + "https://coursera.com/learn/python?ref=nav#top", + "HTTPS://WWW.Coursera.COM/XYZ", + "http://coursera.com:8080/learn", + "coursera.com:443", + "coursera.com.", + " coursera.com ", + } + + module := newTestModule(t, "https://example.com/v1/responses", nil) + want := string(module.cacheKey("coursera.com")) + + for _, variant := range variants { + t.Run(variant, func(t *testing.T) { + assert.Equal(t, "coursera.com", normalizeDomain(variant)) + assert.Equal(t, want, string(module.cacheKey(normalizeDomain(variant))), + "%q must share a cache entry with coursera.com", variant) + }) + } +} + +// TestRequestVariantsShareCacheKey covers the same property end to end: the +// domain may arrive on any of several ORTB fields, in any spelling. +func TestRequestVariantsShareCacheKey(t *testing.T) { + requests := map[string]*openrtb2.BidRequest{ + "site.domain": {Site: &openrtb2.Site{Domain: "coursera.com"}}, + "site.domain with www": {Site: &openrtb2.Site{Domain: "WWW.Coursera.com"}}, + "site.page url": {Site: &openrtb2.Site{Page: "https://www.coursera.com/learn/python"}}, + "site.page bare host": {Site: &openrtb2.Site{Page: "coursera.com/learn"}}, + "site.publisher.domain": {Site: &openrtb2.Site{Publisher: &openrtb2.Publisher{Domain: "coursera.com."}}}, + "app.domain": {App: &openrtb2.App{Domain: "https://coursera.com"}}, + "dooh.domain": {DOOH: &openrtb2.DOOH{Domain: "www.coursera.com:8080"}}, + } + + module := newTestModule(t, "https://example.com/v1/responses", nil) + want := string(module.cacheKey("coursera.com")) + + for name, request := range requests { + t.Run(name, func(t *testing.T) { + domain := resolveDomain(request) + assert.Equal(t, "coursera.com", domain) + assert.Equal(t, want, string(module.cacheKey(domain))) + }) + } +} + +func TestResolveDomain(t *testing.T) { + tests := []struct { + name string + request *openrtb2.BidRequest + want string + }{ + {"nil request", nil, ""}, + {"empty request", &openrtb2.BidRequest{}, ""}, + { + name: "site domain wins", + request: &openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com", Page: "https://other.com/x"}}, + want: "coursera.com", + }, + { + name: "falls back to site page", + request: &openrtb2.BidRequest{Site: &openrtb2.Site{Page: "https://www.coursera.com/learn/python"}}, + want: "coursera.com", + }, + { + name: "falls back to site publisher domain", + request: &openrtb2.BidRequest{Site: &openrtb2.Site{ + Publisher: &openrtb2.Publisher{Domain: "publisher.com"}, + }}, + want: "publisher.com", + }, + { + name: "skips unusable site page", + request: &openrtb2.BidRequest{Site: &openrtb2.Site{ + Page: "http://localhost:8080/test", + Publisher: &openrtb2.Publisher{Domain: "publisher.com"}, + }}, + want: "publisher.com", + }, + { + name: "app domain", + request: &openrtb2.BidRequest{App: &openrtb2.App{Domain: "example.com", Bundle: "com.example.app"}}, + want: "example.com", + }, + { + name: "app bundle fallback", + request: &openrtb2.BidRequest{App: &openrtb2.App{Bundle: "com.example.app"}}, + want: "com.example.app", + }, + { + name: "app skips numeric ios bundle", + request: &openrtb2.BidRequest{App: &openrtb2.App{ + Bundle: "123456789", + Publisher: &openrtb2.Publisher{Domain: "publisher.com"}, + }}, + want: "publisher.com", + }, + { + name: "dooh domain", + request: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{Domain: "screens.example.com"}}, + want: "screens.example.com", + }, + { + name: "dooh publisher fallback", + request: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{ + Publisher: &openrtb2.Publisher{Domain: "publisher.com"}, + }}, + want: "publisher.com", + }, + { + name: "site takes precedence over app", + request: &openrtb2.BidRequest{ + Site: &openrtb2.Site{Domain: "site.com"}, + App: &openrtb2.App{Domain: "app.com"}, + }, + want: "site.com", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, resolveDomain(test.request)) + }) + } +} + +func TestEnrichWritesSegments(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + segs := segments{Content22: []string{"132", "148"}} + + request := &openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}} + assert.True(t, module.enrich(request, segs)) + + require.NotNil(t, request.Site.Content) + require.Len(t, request.Site.Content.Data, 1) + + data := request.Site.Content.Data[0] + assert.Equal(t, DefaultDataProviderName, data.Name) + assert.JSONEq(t, `{"segtax":6}`, string(data.Ext)) + assert.Equal(t, []openrtb2.Segment{{ID: "132"}, {ID: "148"}}, data.Segment) +} + +func TestEnrichAllTaxonomies(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + segs := segments{ + Content22: []string{"132"}, + Content10: []string{"IAB5"}, + Audience: []string{"23"}, + } + + request := &openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}} + assert.True(t, module.enrich(request, segs)) + + require.Len(t, request.Site.Content.Data, 2) + assert.JSONEq(t, `{"segtax":6}`, string(request.Site.Content.Data[0].Ext)) + assert.JSONEq(t, `{"segtax":1}`, string(request.Site.Content.Data[1].Ext)) + assert.Equal(t, []openrtb2.Segment{{ID: "IAB5"}}, request.Site.Content.Data[1].Segment) + + require.NotNil(t, request.User) + require.Len(t, request.User.Data, 1) + assert.JSONEq(t, `{"segtax":4}`, string(request.User.Data[0].Ext)) + assert.Equal(t, []openrtb2.Segment{{ID: "23"}}, request.User.Data[0].Segment) +} + +func TestEnrichPreservesExistingData(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + + request := &openrtb2.BidRequest{Site: &openrtb2.Site{ + Domain: "coursera.com", + Content: &openrtb2.Content{ + Title: "existing content", + Data: []openrtb2.Data{{ + Name: "other-provider.com", + Ext: []byte(`{"segtax":6}`), + Segment: []openrtb2.Segment{{ID: "999"}}, + }}, + }, + }} + + assert.True(t, module.enrich(request, segments{Content22: []string{"132"}})) + + assert.Equal(t, "existing content", request.Site.Content.Title) + require.Len(t, request.Site.Content.Data, 2) + assert.Equal(t, "other-provider.com", request.Site.Content.Data[0].Name) + assert.Equal(t, DefaultDataProviderName, request.Site.Content.Data[1].Name) +} + +func TestEnrichIsIdempotent(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + segs := segments{Content22: []string{"132"}, Audience: []string{"23"}} + request := &openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}} + + assert.True(t, module.enrich(request, segs)) + assert.False(t, module.enrich(request, segs), "a second pass must not append duplicates") + + assert.Len(t, request.Site.Content.Data, 1) + assert.Len(t, request.User.Data, 1) +} + +func TestEnrichAppAndDooh(t *testing.T) { + tests := []struct { + name string + request *openrtb2.BidRequest + content func(*openrtb2.BidRequest) *openrtb2.Content + wantKey []string + }{ + { + name: "app", + request: &openrtb2.BidRequest{App: &openrtb2.App{Bundle: "com.example.app"}}, + content: func(r *openrtb2.BidRequest) *openrtb2.Content { return r.App.Content }, + wantKey: []string{"bidrequest", "app", "content", "data"}, + }, + { + name: "dooh", + request: &openrtb2.BidRequest{DOOH: &openrtb2.DOOH{Domain: "screens.example.com"}}, + content: func(r *openrtb2.BidRequest) *openrtb2.Content { return r.DOOH.Content }, + wantKey: []string{"bidrequest", "dooh", "content", "data"}, + }, + } + + module := newTestModule(t, "https://example.com/v1/responses", nil) + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.True(t, module.enrich(test.request, segments{Content22: []string{"132"}})) + + content := test.content(test.request) + require.NotNil(t, content) + require.Len(t, content.Data, 1) + assert.JSONEq(t, `{"segtax":6}`, string(content.Data[0].Ext)) + assert.Equal(t, test.wantKey, mutationKey(test.request)) + }) + } +} + +func TestEnrichNoOpCases(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + + t.Run("nil request", func(t *testing.T) { + assert.False(t, module.enrich(nil, segments{Content22: []string{"132"}})) + }) + + t.Run("empty segments leave request untouched", func(t *testing.T) { + request := &openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}} + assert.False(t, module.enrich(request, segments{})) + assert.Nil(t, request.Site.Content, "no content object should be created") + assert.Nil(t, request.User) + }) + + t.Run("no distribution channel", func(t *testing.T) { + request := &openrtb2.BidRequest{} + assert.False(t, module.enrich(request, segments{Content22: []string{"132"}})) + assert.Equal(t, []string{"bidrequest"}, mutationKey(request)) + }) + + t.Run("audience only does not create content", func(t *testing.T) { + request := &openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}} + assert.True(t, module.enrich(request, segments{Audience: []string{"23"}})) + assert.Nil(t, request.Site.Content) + assert.Len(t, request.User.Data, 1) + }) + + t.Run("existing user object is reused", func(t *testing.T) { + request := &openrtb2.BidRequest{ + Site: &openrtb2.Site{Domain: "coursera.com"}, + User: &openrtb2.User{ID: "user-1"}, + } + assert.True(t, module.enrich(request, segments{Audience: []string{"23"}})) + assert.Equal(t, "user-1", request.User.ID) + assert.Len(t, request.User.Data, 1) + }) +} + +func TestHasDataEntry(t *testing.T) { + tests := []struct { + name string + data []openrtb2.Data + segtax int + want bool + }{ + {"empty", nil, segtaxContent22, false}, + { + name: "matching name and segtax", + data: []openrtb2.Data{{Name: DefaultDataProviderName, Ext: []byte(`{"segtax":6}`)}}, + segtax: segtaxContent22, + want: true, + }, + { + name: "same name different segtax", + data: []openrtb2.Data{{Name: DefaultDataProviderName, Ext: []byte(`{"segtax":1}`)}}, + segtax: segtaxContent22, + want: false, + }, + { + name: "different provider", + data: []openrtb2.Data{{Name: "other.com", Ext: []byte(`{"segtax":6}`)}}, + segtax: segtaxContent22, + want: false, + }, + { + name: "no ext", + data: []openrtb2.Data{{Name: DefaultDataProviderName}}, + segtax: segtaxContent22, + want: false, + }, + { + name: "unparseable ext is ignored", + data: []openrtb2.Data{{Name: DefaultDataProviderName, Ext: []byte(`not json`)}}, + segtax: segtaxContent22, + want: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, hasDataEntry(test.data, DefaultDataProviderName, test.segtax)) + }) + } +} + +func TestBuildDataRejectsEmptyIDs(t *testing.T) { + module := newTestModule(t, "https://example.com/v1/responses", nil) + _, ok := module.buildData(nil, nil, segtaxContent22) + assert.False(t, ok) +} + +func TestIsAllDigitsAndDots(t *testing.T) { + assert.True(t, isAllDigitsAndDots("192.168.1.1")) + assert.True(t, isAllDigitsAndDots("123")) + assert.False(t, isAllDigitsAndDots("com.example.app")) + assert.False(t, isAllDigitsAndDots("a1.com")) +} diff --git a/modules/zerogpu/rtd/module.go b/modules/zerogpu/rtd/module.go new file mode 100644 index 00000000000..af19792ba53 --- /dev/null +++ b/modules/zerogpu/rtd/module.go @@ -0,0 +1,171 @@ +// Package rtd implements the ZeroGPU Real Time Data module for Prebid Server. +// +// The module resolves the publisher domain from an incoming OpenRTB request, +// classifies it against the IAB content taxonomy using ZeroGPU's +// zlm-v1-iab-domain-classifier model, and injects the resulting categories into +// {site,app,dooh}.content.data so every bidder in the auction can read them. +// +// Classification results are cached in-process, and every failure mode is +// fail-open: a slow or unavailable ZeroGPU API leaves the auction unenriched +// but never delays or rejects it. +package rtd + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "time" + + "github.com/coocood/freecache" + "github.com/prebid/prebid-server/v4/hooks/hookanalytics" + "github.com/prebid/prebid-server/v4/hooks/hookstage" + "github.com/prebid/prebid-server/v4/modules/moduledeps" +) + +// Builder is the module entry point invoked by Prebid Server at startup. +func Builder(rawCfg json.RawMessage, deps moduledeps.ModuleDeps) (interface{}, error) { + cfg, err := newConfig(rawCfg) + if err != nil { + return nil, err + } + + httpClient := &http.Client{Timeout: time.Duration(cfg.Timeout) * time.Millisecond} + if deps.HTTPClient != nil { + // Reuse the host's pooled transport so connections are shared. + httpClient.Transport = deps.HTTPClient.Transport + } + + bgCtx, bgCancel := context.WithCancel(context.Background()) + + return &Module{ + cfg: cfg, + httpClient: httpClient, + cache: freecache.NewCache(cfg.CacheSize), + bgCtx: bgCtx, + bgCancel: bgCancel, + }, nil +} + +// Module implements the ZeroGPU RTD module. +type Module struct { + cfg Config + httpClient *http.Client + cache *freecache.Cache + + // bgCtx bounds the lifetime of background cache warm-ups. It is deliberately + // independent of any hook context, which is cancelled at the execution + // plan's group timeout. + bgCtx context.Context + bgCancel context.CancelFunc + wg sync.WaitGroup + + // inFlight collapses concurrent warm-ups for the same domain. + inFlight sync.Map +} + +var ( + _ hookstage.ProcessedAuctionRequest = (*Module)(nil) + _ shutdowner = (*Module)(nil) +) + +// shutdowner mirrors modules.Shutdowner, which the host calls on teardown. +type shutdowner interface { + Shutdown() error +} + +// Shutdown cancels any in-flight cache warm-ups and waits for them to finish. +func (m *Module) Shutdown() error { + m.bgCancel() + m.wg.Wait() + return nil +} + +const analyticsActivity = "zerogpu-rtd-domain-classification" + +// HandleProcessedAuctionHook enriches the request from cached classifications +// and stages a mutation adding the resulting IAB segments. +// +// This stage is chosen deliberately: it is the last point at which the request +// is still shared by every bidder, stored requests have already been merged, +// and account-level config is available. Running at bidder_request instead +// would warm the same domain once per bidder. +// +// The hook context is intentionally unused. Nothing on this path performs I/O, +// so the hook cannot time out and adds no latency to the auction. +func (m *Module) HandleProcessedAuctionHook( + _ context.Context, + miCtx hookstage.ModuleInvocationContext, + payload hookstage.ProcessedAuctionRequestPayload, +) (hookstage.HookResult[hookstage.ProcessedAuctionRequestPayload], error) { + var result hookstage.HookResult[hookstage.ProcessedAuctionRequestPayload] + + if payload.Request == nil || payload.Request.BidRequest == nil { + return result, nil + } + if !m.cfg.AccountFilter.isAllowed(miCtx.AccountID) { + return result, nil + } + + domain := resolveDomain(payload.Request.BidRequest) + if domain == "" { + result.AnalyticsTags = skippedTags("no domain available on the request") + return result, nil + } + + // The cache is the only source consulted on the auction path. A miss + // schedules a background warm-up and leaves this auction unenriched rather + // than making bidders wait on a network round trip. + segs, cached := m.lookup(domain) + if !cached { + m.warm(domain) + result.AnalyticsTags = skippedTags("domain not yet cached; warming in the background") + return result, nil + } + if segs.isEmpty() { + result.AnalyticsTags = skippedTags("no categories available for this domain") + return result, nil + } + + // Enrichment is applied inside the mutation rather than here so that the + // change is recorded in the module trace and can be reverted by core. + result.ChangeSet.AddMutation( + func(p hookstage.ProcessedAuctionRequestPayload) (hookstage.ProcessedAuctionRequestPayload, error) { + m.enrich(p.Request.BidRequest, segs) + return p, nil + }, + hookstage.MutationAdd, + mutationKey(payload.Request.BidRequest)..., + ) + + result.AnalyticsTags = successTags(domain, segs) + return result, nil +} + +func successTags(domain string, segs segments) hookanalytics.Analytics { + return activity(hookanalytics.ActivityStatusSuccess, hookanalytics.ResultStatusModify, map[string]interface{}{ + "domain": domain, + "content_2_2_count": len(segs.Content22), + "content_1_0_count": len(segs.Content10), + "audience_count": len(segs.Audience), + }) +} + +func skippedTags(reason string) hookanalytics.Analytics { + return activity(hookanalytics.ActivityStatusSuccess, hookanalytics.ResultStatusAllow, map[string]interface{}{ + "reason": reason, + }) +} + +func activity(status hookanalytics.ActivityStatus, resultStatus hookanalytics.ResultStatus, values map[string]interface{}) hookanalytics.Analytics { + return hookanalytics.Analytics{ + Activities: []hookanalytics.Activity{{ + Name: analyticsActivity, + Status: status, + Results: []hookanalytics.Result{{ + Status: resultStatus, + Values: values, + }}, + }}, + } +} diff --git a/modules/zerogpu/rtd/module_test.go b/modules/zerogpu/rtd/module_test.go new file mode 100644 index 00000000000..6f8d3888ef8 --- /dev/null +++ b/modules/zerogpu/rtd/module_test.go @@ -0,0 +1,415 @@ +package rtd + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/prebid/openrtb/v20/openrtb2" + "github.com/prebid/prebid-server/v4/hooks/hookanalytics" + "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) { + deps := moduledeps.ModuleDeps{HTTPClient: http.DefaultClient} + + built, err := Builder(json.RawMessage(`{"enabled": true, "api_key": "test-key"}`), deps) + require.NoError(t, err) + require.IsType(t, &Module{}, built) + + module := built.(*Module) + assert.Equal(t, DefaultEndpoint, module.cfg.Endpoint) + assert.Equal(t, DefaultModel, module.cfg.Model) + assert.Equal(t, "test-key", module.cfg.APIKey) + require.NotNil(t, module.httpClient) + assert.Equal(t, http.DefaultClient.Transport, module.httpClient.Transport) + assert.NotNil(t, module.cache) +} + +func TestBuilderWithoutHostHTTPClient(t *testing.T) { + built, err := Builder(json.RawMessage(`{"api_key": "test-key"}`), moduledeps.ModuleDeps{}) + require.NoError(t, err) + + module := built.(*Module) + assert.Nil(t, module.httpClient.Transport, "falls back to the default transport") +} + +func TestBuilderInvalidConfig(t *testing.T) { + tests := []struct { + name string + cfg string + wantErr string + }{ + {"missing api key", `{"enabled": true}`, "api_key is required"}, + {"malformed json", `{"api_key":`, "failed to parse config"}, + {"bad endpoint", `{"api_key":"k","endpoint":"ftp://x/y"}`, "http or https"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + built, err := Builder(json.RawMessage(test.cfg), moduledeps.ModuleDeps{HTTPClient: http.DefaultClient}) + require.Error(t, err) + assert.Nil(t, built) + assert.Contains(t, err.Error(), test.wantErr) + }) + } +} + +func TestHandleProcessedAuctionHookEnrichesRequest(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + primeCache(t, module, "coursera.com") + + payload := newPayload(&openrtb2.BidRequest{ + ID: "req-1", + Site: &openrtb2.Site{Page: "https://www.coursera.com/learn/python"}, + }) + + result, err := module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, payload) + require.NoError(t, err) + assert.False(t, result.Reject) + require.Len(t, result.ChangeSet.Mutations(), 1) + + mutation := result.ChangeSet.Mutations()[0] + assert.Equal(t, hookstage.MutationAdd, mutation.Type()) + assert.Equal(t, []string{"bidrequest", "site", "content", "data"}, mutation.Key()) + + applyMutations(t, result, payload) + + require.NotNil(t, payload.Request.Site.Content) + require.Len(t, payload.Request.Site.Content.Data, 1) + data := payload.Request.Site.Content.Data[0] + assert.Equal(t, DefaultDataProviderName, data.Name) + assert.JSONEq(t, `{"segtax":6}`, string(data.Ext)) + assert.Equal(t, []openrtb2.Segment{{ID: "132"}, {ID: "148"}}, data.Segment) + + values := activityValues(t, result.AnalyticsTags, hookanalytics.ActivityStatusSuccess) + assert.Equal(t, "coursera.com", values["domain"]) + assert.Equal(t, 2, values["content_2_2_count"]) +} + +// TestHandleProcessedAuctionHookNeverBlocks is the core guarantee of the async +// design: an uncached domain returns immediately and unenriched, and only then +// is the cache warmed for subsequent auctions. +func TestHandleProcessedAuctionHookNeverBlocks(t *testing.T) { + release := make(chan struct{}) + var served int32 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-release // the API is arbitrarily slow + atomic.AddInt32(&served, 1) + _, _ = w.Write([]byte(responsesEnvelope(classificationJSON))) + })) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + start := time.Now() + result, err := module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, payload) + elapsed := time.Since(start) + + require.NoError(t, err) + assert.Empty(t, result.ChangeSet.Mutations(), "an uncached domain must not enrich") + assert.Nil(t, payload.Request.Site.Content) + assert.Less(t, elapsed, 50*time.Millisecond, "the hook must not wait on the API") + assert.Zero(t, atomic.LoadInt32(&served), "the API has not responded yet") + + values := activityValues(t, result.AnalyticsTags, hookanalytics.ActivityStatusSuccess) + assert.Equal(t, "domain not yet cached; warming in the background", values["reason"]) + + // Let the warm-up finish; the next auction on this domain is enriched. + close(release) + awaitWarmUps(module) + + next := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + result, err = module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, next) + require.NoError(t, err) + require.Len(t, result.ChangeSet.Mutations(), 1) +} + +func TestHandleProcessedAuctionHookFailsOpen(t *testing.T) { + tests := []struct { + name string + status int + body string + }{ + {"bad request", http.StatusBadRequest, ``}, + {"unauthorized", http.StatusUnauthorized, ``}, + {"forbidden", http.StatusForbidden, ``}, + {"insufficient quota", statusInsufficientQuota, ``}, + {"server error", http.StatusInternalServerError, ``}, + {"unparseable body", http.StatusOK, `garbage`}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := newClassifierServer(t, test.status, test.body) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + // First auction: uncached, so it schedules a warm-up that fails. + result, err := module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, payload) + require.NoError(t, err) + assert.False(t, result.Reject) + assert.Empty(t, result.ChangeSet.Mutations()) + + awaitWarmUps(module) + + // Second auction: the failure is cached, so the auction still + // proceeds cleanly with no mutation and no error. + next := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + result, err = module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, next) + require.NoError(t, err) + assert.False(t, result.Reject) + assert.Empty(t, result.ChangeSet.Mutations()) + assert.Nil(t, next.Request.Site.Content) + + values := activityValues(t, result.AnalyticsTags, hookanalytics.ActivityStatusSuccess) + assert.Equal(t, "no categories available for this domain", values["reason"]) + }) + } +} + +func TestHandleProcessedAuctionHookIgnoresHookContextCancellation(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + primeCache(t, module, "coursera.com") + + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + // The hook does no I/O, so even an already-cancelled context - which is + // what a zero group timeout produces - must still enrich. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + result, err := module.HandleProcessedAuctionHook(ctx, hookstage.ModuleInvocationContext{}, payload) + + require.NoError(t, err) + assert.Len(t, result.ChangeSet.Mutations(), 1) +} + +// TestWarmSurvivesHookContextCancellation guards the reason warm-ups use the +// module's own context: a warm-up tied to the hook context would be killed at +// the group timeout and the domain would never become cached. +func TestWarmSurvivesHookContextCancellation(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + ctx, cancel := context.WithCancel(context.Background()) + _, err := module.HandleProcessedAuctionHook(ctx, hookstage.ModuleInvocationContext{}, payload) + require.NoError(t, err) + cancel() // the auction ends immediately, as it would at a group timeout + + awaitWarmUps(module) + + segs, cached := module.lookup("coursera.com") + assert.True(t, cached) + assert.Equal(t, []string{"132", "148"}, segs.Content22) +} + +func TestHandleProcessedAuctionHookSkips(t *testing.T) { + tests := []struct { + name string + miCtx hookstage.ModuleInvocationContext + payload hookstage.ProcessedAuctionRequestPayload + mutate func(*Config) + wantTag bool + wantValue string + }{ + { + name: "nil request wrapper", + payload: hookstage.ProcessedAuctionRequestPayload{}, + }, + { + name: "nil bid request", + payload: hookstage.ProcessedAuctionRequestPayload{Request: &openrtb_ext.RequestWrapper{}}, + }, + { + name: "account not on allow list", + miCtx: hookstage.ModuleInvocationContext{AccountID: "9999"}, + payload: newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}), + mutate: func(c *Config) { c.AccountFilter.AllowList = []string{"1001"} }, + }, + { + name: "no resolvable domain", + payload: newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Page: "http://localhost/test"}}), + wantTag: true, + wantValue: "no domain available on the request", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("the classifier must not be called") + })) + defer server.Close() + + module := newTestModule(t, server.URL, test.mutate) + result, err := module.HandleProcessedAuctionHook(context.Background(), test.miCtx, test.payload) + + require.NoError(t, err) + assert.Empty(t, result.ChangeSet.Mutations()) + + if test.wantTag { + values := activityValues(t, result.AnalyticsTags, hookanalytics.ActivityStatusSuccess) + assert.Equal(t, test.wantValue, values["reason"]) + } else { + assert.Empty(t, result.AnalyticsTags.Activities) + } + }) + } +} + +func TestHandleProcessedAuctionHookAllowsListedAccount(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, func(c *Config) { + c.AccountFilter.AllowList = []string{"1001"} + }) + primeCache(t, module, "coursera.com") + + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + result, err := module.HandleProcessedAuctionHook( + context.Background(), + hookstage.ModuleInvocationContext{AccountID: "1001"}, + payload, + ) + require.NoError(t, err) + assert.Len(t, result.ChangeSet.Mutations(), 1) +} + +func TestHandleProcessedAuctionHookNoQualifyingSegments(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + // A min_score above every score in the fixture filters everything out, so + // the domain caches as an empty - but present - result. + module := newTestModule(t, server.URL, func(c *Config) { c.MinScore = 0.999 }) + primeCache(t, module, "coursera.com") + + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + result, err := module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, payload) + require.NoError(t, err) + + assert.Empty(t, result.ChangeSet.Mutations()) + assert.Nil(t, payload.Request.Site.Content) + + values := activityValues(t, result.AnalyticsTags, hookanalytics.ActivityStatusSuccess) + assert.Equal(t, "no categories available for this domain", values["reason"]) +} + +func TestHandleProcessedAuctionHookAllTaxonomies(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, func(c *Config) { + c.EnrichContent10 = true + c.EnrichUserAudience = true + }) + primeCache(t, module, "coursera.com") + + payload := newPayload(&openrtb2.BidRequest{Site: &openrtb2.Site{Domain: "coursera.com"}}) + + result, err := module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, payload) + require.NoError(t, err) + applyMutations(t, result, payload) + + require.Len(t, payload.Request.Site.Content.Data, 2) + require.Len(t, payload.Request.User.Data, 1) + + values := activityValues(t, result.AnalyticsTags, hookanalytics.ActivityStatusSuccess) + assert.Equal(t, 2, values["content_2_2_count"]) + assert.Equal(t, 2, values["content_1_0_count"]) + assert.Equal(t, 2, values["audience_count"]) +} + +// TestHandleProcessedAuctionHookWithFPDBidders covers the interaction with +// firstpartydata.ExtractOpenRtbGlobalFPD, which strips and redistributes +// site.content.data per bidder only when ext.prebid.data.bidders is set. The +// module's job is to inject the same segments either way; core decides where +// they end up. +func TestHandleProcessedAuctionHookWithFPDBidders(t *testing.T) { + tests := []struct { + name string + ext json.RawMessage + }{ + {"without ext.prebid.data.bidders", nil}, + {"with ext.prebid.data.bidders", json.RawMessage(`{"prebid":{"data":{"bidders":["appnexus"]}}}`)}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := newClassifierServer(t, http.StatusOK, responsesEnvelope(classificationJSON)) + defer server.Close() + + module := newTestModule(t, server.URL, nil) + primeCache(t, module, "coursera.com") + + payload := newPayload(&openrtb2.BidRequest{ + ID: "req-1", + Site: &openrtb2.Site{Domain: "coursera.com"}, + Ext: test.ext, + }) + + result, err := module.HandleProcessedAuctionHook(context.Background(), hookstage.ModuleInvocationContext{}, payload) + require.NoError(t, err) + applyMutations(t, result, payload) + + require.Len(t, payload.Request.Site.Content.Data, 1) + assert.JSONEq(t, `{"segtax":6}`, string(payload.Request.Site.Content.Data[0].Ext)) + + // Rebuilding must preserve the injected segments. + require.NoError(t, payload.Request.RebuildRequest()) + require.Len(t, payload.Request.Site.Content.Data, 1) + }) + } +} + +func newPayload(request *openrtb2.BidRequest) hookstage.ProcessedAuctionRequestPayload { + return hookstage.ProcessedAuctionRequestPayload{ + Request: &openrtb_ext.RequestWrapper{BidRequest: request}, + } +} + +// applyMutations runs the staged mutations the way the hook executor does. +func applyMutations(t *testing.T, result hookstage.HookResult[hookstage.ProcessedAuctionRequestPayload], payload hookstage.ProcessedAuctionRequestPayload) { + t.Helper() + for _, mutation := range result.ChangeSet.Mutations() { + _, err := mutation.Apply(payload) + require.NoError(t, err) + } +} + +// activityValues asserts a single analytics activity with the expected status +// and returns its values map. +func activityValues(t *testing.T, tags hookanalytics.Analytics, want hookanalytics.ActivityStatus) map[string]interface{} { + t.Helper() + require.Len(t, tags.Activities, 1) + activity := tags.Activities[0] + assert.Equal(t, analyticsActivity, activity.Name) + assert.Equal(t, want, activity.Status) + require.Len(t, activity.Results, 1) + return activity.Results[0].Values +} diff --git a/modules/zerogpu/rtd/sample/pbs_example.json b/modules/zerogpu/rtd/sample/pbs_example.json new file mode 100644 index 00000000000..7196ca0e480 --- /dev/null +++ b/modules/zerogpu/rtd/sample/pbs_example.json @@ -0,0 +1,49 @@ +{ + "hooks": { + "enabled": true, + "modules": { + "zerogpu": { + "rtd": { + "enabled": true, + "api_key": "", + "endpoint": "https://api.zerogpu.ai/v1/responses", + "model": "zlm-v1-iab-domain-classifier", + "timeout_ms": 2000, + "cache_ttl_seconds": 86400, + "negative_cache_ttl_seconds": 300, + "retry_cache_ttl_seconds": 30, + "cache_size": 10485760, + "min_score": 0.5, + "max_segments": 0, + "data_provider_name": "zerogpu.ai", + "enrich_content_1_0": false, + "enrich_user_audience": false, + "account_filter": { + "allow_list": [] + } + } + } + }, + "host_execution_plan": { + "endpoints": { + "/openrtb2/auction": { + "stages": { + "processed_auction_request": { + "groups": [ + { + "timeout": 10, + "hook_sequence": [ + { + "module_code": "zerogpu.rtd", + "hook_impl_code": "zerogpu-rtd-processed-auction-request" + } + ] + } + ] + } + } + } + } + } + } +}