diff --git a/core/container.go b/core/container.go index ff9311ee..09cccf68 100644 --- a/core/container.go +++ b/core/container.go @@ -1,3 +1,4 @@ +// Package core provides the micro-kernel service bus, generic IoC container, and runtime extensions. package core import ( diff --git a/core/context.go b/core/context.go index 797f2180..0c72ce2a 100644 --- a/core/context.go +++ b/core/context.go @@ -36,6 +36,8 @@ type Context struct { // NewContext creates a new root Context wrapping a standard Go context. // If base is nil, context.Background() is used by default. +// +//nolint:contextcheck func NewContext(base context.Context) *Context { if base == nil { base = context.Background() @@ -135,6 +137,8 @@ func (c *Context) Fork() *Context { } // ForkWithContext creates a child Context using a specific standard Go context. +// +//nolint:contextcheck func (c *Context) ForkWithContext(base context.Context) *Context { if base == nil { base = c.goCtx diff --git a/core/contracts/cache.go b/core/contracts/cache.go index 2777bfef..fc9a0475 100644 --- a/core/contracts/cache.go +++ b/core/contracts/cache.go @@ -1,3 +1,4 @@ +// Package contracts defines unified service interfaces and DTOs for cross-plugin communication. package contracts import ( diff --git a/core/contracts/database.go b/core/contracts/database.go index 73c06c50..1e25efaf 100644 --- a/core/contracts/database.go +++ b/core/contracts/database.go @@ -1,3 +1,4 @@ +// Package contracts defines unified service interfaces and DTOs for cross-plugin communication. package contracts import ( diff --git a/core/contracts/logger.go b/core/contracts/logger.go index 7eb1d260..3af38600 100644 --- a/core/contracts/logger.go +++ b/core/contracts/logger.go @@ -1,3 +1,4 @@ +// Package contracts defines unified service interfaces and DTOs for cross-plugin communication. package contracts import ( diff --git a/core/contracts/storage.go b/core/contracts/storage.go index 4dcad26b..961a994a 100644 --- a/core/contracts/storage.go +++ b/core/contracts/storage.go @@ -1,3 +1,4 @@ +// Package contracts defines unified service interfaces and DTOs for cross-plugin communication. package contracts import ( diff --git a/core/events.go b/core/events.go index 2e9b8534..f4ac6eb8 100644 --- a/core/events.go +++ b/core/events.go @@ -9,6 +9,8 @@ import ( "sync/atomic" ) +const maxHandlerParams = 2 + var ctxInterfaceType = reflect.TypeFor[context.Context]() var errInterfaceType = reflect.TypeFor[error]() @@ -62,7 +64,7 @@ func (b *EventBus) On(topic string, handler any) Disposer { } numIn := fnType.NumIn() - if numIn > 2 { + if numIn > maxHandlerParams { panic(fmt.Sprintf("core/events: handler has %d parameters, maximum 2 supported (ctx, event)", numIn)) } @@ -150,6 +152,8 @@ func Subscribe[T any](bus *EventBus, topic string, handler func(ctx context.Cont // Emit publishes an event to all subscribers of the specified topic. // Handlers are executed synchronously. If any handler panics or returns an error, // the error is collected and returned via errors.Join. +// +//nolint:contextcheck func (b *EventBus) Emit(ctx context.Context, topic string, payload any) error { if ctx == nil { ctx = context.Background() diff --git a/core/extpoints/migration.go b/core/extpoints/migration.go index 322a6a34..268c1746 100644 --- a/core/extpoints/migration.go +++ b/core/extpoints/migration.go @@ -1,3 +1,4 @@ +// Package extpoints defines extension points for router, migrations, tasks, schedules, and settings. package extpoints import ( diff --git a/core/types.go b/core/types.go index 33beb287..59b1d670 100644 --- a/core/types.go +++ b/core/types.go @@ -72,18 +72,38 @@ type Driver interface { // Disposer is a cleanup function executed when a Context is disposed. type Disposer func() error -// Re-exported extension point types from core/extpoints for convenient usage. -type ( - RouterExtension = extpoints.RouterExtension - RouteDefinition = extpoints.RouteDefinition - MigrationExtension = extpoints.MigrationExtension - MigrationEntry = extpoints.MigrationEntry - TaskExtension = extpoints.TaskExtension - TaskDefinition = extpoints.TaskDefinition - TaskOption = extpoints.TaskOption - ScheduleExtension = extpoints.ScheduleExtension - ScheduleDefinition = extpoints.ScheduleDefinition - ScheduleOption = extpoints.ScheduleOption - SettingExtension = extpoints.SettingExtension - SettingSchema = extpoints.SettingSchema -) +// RouterExtension re-exports extpoints.RouterExtension. +type RouterExtension = extpoints.RouterExtension + +// RouteDefinition re-exports extpoints.RouteDefinition. +type RouteDefinition = extpoints.RouteDefinition + +// MigrationExtension re-exports extpoints.MigrationExtension. +type MigrationExtension = extpoints.MigrationExtension + +// MigrationEntry re-exports extpoints.MigrationEntry. +type MigrationEntry = extpoints.MigrationEntry + +// TaskExtension re-exports extpoints.TaskExtension. +type TaskExtension = extpoints.TaskExtension + +// TaskDefinition re-exports extpoints.TaskDefinition. +type TaskDefinition = extpoints.TaskDefinition + +// TaskOption re-exports extpoints.TaskOption. +type TaskOption = extpoints.TaskOption + +// ScheduleExtension re-exports extpoints.ScheduleExtension. +type ScheduleExtension = extpoints.ScheduleExtension + +// ScheduleDefinition re-exports extpoints.ScheduleDefinition. +type ScheduleDefinition = extpoints.ScheduleDefinition + +// ScheduleOption re-exports extpoints.ScheduleOption. +type ScheduleOption = extpoints.ScheduleOption + +// SettingExtension re-exports extpoints.SettingExtension. +type SettingExtension = extpoints.SettingExtension + +// SettingSchema re-exports extpoints.SettingSchema. +type SettingSchema = extpoints.SettingSchema diff --git a/plugins/drivers/driver_asynq_cron/plugin.go b/plugins/drivers/driver_asynq_cron/plugin.go index 325528d2..aa0c8dd1 100644 --- a/plugins/drivers/driver_asynq_cron/plugin.go +++ b/plugins/drivers/driver_asynq_cron/plugin.go @@ -1,3 +1,4 @@ +// Package driver_asynq_cron provides the Asynq cron schedule driver plugin for Cordis. package driver_asynq_cron import ( diff --git a/plugins/drivers/driver_asynq_worker/plugin.go b/plugins/drivers/driver_asynq_worker/plugin.go index ef9bf3cf..ad6d1265 100644 --- a/plugins/drivers/driver_asynq_worker/plugin.go +++ b/plugins/drivers/driver_asynq_worker/plugin.go @@ -1,3 +1,4 @@ +// Package driver_asynq_worker provides the Asynq worker driver plugin for Cordis. package driver_asynq_worker import ( @@ -120,7 +121,7 @@ func (p *Plugin) Type() core.DriverType { } // Start boots the Asynq worker server and starts processing background tasks. -func (p *Plugin) Start(ctx context.Context) error { +func (p *Plugin) Start(_ context.Context) error { p.mu.Lock() defer p.mu.Unlock() @@ -208,10 +209,10 @@ func toAsynqHandler(h any) (asynq.Handler, error) { } switch fn := h.(type) { - case asynq.Handler: - return fn, nil case asynq.HandlerFunc: return fn, nil + case asynq.Handler: + return fn, nil case func(context.Context, *asynq.Task) error: return asynq.HandlerFunc(fn), nil case func(context.Context, []byte) error: diff --git a/plugins/drivers/driver_http/plugin.go b/plugins/drivers/driver_http/plugin.go index cc3f049a..e15fef49 100644 --- a/plugins/drivers/driver_http/plugin.go +++ b/plugins/drivers/driver_http/plugin.go @@ -1,3 +1,4 @@ +// Package driver_http provides the Gin HTTP web server driver plugin for Cordis. package driver_http import ( @@ -172,6 +173,8 @@ func (p *Plugin) Start(ctx context.Context) error { } // Stop gracefully stops the HTTP server. +// +//nolint:contextcheck func (p *Plugin) Stop(ctx context.Context) error { p.mu.Lock() defer p.mu.Unlock() diff --git a/plugins/infra/cache/plugin.go b/plugins/infra/cache/plugin.go new file mode 100644 index 00000000..408f3d72 --- /dev/null +++ b/plugins/infra/cache/plugin.go @@ -0,0 +1,265 @@ +// Package cache provides the multi-tier caching infrastructure plugin for Cordis. +package cache + +import ( + "context" + "encoding/json" + "errors" + "sync" + "time" + + "github.com/Rain-kl/Wavelet/core" + "github.com/Rain-kl/Wavelet/core/contracts" + "github.com/Rain-kl/Wavelet/internal/infra/persistence" + "github.com/Rain-kl/Wavelet/pkg/cache/ram" + "github.com/Rain-kl/Wavelet/pkg/util" + "github.com/redis/go-redis/v9" +) + +const ( + defaultRAMCapacity = 10000 + defaultPubSubChannel = "wavelet:cache:invalidation" +) + +type ramEntry struct { + data []byte + expireAt time.Time +} + +// Option configures the cache plugin. +type Option func(*Plugin) + +// WithRedis sets an explicit Redis client instance. +func WithRedis(client redis.UniversalClient) Option { + return func(p *Plugin) { + p.redisClient = client + } +} + +// WithKeyPrefix sets a custom Redis key prefix. +func WithKeyPrefix(prefix string) Option { + return func(p *Plugin) { + p.keyPrefix = prefix + } +} + +// WithRAMCapacity sets the maximum capacity for the L1 RAM cache. +func WithRAMCapacity(capacity int) Option { + return func(p *Plugin) { + p.ramCapacity = capacity + } +} + +// Plugin implements core.Plugin to provide contracts.CacheService. +type Plugin struct { + redisClient redis.UniversalClient + keyPrefix string + ramCapacity int +} + +// New creates a new cache infrastructure plugin. +func New(opts ...Option) *Plugin { + p := &Plugin{ + ramCapacity: defaultRAMCapacity, + } + for _, opt := range opts { + if opt != nil { + opt(p) + } + } + return p +} + +// Name returns the unique identifier of the cache plugin. +func (p *Plugin) Name() string { + return "cache" +} + +// Apply mounts the multi-layer cache service into the Context. +func (p *Plugin) Apply(ctx *core.Context) error { + redisClient := p.redisClient + if redisClient == nil { + redisClient = db.Redis + } + + ramCache, err := ram.New[string, ramEntry](ram.Options{ + MaximumSize: p.ramCapacity, + }) + if err != nil { + return err + } + + svc := &cacheServiceImpl{ + ramCache: ramCache, + redisClient: redisClient, + keyPrefix: p.keyPrefix, + pubSubChannel: defaultPubSubChannel, + stopCh: make(chan struct{}), + } + + if redisClient != nil { + svc.startPubSubListener() + ctx.OnDispose(func() error { + svc.stopPubSubListener() + return nil + }) + } + + core.Provide[contracts.CacheService](ctx, svc) + return nil +} + +type cacheServiceImpl struct { + ramCache *ram.Cache[string, ramEntry] + redisClient redis.UniversalClient + keyPrefix string + pubSubChannel string + + subOnce sync.Once + stopOnce sync.Once + stopCh chan struct{} + pubsub *redis.PubSub +} + +func (s *cacheServiceImpl) prefixedKey(key string) string { + if s.keyPrefix != "" { + return s.keyPrefix + key + } + return db.PrefixedKey(key) +} + +func (s *cacheServiceImpl) startPubSubListener() { + if s.redisClient == nil { + return + } + + s.subOnce.Do(func() { + pubsub := s.redisClient.Subscribe(context.Background(), s.pubSubChannel) + s.pubsub = pubsub + + util.Go(func() { + ch := pubsub.Channel() + for { + select { + case <-s.stopCh: + return + case msg, ok := <-ch: + if !ok { + return + } + if msg != nil && msg.Payload != "" { + s.ramCache.Invalidate(msg.Payload) + } + } + } + }) + }) +} + +func (s *cacheServiceImpl) stopPubSubListener() { + s.stopOnce.Do(func() { + close(s.stopCh) + if s.pubsub != nil { + _ = s.pubsub.Close() + } + }) +} + +func (s *cacheServiceImpl) Get(ctx context.Context, key string, target any) error { + // 1. Check L1 RAM cache + if entry, ok := s.ramCache.GetIfPresent(key); ok { + if entry.expireAt.IsZero() || time.Now().Before(entry.expireAt) { + return json.Unmarshal(entry.data, target) + } + // Expired in L1 RAM + s.ramCache.Invalidate(key) + } + + // 2. Check L2 Redis cache + if s.redisClient != nil { + data, err := s.redisClient.Get(ctx, s.prefixedKey(key)).Bytes() + if err == nil { + // Backfill L1 RAM cache + s.ramCache.Set(key, ramEntry{ + data: data, + }) + return json.Unmarshal(data, target) + } else if !errors.Is(err, redis.Nil) { + return err + } + } + + return contracts.ErrCacheMiss +} + +func (s *cacheServiceImpl) Set(ctx context.Context, key string, value any, ttl time.Duration) error { + data, err := json.Marshal(value) + if err != nil { + return err + } + + var expireAt time.Time + if ttl > 0 { + expireAt = time.Now().Add(ttl) + } + + // 1. Write L1 RAM cache + s.ramCache.Set(key, ramEntry{ + data: data, + expireAt: expireAt, + }) + + // 2. Write L2 Redis cache + if s.redisClient != nil { + if err := s.redisClient.Set(ctx, s.prefixedKey(key), data, ttl).Err(); err != nil { + return err + } + } + + return nil +} + +func (s *cacheServiceImpl) Delete(ctx context.Context, key string) error { + // 1. Evict L1 RAM + s.ramCache.Invalidate(key) + + // 2. Evict L2 Redis and broadcast invalidation to cluster nodes + if s.redisClient != nil { + if err := s.redisClient.Del(ctx, s.prefixedKey(key)).Err(); err != nil && !errors.Is(err, redis.Nil) { + return err + } + _ = s.redisClient.Publish(ctx, s.pubSubChannel, key).Err() + } + + return nil +} + +func (s *cacheServiceImpl) Invalidate(ctx context.Context, key string) error { + return s.Delete(ctx, key) +} + +func (s *cacheServiceImpl) GetOrSet(ctx context.Context, key string, target any, ttl time.Duration, loader func() (any, error)) error { + err := s.Get(ctx, key, target) + if err == nil { + return nil + } + if !errors.Is(err, contracts.ErrCacheMiss) { + return err + } + + val, err := loader() + if err != nil { + return err + } + + if err := s.Set(ctx, key, val, ttl); err != nil { + return err + } + + data, err := json.Marshal(val) + if err != nil { + return err + } + + return json.Unmarshal(data, target) +} diff --git a/plugins/infra/cache/plugin_test.go b/plugins/infra/cache/plugin_test.go new file mode 100644 index 00000000..d34ecf79 --- /dev/null +++ b/plugins/infra/cache/plugin_test.go @@ -0,0 +1,82 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package cache_test + +import ( + "context" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "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/plugins/infra/cache" +) + +func TestCachePluginOperations(t *testing.T) { + mr, err := miniredis.Run() + require.NoError(t, err) + defer mr.Close() + + rdb := redis.NewClient(&redis.Options{ + Addr: mr.Addr(), + }) + defer func() { _ = rdb.Close() }() + + p := cache.New( + cache.WithRedis(rdb), + cache.WithKeyPrefix("app:"), + cache.WithRAMCapacity(500), + ) + assert.Equal(t, "cache", p.Name()) + + ctx := core.NewContext(context.Background()) + require.NoError(t, p.Apply(ctx)) + + svc, err := core.Inject[contracts.CacheService](ctx) + require.NoError(t, err) + + type Data struct { + Value string `json:"value"` + } + + testCtx := context.Background() + + // 1. ErrCacheMiss + var out Data + err = svc.Get(testCtx, "missing", &out) + assert.ErrorIs(t, err, contracts.ErrCacheMiss) + + // 2. Set & Get + in := Data{Value: "hello"} + require.NoError(t, svc.Set(testCtx, "key1", in, 5*time.Minute)) + + require.NoError(t, svc.Get(testCtx, "key1", &out)) + assert.Equal(t, "hello", out.Value) + + // 3. GetOrSet + var target Data + err = svc.GetOrSet(testCtx, "key1", &target, time.Minute, func() (any, error) { + return Data{Value: "from_loader"}, nil + }) + require.NoError(t, err) + assert.Equal(t, "hello", target.Value) + + err = svc.GetOrSet(testCtx, "key2", &target, time.Minute, func() (any, error) { + return Data{Value: "from_loader"}, nil + }) + require.NoError(t, err) + assert.Equal(t, "from_loader", target.Value) + + // 4. Invalidate / Delete + require.NoError(t, svc.Invalidate(testCtx, "key1")) + err = svc.Get(testCtx, "key1", &out) + assert.ErrorIs(t, err, contracts.ErrCacheMiss) + + require.NoError(t, ctx.Dispose()) +} diff --git a/plugins/infra/database/plugin.go b/plugins/infra/database/plugin.go new file mode 100644 index 00000000..7b6012aa --- /dev/null +++ b/plugins/infra/database/plugin.go @@ -0,0 +1,99 @@ +// Package database provides the relational database infrastructure plugin for Cordis. +package database + +import ( + "context" + + "github.com/Rain-kl/Wavelet/core" + "github.com/Rain-kl/Wavelet/core/contracts" + "github.com/Rain-kl/Wavelet/internal/infra/persistence" + "gorm.io/gorm" +) + +// Option configures the database plugin. +type Option func(*Plugin) + +// WithDB configures an explicit *gorm.DB instance for the plugin. +func WithDB(d *gorm.DB) Option { + return func(p *Plugin) { + p.db = d + } +} + +// WithNamedDB registers a named secondary database connection. +func WithNamedDB(name string, d *gorm.DB) Option { + return func(p *Plugin) { + if p.namedDBs == nil { + p.namedDBs = make(map[string]*gorm.DB) + } + p.namedDBs[name] = d + } +} + +// Plugin implements core.Plugin to provide contracts.DBService into the Cordis micro-kernel. +type Plugin struct { + db *gorm.DB + namedDBs map[string]*gorm.DB +} + +// New creates a new database infrastructure plugin. +func New(opts ...Option) *Plugin { + p := &Plugin{ + namedDBs: make(map[string]*gorm.DB), + } + for _, opt := range opts { + if opt != nil { + opt(p) + } + } + return p +} + +// Name returns the unique identifier of the database plugin. +func (p *Plugin) Name() string { + return "database" +} + +// Apply mounts the database service into the Context. +func (p *Plugin) Apply(ctx *core.Context) error { + targetDB := p.db + if targetDB == nil { + targetDB = db.DB(context.Background()) + } + + svc := &dbServiceImpl{ + primary: targetDB, + namedDBs: p.namedDBs, + } + + core.Provide[contracts.DBService](ctx, svc) + return nil +} + +type dbServiceImpl struct { + primary *gorm.DB + namedDBs map[string]*gorm.DB +} + +func (s *dbServiceImpl) GORM() *gorm.DB { + if s.primary != nil { + return s.primary + } + return db.DB(context.Background()) +} + +func (s *dbServiceImpl) DB(ctx context.Context) *gorm.DB { + if s.primary != nil { + return s.primary.WithContext(ctx) + } + return db.DB(ctx) +} + +func (s *dbServiceImpl) Named(name string) *gorm.DB { + if s.namedDBs != nil { + if d, ok := s.namedDBs[name]; ok && d != nil { + return d + } + } + return s.GORM() +} diff --git a/plugins/infra/database/plugin_test.go b/plugins/infra/database/plugin_test.go new file mode 100644 index 00000000..6f5243ea --- /dev/null +++ b/plugins/infra/database/plugin_test.go @@ -0,0 +1,62 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package database_test + +import ( + "context" + "path/filepath" + "testing" + + "github.com/glebarez/sqlite" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + + "github.com/Rain-kl/Wavelet/core" + "github.com/Rain-kl/Wavelet/core/contracts" + "github.com/Rain-kl/Wavelet/plugins/infra/database" +) + +type Item struct { + ID uint64 `gorm:"primaryKey"` + Name string +} + +func TestDatabasePlugin(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "db_test.db") + gdb, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, gdb.AutoMigrate(&Item{})) + + namedPath := filepath.Join(t.TempDir(), "named_test.db") + namedDB, err := gorm.Open(sqlite.Open(namedPath), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, namedDB.AutoMigrate(&Item{})) + + p := database.New( + database.WithDB(gdb), + database.WithNamedDB("analytics", namedDB), + ) + assert.Equal(t, "database", p.Name()) + + ctx := core.NewContext(context.Background()) + require.NoError(t, p.Apply(ctx)) + + svc, err := core.Inject[contracts.DBService](ctx) + require.NoError(t, err) + require.NotNil(t, svc) + + assert.Equal(t, gdb, svc.GORM()) + assert.NotNil(t, svc.DB(context.Background())) + assert.Equal(t, namedDB, svc.Named("analytics")) + assert.Equal(t, gdb, svc.Named("non_existent")) + + // Verify DB write + item := Item{ID: 1, Name: "TestItem"} + require.NoError(t, svc.DB(context.Background()).Create(&item).Error) + + var retrieved Item + require.NoError(t, svc.GORM().First(&retrieved, 1).Error) + assert.Equal(t, "TestItem", retrieved.Name) +} diff --git a/plugins/infra/infra_test.go b/plugins/infra/infra_test.go new file mode 100644 index 00000000..650b2d32 --- /dev/null +++ b/plugins/infra/infra_test.go @@ -0,0 +1,323 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package infra_test + +import ( + "bytes" + "context" + "io" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/glebarez/sqlite" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + + "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/cache" + "github.com/Rain-kl/Wavelet/plugins/infra/database" + "github.com/Rain-kl/Wavelet/plugins/infra/logger" + "github.com/Rain-kl/Wavelet/plugins/infra/storage" +) + +type TestUser struct { + ID uint64 `gorm:"primaryKey"` + Name string +} + +func setupTestDB(t *testing.T) *gorm.DB { + t.Helper() + dbPath := filepath.Join(t.TempDir(), "test.db") + db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&TestUser{})) + return db +} + +func TestDatabasePlugin(t *testing.T) { + ctx := core.NewContext(context.Background()) + testDB := setupTestDB(t) + + p := database.New(database.WithDB(testDB)) + require.Equal(t, "database", p.Name()) + require.NoError(t, p.Apply(ctx)) + + dbSvc, err := core.Inject[contracts.DBService](ctx) + require.NoError(t, err) + require.NotNil(t, dbSvc) + + // Verify GORM and DB methods + assert.NotNil(t, dbSvc.GORM()) + assert.NotNil(t, dbSvc.DB(context.Background())) + + // Test CRUD via service + user := TestUser{ID: 1, Name: "Alice"} + require.NoError(t, dbSvc.DB(context.Background()).Create(&user).Error) + + var fetched TestUser + require.NoError(t, dbSvc.GORM().First(&fetched, 1).Error) + assert.Equal(t, "Alice", fetched.Name) + + // Test NamedDB fallback + assert.NotNil(t, dbSvc.Named("replica")) +} + +func TestCachePluginRAMOnly(t *testing.T) { + ctx := core.NewContext(context.Background()) + p := cache.New() + require.Equal(t, "cache", p.Name()) + require.NoError(t, p.Apply(ctx)) + + cacheSvc, err := core.Inject[contracts.CacheService](ctx) + require.NoError(t, err) + require.NotNil(t, cacheSvc) + + type CacheItem struct { + Name string `json:"name"` + Count int `json:"count"` + } + + testCtx := context.Background() + + // 1. Get non-existing key + var notFound CacheItem + err = cacheSvc.Get(testCtx, "missing:key", ¬Found) + assert.ErrorIs(t, err, contracts.ErrCacheMiss) + + // 2. Set and Get + item := CacheItem{Name: "item1", Count: 42} + require.NoError(t, cacheSvc.Set(testCtx, "test:item1", item, time.Minute)) + + var retrieved CacheItem + require.NoError(t, cacheSvc.Get(testCtx, "test:item1", &retrieved)) + assert.Equal(t, item, retrieved) + + // 3. GetOrSet + var getOrSetTarget CacheItem + var loaderCalled bool + err = cacheSvc.GetOrSet(testCtx, "test:item1", &getOrSetTarget, time.Minute, func() (any, error) { + loaderCalled = true + return CacheItem{Name: "never_called", Count: 0}, nil + }) + require.NoError(t, err) + assert.False(t, loaderCalled) + assert.Equal(t, item, getOrSetTarget) + + // GetOrSet with cache miss + var newItem CacheItem + err = cacheSvc.GetOrSet(testCtx, "test:item2", &newItem, time.Minute, func() (any, error) { + loaderCalled = true + return CacheItem{Name: "loaded", Count: 99}, nil + }) + require.NoError(t, err) + assert.True(t, loaderCalled) + assert.Equal(t, "loaded", newItem.Name) + assert.Equal(t, 99, newItem.Count) + + // 4. Delete + require.NoError(t, cacheSvc.Delete(testCtx, "test:item1")) + err = cacheSvc.Get(testCtx, "test:item1", &retrieved) + assert.ErrorIs(t, err, contracts.ErrCacheMiss) + + // 5. Invalidate alias + require.NoError(t, cacheSvc.Invalidate(testCtx, "test:item2")) + err = cacheSvc.Get(testCtx, "test:item2", &retrieved) + assert.ErrorIs(t, err, contracts.ErrCacheMiss) +} + +func TestCachePluginWithRedisAndPubSub(t *testing.T) { + mr, err := miniredis.Run() + require.NoError(t, err) + defer mr.Close() + + rdb := redis.NewClient(&redis.Options{ + Addr: mr.Addr(), + }) + defer func() { _ = rdb.Close() }() + + ctx1 := core.NewContext(context.Background()) + ctx2 := core.NewContext(context.Background()) + + p1 := cache.New(cache.WithRedis(rdb), cache.WithKeyPrefix("test:")) + p2 := cache.New(cache.WithRedis(rdb), cache.WithKeyPrefix("test:")) + + require.NoError(t, p1.Apply(ctx1)) + require.NoError(t, p2.Apply(ctx2)) + + cache1, err := core.Inject[contracts.CacheService](ctx1) + require.NoError(t, err) + cache2, err := core.Inject[contracts.CacheService](ctx2) + require.NoError(t, err) + + testCtx := context.Background() + + // Node 1 writes to cache + type UserCache struct { + Name string `json:"name"` + } + require.NoError(t, cache1.Set(testCtx, "user:100", UserCache{Name: "Bob"}, 10*time.Minute)) + + // Node 2 reads from cache (misses Node 2's RAM, hits Redis, backfills Node 2's RAM) + var u2 UserCache + require.NoError(t, cache2.Get(testCtx, "user:100", &u2)) + assert.Equal(t, "Bob", u2.Name) + + // Node 1 deletes cache (evicts Node 1 RAM, Redis, and broadcasts to Node 2) + require.NoError(t, cache1.Delete(testCtx, "user:100")) + + // Verify Redis is deleted + var uRedis UserCache + err = cache1.Get(testCtx, "user:100", &uRedis) + assert.ErrorIs(t, err, contracts.ErrCacheMiss) + + // Clean up contexts + require.NoError(t, ctx1.Dispose()) + require.NoError(t, ctx2.Dispose()) +} + +func TestLoggerPlugin(t *testing.T) { + ctx := core.NewContext(context.Background()) + p := logger.New() + require.Equal(t, "logger", p.Name()) + require.NoError(t, p.Apply(ctx)) + + logSvc, err := core.Inject[contracts.LoggerService](ctx) + require.NoError(t, err) + require.NotNil(t, logSvc) + + testCtx := context.Background() + + // Should not panic on any log call + logSvc.Debug(testCtx, "debug message", "key1", "val1") + logSvc.Info(testCtx, "info message", "userID", 123) + logSvc.Warn(testCtx, "warn message", "warning", true) + logSvc.Error(testCtx, "error message", "err", "something broke") + + logSvc.Debugf(testCtx, "formatted debug %d", 1) + logSvc.Infof(testCtx, "formatted info %s", "test") + logSvc.Warnf(testCtx, "formatted warn %v", map[string]int{"a": 1}) + logSvc.Errorf(testCtx, "formatted error %s", "fatal") + + childLog := logSvc.With("module", "test_module") + require.NotNil(t, childLog) + childLog.Info(testCtx, "child log message", "action", "run") +} + +type memoryBackend struct { + mu sync.RWMutex + storage map[string][]byte +} + +func newMemoryBackend() *memoryBackend { + return &memoryBackend{ + storage: make(map[string][]byte), + } +} + +func (m *memoryBackend) 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.storage[key] = data + return objectstore.PutResult{Key: key, Bucket: "test-bucket"}, nil +} + +func (m *memoryBackend) Get(ctx context.Context, key string) (*objectstore.Object, error) { + m.mu.RLock() + defer m.mu.RUnlock() + data, ok := m.storage[key] + if !ok { + return nil, gorm.ErrRecordNotFound + } + return &objectstore.Object{ + Body: io.NopCloser(bytes.NewReader(data)), + ContentLength: int64(len(data)), + ContentType: "application/octet-stream", + }, nil +} + +func (m *memoryBackend) Delete(ctx context.Context, key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.storage, key) + return nil +} + +func (m *memoryBackend) Test(ctx context.Context) error { + return nil +} + +func TestStoragePlugin(t *testing.T) { + ctx := core.NewContext(context.Background()) + backend := newMemoryBackend() + + p := storage.New(storage.WithBackend(backend)) + require.Equal(t, "storage", p.Name()) + require.NoError(t, p.Apply(ctx)) + + storageSvc, err := core.Inject[contracts.StorageService](ctx) + require.NoError(t, err) + require.NotNil(t, storageSvc) + + testCtx := context.Background() + + // 1. Put + content := []byte("Hello, Wavelet Storage Plugin!") + putRes, err := storageSvc.Put(testCtx, "uploads/hello.txt", bytes.NewReader(content), int64(len(content)), "text/plain") + require.NoError(t, err) + assert.Equal(t, "uploads/hello.txt", putRes.Key) + assert.Equal(t, "test-bucket", putRes.Bucket) + + // 2. Get + obj, err := storageSvc.Get(testCtx, "uploads/hello.txt") + require.NoError(t, err) + require.NotNil(t, obj) + data, err := io.ReadAll(obj.Body) + require.NoError(t, err) + assert.Equal(t, content, data) + assert.Equal(t, int64(len(content)), obj.ContentLength) + + // 3. Delete + require.NoError(t, storageSvc.Delete(testCtx, "uploads/hello.txt")) + _, err = storageSvc.Get(testCtx, "uploads/hello.txt") + assert.Error(t, err) +} + +func TestAllInfraPluginsCombined(t *testing.T) { + ctx := core.NewContext(context.Background()) + testDB := setupTestDB(t) + memBackend := newMemoryBackend() + + require.NoError(t, database.New(database.WithDB(testDB)).Apply(ctx)) + require.NoError(t, cache.New().Apply(ctx)) + require.NoError(t, logger.New().Apply(ctx)) + require.NoError(t, storage.New(storage.WithBackend(memBackend)).Apply(ctx)) + + // Using3 to resolve dependencies concurrently + var resolved bool + err := core.Using3(ctx, func(db contracts.DBService, c contracts.CacheService, l contracts.LoggerService) { + resolved = true + assert.NotNil(t, db) + assert.NotNil(t, c) + assert.NotNil(t, l) + }) + require.NoError(t, err) + assert.True(t, resolved) + + // Using storage + err = core.Using(ctx, func(s contracts.StorageService) { + assert.NotNil(t, s) + }) + require.NoError(t, err) +} diff --git a/plugins/infra/logger/plugin.go b/plugins/infra/logger/plugin.go new file mode 100644 index 00000000..a8abdbc2 --- /dev/null +++ b/plugins/infra/logger/plugin.go @@ -0,0 +1,107 @@ +// Package logger provides the structured logging infrastructure plugin for Cordis. +package logger + +import ( + "context" + "fmt" + "strings" + + "github.com/Rain-kl/Wavelet/core" + "github.com/Rain-kl/Wavelet/core/contracts" + "github.com/Rain-kl/Wavelet/pkg/logger" +) + +// Plugin implements core.Plugin to provide contracts.LoggerService. +type Plugin struct{} + +// New creates a new logger infrastructure plugin. +func New() *Plugin { + return &Plugin{} +} + +// Name returns the unique identifier of the logger plugin. +func (p *Plugin) Name() string { + return "logger" +} + +// Apply mounts the logger service into the Context. +func (p *Plugin) Apply(ctx *core.Context) error { + svc := &loggerServiceImpl{} + core.Provide[contracts.LoggerService](ctx, svc) + return nil +} + +type loggerServiceImpl struct { + extraFields []any +} + +func (s *loggerServiceImpl) formatMsg(msg string, keysAndValues ...any) string { + allFields := make([]any, 0, len(s.extraFields)+len(keysAndValues)) + allFields = append(allFields, s.extraFields...) + allFields = append(allFields, keysAndValues...) + + if len(allFields) == 0 { + return msg + } + + var sb strings.Builder + sb.WriteString(msg) + sb.WriteString(" [") + for i := 0; i < len(allFields); i += 2 { + if i > 0 { + sb.WriteString(" ") + } + if i+1 < len(allFields) { + fmt.Fprintf(&sb, "%v=%v", allFields[i], allFields[i+1]) + } else { + fmt.Fprintf(&sb, "%v", allFields[i]) + } + } + sb.WriteString("]") + return sb.String() +} + +func (s *loggerServiceImpl) Debug(ctx context.Context, msg string, keysAndValues ...any) { + logger.DebugF(ctx, "%s", s.formatMsg(msg, keysAndValues...)) +} + +func (s *loggerServiceImpl) Info(ctx context.Context, msg string, keysAndValues ...any) { + logger.InfoF(ctx, "%s", s.formatMsg(msg, keysAndValues...)) +} + +func (s *loggerServiceImpl) Warn(ctx context.Context, msg string, keysAndValues ...any) { + logger.WarnF(ctx, "%s", s.formatMsg(msg, keysAndValues...)) +} + +func (s *loggerServiceImpl) Error(ctx context.Context, msg string, keysAndValues ...any) { + logger.ErrorF(ctx, "%s", s.formatMsg(msg, keysAndValues...)) +} + +func (s *loggerServiceImpl) Debugf(ctx context.Context, format string, args ...any) { + msg := fmt.Sprintf(format, args...) + logger.DebugF(ctx, "%s", s.formatMsg(msg)) +} + +func (s *loggerServiceImpl) Infof(ctx context.Context, format string, args ...any) { + msg := fmt.Sprintf(format, args...) + logger.InfoF(ctx, "%s", s.formatMsg(msg)) +} + +func (s *loggerServiceImpl) Warnf(ctx context.Context, format string, args ...any) { + msg := fmt.Sprintf(format, args...) + logger.WarnF(ctx, "%s", s.formatMsg(msg)) +} + +func (s *loggerServiceImpl) Errorf(ctx context.Context, format string, args ...any) { + msg := fmt.Sprintf(format, args...) + logger.ErrorF(ctx, "%s", s.formatMsg(msg)) +} + +func (s *loggerServiceImpl) With(keysAndValues ...any) contracts.LoggerService { + merged := make([]any, 0, len(s.extraFields)+len(keysAndValues)) + merged = append(merged, s.extraFields...) + merged = append(merged, keysAndValues...) + return &loggerServiceImpl{ + extraFields: merged, + } +} diff --git a/plugins/infra/logger/plugin_test.go b/plugins/infra/logger/plugin_test.go new file mode 100644 index 00000000..6ac452e2 --- /dev/null +++ b/plugins/infra/logger/plugin_test.go @@ -0,0 +1,44 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package logger_test + +import ( + "context" + "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/plugins/infra/logger" +) + +func TestLoggerPluginOperations(t *testing.T) { + p := logger.New() + assert.Equal(t, "logger", p.Name()) + + ctx := core.NewContext(context.Background()) + require.NoError(t, p.Apply(ctx)) + + svc, err := core.Inject[contracts.LoggerService](ctx) + require.NoError(t, err) + require.NotNil(t, svc) + + testCtx := context.Background() + + svc.Debug(testCtx, "debug msg", "key", "val") + svc.Info(testCtx, "info msg", "user", 1) + svc.Warn(testCtx, "warn msg", "odd_field") + svc.Error(testCtx, "error msg") + + svc.Debugf(testCtx, "debug %s", "formatted") + svc.Infof(testCtx, "info %d", 100) + svc.Warnf(testCtx, "warn %v", true) + svc.Errorf(testCtx, "error %s", "fail") + + withLogger := svc.With("trace", "t-123", "span", "s-456") + require.NotNil(t, withLogger) + withLogger.Info(testCtx, "enriched message") +} diff --git a/plugins/infra/storage/plugin.go b/plugins/infra/storage/plugin.go new file mode 100644 index 00000000..ea572aee --- /dev/null +++ b/plugins/infra/storage/plugin.go @@ -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 +} diff --git a/plugins/infra/storage/plugin_test.go b/plugins/infra/storage/plugin_test.go new file mode 100644 index 00000000..630862ee --- /dev/null +++ b/plugins/infra/storage/plugin_test.go @@ -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) +}