fix(subscriptions): reject concurrent duplicate rules

Fixes #66

internal/service/subscription.go

@@ -98,8 +98,17 @@ func (s *SubscriptionService) Create(ctx context.Context, sub *model.Subscriptio
 		return errors.New("name and feed_url required")
 	}
 	normalizeSubscriptionDefaults(sub)
+	model.RefreshSubscriptionIdentity(sub)
+	if duplicate, err := s.subscriptionDuplicate(ctx, sub, ""); err != nil {
+		return err
+	} else if duplicate != nil {
+		return newSubscriptionAlreadyExistsError(duplicate.ID)
+	}
 	enabled := sub.Enabled
 	if err := s.repo.Subscription.Create(ctx, sub); err != nil {
+		if duplicate, lookupErr := s.subscriptionDuplicate(ctx, sub, ""); lookupErr == nil && duplicate != nil {
+			return newSubscriptionAlreadyExistsError(duplicate.ID)
+		}
 		return err
 	}
 	if !enabled {

internal/service/subscription_archive.go

@@ -23,6 +23,18 @@ func (s *SubscriptionService) Restore(ctx context.Context, id string) (*model.Su
 	if err := s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).First(&sub).Error; err != nil {
 		return nil, err
 	}
+	sub.Enabled = true
+	sub.ArchivedAt = nil
+	sub.ArchiveReason = ""
+	sub.DeletedAt.Valid = false
+	sub.TotalEpisodes = 0
+	normalizeSubscriptionDefaults(&sub)
+	model.RefreshSubscriptionIdentity(&sub)
+	if duplicate, err := s.subscriptionDuplicate(ctx, &sub, sub.ID); err != nil {
+		return nil, err
+	} else if duplicate != nil {
+		return nil, newSubscriptionAlreadyExistsError(duplicate.ID)
+	}
 	if err := s.repo.DB.WithContext(ctx).Unscoped().Model(&model.Subscription{}).
 		Where("id = ?", id).
 		Updates(map[string]any{
@@ -30,12 +42,16 @@ func (s *SubscriptionService) Restore(ctx context.Context, id string) (*model.Su
 			"archived_at":    nil,
 			"archive_reason": "",
 			"deleted_at":     nil,
+			"identity_key":   sub.IdentityKey,
 			// 重置为 0:此前可能被 feed 低估并锁死(updateSubscriptionTotalEpisodes
 			// 只增不减,resolveSubscriptionTotalEpisodes 见 >0 即不再回查元数据)。
 			// 归零后下次 run 会从 TMDb/豆瓣等权威源重算真实总集数,避免恢复后
 			// 因"误判已无缺集"而不再搜索资源。
 			"total_episodes": 0,
 		}).Error; err != nil {
+		if duplicate, lookupErr := s.subscriptionDuplicate(ctx, &sub, sub.ID); lookupErr == nil && duplicate != nil {
+			return nil, newSubscriptionAlreadyExistsError(duplicate.ID)
+		}
 		return nil, err
 	}
 	if s.repo.Setting != nil {

internal/service/subscription_identity.go

@@ -0,0 +1,89 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/ShukeBta/MediaStationGo/internal/model"
+)
+
+var ErrSubscriptionAlreadyExists = errors.New("subscription already exists")
+
+type SubscriptionAlreadyExistsError struct {
+	ExistingID string
+}
+
+func (e *SubscriptionAlreadyExistsError) Error() string {
+	if e == nil || e.ExistingID == "" {
+		return ErrSubscriptionAlreadyExists.Error()
+	}
+	return fmt.Sprintf("%s: %s", ErrSubscriptionAlreadyExists, e.ExistingID)
+}
+
+func (e *SubscriptionAlreadyExistsError) Unwrap() error {
+	return ErrSubscriptionAlreadyExists
+}
+
+func newSubscriptionAlreadyExistsError(existingID string) error {
+	return &SubscriptionAlreadyExistsError{ExistingID: existingID}
+}
+
+func SubscriptionAlreadyExistsID(err error) string {
+	var conflict *SubscriptionAlreadyExistsError
+	if errors.As(err, &conflict) && conflict != nil {
+		return conflict.ExistingID
+	}
+	return ""
+}
+
+func (s *SubscriptionService) subscriptionDuplicate(ctx context.Context, sub *model.Subscription, excludeID string) (*model.Subscription, error) {
+	if s == nil || s.repo == nil || s.repo.Subscription == nil || sub == nil {
+		return nil, nil
+	}
+	return s.repo.Subscription.FindActiveByIdentity(ctx, sub.UserID, sub.IdentityKey, excludeID)
+}
+
+// Update applies API patch fields while recomputing the functional identity.
+// The database partial unique index remains the final concurrency guard.
+func (s *SubscriptionService) Update(ctx context.Context, id string, updates map[string]any) error {
+	if s == nil || s.repo == nil || s.repo.DB == nil {
+		return errors.New("subscription service unavailable")
+	}
+	var sub model.Subscription
+	if err := s.repo.DB.WithContext(ctx).Where("id = ?", id).First(&sub).Error; err != nil {
+		return err
+	}
+	raw, err := json.Marshal(updates)
+	if err != nil {
+		return err
+	}
+	if err := json.Unmarshal(raw, &sub); err != nil {
+		return err
+	}
+	if sub.Name == "" || sub.FeedURL == "" {
+		return errors.New("name and feed_url required")
+	}
+	normalizeSubscriptionDefaults(&sub)
+	model.RefreshSubscriptionIdentity(&sub)
+	if duplicate, err := s.subscriptionDuplicate(ctx, &sub, sub.ID); err != nil {
+		return err
+	} else if duplicate != nil {
+		return newSubscriptionAlreadyExistsError(duplicate.ID)
+	}
+
+	updates["search_mode"] = sub.SearchMode
+	updates["resolution"] = sub.Resolution
+	updates["wash_priority"] = sub.WashPriority
+	updates["priority"] = sub.Priority
+	updates["identity_key"] = sub.IdentityKey
+	if err := s.repo.DB.WithContext(ctx).Model(&model.Subscription{}).
+		Where("id = ?", sub.ID).Updates(updates).Error; err != nil {
+		if duplicate, lookupErr := s.subscriptionDuplicate(ctx, &sub, sub.ID); lookupErr == nil && duplicate != nil {
+			return newSubscriptionAlreadyExistsError(duplicate.ID)
+		}
+		return err
+	}
+	return nil
+}

internal/service/subscription_identity_test.go

@@ -0,0 +1,114 @@
+package service
+
+import (
+	"errors"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/ShukeBta/MediaStationGo/internal/config"
+	"github.com/ShukeBta/MediaStationGo/internal/database"
+	"github.com/ShukeBta/MediaStationGo/internal/model"
+	"github.com/ShukeBta/MediaStationGo/internal/repository"
+)
+
+func TestSubscriptionCreateRejectsConcurrentDuplicateRules(t *testing.T) {
+	cfg := &config.Config{}
+	cfg.Database.Type = "sqlite"
+	cfg.Database.DBPath = filepath.Join(t.TempDir(), "subscriptions.db")
+	cfg.Database.WALMode = true
+	cfg.Database.BusyTimeout = 5000
+	cfg.Database.MaxOpenConns = 4
+	db, err := database.Open(cfg, zap.NewNop())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if sqlDB, dbErr := db.DB(); dbErr == nil {
+		t.Cleanup(func() { _ = sqlDB.Close() })
+	}
+	if err := database.AutoMigrate(db); err != nil {
+		t.Fatal(err)
+	}
+	repos := repository.New(db)
+	svc := NewSubscriptionService(cfg, zap.NewNop(), repos, nil, nil, NewHub(zap.NewNop()))
+
+	start := make(chan struct{})
+	results := make(chan error, 2)
+	var wg sync.WaitGroup
+	for i := 0; i < 2; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			results <- svc.Create(t.Context(), &model.Subscription{
+				UserID:     "user-1",
+				Name:       "Example Show",
+				FeedURL:    "site-search://search?keyword=Example+Show",
+				Filter:     "Example Show",
+				MediaType:  "tv",
+				Resolution: "1080p",
+				Enabled:    true,
+			})
+		}()
+	}
+	close(start)
+	wg.Wait()
+	close(results)
+
+	var created, conflicts int
+	for createErr := range results {
+		switch {
+		case createErr == nil:
+			created++
+		case errors.Is(createErr, ErrSubscriptionAlreadyExists):
+			conflicts++
+		default:
+			t.Fatalf("unexpected create error: %v", createErr)
+		}
+	}
+	if created != 1 || conflicts != 1 {
+		t.Fatalf("created=%d conflicts=%d, want 1/1", created, conflicts)
+	}
+	var count int64
+	if err := db.Model(&model.Subscription{}).Where("archived_at IS NULL").Count(&count).Error; err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 {
+		t.Fatalf("active subscription count = %d, want 1", count)
+	}
+}
+
+func TestSubscriptionUpdateAndRestoreReturnExistingConflict(t *testing.T) {
+	db := newServiceTestDB(t, &model.Subscription{})
+	if err := database.AutoMigrate(db); err != nil {
+		t.Fatal(err)
+	}
+	repos := repository.New(db)
+	svc := NewSubscriptionService(&config.Config{}, zap.NewNop(), repos, nil, nil, NewHub(zap.NewNop()))
+	first := &model.Subscription{UserID: "user-1", Name: "Example", FeedURL: "rss://example", Filter: "Example", Resolution: "1080p", Enabled: true}
+	second := &model.Subscription{UserID: "user-1", Name: "Example", FeedURL: "rss://example", Filter: "Example", Resolution: "2160p", Enabled: true}
+	if err := svc.Create(t.Context(), first); err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.Create(t.Context(), second); err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.Update(t.Context(), second.ID, map[string]any{"resolution": "1080p"}); !errors.Is(err, ErrSubscriptionAlreadyExists) || SubscriptionAlreadyExistsID(err) != first.ID {
+		t.Fatalf("update error = %v, want conflict with %s", err, first.ID)
+	}
+
+	archivedAt := time.Now()
+	archived := *first
+	archived.ID = ""
+	archived.ArchivedAt = &archivedAt
+	archived.Enabled = false
+	if err := repos.Subscription.Create(t.Context(), &archived); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := svc.Restore(t.Context(), archived.ID); !errors.Is(err, ErrSubscriptionAlreadyExists) || SubscriptionAlreadyExistsID(err) != first.ID {
+		t.Fatalf("restore error = %v, want conflict with %s", err, first.ID)
+	}
+}

internal/service/subscription_metadata_prepare.go

@@ -16,7 +16,12 @@ func (s *SubscriptionService) prepareSubscriptionForRun(ctx context.Context, sub
 	}
 	normalizeSubscriptionDefaults(sub)
 	updates := map[string]any{}
-	if s.fillSubscriptionRunMetadata(ctx, sub, updates); len(updates) > 0 && s.repo != nil && s.repo.DB != nil {
+	s.fillSubscriptionRunMetadata(ctx, sub, updates)
+	if identityKey := model.SubscriptionIdentityKey(sub); identityKey != sub.IdentityKey {
+		sub.IdentityKey = identityKey
+		updates["identity_key"] = identityKey
+	}
+	if len(updates) > 0 && s.repo != nil && s.repo.DB != nil {
 		if err := s.repo.DB.WithContext(ctx).Model(&model.Subscription{}).Where("id = ?", sub.ID).Updates(updates).Error; err != nil && s.log != nil {
 			s.log.Debug("subscription metadata prepare persist failed", zap.String("id", sub.ID), zap.Error(err))
 		}