tcpforwarding.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454
  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. "golang.org/x/sync/errgroup"
  10. "maps"
  11. "net"
  12. "sort"
  13. )
  14. type TcpforwardingService interface {
  15. GetTcpforwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.TcpForwardingDataRequest, error)
  16. AddTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error
  17. EditTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error
  18. DeleteTcpForwarding(ctx context.Context, req v1.DeleteTcpForwardingRequest) error
  19. GetTcpForwardingAllIpsByHostId(ctx context.Context, req v1.GetForwardingRequest) ([]v1.TcpForwardingDataRequest, error)
  20. }
  21. func NewTcpforwardingService(
  22. service *Service,
  23. tcpforwardingRepository repository.TcpforwardingRepository,
  24. parser ParserService,
  25. required RequiredService,
  26. crawler CrawlerService,
  27. globalRep repository.GlobalLimitRepository,
  28. hostRep repository.HostRepository,
  29. wafformatter WafFormatterService,
  30. cdn CdnService,
  31. ) TcpforwardingService {
  32. return &tcpforwardingService{
  33. Service: service,
  34. tcpforwardingRepository: tcpforwardingRepository,
  35. parser: parser,
  36. required: required,
  37. crawler: crawler,
  38. globalRep: globalRep,
  39. hostRep: hostRep,
  40. wafformatter: wafformatter,
  41. cdn: cdn,
  42. }
  43. }
  44. type tcpforwardingService struct {
  45. *Service
  46. tcpforwardingRepository repository.TcpforwardingRepository
  47. parser ParserService
  48. required RequiredService
  49. crawler CrawlerService
  50. globalRep repository.GlobalLimitRepository
  51. hostRep repository.HostRepository
  52. wafformatter WafFormatterService
  53. cdn CdnService
  54. }
  55. func (s *tcpforwardingService) GetTcpforwarding(ctx context.Context, req v1.GetForwardingRequest) (v1.TcpForwardingDataRequest, error) {
  56. var tcpForwarding model.Tcpforwarding
  57. var backend model.TcpForwardingRule
  58. var err error
  59. g, gCtx := errgroup.WithContext(ctx)
  60. g.Go(func() error {
  61. res, e := s.tcpforwardingRepository.GetTcpforwarding(gCtx, int64(req.Id))
  62. if e != nil {
  63. return fmt.Errorf("GetTcpforwarding failed: %w", e)
  64. }
  65. if res != nil {
  66. tcpForwarding = *res
  67. }
  68. return nil
  69. })
  70. g.Go(func() error {
  71. res, e := s.tcpforwardingRepository.GetTcpForwardingIpsByID(gCtx, req.Id)
  72. if e != nil {
  73. return fmt.Errorf("GetTcpforwardingIps failed: %w", e)
  74. }
  75. if res != nil {
  76. backend = *res
  77. }
  78. return nil
  79. })
  80. if err = g.Wait(); err != nil {
  81. return v1.TcpForwardingDataRequest{}, err
  82. }
  83. return v1.TcpForwardingDataRequest{
  84. Id: tcpForwarding.Id,
  85. Port: tcpForwarding.Port,
  86. Comment: tcpForwarding.Comment,
  87. BackendList: backend.BackendList,
  88. }, nil
  89. }
  90. func (s *tcpforwardingService) buildTcpForwardingModel(req *v1.TcpForwardingDataRequest, ruleId int, require RequireResponse) *model.Tcpforwarding {
  91. return &model.Tcpforwarding{
  92. HostId: require.HostId,
  93. CdnWebId: ruleId,
  94. Port: req.Port,
  95. Comment: req.Comment,
  96. }
  97. }
  98. func (s *tcpforwardingService) buildTcpRuleModel(reqData *v1.TcpForwardingDataRequest, require RequireResponse, localDbId int, cdnOriginIds map[string]int64) *model.TcpForwardingRule {
  99. return &model.TcpForwardingRule{
  100. Uid: require.Uid,
  101. HostId: require.HostId,
  102. TcpId: localDbId, // 关联到本地数据库的主记录 ID
  103. CdnOriginIds: cdnOriginIds,
  104. BackendList: reqData.BackendList,
  105. }
  106. }
  107. func (s *tcpforwardingService) prepareWafData(ctx context.Context, req *v1.TcpForwardingRequest) (RequireResponse, v1.Website, error) {
  108. // 1. 获取必要的全局信息
  109. require, err := s.wafformatter.Require(ctx, v1.GlobalRequire{
  110. HostId: req.HostId,
  111. Uid: req.Uid,
  112. Comment: req.TcpForwardingData.Comment,
  113. })
  114. if err != nil {
  115. return RequireResponse{}, v1.Website{}, err
  116. }
  117. if require.GatewayGroupId == 0 || require.Uid == 0 {
  118. return RequireResponse{}, v1.Website{}, fmt.Errorf("请先配置实例")
  119. }
  120. var jsonData v1.TypeJSON
  121. jsonData.IsOn = true
  122. for _, v := range require.GatewayIps {
  123. jsonData.Listen = append(jsonData.Listen, v1.Listen{
  124. Protocol: "tcp",
  125. Host: v,
  126. Port: req.TcpForwardingData.Port,
  127. })
  128. }
  129. byteData, err := json.Marshal(jsonData)
  130. if err != nil {
  131. return RequireResponse{}, v1.Website{}, err
  132. }
  133. formData := v1.Website{
  134. UserId: int64(require.CdnUid),
  135. Type: "tcpProxy",
  136. Name: require.Tag,
  137. Description: req.TcpForwardingData.Comment,
  138. TcpJSON: byteData,
  139. ServerGroupIds: []int64{int64(require.GroupId)},
  140. UserPlanId: int64(require.RuleId),
  141. NodeClusterId: 1,
  142. }
  143. return require, formData, nil
  144. }
  145. func (s *tcpforwardingService) AddTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error {
  146. require, formData, err := s.prepareWafData(ctx, req)
  147. if err != nil {
  148. return err
  149. }
  150. err = s.wafformatter.validateWafPortCount(ctx, require.HostId)
  151. if err != nil {
  152. return err
  153. }
  154. tcpId, err := s.cdn.CreateWebsite(ctx, formData)
  155. if err != nil {
  156. return err
  157. }
  158. // 添加源站
  159. cdnOriginIds := make(map[string]int64)
  160. for _, v := range req.TcpForwardingData.BackendList{
  161. id, err := s.wafformatter.AddOrigin(ctx, v1.WebJson{
  162. ApiType: "tcp",
  163. BackendList: v,
  164. Comment: req.TcpForwardingData.Comment,
  165. })
  166. if err != nil {
  167. return err
  168. }
  169. cdnOriginIds[v] = id
  170. }
  171. // 添加源站到网站
  172. for _, v := range cdnOriginIds {
  173. err = s.cdn.AddServerOrigin(ctx, tcpId, v)
  174. if err != nil {
  175. return err
  176. }
  177. }
  178. tcpModel := s.buildTcpForwardingModel(&req.TcpForwardingData, int(tcpId), require)
  179. id, err := s.tcpforwardingRepository.AddTcpforwarding(ctx, tcpModel)
  180. if err != nil {
  181. return err
  182. }
  183. TcpRuleModel := s.buildTcpRuleModel(&req.TcpForwardingData, require, id, cdnOriginIds)
  184. if _, err = s.tcpforwardingRepository.AddTcpforwardingIps(ctx, *TcpRuleModel); err != nil {
  185. return err
  186. }
  187. // 异步任务:将源站IP添加到白名单
  188. var ips []string
  189. if req.TcpForwardingData.BackendList != nil {
  190. for _, v := range req.TcpForwardingData.BackendList {
  191. ip, _, err := net.SplitHostPort(v)
  192. if err != nil {
  193. return err
  194. }
  195. ips = append(ips, ip)
  196. }
  197. go s.wafformatter.PublishIpWhitelistTask(ips, "add","","white")
  198. }
  199. return nil
  200. }
  201. func (s *tcpforwardingService) EditTcpForwarding(ctx context.Context, req *v1.TcpForwardingRequest) error {
  202. require, formData, err := s.prepareWafData(ctx, req)
  203. if err != nil {
  204. return err
  205. }
  206. oldData, err := s.tcpforwardingRepository.GetTcpforwarding(ctx, int64(req.TcpForwardingData.Id))
  207. if err != nil {
  208. return err
  209. }
  210. //修改网站端口
  211. if oldData.Port != req.TcpForwardingData.Port {
  212. err = s.cdn.EditServerType(ctx, v1.EditWebsite{
  213. Id: int64(oldData.CdnWebId),
  214. TypeJSON: formData.TcpJSON,
  215. }, "tcp")
  216. if err != nil {
  217. return err
  218. }
  219. }
  220. //修改网站名字
  221. if oldData.Comment != req.TcpForwardingData.Comment {
  222. nodeId, err := s.globalRep.GetNodeId(ctx, oldData.CdnWebId)
  223. err = s.cdn.EditServerBasic(ctx, int64(oldData.CdnWebId), require.Tag, nodeId)
  224. if err != nil {
  225. return err
  226. }
  227. }
  228. // 异步任务:将IP添加到白名单
  229. ipData, err := s.tcpforwardingRepository.GetTcpForwardingIpsByID(ctx, req.TcpForwardingData.Id)
  230. if err != nil {
  231. return err
  232. }
  233. addedIps, removedIps, err := s.wafformatter.WashEditWafIp(ctx,req.TcpForwardingData.BackendList, ipData.BackendList)
  234. if err != nil {
  235. return err
  236. }
  237. if len(addedIps) > 0 {
  238. go s.wafformatter.PublishIpWhitelistTask(addedIps, "add","","white")
  239. }
  240. if len(removedIps) > 0 {
  241. ipsToDelist, err := s.wafformatter.WashDelIps(ctx, removedIps)
  242. if err != nil {
  243. return err
  244. }
  245. // 4. 如果有需要处理的IP,则批量发布一次任务
  246. if len(ipsToDelist) > 0 {
  247. go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white")
  248. }
  249. }
  250. //修改源站
  251. addOrigins, delOrigins := s.wafformatter.findIpDifferences(ipData.BackendList, req.TcpForwardingData.BackendList)
  252. addedIds := make(map[string]int64)
  253. for _, v := range addOrigins {
  254. id, err := s.wafformatter.AddOrigin(ctx,v1.WebJson{
  255. ApiType: "tcp",
  256. BackendList: v,
  257. Comment: req.TcpForwardingData.Comment,
  258. })
  259. if err != nil {
  260. return err
  261. }
  262. addedIds[v] = id
  263. }
  264. for _, v := range addedIds {
  265. err = s.cdn.AddServerOrigin(ctx, int64(oldData.CdnWebId), v)
  266. if err != nil {
  267. return err
  268. }
  269. }
  270. maps.Copy(ipData.CdnOriginIds, addedIds)
  271. for k, v := range ipData.CdnOriginIds {
  272. for _, ip := range delOrigins {
  273. if k == ip {
  274. err = s.cdn.DelServerOrigin(ctx, int64(oldData.CdnWebId), v)
  275. if err != nil {
  276. return err
  277. }
  278. delete(ipData.CdnOriginIds, k)
  279. }
  280. }
  281. }
  282. tcpModel := s.buildTcpForwardingModel(&req.TcpForwardingData,oldData.CdnWebId, require)
  283. tcpModel.Id = req.TcpForwardingData.Id
  284. if err = s.tcpforwardingRepository.EditTcpforwarding(ctx, tcpModel); err != nil {
  285. return err
  286. }
  287. TcpRuleModel := s.buildTcpRuleModel(&req.TcpForwardingData, require, req.TcpForwardingData.Id, ipData.CdnOriginIds)
  288. if err = s.tcpforwardingRepository.EditTcpforwardingIps(ctx, *TcpRuleModel); err != nil {
  289. return err
  290. }
  291. return nil
  292. }
  293. func (s *tcpforwardingService) DeleteTcpForwarding(ctx context.Context, req v1.DeleteTcpForwardingRequest) error {
  294. for _, Id := range req.Ids {
  295. oldData, err := s.tcpforwardingRepository.GetTcpforwarding(ctx, int64(Id))
  296. if err != nil {
  297. return err
  298. }
  299. err = s.cdn.DelServer(ctx, int64(oldData.CdnWebId))
  300. if err != nil {
  301. return err
  302. }
  303. // 删除白名单
  304. var ips []string
  305. ipData, err := s.tcpforwardingRepository.GetTcpForwardingIpsByID(ctx, Id)
  306. if err != nil {
  307. return err
  308. }
  309. ips, err = s.wafformatter.WashDeleteWafIp(ctx, ipData.BackendList)
  310. if err != nil {
  311. return err
  312. }
  313. if len(ips) > 0 {
  314. ipsToDelist, err := s.wafformatter.WashDelIps(ctx, ips)
  315. if err != nil {
  316. return err
  317. }
  318. // 4. 如果有需要处理的IP,则批量发布一次任务
  319. if len(ipsToDelist) > 0 {
  320. go s.wafformatter.PublishIpWhitelistTask(ipsToDelist, "del", "0", "white")
  321. }
  322. }
  323. if err = s.tcpforwardingRepository.DeleteTcpforwarding(ctx, int64(Id)); err != nil {
  324. return err
  325. }
  326. if err = s.tcpforwardingRepository.DeleteTcpForwardingIpsById(ctx, Id); err != nil {
  327. return err
  328. }
  329. }
  330. return nil
  331. }
  332. func (s *tcpforwardingService) GetTcpForwardingAllIpsByHostId(ctx context.Context, req v1.GetForwardingRequest) ([]v1.TcpForwardingDataRequest, error) {
  333. type CombinedResult struct {
  334. Id int
  335. Forwarding *model.Tcpforwarding
  336. BackendRule *model.TcpForwardingRule
  337. Err error // 如果此ID的处理出错,则携带错误
  338. }
  339. g,gCtx := errgroup.WithContext(ctx)
  340. ids, err := s.tcpforwardingRepository.GetTcpForwardingAllIdsByID(gCtx, req.HostId)
  341. if err != nil {
  342. return nil, fmt.Errorf("GetTcpForwardingAllIds failed: %w", err)
  343. }
  344. if len(ids) == 0 {
  345. return nil, nil
  346. }
  347. resChan := make(chan CombinedResult, len(ids))
  348. g.Go(func() error {
  349. for _, idVal := range ids {
  350. currentID := idVal
  351. g.Go(func() error {
  352. var wf *model.Tcpforwarding
  353. var bk *model.TcpForwardingRule
  354. var localErr error
  355. wf, localErr = s.tcpforwardingRepository.GetTcpforwarding(gCtx, int64(currentID))
  356. if localErr != nil {
  357. resChan <- CombinedResult{Id: currentID, Err: localErr}
  358. return localErr
  359. }
  360. bk, localErr = s.tcpforwardingRepository.GetTcpForwardingIpsByID(gCtx, currentID)
  361. if localErr != nil {
  362. resChan <- CombinedResult{Id: currentID, Err: localErr}
  363. return localErr
  364. }
  365. resChan <- CombinedResult{Id: currentID, Forwarding: wf, BackendRule: bk}
  366. return nil
  367. })
  368. }
  369. return nil
  370. })
  371. groupErr := g.Wait()
  372. close(resChan)
  373. if groupErr != nil {
  374. return nil, groupErr
  375. }
  376. res := make([]v1.TcpForwardingDataRequest, 0, len(ids))
  377. for r := range resChan {
  378. if r.Err != nil {
  379. return nil, fmt.Errorf("received error from goroutine for ID %d: %w", r.Id, r.Err)
  380. }
  381. if r.Forwarding == nil {
  382. return nil,fmt.Errorf("received nil forwarding from goroutine for ID %d", r.Id)
  383. }
  384. dataReq := v1.TcpForwardingDataRequest{
  385. Id: r.Forwarding.Id,
  386. Port: r.Forwarding.Port,
  387. Comment: r.Forwarding.Comment,
  388. }
  389. if r.BackendRule != nil {
  390. dataReq.BackendList = r.BackendRule.BackendList
  391. }
  392. res = append(res, dataReq)
  393. }
  394. sort.Slice(res, func(i, j int) bool {
  395. return res[i].Id > res[j].Id
  396. })
  397. return res, nil
  398. }