Add support for d14n and SubscribeAll mechanism

.gitignore

@@ -5,6 +5,7 @@
 *.so
 *.dylib
 mise.toml
+cmd/server/server
 
 # Test binary, built with `go test -c`
 *.test

Migration.md

@@ -0,0 +1,54 @@
+# Migration
+
+This document describes how to run the Notification Server against different versions of XMTP nodes, especially moving from V3 to V4 version of the XMTP node.
+
+## Legacy - V3
+
+The V3 version of the XMTP node is the version of the XMTP node running in production as of 11. March 2026.
+The source code for the node can be found at https://github.com/xmtp/xmtp-node-go.
+
+This is the default node that Notification Server will expect to find on the other side.
+Notification Server will try to communicate with it using the well established, legacy, V3 GRPC API, found [here](https://github.com/xmtp/proto/blob/main/proto/message_api/v1/message_api.proto).
+
+The Notification Server will issue a `SubscribeAll` call to the V3 server and receive all messages flowing through the network.
+
+## Decentralization (D14N) API - V4
+
+The V4 version is the new, decentralized version of the XMTP node, whose implementation can be found at https://github.com/xmtp/xmtpd.
+Soon it is a goal of XMTP network to completely switch to this node implementation for powering the network, so at some point message traffic will stop flowing through the legacy, v3 node, and will happen only on V4.
+
+The API for the V4 node can be found [here](https://github.com/xmtp/proto/blob/main/proto/xmtpv4/message_api/message_api.proto).
+
+### Running the Notification Server Against D14N Node
+
+In order to run the Notification Server against the new node version, it is sufficient to specify `--d14n` CLI flag.
+All remaining CLI flags are backwards compatible.
+
+This will instruct the Notification Server to connect to the specified node, but will attempt to communicate with it using the new API, and will understand the different data types that the Node will send as a response.
+
+It is possible to run two instances of the Notification Server, each pointed at a different network - V3 and V4.
+Nodes can share the same underlying database from which they can pull installation and subscription data.
+
+### Data Format Difference
+
+As the two networks operate somewhat differently, they have different data formats.
+Note that HTTP delivery mechanism provides more detailed contextual information about the message than APNs and FCM delivery methods.
+
+#### APNs and FCM
+
+Payloads sent by the Notification Server for APNs and FCM are exactly the same, running against the V3 or V4 API.
+
+#### HTTP Delivery
+
+If HTTP Delivery method is selected, Notification Server will POST the message payload to the specified address.
+
+The payload is unchanged if the Notification Server is running against the V3 node.
+If Notification Server is subscribed to a V4 node, the payload is slightly different, and can be found in the `message_v4` field.
+
+If the Notification Server payload is sent to a single server, data origin can be determined based on the data receivedČ
+- `message` - legacy, V3 message
+- `message_v4` - D14N, V4 message
+
+Seeing which field is populated can also make it easy on the data consumer to determine how to interpret the data.
+
+These types are defined and can be inspected in more detail [here](pkg/interfaces/interfaces.go) - check the `SendRequest` type definition.

README.md

@@ -10,6 +10,8 @@ This project is ready to serve as a reference for you to start building.
 
 Many applications will have different needs for push notifications (different delivery providers, different metadata attached to payloads, etc), and this repo is designed to be forked and customized for each application's needs.
 
+For more information about running the notificaiton server against V3 or V4 version of XMTP node, see [here](/Migration.md).
+
 ## Prerequisites
 
 1. Go 1.20
@@ -53,7 +55,7 @@ Here is the output as of 03/01/2024:
 
 ```sh
 Usage:
-  main [OPTIONS]
+  server [OPTIONS]
 
 Application Options:
   -d, --db-connection-string=              Address to database [$DB_CONNECTION_STRING]
@@ -69,6 +71,7 @@ Worker Options:
       --xmtp-listener                      Enable the XMTP listener to actually send notifications. Requires APNSOptions to be configured
       --xmtp-listener-tls                  Whether to connect to XMTP network using TLS
   -x, --xmtp-address=                      Address (including port) of XMTP GRPC server [$XMTP_GRPC_ADDRESS]
+      --d14n                               Use decentralization backend [$XTMP_D14N]
       --num-workers=                       Number of workers used to process messages (default: 50)
 
 APNS Options:
@@ -78,7 +81,8 @@ APNS Options:
       --apns-key-id=                       Key ID associated with APNS credentials [$APNS_KEY_ID]
       --apns-team-id=                      APNS Team ID [$APNS_TEAM_ID]
       --apns-topic=                        Topic to be used on all messages [$APNS_TOPIC]
-      --apns-mode=[development|production] Which APNS servers to deliver to, development or production (default: development) [$APNS_MODE]
+      --apns-mode=[development|production] Which APNS servers to deliver to, development or production (default: development)
+                                           [$APNS_MODE]
 
 FCM Options:
       --fcm-enabled                        Enable FCM sending [$FCM_ENABLED]

cmd/server/main.go

@@ -62,15 +62,24 @@ func main() {
 	}
 
 	db := initDb()
+
 	ctx, cancel := context.WithCancel(context.Background())
-	installationsService := installations.NewInstallationsService(logger, db)
-	subscriptionsService := subscriptions.NewSubscriptionsService(logger, db)
-	var listener *xmtp.Listener
-	var apiServer *api.ApiServer
+
+	installationsService := installations.NewService(logger, db)
+	subscriptionsService := subscriptions.NewService(logger, db)
+
+	var (
+		listener  *xmtp.Listener
+		apiServer *api.ApiServer
+	)
 
 	if opts.Xmtp.ListenerEnabled {
-		deliveryServices := []interfaces.Delivery{}
-		var err error
+
+		// Delivery services: APNs, FCM or HTTP.
+		var (
+			deliveryServices = []interfaces.Delivery{}
+			err              error
+		)
 
 		if opts.Apns.Enabled {
 			apns, err := delivery.NewApnsDelivery(logger, opts.Apns)
@@ -127,12 +136,12 @@ func waitForShutdown() {
 func initDb() *bun.DB {
 	db, err := database.CreateBunDB(opts.DbConnectionString, 10*time.Second)
 	if err != nil {
-		log.Fatal("db creation error", zap.Error(err))
+		logger.Fatal("db creation error", zap.Error(err))
 	}
 
 	err = database.Migrate(context.Background(), db)
 	if err != nil {
-		log.Fatal("db migration error", zap.Error(err))
+		logger.Fatal("db migration error", zap.Error(err))
 	}
 
 	return db

dev/gen-proto

@@ -13,6 +13,16 @@ if [ -n "$(go env GOPATH)" ]; then
     export PATH="${FOUND_GOPATH}/bin:${PATH}"
 fi
 
+# Use a local proto repo path if provided, otherwise pull from the Buf Schema Registry
+# Usage: ./dev/gen-proto [/path/to/local/xmtp/proto]
+if [ -n "${1:-}" ]; then
+    XMTP_PROTO_SOURCE="$1"
+    echo "Using local proto repo: $XMTP_PROTO_SOURCE"
+else
+    XMTP_PROTO_SOURCE="buf.build/xmtp/proto"
+    echo "Using BSR: $XMTP_PROTO_SOURCE"
+fi
+
 rm -rf ./pkg/proto
 buf generate --template proto/buf.gen.yaml
-buf generate buf.build/xmtp/proto
+buf generate "$XMTP_PROTO_SOURCE"

go.mod

@@ -5,47 +5,47 @@ go 1.25
 require (
 	connectrpc.com/connect v1.19.1
 	firebase.google.com/go/v4 v4.15.2
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.0
 	github.com/jessevdk/go-flags v1.6.1
 	github.com/pkg/errors v0.9.1
 	github.com/sideshow/apns2 v0.25.0
-	github.com/stretchr/testify v1.10.0
+	github.com/stretchr/testify v1.11.1
 	github.com/uptrace/bun v1.2.11
 	github.com/uptrace/bun/dialect/pgdialect v1.2.11
 	github.com/uptrace/bun/driver/pgdriver v1.2.11
 	github.com/uptrace/bun/extra/bundebug v1.2.11
 	go.uber.org/zap v1.27.0
-	golang.org/x/net v0.39.0
+	golang.org/x/net v0.48.0
 	google.golang.org/api v0.231.0
-	google.golang.org/genproto/googleapis/api v0.0.0-20250428153025-10db94c68c34
-	google.golang.org/grpc v1.72.0
-	google.golang.org/protobuf v1.36.9
+	google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda
+	google.golang.org/grpc v1.78.0
+	google.golang.org/protobuf v1.36.10
 )
 
 require (
-	cel.dev/expr v0.23.1 // indirect
+	cel.dev/expr v0.24.0 // indirect
 	cloud.google.com/go v0.121.0 // indirect
 	cloud.google.com/go/auth v0.16.1 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
-	cloud.google.com/go/compute/metadata v0.6.0 // indirect
+	cloud.google.com/go/compute/metadata v0.9.0 // indirect
 	cloud.google.com/go/firestore v1.18.0 // indirect
 	cloud.google.com/go/iam v1.5.2 // indirect
 	cloud.google.com/go/longrunning v0.6.7 // indirect
 	cloud.google.com/go/monitoring v1.24.2 // indirect
 	cloud.google.com/go/storage v1.52.0 // indirect
-	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.27.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.51.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0 // indirect
 	github.com/MicahParks/keyfunc v1.9.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/cncf/xds/go v0.0.0-20250326154945-ae57f3c0d45f // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/envoyproxy/go-control-plane/envoy v1.32.4 // indirect
+	github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/envoyproxy/go-control-plane/envoy v1.35.0 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
-	github.com/go-jose/go-jose/v4 v4.1.0 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-jose/go-jose/v4 v4.1.3 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.5.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
@@ -57,33 +57,32 @@ require (
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect
-	github.com/spiffe/go-spiffe/v2 v2.5.0 // indirect
+	github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc // indirect
 	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
-	github.com/zeebo/errs v1.4.0 // indirect
-	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/contrib/detectors/gcp v1.35.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
+	go.opentelemetry.io/contrib/detectors/gcp v1.38.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
-	go.opentelemetry.io/otel v1.35.0 // indirect
-	go.opentelemetry.io/otel/metric v1.35.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
-	go.opentelemetry.io/otel/trace v1.35.0 // indirect
+	go.opentelemetry.io/otel v1.39.0 // indirect
+	go.opentelemetry.io/otel/metric v1.39.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.39.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect
+	go.opentelemetry.io/otel/trace v1.39.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/crypto v0.37.0 // indirect
-	golang.org/x/oauth2 v0.29.0 // indirect
-	golang.org/x/sync v0.13.0 // indirect
-	golang.org/x/sys v0.32.0 // indirect
-	golang.org/x/text v0.24.0 // indirect
-	golang.org/x/time v0.11.0 // indirect
+	golang.org/x/crypto v0.46.0 // indirect
+	golang.org/x/oauth2 v0.32.0 // indirect
+	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sys v0.39.0 // indirect
+	golang.org/x/text v0.32.0 // indirect
+	golang.org/x/time v0.14.0 // indirect
 	google.golang.org/appengine/v2 v2.0.6 // indirect
 	google.golang.org/genproto v0.0.0-20250428153025-10db94c68c34 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251029180050-ab9386a59fda // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	mellium.im/sasl v0.3.2 // indirect
 )