niuniu/sub2api-cn-relay-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(control-plane): harden host-scoped reconcile and acceptance evidence

Browse Source 
- add batch-scoped reconcile_runs persistence and queries
- route batch detail and reconcile writes through batch_id/host_id
- refresh production boards with host-scope acceptance artifacts
- include latest real-host acceptance evidence for self_service and subscription
This commit is contained in:
phamnazage-jpg
2026-05-18 22:22:22 +08:00
parent 71cbaf5fa6
commit 85d495dd16
332 changed files with 5561 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
internal/host/sub2api/accounts.go
View File

@@ -76,13 +76,53 @@ func decodeAccountRefs(body []byte) ([]AccountRef, error) {
var wrapper struct {
Data struct {
Items []AccountRef `json:"items"`
Items []AccountRef `json:"items"`
Results []struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
Success bool `json:"success"`
} `json:"results"`
} `json:"data"`
}
if err := json.Unmarshal(body, &wrapper); err != nil {
if err := json.Unmarshal(body, &wrapper); err == nil {
if len(wrapper.Data.Items) > 0 {
return wrapper.Data.Items, nil
}
if len(wrapper.Data.Results) > 0 {
return decodeBatchAccountResults(wrapper.Data.Results)
}
}
var batch struct {
Results []struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
Success bool `json:"success"`
} `json:"results"`
}
if err := json.Unmarshal(body, &batch); err != nil {
return nil, err
}
return wrapper.Data.Items, nil
return decodeBatchAccountResults(batch.Results)
}
func decodeBatchAccountResults(results []struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
Success bool `json:"success"`
}) ([]AccountRef, error) {
refs := make([]AccountRef, 0, len(results))
for _, item := range results {
if !item.Success {
continue
}
id, err := decodeFlexibleID(item.ID)
if err != nil {
return nil, err
}
refs = append(refs, AccountRef{ID: id, Name: item.Name})
}
return refs, nil
}
func decodeAccountModels(body []byte) ([]AccountModel, error) {

16
internal/host/sub2api/capability_probe.go
View File

@@ -8,27 +8,27 @@ import (
)
func (c *Client) ProbeCapabilities(ctx context.Context) (HostCapabilities, error) {
groups, err := c.probeEndpoint(ctx, http.MethodPost, "/api/v1/admin/groups", map[string]any{})
groups, err := c.probeEndpoint(ctx, http.MethodGet, "/api/v1/admin/groups", nil)
if err != nil {
return HostCapabilities{}, err
}
channels, err := c.probeEndpoint(ctx, http.MethodPost, "/api/v1/admin/channels", map[string]any{})
channels, err := c.probeEndpoint(ctx, http.MethodGet, "/api/v1/admin/channels", nil)
if err != nil {
return HostCapabilities{}, err
}
plans, err := c.probeEndpoint(ctx, http.MethodPost, "/api/v1/admin/payment/plans", map[string]any{})
plans, err := c.probeEndpoint(ctx, http.MethodGet, "/api/v1/admin/payment/plans", nil)
if err != nil {
return HostCapabilities{}, err
}
accounts, err := c.probeEndpoint(ctx, http.MethodPost, "/api/v1/admin/accounts", map[string]any{})
accounts, err := c.probeEndpoint(ctx, http.MethodGet, "/api/v1/admin/accounts", nil)
if err != nil {
return HostCapabilities{}, err
}
accountTest, err := c.probeEndpoint(ctx, http.MethodPost, "/api/v1/admin/accounts/__probe__/test", map[string]any{})
accountTest, err := c.probeEndpoint(ctx, http.MethodGet, "/api/v1/admin/accounts/__probe__/test", nil)
if err != nil {
return HostCapabilities{}, err
}
@@ -38,7 +38,7 @@ func (c *Client) ProbeCapabilities(ctx context.Context) (HostCapabilities, error
return HostCapabilities{}, err
}
subscriptions, err := c.probeEndpoint(ctx, http.MethodPost, "/api/v1/admin/subscriptions/assign", map[string]any{})
subscriptions, err := c.probeEndpoint(ctx, http.MethodGet, "/api/v1/admin/subscriptions/assign", nil)
if err != nil {
return HostCapabilities{}, err
}
@@ -55,7 +55,7 @@ func (c *Client) ProbeCapabilities(ctx context.Context) (HostCapabilities, error
}
func (c *Client) probeEndpoint(ctx context.Context, method, path string, requestBody any) (bool, error) {
statusCode, headers, body, err := c.perform(ctx, method, path, requestBody)
statusCode, _, body, err := c.perform(ctx, method, path, requestBody)
if err != nil {
return false, err
}
@@ -66,7 +66,7 @@ func (c *Client) probeEndpoint(ctx context.Context, method, path string, request
case statusCode == http.StatusUnauthorized || statusCode == http.StatusForbidden:
return false, newHTTPError(method, path, statusCode, body)
case statusCode == http.StatusNotFound || statusCode == http.StatusMethodNotAllowed:
return looksLikeExistingEndpoint(headers, body), nil
return false, nil
case statusCode >= http.StatusBadRequest && statusCode < http.StatusInternalServerError:
return true, nil
default:

8
internal/host/sub2api/client.go
View File

@@ -42,8 +42,10 @@ type HostCapabilities struct {
}
type CreateGroupRequest struct {
Name string `json:"name"`
RateMultiplier float64 `json:"rate_multiplier"`
Name string `json:"name"`
Platform string `json:"platform,omitempty"`
RateMultiplier float64 `json:"rate_multiplier"`
SubscriptionType string `json:"subscription_type,omitempty"`
}
type GroupRef struct {
@@ -108,7 +110,7 @@ type AccountModel struct {
type AssignSubscriptionRequest struct {
UserID string `json:"user_id"`
GroupID string `json:"group_id"`
DurationDays int `json:"duration_days,omitempty"`
DurationDays int `json:"validity_days,omitempty"`
}
type SubscriptionRef struct {

206
internal/host/sub2api/flexible_id.go Normal file
View File

@@ -0,0 +1,206 @@
package sub2api
import (
"bytes"
"encoding/json"
"fmt"
"strconv"
"strings"
)
func decodeFlexibleID(raw json.RawMessage) (string, error) {
if len(raw) == 0 || bytes.Equal(raw, []byte("null")) {
return "", nil
}
var asString string
if err := json.Unmarshal(raw, &asString); err == nil {
return asString, nil
}
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.UseNumber()
var asNumber json.Number
if err := decoder.Decode(&asNumber); err == nil {
return asNumber.String(), nil
}
return "", fmt.Errorf("unsupported id payload: %s", string(raw))
}
func flexibleIDValue(raw string) any {
trimmed := strings.TrimSpace(raw)
if trimmed == "" {
return ""
}
if id, err := strconv.ParseInt(trimmed, 10, 64); err == nil {
return id
}
return trimmed
}
func flexibleIDSliceValues(raw []string) []any {
values := make([]any, 0, len(raw))
for _, item := range raw {
values = append(values, flexibleIDValue(item))
}
return values
}
func (r CreateChannelRequest) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
Name string `json:"name"`
GroupIDs []any `json:"group_ids"`
}{
Name: r.Name,
GroupIDs: flexibleIDSliceValues(r.GroupIDs),
})
}
func (r CreatePlanRequest) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
GroupID any `json:"group_id"`
Name string `json:"name"`
Price float64 `json:"price"`
ValidityDays int `json:"validity_days"`
ValidityUnit string `json:"validity_unit"`
}{
GroupID: flexibleIDValue(r.GroupID),
Name: r.Name,
Price: r.Price,
ValidityDays: r.ValidityDays,
ValidityUnit: r.ValidityUnit,
})
}
func (r CreateAccountRequest) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
Name string `json:"name"`
Platform string `json:"platform"`
Type string `json:"type"`
Credentials map[string]any `json:"credentials"`
GroupIDs []any `json:"group_ids"`
}{
Name: r.Name,
Platform: r.Platform,
Type: r.Type,
Credentials: r.Credentials,
GroupIDs: flexibleIDSliceValues(r.GroupIDs),
})
}
func (r AssignSubscriptionRequest) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
UserID any `json:"user_id"`
GroupID any `json:"group_id"`
DurationDays int `json:"validity_days,omitempty"`
}{
UserID: flexibleIDValue(r.UserID),
GroupID: flexibleIDValue(r.GroupID),
DurationDays: r.DurationDays,
})
}
func (r *NamedResource) UnmarshalJSON(data []byte) error {
var aux struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
id, err := decodeFlexibleID(aux.ID)
if err != nil {
return err
}
r.ID = id
r.Name = aux.Name
return nil
}
func (r *GroupRef) UnmarshalJSON(data []byte) error {
var aux struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
id, err := decodeFlexibleID(aux.ID)
if err != nil {
return err
}
r.ID = id
r.Name = aux.Name
return nil
}
func (r *ChannelRef) UnmarshalJSON(data []byte) error {
var aux struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
id, err := decodeFlexibleID(aux.ID)
if err != nil {
return err
}
r.ID = id
r.Name = aux.Name
return nil
}
func (r *PlanRef) UnmarshalJSON(data []byte) error {
var aux struct {
ID json.RawMessage `json:"id"`
Name string `json:"name"`
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
id, err := decodeFlexibleID(aux.ID)
if err != nil {
return err
}
r.ID = id
r.Name = aux.Name
return nil
}
func (r *AccountRef) UnmarshalJSON(data []byte) error {
var aux struct {
ID json.RawMessage `json:"id"`
Name string `json:"name,omitempty"`
Platform string `json:"platform,omitempty"`
Type string `json:"type,omitempty"`
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
id, err := decodeFlexibleID(aux.ID)
if err != nil {
return err
}
r.ID = id
r.Name = aux.Name
r.Platform = aux.Platform
r.Type = aux.Type
return nil
}
func (r *SubscriptionRef) UnmarshalJSON(data []byte) error {
var aux struct {
ID json.RawMessage `json:"id"`
}
if err := json.Unmarshal(data, &aux); err != nil {
return err
}
id, err := decodeFlexibleID(aux.ID)
if err != nil {
return err
}
r.ID = id
return nil
}

185
internal/host/sub2api/sub2api_test.go
View File

@@ -2,6 +2,7 @@ package sub2api
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
@@ -234,6 +235,15 @@ func TestDecodeNamedResources(t *testing.T) {
t.Fatalf("got %+v", resources)
}
})
t.Run("numeric id", func(t *testing.T) {
resources, err := decodeNamedResources([]byte(`{"data":{"items":[{"id":1,"name":"default"}]}}`))
if err != nil {
t.Fatal(err)
}
if len(resources) != 1 || resources[0].ID != "1" {
t.Fatalf("got %+v", resources)
}
})
t.Run("wrapper with items", func(t *testing.T) {
resources, err := decodeNamedResources([]byte(`{"data":{"items":[{"id":"r2","name":"n2"}]}}`))
if err != nil {
@@ -261,6 +271,15 @@ func TestDecodeAccountRefs(t *testing.T) {
t.Fatalf("got %+v", refs)
}
})
t.Run("numeric id", func(t *testing.T) {
refs, err := decodeAccountRefs([]byte(`{"data":{"items":[{"id":42}]}}`))
if err != nil {
t.Fatal(err)
}
if len(refs) != 1 || refs[0].ID != "42" {
t.Fatalf("got %+v", refs)
}
})
t.Run("wrapper with items", func(t *testing.T) {
refs, err := decodeAccountRefs([]byte(`{"data":{"items":[{"id":"a2"}]}}`))
if err != nil {
@@ -270,6 +289,33 @@ func TestDecodeAccountRefs(t *testing.T) {
t.Fatalf("got %+v", refs)
}
})
t.Run("batch results", func(t *testing.T) {
refs, err := decodeAccountRefs([]byte(`{"success":1,"failed":0,"results":[{"name":"k1","id":123,"success":true}]}`))
if err != nil {
t.Fatal(err)
}
if len(refs) != 1 || refs[0].ID != "123" || refs[0].Name != "k1" {
t.Fatalf("got %+v", refs)
}
})
t.Run("batch results ignores failed items", func(t *testing.T) {
refs, err := decodeAccountRefs([]byte(`{"success":1,"failed":1,"results":[{"name":"k1","id":123,"success":true},{"name":"k2","id":456,"success":false}]}`))
if err != nil {
t.Fatal(err)
}
if len(refs) != 1 || refs[0].ID != "123" {
t.Fatalf("got %+v", refs)
}
})
t.Run("data wrapped batch results", func(t *testing.T) {
refs, err := decodeAccountRefs([]byte(`{"code":0,"message":"success","data":{"failed":0,"results":[{"id":5,"name":"deepseek-01","success":true}],"success":1}}`))
if err != nil {
t.Fatal(err)
}
if len(refs) != 1 || refs[0].ID != "5" || refs[0].Name != "deepseek-01" {
t.Fatalf("got %+v", refs)
}
})
t.Run("invalid json", func(t *testing.T) {
_, err := decodeAccountRefs([]byte(`not json`))
if err == nil {
@@ -439,7 +485,6 @@ func TestNewHTTPError(t *testing.T) {
}
}
func TestPerformWithMockServer(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
@@ -514,6 +559,18 @@ func TestPerformWithMockServer(t *testing.T) {
func TestCreateGroupWithMock(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var req struct {
Name string `json:"name"`
Platform string `json:"platform"`
RateMultiplier float64 `json:"rate_multiplier"`
SubscriptionType string `json:"subscription_type"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
if req.Name != "demo" || req.Platform != "openai" || req.RateMultiplier != 1.0 {
t.Fatalf("unexpected request: %+v", req)
}
w.Write([]byte(`{"data":{"id":"g1","name":"demo"}}`))
}))
defer srv.Close()
@@ -522,7 +579,7 @@ func TestCreateGroupWithMock(t *testing.T) {
if err != nil {
t.Fatal(err)
}
ref, err := client.CreateGroup(context.Background(), CreateGroupRequest{Name: "demo", RateMultiplier: 1.0})
ref, err := client.CreateGroup(context.Background(), CreateGroupRequest{Name: "demo", Platform: "openai", RateMultiplier: 1.0})
if err != nil {
t.Fatal(err)
}
@@ -533,26 +590,58 @@ func TestCreateGroupWithMock(t *testing.T) {
func TestCreateChannelWithMock(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`{"data":{"id":"c1","name":"ch"}}`))
var req struct {
Name string `json:"name"`
GroupIDs []int64 `json:"group_ids"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
if req.Name != "ch" {
t.Fatalf("name = %q, want ch", req.Name)
}
if len(req.GroupIDs) != 1 || req.GroupIDs[0] != 101 {
t.Fatalf("group_ids = %v, want [101]", req.GroupIDs)
}
w.Write([]byte(`{"data":{"id":201,"name":"ch"}}`))
}))
defer srv.Close()
client, _ := NewClient(srv.URL, WithAPIKey("k"))
_, err := client.CreateChannel(context.Background(), CreateChannelRequest{Name: "ch"})
ref, err := client.CreateChannel(context.Background(), CreateChannelRequest{Name: "ch", GroupIDs: []string{"101"}})
if err != nil {
t.Fatal(err)
}
if ref.ID != "201" {
t.Fatalf("id = %q, want 201", ref.ID)
}
}
func TestCreatePlanWithMock(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`{"data":{"id":"p1","name":"plan"}}`))
var req struct {
GroupID int64 `json:"group_id"`
Name string `json:"name"`
Price float64 `json:"price"`
ValidityDays int `json:"validity_days"`
ValidityUnit string `json:"validity_unit"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
if req.GroupID != 101 || req.Name != "plan" || req.Price != 19.9 || req.ValidityDays != 30 || req.ValidityUnit != "day" {
t.Fatalf("unexpected request: %+v", req)
}
w.Write([]byte(`{"data":{"id":301,"name":"plan"}}`))
}))
defer srv.Close()
client, _ := NewClient(srv.URL, WithAPIKey("k"))
_, err := client.CreatePlan(context.Background(), CreatePlanRequest{Name: "plan"})
ref, err := client.CreatePlan(context.Background(), CreatePlanRequest{GroupID: "101", Name: "plan", Price: 19.9, ValidityDays: 30, ValidityUnit: "day"})
if err != nil {
t.Fatal(err)
}
if ref.ID != "301" {
t.Fatalf("id = %q, want 301", ref.ID)
}
}
func TestDeleteWithMock(t *testing.T) {
@@ -586,15 +675,26 @@ func TestDeleteWithMock(t *testing.T) {
func TestAssignSubscriptionWithMock(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`{"data":{"id":"s1"}}`))
var req struct {
UserID int64 `json:"user_id"`
GroupID int64 `json:"group_id"`
DurationDays int `json:"validity_days"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
if req.UserID != 501 || req.GroupID != 101 || req.DurationDays != 30 {
t.Fatalf("unexpected request: %+v", req)
}
w.Write([]byte(`{"data":{"id":401}}`))
}))
defer srv.Close()
client, _ := NewClient(srv.URL, WithAPIKey("k"))
ref, err := client.AssignSubscription(context.Background(), AssignSubscriptionRequest{UserID: "u1"})
ref, err := client.AssignSubscription(context.Background(), AssignSubscriptionRequest{UserID: "501", GroupID: "101", DurationDays: 30})
if err != nil {
t.Fatal(err)
}
if ref.ID != "s1" {
if ref.ID != "401" {
t.Fatalf("id = %q", ref.ID)
}
}
@@ -619,17 +719,39 @@ func TestCheckGatewayAccessWithMock(t *testing.T) {
func TestBatchCreateAccountsWithMock(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`{"data":[{"id":"a1","name":"acct1"}]}`))
var req struct {
Accounts []struct {
Name string `json:"name"`
Platform string `json:"platform"`
Type string `json:"type"`
Credentials map[string]any `json:"credentials"`
GroupIDs []int64 `json:"group_ids"`
} `json:"accounts"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
if len(req.Accounts) != 1 {
t.Fatalf("accounts len = %d, want 1", len(req.Accounts))
}
acct := req.Accounts[0]
if acct.Name != "acct1" || acct.Platform != "openai" || acct.Type != "apikey" {
t.Fatalf("unexpected account metadata: %+v", acct)
}
if len(acct.GroupIDs) != 1 || acct.GroupIDs[0] != 101 {
t.Fatalf("group_ids = %v, want [101]", acct.GroupIDs)
}
w.Write([]byte(`{"data":[{"id":601,"name":"acct1"}]}`))
}))
defer srv.Close()
client, _ := NewClient(srv.URL, WithAPIKey("k"))
refs, err := client.BatchCreateAccounts(context.Background(), BatchCreateAccountsRequest{
Accounts: []CreateAccountRequest{{Name: "acct1"}},
Accounts: []CreateAccountRequest{{Name: "acct1", Platform: "openai", Type: "apikey", GroupIDs: []string{"101"}, Credentials: map[string]any{"api_key": "sk-test", "base_url": "https://api.example.com"}}},
})
if err != nil {
t.Fatal(err)
}
if len(refs) != 1 || refs[0].ID != "a1" {
if len(refs) != 1 || refs[0].ID != "601" {
t.Fatalf("got %+v", refs)
}
}
@@ -638,7 +760,9 @@ func TestProbeCapabilitiesWithMock(t *testing.T) {
callCount := 0
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
callCount++
w.WriteHeader(http.StatusCreated)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"data":[]}`))
}))
defer srv.Close()
client, _ := NewClient(srv.URL, WithAPIKey("k"))
@@ -649,6 +773,41 @@ func TestProbeCapabilitiesWithMock(t *testing.T) {
if !caps.Groups || !caps.Channels || !caps.Plans || !caps.Accounts || !caps.AccountTest || !caps.AccountModels || !caps.Subscriptions {
t.Fatalf("all capabilities should be true, got %+v", caps)
}
if callCount != 7 {
t.Fatalf("callCount = %d, want 7", callCount)
}
}
func TestProbeCapabilitiesDoesNotTreat404AsSupportForAccountOrSubscriptionRoutes(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/admin/groups", "/api/v1/admin/channels", "/api/v1/admin/payment/plans", "/api/v1/admin/accounts":
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"data":[]}`))
case "/api/v1/admin/accounts/__probe__/test", "/api/v1/admin/accounts/__probe__/models", "/api/v1/admin/subscriptions/assign":
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte(`{"error":"not found"}`))
default:
w.WriteHeader(http.StatusNotFound)
}
}))
defer srv.Close()
client, _ := NewClient(srv.URL, WithAPIKey("k"))
caps, err := client.ProbeCapabilities(context.Background())
if err != nil {
t.Fatalf("ProbeCapabilities() error = %v", err)
}
if caps.AccountTest {
t.Fatal("AccountTest = true, want false on 404 probe route")
}
if caps.AccountModels {
t.Fatal("AccountModels = true, want false on 404 probe route")
}
if caps.Subscriptions {
t.Fatal("Subscriptions = true, want false on 404 probe route")
}
}
func TestListManagedResourcesWithMock(t *testing.T) {