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 }