package webhooksync import ( "context" "crypto/rand" "encoding/hex" "fmt" "log/slog" "net/http" "strings" "github.com/google/uuid" "github.com/Silo-Server/silo-server/internal/historyimport" "github.com/Silo-Server/silo-server/internal/userstore" "github.com/Silo-Server/silo-server/internal/watchstate" ) type Service struct { repo *Repository importRepo *historyimport.Repository matcher *historyimport.Matcher watch *watchstate.Service providers map[string]Provider } func NewService(repo *Repository, importRepo *historyimport.Repository, storeProvider userstore.UserStoreProvider) *Service { plexClient := historyimport.NewPlexClient() return &Service{ repo: repo, importRepo: importRepo, matcher: historyimport.NewMatcher(importRepo), watch: watchstate.NewService(storeProvider), providers: map[string]Provider{ ProviderPlex: NewPlexProvider(plexClient), ProviderEmby: NewEmbyProvider(), ProviderJellyfin: NewJellyfinProvider(), }, } } func (s *Service) SetStableIdentityResolver(identity *watchstate.StableIdentityResolver) { if s == nil || s.watch == nil { return } s.watch = s.watch.WithStableIdentityResolver(identity) } func (s *Service) provider(id string) (Provider, error) { provider, ok := s.providers[id] if !ok { return nil, fmt.Errorf("unsupported provider %q", id) } return provider, nil } func (s *Service) ListConnections(ctx context.Context, userID int) ([]Connection, error) { connections, err := s.repo.ListConnections(ctx, userID) if err != nil { return nil, err } for i := range connections { provider, providerErr := s.provider(connections[i].Provider) if providerErr != nil { continue } if _, available, err := provider.DiscoverUsers(ctx, &connections[i], nil); err == nil { connections[i].AccountDiscoveryAvailable = available if err := s.repo.SetDiscoveryAvailable(ctx, connections[i].ID, available); err != nil { slog.WarnContext(ctx, "webhook sync: failed to persist discovery availability", "component", "webhooksync", "connection_id", connections[i].ID, "error", err) } } } return connections, nil } func (s *Service) CreateConnection(ctx context.Context, userID int, input CreateConnectionInput, baseURL string) (*CreateConnectionResult, error) { provider, err := s.provider(input.Provider) if err != nil { return nil, err } if err := provider.ValidateCreateInput(input); err != nil { return nil, err } exists, err := s.repo.ProfileExistsForUser(ctx, userID, input.DefaultProfileID) if err != nil { return nil, err } if !exists { return nil, historyimport.ErrProfileNotFound } secret, err := randomSecret() if err != nil { return nil, err } conn, err := s.repo.CreateConnection(ctx, Connection{ ID: uuid.NewString(), UserID: userID, Provider: input.Provider, ServerID: input.ServerID, ServerName: input.ServerName, BaseURL: input.BaseURL, AccessToken: input.AccessToken, DefaultProfileID: input.DefaultProfileID, WebhookSecret: secret, }) if err != nil { return nil, err } if userID, userName, ok, err := provider.DefaultUser(ctx, conn, input); err != nil { return nil, err } else if ok { if _, err := s.repo.CreateDefaultMapping(ctx, conn.ID, userID, userName, input.DefaultProfileID); err != nil { return nil, err } } return &CreateConnectionResult{ Connection: *conn, WebhookURL: buildWebhookURL(baseURL, secret), }, nil } func (s *Service) UpdateConnection(ctx context.Context, userID int, id string, input UpdateConnectionInput) (*Connection, error) { if input.ServerName != nil && strings.TrimSpace(*input.ServerName) == "" { return nil, fmt.Errorf("server_name cannot be empty") } if input.DefaultProfileID != nil && *input.DefaultProfileID != "" { exists, err := s.repo.ProfileExistsForUser(ctx, userID, *input.DefaultProfileID) if err != nil { return nil, err } if !exists { return nil, historyimport.ErrProfileNotFound } } return s.repo.UpdateConnection(ctx, userID, id, input) } func (s *Service) DeleteConnection(ctx context.Context, userID int, id string) error { return s.repo.DeleteConnection(ctx, userID, id) } func (s *Service) RotateWebhook(ctx context.Context, userID int, id, baseURL string) (*RotateWebhookResult, error) { secret, err := randomSecret() if err != nil { return nil, err } if err := s.repo.UpdateWebhookSecret(ctx, userID, id, secret); err != nil { return nil, err } return &RotateWebhookResult{WebhookURL: buildWebhookURL(baseURL, secret)}, nil } func (s *Service) GetProfileMappings(ctx context.Context, userID int, id string) (*ProfileMappingsResponse, error) { conn, err := s.repo.GetConnection(ctx, userID, id) if err != nil { return nil, err } mappings, err := s.repo.ListMappings(ctx, conn.ID) if err != nil { return nil, err } provider, err := s.provider(conn.Provider) if err != nil { return nil, err } discovered, available, discoverErr := provider.DiscoverUsers(ctx, conn, mappings) if discoverErr == nil { conn.AccountDiscoveryAvailable = available _ = s.repo.SetDiscoveryAvailable(ctx, conn.ID, available) } else { discovered = mappingsToDiscoveredUsers(mappings) } if len(discovered) == 0 { discovered = mappingsToDiscoveredUsers(mappings) } return &ProfileMappingsResponse{ Mappings: mappings, DiscoveredUsers: discovered, AccountDiscoveryAvailable: conn.AccountDiscoveryAvailable, }, nil } func (s *Service) UpdateProfileMappings(ctx context.Context, userID int, id string, input UpdateProfileMappingsInput) ([]ProfileMapping, error) { conn, err := s.repo.GetConnection(ctx, userID, id) if err != nil { return nil, err } for _, mapping := range input.Mappings { if mapping.SiloProfileID == nil || *mapping.SiloProfileID == "" { continue } exists, err := s.repo.ProfileExistsForUser(ctx, userID, *mapping.SiloProfileID) if err != nil { return nil, err } if !exists { return nil, historyimport.ErrProfileNotFound } } return s.repo.ReplaceMappings(ctx, conn.ID, input.Mappings) } func (s *Service) ListEventLogs(ctx context.Context, userID int, id string, limit int) ([]WebhookEventLog, error) { conn, err := s.repo.GetConnection(ctx, userID, id) if err != nil { return nil, err } return s.repo.ListEventLogs(ctx, conn.ID, limit) } func (s *Service) CreateEventLog(ctx context.Context, entry WebhookEventLog) (*WebhookEventLog, error) { return s.repo.CreateEventLog(ctx, entry) } func (s *Service) ProcessWebhook(ctx context.Context, secret string, r *http.Request) (*ProcessWebhookResult, error) { conn, err := s.repo.GetConnectionBySecret(ctx, secret) if err != nil { return nil, err } result := &ProcessWebhookResult{ ConnectionID: conn.ID, Provider: conn.Provider, } provider, err := s.provider(conn.Provider) if err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Provider configuration failed") } event, err := provider.ParseWebhook(ctx, conn, r) if err != nil { result.Outcome = OutcomeRejected result.Summary = "Rejected invalid " + conn.Provider + " webhook payload" result.ErrorMessage = err.Error() _ = s.repo.MarkWebhookError(ctx, conn.ID, err.Error()) return result, err } if event != nil { result.EventKind = event.EventKind result.Action = event.Action result.UserID = event.UserID result.UserName = event.UserName result.ExternalItemID = event.ExternalItemID result.MediaKind = event.MediaKind } if err := s.repo.MarkWebhookReceived(ctx, conn.ID); err != nil { slog.WarnContext(ctx, "webhook sync: failed to mark receipt", "component", "webhooksync", "connection_id", conn.ID, "error", err) } if event == nil || !event.Apply { result.Outcome = OutcomeIgnored result.Summary = webhookResultSummary(event, "Ignored unsupported webhook event") return result, nil } if err := s.repo.UpsertSeenUser(ctx, conn.ID, event.UserID, event.UserName); err != nil { slog.WarnContext(ctx, "webhook sync: failed to upsert seen external user", "component", "webhooksync", "connection_id", conn.ID, "external_user_id", event.UserID, "error", err) } if mapping, err := s.repo.GetMappingByUser(ctx, conn.ID, event.UserID); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to resolve profile mapping") } else if profileID, ok := resolveWebhookProfileID(mapping); ok { result.ProfileID = profileID } else { result.Outcome = OutcomeSkipped result.Summary = "Skipped because external user is not linked to a Silo profile" return result, nil } profileID := result.ProfileID record := event.Record.toHistoryImportRecord() match, _, err := s.matcher.Match(ctx, record) if err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to match webhook item against the catalog") } if match == nil { result.Outcome = OutcomeUnmatched result.Summary = "Received webhook event but found no matching Silo item" return result, nil } result.MatchedMediaItemID = match.MediaItemID result.MatchedMediaItemTitle = match.Title switch event.Action { case ActionMarkUnplayed: if state, err := s.repo.GetItemState(ctx, conn.ID, event.UserID, event.ExternalItemID); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to load existing item state") } else if state != nil && !event.OccurredAt.After(state.LastEventAt) { result.Outcome = OutcomeSkipped result.Summary = "Skipped stale mark-unplayed event" return result, nil } if err := s.watch.RecordImportedMarkUnplayed(ctx, conn.UserID, profileID, match.MediaItemID, event.OccurredAt); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to apply mark-unplayed event") } if err := s.repo.UpsertItemState(ctx, ItemState{ ConnectionID: conn.ID, ExternalUserID: event.UserID, ExternalItemID: event.ExternalItemID, MediaItemID: match.MediaItemID, LastEventAt: event.OccurredAt, LastCompleted: false, LastPositionSecond: 0, }); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to persist mark-unplayed item state") } result.Outcome = OutcomeApplied result.Summary = "Applied mark-unplayed event" return result, nil case ActionAddFavorite: if err := s.watch.SetFavorite(ctx, conn.UserID, profileID, match.MediaItemID, true); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to add favorite from webhook event") } result.Outcome = OutcomeApplied result.Summary = "Applied favorite add event" return result, nil case ActionRemoveFavorite: if err := s.watch.SetFavorite(ctx, conn.UserID, profileID, match.MediaItemID, false); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to remove favorite from webhook event") } result.Outcome = OutcomeApplied result.Summary = "Applied favorite removal event" return result, nil case ActionToggleFavorite: if _, err := s.watch.ToggleFavorite(ctx, conn.UserID, profileID, match.MediaItemID); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to toggle favorite from webhook event") } result.Outcome = OutcomeApplied result.Summary = "Applied favorite toggle event" return result, nil case "", ActionImportProgress: default: result.Outcome = OutcomeIgnored result.Summary = "Ignored unsupported webhook action" return result, nil } localProgress, err := s.importRepo.GetProgress(ctx, conn.UserID, profileID, match.MediaItemID) if err == nil && localProgress != nil && !record.UpdatedAt.After(localProgress.UpdatedAt) { result.Outcome = OutcomeSkipped result.Summary = "Skipped because local watch progress is newer" return result, nil } state, err := s.repo.GetItemState(ctx, conn.ID, event.UserID, event.ExternalItemID) if err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to load existing item state") } if shouldSkipEvent(state, event) { result.Outcome = OutcomeSkipped result.Summary = "Skipped stale or duplicate progress event" return result, nil } _, err = s.watch.RecordImportedWatch( ctx, conn.UserID, profileID, match.MediaItemID, record.DurationSeconds, importedPosition(record), record.Played, record.UpdatedAt, record.LastPlayedAt, ) if err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to apply imported watch progress") } if err := s.repo.UpsertItemState(ctx, ItemState{ ConnectionID: conn.ID, ExternalUserID: event.UserID, ExternalItemID: event.ExternalItemID, MediaItemID: match.MediaItemID, LastEventAt: event.OccurredAt, LastCompleted: event.Completed, LastPositionSecond: event.PositionSeconds, }); err != nil { return s.failWebhook(ctx, conn.ID, result, err, "Failed to persist imported watch progress") } result.Outcome = OutcomeApplied result.Summary = "Applied watch progress event" return result, nil } func (s *Service) failWebhook(ctx context.Context, connectionID string, result *ProcessWebhookResult, err error, summary string) (*ProcessWebhookResult, error) { if result == nil { result = &ProcessWebhookResult{ConnectionID: connectionID} } result.Outcome = OutcomeError result.Summary = summary result.ErrorMessage = err.Error() _ = s.repo.MarkWebhookError(ctx, connectionID, err.Error()) return result, err } func webhookResultSummary(event *CanonicalEvent, fallback string) string { if event != nil && strings.TrimSpace(event.Summary) != "" { return event.Summary } return fallback } func mappingsToDiscoveredUsers(mappings []ProfileMapping) []DiscoveredUser { if len(mappings) == 0 { return nil } out := make([]DiscoveredUser, 0, len(mappings)) for _, mapping := range mappings { out = append(out, DiscoveredUser{ ExternalUserID: mapping.ExternalUserID, ExternalUserName: mapping.ExternalUserName, }) } return out } func shouldSkipEvent(state *ItemState, event *CanonicalEvent) bool { if state == nil { return false } if event.OccurredAt.After(state.LastEventAt) { return false } if !state.LastCompleted && event.Completed { return false } if event.PositionSeconds >= state.LastPositionSecond+5 { return false } return true } func resolveWebhookProfileID(mapping *ProfileMapping) (string, bool) { if mapping == nil || mapping.SiloProfileID == nil || strings.TrimSpace(*mapping.SiloProfileID) == "" { return "", false } return *mapping.SiloProfileID, true } func buildWebhookURL(baseURL, secret string) string { if baseURL == "" { return webhookSyncPathPrefix + secret } return strings.TrimRight(baseURL, "/") + webhookSyncPathPrefix + secret } func randomSecret() (string, error) { buf := make([]byte, 24) if _, err := rand.Read(buf); err != nil { return "", fmt.Errorf("generate webhook secret: %w", err) } return hex.EncodeToString(buf), nil } func importedPosition(record historyimport.Record) float64 { if record.Played && record.DurationSeconds > 0 { return record.DurationSeconds } return record.PositionSeconds } func (r CanonicalRecord) toHistoryImportRecord() historyimport.Record { return historyimport.Record{ ExternalID: r.ExternalID, Kind: r.Kind, Title: r.Title, Year: r.Year, IMDbID: r.IMDbID, TMDBID: r.TMDBID, TVDBID: r.TVDBID, SeriesTitle: r.SeriesTitle, SeriesYear: r.SeriesYear, SeriesIMDbID: r.SeriesIMDbID, SeriesTMDBID: r.SeriesTMDBID, SeriesTVDBID: r.SeriesTVDBID, SeasonNumber: r.SeasonNumber, EpisodeNumber: r.EpisodeNumber, Played: r.Played, LastPlayedAt: r.LastPlayedAt, PositionSeconds: r.PositionSeconds, DurationSeconds: r.DurationSeconds, UpdatedAt: r.UpdatedAt, } } type Provider interface { ID() string ValidateCreateInput(input CreateConnectionInput) error DefaultUser(ctx context.Context, conn *Connection, input CreateConnectionInput) (externalUserID string, externalUserName string, ok bool, err error) DiscoverUsers(ctx context.Context, conn *Connection, mappings []ProfileMapping) ([]DiscoveredUser, bool, error) ParseWebhook(ctx context.Context, conn *Connection, r *http.Request) (*CanonicalEvent, error) }