485 lines
15 KiB
Go
485 lines
15 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/billing"
|
|
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/identity"
|
|
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/jobs"
|
|
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/orchestration"
|
|
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/publicapi"
|
|
)
|
|
|
|
type JobsPublicAuthenticator interface {
|
|
Authenticate(*http.Request) (publicapi.PublicClient, string, error)
|
|
AssertInternalWorker(*http.Request) error
|
|
}
|
|
type JobBuildInput struct {
|
|
Scope jobs.Scope
|
|
Capability string
|
|
Body map[string]any
|
|
IdempotencyKey string
|
|
Origin string
|
|
}
|
|
type JobBuilder interface {
|
|
Build(context.Context, JobBuildInput) (jobs.CreateCommand, error)
|
|
}
|
|
|
|
// ProviderCommandBuilder is the shared seam used by creation, retry, and quote
|
|
// paths. Implementations may resolve runtime provider settings per request.
|
|
type ProviderCommandBuilder interface {
|
|
Build(context.Context, string, string, string, string, map[string]any) (jobs.CreateCommand, error)
|
|
}
|
|
|
|
type JobCreationCoordinator interface {
|
|
CreatePlatform(context.Context, identity.Session, orchestration.CreationInput) (jobs.Job, bool, error)
|
|
CreatePublic(context.Context, orchestration.CreationInput) (jobs.Job, bool, error)
|
|
}
|
|
|
|
type PlatformJobRetryCoordinator interface {
|
|
RetryPlatform(context.Context, identity.Session, jobs.Job) (jobs.Job, error)
|
|
}
|
|
|
|
type JobTickLimiter interface {
|
|
TickLimit(context.Context, string, int) (jobs.TickResult, error)
|
|
}
|
|
|
|
// ProviderBuilderAdapter exposes jobs.ProviderJobBuilder at the HTTP seam.
|
|
type ProviderBuilderAdapter struct{ Builder ProviderCommandBuilder }
|
|
|
|
func (a ProviderBuilderAdapter) Build(ctx context.Context, in JobBuildInput) (jobs.CreateCommand, error) {
|
|
return a.Builder.Build(ctx, in.Scope.OwnerID, in.Scope.ExternalClientID, in.Capability, in.IdempotencyKey, in.Body)
|
|
}
|
|
|
|
type JobsDependencies struct {
|
|
Service *jobs.Service
|
|
Platform *PlatformAuthorizer
|
|
Public JobsPublicAuthenticator
|
|
Builder JobBuilder
|
|
Creation JobCreationCoordinator
|
|
Refunds jobs.RefundPort
|
|
Artifacts jobs.ArtifactDeleter
|
|
Worker JobTickLimiter
|
|
}
|
|
type JobsConfig struct {
|
|
MaxJSONBytes int64
|
|
NewID func() string
|
|
}
|
|
type jobsHandler struct {
|
|
dependencies JobsDependencies
|
|
config JobsConfig
|
|
}
|
|
|
|
func NewJobsHandler(d JobsDependencies, c JobsConfig) (http.Handler, error) {
|
|
if d.Service == nil || d.Platform == nil || d.Public == nil || d.Builder == nil {
|
|
return nil, errors.New("jobs HTTP dependencies are not configured")
|
|
}
|
|
if c.MaxJSONBytes <= 0 {
|
|
c.MaxJSONBytes = 1 << 20
|
|
}
|
|
if c.NewID == nil {
|
|
c.NewID = randomJobID
|
|
}
|
|
return &jobsHandler{d, c}, nil
|
|
}
|
|
|
|
func (h *jobsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
route, id := matchJobRoute(r.URL.Path)
|
|
if route == "" {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
allow := jobAllow(route)
|
|
if r.Method == http.MethodOptions {
|
|
w.Header().Set("Allow", allow)
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
if !methodAllowed(allow, r.Method) {
|
|
w.Header().Set("Allow", allow)
|
|
w.WriteHeader(http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
switch route {
|
|
case "platform-image-list":
|
|
h.platformCollection(w, r, "image.generate")
|
|
case "platform-video-list":
|
|
h.platformCollection(w, r, "video.generate")
|
|
case "platform-image-item":
|
|
h.platformItem(w, r, id, false)
|
|
case "platform-video-item":
|
|
h.platformItem(w, r, id, true)
|
|
case "platform-retry":
|
|
h.platformRetry(w, r, id)
|
|
case "public-list":
|
|
h.publicCollection(w, r)
|
|
case "public-item":
|
|
h.publicGet(w, r, id)
|
|
case "public-cancel":
|
|
h.publicCancel(w, r, id)
|
|
case "worker":
|
|
h.workerTick(w, r)
|
|
}
|
|
}
|
|
|
|
func (h *jobsHandler) platformCollection(w http.ResponseWriter, r *http.Request, capability string) {
|
|
session, err := h.dependencies.Platform.Authorize(r, PlatformApp)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
scope := jobs.Scope{OwnerID: session.User.ID}
|
|
if r.Method == http.MethodGet {
|
|
items, err := h.dependencies.Service.List(r.Context(), jobs.ListFilter{Scope: scope, Capability: capability, Limit: 200})
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
if items == nil {
|
|
items = []jobs.Job{}
|
|
}
|
|
writeJSON(w, 200, map[string]any{"jobs": items})
|
|
return
|
|
}
|
|
h.create(w, r, scope, capability, false, &session)
|
|
}
|
|
func (h *jobsHandler) platformItem(w http.ResponseWriter, r *http.Request, id string, video bool) {
|
|
session, err := h.dependencies.Platform.Authorize(r, PlatformApp)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
scope := jobs.Scope{OwnerID: session.User.ID}
|
|
if r.Method == http.MethodGet {
|
|
j, err := h.dependencies.Service.Get(r.Context(), scope, id)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"job": j})
|
|
return
|
|
}
|
|
j, err := h.dependencies.Service.Get(r.Context(), scope, id)
|
|
if err != nil || ((j.Capability == "video.generate") != video) {
|
|
writeJSON(w, 404, map[string]string{"error": "任务不存在"})
|
|
return
|
|
}
|
|
if !j.Status.Terminal() {
|
|
j, err = h.dependencies.Service.Cancel(r.Context(), scope, id, h.dependencies.Refunds)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
}
|
|
deletedAssets := []string{}
|
|
if h.dependencies.Artifacts != nil {
|
|
deletedAssets, err = h.dependencies.Artifacts.DeleteOutputs(r.Context(), j)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
}
|
|
_, err = h.dependencies.Service.Delete(r.Context(), scope, id, nil)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"ok": true, "deletedJobId": id, "deletedAssetIds": deletedAssets})
|
|
}
|
|
func (h *jobsHandler) platformRetry(w http.ResponseWriter, r *http.Request, id string) {
|
|
session, err := h.dependencies.Platform.Authorize(r, PlatformApp)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
original, err := h.dependencies.Service.Get(r.Context(), jobs.Scope{OwnerID: session.User.ID}, id)
|
|
if err != nil || original.Capability != "image.generate" {
|
|
writeJSON(w, 404, map[string]string{"error": "任务不存在"})
|
|
return
|
|
}
|
|
retryCoordinator, ok := h.dependencies.Creation.(PlatformJobRetryCoordinator)
|
|
if !ok {
|
|
writeJSON(w, 500, map[string]string{"error": "Internal server error."})
|
|
return
|
|
}
|
|
j, err := retryCoordinator.RetryPlatform(r.Context(), session, original)
|
|
if err != nil {
|
|
writeJobError(w, err, false)
|
|
return
|
|
}
|
|
writeJSON(w, 202, map[string]any{"job": j})
|
|
}
|
|
|
|
func (h *jobsHandler) publicCollection(w http.ResponseWriter, r *http.Request) {
|
|
client, owner, err := h.dependencies.Public.Authenticate(r)
|
|
if err != nil {
|
|
writeJobError(w, err, true)
|
|
return
|
|
}
|
|
scope := jobs.Scope{OwnerID: owner, ExternalClientID: client.ID}
|
|
if r.Method == http.MethodPost {
|
|
h.create(w, r, scope, "", true, nil)
|
|
return
|
|
}
|
|
q := r.URL.Query()
|
|
limit, _ := strconv.Atoi(q.Get("limit"))
|
|
var before *time.Time
|
|
if raw := q.Get("before"); raw != "" {
|
|
if parsed, e := time.Parse(time.RFC3339, raw); e == nil {
|
|
before = &parsed
|
|
}
|
|
}
|
|
items, err := h.dependencies.Service.List(r.Context(), jobs.ListFilter{Scope: scope, Status: jobs.Status(q.Get("status")), Capability: q.Get("capability"), Limit: limit, Before: before})
|
|
if err != nil {
|
|
writeJobError(w, err, true)
|
|
return
|
|
}
|
|
if items == nil {
|
|
items = []jobs.Job{}
|
|
}
|
|
writeJSON(w, 200, map[string]any{"jobs": items})
|
|
}
|
|
func (h *jobsHandler) publicGet(w http.ResponseWriter, r *http.Request, id string) {
|
|
_, scope, ok := h.publicScope(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
j, err := h.dependencies.Service.Get(r.Context(), scope, id)
|
|
if err != nil {
|
|
writeJobError(w, err, true)
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"job": j})
|
|
}
|
|
func (h *jobsHandler) publicCancel(w http.ResponseWriter, r *http.Request, id string) {
|
|
_, scope, ok := h.publicScope(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
j, err := h.dependencies.Service.Cancel(r.Context(), scope, id, h.dependencies.Refunds)
|
|
if err != nil {
|
|
writeJobError(w, err, true)
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"job": j})
|
|
}
|
|
func (h *jobsHandler) publicScope(w http.ResponseWriter, r *http.Request) (publicapi.PublicClient, jobs.Scope, bool) {
|
|
c, o, e := h.dependencies.Public.Authenticate(r)
|
|
if e != nil {
|
|
writeJobError(w, e, true)
|
|
return publicapi.PublicClient{}, jobs.Scope{}, false
|
|
}
|
|
return c, jobs.Scope{OwnerID: o, ExternalClientID: c.ID}, true
|
|
}
|
|
func (h *jobsHandler) create(w http.ResponseWriter, r *http.Request, scope jobs.Scope, capability string, public bool, session *identity.Session) {
|
|
body := map[string]any{}
|
|
if !decodeJobJSON(w, r, h.config.MaxJSONBytes, &body, public) {
|
|
return
|
|
}
|
|
if capability == "" {
|
|
capability = stringValueHTTP(body["capability"])
|
|
if capability == "" {
|
|
capability = "image.generate"
|
|
}
|
|
}
|
|
if priority, exists := body["priority"]; exists {
|
|
body["priority"] = float64(jobs.NormalizePriority(jobHTTPInt(priority)))
|
|
}
|
|
if public {
|
|
if webhook := strings.TrimSpace(stringValueHTTP(body["webhookUrl"])); webhook != "" {
|
|
parsed, parseErr := url.Parse(webhook)
|
|
if parseErr != nil || !parsed.IsAbs() || parsed.Host == "" || parsed.User != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "webhookUrl must be an HTTP or HTTPS URL."})
|
|
return
|
|
}
|
|
body["webhookUrl"] = webhook
|
|
}
|
|
}
|
|
key := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
|
|
if key == "" {
|
|
key = stringValueHTTP(body["idempotencyKey"])
|
|
}
|
|
var created jobs.Job
|
|
var reused bool
|
|
var err error
|
|
if h.dependencies.Creation != nil {
|
|
input := orchestration.CreationInput{OwnerID: scope.OwnerID, ExternalClientID: scope.ExternalClientID, Capability: capability, Body: body, IdempotencyKey: key}
|
|
if public {
|
|
created, reused, err = h.dependencies.Creation.CreatePublic(r.Context(), input)
|
|
} else if session != nil {
|
|
created, reused, err = h.dependencies.Creation.CreatePlatform(r.Context(), *session, input)
|
|
} else {
|
|
err = errors.New("platform creation session is unavailable")
|
|
}
|
|
} else {
|
|
cmd, buildErr := h.dependencies.Builder.Build(r.Context(), JobBuildInput{Scope: scope, Capability: capability, Body: body, IdempotencyKey: key, Origin: absoluteRequestURL(r)})
|
|
if buildErr != nil {
|
|
writeJobError(w, buildErr, public)
|
|
return
|
|
}
|
|
created, reused, err = h.dependencies.Service.Create(r.Context(), cmd)
|
|
}
|
|
if err != nil {
|
|
writeJobError(w, err, public)
|
|
return
|
|
}
|
|
status := 202
|
|
if reused {
|
|
status = 200
|
|
}
|
|
response := map[string]any{"job": created}
|
|
if public {
|
|
response["reused"] = reused
|
|
}
|
|
writeJSON(w, status, response)
|
|
}
|
|
|
|
func jobHTTPInt(value any) int {
|
|
switch typed := value.(type) {
|
|
case float64:
|
|
return int(typed)
|
|
case int:
|
|
return typed
|
|
case json.Number:
|
|
parsed, _ := strconv.Atoi(typed.String())
|
|
return parsed
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
func (h *jobsHandler) workerTick(w http.ResponseWriter, r *http.Request) {
|
|
if err := h.dependencies.Public.AssertInternalWorker(r); err != nil {
|
|
writeJobError(w, err, true)
|
|
return
|
|
}
|
|
if h.dependencies.Worker == nil {
|
|
writeJSON(w, 500, map[string]string{"error": "Internal server error."})
|
|
return
|
|
}
|
|
input := struct {
|
|
WorkerID string `json:"workerId"`
|
|
Limit int `json:"limit"`
|
|
}{}
|
|
if !decodeJobJSON(w, r, h.config.MaxJSONBytes, &input, true) {
|
|
return
|
|
}
|
|
if input.WorkerID == "" {
|
|
input.WorkerID = h.config.NewID()
|
|
}
|
|
result, err := h.dependencies.Worker.TickLimit(r.Context(), input.WorkerID, input.Limit)
|
|
if err != nil {
|
|
writeJobError(w, err, true)
|
|
return
|
|
}
|
|
writeJSON(w, 200, result)
|
|
}
|
|
|
|
func matchJobRoute(p string) (string, string) {
|
|
switch p {
|
|
case "/api/generations/image":
|
|
return "platform-image-list", ""
|
|
case "/api/generations/video":
|
|
return "platform-video-list", ""
|
|
case "/api/v1/jobs":
|
|
return "public-list", ""
|
|
case "/api/internal/worker/tick":
|
|
return "worker", ""
|
|
}
|
|
parts := strings.Split(strings.Trim(p, "/"), "/")
|
|
if len(parts) == 4 && parts[0] == "api" && parts[1] == "generations" && (parts[2] == "image" || parts[2] == "video") {
|
|
return "platform-" + parts[2] + "-item", parts[3]
|
|
}
|
|
if len(parts) == 5 && parts[0] == "api" && parts[1] == "generations" && parts[2] == "image" && parts[4] == "retry" {
|
|
return "platform-retry", parts[3]
|
|
}
|
|
if len(parts) == 5 && parts[0] == "api" && parts[1] == "v1" && parts[2] == "jobs" && parts[4] == "cancel" {
|
|
return "public-cancel", parts[3]
|
|
}
|
|
if len(parts) == 4 && parts[0] == "api" && parts[1] == "v1" && parts[2] == "jobs" {
|
|
return "public-item", parts[3]
|
|
}
|
|
return "", ""
|
|
}
|
|
func jobAllow(route string) string {
|
|
switch route {
|
|
case "platform-image-list", "platform-video-list", "public-list":
|
|
return "GET, POST"
|
|
case "platform-image-item", "platform-video-item":
|
|
return "GET, DELETE"
|
|
case "public-item":
|
|
return "GET"
|
|
default:
|
|
return "POST"
|
|
}
|
|
}
|
|
func decodeJobJSON(w http.ResponseWriter, r *http.Request, limit int64, target any, public bool) bool {
|
|
r.Body = http.MaxBytesReader(w, r.Body, limit)
|
|
if err := json.NewDecoder(r.Body).Decode(target); err != nil {
|
|
message := "Invalid request body."
|
|
if !public {
|
|
message = "请求参数无效"
|
|
}
|
|
writeJSON(w, 400, map[string]string{"error": message})
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
func writeJobError(w http.ResponseWriter, err error, public bool) {
|
|
var domain *jobs.Error
|
|
if errors.As(err, &domain) {
|
|
message := domain.Message
|
|
if !public && domain.Kind == jobs.ErrorNotFound {
|
|
message = "Generation job not found."
|
|
}
|
|
writeJSON(w, domain.Status, map[string]string{"error": message})
|
|
return
|
|
}
|
|
var auth *publicapi.AuthError
|
|
if errors.As(err, &auth) {
|
|
writeJSON(w, auth.Status, map[string]string{"error": auth.Message})
|
|
return
|
|
}
|
|
var platform *PlatformAuthError
|
|
if errors.As(err, &platform) {
|
|
writeJSON(w, platform.Status, map[string]string{"error": platform.Message})
|
|
return
|
|
}
|
|
var billingStatus *billing.StatusError
|
|
if errors.As(err, &billingStatus) && (billingStatus.Status == 402 || billingStatus.Status == 409) {
|
|
message := "生成任务计费失败。"
|
|
if public {
|
|
message = "Generation job billing failed."
|
|
} else if errors.Is(err, billing.ErrInsufficientBalance) {
|
|
message = "余额不足，请先充值。"
|
|
} else if errors.Is(err, billing.ErrOrganizationUnavailable) {
|
|
message = "账号未绑定有效组织，请联系管理员。"
|
|
}
|
|
writeJSON(w, billingStatus.Status, map[string]string{"error": message})
|
|
return
|
|
}
|
|
if errors.Is(err, billing.ErrProviderUnavailable) {
|
|
message := "计费服务暂不可用，请稍后重试。"
|
|
if public {
|
|
message = "Generation billing is temporarily unavailable."
|
|
}
|
|
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": message})
|
|
return
|
|
}
|
|
writeJSON(w, 500, map[string]string{"error": "Internal server error."})
|
|
}
|
|
func randomJobID() string {
|
|
b := make([]byte, 16)
|
|
_, _ = rand.Read(b)
|
|
return "job-" + hex.EncodeToString(b)
|
|
}
|
|
func stringValueHTTP(v any) string { s, _ := v.(string); return strings.TrimSpace(s) }
|