mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 12:02:27 +08:00
Merge pull request #4427 from yan9651688/feat/channel-monitor-one-click-copy
feat(channel-monitor): add safe one-click duplication
This commit is contained in:
@@ -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")
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
+61
@@ -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)
|
||||
}
|
||||
@@ -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,
|
||||
))
|
||||
}
|
||||
@@ -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 个配置」用)。
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user