diff --git a/.env.example b/.env.example index e2e32b9..10b16e8 100644 --- a/.env.example +++ b/.env.example @@ -18,4 +18,5 @@ SUMSUB_APP_TOKEN= SUMSUB_SECRET_KEY= SUMSUB_JWT_SECRET_KEY= -EMAIL_TEMPLATES_PATH="./templates/html/" \ No newline at end of file +EMAIL_TEMPLATES_PATH="./templates/html/" +EE_CHAINSTORE_API_URL= diff --git a/cmd/main.go b/cmd/main.go index ad02121..03bd828 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -86,6 +86,19 @@ func startApi(ctx *cli.Context) error { c.Start() } + offlineNodeTiming, found := config.Config.GetOfflineNodesCronJobTiming(nodeAddress) + if found { + if err := service.ValidateOfflineNodesNotifierConfig(); err != nil { + return errors.New("invalid offline nodes notifier config: " + err.Error()) + } + c := cron.New(cron.WithChain(cron.SkipIfStillRunning(cron.DefaultLogger))) + _, err = c.AddFunc(offlineNodeTiming, service.NotifyOfflineLinkedNodes) + if err != nil { + return errors.New("error while starting offline nodes cronjob: " + err.Error()) + } + c.Start() + } + monthlyNodeTiming, found := config.Config.GetMonthlyCronJobTiming(nodeAddress) if found { c := cron.New() diff --git a/config/config.devnet.json b/config/config.devnet.json index 7f93e3b..59d7ae0 100644 --- a/config/config.devnet.json +++ b/config/config.devnet.json @@ -62,6 +62,7 @@ ] }, "DeeployApi": "https://devnet-deeploy-api.ratio1.ai/get_oracle_job_details", + "OraclesApi": "https://devnet-oracle.ratio1.ai", "CronJobTiming": {}, "ChainID": 84532, "BuyLimitUSD": { diff --git a/config/config.go b/config/config.go index 6e481fc..8c2fafc 100644 --- a/config/config.go +++ b/config/config.go @@ -29,6 +29,7 @@ type GeneralConfig struct { Oblio Oblio Infura Infura DeeployApi string + OraclesApi string NDContractAddress string R1ContractAddress string USDCContractAddress string @@ -38,6 +39,7 @@ type GeneralConfig struct { TeamAddresses []string BuyLicenseInvoiceCronJobTiming map[string]string DailyCronJobTiming map[string]string + OfflineNodesCronJobTiming map[string]string MonthlyCronJobTiming map[string]string AdminAddresses []string EmailTemplatesPath string @@ -292,3 +294,8 @@ func (c *GeneralConfig) GetMonthlyCronJobTiming(nodeAddress string) (string, boo nodeTiming, found := c.MonthlyCronJobTiming[nodeAddress] return nodeTiming, found } + +func (c *GeneralConfig) GetOfflineNodesCronJobTiming(nodeAddress string) (string, bool) { + nodeTiming, found := c.OfflineNodesCronJobTiming[nodeAddress] + return nodeTiming, found +} diff --git a/config/config.mainnet.json b/config/config.mainnet.json index 592ffb5..b14bc8a 100644 --- a/config/config.mainnet.json +++ b/config/config.mainnet.json @@ -96,6 +96,14 @@ "0x2539fDD57f93b267E58d5f2E6F77063C0230F6F4": "40 * * * *", "0xdc4fDFd5B86aeA7BaB17d4742B7c39A2728Ff59B": "50 * * * *" }, + "OfflineNodesCronJobTiming": { + "0xe240d9cf8893d6bE9fb3Ac4C9CE1E504343b64a0": "05 * * * *", + "0xA48C80afb9eC73A9F6731a9Cea6774Ed6249a922": "15 * * * *", + "0x3B373897136687af87DDbF65662Eb7004090eAF8": "25 * * * *", + "0x7388fC301eb7CF6743ecF9e6c781210758669bAD": "35 * * * *", + "0x2539fDD57f93b267E58d5f2E6F77063C0230F6F4": "45 * * * *", + "0xdc4fDFd5B86aeA7BaB17d4742B7c39A2728Ff59B": "55 * * * *" + }, "MonthlyCronJobTiming": { "0xe240d9cf8893d6bE9fb3Ac4C9CE1E504343b64a0": "0 08 1 * *", "0xA48C80afb9eC73A9F6731a9Cea6774Ed6249a922": "0 10 1 * *", @@ -114,6 +122,7 @@ "CspUrl": "https://deeploy.ratio1.ai/account?tab=invoicing" }, "DeeployApi": "https://deeploy-api.ratio1.ai/get_oracle_job_details", + "OraclesApi": "https://oracle.ratio1.ai", "ViesApi": { "BaseUrl": "https://viesapi.eu/api" } diff --git a/config/config.testnet.json b/config/config.testnet.json index 2b1799e..be6868f 100644 --- a/config/config.testnet.json +++ b/config/config.testnet.json @@ -62,6 +62,7 @@ ] }, "DeeployApi": "https://testnet-deeploy-api.ratio1.ai/get_oracle_job_details", + "OraclesApi": "https://testnet-oracle.ratio1.ai", "CronJobTiming": {}, "ChainID": 84532, "BuyLimitUSD": { diff --git a/deployScripts/deploy_devnet.py b/deployScripts/deploy_devnet.py index 9adc45a..002b31b 100644 --- a/deployScripts/deploy_devnet.py +++ b/deployScripts/deploy_devnet.py @@ -42,6 +42,7 @@ 'SUMSUB_JWT_SECRET_KEY':'', 'EMAIL_TEMPLATES_PATH':'./templates/html/', + 'EE_CHAINSTORE_API_URL':'', } # Defining the assets of the application. @@ -73,4 +74,4 @@ session.run( wait=30, # wait for the user to stop the execution close_pipelines=True # when the user stops the execution, the remote edge-node pipelines will be closed - ) \ No newline at end of file + ) diff --git a/deployScripts/deploy_mainnet.py b/deployScripts/deploy_mainnet.py index 6d61132..d3c64e6 100644 --- a/deployScripts/deploy_mainnet.py +++ b/deployScripts/deploy_mainnet.py @@ -50,6 +50,7 @@ 'INFURA_SECRET':'', 'EMAIL_TEMPLATES_PATH':'./templates/html/', + 'EE_CHAINSTORE_API_URL':'', } # Defining the assets of the application. @@ -81,4 +82,4 @@ session.run( wait=True, # wait for the user to stop the execution close_pipelines=False # when the user stops the execution, the remote edge-node pipelines will be closed - ) \ No newline at end of file + ) diff --git a/deployScripts/deploy_testnet.py b/deployScripts/deploy_testnet.py index 9adc45a..002b31b 100644 --- a/deployScripts/deploy_testnet.py +++ b/deployScripts/deploy_testnet.py @@ -42,6 +42,7 @@ 'SUMSUB_JWT_SECRET_KEY':'', 'EMAIL_TEMPLATES_PATH':'./templates/html/', + 'EE_CHAINSTORE_API_URL':'', } # Defining the assets of the application. @@ -73,4 +74,4 @@ session.run( wait=30, # wait for the user to stop the execution close_pipelines=True # when the user stops the execution, the remote edge-node pipelines will be closed - ) \ No newline at end of file + ) diff --git a/service/emailService.go b/service/emailService.go index ea6f1f6..883ba48 100644 --- a/service/emailService.go +++ b/service/emailService.go @@ -26,6 +26,7 @@ const ( subjectNewBuyLicenseInvoice = "A new buy license invoice has been sent" subjectNewInvoiceDraft = "New draft invoices have been issued" subjectJobsEndingSoon = "Ratio1 - Jobs ending soon" + subjectNodesOffline = "Ratio1 - Linked nodes offline for more than 24h" ) var ( @@ -239,6 +240,64 @@ func SendJobsEndingEmail(email string, jobs []EndingJob) error { return callSendEmail(email, subjectJobsEndingSoon, body.String()) } +func SendOfflineNodesEmail(email string, nodes []OfflineNodeAlert) error { + if len(nodes) == 0 { + return nil + } + + tmpl, err := templates.GetNodesOfflineEmailTemplate() + if err != nil { + return errors.New("error while retrieving email template: " + err.Error()) + } + + type offlineNodeTemplateRow struct { + Alias string + Address string + OfflineFor string + } + + rows := make([]offlineNodeTemplateRow, 0, len(nodes)) + for _, node := range nodes { + rows = append(rows, offlineNodeTemplateRow{ + Alias: node.NodeAlias, + Address: node.NodeAddress, + OfflineFor: formatOfflineDuration(node.OfflineSeconds), + }) + } + + var body bytes.Buffer + err = tmpl.Execute(&body, struct { + Nodes []offlineNodeTemplateRow + NodesCount int + }{ + Nodes: rows, + NodesCount: len(rows), + }) + if err != nil { + return errors.New("error while executing email template: " + err.Error()) + } + + return callSendEmail(email, subjectNodesOffline, body.String()) +} + +func formatOfflineDuration(seconds int64) string { + if seconds <= 0 { + return "0m" + } + + days := seconds / 86400 + hours := (seconds % 86400) / 3600 + minutes := (seconds % 3600) / 60 + + if days > 0 { + return fmt.Sprintf("%dd %dh %dm", days, hours, minutes) + } + if hours > 0 { + return fmt.Sprintf("%dh %dm", hours, minutes) + } + return fmt.Sprintf("%dm", minutes) +} + func callSendTextEmail(email, subject, text string) error { msg := EmailMessage{ From: config.Config.Mail.FromEmail, diff --git a/service/offlineNodesNotifier.go b/service/offlineNodesNotifier.go new file mode 100644 index 0000000..6f75fa8 --- /dev/null +++ b/service/offlineNodesNotifier.go @@ -0,0 +1,419 @@ +package service + +import ( + "context" + "errors" + "fmt" + "math" + "os" + "sort" + "strconv" + "strings" + "time" + + "github.com/NaeuralEdgeProtocol/ratio1-backend/config" + "github.com/NaeuralEdgeProtocol/ratio1-backend/process" + "github.com/NaeuralEdgeProtocol/ratio1-backend/storage" + "github.com/Ratio1/edge_sdk_go/pkg/cstore" +) + +const ( + offlineNodeThreshold = 24 * time.Hour + ownerNotificationInterval = 24 * time.Hour + offlineNodesCStoreHashKey = "ratio1:backend:offline_nodes_notifier:v1" + offlineNotifierStoreTimeout = 30 * time.Second +) + +var ( + fetchOracleNodesListFn = fetchOracleNodesList + resolveNodeOwnersFn = getNodeOwners + getAccountByAddressFn = storage.GetAccountByAddress + getNotificationEmailFn = storage.GetAccountNotificationEmailByAddress + sendOfflineNodesEmailFn = SendOfflineNodesEmail + newCStoreClientFromEnvFn = cstore.NewFromEnv + newOwnerNotificationStoreFn = newCStoreOwnerNotificationStore + oracleHTTPGetFn = process.HttpGet +) + +type OfflineNodeAlert struct { + NodeAlias string + NodeAddress string + OfflineSeconds int64 +} + +func ValidateOfflineNodesNotifierConfig() error { + if strings.TrimSpace(config.Config.OraclesApi) == "" { + return errors.New("oracles api url is not configured") + } + if strings.TrimSpace(os.Getenv("EE_CHAINSTORE_API_URL")) == "" { + return errors.New("EE_CHAINSTORE_API_URL is not set") + } + if _, err := newCStoreClientFromEnvFn(); err != nil { + return errors.New("invalid CStore configuration: " + err.Error()) + } + return nil +} + +type oracleNodesListResponse struct { + Result oracleNodesListResult `json:"result"` +} + +type oracleNodesListResult struct { + Nodes map[string]oracleNodeRaw `json:"nodes"` + Error any `json:"error"` +} + +type oracleNodeRaw struct { + Alias string `json:"alias"` + EthAddress string `json:"eth_address"` + EthAddr string `json:"eth_addr"` + LastState string `json:"last_state"` + LastSeenAgo any `json:"last_seen_ago"` + NodeIsOnline *bool `json:"node_is_online"` + IsOnline *bool `json:"is_online"` +} + +type ownerNotificationStore interface { + Sync(ctx context.Context) error + LastSent(ctx context.Context, ownerAddress string) (time.Time, bool, error) + SetLastSent(ctx context.Context, ownerAddress string, sentAt time.Time) error +} + +type cstoreOwnerNotificationStore struct { + client *cstore.Client +} + +type cstoreOwnerLastSent struct { + LastSentUnix int64 `json:"last_sent_unix"` +} + +func NotifyOfflineLinkedNodes() { + offlineNodes, err := fetchOracleNodesListFn() + if err != nil { + log.Error("offline nodes notifier failed to fetch oracle nodes list: %s", err.Error()) + return + } + + if len(offlineNodes) == 0 { + return + } + + store, err := newOwnerNotificationStoreFn() + if err != nil { + log.Error("offline nodes notifier failed to initialize cstore: %s", err.Error()) + return + } + + syncCtx, cancelSync := context.WithTimeout(context.Background(), offlineNotifierStoreTimeout) + err = store.Sync(syncCtx) + cancelSync() + if err != nil { + log.Error("offline nodes notifier failed to sync cstore state: %s", err.Error()) + return + } + + nodeAddresses := make([]string, 0, len(offlineNodes)) + nodeByAddress := make(map[string]OfflineNodeAlert, len(offlineNodes)) + for _, node := range offlineNodes { + address := strings.ToLower(strings.TrimSpace(node.NodeAddress)) + if address == "" { + continue + } + nodeAddresses = append(nodeAddresses, address) + nodeByAddress[address] = node + } + if len(nodeAddresses) == 0 { + return + } + + ownersByNode, err := resolveNodeOwnersFn(nodeAddresses) + if err != nil { + log.Error("offline nodes notifier failed to resolve node owners: %s", err.Error()) + return + } + + nodesByOwner := make(map[string][]OfflineNodeAlert) + for nodeAddress, ownerAddress := range ownersByNode { + owner := strings.TrimSpace(ownerAddress) + if owner == "" || strings.EqualFold(owner, "0x0000000000000000000000000000000000000000") { + continue + } + node, found := nodeByAddress[strings.ToLower(strings.TrimSpace(nodeAddress))] + if !found { + continue + } + nodesByOwner[owner] = append(nodesByOwner[owner], node) + } + + now := time.Now().UTC() + for ownerAddress, ownerNodes := range nodesByOwner { + emails, err := getConfirmedAccountEmails(ownerAddress) + if err != nil { + log.Error("offline nodes notifier failed account lookup for %s: %s", ownerAddress, err.Error()) + continue + } + if len(emails) == 0 { + continue + } + + sort.Slice(ownerNodes, func(i, j int) bool { + left := ownerNodes[i] + right := ownerNodes[j] + if left.OfflineSeconds != right.OfflineSeconds { + return left.OfflineSeconds > right.OfflineSeconds + } + return strings.Compare(left.NodeAddress, right.NodeAddress) < 0 + }) + + lastSentCtx, cancelLastSent := context.WithTimeout(context.Background(), offlineNotifierStoreTimeout) + lastSent, hasLastSent, err := store.LastSent(lastSentCtx, strings.ToLower(ownerAddress)) + cancelLastSent() + if err != nil { + log.Error("offline nodes notifier failed cstore read for %s: %s", ownerAddress, err.Error()) + continue + } + + if hasLastSent && now.Sub(lastSent) < ownerNotificationInterval { + continue + } + + sentCount := 0 + for _, email := range emails { + err = sendOfflineNodesEmailFn(email, ownerNodes) + if err != nil { + log.Error("offline nodes notifier failed to send email to %s: %s", email, err.Error()) + continue + } + sentCount++ + } + if sentCount == 0 { + continue + } + + setLastSentCtx, cancelSetLastSent := context.WithTimeout(context.Background(), offlineNotifierStoreTimeout) + err = store.SetLastSent(setLastSentCtx, strings.ToLower(ownerAddress), now) + cancelSetLastSent() + if err != nil { + log.Error("offline nodes notifier failed cstore write for %s: %s", ownerAddress, err.Error()) + continue + } + } +} + +func fetchOracleNodesList() ([]OfflineNodeAlert, error) { + url := strings.TrimSuffix(strings.TrimSpace(config.Config.OraclesApi), "/") + "/nodes_list" + if strings.TrimSpace(config.Config.OraclesApi) == "" { + return nil, errors.New("oracles api url is not configured") + } + + var response oracleNodesListResponse + err := oracleHTTPGetFn(url, &response) + if err != nil { + return nil, errors.New("error while requesting nodes list: " + err.Error()) + } + + if response.Result.Error != nil { + errMsg := strings.TrimSpace(fmt.Sprintf("%v", response.Result.Error)) + if errMsg != "" && errMsg != "" { + return nil, errors.New("oracles api returned error: " + errMsg) + } + } + + if len(response.Result.Nodes) == 0 { + return nil, nil + } + + offlineNodes := make([]OfflineNodeAlert, 0) + for _, node := range response.Result.Nodes { + ethAddress := strings.TrimSpace(node.EthAddress) + if ethAddress == "" { + ethAddress = strings.TrimSpace(node.EthAddr) + } + if ethAddress == "" { + continue + } + + isOffline, hasState := parseNodeOfflineState(node) + if !hasState || !isOffline { + continue + } + + lastSeenSeconds, ok := parseLastSeenSeconds(node.LastSeenAgo) + if !ok { + continue + } + if lastSeenSeconds <= int64(offlineNodeThreshold.Seconds()) { + continue + } + + alias := strings.TrimSpace(node.Alias) + if alias == "" { + alias = "unknown" + } + + offlineNodes = append(offlineNodes, OfflineNodeAlert{ + NodeAlias: alias, + NodeAddress: ethAddress, + OfflineSeconds: lastSeenSeconds, + }) + } + + return offlineNodes, nil +} + +func parseNodeOfflineState(node oracleNodeRaw) (bool, bool) { + if node.NodeIsOnline != nil { + return !(*node.NodeIsOnline), true + } + if node.IsOnline != nil { + return !(*node.IsOnline), true + } + state := strings.TrimSpace(strings.ToLower(node.LastState)) + switch state { + case "offline": + return true, true + case "online": + return false, true + default: + return false, false + } +} + +func parseLastSeenSeconds(raw any) (int64, bool) { + switch v := raw.(type) { + case float64: + if math.IsNaN(v) || math.IsInf(v, 0) { + return 0, false + } + return int64(v), true + case float32: + if math.IsNaN(float64(v)) || math.IsInf(float64(v), 0) { + return 0, false + } + return int64(v), true + case int: + return int64(v), true + case int8: + return int64(v), true + case int16: + return int64(v), true + case int32: + return int64(v), true + case int64: + return v, true + case uint: + return int64(v), true + case uint8: + return int64(v), true + case uint16: + return int64(v), true + case uint32: + return int64(v), true + case uint64: + if v > math.MaxInt64 { + return 0, false + } + return int64(v), true + case string: + return parseLastSeenSecondsFromString(v) + default: + return 0, false + } +} + +func parseLastSeenSecondsFromString(raw string) (int64, bool) { + s := strings.TrimSpace(raw) + if s == "" { + return 0, false + } + + if numericValue, err := strconv.ParseFloat(s, 64); err == nil { + return int64(numericValue), true + } + + if durationValue, err := time.ParseDuration(s); err == nil { + return int64(durationValue.Seconds()), true + } + + if strings.Count(s, ":") == 2 || strings.Count(s, ":") == 1 { + parts := strings.Split(s, ":") + total := int64(0) + multiplier := int64(1) + for i := len(parts) - 1; i >= 0; i-- { + part := strings.TrimSpace(parts[i]) + component, err := strconv.ParseInt(part, 10, 64) + if err != nil { + return 0, false + } + total += component * multiplier + multiplier *= 60 + } + return total, true + } + + return 0, false +} + +func getConfirmedAccountEmails(ownerAddress string) ([]string, error) { + address := strings.TrimSpace(ownerAddress) + if address == "" { + return nil, nil + } + + account, found, err := getAccountByAddressFn(address) + if err != nil { + return nil, err + } + if !found && strings.ToLower(address) != address { + account, found, err = getAccountByAddressFn(strings.ToLower(address)) + if err != nil { + return nil, err + } + } + if !found || account == nil { + return nil, nil + } + + notificationEmail, found, err := getNotificationEmailFn(account.Address) + if err != nil { + return nil, err + } + if !found { + notificationEmail = nil + } + + return notificationEmailsForAccount(account, notificationEmail), nil +} + +func newCStoreOwnerNotificationStore() (ownerNotificationStore, error) { + client, err := newCStoreClientFromEnvFn() + if err != nil { + return nil, err + } + return &cstoreOwnerNotificationStore{ + client: client, + }, nil +} + +func (s *cstoreOwnerNotificationStore) Sync(ctx context.Context) error { + _, err := s.client.GetStatus(ctx) + return err +} + +func (s *cstoreOwnerNotificationStore) LastSent(ctx context.Context, ownerAddress string) (time.Time, bool, error) { + var entry cstoreOwnerLastSent + item, err := s.client.HGet(ctx, offlineNodesCStoreHashKey, ownerAddress, &entry) + if err != nil { + return time.Time{}, false, err + } + if item == nil || entry.LastSentUnix <= 0 { + return time.Time{}, false, nil + } + return time.Unix(entry.LastSentUnix, 0).UTC(), true, nil +} + +func (s *cstoreOwnerNotificationStore) SetLastSent(ctx context.Context, ownerAddress string, sentAt time.Time) error { + return s.client.HSet(ctx, offlineNodesCStoreHashKey, ownerAddress, cstoreOwnerLastSent{ + LastSentUnix: sentAt.UTC().Unix(), + }, nil) +} diff --git a/service/offlineNodesNotifier_test.go b/service/offlineNodesNotifier_test.go new file mode 100644 index 0000000..42ec3ed --- /dev/null +++ b/service/offlineNodesNotifier_test.go @@ -0,0 +1,423 @@ +package service + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/NaeuralEdgeProtocol/ratio1-backend/config" + "github.com/NaeuralEdgeProtocol/ratio1-backend/model" + "github.com/NaeuralEdgeProtocol/ratio1-backend/process" + "github.com/stretchr/testify/require" +) + +func boolPtr(value bool) *bool { + return &value +} + +type mockOwnerNotificationStore struct { + syncErr error + lastSentByKey map[string]time.Time + lastSentErr error + failLastSentNow bool + setErr error + failSetNow bool + setCalls int +} + +func (m *mockOwnerNotificationStore) Sync(ctx context.Context) error { + return m.syncErr +} + +func (m *mockOwnerNotificationStore) LastSent(ctx context.Context, ownerAddress string) (time.Time, bool, error) { + if m.failLastSentNow { + m.failLastSentNow = false + return time.Time{}, false, errors.New("last sent unavailable") + } + if m.lastSentErr != nil { + return time.Time{}, false, m.lastSentErr + } + value, found := m.lastSentByKey[ownerAddress] + return value, found, nil +} + +func (m *mockOwnerNotificationStore) SetLastSent(ctx context.Context, ownerAddress string, sentAt time.Time) error { + if m.failSetNow { + m.failSetNow = false + return errors.New("set unavailable") + } + if m.setErr != nil { + return m.setErr + } + m.setCalls++ + if m.lastSentByKey == nil { + m.lastSentByKey = make(map[string]time.Time) + } + m.lastSentByKey[ownerAddress] = sentAt + return nil +} + +func TestFetchOracleNodesListFiltersOfflineNodes(t *testing.T) { + previousHTTPGet := oracleHTTPGetFn + defer func() { + oracleHTTPGetFn = previousHTTPGet + }() + + oracleHTTPGetFn = func(url string, castTarget interface{}, headers ...process.HttpHeaderPair) error { + require.Equal(t, "https://oracle.test/nodes_list", url) + response := castTarget.(*oracleNodesListResponse) + response.Result.Nodes = map[string]oracleNodeRaw{ + "node1": {Alias: "alpha", EthAddr: "0x1111111111111111111111111111111111111111", IsOnline: boolPtr(false), LastState: "2026-06-08 10:00:00", LastSeenAgo: "25:01:01"}, + "node2": {Alias: "beta", EthAddr: "0x2222222222222222222222222222222222222222", IsOnline: boolPtr(false), LastState: "2026-06-08 12:00:00", LastSeenAgo: "24:00:00"}, + "node3": {Alias: "gamma", EthAddr: "0x3333333333333333333333333333333333333333", IsOnline: boolPtr(true), LastState: "2026-06-09 10:00:00", LastSeenAgo: 99999}, + "node4": {Alias: "delta", EthAddr: "0x4444444444444444444444444444444444444444", LastSeenAgo: 92000}, + } + return nil + } + + oldOraclesAPI := config.Config.OraclesApi + config.Config.OraclesApi = "https://oracle.test" + defer func() { + config.Config.OraclesApi = oldOraclesAPI + }() + + nodes, err := fetchOracleNodesList() + require.NoError(t, err) + require.Len(t, nodes, 1) + require.Equal(t, "alpha", nodes[0].NodeAlias) + require.Equal(t, "0x1111111111111111111111111111111111111111", nodes[0].NodeAddress) + require.True(t, nodes[0].OfflineSeconds > int64(offlineNodeThreshold.Seconds())) +} + +func TestValidateOfflineNodesNotifierConfigRequiresCStore(t *testing.T) { + previousNewCStoreClient := newCStoreClientFromEnvFn + defer func() { + newCStoreClientFromEnvFn = previousNewCStoreClient + }() + + oldOraclesAPI := config.Config.OraclesApi + config.Config.OraclesApi = "https://oracle.test" + defer func() { + config.Config.OraclesApi = oldOraclesAPI + }() + + t.Setenv("EE_CHAINSTORE_API_URL", "") + err := ValidateOfflineNodesNotifierConfig() + require.ErrorContains(t, err, "EE_CHAINSTORE_API_URL is not set") +} + +func TestNotifyOfflineLinkedNodesFailClosedWhenStoreInitFails(t *testing.T) { + previousFetch := fetchOracleNodesListFn + previousStore := newOwnerNotificationStoreFn + previousResolve := resolveNodeOwnersFn + previousGetAccount := getAccountByAddressFn + previousGetNotificationEmail := getNotificationEmailFn + previousSend := sendOfflineNodesEmailFn + defer func() { + fetchOracleNodesListFn = previousFetch + newOwnerNotificationStoreFn = previousStore + resolveNodeOwnersFn = previousResolve + getAccountByAddressFn = previousGetAccount + getNotificationEmailFn = previousGetNotificationEmail + sendOfflineNodesEmailFn = previousSend + }() + + fetchOracleNodesListFn = func() ([]OfflineNodeAlert, error) { + return []OfflineNodeAlert{ + {NodeAlias: "alpha", NodeAddress: "0x1111111111111111111111111111111111111111", OfflineSeconds: 100000}, + }, nil + } + newOwnerNotificationStoreFn = func() (ownerNotificationStore, error) { + return nil, errors.New("store unavailable") + } + + resolveCalled := false + resolveNodeOwnersFn = func(nodes []string) (map[string]string, error) { + resolveCalled = true + return map[string]string{}, nil + } + + sendCalled := false + sendOfflineNodesEmailFn = func(email string, nodes []OfflineNodeAlert) error { + sendCalled = true + return nil + } + + NotifyOfflineLinkedNodes() + + require.False(t, resolveCalled) + require.False(t, sendCalled) +} + +func TestNotifyOfflineLinkedNodesGroupsAndThrottlesOwners(t *testing.T) { + previousFetch := fetchOracleNodesListFn + previousStore := newOwnerNotificationStoreFn + previousResolve := resolveNodeOwnersFn + previousGetAccount := getAccountByAddressFn + previousGetNotificationEmail := getNotificationEmailFn + previousSend := sendOfflineNodesEmailFn + defer func() { + fetchOracleNodesListFn = previousFetch + newOwnerNotificationStoreFn = previousStore + resolveNodeOwnersFn = previousResolve + getAccountByAddressFn = previousGetAccount + getNotificationEmailFn = previousGetNotificationEmail + sendOfflineNodesEmailFn = previousSend + }() + + fetchOracleNodesListFn = func() ([]OfflineNodeAlert, error) { + return []OfflineNodeAlert{ + {NodeAlias: "alpha", NodeAddress: "0x1111111111111111111111111111111111111111", OfflineSeconds: 100000}, + {NodeAlias: "beta", NodeAddress: "0x2222222222222222222222222222222222222222", OfflineSeconds: 110000}, + {NodeAlias: "gamma", NodeAddress: "0x3333333333333333333333333333333333333333", OfflineSeconds: 120000}, + }, nil + } + + mockStore := &mockOwnerNotificationStore{ + lastSentByKey: map[string]time.Time{ + "0xowner2": time.Now().UTC().Add(-2 * time.Hour), + }, + } + newOwnerNotificationStoreFn = func() (ownerNotificationStore, error) { + return mockStore, nil + } + + resolvedAddresses := make([]string, 0) + resolveNodeOwnersFn = func(nodes []string) (map[string]string, error) { + resolvedAddresses = append(resolvedAddresses, nodes...) + return map[string]string{ + "0x1111111111111111111111111111111111111111": "0xowner1", + "0x2222222222222222222222222222222222222222": "0xowner1", + "0x3333333333333333333333333333333333333333": "0xowner2", + }, nil + } + + getAccountByAddressFn = func(address string) (*model.Account, bool, error) { + switch address { + case "0xowner1": + email := "owner1@example.com" + return &model.Account{ + Address: address, + Email: &email, + EmailConfirmed: true, + }, true, nil + case "0xowner2": + email := "owner2@example.com" + return &model.Account{ + Address: address, + Email: &email, + EmailConfirmed: true, + }, true, nil + default: + return nil, false, nil + } + } + + getNotificationEmailFn = func(address string) (*model.AccountNotificationEmail, bool, error) { + if address != "0xowner1" { + return nil, false, nil + } + email := "alerts@example.com" + return &model.AccountNotificationEmail{ + AccountAddress: address, + Email: &email, + EmailConfirmed: true, + }, true, nil + } + + sendCalls := 0 + var sentEmails []string + var sentNodes []OfflineNodeAlert + sendOfflineNodesEmailFn = func(email string, nodes []OfflineNodeAlert) error { + sendCalls++ + sentEmails = append(sentEmails, email) + sentNodes = append([]OfflineNodeAlert(nil), nodes...) + return nil + } + + NotifyOfflineLinkedNodes() + + require.Len(t, resolvedAddresses, 3) + require.Equal(t, 2, sendCalls) + require.ElementsMatch(t, []string{"owner1@example.com", "alerts@example.com"}, sentEmails) + require.Len(t, sentNodes, 2) + require.Equal(t, 1, mockStore.setCalls) +} + +func TestNotifyOfflineLinkedNodesSkipsSetWhenEmailFails(t *testing.T) { + previousFetch := fetchOracleNodesListFn + previousStore := newOwnerNotificationStoreFn + previousResolve := resolveNodeOwnersFn + previousGetAccount := getAccountByAddressFn + previousGetNotificationEmail := getNotificationEmailFn + previousSend := sendOfflineNodesEmailFn + defer func() { + fetchOracleNodesListFn = previousFetch + newOwnerNotificationStoreFn = previousStore + resolveNodeOwnersFn = previousResolve + getAccountByAddressFn = previousGetAccount + getNotificationEmailFn = previousGetNotificationEmail + sendOfflineNodesEmailFn = previousSend + }() + + fetchOracleNodesListFn = func() ([]OfflineNodeAlert, error) { + return []OfflineNodeAlert{ + {NodeAlias: "alpha", NodeAddress: "0x1111111111111111111111111111111111111111", OfflineSeconds: 100000}, + }, nil + } + + mockStore := &mockOwnerNotificationStore{ + lastSentByKey: map[string]time.Time{}, + } + newOwnerNotificationStoreFn = func() (ownerNotificationStore, error) { + return mockStore, nil + } + + resolveNodeOwnersFn = func(nodes []string) (map[string]string, error) { + return map[string]string{ + "0x1111111111111111111111111111111111111111": "0xowner1", + }, nil + } + + getAccountByAddressFn = func(address string) (*model.Account, bool, error) { + email := "owner1@example.com" + return &model.Account{ + Address: address, + Email: &email, + EmailConfirmed: true, + }, true, nil + } + + sendOfflineNodesEmailFn = func(email string, nodes []OfflineNodeAlert) error { + return errors.New("email send failed") + } + + NotifyOfflineLinkedNodes() + + require.Equal(t, 0, mockStore.setCalls) +} + +func TestNotifyOfflineLinkedNodesContinuesAfterOwnerLastSentError(t *testing.T) { + previousFetch := fetchOracleNodesListFn + previousStore := newOwnerNotificationStoreFn + previousResolve := resolveNodeOwnersFn + previousGetAccount := getAccountByAddressFn + previousGetNotificationEmail := getNotificationEmailFn + previousSend := sendOfflineNodesEmailFn + defer func() { + fetchOracleNodesListFn = previousFetch + newOwnerNotificationStoreFn = previousStore + resolveNodeOwnersFn = previousResolve + getAccountByAddressFn = previousGetAccount + getNotificationEmailFn = previousGetNotificationEmail + sendOfflineNodesEmailFn = previousSend + }() + + fetchOracleNodesListFn = func() ([]OfflineNodeAlert, error) { + return []OfflineNodeAlert{ + {NodeAlias: "alpha", NodeAddress: "0x1111111111111111111111111111111111111111", OfflineSeconds: 100000}, + {NodeAlias: "beta", NodeAddress: "0x2222222222222222222222222222222222222222", OfflineSeconds: 110000}, + }, nil + } + + mockStore := &mockOwnerNotificationStore{ + lastSentByKey: map[string]time.Time{}, + failLastSentNow: true, + } + newOwnerNotificationStoreFn = func() (ownerNotificationStore, error) { + return mockStore, nil + } + + resolveNodeOwnersFn = func(nodes []string) (map[string]string, error) { + return map[string]string{ + "0x1111111111111111111111111111111111111111": "0xowner1", + "0x2222222222222222222222222222222222222222": "0xowner2", + }, nil + } + + getAccountByAddressFn = func(address string) (*model.Account, bool, error) { + email := address + "@example.com" + return &model.Account{ + Address: address, + Email: &email, + EmailConfirmed: true, + }, true, nil + } + getNotificationEmailFn = func(address string) (*model.AccountNotificationEmail, bool, error) { + return nil, false, nil + } + + sendCalls := 0 + sendOfflineNodesEmailFn = func(email string, nodes []OfflineNodeAlert) error { + sendCalls++ + return nil + } + + NotifyOfflineLinkedNodes() + + require.Equal(t, 1, sendCalls) + require.Equal(t, 1, mockStore.setCalls) +} + +func TestNotifyOfflineLinkedNodesContinuesAfterOwnerSetLastSentError(t *testing.T) { + previousFetch := fetchOracleNodesListFn + previousStore := newOwnerNotificationStoreFn + previousResolve := resolveNodeOwnersFn + previousGetAccount := getAccountByAddressFn + previousGetNotificationEmail := getNotificationEmailFn + previousSend := sendOfflineNodesEmailFn + defer func() { + fetchOracleNodesListFn = previousFetch + newOwnerNotificationStoreFn = previousStore + resolveNodeOwnersFn = previousResolve + getAccountByAddressFn = previousGetAccount + getNotificationEmailFn = previousGetNotificationEmail + sendOfflineNodesEmailFn = previousSend + }() + + fetchOracleNodesListFn = func() ([]OfflineNodeAlert, error) { + return []OfflineNodeAlert{ + {NodeAlias: "alpha", NodeAddress: "0x1111111111111111111111111111111111111111", OfflineSeconds: 100000}, + {NodeAlias: "beta", NodeAddress: "0x2222222222222222222222222222222222222222", OfflineSeconds: 110000}, + }, nil + } + + mockStore := &mockOwnerNotificationStore{ + lastSentByKey: map[string]time.Time{}, + failSetNow: true, + } + newOwnerNotificationStoreFn = func() (ownerNotificationStore, error) { + return mockStore, nil + } + + resolveNodeOwnersFn = func(nodes []string) (map[string]string, error) { + return map[string]string{ + "0x1111111111111111111111111111111111111111": "0xowner1", + "0x2222222222222222222222222222222222222222": "0xowner2", + }, nil + } + + getAccountByAddressFn = func(address string) (*model.Account, bool, error) { + email := address + "@example.com" + return &model.Account{ + Address: address, + Email: &email, + EmailConfirmed: true, + }, true, nil + } + getNotificationEmailFn = func(address string) (*model.AccountNotificationEmail, bool, error) { + return nil, false, nil + } + + sendCalls := 0 + sendOfflineNodesEmailFn = func(email string, nodes []OfflineNodeAlert) error { + sendCalls++ + return nil + } + + NotifyOfflineLinkedNodes() + + require.Equal(t, 2, sendCalls) + require.Equal(t, 1, mockStore.setCalls) +} diff --git a/service/statsService.go b/service/statsService.go index e799d4a..eca3929 100644 --- a/service/statsService.go +++ b/service/statsService.go @@ -128,9 +128,20 @@ func DailyGetStats() { allJobsDetails[a.JobId] = nil } + jobIDs := make([]string, 0, len(allJobsDetails)) for k := range allJobsDetails { - prevAlloc, err := storage.GetAllocationByJobIDForJobDetails(k) - if err != nil && prevAlloc == nil { + jobIDs = append(jobIDs, k) + } + + prevAllocations, err := storage.GetAllocationsByJobIDsForJobDetails(jobIDs) + if err != nil { + fmt.Println("error getting allocations for job details: " + err.Error()) + return + } + + for k := range allJobsDetails { + prevAlloc, ok := prevAllocations[k] + if !ok { res, err := GetJobDetails(k, config.Config.DeeployApi) if err != nil { continue diff --git a/storage/allocationStorer.go b/storage/allocationStorer.go index 9280b54..6e3bb8c 100644 --- a/storage/allocationStorer.go +++ b/storage/allocationStorer.go @@ -128,16 +128,43 @@ func GetAllocationsByDraftId(draftId string) ([]model.Allocation, error) { } func GetAllocationByJobIDForJobDetails(jobId string) (*model.Allocation, error) { + allocations, err := GetAllocationsByJobIDsForJobDetails([]string{jobId}) + if err != nil { + return nil, err + } + + allocation, ok := allocations[jobId] + if !ok { + return nil, gorm.ErrRecordNotFound + } + + return allocation, nil +} + +func GetAllocationsByJobIDsForJobDetails(jobIDs []string) (map[string]*model.Allocation, error) { + allocationsByJobID := make(map[string]*model.Allocation) + if len(jobIDs) == 0 { + return allocationsByJobID, nil + } + db, err := GetDB() if err != nil { return nil, err } - var allocation model.Allocation - txRead := db.Where("job_id = ? AND job_name IS NOT NULL AND job_name <> ''", jobId).First(&allocation) // Retrieve the allocation with a non-null, non-empty job name. + var allocations []model.Allocation + txRead := db. + Where("job_id IN ? AND job_name IS NOT NULL AND job_name <> ''", jobIDs). + Order("job_id, block_number DESC, allocation_creation DESC, id DESC"). + Distinct("ON (job_id) *"). + Find(&allocations) if txRead.Error != nil { return nil, txRead.Error } - return &allocation, nil + for i := range allocations { + allocationsByJobID[allocations[i].JobId] = &allocations[i] + } + + return allocationsByJobID, nil } diff --git a/storage/allocationStorer_test.go b/storage/allocationStorer_test.go new file mode 100644 index 0000000..babce04 --- /dev/null +++ b/storage/allocationStorer_test.go @@ -0,0 +1,71 @@ +package storage + +import ( + "fmt" + "testing" + "time" + + "github.com/NaeuralEdgeProtocol/ratio1-backend/model" + "github.com/stretchr/testify/require" +) + +func TestGetAllocationsByJobIDsForJobDetails(t *testing.T) { + db, err := GetDB() + require.NoError(t, err) + + now := time.Now().UTC() + suffix := now.UnixNano() + jobID1 := fmt.Sprintf("job-details-%d-1", suffix) + jobID2 := fmt.Sprintf("job-details-%d-2", suffix) + missingJobID := fmt.Sprintf("job-details-%d-missing", suffix) + jobIDs := []string{jobID1, jobID2, missingJobID} + + t.Cleanup(func() { + require.NoError(t, db.Where("job_id IN ?", jobIDs).Delete(&model.Allocation{}).Error) + }) + + allocations := []model.Allocation{ + allocationForJobDetailsTest(jobID1, "old job", model.JobType(1), "old project", 10, now.Add(-2*time.Hour)), + allocationForJobDetailsTest(jobID1, "latest job", model.JobType(2), "latest project", 11, now.Add(-time.Hour)), + allocationForJobDetailsTest(jobID1, "", model.JobType(3), "ignored project", 12, now), + allocationForJobDetailsTest(jobID2, "same block older", model.JobType(4), "same block old project", 20, now.Add(-30*time.Minute)), + allocationForJobDetailsTest(jobID2, "same block newer", model.JobType(5), "same block new project", 20, now.Add(-20*time.Minute)), + } + require.NoError(t, db.Create(&allocations).Error) + + result, err := GetAllocationsByJobIDsForJobDetails(jobIDs) + require.NoError(t, err) + require.Len(t, result, 2) + require.NotContains(t, result, missingJobID) + + require.Equal(t, "latest job", result[jobID1].JobName) + require.Equal(t, model.JobType(2), result[jobID1].JobType) + require.Equal(t, "latest project", result[jobID1].ProjectName) + + require.Equal(t, "same block newer", result[jobID2].JobName) + require.Equal(t, model.JobType(5), result[jobID2].JobType) + require.Equal(t, "same block new project", result[jobID2].ProjectName) +} + +func TestGetAllocationsByJobIDsForJobDetailsEmptyInput(t *testing.T) { + result, err := GetAllocationsByJobIDsForJobDetails(nil) + require.NoError(t, err) + require.Empty(t, result) +} + +func allocationForJobDetailsTest(jobID, jobName string, jobType model.JobType, projectName string, blockNumber int64, allocationCreation time.Time) model.Allocation { + return model.Allocation{ + AllocationCreation: allocationCreation, + BlockNumber: blockNumber, + TxHash: "0x0000000000000000000000000000000000000000000000000000000000000000", + JobId: jobID, + JobName: jobName, + JobType: jobType, + ProjectName: projectName, + NodeAddress: "0x0000000000000000000000000000000000000000", + UserAddress: "0x0000000000000000000000000000000000000000", + CspAddress: "0x0000000000000000000000000000000000000000", + CspOwner: "0x0000000000000000000000000000000000000000", + UsdcAmountPayed: "0", + } +} diff --git a/templates/cache.go b/templates/cache.go index 366a772..215ca00 100644 --- a/templates/cache.go +++ b/templates/cache.go @@ -15,6 +15,7 @@ var ( kycFinalRejectedEmailTemplate *template.Template accountResettedEmailTemplate *template.Template jobsEndingEmailTemplate *template.Template + nodesOfflineEmailTemplate *template.Template invoiceDraftTemplate *template.Template operatorDraftTemplate *template.Template @@ -51,6 +52,10 @@ func LoadAndCacheTemplates() { if err != nil { panic(err) } + nodesOffline, err := LoadNodesOfflineEmailTemplate() + if err != nil { + panic(err) + } invoiceDraftFile, err := LoadInvoiceDraftTemplate() if err != nil { @@ -74,6 +79,7 @@ func LoadAndCacheTemplates() { kycFinalRejectedEmailTemplate = finalReject accountResettedEmailTemplate = accountResetted jobsEndingEmailTemplate = jobsEnding + nodesOfflineEmailTemplate = nodesOffline invoiceDraftTemplate = invoiceDraftFile operatorDraftTemplate = operatorDraftFile @@ -109,6 +115,10 @@ func GetJobsEndingEmailTemplate() (*template.Template, error) { return getOrSetTemplate(LoadJobsEndingEmailTemplate, jobsEndingEmailTemplate) } +func GetNodesOfflineEmailTemplate() (*template.Template, error) { + return getOrSetTemplate(LoadNodesOfflineEmailTemplate, nodesOfflineEmailTemplate) +} + func GetInvoiceDraftTemplate() (*template.Template, error) { return getOrSetTemplate(LoadInvoiceDraftTemplate, invoiceDraftTemplate) } diff --git a/templates/cache_test.go b/templates/cache_test.go index dd42d43..1b587df 100644 --- a/templates/cache_test.go +++ b/templates/cache_test.go @@ -3,10 +3,12 @@ package templates import ( "testing" + "github.com/NaeuralEdgeProtocol/ratio1-backend/config" "github.com/stretchr/testify/require" ) func init() { + config.Config.EmailTemplatesPath = "../templates/html/" LoadAndCacheTemplates() } @@ -34,4 +36,8 @@ func TestTemplateGetters(t *testing.T) { c, err = GetJobsEndingEmailTemplate() require.Nil(t, err) require.Equal(t, c.Name(), emailJobsEndingFile) + + c, err = GetNodesOfflineEmailTemplate() + require.Nil(t, err) + require.Equal(t, c.Name(), emailNodesOfflineFile) } diff --git a/templates/html/email.nodes.offline.html b/templates/html/email.nodes.offline.html new file mode 100644 index 0000000..ac4271b --- /dev/null +++ b/templates/html/email.nodes.offline.html @@ -0,0 +1,535 @@ + + + + + + Mailto + + + + + + + + + +
+ + + + + + + +
+   + + + + + +
+
+   +
+
+ + + + + +
+
+   +
+ + img + +
+   +
+
+ + + + + +
+ + + Linked nodes offline for more than 24h + +
+   +
+ + + We detected {{ .NodesCount }} linked {{ if eq .NodesCount 1 }}node{{ else }}nodes{{ end }} {{ if eq .NodesCount 1 }}that has{{ else }}that have{{ end }} been offline for more than 24 hours. + + +
+   +
+
+   +
+
+ + + + + +
+ + + + + + + {{ range .Nodes }} + + + + + + {{ end }} +
Node AliasNode AddressOffline For
{{ .Alias }}{{ .Address }}{{ .OfflineFor }}
+
+   +
+
+ + + + + +
+
+   +
+ + You will receive this email at most once every 24 hours while at least one linked node remains offline.
+ You can unlink nodes from app.ratio1.ai to stop receiving these notifications.
+ If you received this email by mistake, simply delete it. + +
+   +
+
+ + + + + +
+
+   +
+ + + + +
+ +
+   +
+ + + NAEURAL SRL © Strada Eufrosina Popescu, + Nr. 61, Bucureşti, Romania + + +
+   +
+ + contact@ratio1.ai + + + + + +
+   +
+ + + + + + + + +
+ + img + + +   + + + img + + +   + + + img + +
+
+   +
+
+
+ 		+   +
+ +
+ + + diff --git a/templates/util.go b/templates/util.go index 5cebbbf..a497087 100644 --- a/templates/util.go +++ b/templates/util.go @@ -15,6 +15,7 @@ const ( emailKycConfirmedFile = "email.kyc.confirmed.html" emailAccountResettedFile = "email.account.resetted.html" emailJobsEndingFile = "email.jobs.ending.html" + emailNodesOfflineFile = "email.nodes.offline.html" invoiceDraftFile = "invoice.draft.html" emailOperatorFile = "email.operator.draft.html" @@ -71,6 +72,10 @@ func LoadJobsEndingEmailTemplate() (*template.Template, error) { return loadTemplate(emailJobsEndingFile) } +func LoadNodesOfflineEmailTemplate() (*template.Template, error) { + return loadTemplate(emailNodesOfflineFile) +} + func LoadInvoiceDraftTemplate() (*template.Template, error) { return loadInvoiceTemplate(invoiceDraftFile) }