dashboard-cdc-worker/internal/storage/recharge_reconcile.go

package storage

import (
	"context"
	"database/sql"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/shopspring/decimal"

	"dashboard-cdc-worker/internal/dashboard"
)

const (
	reconcileOfficialSource = "likei.order_purchase_history"
	reconcileMifapaySource  = "likei.order_user_purchase_pay"
)

type officialRechargeRow struct {
	id           string
	userID       int64
	sysOrigin    string
	payPlatform  string
	unitPrice    decimal.Decimal
	coinQuantity decimal.Decimal
	eventTime    time.Time
	user         dashboard.UserCountry
}

type mifapayRechargeRow struct {
	id          string
	userID      int64
	sysOrigin   string
	factoryCode string
	amount      decimal.Decimal
	eventTime   time.Time
	user        dashboard.UserCountry
}

func (r *MySQLRepository) BuildRechargeReconcileEvents(ctx context.Context, since time.Time,
	limit int) ([]dashboard.Event, error) {
	if limit <= 0 {
		limit = 5000
	}
	officialRows, err := r.listOfficialRechargeRows(ctx, since, limit)
	if err != nil {
		return nil, err
	}
	remaining := limit - len(officialRows)
	if remaining <= 0 {
		remaining = 1
	}
	mifapayRows, err := r.listMifapayRechargeRows(ctx, since, remaining)
	if err != nil {
		return nil, err
	}
	events := make([]dashboard.Event, 0, len(officialRows)+len(mifapayRows))
	for _, row := range officialRows {
		if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin {
			continue
		}
		contribution := dashboard.Contribution{
			SourceSchema:             "likei",
			SourceTable:              "order_purchase_history",
			SourcePK:                 row.id,
			SourceAction:             "insert",
			DeltaSign:                1,
			SysOrigin:                row.sysOrigin,
			UserID:                   row.userID,
			Country:                  row.user.Country,
			EventTime:                row.eventTime,
			UserCreatedAt:            row.user.CreatedAt,
			NewUserRechargeCandidate: row.unitPrice,
			RechargeDetail:           officialRechargeDetail(row),
		}
		if row.payPlatform != "MIFA_PAY" {
			contribution.OfficialRecharge = row.unitPrice
		}
		if row.payPlatform == "GOOGLE" {
			contribution.GoogleRecharge = row.unitPrice
			contribution.UserRecharge = row.unitPrice
		}
		events = append(events, dashboard.Event{
			EventKey:      rechargeReconcileEventKey(reconcileOfficialSource, row.id),
			Source:        reconcileOfficialSource,
			Contributions: []dashboard.Contribution{contribution},
		})
	}
	for _, row := range mifapayRows {
		if !r.allowedOrigin(row.sysOrigin) || row.user.IsDeleted || row.user.SysOrigin != row.sysOrigin {
			continue
		}
		contribution := dashboard.Contribution{
			SourceSchema:             "likei",
			SourceTable:              "order_user_purchase_pay",
			SourcePK:                 row.id,
			SourceAction:             "insert",
			DeltaSign:                1,
			SysOrigin:                row.sysOrigin,
			UserID:                   row.userID,
			Country:                  row.user.Country,
			EventTime:                row.eventTime,
			UserCreatedAt:            row.user.CreatedAt,
			NewUserRechargeCandidate: row.amount,
			UserRecharge:             row.amount,
			RechargeDetail:           mifapayRechargeDetail(row),
		}
		if row.factoryCode == "MIFA_PAY" {
			contribution.MifapayRecharge = row.amount
		}
		events = append(events, dashboard.Event{
			EventKey:      rechargeReconcileEventKey(reconcileMifapaySource, row.id),
			Source:        reconcileMifapaySource,
			Contributions: []dashboard.Contribution{contribution},
		})
	}
	return events, nil
}

func (r *MySQLRepository) StartRechargeReconciler(ctx context.Context, logger *slog.Logger,
	projector interface {
		ApplyEvents(context.Context, []dashboard.Event) error
	}) {
	if !r.cfg.Recharge.Enabled || r.cfg.Recharge.Interval <= 0 {
		return
	}
	run := func() {
		since := currentStatDayStart(r.cfg.Dashboard.StatTimezones, r.cfg.Recharge.CurrentDayLag)
		events, err := r.BuildRechargeReconcileEvents(ctx, since, r.cfg.Recharge.BatchLimit)
		if err != nil {
			if r.metrics != nil {
				r.metrics.RecordError(err)
			}
			logger.Error("dashboard recharge reconcile query failed", "error", err)
			return
		}
		if len(events) == 0 {
			return
		}
		if err := projector.ApplyEvents(ctx, events); err != nil {
			if r.metrics != nil {
				r.metrics.RecordError(err)
			}
			logger.Error("dashboard recharge reconcile apply failed", "events", len(events), "error", err)
			return
		}
		logger.Info("dashboard recharge reconcile applied", "events", len(events), "since", since)
	}
	go func() {
		run()
		ticker := time.NewTicker(r.cfg.Recharge.Interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				run()
			}
		}
	}()
}

