edge/pkg/module/blob/module.go

package blob

import (
	"context"
	"fmt"
	"io"
	"mime/multipart"
	"os"
	"sort"

	"forge.cadoles.com/arcad/edge/pkg/app"
	"forge.cadoles.com/arcad/edge/pkg/bus"
	"forge.cadoles.com/arcad/edge/pkg/module/util"
	"forge.cadoles.com/arcad/edge/pkg/storage"
	"github.com/dop251/goja"
	"github.com/pkg/errors"
	"gitlab.com/wpetit/goweb/logger"
)

const (
	DefaultBlobBucket string = "default"
)

type Module struct {
	server *app.Server
	bus    bus.Bus
	store  storage.BlobStore
}

func (m *Module) Name() string {
	return "blob"
}

func (m *Module) Export(export *goja.Object) {
	funcs := map[string]any{
		"listBuckets":   m.listBuckets,
		"deleteBucket":  m.deleteBucket,
		"getBucketSize": m.getBucketSize,
		"listBlobs":     m.listBlobs,
		"getBlobInfo":   m.getBlobInfo,
		"readBlob":      m.readBlob,
		"writeBlob":     m.writeBlob,
		"deleteBlob":    m.deleteBlob,
	}

	for name, fn := range funcs {
		if err := export.Set(name, fn); err != nil {
			panic(errors.Wrapf(err, "could not set '%s' function", name))
		}
	}

	if err := export.Set("DEFAULT_BUCKET", DefaultBlobBucket); err != nil {
		panic(errors.Wrap(err, "could not set 'DEFAULT_BUCKET' property"))
	}
}

func (m *Module) listBuckets(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)

	buckets, err := m.store.ListBuckets(ctx)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	defaultBucketIndex := sort.SearchStrings(buckets, DefaultBlobBucket)
	if defaultBucketIndex == 0 {
		buckets = append(buckets, DefaultBlobBucket)
	} else {
		buckets[defaultBucketIndex] = DefaultBlobBucket
	}

	return rt.ToValue(buckets)
}

