Add support to Mia-Platform Console webhook events

.golangci.yaml

@@ -40,7 +40,6 @@ linters:
   - tenv
   - thelper
   - unconvert
-  - unparam
   - unused
   - usestdlibvars
   - whitespace
@@ -59,8 +58,6 @@ linters-settings:
         yaml: camel
   tenv:
     all: true
-  unparam:
-    check-exported: false
   mnd:
     checks:
     - case

internal/config/config.schema.json

@@ -13,7 +13,8 @@
               "type": {
                 "type": "string",
                 "enum": [
-                  "jira"
+                  "jira",
+                  "console"
                 ]
               },
               "webhookPath": {

internal/config/config_test.go

@@ -16,6 +16,7 @@
 package config
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -72,12 +73,41 @@ func TestLoadServiceConfiguration(t *testing.T) {
 			},
 			expectedSinkConfig:      getExpectedSinkConfig(t),
 			expectedProcessorConfig: getExpectedProcessorConfig(t),
-			expectedSourceConfig:    getExpectedSourceConfig(t),
+			expectedSourceConfig:    getExpectedSourceConfig(t, "jira"),
 		},
 		"invalid config if integrations is empty": {
 			path:          "./testdata/empty-integrations.json",
 			expectedError: "configuration not valid: json schema validation errors:",
 		},
+		"console config is parsed correctly": {
+			path: "./testdata/console-config.json",
+			expectedContent: &Configuration{
+				Integrations: []Integration{
+					{
+						Source: GenericConfig{
+							Type: "console",
+						},
+						Pipelines: []Pipeline{
+							{
+								Processors: Processors{
+									{
+										Type: "mapper",
+									},
+								},
+								Sinks: Sinks{
+									{
+										Type: "mongo",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedSinkConfig:      getExpectedSinkConfig(t),
+			expectedProcessorConfig: getExpectedProcessorConfig(t),
+			expectedSourceConfig:    getExpectedSourceConfig(t, "console"),
+		},
 	}
 
 	for testName, test := range tests {
@@ -178,16 +208,16 @@ func getExpectedProcessorConfig(t *testing.T) string {
 }`
 }
 
-func getExpectedSourceConfig(t *testing.T) string {
+func getExpectedSourceConfig(t *testing.T, sourceType string) string {
 	t.Helper()
 
-	return `{
-	"type": "jira",
+	return fmt.Sprintf(`{
+	"type": "%s",
 	"webhookPath": "/custom-webhook-path",
 	"authentication": {
 		"secret": {
 			"fromFile": "testdata/secret"
 		}
 	}
-}`
+}`, sourceType)
 }

internal/config/testdata/console-config.json

@@ -0,0 +1,39 @@
+{
+  "integrations": [
+    {
+      "source": {
+        "type": "console",
+        "webhookPath": "/custom-webhook-path",
+        "authentication": {
+          "secret": {
+            "fromFile": "testdata/secret"
+          }
+        }
+      },
+      "pipelines": [
+        {
+          "processors": [
+            {
+              "type": "mapper",
+              "outputEvent": {
+                "key": "{{ issue.key }}",
+                "summary": "{{ issue.fields.summary }}",
+                "createdAt": "{{ issue.fields.created }}",
+                "description": "{{ issue.fields.description }}"
+              }
+            }
+          ],
+          "sinks": [
+            {
+              "type": "mongo",
+              "url": {
+                "fromEnv": "TEST_LOAD_SERVICE_MONGO_URL"
+              },
+              "collection": "my-collection"
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

internal/pipeline/pipeline.go

@@ -18,12 +18,12 @@ package pipeline
 import (
 	"context"
 	"errors"
-	"reflect"
 
 	"github.com/mia-platform/integration-connector-agent/internal/entities"
 	"github.com/mia-platform/integration-connector-agent/internal/processors"
 	"github.com/mia-platform/integration-connector-agent/internal/processors/filter"
 	"github.com/mia-platform/integration-connector-agent/internal/sinks"
+	"github.com/mia-platform/integration-connector-agent/internal/utils"
 
 	"github.com/sirupsen/logrus"
 )
@@ -45,7 +45,7 @@ func (p Pipeline) AddMessage(data entities.PipelineEvent) {
 }
 
 func (p Pipeline) Start(ctx context.Context) error {
-	if isNil(p.sinks) {
+	if utils.IsNil(p.sinks) {
 		return ErrWriterNotDefined
 	}
 
@@ -112,7 +112,3 @@ func New(logger *logrus.Logger, p *processors.Processors, sinks sinks.Sink[entit
 
 	return pipeline, nil
 }
-
-func isNil(i any) bool {
-	return i == nil || (reflect.ValueOf(i).Kind() == reflect.Ptr && reflect.ValueOf(i).IsNil())
-}

internal/server/integrations.go

@@ -28,7 +28,6 @@ import (
 	"github.com/mia-platform/integration-connector-agent/internal/sinks/mongo"
 	"github.com/mia-platform/integration-connector-agent/internal/sources"
 	"github.com/mia-platform/integration-connector-agent/internal/sources/jira"
-	"github.com/mia-platform/integration-connector-agent/internal/sources/webhook"
 
 	swagger "github.com/davidebianchi/gswagger"
 	"github.com/gofiber/fiber/v2"
@@ -68,15 +67,9 @@ func setupPipelines(ctx context.Context, log *logrus.Logger, cfg *config.Configu
 		source := cfgIntegration.Source
 		switch source.Type {
 		case sources.Jira:
-			jiraConfig, err := config.GetConfig[*jira.Config](cfgIntegration.Source)
-			if err != nil {
-				return err
-			}
-
-			if err := webhook.SetupService(ctx, oasRouter, &jiraConfig.Configuration, pg); err != nil {
+			if err := jira.AddSourceToRouter(ctx, source, pg, oasRouter); err != nil {
 				return fmt.Errorf("%w: %s", errSetupSource, err)
 			}
-
 		case "test":
 			// do nothing only for testing
 			return nil

internal/sinks/fake/fake.go

@@ -88,6 +88,13 @@ func (f *Writer) Calls() Calls {
 	return f.stub
 }
 
+func (f *Writer) ResetCalls() {
+	f.mtx.Lock()
+	defer f.mtx.Unlock()
+
+	f.stub = Calls{}
+}
+
 func (f *Writer) AddMock(mock Mock) {
 	f.mtx.Lock()
 	defer f.mtx.Unlock()

internal/sinks/fake/fake_test.go

@@ -67,6 +67,21 @@ func TestImplementWriter(t *testing.T) {
 		}, f.Calls().LastCall())
 	})
 
+	t.Run("ResetCalls clean calls", func(t *testing.T) {
+		f := New(config)
+
+		event := &entities.Event{
+			ID:            "id",
+			OperationType: entities.Write,
+		}
+		err := f.WriteData(context.Background(), event)
+		require.NoError(t, err)
+
+		require.Len(t, f.Calls(), 1)
+		f.ResetCalls()
+		require.Len(t, f.Calls(), 0)
+	})
+
 	t.Run("mock error write", func(t *testing.T) {
 		f := New(config)
 

internal/sinks/mongo/mongo.go

@@ -172,10 +172,9 @@ func (w *Writer[T]) Delete(ctx context.Context, data T) error {
 		return err
 	}
 
-	opts := options.Delete()
 	result, err := w.client.Database(w.database).
 		Collection(w.collection).
-		DeleteOne(ctxWithCancel, queryFilter, opts)
+		DeleteOne(ctxWithCancel, queryFilter)
 	if err != nil {
 		return err
 	}