feat: search provider injection + easysearch cluster CRUD module

core/config/system.go

@@ -72,7 +72,7 @@ type NetworkConfig struct {
 	SkipOccupiedPort bool   `config:"skip_occupied_port" json:"skip_occupied_port,omitempty" elastic_mapping:"skip_occupied_port: { type: boolean }"`
 	//ReusePort is nil when reuse_port is not configured, so applications can
 	//tell "unset" apart from an explicit false and apply their own default.
-	ReusePort        *bool  `config:"reuse_port" json:"reuse_port,omitempty" elastic_mapping:"reuse_port: { type: boolean }"`
+	ReusePort *bool `config:"reuse_port" json:"reuse_port,omitempty" elastic_mapping:"reuse_port: { type: boolean }"`
 }
 
 // Helper function to report whether SO_REUSEPORT is explicitly enabled.
@@ -164,6 +164,8 @@ type SystemConfig struct {
 
 	PathConfig PathConfig `config:"path"`
 
+	ORMConfig ORMConfig `config:"orm"`
+
 	LoggingConfig LoggingConfig `config:"log"`
 
 	AllowMultiInstance bool `config:"allow_multi_instance"`
@@ -183,6 +185,10 @@ type SystemConfig struct {
 	HTTPClientConfig map[string]HTTPClientConfig `config:"http_client"`
 }
 
+type ORMConfig struct {
+	Enabled bool `config:"enabled"`
+}
+
 type CookieConfig struct {
 	Store         string `config:"store"`      //cookie/filesystem
 	StorePath     string `config:"store_path"` //filesystem only

core/elastic/api.go

@@ -170,7 +170,8 @@ type ScrollAPI interface {
 type ScriptAPI interface {
 	ScriptExists(scriptName string) (bool, error)
 	PutScript(scriptName string, script []byte) ([]byte, error)
-	SearchByTemplate(indexName, scriptName string, params map[string]interface{}) (*SearchResponse, error)
+	// SearchByTemplate is declared on TemplateAPI (same signature); adapters
+	// implement it once and satisfy both via embedding.
 	//GetScript(scriptName string)([]byte,error)
 	//DeleteScript(scriptName string)([]byte,error)
 }

core/elastic/client_provider.go

@@ -0,0 +1,156 @@
+/* Copyright © INFINI LTD. All rights reserved. */
+
+package elastic
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Client provider injection.
+//
+// The client factory (version probing + adapter selection, e.g.
+// modules/elastic/common.InitClientWithConfig) lives in modules/elastic,
+// which depends on this package — so it cannot be imported here without a
+// dependency cycle. Instead, the module registers its factory once at setup
+// time and this package exposes the config-keyed client cache on top of it.
+//
+// GetOrCreateClient therefore gives any caller (inline/stream config, a
+// storage-backed cluster, or an ad-hoc probe config) a live, self-contained
+// client without registering the cluster by ID first.
+// ──────────────────────────────────────────────────────────────────────────
+
+// ClientProvider builds a live client from a connection config. Registered
+// by modules/elastic at module setup (before its enabled check, so apps that
+// disable the elastic module but still talk to ES keep working).
+type ClientProvider func(cfg ElasticsearchConfig) (API, error)
+
+var (
+	clientProvider   ClientProvider
+	clientProviderMu sync.RWMutex
+)
+
+// RegisterClientProvider installs the client factory. Panics on nil or on a
+// duplicate registration — double wiring is a programming error.
+func RegisterClientProvider(p ClientProvider) {
+	if p == nil {
+		panic("elastic: RegisterClientProvider called with nil provider")
+	}
+	clientProviderMu.Lock()
+	defer clientProviderMu.Unlock()
+	if clientProvider != nil {
+		panic("elastic: client provider already registered")
+	}
+	clientProvider = p
+}
+
+// registeredProvider returns the installed factory, or an error explaining
+// what is missing (clearer than a nil-pointer panic at the call site).
+func registeredProvider() (ClientProvider, error) {
+	clientProviderMu.RLock()
+	defer clientProviderMu.RUnlock()
+	if clientProvider == nil {
+		return nil, errors.New("no client provider registered; modules/elastic registers one via elastic.RegisterClientProvider at setup")
+	}
+	return clientProvider, nil
+}
+
+// clientCache reuses clients across calls that hand in identical configs.
+// Keyed by connection identity (see configCacheKey), not by cluster ID.
+// Bounded eviction (LRU/TTL) can be layered on later; for now a simple map
+// since the number of distinct connection identities is small.
+var (
+	clientCache   = map[string]API{}
+	clientCacheMu sync.RWMutex
+)
+
+// GetOrCreateClient returns a live ES client for the given config, building
+// one on the first request and reusing it for subsequent calls with the same
+// connection identity.
+//
+// Unlike GetElasticClient(clusterID), the cluster does NOT need to be
+// registered: the returned client is self-contained — its metadata is built
+// from the config and attached via SetMetadata, so it never falls back to the
+// ID-based registry (which would panic if the cluster isn't registered).
+//
+// Set cfg.Version (and cfg.Distribution) to skip the network version probe
+// when the caller already knows them; otherwise the provider probes the
+// cluster once (on cache miss) to select the right version-specific adapter.
+func GetOrCreateClient(cfg ElasticsearchConfig) (API, error) {
+	provider, err := registeredProvider()
+	if err != nil {
+		return nil, err
+	}
+
+	key := configCacheKey(cfg)
+
+	clientCacheMu.RLock()
+	if c, ok := clientCache[key]; ok {
+		clientCacheMu.RUnlock()
+		return c, nil
+	}
+	clientCacheMu.RUnlock()
+
+	client, err := provider(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	// Make the client self-contained: attach a metadata built from the config
+	// so GetMetadata() returns it directly instead of looking the config up by
+	// ID (and panicking if the cluster isn't registered).
+	meta := &ElasticsearchMetadata{Config: &cfg}
+	if s, ok := client.(interface {
+		SetMetadata(*ElasticsearchMetadata)
+	}); ok {
+		s.SetMetadata(meta)
+	}
+
+	clientCacheMu.Lock()
+	// A concurrent caller may have built one first; keep the winner.
+	if existing, ok := clientCache[key]; ok {
+		clientCacheMu.Unlock()
+		return existing, nil
+	}
+	clientCache[key] = client
+	clientCacheMu.Unlock()
+	return client, nil
+}
+
+// InvalidateClient drops the cached client (if any) for the given config, so
+// the next GetOrCreateClient rebuilds it — e.g. after a cluster's endpoint or
+// credentials change.
+func InvalidateClient(cfg ElasticsearchConfig) {
+	key := configCacheKey(cfg)
+	clientCacheMu.Lock()
+	delete(clientCache, key)
+	clientCacheMu.Unlock()
+}
+
+// ResetClientCacheForTest clears the client cache and unregisters the client
+// provider. Test-only.
+func ResetClientCacheForTest() {
+	clientCacheMu.Lock()
+	clientCache = map[string]API{}
+	clientCacheMu.Unlock()
+	clientProviderMu.Lock()
+	clientProvider = nil
+	clientProviderMu.Unlock()
+}
+
+// configCacheKey returns a canonical identity for a config's connection: two
+// configs that reach the same cluster with the same auth/version share a key
+// (and thus a client). Secrets are part of the key (held in memory only, never
+// logged) so different credentials get different clients.
+func configCacheKey(cfg ElasticsearchConfig) string {
+	var b strings.Builder
+	fmt.Fprintf(&b, "%s|%s|%s|", cfg.GetAnyEndpoint(), strings.Join(cfg.Endpoints, ","), cfg.Distribution)
+	if cfg.BasicAuth != nil {
+		fmt.Fprintf(&b, "%s:%s|", cfg.BasicAuth.Username, cfg.BasicAuth.Password.Get())
+	}
+	fmt.Fprintf(&b, "%s|%s", cfg.Token.Get(), cfg.Version)
+	return b.String()
+}

core/elastic/client_provider_test.go

@@ -0,0 +1,160 @@
+/* Copyright © INFINI LTD. All rights reserved. */
+
+package elastic
+
+import (
+	"errors"
+	"testing"
+)
+
+func mkCfg(endpoint, version string) ElasticsearchConfig {
+	c := ElasticsearchConfig{}
+	c.ID = "test-" + endpoint
+	c.Name = endpoint
+	c.Endpoint = endpoint
+	c.Distribution = Elasticsearch
+	c.Version = version // set → provider skips the network version probe
+	c.Enabled = true
+	return c
+}
+
+// stubClient satisfies the fat API interface by embedding it (nil); tests
+// only exercise pointer identity and metadata plumbing, never API calls.
+type stubClient struct {
+	API
+	meta *ElasticsearchMetadata
+}
+
+func (s *stubClient) SetMetadata(m *ElasticsearchMetadata) { s.meta = m }
+func (s *stubClient) GetMetadata() *ElasticsearchMetadata  { return s.meta }
+
+// stubProvider returns a fresh client per call so tests can tell builds apart.
+func stubProvider(cfg ElasticsearchConfig) (API, error) {
+	return &stubClient{}, nil
+}
+
+func TestConfigCacheKey(t *testing.T) {
+	a := mkCfg("http://a:9200", "8.0.0")
+	a2 := mkCfg("http://a:9200", "8.0.0")
+	b := mkCfg("http://b:9200", "8.0.0")
+
+	if configCacheKey(a) != configCacheKey(a2) {
+		t.Fatal("identical configs must share a key")
+	}
+	if configCacheKey(a) == configCacheKey(b) {
+		t.Fatal("different endpoints must produce different keys")
+	}
+
+	a3 := mkCfg("http://a:9200", "9.0.0")
+	if configCacheKey(a) == configCacheKey(a3) {
+		t.Fatal("different versions must produce different keys")
+	}
+}
+
+func TestGetOrCreateClient_CachesByConfig(t *testing.T) {
+	ResetClientCacheForTest()
+	RegisterClientProvider(stubProvider)
+	cfg := mkCfg("http://x:9200", "8.0.0")
+
+	c1, err := GetOrCreateClient(cfg)
+	if err != nil {
+		t.Fatalf("first GetOrCreateClient: %v", err)
+	}
+	c2, err := GetOrCreateClient(cfg)
+	if err != nil {
+		t.Fatalf("second GetOrCreateClient: %v", err)
+	}
+	if c1 != c2 {
+		t.Fatal("same config should return the same cached client pointer")
+	}
+
+	other := mkCfg("http://y:9200", "8.0.0")
+	c3, err := GetOrCreateClient(other)
+	if err != nil {
+		t.Fatalf("GetOrCreateClient(other): %v", err)
+	}
+	if c1 == c3 {
+		t.Fatal("different config should return a different client")
+	}
+}
+
+func TestGetOrCreateClient_SelfContained(t *testing.T) {
+	ResetClientCacheForTest()
+	RegisterClientProvider(stubProvider)
+	cfg := mkCfg("http://self:9200", "8.0.0")
+
+	client, err := GetOrCreateClient(cfg)
+	if err != nil {
+		t.Fatalf("GetOrCreateClient: %v", err)
+	}
+
+	// The client must carry its own metadata built from the config — it should
+	// NOT need the cluster to be registered by ID (no RegisterInstance ran).
+	mg, ok := client.(interface {
+		GetMetadata() *ElasticsearchMetadata
+	})
+	if !ok {
+		t.Fatal("client does not expose GetMetadata")
+	}
+	md := mg.GetMetadata()
+	if md == nil || md.Config == nil || md.Config.Endpoint != "http://self:9200" {
+		t.Fatalf("self-contained metadata not set from config: %+v", md)
+	}
+}
+
+func TestGetOrCreateClient_ProviderErrorsPropagate(t *testing.T) {
+	ResetClientCacheForTest()
+	RegisterClientProvider(func(cfg ElasticsearchConfig) (API, error) {
+		return nil, errors.New("boom")
+	})
+	if _, err := GetOrCreateClient(mkCfg("http://err:9200", "8.0.0")); err == nil {
+		t.Fatal("provider error must propagate")
+	}
+}
+
+func TestGetOrCreateClient_NoProviderRegistered(t *testing.T) {
+	ResetClientCacheForTest()
+	_, err := GetOrCreateClient(mkCfg("http://none:9200", "8.0.0"))
+	if err == nil || !contains(err.Error(), "no client provider registered") {
+		t.Fatalf("expected a clear missing-provider error, got: %v", err)
+	}
+}
+
+func TestRegisterClientProvider_DoubleRegistrationPanics(t *testing.T) {
+	ResetClientCacheForTest()
+	RegisterClientProvider(stubProvider)
+	defer func() {
+		if recover() == nil {
+			t.Fatal("duplicate registration must panic")
+		}
+		ResetClientCacheForTest()
+	}()
+	RegisterClientProvider(stubProvider)
+}
+
+func TestInvalidateClient(t *testing.T) {
+	ResetClientCacheForTest()
+	RegisterClientProvider(stubProvider)
+	cfg := mkCfg("http://inv:9200", "8.0.0")
+
+	c1, _ := GetOrCreateClient(cfg)
+	InvalidateClient(cfg)
+	c2, _ := GetOrCreateClient(cfg)
+
+	if c1 == c2 {
+		t.Fatal("InvalidateClient should force a rebuild on next GetOrCreateClient")
+	}
+}
+
+func contains(s, sub string) bool {
+	return len(s) >= len(sub) && (s == sub || len(sub) == 0 || indexOf(s, sub) >= 0)
+}
+
+func indexOf(s, sub string) int {
+	for i := 0; i+len(sub) <= len(s); i++ {
+		if s[i:i+len(sub)] == sub {
+			return i
+		}
+	}
+	return -1
+}

core/elastic/view.go

@@ -33,10 +33,10 @@ type View struct {
 	ID              string    `json:"-" elastic_meta:"_id" elastic_mapping:"id: { type: keyword }"`
 	ClusterID       string    `json:"cluster_id" elastic_mapping:"cluster_id:{type:keyword}"`
 	Title           string    `json:"title" elastic_mapping:"title:{type:keyword}"`
-	ViewName        string    `json:"viewName" elastic_mapping:"view_name:{type:text}`
+	ViewName        string    `json:"viewName" elastic_mapping:"view_name:{type:text}"`
 	TimeFieldName   string    `json:"timeFieldName" elastic_mapping:"timeFieldName:{type:keyword}"`
 	Fields          string    `json:"fields" elastic_mapping:"fields:{type:text}"`
-	FieldFormatMap  string    `json:"fieldFormatMap" elastic_mapping:"fields:{type:text}`
+	FieldFormatMap  string    `json:"fieldFormatMap" elastic_mapping:"fields:{type:text}"`
 	UpdatedAt       time.Time `json:"updated_at,omitempty" elastic_mapping:"updated_at:{type:date}"`
 	DefaultLayoutID string    `json:"default_layout_id" elastic_mapping:"default_layout_id:{type:keyword}"`
 	ComplexFields   string    `json:"complex_fields" elastic_mapping:"complex_fields:{type:text}"`

core/env/env.go

@@ -335,6 +335,7 @@ func GetDefaultSystemConfig() config.SystemConfig {
 			Log:    "log",
 			Config: "config",
 		},
+		ORMConfig: config.ORMConfig{Enabled: true},
 
 		AllowMultiInstance: false,
 		MaxNumOfInstance:   5,

core/orm/orm.go

@@ -704,6 +704,16 @@ func saveOrUpdate(ctx *Context, o interface{}, delta util.MapStr, opType Operati
 
 		if exists {
 			if mergePartial && deltaNotEmpty {
+				// Seed the target with the stored state before overlaying the
+				// delta, so fields absent from the delta keep their stored
+				// values on full-replace stores (e.g. sqlite rewrites the whole
+				// row). Without this, a partial update would wipe every field
+				// not mentioned in the delta.
+				prevValue := reflect.ValueOf(prev)
+				if prevValue.Kind() == reflect.Ptr && !prevValue.IsNil() &&
+					prevValue.Type().Elem() == rValue.Type().Elem() && rValue.Elem().CanSet() {
+					rValue.Elem().Set(prevValue.Elem())
+				}
 				if err := mergeMapToStruct(delta, rValue); err != nil {
 					return err
 				}