CTV Module Migration

analytics/pubmatic/logger.go

@@ -562,10 +562,10 @@ func getAdPodSlot(adPodConfig *models.AdPod) *AdPodSlot {
 	}
 
 	adPodSlot := AdPodSlot{
-		MinAds:                      adPodConfig.MinAds,
-		MaxAds:                      adPodConfig.MaxAds,
-		MinDuration:                 adPodConfig.MinDuration,
-		MaxDuration:                 adPodConfig.MaxDuration,
+		MinAds:                      *adPodConfig.MinAds,
+		MaxAds:                      *adPodConfig.MaxAds,
+		MinDuration:                 *adPodConfig.MinDuration,
+		MaxDuration:                 *adPodConfig.MaxDuration,
 		AdvertiserExclusionPercent:  *adPodConfig.AdvertiserExclusionPercent,
 		IABCategoryExclusionPercent: *adPodConfig.IABCategoryExclusionPercent,
 	}

endpoints/openrtb2/auction.go

@@ -1,6 +1,7 @@
 package openrtb2
 
 import (
+	"bytes"
 	"compress/gzip"
 	"context"
 	"encoding/json"
@@ -411,19 +412,33 @@ func sendAuctionResponse(
 	}
 
 	// Fixes #231
-	enc := json.NewEncoder(w)
+	w.Header().Set("Content-Type", "application/json")
+
+	// Create a buffer to hold the JSON response
+	var respBuf bytes.Buffer
+
+	// Create a new JSON encoder and set the SetEscapeHTML option to false
+	enc := json.NewEncoder(&respBuf)
 	enc.SetEscapeHTML(false)
 
-	w.Header().Set("Content-Type", "application/json")
+	// Encode the response
+	if err := enc.Encode(response); err != nil {
+		ao.Errors = append(ao.Errors, fmt.Errorf("/openrtb2/auction Failed to encode response: %v", err))
+		return labels, ao
+	}
+
+	rawResponse, headers := hookExecutor.ExecuteExitPointStage(respBuf.Bytes(), w.Header())
+	for key, values := range headers {
+		w.Header()[key] = values
+	}
 
-	// If an error happens when encoding the response, there isn't much we can do.
+	// If an error happens when writing the response, there isn't much we can do.
 	// If we've sent _any_ bytes, then Go would have sent the 200 status code first.
 	// That status code can't be un-sent... so the best we can do is log the error.
-	if err := enc.Encode(response); err != nil {
+	if _, err := w.Write(rawResponse); err != nil {
 		labels.RequestStatus = metrics.RequestStatusNetworkErr
 		ao.Errors = append(ao.Errors, fmt.Errorf("/openrtb2/auction Failed to send response: %v", err))
 	}
-
 	return labels, ao
 }
 

endpoints/openrtb2/test_utils.go

@@ -1452,6 +1452,7 @@ type mockPlanBuilder struct {
 	rawBidderResponsePlan        hooks.Plan[hookstage.RawBidderResponse]
 	allProcessedBidResponsesPlan hooks.Plan[hookstage.AllProcessedBidResponses]
 	auctionResponsePlan          hooks.Plan[hookstage.AuctionResponse]
+	exitPointPlan                hooks.Plan[hookstage.ExitPoint]
 }
 
 func (m mockPlanBuilder) PlanForEntrypointStage(_ string) hooks.Plan[hookstage.Entrypoint] {
@@ -1486,6 +1487,10 @@ func (m mockPlanBuilder) PlanForAuctionResponseStage(_ string, _ *config.Account
 	return m.auctionResponsePlan
 }
 
+func (m mockPlanBuilder) PlanForExitPointStage(_ string, _ *config.Account) hooks.Plan[hookstage.ExitPoint] {
+	return m.exitPointPlan
+}
+
 func makePlan[H any](hook H) hooks.Plan[H] {
 	return hooks.Plan[H]{
 		{

go.mod

@@ -110,4 +110,4 @@ replace github.com/prebid/openrtb/v20 => github.com/PubMatic-OpenWrap/prebid-ope
 
 replace github.com/beevik/etree v1.0.2 => github.com/PubMatic-OpenWrap/etree v1.0.2-0.20241125102329-0b5c47d99ad5
 
-replace github.com/beevik/etree/110 => github.com/beevik/etree v1.1.0
+replace github.com/beevik/etree/110 => github.com/beevik/etree v1.1.0

go.sum

@@ -1042,4 +1042,4 @@ honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
-rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
+rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=

hooks/empty_plan.go

@@ -40,3 +40,7 @@ func (e EmptyPlanBuilder) PlanForAllProcessedBidResponsesStage(endpoint string,
 func (e EmptyPlanBuilder) PlanForAuctionResponseStage(endpoint string, account *config.Account) Plan[hookstage.AuctionResponse] {
 	return nil
 }
+
+func (e EmptyPlanBuilder) PlanForExitPointStage(endpoint string, account *config.Account) Plan[hookstage.ExitPoint] {
+	return nil
+}

hooks/hookexecution/executor.go

@@ -41,6 +41,7 @@ type StageExecutor interface {
 	ExecuteAllProcessedBidResponsesStage(adapterBids map[openrtb_ext.BidderName]*entities.PbsOrtbSeatBid)
 	ExecuteAuctionResponseStage(response *openrtb2.BidResponse)
 	ExecuteBeforeRequestValidationStage(req *openrtb2.BidRequest) *RejectError
+	ExecuteExitPointStage(response []byte, headers http.Header) ([]byte, http.Header)
 }
 
 type HookStageExecutor interface {
@@ -326,6 +327,34 @@ func (e *hookExecutor) ExecuteAuctionResponseStage(response *openrtb2.BidRespons
 	e.pushStageOutcome(outcome)
 }
 
+func (e *hookExecutor) ExecuteExitPointStage(response []byte, headers http.Header) ([]byte, http.Header) {
+	plan := e.planBuilder.PlanForExitPointStage(e.endpoint, e.account)
+	if len(plan) == 0 {
+		return response, headers
+	}
+
+	handler := func(
+		ctx context.Context,
+		moduleCtx hookstage.ModuleInvocationContext,
+		hook hookstage.ExitPoint,
+		payload hookstage.ExitPointPayload,
+	) (hookstage.HookResult[hookstage.ExitPointPayload], error) {
+		return hook.HandleExitPointHook(ctx, moduleCtx, payload)
+	}
+
+	stageName := hooks.StageAuctionResponse.String()
+	executionCtx := e.newContext(stageName)
+	payload := hookstage.ExitPointPayload{RawResponse: response}
+
+	outcome, _, contexts, _ := executeStage(executionCtx, plan, payload, handler, e.metricEngine)
+	outcome.Entity = entityAuctionResponse
+	outcome.Stage = stageName
+
+	e.saveModuleContexts(contexts)
+	e.pushStageOutcome(outcome)
+	return payload.RawResponse, payload.Headers
+}
+
 func (e *hookExecutor) newContext(stage string) executionContext {
 	return executionContext{
 		account:         e.account,
@@ -389,3 +418,7 @@ func (executor EmptyHookExecutor) ExecuteAuctionResponseStage(_ *openrtb2.BidRes
 func (executor EmptyHookExecutor) ExecuteBeforeRequestValidationStage(_ *openrtb2.BidRequest) *RejectError {
 	return nil
 }
+
+func (executor EmptyHookExecutor) ExecuteExitPointStage(response []byte, headers http.Header) ([]byte, http.Header) {
+	return nil, nil
+}

hooks/hookstage/exitpoint.go

@@ -0,0 +1,22 @@
+package hookstage
+
+import (
+	"context"
+	"net/http"
+)
+
+type ExitPoint interface {
+	HandleExitPointHook(
+		context.Context,
+		ModuleInvocationContext,
+		ExitPointPayload,
+	) (HookResult[ExitPointPayload], error)
+}
+
+// ExitPointPayload represents the payload data for the exit point hook stage.
+// It contains the raw response body and the associated HTTP headers.
+// Hooks are allowed to modify response using mutations.
+type ExitPointPayload struct {
+	RawResponse []byte
+	Headers     http.Header
+}

hooks/plan.go

@@ -19,6 +19,7 @@ const (
 	StageRawBidderResponse        Stage = "raw_bidder_response"
 	StageAllProcessedBidResponses Stage = "all_processed_bid_responses"
 	StageAuctionResponse          Stage = "auction_response"
+	StageExitPoint                Stage = "exitpoint"
 )
 
 func (s Stage) String() string {
@@ -42,6 +43,7 @@ type ExecutionPlanBuilder interface {
 	PlanForRawBidderResponseStage(endpoint string, account *config.Account) Plan[hookstage.RawBidderResponse]
 	PlanForAllProcessedBidResponsesStage(endpoint string, account *config.Account) Plan[hookstage.AllProcessedBidResponses]
 	PlanForAuctionResponseStage(endpoint string, account *config.Account) Plan[hookstage.AuctionResponse]
+	PlanForExitPointStage(endpoint string, account *config.Account) Plan[hookstage.ExitPoint]
 }
 
 // Plan represents a slice of groups of hooks of a specific type grouped in the established order.
@@ -167,6 +169,16 @@ func (p PlanBuilder) PlanForAuctionResponseStage(endpoint string, account *confi
 	)
 }
 
+func (p PlanBuilder) PlanForExitPointStage(endpoint string, account *config.Account) Plan[hookstage.ExitPoint] {
+	return getMergedPlan(
+		p.hooks,
+		account,
+		endpoint,
+		StageExitPoint,
+		p.repo.GetExitPointHook,
+	)
+}
+
 type hookFn[T any] func(moduleName string) (T, bool)
 
 func getMergedPlan[T any](

hooks/repo.go

@@ -22,6 +22,7 @@ type HookRepository interface {
 	GetRawBidderResponseHook(id string) (hookstage.RawBidderResponse, bool)
 	GetAllProcessedBidResponsesHook(id string) (hookstage.AllProcessedBidResponses, bool)
 	GetAuctionResponseHook(id string) (hookstage.AuctionResponse, bool)
+	GetExitPointHook(id string) (hookstage.ExitPoint, bool)
 }
 
 // NewHookRepository returns a new instance of the HookRepository interface.
@@ -51,6 +52,7 @@ type hookRepository struct {
 	rawBidderResponseHooks       map[string]hookstage.RawBidderResponse
 	allProcessedBidResponseHooks map[string]hookstage.AllProcessedBidResponses
 	auctionResponseHooks         map[string]hookstage.AuctionResponse
+	exitPointHooks               map[string]hookstage.ExitPoint
 }
 
 func (r *hookRepository) GetEntrypointHook(id string) (hookstage.Entrypoint, bool) {
@@ -85,6 +87,10 @@ func (r *hookRepository) GetAuctionResponseHook(id string) (hookstage.AuctionRes
 	return getHook(r.auctionResponseHooks, id)
 }
 
+func (r *hookRepository) GetExitPointHook(id string) (hookstage.ExitPoint, bool) {
+	return getHook(r.exitPointHooks, id)
+}
+
 func (r *hookRepository) add(id string, hook interface{}) error {
 	var hasAnyHooks bool
 	var err error
@@ -145,6 +151,13 @@ func (r *hookRepository) add(id string, hook interface{}) error {
 		}
 	}
 
+	if h, ok := hook.(hookstage.ExitPoint); ok {
+		hasAnyHooks = true
+		if r.exitPointHooks, err = addHook(r.exitPointHooks, h, id); err != nil {
+			return err
+		}
+	}
+
 	if !hasAnyHooks {
 		return fmt.Errorf(`hook "%s" does not implement any supported hook interface`, id)
 	}