whm/ai_site
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: heal dead sync channels and block deleting system default (Z12h)

Browse Source 
Ensure default channel and rebind Agent/Binding on list, ticket, agents/me, and ensure-binding so SyncPage does not stay on deleted-channel errors.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
whm
2026-08-06 08:57:30 +08:00
parent 2a7da53769
commit 362dcee242
8 changed files with 309 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
platform/internal/handler/sync.go
View File

@@ -9,6 +9,7 @@ import (
"aijianzhan/platform/internal/audit"
"aijianzhan/platform/internal/authx"
"aijianzhan/platform/internal/dbsync"
"aijianzhan/platform/internal/logic/applogic"
"aijianzhan/platform/internal/svc"
"github.com/zeromicro/go-zero/rest/httpx"
@@ -41,7 +42,12 @@ func syncListHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
if !requireDBSync(svcCtx, w) {
return
}
list, err := svcCtx.DBSync.Store().ListChannelsByTenant(syncTenantID(r))
tid := syncTenantID(r)
if _, err := applogic.HealTenantSyncBind(r.Context(), svcCtx, tid); err != nil {
// 列表仍继续，避免自愈失败挡运维
_ = err
}
list, err := svcCtx.DBSync.Store().ListChannelsByTenant(tid)
if err != nil {
authx.WriteError(w, http.StatusBadRequest, err.Error())
return
@@ -109,15 +115,26 @@ func syncDeleteHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return
}
id := pathvar.Vars(r)["id"]
if _, err := svcCtx.DBSync.Store().GetChannelForTenant(id, syncTenantID(r)); err != nil {
tid := syncTenantID(r)
ch, err := svcCtx.DBSync.Store().GetChannelForTenant(id, tid)
if err != nil {
authx.WriteError(w, http.StatusNotFound, err.Error())
return
}
// Z12h-1：公司默认同步通道禁止删除（账号落点依赖它）
if ch.IsSystemDefault {
authx.WriteError(w, http.StatusBadRequest, "公司默认同步通道不可删除；账号绑定落点依赖此通道。若异常请刷新页面自动修复")
return
}
svcCtx.DBSync.StopChannel(id)
if err := svcCtx.DBSync.Store().DeleteChannel(id); err != nil {
authx.WriteError(w, http.StatusBadRequest, err.Error())
return
}
// 若误删后仍有 Binding 挂死，heal 会挂回默认通道
if _, hErr := applogic.HealTenantSyncBind(r.Context(), svcCtx, tid); hErr != nil {
_ = hErr
}
httpx.OkJson(w, map[string]any{"ok": true})
}
}

18
platform/internal/handler/sync_binding.go
View File

