Merge pull request #6773 from TheThingsNetwork/issue/6751-notificatio…

…n-fanout-opt

Organization `fanout_notifications` option

CHANGELOG.md

@@ -42,6 +42,9 @@ For details about compatibility between different releases, see the **Commitment
   - This enables CUPS-enabled gateways to change their LNS before the periodic CUPS lookup occurs.
 - The LoRa Basics Station discovery endpoint now verifies the authorization credentials of the caller.
   - This enables the gateways to migrate to another instance gracefully while using CUPS.
+- Organizations can now opt out from sending administrative and technical notifications to its members.
+  - New organizations do not send administrative and technical notifications to its members by default.
+  - To alter the behavior update the organization's `fanout_notifications` field.
 
 ### Fixed
 

api/ttn/lorawan/v3/api.md

@@ -9236,6 +9236,7 @@ is used to manage OAuth client authorizations for users.
 | `contact_info` | [`ContactInfo`](#ttn.lorawan.v3.ContactInfo) | repeated | Contact information for this organization. Typically used to indicate who to contact with security/billing questions about the organization. This field is deprecated. Use administrative_contact and technical_contact instead. |
 | `administrative_contact` | [`OrganizationOrUserIdentifiers`](#ttn.lorawan.v3.OrganizationOrUserIdentifiers) |  |  |
 | `technical_contact` | [`OrganizationOrUserIdentifiers`](#ttn.lorawan.v3.OrganizationOrUserIdentifiers) |  |  |
+| `fanout_notifications` | [`bool`](#bool) |  | Determines if a notification will be sent to the collaborators. If false it, notifications will be sent only to the administrative or technical contact. |
 
 #### Field Rules
 

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

@@ -13811,6 +13811,10 @@
                     },
                     "technical_contact": {
                       "$ref": "#/definitions/v3OrganizationOrUserIdentifiers"
+                    },
+                    "fanout_notifications": {
+                      "type": "boolean",
+                      "description": "Determines if a notification will be sent to the collaborators. If false it, notifications will be sent only to the\nadministrative or technical contact."
                     }
                   }
                 },
@@ -25609,6 +25613,10 @@
         },
         "technical_contact": {
           "$ref": "#/definitions/v3OrganizationOrUserIdentifiers"
+        },
+        "fanout_notifications": {
+          "type": "boolean",
+          "description": "Determines if a notification will be sent to the collaborators. If false it, notifications will be sent only to the\nadministrative or technical contact."
         }
       }
     },

api/ttn/lorawan/v3/organization.proto

