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": {}, } 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": {}, } var funnelDefault = []string{ "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) }