500 lines
12 KiB
Go
500 lines
12 KiB
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.E(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.E(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.E(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.E(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()
|
||
|
|
|
||
|
|
go func() {
|
||
|
|
err := m.bus.Reply(ctx, MessageNamespaceUploadRequest, func(msg bus.Message) (bus.Message, error) {
|
||
|
|
uploadRequest, ok := msg.(*MessageUploadRequest)
|
||
|
|
if !ok {
|
||
|
|
return nil, errors.Wrapf(bus.ErrUnexpectedMessage, "expected message upload request, got '%T'", msg)
|
||
|
|
}
|
||
|
|
|
||
|
|
res, err := m.handleUploadRequest(uploadRequest)
|
||
|
|
if err != nil {
|
||
|
|
logger.Error(ctx, "could not handle upload request", logger.E(errors.WithStack(err)))
|
||
|
|
|
||
|
|
return nil, errors.WithStack(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
logger.Debug(ctx, "upload request response", logger.F("response", res))
|
||
|
|
|
||
|
|
return res, nil
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
panic(errors.WithStack(err))
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
err := m.bus.Reply(ctx, MessageNamespaceDownloadRequest, func(msg bus.Message) (bus.Message, error) {
|
||
|
|
downloadRequest, ok := msg.(*MessageDownloadRequest)
|
||
|
|
if !ok {
|
||
|
|
return nil, errors.Wrapf(bus.ErrUnexpectedMessage, "expected message download request, got '%T'", msg)
|
||
|
|
}
|
||
|
|
|
||
|
|
res, err := m.handleDownloadRequest(downloadRequest)
|
||
|
|
if err != nil {
|
||
|
|
logger.Error(ctx, "could not handle download request", logger.E(errors.WithStack(err)))
|
||
|
|
|
||
|
|
return nil, errors.WithStack(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
return res, nil
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
panic(errors.WithStack(err))
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (m *Module) handleUploadRequest(req *MessageUploadRequest) (*MessageUploadResponse, error) {
|
||
|
|
blobID := storage.NewBlobID()
|
||
|
|
res := NewMessageUploadResponse(req.RequestID)
|
||
|
|
|
||
|
|
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.Export().(map[string]interface{})
|
||
|
|
if !ok {
|
||
|
|
return nil, errors.Errorf(
|
||
|
|
"unexpected onBlobUpload result: expected 'map[string]interface{}', got '%T'",
|
||
|
|
rawResult.Export(),
|
||
|
|
)
|
||
|
|
}
|
||
|
|
|
||
|
|
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.E(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.E(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.E(errors.WithStack(err)))
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
defer func() {
|
||
|
|
if err := writer.Close(); err != nil {
|
||
|
|
logger.Error(ctx, "could not close writer", logger.E(errors.WithStack(err)))
|
||
|
|
}
|
||
|
|
}()
|
||
|
|
|
||
|
|
if _, err := io.Copy(writer, file); err != nil {
|
||
|
|
return errors.WithStack(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (m *Module) handleDownloadRequest(req *MessageDownloadRequest) (*MessageDownloadResponse, error) {
|
||
|
|
res := NewMessageDownloadResponse(req.RequestID)
|
||
|
|
|
||
|
|
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.Export().(map[string]interface{})
|
||
|
|
if !ok {
|
||
|
|
return nil, errors.Errorf(
|
||
|
|
"unexpected onBlobDownload result: expected 'map[string]interface{}', got '%T'",
|
||
|
|
rawResult.Export(),
|
||
|
|
)
|
||
|
|
}
|
||
|
|
|
||
|
|
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.E(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
|
||
|
|
}
|