完善生图5.0的功能

This commit is contained in:
andy committed 2026-08-25 14:06:04 +08:00
1 parent 88a4268d40
commit 678d213f40
31 files changed
+3281 -107

No files matched your search

+76 -7
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"log"
"strings"
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/billing"
@@ -161,8 +162,15 @@ func (c *CreationCoordinator) charge(ctx context.Context, job jobs.Job, reused b
return c.writeCharge(ctx, job, reused, charge.raw)
}
var use usageContext
if json.Unmarshal(job.UsageContext, &use) != nil || use.OrganizationID == "" || charge.AmountFen <= 0 {
return c.failCharge(ctx, job, reused, errors.New("charge generation job"), charge.raw)
if json.Unmarshal(job.UsageContext, &use) != nil || charge.AmountFen <= 0 {
cause := billing.ErrProviderUnavailable
logGenerationBillingFailure(job.ID, "validate", cause, cause)
return c.failCharge(ctx, job, reused, cause, charge.raw)
}
if use.OrganizationID == "" {
cause := &billing.StatusError{Status: 409, Err: billing.ErrOrganizationUnavailable}
logGenerationBillingFailure(job.ID, "validate", cause, cause)
return c.failCharge(ctx, job, reused, cause, charge.raw)
}
request := billing.ChargeRequest{OrganizationID: use.OrganizationID, AccountID: use.AccountID, JobID: job.ID, AmountFen: charge.AmountFen, Description: capabilityLabel(job.Capability) + " · " + job.ReqKey, Metadata: map[string]any{"quote": cloneMap(charge.raw), "accountName": use.DisplayName, "organizationName": use.OrganizationName}}
if atomic, ok := c.state.(AtomicCreationCharger); ok {
@@ -173,16 +181,22 @@ func (c *CreationCoordinator) charge(ctx context.Context, job jobs.Job, reused b
encoded, atomicErr := atomic.ChargeAndActivateCreation(ctx, request, pending)
if atomicErr != nil {
if errors.Is(atomicErr, billing.ErrCommitOutcomeUnknown) {
return jobs.Job{}, reused, errors.New("charge generation job")
publicErr := errors.New("charge generation job")
logGenerationBillingFailure(job.ID, "commit", atomicErr, publicErr)
return jobs.Job{}, reused, publicErr
}
return c.failCharge(ctx, job, reused, safeBillingError(atomicErr), charge.raw)
safeErr := safeBillingError(atomicErr)
logGenerationBillingFailure(job.ID, "charge", atomicErr, safeErr)
return c.failCharge(ctx, job, reused, safeErr, charge.raw)
}
job.Billing = encoded
return job, reused, nil
}
posting, err := c.charges.Charge(ctx, request)
if err != nil {
return c.failCharge(ctx, job, reused, safeBillingError(err), charge.raw)
safeErr := safeBillingError(err)
logGenerationBillingFailure(job.ID, "charge", err, safeErr)
return c.failCharge(ctx, job, reused, safeErr, charge.raw)
}
charge.raw["status"] = "charged"
charge.raw["ledgerEntryId"] = posting.LedgerID
@@ -205,7 +219,8 @@ func (c *CreationCoordinator) failCharge(ctx context.Context, job jobs.Job, reus
job.Status = jobs.StatusFailed
job.Error = &jobs.JobError{Message: "generation charge failed", Retryable: false}
if err := c.state.FailCreation(ctx, job); err != nil {
return jobs.Job{}, reused, errors.New("persist failed generation charge")
logGenerationBillingFailure(job.ID, "persist_failure", err, billing.ErrProviderUnavailable)
return jobs.Job{}, reused, billing.ErrProviderUnavailable
}
return job, reused, cause
}
@@ -231,10 +246,64 @@ func safeBillingError(err error) error {
if strings.Contains(message, "BILLING_IDEMPOTENCY_PAYLOAD_MISMATCH") {
return &billing.StatusError{Status: 409, Err: billing.ErrIdempotencyConflict}
}
return errors.New("charge generation job")
return billing.ErrProviderUnavailable
}
}
func logGenerationBillingFailure(jobID, phase string, rawErr, publicErr error) {
log.Printf(
"zhinian-api generation billing failed jobId=%s phase=%s status=%d code=%q errorClass=%s",
jobID,
phase,
billing.HTTPStatus(publicErr),
billingSQLState(rawErr),
generationBillingErrorClass(rawErr, publicErr),
)
}
func generationBillingErrorClass(rawErr, publicErr error) string {
switch {
case errors.Is(publicErr, billing.ErrInsufficientBalance):
return "insufficient_balance"
case errors.Is(publicErr, billing.ErrIdempotencyConflict):
return "idempotency_conflict"
case errors.Is(publicErr, billing.ErrOrganizationUnavailable):
return "organization_unavailable"
case errors.Is(rawErr, billing.ErrCommitOutcomeUnknown):
return "commit_outcome_unknown"
case errors.Is(rawErr, context.DeadlineExceeded):
return "timeout"
case errors.Is(rawErr, context.Canceled):
return "canceled"
case billingSQLState(rawErr) != "":
return "database"
case errors.Is(publicErr, billing.ErrProviderUnavailable):
return "unavailable"
default:
return "unknown"
}
}
func billingSQLState(err error) string {
type sqlStateCarrier interface{ SQLState() string }
var carrier sqlStateCarrier
if !errors.As(err, &carrier) {
return ""
}
code := strings.TrimSpace(carrier.SQLState())
if len(code) != 5 {
return ""
}
for _, char := range code {
isDigit := char >= '0' && char <= '9'
isUppercase := char >= 'A' && char <= 'Z'
if !isDigit && !isUppercase {
return ""
}
}
return code
}
func retryEngine(provider, capability string) string {
if capability == "video.generate" {
if provider == "seedance" || provider == "bailian" {
@@ -1,10 +1,13 @@
package orchestration
import (
"bytes"
"context"
"encoding/json"
"errors"
"log"
"reflect"
"strings"
"testing"
"time"
@@ -233,6 +236,30 @@ func TestCreationCoordinatorUsesAtomicChargeAndActivationWhenStoreSupportsIt(t *
}
}
func TestCreationCoordinatorMapsAtomicInsufficientBalanceBeforeDispatch(t *testing.T) {
creator := &creationStoreStub{}
quoter := &quoterStub{quote: &billing.Quote{AmountFen: 39, Currency: "CNY"}}
state := &atomicCreationStateStub{err: errors.New("charge and activate: BILLING_INSUFFICIENT_BALANCE")}
coordinator := NewCreationCoordinator(platformBuilderStub{}, creator, quoter, &chargeLedgerStub{}, state)
_, _, err := coordinator.CreatePlatform(context.Background(), identity.Session{User: identity.User{ID: "account", OrganizationID: "org"}}, CreationInput{Capability: "image.generate", Body: map[string]any{"prompt": "fresh"}})
if billing.HTTPStatus(err) != 402 || state.failed.Status != jobs.StatusFailed {
t.Fatalf("error=%v status=%d failed=%#v", err, billing.HTTPStatus(err), state.failed)
}
}
func TestCreationCoordinatorReturnsActionableErrorWithoutBillingOrganization(t *testing.T) {
creator := &creationStoreStub{}
quoter := &quoterStub{quote: &billing.Quote{AmountFen: 39, Currency: "CNY"}}
state := &creationStateStub{}
coordinator := NewCreationCoordinator(platformBuilderStub{}, creator, quoter, &chargeLedgerStub{}, state)
_, _, err := coordinator.CreatePlatform(context.Background(), identity.Session{User: identity.User{ID: "account", Role: "user"}}, CreationInput{Capability: "image.generate", Body: map[string]any{"prompt": "fresh"}})
if billing.HTTPStatus(err) != 409 || state.failed.Status != jobs.StatusFailed {
t.Fatalf("error=%v status=%d failed=%#v", err, billing.HTTPStatus(err), state.failed)
}
}
func TestCreationCoordinatorRetryRebuildsAnyOwnedImageJobWithFreshQuote(t *testing.T) {
creator := &creationStoreStub{}
quoter := &quoterStub{quote: &billing.Quote{AmountFen: 41, Currency: "CNY"}}
@@ -266,7 +293,7 @@ func TestSafeBillingErrorMapsPostgresWalletFailures(t *testing.T) {
}{
{name: "insufficient", err: errors.New("charge and activate: BILLING_INSUFFICIENT_BALANCE"), want: 402},
{name: "idempotency", err: errors.New("charge and activate: BILLING_IDEMPOTENCY_PAYLOAD_MISMATCH"), want: 409},
{name: "unknown", err: errors.New("database unavailable"), want: 500},
{name: "unknown", err: errors.New("database unavailable"), want: 503},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
@@ -274,13 +301,40 @@ func TestSafeBillingErrorMapsPostgresWalletFailures(t *testing.T) {
if status := billing.HTTPStatus(got); status != test.want {
t.Fatalf("status = %d, want %d (error %v)", status, test.want, got)
}
if test.want == 500 && got.Error() != "charge generation job" {
if test.want == 503 && got.Error() == test.err.Error() {
t.Fatalf("unknown error leaked: %v", got)
}
})
}
}
func TestGenerationBillingFailureLogKeepsOnlySafeClassification(t *testing.T) {
var output bytes.Buffer
previousWriter, previousFlags := log.Writer(), log.Flags()
log.SetOutput(&output)
log.SetFlags(0)
defer func() {
log.SetOutput(previousWriter)
log.SetFlags(previousFlags)
}()
logGenerationBillingFailure(
"job-safe-log",
"charge",
&billingSQLStateTestError{code: "42501", message: "database password leaked"},
billing.ErrProviderUnavailable,
)
got := output.String()
for _, expected := range []string{"jobId=job-safe-log", "phase=charge", "status=503", `code="42501"`, "errorClass=database"} {
if !strings.Contains(got, expected) {
t.Fatalf("safe log %q missing %q", got, expected)
}
}
if strings.Contains(got, "password") || strings.Contains(got, "leaked") {
t.Fatalf("safe log leaked raw error: %q", got)
}
}
func TestCreationCoordinatorPublicCreationNeverQuotesOrCharges(t *testing.T) {
creator := &creationStoreStub{}
quoter := &quoterStub{}
@@ -457,12 +511,21 @@ type creationStateStub struct {
type atomicCreationStateStub struct {
creationStateStub
charged json.RawMessage
err error
calls int
}
type billingSQLStateTestError struct {
code string
message string
}
func (e *billingSQLStateTestError) Error() string { return e.message }
func (e *billingSQLStateTestError) SQLState() string { return e.code }
func (s *atomicCreationStateStub) ChargeAndActivateCreation(_ context.Context, _ billing.ChargeRequest, _ json.RawMessage) (json.RawMessage, error) {
s.calls++
return append(json.RawMessage(nil), s.charged...), nil
return append(json.RawMessage(nil), s.charged...), s.err
}
func (s *creationStateStub) WriteBilling(_ context.Context, _ string, _ json.RawMessage) error {
+129 -5
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"strconv"
"strings"
"time"
@@ -23,8 +24,9 @@ type FencedSettlementStateWriter interface {
WriteBillingFenced(context.Context, string, json.RawMessage, jobs.Status, string) error
}
// SettlementProcessor adds Seedance actual-usage settlement to a jobs
// processor. Other providers and non-successful advances pass through.
// SettlementProcessor adds actual-usage settlement for Seedance videos and
// Seedream layer decomposition. Other providers and non-successful advances
// pass through.
type SettlementProcessor struct {
inner jobs.Processor
ledger SettlementLedger
@@ -44,9 +46,15 @@ func (p *SettlementProcessor) Advance(ctx context.Context, job jobs.Job) (jobs.J
return jobs.Job{}, errors.New("advance generation job")
}
advanced, err := p.inner.Advance(ctx, job)
if err != nil || advanced.Provider != "seedance" || advanced.Status != jobs.StatusSucceeded || len(advanced.Billing) == 0 {
if err != nil || advanced.Status != jobs.StatusSucceeded || len(advanced.Billing) == 0 {
return advanced, err
}
if advanced.Provider == "seedream" && seedreamLayerRequest(advanced.RequestPayload) {
return p.settleSeedreamLayers(ctx, advanced)
}
if advanced.Provider != "seedance" {
return advanced, nil
}
var charge billingState
if json.Unmarshal(advanced.Billing, &charge) != nil {
@@ -131,13 +139,84 @@ func (p *SettlementProcessor) Advance(ctx context.Context, job jobs.Job) (jobs.J
return p.write(ctx, advanced, charge.raw)
}
func (p *SettlementProcessor) settleSeedreamLayers(ctx context.Context, job jobs.Job) (jobs.Job, error) {
var charge billingState
if json.Unmarshal(job.Billing, &charge) != nil {
return jobs.Job{}, errors.New("settle seedream layer charge")
}
status, _ := charge.raw["settlementStatus"].(string)
if status == "settled" || status == "estimated" {
return job, nil
}
var use usageContext
if len(job.UsageContext) != 0 && json.Unmarshal(job.UsageContext, &use) != nil {
return jobs.Job{}, errors.New("settle seedream layer charge")
}
quotaExempt := charge.QuotaExempt || use.Source == "platform" && use.Role == "super_admin"
chargeReady := charge.Status == "charged" || quotaExempt && charge.Status == "not_charged"
if !chargeReady || !quotaExempt && use.OrganizationID == "" {
return job, nil
}
images := seedreamLayerImages(job.ResponsePayload)
if len(images) == 0 {
charge.raw["settlementStatus"] = "estimated"
charge.raw["settlementReason"] = "provider_layer_metadata_unavailable"
charge.raw["settledAt"] = p.now().UTC().Format(time.RFC3339Nano)
return p.write(ctx, job, charge.raw)
}
actualAmount, err := billing.CalculateSeedreamLayerAmountFen(images, numberOrZero(charge.raw["markupMultiplier"]))
if err != nil {
return jobs.Job{}, errors.New("settle seedream layer charge")
}
delta := actualAmount - charge.AmountFen
settledAt := p.now().UTC()
var settlementLedgerID string
if delta != 0 && !quotaExempt {
if p.ledger == nil {
return jobs.Job{}, errors.New("settle seedream layer charge")
}
description := capabilityLabel(job.Capability) + "图层拆分实际费用退回"
if delta > 0 {
description = capabilityLabel(job.Capability) + "图层拆分实际费用补扣"
}
posting, postErr := p.ledger.Settle(ctx, billing.SettlementRequest{
OrganizationID: use.OrganizationID, AccountID: use.AccountID, JobID: job.ID,
DeltaFen: delta, Description: description,
Metadata: map[string]any{
"operation": "seedream_layer_actual_settlement", "reservedAmountFen": charge.AmountFen,
"actualAmountFen": actualAmount, "outputCount": len(images), "model": job.ReqKey,
},
})
if postErr != nil {
return jobs.Job{}, errors.New("settle seedream layer charge")
}
settlementLedgerID = posting.LedgerID
if !posting.CreatedAt.IsZero() {
settledAt = posting.CreatedAt.UTC()
}
}
charge.raw["amountFen"] = actualAmount
charge.raw["settlementStatus"] = "settled"
charge.raw["settledAt"] = settledAt.Format(time.RFC3339Nano)
if settlementLedgerID != "" {
charge.raw["settlementLedgerEntryId"] = settlementLedgerID
} else {
delete(charge.raw, "settlementLedgerEntryId")
}
charge.raw["providerUsage"] = map[string]any{
"model": job.ReqKey, "layerDecomposition": true, "generatedImages": len(images),
}
return p.write(ctx, job, charge.raw)
}
func (p *SettlementProcessor) write(ctx context.Context, job jobs.Job, snapshot map[string]any) (jobs.Job, error) {
if p.state == nil {
return jobs.Job{}, errors.New("persist seedance generation settlement")
return jobs.Job{}, errors.New("persist generation settlement")
}
encoded, err := json.Marshal(snapshot)
if err != nil {
return jobs.Job{}, errors.New("persist seedance generation settlement")
return jobs.Job{}, errors.New("persist generation settlement")
}
var writeErr error
if fenced, ok := p.state.(FencedSettlementStateWriter); ok && job.LockedBy != "" {
@@ -163,6 +242,51 @@ func seedanceCompletionTokens(payload json.RawMessage) int64 {
return value
}
func seedreamLayerRequest(payload json.RawMessage) bool {
var request struct {
Settings map[string]any `json:"settings"`
}
if json.Unmarshal(payload, &request) != nil {
return false
}
value, _ := request.Settings["layerDecomposition"].(bool)
creationMode, _ := request.Settings["creationMode"].(string)
return value || strings.EqualFold(strings.TrimSpace(creationMode), "layers")
}
func seedreamLayerImages(payload json.RawMessage) []billing.SeedreamLayerImage {
var response struct {
Raw json.RawMessage `json:"raw"`
}
if json.Unmarshal(payload, &response) != nil || len(response.Raw) == 0 {
return nil
}
var providerResponse struct {
Data []struct {
Size string `json:"size"`
} `json:"data"`
}
if json.Unmarshal(response.Raw, &providerResponse) != nil || len(providerResponse.Data) == 0 || len(providerResponse.Data) > 17 {
return nil
}
images := make([]billing.SeedreamLayerImage, 0, len(providerResponse.Data))
for _, output := range providerResponse.Data {
parts := strings.FieldsFunc(strings.TrimSpace(output.Size), func(character rune) bool {
return character == 'x' || character == 'X' || character == '×' || character == '*'
})
if len(parts) != 2 {
return nil
}
width, widthErr := strconv.Atoi(strings.TrimSpace(parts[0]))
height, heightErr := strconv.Atoi(strings.TrimSpace(parts[1]))
if widthErr != nil || heightErr != nil || width <= 0 || height <= 0 {
return nil
}
images = append(images, billing.SeedreamLayerImage{Width: width, Height: height})
}
return images
}
func seedanceResolution(snapshot map[string]any) string {
parameters, _ := snapshot["parameters"].(map[string]any)
resolution, _ := parameters["resolution"].(string)
@@ -102,6 +102,28 @@ func TestSettlementProcessorLeavesNonSeedanceSuccessAlone(t *testing.T) {
}
}
func TestSettlementProcessorRefundsUnusedSeedreamLayerReservation(t *testing.T) {
state := &settlementStateStub{}
ledger := &settlementLedgerStub{posting: billing.WalletPosting{LedgerID: "layer-settlement", CreatedAt: time.Date(2026, 8, 21, 8, 0, 0, 0, time.UTC)}}
advanced := jobs.Job{
ID: "layer-job", Provider: "seedream", ReqKey: billing.Seedream50ProModel, Capability: "image.generate", Status: jobs.StatusSucceeded,
RequestPayload: json.RawMessage(`{"settings":{"creationMode":"layers","layerDecomposition":true,"size":"2K"}}`),
ResponsePayload: json.RawMessage(`{"status":"succeeded","raw":{"data":[{"size":"2048x2048","z_index":0},{"size":"1273x265","z_index":1}]}}`),
Billing: json.RawMessage(`{"status":"charged","amountFen":612,"reservedAmountFen":612,"markupMultiplier":1.2,"settlementStatus":"pending"}`),
UsageContext: json.RawMessage(`{"organizationId":"org-1","accountId":"account-1"}`),
}
processor := NewSettlementProcessor(settlementProcessorStub{job: advanced}, ledger, state, time.Now)
got, err := processor.Advance(context.Background(), jobs.Job{ID: "layer-job", Provider: "seedream", Status: jobs.StatusRunning})
if err != nil || ledger.calls != 1 || ledger.request.DeltaFen != -558 || ledger.request.Metadata["actualAmountFen"] != int64(54) {
t.Fatalf("job=%#v request=%#v calls=%d err=%v", got, ledger.request, ledger.calls, err)
}
var snapshot map[string]any
_ = json.Unmarshal(got.Billing, &snapshot)
if snapshot["amountFen"] != float64(54) || snapshot["settlementStatus"] != "settled" || snapshot["settlementLedgerEntryId"] != "layer-settlement" {
t.Fatalf("billing snapshot=%#v", snapshot)
}
}
func seedanceSettlementJob(response, billingJSON string) jobs.Job {
return jobs.Job{
ID: "job-1", Provider: "seedance", Capability: "video.generate", Status: jobs.StatusSucceeded,