package store import ( "context" "encoding/json" "errors" "fmt" "time" "github.com/jackc/pgx/v5" ) type RecommendationAction struct { ItemID string `json:"itemId"` Action string `json:"action"` UpdatedAt time.Time `json:"updatedAt"` } type ItemExposureStat struct { ItemID string Impressions int Focuses int Selects int LastShown time.Time } func (s *Store) HouseholdCompletionScores( ctx context.Context, since time.Time, ) (map[string]float64, error) { rows, err := s.pool.Query(ctx, ` SELECT coalesce(nullif(emby_series_id, ''), emby_item_id) AS item_id, count(DISTINCT lower(username))::float8 FROM tracearr_sessions WHERE started_at >= $1 AND (watched OR ( total_duration_ms > 0 AND progress_ms::float8 / total_duration_ms >= 0.9 )) AND coalesce(nullif(emby_series_id, ''), emby_item_id) <> '' GROUP BY item_id`, since) if err != nil { return nil, fmt.Errorf("store: household completion scores: %w", err) } defer rows.Close() out := map[string]float64{} maxScore := 0.0 for rows.Next() { var id string var score float64 if err := rows.Scan(&id, &score); err != nil { return nil, err } out[id] = score if score > maxScore { maxScore = score } } if maxScore > 0 { for id, score := range out { out[id] = score / maxScore } } return out, rows.Err() } func (s *Store) WeightedRecommendationProfile( ctx context.Context, userID string, ) (json.RawMessage, error) { var raw []byte err := s.pool.QueryRow(ctx, ` SELECT weighted_profile FROM recommendation_user_profiles WHERE emby_user_id = $1`, userID).Scan(&raw) if errors.Is(err, pgx.ErrNoRows) { return json.RawMessage(`{}`), nil } if err != nil { return nil, fmt.Errorf("store: weighted recommendation profile: %w", err) } return json.RawMessage(raw), nil } func (s *Store) SetRecommendationAction( ctx context.Context, userID, itemID, action string, ) error { if action != "more_like_this" && action != "not_for_me" { return fmt.Errorf("store: invalid recommendation action %q", action) } _, err := s.pool.Exec(ctx, ` INSERT INTO recommendation_actions (emby_user_id, item_id, action, updated_at) VALUES ($1,$2,$3,now()) ON CONFLICT (emby_user_id, item_id) DO UPDATE SET action = EXCLUDED.action, updated_at = now()`, userID, itemID, action) if err != nil { return fmt.Errorf("store: set recommendation action: %w", err) } return nil } func (s *Store) ClearRecommendationAction( ctx context.Context, userID, itemID string, ) error { _, err := s.pool.Exec(ctx, ` DELETE FROM recommendation_actions WHERE emby_user_id = $1 AND item_id = $2`, userID, itemID) return err } func (s *Store) RecommendationActions( ctx context.Context, userID string, ) ([]RecommendationAction, error) { rows, err := s.pool.Query(ctx, ` SELECT item_id, action, updated_at FROM recommendation_actions WHERE emby_user_id = $1`, userID) if err != nil { return nil, fmt.Errorf("store: recommendation actions: %w", err) } defer rows.Close() out := []RecommendationAction{} for rows.Next() { var value RecommendationAction if err := rows.Scan(&value.ItemID, &value.Action, &value.UpdatedAt); err != nil { return nil, err } out = append(out, value) } return out, rows.Err() } func (s *Store) RecommendationOnboarding( ctx context.Context, userID string, ) (json.RawMessage, error) { var raw []byte err := s.pool.QueryRow(ctx, ` SELECT preferences FROM recommendation_onboarding WHERE emby_user_id = $1`, userID).Scan(&raw) if errors.Is(err, pgx.ErrNoRows) { return json.RawMessage(`{}`), nil } return json.RawMessage(raw), err } func (s *Store) SetRecommendationOnboarding( ctx context.Context, userID string, preferences json.RawMessage, ) error { _, err := s.pool.Exec(ctx, ` INSERT INTO recommendation_onboarding (emby_user_id, preferences, updated_at) VALUES ($1,$2::jsonb,now()) ON CONFLICT (emby_user_id) DO UPDATE SET preferences = EXCLUDED.preferences, updated_at = now()`, userID, string(preferences)) return err } // UserItemExposures is deliberately item-scoped. A row impression with no item id is // useful for row ordering but cannot be used to claim a particular poster was ignored. func (s *Store) UserItemExposures( ctx context.Context, userID string, since time.Time, ) ([]ItemExposureStat, error) { rows, err := s.pool.Query(ctx, ` SELECT item_id, count(*) FILTER (WHERE event = 'impression'), count(*) FILTER (WHERE event = 'focus'), count(*) FILTER (WHERE event = 'select'), max(occurred_at) FROM row_events WHERE emby_user_id = $1 AND occurred_at >= $2 AND item_id <> '' GROUP BY item_id`, userID, since) if err != nil { return nil, fmt.Errorf("store: user item exposures: %w", err) } defer rows.Close() out := []ItemExposureStat{} for rows.Next() { var value ItemExposureStat if err := rows.Scan( &value.ItemID, &value.Impressions, &value.Focuses, &value.Selects, &value.LastShown, ); err != nil { return nil, err } out = append(out, value) } return out, rows.Err() }