366 lines
12 KiB
Go
366 lines
12 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// BrowsingCandidates returns library items the user actively focused or selected,
|
|
// strongest first. Impressions are intentionally excluded: merely scrolling past a row
|
|
// is not evidence of taste.
|
|
func (s *Store) BrowsingCandidates(
|
|
ctx context.Context,
|
|
userID string,
|
|
since time.Time,
|
|
limit int,
|
|
) ([]json.RawMessage, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT li.payload
|
|
FROM row_events re
|
|
JOIN library_items li ON li.id = re.item_id
|
|
WHERE re.emby_user_id = $1
|
|
AND re.occurred_at >= $2
|
|
AND re.event IN ('focus', 'select')
|
|
GROUP BY li.id, li.payload
|
|
ORDER BY
|
|
count(*) FILTER (WHERE re.event = 'select') * 20 +
|
|
count(*) FILTER (WHERE re.event = 'focus') * 2 +
|
|
coalesce(sum(re.dwell_ms), 0) / 10000 DESC,
|
|
max(re.occurred_at) DESC
|
|
LIMIT $3`, userID, since, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: browsing candidates: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := []json.RawMessage{}
|
|
for rows.Next() {
|
|
var payload []byte
|
|
if err := rows.Scan(&payload); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, json.RawMessage(payload))
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// RowEvent is one reported interaction with a home-screen row.
|
|
type RowEvent struct {
|
|
OccurredAt time.Time
|
|
UserID string
|
|
RowID string
|
|
RowKind string
|
|
Event string
|
|
ItemID string
|
|
DwellMs int
|
|
}
|
|
|
|
// JourneyEvent is one significant step through the app. All descriptive fields are
|
|
// controlled vocabulary; ItemID is the only content identity retained.
|
|
type JourneyEvent struct {
|
|
ID int64 `json:"id"`
|
|
OccurredAt time.Time `json:"occurredAt"`
|
|
UserID string `json:"userId"`
|
|
JourneyID string `json:"journeyId"`
|
|
Sequence int `json:"sequence"`
|
|
Category string `json:"category"`
|
|
Action string `json:"action"`
|
|
Screen string `json:"screen"`
|
|
Feature string `json:"feature"`
|
|
Source string `json:"source"`
|
|
Target string `json:"target"`
|
|
ItemID string `json:"itemId,omitempty"`
|
|
ItemType string `json:"itemType,omitempty"`
|
|
Outcome string `json:"outcome,omitempty"`
|
|
}
|
|
|
|
type AnalyticsUser struct {
|
|
UserID string `json:"userId"`
|
|
Username string `json:"username"`
|
|
Events int64 `json:"events"`
|
|
Journeys int64 `json:"journeys"`
|
|
LastActiveAt time.Time `json:"lastActiveAt"`
|
|
}
|
|
|
|
type FeatureStat struct {
|
|
Feature string `json:"feature"`
|
|
Uses int64 `json:"uses"`
|
|
LastUsedAt time.Time `json:"lastUsedAt"`
|
|
}
|
|
|
|
type PathStat struct {
|
|
From string `json:"from"`
|
|
To string `json:"to"`
|
|
Count int64 `json:"count"`
|
|
}
|
|
|
|
// Event kinds. Impressions say a row was drawn; focus says the remote actually landed
|
|
// on it and for how long; select says something was opened from it.
|
|
const (
|
|
RowEventImpression = "impression"
|
|
RowEventFocus = "focus"
|
|
RowEventSelect = "select"
|
|
)
|
|
|
|
// RowStat is the aggregate the admin page renders.
|
|
type RowStat struct {
|
|
RowID string `json:"rowId"`
|
|
RowKind string `json:"rowKind"`
|
|
Impressions int64 `json:"impressions"`
|
|
Focuses int64 `json:"focuses"`
|
|
Selects int64 `json:"selects"`
|
|
DwellMs int64 `json:"dwellMs"`
|
|
Viewers int64 `json:"viewers"`
|
|
SelectRate float64 `json:"selectRate"`
|
|
}
|
|
|
|
// UserRowStats is the per-profile counterpart to the admin aggregate. It gives the
|
|
// home composer enough evidence to gently demote shelves that a viewer repeatedly
|
|
// passes over without turning a couple of accidental focus moves into a preference.
|
|
func (s *Store) UserRowStats(
|
|
ctx context.Context,
|
|
userID string,
|
|
since time.Time,
|
|
) ([]RowStat, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT row_id,
|
|
(array_agg(row_kind ORDER BY occurred_at DESC))[1] AS row_kind,
|
|
count(*) FILTER (WHERE event = 'impression') AS impressions,
|
|
count(*) FILTER (WHERE event = 'focus') AS focuses,
|
|
count(*) FILTER (WHERE event = 'select') AS selects,
|
|
coalesce(sum(dwell_ms), 0) AS dwell_ms
|
|
FROM row_events
|
|
WHERE emby_user_id = $1 AND occurred_at >= $2
|
|
GROUP BY row_id`, userID, since)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: user row stats: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
stats := []RowStat{}
|
|
for rows.Next() {
|
|
var stat RowStat
|
|
if err := rows.Scan(
|
|
&stat.RowID, &stat.RowKind, &stat.Impressions, &stat.Focuses,
|
|
&stat.Selects, &stat.DwellMs,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
stat.Viewers = 1
|
|
if stat.Impressions > 0 {
|
|
stat.SelectRate = float64(stat.Selects) / float64(stat.Impressions)
|
|
}
|
|
stats = append(stats, stat)
|
|
}
|
|
return stats, rows.Err()
|
|
}
|
|
|
|
func (s *Store) InsertRowEvents(ctx context.Context, events []RowEvent) error {
|
|
if len(events) == 0 {
|
|
return nil
|
|
}
|
|
batch := &pgx.Batch{}
|
|
for _, event := range events {
|
|
batch.Queue(`
|
|
INSERT INTO row_events (occurred_at, emby_user_id, row_id, row_kind, event, item_id, dwell_ms)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7)`,
|
|
event.OccurredAt, event.UserID, event.RowID, event.RowKind,
|
|
event.Event, event.ItemID, event.DwellMs)
|
|
}
|
|
|
|
results := s.pool.SendBatch(ctx, batch)
|
|
defer results.Close()
|
|
for range events {
|
|
if _, err := results.Exec(); err != nil {
|
|
return fmt.Errorf("store: insert row events: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) InsertJourneyEvents(ctx context.Context, events []JourneyEvent) error {
|
|
if len(events) == 0 {
|
|
return nil
|
|
}
|
|
batch := &pgx.Batch{}
|
|
for _, event := range events {
|
|
batch.Queue(`
|
|
INSERT INTO journey_events
|
|
(occurred_at, emby_user_id, journey_id, sequence, category, action, screen,
|
|
feature, source, target, item_id, item_type, outcome)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
|
|
ON CONFLICT (emby_user_id, journey_id, sequence) DO NOTHING`,
|
|
event.OccurredAt, event.UserID, event.JourneyID, event.Sequence,
|
|
event.Category, event.Action, event.Screen, event.Feature, event.Source,
|
|
event.Target, event.ItemID, event.ItemType, event.Outcome)
|
|
}
|
|
results := s.pool.SendBatch(ctx, batch)
|
|
defer results.Close()
|
|
for range events {
|
|
if _, err := results.Exec(); err != nil {
|
|
return fmt.Errorf("store: insert journey events: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) AnalyticsUsers(ctx context.Context, since time.Time) ([]AnalyticsUser, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT je.emby_user_id,
|
|
coalesce((array_agg(s.username ORDER BY s.last_seen_at DESC)
|
|
FILTER (WHERE s.username IS NOT NULL))[1], ''),
|
|
count(DISTINCT je.id), count(DISTINCT je.journey_id), max(je.occurred_at)
|
|
FROM journey_events je
|
|
LEFT JOIN sessions s ON s.emby_user_id = je.emby_user_id
|
|
WHERE je.occurred_at >= $1
|
|
GROUP BY je.emby_user_id ORDER BY max(je.occurred_at) DESC`, since)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: analytics users: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := []AnalyticsUser{}
|
|
for rows.Next() {
|
|
var value AnalyticsUser
|
|
if err := rows.Scan(&value.UserID, &value.Username, &value.Events, &value.Journeys, &value.LastActiveAt); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, value)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) UserFeatureStats(ctx context.Context, userID string, since time.Time) ([]FeatureStat, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT feature, count(*), max(occurred_at) FROM journey_events
|
|
WHERE emby_user_id=$1 AND occurred_at >= $2 AND feature <> ''
|
|
AND action NOT IN ('screen_view', 'journey_start', 'journey_end')
|
|
GROUP BY feature ORDER BY count(*) DESC, feature`, userID, since)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: user feature stats: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := []FeatureStat{}
|
|
for rows.Next() {
|
|
var v FeatureStat
|
|
if err := rows.Scan(&v.Feature, &v.Uses, &v.LastUsedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, v)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) UserPaths(ctx context.Context, userID string, since time.Time) ([]PathStat, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
WITH ordered AS (
|
|
SELECT id, journey_id, sequence, action, occurred_at,
|
|
coalesce(nullif(target,''), nullif(screen,''), feature) AS node,
|
|
lag(coalesce(nullif(target,''), nullif(screen,''), feature)) OVER
|
|
(PARTITION BY journey_id ORDER BY sequence, occurred_at, id) AS previous
|
|
FROM journey_events WHERE emby_user_id=$1 AND occurred_at >= $2
|
|
), path_steps AS (
|
|
SELECT previous AS from_node, node AS to_node FROM ordered
|
|
WHERE previous IS NOT NULL AND node IS NOT NULL AND previous <> node
|
|
), last_steps AS (
|
|
SELECT DISTINCT ON (journey_id) journey_id, node, action, occurred_at
|
|
FROM ordered ORDER BY journey_id, sequence DESC, occurred_at DESC, id DESC
|
|
), all_steps AS (
|
|
SELECT from_node, to_node FROM path_steps
|
|
UNION ALL
|
|
SELECT node, 'abandoned' FROM last_steps
|
|
WHERE action <> 'journey_end' AND node <> ''
|
|
AND occurred_at < now() - interval '30 minutes'
|
|
)
|
|
SELECT from_node, to_node, count(*) FROM all_steps
|
|
GROUP BY from_node, to_node ORDER BY count(*) DESC, from_node, to_node LIMIT 20`, userID, since)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: user paths: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := []PathStat{}
|
|
for rows.Next() {
|
|
var v PathStat
|
|
if err := rows.Scan(&v.From, &v.To, &v.Count); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, v)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) UserJourneyEvents(ctx context.Context, userID string, since time.Time, limit int) ([]JourneyEvent, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT id, occurred_at, emby_user_id, journey_id, sequence, category, action,
|
|
screen, feature, source, target, item_id, item_type, outcome
|
|
FROM journey_events WHERE emby_user_id=$1 AND occurred_at >= $2
|
|
ORDER BY occurred_at DESC, sequence DESC LIMIT $3`, userID, since, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: user journey events: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := []JourneyEvent{}
|
|
for rows.Next() {
|
|
var v JourneyEvent
|
|
if err := rows.Scan(&v.ID, &v.OccurredAt, &v.UserID, &v.JourneyID, &v.Sequence, &v.Category, &v.Action, &v.Screen, &v.Feature, &v.Source, &v.Target, &v.ItemID, &v.ItemType, &v.Outcome); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, v)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// RowStats aggregates engagement since a point in time, busiest row first.
|
|
//
|
|
// Dwell is the interesting number: impressions only say a row was on screen, whereas
|
|
// dwell says someone actually stopped there.
|
|
func (s *Store) RowStats(ctx context.Context, since time.Time) ([]RowStat, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT row_id,
|
|
(array_agg(row_kind ORDER BY occurred_at DESC))[1] AS row_kind,
|
|
count(*) FILTER (WHERE event = 'impression') AS impressions,
|
|
count(*) FILTER (WHERE event = 'focus') AS focuses,
|
|
count(*) FILTER (WHERE event = 'select') AS selects,
|
|
coalesce(sum(dwell_ms), 0) AS dwell_ms,
|
|
count(DISTINCT emby_user_id) AS viewers
|
|
FROM row_events
|
|
WHERE occurred_at >= $1
|
|
GROUP BY row_id
|
|
ORDER BY dwell_ms DESC, impressions DESC`, since)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("store: row stats: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
stats := []RowStat{}
|
|
for rows.Next() {
|
|
var stat RowStat
|
|
if err := rows.Scan(&stat.RowID, &stat.RowKind, &stat.Impressions, &stat.Focuses,
|
|
&stat.Selects, &stat.DwellMs, &stat.Viewers); err != nil {
|
|
return nil, err
|
|
}
|
|
if stat.Impressions > 0 {
|
|
stat.SelectRate = float64(stat.Selects) / float64(stat.Impressions)
|
|
}
|
|
stats = append(stats, stat)
|
|
}
|
|
return stats, rows.Err()
|
|
}
|
|
|
|
// PruneRowEvents drops raw events past their retention window. Aggregates are computed
|
|
// at read time, so nothing is preserved once the events go — which is the point: this is
|
|
// engagement telemetry for tuning rows, not a permanent record of what people watched.
|
|
func (s *Store) PruneRowEvents(ctx context.Context, olderThan time.Duration) (int64, error) {
|
|
interval := fmt.Sprintf("%d seconds", int64(olderThan.Seconds()))
|
|
rows, err := s.pool.Exec(ctx, `DELETE FROM row_events WHERE occurred_at < now() - $1::interval`, interval)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
journeys, err := s.pool.Exec(ctx, `DELETE FROM journey_events WHERE occurred_at < now() - $1::interval`, interval)
|
|
if err != nil {
|
|
return rows.RowsAffected(), err
|
|
}
|
|
return rows.RowsAffected() + journeys.RowsAffected(), nil
|
|
}
|