webforwarding.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712
  1. package waf
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. v1 "github.com/go-nunu/nunu-layout-advanced/api/v1"
  7. "github.com/go-nunu/nunu-layout-advanced/internal/model"
  8. "github.com/go-nunu/nunu-layout-advanced/internal/repository/api/waf"
  9. "github.com/go-nunu/nunu-layout-advanced/internal/service"
  10. "github.com/go-nunu/nunu-layout-advanced/internal/service/api/flexCdn"
  11. "github.com/go-nunu/nunu-layout-advanced/pkg/rabbitmq"
  12. "golang.org/x/sync/errgroup"
  13. "maps"
  14. "net"
  15. "sort"
  16. )
  17. type WebForwardingService interface {
  18. GetWebForwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.WebForwardingDataRequest, error)
  19. GetWebForwardingWafWebAllIps(ctx context.Context, req v1.GetForwardingRequest) ([]v1.WebForwardingDataRequest, error)
  20. AddWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) (int, error)
  21. EditWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error
  22. DeleteWebForwarding(ctx context.Context, req v1.DeleteWebForwardingRequest) error
  23. }
  24. func NewWebForwardingService(
  25. service *service.Service,
  26. required service.RequiredService,
  27. webForwardingRepository waf.WebForwardingRepository,
  28. crawler service.CrawlerService,
  29. parser service.ParserService,
  30. wafformatter WafFormatterService,
  31. aoDun service.AoDunService,
  32. mq *rabbitmq.RabbitMQ,
  33. gatewayIp GatewayipService,
  34. globalLimitRep waf.GlobalLimitRepository,
  35. cdn flexCdn.CdnService,
  36. proxy flexCdn.ProxyService,
  37. sslCert flexCdn.SslCertService,
  38. websocket flexCdn.WebsocketService,
  39. cc CcService,
  40. ccIpList CcIpListService,
  41. aidedWeb AidedWebService,
  42. ) WebForwardingService {
  43. return &webForwardingService{
  44. Service: service,
  45. webForwardingRepository: webForwardingRepository,
  46. required: required,
  47. parser: parser,
  48. crawler: crawler,
  49. wafformatter: wafformatter,
  50. aoDun: aoDun,
  51. mq: mq,
  52. gatewayIp: gatewayIp,
  53. cdn: cdn,
  54. globalLimitRep: globalLimitRep,
  55. proxy: proxy,
  56. sslCert: sslCert,
  57. websocket: websocket,
  58. cc: cc,
  59. ccIpList: ccIpList,
  60. aidedWeb: aidedWeb,
  61. }
  62. }
  63. type webForwardingService struct {
  64. *service.Service
  65. webForwardingRepository waf.WebForwardingRepository
  66. required service.RequiredService
  67. parser service.ParserService
  68. crawler service.CrawlerService
  69. wafformatter WafFormatterService
  70. aoDun service.AoDunService
  71. mq *rabbitmq.RabbitMQ
  72. gatewayIp GatewayipService
  73. cdn flexCdn.CdnService
  74. globalLimitRep waf.GlobalLimitRepository
  75. proxy flexCdn.ProxyService
  76. sslCert flexCdn.SslCertService
  77. websocket flexCdn.WebsocketService
  78. cc CcService
  79. ccIpList CcIpListService
  80. aidedWeb AidedWebService
  81. }
  82. func (s *webForwardingService) GetWebForwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.WebForwardingDataRequest, error) {
  83. var webForwarding model.WebForwarding
  84. var backend model.WebForwardingRule
  85. g, gCtx := errgroup.WithContext(ctx)
  86. g.Go(func() error {
  87. res, e := s.webForwardingRepository.GetWebForwarding(gCtx, int64(req.Id))
  88. if e != nil {
  89. // 直接返回错误,errgroup 会捕获它
  90. return fmt.Errorf("GetWebForwarding failed: %w", e)
  91. }
  92. if res != nil {
  93. webForwarding = *res
  94. }
  95. return nil
  96. })
  97. g.Go(func() error {
  98. res, e := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, req.Id)
  99. if e != nil {
  100. return fmt.Errorf("GetWebForwardingByID failed: %w", e)
  101. }
  102. if res != nil {
  103. backend = *res
  104. }
  105. return nil
  106. })
  107. if err := g.Wait(); err != nil {
  108. return v1.WebForwardingDataRequest{}, err
  109. }
  110. return v1.WebForwardingDataRequest{
  111. Id: webForwarding.Id,
  112. Port: webForwarding.Port,
  113. Domain: webForwarding.Domain,
  114. IsHttps: webForwarding.IsHttps,
  115. Comment: webForwarding.Comment,
  116. BackendList: backend.BackendList,
  117. HttpsKey: webForwarding.HttpsKey,
  118. HttpsCert: webForwarding.HttpsCert,
  119. Proxy: webForwarding.Proxy,
  120. CcConfig: v1.CcConfigRequest{
  121. IsOn: webForwarding.Cc,
  122. ThresholdMethod: webForwarding.ThresholdMethod,
  123. Level: webForwarding.Level,
  124. Limit5s: webForwarding.Limit5s,
  125. Limit60s: webForwarding.Limit60s,
  126. Limit300s: webForwarding.Limit300s,
  127. },
  128. }, nil
  129. }
  130. // AddWebForwarding 添加Web转发配置
  131. // 该函数负责创建完整的Web转发配置,包括:
  132. // 1. 数据验证和预处理
  133. // 2. SSL证书管理
  134. // 3. CDN网站创建和配置
  135. // 4. 源站服务器添加
  136. // 5. 各种功能开启(WebSocket、Proxy、日志、CC防护等)
  137. // 6. 数据库记录保存
  138. // 7. 白名单任务发布
  139. func (s *webForwardingService) AddWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) (int, error) {
  140. // 1. 数据准备和验证
  141. require, formData, err := s.aidedWeb.PrepareWafData(ctx, req)
  142. if err != nil {
  143. return 0, err
  144. }
  145. if err := s.aidedWeb.ValidateAddRequest(ctx, req, require); err != nil {
  146. return 0, err
  147. }
  148. // 2. 处理SSL证书
  149. if err := s.aidedWeb.ProcessSSLCertificate(ctx, req, require, formData); err != nil {
  150. return 0, err
  151. }
  152. // 3. 创建CDN网站
  153. webId, err := s.aidedWeb.CreateCdnWebsite(ctx, req, require, formData)
  154. if err != nil {
  155. return 0, err
  156. }
  157. // 4. 配置WebSocket
  158. if err := s.aidedWeb.ConfigureWebsocket(ctx, webId); err != nil {
  159. return 0, err
  160. }
  161. // 5. 添加源站到网站
  162. cdnOriginIds, err := s.aidedWeb.AddOriginsToWebsite(ctx, req, webId)
  163. if err != nil {
  164. return 0, err
  165. }
  166. // 6. 配置各种功能
  167. if err := s.aidedWeb.ConfigureProxyProtocol(ctx, req, webId); err != nil {
  168. return 0, err
  169. }
  170. if err := s.aidedWeb.EditLog(ctx, webId); err != nil {
  171. return 0, err
  172. }
  173. if err := s.aidedWeb.ConfigureCCProtection(ctx, req, webId); err != nil {
  174. return 0, err
  175. }
  176. if err := s.aidedWeb.ConfigureWafFirewall(ctx, webId, require.GroupId); err != nil {
  177. return 0, err
  178. }
  179. // 7. 保存到数据库
  180. id, err := s.aidedWeb.SaveToDatabase(ctx, req, require, webId, cdnOriginIds)
  181. if err != nil {
  182. return 0, err
  183. }
  184. // 8. 处理异步任务
  185. s.aidedWeb.ProcessAsyncTasks(ctx, req, require)
  186. return id, nil
  187. }
  188. func (s *webForwardingService) EditWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error {
  189. err := s.wafformatter.validateWafDomainCount(ctx, v1.GlobalRequire{
  190. HostId: req.HostId,
  191. Domain: req.WebForwardingData.Domain,
  192. Comment: req.WebForwardingData.Comment,
  193. Uid: req.Uid,
  194. })
  195. if err != nil {
  196. return err
  197. }
  198. oldData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(req.WebForwardingData.Id))
  199. if err != nil {
  200. return err
  201. }
  202. req.WebForwardingData.SslCertId = int64(oldData.SslCertId)
  203. req.WebForwardingData.SslPolicyId = int64(oldData.SslPolicyId)
  204. require, formData, err := s.aidedWeb.PrepareWafData(ctx, req)
  205. if err != nil {
  206. return err
  207. }
  208. // 验证端口重复
  209. protocol := s.aidedWeb.GetProtocolType(req.WebForwardingData.IsHttps)
  210. err = s.wafformatter.VerifyPort(ctx, protocol, int64(req.WebForwardingData.Id), req.WebForwardingData.Port, int64(require.HostId), req.WebForwardingData.Domain)
  211. if err != nil {
  212. return err
  213. }
  214. //修改网站端口
  215. if oldData.Port != req.WebForwardingData.Port || oldData.IsHttps != req.WebForwardingData.IsHttps || oldData.HttpsCert != req.WebForwardingData.HttpsCert || oldData.HttpsKey != req.WebForwardingData.HttpsKey {
  216. var apiType string
  217. var closeType string
  218. // 修改证书
  219. if oldData.HttpsCert != req.WebForwardingData.HttpsCert || oldData.HttpsKey != req.WebForwardingData.HttpsKey {
  220. err = s.sslCert.EditSSLCert(ctx, v1.SSL{
  221. Name: req.WebForwardingData.Domain,
  222. CertId: oldData.SslCertId,
  223. CertData: req.WebForwardingData.HttpsCert,
  224. KeyData: req.WebForwardingData.HttpsKey,
  225. CdnUserId: require.CdnUid,
  226. Domain: req.WebForwardingData.Domain,
  227. Description: req.WebForwardingData.Comment,
  228. })
  229. if err != nil {
  230. return err
  231. }
  232. }
  233. // 切换协议
  234. var typeConfig v1.TypeJSON
  235. var closeConfig v1.TypeJSON
  236. if s.aidedWeb.IsHttpsProtocol(req.WebForwardingData.IsHttps) {
  237. typeConfig = formData.HttpsJSON
  238. closeConfig = formData.HttpJSON
  239. apiType = s.aidedWeb.GetProtocolType(req.WebForwardingData.IsHttps)
  240. closeType = s.aidedWeb.GetProtocolType(0) // HTTP
  241. } else {
  242. typeConfig = formData.HttpJSON
  243. closeConfig = formData.HttpsJSON
  244. apiType = s.aidedWeb.GetProtocolType(req.WebForwardingData.IsHttps)
  245. closeType = s.aidedWeb.GetProtocolType(1) // HTTPS
  246. }
  247. typeJson,err := json.Marshal(typeConfig)
  248. if err != nil {
  249. return err
  250. }
  251. closeJson,err := json.Marshal(closeConfig)
  252. if err != nil {
  253. return err
  254. }
  255. // 切换协议
  256. err = s.cdn.EditServerType(ctx, v1.EditWebsite{
  257. Id: int64(oldData.CdnWebId),
  258. TypeJSON: typeJson,
  259. }, apiType)
  260. if err != nil {
  261. return err
  262. }
  263. err = s.cdn.EditServerType(ctx, v1.EditWebsite{
  264. Id: int64(oldData.CdnWebId),
  265. TypeJSON: closeJson,
  266. }, closeType)
  267. if err != nil {
  268. return err
  269. }
  270. }
  271. //修改网站域名
  272. if oldData.Domain != req.WebForwardingData.Domain {
  273. type serverName struct {
  274. Name string `json:"name" form:"name"`
  275. Type string `json:"type" form:"type"`
  276. }
  277. var serverData []serverName
  278. var serverJson []byte
  279. serverData = append(serverData, serverName{
  280. Name: req.WebForwardingData.Domain,
  281. Type: "full",
  282. })
  283. serverJson, err = json.Marshal(serverData)
  284. if err != nil {
  285. return err
  286. }
  287. err = s.cdn.EditServerName(ctx, v1.EditServerNames{
  288. ServerId: int64(oldData.CdnWebId),
  289. ServerNamesJSON: serverJson,
  290. })
  291. if err != nil {
  292. return err
  293. }
  294. }
  295. //修改网站名字
  296. if oldData.Comment != req.WebForwardingData.Comment {
  297. nodeId,err := s.globalLimitRep.GetNodeId(ctx, oldData.CdnWebId)
  298. if err != nil {
  299. return err
  300. }
  301. err = s.cdn.EditServerBasic(ctx, int64(oldData.CdnWebId), require.Tag,nodeId)
  302. if err != nil {
  303. return err
  304. }
  305. }
  306. //修改Proxy
  307. if oldData.Proxy != req.WebForwardingData.Proxy {
  308. err = s.proxy.EditProxy(ctx, int64(oldData.CdnWebId), v1.ProxyProtocolJSON{
  309. IsOn: req.WebForwardingData.Proxy,
  310. Version: 1,
  311. })
  312. if err != nil {
  313. return err
  314. }
  315. }
  316. // 修改CC配置
  317. err = s.cc.EditCcConfig(ctx, int64(oldData.CdnWebId), v1.CcConfigRequest{
  318. IsOn: req.WebForwardingData.CcConfig.IsOn,
  319. Level: req.WebForwardingData.CcConfig.Level,
  320. Limit5s: req.WebForwardingData.CcConfig.Limit5s,
  321. Limit60s: req.WebForwardingData.CcConfig.Limit60s,
  322. Limit300s: req.WebForwardingData.CcConfig.Limit300s,
  323. ThresholdMethod: req.WebForwardingData.CcConfig.ThresholdMethod,
  324. })
  325. if err != nil {
  326. return err
  327. }
  328. // 将域名添加到白名单
  329. webData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(req.WebForwardingData.Id))
  330. if err != nil {
  331. return err
  332. }
  333. // 异步任务:将域名添加到白名单
  334. if webData.Domain != req.WebForwardingData.Domain {
  335. firstIp, err := s.gatewayIp.GetGatewayipByHostIdFirst(ctx, int64(req.HostId), int64(req.Uid))
  336. if err != nil {
  337. return err
  338. }
  339. doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, req.WebForwardingData.Domain)
  340. if err != nil {
  341. return err
  342. }
  343. oldDomain, err := s.wafformatter.ConvertToWildcardDomain(ctx, webData.Domain)
  344. if err != nil {
  345. return err
  346. }
  347. if len(require.GatewayIps) == 0 {
  348. return fmt.Errorf("网关组不存在")
  349. }
  350. // 查找域名数量,如果数量小于2,删除旧域名
  351. count, err := s.webForwardingRepository.GetDomainCount(ctx, req.HostId, webData.Domain)
  352. if err != nil {
  353. return err
  354. }
  355. if count < 2 {
  356. go s.wafformatter.PublishDomainWhitelistTask(oldDomain, firstIp, "del")
  357. }
  358. go s.wafformatter.PublishDomainWhitelistTask(doMain, firstIp, "add")
  359. }
  360. // IP过白
  361. ipData, err := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, req.WebForwardingData.Id)
  362. if err != nil {
  363. return err
  364. }
  365. var oldIps []string
  366. var newIps []string
  367. for _, v := range ipData.BackendList {
  368. ip, _, err := net.SplitHostPort(v.Addr)
  369. if err != nil {
  370. return err
  371. }
  372. oldIps = append(oldIps, ip)
  373. }
  374. for _, v := range req.WebForwardingData.BackendList {
  375. ip, _, err := net.SplitHostPort(v.Addr)
  376. if err != nil {
  377. return err
  378. }
  379. newIps = append(newIps, ip)
  380. }
  381. addedIps, removedIps := s.wafformatter.findIpDifferences(oldIps, newIps)
  382. if len(addedIps) > 0 {
  383. go s.wafformatter.PublishIpWhitelistTask(addedIps, "add", "", "white")
  384. }
  385. // IP过白
  386. if len(removedIps) > 0 {
  387. // 1. 一次性获取所有相关IP的数量
  388. ipsToDelist, err := s.wafformatter.WashDelIps(ctx, removedIps)
  389. if err != nil {
  390. return err
  391. }
  392. // 4. 如果有需要处理的IP,则批量发布一次任务
  393. if len(ipsToDelist) > 0 {
  394. go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white")
  395. }
  396. }
  397. //修改源站
  398. addOrigins, delOrigins := s.aidedWeb.FindDifferenceList(ipData.BackendList, req.WebForwardingData.BackendList)
  399. addedIds := make(map[string]int64)
  400. for _, v := range addOrigins {
  401. apiType := s.aidedWeb.GetProtocolType(v.IsHttps)
  402. id, err := s.wafformatter.AddOrigin(ctx, v1.WebJson{
  403. ApiType: apiType,
  404. BackendList: v.Addr,
  405. Host: v.CustomHost,
  406. Comment: req.WebForwardingData.Comment,
  407. })
  408. if err != nil {
  409. return err
  410. }
  411. addedIds[v.Addr] = id
  412. }
  413. for _, v := range addedIds {
  414. err = s.cdn.AddServerOrigin(ctx, int64(oldData.CdnWebId), v)
  415. if err != nil {
  416. return err
  417. }
  418. }
  419. for k, v := range ipData.CdnOriginIds {
  420. for _, ip := range delOrigins {
  421. if k == ip.Addr {
  422. err = s.cdn.DelServerOrigin(ctx, int64(oldData.CdnWebId), v)
  423. if err != nil {
  424. return err
  425. }
  426. delete(ipData.CdnOriginIds, k)
  427. }
  428. }
  429. }
  430. maps.Copy(ipData.CdnOriginIds, addedIds)
  431. webModel := s.aidedWeb.BuildWebForwardingModel(&req.WebForwardingData, req.WebForwardingData.CdnWebId, require)
  432. webModel.Id = req.WebForwardingData.Id
  433. if err = s.webForwardingRepository.EditWebForwarding(ctx, webModel); err != nil {
  434. return err
  435. }
  436. webRuleModel := s.aidedWeb.BuildWebRuleModel(&req.WebForwardingData, require, req.WebForwardingData.Id, ipData.CdnOriginIds)
  437. if err = s.webForwardingRepository.EditWebForwardingIps(ctx, *webRuleModel); err != nil {
  438. return err
  439. }
  440. return nil
  441. }
  442. func (s *webForwardingService) DeleteWebForwarding(ctx context.Context, req v1.DeleteWebForwardingRequest) error {
  443. for _, Id := range req.Ids {
  444. oldData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(Id))
  445. if err != nil {
  446. return err
  447. }
  448. if oldData.HostId != req.HostId {
  449. return fmt.Errorf("用户权限不足")
  450. }
  451. err = s.cdn.DelServer(ctx, int64(oldData.CdnWebId))
  452. if err != nil {
  453. return err
  454. }
  455. // 异步任务:将域名添加到白名单
  456. firstIp, err := s.gatewayIp.GetGatewayipByHostIdFirst(ctx, int64(oldData.HostId), int64(req.Uid))
  457. if err != nil {
  458. return err
  459. }
  460. if oldData.Domain != "" {
  461. doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, oldData.Domain)
  462. if err != nil {
  463. return err
  464. }
  465. go s.wafformatter.PublishDomainWhitelistTask(doMain, firstIp, "del")
  466. }
  467. // IP过白
  468. ipData, err := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, Id)
  469. if err != nil {
  470. return err
  471. }
  472. var ips []string
  473. if ipData != nil && len(ipData.BackendList) > 0 {
  474. for _, v := range ipData.BackendList {
  475. ip, _, err := net.SplitHostPort(v.Addr)
  476. if err != nil {
  477. return err
  478. }
  479. ips = append(ips, ip)
  480. }
  481. }
  482. if len(ips) > 0 {
  483. ipsToDelist, err := s.wafformatter.WashDelIps(ctx, ips)
  484. if err != nil {
  485. return err
  486. }
  487. // 4. 如果有需要处理的IP,则批量发布一次任务
  488. if len(ipsToDelist) > 0 {
  489. go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white")
  490. }
  491. }
  492. // 删除ssl
  493. if oldData.SslCertId != 0 {
  494. err = s.cdn.DelSSLCert(ctx, int64(oldData.SslCertId))
  495. if err != nil {
  496. return err
  497. }
  498. err = s.sslCert.EditSslPolicy(ctx, int64(oldData.SslPolicyId), []int64{int64(oldData.SslCertId)}, "del")
  499. if err != nil {
  500. return err
  501. }
  502. }
  503. if err = s.webForwardingRepository.DeleteWebForwarding(ctx, int64(Id)); err != nil {
  504. return err
  505. }
  506. if err = s.webForwardingRepository.DeleteWebForwardingIpsById(ctx, Id); err != nil {
  507. return err
  508. }
  509. }
  510. return nil
  511. }
  512. func (s *webForwardingService) GetWebForwardingWafWebAllIps(ctx context.Context, req v1.GetForwardingRequest) ([]v1.WebForwardingDataRequest, error) {
  513. type CombinedResult struct {
  514. Id int
  515. Forwarding *model.WebForwarding
  516. BackendRule *model.WebForwardingRule
  517. Err error // 如果此ID的处理出错,则携带错误
  518. }
  519. g, gCtx := errgroup.WithContext(ctx)
  520. ids, err := s.webForwardingRepository.GetWebForwardingWafWebAllIds(gCtx, req.HostId)
  521. if err != nil {
  522. return nil, fmt.Errorf("GetWebForwardingWafWebAllIds failed: %w", err)
  523. }
  524. if len(ids) == 0 {
  525. return nil, nil // 没有ID,直接返回空切片
  526. }
  527. // 创建一个通道来接收每个ID的处理结果
  528. // 通道缓冲区大小设为ID数量,这样发送者不会因为接收者慢而阻塞(在所有goroutine都启动后)
  529. resultsChan := make(chan CombinedResult, len(ids))
  530. for _, idVal := range ids {
  531. currentID := idVal // 捕获循环变量
  532. g.Go(func() error {
  533. var wf *model.WebForwarding
  534. var bk *model.WebForwardingRule
  535. var localErr error
  536. // 1. 获取 WebForwarding 信息
  537. wf, localErr = s.webForwardingRepository.GetWebForwarding(gCtx, int64(currentID))
  538. if localErr != nil {
  539. // 发送错误到通道,并由 errgroup 捕获
  540. // errgroup 会处理第一个非nil错误,并取消其他 goroutine
  541. resultsChan <- CombinedResult{Id: currentID, Err: fmt.Errorf("GetWebForwarding for id %d failed: %w", currentID, localErr)}
  542. return localErr // 返回错误给 errgroup
  543. }
  544. if wf == nil { // 正常情况下,如果没错误,wf不应为nil,但防御性检查
  545. localErr = fmt.Errorf("GetWebForwarding for id %d returned nil data without error", currentID)
  546. resultsChan <- CombinedResult{Id: currentID, Err: localErr}
  547. return localErr
  548. }
  549. // 2. 获取 Backend IP 信息
  550. // 注意:这里我们允许 GetWebForwardingIpsByID 可能返回 nil 数据(例如没有规则)而不是错误
  551. // 如果它也可能返回错误,则处理方式与上面类似
  552. bk, localErr = s.webForwardingRepository.GetWebForwardingIpsByID(gCtx, currentID)
  553. if localErr != nil {
  554. // 如果获取IP信息失败是一个致命错误,则也应返回错误
  555. // 如果允许部分成功(比如有WebForwarding但没有IP信息),则可以不将此视为errgroup的错误
  556. // 这里假设它也是一个需要errgroup捕获的错误
  557. resultsChan <- CombinedResult{Id: currentID, Forwarding: wf, Err: fmt.Errorf("GetWebForwardingIpsByID for id %d failed: %w", currentID, localErr)}
  558. return localErr // 返回错误给 errgroup
  559. }
  560. // bk 可能是 nil 如果没有错误且没有规则,这取决于业务逻辑
  561. // 发送成功的结果到通道
  562. resultsChan <- CombinedResult{Id: currentID, Forwarding: wf, BackendRule: bk}
  563. return nil // 此goroutine成功
  564. })
  565. }
  566. // 等待所有goroutine完成
  567. groupErr := g.Wait()
  568. // 关闭通道,表示所有发送者都已完成
  569. // 这一步很重要,这样下面的 range 循环才能正常结束
  570. close(resultsChan)
  571. // 如果 errgroup 捕获到任何错误,优先返回该错误
  572. if groupErr != nil {
  573. // 虽然errgroup已经出错了,但通道中可能已经有一些结果(来自出错前成功或出错的goroutine)
  574. // 我们需要排空通道以避免goroutine泄漏(如果它们在发送时阻塞)
  575. // 但由于我们优先返回groupErr,这些结果将被丢弃。
  576. // 在这种设计下,通常任何一个子任务失败都会导致整个操作失败。
  577. return nil, groupErr
  578. }
  579. // 如果没有错误,收集所有成功的结果
  580. finalResults := make([]v1.WebForwardingDataRequest, 0, len(ids))
  581. for res := range resultsChan {
  582. // 再次检查通道中的错误,尽管 errgroup 应该已经捕获了
  583. // 但这是一种更细致的错误处理,以防万一有goroutine在errgroup.Wait()前发送了错误但未被errgroup捕获
  584. // (理论上,如果goroutine返回了错误,errgroup会处理)
  585. // 主要目的是处理 res.forwarding 为 nil 的情况 (如果上面允许不返回错误)
  586. if res.Err != nil {
  587. // 如果到这里还有错误,说明逻辑可能有问题,或者我们决定忽略某些类型的子错误
  588. // 在此示例中,因为 g.Wait() 没有错误,所以这里的 res.err 应该是nil
  589. // 如果不是,那么可能是goroutine在return nil前发送了带有错误的res。
  590. // 严格来说,如果errgroup没有错误,这里res.err也应该是nil
  591. // 但以防万一,我们可以记录日志
  592. return nil, fmt.Errorf("received error from goroutine for ID %d: %w", res.Id, res.Err)
  593. }
  594. if res.Forwarding == nil {
  595. return nil, fmt.Errorf("received nil forwarding from goroutine for ID %d", res.Id)
  596. }
  597. dataReq := v1.WebForwardingDataRequest{
  598. Id: res.Forwarding.Id,
  599. Port: res.Forwarding.Port,
  600. Domain: res.Forwarding.Domain,
  601. IsHttps: res.Forwarding.IsHttps,
  602. Comment: res.Forwarding.Comment,
  603. HttpsKey: res.Forwarding.HttpsKey,
  604. HttpsCert: res.Forwarding.HttpsCert,
  605. Proxy: res.Forwarding.Proxy,
  606. CcConfig: v1.CcConfigRequest{
  607. IsOn: res.Forwarding.Cc,
  608. ThresholdMethod: res.Forwarding.ThresholdMethod,
  609. Level: res.Forwarding.Level,
  610. Limit5s: res.Forwarding.Limit5s,
  611. Limit60s: res.Forwarding.Limit60s,
  612. Limit300s: res.Forwarding.Limit300s,
  613. },
  614. }
  615. if res.BackendRule != nil { // 只有当 BackendRule 存在时才填充相关字段
  616. dataReq.BackendList = res.BackendRule.BackendList
  617. }
  618. finalResults = append(finalResults, dataReq)
  619. }
  620. sort.Slice(finalResults, func(i, j int) bool {
  621. return finalResults[i].Id > finalResults[j].Id
  622. })
  623. return finalResults, nil
  624. }