jackyu66gitandCursor
7ab9add5dd
落地埋点 ingest/数据看板、首页宫格 CMS 与测评上下架；含账号引导、问答流式与免责声明去重，以及 review P1 审计同事务修复。 Co-authored-by: Cursor <cursoragent@cursor.com>
314 lines
8.0 KiB
Go
314 lines
8.0 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// AnalyticsRepo persists behavior events and sessions.
|
|
type AnalyticsRepo struct {
|
|
Pool *pgxpool.Pool
|
|
}
|
|
|
|
// AnalyticsEventRow is one ingest event after validation.
|
|
type AnalyticsEventRow struct {
|
|
SessionID string
|
|
UserID uuid.UUID
|
|
Name string
|
|
PagePath string
|
|
Props json.RawMessage
|
|
ClientTS time.Time
|
|
}
|
|
|
|
// UpsertSession creates or refreshes a session row.
|
|
func (r *AnalyticsRepo) UpsertSession(
|
|
ctx context.Context,
|
|
sessionID, deviceKey string,
|
|
userID uuid.UUID,
|
|
startedAt time.Time,
|
|
) error {
|
|
_, err := r.Pool.Exec(ctx, `
|
|
INSERT INTO analytics_sessions(session_id, device_key, user_id, started_at)
|
|
VALUES ($1,$2,$3,$4)
|
|
ON CONFLICT (session_id) DO UPDATE SET
|
|
device_key = EXCLUDED.device_key,
|
|
user_id = COALESCE(EXCLUDED.user_id, analytics_sessions.user_id)`,
|
|
sessionID, deviceKey, userID, startedAt,
|
|
)
|
|
return err
|
|
}
|
|
|
|
// EndSession updates session end fields.
|
|
func (r *AnalyticsRepo) EndSession(
|
|
ctx context.Context,
|
|
sessionID string,
|
|
endedAt time.Time,
|
|
exitPage string,
|
|
durationMs int,
|
|
) error {
|
|
_, err := r.Pool.Exec(ctx, `
|
|
UPDATE analytics_sessions
|
|
SET ended_at=$2, exit_page=$3, duration_ms=$4
|
|
WHERE session_id=$1`,
|
|
sessionID, endedAt, emptyToNil(exitPage), durationMs,
|
|
)
|
|
return err
|
|
}
|
|
|
|
// InsertEvents bulk-inserts event rows.
|
|
func (r *AnalyticsRepo) InsertEvents(ctx context.Context, rows []AnalyticsEventRow) error {
|
|
if len(rows) == 0 {
|
|
return nil
|
|
}
|
|
for _, row := range rows {
|
|
uid := interface{}(nil)
|
|
if row.UserID != uuid.Nil {
|
|
uid = row.UserID
|
|
}
|
|
page := emptyToNil(row.PagePath)
|
|
props := row.Props
|
|
if len(props) == 0 {
|
|
props = []byte("{}")
|
|
}
|
|
if _, err := r.Pool.Exec(ctx, `
|
|
INSERT INTO analytics_events(session_id, user_id, name, page_path, props, client_ts)
|
|
VALUES ($1,$2,$3,$4,$5,$6)`,
|
|
row.SessionID, uid, row.Name, page, props, row.ClientTS,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// OverviewAgg is admin overview metrics.
|
|
type OverviewAgg struct {
|
|
DAU int `json:"dau"`
|
|
NewUsers int `json:"new_users"`
|
|
Sessions int `json:"sessions"`
|
|
AvgSessionMs float64 `json:"avg_session_ms"`
|
|
Series []DayPoint `json:"series"`
|
|
}
|
|
|
|
// DayPoint is one day in a trend series.
|
|
type DayPoint struct {
|
|
Day string `json:"day"`
|
|
DAU int `json:"dau"`
|
|
Sessions int `json:"sessions"`
|
|
}
|
|
|
|
// PageAgg is per-page metrics.
|
|
type PageAgg struct {
|
|
PagePath string `json:"page_path"`
|
|
PV int `json:"pv"`
|
|
UV int `json:"uv"`
|
|
AvgDwellMs float64 `json:"avg_dwell_ms"`
|
|
ExitCount int `json:"exit_count"`
|
|
}
|
|
|
|
// ExitAgg is exit page ranking.
|
|
type ExitAgg struct {
|
|
ExitPage string `json:"exit_page"`
|
|
Count int `json:"count"`
|
|
}
|
|
|
|
// ClickAgg is click ranking.
|
|
type ClickAgg struct {
|
|
ElementID string `json:"element_id"`
|
|
Count int `json:"count"`
|
|
}
|
|
|
|
// FunnelStep is one named funnel count.
|
|
type FunnelStep struct {
|
|
Name string `json:"name"`
|
|
Count int `json:"count"`
|
|
}
|
|
|
|
func (r *AnalyticsRepo) Overview(ctx context.Context, from, to time.Time) (*OverviewAgg, error) {
|
|
out := &OverviewAgg{}
|
|
err := r.Pool.QueryRow(ctx, `
|
|
SELECT count(DISTINCT user_id)::int
|
|
FROM analytics_events
|
|
WHERE received_at >= $1 AND received_at < $2 AND user_id IS NOT NULL`,
|
|
from, to,
|
|
).Scan(&out.DAU)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
err = r.Pool.QueryRow(ctx, `
|
|
SELECT count(*)::int FROM users
|
|
WHERE created_at >= $1 AND created_at < $2 AND deleted_at IS NULL`,
|
|
from, to,
|
|
).Scan(&out.NewUsers)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
err = r.Pool.QueryRow(ctx, `
|
|
SELECT count(*)::int,
|
|
COALESCE(avg(duration_ms) FILTER (WHERE duration_ms IS NOT NULL), 0)::float8
|
|
FROM analytics_sessions
|
|
WHERE started_at >= $1 AND started_at < $2`,
|
|
from, to,
|
|
).Scan(&out.Sessions, &out.AvgSessionMs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := r.Pool.Query(ctx, `
|
|
SELECT to_char(d, 'YYYY-MM-DD') AS day,
|
|
COALESCE((
|
|
SELECT count(DISTINCT e.user_id)::int FROM analytics_events e
|
|
WHERE e.received_at >= d AND e.received_at < d + interval '1 day'
|
|
AND e.user_id IS NOT NULL
|
|
), 0) AS dau,
|
|
COALESCE((
|
|
SELECT count(*)::int FROM analytics_sessions s
|
|
WHERE s.started_at >= d AND s.started_at < d + interval '1 day'
|
|
), 0) AS sessions
|
|
FROM generate_series($1::timestamptz, $2::timestamptz - interval '1 day', interval '1 day') AS d
|
|
ORDER BY d`,
|
|
from, to,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var p DayPoint
|
|
if err := rows.Scan(&p.Day, &p.DAU, &p.Sessions); err != nil {
|
|
return nil, err
|
|
}
|
|
out.Series = append(out.Series, p)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *AnalyticsRepo) Pages(ctx context.Context, from, to time.Time) ([]PageAgg, error) {
|
|
rows, err := r.Pool.Query(ctx, `
|
|
WITH views AS (
|
|
SELECT coalesce(nullif(page_path,''), props->>'page_path') AS path,
|
|
count(*)::int AS pv,
|
|
count(DISTINCT user_id)::int AS uv
|
|
FROM analytics_events
|
|
WHERE name='page_view' AND received_at >= $1 AND received_at < $2
|
|
GROUP BY 1
|
|
),
|
|
dwells AS (
|
|
SELECT coalesce(nullif(page_path,''), props->>'page_path') AS path,
|
|
avg(NULLIF((props->>'dwell_ms')::float8, 'NaN'))::float8 AS avg_dwell
|
|
FROM analytics_events
|
|
WHERE name='page_leave' AND received_at >= $1 AND received_at < $2
|
|
GROUP BY 1
|
|
),
|
|
exits AS (
|
|
SELECT coalesce(nullif(exit_page,''), '') AS path, count(*)::int AS n
|
|
FROM analytics_sessions
|
|
WHERE ended_at >= $1 AND ended_at < $2 AND exit_page IS NOT NULL AND exit_page <> ''
|
|
GROUP BY 1
|
|
)
|
|
SELECT coalesce(v.path, d.path, e.path) AS page_path,
|
|
coalesce(v.pv, 0), coalesce(v.uv, 0),
|
|
coalesce(d.avg_dwell, 0), coalesce(e.n, 0)
|
|
FROM views v
|
|
FULL OUTER JOIN dwells d ON v.path = d.path
|
|
FULL OUTER JOIN exits e ON coalesce(v.path, d.path) = e.path
|
|
WHERE coalesce(v.path, d.path, e.path) IS NOT NULL
|
|
AND coalesce(v.path, d.path, e.path) <> ''
|
|
ORDER BY coalesce(v.pv, 0) DESC
|
|
LIMIT 50`,
|
|
from, to,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []PageAgg
|
|
for rows.Next() {
|
|
var p PageAgg
|
|
if err := rows.Scan(&p.PagePath, &p.PV, &p.UV, &p.AvgDwellMs, &p.ExitCount); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, p)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *AnalyticsRepo) Exits(ctx context.Context, from, to time.Time) ([]ExitAgg, error) {
|
|
rows, err := r.Pool.Query(ctx, `
|
|
SELECT exit_page, count(*)::int
|
|
FROM analytics_sessions
|
|
WHERE ended_at >= $1 AND ended_at < $2
|
|
AND exit_page IS NOT NULL AND exit_page <> ''
|
|
GROUP BY exit_page
|
|
ORDER BY count(*) DESC
|
|
LIMIT 30`,
|
|
from, to,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []ExitAgg
|
|
for rows.Next() {
|
|
var e ExitAgg
|
|
if err := rows.Scan(&e.ExitPage, &e.Count); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, e)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *AnalyticsRepo) Clicks(ctx context.Context, from, to time.Time) ([]ClickAgg, error) {
|
|
rows, err := r.Pool.Query(ctx, `
|
|
SELECT coalesce(props->>'element_id', '') AS eid, count(*)::int
|
|
FROM analytics_events
|
|
WHERE name='ui_click' AND received_at >= $1 AND received_at < $2
|
|
AND coalesce(props->>'element_id','') <> ''
|
|
GROUP BY 1
|
|
ORDER BY count(*) DESC
|
|
LIMIT 30`,
|
|
from, to,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []ClickAgg
|
|
for rows.Next() {
|
|
var c ClickAgg
|
|
if err := rows.Scan(&c.ElementID, &c.Count); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *AnalyticsRepo) Funnel(ctx context.Context, from, to time.Time, names []string) ([]FunnelStep, error) {
|
|
out := make([]FunnelStep, 0, len(names))
|
|
for _, name := range names {
|
|
var n int
|
|
err := r.Pool.QueryRow(ctx, `
|
|
SELECT count(*)::int FROM analytics_events
|
|
WHERE name=$1 AND received_at >= $2 AND received_at < $3`,
|
|
name, from, to,
|
|
).Scan(&n)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, FunnelStep{Name: name, Count: n})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func emptyToNil(s string) interface{} {
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|