From f85a8c61a49fe590ed5c9f1e3a068973b1593265 Mon Sep 17 00:00:00 2001 From: Miguel de la Cruz Date: Thu, 13 Feb 2025 12:21:46 +0100 Subject: [PATCH] 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 Co-authored-by: Miguel de la Cruz --- .../api4/custom_profile_attributes_test.go | 92 ++++++++++ .../channels/app/custom_profile_attributes.go | 79 +++++---- .../channels/app/properties/property_value.go | 13 ++ .../channels/store/retrylayer/retrylayer.go | 29 ++- .../store/sqlstore/property_value_store.go | 91 ++++++++++ server/channels/store/store.go | 5 +- .../storetest/mocks/PropertyFieldStore.go | 12 +- .../storetest/mocks/PropertyValueStore.go | 42 ++++- .../store/storetest/property_value_store.go | 165 ++++++++++++++++++ .../channels/store/timerlayer/timerlayer.go | 24 ++- server/i18n/en.json | 12 +- server/public/model/websocket_message.go | 4 + 12 files changed, 505 insertions(+), 63 deletions(-) diff --git a/server/channels/api4/custom_profile_attributes_test.go b/server/channels/api4/custom_profile_attributes_test.go index 37e278bdfb..de8daafc5c 100644 --- a/server/channels/api4/custom_profile_attributes_test.go +++ b/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) { diff --git a/server/channels/app/custom_profile_attributes.go b/server/channels/app/custom_profile_attributes.go index a3cd1fafed..caf9ee23fb 100644 --- a/server/channels/app/custom_profile_attributes.go +++ b/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 } diff --git a/server/channels/app/properties/property_value.go b/server/channels/app/properties/property_value.go index 7082ace2d7..dbb79afed9 100644 --- a/server/channels/app/properties/property_value.go +++ b/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) } diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index cd629956cc..a6f738d80b 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/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 } diff --git a/server/channels/store/sqlstore/property_value_store.go b/server/channels/store/sqlstore/property_value_store.go index 1e39cb0c79..d9d12603a3 100644 --- a/server/channels/store/sqlstore/property_value_store.go +++ b/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"). diff --git a/server/channels/store/store.go b/server/channels/store/store.go index 4683e52aea..88f2d29fe0 100644 --- a/server/channels/store/store.go +++ b/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 } diff --git a/server/channels/store/storetest/mocks/PropertyFieldStore.go b/server/channels/store/storetest/mocks/PropertyFieldStore.go index 4afd0e81ff..b32b267da7 100644 --- a/server/channels/store/storetest/mocks/PropertyFieldStore.go +++ b/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) } diff --git a/server/channels/store/storetest/mocks/PropertyValueStore.go b/server/channels/store/storetest/mocks/PropertyValueStore.go index d5b3827ab8..0218927b43 100644 --- a/server/channels/store/storetest/mocks/PropertyValueStore.go +++ b/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) } diff --git a/server/channels/store/storetest/property_value_store.go b/server/channels/store/storetest/property_value_store.go index 41a4ba3cf0..b360380b62 100644 --- a/server/channels/store/storetest/property_value_store.go +++ b/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()) diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index 4d4a8b6580..0867090d74 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/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() diff --git a/server/i18n/en.json b/server/i18n/en.json index ae5b7c51ed..ec9892aa76 100644 --- a/server/i18n/en.json +++ b/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", diff --git a/server/public/model/websocket_message.go b/server/public/model/websocket_message.go index 5736320be7..fd1ca2dda9 100644 --- a/server/public/model/websocket_message.go +++ b/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"