fix: move Gql from broker to consumer

.github/workflows/go.yaml

@@ -23,13 +23,13 @@ jobs:
           cache: false
 
       - name: Install Tools
-        working-directory: "src"
+        working-directory: "internal"
         run: make tools
 
       - name: Lint
-        working-directory: "src"
+        working-directory: "internal"
         run: make lint
 
       - name: Error Check
-        working-directory: "src"
+        working-directory: "internal"
         run: make errors

.github/workflows/release.yaml

@@ -22,18 +22,18 @@ jobs:
         with:
           go-version: "stable"
           cache-dependency-path: |
-            src/go.sum
+            internal/go.sum
 
       - name: Build
-        run: cd src && make prod
+        run: cd internal && make prod
 
       - name: "Run release script"
         run: node scripts/docker-release.mjs ${{ github.ref_name }}
 
       - name: "Release"
         uses: softprops/action-gh-release@v1
         with:
-          files: ./src/bin/*
+          files: ./internal/bin/*
           name: Unchained ${{ github.ref_name}}
           tag_name: ${{ github.ref_name }}
           prerelease: ${{ contains(github.ref_name, 'alpha') || contains(github.ref_name, 'beta') }}

.gitignore

@@ -17,6 +17,6 @@ docker/data
 docker/pgadmin
 .clinic
 node_trace.*
-src/bin
+internal/bin
 conf.*.private
 .idea

conf.worker.yaml.template

@@ -2,8 +2,9 @@ system:
   log: info
   name: <name>
 
-broker:
-  uri: wss://shinobi.brokers.kenshi.io
+network:
+  bind: 127.0.0.1:9123
+  broker-uri: ws://127.0.0.1:9123
 
 rpc:
   - name: ethereum

go.work

@@ -1,6 +1,6 @@
 go 1.22.1
 
 use (
-	./src
+	./internal
 	./zk
 )

go.work.sum

@@ -15,8 +15,6 @@ github.com/Azure/azure-sdk-for-go/sdk/internal v1.3.0/go.mod h1:okt5dMMTOFjX/aov
 github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.2.0 h1:gggzg0SUMs6SQbEw+3LoSsYf9YMjkupeAnHMX8O9mmY=
 github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.2.0/go.mod h1:+6KLcKIVgxoBDMqMO/Nvy7bZ9a0nbU3I1DtFQK3YvB4=
 github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDOSA=
-github.com/BurntSushi/toml v1.3.2 h1:o7IhLm0Msx3BaB+n3Ag7L8EVlByGnpq14C4YWiu/gL8=
-github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
 github.com/StackExchange/wmi v1.2.1/go.mod h1:rcmrprowKIVzvc+NUiLncP2uuArMWLCbu9SBzvHz7e8=
 github.com/aead/siphash v1.0.1 h1:FwHfE/T45KPKYuuSAKyyvE+oPWcaQ+CUmFW0bPlM+kg=
 github.com/alecthomas/kong v0.7.0 h1:YIjJUiR7AcmHxL87UlbPn0gyIGwl4+nYND0OQ4ojP7k=

src/.golangci.yml → internal/.golangci.yml

@@ -4,7 +4,7 @@ run:
 linters-settings:
   cyclop:
     max-complexity: 35
-    package-average: 13.0
+    package-average: 25.0
 
   gosec:
     excludes:

internal/cmd/broker.go

@@ -0,0 +1,45 @@
+package cmd
+
+import (
+	"github.com/KenshiTech/unchained/config"
+	"github.com/KenshiTech/unchained/constants"
+	"github.com/KenshiTech/unchained/crypto/bls"
+	"github.com/KenshiTech/unchained/ethereum"
+	"github.com/KenshiTech/unchained/log"
+	"github.com/KenshiTech/unchained/pos"
+	"github.com/KenshiTech/unchained/transport/server"
+	"github.com/KenshiTech/unchained/transport/server/websocket"
+
+	"github.com/spf13/cobra"
+)
+
+var brokerCmd = &cobra.Command{
+	Use:   "broker",
+	Short: "Run the Unchained client in broker mode",
+	Long:  `Run the Unchained client in broker mode`,
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Start(config.App.System.Log)
+		log.Logger.
+			With("Version", constants.Version).
+			With("Protocol", constants.ProtocolVersion).
+			Info("Running Unchained | Broker")
+
+		err := config.Load(configPath, secretsPath)
+		if err != nil {
+			panic(err)
+		}
+
+		bls.InitClientIdentity()
+
+		ethRPC := ethereum.New()
+		pos.New(ethRPC)
+
+		server.New(
+			websocket.WithWebsocket(),
+		)
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(brokerCmd)
+}

src/cmd/consumer.go → internal/cmd/consumer.go

@@ -1,75 +1,66 @@
-/*
-Copyright © 2024 NAME HERE <EMAIL ADDRESS>
-*/
 package cmd
 
 import (
 	"github.com/KenshiTech/unchained/config"
 	"github.com/KenshiTech/unchained/constants"
-	"github.com/KenshiTech/unchained/consumers"
 	"github.com/KenshiTech/unchained/crypto/bls"
 	"github.com/KenshiTech/unchained/db"
 	"github.com/KenshiTech/unchained/ethereum"
 	"github.com/KenshiTech/unchained/log"
-	"github.com/KenshiTech/unchained/net/client"
-	"github.com/KenshiTech/unchained/plugins/correctness"
-	"github.com/KenshiTech/unchained/plugins/logs"
-	"github.com/KenshiTech/unchained/plugins/uniswap"
 	"github.com/KenshiTech/unchained/pos"
-
+	correctnessService "github.com/KenshiTech/unchained/service/correctness"
+	evmlogService "github.com/KenshiTech/unchained/service/evmlog"
+	uniswapService "github.com/KenshiTech/unchained/service/uniswap"
+	"github.com/KenshiTech/unchained/transport/client"
+	"github.com/KenshiTech/unchained/transport/client/handler"
+	"github.com/KenshiTech/unchained/transport/server"
+	"github.com/KenshiTech/unchained/transport/server/gql"
 	"github.com/spf13/cobra"
 )
 
