refactor(entities/source): refactor source object

Author: vm-001

admin/api/sources.go

@@ -40,6 +40,12 @@ func (api *API) CreateSource(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if source.Type == "http" {
+		if source.Config.HTTP.Path == "" {
+			source.Config.HTTP.Path = "/" + utils.UUIDShort()
+		}
+	}
+
 	source.WorkspaceId = ucontext.GetWorkspaceID(r.Context())
 	err := api.db.SourcesWS.Insert(r.Context(), &source)
 	api.assert(err)

db/entities/source.go

@@ -19,16 +19,33 @@ func (m CustomResponse) Value() (driver.Value, error) {
 	return json.Marshal(m)
 }
 
+type SourceConfig struct {
+	HTTP HttpSourceConfig `json:"http"`
+}
+
+func (m *SourceConfig) Scan(src interface{}) error {
+	return json.Unmarshal(src.([]byte), m)
+}
+
+func (m SourceConfig) Value() (driver.Value, error) {
+	return json.Marshal(m)
+}
+
+type HttpSourceConfig struct {
+	Path     string          `json:"path"`
+	Methods  Strings         `json:"methods"`
+	Response *CustomResponse `json:"response"`
+}
+
 type Source struct {
-	ID        string          `json:"id" db:"id"`
-	Name      *string         `json:"name" db:"name"`
-	Enabled   bool            `json:"enabled" db:"enabled"`
-	Path      string          `json:"path" db:"path"`
-	Methods   Strings         `json:"methods" db:"methods"`
-	Async     bool            `json:"async" db:"async"`
-	Response  *CustomResponse `json:"response" db:"response"`
-	Metadata  Metadata        `json:"metadata" db:"metadata"`
-	RateLimit *RateLimit      `json:"rate_limit" yaml:"rate_limit" db:"rate_limit"`
+	ID        string       `json:"id" db:"id"`
+	Name      *string      `json:"name" db:"name"`
+	Enabled   bool         `json:"enabled" db:"enabled"`
+	Type      string       `json:"type" db:"type"`
+	Config    SourceConfig `json:"config" db:"config"`
+	Async     bool         `json:"async" db:"async"`
+	Metadata  Metadata     `json:"metadata" db:"metadata"`
+	RateLimit *RateLimit   `json:"rate_limit" yaml:"rate_limit" db:"rate_limit"`
 
 	BaseModel `yaml:"-"`
 }

db/migrations/1762423418_source_config.down.sql

@@ -0,0 +1,11 @@
+ALTER TABLE IF EXISTS ONLY "sources" ADD COLUMN IF NOT EXISTS "path" TEXT;
+ALTER TABLE IF EXISTS ONLY "sources" ADD COLUMN IF NOT EXISTS "methods" TEXT[];
+ALTER TABLE IF EXISTS ONLY "sources" ADD COLUMN IF NOT EXISTS "response" JSONB;
+
+UPDATE sources SET
+  "path" = (config->'http'->>'path'),
+  "methods" = ARRAY(SELECT jsonb_array_elements_text(config->'http'->'methods')),
+  "response" = (config->'http'->'response');
+
+ALTER TABLE IF EXISTS ONLY "sources" DROP COLUMN IF EXISTS "type";
+ALTER TABLE IF EXISTS ONLY "sources" DROP COLUMN IF EXISTS "config";

db/migrations/1762423418_source_config.up.sql

@@ -0,0 +1,8 @@
+ALTER TABLE IF EXISTS ONLY "sources" ADD COLUMN IF NOT EXISTS "type" varchar(20);
+ALTER TABLE IF EXISTS ONLY "sources" ADD COLUMN IF NOT EXISTS "config" JSONB NOT NULL DEFAULT '{}'::jsonb;
+
+UPDATE sources SET "type" = 'http', "config" = jsonb_build_object('http', jsonb_build_object('methods', methods, 'path', path, 'response', response));
+
+ALTER TABLE IF EXISTS ONLY "sources" DROP COLUMN IF EXISTS "path";
+ALTER TABLE IF EXISTS ONLY "sources" DROP COLUMN IF EXISTS "methods";
+ALTER TABLE IF EXISTS ONLY "sources" DROP COLUMN IF EXISTS "response";

examples/function/webhookx-function-sample.yml

@@ -14,12 +14,15 @@ endpoints:
 
 sources:
   - name: github-source
-    path: /github
-    methods: [ "POST" ]
-    response:
-      code: 200
-      content_type: application/json
-      body: '{"message": "OK"}'
+    type: http
+    config:
+      http:
+        path: /github
+        methods: [ "POST" ]
+        response:
+          code: 200
+          content_type: application/json
+          body: '{"message": "OK"}'
     plugins:
       - name: function
         config:

openapi.yml

@@ -877,33 +877,21 @@ components:
         enabled:
           type: boolean
           default: true
-        path:
+        type:
           type: string
-        methods:
-          type: array
-          items:
-            type: string
-            enum: [ GET, POST, PUT, DELETE, PATCH ]
-          minItems: 1
+          enum: [ "http" ]
+          default: http
+        config:
+          type: object
+          properties:
+            http:
+              $ref: "#/components/schemas/HTTPSourceConfig"
+          required:
+            - http
         async:
           type: boolean
           description: "Whether to ingest events asynchronously through the queue"
           default: false
-        response:
-          type: object
-          nullable: true
-          properties:
-            code:
-              type: integer
-              minimum: 200
-              maximum: 599
-            content_type:
-              type: string
-            body:
-              type: string
-          required:
-            - code
-            - content_type
         metadata:
           $ref: "#/components/schemas/Metadata"
         rate_limit:
@@ -915,8 +903,7 @@ components:
           type: integer
           readOnly: true
       required:
-        - path
-        - methods
+        - config
 
     Plugin:
       type: object
@@ -1049,3 +1036,33 @@ components:
                 type: string
                 format: jsonschema
                 maxLength: 1048576
+
+    HTTPSourceConfig:
+      type: object
+      default:
+        methods: [ "POST" ]
+      properties:
+        path:
+          type: string
+        methods:
+          type: array
+          items:
+            type: string
+            enum: [ GET, POST, PUT, DELETE, PATCH ]
+          minItems: 1
+          default: [ "POST" ]
+        response:
+          type: object
+          nullable: true
+          properties:
+            code:
+              type: integer
+              minimum: 200
+              maximum: 599
+            content_type:
+              type: string
+            body:
+              type: string
+          required:
+            - code
+            - content_type

proxy/gateway.go

@@ -169,8 +169,8 @@ func (gw *Gateway) buildRouter(version string) {
 	routes := make([]*router.Route, 0)
 	for _, source := range sources {
 		route := router.Route{
-			Paths:   []string{source.Path},
-			Methods: source.Methods,
+			Paths:   []string{source.Config.HTTP.Path},
+			Methods: source.Config.HTTP.Methods,
 			Handler: source,
 		}
 		routes = append(routes, &route)
@@ -204,7 +204,7 @@ func (gw *Gateway) handle(w http.ResponseWriter, r *http.Request) bool {
 		span.SetAttributes(attribute.String("source.name", utils.PointerValue(source.Name)))
 		span.SetAttributes(attribute.String("source.workspace_id", source.WorkspaceId))
 		span.SetAttributes(attribute.Bool("source.async", source.Async))
-		span.SetAttributes(semconv.HTTPRoute(source.Path))
+		span.SetAttributes(semconv.HTTPRoute(source.Config.HTTP.Path))
 		defer span.End()
 		ctx = tracingCtx
 	}
@@ -306,9 +306,9 @@ func (gw *Gateway) handle(w http.ResponseWriter, r *http.Request) bool {
 		headers[constants.HeaderEventId] = event.ID
 	}
 
-	if source.Response != nil {
-		headers["Content-Type"] = source.Response.ContentType
-		exit(w, source.Response.Code, source.Response.Body, headers)
+	if source.Config.HTTP.Response != nil {
+		headers["Content-Type"] = source.Config.HTTP.Response.ContentType
+		exit(w, source.Config.HTTP.Response.Code, source.Config.HTTP.Response.Body, headers)
 		return true
 	}
 

test/README.md

@@ -12,3 +12,26 @@ make deps
 ```shell
 make test-integration
 ```
+
+
+# How-To
+
+#### How to run specific tests?
+
+`FIt`, `FContext`.
+
+```
+Context("some specs you're debugging", func() {
+  It("might be failing", func() { ... })
+  FIt("might also be failing", func() { ... })
+})
+
+```
+
+```
+ginkgo
+```
+
+This will run only test "might also be failing", skip the rest.
+
+See https://onsi.github.io/ginkgo/#focused-specs.

test/admin/sources_test.go

@@ -38,10 +38,7 @@ var _ = Describe("/sources", Ordered, func() {
 	Context("POST", func() {
 		It("creates a source", func() {
 			resp, err := adminClient.R().
-				SetBody(map[string]interface{}{
-					"path":    "/v1",
-					"methods": []string{"POST"},
-				}).
+				SetBody(`{ "type": "http", "config": { "http": { "path": "" } }}`).
 				SetResult(entities.Source{}).
 				Post("/workspaces/default/sources")
 			assert.Nil(GinkgoT(), err)
@@ -51,10 +48,10 @@ var _ = Describe("/sources", Ordered, func() {
 			result := resp.Result().(*entities.Source)
 			assert.NotNil(GinkgoT(), result.ID)
 			assert.Equal(GinkgoT(), true, result.Enabled)
-			assert.Equal(GinkgoT(), "/v1", result.Path)
-			assert.EqualValues(GinkgoT(), []string{"POST"}, result.Methods)
+			assert.True(GinkgoT(), len(result.Config.HTTP.Path) > 0) // auto generate path
+			assert.EqualValues(GinkgoT(), []string{"POST"}, result.Config.HTTP.Methods)
 			assert.Equal(GinkgoT(), false, result.Async)
-			assert.True(GinkgoT(), nil == result.Response)
+			assert.True(GinkgoT(), nil == result.Config.HTTP.Response)
 
 			e, err := db.Sources.Get(context.TODO(), result.ID)
 			assert.Nil(GinkgoT(), err)
@@ -70,7 +67,7 @@ var _ = Describe("/sources", Ordered, func() {
 				assert.Nil(GinkgoT(), err)
 				assert.Equal(GinkgoT(), 400, resp.StatusCode())
 				assert.Equal(GinkgoT(),
-					`{"message":"Request Validation","error":{"message":"request validation","fields":{"methods":"required field missing","path":"required field missing"}}}`,
+					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":"required field missing"}}}`,
 					string(resp.Body()))
 			})
 		})
@@ -166,14 +163,19 @@ var _ = Describe("/sources", Ordered, func() {
 			It("updates by id", func() {
 				resp, err := adminClient.R().
 					SetBody(map[string]interface{}{
-						"path":    "/v1",
-						"methods": []string{"GET", "POST", "PUT", "DELETE"},
-						"async":   true,
-						"response": map[string]interface{}{
-							"code":         200,
-							"content_type": "text/plain",
-							"body":         "OK",
+						"type": "http",
+						"config": map[string]interface{}{
+							"http": map[string]interface{}{
+								"path":    "/v1",
+								"methods": []string{"GET", "POST", "PUT", "DELETE"},
+								"response": map[string]interface{}{
+									"code":         200,
+									"content_type": "text/plain",
+									"body":         "OK",
+								},
+							},
 						},
+						"async": true,
 					}).
 					SetResult(entities.Source{}).
 					Put("/workspaces/default/sources/" + entity.ID)
@@ -183,14 +185,14 @@ var _ = Describe("/sources", Ordered, func() {
 				result := resp.Result().(*entities.Source)
 
 				assert.Equal(GinkgoT(), entity.ID, result.ID)
-				assert.Equal(GinkgoT(), "/v1", result.Path)
-				assert.EqualValues(GinkgoT(), []string{"GET", "POST", "PUT", "DELETE"}, result.Methods)
+				assert.Equal(GinkgoT(), "/v1", result.Config.HTTP.Path)
+				assert.EqualValues(GinkgoT(), []string{"GET", "POST", "PUT", "DELETE"}, result.Config.HTTP.Methods)
 				assert.Equal(GinkgoT(), true, result.Async)
 				assert.EqualValues(GinkgoT(), &entities.CustomResponse{
 					Code:        200,
 					ContentType: "text/plain",
 					Body:        "OK",
-				}, result.Response)
+				}, result.Config.HTTP.Response)
 			})
 		})
 

test/cmd/admin_test.go

@@ -87,11 +87,11 @@ var _ = Describe("admin", Ordered, func() {
 				assert.NoError(GinkgoT(), err)
 				assert.Equal(GinkgoT(), "default-source", *source.Name)
 				assert.Equal(GinkgoT(), true, source.Enabled)
-				assert.Equal(GinkgoT(), "/", source.Path)
-				assert.Equal(GinkgoT(), []string{"POST"}, []string(source.Methods))
-				assert.Equal(GinkgoT(), 200, source.Response.Code)
-				assert.Equal(GinkgoT(), "application/json", source.Response.ContentType)
-				assert.Equal(GinkgoT(), `{"message": "OK"}`, source.Response.Body)
+				assert.Equal(GinkgoT(), "/", source.Config.HTTP.Path)
+				assert.Equal(GinkgoT(), []string{"POST"}, []string(source.Config.HTTP.Methods))
+				assert.Equal(GinkgoT(), 200, source.Config.HTTP.Response.Code)
+				assert.Equal(GinkgoT(), "application/json", source.Config.HTTP.Response.ContentType)
+				assert.Equal(GinkgoT(), `{"message": "OK"}`, source.Config.HTTP.Response.Body)
 
 				plugins, err := db.Plugins.ListEndpointPlugin(context.TODO(), endpoint.ID)
 				assert.NoError(GinkgoT(), err)