feat: wnfs stream verify system

.github/workflows/docker.yml

@@ -16,32 +16,36 @@ env:
 permissions:
   id-token: write # This is required for requesting the JWT
   contents: read # This is required for actions/checkout
+  packages: write
 
 jobs:
   build:
     name: Build for multi-platform
     if: github.repository_owner == 'dataverse-os' 
     runs-on: ubuntu-latest
+    permissions:
+      packages: write
     steps:
       - name: checkout
         uses: actions/checkout@v3
 
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v1
 
-      - name: Login to DockerHub
-        if: github.event_name != 'pull_request'
+      - name: Login to GitHub Container Registry
         uses: docker/login-action@v2
         with:
-          username: dataverseos
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ github.token }}
 
-      - name: Docker metaw
+      - name: Docker meta
         id: meta
         uses: docker/metadata-action@v4
         with:
           # list of Docker images to use as base name for tags
-          images: dataverseos/dapp-backend
+          images: |
+            ghcr.io/dataverse-os/dapp-backend
           # generate Docker tags based on the following events/attributes
           tags: |
             type=schedule
@@ -59,7 +63,7 @@ jobs:
         uses: docker/build-push-action@v3
         with:
           context: .
-          push: ${{ github.event_name != 'pull_request' }}
+          push: true
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
 

.github/workflows/golangci-lint.yml

@@ -37,3 +37,5 @@ jobs:
 
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3
+        with:
+          args: --timeout=30m

.github/workflows/prerun-docker.yml

@@ -55,5 +55,4 @@ jobs:
           context: ./prerun
           push: ${{ github.event_name != 'pull_request' }}
           tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          platforms: ${{ matrix.PLATFORM }}
+          labels: ${{ steps.meta.outputs.labels }}

ceramic/collection.go

@@ -0,0 +1,8 @@
+package ceramic
+
+type Collection struct {
+	Edges []struct {
+		Cursor string      `json:"cursor"`
+		Node   StreamState `json:"node"`
+	} `json:"edges"`
+}

ceramic/commit_utils.go

@@ -0,0 +1,31 @@
+package ceramic
+
+import (
+	"io"
+
+	"github.com/ipfs/go-cid"
+	"github.com/ipld/go-ipld-prime/codec/dagcbor"
+	"github.com/ipld/go-ipld-prime/datamodel"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/ipld/go-ipld-prime/node/basicnode"
+	"github.com/samber/lo"
+)
+
+func MustParseLink(nd datamodel.Node, key string) cid.Cid {
+	link := lo.Must(lo.Must(nd.LookupByString(key)).AsLink()).(cidlink.Link)
+	return link.Cid
+}
+
+func DecodeDagCborNodeDataFromReader(reader io.Reader) (nd datamodel.Node, err error) {
+	builder := basicnode.Prototype.Map.NewBuilder()
+	if err = dagcbor.Decode(builder, reader); err != nil {
+		return
+	}
+	nd = builder.Build()
+	return
+}
+
+func ContainField(nd datamodel.Node, key string) bool {
+	v, err := nd.LookupByString(key)
+	return err == nil && v != nil
+}

ceramic/commits.go

