feat: update pay notify

This commit is contained in:
Rogee
2025-04-30 20:54:37 +08:00
parent 67d6f3ec33
commit d6c5a2677c
48 changed files with 444 additions and 161 deletions

View File

@@ -2,9 +2,9 @@ package publishers
import (
"encoding/json"
"quyun/app/events"
"go.ipao.vip/atom/contracts"
"quyun/app/events"
)
var _ contracts.EventPublisher = (*UserRegister)(nil)

View File

@@ -2,11 +2,11 @@ package subscribers
import (
"encoding/json"
"go.ipao.vip/atom/contracts"
"quyun/app/events"
"quyun/app/events/publishers"
"go.ipao.vip/atom/contracts"
"github.com/ThreeDotsLabs/watermill/message"
"github.com/sirupsen/logrus"
)

View File

@@ -4,7 +4,6 @@ import (
"errors"
"fmt"
"path/filepath"
"quyun/app/jobs"
"quyun/app/models"
"quyun/database/schemas/public/model"

View File

@@ -2,14 +2,13 @@ package http
import (
"net/url"
"time"
"quyun/app/models"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/pkg/utils"
"quyun/providers/jwt"
"quyun/providers/wechat"
"time"
"github.com/gofiber/fiber/v3"
"github.com/pkg/errors"

View File

@@ -1,16 +1,13 @@
package http
import (
"fmt"
"net/http"
"quyun/app/jobs"
"quyun/providers/job"
"quyun/providers/wepay"
"github.com/go-pay/gopay"
"github.com/go-pay/gopay/wechat/v3"
"github.com/go-pay/util/js"
"github.com/gofiber/fiber/v3"
log "github.com/sirupsen/logrus"
)
@@ -28,31 +25,13 @@ type pays struct {
func (ctl *pays) Callback(ctx fiber.Ctx, channel string) error {
log := log.WithField("method", "pays.Callback")
body := ctx.Body()
si := &wechat.SignInfo{
HeaderTimestamp: ctx.Get(wechat.HeaderTimestamp),
HeaderNonce: ctx.Get(wechat.HeaderNonce),
HeaderSignature: ctx.Get(wechat.HeaderSignature),
HeaderSerial: ctx.Get(wechat.HeaderSerial),
SignBody: string(body),
}
notifyReq := &wechat.V3NotifyReq{SignInfo: si}
if err := js.UnmarshalBytes(body, notifyReq); err != nil {
log.Errorf("json unmarshal error:%v", err)
return ctx.Status(http.StatusBadRequest).JSON(fiber.Map{"error": fmt.Sprintf("json unmarshal error:%v", err)})
}
// 获取微信平台证书
certMap := ctl.wepay.WxPublicKeyMap()
// 验证异步通知的签名
err := notifyReq.VerifySignByPKMap(certMap)
notify, err := ctl.wepay.ParseNotify(ctx)
if err != nil {
log.Errorf("verify sign error:%v", err)
return ctx.Status(http.StatusBadRequest).JSON(fiber.Map{"error": "Invalid signature"})
log.Errorf("ParseNotify error:%v", err)
return ctx.Status(http.StatusBadRequest).JSON(fiber.Map{"error": "Failed to parse notify"})
}
if err := ctl.job.Add(&jobs.WechatCallback{NotifyReq: notifyReq}); err != nil {
if err := ctl.job.Add(&jobs.WechatPayNotify{PayNotify: notify}); err != nil {
log.Errorf("add job error:%v", err)
return ctx.Status(http.StatusInternalServerError).JSON(fiber.Map{"error": "Failed to add job"})
}

View File

@@ -2,14 +2,13 @@ package http
import (
_ "embed"
"time"
"quyun/app/models"
"quyun/app/requests"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/providers/ali"
"quyun/providers/wepay"
"time"
"github.com/go-pay/gopay/wechat/v3"
"github.com/gofiber/fiber/v3"

View File

@@ -1,11 +1,10 @@
package http
import (
"strings"
"time"
"quyun/app/models"
"quyun/database/schemas/public/model"
"strings"
"time"
"github.com/gofiber/fiber/v3"
)

View File

@@ -1,11 +1,10 @@
package http
import (
"time"
"quyun/app/models"
"quyun/database/schemas/public/model"
"quyun/providers/wechat"
"time"
"github.com/gofiber/fiber/v3"
"github.com/gofiber/fiber/v3/log"

View File

@@ -4,12 +4,11 @@ import (
"context"
"os"
"path/filepath"
"time"
"quyun/app/models"
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"time"
. "github.com/riverqueue/river"
log "github.com/sirupsen/logrus"
@@ -18,7 +17,7 @@ import (
_ "go.ipao.vip/atom/contracts"
)
var _ contracts.JobArgs = (*WechatCallback)(nil)
var _ contracts.JobArgs = (*DownloadFromAliOSS)(nil)
type DownloadFromAliOSS struct {
MediaHash string `json:"media_hash"`

View File

@@ -2,13 +2,12 @@ package jobs
import (
"context"
"testing"
"quyun/app/models"
"quyun/app/service/testx"
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"testing"
. "github.com/riverqueue/river"
. "github.com/smartystreets/goconvey/convey"

View File

@@ -147,7 +147,7 @@ func Provide(opts ...opt.Option) error {
if err := container.Container.Provide(func(
__job *job.Job,
) (contracts.Initial, error) {
obj := &WechatCallbackWorker{}
obj := &WechatPayNotifyWorker{}
if err := river.AddWorkerSafely(__job.Workers, obj); err != nil {
return nil, err
}

View File

@@ -2,8 +2,6 @@ package jobs
import (
"context"
"time"
"quyun/app/models"
"quyun/database/fields"
"quyun/database/schemas/public/model"
@@ -11,6 +9,7 @@ import (
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"time"
. "github.com/riverqueue/river"
"github.com/samber/lo"

View File

@@ -3,13 +3,12 @@ package jobs
import (
"context"
"path/filepath"
"time"
"quyun/app/models"
"quyun/database/fields"
"quyun/pkg/utils"
"quyun/providers/app"
"quyun/providers/job"
"time"
. "github.com/riverqueue/river"
log "github.com/sirupsen/logrus"

View File

@@ -4,8 +4,6 @@ import (
"context"
"os"
"path/filepath"
"time"
"quyun/app/models"
"quyun/database/fields"
"quyun/database/schemas/public/model"
@@ -13,6 +11,7 @@ import (
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"time"
. "github.com/riverqueue/river"
log "github.com/sirupsen/logrus"

View File

@@ -3,8 +3,6 @@ package jobs
import (
"context"
"path/filepath"
"time"
"quyun/app/models"
"quyun/database/fields"
"quyun/database/schemas/public/model"
@@ -12,6 +10,7 @@ import (
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"time"
. "github.com/riverqueue/river"
log "github.com/sirupsen/logrus"

View File

@@ -1,49 +0,0 @@
package jobs
import (
"context"
"time"
"github.com/go-pay/gopay/wechat/v3"
. "github.com/riverqueue/river"
log "github.com/sirupsen/logrus"
_ "go.ipao.vip/atom"
"go.ipao.vip/atom/contracts"
_ "go.ipao.vip/atom/contracts"
)
var _ contracts.JobArgs = (*WechatCallback)(nil)
type WechatCallback struct {
NotifyReq *wechat.V3NotifyReq `json:"notify_req"`
}
func (s WechatCallback) InsertOpts() InsertOpts {
return InsertOpts{
Queue: QueueDefault,
Priority: PriorityDefault,
}
}
func (WechatCallback) Kind() string { return "wechat_callback" }
func (a WechatCallback) UniqueID() string { return a.Kind() }
var _ Worker[WechatCallback] = (*WechatCallbackWorker)(nil)
// @provider(job)
type WechatCallbackWorker struct {
WorkerDefaults[WechatCallback]
}
func (w *WechatCallbackWorker) Work(ctx context.Context, job *Job[WechatCallback]) error {
log := log.WithField("job", job.Args.Kind())
log.Infof("[Start] Working on job with strings: %+v", job.Args)
defer log.Infof("[End] Finished %s", job.Args.Kind())
return nil
}
func (w *WechatCallbackWorker) NextRetry(job *Job[WechatCallback]) time.Time {
return time.Now().Add(30 * time.Second)
}

View File

@@ -0,0 +1,108 @@
package jobs
import (
"context"
"fmt"
"quyun/app/models"
"quyun/database/fields"
"quyun/providers/wepay"
"time"
"github.com/pkg/errors"
. "github.com/riverqueue/river"
log "github.com/sirupsen/logrus"
_ "go.ipao.vip/atom"
"go.ipao.vip/atom/contracts"
_ "go.ipao.vip/atom/contracts"
)
var _ contracts.JobArgs = (*WechatPayNotify)(nil)
type WechatPayNotify struct {
PayNotify *wepay.PayNotify `json:"notify_req"`
}
func (s WechatPayNotify) InsertOpts() InsertOpts {
return InsertOpts{
Queue: QueueDefault,
Priority: PriorityDefault,
}
}
func (WechatPayNotify) Kind() string { return "wechat_callback" }
func (a WechatPayNotify) UniqueID() string { return a.Kind() }
var _ Worker[WechatPayNotify] = (*WechatPayNotifyWorker)(nil)
// @provider(job)
type WechatPayNotifyWorker struct {
WorkerDefaults[WechatPayNotify]
}
func (w *WechatPayNotifyWorker) Work(ctx context.Context, job *Job[WechatPayNotify]) error {
log := log.WithField("job", job.Args.Kind())
log.Infof("[Start] Working on job with strings: %+v", job.Args)
defer log.Infof("[End] Finished %s", job.Args.Kind())
notify := job.Args.PayNotify
if notify.TradeState != "SUCCESS" {
log.Warnf("TradeState is not SUCCESS for order %s", notify.OutTradeNo)
return JobCancel(errors.New("TradeState is not SUCCESS"))
}
order, err := models.Orders.GetByOrderNo(context.Background(), notify.OutTradeNo)
if err != nil {
log.Errorf("GetByOrderNo error:%v", err)
return err
}
if order.Status != fields.OrderStatusPending {
log.Infof("Order %s is paid, processing...", job.Args.PayNotify.OutTradeNo)
return JobCancel(fmt.Errorf("Order already paid, currently status: %d", order.Status))
}
needToPay := order.Price * int64(order.Discount) / 100
if notify.Amount.Total != needToPay {
log.Errorf("Order %s amount mismatch: expected %d, got %d", job.Args.PayNotify.OutTradeNo, needToPay, notify.Amount.Total)
return fmt.Errorf("amount mismatch for order %s", job.Args.PayNotify.OutTradeNo)
}
order.TransactionID = notify.TransactionID
order.Currency = notify.Amount.Currency
order.PaymentMethod = notify.TradeType
order.Status = fields.OrderStatusCompleted
order.Meta = fields.ToJson(fields.OrderMeta{
PayNotify: *notify,
})
log.Infof("Updated order details: %+v", order)
tx, err := models.Transaction(ctx)
if err != nil {
return errors.Wrap(err, "Transaction error")
}
defer tx.Rollback()
if err := models.Users.BuyPosts(context.Background(), order.UserID, order.PostID, order.Price); err != nil {
log.Errorf("BuyPosts error:%v", err)
return errors.Wrap(err, "BuyPosts error")
}
if err := models.Orders.Update(context.Background(), order); err != nil {
log.Errorf("Update order error:%v", err)
return errors.Wrap(err, "Update order error")
}
if err := tx.Commit(); err != nil {
log.Errorf("Commit error:%v", err)
return errors.Wrap(err, "Commit error")
}
log.Infof("Successfully processed order %s", notify.OutTradeNo)
return nil
}
func (w *WechatPayNotifyWorker) NextRetry(job *Job[WechatPayNotify]) time.Time {
return time.Now().Add(30 * time.Second)
}

View File

@@ -0,0 +1,86 @@
package jobs
import (
"context"
"encoding/json"
"quyun/app/models"
"quyun/app/service/testx"
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"quyun/providers/wepay"
"testing"
. "github.com/riverqueue/river"
. "github.com/smartystreets/goconvey/convey"
"github.com/stretchr/testify/suite"
_ "go.ipao.vip/atom"
"go.ipao.vip/atom/contracts"
"go.uber.org/dig"
)
type WechatPayNotifySuiteInjectParams struct {
dig.In
Initials []contracts.Initial `group:"initials"` // nolint:structcheck
Job *job.Job
Oss *ali.OSSClient
App *app.Config
}
type WechatPayNotifySuite struct {
suite.Suite
WechatPayNotifySuiteInjectParams
}
func Test_WechatPayNotify(t *testing.T) {
providers := testx.Default().With(Provide, models.Provide)
testx.Serve(providers, t, func(p WechatPayNotifySuiteInjectParams) {
suite.Run(t, &WechatPayNotifySuite{WechatPayNotifySuiteInjectParams: p})
})
}
func (t *WechatPayNotifySuite) Test_Work() {
Convey("test_work", t.T(), func() {
Convey("step 1", func() {
notify := `{
"mchid": "1702644947",
"appid": "wx47649361b6eba174",
"out_trade_no": "20250430192543",
"transaction_id": "4200002602202504300651871941",
"trade_type": "JSAPI",
"trade_state": "SUCCESS",
"trade_state_desc": "支付成功",
"bank_type": "OTHERS",
"attach": "",
"success_time": "2025-04-30T19:25:51+08:00",
"payer": {
"openid": "o5Bzk644x3LOMJsKSZRlqWin74IU"
},
"amount": {
"total": 1,
"payer_total": 1,
"currency": "CNY",
"payer_currency": "CNY"
}
}`
var payNotify wepay.PayNotify
err := json.Unmarshal([]byte(notify), &payNotify)
So(err, ShouldBeNil)
job := &Job[WechatPayNotify]{
Args: WechatPayNotify{
PayNotify: &payNotify,
},
}
worker := &WechatPayNotifyWorker{}
err = worker.Work(context.Background(), job)
So(err, ShouldBeNil)
})
})
}

View File

@@ -2,11 +2,10 @@ package middlewares
import (
"net/url"
"strings"
"time"
"quyun/app/models"
"quyun/pkg/utils"
"strings"
"time"
"github.com/gofiber/fiber/v3"
log "github.com/sirupsen/logrus"

View File

@@ -2,12 +2,11 @@ package models
import (
"context"
"time"
"quyun/app/requests"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"time"
. "github.com/go-jet/jet/v2/postgres"
"github.com/samber/lo"

View File

@@ -5,14 +5,13 @@ import (
"fmt"
"math"
"math/rand"
"testing"
"time"
"quyun/app/requests"
"quyun/app/service/testx"
"quyun/database"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"testing"
"time"
. "github.com/smartystreets/goconvey/convey"
"go.ipao.vip/atom/contracts"

View File

@@ -3,12 +3,11 @@ package models
import (
"context"
"fmt"
"time"
"quyun/app/requests"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"time"
. "github.com/go-jet/jet/v2/postgres"
"github.com/pkg/errors"
@@ -243,3 +242,68 @@ func (m *ordersModel) SumAmount(ctx context.Context) (int64, error) {
return cnt.Cnt, nil
}
// GetByOrderNo
func (m *ordersModel) GetByOrderNo(ctx context.Context, orderNo string) (*model.Orders, error) {
tbl := table.Orders
stmt := tbl.
SELECT(tbl.AllColumns).
WHERE(
tbl.OrderNo.EQ(String(orderNo)),
)
m.log.Infof("sql: %s", stmt.DebugSql())
var order model.Orders
err := stmt.QueryContext(ctx, db, &order)
if err != nil {
m.log.Errorf("error querying order by orderNo: %v", err)
return nil, err
}
return &order, nil
}
// SetTranscationID
func (m *ordersModel) SetTranscationID(ctx context.Context, id int64, transactionID string) error {
tbl := table.Orders
stmt := tbl.
UPDATE(tbl.TransactionID).
SET(String(transactionID)).
WHERE(
tbl.ID.EQ(Int64(id)),
)
m.log.Infof("sql: %s", stmt.DebugSql())
if _, err := stmt.ExecContext(ctx, db); err != nil {
m.log.Errorf("error set order transaction ID: %v", err)
return err
}
return nil
}
// Update
func (m *ordersModel) Update(ctx context.Context, order *model.Orders) error {
tbl := table.Orders
stmt := tbl.
UPDATE(
tbl.MutableColumns.Except(
tbl.OrderNo,
tbl.Price,
tbl.Discount,
tbl.SubOrderNo,
tbl.PostID,
tbl.UserID,
),
).
MODEL(order).
WHERE(
tbl.ID.EQ(Int64(order.ID)),
)
m.log.Infof("sql: %s", stmt.DebugSql())
if _, err := stmt.ExecContext(ctx, db); err != nil {
m.log.Errorf("error updating order: %v", err)
return err
}
return nil
}

View File

@@ -2,11 +2,10 @@ package models
import (
"context"
"testing"
"quyun/app/service/testx"
"quyun/database"
"quyun/database/schemas/public/table"
"testing"
. "github.com/smartystreets/goconvey/convey"
"go.ipao.vip/atom/contracts"

View File

@@ -3,11 +3,10 @@ package models
import (
"context"
"errors"
"time"
"quyun/app/requests"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"time"
. "github.com/go-jet/jet/v2/postgres"
"github.com/go-jet/jet/v2/qrm"

View File

@@ -4,13 +4,12 @@ import (
"context"
"fmt"
"math/rand"
"testing"
"quyun/app/service/testx"
"quyun/database"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"testing"
. "github.com/smartystreets/goconvey/convey"
"go.ipao.vip/atom/contracts"

View File

@@ -2,12 +2,12 @@ package models
import (
"context"
"time"
"database/sql"
"quyun/app/requests"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"time"
. "github.com/go-jet/jet/v2/postgres"
"github.com/go-jet/jet/v2/qrm"
@@ -16,6 +16,10 @@ import (
"github.com/sirupsen/logrus"
)
func Transaction(ctx context.Context) (*sql.Tx, error) {
return db.Begin()
}
// @provider
type usersModel struct {
log *logrus.Entry `inject:"false"`
@@ -404,3 +408,25 @@ func (m *usersModel) UpdateUserToken(ctx context.Context, id int64, token fields
}
return nil
}
// BuyPosts
func (m *usersModel) BuyPosts(ctx context.Context, userID, postID, price int64) error {
tbl := table.UserPosts
stmt := tbl.
INSERT(tbl.MutableColumns).
MODEL(&model.UserPosts{
UserID: userID,
PostID: postID,
Price: price,
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
})
m.log.Infof("sql: %s", stmt.DebugSql())
if _, err := stmt.ExecContext(ctx, db); err != nil {
m.log.Errorf("error inserting user post: %v", err)
return err
}
return nil
}

View File

@@ -2,13 +2,12 @@ package models
import (
"context"
"testing"
"quyun/app/service/testx"
"quyun/database"
"quyun/database/fields"
"quyun/database/schemas/public/model"
"quyun/database/schemas/public/table"
"testing"
"github.com/samber/lo"
. "github.com/smartystreets/goconvey/convey"

View File

@@ -9,12 +9,11 @@ import (
"os"
"os/exec"
"path/filepath"
"quyun/app/service"
"strconv"
"strings"
"time"
"quyun/app/service"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"

View File

@@ -2,16 +2,16 @@ package event
import (
"context"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
"quyun/app/events/subscribers"
"quyun/app/service"
"quyun/providers/app"
"quyun/providers/event"
"quyun/providers/postgres"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"go.uber.org/dig"

View File

@@ -1,15 +1,16 @@
package grpc
import (
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
"quyun/app/grpc/users"
"quyun/app/service"
"quyun/providers/app"
"quyun/providers/grpc"
"quyun/providers/postgres"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"go.uber.org/dig"

View File

@@ -4,13 +4,10 @@ import (
"context"
"mime"
"path/filepath"
appHttp "quyun/app/http"
"quyun/app/jobs"
"quyun/app/middlewares"
"quyun/app/models"
"quyun/app/service"
_ "quyun/docs"
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/hashids"
@@ -22,6 +19,10 @@ import (
"quyun/providers/wechat"
"quyun/providers/wepay"
appHttp "quyun/app/http"
_ "quyun/docs"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"

View File

@@ -3,7 +3,6 @@ package migrate
import (
"context"
"database/sql"
"quyun/app/service"
"quyun/database"
"quyun/providers/postgres"

View File

@@ -2,16 +2,16 @@ package queue
import (
"context"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
"quyun/app/jobs"
"quyun/app/service"
"quyun/providers/app"
"quyun/providers/job"
"quyun/providers/postgres"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"go.uber.org/dig"

View File

@@ -1,9 +1,10 @@
package service
import (
"go.ipao.vip/atom/container"
"quyun/providers/app"
"quyun/providers/event"
"go.ipao.vip/atom/container"
)
func Default(providers ...container.ProviderContainer) container.Providers {

View File

@@ -2,13 +2,12 @@ package testx
import (
"os"
"testing"
"quyun/providers/ali"
"quyun/providers/app"
"quyun/providers/job"
"quyun/providers/postgres"
"quyun/providers/wechat"
"testing"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"