func (m *Module) writeBlob(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)
	blobID := assertBlobID(call.Argument(2), rt)
	rawData := call.Argument(3).Export()

	var data []byte
	switch typ := rawData.(type) {
	case []byte:
		data = typ
	case string:
		data = []byte(typ)
	default:
		data = []byte(fmt.Sprintf("%v", typ))
	}

	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	defer func() {
		if err := bucket.Close(); err != nil {
			logger.Error(ctx, "could not close bucket", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	writer, err := bucket.NewWriter(ctx, blobID)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	defer func() {
		if err := writer.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
			logger.Error(ctx, "could not close blob writer", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	if _, err := writer.Write(data); err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	return nil
}

func (m *Module) getBlobInfo(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)
	blobID := assertBlobID(call.Argument(2), rt)

	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	defer func() {
		if err := bucket.Close(); err != nil {
			logger.Error(ctx, "could not close bucket", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	blobInfo, err := bucket.Get(ctx, blobID)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	return rt.ToValue(toGojaBlobInfo(blobInfo))
}

func (m *Module) readBlob(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)
	blobID := assertBlobID(call.Argument(2), rt)

	reader, _, err := m.openBlob(ctx, bucketName, blobID)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	defer func() {
		if err := reader.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
			logger.Error(ctx, "could not close blob reader", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	data, err := io.ReadAll(reader)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	return rt.ToValue(rt.NewArrayBuffer(data))
}

func (m *Module) deleteBlob(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)
	blobID := assertBlobID(call.Argument(2), rt)

	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	if err := bucket.Delete(ctx, blobID); err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	return nil
}

func (m *Module) listBlobs(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)

	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	blobInfos, err := bucket.List(ctx)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	gojaBlobInfos := make([]blobInfo, len(blobInfos))

	for i, b := range blobInfos {
		gojaBlobInfos[i] = toGojaBlobInfo(b)
	}

	return rt.ToValue(gojaBlobInfos)
}

func (m *Module) deleteBucket(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)

	if err := m.store.DeleteBucket(ctx, bucketName); err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	return nil
}

func (m *Module) getBucketSize(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
	ctx := util.AssertContext(call.Argument(0), rt)
	bucketName := util.AssertString(call.Argument(1), rt)

	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	size, err := bucket.Size(ctx)
	if err != nil {
		panic(rt.ToValue(errors.WithStack(err)))
	}

	return rt.ToValue(size)
}

func (m *Module) handleMessages() {
	ctx := context.Background()

	uploadRequestErrs := m.bus.Reply(ctx, AddressUpload, func(env bus.Envelope) (any, error) {
		uploadRequest, ok := env.Message().(*UploadRequest)
		if !ok {
			return nil, errors.Wrapf(bus.ErrUnexpectedMessage, "expected message upload request, got '%T'", env.Message())
		}

		res, err := m.handleUploadRequest(uploadRequest)
		if err != nil {
			logger.Error(ctx, "could not handle upload request", logger.CapturedE(errors.WithStack(err)))

			return nil, errors.WithStack(err)
		}

		logger.Debug(ctx, "upload request response", logger.F("response", res))

		return res, nil
	})

	go func() {
		for err := range uploadRequestErrs {
			logger.Error(ctx, "error while replying to upload requests", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	downloadRequestErrs := m.bus.Reply(ctx, AddressDownload, func(env bus.Envelope) (any, error) {
		downloadRequest, ok := env.Message().(*DownloadRequest)
		if !ok {
			return nil, errors.Wrapf(bus.ErrUnexpectedMessage, "expected message download request, got '%T'", env.Message())
		}

		res, err := m.handleDownloadRequest(downloadRequest)
		if err != nil {
			logger.Error(ctx, "could not handle download request", logger.CapturedE(errors.WithStack(err)))

			return nil, errors.WithStack(err)
		}

		return res, nil
	})

	for err := range downloadRequestErrs {
		logger.Fatal(ctx, "error while replying to download requests", logger.CapturedE(errors.WithStack(err)))
	}
}

func (m *Module) handleUploadRequest(req *UploadRequest) (*UploadResponse, error) {
	blobID := storage.NewBlobID()
	res := &UploadResponse{}

	ctx := logger.With(req.Context, logger.F("blobID", blobID))

	blobInfo := map[string]interface{}{
		"size":        req.FileHeader.Size,
		"filename":    req.FileHeader.Filename,
		"contentType": req.FileHeader.Header.Get("Content-Type"),
	}

	rawResult, err := m.server.ExecFuncByName(ctx, "onBlobUpload", ctx, blobID, blobInfo, req.Metadata)
	if err != nil {
		if errors.Is(err, app.ErrFuncDoesNotExist) {
			res.Allow = false

			return res, nil
		}

		return nil, errors.WithStack(err)
	}

	result, ok := rawResult.(map[string]interface{})
	if !ok {
		return nil, errors.Errorf(
			"unexpected onBlobUpload result: expected 'map[string]interface{}', got '%T'",
			rawResult,
		)
	}

	var allow bool

	rawAllow, exists := result["allow"]
	if !exists {
		allow = false
	} else {
		allow, ok = rawAllow.(bool)
		if !ok {
			return nil, errors.Errorf("invalid 'allow' result property: got type '%T', expected type '%T'", rawAllow, false)
		}
	}

	res.Allow = allow

	if res.Allow {
		bucket := DefaultBlobBucket

		rawBucket, exists := result["bucket"]
		if exists {
			bucket, ok = rawBucket.(string)
			if !ok {
				return nil, errors.Errorf("invalid 'bucket' result property: got type '%T', expected type '%T'", bucket, "")
			}
		}

		if err := m.saveBlob(ctx, bucket, blobID, *req.FileHeader); err != nil {
			return nil, errors.WithStack(err)
		}

		res.Bucket = bucket
		res.BlobID = blobID
	}

	return res, nil
}

func (m *Module) saveBlob(ctx context.Context, bucketName string, blobID storage.BlobID, fileHeader multipart.FileHeader) error {
	file, err := fileHeader.Open()
	if err != nil {
		return errors.WithStack(err)
	}

	defer func() {
		if err := file.Close(); err != nil {
			logger.Error(ctx, "could not close file", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		return errors.WithStack(err)
	}

	defer func() {
		if err := bucket.Close(); err != nil {
			logger.Error(ctx, "could not close bucket", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	writer, err := bucket.NewWriter(ctx, blobID)
	if err != nil {
		return errors.WithStack(err)
	}

	defer func() {
		if err := file.Close(); err != nil {
			logger.Error(ctx, "could not close file", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	defer func() {
		if err := writer.Close(); err != nil {
			logger.Error(ctx, "could not close writer", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	if _, err := io.Copy(writer, file); err != nil {
		return errors.WithStack(err)
	}

	return nil
}

func (m *Module) handleDownloadRequest(req *DownloadRequest) (*DownloadResponse, error) {
	res := &DownloadResponse{}

	rawResult, err := m.server.ExecFuncByName(req.Context, "onBlobDownload", req.Context, req.Bucket, req.BlobID)
	if err != nil {
		if errors.Is(err, app.ErrFuncDoesNotExist) {
			res.Allow = false

			return res, nil
		}

		return nil, errors.WithStack(err)
	}

	result, ok := rawResult.(map[string]interface{})
	if !ok {
		return nil, errors.Errorf(
			"unexpected onBlobDownload result: expected 'map[string]interface{}', got '%T'",
			rawResult,
		)
	}

	var allow bool

	rawAllow, exists := result["allow"]
	if !exists {
		allow = false
	} else {
		allow, ok = rawAllow.(bool)
		if !ok {
			return nil, errors.Errorf("invalid 'allow' result property: got type '%T', expected type '%T'", rawAllow, false)
		}
	}

	res.Allow = allow

	reader, info, err := m.openBlob(req.Context, req.Bucket, req.BlobID)
	if err != nil && !errors.Is(err, storage.ErrBlobNotFound) {
		return nil, errors.WithStack(err)
	}

	if reader != nil {
		res.Blob = reader
	}

	if info != nil {
		res.BlobInfo = info
	}

	return res, nil
}

func (m *Module) openBlob(ctx context.Context, bucketName string, blobID storage.BlobID) (io.ReadSeekCloser, storage.BlobInfo, error) {
	bucket, err := m.store.OpenBucket(ctx, bucketName)
	if err != nil {
		return nil, nil, errors.WithStack(err)
	}

	defer func() {
		if err := bucket.Close(); err != nil {
			logger.Error(ctx, "could not close bucket", logger.CapturedE(errors.WithStack(err)), logger.F("bucket", bucket))
		}
	}()

	info, err := bucket.Get(ctx, blobID)
	if err != nil {
		return nil, nil, errors.WithStack(err)
	}

	reader, err := bucket.NewReader(ctx, blobID)
	if err != nil {
		return nil, nil, errors.WithStack(err)
	}

	return reader, info, nil
}

func ModuleFactory(bus bus.Bus, store storage.BlobStore) app.ServerModuleFactory {
	return func(server *app.Server) app.ServerModule {
		mod := &Module{
			store:  store,
			bus:    bus,
			server: server,
		}

		go mod.handleMessages()

		return mod
	}
}

func assertBlobID(value goja.Value, rt *goja.Runtime) storage.BlobID {
	blobID, ok := value.Export().(storage.BlobID)
	if !ok {
		rawBlobID, ok := value.Export().(string)
		if !ok {
			panic(rt.NewTypeError(fmt.Sprintf("blob id  must be a blob or a string, got '%T'", value.Export())))
		}

		blobID = storage.BlobID(rawBlobID)
	}

	return blobID
}