Merge pull request #36 from PowerLoom/feat/migrate-to-onchain-arch

Snapshotter status report structure update and support for http IPFS URL

Author: xadahiya

go/caching/caching.go

@@ -15,7 +15,7 @@ type DbCache interface {
 	CheckIfProjectExists(ctx context.Context, projectID string) (bool, error)
 	StoreProjects(background context.Context, projects []string) error
 	AddUnfinalizedSnapshotCID(ctx context.Context, msg *datamodel.PayloadCommitMessage) error
-	AddSnapshotterStatusReport(ctx context.Context, epochId int, projectId string, report *datamodel.SnapshotterStatusReport) error
+	AddSnapshotterStatusReport(ctx context.Context, epochId int, projectId string, report *datamodel.SnapshotterStatusReport, incrCount bool) error
 	StoreLastFinalizedEpoch(ctx context.Context, projectID string, epochId int) error
 	StoreFinalizedSnapshot(ctx context.Context, msg *datamodel.PowerloomSnapshotFinalizedMessage) error
 	GetFinalizedSnapshotAtEpochID(ctx context.Context, projectID string, epochId int) (*datamodel.PowerloomSnapshotFinalizedMessage, error)

go/caching/redis.go

@@ -194,10 +194,29 @@ func (r *RedisCache) AddUnfinalizedSnapshotCID(ctx context.Context, msg *datamod
 }
 
 // AddSnapshotterStatusReport adds the snapshotter's status report to the given project and epoch ID.
-func (r *RedisCache) AddSnapshotterStatusReport(ctx context.Context, epochId int, projectId string, report *datamodel.SnapshotterStatusReport) error {
+func (r *RedisCache) AddSnapshotterStatusReport(ctx context.Context, epochId int, projectId string, report *datamodel.SnapshotterStatusReport, incrCount bool) error {
 	key := fmt.Sprintf(redisutils.REDIS_KEY_SNAPSHOTTER_STATUS_REPORT, projectId)
 
+	storedReport := new(datamodel.SnapshotterStatusReport)
+
+	reportJsonString, err := r.readClient.HGet(ctx, key, strconv.Itoa(epochId)).Result()
+	if err == nil || reportJsonString != "" {
+		_ = json.Unmarshal([]byte(reportJsonString), storedReport)
+	}
+
 	if report != nil {
+		if storedReport.SubmittedSnapshotCid != "" {
+			report.SubmittedSnapshotCid = storedReport.SubmittedSnapshotCid
+		}
+
+		if storedReport.Reason != "" {
+			report.Reason = storedReport.Reason
+		}
+
+		if storedReport.FinalizedSnapshotCid != "" {
+			report.FinalizedSnapshotCid = storedReport.FinalizedSnapshotCid
+		}
+
 		reportJson, err := json.Marshal(report)
 		if err != nil {
 			log.WithError(err).Error("failed to marshal snapshotter status report")
@@ -222,9 +241,11 @@ func (r *RedisCache) AddSnapshotterStatusReport(ctx context.Context, epochId int
 		key = fmt.Sprintf(redisutils.REDIS_KEY_TOTAL_SUCCESSFUL_SNAPSHOT_COUNT, projectId)
 	}
 
-	err := r.writeClient.Incr(ctx, key).Err()
-	if err != nil {
-		log.WithError(err).Error("failed to increment total missed snapshot count")
+	if incrCount {
+		err = r.writeClient.Incr(ctx, key).Err()
+		if err != nil {
+			log.WithError(err).Error("failed to increment total missed snapshot count")
+		}
 	}
 
 	log.Debug("added snapshotter status report in redis")

go/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/ethereum/go-ethereum v1.11.6
 	github.com/go-playground/validator/v10 v10.14.0
 	github.com/go-redis/redis/v8 v8.11.5
+	github.com/google/uuid v1.3.0
 	github.com/hashicorp/go-retryablehttp v0.7.2
 	github.com/ipfs/go-cid v0.4.1
 	github.com/ipfs/go-ipfs-api v0.6.0
@@ -35,7 +36,6 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-stack/stack v1.8.1 // indirect
-	github.com/google/uuid v1.3.0 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/holiman/uint256 v1.2.2-0.20230321075855-87b91420868c // indirect

go/goutils/datamodel/data_model.go

@@ -68,9 +68,12 @@ type SnapshotRelayerPayload struct {
 }
 
 type SnapshotterStatusReport struct {
-	SubmittedSnapshotCid string                  `json:"submittedSnapshotCid"`
+	SubmittedSnapshotCid string                  `json:"submittedSnapshotCid,omitempty"`
+	SubmittedSnapshot    map[string]interface{}  `json:"submittedSnapshot,omitempty"`
 	FinalizedSnapshotCid string                  `json:"finalizedSnapshotCid"`
+	FinalizedSnapshot    map[string]interface{}  `json:"finalizedSnapshot,omitempty"`
 	State                SnapshotSubmissionState `json:"state"`
+	Reason               string                  `json:"reason"`
 }
 
 type UnfinalizedSnapshot struct {
@@ -87,3 +90,11 @@ type SnapshotterIssue struct {
 	TimeOfReporting string `json:"timeOfReporting"`
 	Extra           string `json:"extra"`
 }
+
+type SnapshotSubmittedEventMessage struct {
+	SnapshotCid string `json:"snapshotCid"`
+	EpochId     int    `json:"epochId"`
+	ProjectId   string `json:"projectId"`
+	BroadcastId string `json:"broadcastId"`
+	Timestamp   int64  `json:"timestamp"`
+}

go/goutils/ipfsutils/ipfs_utils.go

@@ -4,8 +4,11 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
+	"net/url"
+	"strings"
 	"time"
 
 	shell "github.com/ipfs/go-ipfs-api"
@@ -30,16 +33,16 @@ type IpfsClient struct {
 func InitClient(settingsObj *settings.SettingsObj) *IpfsClient {
 	writeUrl := settingsObj.IpfsConfig.URL
 
-	writeUrl, err := ParseMultiAddrURL(writeUrl)
+	writeUrl, err := ParseURL(writeUrl)
 	if err != nil {
-		log.WithError(err).Fatal("failed to parse IPFS write multiaddr URL: ", writeUrl)
+		log.WithError(err).Fatal("failed to parse IPFS write URL: ", writeUrl)
 	}
 
 	readUrl := settingsObj.IpfsConfig.ReaderURL
 
-	readUrl, err = ParseMultiAddrURL(readUrl)
+	readUrl, err = ParseURL(readUrl)
 	if err != nil {
-		log.WithError(err).Fatal("failed to parse IPFS read multiaddr URL: ", readUrl)
+		log.WithError(err).Fatal("failed to parse IPFS read URL: ", readUrl)
 	}
 
 	ipfsReadHTTPClient := httpclient.GetIPFSReadHTTPClient(settingsObj)
@@ -127,27 +130,28 @@ func (e *UnsupportedMultiaddrError) Error() string {
 	return fmt.Sprintf("unsupported multiaddr url pattern: %s", e.URL)
 }
 
-func ParseMultiAddrURL(url string) (string, error) {
+// ParseURL tries to parse a multiaddr URL, if the url is not multiaddr it tries to parse http url.
+func ParseURL(ipfsUrl string) (string, error) {
 	parts := make([]string, 0) // [host,port,scheme]
 
-	if multiaddr, err := ma.NewMultiaddr(url); err == nil {
+	if multiaddr, err := ma.NewMultiaddr(ipfsUrl); err == nil {
 		addrSplits := ma.Split(multiaddr)
 
 		// host and port are required
 		if len(addrSplits) < 2 {
-			return "", &UnsupportedMultiaddrError{URL: url}
+			return "", &UnsupportedMultiaddrError{URL: ipfsUrl}
 		}
 
 		for index, addr := range addrSplits {
 			component, _ := ma.SplitFirst(addr)
 			if index == 1 && component.Protocol().Code != ma.P_TCP {
-				return "", &UnsupportedMultiaddrError{URL: url}
+				return "", &UnsupportedMultiaddrError{URL: ipfsUrl}
 			}
 
 			// check if scheme is present
 			if index == 2 {
 				if component.Protocol().Code != ma.P_HTTP && component.Protocol().Code != ma.P_HTTPS {
-					return "", &UnsupportedMultiaddrError{URL: url}
+					return "", &UnsupportedMultiaddrError{URL: ipfsUrl}
 				}
 
 				parts = append(parts, component.Protocol().Name)
@@ -159,23 +163,37 @@ func ParseMultiAddrURL(url string) (string, error) {
 		}
 
 		if len(parts) < 2 {
-			return "", &UnsupportedMultiaddrError{URL: url}
+			return "", &UnsupportedMultiaddrError{URL: ipfsUrl}
 		}
 
 		// join host and port
-		url = net.JoinHostPort(parts[0], parts[1])
+		ipfsUrl = net.JoinHostPort(parts[0], parts[1])
 
 		// add scheme if present
 		if len(parts) >= 3 {
-			url = fmt.Sprintf("%s://%s", parts[2], url)
+			ipfsUrl = fmt.Sprintf("%s://%s", parts[2], ipfsUrl)
 		} else {
-			url = fmt.Sprintf("http://%s", url) // default to http if scheme is not present
+			ipfsUrl = fmt.Sprintf("http://%s", ipfsUrl) // default to http if scheme is not present
 		}
 	} else {
-		return "", err
+		// parse http url
+		parsedURL, err := url.ParseRequestURI(ipfsUrl)
+		if err != nil {
+			return "", err
+		}
+
+		// check if scheme is http or https
+		if !strings.EqualFold(parsedURL.Scheme, "http") && !strings.EqualFold(parsedURL.Scheme, "https") {
+			return "", fmt.Errorf("unsupported scheme: %s", parsedURL.Scheme)
+		}
+
+		// check if host is present
+		if parsedURL.Host == "" {
+			return "", errors.New("host is required in url")
+		}
 	}
 
-	return url, nil
+	return ipfsUrl, nil
 }
 
 func (client *IpfsClient) UploadSnapshotToIPFS(payloadCommit *datamodel.PayloadCommitMessage) error {

go/goutils/settings/settings.go

@@ -38,11 +38,16 @@ type (
 			Core struct {
 				Exchange              string `json:"exchange"`
 				CommitPayloadExchange string `json:"commit_payload_exchange"`
+				EventDetectorExchange string `json:"event_detector_exchange"`
 			} `json:"core"`
 			PayloadCommit struct {
 				QueueNamePrefix  string `json:"queue_name_prefix"`
 				RoutingKeyPrefix string `json:"routing_key_prefix"`
 			} `json:"payload_commit"`
+			EventDetector struct {
+				QueueNamePrefix  string `json:"queue_name_prefix"`
+				RoutingKeyPrefix string `json:"routing_key_prefix"`
+			} `json:"event_detector"`
 		} `json:"setup"`
 	}