diff --git a/supply-api/internal/app/runtime.go b/supply-api/internal/app/runtime.go index 9a8ec322..bb8ed1af 100644 --- a/supply-api/internal/app/runtime.go +++ b/supply-api/internal/app/runtime.go @@ -81,8 +81,8 @@ type runtimeStartupViews struct { } type runtimeFactory struct { - newDB func(ctx context.Context, cfg config.DatabaseConfig) (*repository.DB, error) - newRedisCache func(cfg config.RedisConfig) (*cache.RedisCache, error) + newDB func(ctx context.Context, cfg config.DatabaseConfig) (*repository.DB, error) + newRedisCache func(cfg config.RedisConfig) (*cache.RedisCache, error) newSMSVerifier func(cfg config.SMSConfig) (domain.SMSVerifier, error) } @@ -359,7 +359,7 @@ func buildSecurityBundle( Issuer: cfg.Token.Issuer, CacheTTL: cfg.Token.RevocationCacheTTL, Enabled: env != "dev", - BruteForceMaxAttempts: 5, // MED-12: 暴力破解保护，默认5次失败后锁定 + BruteForceMaxAttempts: 5, // MED-12: 暴力破解保护，默认5次失败后锁定 BruteForceLockoutDuration: 15 * time.Minute, // MED-12: 默认锁定15分钟 }, tokenCache, tokenBackend, adapter.NewAuditEmitterAdapter(auditStore)), revocationSubscriber: revocationSubscriber, @@ -478,7 +478,6 @@ func defaultRuntimeTuning() runtimeTuning { partitionedTables: []string{ "audit_events", "supply_usage_records", - "supply_idempotency_records", }, } } diff --git a/supply-api/internal/app/runtime_test.go b/supply-api/internal/app/runtime_test.go index d73b5472..f3d10ac1 100644 --- a/supply-api/internal/app/runtime_test.go +++ b/supply-api/internal/app/runtime_test.go @@ -662,6 +662,16 @@ func TestBuildRuntimeStartupViews_GroupsHTTPAndBackgroundDependencies(t *testing } } +func TestDefaultRuntimeTuning_ExcludesIdempotencyFromPartitionedTables(t *testing.T) { + tuning := defaultRuntimeTuning() + + for _, tableName := range tuning.partitionedTables { + if tableName == "supply_idempotency_records" { + t.Fatal("expected idempotency table to be excluded from partition maintenance") + } + } +} + func TestBuildRuntime_GroupsResourcesAndStartupViews(t *testing.T) { runtime, err := buildRuntimeWithFactory(RuntimeOptions{ Env: "dev", @@ -1182,13 +1192,13 @@ func testRuntimeConfig() *config.Config { ConnMaxLifetime: time.Minute, ConnMaxIdleTime: time.Minute, }, - Redis: config.RedisConfig{ - Host: "127.0.0.1", - Port: 6379, - Password: "", - DB: 0, - PoolSize: 2, - }, + Redis: config.RedisConfig{ + Host: "127.0.0.1", + Port: 6379, + Password: "", + DB: 0, + PoolSize: 2, + }, Token: config.TokenConfig{ SecretKey: "runtime-test-secret", Algorithm: "HS256", diff --git a/supply-api/internal/middleware/idempotency.go b/supply-api/internal/middleware/idempotency.go index 69c59cd4..45539a5f 100644 --- a/supply-api/internal/middleware/idempotency.go +++ b/supply-api/internal/middleware/idempotency.go @@ -22,14 +22,21 @@ type IdempotencyConfig struct { Enabled bool // 是否启用幂等 } +type idempotencyRepository interface { + GetByKey(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey string) (*repository.IdempotencyRecord, error) + UpdateSuccess(ctx context.Context, id int64, responseCode int, responseBody json.RawMessage) error + UpdateFailed(ctx context.Context, id int64, responseCode int, responseBody json.RawMessage) error + AcquireLock(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey, requestID, payloadHash string, ttl time.Duration) (*repository.IdempotencyRecord, error) +} + // IdempotencyMiddleware 幂等中间件 type IdempotencyMiddleware struct { - idempotencyRepo *repository.IdempotencyRepository + idempotencyRepo idempotencyRepository config IdempotencyConfig } // NewIdempotencyMiddleware 创建幂等中间件 -func NewIdempotencyMiddleware(repo *repository.IdempotencyRepository, config IdempotencyConfig) *IdempotencyMiddleware { +func NewIdempotencyMiddleware(repo idempotencyRepository, config IdempotencyConfig) *IdempotencyMiddleware { if config.TTL == 0 { config.TTL = 24 * time.Hour } @@ -167,18 +174,21 @@ func (m *IdempotencyMiddleware) Wrap(handler IdempotentHandler) http.HandlerFunc // 使用AcquireLock获取锁 requestID := r.Header.Get("X-Request-Id") - lockedRecord, err := m.idempotencyRepo.AcquireLock(ctx, idempKey.TenantID, idempKey.OperatorID, idempKey.APIPath, idempKey.Key, m.config.TTL) + lockedRecord, err := m.idempotencyRepo.AcquireLock( + ctx, + idempKey.TenantID, + idempKey.OperatorID, + idempKey.APIPath, + idempKey.Key, + requestID, + payloadHash, + m.config.TTL, + ) if err != nil { writeIdempotencyError(w, http.StatusInternalServerError, "IDEMPOTENCY_LOCK_FAILED", err.Error()) return } - // 更新记录中的request_id和payload_hash - if lockedRecord.ID != 0 && (lockedRecord.RequestID == "" || lockedRecord.PayloadHash == "") { - lockedRecord.RequestID = requestID - lockedRecord.PayloadHash = payloadHash - } - // 创建包装器以捕获实际的状态码和响应体 wrappedWriter := &statusCapturingResponseWriter{ResponseWriter: w} diff --git a/supply-api/internal/middleware/idempotency_test.go b/supply-api/internal/middleware/idempotency_test.go index 86e091db..9ebed68d 100644 --- a/supply-api/internal/middleware/idempotency_test.go +++ b/supply-api/internal/middleware/idempotency_test.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -17,11 +18,13 @@ import ( // MockIdempotencyRepository 模拟幂等仓储 type MockIdempotencyRepository struct { records map[string]*repository.IdempotencyRecord + nextID int64 } func NewMockIdempotencyRepository() *MockIdempotencyRepository { return &MockIdempotencyRepository{ records: make(map[string]*repository.IdempotencyRecord), + nextID: 1, } } @@ -37,41 +40,62 @@ func (r *MockIdempotencyRepository) GetByKey(ctx context.Context, tenantID, oper func (r *MockIdempotencyRepository) Create(ctx context.Context, record *repository.IdempotencyRecord) error { key := buildKey(record.TenantID, record.OperatorID, record.APIPath, record.IdempotencyKey) + if record.ID == 0 { + record.ID = r.nextID + r.nextID++ + } r.records[key] = record return nil } func (r *MockIdempotencyRepository) UpdateSuccess(ctx context.Context, id int64, responseCode int, responseBody json.RawMessage) error { + for _, record := range r.records { + if record.ID == id { + record.ResponseCode = responseCode + record.ResponseBody = append(json.RawMessage(nil), responseBody...) + record.Status = repository.IdempotencyStatusSucceeded + record.UpdatedAt = time.Now() + return nil + } + } return nil } func (r *MockIdempotencyRepository) UpdateFailed(ctx context.Context, id int64, responseCode int, responseBody json.RawMessage) error { + for _, record := range r.records { + if record.ID == id { + record.ResponseCode = responseCode + record.ResponseBody = append(json.RawMessage(nil), responseBody...) + record.Status = repository.IdempotencyStatusFailed + record.UpdatedAt = time.Now() + return nil + } + } return nil } -func (r *MockIdempotencyRepository) AcquireLock(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey string, ttl time.Duration) (*repository.IdempotencyRecord, error) { +func (r *MockIdempotencyRepository) AcquireLock(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey, requestID, payloadHash string, ttl time.Duration) (*repository.IdempotencyRecord, error) { key := buildKey(tenantID, operatorID, apiPath, idempotencyKey) record := &repository.IdempotencyRecord{ + ID: r.nextID, TenantID: tenantID, OperatorID: operatorID, APIPath: apiPath, IdempotencyKey: idempotencyKey, - RequestID: "test-request-id", - PayloadHash: "", + RequestID: requestID, + PayloadHash: payloadHash, Status: repository.IdempotencyStatusProcessing, ExpiresAt: time.Now().Add(ttl), + CreatedAt: time.Now(), + UpdatedAt: time.Now(), } + r.nextID++ r.records[key] = record return record, nil } func buildKey(tenantID, operatorID int64, apiPath, idempotencyKey string) string { - return strings.Join([]string{ - string(rune(tenantID)), - string(rune(operatorID)), - apiPath, - idempotencyKey, - }, ":") + return fmt.Sprintf("%d:%d:%s:%s", tenantID, operatorID, apiPath, idempotencyKey) } func TestComputePayloadHash(t *testing.T) { @@ -231,3 +255,90 @@ func TestIdempotentHandler_EnabledWithoutRepositoryReturnsServiceUnavailable(t * t.Fatalf("expected status 503, got %d body=%s", w.Code, w.Body.String()) } } + +func TestIdempotencyMiddleware_ReplaysSucceededRequestWithSamePayload(t *testing.T) { + repo := NewMockIdempotencyRepository() + callCount := 0 + + testHandler := func(ctx context.Context, w http.ResponseWriter, r *http.Request, record *repository.IdempotencyRecord) error { + callCount++ + w.WriteHeader(http.StatusCreated) + return json.NewEncoder(w).Encode(map[string]any{ + "request_id": record.RequestID, + "status": "created", + }) + } + + middleware := &IdempotencyMiddleware{ + idempotencyRepo: repo, + config: IdempotencyConfig{ + Enabled: true, + TTL: 24 * time.Hour, + }, + } + handler := middleware.Wrap(testHandler) + + firstReq := httptest.NewRequest(http.MethodPost, "/api/v1/supply/accounts", strings.NewReader(`{"name":"acct-1"}`)) + firstReq = firstReq.WithContext(WithOperatorID(WithTenantID(firstReq.Context(), 1001), 2001)) + firstReq.Header.Set("X-Request-Id", "req-1") + firstReq.Header.Set("Idempotency-Key", "idem-key-12345678") + firstRec := httptest.NewRecorder() + handler.ServeHTTP(firstRec, firstReq) + if firstRec.Code != http.StatusCreated { + t.Fatalf("expected first request to create resource, got=%d body=%s", firstRec.Code, firstRec.Body.String()) + } + + secondReq := httptest.NewRequest(http.MethodPost, "/api/v1/supply/accounts", strings.NewReader(`{"name":"acct-1"}`)) + secondReq = secondReq.WithContext(WithOperatorID(WithTenantID(secondReq.Context(), 1001), 2001)) + secondReq.Header.Set("X-Request-Id", "req-2") + secondReq.Header.Set("Idempotency-Key", "idem-key-12345678") + secondRec := httptest.NewRecorder() + handler.ServeHTTP(secondRec, secondReq) + if secondRec.Code != http.StatusCreated { + t.Fatalf("expected replay request to return original success, got=%d body=%s", secondRec.Code, secondRec.Body.String()) + } + if secondRec.Header().Get("X-Idempotent-Replay") != "true" { + t.Fatalf("expected replay header to be set, got headers=%v", secondRec.Header()) + } + if callCount != 1 { + t.Fatalf("expected handler to run once, got=%d", callCount) + } +} + +func TestIdempotencyMiddleware_RejectsDifferentPayloadForSameKey(t *testing.T) { + repo := NewMockIdempotencyRepository() + + testHandler := func(ctx context.Context, w http.ResponseWriter, r *http.Request, record *repository.IdempotencyRecord) error { + w.WriteHeader(http.StatusCreated) + return json.NewEncoder(w).Encode(map[string]string{"status": "created"}) + } + + middleware := NewIdempotencyMiddleware(repo, IdempotencyConfig{ + Enabled: true, + TTL: 24 * time.Hour, + }) + handler := middleware.Wrap(testHandler) + + firstReq := httptest.NewRequest(http.MethodPost, "/api/v1/supply/accounts", strings.NewReader(`{"name":"acct-1"}`)) + firstReq = firstReq.WithContext(WithOperatorID(WithTenantID(firstReq.Context(), 1001), 2001)) + firstReq.Header.Set("X-Request-Id", "req-a") + firstReq.Header.Set("Idempotency-Key", "idem-key-87654321") + first := httptest.NewRecorder() + handler.ServeHTTP(first, firstReq) + if first.Code != http.StatusCreated { + t.Fatalf("expected first request success, got=%d body=%s", first.Code, first.Body.String()) + } + + secondReq := httptest.NewRequest(http.MethodPost, "/api/v1/supply/accounts", strings.NewReader(`{"name":"acct-2"}`)) + secondReq = secondReq.WithContext(WithOperatorID(WithTenantID(secondReq.Context(), 1001), 2001)) + secondReq.Header.Set("X-Request-Id", "req-b") + secondReq.Header.Set("Idempotency-Key", "idem-key-87654321") + second := httptest.NewRecorder() + handler.ServeHTTP(second, secondReq) + if second.Code != http.StatusConflict { + t.Fatalf("expected payload mismatch conflict, got=%d body=%s", second.Code, second.Body.String()) + } + if !strings.Contains(second.Body.String(), "IDEMPOTENCY_PAYLOAD_MISMATCH") { + t.Fatalf("expected payload mismatch code, got body=%s", second.Body.String()) + } +} diff --git a/supply-api/internal/repository/idempotency.go b/supply-api/internal/repository/idempotency.go index 419a14f5..8dff9777 100644 --- a/supply-api/internal/repository/idempotency.go +++ b/supply-api/internal/repository/idempotency.go @@ -51,7 +51,7 @@ func NewIdempotencyRepository(pool *pgxpool.Pool) *IdempotencyRepository { func (r *IdempotencyRepository) GetByKey(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey string) (*IdempotencyRecord, error) { query := ` SELECT id, tenant_id, operator_id, api_path, idempotency_key, - request_id, payload_hash, response_code, response_body, + request_id, payload_hash, COALESCE(response_code, 0), COALESCE(response_body, 'null'::jsonb), status, expires_at, created_at, updated_at FROM supply_idempotency_records WHERE tenant_id = $1 AND operator_id = $2 AND api_path = $3 AND idempotency_key = $4 @@ -151,7 +151,7 @@ func (r *IdempotencyRepository) DeleteExpired(ctx context.Context) (int64, error func (r *IdempotencyRepository) GetByRequestID(ctx context.Context, requestID string) (*IdempotencyRecord, error) { query := ` SELECT id, tenant_id, operator_id, api_path, idempotency_key, - request_id, payload_hash, response_code, response_body, + request_id, payload_hash, COALESCE(response_code, 0), COALESCE(response_body, 'null'::jsonb), status, expires_at, created_at, updated_at FROM supply_idempotency_records WHERE request_id = $1 @@ -194,17 +194,19 @@ func (r *IdempotencyRepository) CheckExists(ctx context.Context, tenantID, opera } // AcquireLock 尝试获取幂等锁（用于创建记录） -func (r *IdempotencyRepository) AcquireLock(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey string, ttl time.Duration) (*IdempotencyRecord, error) { +func (r *IdempotencyRepository) AcquireLock(ctx context.Context, tenantID, operatorID int64, apiPath, idempotencyKey, requestID, payloadHash string, ttl time.Duration) (*IdempotencyRecord, error) { + now := time.Now().UTC() + // 先尝试插入 record := &IdempotencyRecord{ TenantID: tenantID, OperatorID: operatorID, APIPath: apiPath, IdempotencyKey: idempotencyKey, - RequestID: "", // 稍后填充 - PayloadHash: "", // 稍后填充 + RequestID: requestID, + PayloadHash: payloadHash, Status: IdempotencyStatusProcessing, - ExpiresAt: time.Now().Add(ttl), + ExpiresAt: now.Add(ttl), } query := ` @@ -219,16 +221,19 @@ func (r *IdempotencyRepository) AcquireLock(ctx context.Context, tenantID, opera request_id = EXCLUDED.request_id, payload_hash = EXCLUDED.payload_hash, status = EXCLUDED.status, + response_code = NULL, + response_body = NULL, expires_at = EXCLUDED.expires_at, - updated_at = now() - WHERE supply_idempotency_records.expires_at <= $8 - RETURNING id, created_at, updated_at, status + updated_at = $9 + WHERE supply_idempotency_records.expires_at <= $9 + RETURNING id, request_id, payload_hash, expires_at, created_at, updated_at, status ` err := r.pool.QueryRow(ctx, query, record.TenantID, record.OperatorID, record.APIPath, record.IdempotencyKey, record.RequestID, record.PayloadHash, record.Status, record.ExpiresAt, - ).Scan(&record.ID, &record.CreatedAt, &record.UpdatedAt, &record.Status) + now, + ).Scan(&record.ID, &record.RequestID, &record.PayloadHash, &record.ExpiresAt, &record.CreatedAt, &record.UpdatedAt, &record.Status) if err != nil { // 可能是重复插入 diff --git a/supply-api/internal/repository/idempotency_integration_test.go b/supply-api/internal/repository/idempotency_integration_test.go new file mode 100644 index 00000000..da971eed --- /dev/null +++ b/supply-api/internal/repository/idempotency_integration_test.go @@ -0,0 +1,161 @@ +//go:build integration +// +build integration + +package repository + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestIdempotencyRepositorySchemaContract(t *testing.T) { + if testing.Short() { + t.Skip("integration only") + } + + pool := getTestDB(t) + if pool == nil { + return + } + + requireColumns(t, pool, "supply_idempotency_records", []string{ + "id", "tenant_id", "operator_id", "api_path", "idempotency_key", + "request_id", "payload_hash", "response_code", "response_body", + "status", "expires_at", "created_at", "updated_at", + }) + requireRelationKind(t, pool, "supply_idempotency_records", "r") + requireUniqueConstraint(t, pool, "supply_idempotency_records", []string{ + "tenant_id", "operator_id", "api_path", "idempotency_key", + }) +} + +func TestIdempotencyRepository_AcquireLock_Integration(t *testing.T) { + if testing.Short() { + t.Skip("integration only") + } + + pool := getTestDB(t) + if pool == nil { + return + } + + repo := NewIdempotencyRepository(pool) + ctx := context.Background() + apiPath := "/supply/accounts" + idempotencyKey := "idem-lock-integration-20260420" + firstPayloadHash := strings.Repeat("a", 64) + secondPayloadHash := strings.Repeat("b", 64) + + cleanupIdempotencyRecords(t, pool, apiPath, idempotencyKey) + + first, err := repo.AcquireLock(ctx, 1001, 2001, apiPath, idempotencyKey, "req-idem-1", firstPayloadHash, 24*time.Hour) + if err != nil { + t.Fatalf("first acquire lock failed: %v", err) + } + if first == nil || first.ID == 0 { + t.Fatalf("expected first lock record with id, got %#v", first) + } + if first.RequestID != "req-idem-1" { + t.Fatalf("unexpected first request id: got=%s want=req-idem-1", first.RequestID) + } + if first.PayloadHash != firstPayloadHash { + t.Fatalf("unexpected first payload hash: got=%s want=%s", first.PayloadHash, firstPayloadHash) + } + if first.Status != IdempotencyStatusProcessing { + t.Fatalf("unexpected first lock status: got=%s want=%s", first.Status, IdempotencyStatusProcessing) + } + + persisted, err := repo.GetByKey(ctx, 1001, 2001, apiPath, idempotencyKey) + if err != nil { + t.Fatalf("get by key after first acquire failed: %v", err) + } + if persisted == nil { + t.Fatal("expected persisted idempotency record after first acquire") + } + if persisted.RequestID != "req-idem-1" { + t.Fatalf("unexpected persisted request id: got=%s want=req-idem-1", persisted.RequestID) + } + if persisted.PayloadHash != firstPayloadHash { + t.Fatalf("unexpected persisted payload hash: got=%s want=%s", persisted.PayloadHash, firstPayloadHash) + } + + second, err := repo.AcquireLock(ctx, 1001, 2001, apiPath, idempotencyKey, "req-idem-2", secondPayloadHash, 24*time.Hour) + if err != nil { + t.Fatalf("second acquire lock failed: %v", err) + } + if second == nil { + t.Fatal("expected second lock record") + } + if second.ID != first.ID { + t.Fatalf("expected second acquire to return existing record: first=%d second=%d", first.ID, second.ID) + } + if second.RequestID != "req-idem-1" { + t.Fatalf("expected active lock replay to preserve original request id: got=%s want=req-idem-1", second.RequestID) + } + if second.PayloadHash != firstPayloadHash { + t.Fatalf("expected active lock replay to preserve original payload hash: got=%s want=%s", second.PayloadHash, firstPayloadHash) + } +} + +func requireRelationKind(t *testing.T, pool *pgxpool.Pool, table, want string) { + t.Helper() + + var relkind string + err := pool.QueryRow(context.Background(), ` + SELECT c.relkind::text + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname = $1 + `, table).Scan(&relkind) + if err != nil { + t.Fatalf("检查表 %s relkind 失败: %v", table, err) + } + if relkind != want { + t.Fatalf("表 %s 类型错误: got=%s want=%s", table, relkind, want) + } +} + +func requireUniqueConstraint(t *testing.T, pool *pgxpool.Pool, table string, columns []string) { + t.Helper() + + var exists bool + err := pool.QueryRow(context.Background(), ` + SELECT EXISTS ( + SELECT 1 + FROM pg_constraint c + JOIN pg_class tbl ON tbl.oid = c.conrelid + JOIN pg_namespace n ON n.oid = tbl.relnamespace + JOIN LATERAL ( + SELECT string_agg(att.attname, ',' ORDER BY u.ordinality) AS cols + FROM unnest(c.conkey) WITH ORDINALITY AS u(attnum, ordinality) + JOIN pg_attribute att ON att.attrelid = c.conrelid AND att.attnum = u.attnum + ) cols ON TRUE + WHERE n.nspname = 'public' + AND tbl.relname = $1 + AND c.contype = 'u' + AND cols.cols = $2 + ) + `, table, strings.Join(columns, ",")).Scan(&exists) + if err != nil { + t.Fatalf("检查表 %s 唯一约束失败: %v", table, err) + } + if !exists { + t.Fatalf("表 %s 缺少唯一约束: %s", table, strings.Join(columns, ",")) + } +} + +func cleanupIdempotencyRecords(t *testing.T, pool *pgxpool.Pool, apiPath, idempotencyKey string) { + t.Helper() + + _, err := pool.Exec(context.Background(), ` + DELETE FROM supply_idempotency_records + WHERE api_path = $1 AND idempotency_key = $2 + `, apiPath, idempotencyKey) + if err != nil { + t.Fatalf("清理幂等记录失败: %v", err) + } +} diff --git a/supply-api/internal/repository/partition_manager.go b/supply-api/internal/repository/partition_manager.go index 2d5e8e4b..e6cf0e85 100644 --- a/supply-api/internal/repository/partition_manager.go +++ b/supply-api/internal/repository/partition_manager.go @@ -10,11 +10,11 @@ import ( // PartitionConfig 分区配置 type PartitionConfig struct { - TableName string - PartitionType string // RANGE, LIST - PartitionKey string - RetentionMonths int // 0 = 永久保留 - PreCreateMonths int + TableName string + PartitionType string // RANGE, LIST + PartitionKey string + RetentionMonths int // 0 = 永久保留 + PreCreateMonths int } // PartitionManager 分区管理器 @@ -29,25 +29,18 @@ func NewPartitionManager(pool *pgxpool.Pool) *PartitionManager { pool: pool, config: map[string]*PartitionConfig{ "audit_events": { - TableName: "audit_events", - PartitionType: "RANGE", - PartitionKey: "timestamp", - RetentionMonths: 12, - PreCreateMonths: 3, + TableName: "audit_events", + PartitionType: "RANGE", + PartitionKey: "timestamp", + RetentionMonths: 12, + PreCreateMonths: 3, }, "supply_usage_records": { - TableName: "supply_usage_records", - PartitionType: "RANGE", - PartitionKey: "started_at", - RetentionMonths: 3, - PreCreateMonths: 3, - }, - "supply_idempotency_records": { - TableName: "supply_idempotency_records", - PartitionType: "RANGE", - PartitionKey: "expires_at", - RetentionMonths: 1, // 保留1个月 - PreCreateMonths: 1, + TableName: "supply_usage_records", + PartitionType: "RANGE", + PartitionKey: "started_at", + RetentionMonths: 3, + PreCreateMonths: 3, }, }, } diff --git a/supply-api/internal/repository/partition_manager_test.go b/supply-api/internal/repository/partition_manager_test.go new file mode 100644 index 00000000..af31db6a --- /dev/null +++ b/supply-api/internal/repository/partition_manager_test.go @@ -0,0 +1,11 @@ +package repository + +import "testing" + +func TestNewPartitionManager_ExcludesIdempotencyTable(t *testing.T) { + manager := NewPartitionManager(nil) + + if _, ok := manager.config["supply_idempotency_records"]; ok { + t.Fatal("expected idempotency table to be excluded from partition manager config") + } +} diff --git a/supply-api/sql/postgresql/partition_strategy_v1.sql b/supply-api/sql/postgresql/partition_strategy_v1.sql index 15bd9352..5ca84c5c 100644 --- a/supply-api/sql/postgresql/partition_strategy_v1.sql +++ b/supply-api/sql/postgresql/partition_strategy_v1.sql @@ -105,7 +105,7 @@ BEGIN END; $$ LANGUAGE plpgsql; --- ==================== 3. supply_idempotency_records 分区 (按月分区，保留7天) ==================== +-- ==================== 3. supply_idempotency_records 幂等表 (非分区，保留7天) ==================== CREATE TABLE IF NOT EXISTS supply_idempotency_records ( id BIGSERIAL, @@ -121,29 +121,16 @@ CREATE TABLE IF NOT EXISTS supply_idempotency_records ( expires_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (id, expires_at) -) PARTITION BY RANGE (expires_at); + PRIMARY KEY (id), + CONSTRAINT uq_supply_idempotency_records_key + UNIQUE (tenant_id, operator_id, api_path, idempotency_key) +); +-- 向后兼容保留函数名；幂等表不再分区。 CREATE OR REPLACE FUNCTION create_idempotency_partition(partition_date DATE) RETURNS VOID AS $$ -DECLARE - partition_name TEXT; - start_date DATE; - end_date DATE; BEGIN - start_date := date_trunc('month', partition_date)::DATE; - end_date := (start_date + INTERVAL '1 month')::DATE; - partition_name := 'supply_idempotency_records_' || to_char(start_date, 'YYYY_MM'); - - IF NOT EXISTS ( - SELECT 1 FROM pg_class WHERE relname = partition_name - ) THEN - EXECUTE format( - 'CREATE TABLE %I PARTITION OF supply_idempotency_records FOR VALUES FROM (%L) TO (%L)', - partition_name, start_date, end_date - ); - RAISE NOTICE 'Created partition: %', partition_name; - END IF; + RAISE NOTICE 'supply_idempotency_records is no longer partitioned; skip partition setup for %', partition_date; END; $$ LANGUAGE plpgsql; @@ -231,13 +218,15 @@ CREATE INDEX IF NOT EXISTS idx_audit_events_object ON audit_events(object_type, CREATE INDEX IF NOT EXISTS idx_usage_records_order_id ON supply_usage_records(order_id); CREATE INDEX IF NOT EXISTS idx_usage_records_started_at ON supply_usage_records(started_at); +CREATE INDEX IF NOT EXISTS idx_idempotency_request_id ON supply_idempotency_records(request_id); CREATE INDEX IF NOT EXISTS idx_idempotency_expires_at ON supply_idempotency_records(expires_at); +CREATE INDEX IF NOT EXISTS idx_idempotency_status_expires ON supply_idempotency_records(status, expires_at); -- ==================== 8. 注释 ==================== COMMENT ON TABLE audit_events IS '审计事件表 - 按月分区，保留12个月'; COMMENT ON TABLE supply_usage_records IS '使用记录表 - 按月分区，保留3个月'; -COMMENT ON TABLE supply_idempotency_records IS '幂等记录表 - 按月分区，保留7天以上'; +COMMENT ON TABLE supply_idempotency_records IS '幂等记录表 - 非分区唯一表，保留7天以上'; -- 创建pg_cron作业定期维护分区（需要扩展 pg_cron） -- SELECT cron.schedule('partition-maintenance', '0 0 * * *', 'SELECT ensure_future_partitions()'); diff --git a/supply-api/sql/postgresql/supply_idempotency_records_partitioned_to_plain_v2.sql b/supply-api/sql/postgresql/supply_idempotency_records_partitioned_to_plain_v2.sql new file mode 100644 index 00000000..f4785005 --- /dev/null +++ b/supply-api/sql/postgresql/supply_idempotency_records_partitioned_to_plain_v2.sql @@ -0,0 +1,77 @@ +-- Migrate supply_idempotency_records from partitioned table to plain unique table. +-- This migration is intended for environments that previously applied partition_strategy_v1.sql +-- with supply_idempotency_records partitioned by expires_at. + +BEGIN; + +LOCK TABLE supply_idempotency_records IN ACCESS EXCLUSIVE MODE; + +CREATE TABLE IF NOT EXISTS supply_idempotency_records_v2 ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT NOT NULL, + operator_id BIGINT NOT NULL, + api_path VARCHAR(200) NOT NULL, + idempotency_key VARCHAR(128) NOT NULL, + request_id VARCHAR(64) NOT NULL, + payload_hash CHAR(64) NOT NULL, + response_code INT, + response_body JSONB, + status VARCHAR(20) NOT NULL DEFAULT 'processing', + expires_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT uq_supply_idempotency_records_v2_key + UNIQUE (tenant_id, operator_id, api_path, idempotency_key) +); + +INSERT INTO supply_idempotency_records_v2 ( + tenant_id, + operator_id, + api_path, + idempotency_key, + request_id, + payload_hash, + response_code, + response_body, + status, + expires_at, + created_at, + updated_at +) +SELECT DISTINCT ON (tenant_id, operator_id, api_path, idempotency_key) + tenant_id, + operator_id, + api_path, + idempotency_key, + request_id, + payload_hash, + response_code, + response_body, + status, + expires_at, + created_at, + updated_at +FROM supply_idempotency_records +ORDER BY + tenant_id, + operator_id, + api_path, + idempotency_key, + CASE WHEN expires_at > CURRENT_TIMESTAMP THEN 0 ELSE 1 END, + expires_at DESC, + updated_at DESC, + id DESC; + +ALTER TABLE supply_idempotency_records RENAME TO supply_idempotency_records_partitioned_legacy; +ALTER TABLE supply_idempotency_records_v2 RENAME TO supply_idempotency_records; + +CREATE INDEX IF NOT EXISTS idx_idempotency_request_id + ON supply_idempotency_records (request_id); +CREATE INDEX IF NOT EXISTS idx_idempotency_expires_at + ON supply_idempotency_records (expires_at); +CREATE INDEX IF NOT EXISTS idx_idempotency_status_expires + ON supply_idempotency_records (status, expires_at); + +COMMENT ON TABLE supply_idempotency_records IS '幂等记录表 - 非分区唯一表，保留7天以上'; + +COMMIT;