feature: Pause application webhooks

CHANGELOG.md

@@ -34,6 +34,7 @@ For details about compatibility between different releases, see the **Commitment
  - `ListGatewaysRequest` and `ListEndDevicesRequest` RPCs have a new `Filter` field that supports an `updated_since` timestamp.
 - Preparation for universal rights assigned to users.
  - This requires a database schema migration (`ttn-lw-stack is-db migrate`).
+- Option to pause application webhooks.
 
 ### Deprecated
 

api/ttn/lorawan/v3/api.md

@@ -1,4 +1,4 @@
 <a name="top"></a>
 
 # API Documentation
 
@@ -2073,6 +2073,7 @@
 | `service_data` | [`ApplicationWebhook.Message`](#ttn.lorawan.v3.ApplicationWebhook.Message) | | |
 | `health_status` | [`ApplicationWebhookHealth`](#ttn.lorawan.v3.ApplicationWebhookHealth) | | |
 | `field_mask` | [`google.protobuf.FieldMask`](#google.protobuf.FieldMask) | | |
+| `paused` | [`bool`](#bool) | | Set to temporarily pause forwarding uplink data to this end point and receiving downlinks from this end point. |
 
 #### Field Rules
 

api/ttn/lorawan/v3/api.swagger.json

@@ -1,4 +1,4 @@
 {
  "swagger": "2.0",
  "info": {
  "title": "The Things Stack for LoRaWAN v3 API",
@@ -20844,6 +20844,10 @@
  },
  "field_mask": {
  "type": "string"
+ },
+ "paused": {
+ "type": "boolean",
+ "description": "Set to temporarily pause forwarding uplink data to this end point and receiving downlinks from this end point."
  }
  }
  },
@@ -20983,6 +20987,10 @@
  },
  "field_mask": {
  "type": "string"
+ },
+ "paused": {
+ "type": "boolean",
+ "description": "Set to temporarily pause forwarding uplink data to this end point and receiving downlinks from this end point."
  }
  }
  },

api/ttn/lorawan/v3/applicationserver_web.proto

@@ -227,7 +227,10 @@ message ApplicationWebhook {
  reserved 23;
  reserved "queue";
 
- // next: 24
+ // Set to temporarily pause forwarding uplink data to this end point and receiving downlinks from this end point.
+ bool paused = 24;
+
+ // next: 25
 }
 
 message ApplicationWebhooks {

pkg/applicationserver/io/web/webhooks.go

@@ -52,6 +52,7 @@ var webhookFanOutFieldMask = []string{
  "join_accept",
  "location_solved",
  "service_data",
+ "paused",
  "uplink_message",
  "uplink_normalized",
 }
@@ -135,6 +136,12 @@ func (w *webhooks) handleUp(ctx context.Context, msg *ttnpb.ApplicationUp) error
  Health: hook.HealthStatus,
  })
  ctx = log.NewContextWithField(ctx, "hook", hook.Ids.WebhookId)
