digital-psychology/apps/api/internal/service/analytics/service.go

package analytics

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/yuxingu/digital-psychology/apps/api/internal/repository"
)

var (
	ErrTooManyItems = errors.New("too many items")
	ErrInvalidBatch = errors.New("invalid batch")
)

const maxBatch = 100

// Service handles ingest validation and admin aggregates.
type Service struct {
	Repo *repository.AnalyticsRepo
}

// EventIn is one client event.
type EventIn struct {
	Name      string                 `json:"name"`
	SessionID string                 `json:"session_id"`
	PagePath  string                 `json:"page_path"`
	ClientTS  any                    `json:"client_ts"`
	Props     map[string]interface{} `json:"props"`
}

// IngestResult summarizes a successful write.
type IngestResult struct {
	Accepted int `json:"accepted"`
}

var allowedNames = map[string]struct{}{
	"session_start": {}, "session_end": {}, "page_view": {}, "page_leave": {}, "ui_click": {},
	"home_cta_portrait": {}, "portrait_completed": {}, "deep_access_clicked": {},
	"purchase_completed": {}, "relation_completed": {}, "synastry_completed": {},
	"synastry_invite_created": {}, "synastry_invite_accepted": {}, "synastry_nearby_opened": {},
	"star_wheel_viewed": {}, "companion_viewed": {}, "mood_saved": {},
	"cards_scene_selected": {}, "cards_drawn": {}, "cards_quota_exhausted": {},
	"star_completed": {}, "rhythm_completed": {},
	"auth_register": {}, "auth_login": {}, "auth_logout": {},
	"avatar_sheet_opened": {}, "avatar_upload_succeeded": {}, "avatar_upload_failed": {},
	"nickname_updated": {},
	"scale_list_viewed": {}, "scale_bank_category_viewed": {},
	"scale_started": {}, "scale_completed": {}, "scale_result_reopened": {},
	"scale_retake_clicked": {}, "scale_locked_viewed": {}, "scale_share_clicked": {},
	"scale_cta_relation": {}, "scale_cta_ask": {},
	"growth_plan_viewed": {}, "growth_plan_created": {}, "growth_plan_checkin": {},
}

var allowedPropKeys = map[string]struct{}{
	"page_path": {}, "page_title": {}, "referrer_path": {}, "dwell_ms": {},
	"element_id": {}, "exit_page": {}, "duration_ms": {}, "cold": {}, "app_ver": {},
	"source": {}, "kind": {}, "surface": {}, "label": {}, "plan": {}, "count": {},
	"depth": {}, "scene": {}, "score": {}, "planet": {}, "report_id": {},
	"slug": {}, "category": {}, "reason": {}, "access": {},
}

var funnelDefault = []string{
	"scale_list_viewed", "scale_started", "scale_completed",
	"portrait_completed", "deep_access_clicked", "purchase_completed",
}

// Ingest validates and persists a batch. Invalid batch → ErrInvalidBatch (整批 400).
func (s *Service) Ingest(
	ctx context.Context,
	userID uuid.UUID,
	deviceKey string,
	items []EventIn,
) (*IngestResult, error) {
	if len(items) == 0 {
		return nil, ErrInvalidBatch
	}
	if len(items) > maxBatch {
		return nil, ErrTooManyItems
	}
	rows := make([]repository.AnalyticsEventRow, 0, len(items))
	for i := range items {
		row, end, err := normalizeItem(&items[i])
		if err != nil {
			return nil, ErrInvalidBatch
		}
		if err := s.Repo.UpsertSession(ctx, row.SessionID, deviceKey, userID, row.ClientTS); err != nil {
			return nil, err
		}
		if end != nil {
			if err := s.Repo.EndSession(ctx, row.SessionID, row.ClientTS, end.ExitPage, end.DurationMs); err != nil {
				return nil, err
			}
		}
		row.UserID = userID
		rows = append(rows, *row)
	}
	if err := s.Repo.InsertEvents(ctx, rows); err != nil {
		return nil, err
	}
	return &IngestResult{Accepted: len(rows)}, nil
}

type sessionEnd struct {
	ExitPage   string
	DurationMs int
}

