Merge branch 'main' of https://github.com/SkySingh04/fractal

Author: SkySingh04

.gitignore

@@ -1,2 +1,4 @@
 .env
-myenv/
+myenv/
+firebaseConfig.json
+tests/firebaseConfig.json   

Readme.MD

@@ -155,28 +155,25 @@ Depending on the data source, fields can be named using:
 The rules can be embedded into the YAML configuration for pipelines:
 
 ```yaml
+pipeline:
+  error-handling:
+   strategy: LOG_AND_CONTINUE
 inputconfig:
-  url: wss://echo.websocket.org
-inputmethod: WebSocket
-
+   csvsourcefilename: sample.csv
+   inputmethod: CSV
 outputconfig:
-  url: wss://echo.websocket.org
-outputmethod: WebSocket
-
-validation:
-  rules: |
-    FIELD("message") REQUIRED
-    FIELD("timestamp") TYPE(INT) RANGE(1609459200, 1704067200)
-
-transformation:
-  rules: |
-    ADD_FIELD("processed_at", CURRENT_TIME())
-    IF FIELD("message") MATCHES("^hello") THEN RENAME("message", "greeting")
-    MAP("status", {"0": "inactive", "1": "active"})
-
-error_handling:
-  on_error: LOG_AND_CONTINUE
+   csvdestinationfilename: test.csv
+   outputmethod: CSV
+cronjob:
+   repetition_interval: "1h"
+monitoring:
+   job_status:"pending"
+transformations:
+   -ADD_FIELD("processed_at", CURRENT_TIME())
+validations:
+   -FIELD("age") RANGE(30,35)
 ```
+
 ---
 
 # Adding a New Integration

config.yaml

@@ -1,16 +1,9 @@
 errorhandling:
     strategy: LOG_AND_CONTINUE
 inputconfig:
-    csvsourcefilename: sample.csv
-inputmethod: CSV
+    url: sdfxghjuk
+inputmethod: WebSocket
 outputconfig:
-    csvdestinationfilename: samplenew.csv
-outputmethod: CSV
-cronjob:
-  cronexpression: "@every 5s"  # Every 5 seconds
-  jobname: "Data Fetch and Process"
-  task: "process_csv"
-transformations: |
-    ADD_FIELD("processed_at", CURRENT_TIME())
-validations: |
-    FIELD("age") TYPE(INT) RANGE(30, 35)
+    topic: bfhtd
+    url: dfghjkl
+outputmethod: Kafka

firebaseConfig.json

