Enterprise blobber

.github/workflows/build-sdks.yml

@@ -27,7 +27,7 @@ jobs:
       - name: Set up Go 1.22
         uses: actions/setup-go@v3
         with:
-          go-version: 1.22
+          go-version: 1.22.5
 
       - name: Clean build
         run: make clean-mobilesdk
@@ -99,7 +99,7 @@ jobs:
       - name: Set up Go 1.22
         uses: actions/setup-go@v3
         with:
-          go-version: 1.22
+          go-version: 1.22.5
 
       - name: Install deps
         run: |
@@ -202,7 +202,7 @@ jobs:
       - name: Set up Go 1.22
         uses: actions/setup-go@v3
         with:
-          go-version: 1.22
+          go-version: 1.22.5
 
       - name: Clean build
         run: make clean-mobilesdk
@@ -274,7 +274,7 @@ jobs:
       - name: Set up Go 1.22
         uses: actions/setup-go@v3
         with:
-          go-version: 1.22
+          go-version: 1.22.5
 
       - name: Install deps
         run: |
@@ -338,7 +338,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v3
         with:
-          go-version: 1.21.5
+          go-version: 1.22.5
 
       - name: Checkout
         uses: actions/checkout@v2
@@ -349,7 +349,7 @@ jobs:
           sudo apt-get -y install build-essential nghttp2 libnghttp2-dev libssl-dev wget
 
       - name: Build
-        run:  docker run --rm -v $PWD:/gosdk -w /gosdk golang:1.21.5 make wasm-build
+        run:  docker run --rm -v $PWD:/gosdk -w /gosdk golang:1.22.5 make wasm-build
 
       - name: 'Upload Artifact'
         uses: actions/upload-artifact@v3

.github/workflows/tests.yml

@@ -26,7 +26,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v3
         with:
-          go-version: 1.21.5
+          go-version: 1.22.5
 
       - uses: actions/checkout@v3
 
@@ -51,7 +51,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v3
         with:
-          go-version: 1.21.5
+          go-version: 1.22.5
 
       - name: Install deps
         run: |
@@ -170,7 +170,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v3
         with:
-          go-version: 1.21.5
+          go-version: 1.22.5
 
       - uses: actions/setup-node@v2
         with:

go.mod

@@ -1,6 +1,6 @@
 module github.com/0chain/gosdk
 
