memby/server/internal/store/analytics.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
}

// 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
}

// 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) {
	tag, err := s.pool.Exec(ctx,
		`DELETE FROM row_events WHERE occurred_at < now() - $1::interval`,
		fmt.Sprintf("%d seconds", int64(olderThan.Seconds())))
	if err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}