@@ -0,0 +1,264 @@
+package ceramic
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/ceramicnetwork/go-dag-jose/dagjose"
+	"github.com/ipfs/boxo/coreiface/path"
+	"github.com/ipfs/go-cid"
+	"github.com/ipld/go-ipld-prime/datamodel"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/samber/lo"
+	"github.com/tidwall/sjson"
+)
+
+type Commit interface {
+	NodeDataDecoder
+}
+
+type NodeDataDecoder interface {
+	DecodeFromNodeData(nd datamodel.Node) (err error)
+}
+
+var CommitTypeDelecters = []CommitTypeDelecter{
+	&AnchorCommit{},
+}
+
+type CommitWithPayload interface {
+	LoadPayload() (payload CommitPayload, err error)
+}
+
+type CommitPayload interface {
+	Commit
+	DelectType(nd datamodel.Node) bool
+	ApplyToStream(state *StreamState) (err error)
+}
+
+type CommitTypeDelecter interface {
+	DelectType(nd datamodel.Node) bool
+}
+
+type CommitHeader struct {
+	Model       StreamId // raw as StreamID encoded as byte array
+	Controllers []string
+	Unique      []byte
+}
+
+func (header *CommitHeader) DecodeFromNodeData(nd datamodel.Node) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("recovered in %s", r)
+			return
+		}
+	}()
+	header.Model = lo.Must(CastStreamID(
+		lo.Must(lo.Must(nd.LookupByString("model")).AsBytes()),
+	))
+	header.Unique = lo.Must(lo.Must(nd.LookupByString("unique")).AsBytes())
+	iter := lo.Must(nd.LookupByString("controllers")).ListIterator()
+	for !iter.Done() {
+		_, n, _ := iter.Next()
+		header.Controllers = append(header.Controllers, lo.Must(n.AsString()))
+	}
+	return
+}
+
+var _ CommitPayload = (*GenesisCommitPayload)(nil)
+
+type GenesisCommitPayload struct {
+	Header CommitHeader
+	Data   json.RawMessage
+}
+
+func (payload *GenesisCommitPayload) GetData() []byte {
+	return payload.Data
+}
+
+func (payload *GenesisCommitPayload) ApplyToStream(state *StreamState) (err error) {
+	state.Content = payload.Data
+	if state.Metadata, err = json.Marshal(map[string]any{
+		"controllers": payload.Header.Controllers,
+		"model":       payload.Header.Model,
+	}); err != nil {
+		return
+	}
+	return
+}
+
+func (*GenesisCommitPayload) DelectType(nd datamodel.Node) bool {
+	return ContainField(nd, "header") && !ContainField(nd, "id")
+}
+
+func (payload *GenesisCommitPayload) DecodeFromNodeData(nd datamodel.Node) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("recovered in %s", r)
+			return
+		}
+	}()
+	var buf bytes.Buffer
+	if err = dagjsonEncodeOption.Encode(lo.Must(nd.LookupByString("data")), &buf); err != nil {
+		return
+	}
+	payload.Data = buf.Bytes()
+	if headerNode, e := nd.LookupByString("header"); e == nil && !headerNode.IsNull() {
+		if err = payload.Header.DecodeFromNodeData(headerNode); err != nil {
+			return
+		}
+	}
+	return
+}
+
+var _ CommitPayload = (*DataCommitPayload)(nil)
+
+type DataCommitPayload struct {
+	ID     cid.Cid // link to init event
+	Prev   cid.Cid
+	Header *CommitHeader
+	Pathes []Patch
+}
+
+func (payload *DataCommitPayload) DecodeFromNodeData(nd datamodel.Node) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("recovered in %s", r)
+			return
+		}
+	}()
+	payload.ID = MustParseLink(nd, "id")
+	payload.Prev = MustParseLink(nd, "prev")
+
+	var buf bytes.Buffer
+	if err = dagjsonEncodeOption.Encode(lo.Must(nd.LookupByString("data")), &buf); err != nil {
+		return
+	}
+	if err = json.Unmarshal(buf.Bytes(), &payload.Pathes); err != nil {
+		return
+	}
+	return
+}
+
+func (DataCommitPayload) DelectType(nd datamodel.Node) bool {
+	return ContainField(nd, "prev") && !ContainField(nd, "proof")
+}
+
+type Patch struct {
+	Operation string `json:"op"`
+	Path      string `json:"path"`
+	Value     any    `json:"value"`
+}
+
+func (commit *DataCommitPayload) ApplyToStream(state *StreamState) (err error) {
+	for _, v := range commit.Pathes {
+		path := strings.ReplaceAll(v.Path[1:], "/", ".")
+		switch v.Operation {
+		case "add", "replace":
+			if state.Content, err = sjson.SetBytes(state.Content, path, v.Value); err != nil {
+				return
+			}
+		case "remove":
+			if state.Content, err = sjson.DeleteBytes(state.Content, path); err != nil {
+				return
+			}
+		}
+	}
+	return
+}
+
+var _ NodeDataDecoder = (*AnchorCommit)(nil)
+
+type AnchorCommit struct {
+	ID    cid.Cid // link to init event
+	Prev  cid.Cid
+	Proof cid.Cid
+	Path  string
+}
+
+func (*AnchorCommit) ApplyToStream(state *StreamState) (err error) {
+	return
+}
+
+func (*AnchorCommit) DelectType(nd datamodel.Node) bool {
+	return ContainField(nd, "proof")
+}
+
+func (commit *AnchorCommit) DecodeFromNodeData(nd datamodel.Node) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("recovered in %s", r)
+			return
+		}
+	}()
+	commit.ID = MustParseLink(nd, "id")
+	commit.Prev = MustParseLink(nd, "prev")
+	commit.Proof = MustParseLink(nd, "proof")
+	commit.Path = lo.Must(lo.Must(nd.LookupByString("path")).AsString())
+	return
+}
+
+var _ NodeDataDecoder = (*SignedCommit)(nil)
+
+type SignedCommit struct {
+	Link       cid.Cid
+	Payload    cid.Cid
+	Signatures []Signature
+}
+
+func (commit *SignedCommit) LoadPayload(ctx context.Context, impl IpfsImpl) (payload CommitPayload, err error) {
+	var (
+		blkReader io.Reader
+		nd        datamodel.Node
+	)
+	if blkReader, err = impl.blockAPI.Get(ctx, path.IpfsPath(commit.Payload)); err != nil {
+		return
+	}
+	if nd, err = DecodeDagCborNodeDataFromReader(blkReader); err != nil {
+		return
+	}
+	if ContainField(nd, "prev") {
+		payload = &DataCommitPayload{}
+	} else {
+		payload = &GenesisCommitPayload{}
+	}
+	if err = payload.DecodeFromNodeData(nd); err != nil {
+		return
+	}
+	return
+}
+
+func (commit *SignedCommit) DecodeFromNodeData(nd datamodel.Node) (err error) {
+	dagJws := nd.(dagjose.DecodedJWS)
+	payloadData, err := decodeBase64Url(dagJws.FieldPayload())
+	if err != nil {
+		return
+	}
+	if commit.Payload, err = cid.Cast(payloadData); err != nil {
+		return
+	}
+	if link := dagJws.FieldLink(); link.Exists() {
+		if link, ok := link.Must().Link().(cidlink.Link); ok {
+			commit.Link = link.Cid
+		}
+	}
+	signatureIter := dagJws.FieldSignatures().Must().Iterator()
+	for !signatureIter.Done() {
+		_, item := signatureIter.Next()
+		sig := Signature{}
+		if protected := item.FieldProtected(); protected.Exists() {
+			if sig.Protected, err = decodeBase64Url(protected.Must()); err != nil {
+				return
+			}
+		}
+		if sig.Signature, err = decodeBase64Url(item.FieldSignature()); err != nil {
+			return
+		}
+
+		commit.Signatures = append(commit.Signatures, sig)
+	}
+	return
+}