webforwarding.go 22 KB

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