-go 1.21
+go 1.22.5
 
 require (
 	github.com/0chain/common v0.0.6-0.20230127095721-8df4d1d72565

wasmsdk/blobber.go

@@ -998,8 +998,9 @@ func upload(allocationID, remotePath string, fileBytes, thumbnailBytes []byte, w
 // 		- remotePath : remote path of the file
 // 		- authTicket : auth ticket of the file, if the file is shared
 // 		- lookupHash : lookup hash of the file, which is used to locate the file if remotepath and allocation id are not provided
+// 		- writeChunkFuncName : callback function name to write the chunk, if empty the function will return arrayBuffer otherwise will return nil
 
-func downloadBlocks(allocId string, remotePath, authTicket, lookupHash string, startBlock, endBlock int64) ([]byte, error) {
+func downloadBlocks(allocId, remotePath, authTicket, lookupHash, writeChunkFuncName string, startBlock, endBlock int64) ([]byte, error) {
 
 	if len(remotePath) == 0 && len(authTicket) == 0 {
 		return nil, RequiredArg("remotePath/authTicket")
@@ -1017,37 +1018,58 @@ func downloadBlocks(allocId string, remotePath, authTicket, lookupHash string, s
 		statusBar = &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 	)
 
-	pathHash := encryption.FastHash(remotePath)
-	fs, err := sys.Files.Open(pathHash)
-	if err != nil {
-		return nil, fmt.Errorf("could not open local file: %v", err)
+	if lookupHash == "" {
+		lookupHash = getLookupHash(allocId, remotePath)
 	}
 
-	mf, _ := fs.(*sys.MemFile)
-	if mf == nil {
-		return nil, fmt.Errorf("invalid memfile")
-	}
+	var fh sys.File
+	if writeChunkFuncName == "" {
+		pathHash := encryption.FastHash(fmt.Sprintf("%s:%d:%d", lookupHash, startBlock, endBlock))
+		fs, err := sys.Files.Open(pathHash)
+		if err != nil {
+			return nil, fmt.Errorf("could not open local file: %v", err)
+		}
 
-	defer sys.Files.Remove(pathHash) //nolint
+		mf, _ := fs.(*sys.MemFile)
+		if mf == nil {
+			return nil, fmt.Errorf("invalid memfile")
+		}
+		fh = mf
+		defer sys.Files.Remove(pathHash) //nolint
+	} else {
+		fh = jsbridge.NewFileCallbackWriter(writeChunkFuncName)
+	}
 
 	wg.Add(1)
 	if authTicket != "" {
-		err = alloc.DownloadByBlocksToFileHandlerFromAuthTicket(mf, authTicket, lookupHash, startBlock, endBlock, 100, remotePath, false, statusBar, true)
+		err = alloc.DownloadByBlocksToFileHandlerFromAuthTicket(fh, authTicket, lookupHash, startBlock, endBlock, 100, remotePath, false, statusBar, true, sdk.WithFileCallback(
+			func() {
+				fh.Close() //nolint:errcheck
+			},
+		))
 	} else {
 		err = alloc.DownloadByBlocksToFileHandler(
-			mf,
+			fh,
 			remotePath,
 			startBlock,
 			endBlock,
 			100,
 			false,
-			statusBar, true)
+			statusBar, true, sdk.WithFileCallback(
+				func() {
+					fh.Close() //nolint:errcheck
+				},
+			))
 	}
 	if err != nil {
 		return nil, err
 	}
 	wg.Wait()
-	return mf.Buffer, nil
+	var buf []byte
+	if mf, ok := fh.(*sys.MemFile); ok {
+		buf = mf.Buffer
+	}
+	return buf, nil
 }
 
 // getBlobbers get list of active blobbers, and format them as array json string
@@ -1086,7 +1108,19 @@ func repairAllocation(allocationID, callbackFuncName string) error {
 		return err
 	}
 	wg.Wait()
-	return statusBar.err
+	if statusBar.err != nil {
+		fmt.Println("Error in repair allocation: ", statusBar.err)
+		return statusBar.err
+	}
+	status, _, err := alloc.CheckAllocStatus()
+	if err != nil {
+		return err
+	}
+	if status == sdk.Repair || status == sdk.Broken {
+		fmt.Println("allocation repair failed")
+		return errors.New("allocation repair failed")
+	}
+	return nil
 }
 
 // checkAllocStatus check the status of the allocation, either it is ok, needs repair or broken

wasmsdk/common.go

@@ -89,3 +89,11 @@ func (h *hasher) WriteToValidationMT(_ []byte) error {
 func (h *hasher) Finalize() error {
 	return nil
 }
+
+func (h *hasher) GetBlockHash() (string, error) {
+	return "", nil
+}
+
+func (h *hasher) WriteToBlockHasher(buf []byte) error {
+	return nil
+}

wasmsdk/jsbridge/file_writer.go

@@ -5,9 +5,11 @@ package jsbridge
 
 import (
 	"errors"
-	"io"
 	"io/fs"
 	"syscall/js"
+
+	"github.com/0chain/gosdk/core/common"
+	"github.com/valyala/bytebufferpool"
 )
 
 type FileWriter struct {
@@ -31,29 +33,42 @@ func (w *FileWriter) Write(p []byte) (int, error) {
 
 	//copy bytes to buf
 	if w.bufWriteOffset+len(p) > len(w.buf) {
-		w.writeError = true
-		return 0, io.ErrShortWrite
+		err := w.flush()
+		if err != nil {
+			return 0, err
+		}
 	}
 	n := copy(w.buf[w.bufWriteOffset:], p)
 	w.bufWriteOffset += n
 	if w.bufWriteOffset == len(w.buf) {
 		//write to file
-		if w.bufLen != len(w.buf) {
-			w.bufLen = len(w.buf)
-			w.uint8Array = js.Global().Get("Uint8Array").New(w.bufLen)
+		err := w.flush()
+		if err != nil {
+			return 0, err
 		}
-		js.CopyBytesToJS(w.uint8Array, w.buf)
-		_, err := Await(w.writableStream.Call("write", w.uint8Array))
-		if len(err) > 0 && !err[0].IsNull() {
-			w.writeError = true
-			return 0, errors.New("file_writer: " + err[0].String())
-		}
-		//reset buffer
-		w.bufWriteOffset = 0
 	}
 	return len(p), nil
 }
 
+func (w *FileWriter) flush() error {
+	if w.bufWriteOffset == 0 {
+		return nil
+	}
+	if w.bufLen != w.bufWriteOffset {
+		w.bufLen = w.bufWriteOffset
+		w.uint8Array = js.Global().Get("Uint8Array").New(w.bufLen)
+	}
+	js.CopyBytesToJS(w.uint8Array, w.buf[:w.bufWriteOffset])
+	_, err := Await(w.writableStream.Call("write", w.uint8Array))
+	if len(err) > 0 && !err[0].IsNull() {
+		w.writeError = true
+		return errors.New("file_writer: " + err[0].String())
+	}
+	//reset buffer
+	w.bufWriteOffset = 0
+	return nil
+}
+
 // func (w *FileWriter) WriteAt(p []byte, offset int64) (int, error) {
 // 	uint8Array := js.Global().Get("Uint8Array").New(len(p))
 // 	js.CopyBytesToJS(uint8Array, p)
@@ -148,3 +163,74 @@ func NewFileWriterFromHandle(dirHandler js.Value, name string) (*FileWriter, err
 		fileHandle:     fileHandler[0],
 	}, nil
 }
+
+type FileCallbackWriter struct {
+	writeChunk js.Value
+	buf        []byte
+	offset     int64
+}
+
+const bufCallbackCap = 4 * 1024 * 1024 //4MB
+
+func NewFileCallbackWriter(writeChunkFuncName string) *FileCallbackWriter {
+	writeChunk := js.Global().Get(writeChunkFuncName)
+	return &FileCallbackWriter{
+		writeChunk: writeChunk,
+	}
+}
+
+func (wc *FileCallbackWriter) Write(p []byte) (int, error) {
+	if len(wc.buf) == 0 {
+		buff := common.MemPool.Get()
+		if cap(buff.B) < bufCallbackCap {
+			buff.B = make([]byte, 0, bufCallbackCap)
+		}
+		wc.buf = buff.B
+	}
+	if len(wc.buf)+len(p) > cap(wc.buf) {
+		uint8Array := js.Global().Get("Uint8Array").New(len(wc.buf))
+		js.CopyBytesToJS(uint8Array, wc.buf)
+		_, err := Await(wc.writeChunk.Invoke(uint8Array, wc.offset))
+		if len(err) > 0 && !err[0].IsNull() {
+			return 0, errors.New("file_writer: " + err[0].String())
+		}
+		wc.offset += int64(len(wc.buf))
+		wc.buf = wc.buf[:0]
+	}
+	wc.buf = append(wc.buf, p...)
+	return len(p), nil
+}
+
+func (wc *FileCallbackWriter) Close() error {
+	if len(wc.buf) > 0 {
+		uint8Array := js.Global().Get("Uint8Array").New(len(wc.buf))
+		js.CopyBytesToJS(uint8Array, wc.buf)
+		_, err := Await(wc.writeChunk.Invoke(uint8Array, wc.offset))
+		if len(err) > 0 && !err[0].IsNull() {
+			return errors.New("file_writer: " + err[0].String())
+		}
+		wc.offset += int64(len(wc.buf))
+		wc.buf = wc.buf[:0]
+	}
+	buff := &bytebufferpool.ByteBuffer{
+		B: wc.buf,
+	}
+	common.MemPool.Put(buff)
+	return nil
+}
+
+func (wc *FileCallbackWriter) Read(p []byte) (int, error) {
+	return 0, errors.New("file_writer: not supported")
+}
+
+func (wc *FileCallbackWriter) Seek(offset int64, whence int) (int64, error) {
+	return 0, nil
+}
+
+func (wc *FileCallbackWriter) Sync() error {
+	return nil
+}
+
+func (wc *FileCallbackWriter) Stat() (fs.FileInfo, error) {
+	return nil, nil
+}

wasmsdk/jsbridge/template_data.go

@@ -51,7 +51,7 @@ func buildJS(args, env []string, wasmPath string, tpl []byte) (string, error) {
 	if suffix == "" {
 		suffix = "dev"
 	}
-	cdnPath := fmt.Sprintf("https://d2os1u2xwjukgr.cloudfront.net/%s/zcn.wasm", suffix)
+	cdnPath := fmt.Sprintf("https://webapps-staticfiles.s3.us-east-2.amazonaws.com/%s/enterprise-zcn.wasm", suffix)
 	data := templateData{
 		Path:         cdnPath,
 		Args:         args,

wasmsdk/jsbridge/webworker.go

@@ -81,7 +81,7 @@ func NewWasmWebWorker(blobberID, blobberURL, clientID, clientKey, peerPublicKey,
 			"IS_SPLIT=" + strconv.FormatBool(isSplit),
 			"MNEMONIC=" + mnemonic,
 			"ZAUTH_SERVER=" + gZauthServer},
-		Path:        "zcn.wasm",
+		Path:        "enterprise-zcn.wasm",
 		subscribers: make(map[string]chan worker.MessageEvent),
 	}
 

wasmsdk/jsbridge/zcnworker.js.tpl

@@ -1,4 +1,4 @@
-importScripts('https://cdn.jsdelivr.net/gh/golang/go@go1.21.5/misc/wasm/wasm_exec.js','https://cdn.jsdelivr.net/gh/herumi/[email protected]/browser/bls.js');
+importScripts('https://cdn.jsdelivr.net/gh/golang/go@go1.22.5/misc/wasm/wasm_exec.js','https://cdn.jsdelivr.net/gh/herumi/[email protected]/browser/bls.js');
 
 const go = new Go();
 go.argv = {{.ArgsToJS}}

wasmsdk/player_file.go

@@ -66,7 +66,7 @@ func (p *FilePlayer) download(startBlock int64) {
 	}
 	fmt.Println("start:", startBlock, "end:", endBlock, "numBlocks:", p.numBlocks, "total:", p.playlistFile.NumBlocks)
 
-	data, err := downloadBlocks(p.allocationObj.ID, p.remotePath, p.authTicket, p.lookupHash, startBlock, endBlock)
+	data, err := downloadBlocks(p.allocationObj.ID, p.remotePath, p.authTicket, p.lookupHash, "", startBlock, endBlock)
 	// data, err := downloadBlocks2(int(startBlock), int(endBlock), p.allocationObj, p.remotePath)
 	if err != nil {
 		PrintError(err.Error())