Adds websocket messages to Custom Profile Attributes (#30163)

* Adds websocket messages to Custom Profile Attributes

The app layer now fires a websocket event as part of the operations
over Custom Profile Attribute fields and values. It updates as well
the Patch method for CPA values so all the changes are commited as
part of the same transaction.

To be able to do this last operation, the change adds methods to
upsert CPA values in both the store and the property service.

* Fix i18n strings

---------

Co-authored-by: Mattermost Build <build@mattermost.com>
Co-authored-by: Miguel de la Cruz <miguel@ctrlz.es>

server/channels/api4/custom_profile_attributes_test.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/mattermost/mattermost/server/public/model"
 	"github.com/stretchr/testify/require"
@@ -54,6 +55,8 @@ func TestCreateCPAField(t *testing.T) {
 	}, "an invalid field should be rejected")
 
 	th.TestForSystemAdminAndLocal(t, func(t *testing.T, client *model.Client4) {
+		webSocketClient := th.CreateConnectedWebSocketClient(t)
+
 		name := model.NewId()
 		field := &model.PropertyField{
 			Name:  fmt.Sprintf("  %s\t", name), // name should be sanitized
@@ -67,6 +70,27 @@ func TestCreateCPAField(t *testing.T) {
 		require.NotZero(t, createdField.ID)
 		require.Equal(t, name, createdField.Name)
 		require.Equal(t, "default", createdField.Attrs["visibility"])
+
+		t.Run("a websocket event should be fired as part of the field creation", func(t *testing.T) {
+			var wsField model.PropertyField
+			require.Eventually(t, func() bool {
+				select {
+				case event := <-webSocketClient.EventChannel:
+					if event.EventType() == model.WebsocketEventCPAFieldCreated {
+						fieldData, err := json.Marshal(event.GetData()["field"])
+						require.NoError(t, err)
+						require.NoError(t, json.Unmarshal(fieldData, &wsField))
+						return true
+					}
+				default:
+					return false
+				}
+				return false
+			}, 5*time.Second, 100*time.Millisecond)
+
+			require.NotEmpty(t, wsField.ID)
+			require.Equal(t, createdField, &wsField)
+		})
 	}, "a user with admin permissions should be able to create the field")
 }
 
@@ -149,6 +173,8 @@ func TestPatchCPAField(t *testing.T) {
 	})
 
 	th.TestForSystemAdminAndLocal(t, func(t *testing.T, client *model.Client4) {
+		webSocketClient := th.CreateConnectedWebSocketClient(t)
+
 		field := &model.PropertyField{
 			Name: model.NewId(),
 			Type: model.PropertyFieldTypeText,
@@ -163,6 +189,27 @@ func TestPatchCPAField(t *testing.T) {
 		CheckOKStatus(t, resp)
 		require.NoError(t, err)
 		require.Equal(t, newName, patchedField.Name)
+
+		t.Run("a websocket event should be fired as part of the field patch", func(t *testing.T) {
+			var wsField model.PropertyField
+			require.Eventually(t, func() bool {
+				select {
+				case event := <-webSocketClient.EventChannel:
+					if event.EventType() == model.WebsocketEventCPAFieldUpdated {
+						fieldData, err := json.Marshal(event.GetData()["field"])
+						require.NoError(t, err)
+						require.NoError(t, json.Unmarshal(fieldData, &wsField))
+						return true
+					}
+				default:
+					return false
+				}
+				return false
+			}, 5*time.Second, 100*time.Millisecond)
+
+			require.NotEmpty(t, wsField.ID)
+			require.Equal(t, patchedField, &wsField)
+		})
 	}, "a user with admin permissions should be able to patch the field")
 }
 
@@ -197,6 +244,8 @@ func TestDeleteCPAField(t *testing.T) {
 	})
 
 	th.TestForSystemAdminAndLocal(t, func(t *testing.T, client *model.Client4) {
+		webSocketClient := th.CreateConnectedWebSocketClient(t)
+
 		field := &model.PropertyField{
 			Name: model.NewId(),
 			Type: model.PropertyFieldTypeText,
@@ -213,6 +262,26 @@ func TestDeleteCPAField(t *testing.T) {
 		deletedField, appErr := th.App.GetCPAField(createdField.ID)
 		require.Nil(t, appErr)
 		require.NotZero(t, deletedField.DeleteAt)
+
+		t.Run("a websocket event should be fired as part of the field deletion", func(t *testing.T) {
+			var fieldID string
+			require.Eventually(t, func() bool {
+				select {
+				case event := <-webSocketClient.EventChannel:
+					if event.EventType() == model.WebsocketEventCPAFieldDeleted {
+						var ok bool
+						fieldID, ok = event.GetData()["field_id"].(string)
+						require.True(t, ok)
+						return true
+					}
+				default:
+					return false
+				}
+				return false
+			}, 5*time.Second, 100*time.Millisecond)
+
+			require.Equal(t, createdField.ID, fieldID)
+		})
 	}, "a user with admin permissions should be able to delete the field")
 }
 
@@ -470,6 +539,8 @@ func TestPatchCPAValues(t *testing.T) {
 	th.App.Srv().SetLicense(model.NewTestLicenseSKU(model.LicenseShortSkuEnterprise))
 
 	t.Run("any team member should be able to create their own values", func(t *testing.T) {
+		webSocketClient := th.CreateConnectedWebSocketClient(t)
+
 		values := map[string]json.RawMessage{}
 		value := "Field Value"
 		values[createdField.ID] = json.RawMessage(fmt.Sprintf(`"  %s "`, value)) // value should be sanitized
@@ -490,6 +561,27 @@ func TestPatchCPAValues(t *testing.T) {
 		actualValue = ""
 		require.NoError(t, json.Unmarshal(values[createdField.ID], &actualValue))
 		require.Equal(t, value, actualValue)
+
+		t.Run("a websocket event should be fired as part of the value changes", func(t *testing.T) {
+			var wsValues map[string]json.RawMessage
+			require.Eventually(t, func() bool {
+				select {
+				case event := <-webSocketClient.EventChannel:
+					if event.EventType() == model.WebsocketEventCPAValuesUpdated {
+						valuesData, err := json.Marshal(event.GetData()["values"])
+						require.NoError(t, err)
+						require.NoError(t, json.Unmarshal(valuesData, &wsValues))
+						return true
+					}
+				default:
+					return false
+				}
+				return false
+			}, 5*time.Second, 100*time.Millisecond)
+
+			require.NotEmpty(t, wsValues)
+			require.Equal(t, patchedValues, wsValues)
+		})
 	})
 
 	t.Run("any team member should be able to patch their own values", func(t *testing.T) {

server/channels/app/custom_profile_attributes.go

@@ -97,6 +97,10 @@ func (a *App) CreateCPAField(field *model.PropertyField) (*model.PropertyField,
 		}
 	}
 
+	message := model.NewWebSocketEvent(model.WebsocketEventCPAFieldCreated, "", "", "", nil, "")
+	message.Add("field", newField)
+	a.Publish(message)
+
 	return newField, nil
 }
 
@@ -122,6 +126,10 @@ func (a *App) PatchCPAField(fieldID string, patch *model.PropertyFieldPatch) (*m
 		}
 	}
 
+	message := model.NewWebSocketEvent(model.WebsocketEventCPAFieldUpdated, "", "", "", nil, "")
+	message.Add("field", patchedField)
+	a.Publish(message)
+
 	return patchedField, nil
 }
 
@@ -150,6 +158,10 @@ func (a *App) DeleteCPAField(id string) *model.AppError {
 		}
 	}
 
