feat: edfs refactor kafka and nats with a datasource per field

Makefile

@@ -10,13 +10,15 @@ setup-dev-tools: setup-build-tools
 	go install honnef.co/go/tools/cmd/staticcheck@2024.1.1
 	go install github.com/yannh/kubeconform/cmd/kubeconform@v0.6.3
 	go install github.com/norwoodj/helm-docs/cmd/helm-docs@v1.11.3
+	go install github.com/vektra/mockery/v3@v3.3.1
 
 prerequisites: setup-dev-tools
 	go version
 	pnpm -v
 	node -v
 	docker -v
 	dbmate -v
+	mockery version
 
 infra-up: dc-dev
 

demo/cmd/mood/main.go

@@ -3,11 +3,12 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood"
 	"log"
 	"net/http"
 	"os"
 
+	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood"
+
 	"github.com/99designs/gqlgen/graphql"
 	"github.com/99designs/gqlgen/graphql/handler/debug"
 	"github.com/99designs/gqlgen/graphql/playground"
@@ -31,7 +32,9 @@ func main() {
 		port = defaultPort
 	}
 
-	srv := subgraphs.NewDemoServer(mood.NewSchema(nil))
+	srv := subgraphs.NewDemoServer(mood.NewSchema(nil, func(name string) string {
+		return name
+	}))
 
 	srv.Use(&debug.Tracer{})
 	srv.Use(otelgqlgen.Middleware(otelgqlgen.WithCreateSpanFromFields(func(ctx *graphql.FieldContext) bool {

demo/go.mod

@@ -150,6 +150,7 @@ require (
 	github.com/twmb/franz-go/pkg/kmsg v1.7.0 // indirect
 	github.com/urfave/cli/v2 v2.27.5 // indirect
 	github.com/wundergraph/astjson v0.0.0-20250106123708-be463c97e083 // indirect
+	github.com/wundergraph/graphql-go-tools/v2 v2.0.0-rc.183 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
 	github.com/xeipuuv/gojsonschema v1.2.0 // indirect

demo/pkg/subgraphs/availability/availability.go

@@ -2,13 +2,13 @@ package availability
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/availability/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/availability/subgraph/generated"
 )
 
-func NewSchema(pubSubBySourceName map[string]pubsub_datasource.NatsPubSub, pubSubName func(string) string) graphql.ExecutableSchema {
+func NewSchema(pubSubBySourceName map[string]nats.Adapter, pubSubName func(string) string) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: pubSubBySourceName,
 		GetPubSubName:          pubSubName,

demo/pkg/subgraphs/availability/subgraph/resolver.go

@@ -1,14 +1,14 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.Adapter
 	GetPubSubName          func(string) string
 }

demo/pkg/subgraphs/countries/countries.go

@@ -2,13 +2,13 @@ package countries
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/countries/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/countries/subgraph/generated"
 )
 
-func NewSchema(pubSubBySourceName map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(pubSubBySourceName map[string]nats.Adapter) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: pubSubBySourceName,
 	}})

demo/pkg/subgraphs/countries/subgraph/resolver.go

@@ -1,8 +1,9 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"sync"
+
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -11,5 +12,5 @@ import (
 
 type Resolver struct {
 	mux                    sync.Mutex
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.Adapter
 }

demo/pkg/subgraphs/employees/employees.go

@@ -2,13 +2,12 @@ package employees
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
-
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/employees/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/employees/subgraph/generated"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.Adapter) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
 		EmployeesData:          subgraph.Employees,

demo/pkg/subgraphs/employees/subgraph/resolver.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/employees/subgraph/model"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -15,7 +15,7 @@ import (
 
 type Resolver struct {
 	mux                    sync.Mutex
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.Adapter
 	EmployeesData          []*model.Employee
 }
 

demo/pkg/subgraphs/family/family.go

@@ -2,13 +2,13 @@ package family
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/family/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/family/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.Adapter) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
 	}})

demo/pkg/subgraphs/family/subgraph/resolver.go

@@ -1,13 +1,13 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.Adapter
 }

demo/pkg/subgraphs/hobbies/hobbies.go

@@ -2,13 +2,13 @@ package hobbies
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/hobbies/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/hobbies/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.Adapter) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
 	}})

demo/pkg/subgraphs/hobbies/subgraph/resolver.go

@@ -1,18 +1,18 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"reflect"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/hobbies/subgraph/model"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.Adapter
 }
 
 func (r *Resolver) Employees(hobby model.Hobby) ([]*model.Employee, error) {

demo/pkg/subgraphs/mood/mood.go

@@ -2,14 +2,15 @@ package mood
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/mood/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.Adapter, getPubSubName func(string) string) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID: natsPubSubByProviderID,
+		GetPubSubName:          getPubSubName,
 	}})
 }

demo/pkg/subgraphs/mood/subgraph/resolver.go

@@ -1,14 +1,14 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
 //
 // It serves as dependency injection for your app, add any dependencies you require here.
 
 type Resolver struct {
-	NatsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID map[string]nats.Adapter
 	GetPubSubName          func(string) string
 }

demo/pkg/subgraphs/products/products.go

@@ -2,13 +2,13 @@ package products
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.Adapter) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID:       natsPubSubByProviderID,
 		TopSecretFederationFactsData: subgraph.TopSecretFederationFacts,

demo/pkg/subgraphs/products/subgraph/resolver.go

@@ -1,9 +1,10 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph/model"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"sync"
+
+	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products/subgraph/model"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -12,6 +13,6 @@ import (
 
 type Resolver struct {
 	mux                          sync.Mutex
-	NatsPubSubByProviderID       map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID       map[string]nats.Adapter
 	TopSecretFederationFactsData []model.TopSecretFact
 }

demo/pkg/subgraphs/products_fg/products.go

@@ -2,13 +2,13 @@ package products_fg
 
 import (
 	"github.com/99designs/gqlgen/graphql"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph"
 	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph/generated"
 )
 
-func NewSchema(natsPubSubByProviderID map[string]pubsub_datasource.NatsPubSub) graphql.ExecutableSchema {
+func NewSchema(natsPubSubByProviderID map[string]nats.Adapter) graphql.ExecutableSchema {
 	return generated.NewExecutableSchema(generated.Config{Resolvers: &subgraph.Resolver{
 		NatsPubSubByProviderID:       natsPubSubByProviderID,
 		TopSecretFederationFactsData: subgraph.TopSecretFederationFacts,

demo/pkg/subgraphs/products_fg/subgraph/resolver.go

@@ -1,9 +1,10 @@
 package subgraph
 
 import (
-	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph/model"
-	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/pubsub_datasource"
 	"sync"
+
+	"github.com/wundergraph/cosmo/demo/pkg/subgraphs/products_fg/subgraph/model"
+	"github.com/wundergraph/cosmo/router/pkg/pubsub/nats"
 )
 
 // This file will not be regenerated automatically.
@@ -12,6 +13,6 @@ import (
 
 type Resolver struct {
 	mux                          sync.Mutex
-	NatsPubSubByProviderID       map[string]pubsub_datasource.NatsPubSub
+	NatsPubSubByProviderID       map[string]nats.Adapter
 	TopSecretFederationFactsData []model.TopSecretFact
 }