Feature/22 test handling incoming events - Intermediary Commit (#28)

* Checkpoint before the weekend

* Update location for SetupPostgresDB

go.mod

@@ -10,7 +10,7 @@ require (
 )
 
 require (
-	github.com/ethereum/go-ethereum v1.10.17-0.20220323200026-535f25d65fa0 // indirect
+	github.com/ethereum/go-ethereum v1.10.17 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.10.0 // indirect
 	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
@@ -25,10 +25,12 @@ require (
 	github.com/minio/sha256-simd v1.0.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 // indirect
+	github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc // indirect
+	github.com/r3labs/sse/v2 v2.7.7 // indirect
 	github.com/thomaso-mirodin/intmath v0.0.0-20160323211736-5dc6d854e46e // indirect
 	golang.org/x/crypto v0.0.0-20220427172511-eb4f295cb31f // indirect
 	golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 // indirect
-	google.golang.org/genproto v0.0.0-20220503193339-ba3ae3f07e29 // indirect
+	google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3 // indirect
 	google.golang.org/grpc v1.46.0 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/cenkalti/backoff.v1 v1.1.0 // indirect
@@ -41,12 +43,12 @@ require (
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/jackc/pgx/v4 v4.16.0
+	github.com/julienschmidt/httprouter v1.3.0
 	github.com/magiconair/properties v1.8.6 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.0 // indirect
 	github.com/prysmaticlabs/prysm v1.4.2-0.20220504145118-df695346a53c
-	github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc
 	github.com/spf13/afero v1.8.2 // indirect
 	github.com/spf13/cast v1.4.1 // indirect
 	github.com/spf13/cobra v1.4.0

go.sum

@@ -132,6 +132,8 @@ github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/ethereum/go-ethereum v1.10.17-0.20220323200026-535f25d65fa0 h1:dZ/6iVmQ9XIKyp5V8TGoQmnpckp5bd2y/No31jZGhZU=
 github.com/ethereum/go-ethereum v1.10.17-0.20220323200026-535f25d65fa0/go.mod h1:Lt5WzjM07XlXc95YzrhosmR4J9Ahd6X2wyEV2SvGhk0=
+github.com/ethereum/go-ethereum v1.10.17 h1:XEcumY+qSr1cZQaWsQs5Kck3FHB0V2RiMHPdTBJ+oT8=
+github.com/ethereum/go-ethereum v1.10.17/go.mod h1:Lt5WzjM07XlXc95YzrhosmR4J9Ahd6X2wyEV2SvGhk0=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/ferranbt/fastssz v0.0.0-20220303160658-88bb965b6747 h1:K2Bt7NSX8x/5MD2RiO7cPLy21dBgnQ84r9uR0QYoHrE=
 github.com/ferranbt/fastssz v0.0.0-20220303160658-88bb965b6747/go.mod h1:S8yiDeAXy8f88W4Ul+0dBMPx49S05byYbmZD6Uv94K4=
@@ -248,6 +250,7 @@ github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORR
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/graph-gophers/graphql-go v1.3.0/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
+github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.10.0 h1:ESEyqQqXXFIcImj/BE8oKEX37Zsuceb2cZI+EL/zNCY=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.10.0/go.mod h1:XnLCLFp3tjoZJszVKjfpyAK6J8sYIcQXWQxmqLWF21I=
@@ -277,6 +280,7 @@ github.com/influxdata/promql/v2 v2.12.0/go.mod h1:fxOPu+DY0bqCTCECchSRtWfc+0X19y
 github.com/influxdata/roaring v0.4.13-0.20180809181101-fc520f41fab6/go.mod h1:bSgUQ7q5ZLSO+bKBGqJiCBGAl+9DxyW63zLTujjUlOE=
 github.com/influxdata/tdigest v0.0.0-20181121200506-bf2b5ad3c0a9/go.mod h1:Js0mqiSBE6Ffsg94weZZ2c+v/ciT8QRHFOap7EKDrR0=
 github.com/influxdata/usage-client v0.0.0-20160829180054-6d3895376368/go.mod h1:Wbbw6tYNvwa5dlB6304Sd+82Z3f7PmVZHVKU637d4po=
+github.com/jackc/chunkreader v1.0.0 h1:4s39bBR8ByfqH+DKm8rQA3E1LHZWB9XWcrz8fqaZbe0=
 github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9QQ7T3kePo=
 github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
 github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
@@ -300,6 +304,7 @@ github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5W
 github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgproto3 v1.1.0 h1:FYYE4yRw+AgI8wXIinMlNjBbp/UitDJwfj5LqqewP1A=
 github.com/jackc/pgproto3 v1.1.0/go.mod h1:eR5FA3leWg7p9aeAqi37XOTgTIbkABlvcPB3E5rlc78=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190420180111-c116219b62db/go.mod h1:bhq50y+xrl9n5mRYyCBFKkpRVTLYJVWeCc+mEAI3yXA=
 github.com/jackc/pgproto3/v2 v2.0.0-alpha1.0.20190609003834-432c2951c711/go.mod h1:uH0AWtUmuShn0bcesswc4aBTWGvw0cAxIJp+6OB//Wg=
@@ -353,6 +358,8 @@ github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/X
 github.com/jsternberg/zap-logfmt v1.0.0/go.mod h1:uvPs/4X51zdkcm5jXl5SYoN+4RK21K8mysFmDaM/h+o=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
+github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4dN7jwJOQ1U=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/jwilder/encoding v0.0.0-20170811194829-b4e1701a28ef/go.mod h1:Ct9fl0F6iIOGgxJ5npU/IUOhOhqlVrGjyIZc8/MagT0=
 github.com/karalabe/usb v0.0.2/go.mod h1:Od972xHfMJowv7NGVDiWVxk2zxnWgjLlJzE+F4F7AGU=
@@ -477,6 +484,8 @@ github.com/prysmaticlabs/prysm v1.4.2-0.20220504145118-df695346a53c h1:OBWrQQlyA
 github.com/prysmaticlabs/prysm v1.4.2-0.20220504145118-df695346a53c/go.mod h1:z8vCgEaKYKD+6R5MRZvIOEw8aAsFydBidRbxvqtMpcU=
 github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc h1:zAsgcP8MhzAbhMnB1QQ2O7ZhWYVGYSR2iVcjzQuPV+o=
 github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc/go.mod h1:S8xSOnV3CgpNrWd0GQ/OoQfMtlg2uPRSuTzcSGrzwK8=
+github.com/r3labs/sse/v2 v2.7.7 h1:SRXB3/N22d59LAtLvbVzmXp1Z3IQg3jb8OziZHv/sas=
+github.com/r3labs/sse/v2 v2.7.7/go.mod h1:Igau6Whc+F17QUgML1fYe1VPZzTV6EMCnYktEmkNJ7I=
 github.com/retailnext/hllpp v1.0.1-0.20180308014038-101a6d2f8b52/go.mod h1:RDpi1RftBQPUCDRw6SmxeaREsAaRKnOclghuzp/WRzc=
 github.com/rjeczalik/notify v0.9.1/go.mod h1:rKwnCoCGeuQnwBtTSPL9Dad03Vh2n40ePRrjvIXnJho=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
@@ -916,6 +925,8 @@ google.golang.org/genproto v0.0.0-20210108203827-ffc7fda8c3d7/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20210226172003-ab064af71705/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20220503193339-ba3ae3f07e29 h1:DJUvgAPiJWeMBiT+RzBVcJGQN7bAEWS5UEoMshES9xs=
 google.golang.org/genproto v0.0.0-20220503193339-ba3ae3f07e29/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4=
+google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3 h1:q1kiSVscqoDeqTF27eQ2NnLLDmqF0I373qQNXYMy0fo=
+google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=

internal/boot/boot.go

@@ -8,7 +8,6 @@ import (
 	"github.com/vulcanize/ipld-ethcl-indexer/pkg/beaconclient"
 	"github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql"
 	"github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql/postgres"
-	"github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper"
 )
 
 var (
@@ -18,36 +17,6 @@ var (
 	BC            *beaconclient.BeaconClient = &beaconclient.BeaconClient{}
 )
 
-// A simple wrapper to create a DB object to use.
-func SetupPostgresDb(dbHostname string, dbPort int, dbName string, dbUsername string, dbPassword string, driverName string) (sql.Database, error) {
-	log.Debug("Resolving Driver Type")
-	DbDriver, err := postgres.ResolveDriverType(driverName)
-	if err != nil {
-		log.WithFields(log.Fields{
-			"err":                  err,
-			"driver_name_provided": driverName,
-		}).Error("Can't resolve driver type")
-	}
-	log.Info("Using Driver:", DbDriver)
-
-	postgresConfig := postgres.Config{
-		Hostname:     dbHostname,
-		Port:         dbPort,
-		DatabaseName: dbName,
-		Username:     dbUsername,
-		Password:     dbPassword,
-		Driver:       DbDriver,
-	}
-	DB, err = postgres.NewPostgresDB(postgresConfig)
-
-	if err != nil {
-		loghelper.LogError(err).Error("Unable to connect to the DB")
-		return nil, err
-	}
-	return DB, err
-
-}
-
 // This function will perform some boot operations. If any steps fail, the application will fail to start.
 // Keep in mind that the DB connection can be lost later in the lifecycle of the application or
 // it might not be able to connect to the beacon client.
@@ -69,7 +38,7 @@ func BootApplication(ctx context.Context, dbHostname string, dbPort int, dbName
 	}
 
 	log.Debug("Setting up DB connection")
-	DB, err := SetupPostgresDb(dbHostname, dbPort, dbName, dbUsername, dbPassword, driverName)
+	DB, err := postgres.SetupPostgresDb(dbHostname, dbPort, dbName, dbUsername, dbPassword, driverName)
 	if err != nil {
 		return nil, nil, err
 	}

pkg/beaconclient/beaconclient.go

@@ -12,10 +12,10 @@ import (
 // TODO: Use prysms config values instead of hardcoding them here.
 var (
 	bcHealthEndpoint     = "/eth/v1/node/health"               // Endpoint used for the healthcheck
-	bcHeadTopicEndpoint  = "/eth/v1/events?topics=head"        // Endpoint used to subscribe to the head of the chain
+	BcHeadTopicEndpoint  = "/eth/v1/events?topics=head"        // Endpoint used to subscribe to the head of the chain
 	bcReorgTopicEndpoint = "/eth/v1/events?topics=chain_reorg" // Endpoint used to subscribe to the head of the chain
-	bcBlockQueryEndpoint = "/eth/v2/beacon/blocks/"            // Endpoint to query individual Blocks
+	BcBlockQueryEndpoint = "/eth/v2/beacon/blocks/"            // Endpoint to query individual Blocks
-	bcStateQueryEndpoint = "/eth/v2/debug/beacon/states/"      // Endpoint to query individual States
+	BcStateQueryEndpoint = "/eth/v2/debug/beacon/states/"      // Endpoint to query individual States
 	bcSlotsPerEpoch      = 32                                  // Number of slots in a single Epoch
 	//bcSlotPerHistoricalVector = 8192                                // The number of slots in a historic vector.
 	//bcFinalizedTopicEndpoint  = "/eth/v1/events?topics=finalized_checkpoint" // Endpoint used to subscribe to the head of the chain
@@ -61,7 +61,7 @@ func CreateBeaconClient(ctx context.Context, connectionProtocol string, bcAddres
 	return &BeaconClient{
 		Context:        ctx,
 		ServerEndpoint: endpoint,
-		HeadTracking:   createSseEvent[Head](endpoint, bcHeadTopicEndpoint),
+		HeadTracking:   createSseEvent[Head](endpoint, BcHeadTopicEndpoint),
 		ReOrgTracking:  createSseEvent[ChainReorg](endpoint, bcReorgTopicEndpoint),
 		//FinalizationTracking: createSseEvent[FinalizedCheckpoint](endpoint, bcFinalizedTopicEndpoint),
 	}

pkg/beaconclient/capturehead_test.go

@@ -0,0 +1,198 @@
+package beaconclient_test
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	"github.com/sirupsen/logrus"
+
+	//	. "github.com/onsi/gomega"
+	"github.com/r3labs/sse/v2"
+	"github.com/vulcanize/ipld-ethcl-indexer/pkg/beaconclient"
+)
+
+type Message struct {
+	HeadMessage       beaconclient.Head       // The head messsage that will be streamed to the BeaconClient
+	ReorgMessage      beaconclient.ChainReorg // The reorg messsage that will be streamed to the BeaconClient
+	TestNotes         string                  // A small explanation of the purpose this structure plays in the testing landscape.
+	SignedBeaconBlock string                  // The file path output of an SSZ encoded SignedBeaconBlock.
+	BeaconState       string                  // The file path output of an SSZ encoded BeaconState.
+	SuccessfulDBQuery string                  // A string that indicates what a query to the DB should output to pass the test.
+}
+
+var TestEvents map[string]*Message
+
+var _ = Describe("Capturehead", func() {
+	TestEvents = map[string]*Message{
+		"100": {
+			HeadMessage: beaconclient.Head{
+				Slot:                      "100",
+				Block:                     "0x582187e97f7520bb69eea014c3834c964c45259372a0eaaea3f032013797996b",
+				State:                     "0xf286a0379c0386a3c7be28d05d829f8eb7b280cc9ede15449af20ebcd06a7a56",
+				CurrentDutyDependentRoot:  "",
+				PreviousDutyDependentRoot: "",
+				EpochTransition:           false,
+				ExecutionOptimistic:       false,
+			},
+			TestNotes:         "This is a simple, easy to process block.",
+			SignedBeaconBlock: filepath.Join("data", "100", "signed-beacon-block.ssz"),
+			BeaconState:       filepath.Join("data", "100", "beacon-state.ssz"),
+		},
+		"101": {
+			HeadMessage: beaconclient.Head{
+				Slot:                      "101",
+				Block:                     "0xabe1a972e512182d04f0d4a5c9c25f9ee57c2e9d0ff3f4c4c82fd42d13d31083",
+				State:                     "0xcb04aa2edbf13c7bb7e7bd9b621ced6832e0075e89147352eac3019a824ce847",
+				CurrentDutyDependentRoot:  "",
+				PreviousDutyDependentRoot: "",
+				EpochTransition:           false,
+				ExecutionOptimistic:       false,
+			},
+			TestNotes:         "This is a simple, easy to process block.",
+			SignedBeaconBlock: filepath.Join("data", "101", "signed-beacon-block.ssz"),
+			BeaconState:       filepath.Join("data", "101", "beacon-state.ssz"),
+		},
+	}
+
+	// We might also want to add an integration test that will actually process a single event, then end.
+	// This will help us know that our models match that actual data being served from the beacon node.
+
+	Describe("Receiving New Head SSE messages", Label("unit"), func() {
+		Context("Correctly formatted", Label("dry"), func() {
+			It("Should turn it into a struct successfully.", func() {
+				server := createSseServer()
+				logrus.Info("DONE!")
+				client := sse.NewClient("http://localhost:8080" + beaconclient.BcHeadTopicEndpoint)
+
+				logrus.Info("DONE!")
+				ch := make(chan *sse.Event)
+				go client.SubscribeChanRaw(ch)
+
+				time.Sleep(2 * time.Second)
+				logrus.Info("DONE!")
+				sendMessageToStream(server, []byte("hello"))
+				client.Unsubscribe(ch)
+				val := <-ch
+
+				logrus.Info("DONE!")
+				logrus.Info(val)
+			})
+		})
+		//Context("A single incorrectly formatted", func() {
+		//	It("Should create an error, maybe also add the projected slot to the knownGaps table......")
+		//})
+		//Context("An incorrectly formatted message sandwiched between correctly formatted messages", func() {
+		//	It("Should create an error, maybe also add the projected slot to the knownGaps table......")
+		//})
+	})
+
+	//Describe("Receiving New Reorg SSE messages", Label("unit"), func() {
+	//	Context("Reorg slot is already in the DB", func() {
+	//		It("The previous block should be marked as 'forked', the new block should be the only one marked as 'proposed'.")
+	//	})
+	//	Context("Multiple reorgs have occurred on this slot", func() {
+	//		It("The previous blocks should be marked as 'forked', the new block should be the only one marked as 'proposed'.")
+	//	})
+	//	Context("Reorg slot in not already in the DB", func() {
+	//		It("Should simply have the correct slot in the DB.")
+	//	})
+
+	//})
+
+	//Describe("Querying SignedBeaconBlock and Beacon State", Label("unit"), func() {
+	//	Context("When the slot is properly served by the beacon node", func() {
+	//		It("Should provide a successful response.")
+	//	})
+	//	Context("When there is a skipped slot", func() {
+	//		It("Should indicate that the slot was skipped")
+	//		// Future use case.
+
+	//	})
+	//	Context("When the slot is not properly served", func() {
+	//		It("Should return an error, and add the slot to the knownGaps table.")
+	//	})
+	//})
+
+	//Describe("Receiving properly formatted Head SSE events.", Label("unit"), func() {
+	//	Context("In sequential order", func() {
+	//		It("Should write each event to the DB successfully.")
+	//	})
+	//	Context("With gaps in slots", func() {
+	//		It("Should add the slots in between to the knownGaps table")
+	//	})
+	//	Context("With a repeat slot", func() {
+	//		It("Should recognize the reorg and process it.")
+	//	})
+	//	Context("With the previousBlockHash not matching the parentBlockHash", func() {
+	//		It("Should recognize the reorg and add the previous slot to knownGaps table.")
+	//	})
+	//	Context("Out of order", func() {
+	//		It("Not sure what it should do....")
+	//	})
+	//	Context("With a skipped slot", func() {
+	//		It("Should recognize the slot as skipped and continue without error.")
+	//		// Future use case
+	//	})
+	//})
+})
+
+// Create a new Sse.Server.
+func createSseServer() *sse.Server {
+	// server := sse.New()
+	// server.CreateStream("")
+
+	mux := http.NewServeMux()
+	//mux.HandleFunc(beaconclient.BcHeadTopicEndpoint, func(w http.ResponseWriter, r *http.Request) {
+	//	go func() {
+	//		// Received Browser Disconnection
+	//		<-r.Context().Done()
+	//		println("The client is disconnected here")
+	//		return
+	//	}()
+
+	//	server.ServeHTTP(w, r)
+	//})
+	mux.HandleFunc(beaconclient.BcStateQueryEndpoint, provideState)
+	mux.HandleFunc(beaconclient.BcBlockQueryEndpoint, provideBlock)
+	go http.ListenAndServe(":8080", mux)
+	return server
+}
+
+// Send messages to the stream.
+func sendMessageToStream(server *sse.Server, data []byte) {
+	server.Publish("", &sse.Event{
+		Data: data,
+	})
+	logrus.Info("publish complete")
+}
+
+// A function to mimic querying the state from the beacon node. We simply get the SSZ file are return it.
+func provideState(w http.ResponseWriter, req *http.Request) {
+	path := strings.Split(req.URL.Path, "/")
+	slot := path[len(path)-1]
+	slotFile := "data/" + slot + "/beacon-state.ssz"
+	dat, err := os.ReadFile(slotFile)
+	if err != nil {
+		fmt.Fprintf(w, "Can't find the slot file, %s", slotFile)
+	}
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Write(dat)
+}
+
+// A function to mimic querying the state from the beacon node. We simply get the SSZ file are return it.
+func provideBlock(w http.ResponseWriter, req *http.Request) {
+	path := strings.Split(req.URL.Path, "/")
+	slot := path[len(path)-1]
+	slotFile := "data/" + slot + "/signed-beacon-block.ssz"
+	dat, err := os.ReadFile(slotFile)
+	if err != nil {
+		fmt.Fprintf(w, "Can't find the slot file, %s", slotFile)
+	}
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Write(dat)
+}

pkg/beaconclient/processslot.go

@@ -116,7 +116,7 @@ func (ps *ProcessSlot) getSignedBeaconBlock(serverAddress string) error {
 		log.Error(MissingIdentifiedError)
 		return fmt.Errorf(MissingIdentifiedError)
 	}
-	blockEndpoint := serverAddress + bcBlockQueryEndpoint + blockIdentifier
+	blockEndpoint := serverAddress + BcBlockQueryEndpoint + blockIdentifier
 	var err error
 	var rc int
 	ps.SszSignedBeaconBlock, rc, err = querySsz(blockEndpoint, strconv.Itoa(ps.Slot))
@@ -156,7 +156,7 @@ func (ps *ProcessSlot) getBeaconState(serverEndpoint string) error {
 		log.Error(MissingIdentifiedError)
 		return fmt.Errorf(MissingIdentifiedError)
 	}
-	stateEndpoint := serverEndpoint + bcStateQueryEndpoint + stateIdentifier
+	stateEndpoint := serverEndpoint + BcStateQueryEndpoint + stateIdentifier
 	ps.SszBeaconState, _, _ = querySsz(stateEndpoint, strconv.Itoa(ps.Slot))
 
 	ps.FullBeaconState = new(st.BeaconState)

pkg/database/sql/postgres/database.go

@@ -6,6 +6,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 	"github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql"
+	"github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper"
 )
 
 var _ sql.Database = &DB{}
@@ -23,6 +24,35 @@ func NewPostgresDB(c Config) (*DB, error) {
 	return &DB{driver}, nil
 }
 
+// A simple wrapper to create a DB object to use.
+func SetupPostgresDb(dbHostname string, dbPort int, dbName string, dbUsername string, dbPassword string, driverName string) (sql.Database, error) {
+	log.Debug("Resolving Driver Type")
+	DbDriver, err := ResolveDriverType(driverName)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"err":                  err,
+			"driver_name_provided": driverName,
+		}).Error("Can't resolve driver type")
+	}
+	log.Info("Using Driver:", DbDriver)
+
+	postgresConfig := Config{
+		Hostname:     dbHostname,
+		Port:         dbPort,
+		DatabaseName: dbName,
+		Username:     dbUsername,
+		Password:     dbPassword,
+		Driver:       DbDriver,
+	}
+	DB, err := NewPostgresDB(postgresConfig)
+
+	if err != nil {
+		loghelper.LogError(err).Error("Unable to connect to the DB")
+		return nil, err
+	}
+	return DB, err
+}
+
 // Create a driver based on the config
 func createDriver(c Config) (*pgxDriver, error) {
 	switch c.Driver {