-// consumerCmd represents the consumer command.
 var consumerCmd = &cobra.Command{
 	Use:   "consumer",
 	Short: "Run the Unchained client in consumer mode",
 	Long:  `Run the Unchained client in consumer mode`,
 
 	PreRun: func(cmd *cobra.Command, args []string) {
-		config.App.Broker.URI = cmd.Flags().Lookup("broker").Value.String()
+		config.App.Network.BrokerURI = cmd.Flags().Lookup("broker").Value.String()
 	},
 
 	Run: func(cmd *cobra.Command, args []string) {
+		log.Start(config.App.System.Log)
+		log.Logger.
+			With("Version", constants.Version).
+			With("Protocol", constants.ProtocolVersion).
+			Info("Running Unchained | Consumer")
 
 		err := config.Load(configPath, secretsPath)
 		if err != nil {
 			panic(err)
 		}
 
-		log.Start(config.App.System.Log)
-
-		log.Logger.
-			With("Version", constants.Version).
-			With("Protocol", constants.ProtocolVersion).
-			Info("Running Unchained")
-
-		ethereum.Start()
 		bls.InitClientIdentity()
-		pos.Start()
+
+		ethRPC := ethereum.New()
+		pos := pos.New(ethRPC)
 		db.Start()
-		uniswap.New()
-		correctness.New()
-		logs.New()
-		client.StartClient()
-		consumers.StartConsumer()
-		client.Listen()
+
+		server.New(
+			gql.WithGraphQL(),
+		)
+
+		correctnessService := correctnessService.New(ethRPC)
+		evmLogService := evmlogService.New(ethRPC, pos)
+		uniswapService := uniswapService.New(ethRPC, pos)
+
+		handler := handler.New(correctnessService, uniswapService, evmLogService)
+		client.Consume(handler)
 	},
 }
 
 func init() {
 	rootCmd.AddCommand(consumerCmd)
 
-	// Here you will define your flags and configuration settings.
-
-	// Cobra supports Persistent Flags which will work for this command
-	// and all subcommands, e.g.:
-	// consumerCmd.PersistentFlags().String("foo", "", "A help for foo")
-
-	// Cobra supports local flags which will only run when this command
-	// is called directly, e.g.:
-	// consumerCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
-
 	consumerCmd.Flags().StringP(
 		"broker",
 		"b",

src/cmd/root.go → internal/cmd/root.go

@@ -15,13 +15,11 @@ var secretsPath string
 var contextPath string
 var printVersion bool
 
-// rootCmd represents the base command when called without any subcommands.
 var rootCmd = &cobra.Command{
 	Use:   "unchained",
 	Short: "Unchained is the universal data validation and processing protocol",
 	Long:  `Unchained is the universal data validation and processing protocol`,
-	// Uncomment the following line if your bare application
-	// has an action associated with it:
+
 	Run: func(cmd *cobra.Command, args []string) {
 		if printVersion {
 			fmt.Println(constants.Version)
@@ -31,8 +29,6 @@ var rootCmd = &cobra.Command{
 	},
 }
 
-// Execute adds all child commands to the root command and sets flags appropriately.
-// This is called by main.main(). It only needs to happen once to the rootCmd.
 func Execute() {
 	err := rootCmd.Execute()
 	if err != nil {
@@ -41,14 +37,6 @@ func Execute() {
 }
 
 func init() {
-	// Here you will define your flags and configuration settings.
-	// Cobra supports persistent flags, which, if defined here,
-	// will be global for your application.
-
-	// rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.unchained.yaml)")
-
-	// Cobra also supports local flags, which will only run
-	// when this action is called directly.
 	rootCmd.Flags().BoolVarP(&printVersion, "version", "v", false, "Print the Unchained version number and die")
 
 	rootCmd.PersistentFlags().StringVarP(&configPath, "config", "c", "./conf.yaml", "Config file")

src/cmd/worker.go → internal/cmd/worker.go

@@ -1,21 +1,19 @@
-/*
-Copyright © 2024 NAME HERE <EMAIL ADDRESS>
-*/
 package cmd
 
 import (
 	"github.com/KenshiTech/unchained/config"
 	"github.com/KenshiTech/unchained/constants"
 	"github.com/KenshiTech/unchained/crypto/bls"
-	"github.com/KenshiTech/unchained/db"
 	"github.com/KenshiTech/unchained/ethereum"
 	"github.com/KenshiTech/unchained/log"
-	"github.com/KenshiTech/unchained/net/client"
 	"github.com/KenshiTech/unchained/persistence"
-	"github.com/KenshiTech/unchained/plugins/logs"
-	"github.com/KenshiTech/unchained/plugins/uniswap"
 	"github.com/KenshiTech/unchained/pos"
-
+	"github.com/KenshiTech/unchained/scheduler"
+	correctnessService "github.com/KenshiTech/unchained/service/correctness"
+	evmlogService "github.com/KenshiTech/unchained/service/evmlog"
+	uniswapService "github.com/KenshiTech/unchained/service/uniswap"
+	"github.com/KenshiTech/unchained/transport/client"
+	"github.com/KenshiTech/unchained/transport/client/handler"
 	"github.com/spf13/cobra"
 )
 
@@ -26,49 +24,44 @@ var workerCmd = &cobra.Command{
 	Long:  `Run the Unchained client in worker mode`,
 
 	PreRun: func(cmd *cobra.Command, args []string) {
-		config.App.Broker.URI = cmd.Flags().Lookup("broker").Value.String()
+		config.App.Network.BrokerURI = cmd.Flags().Lookup("broker").Value.String()
 	},
 
 	Run: func(cmd *cobra.Command, args []string) {
-		err := config.Load(configPath, secretsPath)
-		if err != nil {
-			panic(err)
-		}
-
 		log.Start(config.App.System.Log)
 		log.Logger.
 			With("Version", constants.Version).
 			With("Protocol", constants.ProtocolVersion).
-			Info("Running Unchained")
+			Info("Running Unchained | Worker")
+
+		err := config.Load(configPath, secretsPath)
+		if err != nil {
+			panic(err)
+		}
 
-		ethereum.Start()
 		bls.InitClientIdentity()
-		pos.Start()
-		db.Start()
-		client.StartClient()
 
-		uniswap.Listen()
-		logs.New()
-		logs.Listen()
+		ethRPC := ethereum.New()
+		pos := pos.New(ethRPC)
+		badger := persistence.New(contextPath)
+
+		correctnessService := correctnessService.New(ethRPC)
+		evmLogService := evmlogService.New(ethRPC, pos)
+		uniswapService := uniswapService.New(ethRPC, pos)
 
-		persistence.Start(contextPath)
-		client.Listen()
+		scheduler.New(
+			scheduler.WithEthLogs(evmLogService, ethRPC, badger),
+			scheduler.WithUniswapEvents(uniswapService, ethRPC),
+		)
+
+		handler := handler.New(correctnessService, uniswapService, evmLogService)
+		client.Consume(handler)
 	},
 }
 
 func init() {
 	rootCmd.AddCommand(workerCmd)
 
-	// Here you will define your flags and configuration settings.
-
-	// Cobra supports Persistent Flags which will work for this command
-	// and all subcommands, e.g.:
-	// workerCmd.PersistentFlags().String("foo", "", "A help for foo")
-
-	// Cobra supports local flags which will only run when this command
-	// is called directly, e.g.:
-	// workerCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
-
 	workerCmd.Flags().StringP(
 		"broker",
 		"b",