silo-server/internal/notifications/interest_hooks.go

package notifications

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/Silo-Server/silo-server/internal/userstore"
)

// WrapUserStoreProvider decorates the shared user-store provider so every
// favorites, watchlist, watch-progress, and watch-history mutation —
// regardless of which path performed it (REST handlers, jellycompat, history
// imports, playback stop, watch sync) — queues an interest recompute. Hooking
// the lowest shared layer keeps the seven-plus mutation call sites hook-free
// and drift-free.
//
// Progress writes queue only on state *transitions* (a row appearing, the
// in-progress flag flipping, completion crossing, rows being cleared):
// progress sync ticks fire continuously during playback on a busy server, and
// recomputing interest on every tick would be a pointless hot write path.
func WrapUserStoreProvider(inner userstore.UserStoreProvider, system *System) userstore.UserStoreProvider {
	if inner == nil || system == nil {
		return inner
	}
	return &interestTrackingProvider{inner: inner, system: system}
}

type interestTrackingProvider struct {
	inner  userstore.UserStoreProvider
	system *System
}

func (p *interestTrackingProvider) ForUser(ctx context.Context, userID int) (userstore.UserStore, error) {
	store, err := p.inner.ForUser(ctx, userID)
	if err != nil || store == nil {
		return store, err
	}
	tracked := &interestTrackingStore{UserStore: store, userID: userID, system: p.system, updater: p.system.Interest}
	// Preserve the DeviceRegistry interface upgrade some callers probe for.
	if registry, ok := store.(userstore.DeviceRegistry); ok {
		return &interestTrackingStoreWithDevices{
			interestTrackingStore: tracked,
			DeviceRegistry:        registry,
		}, nil
	}
	return tracked, nil
}

func (p *interestTrackingProvider) Close() error {
	return p.inner.Close()
}

type interestTrackingStore struct {
	userstore.UserStore
	userID  int
	system  *System
	updater *InterestUpdater
}

type interestTrackingStoreWithDevices struct {
	*interestTrackingStore
	userstore.DeviceRegistry
}

var _ userstore.SettingValueCompareAndSetter = (*interestTrackingStore)(nil)
var _ userstore.SettingMutationTransactioner = (*interestTrackingStore)(nil)
var _ userstore.SettingValueCompareAndSetter = (*interestTrackingStoreWithDevices)(nil)
var _ userstore.SettingMutationTransactioner = (*interestTrackingStoreWithDevices)(nil)

// WithPreferenceSettingsTransaction preserves the optional atomic-settings
// capability of the wrapped store. Preference writes do not affect interest
// signals, so the transaction can pass through unchanged; keeping the method
// on the decorator is what lets settings handlers reach the real backend's
// transaction boundary in production.
func (s *interestTrackingStore) WithPreferenceSettingsTransaction(
	ctx context.Context,
	fn func(userstore.PreferenceSettingsWriter) error,
) error {
	transactioner, ok := s.UserStore.(userstore.PreferenceSettingsTransactioner)
	if !ok {
		return fmt.Errorf("wrapped user store does not support atomic preference settings synchronization")
	}
	return transactioner.WithPreferenceSettingsTransaction(ctx, fn)
}

// CompareAndSetSettingValue preserves the semantic-document CAS capability of
// the wrapped store. Settings writes do not affect notification interests, so
// this decorator must not intercept or downgrade the backend primitive.
func (s *interestTrackingStore) CompareAndSetSettingValue(
	ctx context.Context,
	id userstore.SettingIdentity,
	value json.RawMessage,
	expectedRevision int64,
) (*userstore.SettingValue, error) {
	cas, ok := s.UserStore.(userstore.SettingValueCompareAndSetter)
	if !ok {
		return nil, fmt.Errorf("wrapped user store does not support atomic setting updates")
	}
	return cas.CompareAndSetSettingValue(ctx, id, value, expectedRevision)
}

// WithSettingMutationTransaction preserves the durable setting+receipt
// transaction used by mutation IDs. Passing the transaction-scoped writer
// through unchanged keeps both operations on the concrete backend transaction.
func (s *interestTrackingStore) WithSettingMutationTransaction(
	ctx context.Context,
	mutationID string,
	fn func(userstore.SettingMutationWriter) error,
) error {
	transactioner, ok := s.UserStore.(userstore.SettingMutationTransactioner)
	if !ok {
		return fmt.Errorf("wrapped user store does not support atomic idempotent setting mutations")
	}
	return transactioner.WithSettingMutationTransaction(ctx, mutationID, fn)
}