@@ -6,7 +6,6 @@ toolchain go1.22.9
 
 require (
 	firebase.google.com/go v3.13.0+incompatible
-	github.com/golang/mock v1.1.1
 	github.com/jlaffaye/ftp v0.2.0
 	github.com/manifoldco/promptui v0.9.0
 	github.com/pkg/sftp v1.13.7
@@ -54,7 +53,7 @@ require (
 
 require (
 	cloud.google.com/go v0.116.0 // indirect
-	cloud.google.com/go/auth v0.9.9 // indirect
+	cloud.google.com/go/auth v0.11.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.4 // indirect
 	cloud.google.com/go/compute/metadata v0.5.2 // indirect
 	cloud.google.com/go/iam v1.2.1 // indirect
@@ -122,13 +121,13 @@ require (
 	go.opentelemetry.io/otel/trace v1.32.0
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/mock v0.5.0 // indirect
-	golang.org/x/crypto v0.28.0
+	golang.org/x/crypto v0.31.0
 	golang.org/x/net v0.30.0 // indirect
 	golang.org/x/oauth2 v0.24.0 // indirect
-	golang.org/x/sync v0.8.0 // indirect
-	golang.org/x/sys v0.27.0 // indirect
-	golang.org/x/term v0.25.0 // indirect
-	golang.org/x/text v0.19.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/term v0.27.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.7.0 // indirect
 	google.golang.org/api v0.203.0
 	google.golang.org/genproto v0.0.0-20241015192408-796eee8c2d53 // indirect

go.mod

@@ -1,8 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.116.0 h1:B3fRrSDkLRt5qSHWe40ERJvhvnQwdZiHu0bJOpldweE=
 cloud.google.com/go v0.116.0/go.mod h1:cEPSRWPzZEswwdr9BxE6ChEn01dWlTaF05LiC2Xs70U=
-cloud.google.com/go/auth v0.9.9 h1:BmtbpNQozo8ZwW2t7QJjnrQtdganSdmqeIBxHxNkEZQ=
-cloud.google.com/go/auth v0.9.9/go.mod h1:xxA5AqpDrvS+Gkmo9RqrGGRh6WSNKKOXhY3zNOr38tI=
+cloud.google.com/go/auth v0.11.0 h1:Ic5SZz2lsvbYcWT5dfjNWgw6tTlGi2Wc8hyQSC9BstA=
+cloud.google.com/go/auth v0.11.0/go.mod h1:xxA5AqpDrvS+Gkmo9RqrGGRh6WSNKKOXhY3zNOr38tI=
 cloud.google.com/go/auth/oauth2adapt v0.2.4 h1:0GWE/FUsXhf6C+jAkWgYm7X9tK8cuEIfy19DBn6B6bY=
 cloud.google.com/go/auth/oauth2adapt v0.2.4/go.mod h1:jC/jOpwFP6JBxhB3P5Rr0a9HLMC/Pe3eaL4NmdvqPtc=
 cloud.google.com/go/compute/metadata v0.5.2 h1:UxK4uu/Tn+I3p2dYWTfiX4wva7aYlKixAHn3fyqngqo=
@@ -92,7 +92,6 @@ github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfU
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
 github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
-github.com/golang/mock v1.1.1 h1:G5FRp8JnTd7RQH5kemVNlMeyXQAztQ3mOWV95KxsXH8=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -331,8 +330,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
-golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw=
-golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20231108232855-2478ac86f678 h1:mchzmB1XO2pMaKFRqk/+MV3mgGG96aqaPXaMifQU47w=
 golang.org/x/exp v0.0.0-20231108232855-2478ac86f678/go.mod h1:zk2irFbV9DP96SEBUUAy67IdHUaZuSnrz1n472HUCLE=
@@ -372,8 +371,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
-golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181122145206-62eef0e2fa9b/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -390,16 +389,16 @@ golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s=
-golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
-golang.org/x/term v0.25.0 h1:WtHI/ltw4NvSUig5KARz9h521QvRC8RmF/cuYqifU24=
-golang.org/x/term v0.25.0/go.mod h1:RPyXicDX+6vLxogjjRxjgD2TKtmAO6NZBsBRfrOLu7M=
+golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
+golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
@@ -408,8 +407,8 @@ golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/text v0.19.0 h1:kTxAhCbGbxhK0IwgSKiMO5awPoDQ0RpfiVYBfK860YM=
-golang.org/x/text v0.19.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ=
 golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

go.sum

@@ -3,6 +3,7 @@ package integrations
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log"
 	"sync"
 
@@ -108,61 +109,46 @@ func (m MongoDBDestination) SendData(data interface{}, req interfaces.Request) e
 	}
 	logger.Infof("Connecting to MongoDB destination...")
 
+	// Initialize MongoDB client
 	clientOptions := options.Client().ApplyURI(req.TargetMongoDBConnString)
 	client, err := mongo.Connect(context.TODO(), clientOptions)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to connect to MongoDB: %w", err)
 	}
 	defer func() {
 		if err = client.Disconnect(context.TODO()); err != nil {
 			logger.Errorf("Error disconnecting MongoDB client: %v", err)
 		}
 	}()
 
-	collection := client.Database(req.TargetMongoDBDatabase).Collection(req.TargetMongoDBCollection)
-
-	// Assert that data is a slice of bson.M
-	dataSlice, ok := data.([]bson.M)
-	if !ok {
-		logger.Errorf("data must be a slice of bson.M representing documents")
-		return errors.New("invalid data format: expected []bson.M")
+	// Transform data to BSON
+	bsonData, err := TransformDataToBSON(data)
+	if err != nil {
+		return fmt.Errorf("data transformation failed: %w", err)
 	}
 
-	// Buffered channel for sending documents
-	dataChannel := make(chan bson.M, bufferSize)
-	errorChannel := make(chan error, bufferSize)
-	var wg sync.WaitGroup
-
-	// Goroutines for worker pool
-	for i := 0; i < bufferSize; i++ { // Worker pool
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for doc := range dataChannel {
-				if _, err := collection.InsertOne(context.TODO(), doc); err != nil {
-					logger.Errorf("Error inserting into collection %s: %v", req.TargetMongoDBCollection, err)
-					errorChannel <- err
-				} else {
-					logger.Infof("Data sent to MongoDB target collection %s: %v", req.TargetMongoDBCollection, doc)
-				}
-			}
-		}()
-	}
+	// Access database and collection
+	collection := client.Database(req.TargetMongoDBDatabase).Collection(req.TargetMongoDBCollection)
 
-	// Feed data into the channel
-	go func() {
-		for _, doc := range dataSlice {
-			dataChannel <- doc
+	// Insert data into MongoDB
+	if len(bsonData) == 1 {
+		// Insert a single document
+		_, err = collection.InsertOne(context.TODO(), bsonData[0])
+		if err != nil {
+			return fmt.Errorf("failed to insert document: %w", err)
 		}
-		close(dataChannel)
-	}()
 
-	wg.Wait()
-	close(errorChannel)
-
-	// Check for errors in the error channel
-	if len(errorChannel) > 0 {
-		return errors.New("one or more errors occurred while inserting data")
+	} else {
+		// Insert multiple documents
+		docs := make([]interface{}, len(bsonData))
+		for i, doc := range bsonData {
+			docs[i] = doc
+		}
+		_, err = collection.InsertMany(context.TODO(), docs)
+		if err != nil {
+			return fmt.Errorf("failed to insert documents: %w", err)
+		}
+		logger.Infof("Successfully inserted %d documents into MongoDB collection %s", len(bsonData), req.TargetMongoDBCollection)
 	}
 
 	return nil
@@ -173,3 +159,20 @@ func init() {
 	registry.RegisterSource("MongoDB", MongoDBSource{})
 	registry.RegisterDestination("MongoDB", MongoDBDestination{})
 }
+
+func TransformDataToBSON(data interface{}) ([]bson.M, error) {
+	switch v := data.(type) {
+	case map[string]interface{}: // Single document
+		return []bson.M{v}, nil
+	case []map[string]interface{}: // Multiple documents
+		result := make([]bson.M, len(v))
+		for i, item := range v {
+			result[i] = item
+		}
+		return result, nil
+	case []bson.M: // Already in bson.M
+		return v, nil
+	default:
+		return nil, fmt.Errorf("unsupported data format: %T", v)
+	}
+}