383 lines
12 KiB
Go
383 lines
12 KiB
Go
package httpserver
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"mime"
|
|
"net/http"
|
|
"net/url"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
|
|
"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
)
|
|
|
|
type endpoint func(*http.Request, auth.Token) (any, int, error)
|
|
|
|
func registerTasks(mux *http.ServeMux, s *service.Service) {
|
|
mux.HandleFunc("POST /v1/tasks:batch", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
if len(r.Header.Values("Idempotency-Key")) != 0 {
|
|
return nil, 0, fault.Invalid("batch requires per-item idempotency_key, not an Idempotency-Key header")
|
|
}
|
|
var in service.BatchSubmitInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.SubmitBatch(r.Context(), t, in)
|
|
return out, http.StatusOK, err
|
|
}))
|
|
mux.HandleFunc("POST /v1/tasks", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
var in service.SubmitInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.Submit(r.Context(), t, key, in)
|
|
return out, createdStatus(out.Deduplicated), err
|
|
}))
|
|
mux.HandleFunc("GET /v1/tasks/{id}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
out, err := s.GetTask(r.Context(), t, core.ID(r.PathValue("id")))
|
|
return out, 200, err
|
|
}))
|
|
mux.HandleFunc("GET /v1/tasks", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
q, queryErr := url.ParseQuery(r.URL.RawQuery)
|
|
if queryErr != nil {
|
|
return nil, 0, fault.Invalid("invalid query encoding")
|
|
}
|
|
for k, v := range q {
|
|
if len(v) != 1 || (k != "queue_id" && k != "state" && k != "limit" && k != "cursor" && k != "business_key" && k != "created_from" && k != "created_before") {
|
|
return nil, 0, fault.Invalid("invalid query parameter")
|
|
}
|
|
}
|
|
limit := 0
|
|
if q.Has("limit") {
|
|
var err error
|
|
limit, err = strconv.Atoi(q.Get("limit"))
|
|
if err != nil || limit <= 0 {
|
|
return nil, 0, fault.Invalid("invalid limit")
|
|
}
|
|
}
|
|
in := service.ListInput{QueueID: core.ID(q.Get("queue_id")), State: pulse.TaskState(q.Get("state")), Limit: limit, Cursor: q.Get("cursor"), BusinessKey: q.Get("business_key")}
|
|
if q.Has("business_key") && in.BusinessKey == "" {
|
|
return nil, 0, fault.Invalid("business_key filter must not be empty")
|
|
}
|
|
for name, field := range map[string]**time.Time{"created_from": &in.CreatedFrom, "created_before": &in.CreatedBefore} {
|
|
if q.Has(name) {
|
|
v, err := time.Parse(time.RFC3339Nano, q.Get(name))
|
|
if err != nil {
|
|
return nil, 0, fault.Invalid("invalid creation time bound")
|
|
}
|
|
*field = &v
|
|
}
|
|
}
|
|
out, err := s.ListTasks(r.Context(), t, in)
|
|
return out, 200, err
|
|
}))
|
|
mux.HandleFunc("POST /v1/tasks/{command}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
command := r.PathValue("command")
|
|
if strings.HasSuffix(command, ":review") {
|
|
var in service.ReviewInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.ReviewTask(r.Context(), t, core.ID(strings.TrimSuffix(command, ":review")), key, in)
|
|
return out, 200, err
|
|
}
|
|
if strings.HasSuffix(command, ":retry") {
|
|
var in service.ManualRetryInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.RetryTask(r.Context(), t, core.ID(strings.TrimSuffix(command, ":retry")), key, in)
|
|
return out, createdStatus(out.Deduplicated), err
|
|
}
|
|
if !strings.HasSuffix(command, ":cancel") {
|
|
return nil, 0, fault.NotFound
|
|
}
|
|
var in service.CancelInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.CancelTask(r.Context(), t, core.ID(strings.TrimSuffix(command, ":cancel")), key, in)
|
|
status := 200
|
|
if out.Task.State == pulse.TaskRunning {
|
|
status = 202
|
|
}
|
|
return out, status, err
|
|
}))
|
|
mux.HandleFunc("POST /v1/queues", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
var in service.QueueInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.CreateQueue(r.Context(), t, key, in)
|
|
return out, createdStatus(out.Deduplicated), err
|
|
}))
|
|
mux.HandleFunc("GET /v1/queues/{id}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
out, err := s.GetQueue(r.Context(), t, core.ID(r.PathValue("id")))
|
|
return out, 200, err
|
|
}))
|
|
mux.HandleFunc("POST /v1/queues/{command}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
|
|
command := r.PathValue("command")
|
|
if strings.HasSuffix(command, ":retire") {
|
|
if err := noEventQuery(r); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
var in service.RetireQueueInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.RetireQueue(r.Context(), t, core.ID(strings.TrimSuffix(command, ":retire")), key, in)
|
|
return out, 200, err
|
|
}
|
|
if !strings.HasSuffix(command, ":configure") {
|
|
return nil, 0, fault.NotFound
|
|
}
|
|
var in service.QueueConfigInput
|
|
if err := decodeBody(r, &in); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
// 完整替换命令要求 paused 显式给出，避免遗漏被解释为解除暂停。
|
|
key, err := idempotencyKey(r)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
out, err := s.ConfigureQueue(r.Context(), t, core.ID(strings.TrimSuffix(command, ":configure")), key, in)
|
|
return out, 200, err
|
|
}))
|
|
}
|
|
func createdStatus(duplicate bool) int {
|
|
if duplicate {
|
|
return http.StatusOK
|
|
}
|
|
return http.StatusCreated
|
|
}
|
|
func idempotencyKey(r *http.Request) (string, error) {
|
|
values := r.Header.Values("Idempotency-Key")
|
|
if len(values) != 1 {
|
|
return "", fault.Invalid("exactly one Idempotency-Key is required")
|
|
}
|
|
return values[0], nil
|
|
}
|
|
func wrap(next endpoint) http.HandlerFunc {
|
|
return func(w http.ResponseWriter, r *http.Request) {
|
|
id, err := core.NewID()
|
|
if err != nil {
|
|
writeError(w, "", fault.Unavailable)
|
|
return
|
|
}
|
|
w.Header().Set("X-Request-ID", string(id))
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
w.Header().Set("X-Content-Type-Options", "nosniff")
|
|
values := r.Header.Values("Authorization")
|
|
var token auth.Token
|
|
if len(values) != 1 {
|
|
writeError(w, string(id), fault.Unauthorized)
|
|
return
|
|
}
|
|
scheme, value, ok := strings.Cut(values[0], " ")
|
|
if !ok || !strings.EqualFold(scheme, "Bearer") {
|
|
writeError(w, string(id), fault.Unauthorized)
|
|
return
|
|
}
|
|
token, err = auth.Parse(value)
|
|
if err != nil {
|
|
writeError(w, string(id), fault.Unauthorized)
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second)
|
|
defer cancel()
|
|
data, status, err := next(r.WithContext(ctx), scopedConsoleToken(r, token))
|
|
if err != nil {
|
|
if fault.Code(err) == "UNAVAILABLE" {
|
|
slog.Error("Orbit API operation failed", "request_id", id, "code", "UNAVAILABLE")
|
|
}
|
|
writeError(w, string(id), err)
|
|
return
|
|
}
|
|
writeJSON(w, status, map[string]any{"request_id": id, "data": data})
|
|
}
|
|
}
|
|
func writeJSON(w http.ResponseWriter, status int, body any) {
|
|
raw, err := jsonwire.Marshal(body)
|
|
if err != nil {
|
|
status = 503
|
|
raw = []byte(`{"error":{"code":"UNAVAILABLE","message":"operation unavailable"}}`)
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(status)
|
|
_, _ = w.Write(append(raw, '\n'))
|
|
}
|
|
func writeError(w http.ResponseWriter, id string, err error) {
|
|
var known *fault.Error
|
|
if !errors.As(err, &known) {
|
|
known = fault.Unavailable
|
|
}
|
|
status := map[string]int{"INVALID_ARGUMENT": 400, "UNAUTHENTICATED": 401, "PERMISSION_DENIED": 403, "NOT_FOUND": 404, "STATE_CONFLICT": 409, "RESOURCE_RETIRED": 409, "RESOURCE_BUSY": 409, "IDEMPOTENCY_CONFLICT": 409, "BODY_TOO_LARGE": 413, "UNSUPPORTED_MEDIA_TYPE": 415, "DELIVERY_LEASE_LOST": 409, "EVENT_HISTORY_EXPIRED": 410, "SUBSCRIPTION_LIMIT": 409, "WORKFLOW_DEFINITION_LIMIT": 409, "WORKFLOW_VERSION_LIMIT": 409, "WORKFLOW_INSTANCE_LIMIT": 409, "WORKFLOW_SIGNAL_LIMIT": 409}[known.Code]
|
|
if status == 0 {
|
|
status = 503
|
|
}
|
|
if known.Code == "QUEUE_FULL" || known.Code == "APPLICATION_FULL" {
|
|
status = http.StatusTooManyRequests
|
|
w.Header().Set("Retry-After", "1")
|
|
}
|
|
if status == 401 {
|
|
w.Header().Set("WWW-Authenticate", `Bearer realm="orbit"`)
|
|
}
|
|
writeJSON(w, status, map[string]any{"request_id": id, "error": known})
|
|
}
|
|
func decodeBody(r *http.Request, out any) error {
|
|
content, _, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
|
|
if err != nil || content != "application/json" {
|
|
return fault.New("UNSUPPORTED_MEDIA_TYPE", "Content-Type must be application/json")
|
|
}
|
|
const maxBody = 512 << 10
|
|
raw, err := io.ReadAll(io.LimitReader(r.Body, maxBody+1))
|
|
if err != nil {
|
|
return fault.Invalid("cannot read request body")
|
|
}
|
|
if len(raw) > maxBody {
|
|
return fault.New("BODY_TOO_LARGE", "request body exceeds 512 KiB")
|
|
}
|
|
if _, err := jsonvalue.Canonical(raw); err != nil {
|
|
return fault.Invalid("invalid JSON, duplicate keys, or unsupported depth/numeric range")
|
|
}
|
|
if err := exactFields(raw, reflect.TypeOf(out).Elem()); err != nil {
|
|
return fault.Invalid("unknown field, null, or invalid object structure")
|
|
}
|
|
decoder := json.NewDecoder(bytes.NewReader(raw))
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(out); err != nil {
|
|
return fault.Invalid("invalid field value or type")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 精确匹配结构字段，防止 encoding/json 将 Handler 和 handler 解释为同一个字段。
|
|
func exactFields(raw []byte, t reflect.Type) error {
|
|
if bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
|
|
if t.Kind() == reflect.Pointer {
|
|
return nil
|
|
}
|
|
return errors.New("unexpected null")
|
|
}
|
|
if t.Kind() == reflect.Pointer {
|
|
return exactFields(raw, t.Elem())
|
|
}
|
|
if t == reflect.TypeOf(time.Time{}) || t == reflect.TypeOf(json.RawMessage{}) {
|
|
return nil
|
|
}
|
|
if t.Kind() == reflect.Slice || t.Kind() == reflect.Array {
|
|
var values []json.RawMessage
|
|
if err := json.Unmarshal(raw, &values); err != nil {
|
|
return err
|
|
}
|
|
for _, value := range values {
|
|
if err := exactFields(value, t.Elem()); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
if t.Kind() == reflect.Map {
|
|
var values map[string]json.RawMessage
|
|
if err := json.Unmarshal(raw, &values); err != nil {
|
|
return err
|
|
}
|
|
for _, value := range values {
|
|
if err := exactFields(value, t.Elem()); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
if t.Kind() != reflect.Struct {
|
|
return nil
|
|
}
|
|
var object map[string]json.RawMessage
|
|
if err := json.Unmarshal(raw, &object); err != nil {
|
|
return err
|
|
}
|
|
names := map[string]reflect.Type{}
|
|
nullable := map[string]bool{}
|
|
var collect func(reflect.Type) error
|
|
collect = func(typ reflect.Type) error {
|
|
for i := 0; i < typ.NumField(); i++ {
|
|
field := typ.Field(i)
|
|
name := strings.Split(field.Tag.Get("json"), ",")[0]
|
|
if field.Anonymous && name == "" && field.Type.Kind() == reflect.Struct {
|
|
if err := collect(field.Type); err != nil {
|
|
return err
|
|
}
|
|
continue
|
|
}
|
|
if name == "" || name == "-" {
|
|
continue
|
|
}
|
|
if _, exists := names[name]; exists {
|
|
return errors.New("ambiguous embedded JSON field")
|
|
}
|
|
names[name] = field.Type
|
|
nullable[name] = field.Tag.Get("nullable") == "true"
|
|
if field.Tag.Get("required") == "true" {
|
|
if _, ok := object[name]; !ok {
|
|
return errors.New("missing required field")
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
if err := collect(t); err != nil {
|
|
return err
|
|
}
|
|
|
|
for name, value := range object {
|
|
field, ok := names[name]
|
|
if !ok {
|
|
return errors.New("unknown field")
|
|
}
|
|
if nullable[name] && bytes.Equal(bytes.TrimSpace(value), []byte("null")) {
|
|
continue
|
|
}
|
|
if err := exactFields(value, field); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|