@@ -6,6 +6,7 @@ import (
"aijianzhan/platform/internal/authx"
"aijianzhan/platform/internal/dbsync"
"aijianzhan/platform/internal/logic/applogic"
"aijianzhan/platform/internal/svc"
"github.com/zeromicro/go-zero/rest/httpx"
@@ -16,8 +17,11 @@ func syncBindingsListHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
if !requireDBSync(svcCtx, w) {
return
}
localID := r.URL.Query().Get("local_database_id")
tid := syncTenantID(r)
if _, err := applogic.HealTenantSyncBind(r.Context(), svcCtx, tid); err != nil {
_ = err
}
localID := r.URL.Query().Get("local_database_id")
var (
list []dbsync.Binding
err error
@@ -74,6 +78,18 @@ func syncBindingsEnsureHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return
}
}
// Z12h-2：通道空/已删 → 静默挂回公司默认同步
needHeal := body.ChannelID == ""
if !needHeal {
if _, e := svcCtx.DBSync.Store().GetChannelForTenant(body.ChannelID, body.TenantID); e != nil {
needHeal = true
}
}
if needHeal {
if healed, hErr := applogic.HealTenantSyncBind(r.Context(), svcCtx, body.TenantID); hErr == nil && healed != nil && healed.ChannelID != "" {
body.ChannelID = healed.ChannelID
}
}
saved, err := svcCtx.DBSync.Store().EnsureBinding(body)
if err != nil {
authx.WriteError(w, http.StatusBadRequest, err.Error())

62
platform/internal/logic/applogic/agent_sync_bind.go
View File

@@ -47,45 +47,59 @@ func fillUserSyncOnToken(svcCtx *svc.ServiceContext, resp *types.TokenResp, u *u
resp.SyncBound = resp.ChannelID != "" && resp.OnlineDBID != ""
}
// ensureAgentSyncBind Z12c：启用/创建时若无通道则自动创建公司默认同步通道并写回智能体。
// ensureAgentSyncBind Z12c/Z12h：无通道或通道已删除时，创建/复用默认同步通道并写回智能体。
func (l *AgentAdminLogic) ensureAgentSyncBind(acc *agentstore.Account) (*agentstore.Account, error) {
if acc == nil {
return nil, fmt.Errorf("agent nil")
}
if strings.TrimSpace(acc.ChannelID) != "" && strings.TrimSpace(acc.OnlineDBID) != "" {
return acc, nil
}
if l.svcCtx.DBSync == nil {
return acc, nil
}
cfg := l.svcCtx.Config.DBSync
driver := dbsync.Driver(strings.TrimSpace(cfg.DefaultRemoteDriver))
if driver == "" {
driver = dbsync.DriverPostgres
alive := channelAlive(l.svcCtx.DBSync.Store(), acc.TenantID, acc.ChannelID)
if alive && strings.TrimSpace(acc.OnlineDBID) != "" {
return acc, nil
}
ch, err := l.svcCtx.DBSync.EnsureAndStartSystemDefaultChannel(dbsync.DefaultChannelOpts{
TenantID: acc.TenantID,
AgentID: acc.AgentID,
Name: fmt.Sprintf("默认同步 · %s", acc.Name),
RemoteDriver: driver,
RemoteDSN: strings.TrimSpace(cfg.DefaultRemoteDSN),
DatabaseName: acc.DatabaseName,
})
if err != nil {
return nil, fmt.Errorf("ensure default channel: %w", err)
}
online := dbsync.ResolveOnlineDBID(acc.OnlineDBID, ch.ID)
dbName := strings.TrimSpace(acc.DatabaseName)
if dbName == "" {
dbName = fmt.Sprintf("agent_%d", acc.AgentID)
if _, err := HealTenantSyncBind(l.ctx, l.svcCtx, acc.TenantID); err != nil {
return nil, fmt.Errorf("heal sync bind: %w", err)
}
st, err := l.store()
if err != nil {
return nil, err
}
updated, err := st.AttachSyncBind(l.ctx, acc.AgentID, acc.TenantID, ch.ID, online, dbName, acc.Status == agentstore.StatusActive || acc.Status == "")
updated, err := st.Get(l.ctx, acc.TenantID, acc.AgentID)
if err != nil {
return nil, err
}
// 若 heal 未覆盖该智能体（例如无 Binding 仅 Agent），再显式挂一次
if !channelAlive(l.svcCtx.DBSync.Store(), updated.TenantID, updated.ChannelID) || strings.TrimSpace(updated.OnlineDBID) == "" {
cfg := l.svcCtx.Config.DBSync
driver := dbsync.Driver(strings.TrimSpace(cfg.DefaultRemoteDriver))
if driver == "" {
driver = dbsync.DriverPostgres
}
ch, err := l.svcCtx.DBSync.EnsureAndStartSystemDefaultChannel(dbsync.DefaultChannelOpts{
TenantID: acc.TenantID,
AgentID: acc.AgentID,
Name: fmt.Sprintf("默认同步 · %s", acc.Name),
RemoteDriver: driver,
RemoteDSN: strings.TrimSpace(cfg.DefaultRemoteDSN),
DatabaseName: acc.DatabaseName,
})
if err != nil {
return nil, fmt.Errorf("ensure default channel: %w", err)
}
online := strings.TrimSpace(updated.OnlineDBID)
if online == "" {
online = dbsync.ResolveOnlineDBID("", ch.ID)
}
dbName := strings.TrimSpace(updated.DatabaseName)
if dbName == "" {
dbName = fmt.Sprintf("agent_%d", updated.AgentID)
}
updated, err = st.AttachSyncBind(l.ctx, updated.AgentID, updated.TenantID, ch.ID, online, dbName, updated.Status == agentstore.StatusActive || updated.Status == "")
if err != nil {
return nil, err
}
}
return updated, nil
}

7
platform/internal/logic/applogic/agents.go
View File

@@ -243,6 +243,13 @@ func (l *AuthLogic) IssueClientCredentials(clientID, clientSecret string) (*type
if err != nil {
return nil, err
}
if l.svcCtx.DBSync != nil {
if _, hErr := HealTenantSyncBind(l.ctx, l.svcCtx, acc.TenantID); hErr == nil {
if refreshed, gErr := l.svcCtx.Agents.Get(l.ctx, acc.TenantID, acc.AgentID); gErr == nil && refreshed != nil {
acc = refreshed
}
}
}
token, exp, err := authx.IssueAgentToken(l.svcCtx.JWT, acc.TenantID, acc.AgentID, acc.Perms)
if err != nil {
return nil, err

39
platform/internal/logic/applogic/bind_flow.go
View File

@@ -3,6 +3,7 @@ package applogic
import (
"encoding/json"
"fmt"
"log"
"strings"
"time"
@@ -14,7 +15,7 @@ import (
"aijianzhan/platform/internal/userstore"
)
// AgentMe Z12b：智能体自查绑定（无需「管理智能体」）。
// AgentMe Z12b/Z12h：智能体自查绑定；通道缺失时自愈。
func (l *AuthLogic) AgentMe() (*agentstore.Account, error) {
if !authx.IsAgent(authx.Role(l.ctx)) {
return nil, fmt.Errorf("仅智能体可访问")
@@ -28,6 +29,13 @@ func (l *AuthLogic) AgentMe() (*agentstore.Account, error) {
if err != nil {
return nil, err
}
if l.svcCtx.DBSync != nil {
if _, hErr := HealTenantSyncBind(l.ctx, l.svcCtx, tid); hErr != nil {
log.Printf("agent me heal: %v", hErr)
} else if refreshed, gErr := l.svcCtx.Agents.Get(l.ctx, tid, aid); gErr == nil && refreshed != nil {
acc = refreshed
}
}
return acc, nil
}
@@ -52,30 +60,13 @@ func (l *AuthLogic) CreateBindCode(req BindCodeCreateReq) (*bindcodestore.BindCo
online := strings.TrimSpace(req.OnlineDBID)
dbName := strings.TrimSpace(req.DatabaseName)
if channelID == "" && l.svcCtx.DBSync != nil {
if ch, err := l.svcCtx.DBSync.Store().FindSystemDefaultChannel(tid); err == nil && ch != nil {
channelID = ch.ID
// Z12h：生成绑定码前确保默认通道存在（含误删自愈）
if healed, err := HealTenantSyncBind(l.ctx, l.svcCtx, tid); err != nil {
return nil, fmt.Errorf("无默认同步通道：%w", err)
} else if healed != nil && strings.TrimSpace(healed.ChannelID) != "" {
channelID = healed.ChannelID
if online == "" {
online = dbsync.ResolveOnlineDBID("", ch.ID)
}
_ = l.svcCtx.DBSync.StartChannel(ch.ID) // 已有默认同步通道也保持运行中
} else {
// 尝试创建默认同步通道
cfg := l.svcCtx.Config.DBSync
driver := dbsync.Driver(strings.TrimSpace(cfg.DefaultRemoteDriver))
if driver == "" {
driver = dbsync.DriverPostgres
}
saved, err := l.svcCtx.DBSync.EnsureAndStartSystemDefaultChannel(dbsync.DefaultChannelOpts{
TenantID: tid,
RemoteDriver: driver,
RemoteDSN: strings.TrimSpace(cfg.DefaultRemoteDSN),
})
if err != nil {
return nil, fmt.Errorf("无默认同步通道：%w", err)
}
channelID = saved.ID
if online == "" {
online = dbsync.ResolveOnlineDBID("", saved.ID)
online = dbsync.ResolveOnlineDBID("", channelID)
}
}
}

131
platform/internal/logic/applogic/sync_heal.go Normal file
View File

@@ -0,0 +1,131 @@
package applogic
import (
"context"
"fmt"
"log"
"strings"
"sync"
"aijianzhan/platform/internal/agentstore"
"aijianzhan/platform/internal/dbsync"
"aijianzhan/platform/internal/svc"
)
// HealResult Z12h 自愈结果。
type HealResult struct {
ChannelID string `json:"channel_id"`
BindingsFixed int `json:"bindings_fixed"`
AgentsFixed int `json:"agents_fixed"`
ChannelCreated bool `json:"channel_created,omitempty"`
}
var healTenantMu sync.Map // tenantID -> *sync.Mutex；避免 SyncPage 并行列表双建默认通道
func lockTenantHeal(tenantID int64) func() {
v, _ := healTenantMu.LoadOrStore(tenantID, &sync.Mutex{})
mu := v.(*sync.Mutex)
mu.Lock()
return mu.Unlock
}
// channelAlive 通道是否仍属于该租户。
func channelAlive(store *dbsync.FileStore, tenantID int64, channelID string) bool {
id := strings.TrimSpace(channelID)
if id == "" || store == nil || tenantID <= 0 {
return false
}
_, err := store.GetChannelForTenant(id, tenantID)
return err == nil
}
func agentNeedsSyncHeal(store *dbsync.FileStore, a agentstore.Account) bool {
cid := strings.TrimSpace(a.ChannelID)
alive := channelAlive(store, a.TenantID, cid)
if cid != "" {
if !alive {
return true // 挂死通道
}
return strings.TrimSpace(a.OnlineDBID) == "" // 通道在但缺 online
}
// 无通道：仅 active（与 Z12c 启用即绑一致），勿动 pending
return a.Status == agentstore.StatusActive || a.Status == ""
}
// HealTenantSyncBind Z12h：确保公司默认同步通道存在并运行；重挂挂死通道的 Agent/Binding。
// online_db_id 有值则保留（同一 remote 上的逻辑库名不变）；仅改 channel_id。
func HealTenantSyncBind(ctx context.Context, svcCtx *svc.ServiceContext, tenantID int64) (*HealResult, error) {
if svcCtx == nil || svcCtx.DBSync == nil || tenantID <= 0 {
return nil, nil
}
unlock := lockTenantHeal(tenantID)
defer unlock()
cfg := svcCtx.Config.DBSync
driver := dbsync.Driver(strings.TrimSpace(cfg.DefaultRemoteDriver))
if driver == "" {
driver = dbsync.DriverPostgres
}
before, _ := svcCtx.DBSync.Store().FindSystemDefaultChannel(tenantID)
ch, err := svcCtx.DBSync.EnsureAndStartSystemDefaultChannel(dbsync.DefaultChannelOpts{
TenantID: tenantID,
RemoteDriver: driver,
RemoteDSN: strings.TrimSpace(cfg.DefaultRemoteDSN),
Name: fmt.Sprintf("公司默认同步 #%d", tenantID),
})
if err != nil {
return nil, err
}
out := &HealResult{
ChannelID: ch.ID,
ChannelCreated: before == nil || before.ID != ch.ID,
}
// Binding：通道空或已删除 → 改挂默认通道（保留 online_db_id）
list, err := svcCtx.DBSync.Store().ListBindings(tenantID, "")
if err == nil {
for _, b := range list {
if channelAlive(svcCtx.DBSync.Store(), tenantID, b.ChannelID) {
continue
}
b.ChannelID = ch.ID
if strings.TrimSpace(b.OnlineDBID) == "" {
b.OnlineDBID = dbsync.ResolveOnlineDBID("", ch.ID)
}
if _, e := svcCtx.DBSync.Store().EnsureBinding(b); e != nil {
log.Printf("dbsync heal binding tenant=%d local=%s: %v", tenantID, b.LocalDatabaseID, e)
continue
}
out.BindingsFixed++
}
}
// Agent：挂死通道 / 缺 online / active 无通道 → AttachSyncBind（保留 online_db_id）
if svcCtx.Agents != nil {
agents, err := svcCtx.Agents.List(ctx, tenantID)
if err == nil {
for i := range agents {
a := agents[i]
if !agentNeedsSyncHeal(svcCtx.DBSync.Store(), a) {
continue
}
online := strings.TrimSpace(a.OnlineDBID)
if online == "" {
online = dbsync.ResolveOnlineDBID("", ch.ID)
}
dbName := strings.TrimSpace(a.DatabaseName)
if dbName == "" {
dbName = fmt.Sprintf("agent_%d", a.AgentID)
}
if _, e := svcCtx.Agents.AttachSyncBind(ctx, a.AgentID, tenantID, ch.ID, online, dbName, a.Status == agentstore.StatusActive || a.Status == ""); e != nil {
log.Printf("dbsync heal agent tenant=%d agent=%d: %v", tenantID, a.AgentID, e)
continue
}
out.AgentsFixed++
}
}
}
if out.BindingsFixed > 0 || out.AgentsFixed > 0 || out.ChannelCreated {
log.Printf("dbsync heal tenant=%d channel=%s bindings=%d agents=%d created=%v",
tenantID, ch.ID, out.BindingsFixed, out.AgentsFixed, out.ChannelCreated)
}
return out, nil
}

6
platform/internal/logic/applogic/yuheng_ticket.go
View File

@@ -83,6 +83,12 @@ func (l *AuthLogic) ExchangeYuhengTicket(req YuhengTicketExchangeReq) (*YuhengTi
if err != nil {
return nil, err
}
// Z12h：换票时顺带自愈本租户挂死通道的 Binding/Agent
if _, hErr := HealTenantSyncBind(l.ctx, l.svcCtx, u.TenantID); hErr == nil {
if refreshed, gErr := l.svcCtx.Agents.Get(l.ctx, acc.TenantID, acc.AgentID); gErr == nil && refreshed != nil {
acc = refreshed
}
}
if err := l.ensureAgentSyncPerm(acc); err != nil {
return nil, err
}