webforwarding.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631
  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. amqp "github.com/rabbitmq/amqp091-go"
  11. "go.uber.org/zap"
  12. "golang.org/x/sync/errgroup"
  13. "sort"
  14. "strconv"
  15. "strings"
  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) error
  21. EditWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error
  22. DeleteWebForwarding(ctx context.Context, Ids []int) error
  23. }
  24. func NewWebForwardingService(
  25. service *Service,
  26. required RequiredService,
  27. webForwardingRepository repository.WebForwardingRepository,
  28. crawler CrawlerService,
  29. parser ParserService,
  30. wafformatter WafFormatterService,
  31. aoDun AoDunService,
  32. mq *rabbitmq.RabbitMQ,
  33. gatewayGroupIpRep repository.GateWayGroupIpRepository,
  34. gatewayGroupRep repository.GatewayGroupRepository,
  35. ) WebForwardingService {
  36. return &webForwardingService{
  37. Service: service,
  38. webForwardingRepository: webForwardingRepository,
  39. required: required,
  40. parser: parser,
  41. crawler: crawler,
  42. wafformatter: wafformatter,
  43. aoDun: aoDun,
  44. mq: mq,
  45. gatewayGroupIpRep: gatewayGroupIpRep,
  46. gatewayGroupRep: gatewayGroupRep,
  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. }
  61. func (s *webForwardingService) require(ctx context.Context,req v1.GlobalRequire) (v1.GlobalRequire, error) {
  62. var err error
  63. var res v1.GlobalRequire
  64. g, gCtx := errgroup.WithContext(ctx)
  65. g.Go(func() error {
  66. result, e := s.wafformatter.require(gCtx, req, "web")
  67. if e != nil {
  68. return e
  69. }
  70. res = result
  71. return nil
  72. })
  73. g.Go(func() error {
  74. e := s.wafformatter.validateWafDomainCount(gCtx, req)
  75. if e != nil {
  76. return e
  77. }
  78. return nil
  79. })
  80. if err = g.Wait(); err != nil {
  81. return v1.GlobalRequire{}, err
  82. }
  83. return res, nil
  84. }
  85. func (s *webForwardingService) GetWebForwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.WebForwardingDataRequest, error) {
  86. var webForwarding model.WebForwarding
  87. var backend model.WebForwardingRule
  88. g, gCtx := errgroup.WithContext(ctx)
  89. g.Go(func() error {
  90. res, e := s.webForwardingRepository.GetWebForwarding(gCtx, int64(req.Id))
  91. if e != nil {
  92. // 直接返回错误,errgroup 会捕获它
  93. return fmt.Errorf("GetWebForwarding failed: %w", e)
  94. }
  95. if res != nil {
  96. webForwarding = *res
  97. }
  98. return nil
  99. })
  100. g.Go(func() error {
  101. res, e := s.webForwardingRepository.GetWebForwardingIpsByID(ctx, req.Id)
  102. if e != nil {
  103. return fmt.Errorf("GetWebForwardingByID failed: %w", e)
  104. }
  105. if res != nil {
  106. backend = *res
  107. }
  108. return nil
  109. })
  110. if err := g.Wait(); err != nil {
  111. return v1.WebForwardingDataRequest{}, err
  112. }
  113. return v1.WebForwardingDataRequest{
  114. Id: webForwarding.Id,
  115. WafWebId: webForwarding.WafWebId,
  116. Tag: webForwarding.Tag,
  117. Port: webForwarding.Port,
  118. Domain: webForwarding.Domain,
  119. CustomHost: webForwarding.CustomHost,
  120. WafWebLimitId: webForwarding.WebLimitRuleId,
  121. WafGatewayGroupId: webForwarding.WafGatewayGroupId,
  122. CcCount: webForwarding.CcCount,
  123. CcDuration: webForwarding.CcDuration,
  124. CcBlockCount: webForwarding.CcBlockCount,
  125. CcBlockDuration: webForwarding.CcBlockDuration,
  126. Cc4xxCount: webForwarding.Cc4xxCount,
  127. Cc4xxDuration: webForwarding.Cc4xxDuration,
  128. Cc4xxBlockCount: webForwarding.Cc4xxBlockCount,
  129. Cc4xxBlockDuration: webForwarding.Cc4xxBlockDuration,
  130. Cc5xxCount: webForwarding.Cc5xxCount,
  131. Cc5xxDuration: webForwarding.Cc5xxDuration,
  132. Cc5xxBlockCount: webForwarding.Cc5xxBlockCount,
  133. Cc5xxBlockDuration: webForwarding.Cc5xxBlockDuration,
  134. IsHttps: webForwarding.IsHttps,
  135. Comment: webForwarding.Comment,
  136. BackendList: backend.BackendList,
  137. AllowIpList: backend.AllowIpList,
  138. DenyIpList: backend.DenyIpList,
  139. AccessRule: backend.AccessRule,
  140. HttpsKey: webForwarding.HttpsKey,
  141. HttpsCert: webForwarding.HttpsCert,
  142. }, nil
  143. }
  144. // buildWafFormData 辅助函数,用于构建通用的 formData
  145. func (s *webForwardingService) buildWafFormData(req *v1.WebForwardingDataSend, require v1.GlobalRequire) map[string]interface{} {
  146. // 将BackendList序列化为JSON字符串
  147. backendJSON, err := json.MarshalIndent(req.BackendList, "", " ")
  148. var backendStr interface{}
  149. if err != nil {
  150. // 如果序列化失败,使用空数组
  151. backendStr = "[]"
  152. } else {
  153. // 成功序列化后,使用JSON字符串
  154. backendStr = string(backendJSON)
  155. }
  156. return map[string]interface{}{
  157. "waf_web_id": req.WafWebId,
  158. "tag": require.Tag,
  159. "port": req.Port,
  160. "domain": req.Domain,
  161. "custom_host": req.CustomHost,
  162. "waf_gateway_group_id": require.WafGatewayGroupId,
  163. "waf_web_limit_id": require.LimitRuleId,
  164. "cc_count": req.CcCount,
  165. "cc_duration": req.CcDuration,
  166. "cc_block_count": req.CcBlockCount,
  167. "cc_block_duration": req.CcBlockDuration,
  168. "cc_4xx_count": req.Cc4xxCount,
  169. "cc_4xx_duration": req.Cc4xxDuration,
  170. "cc_4xx_block_count": req.Cc4xxBlockCount,
  171. "cc_4xx_block_duration": req.Cc4xxBlockDuration,
  172. "cc_5xx_count": req.Cc5xxCount,
  173. "cc_5xx_duration": req.Cc5xxDuration,
  174. "cc_5xx_block_count": req.Cc5xxBlockCount,
  175. "cc_5xx_block_duration": req.Cc5xxBlockDuration,
  176. "backend": backendStr,
  177. "allow_ip_list": req.AllowIpList,
  178. "deny_ip_list": req.DenyIpList,
  179. "access_rule": req.AccessRule,
  180. "is_https": req.IsHttps,
  181. "comment": req.Comment,
  182. "https_cert": req.HttpsCert,
  183. "https_key": req.HttpsKey,
  184. }
  185. }
  186. // buildWebForwardingModel 辅助函数,用于构建通用的 WebForwarding 模型
  187. // ruleId 是从 WAF 系统获取的 ID
  188. func (s *webForwardingService) buildWebForwardingModel(req *v1.WebForwardingDataRequest,ruleId int, require v1.GlobalRequire) *model.WebForwarding {
  189. return &model.WebForwarding{
  190. HostId: require.HostId,
  191. WafWebId: ruleId,
  192. Tag: require.Tag,
  193. Port: req.Port,
  194. Domain: req.Domain,
  195. CustomHost: req.CustomHost,
  196. WafGatewayGroupId: require.WafGatewayGroupId,
  197. WebLimitRuleId: require.LimitRuleId,
  198. CcCount: req.CcCount,
  199. CcDuration: req.CcDuration,
  200. CcBlockCount: req.CcBlockCount,
  201. CcBlockDuration: req.CcBlockDuration,
  202. Cc4xxCount: req.Cc4xxCount,
  203. Cc4xxDuration: req.Cc4xxDuration,
  204. Cc4xxBlockCount: req.Cc4xxBlockCount,
  205. Cc4xxBlockDuration: req.Cc4xxBlockDuration,
  206. Cc5xxCount: req.Cc5xxCount,
  207. Cc5xxDuration: req.Cc5xxDuration,
  208. Cc5xxBlockCount: req.Cc5xxBlockCount,
  209. Cc5xxBlockDuration: req.Cc5xxBlockDuration,
  210. IsHttps: req.IsHttps,
  211. Comment: req.Comment,
  212. HttpsCert: req.HttpsCert,
  213. HttpsKey: req.HttpsKey,
  214. }
  215. }
  216. func (s *webForwardingService) buildWebRuleModel(reqData *v1.WebForwardingDataRequest, require v1.GlobalRequire, localDbId int) *model.WebForwardingRule {
  217. return &model.WebForwardingRule{
  218. Uid: require.Uid,
  219. HostId: require.HostId,
  220. WebId: localDbId, // 关联到本地数据库的主记录 ID
  221. BackendList: reqData.BackendList,
  222. AllowIpList: reqData.AllowIpList,
  223. DenyIpList: reqData.DenyIpList,
  224. AccessRule: reqData.AccessRule,
  225. }
  226. }
  227. func (s *webForwardingService) prepareWafData(ctx context.Context, req *v1.WebForwardingRequest) (v1.GlobalRequire, map[string]interface{}, error) {
  228. // 1. 获取必要的全局信息
  229. require, err := s.require(ctx, v1.GlobalRequire{
  230. HostId: req.HostId,
  231. Uid: req.Uid,
  232. Comment: req.WebForwardingData.Comment,
  233. Domain: req.WebForwardingData.Domain,
  234. })
  235. if err != nil {
  236. return v1.GlobalRequire{}, nil, err
  237. }
  238. if require.WafGatewayGroupId == 0 || require.LimitRuleId == 0 {
  239. return v1.GlobalRequire{}, nil, fmt.Errorf("请先配置实例")
  240. }
  241. // 2. 将字符串切片拼接成字符串,用于 WAF API
  242. allowIpListStr := strings.Join(req.WebForwardingData.AllowIpList, "\n")
  243. denyIpListStr := strings.Join(req.WebForwardingData.DenyIpList, "\n")
  244. PortInt, err := strconv.Atoi(req.WebForwardingData.Port)
  245. if err != nil {
  246. return v1.GlobalRequire{}, nil, err
  247. }
  248. // 3. 创建用于构建 WAF 表单的数据结构
  249. formDataBase := v1.WebForwardingDataSend{
  250. Tag: require.Tag,
  251. WafWebId: req.WebForwardingData.WafWebId,
  252. WafGatewayGroupId: require.WafGatewayGroupId,
  253. WafWebLimitId: require.LimitRuleId,
  254. Port: PortInt,
  255. Domain: req.WebForwardingData.Domain,
  256. CustomHost: req.WebForwardingData.CustomHost,
  257. CcCount: req.WebForwardingData.CcCount,
  258. CcDuration: req.WebForwardingData.CcDuration,
  259. CcBlockCount: req.WebForwardingData.CcBlockCount,
  260. CcBlockDuration: req.WebForwardingData.CcBlockDuration,
  261. Cc4xxCount: req.WebForwardingData.Cc4xxCount,
  262. Cc4xxDuration: req.WebForwardingData.Cc4xxDuration,
  263. Cc4xxBlockCount: req.WebForwardingData.Cc4xxBlockCount,
  264. Cc4xxBlockDuration: req.WebForwardingData.Cc4xxBlockDuration,
  265. Cc5xxCount: req.WebForwardingData.Cc5xxCount,
  266. Cc5xxDuration: req.WebForwardingData.Cc5xxDuration,
  267. Cc5xxBlockCount: req.WebForwardingData.Cc5xxBlockCount,
  268. Cc5xxBlockDuration: req.WebForwardingData.Cc5xxBlockDuration,
  269. IsHttps: req.WebForwardingData.IsHttps,
  270. BackendList: req.WebForwardingData.BackendList,
  271. AllowIpList: allowIpListStr,
  272. DenyIpList: denyIpListStr,
  273. AccessRule: req.WebForwardingData.AccessRule,
  274. Comment: req.WebForwardingData.Comment,
  275. HttpsCert: req.WebForwardingData.HttpsCert,
  276. HttpsKey: req.WebForwardingData.HttpsKey,
  277. }
  278. // 4. 构建 WAF 表单数据映射
  279. formData := s.buildWafFormData(&formDataBase, require)
  280. return require, formData, nil
  281. }
  282. func (s *webForwardingService) GetIp(ctx context.Context, gatewayGroupId int) (string, error) {
  283. WafGatewayGroupRuleId, err := s.gatewayGroupRep.GetGatewayGroupByRuleId(ctx, int64(gatewayGroupId))
  284. if err != nil {
  285. return "", err
  286. }
  287. ip, err := s.gatewayGroupIpRep.GetGateWayGroupFirstIpByGatewayGroupId(ctx, WafGatewayGroupRuleId.Id)
  288. if err != nil {
  289. return "", err
  290. }
  291. return ip, nil
  292. }
  293. func (s *webForwardingService) AddWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error {
  294. require, formData, err := s.prepareWafData(ctx, req)
  295. if err != nil {
  296. return err
  297. }
  298. err = s.wafformatter.validateWafPortCount(ctx, require.HostId)
  299. if err != nil {
  300. return err
  301. }
  302. wafWebId, err := s.wafformatter.sendFormData(ctx, "admin/info/waf_web/new", "admin/new/waf_web", formData)
  303. if err != nil {
  304. return err
  305. }
  306. if req.WebForwardingData.Domain != "" {
  307. ip, err := s.GetIp(ctx, require.WafGatewayGroupId)
  308. if err != nil {
  309. return err
  310. }
  311. // 异步任务:将域名添加到白名单
  312. doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, req.WebForwardingData.Domain)
  313. if err != nil {
  314. return err
  315. }
  316. go s.publishDomainWhitelistTask(doMain,ip, "add")
  317. }
  318. webModel := s.buildWebForwardingModel(&req.WebForwardingData, wafWebId, require)
  319. id, err := s.webForwardingRepository.AddWebForwarding(ctx, webModel)
  320. if err != nil {
  321. return err
  322. }
  323. webRuleModel := s.buildWebRuleModel(&req.WebForwardingData, require, id)
  324. //var ips []v1.IpInfo
  325. //for _, v := range req.WebForwardingData.AllowIpList {
  326. // ips = append(ips, v1.IpInfo{
  327. // FType: "allow",
  328. // FStartIp: v,
  329. // FEndIp: v,
  330. // FRemark: "宁波高防IP过白",
  331. // FServerIp: "",
  332. // })
  333. //}
  334. //err = s.aoDun.AddDomainWhiteList(ctx, req.WebForwardingData.Domain)
  335. if _, err = s.webForwardingRepository.AddWebForwardingIps(ctx, *webRuleModel); err != nil {
  336. return err
  337. }
  338. return nil
  339. }
  340. func (s *webForwardingService) EditWebForwarding(ctx context.Context, req *v1.WebForwardingRequest) error {
  341. WafWebId, err := s.webForwardingRepository.GetWebForwardingWafWebIdById(ctx, req.WebForwardingData.Id)
  342. if err != nil {
  343. return err
  344. }
  345. req.WebForwardingData.WafWebId = WafWebId
  346. require, formData, err := s.prepareWafData(ctx, req)
  347. if err != nil {
  348. return err
  349. }
  350. _, err = s.wafformatter.sendFormData(ctx, "admin/info/waf_web/edit?&__goadmin_edit_pk="+strconv.Itoa(req.WebForwardingData.WafWebId), "admin/edit/waf_web", formData)
  351. if err != nil {
  352. return err
  353. }
  354. webData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(req.WebForwardingData.Id))
  355. if err != nil {
  356. return err
  357. }
  358. if webData.Domain != req.WebForwardingData.Domain {
  359. Ip, err := s.GetIp(ctx, webData.WafGatewayGroupId)
  360. if err != nil {
  361. return err
  362. }
  363. // 异步任务:将域名添加到白名单
  364. doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, req.WebForwardingData.Domain)
  365. if err != nil {
  366. return err
  367. }
  368. go s.publishDomainWhitelistTask(webData.Domain, Ip, "del")
  369. go s.publishDomainWhitelistTask(doMain, Ip, "add")
  370. }
  371. webModel := s.buildWebForwardingModel(&req.WebForwardingData, req.WebForwardingData.WafWebId, require)
  372. webModel.Id = req.WebForwardingData.Id
  373. if err = s.webForwardingRepository.EditWebForwarding(ctx, webModel); err != nil {
  374. return err
  375. }
  376. webRuleModel := s.buildWebRuleModel(&req.WebForwardingData, require, req.WebForwardingData.Id)
  377. if err = s.webForwardingRepository.EditWebForwardingIps(ctx, *webRuleModel); err != nil {
  378. return err
  379. }
  380. return nil
  381. }
  382. func (s *webForwardingService) DeleteWebForwarding(ctx context.Context, Ids []int) error {
  383. for _, Id := range Ids {
  384. wafWebId, err := s.webForwardingRepository.GetWebForwardingWafWebIdById(ctx, Id)
  385. if err != nil {
  386. return err
  387. }
  388. webData, err := s.webForwardingRepository.GetWebForwarding(ctx, int64(Id))
  389. if err != nil {
  390. return err
  391. }
  392. if webData.Domain != "" {
  393. ip , err := s.GetIp(ctx, webData.WafGatewayGroupId)
  394. if err != nil {
  395. return err
  396. }
  397. doMain, err := s.wafformatter.ConvertToWildcardDomain(ctx, webData.Domain)
  398. if err != nil {
  399. return err
  400. }
  401. go s.publishDomainWhitelistTask(doMain,ip, "del")
  402. }
  403. _, err = s.crawler.DeleteRule(ctx, wafWebId, "admin/delete/waf_web?page=1&__pageSize=10&__sort=waf_web_id&__sort_type=desc")
  404. if err != nil {
  405. return err
  406. }
  407. if err = s.webForwardingRepository.DeleteWebForwarding(ctx, int64(Id)); err != nil {
  408. return err
  409. }
  410. if err = s.webForwardingRepository.DeleteWebForwardingIpsById(ctx, Id); err != nil {
  411. return err
  412. }
  413. }
  414. return nil
  415. }
  416. func (s *webForwardingService) GetWebForwardingWafWebAllIps(ctx context.Context, req v1.GetForwardingRequest) ([]v1.WebForwardingDataRequest, error) {
  417. type CombinedResult struct {
  418. Id int
  419. Forwarding *model.WebForwarding
  420. BackendRule *model.WebForwardingRule
  421. Err error // 如果此ID的处理出错,则携带错误
  422. }
  423. g, gCtx := errgroup.WithContext(ctx)
  424. ids, err := s.webForwardingRepository.GetWebForwardingWafWebAllIds(gCtx, req.HostId)
  425. if err != nil {
  426. return nil, fmt.Errorf("GetWebForwardingWafWebAllIds failed: %w", err)
  427. }
  428. if len(ids) == 0 {
  429. return nil, nil // 没有ID,直接返回空切片
  430. }
  431. // 创建一个通道来接收每个ID的处理结果
  432. // 通道缓冲区大小设为ID数量,这样发送者不会因为接收者慢而阻塞(在所有goroutine都启动后)
  433. resultsChan := make(chan CombinedResult, len(ids))
  434. for _, idVal := range ids {
  435. currentID := idVal // 捕获循环变量
  436. g.Go(func() error {
  437. var wf *model.WebForwarding
  438. var bk *model.WebForwardingRule
  439. var localErr error
  440. // 1. 获取 WebForwarding 信息
  441. wf, localErr = s.webForwardingRepository.GetWebForwarding(gCtx, int64(currentID))
  442. if localErr != nil {
  443. // 发送错误到通道,并由 errgroup 捕获
  444. // errgroup 会处理第一个非nil错误,并取消其他 goroutine
  445. resultsChan <- CombinedResult{Id: currentID, Err: fmt.Errorf("GetWebForwarding for id %d failed: %w", currentID, localErr)}
  446. return localErr // 返回错误给 errgroup
  447. }
  448. if wf == nil { // 正常情况下,如果没错误,wf不应为nil,但防御性检查
  449. localErr = fmt.Errorf("GetWebForwarding for id %d returned nil data without error", currentID)
  450. resultsChan <- CombinedResult{Id: currentID, Err: localErr}
  451. return localErr
  452. }
  453. // 2. 获取 Backend IP 信息
  454. // 注意:这里我们允许 GetWebForwardingIpsByID 可能返回 nil 数据(例如没有规则)而不是错误
  455. // 如果它也可能返回错误,则处理方式与上面类似
  456. bk, localErr = s.webForwardingRepository.GetWebForwardingIpsByID(gCtx, currentID)
  457. if localErr != nil {
  458. // 如果获取IP信息失败是一个致命错误,则也应返回错误
  459. // 如果允许部分成功(比如有WebForwarding但没有IP信息),则可以不将此视为errgroup的错误
  460. // 这里假设它也是一个需要errgroup捕获的错误
  461. resultsChan <- CombinedResult{Id: currentID, Forwarding: wf, Err: fmt.Errorf("GetWebForwardingIpsByID for id %d failed: %w", currentID, localErr)}
  462. return localErr // 返回错误给 errgroup
  463. }
  464. // bk 可能是 nil 如果没有错误且没有规则,这取决于业务逻辑
  465. // 发送成功的结果到通道
  466. resultsChan <- CombinedResult{Id: currentID, Forwarding: wf, BackendRule: bk}
  467. return nil // 此goroutine成功
  468. })
  469. }
  470. // 等待所有goroutine完成
  471. groupErr := g.Wait()
  472. // 关闭通道,表示所有发送者都已完成
  473. // 这一步很重要,这样下面的 range 循环才能正常结束
  474. close(resultsChan)
  475. // 如果 errgroup 捕获到任何错误,优先返回该错误
  476. if groupErr != nil {
  477. // 虽然errgroup已经出错了,但通道中可能已经有一些结果(来自出错前成功或出错的goroutine)
  478. // 我们需要排空通道以避免goroutine泄漏(如果它们在发送时阻塞)
  479. // 但由于我们优先返回groupErr,这些结果将被丢弃。
  480. // 在这种设计下,通常任何一个子任务失败都会导致整个操作失败。
  481. return nil, groupErr
  482. }
  483. // 如果没有错误,收集所有成功的结果
  484. finalResults := make([]v1.WebForwardingDataRequest, 0, len(ids))
  485. for res := range resultsChan {
  486. // 再次检查通道中的错误,尽管 errgroup 应该已经捕获了
  487. // 但这是一种更细致的错误处理,以防万一有goroutine在errgroup.Wait()前发送了错误但未被errgroup捕获
  488. // (理论上,如果goroutine返回了错误,errgroup会处理)
  489. // 主要目的是处理 res.forwarding 为 nil 的情况 (如果上面允许不返回错误)
  490. if res.Err != nil {
  491. // 如果到这里还有错误,说明逻辑可能有问题,或者我们决定忽略某些类型的子错误
  492. // 在此示例中,因为 g.Wait() 没有错误,所以这里的 res.err 应该是nil
  493. // 如果不是,那么可能是goroutine在return nil前发送了带有错误的res。
  494. // 严格来说,如果errgroup没有错误,这里res.err也应该是nil
  495. // 但以防万一,我们可以记录日志
  496. return nil, fmt.Errorf("received error from goroutine for ID %d: %w", res.Id, res.Err)
  497. }
  498. if res.Forwarding == nil {
  499. return nil, fmt.Errorf("received nil forwarding from goroutine for ID %d", res.Id)
  500. }
  501. dataReq := v1.WebForwardingDataRequest{
  502. Id: res.Forwarding.Id,
  503. Port: res.Forwarding.Port,
  504. Domain: res.Forwarding.Domain,
  505. CustomHost: res.Forwarding.CustomHost,
  506. CcCount: res.Forwarding.CcCount,
  507. CcDuration: res.Forwarding.CcDuration,
  508. CcBlockCount: res.Forwarding.CcBlockCount,
  509. CcBlockDuration: res.Forwarding.CcBlockDuration,
  510. Cc4xxCount: res.Forwarding.Cc4xxCount,
  511. Cc4xxDuration: res.Forwarding.Cc4xxDuration,
  512. Cc4xxBlockCount: res.Forwarding.Cc4xxBlockCount,
  513. Cc4xxBlockDuration: res.Forwarding.Cc4xxBlockDuration,
  514. Cc5xxCount: res.Forwarding.Cc5xxCount,
  515. Cc5xxDuration: res.Forwarding.Cc5xxDuration,
  516. Cc5xxBlockCount: res.Forwarding.Cc5xxBlockCount,
  517. Cc5xxBlockDuration: res.Forwarding.Cc5xxBlockDuration,
  518. IsHttps: res.Forwarding.IsHttps,
  519. Comment: res.Forwarding.Comment,
  520. HttpsKey: res.Forwarding.HttpsKey,
  521. HttpsCert: res.Forwarding.HttpsCert,
  522. }
  523. if res.BackendRule != nil { // 只有当 BackendRule 存在时才填充相关字段
  524. dataReq.BackendList = res.BackendRule.BackendList
  525. dataReq.AllowIpList = res.BackendRule.AllowIpList
  526. dataReq.DenyIpList = res.BackendRule.DenyIpList
  527. dataReq.AccessRule = res.BackendRule.AccessRule
  528. }
  529. finalResults = append(finalResults, dataReq)
  530. }
  531. sort.Slice(finalResults, func(i, j int) bool {
  532. return finalResults[i].Id > finalResults[j].Id
  533. })
  534. return finalResults, nil
  535. }
  536. // publishDomainWhitelistTask is a helper function to publish domain whitelist tasks to RabbitMQ.
  537. // It can handle different actions like "add" or "del".
  538. func (s *webForwardingService) publishDomainWhitelistTask(domain, ip, action string) {
  539. // Define message payload, including the action
  540. type domainTaskPayload struct {
  541. Domain string `json:"domain"`
  542. Ip string `json:"ip"`
  543. Action string `json:"action"`
  544. }
  545. payload := domainTaskPayload{
  546. Domain: domain,
  547. Ip: ip,
  548. Action: action,
  549. }
  550. // Serialize the message
  551. msgBody, err := json.Marshal(payload)
  552. if err != nil {
  553. s.logger.Error("Failed to serialize domain whitelist task message", zap.Error(err), zap.String("domain", domain), zap.String("ip", ip), zap.String("action", action))
  554. return
  555. }
  556. // Get task configuration
  557. taskCfg, ok := s.mq.GetTaskConfig("domain_whitelist")
  558. if !ok {
  559. s.logger.Error("Failed to get 'domain_whitelist' task configuration")
  560. return
  561. }
  562. // Construct the routing key dynamically based on the action
  563. routingKey := fmt.Sprintf("whitelist.domain.%s", action)
  564. // Construct the amqp.Publishing message
  565. publishingMsg := amqp.Publishing{
  566. ContentType: "application/json",
  567. Body: msgBody,
  568. DeliveryMode: amqp.Persistent, // Persistent message
  569. }
  570. // Publish the message
  571. err = s.mq.PublishWithCh(taskCfg.Exchange, routingKey, publishingMsg)
  572. if err != nil {
  573. s.logger.Error("Failed to publish domain whitelist task to MQ", zap.Error(err), zap.String("domain", domain), zap.String("action", action))
  574. } else {
  575. s.logger.Info("Successfully published domain whitelist task to MQ", zap.String("domain", domain), zap.String("action", action))
  576. }
  577. }