mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-04 15:06:37 +08:00
feat(plugins): package database, cache, logger, and storage as infra plugins
This commit is contained in:
ryan
@@ -0,0 +1,145 @@
|
||||
// Package storage provides the object storage and ingestion infrastructure plugin for Cordis.
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/core"
|
||||
"github.com/Rain-kl/Wavelet/core/contracts"
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload/ingest"
|
||||
"github.com/Rain-kl/Wavelet/internal/infra/objectstore"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
)
|
||||
|
||||
// Option configures the storage plugin.
|
||||
type Option func(*Plugin)
|
||||
|
||||
// WithBackend sets an explicit storage backend instance (useful for testing or custom engines).
|
||||
func WithBackend(b objectstore.Backend) Option {
|
||||
return func(p *Plugin) {
|
||||
p.backend = b
|
||||
}
|
||||
}
|
||||
|
||||
// Plugin implements core.Plugin to provide contracts.StorageService.
|
||||
type Plugin struct {
|
||||
backend objectstore.Backend
|
||||
}
|
||||
|
||||
// New creates a new storage infrastructure plugin.
|
||||
func New(opts ...Option) *Plugin {
|
||||
p := &Plugin{}
|
||||
for _, opt := range opts {
|
||||
if opt != nil {
|
||||
opt(p)
|
||||
}
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// Name returns the unique identifier of the storage plugin.
|
||||
func (p *Plugin) Name() string {
|
||||
return "storage"
|
||||
}
|
||||
|
||||
// Apply mounts the storage service into the Context.
|
||||
func (p *Plugin) Apply(ctx *core.Context) error {
|
||||
svc := &storageServiceImpl{
|
||||
backend: p.backend,
|
||||
}
|
||||
core.Provide[contracts.StorageService](ctx, svc)
|
||||
return nil
|
||||
}
|
||||
|
||||
type storageServiceImpl struct {
|
||||
backend objectstore.Backend
|
||||
}
|
||||
|
||||
func (s *storageServiceImpl) getBackend(ctx context.Context) (objectstore.Backend, error) {
|
||||
if s.backend != nil {
|
||||
return s.backend, nil
|
||||
}
|
||||
_, b, err := objectstore.Active(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("storage: get active backend failed: %w", err)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func (s *storageServiceImpl) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (contracts.StoragePutResult, error) {
|
||||
b, err := s.getBackend(ctx)
|
||||
if err != nil {
|
||||
return contracts.StoragePutResult{}, err
|
||||
}
|
||||
|
||||
res, err := b.Put(ctx, key, body, size, contentType)
|
||||
if err != nil {
|
||||
return contracts.StoragePutResult{}, err
|
||||
}
|
||||
|
||||
return contracts.StoragePutResult{
|
||||
Key: res.Key,
|
||||
Bucket: res.Bucket,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *storageServiceImpl) Get(ctx context.Context, key string) (*contracts.StorageObject, error) {
|
||||
b, err := s.getBackend(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
obj, err := b.Get(ctx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &contracts.StorageObject{
|
||||
Key: key,
|
||||
CachePath: obj.CachePath,
|
||||
Body: obj.Body,
|
||||
ContentLength: obj.ContentLength,
|
||||
ContentType: obj.ContentType,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *storageServiceImpl) Delete(ctx context.Context, key string) error {
|
||||
b, err := s.getBackend(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return b.Delete(ctx, key)
|
||||
}
|
||||
|
||||
func (s *storageServiceImpl) Ingest(ctx context.Context, reader io.Reader, opts contracts.IngestOptions) (*contracts.IngestResult, error) {
|
||||
meta := model.UploadMetadata{
|
||||
Extra: opts.Metadata,
|
||||
}
|
||||
|
||||
req := ingest.Request{
|
||||
UserID: opts.UserID,
|
||||
Type: opts.Type,
|
||||
FileName: opts.FileName,
|
||||
MimeType: opts.MimeType,
|
||||
Extension: opts.Extension,
|
||||
Size: opts.Size,
|
||||
Reader: reader,
|
||||
Policy: ingest.Policy(opts.Policy),
|
||||
Metadata: meta,
|
||||
}
|
||||
|
||||
res, err := ingest.Ingest(ctx, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &contracts.IngestResult{
|
||||
ID: res.Upload.ID,
|
||||
Key: res.Upload.FilePath,
|
||||
Created: res.Created,
|
||||
Stored: res.Stored,
|
||||
Resolved: res.Resolved,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package storage_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/core"
|
||||
"github.com/Rain-kl/Wavelet/core/contracts"
|
||||
"github.com/Rain-kl/Wavelet/internal/infra/objectstore"
|
||||
"github.com/Rain-kl/Wavelet/plugins/infra/storage"
|
||||
)
|
||||
|
||||
type mockStorageBackend struct {
|
||||
mu sync.RWMutex
|
||||
files map[string][]byte
|
||||
}
|
||||
|
||||
func newMockStorageBackend() *mockStorageBackend {
|
||||
return &mockStorageBackend{files: make(map[string][]byte)}
|
||||
}
|
||||
|
||||
func (m *mockStorageBackend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (objectstore.PutResult, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
data, err := io.ReadAll(body)
|
||||
if err != nil {
|
||||
return objectstore.PutResult{}, err
|
||||
}
|
||||
m.files[key] = data
|
||||
return objectstore.PutResult{Key: key, Bucket: "mock-bucket"}, nil
|
||||
}
|
||||
|
||||
func (m *mockStorageBackend) Get(ctx context.Context, key string) (*objectstore.Object, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
data, ok := m.files[key]
|
||||
if !ok {
|
||||
return nil, errors.New("file not found")
|
||||
}
|
||||
return &objectstore.Object{
|
||||
Body: io.NopCloser(bytes.NewReader(data)),
|
||||
ContentLength: int64(len(data)),
|
||||
ContentType: "application/octet-stream",
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (m *mockStorageBackend) Delete(ctx context.Context, key string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
delete(m.files, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockStorageBackend) Test(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestStoragePluginOperations(t *testing.T) {
|
||||
backend := newMockStorageBackend()
|
||||
p := storage.New(storage.WithBackend(backend))
|
||||
assert.Equal(t, "storage", p.Name())
|
||||
|
||||
ctx := core.NewContext(context.Background())
|
||||
require.NoError(t, p.Apply(ctx))
|
||||
|
||||
svc, err := core.Inject[contracts.StorageService](ctx)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, svc)
|
||||
|
||||
testCtx := context.Background()
|
||||
|
||||
// Put
|
||||
content := []byte("wavelet storage content")
|
||||
putRes, err := svc.Put(testCtx, "avatar/user1.png", bytes.NewReader(content), int64(len(content)), "image/png")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "avatar/user1.png", putRes.Key)
|
||||
assert.Equal(t, "mock-bucket", putRes.Bucket)
|
||||
|
||||
// Get
|
||||
obj, err := svc.Get(testCtx, "avatar/user1.png")
|
||||
require.NoError(t, err)
|
||||
data, err := io.ReadAll(obj.Body)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, content, data)
|
||||
|
||||
// Delete
|
||||
require.NoError(t, svc.Delete(testCtx, "avatar/user1.png"))
|
||||
_, err = svc.Get(testCtx, "avatar/user1.png")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
Reference in New Issue
Block a user