277 lines
9.6 KiB
Go
277 lines
9.6 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
|
|
"github.com/shopspring/decimal"
|
|
|
|
"dashboard-cdc-worker/internal/dashboard"
|
|
)
|
|
|
|
func (r *MySQLRepository) listSQLLuckyGiftRows(ctx context.Context, start, end time.Time,
|
|
statTimezone string) ([]sqlLuckyGiftRow, error) {
|
|
startUTC := start.In(time.UTC).Format("2006-01-02 15:04:05")
|
|
endUTC := end.In(time.UTC).Format("2006-01-02 15:04:05")
|
|
query := `
|
|
SELECT CAST(t.id AS CHAR), t.user_id, DATE_FORMAT(CONVERT_TZ(t.create_time, '+00:00', ?), '%Y-%m-%d'),
|
|
t.sys_origin, IFNULL(NULLIF(u.country_code, ''), ?),
|
|
IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)),
|
|
t.pay_amount, t.award_amount
|
|
FROM ` + quoteIdent(r.cfg.MySQL.Schema) + `.game_lucky_gift_count t
|
|
JOIN ` + quoteIdent(r.cfg.MySQL.Schema) + `.user_base_info u ON u.id = t.user_id
|
|
WHERE t.create_time >= ?
|
|
AND t.create_time < ?
|
|
AND t.sys_origin IN ` + inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins)) + `
|
|
AND u.origin_sys = t.sys_origin
|
|
AND (u.is_del = 0 OR u.is_del IS NULL)`
|
|
args := []interface{}{timezoneOffset(statTimezone), r.cfg.Dashboard.UnknownCountryCode,
|
|
r.cfg.Dashboard.UnknownCountryCode, startUTC, endUTC}
|
|
args = append(args, stringArgs(r.cfg.Dashboard.SysOrigins)...)
|
|
rows, err := r.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []sqlLuckyGiftRow
|
|
for rows.Next() {
|
|
var item sqlLuckyGiftRow
|
|
var pay, payout decimal.Decimal
|
|
if err := rows.Scan(&item.id, &item.userID, &item.day, &item.sysOrigin, &item.country.Code,
|
|
&item.country.Name, &pay, &payout); err != nil {
|
|
return nil, err
|
|
}
|
|
item.payAmount = pay
|
|
item.payout = payout
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *MySQLRepository) replaceGiftMetricRows(ctx context.Context, tx *sql.Tx,
|
|
statTimezone string, start, end time.Time, rows map[string]*giftMetricRow) error {
|
|
location, err := time.LoadLocation(statTimezone)
|
|
if err != nil {
|
|
location = time.UTC
|
|
}
|
|
for day := start.In(location); day.Before(end); day = day.AddDate(0, 0, 1) {
|
|
dayStart := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, location)
|
|
dayEnd := dayStart.AddDate(0, 0, 1)
|
|
dayKey := dayStart.Format("2006-01-02")
|
|
for _, sysOrigin := range r.cfg.Dashboard.SysOrigins {
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE country_dashboard_period_metric
|
|
SET gift_consume = 0,
|
|
lucky_gift_total_flow = 0,
|
|
lucky_gift_user = 0,
|
|
lucky_gift_payout = 0,
|
|
lucky_gift_anchor_share = 0,
|
|
refreshed_at = NOW()
|
|
WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?`,
|
|
dayKey, statTimezone, sysOrigin); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE country_dashboard_daily_metric
|
|
SET gift_consume = 0,
|
|
lucky_gift_total_flow = 0,
|
|
lucky_gift_user = 0,
|
|
lucky_gift_payout = 0,
|
|
lucky_gift_anchor_share = 0,
|
|
refreshed_at = NOW()
|
|
WHERE stat_date = ? AND sys_origin = ?`,
|
|
dayKey, sysOrigin); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
DELETE FROM country_dashboard_period_user_metric
|
|
WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?
|
|
AND metric_type = ?`, dayKey, statTimezone, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
DELETE FROM country_dashboard_daily_user_metric
|
|
WHERE stat_date = ? AND sys_origin = ? AND metric_type = ?`, dayKey, sysOrigin, dashboard.MetricLuckyGiftUser); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := r.upsertGiftMetricRows(ctx, tx, statTimezone, dayStart, dayEnd, rows); err != nil {
|
|
return err
|
|
}
|
|
if err := r.upsertDailyGiftMetricRows(ctx, tx, dayStart, rows); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *MySQLRepository) upsertGiftMetricRows(ctx context.Context, tx *sql.Tx,
|
|
statTimezone string, dayStart, dayEnd time.Time, rows map[string]*giftMetricRow) error {
|
|
dayKey := dayStart.Format("2006-01-02")
|
|
var metricRows []*giftMetricRow
|
|
for _, row := range rows {
|
|
if row.day != dayKey || row.empty() {
|
|
continue
|
|
}
|
|
if row.sysOrigin == "" {
|
|
row.sysOrigin = defaultSysOrigin(r.cfg.Dashboard.SysOrigins)
|
|
}
|
|
if !r.allowedOrigin(row.sysOrigin) {
|
|
continue
|
|
}
|
|
if row.country.Code == "" {
|
|
row.country.Code = r.cfg.Dashboard.UnknownCountryCode
|
|
}
|
|
if row.country.Name == "" {
|
|
row.country.Name = row.country.Code
|
|
}
|
|
metricRows = append(metricRows, row)
|
|
}
|
|
if len(metricRows) == 0 {
|
|
return nil
|
|
}
|
|
args := make([]interface{}, 0, len(metricRows)*14)
|
|
for _, row := range metricRows {
|
|
args = append(args, dashboard.PeriodDay, dayKey, statTimezone, dayKey, dayStart, dayEnd,
|
|
row.sysOrigin, row.country.Code, row.country.Name, row.giftConsume,
|
|
row.luckyGiftTotalFlow, len(row.luckyGiftUsers), row.luckyGiftPayout,
|
|
row.luckyGiftAnchorShare)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO country_dashboard_period_metric (
|
|
period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
|
|
sys_origin, country_code, country_name, gift_consume, lucky_gift_total_flow,
|
|
lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share
|
|
) VALUES `+placeholders(len(metricRows), 14)+`
|
|
ON DUPLICATE KEY UPDATE
|
|
period_name = VALUES(period_name),
|
|
period_start_date = VALUES(period_start_date),
|
|
period_end_date = VALUES(period_end_date),
|
|
country_name = VALUES(country_name),
|
|
gift_consume = VALUES(gift_consume),
|
|
lucky_gift_total_flow = VALUES(lucky_gift_total_flow),
|
|
lucky_gift_user = VALUES(lucky_gift_user),
|
|
lucky_gift_payout = VALUES(lucky_gift_payout),
|
|
lucky_gift_anchor_share = VALUES(lucky_gift_anchor_share),
|
|
refreshed_at = NOW()`, args...); err != nil {
|
|
return err
|
|
}
|
|
for _, row := range metricRows {
|
|
if len(row.luckyGiftUsers) == 0 {
|
|
continue
|
|
}
|
|
key := periodUserKey{
|
|
periodType: dashboard.PeriodDay, periodKey: dayKey, statTimezone: statTimezone,
|
|
periodName: dayKey, startDate: dayStart, endDate: dayEnd, sysOrigin: row.sysOrigin,
|
|
countryCode: row.country.Code, countryName: row.country.Name, metricType: dashboard.MetricLuckyGiftUser,
|
|
}
|
|
if _, err := bulkInsertPeriodUsers(ctx, tx, key, int64SetToSlice(row.luckyGiftUsers)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *MySQLRepository) upsertDailyGiftMetricRows(ctx context.Context, tx *sql.Tx,
|
|
dayStart time.Time, rows map[string]*giftMetricRow) error {
|
|
dayKey := dayStart.Format("2006-01-02")
|
|
var metricRows []*giftMetricRow
|
|
for _, row := range rows {
|
|
if row.day != dayKey || row.empty() {
|
|
continue
|
|
}
|
|
if row.sysOrigin == "" {
|
|
row.sysOrigin = defaultSysOrigin(r.cfg.Dashboard.SysOrigins)
|
|
}
|
|
if !r.allowedOrigin(row.sysOrigin) {
|
|
continue
|
|
}
|
|
if row.country.Code == "" {
|
|
row.country.Code = r.cfg.Dashboard.UnknownCountryCode
|
|
}
|
|
if row.country.Name == "" {
|
|
row.country.Name = row.country.Code
|
|
}
|
|
metricRows = append(metricRows, row)
|
|
}
|
|
if len(metricRows) == 0 {
|
|
return nil
|
|
}
|
|
args := make([]interface{}, 0, len(metricRows)*10)
|
|
for _, row := range metricRows {
|
|
args = append(args, dayKey, dateNumber(dayStart), row.sysOrigin, row.country.Code, row.country.Name,
|
|
row.giftConsume, row.luckyGiftTotalFlow, len(row.luckyGiftUsers), row.luckyGiftPayout,
|
|
row.luckyGiftAnchorShare)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO country_dashboard_daily_metric (
|
|
stat_date, date_number, sys_origin, country_code, country_name, gift_consume,
|
|
lucky_gift_total_flow, lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share
|
|
) VALUES `+placeholders(len(metricRows), 10)+`
|
|
ON DUPLICATE KEY UPDATE
|
|
country_name = VALUES(country_name),
|
|
gift_consume = VALUES(gift_consume),
|
|
lucky_gift_total_flow = VALUES(lucky_gift_total_flow),
|
|
lucky_gift_user = VALUES(lucky_gift_user),
|
|
lucky_gift_payout = VALUES(lucky_gift_payout),
|
|
lucky_gift_anchor_share = VALUES(lucky_gift_anchor_share),
|
|
refreshed_at = NOW()`, args...); err != nil {
|
|
return err
|
|
}
|
|
for _, row := range metricRows {
|
|
if len(row.luckyGiftUsers) == 0 {
|
|
continue
|
|
}
|
|
key := dailyUserKey{
|
|
day: dayStart, sysOrigin: row.sysOrigin, countryCode: row.country.Code,
|
|
countryName: row.country.Name, metricType: dashboard.MetricLuckyGiftUser,
|
|
}
|
|
if _, err := bulkInsertDailyUsers(ctx, tx, key, int64SetToSlice(row.luckyGiftUsers)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *MySQLRepository) loadSourceUserCountriesByID(ctx context.Context, ids []int64) (map[int64]dashboard.UserCountry, error) {
|
|
result := make(map[int64]dashboard.UserCountry)
|
|
for start := 0; start < len(ids); start += bulkInsertChunkSize {
|
|
end := min(start+bulkInsertChunkSize, len(ids))
|
|
args := make([]interface{}, 0, end-start+2)
|
|
args = append(args, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode)
|
|
for _, id := range ids[start:end] {
|
|
args = append(args, id)
|
|
}
|
|
rows, err := r.db.QueryContext(ctx, `
|
|
SELECT id, origin_sys, IFNULL(NULLIF(country_code, ''), ?),
|
|
IFNULL(NULLIF(country_name, ''), IFNULL(NULLIF(country_code, ''), ?)),
|
|
IFNULL(is_del, 0), create_time, update_time
|
|
FROM `+quoteIdent(r.cfg.MySQL.Schema)+`.user_base_info
|
|
WHERE id IN `+inPlaceholders(end-start), args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for rows.Next() {
|
|
var user dashboard.UserCountry
|
|
var createdAt, updatedAt sql.NullTime
|
|
if err := rows.Scan(&user.UserID, &user.SysOrigin, &user.Country.Code, &user.Country.Name,
|
|
&user.IsDeleted, &createdAt, &updatedAt); err != nil {
|
|
_ = rows.Close()
|
|
return nil, err
|
|
}
|
|
if createdAt.Valid {
|
|
user.CreatedAt = createdAt.Time
|
|
}
|
|
if updatedAt.Valid {
|
|
user.UpdatedAt = updatedAt.Time
|
|
}
|
|
result[user.UserID] = user
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|