Adds PushGatewayClient and plumbing in Pushserver's room consumer.

* Push rules are "always notify".
* Only event_id_only push format is allowed.

cmd/dendrite-demo-libp2p/main.go

@@ -179,7 +179,8 @@ func main() {
 		panic("failed to create new public rooms provider: " + err.Error())
 	}
 
-	psAPI := pushserver.NewInternalAPI(&base.Base, rsAPI)
+	pgClient := base.Base.PushGatewayHTTPClient()
+	psAPI := pushserver.NewInternalAPI(&base.Base, pgClient, rsAPI)
 
 	monolith := setup.Monolith{
 		Config:    base.Base.Cfg,

cmd/dendrite-demo-yggdrasil/main.go

@@ -123,7 +123,8 @@ func main() {
 		base, federation, rsAPI, keyRing, true,
 	)
 
-	psAPI := pushserver.NewInternalAPI(base, rsAPI)
+	pgClient := base.PushGatewayHTTPClient()
+	psAPI := pushserver.NewInternalAPI(base, pgClient, rsAPI)
 	if base.UseHTTPAPIs {
 		pushserver.AddInternalRoutes(base.InternalAPIMux, psAPI)
 		psAPI = base.PushServerHTTPClient()

cmd/dendrite-monolith-server/main.go

@@ -143,7 +143,8 @@ func main() {
 	}
 	rsAPI.SetAppserviceAPI(asAPI)
 
-	psAPI := pushserver.NewInternalAPI(base, rsAPI)
+	pgClient := base.PushGatewayHTTPClient()
+	psAPI := pushserver.NewInternalAPI(base, pgClient, rsAPI)
 	if base.UseHTTPAPIs {
 		pushserver.AddInternalRoutes(base.InternalAPIMux, psAPI)
 		psAPI = base.PushServerHTTPClient()

cmd/dendrite-polylith-multi/personalities/pushserver.go

@@ -22,7 +22,8 @@ import (
 )
 
 func PushServer(base *setup.BaseDendrite, cfg *config.Dendrite, rsAPI roomserverAPI.RoomserverInternalAPI) {
-	intAPI := pushserver.NewInternalAPI(base, rsAPI)
+	pgClient := base.PushGatewayHTTPClient()
+	intAPI := pushserver.NewInternalAPI(base, pgClient, rsAPI)
 
 	pushserver.AddInternalRoutes(base.InternalAPIMux, intAPI)
 

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/docker/go-connections v0.4.0
 	github.com/getsentry/sentry-go v0.11.0
 	github.com/gologme/log v1.2.0
+	github.com/google/go-cmp v0.5.5
 	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/h2non/filetype v1.1.1 // indirect

go.sum

@@ -496,6 +496,7 @@ github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
 github.com/google/go-github v17.0.0+incompatible/go.mod h1:zLgOLi98H3fifZn+44m+umXrS52loVEgC2AApnigrVQ=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=

internal/pushgateway/client.go

@@ -0,0 +1,54 @@
+package pushgateway
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/opentracing/opentracing-go"
+)
+
+type httpClient struct {
+	hc *http.Client
+}
+
+// NewHTTPClient creates a new Push Gateway client that uses an
+// *http.Client.
+func NewHTTPClient(hc *http.Client) Client {
+	return &httpClient{hc: hc}
+}
+
+func (h *httpClient) Notify(ctx context.Context, url string, req *NotifyRequest, resp *NotifyResponse) error {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "Notify")
+	defer span.Finish()
+
+	body, err := json.Marshal(req)
+	if err != nil {
+		return err
+	}
+	hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	hreq.Header.Set("Content-Type", "application/json")
+
+	hresp, err := h.hc.Do(hreq)
+	if err != nil {
+		return err
+	}
+	defer hresp.Body.Close()
+
+	if hresp.StatusCode == http.StatusOK {
+		return json.NewDecoder(hresp.Body).Decode(resp)
+	}
+
+	var errorBody struct {
+		Message string `json:"message"`
+	}
+	if err := json.NewDecoder(hresp.Body).Decode(&errorBody); err == nil {
+		return fmt.Errorf("push gateway: %d from %s: %s", hresp.StatusCode, url, errorBody.Message)
+	}
+	return fmt.Errorf("push gateway: %d from %s", hresp.StatusCode, url)
+}

internal/pushgateway/pushgateway.go

@@ -0,0 +1,58 @@
+package pushgateway
+
+import (
+	"context"
+	"encoding/json"
+)
+
+// A Client is how interactions iwth a Push Gateway is done.
+type Client interface {
+	// Notify sends a notification to the gateway at the given URL.
+	Notify(ctx context.Context, url string, req *NotifyRequest, resp *NotifyResponse) error
+}
+
+type NotifyRequest struct {
+	Notification Notification `json:"notification"` // Required
+}
+
+type NotifyResponse struct {
+	// Rejected is the list of device push keys that were rejected
+	// during the push. The caller should remove the push keys so they
+	// are not used again.
+	Rejected []string `json:"rejected"` // Required
+}
+
+type Notification struct {
+	Content           json.RawMessage `json:"content,omitempty"`
+	Counts            *Counts         `json:"counts,omitempty"`
+	Devices           []*Device       `json:"devices"` // Required
+	EventID           string          `json:"event_id,omitempty"`
+	Prio              Prio            `json:"prio,omitempty"`
+	RoomAlias         string          `json:"room_alias,omitempty"`
+	RoomID            string          `json:"room_id,omitempty"`
+	RoomName          string          `json:"room_name,omitempty"`
+	Sender            string          `json:"sender,omitempty"`
+	SenderDisplayName string          `json:"sender_display_name,omitempty"`
+	Type              string          `json:"type,omitempty"`
+	UserIsTarget      bool            `json:"user_is_target,omitempty"`
+}
+
+type Counts struct {
+	MissedCalls int `json:"missed_calls,omitempty"`
+	Unread      int `json:"unread,omitempty"`
+}
+
+type Device struct {
+	AppID     string                 `json:"app_id"` // Required
+	Data      map[string]interface{} `json:"data,omitempty"`
+	PushKey   string                 `json:"pushkey"` // Required
+	PushKeyTS int64                  `json:"pushkey_ts,omitempty"`
+	Tweaks    map[string]interface{} `json:"tweaks,omitempty"`
+}
+
+type Prio string
+
+const (
+	HighPrio Prio = "high"
+	LowPrio  Prio = "low"
+)