agent/cpa-plugin
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: reconcile current upstream accounts

Browse Source
This commit is contained in:
chuan
2026-08-21 01:05:00 +08:00
parent b2a79b80dc
commit 00f2da4a39
5 changed files with 71 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/plugin/access.go
+1 -1
View File
@@ -353,7 +353,7 @@ func (a *App) syncUpstreamAccounts(ctx context.Context) error {
Disabled: entry.Disabled, Unavailable: entry.Unavailable, Priority: entry.Priority, LastSeenAt: now,
})
}
return store.SyncUpstreamAccounts(ctx, accounts)
return store.ReconcileUpstreamAccounts(ctx, accounts)
}
func firstNonEmpty(values ...string) string {
internal/repository/sqlite_access.go
+20 -4
View File
@@ -282,11 +282,27 @@ func (r *SQLiteUsageRepository) managedKeyModels(ctx context.Context, keyID stri
}
func (r *SQLiteUsageRepository) SyncUpstreamAccounts(ctx context.Context, accounts []managedaccess.UpstreamAccount) error {
return r.syncUpstreamAccounts(ctx, accounts, false)
}
// ReconcileUpstreamAccounts treats accounts as the complete current CPA auth snapshot.
// Historical rows remain available for existing strict bindings, but are excluded from
// new route selection once CPA no longer reports them.
func (r *SQLiteUsageRepository) ReconcileUpstreamAccounts(ctx context.Context, accounts []managedaccess.UpstreamAccount) error {
return r.syncUpstreamAccounts(ctx, accounts, true)
}
func (r *SQLiteUsageRepository) syncUpstreamAccounts(ctx context.Context, accounts []managedaccess.UpstreamAccount, snapshot bool) error {
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("开始同步上游账号: %w", err)
}
defer func() { _ = tx.Rollback() }()
if snapshot {
if _, err := tx.ExecContext(ctx, `UPDATE upstream_accounts SET current=0`); err != nil {
return fmt.Errorf("重置上游账号快照: %w", err)
}
}
for _, account := range accounts {
var existingID string
err := tx.QueryRowContext(ctx, `
@@ -300,14 +316,14 @@ SELECT id FROM upstream_accounts WHERE cpa_auth_id=? OR (cpa_auth_index <> '' AN
existingID = fmt.Sprintf("up_%x", sum[:10])
}
_, err = tx.ExecContext(ctx, `
INSERT INTO upstream_accounts (id, cpa_auth_id, cpa_auth_index, provider, display_name, status, status_message, disabled, unavailable, priority, last_seen_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
INSERT INTO upstream_accounts (id, cpa_auth_id, cpa_auth_index, provider, display_name, status, status_message, disabled, unavailable, current, priority, last_seen_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)
ON CONFLICT(id) DO UPDATE SET cpa_auth_id=excluded.cpa_auth_id, provider=excluded.provider,
cpa_auth_index=CASE WHEN excluded.cpa_auth_index <> '' THEN excluded.cpa_auth_index ELSE upstream_accounts.cpa_auth_index END,
display_name=CASE WHEN excluded.display_name = excluded.cpa_auth_id AND upstream_accounts.display_name <> '' THEN upstream_accounts.display_name ELSE excluded.display_name END,
status=excluded.status,
status_message=excluded.status_message, disabled=excluded.disabled, unavailable=excluded.unavailable,
priority=excluded.priority, last_seen_at=excluded.last_seen_at`, existingID, account.CPAAuthID,
current=1, priority=excluded.priority, last_seen_at=excluded.last_seen_at`, existingID, account.CPAAuthID,
account.CPAAuthIndex, account.Provider, account.DisplayName, account.Status, account.StatusMessage,
account.Disabled, account.Unavailable, account.Priority, formatTime(account.LastSeenAt))
if err != nil {
@@ -324,7 +340,7 @@ func (r *SQLiteUsageRepository) ListUpstreamAccounts(ctx context.Context) ([]man
rows, err := r.readDB.QueryContext(ctx, `
SELECT id, cpa_auth_id, cpa_auth_index, provider, display_name, status, status_message,
disabled, unavailable, priority, last_seen_at
FROM upstream_accounts ORDER BY provider, display_name COLLATE NOCASE`)
FROM upstream_accounts WHERE current=1 ORDER BY provider, display_name COLLATE NOCASE, cpa_auth_index`)
if err != nil {
return nil, fmt.Errorf("查询上游账号: %w", err)
}
internal/repository/sqlite_access_test.go
+40
View File
@@ -86,6 +86,46 @@ func TestManagedKeyStatsUseShanghaiDayBoundary(t *testing.T) {
}
}
func TestReconcileUpstreamAccountsListsOnlyCurrentSnapshot(t *testing.T) {
store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db"))
if err != nil {
t.Fatal(err)
}
defer store.Close()
ctx := context.Background()
old := managedaccess.UpstreamAccount{
CPAAuthID: "codex-user-prolite.json", CPAAuthIndex: "old-index",
Provider: "codex", DisplayName: "user@example.com", LastSeenAt: time.Now().Add(-time.Hour),
}
current := managedaccess.UpstreamAccount{
CPAAuthID: "codex-user-pro.json", CPAAuthIndex: "current-index",
Provider: "codex", DisplayName: "user@example.com", LastSeenAt: time.Now(),
}
if err := store.ReconcileUpstreamAccounts(ctx, []managedaccess.UpstreamAccount{old, current}); err != nil {
t.Fatal(err)
}
initial, err := store.ListUpstreamAccounts(ctx)
if err != nil || len(initial) != 2 {
t.Fatalf("initial accounts = %+v, err=%v", initial, err)
}
var oldID string
for _, account := range initial {
if account.CPAAuthID == old.CPAAuthID {
oldID = account.ID
}
}
if err := store.ReconcileUpstreamAccounts(ctx, []managedaccess.UpstreamAccount{current}); err != nil {
t.Fatal(err)
}
listed, err := store.ListUpstreamAccounts(ctx)
if err != nil || len(listed) != 1 || listed[0].CPAAuthID != current.CPAAuthID {
t.Fatalf("current accounts = %+v, err=%v", listed, err)
}
if historical, err := store.UpstreamAccountByID(ctx, oldID); err != nil || historical.CPAAuthID != old.CPAAuthID {
t.Fatalf("historical account = %+v, err=%v", historical, err)
}
}
func TestManagedKeyStatsCountsOrphanUsageAndAliasLifecycleOnce(t *testing.T) {
store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db"))
if err != nil {
internal/repository/sqlite_usage.go
+2
View File
@@ -149,6 +149,7 @@ CREATE TABLE IF NOT EXISTS upstream_accounts (
status_message TEXT NOT NULL DEFAULT '',
disabled INTEGER NOT NULL DEFAULT 0,
unavailable INTEGER NOT NULL DEFAULT 0,
current INTEGER NOT NULL DEFAULT 1,
priority INTEGER NOT NULL DEFAULT 0,
last_seen_at TEXT NOT NULL
);
@@ -290,6 +291,7 @@ func OpenSQLiteUsage(databasePath string) (*SQLiteUsageRepository, error) {
`ALTER TABLE model_prices ADD COLUMN source_revision TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE model_prices ADD COLUMN source_fetched_at TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE billing_accounts ADD COLUMN lifetime_spent_micros INTEGER NOT NULL DEFAULT 0`,
`ALTER TABLE upstream_accounts ADD COLUMN current INTEGER NOT NULL DEFAULT 1`,
} {
if _, err := db.Exec(migration); err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column") {
_ = db.Close()
internal/web/app/features/keys.js
+8 -1
View File
@@ -467,7 +467,14 @@ function modelRules() {
}
function upstreamLabel(account) {
return `${account.provider || "unknown"} · ${account.display_name || account.cpa_auth_id || account.id}`;
const provider = account.provider || "unknown";
const name = account.display_name || account.cpa_auth_id || account.id;
const duplicated = upstreamAccounts.some(item => item !== account &&
(item.provider || "unknown") === provider &&
(item.display_name || item.cpa_auth_id || item.id) === name);
if (!duplicated) return `${provider} · ${name}`;
const identity = (account.cpa_auth_index || account.cpa_auth_id || account.id).slice(0, 8);
return `${provider} · ${name} · ${identity}`;
}
function routeLabel(key) {