fix[backend](integrations): fixed plugins yaml files integration form…

backend/modules/integrations/repository/tenant_store.go

@@ -7,12 +7,32 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"syscall"
 
 	"github.com/utmstack/utmstack/backend/modules/integrations/connectors"
 	"github.com/utmstack/utmstack/backend/modules/integrations/domain"
 	"gopkg.in/yaml.v3"
 )
 
+// withFileLock acquires an exclusive advisory lock on <path>.lock via flock,
+// runs fn, releases the lock on close. The kernel drops the lock on fd close
+// (including process death), so a crashed writer never leaves a stuck lock.
+// Linux/BSD only; UTMStack runs in Linux containers.
+func withFileLock(path string, fn func() error) error {
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return err
+	}
+	lf, err := os.OpenFile(path+".lock", os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return err
+	}
+	defer lf.Close()
+	if err := syscall.Flock(int(lf.Fd()), syscall.LOCK_EX); err != nil {
+		return err
+	}
+	return fn()
+}
+
 var _ connectors.TenantRepository = (*TenantStore)(nil)
 
 type TenantStore struct {
@@ -37,29 +57,44 @@ var pluginNames = map[string]string{
 	"SOPHOS":       "sophos",
 }
 
-func tenantFileName(module string) string {
+func pluginKey(module string) string {
 	name, ok := pluginNames[strings.ToUpper(module)]
 	if !ok {
 		name = strings.ToLower(module)
 	}
-	return "system_plugins_" + name + ".yaml"
+	return name
+}
+
+func tenantFileName(module string) string {
+	return "system_plugins_" + pluginKey(module) + ".yaml"
 }
 
 func (s *TenantStore) path(module string) string {
 	return filepath.Join(s.dir, tenantFileName(module))
 }
 
+// pluginsFile is the on-disk wrapper: plugins.<name>.tenants: [...].
+type pluginsFile struct {
+	Plugins map[string]pluginEntry `yaml:"plugins"`
+}
+
+type pluginEntry struct {
+	Tenants []domain.Tenant `yaml:"tenants"`
+}
+
 func (s *TenantStore) SetActiveByModule(_ context.Context, module string, active bool) error {
 	if active {
 		return nil
 	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if err := os.Remove(s.path(module)); err != nil && !errors.Is(err, os.ErrNotExist) {
-		return err
-	}
-	return nil
+	return withFileLock(s.path(module), func() error {
+		if err := os.Remove(s.path(module)); err != nil && !errors.Is(err, os.ErrNotExist) {
+			return err
+		}
+		return nil
+	})
 }
 
 func (s *TenantStore) Load(module string) ([]domain.Tenant, error) {
@@ -70,18 +105,28 @@ func (s *TenantStore) Load(module string) ([]domain.Tenant, error) {
 	if err != nil {
 		return nil, err
 	}
-	var tenants []domain.Tenant
-	if err := yaml.Unmarshal(data, &tenants); err != nil {
+	var pf pluginsFile
+	if err := yaml.Unmarshal(data, &pf); err != nil {
 		return nil, err
 	}
-	return tenants, nil
+	entry, ok := pf.Plugins[pluginKey(module)]
+	if !ok {
+		return []domain.Tenant{}, nil
+	}
+	return entry.Tenants, nil
 }
 
 func (s *TenantStore) Save(module string, tenants []domain.Tenant) error {
-	if err := os.MkdirAll(s.dir, 0o755); err != nil {
-		return err
-	}
-	data, err := yaml.Marshal(tenants)
+	return withFileLock(s.path(module), func() error {
+		return s.saveLocked(module, tenants)
+	})
+}
+
+// saveLocked writes the file assuming the caller already holds the file lock.
+func (s *TenantStore) saveLocked(module string, tenants []domain.Tenant) error {
+	key := pluginKey(module)
+	pf := pluginsFile{Plugins: map[string]pluginEntry{key: {Tenants: tenants}}}
+	data, err := yaml.Marshal(pf)
 	if err != nil {
 		return err
 	}
@@ -97,38 +142,42 @@ func (s *TenantStore) Upsert(module string, t domain.Tenant) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	tenants, err := s.Load(module)
-	if err != nil {
-		return err
-	}
-	for i := range tenants {
-		if tenants[i].Name == t.Name {
-			tenants[i] = t
-			return s.Save(module, tenants)
+	return withFileLock(s.path(module), func() error {
+		tenants, err := s.Load(module)
+		if err != nil {
+			return err
 		}
-	}
-	return s.Save(module, append(tenants, t))
+		for i := range tenants {
+			if tenants[i].Name == t.Name {
+				tenants[i] = t
+				return s.saveLocked(module, tenants)
+			}
+		}
+		return s.saveLocked(module, append(tenants, t))
+	})
 }
 
 func (s *TenantStore) Delete(module, name string) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	tenants, err := s.Load(module)
-	if err != nil {
-		return err
-	}
-	out := make([]domain.Tenant, 0, len(tenants))
-	found := false
-	for _, t := range tenants {
-		if t.Name == name {
-			found = true
-			continue
+	return withFileLock(s.path(module), func() error {
+		tenants, err := s.Load(module)
+		if err != nil {
+			return err
 		}
-		out = append(out, t)
-	}
-	if !found {
-		return domain.ErrTenantNotFound
-	}
-	return s.Save(module, out)
+		out := make([]domain.Tenant, 0, len(tenants))
+		found := false
+		for _, t := range tenants {
+			if t.Name == name {
+				found = true
+				continue
+			}
+			out = append(out, t)
+		}
+		if !found {
+			return domain.ErrTenantNotFound
+		}
+		return s.saveLocked(module, out)
+	})
 }

backend/modules/integrations/repository/tenant_store_test.go

@@ -0,0 +1,115 @@
+package repository
+
+import (
+	"os"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/utmstack/utmstack/backend/modules/integrations/domain"
+)
+
+func TestTenantStoreRoundtrip(t *testing.T) {
+	dir := t.TempDir()
+	s := NewTenantStore(dir)
+
+	want := []domain.Tenant{
+		{Name: "t1", Config: map[string]string{"access_key": "AK", "secret": "SK"}},
+		{Name: "t2", Config: map[string]string{"access_key": "AK2"}},
+	}
+	if err := s.Save("AWS_IAM_USER", want); err != nil {
+		t.Fatalf("Save: %v", err)
+	}
+	got, err := s.Load("AWS_IAM_USER")
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	if len(got) != len(want) {
+		t.Fatalf("len(got)=%d want %d", len(got), len(want))
+	}
+	for i := range want {
+		if got[i].Name != want[i].Name {
+			t.Fatalf("[%d] name got %q want %q", i, got[i].Name, want[i].Name)
+		}
+		for k, v := range want[i].Config {
+			if got[i].Config[k] != v {
+				t.Fatalf("[%d] %s got %q want %q", i, k, got[i].Config[k], v)
+			}
+		}
+	}
+
+	data, err := os.ReadFile(s.path("AWS_IAM_USER"))
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	txt := string(data)
+	if !strings.Contains(txt, "plugins:") || !strings.Contains(txt, "aws:") || !strings.Contains(txt, "tenants:") {
+		t.Fatalf("unexpected file layout:\n%s", txt)
+	}
+}
+
+func TestWithFileLockWaits(t *testing.T) {
+	dir := t.TempDir()
+	target := dir + "/f.yaml"
+
+	// pre-hold the flock via a peer fd
+	peer, err := os.OpenFile(target+".lock", os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_EX); err != nil {
+		t.Fatalf("flock: %v", err)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- withFileLock(target, func() error { return nil })
+	}()
+
+	select {
+	case <-done:
+		t.Fatal("withFileLock returned while flock was held")
+	case <-time.After(120 * time.Millisecond):
+	}
+
+	if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_UN); err != nil {
+		t.Fatalf("unlock: %v", err)
+	}
+	_ = peer.Close()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("withFileLock: %v", err)
+		}
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("withFileLock did not return after flock released")
+	}
+}
+
+// TestWithFileLockKernelReleasesOnClose proves the crash-recovery property: if
+// the peer holding the flock closes its fd (equivalent to process death), the
+// kernel releases the lock and a new acquirer proceeds without any TTL/steal.
+func TestWithFileLockKernelReleasesOnClose(t *testing.T) {
+	dir := t.TempDir()
+	target := dir + "/f.yaml"
+
+	peer, err := os.OpenFile(target+".lock", os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if err := syscall.Flock(int(peer.Fd()), syscall.LOCK_EX); err != nil {
+		t.Fatalf("flock: %v", err)
+	}
+	// simulate crash: close fd without explicit unlock
+	_ = peer.Close()
+
+	start := time.Now()
+	if err := withFileLock(target, func() error { return nil }); err != nil {
+		t.Fatalf("withFileLock: %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+		t.Fatalf("lock not released on close: %v", elapsed)
+	}
+}