feat(kling): 接入O1与3.0 Omni兼容接口

新增中国区可灵 V1 AK/SK Omni 协议与 API 2.0 兼容路径，补齐任务隔离、外部任务幂等、参数校验和 OpenAPI 文档。\n\n验证：O1 与 3.0 Omni 真实 V1 任务成功；Go、前端、依赖审计、迁移及 CI 脚本门禁通过。
2026-07-21 23:47:13 +08:00
parent e280c0875c
commit 9d4501bc42
15 changed files with 2113 additions and 23 deletions
@@ -63,6 +63,7 @@ var (
 	ErrBalanceBelowFrozen            = errors.New("wallet balance cannot be below frozen balance")
 	ErrInvalidWalletAmount           = errors.New("wallet amount must be a decimal with at most nine fractional digits")
 	ErrIdempotencyKeyReused          = errors.New("idempotency key was reused for a different request")
+	ErrExternalTaskIDReused          = errors.New("external task id was reused")
 	ErrTaskExecutionLeaseUnavailable = errors.New("task execution lease is unavailable")
 	ErrTaskExecutionLeaseLost        = errors.New("task execution lease was lost")
 	ErrTaskExecutionFinished         = errors.New("task execution already finished")
@@ -465,6 +466,7 @@ type RateLimitWindow struct {
 type CreateTaskInput struct {
 	Kind                   string                `json:"kind"`
 	Model                  string                `json:"model"`
+	ExternalTaskID         string                `json:"externalTaskId,omitempty"`
 	RunMode                string                `json:"runMode"`
 	Async                  bool                  `json:"async"`
 	Request                map[string]any        `json:"request"`
@@ -482,6 +484,7 @@ type CreateTaskResult struct {

 type GatewayTask struct {
 	ID                  string         `json:"id"`
+	ExternalTaskID      string         `json:"externalTaskId,omitempty"`
 	Kind                string         `json:"kind"`
 	RunMode             string         `json:"runMode"`
 	UserID              string         `json:"userId"`
@@ -541,7 +544,7 @@ type GatewayTask struct {
 }

 const gatewayTaskColumns = `
-id::text, kind, run_mode, user_id, COALESCE(gateway_user_id::text, ''), user_source,
+id::text, COALESCE(external_task_id, ''), kind, run_mode, user_id, COALESCE(gateway_user_id::text, ''), user_source,
 COALESCE(gateway_tenant_id::text, ''), COALESCE(tenant_id, ''), COALESCE(tenant_key, ''),
 COALESCE(api_key_id, ''), COALESCE(api_key_name, ''), COALESCE(api_key_prefix, ''),
 COALESCE(user_group_id::text, ''), COALESCE(user_group_key, ''), model,
@@ -1948,15 +1951,15 @@ func (s *Store) CreateTaskIdempotent(ctx context.Context, input CreateTaskInput,

 	task, err := scanGatewayTask(tx.QueryRow(ctx, `
 	INSERT INTO gateway_tasks (
-	  kind, run_mode, user_id, gateway_user_id, user_source, gateway_tenant_id, tenant_id, tenant_key,
+	  external_task_id, kind, run_mode, user_id, gateway_user_id, user_source, gateway_tenant_id, tenant_id, tenant_key,
 	  api_key_id, api_key_name, api_key_prefix, user_group_id, user_group_key,
 	  model, requested_model, request, async_mode, status, result, billings, conversation_id, new_message_count,
 	  idempotency_key_hash, idempotency_request_hash, finished_at
 	)
-	VALUES ($1, $2, $3, NULLIF($4, '')::uuid, COALESCE(NULLIF($5, ''), 'gateway'), NULLIF($6, '')::uuid, NULLIF($7, ''), NULLIF($8, ''), NULLIF($9, ''), NULLIF($10, ''), NULLIF($11, ''), NULLIF($12, '')::uuid, NULLIF($13, ''), $14, $14, $15, $16, $17, $18::jsonb, $19::jsonb, NULLIF($20, '')::uuid, $21, NULLIF($22, ''), NULLIF($23, ''), NULL)
+	VALUES (NULLIF($1, ''), $2, $3, $4, NULLIF($5, '')::uuid, COALESCE(NULLIF($6, ''), 'gateway'), NULLIF($7, '')::uuid, NULLIF($8, ''), NULLIF($9, ''), NULLIF($10, ''), NULLIF($11, ''), NULLIF($12, ''), NULLIF($13, '')::uuid, NULLIF($14, ''), $15, $15, $16, $17, $18, $19::jsonb, $20::jsonb, NULLIF($21, '')::uuid, $22, NULLIF($23, ''), NULLIF($24, ''), NULL)
 	ON CONFLICT (user_id, idempotency_key_hash) WHERE idempotency_key_hash IS NOT NULL DO NOTHING
 	RETURNING `+gatewayTaskColumns,
-		input.Kind, runMode, user.ID, user.GatewayUserID, user.Source, user.GatewayTenantID, user.TenantID, user.TenantKey, user.APIKeyID, user.APIKeyName, user.APIKeyPrefix, user.UserGroupID, user.UserGroupKey, input.Model, requestBody, input.Async, status, resultBody, billingsBody, input.ConversationID, input.NewMessageCount, strings.TrimSpace(input.IdempotencyKeyHash), strings.TrimSpace(input.IdempotencyRequestHash),
+		strings.TrimSpace(input.ExternalTaskID), input.Kind, runMode, user.ID, user.GatewayUserID, user.Source, user.GatewayTenantID, user.TenantID, user.TenantKey, user.APIKeyID, user.APIKeyName, user.APIKeyPrefix, user.UserGroupID, user.UserGroupKey, input.Model, requestBody, input.Async, status, resultBody, billingsBody, input.ConversationID, input.NewMessageCount, strings.TrimSpace(input.IdempotencyKeyHash), strings.TrimSpace(input.IdempotencyRequestHash),
 	))
 	replayed := false
 	if errors.Is(err, pgx.ErrNoRows) && strings.TrimSpace(input.IdempotencyKeyHash) != "" {
@@ -1977,6 +1980,9 @@ FROM gateway_tasks
 WHERE user_id = $1 AND idempotency_key_hash = $2`, user.ID, strings.TrimSpace(input.IdempotencyKeyHash)))
 		replayed = true
 	}
+	if isUniqueViolation(err) && strings.TrimSpace(input.ExternalTaskID) != "" {
+		return CreateTaskResult{}, ErrExternalTaskIDReused
+	}
 	if err != nil {
 		return CreateTaskResult{}, err
 	}
@@ -2041,6 +2047,7 @@ func scanGatewayTask(scanner taskScanner) (GatewayTask, error) {
 	var remoteTaskPayloadBytes []byte
 	if err := scanner.Scan(
 		&task.ID,
+		&task.ExternalTaskID,
 		&task.Kind,
 		&task.RunMode,
 		&task.UserID,
@@ -29,6 +29,16 @@ type TaskListResult struct {
 	PageSize int
 }

+type CompatTaskListFilter struct {
+	Provider    string
+	Version     string
+	Statuses    []string
+	CreatedFrom *time.Time
+	CreatedTo   *time.Time
+	Page        int
+	PageSize    int
+}
+
 func (s *Store) ListTasks(ctx context.Context, user *auth.User, filter TaskListFilter) (TaskListResult, error) {
 	page := filter.Page
 	if page <= 0 {
@@ -146,6 +156,122 @@ LIMIT $8 OFFSET $9`, queryArgs...)
 	}, nil
 }

+func (s *Store) GetCompatTask(ctx context.Context, user *auth.User, provider string, version string, identifier string) (GatewayTask, error) {
+	gatewayUserID := localGatewayUserID(user)
+	apiKeyID := ""
+	userID := ""
+	if user != nil {
+		apiKeyID = strings.TrimSpace(user.APIKeyID)
+		userID = strings.TrimSpace(user.ID)
+	}
+	if gatewayUserID == "" && userID == "" {
+		return GatewayTask{}, ErrLocalUserRequired
+	}
+	task, err := scanGatewayTask(s.pool.QueryRow(ctx, `
+SELECT `+gatewayTaskColumns+`
+FROM gateway_tasks
+WHERE (
+    (NULLIF($1, '')::uuid IS NOT NULL AND gateway_user_id = NULLIF($1, '')::uuid)
+    OR (NULLIF($1, '')::uuid IS NULL AND NULLIF($2, '') IS NOT NULL AND user_id = $2)
+  )
+  AND (NULLIF($3, '') IS NULL OR api_key_id = $3)
+  AND request->>'_compat_provider' = $4
+  AND request->>'_kling_compat_version' = $5
+  AND (id::text = $6 OR external_task_id = $6)
+ORDER BY created_at DESC
+LIMIT 1`, gatewayUserID, userID, apiKeyID, strings.TrimSpace(provider), strings.TrimSpace(version), strings.TrimSpace(identifier)))
+	if err != nil {
+		return GatewayTask{}, err
+	}
+	attempts, err := s.ListTaskAttempts(ctx, task.ID)
+	if err != nil {
+		return GatewayTask{}, err
+	}
+	task.Attempts = attempts
+	return task, nil
+}
+
+func (s *Store) ListCompatTasks(ctx context.Context, user *auth.User, filter CompatTaskListFilter) (TaskListResult, error) {
+	page := filter.Page
+	if page <= 0 {
+		page = 1
+	}
+	pageSize := filter.PageSize
+	if pageSize <= 0 {
+		pageSize = 30
+	}
+	if pageSize > 500 {
+		pageSize = 500
+	}
+	offset := (page - 1) * pageSize
+	gatewayUserID := localGatewayUserID(user)
+	apiKeyID := ""
+	userID := ""
+	if user != nil {
+		apiKeyID = strings.TrimSpace(user.APIKeyID)
+		userID = strings.TrimSpace(user.ID)
+	}
+	if gatewayUserID == "" && userID == "" {
+		return TaskListResult{}, ErrLocalUserRequired
+	}
+	statuses := filter.Statuses
+	if len(statuses) == 0 {
+		statuses = nil
+	}
+	args := []any{
+		gatewayUserID,
+		userID,
+		apiKeyID,
+		strings.TrimSpace(filter.Provider),
+		strings.TrimSpace(filter.Version),
+		nullableTaskListTime(filter.CreatedFrom),
+		nullableTaskListTime(filter.CreatedTo),
+		statuses,
+	}
+	whereSQL := `
+WHERE (
+    (NULLIF($1, '')::uuid IS NOT NULL AND gateway_user_id = NULLIF($1, '')::uuid)
+    OR (NULLIF($1, '')::uuid IS NULL AND NULLIF($2, '') IS NOT NULL AND user_id = $2)
+  )
+  AND (NULLIF($3, '') IS NULL OR api_key_id = $3)
+  AND request->>'_compat_provider' = $4
+  AND request->>'_kling_compat_version' = $5
+  AND ($6::timestamptz IS NULL OR created_at >= $6::timestamptz)
+  AND ($7::timestamptz IS NULL OR created_at <= $7::timestamptz)
+  AND ($8::text[] IS NULL OR status = ANY($8::text[]))`
+	var total int
+	if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_tasks `+whereSQL, args...).Scan(&total); err != nil {
+		return TaskListResult{}, err
+	}
+	queryArgs := append(args, pageSize, offset)
+	rows, err := s.pool.Query(ctx, `
+SELECT `+gatewayTaskColumns+`
+FROM gateway_tasks
+`+whereSQL+`
+ORDER BY created_at DESC
+LIMIT $9 OFFSET $10`, queryArgs...)
+	if err != nil {
+		return TaskListResult{}, err
+	}
+	defer rows.Close()
+	items := make([]GatewayTask, 0)
+	for rows.Next() {
+		task, err := scanGatewayTask(rows)
+		if err != nil {
+			return TaskListResult{}, err
+		}
+		items = append(items, task)
+	}
+	if err := rows.Err(); err != nil {
+		return TaskListResult{}, err
+	}
+	items, err = s.attachTaskAttempts(ctx, items)
+	if err != nil {
+		return TaskListResult{}, err
+	}
+	return TaskListResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil
+}
+
 func nullableTaskListTime(value *time.Time) any {
 	if value == nil {
 		return nil