@@ -91,7 +91,11 @@ message Organization {
   reserved 13;
   reserved "gateway_limit";
 
-  // next: 14
+  // Determines if a notification will be sent to the collaborators. If false it, notifications will be sent only to the
+  // administrative or technical contact.
+  bool fanout_notifications = 14;
+
+  // next: 15
 }
 
 message Organizations {

pkg/identityserver/bunstore/organization_store.go

@@ -51,6 +51,8 @@ type Organization struct {
 
 	TechnicalContactID *string  `bun:"technical_contact_id,type:uuid"`
 	TechnicalContact   *Account `bun:"rel:belongs-to,join:technical_contact_id=id"`
+
+	FanoutNotifications bool `bun:"fanout_notifications"`
 }
 
 // BeforeAppendModel is a hook that modifies the model on SELECT and UPDATE queries.
@@ -71,8 +73,9 @@ func organizationToPB(m *Organization, fieldMask ...string) (*ttnpb.Organization
 		UpdatedAt: timestamppb.New(m.UpdatedAt),
 		DeletedAt: ttnpb.ProtoTime(m.DeletedAt),
 
-		Name:        m.Name,
-		Description: m.Description,
+		Name:                m.Name,
+		Description:         m.Description,
+		FanoutNotifications: m.FanoutNotifications,
 	}
 
 	if len(m.Attributes) > 0 {
@@ -137,8 +140,9 @@ func (s *organizationStore) CreateOrganization(
 		Account: EmbeddedAccount{
 			UID: pb.GetIds().GetOrganizationId(),
 		},
-		Name:        pb.Name,
-		Description: pb.Description,
+		Name:                pb.Name,
+		Description:         pb.Description,
+		FanoutNotifications: pb.FanoutNotifications,
 	}
 
 	if contact := pb.AdministrativeContact; contact != nil {
@@ -229,7 +233,7 @@ func (*organizationStore) selectWithFields(q *bun.SelectQuery, fieldMask store.F
 				return nil, fmt.Errorf("unknown field %q", f)
 			case "ids", "created_at", "updated_at", "deleted_at":
 				// Always selected.
-			case "name", "description":
+			case "name", "description", "fanout_notifications":
 				// Proto name equals model name.
 				columns = append(columns, f)
 			case "attributes":
@@ -441,6 +445,10 @@ func (s *organizationStore) updateOrganizationModel( //nolint:gocyclo
 				model.TechnicalContactID = nil
 			}
 			columns = append(columns, "technical_contact_id")
+
+		case "fanout_notifications":
+			model.FanoutNotifications = pb.FanoutNotifications
+			columns = append(columns, "fanout_notifications")
 		}
 	}
 

pkg/identityserver/notification_registry.go

@@ -78,6 +78,34 @@ func (is *IdentityServer) notifyInternal(ctx context.Context, req *ttnpb.CreateN
 
 var errNoReceiverUserIDs = errors.Define("no_receiver_user_ids", "no receiver users ids")
 
+// getContactReceivers checks if the entityID to provide the appropriate receiverID.
+// If is an user, returns the entityID.
+// If is an organization, it checks if the fanout_notifications is enabled. If enabled returns the organizationID but
+// otherwise it returns the organization's administrative or technical contact.
+func (is *IdentityServer) getContactReceivers(
+	ctx context.Context, entityID *ttnpb.OrganizationOrUserIdentifiers, entityMask []string,
+) (*ttnpb.OrganizationOrUserIdentifiers, error) {
+	if entityID.EntityType() != "organization" {
+		return entityID, nil
+	}
+	org, err := is.store.GetOrganization(
+		ctx, entityID.GetOrganizationIds(), append(entityMask, "fanout_notifications"),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if org.FanoutNotifications {
+		return entityID, nil
+	}
+	if contact := org.GetAdministrativeContact(); contact != nil {
+		return contact, nil
+	}
+	if contact := org.GetTechnicalContact(); contact != nil {
+		return contact, nil
+	}
+	return entityID, nil
+}
+
 func (is *IdentityServer) lookupNotificationReceivers(ctx context.Context, req *ttnpb.CreateNotificationRequest) ([]*ttnpb.UserIdentifiers, error) {
 	var receiverIDs []*ttnpb.OrganizationOrUserIdentifiers
 	err := is.store.Transact(ctx, func(ctx context.Context, st store.Store) error {
@@ -120,11 +148,23 @@ func (is *IdentityServer) lookupNotificationReceivers(ctx context.Context, req *
 				return err
 			}
 			if entity != nil { // NOTE: entity is nil for entities that don't support contacts.
-				if receiversContains(req.Receivers, ttnpb.NotificationReceiver_NOTIFICATION_RECEIVER_ADMINISTRATIVE_CONTACT) && entity.GetAdministrativeContact() != nil {
-					receiverIDs = append(receiverIDs, entity.GetAdministrativeContact())
+				adminContact, err := is.getContactReceivers(
+					ctx, entity.GetAdministrativeContact(), []string{"administrative_contact"},
+				)
+				if err != nil {
+					return err
+				}
+				if adminContact != nil {
+					receiverIDs = append(receiverIDs, adminContact)
+				}
+				techContact, err := is.getContactReceivers(
+					ctx, entity.GetTechnicalContact(), []string{"technical_contact"},
+				)
+				if err != nil {
+					return err
 				}
-				if receiversContains(req.Receivers, ttnpb.NotificationReceiver_NOTIFICATION_RECEIVER_TECHNICAL_CONTACT) && entity.GetTechnicalContact() != nil {
-					receiverIDs = append(receiverIDs, entity.GetTechnicalContact())
+				if techContact != nil {
+					receiverIDs = append(receiverIDs, techContact)
 				}
 			}
 		}

pkg/identityserver/notification_registry_test.go

@@ -15,6 +15,7 @@
 package identityserver
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -61,7 +62,7 @@ func TestNotificationRegistry(t *testing.T) {
 					ttnpb.NotificationReceiver_NOTIFICATION_RECEIVER_ADMINISTRATIVE_CONTACT,
 					ttnpb.NotificationReceiver_NOTIFICATION_RECEIVER_TECHNICAL_CONTACT,
 				},
-			}, is.WithClusterAuth())
+			}, is.Component.WithClusterAuth())
 			if a.So(err, should.BeNil) && a.So(res, should.NotBeNil) {
 				a.So(res.Id, should.NotBeZeroValue)
 			}
@@ -98,3 +99,96 @@ func TestNotificationRegistry(t *testing.T) {
 		}
 	}, withPrivateTestDatabase(p))
 }
+
+// TestNotificationRegistryWithOrganizationFanout contains the test case for notifications that trigger the organization
+// fanout, validating the expected behavior. More details below.
+//
+// Base conditions:
+//   - Three users, the owner and two collaborators.
+//   - Two organizations, one with fanout and another without.
+//   - Two applications, one for each organization. The organization is the owner and admin/tech contact.
+//
+// Expected behavior:
+//   - Create notification for app-1 (no fanout). Only the owner should receive the notification.
+//   - Create notification for app2-2 (has fanout). Both the owner and collaborators should receive the notification.
+func TestNotificationRegistryWithOrganizationFanout(t *testing.T) {
+	t.Parallel()
+	p := &storetest.Population{}
+
+	usr1 := p.NewUser()
+	usr1Key, _ := p.NewAPIKey(usr1.GetEntityIdentifiers(), ttnpb.Right_RIGHT_ALL)
+	usr1Creds := rpcCreds(usr1Key)
+
+	usr2 := p.NewUser()
+	usr2Key, _ := p.NewAPIKey(usr2.GetEntityIdentifiers(), ttnpb.Right_RIGHT_ALL)
+	usr2Creds := rpcCreds(usr2Key)
+	usr3 := p.NewUser()
+	usr3Key, _ := p.NewAPIKey(usr3.GetEntityIdentifiers(), ttnpb.Right_RIGHT_ALL)
+	usr3Creds := rpcCreds(usr3Key)
+
+	org1 := p.NewOrganization(usr1.GetOrganizationOrUserIdentifiers())
+	org2 := p.NewOrganization(usr1.GetOrganizationOrUserIdentifiers())
+	org2.FanoutNotifications = true
+
+	// Register users as collaborators on both organizations.
+	for _, org := range []*ttnpb.Organization{org1, org2} {
+		org := org
+		for _, collab := range []*ttnpb.User{usr2, usr3} {
+			collab := collab
+			p.NewMembership(
+				collab.GetOrganizationOrUserIdentifiers(),
+				org.GetEntityIdentifiers(),
+				ttnpb.Right_RIGHT_ORGANIZATION_ALL,
+			)
+		}
+	}
+
+	// Sends notification to application's admin/tech contacts.
+	// Depending on the organization's fanout setting, the amount of receivers will vary.
+	app1 := p.NewApplication(org1.GetOrganizationOrUserIdentifiers())
+	app2 := p.NewApplication(org2.GetOrganizationOrUserIdentifiers())
+
+	a, ctx := test.New(t)
+	testWithIdentityServer(t, func(is *IdentityServer, cc *grpc.ClientConn) {
+		svc := ttnpb.NewNotificationServiceClient(cc)
+
+		for _, entityID := range []*ttnpb.EntityIdentifiers{
+			app1.GetIds().GetEntityIdentifiers(),
+			app2.GetIds().GetEntityIdentifiers(),
+		} {
+			entityID := entityID
+			res, err := svc.Create(ctx, &ttnpb.CreateNotificationRequest{
+				EntityIds:        entityID,
+				NotificationType: "test_notification",
+				Receivers: []ttnpb.NotificationReceiver{
+					ttnpb.NotificationReceiver_NOTIFICATION_RECEIVER_ADMINISTRATIVE_CONTACT,
+					ttnpb.NotificationReceiver_NOTIFICATION_RECEIVER_TECHNICAL_CONTACT,
+				},
+			}, is.Component.WithClusterAuth())
+			if a.So(err, should.BeNil) && a.So(res, should.NotBeNil) {
+				a.So(res.Id, should.NotBeZeroValue)
+			}
+		}
+
+		// Validates the amount of notifications received by each user.
+		for _, tt := range []struct {
+			userID             *ttnpb.UserIdentifiers
+			notificationAmount int
+			creds              grpc.CallOption
+		}{
+			{userID: usr1.GetIds(), notificationAmount: 2, creds: usr1Creds},
+			{userID: usr2.GetIds(), notificationAmount: 1, creds: usr2Creds},
+			{userID: usr3.GetIds(), notificationAmount: 1, creds: usr3Creds},
+		} {
+			tt := tt
+			ttName := fmt.Sprintf("Expect %s to have %d notifications", tt.userID.GetUserId(), tt.notificationAmount)
+			t.Run(ttName, func(t *testing.T) { // nolint:paralleltest
+				a, ctx := test.New(t)
+				list, err := svc.List(ctx, &ttnpb.ListNotificationsRequest{ReceiverIds: tt.userID}, tt.creds)
+				a.So(err, should.BeNil)
+				a.So(list, should.NotBeNil)
+				a.So(list.Notifications, should.HaveLength, tt.notificationAmount)
+			})
+		}
+	}, withPrivateTestDatabase(p))
+}

pkg/identityserver/store/migrations/20231211000000_organization_notification_fanout.down.sql

@@ -0,0 +1,16 @@
+ALTER TABLE organizations
+DROP COLUMN IF EXISTS fanout_notifications CASCADE;
+
+--bun:split
+CREATE OR REPLACE VIEW organization_accounts AS
+SELECT
+  acc.id AS account_id,
+  acc.created_at AS account_created_at,
+  acc.updated_at AS account_updated_at,
+  acc.deleted_at AS account_deleted_at,
+  acc.uid AS account_uid,
+  org.*
+FROM
+  accounts acc
+  JOIN organizations org ON org.id = acc.account_id
+  AND acc.account_type = 'organization';

pkg/identityserver/store/migrations/20231211000000_organization_notification_fanout.up.sql

@@ -0,0 +1,22 @@
+ALTER TABLE
+  organizations
+ADD
+  COLUMN IF NOT EXISTS fanout_notifications BOOLEAN;
+
+--bun:split
+UPDATE organizations SET fanout_notifications = TRUE;
+
+--bun:split
+CREATE
+OR REPLACE VIEW organization_accounts AS
+SELECT
+  acc.id AS account_id,
+  acc.created_at AS account_created_at,
+  acc.updated_at AS account_updated_at,
+  acc.deleted_at AS account_deleted_at,
+  acc.uid AS account_uid,
+  org.*
+FROM
+  accounts acc
+  JOIN organizations org ON org.id = acc.account_id
+  AND acc.account_type = 'organization';

pkg/identityserver/storetest/organization_store.go

@@ -132,6 +132,7 @@ func (st *StoreTest) TestOrganizationStoreCRUD(t *T) {
 			Attributes:            updatedAttributes,
 			AdministrativeContact: usr2.GetOrganizationOrUserIdentifiers(),
 			TechnicalContact:      usr1.GetOrganizationOrUserIdentifiers(),
+			FanoutNotifications:   true,
 		}, mask)
 		if a.So(err, should.BeNil) && a.So(updated, should.NotBeNil) {
 			a.So(updated.GetIds().GetOrganizationId(), should.Equal, "foo")
@@ -140,6 +141,7 @@ func (st *StoreTest) TestOrganizationStoreCRUD(t *T) {
 			a.So(updated.Attributes, should.Resemble, updatedAttributes)
 			a.So(updated.AdministrativeContact, should.Resemble, usr2.GetOrganizationOrUserIdentifiers())
 			a.So(updated.TechnicalContact, should.Resemble, usr1.GetOrganizationOrUserIdentifiers())
+			a.So(updated.FanoutNotifications, should.BeTrue)
 			a.So(*ttnpb.StdTime(updated.CreatedAt), should.Equal, *ttnpb.StdTime(created.CreatedAt))
 			a.So(*ttnpb.StdTime(updated.UpdatedAt), should.HappenWithin, 5*time.Second, start)
 		}