diff --git a/cli/cmd/followers.go b/cli/cmd/followers.go index 5b0a775..e6e9a85 100644 --- a/cli/cmd/followers.go +++ b/cli/cmd/followers.go @@ -53,10 +53,10 @@ func enrichFollowerProfiles(ctx context.Context, service *store.BlueskyService, } // filterInactive filters follower infos to only include accounts inactive for N days -func filterInactive(ctx context.Context, service *store.BlueskyService, followerInfos []followerInfo, actors []string, inactiveDays int, logger *log.Logger) []followerInfo { +func filterInactive(ctx context.Context, service *store.BlueskyService, cacheRepo *store.CacheRepository, followerInfos []followerInfo, actors []string, inactiveDays int, refresh bool, logger *log.Logger) []followerInfo { logger.Infof("Checking activity status (threshold: %d days)...", inactiveDays) - lastPostDates := service.BatchGetLastPostDates(ctx, actors, 10) + lastPostDates := service.BatchGetLastPostDatesCached(ctx, cacheRepo, actors, 10, refresh) var filtered []followerInfo for i, info := range followerInfos { @@ -82,10 +82,13 @@ func filterInactive(ctx context.Context, service *store.BlueskyService, follower } // filterQuiet filters follower infos to only include quiet posters -func filterQuiet(ctx context.Context, service *store.BlueskyService, followerInfos []followerInfo, actors []string, threshold float64, logger *log.Logger) []followerInfo { - logger.Infof("Computing post rates (threshold: %.2f posts/day, this may take a while)...", threshold) +func filterQuiet(ctx context.Context, service *store.BlueskyService, cacheRepo *store.CacheRepository, followerInfos []followerInfo, actors []string, threshold float64, refresh bool, logger *log.Logger) []followerInfo { + logger.Infof("Computing post rates (threshold: %.2f posts/day)...", threshold) + if refresh { + logger.Infof("Refreshing cache (this may take a while)...") + } - postRates := service.BatchGetPostRates(ctx, actors, 30, 30, 10, func(current, total int) { + postRates := service.BatchGetPostRatesCached(ctx, cacheRepo, actors, 30, 30, 10, refresh, func(current, total int) { if current%10 == 0 || current == total { logger.Infof("Progress: %d/%d accounts analyzed", current, total) } @@ -126,6 +129,11 @@ func ListFollowersAction(ctx context.Context, cmd *cli.Command) error { return fmt.Errorf("not authenticated: run 'skycli login' first") } + cacheRepo, err := reg.GetCacheRepo() + if err != nil { + return fmt.Errorf("failed to get cache repository: %w", err) + } + actor := cmd.String("user") if actor == "" { actor = service.GetDid() @@ -136,6 +144,7 @@ func ListFollowersAction(ctx context.Context, cmd *cli.Command) error { quietPosters := cmd.Bool("quiet") quietThreshold := cmd.Float("threshold") outputFormat := cmd.String("output") + refresh := cmd.Bool("refresh") if limit == 0 { logger.Debugf("Fetching all followers for %v", actor) @@ -197,11 +206,11 @@ func ListFollowersAction(ctx context.Context, cmd *cli.Command) error { followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowers, logger) if inactiveDays > 0 { - followerInfos = filterInactive(ctx, service, followerInfos, actors, inactiveDays, logger) + followerInfos = filterInactive(ctx, service, cacheRepo, followerInfos, actors, inactiveDays, refresh, logger) } if quietPosters { - followerInfos = filterQuiet(ctx, service, followerInfos, actors, quietThreshold, logger) + followerInfos = filterQuiet(ctx, service, cacheRepo, followerInfos, actors, quietThreshold, refresh, logger) } switch outputFormat { @@ -233,6 +242,11 @@ func ListFollowingAction(ctx context.Context, cmd *cli.Command) error { return fmt.Errorf("not authenticated: run 'skycli login' first") } + cacheRepo, err := reg.GetCacheRepo() + if err != nil { + return fmt.Errorf("failed to get cache repository: %w", err) + } + actor := cmd.String("user") if actor == "" { actor = service.GetDid() @@ -242,6 +256,7 @@ func ListFollowingAction(ctx context.Context, cmd *cli.Command) error { quietPosters := cmd.Bool("quiet") quietThreshold := cmd.Float("threshold") outputFormat := cmd.String("output") + refresh := cmd.Bool("refresh") logger.Debugf("Fetching following for actor %v", actor) @@ -282,11 +297,11 @@ func ListFollowingAction(ctx context.Context, cmd *cli.Command) error { followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowing, logger) if inactiveDays > 0 { - followerInfos = filterInactive(ctx, service, followerInfos, actors, inactiveDays, logger) + followerInfos = filterInactive(ctx, service, cacheRepo, followerInfos, actors, inactiveDays, refresh, logger) } if quietPosters { - followerInfos = filterQuiet(ctx, service, followerInfos, actors, quietThreshold, logger) + followerInfos = filterQuiet(ctx, service, cacheRepo, followerInfos, actors, quietThreshold, refresh, logger) } switch outputFormat { @@ -449,19 +464,240 @@ func FollowersDiffAction(ctx context.Context, cmd *cli.Command) error { return fmt.Errorf("not authenticated: run 'skycli login' first") } + snapshotRepo, err := reg.GetSnapshotRepo() + if err != nil { + return fmt.Errorf("failed to get snapshot repository: %w", err) + } + + actor := cmd.String("user") + if actor == "" { + actor = service.GetDid() + } sinceStr := cmd.String("since") untilStr := cmd.String("until") + outputFormat := cmd.String("output") + + // Parse since parameter (date or snapshot ID) + sinceDate, err := time.Parse("2006-01-02", sinceStr) + var baselineSnapshot *store.SnapshotModel + if err != nil { + // Not a date, try as snapshot ID + model, err := snapshotRepo.Get(ctx, sinceStr) + if err != nil { + return fmt.Errorf("invalid --since parameter (not a date or snapshot ID): %w", err) + } + if model == nil { + return fmt.Errorf("snapshot not found: %s", sinceStr) + } + baselineSnapshot = model.(*store.SnapshotModel) + } else { + // Find snapshot by date + baselineSnapshot, err = snapshotRepo.FindByUserTypeAndDate(ctx, actor, "followers", sinceDate) + if err != nil { + return fmt.Errorf("failed to find snapshot: %w", err) + } + if baselineSnapshot == nil { + return fmt.Errorf("no snapshot found for %s on or before %s", actor, sinceStr) + } + } + + logger.Infof("Using baseline snapshot from %s (%d followers)", baselineSnapshot.CreatedAt().Format("2006-01-02 15:04"), baselineSnapshot.TotalCount) + + // Get baseline follower DIDs + baselineDids, err := snapshotRepo.GetActorDids(ctx, baselineSnapshot.ID()) + if err != nil { + return fmt.Errorf("failed to get baseline followers: %w", err) + } + + var comparisonDids []string + var comparisonLabel string + + if untilStr != "" { + // Snapshot-to-snapshot comparison + untilDate, err := time.Parse("2006-01-02", untilStr) + var comparisonSnapshot *store.SnapshotModel + if err != nil { + // Not a date, try as snapshot ID + model, err := snapshotRepo.Get(ctx, untilStr) + if err != nil { + return fmt.Errorf("invalid --until parameter (not a date or snapshot ID): %w", err) + } + if model == nil { + return fmt.Errorf("snapshot not found: %s", untilStr) + } + comparisonSnapshot = model.(*store.SnapshotModel) + } else { + // Find snapshot by date + comparisonSnapshot, err = snapshotRepo.FindByUserTypeAndDate(ctx, actor, "followers", untilDate) + if err != nil { + return fmt.Errorf("failed to find snapshot: %w", err) + } + if comparisonSnapshot == nil { + return fmt.Errorf("no snapshot found for %s on or before %s", actor, untilStr) + } + } + + logger.Infof("Comparing with snapshot from %s (%d followers)", comparisonSnapshot.CreatedAt().Format("2006-01-02 15:04"), comparisonSnapshot.TotalCount) + comparisonLabel = comparisonSnapshot.CreatedAt().Format("2006-01-02 15:04") + + comparisonDids, err = snapshotRepo.GetActorDids(ctx, comparisonSnapshot.ID()) + if err != nil { + return fmt.Errorf("failed to get comparison followers: %w", err) + } + } else { + // Snapshot-to-live comparison + logger.Infof("Fetching current followers for comparison...") + comparisonLabel = "now" + + var allFollowers []store.ActorProfile + cursor := "" + page := 0 + for { + page++ + response, err := service.GetFollowers(ctx, actor, 100, cursor) + if err != nil { + return fmt.Errorf("failed to fetch followers: %w", err) + } + + allFollowers = append(allFollowers, response.Followers...) + + if response.Cursor != "" { + logger.Infof("Fetched page %d (%d followers so far)...", page, len(allFollowers)) + } + + if response.Cursor == "" { + break + } + cursor = response.Cursor + } + + logger.Infof("Fetched %d current followers", len(allFollowers)) - if sinceStr == "" || untilStr == "" { - return fmt.Errorf("both --since and --until are required") + for _, follower := range allFollowers { + comparisonDids = append(comparisonDids, follower.Did) + } + } + + // Calculate diff + baselineSet := make(map[string]bool) + for _, did := range baselineDids { + baselineSet[did] = true + } + + comparisonSet := make(map[string]bool) + for _, did := range comparisonDids { + comparisonSet[did] = true + } + + // New followers: in comparison but not in baseline + var newFollowers []string + for _, did := range comparisonDids { + if !baselineSet[did] { + newFollowers = append(newFollowers, did) + } } - // TODO: Implement snapshot storage and comparison - // This requires a way to store historical follower lists - // Options: SQLite table, JSON files with timestamps, etc. + // Unfollows: in baseline but not in comparison + var unfollows []string + for _, did := range baselineDids { + if !comparisonSet[did] { + unfollows = append(unfollows, did) + } + } - ui.Infoln("Diff functionality requires snapshot storage (not yet implemented)") - ui.Infoln("Consider using 'followers export' to create manual snapshots") + // Output results + switch outputFormat { + case "json": + return outputDiffJSON(newFollowers, unfollows) + case "csv": + return outputDiffCSV(newFollowers, unfollows) + default: + displayDiffTable(baselineSnapshot.CreatedAt().Format("2006-01-02 15:04"), comparisonLabel, len(baselineDids), len(comparisonDids), newFollowers, unfollows) + } + + return nil +} + +func displayDiffTable(baselineLabel, comparisonLabel string, baselineCount, comparisonCount int, newFollowers, unfollows []string) { + ui.Titleln("Follower Diff: %s → %s", baselineLabel, comparisonLabel) + fmt.Println() + + fmt.Printf("Baseline: %d followers\n", baselineCount) + fmt.Printf("Comparison: %d followers\n", comparisonCount) + fmt.Printf("Net change: %+d\n", comparisonCount-baselineCount) + fmt.Println() + + if len(newFollowers) > 0 { + ui.Titleln("New Followers (%d)", len(newFollowers)) + for _, did := range newFollowers { + fmt.Printf(" + %s\n", did) + } + fmt.Println() + } + + if len(unfollows) > 0 { + ui.Titleln("Unfollows (%d)", len(unfollows)) + for _, did := range unfollows { + fmt.Printf(" - %s\n", did) + } + fmt.Println() + } + + if len(newFollowers) == 0 && len(unfollows) == 0 { + ui.Infoln("No changes detected") + } +} + +type diffOutput struct { + NewFollowers []string `json:"newFollowers"` + Unfollows []string `json:"unfollows"` + Summary struct { + BaselineCount int `json:"baselineCount"` + ComparisonCount int `json:"comparisonCount"` + NetChange int `json:"netChange"` + NewCount int `json:"newCount"` + UnfollowCount int `json:"unfollowCount"` + } `json:"summary"` +} + +func outputDiffJSON(newFollowers, unfollows []string) error { + output := diffOutput{ + NewFollowers: newFollowers, + Unfollows: unfollows, + } + if output.NewFollowers == nil { + output.NewFollowers = []string{} + } + if output.Unfollows == nil { + output.Unfollows = []string{} + } + output.Summary.NewCount = len(newFollowers) + output.Summary.UnfollowCount = len(unfollows) + + encoder := json.NewEncoder(os.Stdout) + encoder.SetIndent("", " ") + return encoder.Encode(output) +} + +func outputDiffCSV(newFollowers, unfollows []string) error { + writer := csv.NewWriter(os.Stdout) + defer writer.Flush() + + if err := writer.Write([]string{"type", "did"}); err != nil { + return err + } + + for _, did := range newFollowers { + if err := writer.Write([]string{"new_follower", did}); err != nil { + return err + } + } + + for _, did := range unfollows { + if err := writer.Write([]string{"unfollow", did}); err != nil { + return err + } + } return nil } @@ -483,6 +719,11 @@ func FollowersExportAction(ctx context.Context, cmd *cli.Command) error { return fmt.Errorf("not authenticated: run 'skycli login' first") } + cacheRepo, err := reg.GetCacheRepo() + if err != nil { + return fmt.Errorf("failed to get cache repository: %w", err) + } + actor := cmd.String("user") if actor == "" { actor = service.GetDid() @@ -491,6 +732,7 @@ func FollowersExportAction(ctx context.Context, cmd *cli.Command) error { quietPosters := cmd.Bool("quiet") quietThreshold := cmd.Float("threshold") outputFormat := cmd.String("output") + refresh := cmd.Bool("refresh") logger.Debugf("Exporting followers for actor %v with fmt %v", actor, outputFormat) @@ -521,11 +763,11 @@ func FollowersExportAction(ctx context.Context, cmd *cli.Command) error { followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowers, logger) if inactiveDays > 0 { - followerInfos = filterInactive(ctx, service, followerInfos, actors, inactiveDays, logger) + followerInfos = filterInactive(ctx, service, cacheRepo, followerInfos, actors, inactiveDays, refresh, logger) } if quietPosters { - followerInfos = filterQuiet(ctx, service, followerInfos, actors, quietThreshold, logger) + followerInfos = filterQuiet(ctx, service, cacheRepo, followerInfos, actors, quietThreshold, refresh, logger) } switch outputFormat { @@ -538,6 +780,33 @@ func FollowersExportAction(ctx context.Context, cmd *cli.Command) error { } } +// formatTimeSince formats a time duration into a human-readable string. +// +// Returns +// - "< 1 hour ago" for durations under 1 hour +// - "X hours ago" for under 24 hours +// - "X days ago" for longer durations. +func formatTimeSince(since time.Time) string { + if since.IsZero() { + return "never" + } + + duration := time.Since(since) + hours := duration.Hours() + + if hours < 1 { + return "< 1 hour ago" + } else if hours < 24 { + return fmt.Sprintf("%d hours ago", int(hours)) + } else { + days := int(hours / 24) + if days == 1 { + return "1 day ago" + } + return fmt.Sprintf("%d days ago", days) + } +} + func displayFollowersTable(followers []followerInfo, showInactive bool) { if len(followers) == 0 { ui.Infoln("No followers found") @@ -576,19 +845,11 @@ func displayFollowersTable(followers []followerInfo, showInactive bool) { if showInactive && info.IsQuiet { row = append(row, fmt.Sprintf("%.2f", info.PostsPerDay)) - lastPostInfo := "never" - if info.DaysSincePost >= 0 { - lastPostInfo = fmt.Sprintf("%d days ago", info.DaysSincePost) - } - row = append(row, lastPostInfo) + row = append(row, formatTimeSince(info.LastPostDate)) } else if info.IsQuiet { row = append(row, fmt.Sprintf("%.2f", info.PostsPerDay)) } else if showInactive { - lastPostInfo := "never" - if info.DaysSincePost >= 0 { - lastPostInfo = fmt.Sprintf("%d days ago", info.DaysSincePost) - } - row = append(row, lastPostInfo) + row = append(row, formatTimeSince(info.LastPostDate)) } row = append(row, profileURL) @@ -762,6 +1023,10 @@ func FollowersCommand() *cli.Command { Usage: "Output format: table, json, csv", Value: "table", }, + &cli.BoolFlag{ + Name: "refresh", + Usage: "Force refresh cached data (bypasses 24-hour cache)", + }, }, Action: ListFollowersAction, }, @@ -795,18 +1060,28 @@ func FollowersCommand() *cli.Command { { Name: "diff", Usage: "Compare follower lists between two dates", - UsageText: "Compare follower lists to identify new followers and unfollows. Requires snapshot storage (not yet implemented).", + UsageText: "Compare follower lists to identify new followers and unfollows. Without --until, compares snapshot to current live data.", ArgsUsage: " ", Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "user", + Aliases: []string{"u"}, + Usage: "User handle or DID (defaults to authenticated user)", + }, &cli.StringFlag{ Name: "since", - Usage: "Start date (YYYY-MM-DD)", + Usage: "Start date (YYYY-MM-DD) or snapshot ID", Required: true, }, &cli.StringFlag{ - Name: "until", - Usage: "End date (YYYY-MM-DD)", - Required: true, + Name: "until", + Usage: "End date (YYYY-MM-DD) or snapshot ID (omit to compare with live data)", + }, + &cli.StringFlag{ + Name: "output", + Aliases: []string{"o"}, + Usage: "Output format: table, json, csv", + Value: "table", }, }, Action: FollowersDiffAction, @@ -843,6 +1118,10 @@ func FollowersCommand() *cli.Command { Value: "csv", Required: true, }, + &cli.BoolFlag{ + Name: "refresh", + Usage: "Force refresh cached data (bypasses 24-hour cache)", + }, }, Action: FollowersExportAction, }, @@ -891,6 +1170,10 @@ func FollowingCommand() *cli.Command { Usage: "Output format: table, json, csv", Value: "table", }, + &cli.BoolFlag{ + Name: "refresh", + Usage: "Force refresh cached data (bypasses 24-hour cache)", + }, }, Action: ListFollowingAction, }, diff --git a/cli/internal/registry/registry.go b/cli/internal/registry/registry.go index 4247e65..9fc5de0 100644 --- a/cli/internal/registry/registry.go +++ b/cli/internal/registry/registry.go @@ -15,13 +15,15 @@ var ( // Registry manages singleton instances of repositories and services type Registry struct { - service *store.BlueskyService - sessionRepo *store.SessionRepository - feedRepo *store.FeedRepository - postRepo *store.PostRepository - profileRepo *store.ProfileRepository - initialized bool - mu sync.RWMutex + service *store.BlueskyService + sessionRepo *store.SessionRepository + feedRepo *store.FeedRepository + postRepo *store.PostRepository + profileRepo *store.ProfileRepository + snapshotRepo *store.SnapshotRepository + cacheRepo *store.CacheRepository + initialized bool + mu sync.RWMutex } // Get returns the singleton registry instance @@ -80,6 +82,24 @@ func (r *Registry) Init(ctx context.Context) error { } r.profileRepo = profileRepo + snapshotRepo, err := store.NewSnapshotRepository() + if err != nil { + return &RegistryError{Op: "InitSnapshotRepo", Err: err} + } + if err := snapshotRepo.Init(ctx); err != nil { + return &RegistryError{Op: "InitSnapshotRepo", Err: err} + } + r.snapshotRepo = snapshotRepo + + cacheRepo, err := store.NewCacheRepository() + if err != nil { + return &RegistryError{Op: "InitCacheRepo", Err: err} + } + if err := cacheRepo.Init(ctx); err != nil { + return &RegistryError{Op: "InitCacheRepo", Err: err} + } + r.cacheRepo = cacheRepo + r.service = store.NewBlueskyService("") if sessionRepo.HasValidSession(ctx) { @@ -138,6 +158,18 @@ func (r *Registry) Close() error { } } + if r.snapshotRepo != nil { + if err := r.snapshotRepo.Close(); err != nil { + errs = append(errs, err) + } + } + + if r.cacheRepo != nil { + if err := r.cacheRepo.Close(); err != nil { + errs = append(errs, err) + } + } + r.initialized = false if len(errs) > 0 { @@ -227,6 +259,38 @@ func (r *Registry) GetProfileRepo() (*store.ProfileRepository, error) { return r.profileRepo, nil } +// GetSnapshotRepo returns the SnapshotRepository singleton +func (r *Registry) GetSnapshotRepo() (*store.SnapshotRepository, error) { + r.mu.RLock() + defer r.mu.RUnlock() + + if !r.initialized { + return nil, &RegistryError{Op: "GetSnapshotRepo", Err: errors.New("registry not initialized")} + } + + if r.snapshotRepo == nil { + return nil, &RegistryError{Op: "GetSnapshotRepo", Err: errors.New("snapshot repository not available")} + } + + return r.snapshotRepo, nil +} + +// GetCacheRepo returns the CacheRepository singleton +func (r *Registry) GetCacheRepo() (*store.CacheRepository, error) { + r.mu.RLock() + defer r.mu.RUnlock() + + if !r.initialized { + return nil, &RegistryError{Op: "GetCacheRepo", Err: errors.New("registry not initialized")} + } + + if r.cacheRepo == nil { + return nil, &RegistryError{Op: "GetCacheRepo", Err: errors.New("cache repository not available")} + } + + return r.cacheRepo, nil +} + // IsInitialized returns whether the registry has been initialized func (r *Registry) IsInitialized() bool { r.mu.RLock() diff --git a/cli/internal/store/bluesky.go b/cli/internal/store/bluesky.go index f93aa56..ce4da12 100644 --- a/cli/internal/store/bluesky.go +++ b/cli/internal/store/bluesky.go @@ -8,10 +8,13 @@ import ( "errors" "fmt" "io" + "maps" "net/http" "strings" "sync" "time" + + "github.com/charmbracelet/log" ) const ( @@ -367,7 +370,6 @@ func (s *BlueskyService) GetProfile(ctx context.Context, actor string) (*ActorPr } // SearchActors searches for actors (users) matching the query string. -// Returns actor profiles with pagination support. func (s *BlueskyService) SearchActors(ctx context.Context, query string, limit int, cursor string) (*SearchActorsResponse, error) { urlPath := fmt.Sprintf("/xrpc/app.bsky.actor.searchActors?q=%s&limit=%d", strings.ReplaceAll(query, " ", "+"), limit) if cursor != "" { @@ -465,7 +467,6 @@ func (s *BlueskyService) SetTokens(accessToken, refreshToken string) { } // GetLastPostDate fetches the most recent post date for an actor. -// Returns zero time if the actor has no posts or if an error occurs. func (s *BlueskyService) GetLastPostDate(ctx context.Context, actor string) (time.Time, error) { feed, err := s.GetAuthorFeed(ctx, actor, 1, "") if err != nil { @@ -485,9 +486,8 @@ func (s *BlueskyService) GetLastPostDate(ctx context.Context, actor string) (tim return lastPost, nil } -// BatchGetLastPostDates fetches last post dates for multiple actors concurrently. +// BatchGetLastPostDates fetches last post dates for multiple actors concurrently, as a map of actor DID/handle to their last post date.. // Uses a semaphore to limit concurrent requests to maxConcurrent. -// Returns a map of actor DID/handle to their last post date. func (s *BlueskyService) BatchGetLastPostDates(ctx context.Context, actors []string, maxConcurrent int) map[string]time.Time { results := make(map[string]time.Time) resultsMu := &sync.Mutex{} @@ -517,9 +517,8 @@ func (s *BlueskyService) BatchGetLastPostDates(ctx context.Context, actors []str return results } -// BatchGetProfiles fetches full profiles for multiple actors concurrently. -// Uses a semaphore to limit concurrent requests to maxConcurrent. -// Returns a map of actor DID/handle to their full ActorProfile. +// BatchGetProfiles fetches full profiles for multiple actors concurrently, as a map of actor DID/handle to their full ActorProfile. +// Uses a semaphore to limit concurrent requests to maxConcurrent.. func (s *BlueskyService) BatchGetProfiles(ctx context.Context, actors []string, maxConcurrent int) map[string]*ActorProfile { results := make(map[string]*ActorProfile) resultsMu := &sync.Mutex{} @@ -556,10 +555,10 @@ type PostRate struct { SampleSize int } -// BatchGetPostRates calculates posting rates for multiple actors concurrently. +// BatchGetPostRates calculates posting rates for multiple actors concurrently, as a map of actor DID/handle to their [PostRate] metrics. +// // Samples recent posts from each actor and calculates posts per day over the lookback period. // Uses a semaphore to limit concurrent requests to maxConcurrent. -// Returns a map of actor DID/handle to their PostRate metrics. func (s *BlueskyService) BatchGetPostRates(ctx context.Context, actors []string, sampleSize int, lookbackDays int, maxConcurrent int, progressFn func(current, total int)) map[string]*PostRate { results := make(map[string]*PostRate) resultsMu := &sync.Mutex{} @@ -601,13 +600,11 @@ func (s *BlueskyService) BatchGetPostRates(ctx context.Context, actors []string, return } - // Get the last post date lastPost, err := time.Parse(time.RFC3339, feed.Feed[0].Post.IndexedAt) if err != nil { return } - // Filter posts within lookback window cutoffTime := time.Now().AddDate(0, 0, -lookbackDays) recentPosts := 0 for _, post := range feed.Feed { @@ -744,3 +741,119 @@ func parseJWTExpiry(token string) (time.Time, error) { return time.Unix(claims.Exp, 0), nil } + +// BatchGetPostRatesCached calculates posting rates for multiple actors with caching support. +// +// Checks cache first, falls back to API for cache misses, and saves results to cache. +// If refresh is true, bypasses cache and refetches all data from API. +// +// TODO: Implement per-item TTL for more efficient cache invalidation. +// FIXME: this function signature is ridiculous +func (s *BlueskyService) BatchGetPostRatesCached(ctx context.Context, cacheRepo *CacheRepository, actors []string, sampleSize int, lookbackDays int, maxConcurrent int, refresh bool, progressFn func(current, total int)) map[string]*PostRate { + results := make(map[string]*PostRate) + + // If not refreshing, try to load from cache + var actorsToFetch []string + if !refresh { + cached, err := cacheRepo.GetPostRates(ctx, actors) + if err == nil { + for _, actor := range actors { + if cache, ok := cached[actor]; ok && cache.IsFresh() { + results[actor] = &PostRate{ + PostsPerDay: cache.PostsPerDay, + LastPostDate: cache.LastPostDate, + SampleSize: cache.SampleSize, + } + } else { + actorsToFetch = append(actorsToFetch, actor) + } + } + } else { + actorsToFetch = actors + } + } else { + actorsToFetch = actors + } + + if len(actorsToFetch) > 0 { + apiResults := s.BatchGetPostRates(ctx, actorsToFetch, sampleSize, lookbackDays, maxConcurrent, progressFn) + maps.Copy(results, apiResults) + + var cacheModels []*PostRateCacheModel + for actor, postRate := range apiResults { + cacheModels = append(cacheModels, &PostRateCacheModel{ + ActorDid: actor, + PostsPerDay: postRate.PostsPerDay, + LastPostDate: postRate.LastPostDate, + SampleSize: postRate.SampleSize, + }) + } + + if len(cacheModels) > 0 { + if err := cacheRepo.SavePostRates(ctx, cacheModels); err != nil { + // Log error but don't fail - cache save is non-critical + } + } + } + + return results +} + +// BatchGetLastPostDatesCached fetches last post dates for multiple actors with caching support. +// +// Checks cache first, falls back to API for cache misses, and saves results to cache. +// If refresh is true, bypasses cache and refetches all data from API. +// +// TODO: Implement per-item TTL for more efficient cache invalidation. +// FIXME: this function signature is ridiculous +func (s *BlueskyService) BatchGetLastPostDatesCached(ctx context.Context, cacheRepo *CacheRepository, actors []string, maxConcurrent int, refresh bool) map[string]time.Time { + results := make(map[string]time.Time) + + var actorsToFetch []string + if !refresh { + cached, err := cacheRepo.GetActivities(ctx, actors) + if err == nil { + for _, actor := range actors { + if cache, ok := cached[actor]; ok && cache.IsFresh() { + if cache.HasPosted() { + results[actor] = cache.LastPostDate + } + } else { + actorsToFetch = append(actorsToFetch, actor) + } + } + } else { + actorsToFetch = actors + } + } else { + actorsToFetch = actors + } + + if len(actorsToFetch) > 0 { + apiResults := s.BatchGetLastPostDates(ctx, actorsToFetch, maxConcurrent) + maps.Copy(results, apiResults) + + var cacheModels []*ActivityCacheModel + for _, actor := range actorsToFetch { + lastPostDate, hasPosted := apiResults[actor] + cacheModels = append(cacheModels, &ActivityCacheModel{ + ActorDid: actor, + LastPostDate: lastPostDate, + FetchedAt: time.Now(), + ExpiresAt: time.Now().Add(24 * time.Hour), + }) + + if !hasPosted { + results[actor] = time.Time{} + } + } + + if len(cacheModels) > 0 { + if err := cacheRepo.SaveActivities(ctx, cacheModels); err != nil { + log.Warnf("save failed with error %v", err.Error()) + } + } + } + + return results +} diff --git a/cli/internal/store/cache_model.go b/cli/internal/store/cache_model.go new file mode 100644 index 0000000..3c1c820 --- /dev/null +++ b/cli/internal/store/cache_model.go @@ -0,0 +1,41 @@ +package store + +import "time" + +// PostRateCacheModel represents a cached post rate computation for an actor. +// Stores expensive post rate calculations with TTL support (24 hours default). +type PostRateCacheModel struct { + ActorDid string + PostsPerDay float64 + LastPostDate time.Time + SampleSize int + FetchedAt time.Time + ExpiresAt time.Time +} + +// IsFresh returns true if the cached post rate has not expired. +// Post rates expire after 24 hours by default. +func (m *PostRateCacheModel) IsFresh() bool { + return time.Now().Before(m.ExpiresAt) +} + +// ActivityCacheModel represents cached activity data (last post date) for an actor. +// Stores last post date lookups with TTL support (24 hours default). +type ActivityCacheModel struct { + ActorDid string + LastPostDate time.Time // May be zero if actor has never posted + FetchedAt time.Time + ExpiresAt time.Time +} + +// IsFresh returns true if the cached activity data has not expired. +// Activity data expires after 24 hours by default. +func (m *ActivityCacheModel) IsFresh() bool { + return time.Now().Before(m.ExpiresAt) +} + +// HasPosted returns true if the actor has posted at least once. +// A zero LastPostDate indicates the actor has never posted. +func (m *ActivityCacheModel) HasPosted() bool { + return !m.LastPostDate.IsZero() +} diff --git a/cli/internal/store/cache_repo.go b/cli/internal/store/cache_repo.go new file mode 100644 index 0000000..ff47064 --- /dev/null +++ b/cli/internal/store/cache_repo.go @@ -0,0 +1,475 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "time" + + _ "github.com/mattn/go-sqlite3" + "github.com/stormlightlabs/skypanel/cli/internal/config" +) + +// CacheRepository manages post rate and activity caches using SQLite. +// +// Provides methods for storing and retrieving expensive computation results stored as [PostRateCacheModel] or [ActivityCacheModel]. +type CacheRepository struct { + db *sql.DB +} + +// NewCacheRepository creates a new cache repository with SQLite backend +func NewCacheRepository() (*CacheRepository, error) { + dbPath, err := config.GetCacheDB() + if err != nil { + return nil, err + } + + db, err := sql.Open("sqlite3", dbPath) + if err != nil { + return nil, err + } + + return &CacheRepository{db: db}, nil +} + +// Init ensures database schema is initialized via migrations +func (r *CacheRepository) Init(ctx context.Context) error { + if err := config.EnsureConfigDir(); err != nil { + return err + } + return RunMigrations(r.db) +} + +// Close releases database connection +func (r *CacheRepository) Close() error { + return r.db.Close() +} + +// GetPostRate retrieves cached post rate for an actor +func (r *CacheRepository) GetPostRate(ctx context.Context, actorDid string) (*PostRateCacheModel, error) { + query := ` + SELECT actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at + FROM cached_post_rates + WHERE actor_did = ? AND expires_at > ? + ` + + var cache PostRateCacheModel + var lastPostDate sql.NullTime + + err := r.db.QueryRowContext(ctx, query, actorDid, time.Now()).Scan( + &cache.ActorDid, + &cache.PostsPerDay, + &lastPostDate, + &cache.SampleSize, + &cache.FetchedAt, + &cache.ExpiresAt, + ) + + if lastPostDate.Valid { + cache.LastPostDate = lastPostDate.Time + } + + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, &RepositoryError{Op: "GetPostRate", Err: err} + } + + return &cache, nil +} + +// GetPostRates retrieves cached post rates for multiple actors in a single query, +// as a map of actorDid -> PostRateCacheModel for found entries. +func (r *CacheRepository) GetPostRates(ctx context.Context, actorDids []string) (map[string]*PostRateCacheModel, error) { + if len(actorDids) == 0 { + return make(map[string]*PostRateCacheModel), nil + } + + query := ` + SELECT actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at + FROM cached_post_rates + WHERE actor_did IN (` + buildPlaceholders(len(actorDids)) + `) AND expires_at > ? + ` + + args := make([]interface{}, len(actorDids)+1) + for i, did := range actorDids { + args[i] = did + } + args[len(actorDids)] = time.Now() + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, &RepositoryError{Op: "GetPostRates", Err: err} + } + defer rows.Close() + + result := make(map[string]*PostRateCacheModel) + for rows.Next() { + var cache PostRateCacheModel + var lastPostDate sql.NullTime + + err := rows.Scan( + &cache.ActorDid, + &cache.PostsPerDay, + &lastPostDate, + &cache.SampleSize, + &cache.FetchedAt, + &cache.ExpiresAt, + ) + if err != nil { + return nil, &RepositoryError{Op: "GetPostRates", Err: err} + } + + if lastPostDate.Valid { + cache.LastPostDate = lastPostDate.Time + } + + result[cache.ActorDid] = &cache + } + + return result, rows.Err() +} + +// SavePostRate saves or updates a post rate cache entry +func (r *CacheRepository) SavePostRate(ctx context.Context, cache *PostRateCacheModel) error { + if cache.FetchedAt.IsZero() { + cache.FetchedAt = time.Now() + } + if cache.ExpiresAt.IsZero() { + cache.ExpiresAt = time.Now().Add(24 * time.Hour) + } + + query := ` + INSERT INTO cached_post_rates (actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(actor_did) DO UPDATE SET + posts_per_day = excluded.posts_per_day, + last_post_date = excluded.last_post_date, + sample_size = excluded.sample_size, + fetched_at = excluded.fetched_at, + expires_at = excluded.expires_at + ` + + var lastPostDate interface{} + if !cache.LastPostDate.IsZero() { + lastPostDate = cache.LastPostDate + } + + _, err := r.db.ExecContext(ctx, query, + cache.ActorDid, + cache.PostsPerDay, + lastPostDate, + cache.SampleSize, + cache.FetchedAt, + cache.ExpiresAt, + ) + + if err != nil { + return &RepositoryError{Op: "SavePostRate", Err: err} + } + + return nil +} + +// SavePostRates saves multiple post rate cache entries in a transaction +func (r *CacheRepository) SavePostRates(ctx context.Context, caches []*PostRateCacheModel) error { + if len(caches) == 0 { + return nil + } + + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return &RepositoryError{Op: "SavePostRates", Err: err} + } + defer tx.Rollback() + + stmt, err := tx.PrepareContext(ctx, ` + INSERT INTO cached_post_rates (actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(actor_did) DO UPDATE SET + posts_per_day = excluded.posts_per_day, + last_post_date = excluded.last_post_date, + sample_size = excluded.sample_size, + fetched_at = excluded.fetched_at, + expires_at = excluded.expires_at + `) + if err != nil { + return &RepositoryError{Op: "SavePostRates", Err: err} + } + defer stmt.Close() + + for _, cache := range caches { + if cache.FetchedAt.IsZero() { + cache.FetchedAt = time.Now() + } + if cache.ExpiresAt.IsZero() { + cache.ExpiresAt = time.Now().Add(24 * time.Hour) + } + + var lastPostDate interface{} + if !cache.LastPostDate.IsZero() { + lastPostDate = cache.LastPostDate + } + + _, err := stmt.ExecContext(ctx, + cache.ActorDid, + cache.PostsPerDay, + lastPostDate, + cache.SampleSize, + cache.FetchedAt, + cache.ExpiresAt, + ) + if err != nil { + return &RepositoryError{Op: "SavePostRates", Err: err} + } + } + + if err := tx.Commit(); err != nil { + return &RepositoryError{Op: "SavePostRates", Err: err} + } + + return nil +} + +// DeletePostRate removes a post rate cache entry +func (r *CacheRepository) DeletePostRate(ctx context.Context, actorDid string) error { + query := "DELETE FROM cached_post_rates WHERE actor_did = ?" + _, err := r.db.ExecContext(ctx, query, actorDid) + if err != nil { + return &RepositoryError{Op: "DeletePostRate", Err: err} + } + return nil +} + +// GetActivity retrieves cached activity data for an actor +func (r *CacheRepository) GetActivity(ctx context.Context, actorDid string) (*ActivityCacheModel, error) { + query := ` + SELECT actor_did, last_post_date, fetched_at, expires_at + FROM cached_activity + WHERE actor_did = ? AND expires_at > ? + ` + + var cache ActivityCacheModel + var lastPostDate sql.NullTime + + err := r.db.QueryRowContext(ctx, query, actorDid, time.Now()).Scan( + &cache.ActorDid, + &lastPostDate, + &cache.FetchedAt, + &cache.ExpiresAt, + ) + + if lastPostDate.Valid { + cache.LastPostDate = lastPostDate.Time + } + + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, &RepositoryError{Op: "GetActivity", Err: err} + } + + return &cache, nil +} + +// GetActivities retrieves cached activity data for multiple actors in a single query, +// as a map of actorDid -> ActivityCacheModel for found entries. +func (r *CacheRepository) GetActivities(ctx context.Context, actorDids []string) (map[string]*ActivityCacheModel, error) { + if len(actorDids) == 0 { + return make(map[string]*ActivityCacheModel), nil + } + + query := ` + SELECT actor_did, last_post_date, fetched_at, expires_at + FROM cached_activity + WHERE actor_did IN (` + buildPlaceholders(len(actorDids)) + `) AND expires_at > ? + ` + + args := make([]interface{}, len(actorDids)+1) + for i, did := range actorDids { + args[i] = did + } + args[len(actorDids)] = time.Now() + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, &RepositoryError{Op: "GetActivities", Err: err} + } + defer rows.Close() + + result := make(map[string]*ActivityCacheModel) + for rows.Next() { + var cache ActivityCacheModel + var lastPostDate sql.NullTime + + err := rows.Scan( + &cache.ActorDid, + &lastPostDate, + &cache.FetchedAt, + &cache.ExpiresAt, + ) + if err != nil { + return nil, &RepositoryError{Op: "GetActivities", Err: err} + } + + if lastPostDate.Valid { + cache.LastPostDate = lastPostDate.Time + } + + result[cache.ActorDid] = &cache + } + + return result, rows.Err() +} + +// SaveActivity saves or updates an activity cache entry +func (r *CacheRepository) SaveActivity(ctx context.Context, cache *ActivityCacheModel) error { + if cache.FetchedAt.IsZero() { + cache.FetchedAt = time.Now() + } + if cache.ExpiresAt.IsZero() { + cache.ExpiresAt = time.Now().Add(24 * time.Hour) + } + + query := ` + INSERT INTO cached_activity (actor_did, last_post_date, fetched_at, expires_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(actor_did) DO UPDATE SET + last_post_date = excluded.last_post_date, + fetched_at = excluded.fetched_at, + expires_at = excluded.expires_at + ` + + var lastPostDate interface{} + if !cache.LastPostDate.IsZero() { + lastPostDate = cache.LastPostDate + } + + _, err := r.db.ExecContext(ctx, query, + cache.ActorDid, + lastPostDate, + cache.FetchedAt, + cache.ExpiresAt, + ) + + if err != nil { + return &RepositoryError{Op: "SaveActivity", Err: err} + } + + return nil +} + +// SaveActivities saves multiple activity cache entries in a transaction +func (r *CacheRepository) SaveActivities(ctx context.Context, caches []*ActivityCacheModel) error { + if len(caches) == 0 { + return nil + } + + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return &RepositoryError{Op: "SaveActivities", Err: err} + } + defer tx.Rollback() + + stmt, err := tx.PrepareContext(ctx, ` + INSERT INTO cached_activity (actor_did, last_post_date, fetched_at, expires_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(actor_did) DO UPDATE SET + last_post_date = excluded.last_post_date, + fetched_at = excluded.fetched_at, + expires_at = excluded.expires_at + `) + if err != nil { + return &RepositoryError{Op: "SaveActivities", Err: err} + } + defer stmt.Close() + + for _, cache := range caches { + if cache.FetchedAt.IsZero() { + cache.FetchedAt = time.Now() + } + if cache.ExpiresAt.IsZero() { + cache.ExpiresAt = time.Now().Add(24 * time.Hour) + } + + var lastPostDate interface{} + if !cache.LastPostDate.IsZero() { + lastPostDate = cache.LastPostDate + } + + _, err := stmt.ExecContext(ctx, + cache.ActorDid, + lastPostDate, + cache.FetchedAt, + cache.ExpiresAt, + ) + if err != nil { + return &RepositoryError{Op: "SaveActivities", Err: err} + } + } + + if err := tx.Commit(); err != nil { + return &RepositoryError{Op: "SaveActivities", Err: err} + } + + return nil +} + +// DeleteActivity removes an activity cache entry +func (r *CacheRepository) DeleteActivity(ctx context.Context, actorDid string) error { + query := "DELETE FROM cached_activity WHERE actor_did = ?" + _, err := r.db.ExecContext(ctx, query, actorDid) + if err != nil { + return &RepositoryError{Op: "DeleteActivity", Err: err} + } + return nil +} + +// DeleteExpiredPostRates removes all expired post rate cache entries +func (r *CacheRepository) DeleteExpiredPostRates(ctx context.Context) (int64, error) { + query := "DELETE FROM cached_post_rates WHERE expires_at < ?" + result, err := r.db.ExecContext(ctx, query, time.Now()) + if err != nil { + return 0, &RepositoryError{Op: "DeleteExpiredPostRates", Err: err} + } + + rows, err := result.RowsAffected() + if err != nil { + return 0, &RepositoryError{Op: "DeleteExpiredPostRates", Err: err} + } + + return rows, nil +} + +// DeleteExpiredActivities removes all expired activity cache entries +func (r *CacheRepository) DeleteExpiredActivities(ctx context.Context) (int64, error) { + query := "DELETE FROM cached_activity WHERE expires_at < ?" + result, err := r.db.ExecContext(ctx, query, time.Now()) + if err != nil { + return 0, &RepositoryError{Op: "DeleteExpiredActivities", Err: err} + } + + rows, err := result.RowsAffected() + if err != nil { + return 0, &RepositoryError{Op: "DeleteExpiredActivities", Err: err} + } + + return rows, nil +} + +// buildPlaceholders generates SQL placeholder string for IN queries. +// +// Example: buildPlaceholders(3) returns "?,?,?" +func buildPlaceholders(count int) string { + if count == 0 { + return "" + } + + placeholders := "?" + for i := 1; i < count; i++ { + placeholders += ",?" + } + return placeholders +} diff --git a/cli/internal/store/cache_repo_test.go b/cli/internal/store/cache_repo_test.go new file mode 100644 index 0000000..1f4f2f0 --- /dev/null +++ b/cli/internal/store/cache_repo_test.go @@ -0,0 +1,516 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/stormlightlabs/skypanel/cli/internal/utils" +) + +func TestCacheRepository_Init(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + + err := repo.Init(context.Background()) + if err != nil { + t.Fatalf("Init failed: %v", err) + } + + var count int + err = db.QueryRow("SELECT COUNT(*) FROM cached_post_rates").Scan(&count) + if err != nil { + t.Errorf("cached_post_rates table not created: %v", err) + } + + err = db.QueryRow("SELECT COUNT(*) FROM cached_activity").Scan(&count) + if err != nil { + t.Errorf("cached_activity table not created: %v", err) + } +} + +func TestCacheRepository_SaveAndGetPostRate(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &PostRateCacheModel{ + ActorDid: "did:plc:test123", + PostsPerDay: 2.5, + LastPostDate: time.Now().Add(-2 * time.Hour), + SampleSize: 30, + } + + err := repo.SavePostRate(context.Background(), cache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + if cache.FetchedAt.IsZero() { + t.Error("expected FetchedAt to be set after Save") + } + if cache.ExpiresAt.IsZero() { + t.Error("expected ExpiresAt to be set after Save") + } + + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:test123") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + + if retrieved == nil { + t.Fatal("expected cache entry, got nil") + } + + if retrieved.ActorDid != "did:plc:test123" { + t.Errorf("expected ActorDid 'did:plc:test123', got %s", retrieved.ActorDid) + } + if retrieved.PostsPerDay != 2.5 { + t.Errorf("expected PostsPerDay 2.5, got %f", retrieved.PostsPerDay) + } + if retrieved.SampleSize != 30 { + t.Errorf("expected SampleSize 30, got %d", retrieved.SampleSize) + } +} + +func TestCacheRepository_GetPostRate_NotFound(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:nonexistent") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + + if retrieved != nil { + t.Error("expected nil for nonexistent cache entry") + } +} + +func TestCacheRepository_GetPostRate_Expired(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &PostRateCacheModel{ + ActorDid: "did:plc:expired", + PostsPerDay: 1.0, + SampleSize: 10, + FetchedAt: time.Now().Add(-25 * time.Hour), + ExpiresAt: time.Now().Add(-1 * time.Hour), + } + + err := repo.SavePostRate(context.Background(), cache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:expired") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + + if retrieved != nil { + t.Error("expected nil for expired cache entry") + } +} + +func TestCacheRepository_SavePostRates_Batch(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + caches := []*PostRateCacheModel{ + { + ActorDid: "did:plc:user1", + PostsPerDay: 1.5, + LastPostDate: time.Now().Add(-1 * time.Hour), + SampleSize: 20, + }, + { + ActorDid: "did:plc:user2", + PostsPerDay: 3.0, + LastPostDate: time.Now().Add(-2 * time.Hour), + SampleSize: 30, + }, + { + ActorDid: "did:plc:user3", + PostsPerDay: 0.0, + SampleSize: 0, + }, + } + + err := repo.SavePostRates(context.Background(), caches) + if err != nil { + t.Fatalf("SavePostRates failed: %v", err) + } + + retrieved, err := repo.GetPostRates(context.Background(), []string{"did:plc:user1", "did:plc:user2", "did:plc:user3"}) + if err != nil { + t.Fatalf("GetPostRates failed: %v", err) + } + + if len(retrieved) != 3 { + t.Errorf("expected 3 cache entries, got %d", len(retrieved)) + } + + if cache, ok := retrieved["did:plc:user1"]; ok { + if cache.PostsPerDay != 1.5 { + t.Errorf("expected PostsPerDay 1.5 for user1, got %f", cache.PostsPerDay) + } + } else { + t.Error("expected cache entry for user1") + } + + if cache, ok := retrieved["did:plc:user3"]; ok { + if cache.LastPostDate.IsZero() == false { + t.Error("expected zero LastPostDate for user3 (never posted)") + } + } +} + +func TestCacheRepository_GetPostRates_PartialMatch(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &PostRateCacheModel{ + ActorDid: "did:plc:cached", + PostsPerDay: 2.0, + SampleSize: 25, + } + + err := repo.SavePostRate(context.Background(), cache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + retrieved, err := repo.GetPostRates(context.Background(), []string{"did:plc:cached", "did:plc:notcached"}) + if err != nil { + t.Fatalf("GetPostRates failed: %v", err) + } + + if len(retrieved) != 1 { + t.Errorf("expected 1 cache entry, got %d", len(retrieved)) + } + + if _, ok := retrieved["did:plc:cached"]; !ok { + t.Error("expected cache entry for 'did:plc:cached'") + } + + if _, ok := retrieved["did:plc:notcached"]; ok { + t.Error("did not expect cache entry for 'did:plc:notcached'") + } +} + +func TestCacheRepository_SaveAndGetActivity(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &ActivityCacheModel{ + ActorDid: "did:plc:active", + LastPostDate: time.Now().Add(-3 * time.Hour), + } + + err := repo.SaveActivity(context.Background(), cache) + if err != nil { + t.Fatalf("SaveActivity failed: %v", err) + } + + if cache.FetchedAt.IsZero() { + t.Error("expected FetchedAt to be set after Save") + } + if cache.ExpiresAt.IsZero() { + t.Error("expected ExpiresAt to be set after Save") + } + + retrieved, err := repo.GetActivity(context.Background(), "did:plc:active") + if err != nil { + t.Fatalf("GetActivity failed: %v", err) + } + + if retrieved == nil { + t.Fatal("expected cache entry, got nil") + } + + if retrieved.ActorDid != "did:plc:active" { + t.Errorf("expected ActorDid 'did:plc:active', got %s", retrieved.ActorDid) + } + if retrieved.LastPostDate.IsZero() { + t.Error("expected non-zero LastPostDate") + } + if !retrieved.HasPosted() { + t.Error("expected HasPosted to be true") + } +} + +func TestCacheRepository_SaveActivity_NeverPosted(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &ActivityCacheModel{ + ActorDid: "did:plc:neverposted", + LastPostDate: time.Time{}, + } + + err := repo.SaveActivity(context.Background(), cache) + if err != nil { + t.Fatalf("SaveActivity failed: %v", err) + } + + retrieved, err := repo.GetActivity(context.Background(), "did:plc:neverposted") + if err != nil { + t.Fatalf("GetActivity failed: %v", err) + } + + if retrieved == nil { + t.Fatal("expected cache entry, got nil") + } + + if !retrieved.LastPostDate.IsZero() { + t.Error("expected zero LastPostDate for actor who never posted") + } + if retrieved.HasPosted() { + t.Error("expected HasPosted to be false") + } +} + +func TestCacheRepository_SaveActivities_Batch(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + caches := []*ActivityCacheModel{ + { + ActorDid: "did:plc:actor1", + LastPostDate: time.Now().Add(-1 * time.Hour), + }, + { + ActorDid: "did:plc:actor2", + LastPostDate: time.Now().Add(-5 * time.Hour), + }, + { + ActorDid: "did:plc:actor3", + LastPostDate: time.Time{}, + }, + } + + err := repo.SaveActivities(context.Background(), caches) + if err != nil { + t.Fatalf("SaveActivities failed: %v", err) + } + + retrieved, err := repo.GetActivities(context.Background(), []string{"did:plc:actor1", "did:plc:actor2", "did:plc:actor3"}) + if err != nil { + t.Fatalf("GetActivities failed: %v", err) + } + + if len(retrieved) != 3 { + t.Errorf("expected 3 cache entries, got %d", len(retrieved)) + } + + if cache, ok := retrieved["did:plc:actor1"]; ok { + if cache.LastPostDate.IsZero() { + t.Error("expected non-zero LastPostDate for actor1") + } + } else { + t.Error("expected cache entry for actor1") + } + + if cache, ok := retrieved["did:plc:actor3"]; ok { + if !cache.LastPostDate.IsZero() { + t.Error("expected zero LastPostDate for actor3") + } + } +} + +func TestCacheRepository_Upsert_PostRate(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &PostRateCacheModel{ + ActorDid: "did:plc:upserttest", + PostsPerDay: 1.0, + SampleSize: 10, + } + + err := repo.SavePostRate(context.Background(), cache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + updatedCache := &PostRateCacheModel{ + ActorDid: "did:plc:upserttest", + PostsPerDay: 3.0, + SampleSize: 30, + } + + err = repo.SavePostRate(context.Background(), updatedCache) + if err != nil { + t.Fatalf("Update SavePostRate failed: %v", err) + } + + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:upserttest") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + + if retrieved.PostsPerDay != 3.0 { + t.Errorf("expected PostsPerDay 3.0 after upsert, got %f", retrieved.PostsPerDay) + } + if retrieved.SampleSize != 30 { + t.Errorf("expected SampleSize 30 after upsert, got %d", retrieved.SampleSize) + } +} + +func TestCacheRepository_DeletePostRate(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + cache := &PostRateCacheModel{ + ActorDid: "did:plc:todelete", + PostsPerDay: 2.0, + SampleSize: 20, + } + + err := repo.SavePostRate(context.Background(), cache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + err = repo.DeletePostRate(context.Background(), "did:plc:todelete") + if err != nil { + t.Fatalf("DeletePostRate failed: %v", err) + } + + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:todelete") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + if retrieved != nil { + t.Error("expected nil after DeletePostRate") + } +} + +func TestCacheRepository_DeleteExpiredPostRates(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + freshCache := &PostRateCacheModel{ + ActorDid: "did:plc:fresh", + PostsPerDay: 2.0, + SampleSize: 20, + FetchedAt: time.Now(), + ExpiresAt: time.Now().Add(24 * time.Hour), + } + + expiredCache := &PostRateCacheModel{ + ActorDid: "did:plc:expired", + PostsPerDay: 1.0, + SampleSize: 10, + FetchedAt: time.Now().Add(-25 * time.Hour), + ExpiresAt: time.Now().Add(-1 * time.Hour), + } + + err := repo.SavePostRate(context.Background(), freshCache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + err = repo.SavePostRate(context.Background(), expiredCache) + if err != nil { + t.Fatalf("SavePostRate failed: %v", err) + } + + deleted, err := repo.DeleteExpiredPostRates(context.Background()) + if err != nil { + t.Fatalf("DeleteExpiredPostRates failed: %v", err) + } + + if deleted != 1 { + t.Errorf("expected 1 deleted entry, got %d", deleted) + } + + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:fresh") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + if retrieved == nil { + t.Error("expected fresh cache to still exist") + } + + retrieved, err = repo.GetPostRate(context.Background(), "did:plc:expired") + if err != nil { + t.Fatalf("GetPostRate failed: %v", err) + } + if retrieved != nil { + t.Error("expected expired cache to be deleted") + } +} + +func TestCacheRepository_Close(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &CacheRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + err := repo.Close() + if err != nil { + t.Errorf("Close failed: %v", err) + } +} diff --git a/cli/internal/store/migration_test.go b/cli/internal/store/migration_test.go index 19575b4..069e2e7 100644 --- a/cli/internal/store/migration_test.go +++ b/cli/internal/store/migration_test.go @@ -7,8 +7,6 @@ import ( "github.com/stormlightlabs/skypanel/cli/internal/utils" ) -// TestRunMigrations verifies that migrations are applied correctly to a fresh database. -// It checks that the schema_migrations table is created and all migrations are executed in order. func TestRunMigrations(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -23,8 +21,8 @@ func TestRunMigrations(t *testing.T) { t.Fatalf("schema_migrations table not found: %v", err) } - if count != 3 { - t.Errorf("expected 3 migrations applied, got %d", count) + if count != 4 { + t.Errorf("expected 4 migrations applied, got %d", count) } err = db.QueryRow("SELECT COUNT(*) FROM feeds").Scan(&count) @@ -43,8 +41,6 @@ func TestRunMigrations(t *testing.T) { } } -// TestRunMigrations_Idempotent verifies that running migrations multiple times -// doesn't re-apply already executed migrations. func TestRunMigrations_Idempotent(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -63,12 +59,11 @@ func TestRunMigrations_Idempotent(t *testing.T) { t.Fatalf("failed to query migrations: %v", err) } - if count != 3 { - t.Errorf("expected 3 migrations, got %d", count) + if count != 4 { + t.Errorf("expected 4 migrations, got %d", count) } } -// TestRollback verifies that down migrations correctly revert database changes. func TestRollback(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -102,7 +97,6 @@ func TestRollback(t *testing.T) { } } -// TestRollback_Complete verifies that rolling back to version 0 removes all migrations. func TestRollback_Complete(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -138,7 +132,6 @@ func TestRollback_Complete(t *testing.T) { } } -// TestMigrationOrdering verifies that migrations are applied in correct version order. func TestMigrationOrdering(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -153,7 +146,7 @@ func TestMigrationOrdering(t *testing.T) { } defer rows.Close() - expectedVersions := []int{1, 2, 3} + expectedVersions := []int{1, 2, 3, 4} var actualVersions []int for rows.Next() { @@ -175,7 +168,6 @@ func TestMigrationOrdering(t *testing.T) { } } -// TestGetAppliedMigrations verifies the helper function correctly retrieves applied migrations. func TestGetAppliedMigrations(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -205,15 +197,14 @@ func TestGetAppliedMigrations(t *testing.T) { } } -// TestLoadMigrations verifies that migration files are correctly loaded from embedded FS. func TestLoadMigrations(t *testing.T) { upMigrations, err := loadMigrations("up") if err != nil { t.Fatalf("failed to load up migrations: %v", err) } - if len(upMigrations) != 3 { - t.Errorf("expected 3 up migrations, got %d", len(upMigrations)) + if len(upMigrations) != 4 { + t.Errorf("expected 4 up migrations, got %d", len(upMigrations)) } for i := 1; i < len(upMigrations); i++ { @@ -227,12 +218,11 @@ func TestLoadMigrations(t *testing.T) { t.Fatalf("failed to load down migrations: %v", err) } - if len(downMigrations) != 3 { - t.Errorf("expected 3 down migrations, got %d", len(downMigrations)) + if len(downMigrations) != 4 { + t.Errorf("expected 4 down migrations, got %d", len(downMigrations)) } } -// TestExecuteMigration verifies that SQL is correctly executed. func TestExecuteMigration(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -254,7 +244,6 @@ func TestExecuteMigration(t *testing.T) { } } -// TestRecordAndRemoveMigration verifies the migration tracking functions. func TestRecordAndRemoveMigration(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -289,7 +278,6 @@ func TestRecordAndRemoveMigration(t *testing.T) { } } -// TestMigrationWithForeignKey verifies that foreign key constraints work correctly. func TestMigrationWithForeignKey(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() @@ -327,7 +315,6 @@ func TestMigrationWithForeignKey(t *testing.T) { } } -// TestCreateMigrationsTable verifies the migrations tracking table is created correctly. func TestCreateMigrationsTable(t *testing.T) { db, cleanup := utils.NewTestDB(t) defer cleanup() diff --git a/cli/internal/store/migrations/004_create_cache_tables.down.sql b/cli/internal/store/migrations/004_create_cache_tables.down.sql new file mode 100644 index 0000000..e375a22 --- /dev/null +++ b/cli/internal/store/migrations/004_create_cache_tables.down.sql @@ -0,0 +1,4 @@ +DROP TABLE IF EXISTS cached_activity; +DROP TABLE IF EXISTS cached_post_rates; +DROP TABLE IF EXISTS follower_snapshot_entries; +DROP TABLE IF EXISTS follower_snapshots; diff --git a/cli/internal/store/migrations/004_create_cache_tables.up.sql b/cli/internal/store/migrations/004_create_cache_tables.up.sql new file mode 100644 index 0000000..7e32ed8 --- /dev/null +++ b/cli/internal/store/migrations/004_create_cache_tables.up.sql @@ -0,0 +1,48 @@ +-- Follower snapshots metadata +CREATE TABLE IF NOT EXISTS follower_snapshots ( + id TEXT PRIMARY KEY, + created_at DATETIME NOT NULL, + user_did TEXT NOT NULL, + snapshot_type TEXT NOT NULL, + total_count INTEGER NOT NULL, + expires_at DATETIME NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_snapshots_user_type ON follower_snapshots(user_did, snapshot_type); +CREATE INDEX IF NOT EXISTS idx_snapshots_created ON follower_snapshots(created_at); +CREATE INDEX IF NOT EXISTS idx_snapshots_expires ON follower_snapshots(expires_at); + +-- Snapshot entries (actors in each snapshot) +CREATE TABLE IF NOT EXISTS follower_snapshot_entries ( + snapshot_id TEXT NOT NULL, + actor_did TEXT NOT NULL, + indexed_at TEXT, + PRIMARY KEY(snapshot_id, actor_did), + FOREIGN KEY(snapshot_id) REFERENCES follower_snapshots(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_snapshot_entries_actor ON follower_snapshot_entries(actor_did); + +-- Cached post rate metrics +CREATE TABLE IF NOT EXISTS cached_post_rates ( + actor_did TEXT PRIMARY KEY, + posts_per_day REAL NOT NULL, + last_post_date DATETIME, + sample_size INTEGER NOT NULL, + fetched_at DATETIME NOT NULL, + expires_at DATETIME NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_post_rates_fetched ON cached_post_rates(fetched_at); +CREATE INDEX IF NOT EXISTS idx_post_rates_expires ON cached_post_rates(expires_at); + +-- Cached activity data (last post dates) +CREATE TABLE IF NOT EXISTS cached_activity ( + actor_did TEXT PRIMARY KEY, + last_post_date DATETIME, + fetched_at DATETIME NOT NULL, + expires_at DATETIME NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_activity_fetched ON cached_activity(fetched_at); +CREATE INDEX IF NOT EXISTS idx_activity_expires ON cached_activity(expires_at); diff --git a/cli/internal/store/snapshot_model.go b/cli/internal/store/snapshot_model.go new file mode 100644 index 0000000..2b3659b --- /dev/null +++ b/cli/internal/store/snapshot_model.go @@ -0,0 +1,34 @@ +package store + +import "time" + +// SnapshotModel represents a follower or following snapshot with metadata. +// Stores snapshot metadata with TTL support (24 hours default). +type SnapshotModel struct { + id string + createdAt time.Time + UserDid string + SnapshotType string // "followers" or "following" + TotalCount int + ExpiresAt time.Time +} + +func (m *SnapshotModel) ID() string { return m.id } +func (m *SnapshotModel) CreatedAt() time.Time { return m.createdAt } +func (m *SnapshotModel) UpdatedAt() time.Time { return m.createdAt } // Snapshots are immutable +func (m *SnapshotModel) SetID(id string) { m.id = id } +func (m *SnapshotModel) SetCreatedAt(t time.Time) { m.createdAt = t } +func (m *SnapshotModel) SetUpdatedAt(t time.Time) {} // Snapshots are immutable + +// IsFresh returns true if the snapshot has not expired. Snapshots expire after 24 hours by default. +func (m *SnapshotModel) IsFresh() bool { + return time.Now().Before(m.ExpiresAt) +} + +// SnapshotEntry represents an actor in a snapshot with minimal cached data. +// Linked to [SnapshotModel] via snapshot_id foreign key. +type SnapshotEntry struct { + SnapshotID string + ActorDid string + IndexedAt string // When the follow relationship was indexed by Bluesky +} diff --git a/cli/internal/store/snapshot_repo.go b/cli/internal/store/snapshot_repo.go new file mode 100644 index 0000000..046910d --- /dev/null +++ b/cli/internal/store/snapshot_repo.go @@ -0,0 +1,362 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "time" + + _ "github.com/mattn/go-sqlite3" + "github.com/stormlightlabs/skypanel/cli/internal/config" +) + +// SnapshotRepository implements Repository for SnapshotModel using SQLite. +// Manages follower/following snapshots with entries for diff and historical comparison. +type SnapshotRepository struct { + db *sql.DB +} + +// NewSnapshotRepository creates a new snapshot repository with SQLite backend +func NewSnapshotRepository() (*SnapshotRepository, error) { + dbPath, err := config.GetCacheDB() + if err != nil { + return nil, err + } + + db, err := sql.Open("sqlite3", dbPath) + if err != nil { + return nil, err + } + + return &SnapshotRepository{db: db}, nil +} + +// Init ensures database schema is initialized via migrations +func (r *SnapshotRepository) Init(ctx context.Context) error { + if err := config.EnsureConfigDir(); err != nil { + return err + } + return RunMigrations(r.db) +} + +// Close releases database connection +func (r *SnapshotRepository) Close() error { + return r.db.Close() +} + +// Get retrieves a snapshot by ID +func (r *SnapshotRepository) Get(ctx context.Context, id string) (Model, error) { + query := ` + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at + FROM follower_snapshots + WHERE id = ? + ` + + var snapshot SnapshotModel + var snapshotID string + var createdAt, expiresAt time.Time + + err := r.db.QueryRowContext(ctx, query, id).Scan( + &snapshotID, + &createdAt, + &snapshot.UserDid, + &snapshot.SnapshotType, + &snapshot.TotalCount, + &expiresAt, + ) + + snapshot.SetID(snapshotID) + snapshot.SetCreatedAt(createdAt) + snapshot.ExpiresAt = expiresAt + + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, &RepositoryError{Op: "Get", Err: errors.New("snapshot not found")} + } + return nil, &RepositoryError{Op: "Get", Err: err} + } + + return &snapshot, nil +} + +// List retrieves all snapshots ordered by creation date (newest first) +func (r *SnapshotRepository) List(ctx context.Context) ([]Model, error) { + query := ` + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at + FROM follower_snapshots + ORDER BY created_at DESC + ` + + rows, err := r.db.QueryContext(ctx, query) + if err != nil { + return nil, &RepositoryError{Op: "List", Err: err} + } + defer rows.Close() + + var snapshots []Model + for rows.Next() { + var snapshot SnapshotModel + var snapshotID string + var createdAt, expiresAt time.Time + + err := rows.Scan( + &snapshotID, + &createdAt, + &snapshot.UserDid, + &snapshot.SnapshotType, + &snapshot.TotalCount, + &expiresAt, + ) + if err != nil { + return nil, &RepositoryError{Op: "List", Err: err} + } + + snapshot.SetID(snapshotID) + snapshot.SetCreatedAt(createdAt) + snapshot.ExpiresAt = expiresAt + snapshots = append(snapshots, &snapshot) + } + + return snapshots, rows.Err() +} + +// Save creates a new snapshot (snapshots are immutable, no updates) +func (r *SnapshotRepository) Save(ctx context.Context, model Model) error { + snapshot, ok := model.(*SnapshotModel) + if !ok { + return &RepositoryError{Op: "Save", Err: errors.New("invalid model type: expected *SnapshotModel")} + } + + if snapshot.ID() == "" { + snapshot.SetID(GenerateUUID()) + snapshot.SetCreatedAt(time.Now()) + } + + if snapshot.ExpiresAt.IsZero() { + snapshot.ExpiresAt = time.Now().Add(24 * time.Hour) + } + + query := ` + INSERT INTO follower_snapshots (id, created_at, user_did, snapshot_type, total_count, expires_at) + VALUES (?, ?, ?, ?, ?, ?) + ` + + _, err := r.db.ExecContext(ctx, query, + snapshot.ID(), + snapshot.CreatedAt(), + snapshot.UserDid, + snapshot.SnapshotType, + snapshot.TotalCount, + snapshot.ExpiresAt, + ) + + if err != nil { + return &RepositoryError{Op: "Save", Err: err} + } + + return nil +} + +// Delete removes a snapshot by ID (cascade deletes entries) +func (r *SnapshotRepository) Delete(ctx context.Context, id string) error { + query := "DELETE FROM follower_snapshots WHERE id = ?" + result, err := r.db.ExecContext(ctx, query, id) + if err != nil { + return &RepositoryError{Op: "Delete", Err: err} + } + + rows, err := result.RowsAffected() + if err != nil { + return &RepositoryError{Op: "Delete", Err: err} + } + + if rows == 0 { + return &RepositoryError{Op: "Delete", Err: errors.New("snapshot not found")} + } + + return nil +} + +// FindByUserAndType retrieves the most recent fresh snapshot for a user and type. +func (r *SnapshotRepository) FindByUserAndType(ctx context.Context, userDid, snapshotType string) (*SnapshotModel, error) { + query := ` + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at + FROM follower_snapshots + WHERE user_did = ? AND snapshot_type = ? AND expires_at > ? + ORDER BY created_at DESC + LIMIT 1 + ` + + var snapshot SnapshotModel + var snapshotID string + var createdAt, expiresAt time.Time + + err := r.db.QueryRowContext(ctx, query, userDid, snapshotType, time.Now()).Scan( + &snapshotID, + &createdAt, + &snapshot.UserDid, + &snapshot.SnapshotType, + &snapshot.TotalCount, + &expiresAt, + ) + + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, &RepositoryError{Op: "FindByUserAndType", Err: err} + } + + snapshot.SetID(snapshotID) + snapshot.SetCreatedAt(createdAt) + snapshot.ExpiresAt = expiresAt + + return &snapshot, nil +} + +// FindByUserTypeAndDate retrieves a snapshot for a user, type, and specific date, closest to (but not after) the specified date. +func (r *SnapshotRepository) FindByUserTypeAndDate(ctx context.Context, userDid, snapshotType string, date time.Time) (*SnapshotModel, error) { + query := ` + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at + FROM follower_snapshots + WHERE user_did = ? AND snapshot_type = ? AND created_at <= ? + ORDER BY created_at DESC + LIMIT 1 + ` + + var snapshot SnapshotModel + var snapshotID string + var createdAt, expiresAt time.Time + + err := r.db.QueryRowContext(ctx, query, userDid, snapshotType, date).Scan( + &snapshotID, + &createdAt, + &snapshot.UserDid, + &snapshot.SnapshotType, + &snapshot.TotalCount, + &expiresAt, + ) + + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, &RepositoryError{Op: "FindByUserTypeAndDate", Err: err} + } + + snapshot.SetID(snapshotID) + snapshot.SetCreatedAt(createdAt) + snapshot.ExpiresAt = expiresAt + return &snapshot, nil +} + +// SaveEntry saves a single snapshot entry +func (r *SnapshotRepository) SaveEntry(ctx context.Context, entry *SnapshotEntry) error { + query := ` + INSERT INTO follower_snapshot_entries (snapshot_id, actor_did, indexed_at) + VALUES (?, ?, ?) + ` + + _, err := r.db.ExecContext(ctx, query, entry.SnapshotID, entry.ActorDid, entry.IndexedAt) + if err != nil { + return &RepositoryError{Op: "SaveEntry", Err: err} + } + return nil +} + +// SaveEntries saves multiple snapshot entries in a transaction for efficiency +func (r *SnapshotRepository) SaveEntries(ctx context.Context, entries []*SnapshotEntry) error { + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return &RepositoryError{Op: "SaveEntries", Err: err} + } + defer tx.Rollback() + + stmt, err := tx.PrepareContext(ctx, ` + INSERT INTO follower_snapshot_entries (snapshot_id, actor_did, indexed_at) + VALUES (?, ?, ?) + `) + if err != nil { + return &RepositoryError{Op: "SaveEntries", Err: err} + } + defer stmt.Close() + + for _, entry := range entries { + _, err := stmt.ExecContext(ctx, entry.SnapshotID, entry.ActorDid, entry.IndexedAt) + if err != nil { + return &RepositoryError{Op: "SaveEntries", Err: err} + } + } + + if err := tx.Commit(); err != nil { + return &RepositoryError{Op: "SaveEntries", Err: err} + } + return nil +} + +// GetEntries retrieves all entries for a snapshot +func (r *SnapshotRepository) GetEntries(ctx context.Context, snapshotID string) ([]*SnapshotEntry, error) { + query := ` + SELECT snapshot_id, actor_did, indexed_at + FROM follower_snapshot_entries + WHERE snapshot_id = ? + ` + + rows, err := r.db.QueryContext(ctx, query, snapshotID) + if err != nil { + return nil, &RepositoryError{Op: "GetEntries", Err: err} + } + defer rows.Close() + + var entries []*SnapshotEntry + for rows.Next() { + var entry SnapshotEntry + err := rows.Scan(&entry.SnapshotID, &entry.ActorDid, &entry.IndexedAt) + if err != nil { + return nil, &RepositoryError{Op: "GetEntries", Err: err} + } + entries = append(entries, &entry) + } + return entries, rows.Err() +} + +// GetActorDids retrieves just the actor DIDs for a snapshot (efficient for diffs) +func (r *SnapshotRepository) GetActorDids(ctx context.Context, snapshotID string) ([]string, error) { + query := ` + SELECT actor_did + FROM follower_snapshot_entries + WHERE snapshot_id = ? + ` + + rows, err := r.db.QueryContext(ctx, query, snapshotID) + if err != nil { + return nil, &RepositoryError{Op: "GetActorDids", Err: err} + } + defer rows.Close() + + var dids []string + for rows.Next() { + var did string + err := rows.Scan(&did) + if err != nil { + return nil, &RepositoryError{Op: "GetActorDids", Err: err} + } + dids = append(dids, did) + } + return dids, rows.Err() +} + +// DeleteExpiredSnapshots removes all expired snapshots and their entries +func (r *SnapshotRepository) DeleteExpiredSnapshots(ctx context.Context) (int64, error) { + query := "DELETE FROM follower_snapshots WHERE expires_at < ?" + result, err := r.db.ExecContext(ctx, query, time.Now()) + if err != nil { + return 0, &RepositoryError{Op: "DeleteExpiredSnapshots", Err: err} + } + + rows, err := result.RowsAffected() + if err != nil { + return 0, &RepositoryError{Op: "DeleteExpiredSnapshots", Err: err} + } + return rows, nil +} diff --git a/cli/internal/store/snapshot_repo_test.go b/cli/internal/store/snapshot_repo_test.go new file mode 100644 index 0000000..70d413b --- /dev/null +++ b/cli/internal/store/snapshot_repo_test.go @@ -0,0 +1,581 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/stormlightlabs/skypanel/cli/internal/utils" +) + +func TestSnapshotRepository_Init(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + + err := repo.Init(context.Background()) + if err != nil { + t.Fatalf("Init failed: %v", err) + } + + var count int + err = db.QueryRow("SELECT COUNT(*) FROM follower_snapshots").Scan(&count) + if err != nil { + t.Errorf("follower_snapshots table not created: %v", err) + } + + err = db.QueryRow("SELECT COUNT(*) FROM follower_snapshot_entries").Scan(&count) + if err != nil { + t.Errorf("follower_snapshot_entries table not created: %v", err) + } +} + +func TestSnapshotRepository_SaveAndGet(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:testuser", + SnapshotType: "followers", + TotalCount: 150, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save failed: %v", err) + } + + if snapshot.ID() == "" { + t.Error("expected ID to be set after Save") + } + if snapshot.CreatedAt().IsZero() { + t.Error("expected CreatedAt to be set after Save") + } + if snapshot.ExpiresAt.IsZero() { + t.Error("expected ExpiresAt to be set after Save") + } + + retrieved, err := repo.Get(context.Background(), snapshot.ID()) + if err != nil { + t.Fatalf("Get failed: %v", err) + } + + retrievedSnapshot, ok := retrieved.(*SnapshotModel) + if !ok { + t.Fatal("expected *SnapshotModel") + } + + if retrievedSnapshot.UserDid != "did:plc:testuser" { + t.Errorf("expected UserDid 'did:plc:testuser', got %s", retrievedSnapshot.UserDid) + } + if retrievedSnapshot.SnapshotType != "followers" { + t.Errorf("expected SnapshotType 'followers', got %s", retrievedSnapshot.SnapshotType) + } + if retrievedSnapshot.TotalCount != 150 { + t.Errorf("expected TotalCount 150, got %d", retrievedSnapshot.TotalCount) + } +} + +func TestSnapshotRepository_List(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot1 := &SnapshotModel{ + UserDid: "did:plc:user1", + SnapshotType: "followers", + TotalCount: 100, + } + snapshot2 := &SnapshotModel{ + UserDid: "did:plc:user1", + SnapshotType: "following", + TotalCount: 50, + } + + if err := repo.Save(context.Background(), snapshot1); err != nil { + t.Fatalf("Save snapshot1 failed: %v", err) + } + time.Sleep(10 * time.Millisecond) + if err := repo.Save(context.Background(), snapshot2); err != nil { + t.Fatalf("Save snapshot2 failed: %v", err) + } + + snapshots, err := repo.List(context.Background()) + if err != nil { + t.Fatalf("List failed: %v", err) + } + + if len(snapshots) != 2 { + t.Errorf("expected 2 snapshots, got %d", len(snapshots)) + } + + if s, ok := snapshots[0].(*SnapshotModel); ok { + if s.SnapshotType != "following" { + t.Errorf("expected first snapshot to be 'following', got %s", s.SnapshotType) + } + } +} + +func TestSnapshotRepository_FindByUserAndType(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:alice", + SnapshotType: "followers", + TotalCount: 200, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save failed: %v", err) + } + + retrieved, err := repo.FindByUserAndType(context.Background(), "did:plc:alice", "followers") + if err != nil { + t.Fatalf("FindByUserAndType failed: %v", err) + } + + if retrieved == nil { + t.Fatal("expected snapshot, got nil") + } + + if retrieved.UserDid != "did:plc:alice" { + t.Errorf("expected UserDid 'did:plc:alice', got %s", retrieved.UserDid) + } + if retrieved.SnapshotType != "followers" { + t.Errorf("expected SnapshotType 'followers', got %s", retrieved.SnapshotType) + } +} + +func TestSnapshotRepository_FindByUserAndType_NotFound(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + retrieved, err := repo.FindByUserAndType(context.Background(), "did:plc:nonexistent", "followers") + if err != nil { + t.Fatalf("FindByUserAndType failed: %v", err) + } + + if retrieved != nil { + t.Error("expected nil for nonexistent snapshot") + } +} + +func TestSnapshotRepository_FindByUserAndType_Expired(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:bob", + SnapshotType: "followers", + TotalCount: 100, + } + snapshot.SetID(GenerateUUID()) + snapshot.SetCreatedAt(time.Now().Add(-25 * time.Hour)) + snapshot.ExpiresAt = time.Now().Add(-1 * time.Hour) + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save failed: %v", err) + } + + retrieved, err := repo.FindByUserAndType(context.Background(), "did:plc:bob", "followers") + if err != nil { + t.Fatalf("FindByUserAndType failed: %v", err) + } + + if retrieved != nil { + t.Error("expected nil for expired snapshot") + } +} + +func TestSnapshotRepository_FindByUserTypeAndDate(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + oldSnapshot := &SnapshotModel{ + UserDid: "did:plc:charlie", + SnapshotType: "followers", + TotalCount: 80, + } + oldSnapshot.SetID(GenerateUUID()) + oldSnapshot.SetCreatedAt(time.Now().Add(-48 * time.Hour)) + oldSnapshot.ExpiresAt = time.Now().Add(24 * time.Hour) + + recentSnapshot := &SnapshotModel{ + UserDid: "did:plc:charlie", + SnapshotType: "followers", + TotalCount: 100, + } + recentSnapshot.SetID(GenerateUUID()) + recentSnapshot.SetCreatedAt(time.Now().Add(-12 * time.Hour)) + recentSnapshot.ExpiresAt = time.Now().Add(24 * time.Hour) + + if err := repo.Save(context.Background(), oldSnapshot); err != nil { + t.Fatalf("Save oldSnapshot failed: %v", err) + } + if err := repo.Save(context.Background(), recentSnapshot); err != nil { + t.Fatalf("Save recentSnapshot failed: %v", err) + } + + targetDate := time.Now().Add(-24 * time.Hour) + + retrieved, err := repo.FindByUserTypeAndDate(context.Background(), "did:plc:charlie", "followers", targetDate) + if err != nil { + t.Fatalf("FindByUserTypeAndDate failed: %v", err) + } + + if retrieved == nil { + t.Fatal("expected snapshot, got nil") + } + + if retrieved.TotalCount != 80 { + t.Errorf("expected TotalCount 80 (old snapshot), got %d", retrieved.TotalCount) + } +} + +func TestSnapshotRepository_SaveAndGetEntry(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:testuser", + SnapshotType: "followers", + TotalCount: 1, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save snapshot failed: %v", err) + } + + entry := &SnapshotEntry{ + SnapshotID: snapshot.ID(), + ActorDid: "did:plc:follower1", + IndexedAt: "2024-01-15T10:00:00Z", + } + + err = repo.SaveEntry(context.Background(), entry) + if err != nil { + t.Fatalf("SaveEntry failed: %v", err) + } + + entries, err := repo.GetEntries(context.Background(), snapshot.ID()) + if err != nil { + t.Fatalf("GetEntries failed: %v", err) + } + + if len(entries) != 1 { + t.Errorf("expected 1 entry, got %d", len(entries)) + } + + if entries[0].ActorDid != "did:plc:follower1" { + t.Errorf("expected ActorDid 'did:plc:follower1', got %s", entries[0].ActorDid) + } + if entries[0].IndexedAt != "2024-01-15T10:00:00Z" { + t.Errorf("expected IndexedAt '2024-01-15T10:00:00Z', got %s", entries[0].IndexedAt) + } +} + +func TestSnapshotRepository_SaveEntries_Batch(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:testuser", + SnapshotType: "followers", + TotalCount: 3, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save snapshot failed: %v", err) + } + + entries := []*SnapshotEntry{ + { + SnapshotID: snapshot.ID(), + ActorDid: "did:plc:follower1", + IndexedAt: "2024-01-15T10:00:00Z", + }, + { + SnapshotID: snapshot.ID(), + ActorDid: "did:plc:follower2", + IndexedAt: "2024-01-15T11:00:00Z", + }, + { + SnapshotID: snapshot.ID(), + ActorDid: "did:plc:follower3", + IndexedAt: "2024-01-15T12:00:00Z", + }, + } + + err = repo.SaveEntries(context.Background(), entries) + if err != nil { + t.Fatalf("SaveEntries failed: %v", err) + } + + retrieved, err := repo.GetEntries(context.Background(), snapshot.ID()) + if err != nil { + t.Fatalf("GetEntries failed: %v", err) + } + + if len(retrieved) != 3 { + t.Errorf("expected 3 entries, got %d", len(retrieved)) + } +} + +func TestSnapshotRepository_GetActorDids(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:testuser", + SnapshotType: "followers", + TotalCount: 2, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save snapshot failed: %v", err) + } + + entries := []*SnapshotEntry{ + {SnapshotID: snapshot.ID(), ActorDid: "did:plc:actor1", IndexedAt: "2024-01-15T10:00:00Z"}, + {SnapshotID: snapshot.ID(), ActorDid: "did:plc:actor2", IndexedAt: "2024-01-15T11:00:00Z"}, + } + + err = repo.SaveEntries(context.Background(), entries) + if err != nil { + t.Fatalf("SaveEntries failed: %v", err) + } + + dids, err := repo.GetActorDids(context.Background(), snapshot.ID()) + if err != nil { + t.Fatalf("GetActorDids failed: %v", err) + } + + if len(dids) != 2 { + t.Errorf("expected 2 DIDs, got %d", len(dids)) + } + + didMap := make(map[string]bool) + for _, did := range dids { + didMap[did] = true + } + + if !didMap["did:plc:actor1"] { + t.Error("expected 'did:plc:actor1' in results") + } + if !didMap["did:plc:actor2"] { + t.Error("expected 'did:plc:actor2' in results") + } +} + +func TestSnapshotRepository_Delete_CascadesEntries(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + if _, err := db.Exec("PRAGMA foreign_keys = ON"); err != nil { + t.Fatalf("failed to enable foreign keys: %v", err) + } + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:testuser", + SnapshotType: "followers", + TotalCount: 1, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save snapshot failed: %v", err) + } + + entry := &SnapshotEntry{ + SnapshotID: snapshot.ID(), + ActorDid: "did:plc:follower1", + IndexedAt: "2024-01-15T10:00:00Z", + } + + err = repo.SaveEntry(context.Background(), entry) + if err != nil { + t.Fatalf("SaveEntry failed: %v", err) + } + + err = repo.Delete(context.Background(), snapshot.ID()) + if err != nil { + t.Fatalf("Delete failed: %v", err) + } + + entries, err := repo.GetEntries(context.Background(), snapshot.ID()) + if err != nil { + t.Fatalf("GetEntries failed: %v", err) + } + + if len(entries) != 0 { + t.Errorf("expected 0 entries after cascade delete, got %d", len(entries)) + } +} + +func TestSnapshotRepository_DeleteExpiredSnapshots(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + freshSnapshot := &SnapshotModel{ + UserDid: "did:plc:user1", + SnapshotType: "followers", + TotalCount: 100, + } + freshSnapshot.SetID(GenerateUUID()) + freshSnapshot.SetCreatedAt(time.Now()) + freshSnapshot.ExpiresAt = time.Now().Add(24 * time.Hour) + + expiredSnapshot := &SnapshotModel{ + UserDid: "did:plc:user1", + SnapshotType: "following", + TotalCount: 50, + } + expiredSnapshot.SetID(GenerateUUID()) + expiredSnapshot.SetCreatedAt(time.Now().Add(-25 * time.Hour)) + expiredSnapshot.ExpiresAt = time.Now().Add(-1 * time.Hour) + + if err := repo.Save(context.Background(), freshSnapshot); err != nil { + t.Fatalf("Save freshSnapshot failed: %v", err) + } + if err := repo.Save(context.Background(), expiredSnapshot); err != nil { + t.Fatalf("Save expiredSnapshot failed: %v", err) + } + + deleted, err := repo.DeleteExpiredSnapshots(context.Background()) + if err != nil { + t.Fatalf("DeleteExpiredSnapshots failed: %v", err) + } + + if deleted != 1 { + t.Errorf("expected 1 deleted snapshot, got %d", deleted) + } + + snapshots, err := repo.List(context.Background()) + if err != nil { + t.Fatalf("List failed: %v", err) + } + + if len(snapshots) != 1 { + t.Errorf("expected 1 snapshot remaining, got %d", len(snapshots)) + } + + if s, ok := snapshots[0].(*SnapshotModel); ok { + if s.SnapshotType != "followers" { + t.Errorf("expected remaining snapshot to be 'followers', got %s", s.SnapshotType) + } + } +} + +func TestSnapshotRepository_IsFresh(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + snapshot := &SnapshotModel{ + UserDid: "did:plc:testuser", + SnapshotType: "followers", + TotalCount: 100, + } + + err := repo.Save(context.Background(), snapshot) + if err != nil { + t.Fatalf("Save failed: %v", err) + } + + if !snapshot.IsFresh() { + t.Error("newly saved snapshot should be fresh") + } + + expiredSnapshot := &SnapshotModel{ + UserDid: "did:plc:testuser2", + SnapshotType: "followers", + TotalCount: 50, + } + expiredSnapshot.SetID(GenerateUUID()) + expiredSnapshot.SetCreatedAt(time.Now().Add(-25 * time.Hour)) + expiredSnapshot.ExpiresAt = time.Now().Add(-1 * time.Hour) + + if expiredSnapshot.IsFresh() { + t.Error("expired snapshot should not be fresh") + } +} + +func TestSnapshotRepository_Close(t *testing.T) { + db, cleanup := utils.NewTestDB(t) + defer cleanup() + + repo := &SnapshotRepository{db: db} + if err := repo.Init(context.Background()); err != nil { + t.Fatalf("Init failed: %v", err) + } + + err := repo.Close() + if err != nil { + t.Errorf("Close failed: %v", err) + } +}