476 lines
16 KiB
Go
476 lines
16 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"net"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql"
|
||
)
|
||
|
||
type queryHTTPServer struct {
|
||
repo *mysqlstorage.Repository
|
||
server *http.Server
|
||
listener net.Listener
|
||
}
|
||
|
||
func newQueryHTTPServer(addr string, repo *mysqlstorage.Repository) (*queryHTTPServer, error) {
|
||
addr = strings.TrimSpace(addr)
|
||
if addr == "" {
|
||
return nil, nil
|
||
}
|
||
listener, err := net.Listen("tcp", addr)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s := &queryHTTPServer{repo: repo, listener: listener}
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("/internal/v1/statistics/overview", s.overview)
|
||
mux.HandleFunc("/internal/v1/statistics/platform-grants/users", s.platformGrantUsers)
|
||
mux.HandleFunc("/internal/v1/statistics/platform-grants/records", s.platformGrantRecords)
|
||
mux.HandleFunc("/internal/v1/statistics/app/events", s.appEvents)
|
||
mux.HandleFunc("/internal/v1/statistics/app/funnel", s.appFunnel)
|
||
mux.HandleFunc("/internal/v1/statistics/social/requirements", s.socialRequirements)
|
||
mux.HandleFunc("/internal/v1/statistics/social/coin-seller-recharge-orders", s.socialCoinSellerRechargeOrder)
|
||
mux.HandleFunc("/internal/v1/statistics/self-game/events", s.selfGameEvents)
|
||
mux.HandleFunc("/internal/v1/statistics/self-games/overview", s.selfGameOverview)
|
||
s.server = &http.Server{Handler: mux, ReadHeaderTimeout: 2 * time.Second}
|
||
return s, nil
|
||
}
|
||
|
||
func (s *queryHTTPServer) Run() error {
|
||
if s == nil {
|
||
return nil
|
||
}
|
||
err := s.server.Serve(s.listener)
|
||
if errors.Is(err, http.ErrServerClosed) {
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
|
||
func (s *queryHTTPServer) Close(ctx context.Context) error {
|
||
if s == nil {
|
||
return nil
|
||
}
|
||
return s.server.Shutdown(ctx)
|
||
}
|
||
|
||
func (s *queryHTTPServer) overview(w http.ResponseWriter, r *http.Request) {
|
||
query := r.URL.Query()
|
||
startMS := parseInt64(query.Get("start_ms"))
|
||
endMS := parseInt64(query.Get("end_ms"))
|
||
if startMS <= 0 {
|
||
now := time.Now().UTC()
|
||
startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli()
|
||
}
|
||
if endMS <= startMS {
|
||
endMS = startMS + 24*time.Hour.Milliseconds()
|
||
}
|
||
overview, err := s.repo.QueryOverview(r.Context(), mysqlstorage.OverviewQuery{
|
||
AppCode: query.Get("app_code"),
|
||
StatTZ: query.Get("stat_tz"),
|
||
StartMS: startMS,
|
||
EndMS: endMS,
|
||
SeriesStartMS: parseInt64(query.Get("series_start_ms")),
|
||
SeriesEndMS: parseInt64(query.Get("series_end_ms")),
|
||
CountryID: parseInt64(query.Get("country_id")),
|
||
RegionID: parseInt64(query.Get("region_id")),
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, overview)
|
||
}
|
||
|
||
func (s *queryHTTPServer) platformGrantUsers(w http.ResponseWriter, r *http.Request) {
|
||
query := r.URL.Query()
|
||
pageSize := parsePositiveInt(query.Get("page_size"))
|
||
if pageSize == 0 {
|
||
pageSize = parsePositiveInt(query.Get("pageSize"))
|
||
}
|
||
page, err := s.repo.QueryPlatformGrantCoinUsers(r.Context(), mysqlstorage.PlatformGrantCoinQuery{
|
||
AppCode: query.Get("app_code"),
|
||
StatTZ: query.Get("stat_tz"),
|
||
StartMS: parseInt64(query.Get("start_ms")),
|
||
EndMS: parseInt64(query.Get("end_ms")),
|
||
StatDay: query.Get("stat_day"),
|
||
CountryID: parseInt64(query.Get("country_id")),
|
||
RegionID: parseInt64(query.Get("region_id")),
|
||
EventType: firstNonEmpty(query.Get("source"), query.Get("event_type")),
|
||
Page: parsePositiveInt(query.Get("page")),
|
||
PageSize: pageSize,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, page)
|
||
}
|
||
|
||
func (s *queryHTTPServer) platformGrantRecords(w http.ResponseWriter, r *http.Request) {
|
||
query := r.URL.Query()
|
||
userID := parseInt64(query.Get("user_id"))
|
||
if userID <= 0 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "user_id is required"})
|
||
return
|
||
}
|
||
pageSize := parsePositiveInt(query.Get("page_size"))
|
||
if pageSize == 0 {
|
||
pageSize = parsePositiveInt(query.Get("pageSize"))
|
||
}
|
||
page, err := s.repo.QueryPlatformGrantCoinRecords(r.Context(), mysqlstorage.PlatformGrantCoinQuery{
|
||
AppCode: query.Get("app_code"),
|
||
StatTZ: query.Get("stat_tz"),
|
||
StartMS: parseInt64(query.Get("start_ms")),
|
||
EndMS: parseInt64(query.Get("end_ms")),
|
||
StatDay: query.Get("stat_day"),
|
||
CountryID: parseInt64(query.Get("country_id")),
|
||
RegionID: parseInt64(query.Get("region_id")),
|
||
UserID: userID,
|
||
EventType: firstNonEmpty(query.Get("source"), query.Get("event_type")),
|
||
Page: parsePositiveInt(query.Get("page")),
|
||
PageSize: pageSize,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, page)
|
||
}
|
||
|
||
func (s *queryHTTPServer) appFunnel(w http.ResponseWriter, r *http.Request) {
|
||
query := r.URL.Query()
|
||
startMS := parseInt64(query.Get("start_ms"))
|
||
endMS := parseInt64(query.Get("end_ms"))
|
||
if startMS <= 0 {
|
||
now := time.Now().UTC()
|
||
startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli()
|
||
}
|
||
if endMS <= startMS {
|
||
endMS = startMS + 24*time.Hour.Milliseconds()
|
||
}
|
||
funnel, err := s.repo.QueryAppTrackingFunnel(r.Context(), mysqlstorage.AppTrackingFunnelQuery{
|
||
AppCode: query.Get("app_code"),
|
||
StatTZ: query.Get("stat_tz"),
|
||
StartMS: startMS,
|
||
EndMS: endMS,
|
||
CountryID: parseInt64(query.Get("country_id")),
|
||
RegionID: parseInt64(query.Get("region_id")),
|
||
RegionIDs: parseInt64CSV(query.Get("region_ids")),
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, funnel)
|
||
}
|
||
|
||
func (s *queryHTTPServer) socialRequirements(w http.ResponseWriter, r *http.Request) {
|
||
query := r.URL.Query()
|
||
startMS := parseInt64(query.Get("start_ms"))
|
||
endMS := parseInt64(query.Get("end_ms"))
|
||
if startMS <= 0 {
|
||
now := time.Now().UTC()
|
||
startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli()
|
||
}
|
||
if endMS <= startMS {
|
||
endMS = startMS + 24*time.Hour.Milliseconds()
|
||
}
|
||
requirements, err := s.repo.QuerySocialRequirements(r.Context(), mysqlstorage.SocialRequirementsQuery{
|
||
AppCode: query.Get("app_code"),
|
||
StatTZ: query.Get("stat_tz"),
|
||
StartMS: startMS,
|
||
EndMS: endMS,
|
||
CountryID: parseInt64(query.Get("country_id")),
|
||
RegionID: parseInt64(query.Get("region_id")),
|
||
RegionIDs: parseInt64CSV(query.Get("region_ids")),
|
||
Section: query.Get("section"),
|
||
UserRole: query.Get("user_role"),
|
||
PayerType: query.Get("payer_type"),
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, requirements)
|
||
}
|
||
|
||
type selfGameH5EventRequest struct {
|
||
AppCode string `json:"app_code"`
|
||
EventID string `json:"event_id"`
|
||
EventName string `json:"event_name"`
|
||
GameID string `json:"game_id"`
|
||
Page string `json:"page"`
|
||
SessionID string `json:"session_id"`
|
||
MatchID string `json:"match_id"`
|
||
UserID int64 `json:"user_id"`
|
||
CountryID int64 `json:"country_id"`
|
||
RegionID int64 `json:"region_id"`
|
||
Language string `json:"language"`
|
||
ClientVersion string `json:"client_version"`
|
||
H5Version string `json:"h5_version"`
|
||
EntrySource string `json:"entry_source"`
|
||
DurationMS int64 `json:"duration_ms"`
|
||
Success bool `json:"success"`
|
||
ErrorCode string `json:"error_code"`
|
||
OccurredAtMS int64 `json:"occurred_at_ms"`
|
||
}
|
||
|
||
type appEventsRequest struct {
|
||
Events []appTrackingEventRequest `json:"events"`
|
||
}
|
||
|
||
type socialCoinSellerRechargeOrderRequest struct {
|
||
AppCode string `json:"app_code"`
|
||
EventID string `json:"event_id"`
|
||
OrderID int64 `json:"order_id"`
|
||
VerifyStatus string `json:"verify_status"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
USDMinor int64 `json:"usd_minor"`
|
||
USDMinorAlt int64 `json:"usd_minor_amount"`
|
||
CoinAmount int64 `json:"coin_amount"`
|
||
CountryID int64 `json:"target_country_id"`
|
||
RegionID int64 `json:"target_region_id"`
|
||
CreatedAtMS int64 `json:"created_at_ms"`
|
||
OccurredAtMS int64 `json:"occurred_at_ms"`
|
||
}
|
||
|
||
type appTrackingEventRequest struct {
|
||
AppCode string `json:"app_code"`
|
||
EventID string `json:"event_id"`
|
||
EventName string `json:"event_name"`
|
||
EventType string `json:"event_type"`
|
||
Screen string `json:"screen"`
|
||
TargetType string `json:"target_type"`
|
||
TargetID string `json:"target_id"`
|
||
UserID int64 `json:"user_id"`
|
||
DeviceID string `json:"device_id"`
|
||
SessionID string `json:"session_id"`
|
||
Platform string `json:"platform"`
|
||
AppVersion string `json:"app_version"`
|
||
Language string `json:"language"`
|
||
Timezone string `json:"timezone"`
|
||
CountryID int64 `json:"country_id"`
|
||
RegionID int64 `json:"region_id"`
|
||
DurationMS int64 `json:"duration_ms"`
|
||
Success bool `json:"success"`
|
||
ErrorCode string `json:"error_code"`
|
||
Properties json.RawMessage `json:"properties"`
|
||
OccurredAtMS int64 `json:"occurred_at_ms"`
|
||
}
|
||
|
||
func (s *queryHTTPServer) socialCoinSellerRechargeOrder(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"})
|
||
return
|
||
}
|
||
defer r.Body.Close()
|
||
var body socialCoinSellerRechargeOrderRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid json"})
|
||
return
|
||
}
|
||
app := appcode.Normalize(body.AppCode)
|
||
if app == "" {
|
||
app = "lalu"
|
||
}
|
||
eventID := strings.TrimSpace(body.EventID)
|
||
if eventID == "" && body.OrderID > 0 {
|
||
eventID = "finance_coin_seller_recharge_order:" + strconv.FormatInt(body.OrderID, 10)
|
||
}
|
||
usdMinor := body.USDMinor
|
||
if usdMinor <= 0 {
|
||
usdMinor = body.USDMinorAlt
|
||
}
|
||
if eventID == "" || body.TargetUserID <= 0 || usdMinor <= 0 || body.CreatedAtMS <= 0 || !strings.EqualFold(strings.TrimSpace(body.VerifyStatus), "verified") {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "verified event_id/order_id, target_user_id, usd_minor and created_at_ms are required"})
|
||
return
|
||
}
|
||
// 这个入口只承接 admin finance 已经校验完成的订单快照；statistics-service 不回查 admin 表，避免统计链路拥有财务状态。
|
||
err := s.repo.ConsumeFinanceCoinSellerRechargeOrder(appcode.WithContext(r.Context(), app), mysqlstorage.FinanceCoinSellerRechargeOrderEvent{
|
||
AppCode: app,
|
||
EventID: eventID,
|
||
VerifyStatus: body.VerifyStatus,
|
||
TargetUserID: body.TargetUserID,
|
||
USDMinor: usdMinor,
|
||
CoinAmount: body.CoinAmount,
|
||
CountryID: body.CountryID,
|
||
RegionID: body.RegionID,
|
||
CreatedAtMS: body.CreatedAtMS,
|
||
OccurredAtMS: body.OccurredAtMS,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
|
||
}
|
||
|
||
func (s *queryHTTPServer) appEvents(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"})
|
||
return
|
||
}
|
||
defer r.Body.Close()
|
||
var body appEventsRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid json"})
|
||
return
|
||
}
|
||
if len(body.Events) == 0 || len(body.Events) > 50 {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "events batch is invalid"})
|
||
return
|
||
}
|
||
events := make([]mysqlstorage.AppTrackingEvent, 0, len(body.Events))
|
||
for _, item := range body.Events {
|
||
// Internal ingest 只接受 gateway 已解析过的 App、用户和维度；statistics-service 只负责幂等落表，不回查 owner service。
|
||
events = append(events, mysqlstorage.AppTrackingEvent{
|
||
AppCode: appcode.Normalize(item.AppCode),
|
||
EventID: item.EventID,
|
||
EventName: item.EventName,
|
||
EventType: item.EventType,
|
||
Screen: item.Screen,
|
||
TargetType: item.TargetType,
|
||
TargetID: item.TargetID,
|
||
UserID: item.UserID,
|
||
DeviceID: item.DeviceID,
|
||
SessionID: item.SessionID,
|
||
Platform: item.Platform,
|
||
AppVersion: item.AppVersion,
|
||
Language: item.Language,
|
||
Timezone: item.Timezone,
|
||
CountryID: item.CountryID,
|
||
RegionID: item.RegionID,
|
||
DurationMS: item.DurationMS,
|
||
Success: item.Success,
|
||
ErrorCode: item.ErrorCode,
|
||
Properties: item.Properties,
|
||
OccurredAtMS: item.OccurredAtMS,
|
||
})
|
||
}
|
||
result, err := s.repo.ConsumeAppTrackingEvents(r.Context(), events)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, result)
|
||
}
|
||
|
||
func (s *queryHTTPServer) selfGameEvents(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "method not allowed"})
|
||
return
|
||
}
|
||
defer r.Body.Close()
|
||
var body selfGameH5EventRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid json"})
|
||
return
|
||
}
|
||
body.EventID = strings.TrimSpace(body.EventID)
|
||
body.EventName = strings.TrimSpace(body.EventName)
|
||
if body.EventID == "" || body.EventName == "" {
|
||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "event_id and event_name are required"})
|
||
return
|
||
}
|
||
if body.OccurredAtMS <= 0 {
|
||
body.OccurredAtMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
app := appcode.Normalize(body.AppCode)
|
||
if app == "" {
|
||
app = "lalu"
|
||
}
|
||
// 内部 ingest 只承接 gateway 已解析过的用户、App 和维度字段；statistics-service 只做幂等聚合，不反查业务库。
|
||
err := s.repo.ConsumeSelfGameH5Event(appcode.WithContext(r.Context(), app), mysqlstorage.SelfGameH5Event{
|
||
AppCode: app,
|
||
EventID: body.EventID,
|
||
EventName: body.EventName,
|
||
GameID: body.GameID,
|
||
Page: body.Page,
|
||
SessionID: body.SessionID,
|
||
MatchID: body.MatchID,
|
||
UserID: body.UserID,
|
||
CountryID: body.CountryID,
|
||
RegionID: body.RegionID,
|
||
Language: body.Language,
|
||
ClientVersion: body.ClientVersion,
|
||
H5Version: body.H5Version,
|
||
EntrySource: body.EntrySource,
|
||
DurationMS: body.DurationMS,
|
||
Success: body.Success,
|
||
ErrorCode: body.ErrorCode,
|
||
OccurredAtMS: body.OccurredAtMS,
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
|
||
}
|
||
|
||
func (s *queryHTTPServer) selfGameOverview(w http.ResponseWriter, r *http.Request) {
|
||
query := r.URL.Query()
|
||
startMS := parseInt64(query.Get("start_ms"))
|
||
endMS := parseInt64(query.Get("end_ms"))
|
||
if startMS <= 0 {
|
||
now := time.Now().UTC()
|
||
startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli()
|
||
}
|
||
if endMS <= startMS {
|
||
endMS = startMS + 24*time.Hour.Milliseconds()
|
||
}
|
||
overview, err := s.repo.QuerySelfGameOverview(r.Context(), mysqlstorage.SelfGameOverviewQuery{
|
||
AppCode: query.Get("app_code"),
|
||
StatTZ: query.Get("stat_tz"),
|
||
StartMS: startMS,
|
||
EndMS: endMS,
|
||
GameID: query.Get("game_id"),
|
||
CountryID: parseInt64(query.Get("country_id")),
|
||
RegionID: parseInt64(query.Get("region_id")),
|
||
})
|
||
if err != nil {
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, overview)
|
||
}
|
||
|
||
func parseInt64(value string) int64 {
|
||
parsed, _ := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
|
||
return parsed
|
||
}
|
||
|
||
func parsePositiveInt(value string) int {
|
||
parsed, _ := strconv.Atoi(strings.TrimSpace(value))
|
||
if parsed < 0 {
|
||
return 0
|
||
}
|
||
return parsed
|
||
}
|
||
|
||
func parseInt64CSV(value string) []int64 {
|
||
parts := strings.Split(value, ",")
|
||
out := make([]int64, 0, len(parts))
|
||
for _, part := range parts {
|
||
parsed := parseInt64(part)
|
||
if parsed > 0 {
|
||
out = append(out, parsed)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func writeJSON(w http.ResponseWriter, status int, payload any) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(status)
|
||
_ = json.NewEncoder(w).Encode(payload)
|
||
}
|