package shuffle import ( "context" "crypto/sha256" "encoding/hex" "fmt" "log" "sort" "strconv" "strings" "time" "encoding/json" "io/ioutil" "math/rand" "net/http" gomemcache "github.com/bradfitz/gomemcache/memcache" uuid "github.com/satori/go.uuid" ) // FIXME: There is some issue when going past 0x9 (>0xA) with how // cache is being counted locally // var dbInterval = 0x20 var dbInterval = 0x9 // var dbInterval = 0x4 var PredictableDataTypes = []string{ "app_executions", "childorg_app_executions", "workflow_executions", "workflow_executions_finished", "workflow_executions_failed", "app_executions_failed", "app_executions_cloud", "subflow_executions", "org_sync_actions", "workflow_executions_cloud", "workflow_executions_onprem", "api_usage", "ai_executions", } func HandleGetWidget(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { return } user, err := HandleApiAuthentication(resp, request) if err != nil { log.Printf("[WARNING] Api authentication failed in get widget: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } _ = user var dashboard string var widget string location := strings.Split(request.URL.String(), "/") if location[1] == "api" { if len(location) <= 6 { resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } dashboard = location[4] widget = location[6] } //log.Printf("Should get widget %s in dashboard %s", widget, dashboard) id := uuid.NewV4().String() // Returning some static info for now returnData := Widget{ Success: true, Id: id, Title: widget, Dashboard: dashboard, Data: []WidgetPoint{ WidgetPoint{ Key: widget, Data: []WidgetPointData{ WidgetPointData{ Key: "11/21/2019", Data: 9, MetaData: WidgetMeta{ Color: "#f86a3e", }, }, WidgetPointData{ Key: "11/22/2019", Data: 4, }, WidgetPointData{ Key: "11/24/2019", Data: 12, }, }, }, WidgetPoint{ Key: "Intel", Data: []WidgetPointData{ WidgetPointData{ Key: "11/22/2019", Data: 5, MetaData: WidgetMeta{ Color: "cyan", }, }, WidgetPointData{ Key: "11/23/2019", Data: 8, }, WidgetPointData{ Key: "11/24/2019", Data: 14, }, }, }, }, } newjson, err := json.Marshal(returnData) if err != nil { log.Printf("[ERROR] Failed marshal in get widget: %s", err) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking data"}`))) return } resp.WriteHeader(200) resp.Write(newjson) } // Starts a new webhook func HandleNewWidget(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { return } user, err := HandleApiAuthentication(resp, request) if err != nil { log.Printf("[WARNING] Api authentication failed in set new hook: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } if user.Role == "org-reader" { log.Printf("[WARNING] Org-reader doesn't have access to make new widgets: %s (%s)", user.Username, user.Id) resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "reason": "Read only user"}`)) return } type requestData struct { Id string `json:"id"` Name string `json:"name"` Type string `json:"type"` Start string `json:"start"` Auth string `json:"auth"` Workflow string `json:"workflow"` Environment string `json:"environment"` Description string `json:"description"` CustomResponse string `json:"custom_response"` } body, err := ioutil.ReadAll(request.Body) if err != nil { log.Printf("[WARNING] Body data error in webhook set: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } _ = body /* ctx := GetContext(request) var requestdata requestData err = json.Unmarshal([]byte(body), &requestdata) if err != nil { log.Printf("[WARNING] Failed unmarshaling inputdata for webhook: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } newId := requestdata.Id if len(newId) != 36 { log.Printf("[WARNING] Bad webhook ID: %s", newId) resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "reason": "Invalid Webhook ID: bad formatting"}`)) return } if requestdata.Id == "" || requestdata.Name == "" { resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "reason": "Required fields id and name can't be empty"}`)) return } validTypes := []string{ "webhook", } isTypeValid := false for _, thistype := range validTypes { if requestdata.Type == thistype { isTypeValid = true break } } if !(isTypeValid) { log.Printf("Type %s is not valid. Try any of these: %s", requestdata.Type, strings.Join(validTypes, ", ")) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } // Let remote endpoint handle access checks (shuffler.io) baseUrl := "https://shuffler.io" if len(os.Getenv("SHUFFLE_GCEPROJECT")) > 0 && len(os.Getenv("SHUFFLE_GCEPROJECT_LOCATION")) > 0 { baseUrl = fmt.Sprintf("https://%s.%s.r.appspot.com", os.Getenv("SHUFFLE_GCEPROJECT"), os.Getenv("SHUFFLE_GCEPROJECT_LOCATION")) } currentUrl := fmt.Sprintf("%s/api/v1/hooks/webhook_%s", baseUrl, newId) startNode := requestdata.Start if requestdata.Environment == "cloud" && project.Environment != "cloud" { // https://shuffler.io/v1/hooks/webhook_80184973-3e82-4852-842e-0290f7f34d7c log.Printf("[INFO] Should START a cloud webhook for url %s for startnode %s", currentUrl, startNode) org, err := GetOrg(ctx, user.ActiveOrg.Id) if err != nil { log.Printf("Failed finding org %s: %s", org.Id, err) return } action := CloudSyncJob{ Type: "webhook", Action: "start", OrgId: org.Id, PrimaryItemId: newId, SecondaryItem: startNode, ThirdItem: requestdata.Workflow, FourthItem: requestdata.Auth, } err = executeCloudAction(action, org.SyncConfig.Apikey) if err != nil { log.Printf("[WARNING] Failed cloud action START webhook execution: %s", err) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) return } else { log.Printf("[INFO] Successfully set up cloud action schedule") } } hook := Hook{ Id: newId, Start: startNode, Workflows: []string{requestdata.Workflow}, Info: Info{ Name: requestdata.Name, Description: requestdata.Description, Url: fmt.Sprintf("%s/api/v1/hooks/webhook_%s", baseUrl, newId), }, Type: "webhook", Owner: user.Username, Status: "uninitialized", Actions: []HookAction{ HookAction{ Type: "workflow", Name: requestdata.Name, Id: requestdata.Workflow, Field: "", }, }, Running: false, OrgId: user.ActiveOrg.Id, Environment: requestdata.Environment, Auth: requestdata.Auth, CustomResponse: requestdata.CustomResponse, } hook.Status = "running" hook.Running = true err = SetHook(ctx, hook) if err != nil { log.Printf("[WARNING] Failed setting hook: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } */ newId := "tmp" log.Printf("[INFO] Set up a new widget %s", newId) resp.WriteHeader(200) resp.Write([]byte(`{"success": true}`)) } func GetSpecificStats(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { return } var orgId string var statsKey string location := strings.Split(request.URL.String(), "/") if location[1] == "api" { if len(location) <= 4 { log.Printf("Path too short: %d", len(location)) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } statsKey = location[4] if len(location) > 6 { orgId = location[4] statsKey = location[6] } } // Remove ? from orgId or statsKey orgId = strings.Split(orgId, "?")[0] statsKey = strings.Split(statsKey, "?")[0] if len(statsKey) <= 1 { log.Printf("[WARNING] Invalid stats key: %s", statsKey) resp.WriteHeader(400) resp.Write([]byte(`{"success": false, "reason": "Invalid stats key"}`)) return } statsKey = strings.ToLower(strings.ReplaceAll(statsKey, " ", "_")) user, err := HandleApiAuthentication(resp, request) if err != nil { log.Printf("[WARNING] Api authentication failed in get stats: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } orgId = user.ActiveOrg.Id ctx := GetContext(request) info, err := GetOrgStatistics(ctx, orgId) if err != nil { log.Printf("[WARNING] Failed getting stats in specific stats for org %s: %s", orgId, err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "reason": "Failed getting stats for your org. Maybe not initialized yet?"}`)) return } // Default statDays := 30 // Check for if the query parameter exists if len(request.URL.Query().Get("days")) > 0 { amountQuery := request.URL.Query().Get("days") statDays, err = strconv.Atoi(amountQuery) if err != nil { log.Printf("[WARNING] Failed parsing days query parameter: %s", err) } else { if statDays > 365 { statDays = 365 } } } if debug { log.Printf("[DEBUG] Should get stats for key %s for the last %d days", statsKey, statDays) } totalEntires := 0 totalValue := 0 statEntries := []AdditionalUseConfig{} info.DailyStatistics = append(info.DailyStatistics, DailyStatistics{ Date: time.Now(), Additions: info.Additions, }) allStats := []string{} getTypedValue := func(d DailyStatistics, key string) int64 { switch key { case "app_executions": return d.AppExecutions case "childorg_app_executions": return d.ChildAppExecutions case "app_executions_failed": return d.AppExecutionsFailed case "subflow_executions": return d.SubflowExecutions case "workflow_executions": return d.WorkflowExecutions case "workflow_executions_finished": return d.WorkflowExecutionsFinished case "workflow_executions_failed": return d.WorkflowExecutionsFailed case "org_sync_actions": return d.OrgSyncActions case "workflow_executions_cloud": return d.CloudExecutions case "workflow_executions_onprem": return d.OnpremExecutions case "api_usage": return d.ApiUsage case "ai_executions": return d.AIUsage default: return -1 } } isPredictable := ArrayContains(PredictableDataTypes, statsKey) for _, daily := range info.DailyStatistics { // Check if the date is more than statDays ago shouldAppend := true if daily.Date.Before(time.Now().AddDate(0, 0, -statDays)) { shouldAppend = false } if isPredictable { if shouldAppend { value := getTypedValue(daily, statsKey) if value >= 0 { totalEntires++ totalValue += int(value) statEntries = append(statEntries, AdditionalUseConfig{ Key: statsKey, Value: value, Date: daily.Date, }) } } // Track available keys too for _, k := range PredictableDataTypes { if !ArrayContains(allStats, k) { allStats = append(allStats, k) } } continue } // Custom additions path (original behavior) for _, addition := range daily.Additions { newKey := strings.ToLower(strings.ReplaceAll(addition.Key, " ", "_")) if shouldAppend && newKey == statsKey { totalEntires++ totalValue += int(addition.Value) addition.Key = statsKey addition.Date = daily.Date statEntries = append(statEntries, addition) } if !ArrayContains(allStats, newKey) { allStats = append(allStats, newKey) } } } // If predictable key, also include today's in-memory daily counters (not yet rolled into DailyStatistics) if isPredictable { today := time.Now() var todayValue int64 = 0 switch statsKey { case "app_executions": todayValue = info.DailyAppExecutions case "childorg_app_executions": todayValue = info.DailyChildAppExecutions case "app_executions_failed": todayValue = info.DailyAppExecutionsFailed case "subflow_executions": todayValue = info.DailySubflowExecutions case "workflow_executions": todayValue = info.DailyWorkflowExecutions case "workflow_executions_finished": todayValue = info.DailyWorkflowExecutionsFinished case "workflow_executions_failed": todayValue = info.DailyWorkflowExecutionsFailed case "org_sync_actions": todayValue = info.DailyOrgSyncActions case "workflow_executions_cloud": todayValue = info.DailyCloudExecutions case "workflow_executions_onprem": todayValue = info.DailyOnpremExecutions case "api_usage": todayValue = info.DailyApiUsage case "ai_executions": todayValue = info.DailyAIUsage } // Only append if within window if !today.Before(time.Now().AddDate(0, 0, -statDays)) { statEntries = append(statEntries, AdditionalUseConfig{ Key: statsKey, Value: todayValue, Date: today, }) totalEntires++ totalValue += int(todayValue) } } // Deduplicate and merge same days mergedEntries := []AdditionalUseConfig{} for _, entry := range statEntries { found := false for mergedEntryIndex, mergedEntry := range mergedEntries { if mergedEntry.Date.Day() == entry.Date.Day() && mergedEntry.Date.Month() == entry.Date.Month() && mergedEntry.Date.Year() == entry.Date.Year() { mergedEntries[mergedEntryIndex].Value += entry.Value found = true break } } if !found { mergedEntries = append(mergedEntries, entry) } } statEntries = mergedEntries // Check if entries exist for the last X statDays // Backfill any missing ones so that the number is correct if len(statEntries) < statDays { // Find the missing days missingDays := []time.Time{} for i := 0; i < statDays; i++ { missingDays = append(missingDays, time.Now().AddDate(0, 0, -i)) } // Find the missing entries appended := 0 foundAmount := 0 toAppend := []AdditionalUseConfig{} for _, missingDay := range missingDays { found := false for _, entry := range statEntries { if entry.Date.Day() == missingDay.Day() && entry.Date.Month() == missingDay.Month() && entry.Date.Year() == missingDay.Year() { foundAmount += 1 found = true break } } if !found { appended += 1 toAppend = append(toAppend, AdditionalUseConfig{ Key: statsKey, Value: 0, Date: missingDay, }) } } statEntries = append(statEntries, toAppend...) } // Append cache for right now as it may not be in the DB yet for statEntryIndex, statEntry := range statEntries { if statEntry.Date.Day() == time.Now().Day() && statEntry.Date.Month() == time.Now().Month() && statEntry.Date.Year() == time.Now().Year() { for _, addition := range info.Additions { if addition.Key != statsKey { continue } key := fmt.Sprintf("cache_%s_%s", orgId, addition.Key) cacheItem, err := GetCache(ctx, key) if err == nil { parsedItem := []byte(cacheItem.([]uint8)) increment, err := strconv.Atoi(string(parsedItem)) if err == nil { statEntries[statEntryIndex].Value += int64(increment) totalValue += int(increment) } } break } } } // Sort statentries by date sort.Slice(statEntries, func(i, j int) bool { return statEntries[i].Date.Before(statEntries[j].Date) }) // For debugging stats that don't show up by injecting them /* if debug && totalValue == 0 { log.Printf("[DEBUG] Found %d entries for '%s' with 0 in data. Force-adding data to first entry.", len(statEntries), statsKey) chosenIndex := rand.Intn(len(statEntries)) statEntries[chosenIndex].Value = int64(rand.Intn(10) + 1) } */ marshalledEntries, err := json.Marshal(statEntries) if err != nil { log.Printf("[ERROR] Failed marshal in get org stats: %s", err) resp.WriteHeader(500) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking data for org stats"}`))) return } availableStats, err := json.Marshal(allStats) if err != nil { log.Printf("[ERROR] Failed marshal in get org stats: %s", err) resp.WriteHeader(500) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking data for org stats"}`))) return } //successful := totalValue != 0 successful := true resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": %v, "key": "%s", "total": %d, "available_keys": %s, "entries": %s}`, successful, strings.ReplaceAll(statsKey, "\"", ""), totalValue, string(availableStats), string(marshalledEntries)))) } func HandleGetStatistics(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { return } var orgId string var statsKey string location := strings.Split(request.URL.String(), "/") if location[1] == "api" { // Just falling back if len(location) <= 4 { } else { orgId = location[4] } } user, err := HandleApiAuthentication(resp, request) if err != nil { log.Printf("[WARNING] Api authentication failed in get stats: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } if len(orgId) == 0 { orgId = user.ActiveOrg.Id } org := &Org{} ctx := GetContext(request) if orgId == "public" { if user.SupportAccess { log.Printf("[AUDIT] User %s (%s) is getting org stats for PUBLIC org %s with support access", user.Username, user.Id, orgId) } } else { org, err = GetOrg(ctx, orgId) if err != nil { resp.WriteHeader(403) resp.Write([]byte(`{"success": false, "reason": "Failed getting org stats"}`)) return } userFound := false for _, inneruser := range org.Users { if inneruser.Id == user.Id { userFound = true break } } if user.SupportAccess { log.Printf("[AUDIT] User %s (%s) is getting org stats for %s (%s) with support access", user.Username, user.Id, org.Name, orgId) userFound = true } if !userFound { log.Printf("[WARNING] User %s isn't a part of org %s (get)", user.Id, org.Id) resp.WriteHeader(403) resp.Write([]byte(`{"success": false, "reason": "User doesn't have access to org"}`)) return } } // FIXME: Removed the current stats grabber as it made no sense // to dump it to cache. The point was JUST to grab it in realtime. info, err := GetOrgStatistics(ctx, orgId) if err != nil { log.Printf("[WARNING] Failed getting stats for org %s: %s", orgId, err) //resp.WriteHeader(400) //resp.Write([]byte(`{"success": false, "reason": "Failed getting stats for your org. Maybe not initialized yet?"}`)) //return info.OrgId = orgId info.OrgName = org.Name } // Sideload GCS overflow stats (entries >60 days old archived from Datastore), cached 30 min. if project.Environment == "cloud" && len(orgFileBucket) > 0 { var gcsStats []DailyStatistics gcsCacheKey := fmt.Sprintf("gcs_stats_%s", orgId) if cached, cacheErr := GetCache(ctx, gcsCacheKey); cacheErr == nil { _ = json.Unmarshal([]byte(cached.([]uint8)), &gcsStats) } else { bucketPath := fmt.Sprintf("org_statistics/%s/stats.json", orgId) obj := project.StorageClient.Bucket(orgFileBucket).Object(bucketPath) if gcsReader, gcsErr := obj.NewReader(ctx); gcsErr == nil { gcsBytes, readErr := ioutil.ReadAll(gcsReader) gcsReader.Close() if readErr == nil && len(gcsBytes) > 0 { if unmarshalErr := json.Unmarshal(gcsBytes, &gcsStats); unmarshalErr == nil { _ = SetCache(ctx, gcsCacheKey, gcsBytes, 30) } } } } if len(gcsStats) > 0 { log.Printf("[DEBUG] HandleGetStatistics: merging %d GCS overflow entries for org %s", len(gcsStats), orgId) // Deduplicate by date; Datastore entries win on conflict. dateMapCap := len(gcsStats) if len(info.DailyStatistics) > dateMapCap { dateMapCap = len(info.DailyStatistics) } dateMap := make(map[string]DailyStatistics, dateMapCap) for _, d := range gcsStats { dateMap[d.Date.UTC().Format("2006-01-02")] = d } for _, d := range info.DailyStatistics { dateMap[d.Date.UTC().Format("2006-01-02")] = d } merged := make([]DailyStatistics, 0, len(dateMap)) for _, d := range dateMap { merged = append(merged, d) } info.DailyStatistics = merged } } // Sideload app runs, workflow runs and subflow runs (just in case) // This makes numbers accurate even when less than dbDumpInterval key := fmt.Sprintf("cache_%s_app_executions", orgId) cacheItem, err := GetCache(ctx, key) if err == nil { parsedItem := []byte(cacheItem.([]uint8)) increment, err := strconv.Atoi(string(parsedItem)) if err == nil { info.TotalAppExecutions += int64(increment) info.MonthlyAppExecutions += int64(increment) info.WeeklyAppExecutions += int64(increment) info.DailyAppExecutions += int64(increment) info.HourlyAppExecutions += int64(increment) } } key = fmt.Sprintf("cache_%s_childorg_app_executions", orgId) cacheItem, err = GetCache(ctx, key) if err == nil { parsedItem := []byte(cacheItem.([]uint8)) increment, err := strconv.Atoi(string(parsedItem)) if err == nil { info.TotalChildAppExecutions += int64(increment) info.MonthlyChildAppExecutions += int64(increment) info.WeeklyChildAppExecutions += int64(increment) info.DailyChildAppExecutions += int64(increment) info.HourlyChildAppExecutions += int64(increment) } } key = fmt.Sprintf("cache_%s_workflow_executions", orgId) cacheItem, err = GetCache(ctx, key) if err == nil { parsedItem := []byte(cacheItem.([]uint8)) increment, err := strconv.Atoi(string(parsedItem)) if err == nil { info.TotalWorkflowExecutions += int64(increment) info.MonthlyWorkflowExecutions += int64(increment) info.WeeklyWorkflowExecutions += int64(increment) info.DailyWorkflowExecutions += int64(increment) info.HourlyWorkflowExecutions += int64(increment) } } key = fmt.Sprintf("cache_%s_subflow_executions", orgId) cacheItem, err = GetCache(ctx, key) if err == nil { parsedItem := []byte(cacheItem.([]uint8)) increment, err := strconv.Atoi(string(parsedItem)) if err == nil { info.TotalSubflowExecutions += int64(increment) info.MonthlySubflowExecutions += int64(increment) info.WeeklySubflowExecutions += int64(increment) info.DailySubflowExecutions += int64(increment) info.HourlySubflowExecutions += int64(increment) } } for additionCnt, addition := range info.Additions { key := fmt.Sprintf("cache_%s_%s", orgId, addition.Key) cacheItem, err = GetCache(ctx, key) if err == nil { parsedItem := []byte(cacheItem.([]uint8)) increment, err := strconv.Atoi(string(parsedItem)) if err == nil { info.Additions[additionCnt].Value += int64(increment) } } // In case a lot of use if additionCnt > 10 { break } } _ = statsKey //if len(statsKey) > 0 { // log.Printf("[INFO] Should get stats for key %s", statsKey) //} if len(info.DailyStatistics) > 0 { // Sort the array sort.Slice(info.DailyStatistics, func(i, j int) bool { return info.DailyStatistics[i].Date.Before(info.DailyStatistics[j].Date) }) // Get a max of the last 365 days if len(info.DailyStatistics) > 365 { info.DailyStatistics = info.DailyStatistics[len(info.DailyStatistics)-60:] } } newjson, err := json.Marshal(info) if err != nil { log.Printf("[ERROR] Failed marshal in get org stats: %s", err) resp.WriteHeader(500) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking data for org stats"}`))) return } resp.WriteHeader(200) resp.Write(newjson) } func HandleAppendStatistics(resp http.ResponseWriter, request *http.Request) { // Send in a thing to increment cors := HandleCors(resp, request) if cors { return } user, err := HandleApiAuthentication(resp, request) if err != nil { log.Printf("[WARNING] Api authentication failed in add stats: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return } if user.Role == "org-reader" { log.Printf("[WARNING] Org-reader doesn't have access to add stats: %s (%s)", user.Username, user.Id) resp.WriteHeader(403) resp.Write([]byte(`{"success": false, "reason": "Read only user"}`)) return } body, err := ioutil.ReadAll(request.Body) if err != nil { log.Printf("[WARNING] Failed reading body in add stats: %s", err) resp.WriteHeader(400) resp.Write([]byte(`{"success": false, "reason": "Failed reading body"}`)) return } inputData := AdditionalUseConfig{} err = json.Unmarshal(body, &inputData) if err != nil { log.Printf("[WARNING] Failed unmarshaling inputdata for add stats: %s", err) resp.WriteHeader(400) resp.Write([]byte(`{"success": false, "reason": "Failed unpacking data"}`)) return } if len(inputData.Key) < 3 || len(inputData.Key) > 50 { log.Printf("[WARNING] Invalid input data for add stats: %s", inputData.Key) resp.WriteHeader(400) resp.Write([]byte(`{"success": false, "reason": "'key' has to be a minimum of 3 characters and a maximum of 50"}`)) return } if inputData.Value <= 0 { inputData.Value = 1 } if inputData.Value > 100 { resp.WriteHeader(400) resp.Write([]byte(`{"success": false, "reason": "'value' to increment can be a maximum of 100"}`)) return } if !strings.HasPrefix(inputData.Key, "custom_") { inputData.Key = fmt.Sprintf("custom_%s", inputData.Key) } ctx := GetContext(request) go IncrementCache(ctx, user.ActiveOrg.Id, inputData.Key, int(inputData.Value)) resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Cache incremented by %d"}`, inputData.Value))) } // Rudementary caching system. WILL go wrong at times without sharding. // It's only good for the user in cloud, hence wont bother for a while // Optional input is the amount to increment func IncrementCache(ctx context.Context, orgId, dataType string, amount ...int) { // Check if environment is worker and skip if project.Environment == "worker" { //log.Printf("[DEBUG] Skipping cache increment for worker with datatype %s", dataType) return } if len(orgId) != 36 && orgId != "public" && orgId != "INTERNAL" { log.Printf("[ERROR] Increment Stats with bad OrgId '%s' for type '%s'", orgId, dataType) return } dataType = strings.ToLower(strings.Replace(dataType, " ", "_", -1)) incrementAmount := 1 if len(amount) > 0 { if amount[0] > 0 { incrementAmount = amount[0] } } // Dump to disk every 0x19 // 1. Get the existing value // 2. Update it dbDumpInterval := uint8(dbInterval) key := fmt.Sprintf("cache_%s_%s", orgId, dataType) if len(memcached) > 0 { appendForQuickDump := false if !ArrayContains(PredictableDataTypes, dataType) { appendForQuickDump = true } if appendForQuickDump { // check if the cache already key is indexed in memcache keyItems, err := mc.Get("stat_cache_keys_" + orgId) if err == gomemcache.ErrCacheMiss { keyItem := []string{key} data, err := json.Marshal(keyItem) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) } else { // dump it to memcache item := &gomemcache.Item{ Key: "stat_cache_keys_" + orgId, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting increment cache for key %s: %s", orgId, err) } else { // log.Printf("[DEBUG] Set cache index key for (1) %s", orgId) } } } else { dumpedItems := []string{} err = json.Unmarshal(keyItems.Value, &dumpedItems) if err != nil { log.Printf("[ERROR] Failed unmarshalling item in cache: %s", err) } else { if !ArrayContains(dumpedItems, key) { dumpedItems = append(dumpedItems, key) data, err := json.Marshal(dumpedItems) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) } else { // dump it to memcache item := &gomemcache.Item{ Key: "stat_cache_keys_" + orgId, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting increment cache for key %s: %s", orgId, err) } else { // log.Printf("[DEBUG] Set cache index key for (1) %s", orgId) } } } } } } item, err := mc.Get(key) if err == gomemcache.ErrCacheMiss { incrementItem := IncrementInCache{ Amount: uint64(incrementAmount), CreatedAt: time.Now().Unix(), } data, err := json.Marshal(incrementItem) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) return } item := &gomemcache.Item{ Key: key, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting increment cache for key %s: %s", orgId, err) } } else if err != nil { log.Printf("[ERROR] Failed increment memcache err: %s", err) } else { if item == nil || item.Value == nil { incrementItem := IncrementInCache{ Amount: uint64(incrementAmount), CreatedAt: time.Now().Unix(), } data, err := json.Marshal(incrementItem) if err != nil { log.Printf("[DEBUG] Failed marshalling increment item for cache: %s", err) return } item = &gomemcache.Item{ Key: key, Value: data, Expiration: 86400 * 30, } // log.Printf("[ERROR] Value in DB is nil for cache %s.", dataType) } if len(item.Value) == 1 { // case to use if the cache that was present before // the new changes that introduced the struct to the increment system. // log.Printf("[DEBUG] This is from the older system. num: %+v", item.Value) // num := uint64(item.Value[0]) // num += uint64(incrementAmount) // log.Printf("[DEBUG] new num: %d", num) // there is some bug here. i would much rather lose the data here. num := uint64(incrementAmount) incrementItem := IncrementInCache{ Amount: num, CreatedAt: time.Now().Unix(), } data, err := json.Marshal(incrementItem) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) return } item := &gomemcache.Item{ Key: key, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting increment cache for key %s: %s", orgId, err) return } } else if len(item.Value) > 0 { var incrementedItemInCache IncrementInCache err := json.Unmarshal(item.Value, &incrementedItemInCache) if err != nil { log.Printf("[ERROR] Failed unmarshalling item in cache: %s", err) return } num := incrementedItemInCache.Amount // num += byte(incrementAmount) num += uint64(incrementAmount) //num += []byte{2} incrementedItemInCache.Amount = num // log.Printf("[DEBUG] time.Now().Unix() (%d) - incrementedItemInCache.CreatedAt (%d) = %d", time.Now().Unix(), incrementedItemInCache.CreatedAt, time.Now().Unix()-incrementedItemInCache.CreatedAt) // if num >= dbDumpInterval { // if the cache was created more than a day ago // make it a random number between // (10-60 seconds) randomSeconds := (rand.Intn(50) + 10) * 5 // to make the number longer if time.Now().Unix()-incrementedItemInCache.CreatedAt > int64(randomSeconds) && incrementedItemInCache.Amount > uint64(dbInterval) { // Memcache dump first to keep the counter going for other executions oldNum := num num = 0 incrementedItemInCache.Amount = num incrementedItemInCache.CreatedAt = time.Now().Unix() // log.Printf("[DEBUG] Dumping cache item with key %s which was created at %s is was %d", key, incrementedItemInCache.CreatedAt, oldNum) data, err := json.Marshal(incrementedItemInCache) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) return } // an issue here is that it isn't necessary that num is dbDumpInterval err = IncrementCacheDump(ctx, orgId, dataType, int(oldNum)) if err != nil { log.Printf("[ERROR] Failed dumping cache for key (1) %s: %s", key, err) if strings.Contains(fmt.Sprintf("%s", err), "concurrent transaction") { // log.Printf("[ERROR] Concurrent transaction in cache dump: %s. Storing in cache (%s) instead with new amount: %d", err, key, oldNum) incrementedItemInCache.Amount = oldNum data, err := json.Marshal(incrementedItemInCache) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) } item := &gomemcache.Item{ Key: key, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting inner memcache for key %s: %s", orgId, err) } } else { log.Printf("[ERROR] Failed dumping cache for key %s: %s", key, err) } } else { item := &gomemcache.Item{ Key: key, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting inner memcache for key %s: %s", orgId, err) } } } else { //log.Printf("NOT Dumping!") // this case got apparently overwritten unnecessarily 3 times out of 20. // data gets more lost here due to cache overwrites. // add a random sleep of a few miliseconds here randomSleep := rand.Intn(50) + 10 time.Sleep(time.Duration(randomSleep) * time.Millisecond) // read again and check if it's already not dumped item, err := mc.Get(key) if err != nil { log.Printf("[ERROR] Failed getting cache item for key %s: %s", key, err) return } incrementedItemInCache = IncrementInCache{} err = json.Unmarshal(item.Value, &incrementedItemInCache) if err != nil { log.Printf("[ERROR] Failed unmarshalling item in cache: %s", err) incrementedItemInCache.Amount = num incrementedItemInCache.CreatedAt = time.Now().Unix() } // this means there will be an overwrite! if incrementedItemInCache.Amount == num { // better to update the cache again instead of losing the data incrementedItemInCache.Amount += uint64(incrementAmount) } else if num > incrementedItemInCache.Amount { // we bow to the higher number we have incrementedItemInCache.Amount = num } else if incrementedItemInCache.Amount > num { // this means, a bunch of stats were added in the meantime // bow to the higher number and just increment again incrementedItemInCache.Amount += uint64(incrementAmount) } // log.Printf("[DEBUG] Cache item with key %s which was created at %d is now %d", key, incrementedItemInCache.CreatedAt, incrementedItemInCache.Amount) // log.Printf("[DEBUG] Cache item with key %s which was created at %d is now %d. While num we updated was %d", key, incrementedItemInCache.CreatedAt, incrementedItemInCache.Amount, num) data, err := json.Marshal(incrementedItemInCache) if err != nil { log.Printf("[ERROR] Failed marshalling increment item for cache: %s", err) } item = &gomemcache.Item{ Key: key, Value: data, Expiration: 86400 * 30, } if err := mc.Set(item); err != nil { log.Printf("[ERROR] Failed setting inner memcache for key %s: %s", orgId, err) } } } else { // let's keep this here for now // log.Printf("[ERROR] Length of value in cache key %s is less than 1: %d", key, len(item.Value)) } } } else { // Get the cache, but use requestCache instead of memcache foundItem := 1 item, err := GetCache(ctx, key) if err != nil { if incrementAmount > int(dbDumpInterval) { foundItem = incrementAmount } else { //toIncrement := []byte(fmt.Sprintf("%d", incrementAmount)) //toIncrement := []byte(string(incrementAmount)) foundItem = incrementAmount } //log.Printf("[DEBUG] Increment cache miss for %s", key) } else { // make item into a number if item == nil { log.Printf("[ERROR] Value in DB is nil for cache %s. Setting to 1", dataType) } else { // Parse out int from []uint8 with marshal // String (ASCII): 0x31 -> 1 // int: 0x1 -> 1 //foundData := []byte(item.(int)) foundData := item.([]uint8) foundItem, err = strconv.Atoi(string(foundData)) if err != nil { log.Printf("[ERROR] Stat tracking fail: Failed converting item to int: %s. Datatype: %s", err, dataType) foundItem = incrementAmount //foundItem = foundData } else { foundItem += incrementAmount } } } if foundItem >= int(dbDumpInterval) { // Memcache dump first to keep the counter going for other executions go SetCache(context.Background(), key, []byte(fmt.Sprintf("%x", 0)), 86400) IncrementCacheDump(ctx, orgId, dataType, foundItem) //log.Printf("[DEBUG] Dumping cache for %s with amount %d", key, foundItem) } else { // Set cache //setCacheValue := []byte(strconv.FormatInt(int64(foundItem), 16)) //setCacheValue := []byte(fmt.Sprintf("%d", foundItem)) // FIXME: Something is wrong here past 0x9 :O setCacheValue := []byte(fmt.Sprintf("%x", foundItem)) err = SetCache(ctx, key, setCacheValue, 86400) if err != nil { log.Printf("[ERROR] Failed setting increment cache for key %s: %s", orgId, err) } } return } } // 1. Check list if there is a record for yesterday // 2. If there isn't, set it and clear out the daily records // Also: can we dump a list of apps that run? Maybe a list of them? func handleDailyCacheUpdate(executionInfo *ExecutionInfo) *ExecutionInfo { timeYesterday := time.Now().AddDate(0, 0, -1) timeYesterdayFormatted := timeYesterday.Format("2006-12-02") for _, day := range executionInfo.DailyStatistics { // Check if the day.Date is the same as yesterday and return if it is if day.Date.Format("2006-12-02") == timeYesterdayFormatted { for additionIndex, _ := range executionInfo.Additions { executionInfo.Additions[additionIndex].DailyValue = 0 } return executionInfo } } log.Printf("[DEBUG] Daily stats not updated for %s in org %s today. Only have %d stats so far - running update.", timeYesterday, executionInfo.OrgId, len(executionInfo.DailyStatistics)) // If we get here, we need to update the daily stats newDay := DailyStatistics{ Date: timeYesterday, AppExecutions: executionInfo.DailyAppExecutions, ChildAppExecutions: executionInfo.DailyChildAppExecutions, AppExecutionsFailed: executionInfo.DailyAppExecutionsFailed, SubflowExecutions: executionInfo.DailySubflowExecutions, WorkflowExecutions: executionInfo.DailyWorkflowExecutions, WorkflowExecutionsFinished: executionInfo.DailyWorkflowExecutionsFinished, WorkflowExecutionsFailed: executionInfo.DailyWorkflowExecutionsFailed, OrgSyncActions: executionInfo.DailyOrgSyncActions, CloudExecutions: executionInfo.DailyCloudExecutions, OnpremExecutions: executionInfo.DailyOnpremExecutions, AIUsage: executionInfo.DailyAIUsage, ApiUsage: executionInfo.DailyApiUsage, Additions: executionInfo.Additions, } executionInfo.DailyStatistics = append(executionInfo.DailyStatistics, newDay) // Cleaning up old stuff we don't use for now executionInfo.HourlyAppExecutions = 0 executionInfo.HourlyChildAppExecutions = 0 executionInfo.HourlyAppExecutionsFailed = 0 executionInfo.HourlySubflowExecutions = 0 executionInfo.HourlyWorkflowExecutions = 0 executionInfo.HourlyWorkflowExecutionsFinished = 0 executionInfo.HourlyChildWorkflowExecutions = 0 executionInfo.HourlyWorkflowExecutionsFailed = 0 executionInfo.HourlyOrgSyncActions = 0 executionInfo.HourlyCloudExecutions = 0 executionInfo.HourlyOnpremExecutions = 0 // Reset daily executionInfo.DailyAppExecutions = 0 executionInfo.DailyChildAppExecutions = 0 executionInfo.DailyAppExecutionsFailed = 0 executionInfo.DailySubflowExecutions = 0 executionInfo.DailyWorkflowExecutions = 0 executionInfo.DailyWorkflowExecutionsFinished = 0 executionInfo.DailyChildWorkflowExecutions = 0 executionInfo.DailyWorkflowExecutionsFailed = 0 executionInfo.DailyOrgSyncActions = 0 executionInfo.DailyCloudExecutions = 0 executionInfo.DailyOnpremExecutions = 0 executionInfo.DailyApiUsage = 0 executionInfo.DailyAIUsage = 0 // Weekly executionInfo.WeeklyAppExecutions = 0 executionInfo.WeeklyChildAppExecutions = 0 executionInfo.WeeklyAppExecutionsFailed = 0 executionInfo.WeeklySubflowExecutions = 0 executionInfo.WeeklyWorkflowExecutions = 0 executionInfo.WeeklyWorkflowExecutionsFinished = 0 executionInfo.WeeklyWorkflowExecutionsFailed = 0 executionInfo.WeeklyOrgSyncActions = 0 executionInfo.WeeklyCloudExecutions = 0 executionInfo.WeeklyOnpremExecutions = 0 executionInfo.WeeklyChildWorkflowExecutions = 0 // Cleans up "random" stats as well for additionIndex, _ := range executionInfo.Additions { executionInfo.Additions[additionIndex].Value = 0 executionInfo.Additions[additionIndex].DailyValue = 0 } now := time.Now() currentMonth := int(now.Month()) if executionInfo.LastMonthlyResetMonth != currentMonth { log.Printf("[DEBUG] Resetting monthly stats for org %s on %s", executionInfo.OrgId, now.Format("2006-01-02")) executionInfo.MonthlyAppExecutions = 0 executionInfo.MonthlyChildAppExecutions = 0 executionInfo.MonthlyAppExecutionsFailed = 0 executionInfo.MonthlySubflowExecutions = 0 executionInfo.MonthlyWorkflowExecutions = 0 executionInfo.MonthlyWorkflowExecutionsFinished = 0 executionInfo.MonthlyChildWorkflowExecutions = 0 executionInfo.MonthlyWorkflowExecutionsFailed = 0 executionInfo.MonthlyOrgSyncActions = 0 executionInfo.MonthlyCloudExecutions = 0 executionInfo.MonthlyOnpremExecutions = 0 executionInfo.MonthlyApiUsage = 0 executionInfo.MonthlyAIUsage = 0 executionInfo.MonthlyAgentExecutions = 0 executionInfo.MonthlyAgentTokens = 0 executionInfo.LastMonthlyResetMonth = currentMonth executionInfo.LastUsageAlertThreshold = 0 // Reset all usage alerts to unsent for index := range executionInfo.UsageAlerts { executionInfo.UsageAlerts[index].Email_send = false } } return executionInfo } func generateAlertCacheKey(orgId string, threshold interface{}, emailList []string) string { sortedEmails := make([]string, len(emailList)) copy(sortedEmails, emailList) sort.Strings(sortedEmails) emailsStr := strings.Join(sortedEmails, ",") thresholdStr := fmt.Sprintf("%v", threshold) key := fmt.Sprintf("alert_cache_%s_%s_%s", orgId, thresholdStr, emailsStr) key = strings.ReplaceAll(key, "@", "_at_") key = strings.ReplaceAll(key, ".", "_dot_") key = strings.ReplaceAll(key, " ", "_") // Memcache keys have a 250-character limit. Hash anything that exceeds it. if len(key) > 200 { hash := sha256.Sum256([]byte(key)) key = "alert_cache_" + hex.EncodeToString(hash[:]) } return key } func checkAndSetAlertCache(ctx context.Context, cacheKey string) bool { _, err := GetCache(ctx, cacheKey) if err == nil { return false } now := time.Now() endOfMonth := time.Date(now.Year(), now.Month()+1, 1, 0, 0, 0, 0, now.Location()) remainingMinutes := int32(endOfMonth.Sub(now).Minutes()) if remainingMinutes < 60 { remainingMinutes = 60 } err = SetCache(ctx, cacheKey, []byte("sent"), remainingMinutes) if err != nil { log.Printf("[WARNING] Failed setting alert cache for key %s: %s", cacheKey, err) } return true } func HandleIncrement(dataType string, orgStatistics *ExecutionInfo, increment uint) *ExecutionInfo { appendCustom := false if dataType == "childorg_app_executions" { orgStatistics.TotalChildAppExecutions += int64(increment) orgStatistics.MonthlyChildAppExecutions += int64(increment) orgStatistics.WeeklyChildAppExecutions += int64(increment) orgStatistics.DailyChildAppExecutions += int64(increment) orgStatistics.HourlyChildAppExecutions += int64(increment) } else if dataType == "app_executions" { orgStatistics.TotalAppExecutions += int64(increment) orgStatistics.MonthlyAppExecutions += int64(increment) orgStatistics.WeeklyAppExecutions += int64(increment) orgStatistics.DailyAppExecutions += int64(increment) orgStatistics.HourlyAppExecutions += int64(increment) } else if dataType == "workflow_executions" { orgStatistics.TotalWorkflowExecutions += int64(increment) orgStatistics.MonthlyWorkflowExecutions += int64(increment) orgStatistics.WeeklyWorkflowExecutions += int64(increment) orgStatistics.DailyWorkflowExecutions += int64(increment) orgStatistics.HourlyWorkflowExecutions += int64(increment) } else if dataType == "childorg_workflow_executions" { orgStatistics.TotalChildWorkflowExecutions += int64(increment) orgStatistics.MonthlyChildWorkflowExecutions += int64(increment) orgStatistics.WeeklyChildWorkflowExecutions += int64(increment) orgStatistics.DailyChildWorkflowExecutions += int64(increment) orgStatistics.HourlyChildWorkflowExecutions += int64(increment) } else if dataType == "workflow_executions_finished" { orgStatistics.TotalWorkflowExecutionsFinished += int64(increment) orgStatistics.MonthlyWorkflowExecutionsFinished += int64(increment) orgStatistics.WeeklyWorkflowExecutionsFinished += int64(increment) orgStatistics.DailyWorkflowExecutionsFinished += int64(increment) orgStatistics.HourlyWorkflowExecutionsFinished += int64(increment) } else if dataType == "workflow_executions_failed" { orgStatistics.TotalWorkflowExecutionsFailed += int64(increment) orgStatistics.MonthlyWorkflowExecutionsFailed += int64(increment) orgStatistics.WeeklyWorkflowExecutionsFailed += int64(increment) orgStatistics.DailyWorkflowExecutionsFailed += int64(increment) orgStatistics.HourlyWorkflowExecutionsFailed += int64(increment) } else if dataType == "app_executions_failed" { orgStatistics.TotalAppExecutionsFailed += int64(increment) orgStatistics.MonthlyAppExecutionsFailed += int64(increment) orgStatistics.WeeklyAppExecutionsFailed += int64(increment) orgStatistics.DailyAppExecutionsFailed += int64(increment) orgStatistics.HourlyAppExecutionsFailed += int64(increment) } else if dataType == "subflow_executions" { orgStatistics.TotalSubflowExecutions += int64(increment) orgStatistics.MonthlySubflowExecutions += int64(increment) orgStatistics.WeeklySubflowExecutions += int64(increment) orgStatistics.DailySubflowExecutions += int64(increment) orgStatistics.HourlySubflowExecutions += int64(increment) } else if dataType == "org_sync_actions" { orgStatistics.TotalOrgSyncActions += int64(increment) orgStatistics.MonthlyOrgSyncActions += int64(increment) orgStatistics.WeeklyOrgSyncActions += int64(increment) orgStatistics.DailyOrgSyncActions += int64(increment) orgStatistics.HourlyOrgSyncActions += int64(increment) } else if dataType == "workflow_executions_cloud" { orgStatistics.TotalCloudExecutions += int64(increment) orgStatistics.MonthlyCloudExecutions += int64(increment) orgStatistics.WeeklyCloudExecutions += int64(increment) orgStatistics.DailyCloudExecutions += int64(increment) orgStatistics.HourlyCloudExecutions += int64(increment) } else if dataType == "workflow_executions_onprem" { orgStatistics.TotalOnpremExecutions += int64(increment) orgStatistics.MonthlyOnpremExecutions += int64(increment) orgStatistics.WeeklyOnpremExecutions += int64(increment) orgStatistics.DailyOnpremExecutions += int64(increment) orgStatistics.HourlyOnpremExecutions += int64(increment) } else if dataType == "api_usage" { orgStatistics.TotalApiUsage += int64(increment) orgStatistics.MonthlyApiUsage += int64(increment) orgStatistics.DailyApiUsage += int64(increment) } else if dataType == "ai_executions" { orgStatistics.TotalAIUsage += int64(increment) orgStatistics.MonthlyAIUsage += int64(increment) orgStatistics.DailyAIUsage += int64(increment) } else if dataType == "agent_executions" { orgStatistics.TotalAgentExecutions += int64(increment) orgStatistics.MonthlyAgentExecutions += int64(increment) orgStatistics.DailyAgentExecutions += int64(increment) } else if dataType == "agent_tokens" { orgStatistics.TotalAgentTokens += int64(increment) orgStatistics.MonthlyAgentTokens += int64(increment) orgStatistics.DailyAgentTokens += int64(increment) } else if dataType == "agent_input_tokens" { orgStatistics.TotalAgentInputTokens += int64(increment) orgStatistics.MonthlyAgentInputTokens += int64(increment) orgStatistics.DailyAgentInputTokens += int64(increment) } else if dataType == "agent_output_tokens" { orgStatistics.TotalAgentOutputTokens += int64(increment) orgStatistics.MonthlyAgentOutputTokens += int64(increment) orgStatistics.DailyAgentOutputTokens += int64(increment) } else { //log.Printf("\n\n[ERROR] Unknown data type in stats increment for org %s: %s. Appending to custom list.\n\n", orgStatistics.OrgId, dataType) appendCustom = true } if strings.HasPrefix(dataType, "app_executions") && dataType != "app_executions" { appendCustom = true } if appendCustom { if debug { log.Printf("[DEBUG] Appending custom data type %s for org %s. Amount: %d", dataType, orgStatistics.OrgId, increment) } dataType = strings.ToLower(strings.Replace(dataType, " ", "_", -1)) found := false for additionIndex, addition := range orgStatistics.Additions { if addition.Key != dataType { continue } found = true amount := int64(increment) orgStatistics.Additions[additionIndex].Value += amount //orgStatistics.Additions[additionIndex].DailyValue += amount break } if debug { log.Printf("[DEBUG] After processing custom data type %s for org %s. Amount: %d. Found: %v", dataType, orgStatistics.OrgId, increment, found) } if !found { orgStatistics.Additions = append(orgStatistics.Additions, AdditionalUseConfig{ Key: dataType, Value: int64(increment), //DailyValue: int64(increment), //Date: 0, }) } } //send mail if the app runs more than the set threshold limit ctx := context.Background() orgId := orgStatistics.OrgId //Unmarshal the org details org, err := GetOrg(ctx, orgId) if err != nil { log.Printf("[ERROR] Failed getting org in increment: %s", err) return orgStatistics } //send mail if the app runs more than the set threshold limit emailSend := false if len(org.Id) == 0 { return orgStatistics } for _, alert := range org.Billing.AlertThreshold { found := false for _, statAlert := range orgStatistics.UsageAlerts { if statAlert.Percentage == alert.Percentage && statAlert.Count == alert.Count { found = true break } } if !found { orgStatistics.UsageAlerts = append(orgStatistics.UsageAlerts, AlertThreshold{ Percentage: alert.Percentage, Count: alert.Count, Email_send: alert.Email_send, }) } } for index, AlertThreshold := range org.Billing.AlertThreshold { totalAppExecutions := orgStatistics.MonthlyAppExecutions + orgStatistics.MonthlyChildAppExecutions // Alert should be based on the current month usage, check if monthly reset happened if yes than only send alert monthlyResetMonth := time.Now().Month() shouldSendAlert := false if orgStatistics.LastMonthlyResetMonth == int(monthlyResetMonth) { shouldSendAlert = true } sendAlert := false for _, alerts := range orgStatistics.UsageAlerts { if alerts.Percentage == AlertThreshold.Percentage && alerts.Count == AlertThreshold.Count { sendAlert = alerts.Email_send break } } if int64(AlertThreshold.Count) < totalAppExecutions && !sendAlert && shouldSendAlert { allAdmins := []string{} firstAdmin := "" allShufflerEmails := true for _, user := range org.Users { if user.Role == "admin" { allAdmins = append(allAdmins, user.Username) if firstAdmin == "" && !strings.Contains(user.Username, "shuffler.io") { firstAdmin = user.Username } if !strings.Contains(user.Username, "shuffler.io") { allShufflerEmails = false } } } if allShufflerEmails && firstAdmin == "" && len(allAdmins) > 0 { firstAdmin = allAdmins[0] } if !ArrayContains(allAdmins, "chris@shuffler.io") { allAdmins = append(allAdmins, "chris@shuffler.io") } if !ArrayContains(allAdmins, "jay@shuffler.io") { allAdmins = append(allAdmins, "jay@shuffler.io") } cacheKey := generateAlertCacheKey(orgId, AlertThreshold.Count, allAdmins) if !checkAndSetAlertCache(ctx, cacheKey) { continue } Subject := fmt.Sprintf("[Shuffle]: You've reached the app-runs threshold limit for your account %s", firstAdmin) AppRunsPercentage := float64(totalAppExecutions) / float64(org.SyncFeatures.AppExecutions.Limit) * 100 substitutions := map[string]interface{}{ "app_runs_usage": totalAppExecutions, "app_runs_limit": org.SyncFeatures.AppExecutions.Limit, "app_runs_usage_percentage": int64(AppRunsPercentage), "org_name": org.Name, "org_id": org.Id, "admin_email": firstAdmin, } err = sendMailSendgridV2( []string{"support@shuffler.io"}, Subject, substitutions, false, "d-3678d48b2b7144feb4b0b4cff7045016", allAdmins, ) if err != nil { log.Printf("[ERROR] Failed sending alert mail in increment: %s", err) } else { emailSend = true } if emailSend { org.Billing.AlertThreshold[index].Email_send = true err = SetOrg(ctx, *org, orgId) if err != nil { log.Printf("[ERROR] Failed setting org in increment: %s", err) return orgStatistics } // update the the alert send in the statistics for index, alerts := range orgStatistics.UsageAlerts { if alerts.Percentage == AlertThreshold.Percentage && alerts.Count == AlertThreshold.Count { orgStatistics.UsageAlerts[index].Email_send = true break } } log.Printf("[DEBUG] Successfully sent alert mail for org %s", orgId) } } } // hard limit aleart if org.Billing.AppRunsHardLimit > 0 && orgStatistics.MonthlyAppExecutions > org.Billing.AppRunsHardLimit { // send alert to all admin in the orgs admins := []string{} for _, user := range org.Users { if user.Role == "admin" { admins = append(admins, user.Username) } } cacheKey := generateAlertCacheKey(orgId, "hard_limit", admins) if !checkAndSetAlertCache(ctx, cacheKey) { log.Printf("[DEBUG] Skipping duplicate hard limit alert for org %s - alert sent within last minute", orgId) } else { subject := fmt.Sprintf("App Runs Hard Limit Exceeded for Org %s (%s)", org.Name, org.Id) message := fmt.Sprintf( `Dear Team, Your organization %s (ID: %s) has exceeded the monthly app runs hard limit of %d runs. Current usage: %d app runs. As a result, all workflows have been temporarily blocked until the start of the next billing cycle. To increase your organization's hard limit, please visit the admin panel of the parent organization. If you have any questions, feel free to reach out to us at support@shuffler.io. Note: This is an automated message sent by Shuffle to notify you about the exceeded app runs hard limit. Best regards, The Shuffler Team`, org.Name, org.Id, org.Billing.AppRunsHardLimit, orgStatistics.MonthlyAppExecutions, ) err = sendMailSendgrid(admins, subject, message, false, []string{}) if err != nil { log.Printf("[ERROR] Failed sending alert email to admins of org %s (%s): %s", org.Name, org.Id, err) } } } if dataType == "app_executions" || dataType == "childorg_app_executions" { validationOrg := org validationOrgStatistics := orgStatistics if len(org.CreatorOrg) > 0 { validationOrg, err = GetOrg(ctx, org.CreatorOrg) if err != nil { log.Printf("[ERROR] Failed getting parent org in increment: %s", err) return validationOrgStatistics } validationOrgStatistics, err = GetOrgStatistics(ctx, org.CreatorOrg) if err != nil { log.Printf("[ERROR] Failed getting parent org statistics in increment: %s", err) return validationOrgStatistics } } totalExecutions := float64(validationOrgStatistics.MonthlyAppExecutions) + float64(validationOrgStatistics.MonthlyChildAppExecutions) limit := float64(validationOrg.SyncFeatures.AppExecutions.Limit) percentage := (totalExecutions / limit) * 100 var currentThreshold int64 if percentage >= 50 { currentThreshold = int64((int(percentage) / 50) * 50) } monthlyResetMonth := time.Now().Month() shouldSendAlert := false if orgStatistics.LastMonthlyResetMonth == int(monthlyResetMonth) { shouldSendAlert = true } if currentThreshold >= 50 && currentThreshold > validationOrgStatistics.LastUsageAlertThreshold && shouldSendAlert { allAdmins := []string{} firstAdmin := "" allShufflerEmails := true for _, user := range org.Users { if user.Role == "admin" { allAdmins = append(allAdmins, user.Username) if firstAdmin == "" && !strings.Contains(user.Username, "shuffler.io") { firstAdmin = user.Username } if !strings.Contains(user.Username, "shuffler.io") { allShufflerEmails = false } } } if allShufflerEmails && firstAdmin == "" && len(allAdmins) > 0 { firstAdmin = allAdmins[0] } alertAlreadySet := false // If 50% and 100% alert are already set by user, and alert is send for that threshold, then skip for _, AlertThreshold := range validationOrg.Billing.AlertThreshold { if AlertThreshold.Percentage == int(currentThreshold) && AlertThreshold.Email_send { alertAlreadySet = true break } } newEmailList := []string{} if alertAlreadySet && (currentThreshold == 100 || currentThreshold == 50) { newEmailList = allAdmins } else { newEmailList = []string{"chris@shuffler.io", "jay@shuffler.io"} } // send mail use different subject line as it will sent only to the team Subject := fmt.Sprintf("[Shuffle]: You've reached the app-runs threshold limit for your account %s", firstAdmin) leadInfo := "" if validationOrg.LeadInfo.POV { leadInfo = "POC" } if validationOrg.LeadInfo.Customer { leadInfo = "Customer" } if validationOrg.LeadInfo.IntegrationPartner || validationOrg.LeadInfo.TechPartner || validationOrg.LeadInfo.DistributionPartner || validationOrg.LeadInfo.ServicePartner || validationOrg.LeadInfo.ChannelPartner { leadInfo = "Partner" } if len(leadInfo) > 0 && (currentThreshold > 100) { Subject = fmt.Sprintf("[Shuffle] %s: You've reached the app-runs threshold limit for your account %s", leadInfo, firstAdmin) } if len(leadInfo) == 0 && !ArrayContains(newEmailList, "jay@shuffler.io") { newEmailList = append(newEmailList, "jay@shuffler.io") } if len(leadInfo) == 0 && !ArrayContains(newEmailList, "chris@shuffler.io") { newEmailList = append(newEmailList, "chris@shuffler.io") } cacheKey := generateAlertCacheKey(validationOrg.Id, currentThreshold, newEmailList) if !checkAndSetAlertCache(ctx, cacheKey) { log.Printf("[DEBUG] Skipping duplicate percentage threshold alert for org %s, threshold %d%% - alert sent within last minute", validationOrg.Id, currentThreshold) } else { totalAppExecutions := validationOrgStatistics.MonthlyAppExecutions + validationOrgStatistics.MonthlyChildAppExecutions AppRunsPercentage := float64(totalAppExecutions) / float64(validationOrg.SyncFeatures.AppExecutions.Limit) * 100 substitutions := map[string]interface{}{ "app_runs_usage": totalAppExecutions, "app_runs_limit": validationOrg.SyncFeatures.AppExecutions.Limit, "app_runs_usage_percentage": int64(AppRunsPercentage), "org_name": validationOrg.Name, "org_id": validationOrg.Id, "admin_email": firstAdmin, } if currentThreshold > 100 { substitutions["lead_info"] = leadInfo } err = sendMailSendgridV2( []string{"support@shuffler.io"}, Subject, substitutions, false, "d-3678d48b2b7144feb4b0b4cff7045016", newEmailList, ) if err != nil { log.Printf("[ERROR] Failed sending alert mail for child org in increment (1): %s", err) } else { log.Printf("[DEBUG] Successfully sent alert mail for child org %s to parent org %s (1)", validationOrg.Name, validationOrg.Name) } } if (currentThreshold == 100 || currentThreshold == 50) && len(leadInfo) > 0 { secondEmailList := []string{"chris@shuffler.io", "jay@shuffler.io", "support@shuffler.io"} secondCacheKey := generateAlertCacheKey(validationOrg.Id, fmt.Sprintf("second_%d", currentThreshold), secondEmailList) if !checkAndSetAlertCache(ctx, secondCacheKey) { log.Printf("[DEBUG] Skipping duplicate second alert for org %s, threshold %d%% - alert sent within last minute", validationOrg.Id, currentThreshold) } else { if len(leadInfo) > 0 { Subject = fmt.Sprintf("[Shuffle] %s: You've reached the app-runs threshold limit for your account %s", leadInfo, firstAdmin) } totalAppExecutions := validationOrgStatistics.MonthlyAppExecutions + validationOrgStatistics.MonthlyChildAppExecutions AppRunsPercentage := float64(totalAppExecutions) / float64(validationOrg.SyncFeatures.AppExecutions.Limit) * 100 substitutions := map[string]interface{}{ "app_runs_usage": totalAppExecutions, "app_runs_limit": validationOrg.SyncFeatures.AppExecutions.Limit, "app_runs_usage_percentage": int64(AppRunsPercentage), "org_name": validationOrg.Name, "org_id": validationOrg.Id, "admin_email": firstAdmin, "lead_info": leadInfo, } log.Printf("[DEBUG] Sending second alert mail for child org %s to parent org %s (2)", validationOrg.Name, validationOrg.Name) err = sendMailSendgridV2( []string{"chris@shuffler.io", "jay@shuffler.io", "support@shuffler.io"}, Subject, substitutions, false, "d-3678d48b2b7144feb4b0b4cff7045016", []string{}, ) if err != nil { log.Printf("[ERROR] Failed sending alert mail for child org in increment (2): %s", err) } else { log.Printf("[DEBUG] Successfully sent alert mail for child org %s to parent org %s (2)", validationOrg.Name, validationOrg.Name) } } } orgStatistics.LastUsageAlertThreshold = currentThreshold } } return orgStatistics } func UpdateDetectionStats(ctx context.Context, cacheData CacheKeyData) { //log.Printf("\n\n\nDETECTION STAT UPDATE!!\n\n\n") if len(cacheData.Category) == 0 || cacheData.Category == "default" { return } if len(cacheData.OrgId) == 0 { return } // Handle Detection // We actually do this in 'shuffle-security_incidents' tho category := strings.ToLower(cacheData.Category) if category != "ticket" && category != "detection" && category != "incidents" { //if debug { // log.Printf("[WARNING] Debug: Not a detection or ticket category, skipping detection stats update for category '%s'", category) //} return } // Should we verify the data here? // Look for whether "rule" is set. mappedContent := map[string]interface{}{} err := json.Unmarshal([]byte(cacheData.Value), &mappedContent) if err != nil { log.Printf("[WARNING] Failed unmarshalling detection content for stats update: %s", err) } if mappedContent["rule"] == nil { log.Printf("[WARNING] No rule found in detection content, skipping stats update") return } ruleName := fmt.Sprintf("%v", mappedContent["rule"]) if len(ruleName) == 0 { log.Printf("[WARNING] No rule name found in detection content, skipping stats update") return } detectionStatname := fmt.Sprintf("detection_rule_%s", strings.TrimSpace(strings.ToLower(strings.ReplaceAll(ruleName, " ", "_")))) IncrementCache(ctx, cacheData.OrgId, detectionStatname, 1) if debug { log.Printf("[DEBUG] Incremented detection stat '%s' for org %s", detectionStatname, cacheData.OrgId) } }