+	message := model.NewWebSocketEvent(model.WebsocketEventCPAFieldDeleted, "", "", "", nil, "")
+	message.Add("field_id", id)
+	a.Publish(message)
+
 	return nil
 }
 
@@ -193,49 +205,54 @@ func (a *App) GetCPAValue(valueID string) (*model.PropertyValue, *model.AppError
 }
 
 func (a *App) PatchCPAValue(userID string, fieldID string, value json.RawMessage) (*model.PropertyValue, *model.AppError) {
+	values, appErr := a.PatchCPAValues(userID, map[string]json.RawMessage{fieldID: value})
+	if appErr != nil {
+		return nil, appErr
+	}
+
+	return values[0], nil
+}
+
+func (a *App) PatchCPAValues(userID string, fieldValueMap map[string]json.RawMessage) ([]*model.PropertyValue, *model.AppError) {
 	groupID, err := a.cpaGroupID()
 	if err != nil {
 		return nil, model.NewAppError("PatchCPAValues", "app.custom_profile_attributes.cpa_group_id.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
 	}
 
-	// make sure field exists in this group
-	existingField, appErr := a.GetCPAField(fieldID)
-	if appErr != nil {
-		return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_field_not_found.app_error", nil, "", http.StatusNotFound).Wrap(appErr)
-	} else if existingField.DeleteAt > 0 {
-		return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_field_not_found.app_error", nil, "", http.StatusNotFound)
-	}
-
-	existingValues, appErr := a.ListCPAValues(userID)
-	if appErr != nil {
-		return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_value_list.app_error", nil, "", http.StatusNotFound).Wrap(err)
-	}
-	var existingValue *model.PropertyValue
-	for key, value := range existingValues {
-		if value.FieldID == fieldID {
-			existingValue = existingValues[key]
-			break
+	valuesToUpdate := []*model.PropertyValue{}
+	for fieldID, value := range fieldValueMap {
+		// make sure field exists in this group
+		existingField, appErr := a.GetCPAField(fieldID)
+		if appErr != nil {
+			return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_field_not_found.app_error", nil, "", http.StatusNotFound).Wrap(appErr)
+		} else if existingField.DeleteAt > 0 {
+			return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_field_not_found.app_error", nil, "", http.StatusNotFound)
 		}
-	}
 
-	if existingValue != nil {
-		existingValue.Value = value
-		_, err = a.ch.srv.propertyService.UpdatePropertyValue(existingValue)
-		if err != nil {
-			return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_value_update.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
-		}
-	} else {
-		propertyValue := &model.PropertyValue{
+		value := &model.PropertyValue{
 			GroupID:    groupID,
 			TargetType: "user",
 			TargetID:   userID,
 			FieldID:    fieldID,
 			Value:      value,
 		}
-		existingValue, err = a.ch.srv.propertyService.CreatePropertyValue(propertyValue)
-		if err != nil {
-			return nil, model.NewAppError("PatchCPAValue", "app.custom_profile_attributes.property_value_creation.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
-		}
+		valuesToUpdate = append(valuesToUpdate, value)
 	}
-	return existingValue, nil
+
+	updatedValues, err := a.Srv().propertyService.UpsertPropertyValues(valuesToUpdate)
+	if err != nil {
+		return nil, model.NewAppError("PatchCPAValues", "app.custom_profile_attributes.property_value_upsert.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
+	}
+
+	updatedFieldValueMap := map[string]json.RawMessage{}
+	for _, value := range updatedValues {
+		updatedFieldValueMap[value.FieldID] = value.Value
+	}
+
+	message := model.NewWebSocketEvent(model.WebsocketEventCPAValuesUpdated, "", "", "", nil, "")
+	message.Add("user_id", userID)
+	message.Add("values", updatedFieldValueMap)
+	a.Publish(message)
+
+	return updatedValues, nil
 }

server/channels/app/properties/property_value.go

@@ -36,6 +36,19 @@ func (ps *PropertyService) UpdatePropertyValues(values []*model.PropertyValue) (
 	return ps.valueStore.Update(values)
 }
 
+func (ps *PropertyService) UpsertPropertyValue(value *model.PropertyValue) (*model.PropertyValue, error) {
+	values, err := ps.UpsertPropertyValues([]*model.PropertyValue{value})
+	if err != nil {
+		return nil, err
+	}
+
+	return values[0], nil
+}
+
+func (ps *PropertyService) UpsertPropertyValues(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
+	return ps.valueStore.Upsert(values)
+}
+
 func (ps *PropertyService) DeletePropertyValue(id string) error {
 	return ps.valueStore.Delete(id)
 }

server/channels/store/retrylayer/retrylayer.go

@@ -9054,11 +9054,11 @@ func (s *RetryLayerPropertyFieldStore) SearchPropertyFields(opts model.PropertyF
 
 }
 
-func (s *RetryLayerPropertyFieldStore) Update(field []*model.PropertyField) ([]*model.PropertyField, error) {
+func (s *RetryLayerPropertyFieldStore) Update(fields []*model.PropertyField) ([]*model.PropertyField, error) {
 
 	tries := 0
 	for {
-		result, err := s.PropertyFieldStore.Update(field)
+		result, err := s.PropertyFieldStore.Update(fields)
 		if err == nil {
 			return result, nil
 		}
@@ -9243,11 +9243,32 @@ func (s *RetryLayerPropertyValueStore) SearchPropertyValues(opts model.PropertyV
 
 }
 
-func (s *RetryLayerPropertyValueStore) Update(field []*model.PropertyValue) ([]*model.PropertyValue, error) {
+func (s *RetryLayerPropertyValueStore) Update(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
 
 	tries := 0
 	for {
-		result, err := s.PropertyValueStore.Update(field)
+		result, err := s.PropertyValueStore.Update(values)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
+func (s *RetryLayerPropertyValueStore) Upsert(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
+
+	tries := 0
+	for {
+		result, err := s.PropertyValueStore.Upsert(values)
 		if err == nil {
 			return result, nil
 		}

server/channels/store/sqlstore/property_value_store.go

@@ -189,6 +189,97 @@ func (s *SqlPropertyValueStore) Update(values []*model.PropertyValue) (_ []*mode
 	return values, nil
 }
 
+func (s *SqlPropertyValueStore) Upsert(values []*model.PropertyValue) (_ []*model.PropertyValue, err error) {
+	if len(values) == 0 {
+		return nil, nil
+	}
+
+	transaction, err := s.GetMaster().Beginx()
+	if err != nil {
+		return nil, errors.Wrap(err, "property_value_upsert_begin_transaction")
+	}
+	defer finalizeTransactionX(transaction, &err)
+
+	updatedValues := make([]*model.PropertyValue, len(values))
+	updateTime := model.GetMillis()
+	for i, value := range values {
+		value.PreSave()
+		value.UpdateAt = updateTime
+
+		if err := value.IsValid(); err != nil {
+			return nil, errors.Wrap(err, "property_value_upsert_isvalid")
+		}
+
+		valueJSON := value.Value
+		if s.IsBinaryParamEnabled() {
+			valueJSON = AppendBinaryFlag(valueJSON)
+		}
+
+		builder := s.getQueryBuilder().
+			Insert("PropertyValues").
+			Columns("ID", "TargetID", "TargetType", "GroupID", "FieldID", "Value", "CreateAt", "UpdateAt", "DeleteAt").
+			Values(value.ID, value.TargetID, value.TargetType, value.GroupID, value.FieldID, valueJSON, value.CreateAt, value.UpdateAt, value.DeleteAt)
+
+		if s.DriverName() == model.DatabaseDriverMysql {
+			builder = builder.SuffixExpr(sq.Expr(
+				"ON DUPLICATE KEY UPDATE Value = ?, UpdateAt = ?, DeleteAt = ?",
+				valueJSON,
+				value.UpdateAt,
+				0,
+			))
+
+			if _, err := transaction.ExecBuilder(builder); err != nil {
+				return nil, errors.Wrap(err, "property_value_upsert_exec")
+			}
+
+			// MySQL doesn't support RETURNING, so we need to fetch
+			// the new field to get its ID in case we hit a DUPLICATED
+			// KEY and the value.ID we have is not the right one
+			gBuilder := s.tableSelectQuery.Where(sq.Eq{
+				"GroupID":  value.GroupID,
+				"TargetID": value.TargetID,
+				"FieldID":  value.FieldID,
+				"DeleteAt": 0,
+			})
+
+			var values []*model.PropertyValue
+			if gErr := transaction.SelectBuilder(&values, gBuilder); gErr != nil {
+				return nil, errors.Wrap(gErr, "property_value_upsert_select")
+			}
+
+			if len(values) != 1 {
+				return nil, errors.New("property_value_upsert_select_length")
+			}
+
+			updatedValues[i] = values[0]
+		} else {
+			builder = builder.SuffixExpr(sq.Expr(
+				"ON CONFLICT (GroupID, TargetID, FieldID) WHERE DeleteAt = 0 DO UPDATE SET Value = ?, UpdateAt = ?, DeleteAt = ? RETURNING *",
+				valueJSON,
+				value.UpdateAt,
+				0,
+			))
+
+			var values []*model.PropertyValue
+			if err := transaction.SelectBuilder(&values, builder); err != nil {
+				return nil, errors.Wrapf(err, "failed to upsert property value with id: %s", value.ID)
+			}
+
+			if len(values) != 1 {
+				return nil, errors.New("property_value_upsert_select_length")
+			}
+
+			updatedValues[i] = values[0]
+		}
+	}
+
+	if err := transaction.Commit(); err != nil {
+		return nil, errors.Wrap(err, "property_value_upsert_commit")
+	}
+
+	return updatedValues, nil
+}
+
 func (s *SqlPropertyValueStore) Delete(id string) error {
 	builder := s.getQueryBuilder().
 		Update("PropertyValues").

server/channels/store/store.go

@@ -1090,7 +1090,7 @@ type PropertyFieldStore interface {
 	Get(id string) (*model.PropertyField, error)
 	GetMany(ids []string) ([]*model.PropertyField, error)
 	SearchPropertyFields(opts model.PropertyFieldSearchOpts) ([]*model.PropertyField, error)
-	Update(field []*model.PropertyField) ([]*model.PropertyField, error)
+	Update(fields []*model.PropertyField) ([]*model.PropertyField, error)
 	Delete(id string) error
 }
 
@@ -1099,7 +1099,8 @@ type PropertyValueStore interface {
 	Get(id string) (*model.PropertyValue, error)
 	GetMany(ids []string) ([]*model.PropertyValue, error)
 	SearchPropertyValues(opts model.PropertyValueSearchOpts) ([]*model.PropertyValue, error)
-	Update(field []*model.PropertyValue) ([]*model.PropertyValue, error)
+	Update(values []*model.PropertyValue) ([]*model.PropertyValue, error)
+	Upsert(values []*model.PropertyValue) ([]*model.PropertyValue, error)
 	Delete(id string) error
 	DeleteForField(id string) error
 }

server/channels/store/storetest/mocks/PropertyFieldStore.go

@@ -152,9 +152,9 @@ func (_m *PropertyFieldStore) SearchPropertyFields(opts model.PropertyFieldSearc
 	return r0, r1
 }
 
-// Update provides a mock function with given fields: field
-func (_m *PropertyFieldStore) Update(field []*model.PropertyField) ([]*model.PropertyField, error) {
-	ret := _m.Called(field)
+// Update provides a mock function with given fields: fields
+func (_m *PropertyFieldStore) Update(fields []*model.PropertyField) ([]*model.PropertyField, error) {
+	ret := _m.Called(fields)
 
 	if len(ret) == 0 {
 		panic("no return value specified for Update")
@@ -163,10 +163,10 @@ func (_m *PropertyFieldStore) Update(field []*model.PropertyField) ([]*model.Pro
 	var r0 []*model.PropertyField
 	var r1 error
 	if rf, ok := ret.Get(0).(func([]*model.PropertyField) ([]*model.PropertyField, error)); ok {
-		return rf(field)
+		return rf(fields)
 	}
 	if rf, ok := ret.Get(0).(func([]*model.PropertyField) []*model.PropertyField); ok {
-		r0 = rf(field)
+		r0 = rf(fields)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.PropertyField)
@@ -174,7 +174,7 @@ func (_m *PropertyFieldStore) Update(field []*model.PropertyField) ([]*model.Pro
 	}
 
 	if rf, ok := ret.Get(1).(func([]*model.PropertyField) error); ok {
-		r1 = rf(field)
+		r1 = rf(fields)
 	} else {
 		r1 = ret.Error(1)
 	}

server/channels/store/storetest/mocks/PropertyValueStore.go

@@ -170,9 +170,9 @@ func (_m *PropertyValueStore) SearchPropertyValues(opts model.PropertyValueSearc
 	return r0, r1
 }
 
-// Update provides a mock function with given fields: field
-func (_m *PropertyValueStore) Update(field []*model.PropertyValue) ([]*model.PropertyValue, error) {
-	ret := _m.Called(field)
+// Update provides a mock function with given fields: values
+func (_m *PropertyValueStore) Update(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
+	ret := _m.Called(values)
 
 	if len(ret) == 0 {
 		panic("no return value specified for Update")
@@ -181,10 +181,10 @@ func (_m *PropertyValueStore) Update(field []*model.PropertyValue) ([]*model.Pro
 	var r0 []*model.PropertyValue
 	var r1 error
 	if rf, ok := ret.Get(0).(func([]*model.PropertyValue) ([]*model.PropertyValue, error)); ok {
-		return rf(field)
+		return rf(values)
 	}
 	if rf, ok := ret.Get(0).(func([]*model.PropertyValue) []*model.PropertyValue); ok {
-		r0 = rf(field)
+		r0 = rf(values)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.PropertyValue)
@@ -192,7 +192,37 @@ func (_m *PropertyValueStore) Update(field []*model.PropertyValue) ([]*model.Pro
 	}
 
 	if rf, ok := ret.Get(1).(func([]*model.PropertyValue) error); ok {
-		r1 = rf(field)
+		r1 = rf(values)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Upsert provides a mock function with given fields: values
+func (_m *PropertyValueStore) Upsert(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
+	ret := _m.Called(values)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Upsert")
+	}
+
+	var r0 []*model.PropertyValue
+	var r1 error
+	if rf, ok := ret.Get(0).(func([]*model.PropertyValue) ([]*model.PropertyValue, error)); ok {
+		return rf(values)
+	}
+	if rf, ok := ret.Get(0).(func([]*model.PropertyValue) []*model.PropertyValue); ok {
+		r0 = rf(values)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.PropertyValue)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func([]*model.PropertyValue) error); ok {
+		r1 = rf(values)
 	} else {
 		r1 = ret.Error(1)
 	}

server/channels/store/storetest/property_value_store.go

@@ -22,6 +22,7 @@ func TestPropertyValueStore(t *testing.T, rctx request.CTX, ss store.Store, s Sq
 	t.Run("GetPropertyValue", func(t *testing.T) { testGetPropertyValue(t, rctx, ss) })
 	t.Run("GetManyPropertyValues", func(t *testing.T) { testGetManyPropertyValues(t, rctx, ss) })
 	t.Run("UpdatePropertyValue", func(t *testing.T) { testUpdatePropertyValue(t, rctx, ss) })
+	t.Run("UpsertPropertyValue", func(t *testing.T) { testUpsertPropertyValue(t, rctx, ss) })
 	t.Run("DeletePropertyValue", func(t *testing.T) { testDeletePropertyValue(t, rctx, ss) })
 	t.Run("SearchPropertyValues", func(t *testing.T) { testSearchPropertyValues(t, rctx, ss) })
 	t.Run("DeleteForField", func(t *testing.T) { testDeleteForField(t, rctx, ss) })
@@ -306,6 +307,170 @@ func testUpdatePropertyValue(t *testing.T, _ request.CTX, ss store.Store) {
 	})
 }
 
+func testUpsertPropertyValue(t *testing.T, _ request.CTX, ss store.Store) {
+	t.Run("should fail if the property value is not valid", func(t *testing.T) {
+		value := &model.PropertyValue{
+			TargetID:   "",
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"test value"`),
+		}
+		updatedValue, err := ss.PropertyValue().Upsert([]*model.PropertyValue{value})
+		require.Zero(t, updatedValue)
+		require.ErrorContains(t, err, "model.property_value.is_valid.app_error")
+
+		value.TargetID = model.NewId()
+		value.GroupID = ""
+		updatedValue, err = ss.PropertyValue().Upsert([]*model.PropertyValue{value})
+		require.Zero(t, updatedValue)
+		require.ErrorContains(t, err, "model.property_value.is_valid.app_error")
+	})
+
+	t.Run("should be able to insert new property values", func(t *testing.T) {
+		value1 := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"value 1"`),
+		}
+
+		value2 := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"value 2"`),
+		}
+
+		values, err := ss.PropertyValue().Upsert([]*model.PropertyValue{value1, value2})
+		require.NoError(t, err)
+		require.Len(t, values, 2)
+		require.NotEmpty(t, values[0].ID)
+		require.NotEmpty(t, values[1].ID)
+		require.NotZero(t, values[0].CreateAt)
+		require.NotZero(t, values[1].CreateAt)
+
+		valuesFromStore, err := ss.PropertyValue().GetMany([]string{values[0].ID, values[1].ID})
+		require.NoError(t, err)
+		require.Len(t, valuesFromStore, 2)
+	})
+
+	t.Run("should be able to update existing property values", func(t *testing.T) {
+		// Create initial value
+		value := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"initial value"`),
+		}
+		_, err := ss.PropertyValue().Create(value)
+		require.NoError(t, err)
+		valueID := value.ID
+
+		time.Sleep(10 * time.Millisecond)
+
+		// Update via upsert
+		value.ID = ""
+		value.Value = json.RawMessage(`"updated value"`)
+		values, err := ss.PropertyValue().Upsert([]*model.PropertyValue{value})
+		require.NoError(t, err)
+		require.Len(t, values, 1)
+		require.Equal(t, valueID, values[0].ID)
+		require.Equal(t, json.RawMessage(`"updated value"`), values[0].Value)
+		require.Greater(t, values[0].UpdateAt, values[0].CreateAt)
+
+		// Verify in database
+		updated, err := ss.PropertyValue().Get(valueID)
+		require.NoError(t, err)
+		require.Equal(t, json.RawMessage(`"updated value"`), updated.Value)
+		require.Greater(t, updated.UpdateAt, updated.CreateAt)
+	})
+
+	t.Run("should handle mixed insert and update operations", func(t *testing.T) {
+		// Create first value
+		existingValue := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"existing value"`),
+		}
+		_, err := ss.PropertyValue().Create(existingValue)
+		require.NoError(t, err)
+
+		// Prepare new value
+		newValue := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"new value"`),
+		}
+
+		// Update existing and insert new via upsert
+		existingValue.Value = json.RawMessage(`"updated existing"`)
+		values, err := ss.PropertyValue().Upsert([]*model.PropertyValue{existingValue, newValue})
+		require.NoError(t, err)
+		require.Len(t, values, 2)
+
+		// Verify both values
+		newValueUpserted, err := ss.PropertyValue().Get(newValue.ID)
+		require.NoError(t, err)
+		require.Equal(t, json.RawMessage(`"new value"`), newValueUpserted.Value)
+		existingValueUpserted, err := ss.PropertyValue().Get(existingValue.ID)
+		require.NoError(t, err)
+		require.Equal(t, json.RawMessage(`"updated existing"`), existingValueUpserted.Value)
+	})
+
+	t.Run("should not perform any operation if one of the fields is invalid", func(t *testing.T) {
+		// Create initial valid value
+		existingValue := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    model.NewId(),
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"existing value"`),
+		}
+		_, err := ss.PropertyValue().Create(existingValue)
+		require.NoError(t, err)
+
+		originalValue := *existingValue
+
+		// Prepare an invalid value
+		invalidValue := &model.PropertyValue{
+			TargetID:   model.NewId(),
+			TargetType: "test_type",
+			GroupID:    "", // Invalid: empty group ID
+			FieldID:    model.NewId(),
+			Value:      json.RawMessage(`"new value"`),
+		}
+
+		// Try to update existing and insert invalid via upsert
+		existingValue.Value = json.RawMessage(`"should not update"`)
+		_, err = ss.PropertyValue().Upsert([]*model.PropertyValue{existingValue, invalidValue})
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "model.property_value.is_valid.app_error")
+
+		// Verify the existing value was not changed
+		retrieved, err := ss.PropertyValue().Get(existingValue.ID)
+		require.NoError(t, err)
+		require.Equal(t, originalValue.Value, retrieved.Value)
+		require.Equal(t, originalValue.UpdateAt, retrieved.UpdateAt)
+
+		// Verify the invalid value was not inserted
+		results, err := ss.PropertyValue().SearchPropertyValues(model.PropertyValueSearchOpts{
+			TargetID: invalidValue.TargetID,
+			Page:     0,
+			PerPage:  10,
+		})
+		require.NoError(t, err)
+		require.Empty(t, results)
+	})
+}
+
 func testDeletePropertyValue(t *testing.T, _ request.CTX, ss store.Store) {
 	t.Run("should fail on nonexisting value", func(t *testing.T) {
 		err := ss.PropertyValue().Delete(model.NewId())

server/channels/store/timerlayer/timerlayer.go

@@ -7185,10 +7185,10 @@ func (s *TimerLayerPropertyFieldStore) SearchPropertyFields(opts model.PropertyF
 	return result, err
 }
 
-func (s *TimerLayerPropertyFieldStore) Update(field []*model.PropertyField) ([]*model.PropertyField, error) {
+func (s *TimerLayerPropertyFieldStore) Update(fields []*model.PropertyField) ([]*model.PropertyField, error) {
 	start := time.Now()
 
-	result, err := s.PropertyFieldStore.Update(field)
+	result, err := s.PropertyFieldStore.Update(fields)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {
@@ -7329,10 +7329,10 @@ func (s *TimerLayerPropertyValueStore) SearchPropertyValues(opts model.PropertyV
 	return result, err
 }
 
-func (s *TimerLayerPropertyValueStore) Update(field []*model.PropertyValue) ([]*model.PropertyValue, error) {
+func (s *TimerLayerPropertyValueStore) Update(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
 	start := time.Now()
 
-	result, err := s.PropertyValueStore.Update(field)
+	result, err := s.PropertyValueStore.Update(values)
 
 	elapsed := float64(time.Since(start)) / float64(time.Second)
 	if s.Root.Metrics != nil {
@@ -7345,6 +7345,22 @@ func (s *TimerLayerPropertyValueStore) Update(field []*model.PropertyValue) ([]*
 	return result, err
 }
 
+func (s *TimerLayerPropertyValueStore) Upsert(values []*model.PropertyValue) ([]*model.PropertyValue, error) {
+	start := time.Now()
+
+	result, err := s.PropertyValueStore.Upsert(values)
+
+	elapsed := float64(time.Since(start)) / float64(time.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PropertyValueStore.Upsert", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerReactionStore) BulkGetForPosts(postIds []string) ([]*model.Reaction, error) {
 	start := time.Now()
 

server/i18n/en.json

@@ -5039,16 +5039,8 @@
     "translation": "Unable to update Custom Profile Attribute field"
   },
   {
-    "id": "app.custom_profile_attributes.property_value_creation.app_error",
-    "translation": "Cannot create property value"
-  },
-  {
-    "id": "app.custom_profile_attributes.property_value_list.app_error",
-    "translation": "Unable to retrieve property values"
-  },
-  {
-    "id": "app.custom_profile_attributes.property_value_update.app_error",
-    "translation": "Cannot update property value"
+    "id": "app.custom_profile_attributes.property_value_upsert.app_error",
+    "translation": "Unable to upsert Custom Profile Attribute fields"
   },
   {
     "id": "app.custom_profile_attributes.search_property_fields.app_error",

server/public/model/websocket_message.go

@@ -94,6 +94,10 @@ const (
 	WebsocketScheduledPostCreated                     WebsocketEventType = "scheduled_post_created"
 	WebsocketScheduledPostUpdated                     WebsocketEventType = "scheduled_post_updated"
 	WebsocketScheduledPostDeleted                     WebsocketEventType = "scheduled_post_deleted"
+	WebsocketEventCPAFieldCreated                     WebsocketEventType = "custom_profile_attributes_field_created"
+	WebsocketEventCPAFieldUpdated                     WebsocketEventType = "custom_profile_attributes_field_updated"
+	WebsocketEventCPAFieldDeleted                     WebsocketEventType = "custom_profile_attributes_field_deleted"
+	WebsocketEventCPAValuesUpdated                    WebsocketEventType = "custom_profile_attributes_values_updated"
 
 	WebSocketMsgTypeResponse = "response"
 	WebSocketMsgTypeEvent    = "event"