func normalizeItem(in *EventIn) (*repository.AnalyticsEventRow, *sessionEnd, error) {
	name := strings.TrimSpace(in.Name)
	sid := strings.TrimSpace(in.SessionID)
	if name == "" || sid == "" || len(sid) > 64 {
		return nil, nil, ErrInvalidBatch
	}
	if _, ok := allowedNames[name]; !ok {
		return nil, nil, ErrInvalidBatch
	}
	ts, err := parseClientTS(in.ClientTS)
	if err != nil {
		return nil, nil, ErrInvalidBatch
	}
	page := strings.TrimSpace(in.PagePath)
	if page == "" && in.Props != nil {
		if v, ok := in.Props["page_path"].(string); ok {
			page = strings.TrimSpace(v)
		}
	}
	props, end, err := scrubProps(name, in.Props)
	if err != nil {
		return nil, nil, err
	}
	raw, err := json.Marshal(props)
	if err != nil {
		return nil, nil, ErrInvalidBatch
	}
	return &repository.AnalyticsEventRow{
		SessionID: sid,
		Name:      name,
		PagePath:  page,
		Props:     raw,
		ClientTS:  ts,
	}, end, nil
}

func scrubProps(name string, in map[string]interface{}) (map[string]interface{}, *sessionEnd, error) {
	out := map[string]interface{}{}
	var end *sessionEnd
	if name == "session_end" {
		end = &sessionEnd{}
	}
	for k, v := range in {
		key := strings.TrimSpace(k)
		if _, ok := allowedPropKeys[key]; !ok {
			continue
		}
		if key == "dwell_ms" || key == "duration_ms" {
			n, ok := asNonNegInt(v)
			if !ok {
				return nil, nil, ErrInvalidBatch
			}
			out[key] = n
			if key == "duration_ms" && end != nil {
				end.DurationMs = n
			}
			continue
		}
		if key == "exit_page" {
			s, _ := v.(string)
			s = strings.TrimSpace(s)
			out[key] = s
			if end != nil {
				end.ExitPage = s
			}
			continue
		}
		switch t := v.(type) {
		case string:
			if len(t) > 256 {
				t = t[:256]
			}
			out[key] = t
		case float64:
			out[key] = t
		case bool:
			out[key] = t
		case int:
			out[key] = t
		}
	}
	return out, end, nil
}

func asNonNegInt(v interface{}) (int, bool) {
	switch t := v.(type) {
	case float64:
		if t < 0 || t > 1e9 {
			return 0, false
		}
		return int(t), true
	case int:
		if t < 0 {
			return 0, false
		}
		return t, true
	case string:
		n, err := strconv.Atoi(t)
		if err != nil || n < 0 {
			return 0, false
		}
		return n, true
	default:
		return 0, false
	}
}

func parseClientTS(v any) (time.Time, error) {
	switch t := v.(type) {
	case string:
		ts, err := time.Parse(time.RFC3339Nano, t)
		if err != nil {
			ts, err = time.Parse(time.RFC3339, t)
		}
		if err != nil {
			return time.Time{}, err
		}
		return ts.UTC(), nil
	case float64:
		if t > 1e12 {
			return time.UnixMilli(int64(t)).UTC(), nil
		}
		return time.Unix(int64(t), 0).UTC(), nil
	case nil:
		return time.Now().UTC(), nil
	default:
		return time.Time{}, ErrInvalidBatch
	}
}

// ParseDayRange parses from/to (YYYY-MM-DD inclusive from, exclusive to+1day).
func ParseDayRange(fromStr, toStr string) (time.Time, time.Time, error) {
	if fromStr == "" || toStr == "" {
		to := time.Now().UTC().Truncate(24 * time.Hour).Add(24 * time.Hour)
		from := to.Add(-7 * 24 * time.Hour)
		return from, to, nil
	}
	from, err := time.ParseInLocation("2006-01-02", fromStr, time.UTC)
	if err != nil {
		return time.Time{}, time.Time{}, err
	}
	toDay, err := time.ParseInLocation("2006-01-02", toStr, time.UTC)
	if err != nil {
		return time.Time{}, time.Time{}, err
	}
	to := toDay.Add(24 * time.Hour)
	if !to.After(from) || to.Sub(from) > 93*24*time.Hour {
		return time.Time{}, time.Time{}, ErrInvalidBatch
	}
	return from, to, nil
}

func (s *Service) Overview(ctx context.Context, from, to time.Time) (*repository.OverviewAgg, error) {
	return s.Repo.Overview(ctx, from, to)
}

func (s *Service) Pages(ctx context.Context, from, to time.Time) ([]repository.PageAgg, error) {
	return s.Repo.Pages(ctx, from, to)
}

func (s *Service) Exits(ctx context.Context, from, to time.Time) ([]repository.ExitAgg, error) {
	return s.Repo.Exits(ctx, from, to)
}

func (s *Service) Clicks(ctx context.Context, from, to time.Time) ([]repository.ClickAgg, error) {
	return s.Repo.Clicks(ctx, from, to)
}

func (s *Service) Funnel(ctx context.Context, from, to time.Time) ([]repository.FunnelStep, error) {
	return s.Repo.Funnel(ctx, from, to, funnelDefault)
}