427 lines
14 KiB
Go
427 lines
14 KiB
Go
package dao
|
|
|
|
import (
|
|
"context"
|
|
xsql "database/sql"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"go-common/app/service/main/thumbup/model"
|
|
sql "go-common/library/database/tidb"
|
|
"go-common/library/log"
|
|
xtime "go-common/library/time"
|
|
"go-common/library/xstr"
|
|
|
|
"go-common/library/sync/errgroup"
|
|
)
|
|
|
|
const (
|
|
_tidbBulkSize = 50
|
|
_tidbBusinessesSQL = "SELECT id, name, message_list_type, user_likes_limit, message_likes_limit, enable_originid, user_list_type FROM business WHERE dtime = '0000-00-00 00:00:00'"
|
|
_tidbLikeMidSQL = "SELECT type FROM likes WHERE mid=? AND business_id = ? AND origin_id = ? AND message_id = ?"
|
|
_tidbUserLikeListSQL = "SELECT message_id, mtime FROM likes WHERE mid = ? AND business_id =? AND type = ? ORDER BY mtime desc LIMIT ? OFFSET ?"
|
|
_tidbUserHasLikeSQL = "SELECT message_id, mtime FROM likes WHERE mid = ? AND business_id =? AND type = ? AND message_id in (%s)"
|
|
_tidbItemLikeListSQL = "SELECT mid, mtime FROM likes WHERE business_id =? AND origin_id =? AND message_id =? AND type = ? ORDER BY mtime desc LIMIT ? OFFSET ?"
|
|
_tidbStatSQL = "SELECT message_id, likes_count, dislikes_count, origin_id, likes_change, dislikes_change FROM counts WHERE business_id = ? AND origin_id = ? AND message_id = ?"
|
|
_tidbStatsSQL = "SELECT message_id, likes_count, dislikes_count, likes_change, dislikes_change FROM counts WHERE business_id = ? AND origin_id = 0 AND message_id in (%s)"
|
|
_tidbStatsOriginSQL = "SELECT message_id, likes_count, dislikes_count, likes_change, dislikes_change FROM counts WHERE business_id = ? AND origin_id = ?"
|
|
_tidbUserLikeCountSQL = "SELECT COUNT(*) FROM likes WHERE business_id =? AND mid = ? AND type = ?"
|
|
_tidbUpdateLikeSQL = "INSERT INTO likes (business_id, origin_id, message_id, mid, type, mtime) VALUES (?,?,?,?,?,?) ON DUPLICATE KEY UPDATE type=?, mtime = ?"
|
|
_tidbUpdateCountsSQL = "INSERT INTO counts (business_id, origin_id, message_id, likes_count, dislikes_count, up_mid) VALUES (?,?,?,?,?,?) ON DUPLICATE KEY UPDATE "
|
|
_tidbupdateCountChange = "UPDATE counts SET likes_change = likes_change + ?, dislikes_change = dislikes_change + ? WHERE business_id = ? AND origin_id = ? AND message_id = ?"
|
|
_tidbItemHasLikeSQL = "SELECT mid, mtime FROM likes WHERE business_id = ? AND origin_id = ? AND message_id = ? AND type =? AND mid in (%s)"
|
|
)
|
|
|
|
// Businesses get business list
|
|
func (d *Dao) Businesses(c context.Context) (res []*model.Business, err error) {
|
|
var rows *sql.Rows
|
|
if rows, err = d.businessesStmt.Query(c); err != nil {
|
|
PromError("tidb:业务查询")
|
|
log.Error("tidb.businessesStmt.Query error(%v)", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
b := &model.Business{}
|
|
if err = rows.Scan(&b.ID, &b.Name, &b.MessageListType, &b.UserLikesLimit, &b.MessageLikesLimit, &b.EnableOriginID, &b.UserListType); err != nil {
|
|
PromError("tidb:业务Scan")
|
|
log.Error("tidb.rows.Business.Scan error(%v)", err)
|
|
return
|
|
}
|
|
res = append(res, b)
|
|
}
|
|
err = rows.Err()
|
|
return
|
|
}
|
|
|
|
// LikeState get like state
|
|
func (d *Dao) LikeState(c context.Context, mid, businessID, originID, messageID int64) (res int8, err error) {
|
|
if err = d.likeStateStmt.QueryRow(c, mid, businessID, originID, messageID).Scan(&res); err != nil {
|
|
if err == sql.ErrNoRows {
|
|
err = nil
|
|
return
|
|
}
|
|
PromError("tidb:LikeByMid")
|
|
log.Error("tidbLikeByMid(%d,%d,%d,%d) error(%v)", mid, businessID, originID, messageID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// UserLikeCount count
|
|
func (d *Dao) UserLikeCount(c context.Context, businessID, mid int64, typ int8) (res int, err error) {
|
|
if err = d.userLikeCountStmt.QueryRow(c, businessID, mid, typ).Scan(&res); err != nil {
|
|
PromError("tidb:UserLikeCount")
|
|
log.Error("tidbUserLikeCount(%d,%d,%d) error(%v)", businessID, mid, typ, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// RawItemLikeList item like list
|
|
func (d *Dao) RawItemLikeList(c context.Context, messageID, businessID, originID int64, state int8, start, end int) (res []*model.UserLikeRecord, err error) {
|
|
limit := end - start + 1
|
|
var rows *sql.Rows
|
|
if rows, err = d.itemLikeListStmt.Query(c, businessID, originID, messageID, state, limit, start); err != nil {
|
|
PromError("tidb:项目点赞列表")
|
|
log.Error("tidb.ItemLikeList.Query error(%v,%v,%v,%v,%v, %v)", businessID, originID, messageID, state, limit, err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
b := &model.UserLikeRecord{}
|
|
var t time.Time
|
|
if err = rows.Scan(&b.Mid, &t); err != nil {
|
|
PromError("tidb:业务Scan")
|
|
log.Error("tidb.rows.Business.Scan error(%v)", err)
|
|
return
|
|
}
|
|
b.Time = xtime.Time(t.Unix())
|
|
res = append(res, b)
|
|
}
|
|
err = rows.Err()
|
|
return
|
|
}
|
|
|
|
// RawUserLikeList .
|
|
func (d *Dao) RawUserLikeList(c context.Context, mid, businessID int64, state int8, start, end int) (res []*model.ItemLikeRecord, err error) {
|
|
limit := end - start
|
|
var rows *sql.Rows
|
|
if rows, err = d.userLikeListStmt.Query(c, mid, businessID, state, limit, start); err != nil {
|
|
PromError("tidb:用户点赞列表")
|
|
log.Error("tidb.UserLikeList.Query error(%v,%v,%v,%v,%v)", mid, businessID, state, limit, err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
b := &model.ItemLikeRecord{}
|
|
var t time.Time
|
|
if err = rows.Scan(&b.MessageID, &t); err != nil {
|
|
PromError("tidb:用户Scan")
|
|
log.Error("tidb.rows.Business.Scan error(%v)", err)
|
|
return
|
|
}
|
|
b.Time = xtime.Time(t.Unix())
|
|
res = append(res, b)
|
|
}
|
|
err = rows.Err()
|
|
return
|
|
}
|
|
|
|
// MessageStats .
|
|
func (d *Dao) MessageStats(c context.Context, businessID int64, ids []int64) (res map[int64]*model.Stats, err error) {
|
|
res = make(map[int64]*model.Stats)
|
|
var (
|
|
group = &errgroup.Group{}
|
|
mutex = &sync.Mutex{}
|
|
)
|
|
keysLen := len(ids)
|
|
for i := 0; i < keysLen; i += _tidbBulkSize {
|
|
var keys []int64
|
|
if (i + _tidbBulkSize) > keysLen {
|
|
keys = ids[i:]
|
|
} else {
|
|
keys = ids[i : i+_tidbBulkSize]
|
|
}
|
|
group.Go(func() error {
|
|
statsSQL := fmt.Sprintf(_tidbStatsSQL, xstr.JoinInts(keys))
|
|
rows, e := d.tidb.Query(c, statsSQL, businessID)
|
|
if e != nil {
|
|
err = e
|
|
return nil
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
s := &model.Stats{}
|
|
var likeChange, dislikeChange int64
|
|
e = rows.Scan(&s.ID, &s.Likes, &s.Dislikes, &likeChange, &dislikeChange)
|
|
s.Likes += likeChange
|
|
s.Dislikes += dislikeChange
|
|
if e != nil {
|
|
err = e
|
|
continue
|
|
}
|
|
mutex.Lock()
|
|
res[s.ID] = s
|
|
mutex.Unlock()
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
group.Wait()
|
|
if err != nil {
|
|
PromError("tidb stats Scan")
|
|
log.Error("tidb.rows.Stats.Scan error(%v)", err)
|
|
}
|
|
if len(res) == 0 {
|
|
res = nil
|
|
}
|
|
return
|
|
}
|
|
|
|
// OriginStats .
|
|
func (d *Dao) OriginStats(c context.Context, businessID, originID int64) (res map[int64]*model.Stats, err error) {
|
|
var rows *sql.Rows
|
|
if rows, err = d.statsOriginStmt.Query(c, businessID, originID); err != nil {
|
|
PromError("tidb:计数表")
|
|
log.Error("db.Stats.Query error(%v,%v,%v)", businessID, originID, err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
b := &model.Stats{OriginID: originID}
|
|
var likeChange, dislikeChange int64
|
|
if err = rows.Scan(&b.ID, &b.Likes, &b.Dislikes, &likeChange, &dislikeChange); err != nil {
|
|
PromError("tidb:计数Scan")
|
|
log.Error("tidb.rows.Stats.Scan error(%v)", err)
|
|
return
|
|
}
|
|
b.Likes += likeChange
|
|
b.Dislikes += dislikeChange
|
|
if res == nil {
|
|
res = make(map[int64]*model.Stats)
|
|
}
|
|
res[b.ID] = b
|
|
}
|
|
err = rows.Err()
|
|
return
|
|
}
|
|
|
|
// Stat .
|
|
func (d *Dao) Stat(c context.Context, businessID, originID, messageID int64) (res model.Stats, err error) {
|
|
var likeChange, dislikeChange int64
|
|
if err = d.statStmt.QueryRow(c, businessID, originID, messageID).Scan(&res.ID, &res.Likes, &res.Dislikes, &res.OriginID, &likeChange, &dislikeChange); err != nil {
|
|
if err == sql.ErrNoRows {
|
|
err = nil
|
|
res.ID = messageID
|
|
res.OriginID = originID
|
|
return
|
|
}
|
|
PromError("tidb:查询计数表")
|
|
log.Error("tidb.Stat.Query error(%v,%v,%v,%v)", businessID, originID, messageID, err)
|
|
}
|
|
res.Likes += likeChange
|
|
res.Dislikes += dislikeChange
|
|
return
|
|
}
|
|
|
|
// RawStats stat changes
|
|
func (d *Dao) RawStats(c context.Context, businessID, originID, messageID int64) (res model.RawStats, err error) {
|
|
if err = d.statStmt.QueryRow(c, businessID, originID, messageID).Scan(&res.ID, &res.Likes, &res.Dislikes, &res.OriginID, &res.LikesChange, &res.DislikesChange); err != nil {
|
|
if err == sql.ErrNoRows {
|
|
err = nil
|
|
res.ID = messageID
|
|
res.OriginID = originID
|
|
return
|
|
}
|
|
PromError("tidb:查询change计数表")
|
|
log.Error("tidb.StatChanges.Query error(%v,%v,%v,%v)", businessID, originID, messageID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// TxUpdateCounts update like info via transaction.
|
|
func (d *Dao) tidbTxUpdateCounts(c context.Context, tx *sql.Tx, businessID, originID, messageID int64, likesCount, dislikesCount, upMid int64) (err error) {
|
|
likeSQL := _tidbUpdateCountsSQL
|
|
if (likesCount == 0) && (dislikesCount == 0) {
|
|
return
|
|
}
|
|
if (likesCount != 0) && (dislikesCount == 0) {
|
|
if likesCount > 0 {
|
|
likeSQL += fmt.Sprintf("likes_count = likes_count + %v", likesCount)
|
|
} else {
|
|
likeSQL += fmt.Sprintf("likes_count = likes_count %v", likesCount)
|
|
}
|
|
} else if (likesCount == 0) && (dislikesCount != 0) {
|
|
if dislikesCount > 0 {
|
|
likeSQL += fmt.Sprintf("dislikes_count = dislikes_count + %v", dislikesCount)
|
|
} else {
|
|
likeSQL += fmt.Sprintf("dislikes_count = dislikes_count %v", dislikesCount)
|
|
}
|
|
} else {
|
|
if likesCount > 0 {
|
|
likeSQL += fmt.Sprintf("likes_count = likes_count + %v", likesCount)
|
|
} else {
|
|
likeSQL += fmt.Sprintf("likes_count = likes_count %v", likesCount)
|
|
}
|
|
if dislikesCount > 0 {
|
|
likeSQL += fmt.Sprintf(", dislikes_count = dislikes_count + %v", dislikesCount)
|
|
} else {
|
|
likeSQL += fmt.Sprintf(", dislikes_count = dislikes_count %v", dislikesCount)
|
|
}
|
|
}
|
|
if upMid > 0 {
|
|
likeSQL += fmt.Sprintf(", up_mid = %d", upMid)
|
|
}
|
|
if _, err = tx.Exec(likeSQL, businessID, originID, messageID, likesCount, dislikesCount, upMid); err != nil {
|
|
PromError("tidb:更新计数表")
|
|
log.Error("dao.tidbTxUpdateCounts tx.Exec(%s, %v,%v,%v) error(%v)", likeSQL, businessID, originID, messageID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// tidbTxStat .
|
|
func (d *Dao) tidbTxStat(c context.Context, tx *sql.Tx, businessID, originID, messageID int64) (res model.Stats, err error) {
|
|
var likeChange, dislikeChange int64
|
|
if err = tx.Stmts(d.statStmt).QueryRow(c, businessID, originID, messageID).Scan(&res.ID, &res.Likes, &res.Dislikes, &res.OriginID, &likeChange, &dislikeChange); err != nil {
|
|
if err == sql.ErrNoRows {
|
|
err = nil
|
|
res.ID = messageID
|
|
res.OriginID = originID
|
|
return
|
|
}
|
|
PromError("tidb:查询计数表")
|
|
log.Error("tidb: Tx db.Stat.Query error(%v,%v,%v,%v)", businessID, originID, messageID, err)
|
|
}
|
|
res.Likes += likeChange
|
|
res.Dislikes += dislikeChange
|
|
return
|
|
}
|
|
|
|
// UpdateCount .
|
|
func (d *Dao) UpdateCount(c context.Context, businessID, originID, messageID int64, likeChange, dislikeChange int64) (err error) {
|
|
if _, err = d.updateCountChangeStmt.Exec(c, likeChange, dislikeChange, businessID, originID, messageID); err != nil {
|
|
PromError("tidb:更新change计数表")
|
|
log.Error("db.tidbUpdateCountChange.Exec error(%v,%v,%v,%v)", businessID, originID, messageID, err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// tidbBeginTran begin transaction.
|
|
func (d *Dao) tidbBeginTran(c context.Context) (*sql.Tx, error) {
|
|
return d.tidb.Begin(c)
|
|
}
|
|
|
|
// UpdateCounts .
|
|
func (d *Dao) UpdateCounts(c context.Context, businessID, originID, messageID int64, likeCount, dislikeCount int64, upMid int64) (err error) {
|
|
var tx *sql.Tx
|
|
if tx, err = d.tidbBeginTran(c); err != nil {
|
|
log.Error("tx.BeginTran() error(%v)", err)
|
|
return
|
|
}
|
|
var stat model.Stats
|
|
if stat, err = d.tidbTxStat(c, tx, businessID, originID, messageID); err != nil {
|
|
if err1 := tx.Rollback(); err1 != nil {
|
|
log.Error("tx.Rollback() error(%v)", err1)
|
|
}
|
|
return
|
|
}
|
|
if stat.Likes+likeCount < 0 {
|
|
likeCount = -stat.Likes
|
|
}
|
|
if stat.Dislikes+dislikeCount < 0 {
|
|
dislikeCount = -stat.Dislikes
|
|
}
|
|
if err = d.tidbTxUpdateCounts(c, tx, businessID, originID, messageID, likeCount, dislikeCount, upMid); err != nil {
|
|
if err1 := tx.Rollback(); err1 != nil {
|
|
log.Error("tx.Rollback() error(%v)", err1)
|
|
}
|
|
return
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
PromError("like:更新喜欢计数")
|
|
log.Error("tx.Commit() error(%v)", err)
|
|
return
|
|
}
|
|
return
|
|
}
|
|
|
|
// UpdateUpMids .
|
|
func (d *Dao) UpdateUpMids(c context.Context, businessID int64, data []*model.UpMidsReq) (rows int64, err error) {
|
|
var tx *sql.Tx
|
|
if tx, err = d.tidbBeginTran(c); err != nil {
|
|
log.Error("UpdateUpMids tx.BeginTran() error(%v)", err)
|
|
return
|
|
}
|
|
var res xsql.Result
|
|
for _, u := range data {
|
|
if res, err = tx.Exec("UPDATE counts SET up_mid = ? WHERE business_id = ? AND origin_id = ? AND message_id = ?", u.UpMid, businessID, u.OriginID, u.MessageID); err != nil {
|
|
if err1 := tx.Rollback(); err1 != nil {
|
|
log.Error("UpdateUpMids tx.Rollback() error(%v)", err1)
|
|
}
|
|
return
|
|
}
|
|
aff, _ := res.RowsAffected()
|
|
rows += aff
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
PromError("like:更新UpMids")
|
|
log.Error("UpdateUpMids tx.Commit() error(%v)", err)
|
|
}
|
|
return
|
|
}
|
|
|
|
// ItemHasLike .
|
|
func (d *Dao) ItemHasLike(c context.Context, businessID int64, originID, messageID int64, mids []int64, typ int8) (res map[int64]int64, err error) {
|
|
var rows *sql.Rows
|
|
sqlStr := fmt.Sprintf(_tidbItemHasLikeSQL, xstr.JoinInts(mids))
|
|
if rows, err = d.tidb.Query(c, sqlStr, businessID, originID, messageID, typ); err != nil {
|
|
PromError("tidb:itemHasLike")
|
|
log.Error("tidb.ItemHasLike.Query asc error(%v,%v,%v,%v,%v)", businessID, originID, messageID, typ, err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var t time.Time
|
|
var mid int64
|
|
if res == nil {
|
|
res = make(map[int64]int64)
|
|
}
|
|
if err = rows.Scan(&mid, &t); err != nil {
|
|
PromError("tidb:itemHasLikeScan")
|
|
log.Error("tidb.rows.itemHasLike.Scan error(%v)", err)
|
|
return
|
|
}
|
|
res[mid] = t.Unix()
|
|
}
|
|
err = rows.Err()
|
|
return
|
|
}
|
|
|
|
// UserHasLike .
|
|
func (d *Dao) UserHasLike(c context.Context, businessID, mid int64, messageIDs []int64, typ int8) (res []*model.ItemLikeRecord, err error) {
|
|
var rows *sql.Rows
|
|
sqlStr := fmt.Sprintf(_tidbUserHasLikeSQL, xstr.JoinInts(messageIDs))
|
|
if rows, err = d.tidb.Query(c, sqlStr, mid, businessID, typ); err != nil {
|
|
PromError("tidb:userHasLike")
|
|
log.Error("tidb.UserHasLike.Query asc error(%v,%v,%v,%v,%v)", mid, businessID, messageIDs, typ, err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var t xtime.Time
|
|
var messageID int64
|
|
if err = rows.Scan(&messageID, &t); err != nil {
|
|
PromError("tidb:userHasLikeScan")
|
|
log.Error("tidb.rows.userHasLike.Scan error(%v)", err)
|
|
return
|
|
}
|
|
tmp := &model.ItemLikeRecord{
|
|
MessageID: messageID,
|
|
Time: t,
|
|
}
|
|
res = append(res, tmp)
|
|
}
|
|
err = rows.Err()
|
|
return
|
|
}
|