func (r *MySQLRepository) listOfficialRechargeRows(ctx context.Context, since time.Time,
	limit int) ([]officialRechargeRow, error) {
	rows, err := r.db.QueryContext(ctx, `
SELECT CAST(o.id AS CHAR), o.user_id, o.sys_origin, o.pay_platform, o.unit_price, 0 AS coin_quantity, o.create_time,
       u.id, u.origin_sys, IFNULL(NULLIF(u.country_code, ''), ?),
       IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)),
       IFNULL(u.is_del, 0), u.create_time, u.update_time
FROM order_purchase_history o
JOIN user_base_info u ON u.id = o.user_id
WHERE o.create_time >= ?
  AND o.evn = 'PROD'
  AND o.status = 'COMPLETE'
  AND IFNULL(o.is_trial_period, 0) = 0
ORDER BY o.create_time ASC
LIMIT ?`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode, since, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []officialRechargeRow
	for rows.Next() {
		var item officialRechargeRow
		if err := scanRechargeUser(rows, &item.id, &item.userID, &item.sysOrigin,
			&item.payPlatform, &item.unitPrice, &item.coinQuantity, &item.eventTime,
			&item.user); err != nil {
			return nil, err
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func (r *MySQLRepository) listMifapayRechargeRows(ctx context.Context, since time.Time,
	limit int) ([]mifapayRechargeRow, error) {
	rows, err := r.db.QueryContext(ctx, `
SELECT CAST(o.id AS CHAR), o.user_id, o.sys_origin, o.factory_code, o.compute_usd_amount, o.create_time,
       u.id, u.origin_sys, IFNULL(NULLIF(u.country_code, ''), ?),
       IFNULL(NULLIF(u.country_name, ''), IFNULL(NULLIF(u.country_code, ''), ?)),
       IFNULL(u.is_del, 0), u.create_time, u.update_time
FROM order_user_purchase_pay o
JOIN user_base_info u ON u.id = o.user_id
WHERE o.create_time >= ?
  AND o.evn = 'PROD'
  AND o.pay_status = 'SUCCESSFUL'
  AND o.receipt_type = 'PAYMENT'
  AND o.refund_status = 'NONE'
ORDER BY o.create_time ASC
LIMIT ?`, r.cfg.Dashboard.UnknownCountryCode, r.cfg.Dashboard.UnknownCountryCode, since, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []mifapayRechargeRow
	for rows.Next() {
		var item mifapayRechargeRow
		var countryCode, countryName string
		var isDeleted bool
		var userCreatedAt, userUpdatedAt sql.NullTime
		if err := rows.Scan(&item.id, &item.userID, &item.sysOrigin, &item.factoryCode, &item.amount,
			&item.eventTime, &item.user.UserID, &item.user.SysOrigin, &countryCode,
			&countryName, &isDeleted, &userCreatedAt, &userUpdatedAt); err != nil {
			return nil, err
		}
		item.user.Country = dashboard.Country{Code: countryCode, Name: countryName}
		item.user.IsDeleted = isDeleted
		if userCreatedAt.Valid {
			item.user.CreatedAt = userCreatedAt.Time
		}
		if userUpdatedAt.Valid {
			item.user.UpdatedAt = userUpdatedAt.Time
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

func scanRechargeUser(rows *sql.Rows, id *string, userID *int64, sysOrigin *string,
	payPlatform *string, unitPrice *decimal.Decimal, coinQuantity *decimal.Decimal,
	eventTime *time.Time, user *dashboard.UserCountry) error {
	var countryCode, countryName string
	var isDeleted bool
	var userCreatedAt, userUpdatedAt sql.NullTime
	if err := rows.Scan(id, userID, sysOrigin, payPlatform, unitPrice, coinQuantity, eventTime,
		&user.UserID, &user.SysOrigin, &countryCode, &countryName, &isDeleted,
		&userCreatedAt, &userUpdatedAt); err != nil {
		return err
	}
	user.Country = dashboard.Country{Code: countryCode, Name: countryName}
	user.IsDeleted = isDeleted
	if userCreatedAt.Valid {
		user.CreatedAt = userCreatedAt.Time
	}
	if userUpdatedAt.Valid {
		user.UpdatedAt = userUpdatedAt.Time
	}
	return nil
}

func officialRechargeDetail(row officialRechargeRow) *dashboard.RechargeDetail {
	return &dashboard.RechargeDetail{
		SourceType:    "OFFICIAL",
		RecordID:      row.id,
		SourceChannel: row.payPlatform,
		SysOrigin:     row.user.SysOrigin,
		UserID:        row.userID,
		Country:       row.user.Country,
		Amount:        row.unitPrice,
		CoinQuantity:  row.coinQuantity,
		EventTime:     row.eventTime,
	}
}

func mifapayRechargeDetail(row mifapayRechargeRow) *dashboard.RechargeDetail {
	sourceType := "THIRD_PARTY"
	if row.factoryCode == "MIFA_PAY" {
		sourceType = "MIFAPAY"
	}
	return &dashboard.RechargeDetail{
		SourceType:    sourceType,
		RecordID:      row.id,
		SourceChannel: row.factoryCode,
		SysOrigin:     row.user.SysOrigin,
		UserID:        row.userID,
		Country:       row.user.Country,
		Amount:        row.amount,
		EventTime:     row.eventTime,
	}
}

func rechargeReconcileEventKey(source string, id string) string {
	return fmt.Sprintf("reconcile:%s:%s:complete", source, id)
}

func currentStatDayStart(statTimezones []string, lag time.Duration) time.Time {
	timezone := "Asia/Riyadh"
	if len(statTimezones) > 0 && strings.TrimSpace(statTimezones[0]) != "" {
		timezone = strings.TrimSpace(statTimezones[0])
	}
	location, err := time.LoadLocation(timezone)
	if err != nil {
		location = time.UTC
	}
	now := time.Now().In(location).Add(-lag)
	return time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, location)
}

func (r *MySQLRepository) allowedOrigin(sysOrigin string) bool {
	if len(r.cfg.Dashboard.SysOrigins) == 0 {
		return true
	}
	for _, origin := range r.cfg.Dashboard.SysOrigins {
		if strings.TrimSpace(origin) == sysOrigin {
			return true
		}
	}
	return false
}