kra-new/internal/global/resources_test.go

package global

import (
	"context"
	"io"
	"log/slog"
	"sync"
	"testing"

	"github.com/redis/go-redis/v9"
	"go.mongodb.org/mongo-driver/mongo"
	"gorm.io/gorm"
	platformmq "kra/pkg/mq"
	platformws "kra/pkg/websocket"
)

type storageStub struct{}

func (*storageStub) Open(context.Context, string) (io.ReadCloser, error) { return nil, nil }
func (*storageStub) Delete(context.Context, string) error                { return nil }
func (*storageStub) DeletePrefix(context.Context, string) error          { return nil }

type schedulerStub struct{}

func (*schedulerStub) Start(context.Context) error { return nil }
func (*schedulerStub) Stop(context.Context) error  { return nil }

type mqStub struct{}

func (*mqStub) Register(platformmq.SubscriptionSet) error { return nil }
func (*mqStub) Unregister(string) error                   { return nil }
func (*mqStub) Client(string) platformmq.Client           { return nil }
func (*mqStub) PublishTo(context.Context, string, string, []byte, byte, bool) error {
	return nil
}
func (*mqStub) SubscribeTo(context.Context, string, string, byte, platformmq.Handler) error {
	return nil
}
func (*mqStub) UnsubscribeFrom(context.Context, string, ...string) error { return nil }
func (*mqStub) ConnectedTo(string) bool                                  { return true }

type webSocketStub struct{}

func (*webSocketStub) Enabled() bool                                     { return true }
func (*webSocketStub) Broadcast([]byte) error                            { return nil }
func (*webSocketStub) BroadcastBinary([]byte) error                      { return nil }
func (*webSocketStub) Sessions() ([]*platformws.Session, error)          { return nil, nil }
func (*webSocketStub) Len() int                                          { return 0 }
func (*webSocketStub) Send(*platformws.Session, []byte) error            { return nil }
func (*webSocketStub) SendBinary(*platformws.Session, []byte) error      { return nil }
func (*webSocketStub) OnMessage(func(*platformws.Session, []byte))       {}
func (*webSocketStub) OnBinaryMessage(func(*platformws.Session, []byte)) {}
func (*webSocketStub) OnConnect(func(*platformws.Session))               {}
func (*webSocketStub) OnDisconnect(func(*platformws.Session))            {}

func TestResourceRegistryDefaults(t *testing.T) {
	registry := NewResourceRegistry()
	if registry.Logger() == nil {
		t.Fatal("default logger is nil")
	}
	resources := registry.Snapshot()
	if resources.DB != nil || len(resources.NamedDBs) != 0 || resources.Redis != nil || len(resources.NamedRedis) != 0 || resources.Mongo != nil || resources.Storage != nil || resources.MQ != nil || resources.WebSocket != nil || resources.Scheduler != nil {
		t.Fatalf("unexpected initialized resource: %#v", resources)
	}
}

func TestResourceRegistryReplaceAndSetters(t *testing.T) {
	logger := slog.Default().With("test", true)
	db := &gorm.DB{}
	redisClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	t.Cleanup(func() { _ = redisClient.Close() })
	mongoClient := &mongo.Client{}
	storage := &storageStub{}
	mq := &mqStub{}
	webSocket := &webSocketStub{}
	scheduler := &schedulerStub{}

	registry := NewResourceRegistry()
	registry.Replace(Resources{
		Logger: logger, DB: db, NamedDBs: map[string]*gorm.DB{"reporting": db},
		Redis: redisClient, NamedRedis: map[string]redis.UniversalClient{"cache": redisClient}, Mongo: mongoClient,
		Storage: storage, MQ: mq, WebSocket: webSocket, Scheduler: scheduler,
	})
	resources := registry.Snapshot()
	if resources.Logger != logger || resources.DB != db || resources.Redis != redisClient || resources.Mongo != mongoClient || resources.Storage != storage || resources.MQ != mq || resources.WebSocket != webSocket || resources.Scheduler != scheduler {
		t.Fatalf("replacement snapshot does not match: %#v", resources)
	}

	nextDB := &gorm.DB{}
	registry.SetDB(nextDB)
	if registry.DB() != nextDB {
		t.Fatal("database setter did not publish the new database")
	}
	if registry.Redis() != redisClient || registry.Storage() != storage {
		t.Fatal("database setter changed unrelated resources")
	}
	if registry.NamedDB("reporting") != db || registry.NamedDB("") != nextDB {
		t.Fatal("named database lookup returned the wrong resource")
	}
	if registry.NamedRedis("cache") != redisClient || registry.NamedRedis("") != redisClient {
		t.Fatal("named redis lookup returned the wrong resource")
	}
	inputDBs := map[string]*gorm.DB{"analytics": db}
	registry.SetNamedDBs(inputDBs)
	inputDBs["mutated"] = nextDB
	if registry.NamedDB("mutated") != nil || registry.NamedDB("analytics") != db {
		t.Fatal("named database map was not copied")
	}
	inputRedis := map[string]redis.UniversalClient{"sessions": redisClient}
	registry.SetNamedRedis(inputRedis)
	inputRedis["mutated"] = redisClient
	if registry.NamedRedis("mutated") != nil || registry.NamedRedis("sessions") != redisClient {
		t.Fatal("named redis map was not copied")
	}
}

func TestResourceRegistryNormalizesTypedNilInterfaces(t *testing.T) {
	registry := NewResourceRegistry()
	var redisClient *redis.Client
	var storage *storageStub
	var mq *mqStub
	var webSocket *webSocketStub
	var scheduler *schedulerStub
	registry.Replace(Resources{Redis: redisClient, Storage: storage, MQ: mq, WebSocket: webSocket, Scheduler: scheduler})
	resources := registry.Snapshot()
	if resources.Redis != nil || resources.Storage != nil || resources.MQ != nil || resources.WebSocket != nil || resources.Scheduler != nil {
		t.Fatalf("typed nil resource was retained: %#v", resources)
	}
	if resources.Logger == nil {
		t.Fatal("nil logger was not replaced with the default logger")
	}
}

func TestResourceRegistryConcurrentAccess(t *testing.T) {
	registry := NewResourceRegistry()
	const iterations = 1000
	var wait sync.WaitGroup
	for index := 0; index < 8; index++ {
		wait.Add(1)
		go func() {
			defer wait.Done()
			for step := 0; step < iterations; step++ {
				registry.SetDB(&gorm.DB{})
				_ = registry.Snapshot()
				_ = registry.Logger()
			}
		}()
	}
	wait.Wait()
	if registry.DB() == nil {
		t.Fatal("concurrent setters lost the database")
	}
}