+
+ if hook.Paused {
+ log.FromContext(ctx).Debug("Webhook is paused")
+ continue
+ }
+
  f := func(ctx context.Context) error {
  req, err := NewRequest(ctx, w.downlinks, msg, hook)
  if err != nil {
@@ -185,7 +192,7 @@ func (w *webhooks) handleDown(
  "webhook_id", hookID.WebhookId,
  ))
 
- hook, err := w.registry.Get(ctx, hookID, []string{"format"})
+ hook, err := w.registry.Get(ctx, hookID, []string{"format", "paused"})
  if err != nil {
  webhandlers.Error(res, req, err)
  return
@@ -194,6 +201,11 @@ func (w *webhooks) handleDown(
  webhandlers.Error(res, req, errWebhookNotFound.New())
  return
  }
+ if hook.Paused {
+ logger.Debug("Webhook is paused")
+ res.WriteHeader(http.StatusNotAcceptable)
+ return
+ }
  format, ok := formats[hook.Format]
  if !ok {
  webhandlers.Error(res, req, errFormatNotFound.WithAttributes("format", hook.Format))

pkg/applicationserver/io/web/webhooks_test.go

@@ -149,6 +149,7 @@ func TestWebhooks(t *testing.T) {
  ServiceData: &ttnpb.ApplicationWebhook_Message{
  Path: tc.prefix + "service/data",
  },
+ Paused: false,
  FieldMask: ttnpb.FieldMask(
  "correlation_ids",
  "end_device_ids",
@@ -162,6 +163,7 @@ func TestWebhooks(t *testing.T) {
  "up.downlink_sent",
  "up.join_accept",
  "up.location_solved",
+ "up.paused",
  "up.service_data",
  "up.uplink_message",
  "up.uplink_normalized",
@@ -183,6 +185,7 @@ func TestWebhooks(t *testing.T) {
  "join_accept",
  "location_solved",
  "service_data",
+ "paused",
  "uplink_message",
  "uplink_normalized",
  }, nil
@@ -574,4 +577,202 @@ func TestWebhooks(t *testing.T) {
  }
  })
  })
+
+ //nolint:paralleltest
+ t.Run("PausedUplink", func(t *testing.T) {
+ _, ctx := test.New(t)
+
+ // Create an active webhook.
+ _, err := registry.Set(ctx, ids, nil,
+ func(_ *ttnpb.ApplicationWebhook) (*ttnpb.ApplicationWebhook, []string, error) {
+ return &ttnpb.ApplicationWebhook{
+ Ids: ids,
+ BaseUrl: "https://myapp.com/api/ttn/v3/{/appID,devID}",
+ Format: "json",
+ Paused: false,
+ UplinkMessage: &ttnpb.ApplicationWebhook_Message{Path: "/"},
+ },
+ []string{
+ "ids.application_ids",
+ "ids.webhook_id",
+ "base_url",
+ "format",
+ "paused",
+ "uplink_message",
+ "field_mask",
+ }, nil
+ })
+ if err != nil {
+ t.Fatalf("Failed to set webhook in registry: %s", err)
+ }
+
+ sinkCh := make(chan *http.Request, 1)
+ testSink := mocksink.New(sinkCh)
+
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ c := componenttest.NewComponent(t, &component.Config{})
+ componenttest.StartComponent(t, c)
+ defer c.Close()
+
+ as := mock.NewServer(c)
+ _, err = web.NewWebhooks(ctx, as, registry, testSink, downlinks)
+ if err != nil {
+ t.Fatalf("Unexpected error %v", err)
+ }
+
+ message := &ttnpb.ApplicationUp{
+ EndDeviceIds: registeredDeviceID,
+ Up: &ttnpb.ApplicationUp_UplinkMessage{
+ UplinkMessage: &ttnpb.ApplicationUplink{
+ SessionKeyId: []byte{0x11},
+ FPort: 42,
+ FCnt: 42,
+ FrmPayload: []byte{0x1, 0x2, 0x3},
+ },
+ },
+ }
+
+ a := assertions.New(t)
+ err = as.Publish(ctx, message)
+ if !a.So(err, should.BeNil) {
+ t.FailNow()
+ }
+
+ var req *http.Request
+ select {
+ case req = <-sinkCh:
+ case <-time.After(timeout):
+ t.Fatal("Expected message but nothing received")
+ }
+
+ actualBody, err := stdio.ReadAll(req.Body)
+ if !a.So(err, should.BeNil) {
+ t.FailNow()
+ }
+ expectedBody, err := formatters.JSON.FromUp(message)
+ if !a.So(err, should.BeNil) {
+ t.FailNow()
+ }
+ a.So(actualBody, should.Resemble, expectedBody)
+
+ // Pause the webhook.
+ _, err = registry.Set(ctx, ids, nil,
+ func(_ *ttnpb.ApplicationWebhook) (*ttnpb.ApplicationWebhook, []string, error) {
+ return &ttnpb.ApplicationWebhook{
+ Ids: ids,
+ BaseUrl: "https://myapp.com/api/ttn/v3/{/appID,devID}",
+ Format: "json",
+ Paused: true,
+ },
+ []string{
+ "ids.application_ids",
+ "ids.webhook_id",
+ "base_url",
+ "format",
+ "paused",
+ "field_mask",
+ }, nil
+ })
+ if err != nil {
+ t.Fatalf("Failed to set webhook in registry: %s", err)
+ }
+
+ err = as.Publish(ctx, message)
+ if !a.So(err, should.BeNil) {
+ t.FailNow()
+ }
+
+ select {
+ case req = <-sinkCh:
+ t.Fatalf("Did not expect message but received: %v", req)
+ case <-time.After(timeout):
+ // Webhook was not received.
+ }
+ })
+
+ //nolint:paralleltest
+ t.Run("PausedDownlink", func(t *testing.T) {
+ is, isAddr, closeIS := mockis.New(ctx)
+ defer closeIS()
+
+ is.ApplicationRegistry().Add(ctx, registeredApplicationID, registeredApplicationKey,
+ ttnpb.Right_RIGHT_APPLICATION_SETTINGS_BASIC,
+ ttnpb.Right_RIGHT_APPLICATION_DEVICES_READ,
+ ttnpb.Right_RIGHT_APPLICATION_DEVICES_WRITE,
+ ttnpb.Right_RIGHT_APPLICATION_TRAFFIC_READ,
+ ttnpb.Right_RIGHT_APPLICATION_TRAFFIC_DOWN_WRITE)
+ conf := &component.Config{
+ ServiceBase: config.ServiceBase{
+ GRPC: config.GRPC{
+ Listen: ":0",
+ AllowInsecureForCredentials: true,
+ },
+ Cluster: cluster.Config{
+ IdentityServer: isAddr,
+ },
+ },
+ }
+
+ _, err := registry.Set(ctx, ids, nil,
+ func(_ *ttnpb.ApplicationWebhook) (*ttnpb.ApplicationWebhook, []string, error) {
+ return &ttnpb.ApplicationWebhook{
+ Ids: ids,
+ BaseUrl: "https://myapp.com/api/ttn/v3/{/appID,devID}",
+ Format: "json",
+ Paused: true,
+ UplinkMessage: &ttnpb.ApplicationWebhook_Message{Path: "/"},
+ },
+ []string{
+ "ids.application_ids",
+ "ids.webhook_id",
+ "base_url",
+ "format",
+ "paused",
+ "uplink_message",
+ "field_mask",
+ }, nil
+ })
+ if err != nil {
+ t.Fatalf("Failed to set webhook in registry: %s", err)
+ }
+
+ sinkCh := make(chan *http.Request, 1)
+ testSink := mocksink.New(sinkCh)
+
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ c := componenttest.NewComponent(t, conf)
+ as := mock.NewServer(c)
+ w, err := web.NewWebhooks(ctx, as, registry, testSink, downlinks)
+ if err != nil {
+ t.Fatalf("Unexpected error %v", err)
+ }
+ c.RegisterWeb(w)
+ componenttest.StartComponent(t, c)
+ defer c.Close()
+
+ mustHavePeer(ctx, c, ttnpb.ClusterRole_ENTITY_REGISTRY)
+
+ // Check the status error code when scheduling downlink to a paused webhook.
+ url := fmt.Sprintf("/api/v3/as/applications/%s/webhooks/%s/devices/%s/down/replace",
+ ids.ApplicationIds.ApplicationId, ids.WebhookId, registeredDeviceID.DeviceId,
+ )
+ body := bytes.NewReader([]byte(`{"downlinks":[]}`))
+ req := httptest.NewRequest(http.MethodPost, url, body)
+ req.Header.Set("Content-Type", "application/json")
+ req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", registeredApplicationKey))
+
+ res := httptest.NewRecorder()
+ c.ServeHTTP(res, req)
+ a.So(res.Code, should.Equal, http.StatusNotAcceptable)
+ downlinks, err := as.DownlinkQueueList(ctx, registeredDeviceID)
+ if !a.So(err, should.BeNil) {
+ t.FailNow()
+ }
+
+ a.So(downlinks, should.Resemble, []*ttnpb.ApplicationDownlink{})
+ })
 }