2 files changed, 176 insertions, 0 deletions
diff --git a/userapi/util/devices.go b/userapi/util/devices.go
new file mode 100644
index 00000000..cbf3bd28
--- /dev/null
+++ b/userapi/util/devices.go
@@ -0,0 +1,100 @@
+package util
+
+import (
+	"context"
+
+	"github.com/matrix-org/dendrite/internal/pushgateway"
+	"github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/dendrite/userapi/storage"
+	log "github.com/sirupsen/logrus"
+)
+
+type PusherDevice struct {
+	Device pushgateway.Device
+	Pusher *api.Pusher
+	URL    string
+	Format string
+}
+
+// GetPushDevices pushes to the configured devices of a local user.
+func GetPushDevices(ctx context.Context, localpart string, tweaks map[string]interface{}, db storage.Database) ([]*PusherDevice, error) {
+	pushers, err := db.GetPushers(ctx, localpart)
+	if err != nil {
+		return nil, err
+	}
+
+	devices := make([]*PusherDevice, 0, len(pushers))
+	for _, pusher := range pushers {
+		var url, format string
+		data := pusher.Data
+		switch pusher.Kind {
+		case api.EmailKind:
+			url = "mailto:"
+
+		case api.HTTPKind:
+			// TODO: The spec says only event_id_only is supported,
+			// but Sytests assume "" means "full notification".
+			fmtIface := pusher.Data["format"]
+			var ok bool
+			format, ok = fmtIface.(string)
+			if ok && format != "event_id_only" {
+				log.WithFields(log.Fields{
+					"localpart": localpart,
+					"app_id":    pusher.AppID,
+				}).Errorf("Only data.format event_id_only or empty is supported")
+				continue
+			}
+
+			urlIface := pusher.Data["url"]
+			url, ok = urlIface.(string)
+			if !ok {
+				log.WithFields(log.Fields{
+					"localpart": localpart,
+					"app_id":    pusher.AppID,
+				}).Errorf("No data.url configured for HTTP Pusher")
+				continue
+			}
+			data = mapWithout(data, "url")
+
+		default:
+			log.WithFields(log.Fields{
+				"localpart": localpart,
+				"app_id":    pusher.AppID,
+				"kind":      pusher.Kind,
+			}).Errorf("Unhandled pusher kind")
+			continue
+		}
+
+		devices = append(devices, &PusherDevice{
+			Device: pushgateway.Device{
+				AppID:     pusher.AppID,
+				Data:      data,
+				PushKey:   pusher.PushKey,
+				PushKeyTS: pusher.PushKeyTS,
+				Tweaks:    tweaks,
+			},
+			Pusher: &pusher,
+			URL:    url,
+			Format: format,
+		})
+	}
+
+	return devices, nil
+}
+
+// mapWithout returns a shallow copy of the map, without the given
+// key. Returns nil if the resulting map is empty.
+func mapWithout(m map[string]interface{}, key string) map[string]interface{} {
+	ret := make(map[string]interface{}, len(m))
+	for k, v := range m {
+		// The specification says we do not send "url".
+		if k == key {
+			continue
+		}
+		ret[k] = v
+	}
+	if len(ret) == 0 {
+		return nil
+	}
+	return ret
+}
diff --git a/userapi/util/notify.go b/userapi/util/notify.go
new file mode 100644
index 00000000..ff206bd3
--- /dev/null
+++ b/userapi/util/notify.go
@@ -0,0 +1,76 @@
+package util
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/matrix-org/dendrite/internal/pushgateway"
+	"github.com/matrix-org/dendrite/userapi/storage"
+	"github.com/matrix-org/dendrite/userapi/storage/tables"
+	log "github.com/sirupsen/logrus"
+)
+
+// NotifyUserCountsAsync sends notifications to a local user's
+// notification destinations. Database lookups run synchronously, but
+// a single goroutine is started when talking to the Push
+// gateways. There is no way to know when the background goroutine has
+// finished.
+func NotifyUserCountsAsync(ctx context.Context, pgClient pushgateway.Client, localpart string, db storage.Database) error {
+	pusherDevices, err := GetPushDevices(ctx, localpart, nil, db)
+	if err != nil {
+		return err
+	}
+
+	if len(pusherDevices) == 0 {
+		return nil
+	}
+
+	userNumUnreadNotifs, err := db.GetNotificationCount(ctx, localpart, tables.AllNotifications)
+	if err != nil {
+		return err
+	}
+
+	log.WithFields(log.Fields{
+		"localpart": localpart,
+		"app_id0":   pusherDevices[0].Device.AppID,
+		"pushkey":   pusherDevices[0].Device.PushKey,
+	}).Tracef("Notifying HTTP push gateway about notification counts")
+
+	// TODO: think about bounding this to one per user, and what
+	// ordering guarantees we must provide.
+	go func() {
+		// This background processing cannot be tied to a request.
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		// TODO: we could batch all devices with the same URL, but
+		// Sytest requires consumers/roomserver.go to do it
+		// one-by-one, so we do the same here.
+		for _, pusherDevice := range pusherDevices {
+			// TODO: support "email".
+			if !strings.HasPrefix(pusherDevice.URL, "http") {
+				continue
+			}
+
+			req := pushgateway.NotifyRequest{
+				Notification: pushgateway.Notification{
+					Counts: &pushgateway.Counts{
+						Unread: int(userNumUnreadNotifs),
+					},
+					Devices: []*pushgateway.Device{&pusherDevice.Device},
+				},
+			}
+			if err := pgClient.Notify(ctx, pusherDevice.URL, &req, &pushgateway.NotifyResponse{}); err != nil {
+				log.WithFields(log.Fields{
+					"localpart": localpart,
+					"app_id0":   pusherDevice.Device.AppID,
+					"pushkey":   pusherDevice.Device.PushKey,
+				}).WithError(err).Error("HTTP push gateway request failed")
+				return
+			}
+		}
+	}()
+
+	return nil
+}