Merge pull request #910 from gobitfly/BEDS-473/Notifications_set_settings

Beds 473/notifications set settings

Author: Eisei24

backend/pkg/api/data_access/notifications.go

@@ -7,11 +7,13 @@ import (
 	"strings"
 	"time"
 
+	"github.com/doug-martin/goqu/v9"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/gobitfly/beaconchain/pkg/api/enums"
 	t "github.com/gobitfly/beaconchain/pkg/api/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
+	"github.com/gobitfly/beaconchain/pkg/commons/utils"
 	"github.com/shopspring/decimal"
 	"golang.org/x/sync/errgroup"
 )
@@ -296,19 +298,258 @@ func (d *DataAccessService) GetNotificationSettings(ctx context.Context, userId
 	return result, nil
 }
 func (d *DataAccessService) UpdateNotificationSettingsGeneral(ctx context.Context, userId uint64, settings t.NotificationSettingsGeneral) error {
-	return d.dummy.UpdateNotificationSettingsGeneral(ctx, userId, settings)
+	epoch := utils.TimeToEpoch(time.Now())
+
+	var eventsToInsert []goqu.Record
+	var eventsToDelete []goqu.Expression
+
+	tx, err := d.userWriter.BeginTxx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("error starting db transactions to update general notification settings: %w", err)
+	}
+	defer utils.Rollback(tx)
+
+	// -------------------------------------
+	// Set the "do not disturb" setting
+	_, err = tx.ExecContext(ctx, `
+		UPDATE users 
+		SET notifications_do_not_disturb_ts = 
+		    CASE 
+		        WHEN $1 = 0 THEN NULL
+		        ELSE TO_TIMESTAMP($1)
+		    END 
+		WHERE id = $2`, settings.DoNotDisturbTimestamp, userId)
+	if err != nil {
+		return err
+	}
+
+	// -------------------------------------
+	// Set the notification channels
+	_, err = tx.ExecContext(ctx, `
+		INSERT INTO users_notification_channels (user_id, channel, active)
+    		VALUES ($1, $2, $3), ($1, $4, $5)
+    	ON CONFLICT (user_id, channel) 
+    		DO UPDATE SET active = EXCLUDED.active`,
+		userId, types.EmailNotificationChannel, settings.IsEmailNotificationsEnabled, types.PushNotificationChannel, settings.IsPushNotificationsEnabled)
+	if err != nil {
+		return err
+	}
+
+	// -------------------------------------
+	// Collect the machine and rocketpool events to set and delete
+
+	//Machine events
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsMachineOfflineSubscribed, userId, string(types.MonitoringMachineOfflineEventName), "", epoch, 0)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsMachineStorageUsageSubscribed, userId, string(types.MonitoringMachineDiskAlmostFullEventName), "", epoch, settings.MachineStorageUsageThreshold)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsMachineCpuUsageSubscribed, userId, string(types.MonitoringMachineCpuLoadEventName), "", epoch, settings.MachineCpuUsageThreshold)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsMachineMemoryUsageSubscribed, userId, string(types.MonitoringMachineMemoryUsageEventName), "", epoch, settings.MachineMemoryUsageThreshold)
+
+	// Insert all the events or update the threshold if they already exist
+	if len(eventsToInsert) > 0 {
+		insertDs := goqu.Dialect("postgres").
+			Insert("users_subscriptions").
+			Cols("user_id", "event_name", "event_filter", "created_ts", "created_epoch", "event_threshold").
+			Rows(eventsToInsert).
+			OnConflict(goqu.DoUpdate(
+				"user_id, event_name, event_filter",
+				goqu.Record{"event_threshold": goqu.L("EXCLUDED.event_threshold")},
+			))
+
+		query, args, err := insertDs.Prepared(true).ToSQL()
+		if err != nil {
+			return fmt.Errorf("error preparing query: %v", err)
+		}
+
+		_, err = tx.ExecContext(ctx, query, args...)
+		if err != nil {
+			return err
+		}
+	}
+
+	// Delete all the events
+	if len(eventsToDelete) > 0 {
+		deleteDs := goqu.Dialect("postgres").
+			Delete("users_subscriptions").
+			Where(goqu.Or(eventsToDelete...))
+
+		query, args, err := deleteDs.Prepared(true).ToSQL()
+		if err != nil {
+			return fmt.Errorf("error preparing query: %v", err)
+		}
+
+		_, err = tx.ExecContext(ctx, query, args...)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("error committing tx to update general notification settings: %w", err)
+	}
+	return nil
 }
 func (d *DataAccessService) UpdateNotificationSettingsNetworks(ctx context.Context, userId uint64, chainId uint64, settings t.NotificationSettingsNetwork) error {
-	return d.dummy.UpdateNotificationSettingsNetworks(ctx, userId, chainId, settings)
+	epoch := utils.TimeToEpoch(time.Now())
+
+	networks, err := d.GetAllNetworks()
+	if err != nil {
+		return err
+	}
+
+	networkName := ""
+	for _, network := range networks {
+		if network.ChainId == chainId {
+			networkName = network.Name
+			break
+		}
+	}
+	if networkName == "" {
+		return fmt.Errorf("network with chain id %d to update general notification settings not found", chainId)
+	}
+
+	var eventsToInsert []goqu.Record
+	var eventsToDelete []goqu.Expression
+
+	tx, err := d.userWriter.BeginTxx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("error starting db transactions to update general notification settings: %w", err)
+	}
+	defer utils.Rollback(tx)
+
+	eventName := fmt.Sprintf("%s:%s", networkName, types.NetworkGasAboveThresholdEventName)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsGasAboveSubscribed, userId, eventName, "", epoch, settings.GasAboveThreshold.Div(decimal.NewFromInt(params.GWei)).InexactFloat64())
+	eventName = fmt.Sprintf("%s:%s", networkName, types.NetworkGasBelowThresholdEventName)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsGasBelowSubscribed, userId, eventName, "", epoch, settings.GasBelowThreshold.Div(decimal.NewFromInt(params.GWei)).InexactFloat64())
+	eventName = fmt.Sprintf("%s:%s", networkName, types.NetworkParticipationRateThresholdEventName)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsParticipationRateSubscribed, userId, eventName, "", epoch, settings.ParticipationRateThreshold)
+	eventName = fmt.Sprintf("%s:%s", networkName, types.RocketpoolNewClaimRoundStartedEventName)
+	d.AddOrRemoveEvent(&eventsToInsert, &eventsToDelete, settings.IsNewRewardRoundSubscribed, userId, eventName, "", epoch, 0)
+
+	// Insert all the events or update the threshold if they already exist
+	if len(eventsToInsert) > 0 {
+		insertDs := goqu.Dialect("postgres").
+			Insert("users_subscriptions").
+			Cols("user_id", "event_name", "event_filter", "created_ts", "created_epoch", "event_threshold").
+			Rows(eventsToInsert).
+			OnConflict(goqu.DoUpdate(
+				"user_id, event_name, event_filter",
+				goqu.Record{"event_threshold": goqu.L("EXCLUDED.event_threshold")},
+			))
+
+		query, args, err := insertDs.Prepared(true).ToSQL()
+		if err != nil {
+			return fmt.Errorf("error preparing query: %v", err)
+		}
+
+		_, err = tx.ExecContext(ctx, query, args...)
+		if err != nil {
+			return err
+		}
+	}
+
+	// Delete all the events
+	if len(eventsToDelete) > 0 {
+		deleteDs := goqu.Dialect("postgres").
+			Delete("users_subscriptions").
+			Where(goqu.Or(eventsToDelete...))
+
+		query, args, err := deleteDs.Prepared(true).ToSQL()
+		if err != nil {
+			return fmt.Errorf("error preparing query: %v", err)
+		}
+
+		_, err = tx.ExecContext(ctx, query, args...)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("error committing tx to update general notification settings: %w", err)
+	}
+	return nil
 }
 func (d *DataAccessService) UpdateNotificationSettingsPairedDevice(ctx context.Context, userId uint64, pairedDeviceId string, name string, IsNotificationsEnabled bool) error {
-	return d.dummy.UpdateNotificationSettingsPairedDevice(ctx, userId, pairedDeviceId, name, IsNotificationsEnabled)
+	result, err := d.userWriter.ExecContext(ctx, `
+		UPDATE users_devices 
+		SET 
+			device_name = $1,
+			notify_enabled = $2
+		WHERE user_id = $3 AND device_identifier = $4`,
+		name, IsNotificationsEnabled, userId, pairedDeviceId)
+	if err != nil {
+		return err
+	}
+
+	// TODO: This can be deleted when the API layer has an improved check for the device id
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if rowsAffected == 0 {
+		return fmt.Errorf("device with id %s to update notification settings not found", pairedDeviceId)
+	}
+	return nil
 }
 func (d *DataAccessService) DeleteNotificationSettingsPairedDevice(ctx context.Context, userId uint64, pairedDeviceId string) error {
-	return d.dummy.DeleteNotificationSettingsPairedDevice(ctx, userId, pairedDeviceId)
+	result, err := d.userWriter.ExecContext(ctx, `
+		DELETE FROM users_devices 
+		WHERE user_id = $1 AND device_identifier = $2`,
+		userId, pairedDeviceId)
+	if err != nil {
+		return err
+	}
+
+	// TODO: This can be deleted when the API layer has an improved check for the device id
+	rowsAffected, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if rowsAffected == 0 {
+		return fmt.Errorf("device with id %s to delete not found", pairedDeviceId)
+	}
+	return nil
 }
 func (d *DataAccessService) UpdateNotificationSettingsClients(ctx context.Context, userId uint64, clientId uint64, IsSubscribed bool) (*t.NotificationSettingsClient, error) {
-	return d.dummy.UpdateNotificationSettingsClients(ctx, userId, clientId, IsSubscribed)
+	result := &t.NotificationSettingsClient{Id: clientId, IsSubscribed: IsSubscribed}
+
+	var clientInfo *t.ClientInfo
+
+	clients, err := d.GetAllClients()
+	if err != nil {
+		return nil, err
+	}
+	for _, client := range clients {
+		if client.Id == clientId {
+			clientInfo = &client
+			break
+		}
+	}
+	if clientInfo == nil {
+		return nil, fmt.Errorf("client with id %d to update client notification settings not found", clientId)
+	}
+
+	if IsSubscribed {
+		_, err = d.userWriter.ExecContext(ctx, `
+			INSERT INTO users_subscriptions (user_id, event_name, event_filter, created_ts, created_epoch)
+				VALUES ($1, $2, $3, NOW(), $4)
+			ON CONFLICT (user_id, event_name, event_filter) 
+				DO NOTHING`,
+			userId, types.EthClientUpdateEventName, clientInfo.Name, utils.TimeToEpoch(time.Now()))
+	} else {
+		_, err = d.userWriter.ExecContext(ctx, `DELETE FROM users_subscriptions WHERE user_id = $1 AND event_name = $2 AND event_filter = $3`,
+			userId, types.EthClientUpdateEventName, clientInfo.Name)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	result.Name = clientInfo.Name
+	result.Category = clientInfo.Category
+
+	return result, nil
 }
 func (d *DataAccessService) GetNotificationSettingsDashboards(ctx context.Context, userId uint64, cursor string, colSort t.Sort[enums.NotificationSettingsDashboardColumn], search string, limit uint64) ([]t.NotificationSettingsDashboardsTableRow, *t.Paging, error) {
 	return d.dummy.GetNotificationSettingsDashboards(ctx, userId, cursor, colSort, search, limit)
@@ -319,3 +560,12 @@ func (d *DataAccessService) UpdateNotificationSettingsValidatorDashboard(ctx con
 func (d *DataAccessService) UpdateNotificationSettingsAccountDashboard(ctx context.Context, dashboardId t.VDBIdPrimary, groupId uint64, settings t.NotificationSettingsAccountDashboard) error {
 	return d.dummy.UpdateNotificationSettingsAccountDashboard(ctx, dashboardId, groupId, settings)
 }
+
+func (d *DataAccessService) AddOrRemoveEvent(eventsToInsert *[]goqu.Record, eventsToDelete *[]goqu.Expression, isSubscribed bool, userId uint64, eventName string, eventFilter string, epoch int64, threshold float64) {
+	if isSubscribed {
+		event := goqu.Record{"user_id": userId, "event_name": eventName, "event_filter": eventFilter, "created_ts": goqu.L("NOW()"), "created_epoch": epoch, "event_threshold": threshold}
+		*eventsToInsert = append(*eventsToInsert, event)
+	} else {
+		*eventsToDelete = append(*eventsToDelete, goqu.Ex{"user_id": userId, "event_name": eventName, "event_filter": eventFilter})
+	}
+}