Major update to be up to date with substreams-sink

.github/workflows/docker.yml

@@ -32,7 +32,7 @@ jobs:
       - name: Install Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.19.x
+          go-version: 1.22.x
 
       - uses: actions/cache@v2
         with:

.github/workflows/main.yml

@@ -17,7 +17,7 @@ jobs:
     if: "${{ !startsWith(github.event.head_commit.message, 'GitBook: [#') }}"
     strategy:
       matrix:
-        go-version: [1.19.x]
+        go-version: [1.22.x]
         os: [ubuntu-latest]
     steps:
       - name: Set up Go

CHANGELOG.md

@@ -7,4 +7,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+* Updated to
+
 * `graphload run` now accepts `+<count>` like `+10_000` to run for 10 000 relative to module's start block.

bundler/encoder.go

@@ -3,12 +3,17 @@ package bundler
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/golang/protobuf/proto"
+
+	"google.golang.org/protobuf/proto"
 )
 
 type Encoder func(proto.Message) ([]byte, error)
 
 func JSONLEncode(message proto.Message) ([]byte, error) {
+	return JSONLEncodeAny(message)
+}
+
+func JSONLEncodeAny(message any) ([]byte, error) {
 	buf := []byte{}
 	data, err := json.Marshal(message)
 	if err != nil {

bundler/stats.go

@@ -1,10 +1,11 @@
 package bundler
 
 import (
+	"time"
+
 	"github.com/streamingfast/bstream"
 	"github.com/streamingfast/dmetrics"
 	"go.uber.org/zap"
-	"time"
 )
 
 type boundaryStats struct {
@@ -63,11 +64,11 @@ func (s *boundaryStats) Log() []zap.Field {
 		zap.Duration("boundary_process_duration", s.boundaryProcessTime),
 		zap.Duration("upload_duration", s.uploadedDuration),
 		zap.Duration("data_process_duration", s.procesingDataTime),
-		zap.Float64("avg_upload_dur", s.avgUploadDuration.Average()),
-		zap.Float64("total_upload_dur", s.avgUploadDuration.Total()),
-		zap.Float64("avg_boundary_process_dur", s.avgBoundaryProcessDuration.Average()),
-		zap.Float64("total_boundary_process_dur", s.avgBoundaryProcessDuration.Total()),
-		zap.Float64("avg_data_process_dur", s.avgDataProcessDuration.Average()),
-		zap.Float64("total_data_process_dur", s.avgDataProcessDuration.Total()),
+		zap.Duration("avg_upload_dur", s.avgUploadDuration.Average()),
+		zap.Duration("total_upload_dur", s.avgUploadDuration.Total()),
+		zap.Duration("avg_boundary_process_dur", s.avgBoundaryProcessDuration.Average()),
+		zap.Duration("total_boundary_process_dur", s.avgBoundaryProcessDuration.Total()),
+		zap.Duration("avg_data_process_dur", s.avgDataProcessDuration.Average()),
+		zap.Duration("total_data_process_dur", s.avgDataProcessDuration.Total()),
 	}
 }

cmd/graphload/run.go → cmd/substreams-sink-graph-load/run.go

@@ -21,8 +21,8 @@ import (
 )
 
 const (
-	SUPPORTED_MODULE_TYPE = "sf.substreams.entity.v1.EntityChanges"
-	LEGACY_MODULE_TYPE    = "substreams.entity.v1.EntityChanges"
+	SUPPORTED_MODULE_TYPE = "sf.substreams.sink.entity.v1.EntityChanges"
+	LEGACY_MODULE_TYPE    = "sf.substreams.entity.v1.EntityChanges"
 )
 
 var SinkRunCmd = Command(sinkRunE,

csvprocessor/entity.go

@@ -3,8 +3,8 @@ package csvprocessor
 import (
 	"fmt"
 
-	pbentity "github.com/streamingfast/substreams-graph-load/pb/entity/v1"
 	"github.com/streamingfast/substreams-graph-load/schema"
+	pbentity "github.com/streamingfast/substreams-sink-entity-changes/pb/sf/substreams/sink/entity/v1"
 )
 
 const FieldTypeBigint = "Bigint"
@@ -46,7 +46,7 @@ func (e *Entity) ValidateFields(desc *schema.EntityDesc) error {
 }
 
 func newEntity(in *EntityChangeAtBlockNum, desc *schema.EntityDesc) (*Entity, error) {
-	if in.EntityChange.Operation == pbentity.EntityChange_DELETE {
+	if in.EntityChange.Operation == pbentity.EntityChange_OPERATION_DELETE {
 		return nil, nil
 	}
 

csvprocessor/processor.go

@@ -13,8 +13,8 @@ import (
 	"github.com/streamingfast/dstore"
 	"github.com/streamingfast/logging"
 	"github.com/streamingfast/shutter"
-	pbentity "github.com/streamingfast/substreams-graph-load/pb/entity/v1"
 	"github.com/streamingfast/substreams-graph-load/schema"
+	pbentity "github.com/streamingfast/substreams-sink-entity-changes/pb/sf/substreams/sink/entity/v1"
 	"go.uber.org/zap"
 )
 
@@ -235,7 +235,7 @@ func (p *Processor) processEntityFile(ctx context.Context, filename string) erro
 		prev, found := p.entities[ch.EntityChange.ID]
 
 		switch ch.EntityChange.Operation {
-		case pbentity.EntityChange_CREATE:
+		case pbentity.EntityChange_OPERATION_CREATE:
 			if found {
 				return fmt.Errorf("@%d got CREATE on entity %q but it already exists since block %d", ch.BlockNum, ch.EntityChange.ID, prev.StartBlock)
 			}
@@ -252,7 +252,7 @@ func (p *Processor) processEntityFile(ctx context.Context, filename string) erro
 			}
 			p.entities[ch.EntityChange.ID] = newEnt
 
-		case pbentity.EntityChange_UPDATE:
+		case pbentity.EntityChange_OPERATION_UPDATE:
 			if p.entityDesc.Immutable {
 				if err := newEnt.ValidateFields(p.entityDesc); err != nil {
 					return fmt.Errorf("@%d during UPDATE to an immutable entity: %w", ch.BlockNum, err)
@@ -282,7 +282,7 @@ func (p *Processor) processEntityFile(ctx context.Context, filename string) erro
 			prev.Update(newEnt)
 			p.entities[ch.EntityChange.ID] = prev
 
-		case pbentity.EntityChange_DELETE:
+		case pbentity.EntityChange_OPERATION_DELETE:
 			if p.entityDesc.Immutable {
 				return fmt.Errorf("entity %q got deleted but should be immutable", ch.EntityChange.ID)
 			}
@@ -295,7 +295,7 @@ func (p *Processor) processEntityFile(ctx context.Context, filename string) erro
 			}
 			delete(p.entities, ch.EntityChange.ID)
 
-		case pbentity.EntityChange_FINAL:
+		case pbentity.EntityChange_OPERATION_FINAL:
 			if p.entityDesc.Immutable {
 				continue
 			}

devel/substreams-sink-graphcsv → devel/substreams-sink-graph-load

@@ -19,17 +19,17 @@ main() {
   fi
 
   pushd "$ROOT" &> /dev/null
-    go install -ldflags "-X main.Version=$version -X main.Commit=$commit -X main.IsDirty=$dirty" ./cmd/graphload
+    go install -ldflags "-X main.Version=$version -X main.Commit=$commit -X main.IsDirty=$dirty" ./cmd/substreams-sink-graph-load
   popd &> /dev/null
 
   if [[ $KILL_AFTER != "" ]]; then
-    ${GOPATH:-$HOME/go}/bin/graphload "$@" &
+    ${GOPATH:-$HOME/go}/bin/substreams-sink-graph-load "$@" &
     active_pid=$!
 
     sleep $KILL_AFTER
     kill -s TERM $active_pid &> /dev/null || true
   else
-    exec ${GOPATH:-$HOME/go}/bin/graphload "$@"
+    exec ${GOPATH:-$HOME/go}/bin/substreams-sink-graph-load "$@"
   fi
 }
 

devel/uniswap-v3/start.sh

@@ -8,7 +8,7 @@ main() {
   set -e
 
   out="${OUT_DIR}:-"$ROOT/csv"}"
-  sink="$ROOT/../graphload"
+  sink="$ROOT/../substreams-sink-graph-load"
 
   echo dsn
   set -x