// progressState is the transition-relevant projection of a progress row.
type progressState struct {
	exists     bool
	inProgress bool
	completed  bool
}

func (s *interestTrackingStore) currentProgressState(ctx context.Context, profileID, mediaItemID string) progressState {
	entry, err := s.GetProgress(ctx, profileID, mediaItemID)
	if err != nil || entry == nil {
		return progressState{}
	}
	return progressState{
		exists:     true,
		inProgress: !entry.Completed && entry.PositionSeconds > 0,
		completed:  entry.Completed,
	}
}

func progressStateFromValues(position, duration float64, thresholds userstore.ProgressThresholds) progressState {
	completed := duration > 0 && position/duration > userstore.WatchedFraction(thresholds.WatchedPct)
	return progressState{
		exists:     true,
		inProgress: !completed && position > 0,
		completed:  completed,
	}
}

func (s *interestTrackingStore) queueOnTransition(profileID, mediaItemID string, before, after progressState) {
	if before != after {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
}

// --- Favorites & watchlist: every mutation queues (user-action frequency).

func (s *interestTrackingStore) AddFavorite(ctx context.Context, profileID, mediaItemID string) error {
	err := s.UserStore.AddFavorite(ctx, profileID, mediaItemID)
	if err == nil {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
	return err
}

func (s *interestTrackingStore) AddFavoriteAt(ctx context.Context, profileID, mediaItemID string, addedAt time.Time) (bool, error) {
	inserted, err := s.UserStore.AddFavoriteAt(ctx, profileID, mediaItemID, addedAt)
	if err == nil && inserted {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
	return inserted, err
}

func (s *interestTrackingStore) RemoveFavorite(ctx context.Context, profileID, mediaItemID string) error {
	err := s.UserStore.RemoveFavorite(ctx, profileID, mediaItemID)
	if err == nil {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
	return err
}

func (s *interestTrackingStore) AddToWatchlist(ctx context.Context, profileID, mediaItemID string) error {
	err := s.UserStore.AddToWatchlist(ctx, profileID, mediaItemID)
	if err == nil {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
	return err
}

func (s *interestTrackingStore) RemoveFromWatchlist(ctx context.Context, profileID, mediaItemID string) error {
	err := s.UserStore.RemoveFromWatchlist(ctx, profileID, mediaItemID)
	if err == nil {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
	return err
}

// --- Progress: queue on transitions only.

func (s *interestTrackingStore) UpdateProgress(ctx context.Context, profileID, mediaItemID string, position, duration float64, thresholds userstore.ProgressThresholds) error {
	before := s.currentProgressState(ctx, profileID, mediaItemID)
	err := s.UserStore.UpdateProgress(ctx, profileID, mediaItemID, position, duration, thresholds)
	if err == nil {
		s.queueOnTransition(profileID, mediaItemID, before, progressStateFromValues(position, duration, thresholds))
	}
	return err
}

func (s *interestTrackingStore) SetProgress(ctx context.Context, profileID, mediaItemID string, position, duration float64, thresholds userstore.ProgressThresholds) error {
	before := s.currentProgressState(ctx, profileID, mediaItemID)
	err := s.UserStore.SetProgress(ctx, profileID, mediaItemID, position, duration, thresholds)
	if err == nil {
		s.queueOnTransition(profileID, mediaItemID, before, progressStateFromValues(position, duration, thresholds))
	}
	return err
}

func (s *interestTrackingStore) SetProgressAt(ctx context.Context, profileID, mediaItemID string, position, duration float64, completed bool, updatedAt time.Time) error {
	before := s.currentProgressState(ctx, profileID, mediaItemID)
	err := s.UserStore.SetProgressAt(ctx, profileID, mediaItemID, position, duration, completed, updatedAt)
	if err == nil {
		after := progressState{exists: true, inProgress: !completed && position > 0, completed: completed}
		s.queueOnTransition(profileID, mediaItemID, before, after)
	}
	return err
}

func (s *interestTrackingStore) SetProgressIfNewer(ctx context.Context, profileID, mediaItemID string, position, duration float64, completed bool, updatedAt time.Time) (bool, error) {
	before := s.currentProgressState(ctx, profileID, mediaItemID)
	applied, err := s.UserStore.SetProgressIfNewer(ctx, profileID, mediaItemID, position, duration, completed, updatedAt)
	if err == nil && applied {
		after := progressState{exists: true, inProgress: !completed && position > 0, completed: completed}
		s.queueOnTransition(profileID, mediaItemID, before, after)
	}
	return applied, err
}

func (s *interestTrackingStore) MarkWatched(ctx context.Context, profileID, mediaItemID string, duration float64) error {
	before := s.currentProgressState(ctx, profileID, mediaItemID)
	err := s.UserStore.MarkWatched(ctx, profileID, mediaItemID, duration)
	if err == nil {
		s.queueOnTransition(profileID, mediaItemID, before, progressState{exists: true, completed: true})
	}
	return err
}

func (s *interestTrackingStore) MarkProgressBatch(ctx context.Context, profileID string, mediaItemIDs []string, updatedAt time.Time) error {
	beforeStates, _ := s.ListProgressByMediaItems(ctx, profileID, mediaItemIDs)
	err := s.UserStore.MarkProgressBatch(ctx, profileID, mediaItemIDs, updatedAt)
	if err == nil {
		for _, mediaItemID := range mediaItemIDs {
			if entry, ok := beforeStates[mediaItemID]; ok && entry.Completed {
				continue // already completed: no transition
			}
			s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
		}
	}
	return err
}

func (s *interestTrackingStore) ClearProgress(ctx context.Context, profileID, mediaItemID string) error {
	err := s.UserStore.ClearProgress(ctx, profileID, mediaItemID)
	if err == nil {
		s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
	}
	return err
}

func (s *interestTrackingStore) ClearProgressBatch(ctx context.Context, profileID string, mediaItemIDs []string, updatedAt time.Time) error {
	err := s.UserStore.ClearProgressBatch(ctx, profileID, mediaItemIDs, updatedAt)
	if err == nil {
		for _, mediaItemID := range mediaItemIDs {
			s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
		}
	}
	return err
}

// --- Watch history: history imports and watch-provider syncs may record a
// completed watch without any progress write, so the progress hooks alone
// would never see them. AddHistory (the live playback path) is deliberately
// not hooked: playback always writes progress alongside it, and those writes
// already queue on transitions.

func (s *interestTrackingStore) AddHistoryIfMissing(ctx context.Context, entry userstore.WatchHistoryEntry) (bool, error) {
	created, err := s.UserStore.AddHistoryIfMissing(ctx, entry)
	if err == nil && created && entry.Completed {
		s.updater.QueueItemMutation(s.userID, entry.ProfileID, entry.MediaItemID)
	}
	return created, err
}

func (s *interestTrackingStore) RemoveHistoryItems(ctx context.Context, profileID string, mediaItemIDs []string, removedAt time.Time) error {
	err := s.UserStore.RemoveHistoryItems(ctx, profileID, mediaItemIDs, removedAt)
	if err == nil {
		for _, mediaItemID := range mediaItemIDs {
			s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
		}
	}
	return err
}

func (s *interestTrackingStore) DeleteHistoryBySource(ctx context.Context, profileID string, mediaItemIDs []string, source userstore.WatchHistorySource) error {
	err := s.UserStore.DeleteHistoryBySource(ctx, profileID, mediaItemIDs, source)
	if err == nil {
		for _, mediaItemID := range mediaItemIDs {
			s.updater.QueueItemMutation(s.userID, profileID, mediaItemID)
		}
	}
	return err
}

// DeleteProfile purges notification state alongside the profile itself;
// profiles may live outside Postgres, so no cascade covers these tables.
// The purge is best-effort: a failure is logged, never surfaced as a
// profile-deletion failure (the retention task prunes leftovers).
func (s *interestTrackingStore) DeleteProfile(ctx context.Context, id string) error {
	err := s.UserStore.DeleteProfile(ctx, id)
	if err == nil {
		purgeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
		defer cancel()
		if purgeErr := s.system.PurgeProfile(purgeCtx, id); purgeErr != nil {
			slog.WarnContext(ctx, "notifications: profile purge failed", "component", "notifications", "profile_id", id, "error", purgeErr)
		}
	}
	return err
}