rogee/mp-qvyun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: user buy media

Browse Source
This commit is contained in:
Rogee
2024-12-10 11:21:55 +08:00
parent 21840c3adf
commit e3ef31037c
17 changed files with 201 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
backend/database/migrations/20241128075611_init.sql
View File

@@ -33,6 +33,7 @@ CREATE TABLE
id SERIAL8 PRIMARY KEY,
user_id INT8 NOT NULL,
tenant_id INT8 NOT NULL,
balance INT8 NOT NULL default 0,
created_at timestamp NOT NULL default now()
);
@@ -41,16 +42,6 @@ CREATE INDEX idx_users_tenants_tenant_id ON users_tenants (tenant_id);
-- uniq user_id, tenant_id
CREATE UNIQUE INDEX idx_users_tenants_user_id_tenant_id ON users_tenants (user_id, tenant_id);
CREATE TABLE tenant_user_balances (
id SERIAL8 PRIMARY KEY,
user_id INT8 NOT NULL,
tenant_id INT8 NOT NULL,
balance INT8 NOT NULL
);
CREATE INDEX idx_tenant_user_balance_user_id ON tenant_user_balances (user_id);
CREATE INDEX idx_tenant_user_balance_tenant_id ON tenant_user_balances (tenant_id);
-- table user_balance_history
CREATE TABLE user_balance_histories (
id SERIAL8 PRIMARY KEY,

15
backend/database/models/qvyun/public/model/tenant_user_balances.go
View File

@@ -1,15 +0,0 @@
//
// Code generated by go-jet DO NOT EDIT.
//
// WARNING: Changes to this file may cause incorrect behavior
// and will be lost if the code is regenerated
//
package model
type TenantUserBalances struct {
ID int64 `sql:"primary_key" json:"id"`
UserID int64 `json:"user_id"`
TenantID int64 `json:"tenant_id"`
Balance int64 `json:"balance"`
}

1
backend/database/models/qvyun/public/model/users_tenants.go
View File

@@ -15,5 +15,6 @@ type UsersTenants struct {
ID int64 `sql:"primary_key" json:"id"`
UserID int64 `json:"user_id"`
TenantID int64 `json:"tenant_id"`
Balance int64 `json:"balance"`
CreatedAt time.Time `json:"created_at"`
}

1
backend/database/models/qvyun/public/table/table_use_schema.go
View File

@@ -12,7 +12,6 @@ package table
func UseSchema(schema string) {
Medias = Medias.FromSchema(schema)
Migrations = Migrations.FromSchema(schema)
TenantUserBalances = TenantUserBalances.FromSchema(schema)
Tenants = Tenants.FromSchema(schema)
UserBalanceHistories = UserBalanceHistories.FromSchema(schema)
UserMedias = UserMedias.FromSchema(schema)

84
backend/database/models/qvyun/public/table/tenant_user_balances.go
View File

@@ -1,84 +0,0 @@
//
// Code generated by go-jet DO NOT EDIT.
//
// WARNING: Changes to this file may cause incorrect behavior
// and will be lost if the code is regenerated
//
package table
import (
"github.com/go-jet/jet/v2/postgres"
)
var TenantUserBalances = newTenantUserBalancesTable("public", "tenant_user_balances", "")
type tenantUserBalancesTable struct {
postgres.Table
// Columns
ID postgres.ColumnInteger
UserID postgres.ColumnInteger
TenantID postgres.ColumnInteger
Balance postgres.ColumnInteger
AllColumns postgres.ColumnList
MutableColumns postgres.ColumnList
}
type TenantUserBalancesTable struct {
tenantUserBalancesTable
EXCLUDED tenantUserBalancesTable
}
// AS creates new TenantUserBalancesTable with assigned alias
func (a TenantUserBalancesTable) AS(alias string) *TenantUserBalancesTable {
return newTenantUserBalancesTable(a.SchemaName(), a.TableName(), alias)
}
// Schema creates new TenantUserBalancesTable with assigned schema name
func (a TenantUserBalancesTable) FromSchema(schemaName string) *TenantUserBalancesTable {
return newTenantUserBalancesTable(schemaName, a.TableName(), a.Alias())
}
// WithPrefix creates new TenantUserBalancesTable with assigned table prefix
func (a TenantUserBalancesTable) WithPrefix(prefix string) *TenantUserBalancesTable {
return newTenantUserBalancesTable(a.SchemaName(), prefix+a.TableName(), a.TableName())
}
// WithSuffix creates new TenantUserBalancesTable with assigned table suffix
func (a TenantUserBalancesTable) WithSuffix(suffix string) *TenantUserBalancesTable {
return newTenantUserBalancesTable(a.SchemaName(), a.TableName()+suffix, a.TableName())
}
func newTenantUserBalancesTable(schemaName, tableName, alias string) *TenantUserBalancesTable {
return &TenantUserBalancesTable{
tenantUserBalancesTable: newTenantUserBalancesTableImpl(schemaName, tableName, alias),
EXCLUDED: newTenantUserBalancesTableImpl("", "excluded", ""),
}
}
func newTenantUserBalancesTableImpl(schemaName, tableName, alias string) tenantUserBalancesTable {
var (
IDColumn = postgres.IntegerColumn("id")
UserIDColumn = postgres.IntegerColumn("user_id")
TenantIDColumn = postgres.IntegerColumn("tenant_id")
BalanceColumn = postgres.IntegerColumn("balance")
allColumns = postgres.ColumnList{IDColumn, UserIDColumn, TenantIDColumn, BalanceColumn}
mutableColumns = postgres.ColumnList{UserIDColumn, TenantIDColumn, BalanceColumn}
)
return tenantUserBalancesTable{
Table: postgres.NewTable(schemaName, tableName, alias, allColumns...),
//Columns
ID: IDColumn,
UserID: UserIDColumn,
TenantID: TenantIDColumn,
Balance: BalanceColumn,
AllColumns: allColumns,
MutableColumns: mutableColumns,
}
}

7
backend/database/models/qvyun/public/table/users_tenants.go
View File

@@ -20,6 +20,7 @@ type usersTenantsTable struct {
ID postgres.ColumnInteger
UserID postgres.ColumnInteger
TenantID postgres.ColumnInteger
Balance postgres.ColumnInteger
CreatedAt postgres.ColumnTimestamp
AllColumns postgres.ColumnList
@@ -64,9 +65,10 @@ func newUsersTenantsTableImpl(schemaName, tableName, alias string) usersTenantsT
IDColumn = postgres.IntegerColumn("id")
UserIDColumn = postgres.IntegerColumn("user_id")
TenantIDColumn = postgres.IntegerColumn("tenant_id")
BalanceColumn = postgres.IntegerColumn("balance")
CreatedAtColumn = postgres.TimestampColumn("created_at")
allColumns = postgres.ColumnList{IDColumn, UserIDColumn, TenantIDColumn, CreatedAtColumn}
mutableColumns = postgres.ColumnList{UserIDColumn, TenantIDColumn, CreatedAtColumn}
allColumns = postgres.ColumnList{IDColumn, UserIDColumn, TenantIDColumn, BalanceColumn, CreatedAtColumn}
mutableColumns = postgres.ColumnList{UserIDColumn, TenantIDColumn, BalanceColumn, CreatedAtColumn}
)
return usersTenantsTable{
@@ -76,6 +78,7 @@ func newUsersTenantsTableImpl(schemaName, tableName, alias string) usersTenantsT
ID: IDColumn,
UserID: UserIDColumn,
TenantID: TenantIDColumn,
Balance: BalanceColumn,
CreatedAt: CreatedAtColumn,
AllColumns: allColumns,

43
backend/modules/medias/controller.go
View File

@@ -2,7 +2,6 @@ package medias
import (
"backend/pkg/consts"
"backend/pkg/errorx"
"backend/pkg/pg"
"backend/providers/jwt"
@@ -22,13 +21,13 @@ func (c *Controller) List(ctx fiber.Ctx) error {
filter := ListFilter{}
if err := ctx.Bind().Body(&filter); err != nil {
log.WithError(err).Error("parse body failed")
return ctx.Status(fiber.StatusBadRequest).JSON(errorx.RequestParseError)
return err
}
claim := ctx.Locals(consts.CtxKeyClaim).(*jwt.Claims)
items, err := c.svc.List(ctx.Context(), claim.TenantID, claim.UserID, &filter)
if err != nil {
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
return ctx.JSON(items)
@@ -43,14 +42,14 @@ func (c *Controller) Show(ctx fiber.Ctx) error {
model, err := c.svc.GetMediaByHash(ctx.Context(), claim.TenantID, hash)
if err != nil {
log.WithField("action", "medias.Show").WithError(err).Error("GetMediaByHash")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
resource := c.svc.ModelToListItem(ctx.Context(), model)
resource.Bought, err = c.svc.HasUserBought(ctx.Context(), claim.TenantID, claim.UserID, model.ID)
if err != nil {
log.WithField("action", "medias.Show").WithError(err).Error("HasUserBought")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
return ctx.JSON(resource)
@@ -60,7 +59,7 @@ func (c *Controller) Show(ctx fiber.Ctx) error {
func (c *Controller) MediaIndex(ctx fiber.Ctx) error {
mediaType, err := pg.ParseMediaType(ctx.Params("type"))
if err != nil {
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.RequestParseError)
return err
}
hash := ctx.Params("hash")
@@ -70,19 +69,19 @@ func (c *Controller) MediaIndex(ctx fiber.Ctx) error {
model, err := c.svc.GetMediaByHash(ctx.Context(), claim.TenantID, hash)
if err != nil {
log.WithField("action", "medias.MediaIndex").WithError(err).Error("GetMediaByHash")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
bought, err := c.svc.HasUserBought(ctx.Context(), claim.TenantID, claim.UserID, model.ID)
if err != nil {
log.WithField("action", "medias.MediaIndex").WithError(err).Error("HasUserBought")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
playlist, err := c.svc.GetM3U8(ctx.Context(), claim.TenantID, mediaType, model.Hash, bought)
if err != nil {
log.WithField("action", "medias.MediaIndex").WithError(err).Error("GetMediaPlaylist")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
return ctx.SendString(playlist.String())
@@ -91,14 +90,14 @@ func (c *Controller) MediaIndex(ctx fiber.Ctx) error {
func (c *Controller) MediaSegment(ctx fiber.Ctx) error {
mediaType, err := pg.ParseMediaType(ctx.Params("type"))
if err != nil {
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.RequestParseError)
return err
}
segment := ctx.Params("segment")
segments, err := c.hashIds.DecodeInt64WithError(segment)
if err != nil {
log.WithField("action", "medias.MediaSegment").WithError(err).Error("DecodeInt64WithError")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.RequestParseError)
return err
}
hash := ctx.Params("hash")
@@ -108,9 +107,29 @@ func (c *Controller) MediaSegment(ctx fiber.Ctx) error {
model, err := c.svc.GetMediaByHash(ctx.Context(), claim.TenantID, hash)
if err != nil {
log.WithField("action", "medias.MediaSegment").WithError(err).Error("GetMediaByHash")
return ctx.Status(fiber.StatusInternalServerError).JSON(errorx.InternalError)
return err
}
filepath := c.svc.GetSegmentPath(ctx.Context(), mediaType, model.Hash, segments[0])
return ctx.SendFile(filepath)
}
// Checkout
func (c *Controller) Checkout(ctx fiber.Ctx) error {
hash := ctx.Params("hash")
claim := fiber.Locals[*jwt.Claims](ctx, consts.CtxKeyClaim)
log.Debug(claim)
model, err := c.svc.GetMediaByHash(ctx.Context(), claim.TenantID, hash)
if err != nil {
log.WithField("action", "medias.MediaSegment").WithError(err).Error("GetMediaByHash")
return err
}
if err := c.svc.Checkout(ctx.Context(), claim.TenantID, claim.UserID, model.ID); err != nil {
log.WithField("action", "medias.MediaSegment").WithError(err).Error("Checkout")
return err
}
return ctx.JSON(nil)
}

1
backend/modules/medias/router.go
View File

@@ -29,4 +29,5 @@ func (r *Router) Register(router fiber.Router) {
group.Get(":hash", r.controller.Show)
group.Get(":hash/:type<regex([video|audio])>", r.controller.MediaIndex)
group.Get(":hash/:type<regex([video|audio])>/:segment.ts", r.controller.MediaSegment)
group.Get(":hash/checkout", r.controller.Checkout)
}

92
backend/modules/medias/service.go
View File

@@ -11,6 +11,7 @@ import (
"backend/database/models/qvyun/public/model"
"backend/database/models/qvyun/public/table"
"backend/pkg/errorx"
"backend/pkg/media_store"
"backend/pkg/path"
"backend/pkg/pg"
@@ -314,3 +315,94 @@ func (svc *Service) GetM3U8(ctx context.Context, tenantId int64, types pg.MediaT
func (svc *Service) GetSegmentPath(ctx context.Context, t pg.MediaType, hash string, segment int64) string {
return filepath.Join(svc.storageConfig.Path, hash, t.String(), fmt.Sprintf("%d.ts", segment))
}
func (svc *Service) Checkout(ctx context.Context, tenantId, userId, mediaId int64) error {
log := svc.log.WithField("method", "Checkout")
bought, err := svc.HasUserBought(ctx, tenantId, userId, mediaId)
if err != nil {
return errors.Wrap(err, "check user bought")
}
if bought {
return nil
}
media, err := svc.GetMediaByID(ctx, tenantId, userId, mediaId)
if err != nil {
return errors.Wrap(err, "get media")
}
userBalance, err := svc.GetUserBalance(ctx, tenantId, userId)
if err != nil {
return errors.Wrap(err, "get user balance")
}
if userBalance < media.Price {
return errorx.UserBalanceNotEnough
}
tx, err := svc.db.Begin()
if err != nil {
return errors.Wrap(err, "begin transaction")
}
defer tx.Rollback()
tbl := table.UserMedias
stmt := tbl.
INSERT(tbl.TenantID, tbl.UserID, tbl.MediaID, tbl.Price).
VALUES(Int(tenantId), Int(userId), Int(mediaId), Int(media.Price))
log.Debug(stmt.DebugSql())
if _, err := stmt.ExecContext(ctx, tx); err != nil {
return errors.Wrap(err, "insert user media")
}
// update user balance
tblUserTenants := table.UsersTenants
stmtUserTenants := tblUserTenants.
UPDATE().
SET(
tblUserTenants.Balance.SET(
tblUserTenants.Balance.SUB(Int(media.Price)),
),
).
WHERE(
tblUserTenants.TenantID.EQ(Int(tenantId)).AND(
tblUserTenants.UserID.EQ(Int(userId)),
),
)
log.Debug(stmtUserTenants.DebugSql())
if _, err := stmtUserTenants.ExecContext(ctx, tx); err != nil {
return errors.Wrap(err, "update user balance")
}
if err := tx.Commit(); err != nil {
return errors.Wrap(err, "commit transaction")
}
return nil
}
// GetUserBalance
func (svc *Service) GetUserBalance(ctx context.Context, tenantId, userId int64) (int64, error) {
log := svc.log.WithField("method", "GetUserBalance")
tbl := table.UsersTenants
stmt := tbl.SELECT(tbl.Balance.AS("balance")).WHERE(
tbl.TenantID.EQ(Int(tenantId)).AND(
tbl.UserID.EQ(Int(userId)),
),
)
log.Debug(stmt.DebugSql())
var result struct {
Balance int64
}
if err := stmt.QueryContext(ctx, svc.db, &result); err != nil {
return 0, errors.Wrap(err, "query user balance")
}
return result.Balance, nil
}

17
backend/modules/medias/service_test.go
View File

@@ -11,6 +11,7 @@ import (
"backend/providers/postgres"
"backend/providers/storage"
log "github.com/sirupsen/logrus"
. "github.com/smartystreets/goconvey/convey"
"github.com/stretchr/testify/suite"
"go.uber.org/dig"
@@ -27,6 +28,7 @@ type ServiceTestSuite struct {
}
func Test_DiscoverMedias(t *testing.T) {
log.SetLevel(log.DebugLevel)
providers := testx.Default(
postgres.DefaultProvider(),
storage.DefaultProvider(),
@@ -66,3 +68,18 @@ func (t *ServiceTestSuite) Test_getMediaByHash() {
So(ext, ShouldEqual, "ts")
})
}
func (t *ServiceTestSuite) Test_GetUserBalance() {
Convey("Test_GetUserBalance", t.T(), func() {
balance, err := t.Svc.GetUserBalance(context.Background(), 1, 1)
So(err, ShouldBeNil)
t.T().Logf("balance: %+v", balance)
})
}
func (t *ServiceTestSuite) Test_Checkout() {
Convey("Test_Checkout", t.T(), func() {
err := t.Svc.Checkout(context.TODO(), 1, 1, 1)
So(err, ShouldBeNil)
})
}

30
backend/modules/middlewares/mid_response.go Normal file
View File

@@ -0,0 +1,30 @@
package middlewares
import (
"backend/pkg/errorx"
"github.com/gofiber/fiber/v3"
log "github.com/sirupsen/logrus"
)
func (f *Middlewares) ProcessResponse(c fiber.Ctx) error {
if err := c.Next(); err != nil {
log.WithError(err).Error("process response error")
if e, ok := err.(errorx.Response); ok {
return e.Response(c)
}
if e, ok := err.(*fiber.Error); ok {
return errorx.Response{
StatusCode: e.Code,
Code: e.Code,
Message: e.Message,
}.Response(c)
}
return errorx.Wrap(err).Response(c)
}
return c.Next()
}

2
backend/pkg/dao.go
View File

@@ -1,4 +1,4 @@
package common
package pkg
func WrapLike(v string) string {
return "%" + v + "%"

2
backend/pkg/data_structures.go
View File

@@ -1,4 +1,4 @@
package common
package pkg
import (
"strings"

23
backend/pkg/errorx/error.go
View File

@@ -2,18 +2,31 @@ package errorx
import (
"fmt"
"net/http"
"github.com/gofiber/fiber/v3"
)
type Response struct {
Code int `json:"code"`
Message string `json:"message"`
StatusCode int `json:"-"`
Code int `json:"code"`
Message string `json:"message"`
}
func Wrap(err error) Response {
return Response{http.StatusInternalServerError, http.StatusInternalServerError, err.Error()}
}
func (r Response) Error() string {
return fmt.Sprintf("%d: %s", r.Code, r.Message)
return fmt.Sprintf("[%d] %s", r.Code, r.Message)
}
func (r Response) Response(ctx fiber.Ctx) error {
return ctx.Status(r.Code).JSON(r)
}
var (
RequestParseError = Response{400, "请求解析错误"}
InternalError = Response{500, "内部错误"}
RequestParseError = Response{http.StatusBadRequest, http.StatusBadRequest, "请求解析错误"}
InternalError = Response{http.StatusInternalServerError, http.StatusInternalServerError, "内部错误"}
UserBalanceNotEnough = Response{http.StatusPaymentRequired, 1001, "余额不足，请充值"}
)

1
backend/pkg/service/http/http.go
View File

@@ -64,6 +64,7 @@ func Serve(cmd *cobra.Command, args []string) error {
mid := http.Middlewares
http.Service.Engine.Use(mid.DebugMode)
http.Service.Engine.Use(mid.ProcessResponse)
http.Service.Engine.Use(mid.WeChatVerify)
http.Service.Engine.Use(mid.WeChatAuthUserInfo)
http.Service.Engine.Use(mid.WeChatSilentAuth)

2
backend/pkg/service/tasks/tasks.go
View File

@@ -5,6 +5,7 @@ import (
"backend/modules/commands/store"
"backend/modules/medias"
"backend/providers/app"
"backend/providers/hashids"
"backend/providers/postgres"
"backend/providers/storage"
@@ -18,6 +19,7 @@ func defaultProviders(providers ...container.ProviderContainer) container.Provid
app.DefaultProvider(),
storage.DefaultProvider(),
postgres.DefaultProvider(),
hashids.DefaultProvider(),
}, providers...)
}

1
backend/pkg/session.go
View File

@@ -1 +0,0 @@
package common