refactor: use protocol (ODP) to better align Whitepaper (#581)

Makefile

@@ -115,7 +115,7 @@ image: generate
 # Use `make service_up` to start all required services using docker-compose
 # Use `make service_down` to remove all required services
 #
-# Run a worker service locally to index and structure data from a decentralized source
+# Run a worker service locally to index and structure data from a decentralized protocol
 # into the RSS3 Protocol format. Use `make worker WORKER_ID=<worker-id>`
 worker: generate
 	@if [ -z "$(WORKER_ID)" ]; then \

config/config.go

@@ -17,6 +17,7 @@ import (
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/schema/worker"
 	"github.com/rss3-network/node/schema/worker/federated"
+	"github.com/rss3-network/node/schema/worker/rss"
 	"github.com/rss3-network/protocol-go/schema/network"
 	"github.com/samber/lo"
 	"github.com/spf13/viper"
@@ -262,6 +263,13 @@ func _Setup(configName, configType string, v *viper.Viper) (*File, error) {
 		}
 	}
 
+	// Add extra logic to convert RSS worker string to correct worker type.
+	if configFile.Component.RSS != nil {
+		if rssWorker := rss.GetValueByWorkerStr(configFile.Component.RSS.Worker.Name()); rssWorker != 0 {
+			configFile.Component.RSS.Worker = rssWorker
+		}
+	}
+
 	// Use a function to load the endpoint for each module, because mapstructure doesn't support the use of custom unmarshaler.
 	// Reference https://github.com/mitchellh/mapstructure/issues/115.
 	if err := configFile.LoadModulesEndpoint(); err != nil {

config/config_test.go

@@ -65,8 +65,8 @@ observability:
       endpoint: localhost:4318
 component:
   rss:
-      network: rss
-      worker: rsshub
+      network: rsshub
+      worker: core
       endpoint: https://rsshub.app/
       parameters:
         authentication:
@@ -154,8 +154,8 @@ component:
   },
   "component": {
   "rss": {
-    "network": "rss",
-    "worker": "rsshub",
+    "network": "rsshub",
+    "worker": "core",
     "endpoint": "https://rsshub.app/",
     "parameters": {
       "authentication": {
@@ -249,8 +249,8 @@ insecure = true
 endpoint = "localhost:4318"
 
 [component.rss]
-network = "rss"
-worker = "rsshub"
+network = "rsshub"
+worker = "core"
 endpoint = "https://rsshub.app/"
 
 [component.rss.parameters.authentication]
@@ -317,12 +317,12 @@ var configFileExpected = &File{
 	},
 	Component: &Component{
 		RSS: &Module{
-			Network:    network.RSS,
+			Network:    network.RSSHub,
 			EndpointID: "https://rsshub.app/",
 			Endpoint: Endpoint{
 				URL: "https://rsshub.app/",
 			},
-			Worker: rss.RSSHub,
+			Worker: rss.Core,
 			Parameters: &Parameters{
 				"authentication": map[string]any{
 					"access_code": "def",

config/parameter/network.go

@@ -221,7 +221,7 @@ func buildCurrentEpochCacheKey() string {
 
 // buildNetworkBlockStartCacheKey builds the cache key for the network block start
 func buildNetworkBlockStartCacheKey(network string) string {
-	return fmt.Sprintf("source:network:start:%s", strings.ToLower(network))
+	return fmt.Sprintf("protocol:network:start:%s", strings.ToLower(network))
 }
 
 // InitVSLClient initializes the VSL client

docs/openapi.go

@@ -69,7 +69,7 @@ func generateEnum(file []byte) ([]byte, error) {
 			network.Unknown.String(),
 			network.Bitcoin.String(),
 			network.SatoshiVM.String(),
-			network.RSS.String(),
+			network.RSSHub.String(),
 		}, s)
 	})
 

go.mod

@@ -63,7 +63,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/redis/rueidis v1.0.45
 	github.com/redis/rueidis/rueidiscompat v1.0.45
-	github.com/rss3-network/protocol-go v0.5.10
+	github.com/rss3-network/protocol-go v0.5.12
 	github.com/spf13/afero v1.11.0
 	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7
 	github.com/tidwall/sjson v1.2.5

go.sum

@@ -390,8 +390,8 @@ github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
-github.com/rss3-network/protocol-go v0.5.10 h1:U+CXrw36V7UmuPgW+s7iPFQ5YtB+YIDlOpJwegnZz2A=
-github.com/rss3-network/protocol-go v0.5.10/go.mod h1:npcyduWt86uVbIi77xQaYk8eqltI9XNjk1FMGpjyIq0=
+github.com/rss3-network/protocol-go v0.5.12 h1:hDqTa3Ft0PDM6/IhXGC0kOY4Ogig7TuM3UvQ7U9cM2A=
+github.com/rss3-network/protocol-go v0.5.12/go.mod h1:npcyduWt86uVbIi77xQaYk8eqltI9XNjk1FMGpjyIq0=
 github.com/russross/blackfriday v1.6.0 h1:KqfZb0pUVN2lYqZUYRddxF4OR8ZMURnJIG5Y3VRLtww=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=

internal/engine/data_source.go

@@ -7,7 +7,7 @@ import (
 	"github.com/rss3-network/protocol-go/schema/network"
 )
 
-// DataSource is the interface that wraps the basic methods of a data source.
+// DataSource is the interface that wraps the basic methods of a data protocol.
 type DataSource interface {
 	Network() network.Network
 	State() json.RawMessage

internal/engine/source/activitypub/data_source.go → internal/engine/protocol/activitypub/data_source.go

@@ -24,7 +24,7 @@ var _ engine.DataSource = (*dataSource)(nil)
 
 var DefaultStartTime int64
 
-// dataSource struct defines the fields for the data source
+// dataSource struct defines the fields for the data protocol
 type dataSource struct {
 	config         *config.Module
 	databaseClient database.Client
@@ -42,7 +42,7 @@ func (s *dataSource) State() json.RawMessage {
 	return lo.Must(json.Marshal(s.state))
 }
 
-// Start initializes the data source and starts consuming Kafka messages
+// Start initializes the data protocol and starts consuming Kafka messages
 func (s *dataSource) Start(ctx context.Context, tasksChan chan<- *engine.Tasks, errorChan chan<- error) {
 	if err := s.initialize(); err != nil {
 		errorChan <- fmt.Errorf("initialize dataSource: %w", err)
@@ -176,7 +176,7 @@ func (s *dataSource) buildMastodonMessageTasks(_ context.Context, object activit
 	return &tasks
 }
 
-// NewSource creates a new data source instance
+// NewSource creates a new data protocol instance
 func NewSource(config *config.Module, checkpoint *engine.Checkpoint, databaseClient database.Client) (engine.DataSource, error) {
 	var (
 		state State

internal/engine/source/activitypub/state.go → internal/engine/protocol/activitypub/state.go

@@ -1,6 +1,6 @@
 package activitypub
 
-// State represents the state of the ActivityPub data source.
+// State represents the state of the ActivityPub data protocol.
 type State struct {
 	LastOffset int64 `json:"last_offset"`
 }

internal/engine/source/ethereum/data_source_test.go → internal/engine/protocol/ethereum/data_source_test.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	"github.com/rss3-network/node/internal/engine/source/ethereum"
+	"github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
@@ -36,7 +36,7 @@ func TestSource(t *testing.T) {
 		want      require.ValueAssertionFunc
 		wantError require.ErrorAssertionFunc
 	}{
-		// TODO Implement a solution to configure custom block number ranges for source.
+		// TODO Implement a solution to configure custom block number ranges for protocol.
 		// {
 		// 	name: "From block number 15537393 to 15537398",
 		// 	want: func(t require.TestingT, actual interface{}, msgAndArgs ...interface{}) {

internal/engine/source/factory.go → internal/engine/protocol/factory.go

@@ -1,4 +1,4 @@
-package source
+package protocol
 
 import (
 	"fmt"
@@ -7,28 +7,28 @@ import (
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/database"
 	"github.com/rss3-network/node/internal/engine"
-	"github.com/rss3-network/node/internal/engine/source/activitypub"
-	"github.com/rss3-network/node/internal/engine/source/arweave"
-	"github.com/rss3-network/node/internal/engine/source/ethereum"
-	"github.com/rss3-network/node/internal/engine/source/farcaster"
-	"github.com/rss3-network/node/internal/engine/source/near"
+	"github.com/rss3-network/node/internal/engine/protocol/activitypub"
+	"github.com/rss3-network/node/internal/engine/protocol/arweave"
+	"github.com/rss3-network/node/internal/engine/protocol/ethereum"
+	"github.com/rss3-network/node/internal/engine/protocol/farcaster"
+	"github.com/rss3-network/node/internal/engine/protocol/near"
 	"github.com/rss3-network/protocol-go/schema/network"
 )
 
-// New creates a new source.
+// New creates a new protocol.
 func New(config *config.Module, sourceFilter engine.DataSourceFilter, checkpoint *engine.Checkpoint, databaseClient database.Client, redisClient rueidis.Client) (engine.DataSource, error) {
-	switch config.Network.Source() {
-	case network.EthereumSource:
+	switch config.Network.Protocol() {
+	case network.EthereumProtocol:
 		return ethereum.NewSource(config, sourceFilter, checkpoint, redisClient)
-	case network.ArweaveSource:
+	case network.ArweaveProtocol:
 		return arweave.NewSource(config, sourceFilter, checkpoint, redisClient)
-	case network.FarcasterSource:
+	case network.FarcasterProtocol:
 		return farcaster.NewSource(config, checkpoint, databaseClient)
-	case network.ActivityPubSource:
+	case network.ActivityPubProtocol:
 		return activitypub.NewSource(config, checkpoint, databaseClient)
-	case network.NearSource:
+	case network.NearProtocol:
 		return near.NewSource(config, sourceFilter, checkpoint, redisClient)
 	default:
-		return nil, fmt.Errorf("unsupported network source %s", config.Network)
+		return nil, fmt.Errorf("unsupported network protocol %s", config.Network)
 	}
 }

internal/engine/worker/decentralized/contract/1inch/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/1inch"
@@ -68,7 +68,7 @@ func (w *worker) Types() []schema.Type {
 	}
 }
 
-// Filter returns a filter for source.
+// Filter returns a filter for protocol.
 func (w *worker) Filter() engine.DataSourceFilter {
 	return &source.Filter{
 		LogAddresses: []common.Address{

internal/engine/worker/decentralized/contract/1inch/worker_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	worker "github.com/rss3-network/node/internal/engine/worker/decentralized/contract/1inch"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract/1inch"

internal/engine/worker/decentralized/contract/aave/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/aave"

internal/engine/worker/decentralized/contract/aave/worker_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	worker "github.com/rss3-network/node/internal/engine/worker/decentralized/contract/aave"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract/aave"

internal/engine/worker/decentralized/contract/aavegotchi/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract/aavegotchi"
 	"github.com/rss3-network/node/provider/ethereum/contract/erc20"
@@ -59,7 +59,7 @@ func (w *worker) Types() []schema.Type {
 	}
 }
 
-// Filter filters the source for Aavegotchi.
+// Filter filters the protocol for Aavegotchi.
 func (w *worker) Filter() engine.DataSourceFilter {
 	return &source.Filter{
 		LogAddresses: []common.Address{

internal/engine/worker/decentralized/contract/aavegotchi/worker_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/internal/engine/worker/decentralized/contract/aavegotchi"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/endpoint"

internal/engine/worker/decentralized/contract/arbitrum/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/arbitrum"

internal/engine/worker/decentralized/contract/arbitrum/worker_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	worker "github.com/rss3-network/node/internal/engine/worker/decentralized/contract/arbitrum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/endpoint"

internal/engine/worker/decentralized/contract/base/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/base"
@@ -62,7 +62,7 @@ func (w *worker) Types() []schema.Type {
 	}
 }
 
-// Filter returns a filter for source.
+// Filter returns a filter for protocol.
 func (w *worker) Filter() engine.DataSourceFilter {
 	return &source.Filter{
 		LogAddresses: []common.Address{

internal/engine/worker/decentralized/contract/base/worker_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	worker "github.com/rss3-network/node/internal/engine/worker/decentralized/contract/base"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract/base"

internal/engine/worker/decentralized/contract/benddao/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/benddao"

internal/engine/worker/decentralized/contract/benddao/worker_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	worker "github.com/rss3-network/node/internal/engine/worker/decentralized/contract/benddao"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract/benddao"

internal/engine/worker/decentralized/contract/cow/worker.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/cow"
@@ -148,7 +148,7 @@ func (w *worker) transformSettlementTradeLog(ctx context.Context, task *source.T
 	return actions, nil
 }
 
-// func (w *worker) buildTransactionTransferAction(ctx context.Context, task *source.Task, from, to common.Address, tokenAddress *common.Address, amount *big.Int) (*activityx.Action, error) {
+// func (w *worker) buildTransactionTransferAction(ctx context.Context, task *protocol.Task, from, to common.Address, tokenAddress *common.Address, amount *big.Int) (*activityx.Action, error) {
 //	tokenMetadata, err := w.tokenClient.Lookup(ctx, task.ChainID, tokenAddress, nil, task.Header.Number)
 //	if err != nil {
 //		return nil, fmt.Errorf("lookup token metadata: %w", err)

internal/engine/worker/decentralized/contract/cow/worker_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/rss3-network/node/config"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	worker "github.com/rss3-network/node/internal/engine/worker/decentralized/contract/cow"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/endpoint"

internal/engine/worker/decentralized/contract/crossbell/worker.go

@@ -14,7 +14,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/rss3-network/node/config"
 	"github.com/rss3-network/node/internal/engine"
-	source "github.com/rss3-network/node/internal/engine/source/ethereum"
+	source "github.com/rss3-network/node/internal/engine/protocol/ethereum"
 	"github.com/rss3-network/node/provider/ethereum"
 	"github.com/rss3-network/node/provider/ethereum/contract"
 	"github.com/rss3-network/node/provider/ethereum/contract/crossbell"