// Copyright 2026 Arctel.net // SPDX-License-Identifier: Apache-2.0 // Package storage provides the object storage and ingestion infrastructure plugin for Cordis. package storage import ( "Wavelet/core" "Wavelet/core/contracts" "Wavelet/plugins/infra/storage/diskcache" "Wavelet/plugins/infra/storage/objectstore" "context" "errors" "fmt" "io" ) // 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 { // Bind DBService if db, err := core.Inject[contracts.DBService](ctx); err == nil && db != nil { objectstore.SetDBService(db) diskcache.SetDBService(db) } else { core.When[contracts.DBService](ctx, func(db contracts.DBService) { objectstore.SetDBService(db) diskcache.SetDBService(db) }) } // Bind CacheService if cache, err := core.Inject[contracts.CacheService](ctx); err == nil && cache != nil { objectstore.SetCacheService(cache) } else { core.When[contracts.CacheService](ctx, func(cache contracts.CacheService) { objectstore.SetCacheService(cache) }) } ctx.OnDispose(func() error { objectstore.SetDBService(nil) diskcache.SetDBService(nil) objectstore.SetCacheService(nil) return nil }) 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(_ context.Context, _ io.Reader, _ contracts.IngestOptions) (*contracts.IngestResult, error) { return nil, errors.New("storage: programmatic ingest is managed by domain/upload plugin") }