Force 3 station replications

db/db.go

@@ -1328,6 +1328,34 @@ func InsertNewStation(
 	return newStation, rowsAffected, nil
 }
 
+func GetAllStationsWithNoHA3() ([]models.Station, error) {
+	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
+	defer cancelfunc()
+	conn, err := MetadataDbClient.Client.Acquire(ctx)
+	if err != nil {
+		return []models.Station{}, err
+	}
+	defer conn.Release()
+	query := `SELECT * FROM stations WHERE replicas = 1 OR replicas = 5`
+	stmt, err := conn.Conn().Prepare(ctx, "get_all_stations_with_no_ha_3", query)
+	if err != nil {
+		return []models.Station{}, err
+	}
+	rows, err := conn.Conn().Query(ctx, stmt.Name)
+	if err != nil {
+		return []models.Station{}, err
+	}
+	defer rows.Close()
+	stations, err := pgx.CollectRows(rows, pgx.RowToStructByPos[models.Station])
+	if err != nil {
+		return []models.Station{}, err
+	}
+	if len(stations) == 0 {
+		return []models.Station{}, err
+	}
+	return stations, nil
+}
+
 func GetAllStationsDetailsPerTenant(tenantName string) ([]models.ExtendedStation, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
@@ -1862,6 +1890,26 @@ func UpdateStationsOfDeletedUser(userId int, tenantName string) error {
 	return nil
 }
 
+func UpdateStationsWithNoHA3() error {
+	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
+	defer cancelfunc()
+	conn, err := MetadataDbClient.Client.Acquire(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+	query := `UPDATE stations SET replicas = 3 WHERE replicas = 1 OR replicas = 5`
+	stmt, err := conn.Conn().Prepare(ctx, "update_stations_with_no_ha_3", query)
+	if err != nil {
+		return err
+	}
+	_, err = conn.Conn().Query(ctx, stmt.Name)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func RemoveStationsByTenant(tenantName string) error {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()

main.go

@@ -123,6 +123,11 @@ func runMemphis(s *server.Server) {
 		s.Errorf("Failed initializing integrations: " + err.Error())
 	}
 
+	err = s.Force3ReplicationsPerStation()
+	if err != nil {
+		s.Errorf("Failed forece 3 replications per station: " + err.Error())
+	}
+
 	go func() {
 		s.CreateInternalJetStreamResources()
 		go http_server.InitializeHttpServer(s)

server/memphis_cloud.go

@@ -1860,3 +1860,22 @@ func validateReplicas(replicas int) error {
 
 	return nil
 }
+
+func (s *Server) Force3ReplicationsPerStation() error {
+	return nil
+}
+
+func getStationReplicas(replicas int) int {
+	if replicas <= 0 {
+		return 1
+	} else if replicas == 2 || replicas == 4 {
+		return 3
+	} else if replicas > 5 {
+		return 5
+	}
+	return replicas
+}
+
+func getDefaultReplicas() int {
+	return 1
+}

server/memphis_handlers.go

@@ -48,7 +48,7 @@ type srvMemphis struct {
 	activateSysLogsPubFunc func()
 	fallbackLogQ           *ipQueue[fallbackLog]
 	// jsApiMu                sync.Mutex
-	ws                     memphisWS
+	ws memphisWS
 }
 
 type memphisWS struct {
@@ -76,15 +76,16 @@ func getUserDetailsFromMiddleware(c *gin.Context) (models.User, error) {
 
 func CreateDefaultStation(tenantName string, s *Server, sn StationName, userId int, username string) (models.Station, bool, error) {
 	stationName := sn.Ext()
-	err := s.CreateStream(tenantName, sn, "message_age_sec", 604800, "file", 120000, 1, false)
+	replicas := getDefaultReplicas()
+	err := s.CreateStream(tenantName, sn, "message_age_sec", 604800, "file", 120000, replicas, false)
 	if err != nil {
 		return models.Station{}, false, err
 	}
 
 	schemaName := ""
 	schemaVersionNumber := 0
 
-	newStation, rowsUpdated, err := db.InsertNewStation(stationName, userId, username, "message_age_sec", 604800, "file", 1, schemaName, schemaVersionNumber, 120000, true, models.DlsConfiguration{Poison: true, Schemaverse: true}, false, tenantName)
+	newStation, rowsUpdated, err := db.InsertNewStation(stationName, userId, username, "message_age_sec", 604800, "file", replicas, schemaName, schemaVersionNumber, 120000, true, models.DlsConfiguration{Poison: true, Schemaverse: true}, false, tenantName)
 	if err != nil {
 		return models.Station{}, false, err
 	}

server/memphis_handlers_consumers.go

@@ -168,7 +168,6 @@ func (s *Server) createConsumerDirectCommon(c *client, consumerName, cStationNam
 			shouldSendAnalytics, _ := shouldSendAnalytics()
 			if shouldSendAnalytics {
 				analyticsParams := map[string]interface{}{"station-name": stationName.Ext(), "storage-type": "disk"}
-
 				analytics.SendEvent(user.TenantName, user.Username, analyticsParams, "user-create-station-sdk")
 			}
 		}

server/memphis_handlers_stations.go

@@ -99,17 +99,6 @@ func validateStorageType(storageType string) error {
 	return nil
 }
 
-func getStationReplicas(replicas int) int {
-	if replicas <= 0 {
-		return 1
-	} else if replicas == 2 || replicas == 4 {
-		return 3
-	} else if replicas > 5 {
-		return 5
-	}
-	return replicas
-}
-
 func validateIdempotencyWindow(retentionType string, retentionValue int, idempotencyWindow int64) error {
 	if idempotencyWindow > 86400000 { // 24 hours
 		return errors.New("idempotency window can not exceed 24 hours")
@@ -714,7 +703,6 @@ func (sh StationsHandler) GetAllStations(c *gin.Context) {
 }
 
 func (sh StationsHandler) CreateStation(c *gin.Context) {
-
 	var body models.CreateStationSchema
 	ok := utils.Validate(c, &body, false, nil)
 	if !ok {

server/memphis_handlers_user_mgmt.go

@@ -373,8 +373,8 @@ func (umh UserMgmtHandler) AddUserSignUp(c *gin.Context) {
 
 	newUser, err := db.CreateUser(username, "management", hashedPwdString, fullName, subscription, 1, MEMPHIS_GLOBAL_ACCOUNT, false, "", "", "", "")
 	if err != nil {
-		if strings.Contains(err.Error(), "already exist") {
-			c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "User already exist"})
+		if strings.Contains(err.Error(), "already exists") {
+			c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "User already exists"})
 			return
 		}
 		serv.Errorf("[tenant: %v][user: %v]CreateUserSignUp error at db.CreateUser: %v", user.TenantName, user.Username, err.Error())

ui_src/src/components/createStationForm/index.js

@@ -84,7 +84,7 @@ const CreateStationForm = ({ createStationFormRef, getStartedStateRef, finishUpd
     const history = useHistory();
     const [creationForm] = Form.useForm();
     const [allowEdit, setAllowEdit] = useState(true);
-    const [actualPods, setActualPods] = useState(['No HA (1)']);
+    const [actualPods, setActualPods] = useState(isCloud() ? [' HA (3)'] : ['No HA (1)']);
     const [retentionType, setRetentionType] = useState(retanionOptions[0].value);
     const [idempotencyType, setIdempotencyType] = useState(idempotencyOptions[2]);
     const [schemas, setSchemas] = useState([]);
@@ -164,14 +164,17 @@ const CreateStationForm = ({ createStationFormRef, getStartedStateRef, finishUpd
                     replicas = ['No HA (1)'];
                     break;
                 case 3:
-                    replicas = ['No HA (1)', 'HA (3)'];
+                    if (isCloud()) {
+                        replicas = ['HA (3)'];
+                    } else {
+                        replicas = ['No HA (1)', 'HA (3)'];
+                    }
                     break;
                 case 5:
                     replicas = ['No HA (1)', 'HA (3)', 'Super HA (5)'];
                     break;
                 default:
                     replicas = ['No HA (1)'];
-
                     break;
             }
             setActualPods(replicas);
@@ -286,31 +289,32 @@ const CreateStationForm = ({ createStationFormRef, getStartedStateRef, finishUpd
                         </div>
                     )}
                 </div>
-                <div className="replicas-container">
-                    <TitleComponent
-                        headerTitle="Replicas"
-                        typeTitle="sub-header"
-                        headerDescription="Amount of mirrors per message."
-                        learnMore={true}
-                        link="https://docs.memphis.dev/memphis/memphis/concepts/station#replicas-mirroring"
-                    />
-                    <div>
-                        <Form.Item name="replicas" initialValue={getStartedStateRef?.formFieldsCreateStation?.replicas || actualPods[0]} style={{ height: '50px' }}>
-                            <SelectComponent
-                                colorType="black"
-                                backgroundColorType="none"
-                                borderColorType="gray"
-                                radiusType="semi-round"
-                                height="40px"
-                                popupClassName="select-options"
-                                options={actualPods}
-                                value={getStartedStateRef?.formFieldsCreateStation?.replicas || actualPods[0]}
-                                onChange={(e) => getStarted && updateFormState('replicas', e)}
-                                disabled={!allowEdit}
-                            />
-                        </Form.Item>
-                    </div>
-                </div>
+                {isCloud() ? null :
+                    <div className="replicas-container">
+                        <TitleComponent
+                            headerTitle="Replicas"
+                            typeTitle="sub-header"
+                            headerDescription="Amount of mirrors per message."
+                            learnMore={true}
+                            link="https://docs.memphis.dev/memphis/memphis/concepts/station#replicas-mirroring"
+                        />
+                        <div>
+                            <Form.Item name="replicas" initialValue={getStartedStateRef?.formFieldsCreateStation?.replicas || actualPods[0]} style={{ height: '50px' }}>
+                                <SelectComponent
+                                    colorType="black"
+                                    backgroundColorType="none"
+                                    borderColorType="gray"
+                                    radiusType="semi-round"
+                                    height="40px"
+                                    popupClassName="select-options"
+                                    options={actualPods}
+                                    value={getStartedStateRef?.formFieldsCreateStation?.replicas || actualPods[0]}
+                                    onChange={(e) => getStarted && updateFormState('replicas', e)}
+                                    disabled={!allowEdit}
+                                />
+                            </Form.Item>
+                        </div>
+                    </div>}
                 <div className="idempotency-type">
                     <Form.Item name="idempotency">
                         <div>

ui_src/src/domain/overview/getStarted/index.js

@@ -32,7 +32,7 @@ import CreateStation from './createStation';
 import Reducer from './hooks/reducer';
 import SideStep from './sideStep';
 import Finish from './finish';
-import { capitalizeFirst } from '../../../services/valueConvertor';
+import { capitalizeFirst, isCloud } from '../../../services/valueConvertor';
 
 const steps = [{ stepName: 'Create Station' }, { stepName: 'Create App user' }, { stepName: 'Produce data' }, { stepName: 'Consume data' }, { stepName: 'Finish' }];
 
@@ -73,7 +73,7 @@ const initialState = {
         retention_type: 'message_age_sec',
         retention_value: 604800,
         storage_type: 'file',
-        replicas: 'No HA (1)',
+        replicas: isCloud() ? 'HA (3)' : 'No HA (1)',
         days: 7,
         hours: 0,
         minutes: 0,

ui_src/src/services/valueConvertor.js

@@ -337,6 +337,9 @@ export const tieredStorageTimeValidator = (value) => {
 };
 
 export const replicasConvertor = (value, stringToNumber) => {
+    if (isCloud()) {
+        value = 3
+    }
     if (stringToNumber) {
         switch (value) {
             case 'No HA (1)':