AI 视频生成 API 接入：异步任务与生产可靠性

单条可交付视频成本 = 所有 Attempt 实际费用之和 / ready Job 数
每可交付秒成本 = 所有 Attempt 实际费用之和 / ready Asset 总时长
有效产出率 = ready Job 数 / 可计费 Attempt 数

package videoworkflow

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"crypto/subtle"
	"encoding/base64"
	"errors"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

var (
	ErrOutcomeUnknown = errors.New("provider may have accepted the request")
	ErrRemoteNotFound = errors.New("provider job not found")
	ErrInvalidState   = errors.New("invalid state transition")
)

type State string

const (
	StateAccepted      State = "accepted"
	StateDispatching   State = "dispatching"
	StateSubmitUnknown State = "submit_unknown"
	StateSubmitted     State = "submitted"
	StateRunning       State = "running"
	StateSucceeded     State = "succeeded"
	StateReady         State = "ready"
	StateFailed        State = "failed"
)

var stateOrder = map[State]int{
	StateAccepted:      10,
	StateDispatching:   20,
	StateSubmitUnknown: 25,
	StateSubmitted:     30,
	StateRunning:       40,
	StateSucceeded:     50,
	StateReady:         60,
	StateFailed:        60,
}

type Job struct {
	ID             string
	IdempotencyKey string
	RemoteID       string
	State          State
	Version        int64
	Failure        string
}

type Provider interface {
	Submit(context.Context, string) (string, error)
	LookupByIdempotencyKey(context.Context, string) (string, bool, error)
}

type Controller struct {
	mu          sync.Mutex
	provider    Provider
	jobs        map[string]Job
	jobByRemote map[string]string
	seenEvents  map[string]struct{}
}

func NewController(provider Provider) *Controller {
	return &Controller{
		provider:    provider,
		jobs:        make(map[string]Job),
		jobByRemote: make(map[string]string),
		seenEvents:  make(map[string]struct{}),
	}
}

func (c *Controller) Create(jobID string) Job {
	c.mu.Lock()
	defer c.mu.Unlock()
	if job, ok := c.jobs[jobID]; ok {
		return job
	}

	job := Job{
		ID:             jobID,
		IdempotencyKey: "video:" + jobID + ":attempt:1",
		State:          StateAccepted,
		Version:        1,
	}
	c.jobs[jobID] = job
	return job
}

func (c *Controller) Dispatch(ctx context.Context, jobID string) error {
	c.mu.Lock()
	job, ok := c.jobs[jobID]
	if !ok || job.State != StateAccepted {
		c.mu.Unlock()
		return ErrInvalidState
	}
	job.State = StateDispatching
	job.Version++
	c.jobs[jobID] = job
	c.mu.Unlock()

	remoteID, err := c.provider.Submit(ctx, job.IdempotencyKey)

	c.mu.Lock()
	defer c.mu.Unlock()
	job = c.jobs[jobID]
	if errors.Is(err, ErrOutcomeUnknown) {
		job.State = StateSubmitUnknown
		job.Failure = err.Error()
		job.Version++
		c.jobs[jobID] = job
		return err
	}
	if err != nil {
		job.State = StateFailed
		job.Failure = err.Error()
		job.Version++
		c.jobs[jobID] = job
		return err
	}

	c.bindRemote(&job, remoteID)
	c.jobs[jobID] = job
	return nil
}

func (c *Controller) ReconcileUnknownSubmit(
	ctx context.Context,
	jobID string,
) error {
	c.mu.Lock()
	job, ok := c.jobs[jobID]
	c.mu.Unlock()
	if !ok || job.State != StateSubmitUnknown {
		return ErrInvalidState
	}

	remoteID, found, err := c.provider.LookupByIdempotencyKey(
		ctx,
		job.IdempotencyKey,
	)
	if err != nil {
		return err
	}
	if !found {
		return ErrRemoteNotFound
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	job = c.jobs[jobID]
	if job.State != StateSubmitUnknown {
		return nil
	}
	c.bindRemote(&job, remoteID)
	c.jobs[jobID] = job
	return nil
}

func (c *Controller) bindRemote(job *Job, remoteID string) {
	job.RemoteID = remoteID
	job.State = StateSubmitted
	job.Failure = ""
	job.Version++
	c.jobByRemote[remoteID] = job.ID
}

func (c *Controller) ApplyProviderEvent(
	eventID string,
	remoteID string,
	incoming State,
) error {
	if incoming != StateRunning &&
		incoming != StateSucceeded &&
		incoming != StateFailed {
		return ErrInvalidState
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	jobID, ok := c.jobByRemote[remoteID]
	if !ok {
		return ErrRemoteNotFound
	}
	if _, duplicate := c.seenEvents[eventID]; duplicate {
		return nil
	}
	c.seenEvents[eventID] = struct{}{}

	job := c.jobs[jobID]
	if job.State == StateReady ||
		job.State == StateSucceeded ||
		job.State == StateFailed {
		return nil
	}
	if stateOrder[incoming] <= stateOrder[job.State] {
		return nil
	}
	job.State = incoming
	job.Version++
	c.jobs[jobID] = job
	return nil
}

func VerifyStandardWebhook(
	secret string,
	header http.Header,
	body []byte,
	now time.Time,
	tolerance time.Duration,
) error {
	messageID := header.Get("webhook-id")
	timestamp := header.Get("webhook-timestamp")
	signatures := header.Get("webhook-signature")
	if messageID == "" || timestamp == "" || signatures == "" {
		return errors.New("missing webhook signature headers")
	}

	seconds, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return errors.New("invalid webhook timestamp")
	}
	delta := now.Sub(time.Unix(seconds, 0))
	if delta < -tolerance || delta > tolerance {
		return errors.New("webhook timestamp outside tolerance")
	}

	encodedSecret := strings.TrimPrefix(secret, "whsec_")
	key, err := base64.StdEncoding.DecodeString(encodedSecret)
	if err != nil {
		return errors.New("invalid webhook secret")
	}

	signed := make([]byte, 0, len(messageID)+len(timestamp)+len(body)+2)
	signed = append(signed, messageID...)
	signed = append(signed, '.')
	signed = append(signed, timestamp...)
	signed = append(signed, '.')
	signed = append(signed, body...)

	mac := hmac.New(sha256.New, key)
	_, _ = mac.Write(signed)
	expected := mac.Sum(nil)

	for _, candidate := range strings.Fields(signatures) {
		versionAndValue := strings.SplitN(candidate, ",", 2)
		if len(versionAndValue) != 2 || versionAndValue[0] != "v1" {
			continue
		}
		actual, decodeErr := base64.StdEncoding.DecodeString(versionAndValue[1])
		if decodeErr == nil &&
			subtle.ConstantTimeCompare(actual, expected) == 1 {
			return nil
		}
	}
	return errors.New("webhook signature mismatch")
}