feat(ops): ECR-007 行为分析与 ECR-008 内容运营后台
落地埋点 ingest/数据看板、首页宫格 CMS 与测评上下架；含账号引导、问答流式与免责声明去重，以及 review P1 审计同事务修复。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,313 @@
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user