蛋蛋星球RabbitMq消费项目
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

267 lines
8.4 KiB

  1. package consume
  2. import (
  3. "applet/app/cfg"
  4. "applet/app/db"
  5. utils2 "applet/app/utils"
  6. "applet/app/utils/logx"
  7. "applet/consume/md"
  8. "code.fnuoos.com/EggPlanet/egg_models.git/src/implement"
  9. "code.fnuoos.com/EggPlanet/egg_models.git/src/model"
  10. "code.fnuoos.com/EggPlanet/egg_system_rules.git"
  11. "code.fnuoos.com/EggPlanet/egg_system_rules.git/enum"
  12. md3 "code.fnuoos.com/EggPlanet/egg_system_rules.git/md"
  13. "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule"
  14. md2 "code.fnuoos.com/EggPlanet/egg_system_rules.git/rule/egg_energy/md"
  15. "code.fnuoos.com/EggPlanet/egg_system_rules.git/svc/sys_cfg"
  16. "code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
  17. "context"
  18. "encoding/json"
  19. "errors"
  20. "fmt"
  21. "github.com/go-pay/gopay"
  22. "github.com/go-pay/gopay/alipay"
  23. "github.com/go-pay/xlog"
  24. "github.com/jinzhu/copier"
  25. "github.com/streadway/amqp"
  26. "time"
  27. )
  28. func EggFinWithdrawApplyDataConsume(queue md.MqQueue) {
  29. fmt.Println(">>>>>>>>>>>>EggFinWithdrawApplyDataConsume>>>>>>>>>>>>")
  30. ch, err := rabbit.Cfg.Pool.GetChannel()
  31. if err != nil {
  32. logx.Error(err)
  33. return
  34. }
  35. defer ch.Release()
  36. //1、将自己绑定到交换机上
  37. ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
  38. //2、取出数据进行消费
  39. ch.Qos(1)
  40. delivery := ch.Consume(queue.Name, false)
  41. egg_system_rules.Init(cfg.RedisAddr)
  42. var alipayStruct *InitAlipayStruct
  43. client, err := InitAlipay(alipayStruct)
  44. if err != nil {
  45. fmt.Println("EggFinWithdrawApplyDataConsume:::::", err.Error())
  46. utils2.FilePutContents("EggFinWithdrawApplyDataConsume", utils2.SerializeStr(map[string]interface{}{
  47. "err": err.Error(),
  48. }))
  49. return
  50. }
  51. var res amqp.Delivery
  52. var ok bool
  53. for {
  54. res, ok = <-delivery
  55. if ok == true {
  56. err = handleEggFinWithdrawApplyDataConsume(res.Body, client, ch)
  57. if err != nil {
  58. fmt.Println("EggFinWithdrawApplyDataConsume:::::", err.Error())
  59. utils2.FilePutContents("EggFinWithdrawApplyDataConsume", utils2.SerializeStr(map[string]interface{}{
  60. "body": res.Body,
  61. "err": err.Error(),
  62. }))
  63. var data md2.EggFinWithdrawApplyErrorData
  64. data.ErrorInfo = err.Error()
  65. // 尝试解析数据结构体
  66. var msg *md2.EggFinWithdrawApplyData
  67. err1 := json.Unmarshal(res.Body, &msg)
  68. if err1 != nil {
  69. // 无法解析 直接将[]byte推进队列
  70. data.Ext = res.Body
  71. } else {
  72. // 解析成功
  73. err2 := copier.Copy(&data, &msg)
  74. if err2 != nil {
  75. data.Ext = res.Body
  76. }
  77. }
  78. ch.Publish(md2.EggAppExchange, utils2.SerializeStr(data), md2.EggFinWithdrawApplyError)
  79. }
  80. //_ = res.Reject(false)
  81. err = res.Ack(true)
  82. fmt.Println("err ::: ", err)
  83. } else {
  84. panic(errors.New("error getting message"))
  85. }
  86. }
  87. }
  88. func handleEggFinWithdrawApplyDataConsume(msgData []byte, client *alipay.Client, ch *rabbit.Channel) error {
  89. time.Sleep(time.Duration(100) * time.Millisecond) //休眠100毫秒
  90. // 1.解析mq中queue的数据结构体
  91. var msg *md2.EggFinWithdrawApplyData
  92. err := json.Unmarshal(msgData, &msg)
  93. if err != nil {
  94. return err
  95. }
  96. applyDb := implement.NewFinWithdrawApplyDb(db.Db)
  97. apply, err := applyDb.FinWithdrawApplyGet(msg.Id)
  98. if err != nil {
  99. return err
  100. }
  101. if apply == nil {
  102. return errors.New("提现记录不存在~")
  103. }
  104. // 如果不在队列处理阶段或不是自动处理订单,直接返回
  105. if apply.State != 4 && apply.Type != 2 {
  106. return errors.New("提现记录无需处理~")
  107. }
  108. session := db.Db.NewSession()
  109. defer func() {
  110. session.Close()
  111. if err := recover(); err != nil {
  112. _ = logx.Error(err)
  113. }
  114. }()
  115. session.Begin()
  116. // 2.3 更新 apply 信息
  117. apply.State = 2
  118. affected, err2 := applyDb.UpdateFinWithdrawApplyBySession(session, apply, "state")
  119. if err2 != nil {
  120. return err2
  121. }
  122. if affected == 0 {
  123. return errors.New("更新 apply 状态失败")
  124. }
  125. if msg.WithdrawKind == 1 {
  126. // 3.支付宝提现
  127. alipayUserInfoDb := implement.NewAlipayUserInfoDb(db.Db)
  128. userInfo, err1 := alipayUserInfoDb.GetAlipayUserInfo(msg.Uid)
  129. if err != nil {
  130. return err1
  131. }
  132. bm := make(gopay.BodyMap)
  133. bm.Set("out_biz_no", msg.Id).
  134. Set("trans_amount", msg.Amount).
  135. Set("biz_scene", "DIRECT_TRANSFER").
  136. Set("product_code", "TRANS_ACCOUNT_NO_PWD").
  137. SetBodyMap("payee_info", func(bm gopay.BodyMap) {
  138. bm.Set("identity", userInfo.OpenId)
  139. bm.Set("identity_type", "ALIPAY_OPEN_ID")
  140. })
  141. _, err = client.FundTransUniTransfer(context.Background(), bm)
  142. if err != nil {
  143. // 如果需要重试 推回队尾
  144. if err.Error() == "Lock wait timeout exceeded; try restarting transaction" {
  145. ch.Publish(md2.EggAppExchange, msg, md2.EggFinWithdrawApply)
  146. return nil
  147. }
  148. //TODO::处理提现失败
  149. err = dealFinWithdrawApplyStateForBad(apply)
  150. if err != nil {
  151. return err
  152. }
  153. session.Rollback()
  154. return err
  155. }
  156. } else if msg.WithdrawKind == 2 {
  157. }
  158. return session.Commit()
  159. }
  160. func dealFinWithdrawApplyStateForBad(finWithdrawApply *model.FinWithdrawApply) error {
  161. session := db.Db.NewSession()
  162. defer session.Close()
  163. session.Begin()
  164. finWithdrawApplyDb := implement.NewFinWithdrawApplyDb(db.Db)
  165. finWithdrawApply.State = int(enum.FinWithdrawApplyStateForBad)
  166. updateAffected, err1 := finWithdrawApplyDb.UpdateFinWithdrawApplyBySession(session, finWithdrawApply, "state")
  167. if err1 != nil {
  168. return err1
  169. }
  170. if updateAffected <= 0 {
  171. return errors.New("更新提现单状态失败")
  172. }
  173. dealUserWalletReq := md3.DealUserWalletReq{
  174. Direction: "sub",
  175. Kind: int(enum.UserWithdrawBad),
  176. Title: enum.UserWithdrawBad.String(),
  177. Uid: finWithdrawApply.Uid,
  178. Amount: utils2.StrToFloat64(finWithdrawApply.Amount),
  179. }
  180. err := rule.DealUserWallet(session, dealUserWalletReq)
  181. if err != nil {
  182. session.Rollback()
  183. return err
  184. }
  185. err = session.Commit()
  186. if err != nil {
  187. _ = session.Rollback()
  188. }
  189. return err
  190. }
  191. type InitAlipayStruct struct {
  192. IsProd bool `json:"is_prod" label:"是否生产环境"`
  193. AlipayAppId string `json:"alipay_app_id" label:"支付宝商家应用appid"`
  194. AlipayPrivateKey string `json:"alipay_private_key" label:"支付宝商家应用私钥"`
  195. AlipayPublicKey string `json:"alipay_public_key" label:"支付宝商家应用公钥"`
  196. AlipayPublicContentRSA2 []byte `json:"alipay_public_content_rsa_2" label:"支付宝公钥证书"`
  197. AlipayRootContent []byte `json:"alipay_root_content" label:"支付宝根证书"`
  198. AppPublicContent []byte `json:"app_public_content" label:"应用公钥证书"`
  199. }
  200. // InitAlipay 初始化支付宝客户端
  201. // appid:应用ID
  202. // privateKey:应用私钥,支持PKCS1和PKCS8
  203. // isProd:是否是正式环境,沙箱环境请选择新版沙箱应用。
  204. func InitAlipay(initData *InitAlipayStruct) (client *alipay.Client, err error) {
  205. if initData == nil {
  206. sysCfgDb := sys_cfg.NewSysCfgDb(db.Db)
  207. sysCfgMap := sysCfgDb.SysCfgFindWithDb(enum.AlipayAppId, enum.AlipayPrivateKey, enum.AlipayPublicKey, enum.AlipayPublicContentRSA2, enum.AlipayRootContent, enum.AppPublicContent)
  208. initData = &InitAlipayStruct{
  209. IsProd: true,
  210. AlipayAppId: sysCfgMap[enum.AlipayAppId],
  211. AlipayPrivateKey: sysCfgMap[enum.AlipayPrivateKey],
  212. AlipayPublicKey: sysCfgMap[enum.AlipayPublicKey],
  213. AlipayPublicContentRSA2: []byte(sysCfgMap[enum.AlipayPublicContentRSA2]),
  214. AlipayRootContent: []byte(sysCfgMap[enum.AlipayRootContent]),
  215. AppPublicContent: []byte(sysCfgMap[enum.AppPublicContent]),
  216. }
  217. }
  218. client, err = alipay.NewClient(initData.AlipayAppId, initData.AlipayPrivateKey, initData.IsProd)
  219. if err != nil {
  220. xlog.Error(err)
  221. return
  222. }
  223. // 自定义配置http请求接收返回结果body大小,默认 10MB
  224. client.SetBodySize(10) // 没有特殊需求,可忽略此配置
  225. // 打开Debug开关,输出日志,默认关闭
  226. client.DebugSwitch = gopay.DebugOn
  227. client.SetLocation(alipay.LocationShanghai). // 设置时区,不设置或出错均为默认服务器时间
  228. SetCharset(alipay.UTF8). // 设置字符编码,不设置默认 utf-8
  229. SetSignType(alipay.RSA2) // 设置签名类型,不设置默认 RSA2
  230. // SetAppAuthToken("") //授权token
  231. //SetReturnUrl("https://www.fmm.ink"). // 设置返回URL
  232. // SetNotifyUrl("https://www.fmm.ink"). // 设置异步通知URL
  233. // 传入 支付宝公钥证书 alipayPublicCert.crt 内容
  234. client.AutoVerifySign(initData.AlipayPublicContentRSA2)
  235. // 传入证书内容
  236. err = client.SetCertSnByContent(initData.AppPublicContent, initData.AlipayRootContent, initData.AlipayPublicContentRSA2)
  237. if err != nil {
  238. xlog.Debug("SetCertSn:", err)
  239. return
  240. }
  241. return
  242. }