Add trigger validation

Add trigger validation on sync, removing invalid payload from process queue

Signed-off-by: Eddy Babetto <eddy.babetto@secomind.com>

cmd/appengine/device.go

@@ -1241,7 +1241,7 @@ func getProtoInterface(deviceID string, deviceIdentifierType client.DeviceIdenti
 		// Just a trick to trick the parser into doing the right thing.
 		if isParametricInterface {
 			iface.Mappings = []interfaces.AstarteInterfaceMapping{
-				interfaces.AstarteInterfaceMapping{
+				{
 					Endpoint: "/it/%{is}/parametric",
 				},
 			}

cmd/realm/triggers.go

@@ -17,12 +17,12 @@ package realm
 import (
 	"encoding/json"
 	"fmt"
-	"os"
-	"path/filepath"
-
+	"github.com/astarte-platform/astarte-go/triggers"
 	"github.com/astarte-platform/astartectl/utils"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"os"
+	"path/filepath"
 )
 
 // triggersCmd represents the triggers command
@@ -72,10 +72,10 @@ var triggersDeleteCmd = &cobra.Command{
 }
 
 var triggersSaveCmd = &cobra.Command{
-	Use:   "save [destination-path]",
+	Use:   "save <destination-path>",
 	Short: "Save triggers to a local folder",
 	Long: `Save each trigger in a realm to a local folder. Each trigger will
-be saved in a dedicated file whose name will be in the form '<trigger_name>_v<version>.json'.
+be saved in a dedicated file whose name will be in the form '<trigger_name>.json'.
 When no destination path is set, triggers will be saved in the current working directory.
 This command does not support the --to-curl flag.`,
 	Example: `  astartectl realm-management triggers save`,
@@ -84,7 +84,7 @@ This command does not support the --to-curl flag.`,
 }
 
 var triggersSyncCmd = &cobra.Command{
-	Use:   "sync <interface_files> [...]",
+	Use:   "sync <trigger_files> [...]",
 	Short: "Synchronize triggers",
 	Long: `Synchronize triggers in the realm with the given files.
 All given files will be parsed, and only new triggers will be installed in the
@@ -97,7 +97,7 @@ realm, depending on the realm's state. In order to force triggers update, use --
 func init() {
 
 	RealmManagementCmd.AddCommand(triggersCmd)
-	triggersSyncCmd.Flags().Bool("force", false, "When set, force triggers update")
+	triggersSyncCmd.Flags().Lookup("force").NoOptDefVal = "false"
 	triggersCmd.AddCommand(
 		triggersListCmd,
 		triggersShowCmd,
@@ -115,12 +115,10 @@ func triggersListF(command *cobra.Command, args []string) error {
 }
 
 func triggersShowF(command *cobra.Command, args []string) error {
-
 	triggerName := args[0]
 	triggerDefinition, _ := getTriggerDefinition(realm, triggerName)
 	respJSON, _ := json.MarshalIndent(triggerDefinition, "", "  ")
 	fmt.Println(string(respJSON))
-
 	return nil
 }
 
@@ -226,28 +224,33 @@ func triggersSyncF(command *cobra.Command, args []string) error {
 		os.Exit(1)
 	}
 
-	triggerToInstall := []map[string]interface{}{}
-	triggerToUpdate := []map[string]interface{}{}
+	triggersToInstall := []map[string]interface{}{}
+	triggersToUpdate := []map[string]interface{}{}
+	invalidTriggers := []string{}
 
 	for _, f := range args {
-		triggerFile, err := os.ReadFile(f)
+		triggerFile, err := os.ReadFile(f)	
 		if err != nil {
 			return err
 		}
+		if !validateTrigger(f) {
+			invalidTriggers = append(invalidTriggers, f)
+			continue
+		}
 
-		var astarteTrigger map[string]interface{}
+		var astarteTrigger map[string]triggers.Trigger{}
 		if err = json.Unmarshal(triggerFile, &astarteTrigger); err != nil {
 			return err
 		}
 
 		if _, err := getTriggerDefinition(realm, astarteTrigger["name"].(string)); err != nil {
 			// The trigger does not exist
-			triggerToInstall = append(triggerToInstall, astarteTrigger)
+			triggersToInstall = append(triggersToInstall, astarteTrigger)
 		} else {
-			triggerToUpdate = append(triggerToUpdate, astarteTrigger)
+			triggersToUpdate = append(triggersToUpdate, astarteTrigger)
 		}
 
-		if len(triggerToInstall) == 0 && len(triggerToUpdate) == 0 {
+		if len(triggersToInstall) == 0 && len(triggersToUpdate) == 0 {
 			// All good in the hood
 			fmt.Println("Your realm is in sync with the provided triggers files")
 			return nil
@@ -257,22 +260,21 @@ func triggersSyncF(command *cobra.Command, args []string) error {
 	// Notify the user about what we're about to do
 	list := []string{}
 
-	for _, v := range triggerToInstall {
+	for _, v := range triggersToInstall {
 		list = append(list, v["name"].(string))
 	}
 
 	list_existing := []string{}
 
-	for _, v := range triggerToUpdate {
+	for _, v := range triggersToUpdate {
 		list_existing = append(list_existing, v["name"].(string))
 	}
 
-	// Start syncing.
+	fmt.Printf("The following triggers are invalid and thus will not be processed: %+q \n", invalidTriggers)
 
 	//install new triggers
-	if len(triggerToInstall) > 0 {
+	if len(triggersToInstall) > 0 {
 
-		fmt.Printf("\n")
 		fmt.Printf("The following new triggers will be installed: %+q \n", list)
 		fmt.Printf("\n")
 
@@ -281,52 +283,38 @@ func triggersSyncF(command *cobra.Command, args []string) error {
 			return nil
 		}
 
-		for _, v := range triggerToInstall {
+		for _, v := range triggersToInstall {
 			if err := installTrigger(realm, v); err != nil {
 				fmt.Fprintf(os.Stderr, "Could not install trigger %s: %s\n", v["name"].(string), err)
 			} else {
 				fmt.Printf("trigger %s installed successfully\n", v["name"].(string))
 			}
 		}
 
-		fmt.Printf("\n")
-
 	}
-
-	if len(triggerToUpdate) > 0 {
-
+	if len(triggersToUpdate) > 0 {
 		y, err := command.Flags().GetBool("force")
 		if err != nil {
 			return err
 		}
-
 		if y {
 			fmt.Printf("The following triggers already exists and WILL be DELETED and RECREATED: %+q \n", list_existing)
 			fmt.Printf("\n")
 			if ok, err := utils.AskForConfirmation("Do you want to continue?"); !ok || err != nil {
 				fmt.Printf("aborting")
 				return nil
 			}
-
-			for _, v := range triggerToUpdate {
+			for _, v := range triggersToUpdate {
 				if err := updateTrigger(realm, v["name"].(string), v); err != nil {
 					fmt.Fprintf(os.Stderr, "Could not update trigger %s: %s\n", v["name"].(string), err)
 				} else {
 					fmt.Printf("trigger %s updated successfully\n", v["name"].(string))
 				}
 			}
-			fmt.Printf("\n")
-			fmt.Printf("\n")
-
 		} else {
-
-			// Start syncing.
 			fmt.Printf("The following triggers already exists and WILL NOT be updated: %+q \n", list_existing)
-			fmt.Printf("\n")
-
 		}
 	}
-
 	return nil
 }
 
@@ -351,7 +339,6 @@ func installTrigger(realm string, trigger map[string]interface{}) error {
 }
 
 func updateTrigger(realm string, triggername string, newtrig map[string]interface{}) error {
-
 	deleteTriggercall, err := astarteAPIClient.DeleteTrigger(realm, triggername)
 	if err != nil {
 		return err
@@ -407,8 +394,8 @@ func getTriggerDefinition(realm, triggerName string) (map[string]interface{}, er
 		return nil, err
 	}
 
-	// When we're here in the context of `interfaces sync`, the to-curl flag
-	// is always false (`interfaces sync` has no `--to-curl` flag)
+	// When we're here in the context of `trigger sync`, the to-curl flag
+	// is always false (`trigger sync` has no `--to-curl` flag)
 	// and thus the call will never exit unexpectedly
 	utils.MaybeCurlAndExit(getTriggerCall, astarteAPIClient)
 
@@ -423,3 +410,11 @@ func getTriggerDefinition(realm, triggerName string) (map[string]interface{}, er
 	triggerDefinition, _ := rawTRigger.(map[string]interface{})
 	return triggerDefinition, nil
 }
+
+func validateTrigger(path string) bool {
+	if _, err := triggers.ParseTriggerFrom(path); err != nil {
+		return false
+	} else {
+		return true
+	}
+}