diff --git a/backend/internal/handler/admin/channel_monitor_duplicate_test.go b/backend/internal/handler/admin/channel_monitor_duplicate_test.go new file mode 100644 index 0000000000..85670ca7b4 --- /dev/null +++ b/backend/internal/handler/admin/channel_monitor_duplicate_test.go @@ -0,0 +1,170 @@ +//go:build unit + +package admin + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +type duplicateChannelMonitorHandlerRepoStub struct { + service.ChannelMonitorRepository + source *service.ChannelMonitor + byOperation map[string]*service.ChannelMonitor + createCalls int +} + +func (r *duplicateChannelMonitorHandlerRepoStub) GetByID(_ context.Context, id int64) (*service.ChannelMonitor, error) { + if r.source == nil || r.source.ID != id { + return nil, service.ErrChannelMonitorNotFound + } + return r.source, nil +} + +func (r *duplicateChannelMonitorHandlerRepoStub) Create(_ context.Context, monitor *service.ChannelMonitor) error { + r.createCalls++ + monitor.ID = int64(100 + r.createCalls) + stored := *monitor + if stored.DuplicateOperationID != "" { + if r.byOperation == nil { + r.byOperation = make(map[string]*service.ChannelMonitor) + } + r.byOperation[stored.DuplicateOperationID] = &stored + } + return nil +} + +func (r *duplicateChannelMonitorHandlerRepoStub) FindByDuplicateOperationID(_ context.Context, operationID string) (*service.ChannelMonitor, error) { + monitor := r.byOperation[operationID] + if monitor == nil { + return nil, nil + } + cloned := *monitor + return &cloned, nil +} + +type duplicateChannelMonitorHandlerEncryptor struct{} + +func (duplicateChannelMonitorHandlerEncryptor) Encrypt(plaintext string) (string, error) { + return "ENC:" + plaintext, nil +} + +func (duplicateChannelMonitorHandlerEncryptor) Decrypt(ciphertext string) (string, error) { + return strings.TrimPrefix(ciphertext, "ENC:"), nil +} + +func setupDuplicateChannelMonitorRouter(t *testing.T) (*gin.Engine, *duplicateChannelMonitorHandlerRepoStub) { + t.Helper() + previousCoordinator := service.DefaultIdempotencyCoordinator() + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator( + newMemoryIdempotencyRepoStub(), + service.DefaultIdempotencyConfig(), + )) + t.Cleanup(func() { service.SetDefaultIdempotencyCoordinator(previousCoordinator) }) + + repo := &duplicateChannelMonitorHandlerRepoStub{ + source: &service.ChannelMonitor{ + ID: 42, + Name: "primary", + Provider: service.MonitorProviderOpenAI, + APIMode: service.MonitorAPIModeResponses, + Endpoint: "https://api.example.com", + APIKey: "ENC:top-secret", + PrimaryModel: "gpt-5.4-mini", + ExtraModels: []string{"gpt-5.4"}, + Enabled: true, + IntervalSeconds: 60, + BodyOverrideMode: service.MonitorBodyOverrideModeOff, + }, + } + monitorService := service.NewChannelMonitorService(repo, duplicateChannelMonitorHandlerEncryptor{}) + handler := NewChannelMonitorHandler(monitorService) + + gin.SetMode(gin.TestMode) + router := gin.New() + router.Use(func(c *gin.Context) { + c.Set(string(middleware2.ContextKeyUser), middleware2.AuthSubject{UserID: 77}) + c.Next() + }) + router.POST("/api/v1/admin/channel-monitors/:id/duplicate", handler.Duplicate) + return router, repo +} + +func TestDuplicateChannelMonitorHandlerRedactsKeyAndReplaysRetry(t *testing.T) { + router, repo := setupDuplicateChannelMonitorRouter(t) + + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/channel-monitors/42/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-channel-monitor-42") + first := httptest.NewRecorder() + router.ServeHTTP(first, request) + + require.Equal(t, http.StatusOK, first.Code) + require.Equal(t, 1, repo.createCalls) + require.Contains(t, first.Body.String(), `"name":"primary (Copy)"`) + require.Contains(t, first.Body.String(), `"api_key_masked":"top-***"`) + require.Contains(t, first.Body.String(), `"created_by":77`) + require.Contains(t, first.Body.String(), `"enabled":false`) + require.NotContains(t, first.Body.String(), "top-secret") + + retryRequest := httptest.NewRequest(http.MethodPost, "/api/v1/admin/channel-monitors/42/duplicate", nil) + retryRequest.Header.Set("Idempotency-Key", "duplicate-channel-monitor-42") + retry := httptest.NewRecorder() + router.ServeHTTP(retry, retryRequest) + + require.Equal(t, http.StatusOK, retry.Code) + require.Equal(t, "true", retry.Header().Get("X-Idempotency-Replayed")) + require.Equal(t, 1, repo.createCalls) + require.JSONEq(t, first.Body.String(), retry.Body.String()) +} + +func TestDuplicateChannelMonitorHandlerRejectsInvalidID(t *testing.T) { + router, repo := setupDuplicateChannelMonitorRouter(t) + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/channel-monitors/not-a-number/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-channel-monitor-invalid") + recorder := httptest.NewRecorder() + + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusBadRequest, recorder.Code) + require.Zero(t, repo.createCalls) +} + +func TestDuplicateChannelMonitorHandlerRecoversAfterMarkSucceededFailure(t *testing.T) { + router, repo := setupDuplicateChannelMonitorRouter(t) + idempotencyRepo := &failOnceMarkSucceededRepo{ + memoryIdempotencyRepoStub: newMemoryIdempotencyRepoStub(), + failNext: true, + } + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator( + idempotencyRepo, + service.DefaultIdempotencyConfig(), + )) + + call := func() *httptest.ResponseRecorder { + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/channel-monitors/42/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-channel-monitor-recovery") + recorder := httptest.NewRecorder() + router.ServeHTTP(recorder, request) + return recorder + } + + first := call() + second := call() + + require.Equal(t, http.StatusOK, first.Code) + require.Equal(t, http.StatusOK, second.Code) + require.Equal(t, "true", first.Header().Get("X-Idempotency-Recovered")) + require.Equal(t, "true", second.Header().Get("X-Idempotency-Recovered")) + require.Equal(t, 1, repo.createCalls, "ambiguous retries must not repeat the create side effect") + require.Contains(t, second.Body.String(), `"id":101`) + require.Contains(t, second.Body.String(), `"api_key_masked":"top-***"`) + require.NotContains(t, second.Body.String(), "top-secret") +} diff --git a/backend/internal/handler/admin/channel_monitor_handler.go b/backend/internal/handler/admin/channel_monitor_handler.go index a69b835849..ec5735f0ca 100644 --- a/backend/internal/handler/admin/channel_monitor_handler.go +++ b/backend/internal/handler/admin/channel_monitor_handler.go @@ -1,6 +1,8 @@ package admin import ( + "context" + "log/slog" "strconv" "strings" "time" @@ -333,6 +335,63 @@ func (h *ChannelMonitorHandler) Create(c *gin.Context) { response.Created(c, channelMonitorToResponse(m)) } +// Duplicate POST /api/v1/admin/channel-monitors/:id/duplicate +func (h *ChannelMonitorHandler) Duplicate(c *gin.Context) { + id, ok := ParseChannelMonitorID(c) + if !ok { + return + } + subject, _ := middleware2.GetAuthSubjectFromContext(c) + actorScope := adminActorScope(c) + + result, err := executeAdminIdempotent( + c, + "admin.channel_monitors.duplicate", + struct { + MonitorID int64 `json:"monitor_id"` + }{MonitorID: id}, + service.DefaultWriteIdempotencyTTL(), + func(ctx context.Context) (any, error) { + monitor, err := h.monitorService.Duplicate( + ctx, + id, + subject.UserID, + actorScope, + c.GetHeader("Idempotency-Key"), + ) + if err != nil { + return nil, err + } + return channelMonitorToResponse(monitor), nil + }, + ) + if err != nil { + reason := infraerrors.Reason(err) + if reason == infraerrors.Reason(service.ErrIdempotencyInProgress) || reason == infraerrors.Reason(service.ErrIdempotencyStoreUnavail) { + recovered, recoverErr := h.monitorService.RecoverDuplicate( + c.Request.Context(), + id, + actorScope, + c.GetHeader("Idempotency-Key"), + ) + if recoverErr != nil { + slog.Warn("channel_monitor_duplicate_recovery_failed", "monitor_id", id, "actor_scope", actorScope, "reason", reason, "error", recoverErr) + } else if recovered != nil { + c.Header("X-Idempotency-Recovered", "true") + response.Success(c, channelMonitorToResponse(recovered)) + return + } + } + response.ErrorFrom(c, err) + return + } + + if result != nil && result.Replayed { + c.Header("X-Idempotency-Replayed", "true") + } + response.Success(c, result.Data) +} + // Update PUT /api/v1/admin/channel-monitors/:id func (h *ChannelMonitorHandler) Update(c *gin.Context) { id, ok := ParseChannelMonitorID(c) diff --git a/backend/internal/repository/channel_monitor_duplicate_test.go b/backend/internal/repository/channel_monitor_duplicate_test.go new file mode 100644 index 0000000000..6b189dfe36 --- /dev/null +++ b/backend/internal/repository/channel_monitor_duplicate_test.go @@ -0,0 +1,29 @@ +//go:build unit + +package repository + +import ( + "testing" + + dbent "github.com/Wei-Shaw/sub2api/ent" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +func TestChannelMonitorDuplicateOperationMetadataStaysOutOfRuntimeHeaders(t *testing.T) { + monitor := &service.ChannelMonitor{ + ExtraHeaders: map[string]string{"User-Agent": "Codex"}, + DuplicateOperationID: "operation-digest", + } + + persisted := channelMonitorHeadersForPersistence(monitor) + require.Equal(t, "operation-digest", persisted[service.ChannelMonitorDuplicateOperationIDMetadataKey]) + require.Equal(t, "Codex", persisted["User-Agent"]) + require.NotContains(t, monitor.ExtraHeaders, service.ChannelMonitorDuplicateOperationIDMetadataKey) + + restored := entToServiceMonitor(&dbent.ChannelMonitor{ExtraHeaders: persisted}) + require.Equal(t, "operation-digest", restored.DuplicateOperationID) + require.Equal(t, map[string]string{"User-Agent": "Codex"}, restored.ExtraHeaders) + require.NotContains(t, restored.ExtraHeaders, service.ChannelMonitorDuplicateOperationIDMetadataKey) + require.Equal(t, "operation-digest", persisted[service.ChannelMonitorDuplicateOperationIDMetadataKey], "decoding must not mutate the ent row") +} diff --git a/backend/internal/repository/channel_monitor_repo.go b/backend/internal/repository/channel_monitor_repo.go index c7b33f2bd9..aa8b12ff8b 100644 --- a/backend/internal/repository/channel_monitor_repo.go +++ b/backend/internal/repository/channel_monitor_repo.go @@ -12,6 +12,9 @@ import ( "github.com/Wei-Shaw/sub2api/ent/channelmonitorhistory" "github.com/Wei-Shaw/sub2api/internal/service" "github.com/lib/pq" + + entsql "entgo.io/ent/dialect/sql" + "entgo.io/ent/dialect/sql/sqljson" ) // channelMonitorRepository 实现 service.ChannelMonitorRepository。 @@ -47,7 +50,7 @@ func (r *channelMonitorRepository) Create(ctx context.Context, m *service.Channe SetIntervalSeconds(m.IntervalSeconds). SetJitterSeconds(m.JitterSeconds). SetCreatedBy(m.CreatedBy). - SetExtraHeaders(emptyHeadersIfNilRepo(m.ExtraHeaders)). + SetExtraHeaders(channelMonitorHeadersForPersistence(m)). SetBodyOverrideMode(defaultBodyModeRepo(m.BodyOverrideMode)) if m.TemplateID != nil { builder = builder.SetTemplateID(*m.TemplateID) @@ -66,6 +69,30 @@ func (r *channelMonitorRepository) Create(ctx context.Context, m *service.Channe return nil } +func (r *channelMonitorRepository) FindByDuplicateOperationID(ctx context.Context, operationID string) (*service.ChannelMonitor, error) { + if strings.TrimSpace(operationID) == "" { + return nil, nil + } + client := clientFromContext(ctx, r.client) + row, err := client.ChannelMonitor.Query(). + Where(func(selector *entsql.Selector) { + selector.Where(sqljson.ValueEQ( + channelmonitor.FieldExtraHeaders, + operationID, + sqljson.Path(service.ChannelMonitorDuplicateOperationIDMetadataKey), + )) + }). + Order(dbent.Asc(channelmonitor.FieldID)). + First(ctx) + if dbent.IsNotFound(err) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("find channel monitor duplicate operation: %w", err) + } + return entToServiceMonitor(row), nil +} + func (r *channelMonitorRepository) GetByID(ctx context.Context, id int64) (*service.ChannelMonitor, error) { row, err := r.client.ChannelMonitor.Query(). Where(channelmonitor.IDEQ(id)). @@ -90,7 +117,7 @@ func (r *channelMonitorRepository) Update(ctx context.Context, m *service.Channe SetEnabled(m.Enabled). SetIntervalSeconds(m.IntervalSeconds). SetJitterSeconds(m.JitterSeconds). - SetExtraHeaders(emptyHeadersIfNilRepo(m.ExtraHeaders)). + SetExtraHeaders(channelMonitorHeadersForPersistence(m)). SetBodyOverrideMode(defaultBodyModeRepo(m.BodyOverrideMode)) if m.TemplateID != nil { updater = updater.SetTemplateID(*m.TemplateID) @@ -704,30 +731,33 @@ func entToServiceMonitor(row *dbent.ChannelMonitor) *service.ChannelMonitor { if extras == nil { extras = []string{} } - headers := row.ExtraHeaders - if headers == nil { - headers = map[string]string{} + headers := make(map[string]string, len(row.ExtraHeaders)) + for key, value := range row.ExtraHeaders { + headers[key] = value } + duplicateOperationID := headers[service.ChannelMonitorDuplicateOperationIDMetadataKey] + delete(headers, service.ChannelMonitorDuplicateOperationIDMetadataKey) out := &service.ChannelMonitor{ - ID: row.ID, - Name: row.Name, - Provider: string(row.Provider), - APIMode: defaultAPIModeRepo(row.APIMode), - Endpoint: row.Endpoint, - APIKey: row.APIKeyEncrypted, // 仍为密文，service 层负责解密 - PrimaryModel: row.PrimaryModel, - ExtraModels: extras, - GroupName: row.GroupName, - Enabled: row.Enabled, - IntervalSeconds: row.IntervalSeconds, - JitterSeconds: row.JitterSeconds, - LastCheckedAt: row.LastCheckedAt, - CreatedBy: row.CreatedBy, - CreatedAt: row.CreatedAt, - UpdatedAt: row.UpdatedAt, - ExtraHeaders: headers, - BodyOverrideMode: row.BodyOverrideMode, - BodyOverride: row.BodyOverride, + ID: row.ID, + Name: row.Name, + Provider: string(row.Provider), + APIMode: defaultAPIModeRepo(row.APIMode), + Endpoint: row.Endpoint, + APIKey: row.APIKeyEncrypted, // 仍为密文，service 层负责解密 + PrimaryModel: row.PrimaryModel, + ExtraModels: extras, + GroupName: row.GroupName, + Enabled: row.Enabled, + IntervalSeconds: row.IntervalSeconds, + JitterSeconds: row.JitterSeconds, + LastCheckedAt: row.LastCheckedAt, + CreatedBy: row.CreatedBy, + CreatedAt: row.CreatedAt, + UpdatedAt: row.UpdatedAt, + ExtraHeaders: headers, + BodyOverrideMode: row.BodyOverrideMode, + BodyOverride: row.BodyOverride, + DuplicateOperationID: duplicateOperationID, } if row.TemplateID != nil { id := *row.TemplateID @@ -736,6 +766,23 @@ func entToServiceMonitor(row *dbent.ChannelMonitor) *service.ChannelMonitor { return out } +func channelMonitorHeadersForPersistence(m *service.ChannelMonitor) map[string]string { + if m == nil { + return map[string]string{} + } + headers := make(map[string]string, len(m.ExtraHeaders)+1) + for key, value := range m.ExtraHeaders { + if key == service.ChannelMonitorDuplicateOperationIDMetadataKey { + continue + } + headers[key] = value + } + if operationID := strings.TrimSpace(m.DuplicateOperationID); operationID != "" { + headers[service.ChannelMonitorDuplicateOperationIDMetadataKey] = operationID + } + return headers +} + // emptyHeadersIfNilRepo 与 service.emptyHeadersIfNil 功能一致， // repo 独立一份避免 import 循环。 func emptyHeadersIfNilRepo(h map[string]string) map[string]string { diff --git a/backend/internal/repository/channel_monitor_template_duplicate_metadata_integration_test.go b/backend/internal/repository/channel_monitor_template_duplicate_metadata_integration_test.go new file mode 100644 index 0000000000..d6d9ce4dd8 --- /dev/null +++ b/backend/internal/repository/channel_monitor_template_duplicate_metadata_integration_test.go @@ -0,0 +1,61 @@ +//go:build integration + +package repository + +import ( + "context" + "testing" + + dbent "github.com/Wei-Shaw/sub2api/ent" + "github.com/Wei-Shaw/sub2api/ent/channelmonitor" + "github.com/Wei-Shaw/sub2api/ent/channelmonitorrequesttemplate" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +func TestApplyChannelMonitorTemplatePreservesDuplicateOperationMetadata(t *testing.T) { + tx := testEntTx(t) + ctx := dbent.NewTxContext(context.Background(), tx) + client := tx.Client() + + template, err := client.ChannelMonitorRequestTemplate.Create(). + SetName("duplicate-metadata-template"). + SetProvider(channelmonitorrequesttemplate.ProviderOpenai). + SetAPIMode(service.MonitorAPIModeResponses). + SetExtraHeaders(map[string]string{"User-Agent": "template-client"}). + SetBodyOverrideMode(service.MonitorBodyOverrideModeOff). + Save(ctx) + require.NoError(t, err) + + monitor, err := client.ChannelMonitor.Create(). + SetName("duplicate-copy"). + SetProvider(channelmonitor.ProviderOpenai). + SetAPIMode(service.MonitorAPIModeResponses). + SetEndpoint("https://api.example.com"). + SetAPIKeyEncrypted("encrypted-key"). + SetPrimaryModel("gpt-5.4-mini"). + SetIntervalSeconds(60). + SetCreatedBy(1). + SetTemplateID(template.ID). + SetExtraHeaders(map[string]string{ + "X-Original": "replaced", + service.ChannelMonitorDuplicateOperationIDMetadataKey: "operation-digest", + }). + Save(ctx) + require.NoError(t, err) + + repo := NewChannelMonitorRequestTemplateRepository(integrationEntClient, integrationDB) + affected, err := repo.ApplyToMonitors(ctx, template.ID, []int64{monitor.ID}) + require.NoError(t, err) + require.Equal(t, int64(1), affected) + + stored, err := client.ChannelMonitor.Get(ctx, monitor.ID) + require.NoError(t, err) + require.Equal(t, "template-client", stored.ExtraHeaders["User-Agent"]) + require.NotContains(t, stored.ExtraHeaders, "X-Original") + require.Equal(t, "operation-digest", stored.ExtraHeaders[service.ChannelMonitorDuplicateOperationIDMetadataKey]) + + runtimeMonitor := entToServiceMonitor(stored) + require.Equal(t, "operation-digest", runtimeMonitor.DuplicateOperationID) + require.NotContains(t, runtimeMonitor.ExtraHeaders, service.ChannelMonitorDuplicateOperationIDMetadataKey) +} diff --git a/backend/internal/repository/channel_monitor_template_duplicate_metadata_unit_test.go b/backend/internal/repository/channel_monitor_template_duplicate_metadata_unit_test.go new file mode 100644 index 0000000000..18c566fa19 --- /dev/null +++ b/backend/internal/repository/channel_monitor_template_duplicate_metadata_unit_test.go @@ -0,0 +1,108 @@ +//go:build unit + +package repository + +import ( + "context" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + dbent "github.com/Wei-Shaw/sub2api/ent" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" + + "entgo.io/ent/dialect" + entsql "entgo.io/ent/dialect/sql" +) + +func TestApplyChannelMonitorTemplatePreservesDuplicateOperationMetadataAtomically(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + client := dbent.NewClient(dbent.Driver(entsql.OpenDB(dialect.Postgres, db))) + t.Cleanup(func() { _ = client.Close() }) + + const templateID int64 = 7 + monitorIDs := []int64{41, 42} + + mock.ExpectBegin() + expectChannelMonitorTemplateForApply(mock, templateID) + mock.ExpectExec(`(?s)UPDATE "channel_monitors" SET "body_override" = NULL, "updated_at" = \$1, "api_mode" = \$2, "body_override_mode" = \$3 WHERE .*"template_id" = \$4.*"id" IN \(\$5, \$6\).*"provider" = \$7.*"api_mode" = \$8`). + WithArgs( + sqlmock.AnyArg(), + service.MonitorAPIModeResponses, + service.MonitorBodyOverrideModeOff, + templateID, + monitorIDs[0], + monitorIDs[1], + service.MonitorProviderOpenAI, + service.MonitorAPIModeResponses, + ). + WillReturnResult(sqlmock.NewResult(0, 2)) + mock.ExpectExec(`(?s)UPDATE channel_monitors\s+SET extra_headers = \$1::jsonb \|\| CASE\s+WHEN COALESCE\(extra_headers, '\{\}'::jsonb\) \? \(\$2::text\)\s+THEN jsonb_build_object\(\$2::text, COALESCE\(extra_headers, '\{\}'::jsonb\) -> \(\$2::text\)\)\s+ELSE '\{\}'::jsonb\s+END\s+WHERE template_id = \$3\s+AND id = ANY\(\$4\)\s+AND provider = \$5\s+AND api_mode = \$6`). + WithArgs( + `{"User-Agent":"template-client"}`, + service.ChannelMonitorDuplicateOperationIDMetadataKey, + templateID, + `{41,42}`, + service.MonitorProviderOpenAI, + service.MonitorAPIModeResponses, + ). + WillReturnResult(sqlmock.NewResult(0, 2)) + mock.ExpectCommit() + + repo := NewChannelMonitorRequestTemplateRepository(client, db) + affected, err := repo.ApplyToMonitors(context.Background(), templateID, monitorIDs) + + require.NoError(t, err) + require.Equal(t, int64(2), affected) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestApplyChannelMonitorTemplateRollsBackWhenHeaderRowCountDiffers(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + client := dbent.NewClient(dbent.Driver(entsql.OpenDB(dialect.Postgres, db))) + t.Cleanup(func() { _ = client.Close() }) + + const templateID int64 = 7 + + mock.ExpectBegin() + expectChannelMonitorTemplateForApply(mock, templateID) + mock.ExpectExec(`(?s)UPDATE "channel_monitors" SET .*WHERE `). + WillReturnResult(sqlmock.NewResult(0, 2)) + mock.ExpectExec(`(?s)UPDATE channel_monitors\s+SET extra_headers = \$1::jsonb \|\| CASE.*jsonb_build_object\(\$2::text,.*WHERE template_id = \$3.*AND id = ANY\(\$4\)`). + WithArgs( + `{"User-Agent":"template-client"}`, + service.ChannelMonitorDuplicateOperationIDMetadataKey, + templateID, + `{41,42}`, + service.MonitorProviderOpenAI, + service.MonitorAPIModeResponses, + ). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectRollback() + + repo := NewChannelMonitorRequestTemplateRepository(client, db) + affected, err := repo.ApplyToMonitors(context.Background(), templateID, []int64{41, 42}) + + require.Zero(t, affected) + require.EqualError(t, err, "apply template headers: affected 1 rows, expected 2") + require.NoError(t, mock.ExpectationsWereMet()) +} + +func expectChannelMonitorTemplateForApply(mock sqlmock.Sqlmock, templateID int64) { + now := time.Now() + mock.ExpectQuery(`(?s)SELECT .* FROM "channel_monitor_request_templates" WHERE "channel_monitor_request_templates"\."id" = \$1 LIMIT 2`). + WithArgs(templateID). + WillReturnRows(sqlmock.NewRows([]string{ + "id", "created_at", "updated_at", "name", "provider", "api_mode", "description", + "extra_headers", "body_override_mode", "body_override", + }).AddRow( + templateID, now, now, "monitor-template", service.MonitorProviderOpenAI, + service.MonitorAPIModeResponses, "", []byte(`{"User-Agent":"template-client"}`), + service.MonitorBodyOverrideModeOff, nil, + )) +} diff --git a/backend/internal/repository/channel_monitor_template_repo.go b/backend/internal/repository/channel_monitor_template_repo.go index 3a972360fb..e4886e1f87 100644 --- a/backend/internal/repository/channel_monitor_template_repo.go +++ b/backend/internal/repository/channel_monitor_template_repo.go @@ -3,12 +3,14 @@ package repository import ( "context" "database/sql" + "encoding/json" "fmt" dbent "github.com/Wei-Shaw/sub2api/ent" "github.com/Wei-Shaw/sub2api/ent/channelmonitor" "github.com/Wei-Shaw/sub2api/ent/channelmonitorrequesttemplate" "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/lib/pq" ) // channelMonitorRequestTemplateRepository 实现 service.ChannelMonitorRequestTemplateRepository。 @@ -110,12 +112,38 @@ func (r *channelMonitorRequestTemplateRepository) List(ctx context.Context, para // ApplyToMonitors 把模板当前配置覆盖到 monitorIDs 列表里的关联监控。 // WHERE 双重过滤：template_id = id AND id IN (monitorIDs)，防止用户传了未关联本模板的 id -// 就被覆盖。走 ent UpdateMany 保留 hooks。 +// 就被覆盖。模板字段通过 ent UpdateMany 更新以保留 hooks；extra_headers 在同一事务中 +// 单独合并，以保留仅用于幂等恢复、绝不会发往上游的内部 operation ID。 func (r *channelMonitorRequestTemplateRepository) ApplyToMonitors(ctx context.Context, id int64, monitorIDs []int64) (int64, error) { if len(monitorIDs) == 0 { return 0, nil } - client := clientFromContext(ctx, r.client) + if tx := dbent.TxFromContext(ctx); tx != nil { + return r.applyToMonitorsWithClient(ctx, tx.Client(), id, monitorIDs) + } + + tx, err := r.client.Tx(ctx) + if err != nil { + return 0, fmt.Errorf("begin apply template transaction: %w", err) + } + txCtx := dbent.NewTxContext(ctx, tx) + affected, err := r.applyToMonitorsWithClient(txCtx, tx.Client(), id, monitorIDs) + if err != nil { + _ = tx.Rollback() + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("commit apply template transaction: %w", err) + } + return affected, nil +} + +func (r *channelMonitorRequestTemplateRepository) applyToMonitorsWithClient( + ctx context.Context, + client *dbent.Client, + id int64, + monitorIDs []int64, +) (int64, error) { tpl, err := client.ChannelMonitorRequestTemplate.Query(). Where(channelmonitorrequesttemplate.IDEQ(id)). Only(ctx) @@ -131,7 +159,6 @@ func (r *channelMonitorRequestTemplateRepository) ApplyToMonitors(ctx context.Co channelmonitor.APIModeEQ(defaultAPIModeRepo(tpl.APIMode)), ). SetAPIMode(defaultAPIModeRepo(tpl.APIMode)). - SetExtraHeaders(emptyHeadersIfNilRepo(tpl.ExtraHeaders)). SetBodyOverrideMode(defaultBodyModeRepo(tpl.BodyOverrideMode)) if tpl.BodyOverride != nil { updater = updater.SetBodyOverride(tpl.BodyOverride) @@ -143,7 +170,41 @@ func (r *channelMonitorRequestTemplateRepository) ApplyToMonitors(ctx context.Co if err != nil { return 0, fmt.Errorf("apply template to monitors: %w", err) } - return int64(affected), nil + if affected == 0 { + return 0, nil + } + + templateHeaders := channelMonitorHeadersForPersistence(&service.ChannelMonitor{ + ExtraHeaders: tpl.ExtraHeaders, + }) + templateHeadersJSON, err := json.Marshal(templateHeaders) + if err != nil { + return 0, fmt.Errorf("marshal template headers: %w", err) + } + result, err := client.ExecContext(ctx, ` + UPDATE channel_monitors + SET extra_headers = $1::jsonb || CASE + WHEN COALESCE(extra_headers, '{}'::jsonb) ? ($2::text) + THEN jsonb_build_object($2::text, COALESCE(extra_headers, '{}'::jsonb) -> ($2::text)) + ELSE '{}'::jsonb + END + WHERE template_id = $3 + AND id = ANY($4) + AND provider = $5 + AND api_mode = $6 + `, string(templateHeadersJSON), service.ChannelMonitorDuplicateOperationIDMetadataKey, + id, pq.Array(monitorIDs), string(tpl.Provider), defaultAPIModeRepo(tpl.APIMode)) + if err != nil { + return 0, fmt.Errorf("apply template headers to monitors: %w", err) + } + headersAffected, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("count applied template headers: %w", err) + } + if headersAffected != int64(affected) { + return 0, fmt.Errorf("apply template headers: affected %d rows, expected %d", headersAffected, affected) + } + return headersAffected, nil } // CountAssociatedMonitors 统计关联监控数（UI 展示「N 个配置」用）。 diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index 94867bf592..844f75eae0 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -682,6 +682,7 @@ func registerChannelMonitorRoutes(admin *gin.RouterGroup, h *handler.Handlers) { monitors.GET("", h.Admin.ChannelMonitor.List) monitors.POST("", h.Admin.ChannelMonitor.Create) monitors.GET("/:id", h.Admin.ChannelMonitor.Get) + monitors.POST("/:id/duplicate", h.Admin.ChannelMonitor.Duplicate) monitors.PUT("/:id", h.Admin.ChannelMonitor.Update) monitors.DELETE("/:id", h.Admin.ChannelMonitor.Delete) monitors.POST("/:id/run", h.Admin.ChannelMonitor.Run) diff --git a/backend/internal/service/channel_monitor_duplicate_test.go b/backend/internal/service/channel_monitor_duplicate_test.go new file mode 100644 index 0000000000..e61bb02452 --- /dev/null +++ b/backend/internal/service/channel_monitor_duplicate_test.go @@ -0,0 +1,239 @@ +//go:build unit + +package service + +import ( + "context" + "errors" + "strings" + "testing" + "time" + "unicode/utf8" + + "github.com/stretchr/testify/require" +) + +type duplicateChannelMonitorRepoStub struct { + ChannelMonitorRepository + source *ChannelMonitor + created []*ChannelMonitor + byOperation map[string]*ChannelMonitor + nextID int64 +} + +func (r *duplicateChannelMonitorRepoStub) GetByID(_ context.Context, id int64) (*ChannelMonitor, error) { + if r.source == nil || r.source.ID != id { + return nil, ErrChannelMonitorNotFound + } + return r.source, nil +} + +func (r *duplicateChannelMonitorRepoStub) Create(_ context.Context, monitor *ChannelMonitor) error { + r.nextID++ + monitor.ID = 100 + r.nextID + monitor.CreatedAt = time.Date(2026, time.July, 16, 8, 0, 0, 0, time.UTC) + monitor.UpdatedAt = monitor.CreatedAt + + stored := *monitor + stored.ExtraModels = append([]string(nil), monitor.ExtraModels...) + stored.ExtraHeaders = cloneStringMap(monitor.ExtraHeaders) + stored.BodyOverride = mustCloneJSONMap(monitor.BodyOverride) + if monitor.TemplateID != nil { + templateID := *monitor.TemplateID + stored.TemplateID = &templateID + } + r.created = append(r.created, &stored) + if stored.DuplicateOperationID != "" { + if r.byOperation == nil { + r.byOperation = make(map[string]*ChannelMonitor) + } + r.byOperation[stored.DuplicateOperationID] = &stored + } + return nil +} + +func (r *duplicateChannelMonitorRepoStub) FindByDuplicateOperationID(_ context.Context, operationID string) (*ChannelMonitor, error) { + monitor := r.byOperation[operationID] + if monitor == nil { + return nil, nil + } + cloned := *monitor + cloned.ExtraModels = append([]string(nil), monitor.ExtraModels...) + cloned.ExtraHeaders = cloneStringMap(monitor.ExtraHeaders) + cloned.BodyOverride = mustCloneJSONMap(monitor.BodyOverride) + return &cloned, nil +} + +func cloneStringMap(source map[string]string) map[string]string { + if source == nil { + return nil + } + cloned := make(map[string]string, len(source)) + for key, value := range source { + cloned[key] = value + } + return cloned +} + +func mustCloneJSONMap(source map[string]any) map[string]any { + if source == nil { + return nil + } + cloned, err := cloneChannelMonitorJSONMap(source) + if err != nil { + panic(err) + } + return cloned +} + +type duplicateChannelMonitorEncryptor struct { + decryptErr error + encryptErr error +} + +func (e *duplicateChannelMonitorEncryptor) Encrypt(plaintext string) (string, error) { + if e.encryptErr != nil { + return "", e.encryptErr + } + return "NEW:" + plaintext, nil +} + +func (e *duplicateChannelMonitorEncryptor) Decrypt(ciphertext string) (string, error) { + if e.decryptErr != nil { + return "", e.decryptErr + } + if !strings.HasPrefix(ciphertext, "OLD:") && !strings.HasPrefix(ciphertext, "NEW:") { + return "", errors.New("invalid ciphertext") + } + return strings.TrimPrefix(strings.TrimPrefix(ciphertext, "OLD:"), "NEW:"), nil +} + +func TestDuplicateChannelMonitorCopiesConfigurationAndResetsRuntimeState(t *testing.T) { + lastCheckedAt := time.Date(2026, time.July, 15, 7, 0, 0, 0, time.UTC) + templateID := int64(9) + source := &ChannelMonitor{ + ID: 42, + Name: "primary", + Provider: MonitorProviderOpenAI, + APIMode: MonitorAPIModeResponses, + Endpoint: "https://api.example.com", + APIKey: "OLD:top-secret", + PrimaryModel: "gpt-5.4-mini", + ExtraModels: []string{"gpt-5.4", "gpt-5.3"}, + GroupName: "production", + Enabled: true, + IntervalSeconds: 90, + JitterSeconds: 15, + LastCheckedAt: &lastCheckedAt, + CreatedBy: 4, + CreatedAt: lastCheckedAt.Add(-time.Hour), + UpdatedAt: lastCheckedAt, + TemplateID: &templateID, + ExtraHeaders: map[string]string{"User-Agent": "Codex"}, + BodyOverrideMode: MonitorBodyOverrideModeMerge, + BodyOverride: map[string]any{ + "metadata": map[string]any{"source": "original"}, + }, + } + repo := &duplicateChannelMonitorRepoStub{source: source} + service := NewChannelMonitorService(repo, &duplicateChannelMonitorEncryptor{}) + + duplicate, err := service.Duplicate(context.Background(), source.ID, 77, "admin:77", "copy-primary") + + require.NoError(t, err) + require.Len(t, repo.created, 1) + stored := repo.created[0] + require.NotEqual(t, source.ID, duplicate.ID) + require.Equal(t, "primary (Copy)", duplicate.Name) + require.Equal(t, source.Provider, duplicate.Provider) + require.Equal(t, source.APIMode, duplicate.APIMode) + require.Equal(t, source.Endpoint, duplicate.Endpoint) + require.Equal(t, "top-secret", duplicate.APIKey) + require.Equal(t, "NEW:top-secret", stored.APIKey) + require.Equal(t, source.PrimaryModel, duplicate.PrimaryModel) + require.Equal(t, source.ExtraModels, duplicate.ExtraModels) + require.Equal(t, source.GroupName, duplicate.GroupName) + require.Equal(t, source.IntervalSeconds, duplicate.IntervalSeconds) + require.Equal(t, source.JitterSeconds, duplicate.JitterSeconds) + require.Equal(t, source.TemplateID, duplicate.TemplateID) + require.Equal(t, source.ExtraHeaders, duplicate.ExtraHeaders) + require.Equal(t, source.BodyOverrideMode, duplicate.BodyOverrideMode) + require.Equal(t, source.BodyOverride, duplicate.BodyOverride) + require.False(t, duplicate.Enabled) + require.Nil(t, duplicate.LastCheckedAt) + require.Equal(t, int64(77), duplicate.CreatedBy) + require.False(t, duplicate.APIKeyDecryptFailed) + require.NotEmpty(t, duplicate.DuplicateOperationID) + + duplicate.ExtraModels[0] = "changed" + duplicate.ExtraHeaders["User-Agent"] = "changed" + duplicate.BodyOverride["metadata"].(map[string]any)["source"] = "changed" + *duplicate.TemplateID = 10 + require.Equal(t, []string{"gpt-5.4", "gpt-5.3"}, source.ExtraModels) + require.Equal(t, "Codex", source.ExtraHeaders["User-Agent"]) + require.Equal(t, "original", source.BodyOverride["metadata"].(map[string]any)["source"]) + require.Equal(t, int64(9), *source.TemplateID) + require.Equal(t, "OLD:top-secret", source.APIKey) + require.True(t, source.Enabled) + require.Equal(t, &lastCheckedAt, source.LastCheckedAt) +} + +func TestDuplicateChannelMonitorNamePreservesSuffixWithinSchemaLimit(t *testing.T) { + name := duplicateChannelMonitorName(strings.Repeat("界", 100)) + + require.Equal(t, 100, utf8.RuneCountInString(name)) + require.True(t, strings.HasSuffix(name, " (Copy)")) +} + +func TestDuplicateChannelMonitorRejectsUndecryptableAPIKey(t *testing.T) { + source := &ChannelMonitor{ID: 42, Name: "broken", APIKey: "OLD:broken"} + repo := &duplicateChannelMonitorRepoStub{source: source} + service := NewChannelMonitorService(repo, &duplicateChannelMonitorEncryptor{decryptErr: errors.New("wrong encryption key")}) + + duplicate, err := service.Duplicate(context.Background(), source.ID, 77, "admin:77", "copy-broken") + + require.Nil(t, duplicate) + require.ErrorIs(t, err, ErrChannelMonitorAPIKeyDecryptFailed) + require.Empty(t, repo.created) + require.Equal(t, "OLD:broken", source.APIKey) +} + +func TestDuplicateChannelMonitorRecoversCommittedCopyForSameOperation(t *testing.T) { + source := &ChannelMonitor{ + ID: 42, + Name: "primary", + Provider: MonitorProviderOpenAI, + APIMode: MonitorAPIModeResponses, + Endpoint: "https://api.example.com", + APIKey: "OLD:top-secret", + PrimaryModel: "gpt-5.4-mini", + IntervalSeconds: 60, + BodyOverrideMode: MonitorBodyOverrideModeOff, + } + repo := &duplicateChannelMonitorRepoStub{source: source} + service := NewChannelMonitorService(repo, &duplicateChannelMonitorEncryptor{}) + + first, err := service.Duplicate(context.Background(), source.ID, 77, "admin:77", "stable-key") + require.NoError(t, err) + retry, err := service.Duplicate(context.Background(), source.ID, 77, "admin:77", "stable-key") + require.NoError(t, err) + + require.Len(t, repo.created, 1, "same operation must not create a second monitor") + require.Equal(t, first.ID, retry.ID) + require.Equal(t, "top-secret", retry.APIKey) + require.Equal(t, first.DuplicateOperationID, retry.DuplicateOperationID) + require.NotContains(t, retry.ExtraHeaders, ChannelMonitorDuplicateOperationIDMetadataKey) + + otherActor, err := service.Duplicate(context.Background(), source.ID, 88, "admin:88", "stable-key") + require.NoError(t, err) + require.NotEqual(t, first.ID, otherActor.ID) + require.Len(t, repo.created, 2, "operation identity must include the actor scope") +} + +func TestChannelMonitorDuplicateOperationMetadataKeyCannotBeSubmittedAsHeader(t *testing.T) { + err := validateExtraHeaders(map[string]string{ + ChannelMonitorDuplicateOperationIDMetadataKey: "forged-operation", + }) + + require.Error(t, err) +} diff --git a/backend/internal/service/channel_monitor_service.go b/backend/internal/service/channel_monitor_service.go index b5dea22589..9cc793b43b 100644 --- a/backend/internal/service/channel_monitor_service.go +++ b/backend/internal/service/channel_monitor_service.go @@ -2,8 +2,11 @@ package service import ( "context" + "crypto/sha256" + "encoding/json" "fmt" "log/slog" + "strconv" "strings" "sync" "time" @@ -21,6 +24,7 @@ type ChannelMonitorRepository interface { Update(ctx context.Context, m *ChannelMonitor) error Delete(ctx context.Context, id int64) error List(ctx context.Context, params ChannelMonitorListParams) ([]*ChannelMonitor, int64, error) + FindByDuplicateOperationID(ctx context.Context, operationID string) (*ChannelMonitor, error) // 调度器辅助 ListEnabled(ctx context.Context) ([]*ChannelMonitor, error) @@ -66,6 +70,14 @@ type ChannelMonitorService struct { scheduler MonitorScheduler } +const maxChannelMonitorNameRunes = 100 + +// ChannelMonitorDuplicateOperationIDMetadataKey is stored in the existing +// extra_headers JSON column to avoid a schema migration. The colon makes it an +// invalid HTTP header name, and repository adapters remove it before exposing +// ExtraHeaders to the service layer. +const ChannelMonitorDuplicateOperationIDMetadataKey = "sub2api:duplicate_operation_id" + // NewChannelMonitorService 创建渠道监控服务实例。 func NewChannelMonitorService(repo ChannelMonitorRepository, encryptor SecretEncryptor) *ChannelMonitorService { return &ChannelMonitorService{repo: repo, encryptor: encryptor} @@ -147,6 +159,164 @@ func (s *ChannelMonitorService) Create(ctx context.Context, p ChannelMonitorCrea return m, nil } +// Duplicate creates an independent, disabled copy of an existing monitor. +// The API key stays server-side: it is decrypted only long enough to encrypt a +// fresh ciphertext for the new row. Runtime state and history are not copied. +func (s *ChannelMonitorService) Duplicate( + ctx context.Context, + id, createdBy int64, + actorScope, operationKey string, +) (*ChannelMonitor, error) { + operationID := duplicateChannelMonitorOperationID(id, actorScope, operationKey) + existing, err := s.RecoverDuplicate(ctx, id, actorScope, operationKey) + if err != nil { + return nil, err + } + if existing != nil { + return existing, nil + } + + source, err := s.repo.GetByID(ctx, id) + if err != nil { + return nil, err + } + + plainAPIKey, err := s.decryptAPIKeyForDuplicate(source) + if err != nil { + return nil, err + } + encryptedAPIKey, err := s.encryptor.Encrypt(plainAPIKey) + if err != nil { + return nil, fmt.Errorf("encrypt duplicate channel monitor api key: %w", err) + } + bodyOverride, err := cloneChannelMonitorJSONMap(source.BodyOverride) + if err != nil { + return nil, fmt.Errorf("clone duplicate channel monitor body override: %w", err) + } + + duplicate := &ChannelMonitor{ + Name: duplicateChannelMonitorName(source.Name), + Provider: source.Provider, + APIMode: source.APIMode, + Endpoint: source.Endpoint, + APIKey: encryptedAPIKey, + PrimaryModel: source.PrimaryModel, + ExtraModels: append([]string{}, source.ExtraModels...), + GroupName: source.GroupName, + Enabled: false, + IntervalSeconds: source.IntervalSeconds, + JitterSeconds: source.JitterSeconds, + CreatedBy: createdBy, + TemplateID: cloneInt64Pointer(source.TemplateID), + ExtraHeaders: cloneChannelMonitorHeaders(source.ExtraHeaders), + BodyOverrideMode: source.BodyOverrideMode, + BodyOverride: bodyOverride, + DuplicateOperationID: operationID, + } + if err := s.repo.Create(ctx, duplicate); err != nil { + return nil, fmt.Errorf("duplicate channel monitor: %w", err) + } + + // Match Create/Update response semantics: repository receives ciphertext, + // while handlers receive plaintext only so they can return the masked form. + duplicate.APIKey = plainAPIKey + return duplicate, nil +} + +// RecoverDuplicate performs a read-only lookup for a duplicate that was +// already committed for the same actor, source monitor, and idempotency key. +// It deliberately never repeats the create side effect. +func (s *ChannelMonitorService) RecoverDuplicate( + ctx context.Context, + id int64, + actorScope, operationKey string, +) (*ChannelMonitor, error) { + operationID := duplicateChannelMonitorOperationID(id, actorScope, operationKey) + if operationID == "" { + return nil, nil + } + monitor, err := s.repo.FindByDuplicateOperationID(ctx, operationID) + if err != nil { + return nil, fmt.Errorf("find duplicate channel monitor operation: %w", err) + } + if monitor == nil { + return nil, nil + } + s.decryptInPlace(monitor) + return monitor, nil +} + +func duplicateChannelMonitorOperationID(sourceID int64, actorScope, operationKey string) string { + operationKey = strings.TrimSpace(operationKey) + if operationKey == "" { + return "" + } + actorScope = strings.TrimSpace(actorScope) + if actorScope == "" { + actorScope = "admin:0" + } + payload := "admin.channel_monitors.duplicate\x00" + actorScope + "\x00" + strconv.FormatInt(sourceID, 10) + "\x00" + operationKey + digest := sha256.Sum256([]byte(payload)) + return fmt.Sprintf("%x", digest) +} + +func (s *ChannelMonitorService) decryptAPIKeyForDuplicate(source *ChannelMonitor) (string, error) { + if source == nil || strings.TrimSpace(source.APIKey) == "" { + return "", ErrChannelMonitorAPIKeyDecryptFailed + } + plain, err := s.encryptor.Decrypt(source.APIKey) + if err != nil || strings.TrimSpace(plain) == "" { + slog.Warn("channel_monitor: decrypt api key for duplicate failed", + "monitor_id", source.ID, "error", err) + return "", ErrChannelMonitorAPIKeyDecryptFailed + } + return plain, nil +} + +func duplicateChannelMonitorName(sourceName string) string { + const suffix = " (Copy)" + nameRunes := []rune(strings.TrimSpace(sourceName)) + maxBaseRunes := maxChannelMonitorNameRunes - len([]rune(suffix)) + if len(nameRunes) > maxBaseRunes { + nameRunes = nameRunes[:maxBaseRunes] + } + return string(nameRunes) + suffix +} + +func cloneInt64Pointer(value *int64) *int64 { + if value == nil { + return nil + } + cloned := *value + return &cloned +} + +func cloneChannelMonitorHeaders(source map[string]string) map[string]string { + if source == nil { + return map[string]string{} + } + cloned := make(map[string]string, len(source)) + for key, value := range source { + cloned[key] = value + } + return cloned +} + +func cloneChannelMonitorJSONMap(source map[string]any) (map[string]any, error) { + if source == nil { + return nil, nil + } + payload, err := json.Marshal(source) + if err != nil { + return nil, err + } + cloned := make(map[string]any, len(source)) + if err := json.Unmarshal(payload, &cloned); err != nil { + return nil, err + } + return cloned, nil +} + // validateCreateParams 把 Create 入参的所有校验聚拢为一个函数，避免 Create 主体超过 30 行。 func validateCreateParams(p ChannelMonitorCreateParams) error { if err := validateProvider(p.Provider); err != nil { diff --git a/backend/internal/service/channel_monitor_types.go b/backend/internal/service/channel_monitor_types.go index 0f68bc29e4..20ab1da935 100644 --- a/backend/internal/service/channel_monitor_types.go +++ b/backend/internal/service/channel_monitor_types.go @@ -51,6 +51,12 @@ type ChannelMonitor struct { BodyOverrideMode string // off / merge / replace BodyOverride map[string]any // 仅 mode != off 时使用 + // DuplicateOperationID is internal persistence metadata used to recover an + // already committed duplicate after an ambiguous idempotency-store failure. + // Repository implementations must keep it out of ExtraHeaders so it can + // never be serialized to clients or forwarded to an upstream provider. + DuplicateOperationID string + // APIKeyDecryptFailed 表示 APIKey 字段无法解密（密钥不一致或损坏）。 // 此时 APIKey 为空字符串，runner / RunCheck 必须跳过该监控并提示重填。 APIKeyDecryptFailed bool diff --git a/frontend/src/api/__tests__/admin.channelMonitor.duplicate.spec.ts b/frontend/src/api/__tests__/admin.channelMonitor.duplicate.spec.ts new file mode 100644 index 0000000000..ac5d2b05a8 --- /dev/null +++ b/frontend/src/api/__tests__/admin.channelMonitor.duplicate.spec.ts @@ -0,0 +1,103 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { post } = vi.hoisted(() => ({ + post: vi.fn(), +})) + +vi.mock('@/api/client', () => ({ + apiClient: { post }, +})) + +import { duplicate } from '@/api/admin/channelMonitor' + +describe('admin channel monitor duplicate API', () => { + beforeEach(() => { + localStorage.clear() + sessionStorage.clear() + localStorage.setItem('auth_user', JSON.stringify({ id: 7 })) + post.mockReset() + post.mockResolvedValue({ data: { id: 43, name: 'primary (Copy)' } }) + vi.spyOn(globalThis.crypto, 'randomUUID').mockReturnValue('11111111-1111-4111-8111-111111111111') + }) + + afterEach(() => { + vi.restoreAllMocks() + }) + + it('sends a stable idempotency key with the duplicate request', async () => { + const monitor = await duplicate(42) + + expect(post).toHaveBeenCalledWith('/admin/channel-monitors/42/duplicate', undefined, { + headers: { + 'Idempotency-Key': 'channel-monitor-duplicate-7-42-11111111-1111-4111-8111-111111111111', + }, + }) + expect(monitor).toEqual({ id: 43, name: 'primary (Copy)' }) + expect(sessionStorage.length).toBe(0) + }) + + it('reuses the operation key after an ambiguous failed request', async () => { + post.mockRejectedValueOnce(new Error('network timeout')) + await expect(duplicate(99)).rejects.toThrow('network timeout') + + post.mockResolvedValueOnce({ data: { id: 100, name: 'retry (Copy)' } }) + await duplicate(99) + + expect(post).toHaveBeenCalledTimes(2) + expect(post.mock.calls[1][2].headers).toEqual(post.mock.calls[0][2].headers) + expect(sessionStorage.length).toBe(0) + }) + + it('reuses the operation key after a page reload', async () => { + post.mockRejectedValueOnce(new Error('network timeout')) + await expect(duplicate(77)).rejects.toThrow('network timeout') + const firstHeaders = post.mock.calls[0][2].headers + + vi.resetModules() + post.mockResolvedValueOnce({ data: { id: 78, name: 'reload (Copy)' } }) + const { duplicate: duplicateAfterReload } = await import('@/api/admin/channelMonitor') + await duplicateAfterReload(77) + + expect(post).toHaveBeenCalledTimes(2) + expect(post.mock.calls[1][2].headers).toEqual(firstHeaders) + expect(sessionStorage.length).toBe(0) + }) + + it('does not reuse an operation key across administrators for the same monitor', async () => { + post.mockRejectedValueOnce(new Error('first admin timeout')) + await expect(duplicate(55)).rejects.toThrow('first admin timeout') + const firstAdminHeaders = post.mock.calls[0][2].headers + + localStorage.setItem('auth_user', JSON.stringify({ id: 8 })) + vi.mocked(globalThis.crypto.randomUUID).mockReturnValueOnce( + '22222222-2222-4222-8222-222222222222' + ) + post.mockResolvedValueOnce({ data: { id: 56, name: 'second admin copy' } }) + await duplicate(55) + + expect(post.mock.calls[1][2].headers).not.toEqual(firstAdminHeaders) + expect(post.mock.calls[1][2].headers).toEqual({ + 'Idempotency-Key': 'channel-monitor-duplicate-8-55-22222222-2222-4222-8222-222222222222', + }) + expect(sessionStorage.getItem('sub2api:admin:channel-monitor-duplicate:7:55')).toBe( + firstAdminHeaders['Idempotency-Key'] + ) + expect(sessionStorage.getItem('sub2api:admin:channel-monitor-duplicate:8:55')).toBeNull() + }) + + it('does not persist or reuse keys when the current user cannot be parsed', async () => { + localStorage.setItem('auth_user', '{invalid json') + post.mockRejectedValueOnce(new Error('network timeout')) + await expect(duplicate(66)).rejects.toThrow('network timeout') + const firstHeaders = post.mock.calls[0][2].headers + + vi.mocked(globalThis.crypto.randomUUID).mockReturnValueOnce( + '33333333-3333-4333-8333-333333333333' + ) + post.mockResolvedValueOnce({ data: { id: 67, name: 'fallback copy' } }) + await duplicate(66) + + expect(post.mock.calls[1][2].headers).not.toEqual(firstHeaders) + expect(sessionStorage.length).toBe(0) + }) +}) diff --git a/frontend/src/api/admin/channelMonitor.ts b/frontend/src/api/admin/channelMonitor.ts index de605351e3..5eb78b798f 100644 --- a/frontend/src/api/admin/channelMonitor.ts +++ b/frontend/src/api/admin/channelMonitor.ts @@ -156,6 +156,88 @@ export async function create(params: CreateParams): Promise { return data } +/** + * Duplicate a monitor without exposing its stored API key to the browser. + * Keep the operation key after ambiguous failures so a retry replays the + * original server-side operation instead of creating another monitor. + */ +const duplicateOperationKeys = new Map() + +interface DuplicateOperationScope { + adminID: string + key: string +} + +function getCurrentAdminID(): string | null { + try { + const rawUser = globalThis.localStorage?.getItem('auth_user') + if (!rawUser) return null + + const user: unknown = JSON.parse(rawUser) + if (typeof user !== 'object' || user === null) return null + + const id = (user as { id?: unknown }).id + if (typeof id !== 'number' || !Number.isSafeInteger(id) || id <= 0) return null + return String(id) + } catch { + return null + } +} + +function duplicateOperationScope(id: number): DuplicateOperationScope | null { + const adminID = getCurrentAdminID() + if (!adminID) return null + + return { + adminID, + key: `sub2api:admin:channel-monitor-duplicate:${adminID}:${id}`, + } +} + +function getStoredDuplicateOperationKey(storageKey: string): string | null { + try { + return globalThis.sessionStorage?.getItem(storageKey) ?? null + } catch { + return null + } +} + +function storeDuplicateOperationKey(storageKey: string, key: string | null): void { + try { + if (key) globalThis.sessionStorage?.setItem(storageKey, key) + else globalThis.sessionStorage?.removeItem(storageKey) + } catch { + // In-memory retry protection still works when browser storage is unavailable. + } +} + +export async function duplicate(id: number): Promise { + const scope = duplicateOperationScope(id) + let idempotencyKey = scope + ? duplicateOperationKeys.get(scope.key) ?? getStoredDuplicateOperationKey(scope.key) + : null + if (!idempotencyKey) { + const requestID = globalThis.crypto?.randomUUID?.() ?? `${Date.now()}-${Math.random().toString(36).slice(2)}` + idempotencyKey = `channel-monitor-duplicate-${scope?.adminID ?? 'unknown-admin'}-${id}-${requestID}` + } + if (scope) { + duplicateOperationKeys.set(scope.key, idempotencyKey) + storeDuplicateOperationKey(scope.key, idempotencyKey) + } + + const { data } = await apiClient.post( + `/admin/channel-monitors/${id}/duplicate`, + undefined, + { headers: { 'Idempotency-Key': idempotencyKey } } + ) + + if (scope) { + duplicateOperationKeys.delete(scope.key) + storeDuplicateOperationKey(scope.key, null) + } + return data +} + /** * Update an existing channel monitor. * api_key field: empty string means "do not modify". @@ -199,6 +281,7 @@ export const channelMonitorAPI = { list, get, create, + duplicate, update, del, runNow, diff --git a/frontend/src/components/admin/monitor/MonitorActionsCell.spec.ts b/frontend/src/components/admin/monitor/MonitorActionsCell.spec.ts new file mode 100644 index 0000000000..31f9eca738 --- /dev/null +++ b/frontend/src/components/admin/monitor/MonitorActionsCell.spec.ts @@ -0,0 +1,77 @@ +import { mount } from '@vue/test-utils' +import { describe, expect, it, vi } from 'vitest' + +import type { ChannelMonitor } from '@/api/admin/channelMonitor' +import MonitorActionsCell from '@/components/admin/monitor/MonitorActionsCell.vue' + +vi.mock('vue-i18n', () => ({ + useI18n: () => ({ t: (key: string) => key }), +})) + +function makeMonitor(overrides: Partial = {}): ChannelMonitor { + return { + id: 42, + name: 'primary', + provider: 'openai', + api_mode: 'chat_completions', + endpoint: 'https://api.example.com', + api_key_masked: 'sk-t***', + primary_model: 'gpt-4o-mini', + extra_models: [], + group_name: '', + enabled: true, + interval_seconds: 60, + jitter_seconds: 0, + last_checked_at: null, + created_by: 1, + created_at: '2026-07-16T00:00:00Z', + updated_at: '2026-07-16T00:00:00Z', + primary_status: '', + primary_latency_ms: null, + availability_7d: 0, + extra_models_status: [], + template_id: null, + extra_headers: {}, + body_override_mode: 'off', + body_override: null, + ...overrides, + } +} + +describe('MonitorActionsCell duplicate action', () => { + it('emits the selected monitor when duplicate is clicked', async () => { + const row = makeMonitor() + const wrapper = mount(MonitorActionsCell, { + props: { row, running: false, duplicating: false }, + }) + + await wrapper.get('[data-testid="monitor-duplicate"]').trigger('click') + + expect(wrapper.emitted('duplicate')).toEqual([[row]]) + }) + + it('disables the action while the same monitor is being duplicated', () => { + const wrapper = mount(MonitorActionsCell, { + props: { row: makeMonitor(), running: false, duplicating: true }, + }) + const button = wrapper.get('[data-testid="monitor-duplicate"]') + + expect(button.attributes('disabled')).toBeDefined() + expect(button.attributes('title')).toBe('admin.channelMonitor.duplicating') + expect(button.text()).toContain('admin.channelMonitor.duplicating') + }) + + it('disables the action when the stored API key cannot be decrypted', () => { + const wrapper = mount(MonitorActionsCell, { + props: { + row: makeMonitor({ api_key_decrypt_failed: true }), + running: false, + duplicating: false, + }, + }) + const button = wrapper.get('[data-testid="monitor-duplicate"]') + + expect(button.attributes('disabled')).toBeDefined() + expect(button.attributes('title')).toBe('admin.channelMonitor.duplicateKeyUnavailable') + }) +}) diff --git a/frontend/src/components/admin/monitor/MonitorActionsCell.vue b/frontend/src/components/admin/monitor/MonitorActionsCell.vue index 74aa40172e..324af8ed3c 100644 --- a/frontend/src/components/admin/monitor/MonitorActionsCell.vue +++ b/frontend/src/components/admin/monitor/MonitorActionsCell.vue @@ -8,6 +8,18 @@ {{ t('admin.channelMonitor.runNow') }} +