code: func PubsubReceiver(ctx *router.Context) {
  	c, rw, r := ctx.Context, ctx.Writer, ctx.Request
  	defer r.Body.Close()
  
  	type PubsubMessage struct {
  		Attributes map[string]string `json:"attributes"`
  		Data       []byte            `json:"data"`
  		MessageID  string            `json:"message_id"`
  	}
  	type PubsubPushMessage struct {
  		Message      PubsubMessage `json:"message"`
  		Subscription string        `json:"subscription"`
  	}
  	psm := &PubsubPushMessage{}
  
  	if err := json.NewDecoder(r.Body).Decode(psm); err != nil {
  		logging.WithError(err).Errorf(c, "Failed to parse pubsub message")
  		http.Error(rw, "Failed to parse pubsub message", http.StatusInternalServerError)
  		return
  	}
  
  	eid, cfgName, err := decodeAuthToken(c, psm.Message.Attributes["auth_token"])
  	if err != nil {
  		logging.WithError(err).Errorf(c, "bad auth_token")
  		// Acknowledge this message, since it'll never be valid.
  		rw.WriteHeader(http.StatusNoContent)
  		return
  	}
  
  	// remove "auth_token" from Attributes to avoid having it pass to the
  	// distributor.
  	delete(psm.Message.Attributes, "auth_token")
  
  	err = tumble.RunMutation(c, &NotifyExecution{
  		cfgName, &Notification{eid, psm.Message.Data, psm.Message.Attributes},
  	})
  	if err != nil {
  		// TODO(riannucci): distinguish between transient/non-transient failures.
  		logging.WithError(err).Errorf(c, "failed to NotifyExecution")
  		rw.WriteHeader(http.StatusInternalServerError)
  		return
  	}
  
  	rw.WriteHeader(http.StatusNoContent)
  }
docs: // PubsubReceiver is the HTTP handler that processes incoming pubsub events
  // delivered to topics prepared with TaskDescription.PrepareTopic, and routes
  // them to the appropriate distributor implementation's HandleNotification
  // method.
  //
  // It requires that a Registry be installed in c via WithRegistry.