diff --git a/README.md b/README.md index c22ac9646..6611fb444 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,7 @@ +

+Please pay attention that Memphis.dev is no longer supported officially by the Superstream team (formerly Memphis.dev) and was released to the public. +

+ [![Github (4)](https://github.com/memphisdev/memphis-terraform/assets/107035359/a5fe5d0f-22e1-4445-957d-5ce4464e61b1)](https://memphis.dev)

Discord @@ -8,10 +12,6 @@

-

- Cloud - Academy - Docs - X - YouTube -

 -

@@ -19,8 +19,6 @@ **[Memphis.dev](https://memphis.dev)** Is The First Data Streaming Platform Designed For Backend Developers
To Build Event-driven And Real-time Features Faster Than Ever.
-cloud_native 2 (5) -

@@ -48,7 +46,7 @@ docker compose -f docker-compose.yml -p memphis up -## ✨ Key Features [v1.4.2](https://docs.memphis.dev/memphis/release-notes/releases/v1.4.2-latest) +## ✨ Key Features [v1.4.4](https://docs.memphis.dev/memphis/release-notes/releases/v1.4.4-latest) ![20](https://user-images.githubusercontent.com/70286779/220196529-abb958d2-5c58-4c33-b5e0-40f5446515ad.png) Production-ready message broker in under 3 minutes
diff --git a/analytics/analytics.go b/analytics/analytics.go index 97fa327b9..f57a0f08d 100644 --- a/analytics/analytics.go +++ b/analytics/analytics.go @@ -11,26 +11,7 @@ // A "Service" is a commercial offering, product, hosted, or managed service, that allows third parties (other than your own employees and contractors acting on your behalf) to access and/or use the Licensed Work or a substantial set of the features or functionality of the Licensed Work to third parties as a software-as-a-service, platform-as-a-service, infrastructure-as-a-service or other similar services that compete with Licensor products or services. package analytics -import ( - "encoding/json" - "fmt" - "strconv" - "strings" - "time" - - "github.com/memphisdev/memphis/conf" - "github.com/memphisdev/memphis/db" - - "github.com/gofrs/uuid" - "github.com/memphisdev/memphis.go" -) - -const ( - ACCOUNT_ID = 223671990 - USERNAME = "traces_producer" - PASSWORD = "usersTracesMemphis@1" - HOST = "aws-eu-central-1.cloud.memphis.dev" -) +import () type EventParam struct { Name string `json:"name"` @@ -44,130 +25,12 @@ type EventBody struct { TimeStamp string `json:"timestamp"` } -var configuration = conf.GetConfig() -var deploymentId string -var memphisVersion string -var memphisConnection *memphis.Conn - func InitializeAnalytics(memphisV, customDeploymentId string) error { - acc := conf.MemphisGlobalAccountName - if !conf.GetConfig().USER_PASS_BASED_AUTH { - acc = conf.GlobalAccount - } - - memphisVersion = memphisV - if customDeploymentId != "" { - deploymentId = customDeploymentId - } else { - exist, deployment, err := db.GetSystemKey("deployment_id", acc) - if err != nil { - return err - } else if !exist { - uid, err := uuid.NewV4() - if err != nil { - return err - } - deploymentId = uid.String() - err = db.InsertSystemKey("deployment_id", deploymentId, acc) - if err != nil { - return err - } - } else { - deploymentId = deployment.Value - } - } - - exist, _, err := db.GetSystemKey("analytics", acc) - if err != nil { - return err - } else if !exist { - value := "" - if configuration.ANALYTICS == "true" { - value = "true" - } else { - value = "false" - } - - err = db.InsertSystemKey("analytics", value, acc) - if err != nil { - return err - } - } - - timeout := 0 * time.Minute - if configuration.PROVIDER == "aws" && configuration.REGION == "eu-central-1" { - timeout = 1 * time.Minute - } - time.AfterFunc(timeout, func() { - memphisConnection, err = memphis.Connect(HOST, USERNAME, memphis.Password(PASSWORD), memphis.AccountId(ACCOUNT_ID), memphis.MaxReconnect(500), memphis.ReconnectInterval(1*time.Second)) - if err != nil { - fmt.Printf("InitializeAnalytics: initalize connection failed %s \n", err.Error()) - } else { - memphisConnection.CreateStation("users-traces", memphis.Replicas(3), memphis.TieredStorageEnabled(true), memphis.RetentionTypeOpt(memphis.MaxMessageAgeSeconds), memphis.RetentionVal(14400)) - } - }) - return nil } func Close() { - acc := conf.MemphisGlobalAccountName - if !conf.GetConfig().USER_PASS_BASED_AUTH { - acc = conf.GlobalAccount - } - _, analytics, _ := db.GetSystemKey("analytics", acc) - if analytics.Value == "true" { - memphisConnection.Close() - } } func SendEvent(tenantName, username string, params map[string]interface{}, eventName string) { - distinctId := deploymentId - if configuration.DEV_ENV != "" { - distinctId = "dev" - } - - if eventName != "error" { - tenantName = strings.ReplaceAll(tenantName, "-", "_") // for parsing purposes - if tenantName != "" && username != "" { - distinctId = distinctId + "-" + tenantName + "-" + username - } - } - - var eventMsg []byte - var event *EventBody - var err error - - creationTime := time.Now().Unix() - timestamp := strconv.FormatInt(creationTime, 10) - params["memphis_version"] = memphisVersion - if eventName == "error" { - event = &EventBody{ - DistinctId: distinctId, - Event: "error", - Properties: params, - TimeStamp: timestamp, - } - } else { - event = &EventBody{ - DistinctId: distinctId, - Event: eventName, - Properties: params, - TimeStamp: timestamp, - } - } - - eventMsg, err = json.Marshal(event) - if err != nil { - return - } - if memphisConnection != nil { - err := memphisConnection.Produce("users-traces", "producer_users_traces", eventMsg, []memphis.ProducerOpt{}, []memphis.ProduceOpt{}) - if err != nil { // retry - memphisConnection, err = memphis.Connect(HOST, USERNAME, memphis.Password(PASSWORD), memphis.AccountId(ACCOUNT_ID), memphis.MaxReconnect(500), memphis.ReconnectInterval(1*time.Second)) - if err == nil { - memphisConnection.Produce("users-traces", "producer_users_traces", eventMsg, []memphis.ProducerOpt{}, []memphis.ProduceOpt{}) - } - } - } } diff --git a/db/db.go b/db/db.go index bb2b0370d..21706b33e 100644 --- a/db/db.go +++ b/db/db.go @@ -3013,7 +3013,7 @@ func GetActiveConsumerByCG(consumersGroup string, stationId int) (bool, models.C } defer conn.Release() - query := `SELECT * FROM consumers WHERE consumers_group = $1 AND station_id = $2 AND type = 'application' LIMIT 1` + query := `SELECT * FROM consumers WHERE consumers_group = $1 AND station_id = $2 AND type = 'application' AND is_active = true LIMIT 1` stmt, err := conn.Conn().Prepare(ctx, "get_active_consumer_by_cg", query) if err != nil { return false, models.Consumer{}, err @@ -5901,7 +5901,7 @@ func GetMsgByStationIdAndMsgSeq(stationId, messageSeq, partitionNumber int) (boo return true, message[0], nil } -func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName string, poisonedCgs []string, messageDetails models.MessagePayload, tenantName string, partitionNumber int) (int, bool, error) { +func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName string, poisonedCgs []string, messageDetails models.MessagePayload, tenantName string, partitionNumber int, validationError string) (int, bool, error) { ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second) defer cancelfunc() updated := false @@ -5979,7 +5979,7 @@ func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName strin if tenantName != conf.GlobalAccount { tenantName = strings.ToLower(tenantName) } - rows, err := tx.Query(ctx, stmt.Name, stationId, messageSeq, producerName, poisonedCgs, messageDetails, updatedAt, "poison", "", tenantName, partitionNumber) + rows, err := tx.Query(ctx, stmt.Name, stationId, messageSeq, producerName, poisonedCgs, messageDetails, updatedAt, "poison", validationError, tenantName, partitionNumber) if err != nil { return 0, updated, err } @@ -6008,7 +6008,7 @@ func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName strin } } } else { // then update - query = `UPDATE dls_messages SET poisoned_cgs = ARRAY_APPEND(poisoned_cgs, $1), updated_at = $4 WHERE station_id=$2 AND message_seq=$3 AND not($1 = ANY(poisoned_cgs)) AND tenant_name=$5 RETURNING id` + query = `UPDATE dls_messages SET poisoned_cgs = ARRAY_APPEND(poisoned_cgs, $1), updated_at = $4, validation_error = $6 WHERE station_id=$2 AND message_seq=$3 AND not($1 = ANY(poisoned_cgs)) AND tenant_name=$5 RETURNING id` stmt, err := tx.Prepare(ctx, "update_poisoned_cgs", query) if err != nil { return 0, updated, err @@ -6017,7 +6017,7 @@ func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName strin if tenantName != conf.GlobalAccount { tenantName = strings.ToLower(tenantName) } - rows, err = tx.Query(ctx, stmt.Name, poisonedCgs[0], stationId, messageSeq, updatedAt, tenantName) + rows, err = tx.Query(ctx, stmt.Name, poisonedCgs[0], stationId, messageSeq, updatedAt, tenantName, validationError) if err != nil { return 0, updated, err } @@ -6756,6 +6756,12 @@ func CreateTenant(name, firebaseOrganizationId, encryptrdInternalWSPass, organiz } defer conn.Release() + var colors any + colors = brandData.Colors + if brandData.Colors == nil || len(brandData.Colors) == 0 { + colors = []string{} + } + query := `INSERT INTO tenants (id, name, firebase_organization_id, internal_ws_pass, organization_name, dark_icon, light_icon, brand_colors) VALUES(nextval('tenants_seq'),$1, $2, $3, $4, $5, $6, $7)` @@ -6766,7 +6772,7 @@ func CreateTenant(name, firebaseOrganizationId, encryptrdInternalWSPass, organiz } var tenantId int - rows, err := conn.Conn().Query(ctx, stmt.Name, name, firebaseOrganizationId, encryptrdInternalWSPass, organizationName, brandData.DarkIcon, brandData.LightIcon, brandData.Colors) + rows, err := conn.Conn().Query(ctx, stmt.Name, name, firebaseOrganizationId, encryptrdInternalWSPass, organizationName, brandData.DarkIcon, brandData.LightIcon, colors) if err != nil { return models.Tenant{}, err } @@ -8407,3 +8413,49 @@ func RemoveRoleAndPermissions(roleID []int, tenantName string) error { return nil } + +func RemovePermissionsByTenant(tenantName string) error { + ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second) + defer cancelfunc() + tenantName = strings.ToLower(tenantName) + conn, err := MetadataDbClient.Client.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + + query := `DELETE FROM permissions WHERE tenant_name = $1` + stmt, err := conn.Conn().Prepare(ctx, "remove_permissions_by_tenant", query) + if err != nil { + return err + } + _, err = conn.Conn().Query(ctx, stmt.Name, tenantName) + if err != nil { + return err + } + + return nil +} + +func RemoveRolesByTenant(tenantName string) error { + ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second) + defer cancelfunc() + tenantName = strings.ToLower(tenantName) + conn, err := MetadataDbClient.Client.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + + query := `DELETE FROM roles WHERE tenant_name = $1` + stmt, err := conn.Conn().Prepare(ctx, "remove_roles_by_tenant", query) + if err != nil { + return err + } + _, err = conn.Conn().Query(ctx, stmt.Name, tenantName) + if err != nil { + return err + } + + return nil +} diff --git a/models/dead_letter_station.go b/models/dead_letter_station.go index 782c3cc46..89acf21a5 100644 --- a/models/dead_letter_station.go +++ b/models/dead_letter_station.go @@ -71,6 +71,14 @@ type SchemaVerseDlsMessageSdk struct { PartitionNumber int `json:"partition_number"` } +type NackedDlsMessageSdk struct { + StationName string `json:"station_name"` + Error string `json:"error"` + CgName string `json:"cg_name"` + Seq uint64 `json:"seq"` + Partition int `json:"partition"` +} + type FunctionsDlsMessage struct { StationID int `json:"station_id"` TenantName string `json:"tenant_name"` diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 000000000..dc31c684e --- /dev/null +++ b/package-lock.json @@ -0,0 +1,6 @@ +{ + "name": "memphis", + "lockfileVersion": 3, + "requires": true, + "packages": {} +} diff --git a/package.json b/package.json new file mode 100644 index 000000000..0967ef424 --- /dev/null +++ b/package.json @@ -0,0 +1 @@ +{} diff --git a/server/background_tasks.go b/server/background_tasks.go index 1fb308264..a8a8c1e51 100644 --- a/server/background_tasks.go +++ b/server/background_tasks.go @@ -37,6 +37,9 @@ const NOTIFICATION_EVENTS_SUBJ = "$memphis_notifications" const PM_RESEND_ACK_SUBJ = "$memphis_pm_acks" const TIERED_STORAGE_CONSUMER = "$memphis_tiered_storage_consumer" const DLS_UNACKED_CONSUMER = "$memphis_dls_unacked_consumer" +const NACKED_DLS_SUBJ = "$memphis_nacked_dls" +const NACKED_DLS_INNER_SUBJ = "$memphis_nacked_inner_dls" +const NACKED_DLS_CONSUMER = "$memphis_nacked_dls_consumer" const SCHEMAVERSE_DLS_SUBJ = "$memphis_schemaverse_dls" const SCHEMAVERSE_DLS_INNER_SUBJ = "$memphis_schemaverse_inner_dls" const SCHEMAVERSE_DLS_CONSUMER = "$memphis_schemaverse_dls_consumer" @@ -211,6 +214,19 @@ func (s *Server) ListenForSchemaverseDlsEvents() error { return nil } +func (s *Server) ListenForNackedDlsEvents() error { + err := s.queueSubscribe(s.MemphisGlobalAccountString(), NACKED_DLS_SUBJ, NACKED_DLS_SUBJ+"_group", func(_ *client, subject, reply string, msg []byte) { + go func(msg []byte) { + s.sendInternalAccountMsg(s.MemphisGlobalAccount(), NACKED_DLS_INNER_SUBJ, msg) + }(copyBytes(msg)) + }) + if err != nil { + return err + } + + return nil +} + func (s *Server) ListenForPoisonMsgAcks() error { err := s.queueSubscribe(s.MemphisGlobalAccountString(), PM_RESEND_ACK_SUBJ, PM_RESEND_ACK_SUBJ+"_group", func(_ *client, subject, reply string, msg []byte) { go func(msg []byte) { @@ -311,6 +327,11 @@ func (s *Server) StartBackgroundTasks() error { return errors.New("Failed to subscribing for schemaverse dls" + err.Error()) } + err = s.ListenForNackedDlsEvents() + if err != nil { + return errors.New("Failed to subscribing for nacked dls" + err.Error()) + } + err = s.ListenForPoisonMsgAcks() if err != nil { return errors.New("Failed subscribing for poison message acks: " + err.Error()) @@ -337,6 +358,7 @@ func (s *Server) StartBackgroundTasks() error { } go s.ConsumeSchemaverseDlsMessages() + go s.ConsumeNackedDlsMessages() go s.ConsumeUnackedMsgs() go s.ConsumeFunctionsDlsMessages() go s.ConsumeTieredStorageMsgs() @@ -608,6 +630,77 @@ func (s *Server) ConsumeSchemaverseDlsMessages() { } } +func (s *Server) ConsumeNackedDlsMessages() { + type nackedDlsMsg struct { + Msg []byte + ReplySubject string + } + amount := 1000 + req := []byte(strconv.FormatUint(uint64(amount), 10)) + for { + if DLS_NACKED_CONSUMER_CREATED && DLS_NACKED_STREAM_CREATED { + resp := make(chan nackedDlsMsg) + replySubj := NACKED_DLS_CONSUMER + "_reply_" + s.memphis.nuid.Next() + + // subscribe to schemavers dls messages + sub, err := s.subscribeOnAcc(s.MemphisGlobalAccount(), replySubj, replySubj+"_sid", func(_ *client, subject, reply string, msg []byte) { + go func(subject, reply string, msg []byte) { + // Ignore 409 Exceeded MaxWaiting cases + if reply != _EMPTY_ { + message := nackedDlsMsg{ + Msg: msg, + ReplySubject: reply, + } + resp <- message + } + }(subject, reply, copyBytes(msg)) + }) + if err != nil { + s.Errorf("Failed to subscribe to nacked dls messages: %v", err.Error()) + continue + } + + // send JS API request to get more messages + subject := fmt.Sprintf(JSApiRequestNextT, dlsNackedStream, NACKED_DLS_CONSUMER) + s.sendInternalAccountMsgWithReply(s.MemphisGlobalAccount(), subject, replySubj, nil, req, true) + + s.Debugf("ConsumeNackedDlsMessages: sending fetch request") + + timeout := time.NewTimer(5 * time.Second) + msgs := make([]nackedDlsMsg, 0) + stop := false + for { + if stop { + s.unsubscribeOnAcc(s.MemphisGlobalAccount(), sub) + break + } + select { + case nackedDlsMsg := <-resp: + msgs = append(msgs, nackedDlsMsg) + if len(msgs) == amount { + stop = true + s.Debugf("ConsumeNackedDlsMessages: finished appending %v messages", len(msgs)) + } + case <-timeout.C: + stop = true + s.Debugf("ConsumeNackedDlsMessages: finished because of timer: %v messages", len(msgs)) + } + } + for _, message := range msgs { + msg := message.Msg + err := s.handleNackedDlsMsg(msg) + if err == nil { + // send ack + s.sendInternalAccountMsgWithEcho(s.MemphisGlobalAccount(), message.ReplySubject, []byte(_EMPTY_)) + } + } + } else { + s.Warnf("ConsumeNackedDlsMessages: waiting for consumer and stream to be created") + time.Sleep(2 * time.Second) + } + } +} + func (s *Server) RemoveOldDlsMsgs() { ticker := time.NewTicker(2 * time.Minute) for range ticker.C { diff --git a/server/const.go b/server/const.go index 377fb5662..33fc179dc 100644 --- a/server/const.go +++ b/server/const.go @@ -56,7 +56,7 @@ const ( DEFAULT_GC_PRODUCER_CONSUMER_RETENTION_HOURS = 2 DEFAULT_ENCRYPTION_SECRET_KEY = "thisis32bitlongpassphraseimusing" // MEMPHIS_VERSION is the current version for the memphis server. - MEMPHIS_VERSION = "1.4.3" + MEMPHIS_VERSION = "1.4.4" // ** added by Memphis // PROTO is the currently supported protocol. diff --git a/server/jetstream.go b/server/jetstream.go index 1ec5dc01b..4a2673fcb 100644 --- a/server/jetstream.go +++ b/server/jetstream.go @@ -400,8 +400,8 @@ func (s *Server) enableJetStream(cfg JetStreamConfig) error { s.SetDefaultSystemAccount() } - // ** changed by Memphis - s.Noticef(" __ __ _ _ __ _ __ ") + + s.Noticef(" __ __ _ _ __ _ __ ") s.Noticef(" | \\/ | | | (_) / / | | \\ \\ ") s.Noticef(" | \\ / | ___ _ __ ___ _ __ | |__ _ ___ | | __| | ___ __ __ | | ") s.Noticef(" | |\\/| | / _ \\ | '_ ` _ \\ | '_ \\ | '_ \\ | | / __| / / / _` | / _ \\ \\ \\ / / \\ \\") @@ -1346,6 +1346,8 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro THROUGHPUT_LEGACY_STREAM_EXIST = true case dlsSchemaverseStream: DLS_SCHEMAVERSE_STREAM_CREATED = true + case dlsNackedStream: + DLS_NACKED_STREAM_CREATED = true case integrationsAuditLogsStream: INTEGRATIONS_AUDIT_LOGS_STREAM_CREATED = true case notificationsStreamName: @@ -1386,6 +1388,8 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro DLS_UNACKED_CONSUMER_CREATED = true case dlsSchemaverseStream: DLS_SCHEMAVERSE_CONSUMER_CREATED = true + case dlsNackedStream: + DLS_NACKED_CONSUMER_CREATED = true case notificationsStreamName: NOTIFICATIONS_BUFFER_CONSUMER_CREATED = true case systemTasksStreamName: diff --git a/server/memphis_handlers_consumers.go b/server/memphis_handlers_consumers.go index 370678925..732fc5eca 100644 --- a/server/memphis_handlers_consumers.go +++ b/server/memphis_handlers_consumers.go @@ -201,22 +201,24 @@ func (s *Server) createConsumerDirectCommon(c *client, consumerName, cStationNam } } - var newConsumer models.Consumer + newConsumer := models.Consumer{ + Name: name, + StationId: station.ID, + ConnectionId: connectionId, + ConsumersGroup: consumerGroup, + MaxAckTimeMs: int64(maxAckTime), + MaxMsgDeliveries: maxMsgDeliveries, + StartConsumeFromSeq: startConsumeFromSequence, + LastMessages: lastMessages, + TenantName: tenantName, + PartitionsList: station.PartitionsList, + } if strings.HasPrefix(user.Username, "$") { - newConsumer, err = db.InsertNewConsumer(name, station.ID, "connector", connectionId, consumerGroup, maxAckTime, maxMsgDeliveries, startConsumeFromSequence, lastMessages, tenantName, station.PartitionsList, requestVersion, sdkName, appId) - if err != nil { - serv.Errorf("[tenant: %v]createConsumerDirectCommon at InsertNewConsumer: Consumer %v at station %v :%v", user.TenantName, consumerName, cStationName, err.Error()) - return []int{}, err - } + newConsumer.Type = "connector" } else { - newConsumer, err = db.InsertNewConsumer(name, station.ID, consumerType, connectionId, consumerGroup, maxAckTime, maxMsgDeliveries, startConsumeFromSequence, lastMessages, tenantName, station.PartitionsList, requestVersion, sdkName, appId) - if err != nil { - serv.Errorf("[tenant: %v]createConsumerDirectCommon at InsertNewConsumer: Consumer %v at station %v :%v", user.TenantName, consumerName, cStationName, err.Error()) - return []int{}, err - } + newConsumer.Type = consumerType } - message := "Consumer " + name + " connected" if consumerGroupExist { if requestVersion == 1 { if newConsumer.StartConsumeFromSeq != consumerFromGroup.StartConsumeFromSeq || newConsumer.LastMessages != consumerFromGroup.LastMessages { @@ -260,7 +262,23 @@ func (s *Server) createConsumerDirectCommon(c *client, consumerName, cStationNam return []int{}, err } } + + if strings.HasPrefix(user.Username, "$") { + newConsumer, err = db.InsertNewConsumer(name, station.ID, "connector", connectionId, consumerGroup, maxAckTime, maxMsgDeliveries, startConsumeFromSequence, lastMessages, tenantName, station.PartitionsList, requestVersion, sdkName, appId) + if err != nil { + serv.Errorf("[tenant: %v]createConsumerDirectCommon at InsertNewConsumer: Consumer %v at station %v :%v", user.TenantName, consumerName, cStationName, err.Error()) + return []int{}, err + } + } else { + newConsumer, err = db.InsertNewConsumer(name, station.ID, consumerType, connectionId, consumerGroup, maxAckTime, maxMsgDeliveries, startConsumeFromSequence, lastMessages, tenantName, station.PartitionsList, requestVersion, sdkName, appId) + if err != nil { + serv.Errorf("[tenant: %v]createConsumerDirectCommon at InsertNewConsumer: Consumer %v at station %v :%v", user.TenantName, consumerName, cStationName, err.Error()) + return []int{}, err + } + } + var auditLogs []interface{} + message := "Consumer " + name + " connected" newAuditLog := models.AuditLog{ StationName: stationName.Ext(), Message: message, diff --git a/server/memphis_handlers_dls_messages.go b/server/memphis_handlers_dls_messages.go index 3e5dd0bba..af72acb63 100644 --- a/server/memphis_handlers_dls_messages.go +++ b/server/memphis_handlers_dls_messages.go @@ -128,7 +128,7 @@ func (s *Server) handleNewUnackedMsg(msg []byte) error { Headers: headersJson, } - dlsMsgId, updated, err := db.StorePoisonMsg(station.ID, int(messageSeq), cgName, producedByHeader, poisonedCgs, messageDetails, station.TenantName, partitionNumber) + dlsMsgId, updated, err := db.StorePoisonMsg(station.ID, int(messageSeq), cgName, producedByHeader, poisonedCgs, messageDetails, station.TenantName, partitionNumber, "") if err != nil { serv.Errorf("[tenant: %v]handleNewUnackedMsg at StorePoisonMsg: Error while getting notified about a poison message: %v", station.TenantName, err.Error()) return err @@ -168,13 +168,14 @@ func (s *Server) handleSchemaverseDlsMsg(msg []byte) error { return err } - exist, station, err := db.GetStationByName(message.StationName, tenantName) + stationName := StationNameFromStreamName(message.StationName) + exist, station, err := db.GetStationByName(stationName.Ext(), tenantName) if err != nil { serv.Errorf("[tenant: %v]handleSchemaverseDlsMsg: %v", tenantName, err.Error()) return err } if !exist { - serv.Warnf("[tenant: %v]handleSchemaverseDlsMsg: station %v couldn't been found", tenantName, message.StationName) + serv.Warnf("[tenant: %v]handleSchemaverseDlsMsg: station %v couldn't been found", tenantName, stationName.Ext()) return nil } @@ -198,6 +199,105 @@ func (s *Server) handleSchemaverseDlsMsg(msg []byte) error { return nil } +func (s *Server) handleNackedDlsMsg(msg []byte) error { + tenantName, stringMessage, err := s.getTenantNameAndMessage(msg) + if err != nil { + s.Errorf("handleNackedDlsMsg at getTenantNameAndMessage: %v", err.Error()) + return err + } + var message models.NackedDlsMessageSdk + err = json.Unmarshal([]byte(stringMessage), &message) + if err != nil { + serv.Errorf("[tenant: %v]handleNackedDlsMsg: %v", tenantName, err.Error()) + return err + } + + if message.Partition == 0 { + serv.Errorf("[tenant: %v]handleNackedDlsMsg - missing partition number: %v", tenantName, err.Error()) + return err + } + + stationName := StationNameFromStreamName(message.StationName) + streamName := stationName.Intern() + "$" + strconv.Itoa(message.Partition) + exist, station, err := db.GetStationByName(stationName.Ext(), tenantName) + if err != nil { + serv.Errorf("[tenant: %v]handleNackedDlsMsg: %v", tenantName, err.Error()) + return err + } + if !exist { + serv.Warnf("[tenant: %v]handleNackedDlsMsg: station %v couldn't been found", tenantName, stationName.Ext()) + return nil + } + if !station.DlsConfigurationPoison { + return nil + } + + poisonMessageContent, err := s.memphisGetMessage(tenantName, streamName, uint64(message.Seq)) + if err != nil { + if IsNatsErr(err, JSNoMessageFoundErr) { + return nil + } + serv.Errorf("[tenant: %v]handleNackedDlsMsg at memphisGetMessage: station: %v, Error while getting notified about a poison message: %v", tenantName, stationName.Ext(), err.Error()) + return err + } + + + timeSentTimeStamp := poisonMessageContent.Time + data := poisonMessageContent.Data + lenPayload := len(poisonMessageContent.Data) + len(poisonMessageContent.Header) + headers := poisonMessageContent.Header + var headersJson map[string]string + if headers != nil { + headersJson, err = DecodeHeader(headers) + if err != nil { + serv.Errorf("handleNackedDlsMsg: %v", err.Error()) + return err + } + } + + producedByHeader := _EMPTY_ + poisonedCgs := []string{} + producedByHeader = headersJson["$memphis_producedBy"] + if producedByHeader == _EMPTY_ { + producedByHeader = "unknown" + } + poisonedCgs = append(poisonedCgs, message.CgName) + + messageDetails := models.MessagePayload{ + TimeSent: timeSentTimeStamp, + Size: lenPayload, + Data: hex.EncodeToString(data), + Headers: headersJson, + } + + dlsMsgId, updated, err := db.StorePoisonMsg(station.ID, int(message.Seq), message.CgName, producedByHeader, poisonedCgs, messageDetails, tenantName, message.Partition, message.Error) + if err != nil { + serv.Errorf("[tenant: %v]handleNackedDlsMsg at StorePoisonMsg: Error while getting notified about a poison message: %v", station.TenantName, err.Error()) + return err + } + if !updated { + err = s.sendToDlsStation(station, data, headersJson, "unacked", _EMPTY_) + if err != nil { + serv.Errorf("[tenant: %v]handleNackedDlsMsg at sendToDlsStation: station: %v, Error while getting notified about a poison message: %v", station.TenantName, station.DlsStation, err.Error()) + return err + } + } + + if dlsMsgId == 0 { // nothing to do + return nil + } + + idForUrl := strconv.Itoa(dlsMsgId) + var msgUrl = s.opts.UiHost + "/stations/" + stationName.Ext() + "/" + idForUrl + err = s.SendNotification(station.TenantName, PoisonMessageTitle, "Poison message has been identified, for more details head to: "+msgUrl, PoisonMAlert) + if err != nil { + serv.Warnf("[tenant: %v]handleNackedDlsMsg at SendNotification: Error while sending a poison message notification: %v", station.TenantName, err.Error()) + return nil + } + + return nil +} + func (pmh PoisonMessagesHandler) GetDlsMsgsByStationLight(station models.Station, partitionNumber int) ([]models.LightDlsMessageResponse, []models.LightDlsMessageResponse, []models.LightDlsMessageResponse, int, error) { poisonMessages := make([]models.LightDlsMessageResponse, 0) schemaMessages := make([]models.LightDlsMessageResponse, 0) diff --git a/server/memphis_handlers_rbac.go b/server/memphis_handlers_rbac.go index 98795beb2..8a1b1f65d 100644 --- a/server/memphis_handlers_rbac.go +++ b/server/memphis_handlers_rbac.go @@ -132,6 +132,7 @@ func GetAllMemphisAndNatsInternalSubjects() []string { var subjects []string // Memphis subjects + subjects = append(subjects, NACKED_DLS_SUBJ) subjects = append(subjects, SCHEMAVERSE_DLS_SUBJ) subjects = append(subjects, sdkClientsUpdatesSubject) subjects = append(subjects, PM_RESEND_ACK_SUBJ) diff --git a/server/memphis_handlers_user_mgmt.go b/server/memphis_handlers_user_mgmt.go index 1ba87ef91..36194423a 100644 --- a/server/memphis_handlers_user_mgmt.go +++ b/server/memphis_handlers_user_mgmt.go @@ -176,6 +176,16 @@ func removeTenantResources(tenantName string, user models.User) error { return err } + err = db.RemovePermissionsByTenant(tenantName) + if err != nil { + return err + } + + err = db.RemoveRolesByTenant(tenantName) + if err != nil { + return err + } + err = db.RemoveStationsByTenant(tenantName) if err != nil { return err diff --git a/server/memphis_helper.go b/server/memphis_helper.go index 2828efae9..0db025923 100644 --- a/server/memphis_helper.go +++ b/server/memphis_helper.go @@ -54,6 +54,7 @@ const ( dlsStreamName = "$memphis-%s-dls" dlsUnackedStream = "$memphis_dls_unacked" dlsSchemaverseStream = "$memphis_dls_schemaverse" + dlsNackedStream = "$memphis_dls_nacked" dlsFunctionsStream = "$memphis_dls_functions" dlsResendMessagesStreamNew = "$memphis_dls_%v.%v" dlsResendMessagesStreamOld = "$memphis_dls_%v_%v" @@ -95,6 +96,7 @@ var memphisExportString = `[ {service: "$memphis_integration_updates"}, {service: "$memphis_notifications"}, {service: "$memphis_schemaverse_dls"}, + {service: "$memphis_nacked_dls"}, {service: "$memphis_pm_acks"}, {service: "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>"}, {stream: "$memphis_ws_pubs.>"}, @@ -115,6 +117,7 @@ var memphisImportString = `[ {service: {account: "$memphis", subject: "$memphis_integration_updates"}}, {service: {account: "$memphis", subject: "$memphis_notifications"}}, {service: {account: "$memphis", subject: "$memphis_schemaverse_dls"}}, + {service: {account: "$memphis", subject: "$memphis_nacked_dls"}}, {service: {account: "$memphis", subject: "$memphis_pm_acks"}}, {service: {account: "$memphis", subject: "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>"}}, {stream: {account: "$memphis", subject: "$memphis_ws_pubs.>"}}, @@ -147,6 +150,8 @@ var ( DLS_UNACKED_STREAM_CREATED bool DLS_SCHEMAVERSE_STREAM_CREATED bool DLS_SCHEMAVERSE_CONSUMER_CREATED bool + DLS_NACKED_STREAM_CREATED bool + DLS_NACKED_CONSUMER_CREATED bool DLS_FUNCTIONS_STREAM_CREATED bool DLS_FUNCTIONS_CONSUMER_CREATED bool SYSLOGS_STREAM_CREATED bool @@ -447,6 +452,43 @@ func tryCreateInternalJetStreamResources(s *Server, retentionDur time.Duration, DLS_SCHEMAVERSE_CONSUMER_CREATED = true } + // create nacked dls stream + if !DLS_NACKED_STREAM_CREATED { + err = s.memphisAddStream(s.MemphisGlobalAccountString(), &StreamConfig{ + Name: dlsNackedStream, + Subjects: []string{NACKED_DLS_INNER_SUBJ}, + Retention: WorkQueuePolicy, + MaxAge: time.Hour * 24, + MaxConsumers: -1, + Discard: DiscardOld, + Storage: FileStorage, + Replicas: replicas, + }) + if err != nil && !IsNatsErr(err, JSStreamNameExistErr) { + successCh <- err + return + } + DLS_NACKED_STREAM_CREATED = true + } + + // create nacked dls consumer + if !DLS_NACKED_CONSUMER_CREATED { + cc := ConsumerConfig{ + DeliverPolicy: DeliverAll, + AckPolicy: AckExplicit, + Durable: NACKED_DLS_CONSUMER, + AckWait: time.Duration(80) * time.Second, + MaxAckPending: -1, + MaxDeliver: 10, + } + err = serv.memphisAddConsumer(s.MemphisGlobalAccountString(), dlsNackedStream, &cc) + if err != nil { + successCh <- err + return + } + DLS_NACKED_CONSUMER_CREATED = true + } + // create functions dls stream if shouldCreateFunctionDlsStream() && !DLS_FUNCTIONS_STREAM_CREATED { err = s.memphisAddStream(s.MemphisGlobalAccountString(), &StreamConfig{ diff --git a/ui_src/package-lock.json b/ui_src/package-lock.json index aee69dfe0..44fa2cc85 100644 --- a/ui_src/package-lock.json +++ b/ui_src/package-lock.json @@ -110,6 +110,7 @@ "version": "2.2.1", "resolved": "https://registry.npmjs.org/@ampproject/remapping/-/remapping-2.2.1.tgz", "integrity": "sha512-lFMjJTrFL3j7L9yBxwYfCq2k6qqwHyzuUl/XBnif78PWTJYyL/dfowQHWE3sp6U6ZzqWiiIZnpTMO96zhkjwtg==", + "dev": true, "dependencies": { "@jridgewell/gen-mapping": "^0.3.0", "@jridgewell/trace-mapping": "^0.3.9" @@ -1525,6 +1526,7 @@ "version": "7.23.2", "resolved": "https://registry.npmjs.org/@babel/compat-data/-/compat-data-7.23.2.tgz", "integrity": "sha512-0S9TQMmDHlqAZ2ITT95irXKfxN9bncq8ZCoJhun3nHL/lLUxd2NKBJYoNGWH7S0hz6fRQwWlAWn/ILM0C70KZQ==", + "dev": true, "engines": { "node": ">=6.9.0" } @@ -1533,6 +1535,7 @@ "version": "7.23.2", "resolved": "https://registry.npmjs.org/@babel/core/-/core-7.23.2.tgz", "integrity": "sha512-n7s51eWdaWZ3vGT2tD4T7J6eJs3QoBXydv7vkUM06Bf1cbVD2Kc2UrkzhiQwobfV7NwOnQXYL7UBJ5VPU+RGoQ==", + "dev": true, "dependencies": { "@ampproject/remapping": "^2.2.0", "@babel/code-frame": "^7.22.13", @@ -1626,6 +1629,7 @@ "version": "7.22.15", "resolved": "https://registry.npmjs.org/@babel/helper-compilation-targets/-/helper-compilation-targets-7.22.15.tgz", "integrity": "sha512-y6EEzULok0Qvz8yyLkCvVX+02ic+By2UdOhylwUOvOn9dvYc9mKICJuuU1n1XBI02YWsNsnrY1kc6DVbjcXbtw==", + "dev": true, "dependencies": { "@babel/compat-data": "^7.22.9", "@babel/helper-validator-option": "^7.22.15", @@ -1751,6 +1755,7 @@ "version": "7.23.0", "resolved": "https://registry.npmjs.org/@babel/helper-module-transforms/-/helper-module-transforms-7.23.0.tgz", "integrity": "sha512-WhDWw1tdrlT0gMgUJSlX0IQvoO1eN279zrAUbVB+KpV2c3Tylz8+GnKOLllCS6Z/iZQEyVYxhZVUdPTqs2YYPw==", + "dev": true, "dependencies": { "@babel/helper-environment-visitor": "^7.22.20", "@babel/helper-module-imports": "^7.22.15", @@ -1823,6 +1828,7 @@ "version": "7.22.5", "resolved": "https://registry.npmjs.org/@babel/helper-simple-access/-/helper-simple-access-7.22.5.tgz", "integrity": "sha512-n0H99E/K+Bika3++WNL17POvo4rKWZ7lZEp1Q+fStVbUi8nxPQEBOlTmCOxW/0JsS56SKKQ+ojAe2pHKJHN35w==", + "dev": true, "dependencies": { "@babel/types": "^7.22.5" }, @@ -1873,6 +1879,7 @@ "version": "7.22.15", "resolved": "https://registry.npmjs.org/@babel/helper-validator-option/-/helper-validator-option-7.22.15.tgz", "integrity": "sha512-bMn7RmyFjY/mdECUbgn9eoSY4vqvacUnS9i9vGAGttgFWesO6B4CYWA7XlpbWgBt71iv/hfbPlynohStqnu5hA==", + "dev": true, "engines": { "node": ">=6.9.0" } @@ -1895,6 +1902,7 @@ "version": "7.23.2", "resolved": "https://registry.npmjs.org/@babel/helpers/-/helpers-7.23.2.tgz", "integrity": "sha512-lzchcp8SjTSVe/fPmLwtWVBFC7+Tbn8LGHDVfDp9JGxpAY5opSaEFgt8UQvrnECWOTdji2mOWMz1rOhkHscmGQ==", + "dev": true, "dependencies": { "@babel/template": "^7.22.15", "@babel/traverse": "^7.23.2", @@ -9465,6 +9473,7 @@ "version": "4.22.1", "resolved": "https://registry.npmjs.org/browserslist/-/browserslist-4.22.1.tgz", "integrity": "sha512-FEVc202+2iuClEhZhrWy6ZiAcRLvNMyYcxZ8raemul1DYVOVdFsbqckWLdsixQZCpJlwe77Z3UTalE7jsjnKfQ==", + "dev": true, "funding": [ { "type": "opencollective", @@ -9645,6 +9654,7 @@ "version": "1.0.30001549", "resolved": "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001549.tgz", "integrity": "sha512-qRp48dPYSCYaP+KurZLhDYdVE+yEyht/3NlmcJgVQ2VMGt6JL36ndQ/7rgspdZsJuxDPFIo/OzBT2+GmIJ53BA==", + "dev": true, "funding": [ { "type": "opencollective", @@ -10232,7 +10242,8 @@ "node_modules/convert-source-map": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/convert-source-map/-/convert-source-map-2.0.0.tgz", - "integrity": "sha512-Kvp459HrV2FEJ1CAsi1Ku+MY3kasH19TFykTz2xWmMeq6bk2NU3XXvfJ+Q61m0xktWwt+1HSYf3JZsTms3aRJg==" + "integrity": "sha512-Kvp459HrV2FEJ1CAsi1Ku+MY3kasH19TFykTz2xWmMeq6bk2NU3XXvfJ+Q61m0xktWwt+1HSYf3JZsTms3aRJg==", + "dev": true }, "node_modules/cookie": { "version": "0.5.0", @@ -11635,7 +11646,8 @@ "node_modules/electron-to-chromium": { "version": "1.4.556", "resolved": "https://registry.npmjs.org/electron-to-chromium/-/electron-to-chromium-1.4.556.tgz", - "integrity": "sha512-6RPN0hHfzDU8D56E72YkDvnLw5Cj2NMXZGg3UkgyoHxjVhG99KZpsKgBWMmTy0Ei89xwan+rbRsVB9yzATmYzQ==" + "integrity": "sha512-6RPN0hHfzDU8D56E72YkDvnLw5Cj2NMXZGg3UkgyoHxjVhG99KZpsKgBWMmTy0Ei89xwan+rbRsVB9yzATmYzQ==", + "dev": true }, "node_modules/element-resize-event": { "version": "3.0.6", @@ -11975,6 +11987,7 @@ "version": "3.1.1", "resolved": "https://registry.npmjs.org/escalade/-/escalade-3.1.1.tgz", "integrity": "sha512-k0er2gUkLf8O0zKJiAhmkTnJlTvINGv7ygDNPbeIsX/TJjGJZHuh9B2UxbsaEkmlEo9MfhrSzmhIlhRlI2GXnw==", + "dev": true, "engines": { "node": ">=6" } @@ -13802,6 +13815,7 @@ "version": "1.0.0-beta.2", "resolved": "https://registry.npmjs.org/gensync/-/gensync-1.0.0-beta.2.tgz", "integrity": "sha512-3hN7NaskYvMDLQY55gnW3NQ+mesEAepTqlg+VEbj7zzqEMBVNhzcGYYeqFo/TlYz6eQiFcp1HcsCZO+nGgS8zg==", + "dev": true, "engines": { "node": ">=6.9.0" } @@ -19018,6 +19032,7 @@ "version": "2.2.3", "resolved": "https://registry.npmjs.org/json5/-/json5-2.2.3.tgz", "integrity": "sha512-XmOWe7eyHYH14cLdVPoyg+GOH3rYX++KpzrylJwSW98t3Nk+U8XOl8FWKOgwtzdb8lXGf6zYwDUzeHMWfxasyg==", + "dev": true, "bin": { "json5": "lib/cli.js" }, @@ -19730,6 +19745,7 @@ "version": "5.1.1", "resolved": "https://registry.npmjs.org/lru-cache/-/lru-cache-5.1.1.tgz", "integrity": "sha512-KpNARQA3Iwv+jTA0utUVVbrh+Jlrr1Fv0e56GGzAFOXN7dk/FviaDW8LHmK52DlcH4WP2n6gI8vN1aesBFgo9w==", + "dev": true, "dependencies": { "yallist": "^3.0.2" } @@ -21233,7 +21249,8 @@ "node_modules/node-releases": { "version": "2.0.13", "resolved": "https://registry.npmjs.org/node-releases/-/node-releases-2.0.13.tgz", - "integrity": "sha512-uYr7J37ae/ORWdZeQ1xxMJe3NtdmqMC/JZK+geofDrkLUApKRHPd18/TxtBOJ4A0/+uUIliorNrfYV6s1b02eQ==" + "integrity": "sha512-uYr7J37ae/ORWdZeQ1xxMJe3NtdmqMC/JZK+geofDrkLUApKRHPd18/TxtBOJ4A0/+uUIliorNrfYV6s1b02eQ==", + "dev": true }, "node_modules/normalize-path": { "version": "3.0.0", @@ -27518,6 +27535,7 @@ "version": "6.3.1", "resolved": "https://registry.npmjs.org/semver/-/semver-6.3.1.tgz", "integrity": "sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA==", + "dev": true, "bin": { "semver": "bin/semver.js" } @@ -29477,20 +29495,6 @@ "is-typedarray": "^1.0.0" } }, - "node_modules/typescript": { - "version": "4.9.5", - "resolved": "https://registry.npmjs.org/typescript/-/typescript-4.9.5.tgz", - "integrity": "sha512-1FXk9E2Hm+QzZQ7z+McJiHL4NW1F2EzMu9Nq9i3zAaGqibafqYwCVU6WyWAuyQRRzOlxou8xZSyXLEN8oKj24g==", - "dev": true, - "peer": true, - "bin": { - "tsc": "bin/tsc", - "tsserver": "bin/tsserver" - }, - "engines": { - "node": ">=4.2.0" - } - }, "node_modules/typewise": { "version": "1.0.3", "resolved": "https://registry.npmjs.org/typewise/-/typewise-1.0.3.tgz", @@ -29811,6 +29815,7 @@ "version": "1.0.13", "resolved": "https://registry.npmjs.org/update-browserslist-db/-/update-browserslist-db-1.0.13.tgz", "integrity": "sha512-xebP81SNcPuNpPP3uzeW1NYXxI3rxyJzF3pD6sH4jE7o/IX+WtSpwnVU+qIsDPyk0d3hmFQ7mjqc6AtV604hbg==", + "dev": true, "funding": [ { "type": "opencollective", @@ -31067,7 +31072,8 @@ "node_modules/yallist": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/yallist/-/yallist-3.1.1.tgz", - "integrity": "sha512-a4UGQaWPH59mOXUYnAG2ewncQS4i4F43Tv3JoAM+s2VDAmS9NsK8GpDMLrCHPksFT7h3K6TOoUNn2pb7RoXx4g==" + "integrity": "sha512-a4UGQaWPH59mOXUYnAG2ewncQS4i4F43Tv3JoAM+s2VDAmS9NsK8GpDMLrCHPksFT7h3K6TOoUNn2pb7RoXx4g==", + "dev": true }, "node_modules/yaml": { "version": "1.10.2", diff --git a/ui_src/src/App.js b/ui_src/src/App.js index 6a564f62a..b1a346550 100644 --- a/ui_src/src/App.js +++ b/ui_src/src/App.js @@ -15,7 +15,6 @@ import 'App.scss'; import { Switch, Route, withRouter } from 'react-router-dom'; import React, { useCallback, useContext, useEffect, useRef, useState } from 'react'; import { JSONCodec, StringCodec, connect } from 'nats.ws'; -import { useStiggContext } from '@stigg/react-sdk'; import { useMediaQuery } from 'react-responsive'; import { useHistory } from 'react-router-dom'; import { message, notification } from 'antd'; @@ -27,18 +26,16 @@ import { LOCAL_STORAGE_INTERNAL_WS_PASS, LOCAL_STORAGE_CONNECTION_TOKEN, LOCAL_STORAGE_TOKEN, - LOCAL_STORAGE_USER_PASS_BASED_AUTH, - USER_IMAGE, - LOCAL_STORAGE_PLAN + LOCAL_STORAGE_USER_PASS_BASED_AUTH } from 'const/localStorageConsts'; -import { CLOUD_URL, HANDLE_REFRESH_INTERVAL, WS_PREFIX } from 'config'; -import { isCheckoutCompletedTrue, isCloud } from 'services/valueConvertor'; +import { HANDLE_REFRESH_INTERVAL, WS_PREFIX } from 'config'; +import { isCheckoutCompletedTrue } from 'services/valueConvertor'; import { ReactComponent as InfoNotificationIcon } from 'assets/images/infoNotificationIcon.svg'; import { handleRefreshTokenRequest, httpRequest } from 'services/http'; import { ReactComponent as RedirectIcon } from 'assets/images/redirectIcon.svg'; import { ReactComponent as SuccessIcon } from 'assets/images/successIcon.svg'; import { ReactComponent as CloseIcon } from 'assets/images/closeNotification.svg'; -import { showMessages, useGetAllowedActions } from 'services/genericServices'; +import { showMessages } from 'services/genericServices'; import StationOverview from 'domain/stationOverview'; import { ReactComponent as ErrorIcon } from 'assets/images/errorIcon.svg'; import MessageJourney from 'domain/messageJourney'; @@ -52,21 +49,18 @@ import PrivateRoute from 'PrivateRoute'; import AuthService from 'services/auth'; import Overview from 'domain/overview'; import Loader from 'components/loader'; -import Functions from 'domain/functions'; import { Context } from 'hooks/store'; import pathDomains from 'router'; import Users from 'domain/users'; -import { planType } from "const/globalConst"; +import { planType } from 'const/globalConst'; let SysLogs = undefined; let Login = undefined; let Signup = undefined; -if (!isCloud()) { - SysLogs = require('domain/sysLogs').default; - Login = require('domain/login').default; - Signup = require('domain/signup').default; -} +SysLogs = require('domain/sysLogs').default; +Login = require('domain/login').default; +Signup = require('domain/signup').default; const App = withRouter(() => { const [state, dispatch] = useContext(Context); @@ -76,69 +70,20 @@ const App = withRouter(() => { const urlParams = new URLSearchParams(window.location.search); const firebase_id_token = urlParams.get('firebase_id_token'); const firebase_organization_id = urlParams.get('firebase_organization_id'); - const [cloudLogedIn, setCloudLogedIn] = useState(isCloud() ? false : true); - const [refreshPlan, setRefreshPlan] = useState(isCloud() ? true : false); + const [cloudLogedIn, setCloudLogedIn] = useState(true); + const [refreshPlan, setRefreshPlan] = useState(false); const [persistedNotifications, setPersistedNotifications] = useState(() => { const storedNotifications = JSON.parse(localStorage.getItem('persistedNotifications')); return storedNotifications || []; }); const [displayedNotifications, setDisplayedNotifications] = useState([]); const [systemMessage, setSystemMessage] = useState([]); - const { stigg } = isCloud() && useStiggContext(); - const getAllowedActions = useGetAllowedActions(); const stateRef = useRef([]); stateRef.current = [cloudLogedIn, persistedNotifications]; - const handleLoginWithToken = async () => { - try { - const data = await httpRequest('POST', ApiEndpoints.LOGIN, { firebase_id_token, firebase_organization_id }, {}, {}, false); - if (data) { - stigg.setCustomerId(data.account_name); - localStorage.setItem(USER_IMAGE, data.user_image); - AuthService.saveToLocalStorage(data); - dispatch({ type: 'SET_USER_DATA', payload: data }); - try { - let wsHost = localStorage.getItem(LOCAL_STORAGE_WS_HOST); - wsHost = `${WS_PREFIX}://${wsHost}`; - let conn; - if (localStorage.getItem(LOCAL_STORAGE_USER_PASS_BASED_AUTH) === 'true') { - const account_id = localStorage.getItem(LOCAL_STORAGE_ACCOUNT_ID); - const internal_ws_pass = localStorage.getItem(LOCAL_STORAGE_INTERNAL_WS_PASS); - conn = await connect({ - servers: [wsHost], - user: '$memphis_user$' + account_id, - pass: internal_ws_pass, - timeout: '5000' - }); - } else { - const connection_token = localStorage.getItem(LOCAL_STORAGE_CONNECTION_TOKEN); - conn = await connect({ - servers: [wsHost], - token: '::' + connection_token, - timeout: '5000' - }); - } - dispatch({ type: 'SET_SOCKET_DETAILS', payload: conn }); - } catch (error) { - throw new Error(error); - } - } - history.push('/overview'); - setCloudLogedIn(true); - } catch (error) { - setCloudLogedIn(true); - console.log(error); - } - }; - useEffect(() => { - if (isCloud() && firebase_id_token) { - const fetchData = async () => { - await handleLoginWithToken(); - }; - fetchData(); - } else setCloudLogedIn(true); + setCloudLogedIn(true); }, []); useEffect(() => { @@ -161,8 +106,6 @@ const App = withRouter(() => { } else if (localStorage.getItem(LOCAL_STORAGE_TOKEN)) { const handleRefreshData = await handleRefreshTokenRequest(); dispatch({ type: 'SET_USER_DATA', payload: handleRefreshData }); - isCloud() && stigg.setCustomerId(handleRefreshData.account_name); - isCloud() && localStorage.setItem(LOCAL_STORAGE_PLAN, handleRefreshData.plan); if (handleRefreshData !== '') { if (firstTime) { try { @@ -193,9 +136,7 @@ const App = withRouter(() => { } return true; } - } else { - isCloud() ? window.location.replace(CLOUD_URL) : history.push(pathDomains.signup); - } + } else history.push(pathDomains.signup); }, []); const handleUpdatePlan = async () => { @@ -204,7 +145,6 @@ const App = withRouter(() => { dispatch({ type: 'SET_ENTITLEMENTS', payload: data?.entitelments }); dispatch({ type: 'SET_PLAN_TYPE', payload: data.plan === planType.FREE }); setRefreshPlan(false); - await getAllowedActions(); showMessages('success', 'Your plan has been successfully updated.'); } catch (error) { setRefreshPlan(false); @@ -382,335 +322,157 @@ const App = withRouter(() => { {systemMessage?.length > 0 && displaySystemMessage()}
{' '} - {!authCheck && - cloudLogedIn && - !refreshPlan && - (!isCloud() ? ( - - - - - -
- } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - }>} /> - }>} /> - }>} - /> - }>} - /> - - - - } - > - } - /> - - - - - ) : ( - - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - {/* - - - } - > - } - /> */} - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - - - - } - > - } - /> - }>} /> - }>} /> - }>} - /> - }>} /> - }>} /> - }>} - /> + {!authCheck && cloudLogedIn && !refreshPlan && ( + + + + + + + } + > + } + /> + + + + } + > + } + /> + + + + } + > + } + /> + + + + } + > + } + /> + + + + } + > + } + /> + + + + } + > + } + /> + + + + } + > + } + /> - - - - } - > - } - /> - - - - - ))} + + + + } + > + } + /> + + + + } + > + } + /> + }>} /> + }>} /> + }>} + /> + }>} + /> + + + + } + > + } + /> + + + + + )} ); diff --git a/ui_src/src/PrivateRoute.js b/ui_src/src/PrivateRoute.js index 25a6bd9b7..0a47e8e30 100644 --- a/ui_src/src/PrivateRoute.js +++ b/ui_src/src/PrivateRoute.js @@ -14,22 +14,13 @@ import React from 'react'; import { Route, Redirect } from 'react-router-dom'; import AuthService from 'services/auth'; -import { isCloud } from 'services/valueConvertor'; -import { CLOUD_URL } from 'config'; import pathDomains from 'router'; function PrivateRoute(props) { const { component: Component, ...rest } = props; if (AuthService.isValidToken()) { return Component} />; - } else { - if (isCloud()) { - AuthService.clearLocalStorage(); - window.location.replace(CLOUD_URL); - } else { - return ; - } - } + } else return ; } export default PrivateRoute; diff --git a/ui_src/src/components/Tabs/index.js b/ui_src/src/components/Tabs/index.js index 13c63c5e0..0adee9ccb 100644 --- a/ui_src/src/components/Tabs/index.js +++ b/ui_src/src/components/Tabs/index.js @@ -22,7 +22,6 @@ import React from 'react'; import TooltipComponent from 'components/tooltip/tooltip'; import CheckboxComponent from 'components/checkBox'; import { PriorityHighRounded } from '@material-ui/icons'; -import CloudOnly from 'components/cloudOnly'; const CustomTabs = ({ tabs, onChange, value, disabled, length, tooltip, icon = false, checkbox = false, defaultActiveKey, tabsCounter, icons }) => { return ( @@ -39,10 +38,7 @@ const CustomTabs = ({ tabs, onChange, value, disabled, length, tooltip, icon = f {icons && icons[index] && !icon && icons[index]} {checkbox && } - <> - {`${tab?.name || tab} ${tabsCounter ? `(${tabsCounter[index]})` : ''} `} - {tab?.cloudOnly && } - + <>{`${tab?.name || tab} ${tabsCounter ? `(${tabsCounter[index]})` : ''} `} {length && length[index] && icon && (
diff --git a/ui_src/src/components/cloudModal/index.js b/ui_src/src/components/cloudModal/index.js deleted file mode 100644 index ab7c09a20..000000000 --- a/ui_src/src/components/cloudModal/index.js +++ /dev/null @@ -1,174 +0,0 @@ -// Copyright 2022-2023 The Memphis.dev Authors -// Licensed under the Memphis Business Source License 1.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// Changed License: [Apache License, Version 2.0 (https://www.apache.org/licenses/LICENSE-2.0), as published by the Apache Foundation. -// -// https://github.com/memphisdev/memphis/blob/master/LICENSE -// -// Additional Use Grant: You may make use of the Licensed Work (i) only as part of your own product or service, provided it is not a message broker or a message queue product or service; and (ii) provided that you do not use, provide, distribute, or make available the Licensed Work as a Service. -// A "Service" is a commercial offering, product, hosted, or managed service, that allows third parties (other than your own employees and contractors acting on your behalf) to access and/or use the Licensed Work or a substantial set of the features or functionality of the Licensed Work to third parties as a software-as-a-service, platform-as-a-service, infrastructure-as-a-service or other similar services that compete with Licensor products or services. - -import './style.scss'; - -import React from 'react'; -import { ReactComponent as FunctionIntegrateIcon } from 'assets/images/functionIntegrate.svg'; -import BundleBanner from 'assets/images/banners/bundle1.webp'; -import CloudBanner from 'assets/images/banners/cloud2.webp'; -import FunctionsBanner from 'assets/images/banners/function3.webp'; -import UpgradeBanner from 'assets/images/banners/upgrade4.webp'; -import UpgradePlans from 'components/upgradePlans'; -import Modal from 'components/modal'; -import Button from 'components/button'; -import { sendTrace } from 'services/genericServices'; - -const CloudModal = ({ type, open, handleClose }) => { - const content = { - bundle: { - title: , - subtitle: ( - <> - Get Your Open-Source Today! - - ), - banner: BundleBanner, - leftBtn: 'Learn More', - leftBtnLink: 'https://memphis.dev/open-source-support-bundle/', - rightBtn: 'Book a Call', - rightBtnLink: 'https://meetings.hubspot.com/yaniv-benhemo' - }, - cloud: { - title: , - subtitle: 'Embrace serverless, enjoy peace of mind, and experience enhanced resilience.', - banner: CloudBanner, - leftBtn: 'Learn More', - leftBtnLink: 'https://memphis.dev/memphis-dev-cloud/', - rightBtn: 'Create a free account', - rightBtnLink: 'https://cloud.memphis.dev/' - }, - upgrade: { - title: ( - <> - - - - ), - subtitle: 'To Unlock More Features And Enhance Your Experience!', - banner: UpgradeBanner, - leftBtn: 'Chat With Your Account Manager', - leftBtnLink: 'https://meetings.hubspot.com/yaniv-benhemo/upgrade-call', - rightBtn: 'Upgrade Now', - rightBtnLink: '' - }, - functions: { - title: ( - <> - - - - ), - subtitle: 'Discover A Faster And Smarter Way To Do Event-driven And Stream Processing', - banner: FunctionsBanner, - leftBtn: 'Learn More', - leftBtnLink: 'https://functions.memphis.dev/', - rightBtn: 'Book a demo', - rightBtnLink: 'https://meetings.hubspot.com/yaniv-benhemo/demo-for-memphis-functions' - } - }; - - return ( - - -
- -
- {content[type]?.title} - -
- } - displayButtons={false} - width="560px" - height="390px" - clickOutside={handleClose} - open={open} - className="cloud-modal" - > -
- banner -
- -