bluesky viewer in the terminal

feat: cache & diff command for followers

+316 -33
cli/cmd/followers.go
··· 53 } 54 55 // filterInactive filters follower infos to only include accounts inactive for N days 56 - func filterInactive(ctx context.Context, service *store.BlueskyService, followerInfos []followerInfo, actors []string, inactiveDays int, logger *log.Logger) []followerInfo { 57 logger.Infof("Checking activity status (threshold: %d days)...", inactiveDays) 58 59 - lastPostDates := service.BatchGetLastPostDates(ctx, actors, 10) 60 61 var filtered []followerInfo 62 for i, info := range followerInfos { ··· 82 } 83 84 // filterQuiet filters follower infos to only include quiet posters 85 - func filterQuiet(ctx context.Context, service *store.BlueskyService, followerInfos []followerInfo, actors []string, threshold float64, logger *log.Logger) []followerInfo { 86 - logger.Infof("Computing post rates (threshold: %.2f posts/day, this may take a while)...", threshold) 87 88 - postRates := service.BatchGetPostRates(ctx, actors, 30, 30, 10, func(current, total int) { 89 if current%10 == 0 || current == total { 90 logger.Infof("Progress: %d/%d accounts analyzed", current, total) 91 } ··· 126 return fmt.Errorf("not authenticated: run 'skycli login' first") 127 } 128 129 actor := cmd.String("user") 130 if actor == "" { 131 actor = service.GetDid() ··· 136 quietPosters := cmd.Bool("quiet") 137 quietThreshold := cmd.Float("threshold") 138 outputFormat := cmd.String("output") 139 140 if limit == 0 { 141 logger.Debugf("Fetching all followers for %v", actor) ··· 197 followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowers, logger) 198 199 if inactiveDays > 0 { 200 - followerInfos = filterInactive(ctx, service, followerInfos, actors, inactiveDays, logger) 201 } 202 203 if quietPosters { 204 - followerInfos = filterQuiet(ctx, service, followerInfos, actors, quietThreshold, logger) 205 } 206 207 switch outputFormat { ··· 231 232 if !service.Authenticated() { 233 return fmt.Errorf("not authenticated: run 'skycli login' first") 234 } 235 236 actor := cmd.String("user") ··· 242 quietPosters := cmd.Bool("quiet") 243 quietThreshold := cmd.Float("threshold") 244 outputFormat := cmd.String("output") 245 246 logger.Debugf("Fetching following for actor %v", actor) 247 ··· 282 followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowing, logger) 283 284 if inactiveDays > 0 { 285 - followerInfos = filterInactive(ctx, service, followerInfos, actors, inactiveDays, logger) 286 } 287 288 if quietPosters { 289 - followerInfos = filterQuiet(ctx, service, followerInfos, actors, quietThreshold, logger) 290 } 291 292 switch outputFormat { ··· 449 return fmt.Errorf("not authenticated: run 'skycli login' first") 450 } 451 452 sinceStr := cmd.String("since") 453 untilStr := cmd.String("until") 454 455 - if sinceStr == "" || untilStr == "" { 456 - return fmt.Errorf("both --since and --until are required") 457 } 458 459 - // TODO: Implement snapshot storage and comparison 460 - // This requires a way to store historical follower lists 461 - // Options: SQLite table, JSON files with timestamps, etc. 462 463 - ui.Infoln("Diff functionality requires snapshot storage (not yet implemented)") 464 - ui.Infoln("Consider using 'followers export' to create manual snapshots") 465 466 return nil 467 } ··· 481 482 if !service.Authenticated() { 483 return fmt.Errorf("not authenticated: run 'skycli login' first") 484 } 485 486 actor := cmd.String("user") ··· 491 quietPosters := cmd.Bool("quiet") 492 quietThreshold := cmd.Float("threshold") 493 outputFormat := cmd.String("output") 494 495 logger.Debugf("Exporting followers for actor %v with fmt %v", actor, outputFormat) 496 ··· 521 followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowers, logger) 522 523 if inactiveDays > 0 { 524 - followerInfos = filterInactive(ctx, service, followerInfos, actors, inactiveDays, logger) 525 } 526 527 if quietPosters { 528 - followerInfos = filterQuiet(ctx, service, followerInfos, actors, quietThreshold, logger) 529 } 530 531 switch outputFormat { ··· 538 } 539 } 540 541 func displayFollowersTable(followers []followerInfo, showInactive bool) { 542 if len(followers) == 0 { 543 ui.Infoln("No followers found") ··· 576 577 if showInactive && info.IsQuiet { 578 row = append(row, fmt.Sprintf("%.2f", info.PostsPerDay)) 579 - lastPostInfo := "never" 580 - if info.DaysSincePost >= 0 { 581 - lastPostInfo = fmt.Sprintf("%d days ago", info.DaysSincePost) 582 - } 583 - row = append(row, lastPostInfo) 584 } else if info.IsQuiet { 585 row = append(row, fmt.Sprintf("%.2f", info.PostsPerDay)) 586 } else if showInactive { 587 - lastPostInfo := "never" 588 - if info.DaysSincePost >= 0 { 589 - lastPostInfo = fmt.Sprintf("%d days ago", info.DaysSincePost) 590 - } 591 - row = append(row, lastPostInfo) 592 } 593 594 row = append(row, profileURL) ··· 762 Usage: "Output format: table, json, csv", 763 Value: "table", 764 }, 765 }, 766 Action: ListFollowersAction, 767 }, ··· 795 { 796 Name: "diff", 797 Usage: "Compare follower lists between two dates", 798 - UsageText: "Compare follower lists to identify new followers and unfollows. Requires snapshot storage (not yet implemented).", 799 ArgsUsage: " ", 800 Flags: []cli.Flag{ 801 &cli.StringFlag{ 802 Name: "since", 803 - Usage: "Start date (YYYY-MM-DD)", 804 Required: true, 805 }, 806 &cli.StringFlag{ 807 - Name: "until", 808 - Usage: "End date (YYYY-MM-DD)", 809 - Required: true, 810 }, 811 }, 812 Action: FollowersDiffAction, ··· 843 Value: "csv", 844 Required: true, 845 }, 846 }, 847 Action: FollowersExportAction, 848 }, ··· 890 Aliases: []string{"o"}, 891 Usage: "Output format: table, json, csv", 892 Value: "table", 893 }, 894 }, 895 Action: ListFollowingAction,
··· 53 } 54 55 // filterInactive filters follower infos to only include accounts inactive for N days 56 + func filterInactive(ctx context.Context, service *store.BlueskyService, cacheRepo *store.CacheRepository, followerInfos []followerInfo, actors []string, inactiveDays int, refresh bool, logger *log.Logger) []followerInfo { 57 logger.Infof("Checking activity status (threshold: %d days)...", inactiveDays) 58 59 + lastPostDates := service.BatchGetLastPostDatesCached(ctx, cacheRepo, actors, 10, refresh) 60 61 var filtered []followerInfo 62 for i, info := range followerInfos { ··· 82 } 83 84 // filterQuiet filters follower infos to only include quiet posters 85 + func filterQuiet(ctx context.Context, service *store.BlueskyService, cacheRepo *store.CacheRepository, followerInfos []followerInfo, actors []string, threshold float64, refresh bool, logger *log.Logger) []followerInfo { 86 + logger.Infof("Computing post rates (threshold: %.2f posts/day)...", threshold) 87 + if refresh { 88 + logger.Infof("Refreshing cache (this may take a while)...") 89 + } 90 91 + postRates := service.BatchGetPostRatesCached(ctx, cacheRepo, actors, 30, 30, 10, refresh, func(current, total int) { 92 if current%10 == 0 || current == total { 93 logger.Infof("Progress: %d/%d accounts analyzed", current, total) 94 } ··· 129 return fmt.Errorf("not authenticated: run 'skycli login' first") 130 } 131 132 + cacheRepo, err := reg.GetCacheRepo() 133 + if err != nil { 134 + return fmt.Errorf("failed to get cache repository: %w", err) 135 + } 136 + 137 actor := cmd.String("user") 138 if actor == "" { 139 actor = service.GetDid() ··· 144 quietPosters := cmd.Bool("quiet") 145 quietThreshold := cmd.Float("threshold") 146 outputFormat := cmd.String("output") 147 + refresh := cmd.Bool("refresh") 148 149 if limit == 0 { 150 logger.Debugf("Fetching all followers for %v", actor) ··· 206 followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowers, logger) 207 208 if inactiveDays > 0 { 209 + followerInfos = filterInactive(ctx, service, cacheRepo, followerInfos, actors, inactiveDays, refresh, logger) 210 } 211 212 if quietPosters { 213 + followerInfos = filterQuiet(ctx, service, cacheRepo, followerInfos, actors, quietThreshold, refresh, logger) 214 } 215 216 switch outputFormat { ··· 240 241 if !service.Authenticated() { 242 return fmt.Errorf("not authenticated: run 'skycli login' first") 243 + } 244 + 245 + cacheRepo, err := reg.GetCacheRepo() 246 + if err != nil { 247 + return fmt.Errorf("failed to get cache repository: %w", err) 248 } 249 250 actor := cmd.String("user") ··· 256 quietPosters := cmd.Bool("quiet") 257 quietThreshold := cmd.Float("threshold") 258 outputFormat := cmd.String("output") 259 + refresh := cmd.Bool("refresh") 260 261 logger.Debugf("Fetching following for actor %v", actor) 262 ··· 297 followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowing, logger) 298 299 if inactiveDays > 0 { 300 + followerInfos = filterInactive(ctx, service, cacheRepo, followerInfos, actors, inactiveDays, refresh, logger) 301 } 302 303 if quietPosters { 304 + followerInfos = filterQuiet(ctx, service, cacheRepo, followerInfos, actors, quietThreshold, refresh, logger) 305 } 306 307 switch outputFormat { ··· 464 return fmt.Errorf("not authenticated: run 'skycli login' first") 465 } 466 467 + snapshotRepo, err := reg.GetSnapshotRepo() 468 + if err != nil { 469 + return fmt.Errorf("failed to get snapshot repository: %w", err) 470 + } 471 + 472 + actor := cmd.String("user") 473 + if actor == "" { 474 + actor = service.GetDid() 475 + } 476 sinceStr := cmd.String("since") 477 untilStr := cmd.String("until") 478 + outputFormat := cmd.String("output") 479 480 + // Parse since parameter (date or snapshot ID) 481 + sinceDate, err := time.Parse("2006-01-02", sinceStr) 482 + var baselineSnapshot *store.SnapshotModel 483 + if err != nil { 484 + // Not a date, try as snapshot ID 485 + model, err := snapshotRepo.Get(ctx, sinceStr) 486 + if err != nil { 487 + return fmt.Errorf("invalid --since parameter (not a date or snapshot ID): %w", err) 488 + } 489 + if model == nil { 490 + return fmt.Errorf("snapshot not found: %s", sinceStr) 491 + } 492 + baselineSnapshot = model.(*store.SnapshotModel) 493 + } else { 494 + // Find snapshot by date 495 + baselineSnapshot, err = snapshotRepo.FindByUserTypeAndDate(ctx, actor, "followers", sinceDate) 496 + if err != nil { 497 + return fmt.Errorf("failed to find snapshot: %w", err) 498 + } 499 + if baselineSnapshot == nil { 500 + return fmt.Errorf("no snapshot found for %s on or before %s", actor, sinceStr) 501 + } 502 + } 503 + 504 + logger.Infof("Using baseline snapshot from %s (%d followers)", baselineSnapshot.CreatedAt().Format("2006-01-02 15:04"), baselineSnapshot.TotalCount) 505 + 506 + // Get baseline follower DIDs 507 + baselineDids, err := snapshotRepo.GetActorDids(ctx, baselineSnapshot.ID()) 508 + if err != nil { 509 + return fmt.Errorf("failed to get baseline followers: %w", err) 510 + } 511 + 512 + var comparisonDids []string 513 + var comparisonLabel string 514 + 515 + if untilStr != "" { 516 + // Snapshot-to-snapshot comparison 517 + untilDate, err := time.Parse("2006-01-02", untilStr) 518 + var comparisonSnapshot *store.SnapshotModel 519 + if err != nil { 520 + // Not a date, try as snapshot ID 521 + model, err := snapshotRepo.Get(ctx, untilStr) 522 + if err != nil { 523 + return fmt.Errorf("invalid --until parameter (not a date or snapshot ID): %w", err) 524 + } 525 + if model == nil { 526 + return fmt.Errorf("snapshot not found: %s", untilStr) 527 + } 528 + comparisonSnapshot = model.(*store.SnapshotModel) 529 + } else { 530 + // Find snapshot by date 531 + comparisonSnapshot, err = snapshotRepo.FindByUserTypeAndDate(ctx, actor, "followers", untilDate) 532 + if err != nil { 533 + return fmt.Errorf("failed to find snapshot: %w", err) 534 + } 535 + if comparisonSnapshot == nil { 536 + return fmt.Errorf("no snapshot found for %s on or before %s", actor, untilStr) 537 + } 538 + } 539 + 540 + logger.Infof("Comparing with snapshot from %s (%d followers)", comparisonSnapshot.CreatedAt().Format("2006-01-02 15:04"), comparisonSnapshot.TotalCount) 541 + comparisonLabel = comparisonSnapshot.CreatedAt().Format("2006-01-02 15:04") 542 + 543 + comparisonDids, err = snapshotRepo.GetActorDids(ctx, comparisonSnapshot.ID()) 544 + if err != nil { 545 + return fmt.Errorf("failed to get comparison followers: %w", err) 546 + } 547 + } else { 548 + // Snapshot-to-live comparison 549 + logger.Infof("Fetching current followers for comparison...") 550 + comparisonLabel = "now" 551 + 552 + var allFollowers []store.ActorProfile 553 + cursor := "" 554 + page := 0 555 + for { 556 + page++ 557 + response, err := service.GetFollowers(ctx, actor, 100, cursor) 558 + if err != nil { 559 + return fmt.Errorf("failed to fetch followers: %w", err) 560 + } 561 + 562 + allFollowers = append(allFollowers, response.Followers...) 563 + 564 + if response.Cursor != "" { 565 + logger.Infof("Fetched page %d (%d followers so far)...", page, len(allFollowers)) 566 + } 567 + 568 + if response.Cursor == "" { 569 + break 570 + } 571 + cursor = response.Cursor 572 + } 573 + 574 + logger.Infof("Fetched %d current followers", len(allFollowers)) 575 + 576 + for _, follower := range allFollowers { 577 + comparisonDids = append(comparisonDids, follower.Did) 578 + } 579 + } 580 + 581 + // Calculate diff 582 + baselineSet := make(map[string]bool) 583 + for _, did := range baselineDids { 584 + baselineSet[did] = true 585 + } 586 + 587 + comparisonSet := make(map[string]bool) 588 + for _, did := range comparisonDids { 589 + comparisonSet[did] = true 590 + } 591 + 592 + // New followers: in comparison but not in baseline 593 + var newFollowers []string 594 + for _, did := range comparisonDids { 595 + if !baselineSet[did] { 596 + newFollowers = append(newFollowers, did) 597 + } 598 + } 599 + 600 + // Unfollows: in baseline but not in comparison 601 + var unfollows []string 602 + for _, did := range baselineDids { 603 + if !comparisonSet[did] { 604 + unfollows = append(unfollows, did) 605 + } 606 + } 607 + 608 + // Output results 609 + switch outputFormat { 610 + case "json": 611 + return outputDiffJSON(newFollowers, unfollows) 612 + case "csv": 613 + return outputDiffCSV(newFollowers, unfollows) 614 + default: 615 + displayDiffTable(baselineSnapshot.CreatedAt().Format("2006-01-02 15:04"), comparisonLabel, len(baselineDids), len(comparisonDids), newFollowers, unfollows) 616 + } 617 + 618 + return nil 619 + } 620 + 621 + func displayDiffTable(baselineLabel, comparisonLabel string, baselineCount, comparisonCount int, newFollowers, unfollows []string) { 622 + ui.Titleln("Follower Diff: %s → %s", baselineLabel, comparisonLabel) 623 + fmt.Println() 624 + 625 + fmt.Printf("Baseline: %d followers\n", baselineCount) 626 + fmt.Printf("Comparison: %d followers\n", comparisonCount) 627 + fmt.Printf("Net change: %+d\n", comparisonCount-baselineCount) 628 + fmt.Println() 629 + 630 + if len(newFollowers) > 0 { 631 + ui.Titleln("New Followers (%d)", len(newFollowers)) 632 + for _, did := range newFollowers { 633 + fmt.Printf(" + %s\n", did) 634 + } 635 + fmt.Println() 636 + } 637 + 638 + if len(unfollows) > 0 { 639 + ui.Titleln("Unfollows (%d)", len(unfollows)) 640 + for _, did := range unfollows { 641 + fmt.Printf(" - %s\n", did) 642 + } 643 + fmt.Println() 644 + } 645 + 646 + if len(newFollowers) == 0 && len(unfollows) == 0 { 647 + ui.Infoln("No changes detected") 648 + } 649 + } 650 + 651 + type diffOutput struct { 652 + NewFollowers []string `json:"newFollowers"` 653 + Unfollows []string `json:"unfollows"` 654 + Summary struct { 655 + BaselineCount int `json:"baselineCount"` 656 + ComparisonCount int `json:"comparisonCount"` 657 + NetChange int `json:"netChange"` 658 + NewCount int `json:"newCount"` 659 + UnfollowCount int `json:"unfollowCount"` 660 + } `json:"summary"` 661 + } 662 + 663 + func outputDiffJSON(newFollowers, unfollows []string) error { 664 + output := diffOutput{ 665 + NewFollowers: newFollowers, 666 + Unfollows: unfollows, 667 } 668 + if output.NewFollowers == nil { 669 + output.NewFollowers = []string{} 670 + } 671 + if output.Unfollows == nil { 672 + output.Unfollows = []string{} 673 + } 674 + output.Summary.NewCount = len(newFollowers) 675 + output.Summary.UnfollowCount = len(unfollows) 676 677 + encoder := json.NewEncoder(os.Stdout) 678 + encoder.SetIndent("", " ") 679 + return encoder.Encode(output) 680 + } 681 + 682 + func outputDiffCSV(newFollowers, unfollows []string) error { 683 + writer := csv.NewWriter(os.Stdout) 684 + defer writer.Flush() 685 + 686 + if err := writer.Write([]string{"type", "did"}); err != nil { 687 + return err 688 + } 689 690 + for _, did := range newFollowers { 691 + if err := writer.Write([]string{"new_follower", did}); err != nil { 692 + return err 693 + } 694 + } 695 + 696 + for _, did := range unfollows { 697 + if err := writer.Write([]string{"unfollow", did}); err != nil { 698 + return err 699 + } 700 + } 701 702 return nil 703 } ··· 717 718 if !service.Authenticated() { 719 return fmt.Errorf("not authenticated: run 'skycli login' first") 720 + } 721 + 722 + cacheRepo, err := reg.GetCacheRepo() 723 + if err != nil { 724 + return fmt.Errorf("failed to get cache repository: %w", err) 725 } 726 727 actor := cmd.String("user") ··· 732 quietPosters := cmd.Bool("quiet") 733 quietThreshold := cmd.Float("threshold") 734 outputFormat := cmd.String("output") 735 + refresh := cmd.Bool("refresh") 736 737 logger.Debugf("Exporting followers for actor %v with fmt %v", actor, outputFormat) 738 ··· 763 followerInfos, actors := enrichFollowerProfiles(ctx, service, allFollowers, logger) 764 765 if inactiveDays > 0 { 766 + followerInfos = filterInactive(ctx, service, cacheRepo, followerInfos, actors, inactiveDays, refresh, logger) 767 } 768 769 if quietPosters { 770 + followerInfos = filterQuiet(ctx, service, cacheRepo, followerInfos, actors, quietThreshold, refresh, logger) 771 } 772 773 switch outputFormat { ··· 780 } 781 } 782 783 + // formatTimeSince formats a time duration into a human-readable string. 784 + // 785 + // Returns 786 + // - "< 1 hour ago" for durations under 1 hour 787 + // - "X hours ago" for under 24 hours 788 + // - "X days ago" for longer durations. 789 + func formatTimeSince(since time.Time) string { 790 + if since.IsZero() { 791 + return "never" 792 + } 793 + 794 + duration := time.Since(since) 795 + hours := duration.Hours() 796 + 797 + if hours < 1 { 798 + return "< 1 hour ago" 799 + } else if hours < 24 { 800 + return fmt.Sprintf("%d hours ago", int(hours)) 801 + } else { 802 + days := int(hours / 24) 803 + if days == 1 { 804 + return "1 day ago" 805 + } 806 + return fmt.Sprintf("%d days ago", days) 807 + } 808 + } 809 + 810 func displayFollowersTable(followers []followerInfo, showInactive bool) { 811 if len(followers) == 0 { 812 ui.Infoln("No followers found") ··· 845 846 if showInactive && info.IsQuiet { 847 row = append(row, fmt.Sprintf("%.2f", info.PostsPerDay)) 848 + row = append(row, formatTimeSince(info.LastPostDate)) 849 } else if info.IsQuiet { 850 row = append(row, fmt.Sprintf("%.2f", info.PostsPerDay)) 851 } else if showInactive { 852 + row = append(row, formatTimeSince(info.LastPostDate)) 853 } 854 855 row = append(row, profileURL) ··· 1023 Usage: "Output format: table, json, csv", 1024 Value: "table", 1025 }, 1026 + &cli.BoolFlag{ 1027 + Name: "refresh", 1028 + Usage: "Force refresh cached data (bypasses 24-hour cache)", 1029 + }, 1030 }, 1031 Action: ListFollowersAction, 1032 }, ··· 1060 { 1061 Name: "diff", 1062 Usage: "Compare follower lists between two dates", 1063 + UsageText: "Compare follower lists to identify new followers and unfollows. Without --until, compares snapshot to current live data.", 1064 ArgsUsage: " ", 1065 Flags: []cli.Flag{ 1066 &cli.StringFlag{ 1067 + Name: "user", 1068 + Aliases: []string{"u"}, 1069 + Usage: "User handle or DID (defaults to authenticated user)", 1070 + }, 1071 + &cli.StringFlag{ 1072 Name: "since", 1073 + Usage: "Start date (YYYY-MM-DD) or snapshot ID", 1074 Required: true, 1075 }, 1076 &cli.StringFlag{ 1077 + Name: "until", 1078 + Usage: "End date (YYYY-MM-DD) or snapshot ID (omit to compare with live data)", 1079 + }, 1080 + &cli.StringFlag{ 1081 + Name: "output", 1082 + Aliases: []string{"o"}, 1083 + Usage: "Output format: table, json, csv", 1084 + Value: "table", 1085 }, 1086 }, 1087 Action: FollowersDiffAction, ··· 1118 Value: "csv", 1119 Required: true, 1120 }, 1121 + &cli.BoolFlag{ 1122 + Name: "refresh", 1123 + Usage: "Force refresh cached data (bypasses 24-hour cache)", 1124 + }, 1125 }, 1126 Action: FollowersExportAction, 1127 }, ··· 1169 Aliases: []string{"o"}, 1170 Usage: "Output format: table, json, csv", 1171 Value: "table", 1172 + }, 1173 + &cli.BoolFlag{ 1174 + Name: "refresh", 1175 + Usage: "Force refresh cached data (bypasses 24-hour cache)", 1176 }, 1177 }, 1178 Action: ListFollowingAction,
+71 -7
cli/internal/registry/registry.go
··· 15 16 // Registry manages singleton instances of repositories and services 17 type Registry struct { 18 - service *store.BlueskyService 19 - sessionRepo *store.SessionRepository 20 - feedRepo *store.FeedRepository 21 - postRepo *store.PostRepository 22 - profileRepo *store.ProfileRepository 23 - initialized bool 24 - mu sync.RWMutex 25 } 26 27 // Get returns the singleton registry instance ··· 80 } 81 r.profileRepo = profileRepo 82 83 r.service = store.NewBlueskyService("") 84 85 if sessionRepo.HasValidSession(ctx) { ··· 138 } 139 } 140 141 r.initialized = false 142 143 if len(errs) > 0 { ··· 225 } 226 227 return r.profileRepo, nil 228 } 229 230 // IsInitialized returns whether the registry has been initialized
··· 15 16 // Registry manages singleton instances of repositories and services 17 type Registry struct { 18 + service *store.BlueskyService 19 + sessionRepo *store.SessionRepository 20 + feedRepo *store.FeedRepository 21 + postRepo *store.PostRepository 22 + profileRepo *store.ProfileRepository 23 + snapshotRepo *store.SnapshotRepository 24 + cacheRepo *store.CacheRepository 25 + initialized bool 26 + mu sync.RWMutex 27 } 28 29 // Get returns the singleton registry instance ··· 82 } 83 r.profileRepo = profileRepo 84 85 + snapshotRepo, err := store.NewSnapshotRepository() 86 + if err != nil { 87 + return &RegistryError{Op: "InitSnapshotRepo", Err: err} 88 + } 89 + if err := snapshotRepo.Init(ctx); err != nil { 90 + return &RegistryError{Op: "InitSnapshotRepo", Err: err} 91 + } 92 + r.snapshotRepo = snapshotRepo 93 + 94 + cacheRepo, err := store.NewCacheRepository() 95 + if err != nil { 96 + return &RegistryError{Op: "InitCacheRepo", Err: err} 97 + } 98 + if err := cacheRepo.Init(ctx); err != nil { 99 + return &RegistryError{Op: "InitCacheRepo", Err: err} 100 + } 101 + r.cacheRepo = cacheRepo 102 + 103 r.service = store.NewBlueskyService("") 104 105 if sessionRepo.HasValidSession(ctx) { ··· 158 } 159 } 160 161 + if r.snapshotRepo != nil { 162 + if err := r.snapshotRepo.Close(); err != nil { 163 + errs = append(errs, err) 164 + } 165 + } 166 + 167 + if r.cacheRepo != nil { 168 + if err := r.cacheRepo.Close(); err != nil { 169 + errs = append(errs, err) 170 + } 171 + } 172 + 173 r.initialized = false 174 175 if len(errs) > 0 { ··· 257 } 258 259 return r.profileRepo, nil 260 + } 261 + 262 + // GetSnapshotRepo returns the SnapshotRepository singleton 263 + func (r *Registry) GetSnapshotRepo() (*store.SnapshotRepository, error) { 264 + r.mu.RLock() 265 + defer r.mu.RUnlock() 266 + 267 + if !r.initialized { 268 + return nil, &RegistryError{Op: "GetSnapshotRepo", Err: errors.New("registry not initialized")} 269 + } 270 + 271 + if r.snapshotRepo == nil { 272 + return nil, &RegistryError{Op: "GetSnapshotRepo", Err: errors.New("snapshot repository not available")} 273 + } 274 + 275 + return r.snapshotRepo, nil 276 + } 277 + 278 + // GetCacheRepo returns the CacheRepository singleton 279 + func (r *Registry) GetCacheRepo() (*store.CacheRepository, error) { 280 + r.mu.RLock() 281 + defer r.mu.RUnlock() 282 + 283 + if !r.initialized { 284 + return nil, &RegistryError{Op: "GetCacheRepo", Err: errors.New("registry not initialized")} 285 + } 286 + 287 + if r.cacheRepo == nil { 288 + return nil, &RegistryError{Op: "GetCacheRepo", Err: errors.New("cache repository not available")} 289 + } 290 + 291 + return r.cacheRepo, nil 292 } 293 294 // IsInitialized returns whether the registry has been initialized
+124 -11
cli/internal/store/bluesky.go
··· 8 "errors" 9 "fmt" 10 "io" 11 "net/http" 12 "strings" 13 "sync" 14 "time" 15 ) 16 17 const ( ··· 367 } 368 369 // SearchActors searches for actors (users) matching the query string. 370 - // Returns actor profiles with pagination support. 371 func (s *BlueskyService) SearchActors(ctx context.Context, query string, limit int, cursor string) (*SearchActorsResponse, error) { 372 urlPath := fmt.Sprintf("/xrpc/app.bsky.actor.searchActors?q=%s&limit=%d", strings.ReplaceAll(query, " ", "+"), limit) 373 if cursor != "" { ··· 465 } 466 467 // GetLastPostDate fetches the most recent post date for an actor. 468 - // Returns zero time if the actor has no posts or if an error occurs. 469 func (s *BlueskyService) GetLastPostDate(ctx context.Context, actor string) (time.Time, error) { 470 feed, err := s.GetAuthorFeed(ctx, actor, 1, "") 471 if err != nil { ··· 485 return lastPost, nil 486 } 487 488 - // BatchGetLastPostDates fetches last post dates for multiple actors concurrently. 489 // Uses a semaphore to limit concurrent requests to maxConcurrent. 490 - // Returns a map of actor DID/handle to their last post date. 491 func (s *BlueskyService) BatchGetLastPostDates(ctx context.Context, actors []string, maxConcurrent int) map[string]time.Time { 492 results := make(map[string]time.Time) 493 resultsMu := &sync.Mutex{} ··· 517 return results 518 } 519 520 - // BatchGetProfiles fetches full profiles for multiple actors concurrently. 521 - // Uses a semaphore to limit concurrent requests to maxConcurrent. 522 - // Returns a map of actor DID/handle to their full ActorProfile. 523 func (s *BlueskyService) BatchGetProfiles(ctx context.Context, actors []string, maxConcurrent int) map[string]*ActorProfile { 524 results := make(map[string]*ActorProfile) 525 resultsMu := &sync.Mutex{} ··· 556 SampleSize int 557 } 558 559 - // BatchGetPostRates calculates posting rates for multiple actors concurrently. 560 // Samples recent posts from each actor and calculates posts per day over the lookback period. 561 // Uses a semaphore to limit concurrent requests to maxConcurrent. 562 - // Returns a map of actor DID/handle to their PostRate metrics. 563 func (s *BlueskyService) BatchGetPostRates(ctx context.Context, actors []string, sampleSize int, lookbackDays int, maxConcurrent int, progressFn func(current, total int)) map[string]*PostRate { 564 results := make(map[string]*PostRate) 565 resultsMu := &sync.Mutex{} ··· 601 return 602 } 603 604 - // Get the last post date 605 lastPost, err := time.Parse(time.RFC3339, feed.Feed[0].Post.IndexedAt) 606 if err != nil { 607 return 608 } 609 610 - // Filter posts within lookback window 611 cutoffTime := time.Now().AddDate(0, 0, -lookbackDays) 612 recentPosts := 0 613 for _, post := range feed.Feed { ··· 744 745 return time.Unix(claims.Exp, 0), nil 746 }
··· 8 "errors" 9 "fmt" 10 "io" 11 + "maps" 12 "net/http" 13 "strings" 14 "sync" 15 "time" 16 + 17 + "github.com/charmbracelet/log" 18 ) 19 20 const ( ··· 370 } 371 372 // SearchActors searches for actors (users) matching the query string. 373 func (s *BlueskyService) SearchActors(ctx context.Context, query string, limit int, cursor string) (*SearchActorsResponse, error) { 374 urlPath := fmt.Sprintf("/xrpc/app.bsky.actor.searchActors?q=%s&limit=%d", strings.ReplaceAll(query, " ", "+"), limit) 375 if cursor != "" { ··· 467 } 468 469 // GetLastPostDate fetches the most recent post date for an actor. 470 func (s *BlueskyService) GetLastPostDate(ctx context.Context, actor string) (time.Time, error) { 471 feed, err := s.GetAuthorFeed(ctx, actor, 1, "") 472 if err != nil { ··· 486 return lastPost, nil 487 } 488 489 + // BatchGetLastPostDates fetches last post dates for multiple actors concurrently, as a map of actor DID/handle to their last post date.. 490 // Uses a semaphore to limit concurrent requests to maxConcurrent. 491 func (s *BlueskyService) BatchGetLastPostDates(ctx context.Context, actors []string, maxConcurrent int) map[string]time.Time { 492 results := make(map[string]time.Time) 493 resultsMu := &sync.Mutex{} ··· 517 return results 518 } 519 520 + // BatchGetProfiles fetches full profiles for multiple actors concurrently, as a map of actor DID/handle to their full ActorProfile. 521 + // Uses a semaphore to limit concurrent requests to maxConcurrent.. 522 func (s *BlueskyService) BatchGetProfiles(ctx context.Context, actors []string, maxConcurrent int) map[string]*ActorProfile { 523 results := make(map[string]*ActorProfile) 524 resultsMu := &sync.Mutex{} ··· 555 SampleSize int 556 } 557 558 + // BatchGetPostRates calculates posting rates for multiple actors concurrently, as a map of actor DID/handle to their [PostRate] metrics. 559 + // 560 // Samples recent posts from each actor and calculates posts per day over the lookback period. 561 // Uses a semaphore to limit concurrent requests to maxConcurrent. 562 func (s *BlueskyService) BatchGetPostRates(ctx context.Context, actors []string, sampleSize int, lookbackDays int, maxConcurrent int, progressFn func(current, total int)) map[string]*PostRate { 563 results := make(map[string]*PostRate) 564 resultsMu := &sync.Mutex{} ··· 600 return 601 } 602 603 lastPost, err := time.Parse(time.RFC3339, feed.Feed[0].Post.IndexedAt) 604 if err != nil { 605 return 606 } 607 608 cutoffTime := time.Now().AddDate(0, 0, -lookbackDays) 609 recentPosts := 0 610 for _, post := range feed.Feed { ··· 741 742 return time.Unix(claims.Exp, 0), nil 743 } 744 + 745 + // BatchGetPostRatesCached calculates posting rates for multiple actors with caching support. 746 + // 747 + // Checks cache first, falls back to API for cache misses, and saves results to cache. 748 + // If refresh is true, bypasses cache and refetches all data from API. 749 + // 750 + // TODO: Implement per-item TTL for more efficient cache invalidation. 751 + // FIXME: this function signature is ridiculous 752 + 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 { 753 + results := make(map[string]*PostRate) 754 + 755 + // If not refreshing, try to load from cache 756 + var actorsToFetch []string 757 + if !refresh { 758 + cached, err := cacheRepo.GetPostRates(ctx, actors) 759 + if err == nil { 760 + for _, actor := range actors { 761 + if cache, ok := cached[actor]; ok && cache.IsFresh() { 762 + results[actor] = &PostRate{ 763 + PostsPerDay: cache.PostsPerDay, 764 + LastPostDate: cache.LastPostDate, 765 + SampleSize: cache.SampleSize, 766 + } 767 + } else { 768 + actorsToFetch = append(actorsToFetch, actor) 769 + } 770 + } 771 + } else { 772 + actorsToFetch = actors 773 + } 774 + } else { 775 + actorsToFetch = actors 776 + } 777 + 778 + if len(actorsToFetch) > 0 { 779 + apiResults := s.BatchGetPostRates(ctx, actorsToFetch, sampleSize, lookbackDays, maxConcurrent, progressFn) 780 + maps.Copy(results, apiResults) 781 + 782 + var cacheModels []*PostRateCacheModel 783 + for actor, postRate := range apiResults { 784 + cacheModels = append(cacheModels, &PostRateCacheModel{ 785 + ActorDid: actor, 786 + PostsPerDay: postRate.PostsPerDay, 787 + LastPostDate: postRate.LastPostDate, 788 + SampleSize: postRate.SampleSize, 789 + }) 790 + } 791 + 792 + if len(cacheModels) > 0 { 793 + if err := cacheRepo.SavePostRates(ctx, cacheModels); err != nil { 794 + // Log error but don't fail - cache save is non-critical 795 + } 796 + } 797 + } 798 + 799 + return results 800 + } 801 + 802 + // BatchGetLastPostDatesCached fetches last post dates for multiple actors with caching support. 803 + // 804 + // Checks cache first, falls back to API for cache misses, and saves results to cache. 805 + // If refresh is true, bypasses cache and refetches all data from API. 806 + // 807 + // TODO: Implement per-item TTL for more efficient cache invalidation. 808 + // FIXME: this function signature is ridiculous 809 + func (s *BlueskyService) BatchGetLastPostDatesCached(ctx context.Context, cacheRepo *CacheRepository, actors []string, maxConcurrent int, refresh bool) map[string]time.Time { 810 + results := make(map[string]time.Time) 811 + 812 + var actorsToFetch []string 813 + if !refresh { 814 + cached, err := cacheRepo.GetActivities(ctx, actors) 815 + if err == nil { 816 + for _, actor := range actors { 817 + if cache, ok := cached[actor]; ok && cache.IsFresh() { 818 + if cache.HasPosted() { 819 + results[actor] = cache.LastPostDate 820 + } 821 + } else { 822 + actorsToFetch = append(actorsToFetch, actor) 823 + } 824 + } 825 + } else { 826 + actorsToFetch = actors 827 + } 828 + } else { 829 + actorsToFetch = actors 830 + } 831 + 832 + if len(actorsToFetch) > 0 { 833 + apiResults := s.BatchGetLastPostDates(ctx, actorsToFetch, maxConcurrent) 834 + maps.Copy(results, apiResults) 835 + 836 + var cacheModels []*ActivityCacheModel 837 + for _, actor := range actorsToFetch { 838 + lastPostDate, hasPosted := apiResults[actor] 839 + cacheModels = append(cacheModels, &ActivityCacheModel{ 840 + ActorDid: actor, 841 + LastPostDate: lastPostDate, 842 + FetchedAt: time.Now(), 843 + ExpiresAt: time.Now().Add(24 * time.Hour), 844 + }) 845 + 846 + if !hasPosted { 847 + results[actor] = time.Time{} 848 + } 849 + } 850 + 851 + if len(cacheModels) > 0 { 852 + if err := cacheRepo.SaveActivities(ctx, cacheModels); err != nil { 853 + log.Warnf("save failed with error %v", err.Error()) 854 + } 855 + } 856 + } 857 + 858 + return results 859 + }
+41
cli/internal/store/cache_model.go
···
··· 1 + package store 2 + 3 + import "time" 4 + 5 + // PostRateCacheModel represents a cached post rate computation for an actor. 6 + // Stores expensive post rate calculations with TTL support (24 hours default). 7 + type PostRateCacheModel struct { 8 + ActorDid string 9 + PostsPerDay float64 10 + LastPostDate time.Time 11 + SampleSize int 12 + FetchedAt time.Time 13 + ExpiresAt time.Time 14 + } 15 + 16 + // IsFresh returns true if the cached post rate has not expired. 17 + // Post rates expire after 24 hours by default. 18 + func (m *PostRateCacheModel) IsFresh() bool { 19 + return time.Now().Before(m.ExpiresAt) 20 + } 21 + 22 + // ActivityCacheModel represents cached activity data (last post date) for an actor. 23 + // Stores last post date lookups with TTL support (24 hours default). 24 + type ActivityCacheModel struct { 25 + ActorDid string 26 + LastPostDate time.Time // May be zero if actor has never posted 27 + FetchedAt time.Time 28 + ExpiresAt time.Time 29 + } 30 + 31 + // IsFresh returns true if the cached activity data has not expired. 32 + // Activity data expires after 24 hours by default. 33 + func (m *ActivityCacheModel) IsFresh() bool { 34 + return time.Now().Before(m.ExpiresAt) 35 + } 36 + 37 + // HasPosted returns true if the actor has posted at least once. 38 + // A zero LastPostDate indicates the actor has never posted. 39 + func (m *ActivityCacheModel) HasPosted() bool { 40 + return !m.LastPostDate.IsZero() 41 + }
+475
cli/internal/store/cache_repo.go
···
··· 1 + package store 2 + 3 + import ( 4 + "context" 5 + "database/sql" 6 + "errors" 7 + "time" 8 + 9 + _ "github.com/mattn/go-sqlite3" 10 + "github.com/stormlightlabs/skypanel/cli/internal/config" 11 + ) 12 + 13 + // CacheRepository manages post rate and activity caches using SQLite. 14 + // 15 + // Provides methods for storing and retrieving expensive computation results stored as [PostRateCacheModel] or [ActivityCacheModel]. 16 + type CacheRepository struct { 17 + db *sql.DB 18 + } 19 + 20 + // NewCacheRepository creates a new cache repository with SQLite backend 21 + func NewCacheRepository() (*CacheRepository, error) { 22 + dbPath, err := config.GetCacheDB() 23 + if err != nil { 24 + return nil, err 25 + } 26 + 27 + db, err := sql.Open("sqlite3", dbPath) 28 + if err != nil { 29 + return nil, err 30 + } 31 + 32 + return &CacheRepository{db: db}, nil 33 + } 34 + 35 + // Init ensures database schema is initialized via migrations 36 + func (r *CacheRepository) Init(ctx context.Context) error { 37 + if err := config.EnsureConfigDir(); err != nil { 38 + return err 39 + } 40 + return RunMigrations(r.db) 41 + } 42 + 43 + // Close releases database connection 44 + func (r *CacheRepository) Close() error { 45 + return r.db.Close() 46 + } 47 + 48 + // GetPostRate retrieves cached post rate for an actor 49 + func (r *CacheRepository) GetPostRate(ctx context.Context, actorDid string) (*PostRateCacheModel, error) { 50 + query := ` 51 + SELECT actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at 52 + FROM cached_post_rates 53 + WHERE actor_did = ? AND expires_at > ? 54 + ` 55 + 56 + var cache PostRateCacheModel 57 + var lastPostDate sql.NullTime 58 + 59 + err := r.db.QueryRowContext(ctx, query, actorDid, time.Now()).Scan( 60 + &cache.ActorDid, 61 + &cache.PostsPerDay, 62 + &lastPostDate, 63 + &cache.SampleSize, 64 + &cache.FetchedAt, 65 + &cache.ExpiresAt, 66 + ) 67 + 68 + if lastPostDate.Valid { 69 + cache.LastPostDate = lastPostDate.Time 70 + } 71 + 72 + if err != nil { 73 + if errors.Is(err, sql.ErrNoRows) { 74 + return nil, nil 75 + } 76 + return nil, &RepositoryError{Op: "GetPostRate", Err: err} 77 + } 78 + 79 + return &cache, nil 80 + } 81 + 82 + // GetPostRates retrieves cached post rates for multiple actors in a single query, 83 + // as a map of actorDid -> PostRateCacheModel for found entries. 84 + func (r *CacheRepository) GetPostRates(ctx context.Context, actorDids []string) (map[string]*PostRateCacheModel, error) { 85 + if len(actorDids) == 0 { 86 + return make(map[string]*PostRateCacheModel), nil 87 + } 88 + 89 + query := ` 90 + SELECT actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at 91 + FROM cached_post_rates 92 + WHERE actor_did IN (` + buildPlaceholders(len(actorDids)) + `) AND expires_at > ? 93 + ` 94 + 95 + args := make([]interface{}, len(actorDids)+1) 96 + for i, did := range actorDids { 97 + args[i] = did 98 + } 99 + args[len(actorDids)] = time.Now() 100 + 101 + rows, err := r.db.QueryContext(ctx, query, args...) 102 + if err != nil { 103 + return nil, &RepositoryError{Op: "GetPostRates", Err: err} 104 + } 105 + defer rows.Close() 106 + 107 + result := make(map[string]*PostRateCacheModel) 108 + for rows.Next() { 109 + var cache PostRateCacheModel 110 + var lastPostDate sql.NullTime 111 + 112 + err := rows.Scan( 113 + &cache.ActorDid, 114 + &cache.PostsPerDay, 115 + &lastPostDate, 116 + &cache.SampleSize, 117 + &cache.FetchedAt, 118 + &cache.ExpiresAt, 119 + ) 120 + if err != nil { 121 + return nil, &RepositoryError{Op: "GetPostRates", Err: err} 122 + } 123 + 124 + if lastPostDate.Valid { 125 + cache.LastPostDate = lastPostDate.Time 126 + } 127 + 128 + result[cache.ActorDid] = &cache 129 + } 130 + 131 + return result, rows.Err() 132 + } 133 + 134 + // SavePostRate saves or updates a post rate cache entry 135 + func (r *CacheRepository) SavePostRate(ctx context.Context, cache *PostRateCacheModel) error { 136 + if cache.FetchedAt.IsZero() { 137 + cache.FetchedAt = time.Now() 138 + } 139 + if cache.ExpiresAt.IsZero() { 140 + cache.ExpiresAt = time.Now().Add(24 * time.Hour) 141 + } 142 + 143 + query := ` 144 + INSERT INTO cached_post_rates (actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at) 145 + VALUES (?, ?, ?, ?, ?, ?) 146 + ON CONFLICT(actor_did) DO UPDATE SET 147 + posts_per_day = excluded.posts_per_day, 148 + last_post_date = excluded.last_post_date, 149 + sample_size = excluded.sample_size, 150 + fetched_at = excluded.fetched_at, 151 + expires_at = excluded.expires_at 152 + ` 153 + 154 + var lastPostDate interface{} 155 + if !cache.LastPostDate.IsZero() { 156 + lastPostDate = cache.LastPostDate 157 + } 158 + 159 + _, err := r.db.ExecContext(ctx, query, 160 + cache.ActorDid, 161 + cache.PostsPerDay, 162 + lastPostDate, 163 + cache.SampleSize, 164 + cache.FetchedAt, 165 + cache.ExpiresAt, 166 + ) 167 + 168 + if err != nil { 169 + return &RepositoryError{Op: "SavePostRate", Err: err} 170 + } 171 + 172 + return nil 173 + } 174 + 175 + // SavePostRates saves multiple post rate cache entries in a transaction 176 + func (r *CacheRepository) SavePostRates(ctx context.Context, caches []*PostRateCacheModel) error { 177 + if len(caches) == 0 { 178 + return nil 179 + } 180 + 181 + tx, err := r.db.BeginTx(ctx, nil) 182 + if err != nil { 183 + return &RepositoryError{Op: "SavePostRates", Err: err} 184 + } 185 + defer tx.Rollback() 186 + 187 + stmt, err := tx.PrepareContext(ctx, ` 188 + INSERT INTO cached_post_rates (actor_did, posts_per_day, last_post_date, sample_size, fetched_at, expires_at) 189 + VALUES (?, ?, ?, ?, ?, ?) 190 + ON CONFLICT(actor_did) DO UPDATE SET 191 + posts_per_day = excluded.posts_per_day, 192 + last_post_date = excluded.last_post_date, 193 + sample_size = excluded.sample_size, 194 + fetched_at = excluded.fetched_at, 195 + expires_at = excluded.expires_at 196 + `) 197 + if err != nil { 198 + return &RepositoryError{Op: "SavePostRates", Err: err} 199 + } 200 + defer stmt.Close() 201 + 202 + for _, cache := range caches { 203 + if cache.FetchedAt.IsZero() { 204 + cache.FetchedAt = time.Now() 205 + } 206 + if cache.ExpiresAt.IsZero() { 207 + cache.ExpiresAt = time.Now().Add(24 * time.Hour) 208 + } 209 + 210 + var lastPostDate interface{} 211 + if !cache.LastPostDate.IsZero() { 212 + lastPostDate = cache.LastPostDate 213 + } 214 + 215 + _, err := stmt.ExecContext(ctx, 216 + cache.ActorDid, 217 + cache.PostsPerDay, 218 + lastPostDate, 219 + cache.SampleSize, 220 + cache.FetchedAt, 221 + cache.ExpiresAt, 222 + ) 223 + if err != nil { 224 + return &RepositoryError{Op: "SavePostRates", Err: err} 225 + } 226 + } 227 + 228 + if err := tx.Commit(); err != nil { 229 + return &RepositoryError{Op: "SavePostRates", Err: err} 230 + } 231 + 232 + return nil 233 + } 234 + 235 + // DeletePostRate removes a post rate cache entry 236 + func (r *CacheRepository) DeletePostRate(ctx context.Context, actorDid string) error { 237 + query := "DELETE FROM cached_post_rates WHERE actor_did = ?" 238 + _, err := r.db.ExecContext(ctx, query, actorDid) 239 + if err != nil { 240 + return &RepositoryError{Op: "DeletePostRate", Err: err} 241 + } 242 + return nil 243 + } 244 + 245 + // GetActivity retrieves cached activity data for an actor 246 + func (r *CacheRepository) GetActivity(ctx context.Context, actorDid string) (*ActivityCacheModel, error) { 247 + query := ` 248 + SELECT actor_did, last_post_date, fetched_at, expires_at 249 + FROM cached_activity 250 + WHERE actor_did = ? AND expires_at > ? 251 + ` 252 + 253 + var cache ActivityCacheModel 254 + var lastPostDate sql.NullTime 255 + 256 + err := r.db.QueryRowContext(ctx, query, actorDid, time.Now()).Scan( 257 + &cache.ActorDid, 258 + &lastPostDate, 259 + &cache.FetchedAt, 260 + &cache.ExpiresAt, 261 + ) 262 + 263 + if lastPostDate.Valid { 264 + cache.LastPostDate = lastPostDate.Time 265 + } 266 + 267 + if err != nil { 268 + if errors.Is(err, sql.ErrNoRows) { 269 + return nil, nil 270 + } 271 + return nil, &RepositoryError{Op: "GetActivity", Err: err} 272 + } 273 + 274 + return &cache, nil 275 + } 276 + 277 + // GetActivities retrieves cached activity data for multiple actors in a single query, 278 + // as a map of actorDid -> ActivityCacheModel for found entries. 279 + func (r *CacheRepository) GetActivities(ctx context.Context, actorDids []string) (map[string]*ActivityCacheModel, error) { 280 + if len(actorDids) == 0 { 281 + return make(map[string]*ActivityCacheModel), nil 282 + } 283 + 284 + query := ` 285 + SELECT actor_did, last_post_date, fetched_at, expires_at 286 + FROM cached_activity 287 + WHERE actor_did IN (` + buildPlaceholders(len(actorDids)) + `) AND expires_at > ? 288 + ` 289 + 290 + args := make([]interface{}, len(actorDids)+1) 291 + for i, did := range actorDids { 292 + args[i] = did 293 + } 294 + args[len(actorDids)] = time.Now() 295 + 296 + rows, err := r.db.QueryContext(ctx, query, args...) 297 + if err != nil { 298 + return nil, &RepositoryError{Op: "GetActivities", Err: err} 299 + } 300 + defer rows.Close() 301 + 302 + result := make(map[string]*ActivityCacheModel) 303 + for rows.Next() { 304 + var cache ActivityCacheModel 305 + var lastPostDate sql.NullTime 306 + 307 + err := rows.Scan( 308 + &cache.ActorDid, 309 + &lastPostDate, 310 + &cache.FetchedAt, 311 + &cache.ExpiresAt, 312 + ) 313 + if err != nil { 314 + return nil, &RepositoryError{Op: "GetActivities", Err: err} 315 + } 316 + 317 + if lastPostDate.Valid { 318 + cache.LastPostDate = lastPostDate.Time 319 + } 320 + 321 + result[cache.ActorDid] = &cache 322 + } 323 + 324 + return result, rows.Err() 325 + } 326 + 327 + // SaveActivity saves or updates an activity cache entry 328 + func (r *CacheRepository) SaveActivity(ctx context.Context, cache *ActivityCacheModel) error { 329 + if cache.FetchedAt.IsZero() { 330 + cache.FetchedAt = time.Now() 331 + } 332 + if cache.ExpiresAt.IsZero() { 333 + cache.ExpiresAt = time.Now().Add(24 * time.Hour) 334 + } 335 + 336 + query := ` 337 + INSERT INTO cached_activity (actor_did, last_post_date, fetched_at, expires_at) 338 + VALUES (?, ?, ?, ?) 339 + ON CONFLICT(actor_did) DO UPDATE SET 340 + last_post_date = excluded.last_post_date, 341 + fetched_at = excluded.fetched_at, 342 + expires_at = excluded.expires_at 343 + ` 344 + 345 + var lastPostDate interface{} 346 + if !cache.LastPostDate.IsZero() { 347 + lastPostDate = cache.LastPostDate 348 + } 349 + 350 + _, err := r.db.ExecContext(ctx, query, 351 + cache.ActorDid, 352 + lastPostDate, 353 + cache.FetchedAt, 354 + cache.ExpiresAt, 355 + ) 356 + 357 + if err != nil { 358 + return &RepositoryError{Op: "SaveActivity", Err: err} 359 + } 360 + 361 + return nil 362 + } 363 + 364 + // SaveActivities saves multiple activity cache entries in a transaction 365 + func (r *CacheRepository) SaveActivities(ctx context.Context, caches []*ActivityCacheModel) error { 366 + if len(caches) == 0 { 367 + return nil 368 + } 369 + 370 + tx, err := r.db.BeginTx(ctx, nil) 371 + if err != nil { 372 + return &RepositoryError{Op: "SaveActivities", Err: err} 373 + } 374 + defer tx.Rollback() 375 + 376 + stmt, err := tx.PrepareContext(ctx, ` 377 + INSERT INTO cached_activity (actor_did, last_post_date, fetched_at, expires_at) 378 + VALUES (?, ?, ?, ?) 379 + ON CONFLICT(actor_did) DO UPDATE SET 380 + last_post_date = excluded.last_post_date, 381 + fetched_at = excluded.fetched_at, 382 + expires_at = excluded.expires_at 383 + `) 384 + if err != nil { 385 + return &RepositoryError{Op: "SaveActivities", Err: err} 386 + } 387 + defer stmt.Close() 388 + 389 + for _, cache := range caches { 390 + if cache.FetchedAt.IsZero() { 391 + cache.FetchedAt = time.Now() 392 + } 393 + if cache.ExpiresAt.IsZero() { 394 + cache.ExpiresAt = time.Now().Add(24 * time.Hour) 395 + } 396 + 397 + var lastPostDate interface{} 398 + if !cache.LastPostDate.IsZero() { 399 + lastPostDate = cache.LastPostDate 400 + } 401 + 402 + _, err := stmt.ExecContext(ctx, 403 + cache.ActorDid, 404 + lastPostDate, 405 + cache.FetchedAt, 406 + cache.ExpiresAt, 407 + ) 408 + if err != nil { 409 + return &RepositoryError{Op: "SaveActivities", Err: err} 410 + } 411 + } 412 + 413 + if err := tx.Commit(); err != nil { 414 + return &RepositoryError{Op: "SaveActivities", Err: err} 415 + } 416 + 417 + return nil 418 + } 419 + 420 + // DeleteActivity removes an activity cache entry 421 + func (r *CacheRepository) DeleteActivity(ctx context.Context, actorDid string) error { 422 + query := "DELETE FROM cached_activity WHERE actor_did = ?" 423 + _, err := r.db.ExecContext(ctx, query, actorDid) 424 + if err != nil { 425 + return &RepositoryError{Op: "DeleteActivity", Err: err} 426 + } 427 + return nil 428 + } 429 + 430 + // DeleteExpiredPostRates removes all expired post rate cache entries 431 + func (r *CacheRepository) DeleteExpiredPostRates(ctx context.Context) (int64, error) { 432 + query := "DELETE FROM cached_post_rates WHERE expires_at < ?" 433 + result, err := r.db.ExecContext(ctx, query, time.Now()) 434 + if err != nil { 435 + return 0, &RepositoryError{Op: "DeleteExpiredPostRates", Err: err} 436 + } 437 + 438 + rows, err := result.RowsAffected() 439 + if err != nil { 440 + return 0, &RepositoryError{Op: "DeleteExpiredPostRates", Err: err} 441 + } 442 + 443 + return rows, nil 444 + } 445 + 446 + // DeleteExpiredActivities removes all expired activity cache entries 447 + func (r *CacheRepository) DeleteExpiredActivities(ctx context.Context) (int64, error) { 448 + query := "DELETE FROM cached_activity WHERE expires_at < ?" 449 + result, err := r.db.ExecContext(ctx, query, time.Now()) 450 + if err != nil { 451 + return 0, &RepositoryError{Op: "DeleteExpiredActivities", Err: err} 452 + } 453 + 454 + rows, err := result.RowsAffected() 455 + if err != nil { 456 + return 0, &RepositoryError{Op: "DeleteExpiredActivities", Err: err} 457 + } 458 + 459 + return rows, nil 460 + } 461 + 462 + // buildPlaceholders generates SQL placeholder string for IN queries. 463 + // 464 + // Example: buildPlaceholders(3) returns "?,?,?" 465 + func buildPlaceholders(count int) string { 466 + if count == 0 { 467 + return "" 468 + } 469 + 470 + placeholders := "?" 471 + for i := 1; i < count; i++ { 472 + placeholders += ",?" 473 + } 474 + return placeholders 475 + }
+516
cli/internal/store/cache_repo_test.go
···
··· 1 + package store 2 + 3 + import ( 4 + "context" 5 + "testing" 6 + "time" 7 + 8 + "github.com/stormlightlabs/skypanel/cli/internal/utils" 9 + ) 10 + 11 + func TestCacheRepository_Init(t *testing.T) { 12 + db, cleanup := utils.NewTestDB(t) 13 + defer cleanup() 14 + 15 + repo := &CacheRepository{db: db} 16 + 17 + err := repo.Init(context.Background()) 18 + if err != nil { 19 + t.Fatalf("Init failed: %v", err) 20 + } 21 + 22 + var count int 23 + err = db.QueryRow("SELECT COUNT(*) FROM cached_post_rates").Scan(&count) 24 + if err != nil { 25 + t.Errorf("cached_post_rates table not created: %v", err) 26 + } 27 + 28 + err = db.QueryRow("SELECT COUNT(*) FROM cached_activity").Scan(&count) 29 + if err != nil { 30 + t.Errorf("cached_activity table not created: %v", err) 31 + } 32 + } 33 + 34 + func TestCacheRepository_SaveAndGetPostRate(t *testing.T) { 35 + db, cleanup := utils.NewTestDB(t) 36 + defer cleanup() 37 + 38 + repo := &CacheRepository{db: db} 39 + if err := repo.Init(context.Background()); err != nil { 40 + t.Fatalf("Init failed: %v", err) 41 + } 42 + 43 + cache := &PostRateCacheModel{ 44 + ActorDid: "did:plc:test123", 45 + PostsPerDay: 2.5, 46 + LastPostDate: time.Now().Add(-2 * time.Hour), 47 + SampleSize: 30, 48 + } 49 + 50 + err := repo.SavePostRate(context.Background(), cache) 51 + if err != nil { 52 + t.Fatalf("SavePostRate failed: %v", err) 53 + } 54 + 55 + if cache.FetchedAt.IsZero() { 56 + t.Error("expected FetchedAt to be set after Save") 57 + } 58 + if cache.ExpiresAt.IsZero() { 59 + t.Error("expected ExpiresAt to be set after Save") 60 + } 61 + 62 + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:test123") 63 + if err != nil { 64 + t.Fatalf("GetPostRate failed: %v", err) 65 + } 66 + 67 + if retrieved == nil { 68 + t.Fatal("expected cache entry, got nil") 69 + } 70 + 71 + if retrieved.ActorDid != "did:plc:test123" { 72 + t.Errorf("expected ActorDid 'did:plc:test123', got %s", retrieved.ActorDid) 73 + } 74 + if retrieved.PostsPerDay != 2.5 { 75 + t.Errorf("expected PostsPerDay 2.5, got %f", retrieved.PostsPerDay) 76 + } 77 + if retrieved.SampleSize != 30 { 78 + t.Errorf("expected SampleSize 30, got %d", retrieved.SampleSize) 79 + } 80 + } 81 + 82 + func TestCacheRepository_GetPostRate_NotFound(t *testing.T) { 83 + db, cleanup := utils.NewTestDB(t) 84 + defer cleanup() 85 + 86 + repo := &CacheRepository{db: db} 87 + if err := repo.Init(context.Background()); err != nil { 88 + t.Fatalf("Init failed: %v", err) 89 + } 90 + 91 + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:nonexistent") 92 + if err != nil { 93 + t.Fatalf("GetPostRate failed: %v", err) 94 + } 95 + 96 + if retrieved != nil { 97 + t.Error("expected nil for nonexistent cache entry") 98 + } 99 + } 100 + 101 + func TestCacheRepository_GetPostRate_Expired(t *testing.T) { 102 + db, cleanup := utils.NewTestDB(t) 103 + defer cleanup() 104 + 105 + repo := &CacheRepository{db: db} 106 + if err := repo.Init(context.Background()); err != nil { 107 + t.Fatalf("Init failed: %v", err) 108 + } 109 + 110 + cache := &PostRateCacheModel{ 111 + ActorDid: "did:plc:expired", 112 + PostsPerDay: 1.0, 113 + SampleSize: 10, 114 + FetchedAt: time.Now().Add(-25 * time.Hour), 115 + ExpiresAt: time.Now().Add(-1 * time.Hour), 116 + } 117 + 118 + err := repo.SavePostRate(context.Background(), cache) 119 + if err != nil { 120 + t.Fatalf("SavePostRate failed: %v", err) 121 + } 122 + 123 + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:expired") 124 + if err != nil { 125 + t.Fatalf("GetPostRate failed: %v", err) 126 + } 127 + 128 + if retrieved != nil { 129 + t.Error("expected nil for expired cache entry") 130 + } 131 + } 132 + 133 + func TestCacheRepository_SavePostRates_Batch(t *testing.T) { 134 + db, cleanup := utils.NewTestDB(t) 135 + defer cleanup() 136 + 137 + repo := &CacheRepository{db: db} 138 + if err := repo.Init(context.Background()); err != nil { 139 + t.Fatalf("Init failed: %v", err) 140 + } 141 + 142 + caches := []*PostRateCacheModel{ 143 + { 144 + ActorDid: "did:plc:user1", 145 + PostsPerDay: 1.5, 146 + LastPostDate: time.Now().Add(-1 * time.Hour), 147 + SampleSize: 20, 148 + }, 149 + { 150 + ActorDid: "did:plc:user2", 151 + PostsPerDay: 3.0, 152 + LastPostDate: time.Now().Add(-2 * time.Hour), 153 + SampleSize: 30, 154 + }, 155 + { 156 + ActorDid: "did:plc:user3", 157 + PostsPerDay: 0.0, 158 + SampleSize: 0, 159 + }, 160 + } 161 + 162 + err := repo.SavePostRates(context.Background(), caches) 163 + if err != nil { 164 + t.Fatalf("SavePostRates failed: %v", err) 165 + } 166 + 167 + retrieved, err := repo.GetPostRates(context.Background(), []string{"did:plc:user1", "did:plc:user2", "did:plc:user3"}) 168 + if err != nil { 169 + t.Fatalf("GetPostRates failed: %v", err) 170 + } 171 + 172 + if len(retrieved) != 3 { 173 + t.Errorf("expected 3 cache entries, got %d", len(retrieved)) 174 + } 175 + 176 + if cache, ok := retrieved["did:plc:user1"]; ok { 177 + if cache.PostsPerDay != 1.5 { 178 + t.Errorf("expected PostsPerDay 1.5 for user1, got %f", cache.PostsPerDay) 179 + } 180 + } else { 181 + t.Error("expected cache entry for user1") 182 + } 183 + 184 + if cache, ok := retrieved["did:plc:user3"]; ok { 185 + if cache.LastPostDate.IsZero() == false { 186 + t.Error("expected zero LastPostDate for user3 (never posted)") 187 + } 188 + } 189 + } 190 + 191 + func TestCacheRepository_GetPostRates_PartialMatch(t *testing.T) { 192 + db, cleanup := utils.NewTestDB(t) 193 + defer cleanup() 194 + 195 + repo := &CacheRepository{db: db} 196 + if err := repo.Init(context.Background()); err != nil { 197 + t.Fatalf("Init failed: %v", err) 198 + } 199 + 200 + cache := &PostRateCacheModel{ 201 + ActorDid: "did:plc:cached", 202 + PostsPerDay: 2.0, 203 + SampleSize: 25, 204 + } 205 + 206 + err := repo.SavePostRate(context.Background(), cache) 207 + if err != nil { 208 + t.Fatalf("SavePostRate failed: %v", err) 209 + } 210 + 211 + retrieved, err := repo.GetPostRates(context.Background(), []string{"did:plc:cached", "did:plc:notcached"}) 212 + if err != nil { 213 + t.Fatalf("GetPostRates failed: %v", err) 214 + } 215 + 216 + if len(retrieved) != 1 { 217 + t.Errorf("expected 1 cache entry, got %d", len(retrieved)) 218 + } 219 + 220 + if _, ok := retrieved["did:plc:cached"]; !ok { 221 + t.Error("expected cache entry for 'did:plc:cached'") 222 + } 223 + 224 + if _, ok := retrieved["did:plc:notcached"]; ok { 225 + t.Error("did not expect cache entry for 'did:plc:notcached'") 226 + } 227 + } 228 + 229 + func TestCacheRepository_SaveAndGetActivity(t *testing.T) { 230 + db, cleanup := utils.NewTestDB(t) 231 + defer cleanup() 232 + 233 + repo := &CacheRepository{db: db} 234 + if err := repo.Init(context.Background()); err != nil { 235 + t.Fatalf("Init failed: %v", err) 236 + } 237 + 238 + cache := &ActivityCacheModel{ 239 + ActorDid: "did:plc:active", 240 + LastPostDate: time.Now().Add(-3 * time.Hour), 241 + } 242 + 243 + err := repo.SaveActivity(context.Background(), cache) 244 + if err != nil { 245 + t.Fatalf("SaveActivity failed: %v", err) 246 + } 247 + 248 + if cache.FetchedAt.IsZero() { 249 + t.Error("expected FetchedAt to be set after Save") 250 + } 251 + if cache.ExpiresAt.IsZero() { 252 + t.Error("expected ExpiresAt to be set after Save") 253 + } 254 + 255 + retrieved, err := repo.GetActivity(context.Background(), "did:plc:active") 256 + if err != nil { 257 + t.Fatalf("GetActivity failed: %v", err) 258 + } 259 + 260 + if retrieved == nil { 261 + t.Fatal("expected cache entry, got nil") 262 + } 263 + 264 + if retrieved.ActorDid != "did:plc:active" { 265 + t.Errorf("expected ActorDid 'did:plc:active', got %s", retrieved.ActorDid) 266 + } 267 + if retrieved.LastPostDate.IsZero() { 268 + t.Error("expected non-zero LastPostDate") 269 + } 270 + if !retrieved.HasPosted() { 271 + t.Error("expected HasPosted to be true") 272 + } 273 + } 274 + 275 + func TestCacheRepository_SaveActivity_NeverPosted(t *testing.T) { 276 + db, cleanup := utils.NewTestDB(t) 277 + defer cleanup() 278 + 279 + repo := &CacheRepository{db: db} 280 + if err := repo.Init(context.Background()); err != nil { 281 + t.Fatalf("Init failed: %v", err) 282 + } 283 + 284 + cache := &ActivityCacheModel{ 285 + ActorDid: "did:plc:neverposted", 286 + LastPostDate: time.Time{}, 287 + } 288 + 289 + err := repo.SaveActivity(context.Background(), cache) 290 + if err != nil { 291 + t.Fatalf("SaveActivity failed: %v", err) 292 + } 293 + 294 + retrieved, err := repo.GetActivity(context.Background(), "did:plc:neverposted") 295 + if err != nil { 296 + t.Fatalf("GetActivity failed: %v", err) 297 + } 298 + 299 + if retrieved == nil { 300 + t.Fatal("expected cache entry, got nil") 301 + } 302 + 303 + if !retrieved.LastPostDate.IsZero() { 304 + t.Error("expected zero LastPostDate for actor who never posted") 305 + } 306 + if retrieved.HasPosted() { 307 + t.Error("expected HasPosted to be false") 308 + } 309 + } 310 + 311 + func TestCacheRepository_SaveActivities_Batch(t *testing.T) { 312 + db, cleanup := utils.NewTestDB(t) 313 + defer cleanup() 314 + 315 + repo := &CacheRepository{db: db} 316 + if err := repo.Init(context.Background()); err != nil { 317 + t.Fatalf("Init failed: %v", err) 318 + } 319 + 320 + caches := []*ActivityCacheModel{ 321 + { 322 + ActorDid: "did:plc:actor1", 323 + LastPostDate: time.Now().Add(-1 * time.Hour), 324 + }, 325 + { 326 + ActorDid: "did:plc:actor2", 327 + LastPostDate: time.Now().Add(-5 * time.Hour), 328 + }, 329 + { 330 + ActorDid: "did:plc:actor3", 331 + LastPostDate: time.Time{}, 332 + }, 333 + } 334 + 335 + err := repo.SaveActivities(context.Background(), caches) 336 + if err != nil { 337 + t.Fatalf("SaveActivities failed: %v", err) 338 + } 339 + 340 + retrieved, err := repo.GetActivities(context.Background(), []string{"did:plc:actor1", "did:plc:actor2", "did:plc:actor3"}) 341 + if err != nil { 342 + t.Fatalf("GetActivities failed: %v", err) 343 + } 344 + 345 + if len(retrieved) != 3 { 346 + t.Errorf("expected 3 cache entries, got %d", len(retrieved)) 347 + } 348 + 349 + if cache, ok := retrieved["did:plc:actor1"]; ok { 350 + if cache.LastPostDate.IsZero() { 351 + t.Error("expected non-zero LastPostDate for actor1") 352 + } 353 + } else { 354 + t.Error("expected cache entry for actor1") 355 + } 356 + 357 + if cache, ok := retrieved["did:plc:actor3"]; ok { 358 + if !cache.LastPostDate.IsZero() { 359 + t.Error("expected zero LastPostDate for actor3") 360 + } 361 + } 362 + } 363 + 364 + func TestCacheRepository_Upsert_PostRate(t *testing.T) { 365 + db, cleanup := utils.NewTestDB(t) 366 + defer cleanup() 367 + 368 + repo := &CacheRepository{db: db} 369 + if err := repo.Init(context.Background()); err != nil { 370 + t.Fatalf("Init failed: %v", err) 371 + } 372 + 373 + cache := &PostRateCacheModel{ 374 + ActorDid: "did:plc:upserttest", 375 + PostsPerDay: 1.0, 376 + SampleSize: 10, 377 + } 378 + 379 + err := repo.SavePostRate(context.Background(), cache) 380 + if err != nil { 381 + t.Fatalf("SavePostRate failed: %v", err) 382 + } 383 + 384 + updatedCache := &PostRateCacheModel{ 385 + ActorDid: "did:plc:upserttest", 386 + PostsPerDay: 3.0, 387 + SampleSize: 30, 388 + } 389 + 390 + err = repo.SavePostRate(context.Background(), updatedCache) 391 + if err != nil { 392 + t.Fatalf("Update SavePostRate failed: %v", err) 393 + } 394 + 395 + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:upserttest") 396 + if err != nil { 397 + t.Fatalf("GetPostRate failed: %v", err) 398 + } 399 + 400 + if retrieved.PostsPerDay != 3.0 { 401 + t.Errorf("expected PostsPerDay 3.0 after upsert, got %f", retrieved.PostsPerDay) 402 + } 403 + if retrieved.SampleSize != 30 { 404 + t.Errorf("expected SampleSize 30 after upsert, got %d", retrieved.SampleSize) 405 + } 406 + } 407 + 408 + func TestCacheRepository_DeletePostRate(t *testing.T) { 409 + db, cleanup := utils.NewTestDB(t) 410 + defer cleanup() 411 + 412 + repo := &CacheRepository{db: db} 413 + if err := repo.Init(context.Background()); err != nil { 414 + t.Fatalf("Init failed: %v", err) 415 + } 416 + 417 + cache := &PostRateCacheModel{ 418 + ActorDid: "did:plc:todelete", 419 + PostsPerDay: 2.0, 420 + SampleSize: 20, 421 + } 422 + 423 + err := repo.SavePostRate(context.Background(), cache) 424 + if err != nil { 425 + t.Fatalf("SavePostRate failed: %v", err) 426 + } 427 + 428 + err = repo.DeletePostRate(context.Background(), "did:plc:todelete") 429 + if err != nil { 430 + t.Fatalf("DeletePostRate failed: %v", err) 431 + } 432 + 433 + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:todelete") 434 + if err != nil { 435 + t.Fatalf("GetPostRate failed: %v", err) 436 + } 437 + if retrieved != nil { 438 + t.Error("expected nil after DeletePostRate") 439 + } 440 + } 441 + 442 + func TestCacheRepository_DeleteExpiredPostRates(t *testing.T) { 443 + db, cleanup := utils.NewTestDB(t) 444 + defer cleanup() 445 + 446 + repo := &CacheRepository{db: db} 447 + if err := repo.Init(context.Background()); err != nil { 448 + t.Fatalf("Init failed: %v", err) 449 + } 450 + 451 + freshCache := &PostRateCacheModel{ 452 + ActorDid: "did:plc:fresh", 453 + PostsPerDay: 2.0, 454 + SampleSize: 20, 455 + FetchedAt: time.Now(), 456 + ExpiresAt: time.Now().Add(24 * time.Hour), 457 + } 458 + 459 + expiredCache := &PostRateCacheModel{ 460 + ActorDid: "did:plc:expired", 461 + PostsPerDay: 1.0, 462 + SampleSize: 10, 463 + FetchedAt: time.Now().Add(-25 * time.Hour), 464 + ExpiresAt: time.Now().Add(-1 * time.Hour), 465 + } 466 + 467 + err := repo.SavePostRate(context.Background(), freshCache) 468 + if err != nil { 469 + t.Fatalf("SavePostRate failed: %v", err) 470 + } 471 + 472 + err = repo.SavePostRate(context.Background(), expiredCache) 473 + if err != nil { 474 + t.Fatalf("SavePostRate failed: %v", err) 475 + } 476 + 477 + deleted, err := repo.DeleteExpiredPostRates(context.Background()) 478 + if err != nil { 479 + t.Fatalf("DeleteExpiredPostRates failed: %v", err) 480 + } 481 + 482 + if deleted != 1 { 483 + t.Errorf("expected 1 deleted entry, got %d", deleted) 484 + } 485 + 486 + retrieved, err := repo.GetPostRate(context.Background(), "did:plc:fresh") 487 + if err != nil { 488 + t.Fatalf("GetPostRate failed: %v", err) 489 + } 490 + if retrieved == nil { 491 + t.Error("expected fresh cache to still exist") 492 + } 493 + 494 + retrieved, err = repo.GetPostRate(context.Background(), "did:plc:expired") 495 + if err != nil { 496 + t.Fatalf("GetPostRate failed: %v", err) 497 + } 498 + if retrieved != nil { 499 + t.Error("expected expired cache to be deleted") 500 + } 501 + } 502 + 503 + func TestCacheRepository_Close(t *testing.T) { 504 + db, cleanup := utils.NewTestDB(t) 505 + defer cleanup() 506 + 507 + repo := &CacheRepository{db: db} 508 + if err := repo.Init(context.Background()); err != nil { 509 + t.Fatalf("Init failed: %v", err) 510 + } 511 + 512 + err := repo.Close() 513 + if err != nil { 514 + t.Errorf("Close failed: %v", err) 515 + } 516 + }
+9 -22
cli/internal/store/migration_test.go
··· 7 "github.com/stormlightlabs/skypanel/cli/internal/utils" 8 ) 9 10 - // TestRunMigrations verifies that migrations are applied correctly to a fresh database. 11 - // It checks that the schema_migrations table is created and all migrations are executed in order. 12 func TestRunMigrations(t *testing.T) { 13 db, cleanup := utils.NewTestDB(t) 14 defer cleanup() ··· 23 t.Fatalf("schema_migrations table not found: %v", err) 24 } 25 26 - if count != 3 { 27 - t.Errorf("expected 3 migrations applied, got %d", count) 28 } 29 30 err = db.QueryRow("SELECT COUNT(*) FROM feeds").Scan(&count) ··· 43 } 44 } 45 46 - // TestRunMigrations_Idempotent verifies that running migrations multiple times 47 - // doesn't re-apply already executed migrations. 48 func TestRunMigrations_Idempotent(t *testing.T) { 49 db, cleanup := utils.NewTestDB(t) 50 defer cleanup() ··· 63 t.Fatalf("failed to query migrations: %v", err) 64 } 65 66 - if count != 3 { 67 - t.Errorf("expected 3 migrations, got %d", count) 68 } 69 } 70 71 - // TestRollback verifies that down migrations correctly revert database changes. 72 func TestRollback(t *testing.T) { 73 db, cleanup := utils.NewTestDB(t) 74 defer cleanup() ··· 102 } 103 } 104 105 - // TestRollback_Complete verifies that rolling back to version 0 removes all migrations. 106 func TestRollback_Complete(t *testing.T) { 107 db, cleanup := utils.NewTestDB(t) 108 defer cleanup() ··· 138 } 139 } 140 141 - // TestMigrationOrdering verifies that migrations are applied in correct version order. 142 func TestMigrationOrdering(t *testing.T) { 143 db, cleanup := utils.NewTestDB(t) 144 defer cleanup() ··· 153 } 154 defer rows.Close() 155 156 - expectedVersions := []int{1, 2, 3} 157 var actualVersions []int 158 159 for rows.Next() { ··· 175 } 176 } 177 178 - // TestGetAppliedMigrations verifies the helper function correctly retrieves applied migrations. 179 func TestGetAppliedMigrations(t *testing.T) { 180 db, cleanup := utils.NewTestDB(t) 181 defer cleanup() ··· 205 } 206 } 207 208 - // TestLoadMigrations verifies that migration files are correctly loaded from embedded FS. 209 func TestLoadMigrations(t *testing.T) { 210 upMigrations, err := loadMigrations("up") 211 if err != nil { 212 t.Fatalf("failed to load up migrations: %v", err) 213 } 214 215 - if len(upMigrations) != 3 { 216 - t.Errorf("expected 3 up migrations, got %d", len(upMigrations)) 217 } 218 219 for i := 1; i < len(upMigrations); i++ { ··· 227 t.Fatalf("failed to load down migrations: %v", err) 228 } 229 230 - if len(downMigrations) != 3 { 231 - t.Errorf("expected 3 down migrations, got %d", len(downMigrations)) 232 } 233 } 234 235 - // TestExecuteMigration verifies that SQL is correctly executed. 236 func TestExecuteMigration(t *testing.T) { 237 db, cleanup := utils.NewTestDB(t) 238 defer cleanup() ··· 254 } 255 } 256 257 - // TestRecordAndRemoveMigration verifies the migration tracking functions. 258 func TestRecordAndRemoveMigration(t *testing.T) { 259 db, cleanup := utils.NewTestDB(t) 260 defer cleanup() ··· 289 } 290 } 291 292 - // TestMigrationWithForeignKey verifies that foreign key constraints work correctly. 293 func TestMigrationWithForeignKey(t *testing.T) { 294 db, cleanup := utils.NewTestDB(t) 295 defer cleanup() ··· 327 } 328 } 329 330 - // TestCreateMigrationsTable verifies the migrations tracking table is created correctly. 331 func TestCreateMigrationsTable(t *testing.T) { 332 db, cleanup := utils.NewTestDB(t) 333 defer cleanup()
··· 7 "github.com/stormlightlabs/skypanel/cli/internal/utils" 8 ) 9 10 func TestRunMigrations(t *testing.T) { 11 db, cleanup := utils.NewTestDB(t) 12 defer cleanup() ··· 21 t.Fatalf("schema_migrations table not found: %v", err) 22 } 23 24 + if count != 4 { 25 + t.Errorf("expected 4 migrations applied, got %d", count) 26 } 27 28 err = db.QueryRow("SELECT COUNT(*) FROM feeds").Scan(&count) ··· 41 } 42 } 43 44 func TestRunMigrations_Idempotent(t *testing.T) { 45 db, cleanup := utils.NewTestDB(t) 46 defer cleanup() ··· 59 t.Fatalf("failed to query migrations: %v", err) 60 } 61 62 + if count != 4 { 63 + t.Errorf("expected 4 migrations, got %d", count) 64 } 65 } 66 67 func TestRollback(t *testing.T) { 68 db, cleanup := utils.NewTestDB(t) 69 defer cleanup() ··· 97 } 98 } 99 100 func TestRollback_Complete(t *testing.T) { 101 db, cleanup := utils.NewTestDB(t) 102 defer cleanup() ··· 132 } 133 } 134 135 func TestMigrationOrdering(t *testing.T) { 136 db, cleanup := utils.NewTestDB(t) 137 defer cleanup() ··· 146 } 147 defer rows.Close() 148 149 + expectedVersions := []int{1, 2, 3, 4} 150 var actualVersions []int 151 152 for rows.Next() { ··· 168 } 169 } 170 171 func TestGetAppliedMigrations(t *testing.T) { 172 db, cleanup := utils.NewTestDB(t) 173 defer cleanup() ··· 197 } 198 } 199 200 func TestLoadMigrations(t *testing.T) { 201 upMigrations, err := loadMigrations("up") 202 if err != nil { 203 t.Fatalf("failed to load up migrations: %v", err) 204 } 205 206 + if len(upMigrations) != 4 { 207 + t.Errorf("expected 4 up migrations, got %d", len(upMigrations)) 208 } 209 210 for i := 1; i < len(upMigrations); i++ { ··· 218 t.Fatalf("failed to load down migrations: %v", err) 219 } 220 221 + if len(downMigrations) != 4 { 222 + t.Errorf("expected 4 down migrations, got %d", len(downMigrations)) 223 } 224 } 225 226 func TestExecuteMigration(t *testing.T) { 227 db, cleanup := utils.NewTestDB(t) 228 defer cleanup() ··· 244 } 245 } 246 247 func TestRecordAndRemoveMigration(t *testing.T) { 248 db, cleanup := utils.NewTestDB(t) 249 defer cleanup() ··· 278 } 279 } 280 281 func TestMigrationWithForeignKey(t *testing.T) { 282 db, cleanup := utils.NewTestDB(t) 283 defer cleanup() ··· 315 } 316 } 317 318 func TestCreateMigrationsTable(t *testing.T) { 319 db, cleanup := utils.NewTestDB(t) 320 defer cleanup()
+4
cli/internal/store/migrations/004_create_cache_tables.down.sql
···
··· 1 + DROP TABLE IF EXISTS cached_activity; 2 + DROP TABLE IF EXISTS cached_post_rates; 3 + DROP TABLE IF EXISTS follower_snapshot_entries; 4 + DROP TABLE IF EXISTS follower_snapshots;
+48
cli/internal/store/migrations/004_create_cache_tables.up.sql
···
··· 1 + -- Follower snapshots metadata 2 + CREATE TABLE IF NOT EXISTS follower_snapshots ( 3 + id TEXT PRIMARY KEY, 4 + created_at DATETIME NOT NULL, 5 + user_did TEXT NOT NULL, 6 + snapshot_type TEXT NOT NULL, 7 + total_count INTEGER NOT NULL, 8 + expires_at DATETIME NOT NULL 9 + ); 10 + 11 + CREATE INDEX IF NOT EXISTS idx_snapshots_user_type ON follower_snapshots(user_did, snapshot_type); 12 + CREATE INDEX IF NOT EXISTS idx_snapshots_created ON follower_snapshots(created_at); 13 + CREATE INDEX IF NOT EXISTS idx_snapshots_expires ON follower_snapshots(expires_at); 14 + 15 + -- Snapshot entries (actors in each snapshot) 16 + CREATE TABLE IF NOT EXISTS follower_snapshot_entries ( 17 + snapshot_id TEXT NOT NULL, 18 + actor_did TEXT NOT NULL, 19 + indexed_at TEXT, 20 + PRIMARY KEY(snapshot_id, actor_did), 21 + FOREIGN KEY(snapshot_id) REFERENCES follower_snapshots(id) ON DELETE CASCADE 22 + ); 23 + 24 + CREATE INDEX IF NOT EXISTS idx_snapshot_entries_actor ON follower_snapshot_entries(actor_did); 25 + 26 + -- Cached post rate metrics 27 + CREATE TABLE IF NOT EXISTS cached_post_rates ( 28 + actor_did TEXT PRIMARY KEY, 29 + posts_per_day REAL NOT NULL, 30 + last_post_date DATETIME, 31 + sample_size INTEGER NOT NULL, 32 + fetched_at DATETIME NOT NULL, 33 + expires_at DATETIME NOT NULL 34 + ); 35 + 36 + CREATE INDEX IF NOT EXISTS idx_post_rates_fetched ON cached_post_rates(fetched_at); 37 + CREATE INDEX IF NOT EXISTS idx_post_rates_expires ON cached_post_rates(expires_at); 38 + 39 + -- Cached activity data (last post dates) 40 + CREATE TABLE IF NOT EXISTS cached_activity ( 41 + actor_did TEXT PRIMARY KEY, 42 + last_post_date DATETIME, 43 + fetched_at DATETIME NOT NULL, 44 + expires_at DATETIME NOT NULL 45 + ); 46 + 47 + CREATE INDEX IF NOT EXISTS idx_activity_fetched ON cached_activity(fetched_at); 48 + CREATE INDEX IF NOT EXISTS idx_activity_expires ON cached_activity(expires_at);
+34
cli/internal/store/snapshot_model.go
···
··· 1 + package store 2 + 3 + import "time" 4 + 5 + // SnapshotModel represents a follower or following snapshot with metadata. 6 + // Stores snapshot metadata with TTL support (24 hours default). 7 + type SnapshotModel struct { 8 + id string 9 + createdAt time.Time 10 + UserDid string 11 + SnapshotType string // "followers" or "following" 12 + TotalCount int 13 + ExpiresAt time.Time 14 + } 15 + 16 + func (m *SnapshotModel) ID() string { return m.id } 17 + func (m *SnapshotModel) CreatedAt() time.Time { return m.createdAt } 18 + func (m *SnapshotModel) UpdatedAt() time.Time { return m.createdAt } // Snapshots are immutable 19 + func (m *SnapshotModel) SetID(id string) { m.id = id } 20 + func (m *SnapshotModel) SetCreatedAt(t time.Time) { m.createdAt = t } 21 + func (m *SnapshotModel) SetUpdatedAt(t time.Time) {} // Snapshots are immutable 22 + 23 + // IsFresh returns true if the snapshot has not expired. Snapshots expire after 24 hours by default. 24 + func (m *SnapshotModel) IsFresh() bool { 25 + return time.Now().Before(m.ExpiresAt) 26 + } 27 + 28 + // SnapshotEntry represents an actor in a snapshot with minimal cached data. 29 + // Linked to [SnapshotModel] via snapshot_id foreign key. 30 + type SnapshotEntry struct { 31 + SnapshotID string 32 + ActorDid string 33 + IndexedAt string // When the follow relationship was indexed by Bluesky 34 + }
+362
cli/internal/store/snapshot_repo.go
···
··· 1 + package store 2 + 3 + import ( 4 + "context" 5 + "database/sql" 6 + "errors" 7 + "time" 8 + 9 + _ "github.com/mattn/go-sqlite3" 10 + "github.com/stormlightlabs/skypanel/cli/internal/config" 11 + ) 12 + 13 + // SnapshotRepository implements Repository for SnapshotModel using SQLite. 14 + // Manages follower/following snapshots with entries for diff and historical comparison. 15 + type SnapshotRepository struct { 16 + db *sql.DB 17 + } 18 + 19 + // NewSnapshotRepository creates a new snapshot repository with SQLite backend 20 + func NewSnapshotRepository() (*SnapshotRepository, error) { 21 + dbPath, err := config.GetCacheDB() 22 + if err != nil { 23 + return nil, err 24 + } 25 + 26 + db, err := sql.Open("sqlite3", dbPath) 27 + if err != nil { 28 + return nil, err 29 + } 30 + 31 + return &SnapshotRepository{db: db}, nil 32 + } 33 + 34 + // Init ensures database schema is initialized via migrations 35 + func (r *SnapshotRepository) Init(ctx context.Context) error { 36 + if err := config.EnsureConfigDir(); err != nil { 37 + return err 38 + } 39 + return RunMigrations(r.db) 40 + } 41 + 42 + // Close releases database connection 43 + func (r *SnapshotRepository) Close() error { 44 + return r.db.Close() 45 + } 46 + 47 + // Get retrieves a snapshot by ID 48 + func (r *SnapshotRepository) Get(ctx context.Context, id string) (Model, error) { 49 + query := ` 50 + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at 51 + FROM follower_snapshots 52 + WHERE id = ? 53 + ` 54 + 55 + var snapshot SnapshotModel 56 + var snapshotID string 57 + var createdAt, expiresAt time.Time 58 + 59 + err := r.db.QueryRowContext(ctx, query, id).Scan( 60 + &snapshotID, 61 + &createdAt, 62 + &snapshot.UserDid, 63 + &snapshot.SnapshotType, 64 + &snapshot.TotalCount, 65 + &expiresAt, 66 + ) 67 + 68 + snapshot.SetID(snapshotID) 69 + snapshot.SetCreatedAt(createdAt) 70 + snapshot.ExpiresAt = expiresAt 71 + 72 + if err != nil { 73 + if errors.Is(err, sql.ErrNoRows) { 74 + return nil, &RepositoryError{Op: "Get", Err: errors.New("snapshot not found")} 75 + } 76 + return nil, &RepositoryError{Op: "Get", Err: err} 77 + } 78 + 79 + return &snapshot, nil 80 + } 81 + 82 + // List retrieves all snapshots ordered by creation date (newest first) 83 + func (r *SnapshotRepository) List(ctx context.Context) ([]Model, error) { 84 + query := ` 85 + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at 86 + FROM follower_snapshots 87 + ORDER BY created_at DESC 88 + ` 89 + 90 + rows, err := r.db.QueryContext(ctx, query) 91 + if err != nil { 92 + return nil, &RepositoryError{Op: "List", Err: err} 93 + } 94 + defer rows.Close() 95 + 96 + var snapshots []Model 97 + for rows.Next() { 98 + var snapshot SnapshotModel 99 + var snapshotID string 100 + var createdAt, expiresAt time.Time 101 + 102 + err := rows.Scan( 103 + &snapshotID, 104 + &createdAt, 105 + &snapshot.UserDid, 106 + &snapshot.SnapshotType, 107 + &snapshot.TotalCount, 108 + &expiresAt, 109 + ) 110 + if err != nil { 111 + return nil, &RepositoryError{Op: "List", Err: err} 112 + } 113 + 114 + snapshot.SetID(snapshotID) 115 + snapshot.SetCreatedAt(createdAt) 116 + snapshot.ExpiresAt = expiresAt 117 + snapshots = append(snapshots, &snapshot) 118 + } 119 + 120 + return snapshots, rows.Err() 121 + } 122 + 123 + // Save creates a new snapshot (snapshots are immutable, no updates) 124 + func (r *SnapshotRepository) Save(ctx context.Context, model Model) error { 125 + snapshot, ok := model.(*SnapshotModel) 126 + if !ok { 127 + return &RepositoryError{Op: "Save", Err: errors.New("invalid model type: expected *SnapshotModel")} 128 + } 129 + 130 + if snapshot.ID() == "" { 131 + snapshot.SetID(GenerateUUID()) 132 + snapshot.SetCreatedAt(time.Now()) 133 + } 134 + 135 + if snapshot.ExpiresAt.IsZero() { 136 + snapshot.ExpiresAt = time.Now().Add(24 * time.Hour) 137 + } 138 + 139 + query := ` 140 + INSERT INTO follower_snapshots (id, created_at, user_did, snapshot_type, total_count, expires_at) 141 + VALUES (?, ?, ?, ?, ?, ?) 142 + ` 143 + 144 + _, err := r.db.ExecContext(ctx, query, 145 + snapshot.ID(), 146 + snapshot.CreatedAt(), 147 + snapshot.UserDid, 148 + snapshot.SnapshotType, 149 + snapshot.TotalCount, 150 + snapshot.ExpiresAt, 151 + ) 152 + 153 + if err != nil { 154 + return &RepositoryError{Op: "Save", Err: err} 155 + } 156 + 157 + return nil 158 + } 159 + 160 + // Delete removes a snapshot by ID (cascade deletes entries) 161 + func (r *SnapshotRepository) Delete(ctx context.Context, id string) error { 162 + query := "DELETE FROM follower_snapshots WHERE id = ?" 163 + result, err := r.db.ExecContext(ctx, query, id) 164 + if err != nil { 165 + return &RepositoryError{Op: "Delete", Err: err} 166 + } 167 + 168 + rows, err := result.RowsAffected() 169 + if err != nil { 170 + return &RepositoryError{Op: "Delete", Err: err} 171 + } 172 + 173 + if rows == 0 { 174 + return &RepositoryError{Op: "Delete", Err: errors.New("snapshot not found")} 175 + } 176 + 177 + return nil 178 + } 179 + 180 + // FindByUserAndType retrieves the most recent fresh snapshot for a user and type. 181 + func (r *SnapshotRepository) FindByUserAndType(ctx context.Context, userDid, snapshotType string) (*SnapshotModel, error) { 182 + query := ` 183 + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at 184 + FROM follower_snapshots 185 + WHERE user_did = ? AND snapshot_type = ? AND expires_at > ? 186 + ORDER BY created_at DESC 187 + LIMIT 1 188 + ` 189 + 190 + var snapshot SnapshotModel 191 + var snapshotID string 192 + var createdAt, expiresAt time.Time 193 + 194 + err := r.db.QueryRowContext(ctx, query, userDid, snapshotType, time.Now()).Scan( 195 + &snapshotID, 196 + &createdAt, 197 + &snapshot.UserDid, 198 + &snapshot.SnapshotType, 199 + &snapshot.TotalCount, 200 + &expiresAt, 201 + ) 202 + 203 + if err != nil { 204 + if errors.Is(err, sql.ErrNoRows) { 205 + return nil, nil 206 + } 207 + return nil, &RepositoryError{Op: "FindByUserAndType", Err: err} 208 + } 209 + 210 + snapshot.SetID(snapshotID) 211 + snapshot.SetCreatedAt(createdAt) 212 + snapshot.ExpiresAt = expiresAt 213 + 214 + return &snapshot, nil 215 + } 216 + 217 + // FindByUserTypeAndDate retrieves a snapshot for a user, type, and specific date, closest to (but not after) the specified date. 218 + func (r *SnapshotRepository) FindByUserTypeAndDate(ctx context.Context, userDid, snapshotType string, date time.Time) (*SnapshotModel, error) { 219 + query := ` 220 + SELECT id, created_at, user_did, snapshot_type, total_count, expires_at 221 + FROM follower_snapshots 222 + WHERE user_did = ? AND snapshot_type = ? AND created_at <= ? 223 + ORDER BY created_at DESC 224 + LIMIT 1 225 + ` 226 + 227 + var snapshot SnapshotModel 228 + var snapshotID string 229 + var createdAt, expiresAt time.Time 230 + 231 + err := r.db.QueryRowContext(ctx, query, userDid, snapshotType, date).Scan( 232 + &snapshotID, 233 + &createdAt, 234 + &snapshot.UserDid, 235 + &snapshot.SnapshotType, 236 + &snapshot.TotalCount, 237 + &expiresAt, 238 + ) 239 + 240 + if err != nil { 241 + if errors.Is(err, sql.ErrNoRows) { 242 + return nil, nil 243 + } 244 + return nil, &RepositoryError{Op: "FindByUserTypeAndDate", Err: err} 245 + } 246 + 247 + snapshot.SetID(snapshotID) 248 + snapshot.SetCreatedAt(createdAt) 249 + snapshot.ExpiresAt = expiresAt 250 + return &snapshot, nil 251 + } 252 + 253 + // SaveEntry saves a single snapshot entry 254 + func (r *SnapshotRepository) SaveEntry(ctx context.Context, entry *SnapshotEntry) error { 255 + query := ` 256 + INSERT INTO follower_snapshot_entries (snapshot_id, actor_did, indexed_at) 257 + VALUES (?, ?, ?) 258 + ` 259 + 260 + _, err := r.db.ExecContext(ctx, query, entry.SnapshotID, entry.ActorDid, entry.IndexedAt) 261 + if err != nil { 262 + return &RepositoryError{Op: "SaveEntry", Err: err} 263 + } 264 + return nil 265 + } 266 + 267 + // SaveEntries saves multiple snapshot entries in a transaction for efficiency 268 + func (r *SnapshotRepository) SaveEntries(ctx context.Context, entries []*SnapshotEntry) error { 269 + tx, err := r.db.BeginTx(ctx, nil) 270 + if err != nil { 271 + return &RepositoryError{Op: "SaveEntries", Err: err} 272 + } 273 + defer tx.Rollback() 274 + 275 + stmt, err := tx.PrepareContext(ctx, ` 276 + INSERT INTO follower_snapshot_entries (snapshot_id, actor_did, indexed_at) 277 + VALUES (?, ?, ?) 278 + `) 279 + if err != nil { 280 + return &RepositoryError{Op: "SaveEntries", Err: err} 281 + } 282 + defer stmt.Close() 283 + 284 + for _, entry := range entries { 285 + _, err := stmt.ExecContext(ctx, entry.SnapshotID, entry.ActorDid, entry.IndexedAt) 286 + if err != nil { 287 + return &RepositoryError{Op: "SaveEntries", Err: err} 288 + } 289 + } 290 + 291 + if err := tx.Commit(); err != nil { 292 + return &RepositoryError{Op: "SaveEntries", Err: err} 293 + } 294 + return nil 295 + } 296 + 297 + // GetEntries retrieves all entries for a snapshot 298 + func (r *SnapshotRepository) GetEntries(ctx context.Context, snapshotID string) ([]*SnapshotEntry, error) { 299 + query := ` 300 + SELECT snapshot_id, actor_did, indexed_at 301 + FROM follower_snapshot_entries 302 + WHERE snapshot_id = ? 303 + ` 304 + 305 + rows, err := r.db.QueryContext(ctx, query, snapshotID) 306 + if err != nil { 307 + return nil, &RepositoryError{Op: "GetEntries", Err: err} 308 + } 309 + defer rows.Close() 310 + 311 + var entries []*SnapshotEntry 312 + for rows.Next() { 313 + var entry SnapshotEntry 314 + err := rows.Scan(&entry.SnapshotID, &entry.ActorDid, &entry.IndexedAt) 315 + if err != nil { 316 + return nil, &RepositoryError{Op: "GetEntries", Err: err} 317 + } 318 + entries = append(entries, &entry) 319 + } 320 + return entries, rows.Err() 321 + } 322 + 323 + // GetActorDids retrieves just the actor DIDs for a snapshot (efficient for diffs) 324 + func (r *SnapshotRepository) GetActorDids(ctx context.Context, snapshotID string) ([]string, error) { 325 + query := ` 326 + SELECT actor_did 327 + FROM follower_snapshot_entries 328 + WHERE snapshot_id = ? 329 + ` 330 + 331 + rows, err := r.db.QueryContext(ctx, query, snapshotID) 332 + if err != nil { 333 + return nil, &RepositoryError{Op: "GetActorDids", Err: err} 334 + } 335 + defer rows.Close() 336 + 337 + var dids []string 338 + for rows.Next() { 339 + var did string 340 + err := rows.Scan(&did) 341 + if err != nil { 342 + return nil, &RepositoryError{Op: "GetActorDids", Err: err} 343 + } 344 + dids = append(dids, did) 345 + } 346 + return dids, rows.Err() 347 + } 348 + 349 + // DeleteExpiredSnapshots removes all expired snapshots and their entries 350 + func (r *SnapshotRepository) DeleteExpiredSnapshots(ctx context.Context) (int64, error) { 351 + query := "DELETE FROM follower_snapshots WHERE expires_at < ?" 352 + result, err := r.db.ExecContext(ctx, query, time.Now()) 353 + if err != nil { 354 + return 0, &RepositoryError{Op: "DeleteExpiredSnapshots", Err: err} 355 + } 356 + 357 + rows, err := result.RowsAffected() 358 + if err != nil { 359 + return 0, &RepositoryError{Op: "DeleteExpiredSnapshots", Err: err} 360 + } 361 + return rows, nil 362 + }
+581
cli/internal/store/snapshot_repo_test.go
···
··· 1 + package store 2 + 3 + import ( 4 + "context" 5 + "testing" 6 + "time" 7 + 8 + "github.com/stormlightlabs/skypanel/cli/internal/utils" 9 + ) 10 + 11 + func TestSnapshotRepository_Init(t *testing.T) { 12 + db, cleanup := utils.NewTestDB(t) 13 + defer cleanup() 14 + 15 + repo := &SnapshotRepository{db: db} 16 + 17 + err := repo.Init(context.Background()) 18 + if err != nil { 19 + t.Fatalf("Init failed: %v", err) 20 + } 21 + 22 + var count int 23 + err = db.QueryRow("SELECT COUNT(*) FROM follower_snapshots").Scan(&count) 24 + if err != nil { 25 + t.Errorf("follower_snapshots table not created: %v", err) 26 + } 27 + 28 + err = db.QueryRow("SELECT COUNT(*) FROM follower_snapshot_entries").Scan(&count) 29 + if err != nil { 30 + t.Errorf("follower_snapshot_entries table not created: %v", err) 31 + } 32 + } 33 + 34 + func TestSnapshotRepository_SaveAndGet(t *testing.T) { 35 + db, cleanup := utils.NewTestDB(t) 36 + defer cleanup() 37 + 38 + repo := &SnapshotRepository{db: db} 39 + if err := repo.Init(context.Background()); err != nil { 40 + t.Fatalf("Init failed: %v", err) 41 + } 42 + 43 + snapshot := &SnapshotModel{ 44 + UserDid: "did:plc:testuser", 45 + SnapshotType: "followers", 46 + TotalCount: 150, 47 + } 48 + 49 + err := repo.Save(context.Background(), snapshot) 50 + if err != nil { 51 + t.Fatalf("Save failed: %v", err) 52 + } 53 + 54 + if snapshot.ID() == "" { 55 + t.Error("expected ID to be set after Save") 56 + } 57 + if snapshot.CreatedAt().IsZero() { 58 + t.Error("expected CreatedAt to be set after Save") 59 + } 60 + if snapshot.ExpiresAt.IsZero() { 61 + t.Error("expected ExpiresAt to be set after Save") 62 + } 63 + 64 + retrieved, err := repo.Get(context.Background(), snapshot.ID()) 65 + if err != nil { 66 + t.Fatalf("Get failed: %v", err) 67 + } 68 + 69 + retrievedSnapshot, ok := retrieved.(*SnapshotModel) 70 + if !ok { 71 + t.Fatal("expected *SnapshotModel") 72 + } 73 + 74 + if retrievedSnapshot.UserDid != "did:plc:testuser" { 75 + t.Errorf("expected UserDid 'did:plc:testuser', got %s", retrievedSnapshot.UserDid) 76 + } 77 + if retrievedSnapshot.SnapshotType != "followers" { 78 + t.Errorf("expected SnapshotType 'followers', got %s", retrievedSnapshot.SnapshotType) 79 + } 80 + if retrievedSnapshot.TotalCount != 150 { 81 + t.Errorf("expected TotalCount 150, got %d", retrievedSnapshot.TotalCount) 82 + } 83 + } 84 + 85 + func TestSnapshotRepository_List(t *testing.T) { 86 + db, cleanup := utils.NewTestDB(t) 87 + defer cleanup() 88 + 89 + repo := &SnapshotRepository{db: db} 90 + if err := repo.Init(context.Background()); err != nil { 91 + t.Fatalf("Init failed: %v", err) 92 + } 93 + 94 + snapshot1 := &SnapshotModel{ 95 + UserDid: "did:plc:user1", 96 + SnapshotType: "followers", 97 + TotalCount: 100, 98 + } 99 + snapshot2 := &SnapshotModel{ 100 + UserDid: "did:plc:user1", 101 + SnapshotType: "following", 102 + TotalCount: 50, 103 + } 104 + 105 + if err := repo.Save(context.Background(), snapshot1); err != nil { 106 + t.Fatalf("Save snapshot1 failed: %v", err) 107 + } 108 + time.Sleep(10 * time.Millisecond) 109 + if err := repo.Save(context.Background(), snapshot2); err != nil { 110 + t.Fatalf("Save snapshot2 failed: %v", err) 111 + } 112 + 113 + snapshots, err := repo.List(context.Background()) 114 + if err != nil { 115 + t.Fatalf("List failed: %v", err) 116 + } 117 + 118 + if len(snapshots) != 2 { 119 + t.Errorf("expected 2 snapshots, got %d", len(snapshots)) 120 + } 121 + 122 + if s, ok := snapshots[0].(*SnapshotModel); ok { 123 + if s.SnapshotType != "following" { 124 + t.Errorf("expected first snapshot to be 'following', got %s", s.SnapshotType) 125 + } 126 + } 127 + } 128 + 129 + func TestSnapshotRepository_FindByUserAndType(t *testing.T) { 130 + db, cleanup := utils.NewTestDB(t) 131 + defer cleanup() 132 + 133 + repo := &SnapshotRepository{db: db} 134 + if err := repo.Init(context.Background()); err != nil { 135 + t.Fatalf("Init failed: %v", err) 136 + } 137 + 138 + snapshot := &SnapshotModel{ 139 + UserDid: "did:plc:alice", 140 + SnapshotType: "followers", 141 + TotalCount: 200, 142 + } 143 + 144 + err := repo.Save(context.Background(), snapshot) 145 + if err != nil { 146 + t.Fatalf("Save failed: %v", err) 147 + } 148 + 149 + retrieved, err := repo.FindByUserAndType(context.Background(), "did:plc:alice", "followers") 150 + if err != nil { 151 + t.Fatalf("FindByUserAndType failed: %v", err) 152 + } 153 + 154 + if retrieved == nil { 155 + t.Fatal("expected snapshot, got nil") 156 + } 157 + 158 + if retrieved.UserDid != "did:plc:alice" { 159 + t.Errorf("expected UserDid 'did:plc:alice', got %s", retrieved.UserDid) 160 + } 161 + if retrieved.SnapshotType != "followers" { 162 + t.Errorf("expected SnapshotType 'followers', got %s", retrieved.SnapshotType) 163 + } 164 + } 165 + 166 + func TestSnapshotRepository_FindByUserAndType_NotFound(t *testing.T) { 167 + db, cleanup := utils.NewTestDB(t) 168 + defer cleanup() 169 + 170 + repo := &SnapshotRepository{db: db} 171 + if err := repo.Init(context.Background()); err != nil { 172 + t.Fatalf("Init failed: %v", err) 173 + } 174 + 175 + retrieved, err := repo.FindByUserAndType(context.Background(), "did:plc:nonexistent", "followers") 176 + if err != nil { 177 + t.Fatalf("FindByUserAndType failed: %v", err) 178 + } 179 + 180 + if retrieved != nil { 181 + t.Error("expected nil for nonexistent snapshot") 182 + } 183 + } 184 + 185 + func TestSnapshotRepository_FindByUserAndType_Expired(t *testing.T) { 186 + db, cleanup := utils.NewTestDB(t) 187 + defer cleanup() 188 + 189 + repo := &SnapshotRepository{db: db} 190 + if err := repo.Init(context.Background()); err != nil { 191 + t.Fatalf("Init failed: %v", err) 192 + } 193 + 194 + snapshot := &SnapshotModel{ 195 + UserDid: "did:plc:bob", 196 + SnapshotType: "followers", 197 + TotalCount: 100, 198 + } 199 + snapshot.SetID(GenerateUUID()) 200 + snapshot.SetCreatedAt(time.Now().Add(-25 * time.Hour)) 201 + snapshot.ExpiresAt = time.Now().Add(-1 * time.Hour) 202 + 203 + err := repo.Save(context.Background(), snapshot) 204 + if err != nil { 205 + t.Fatalf("Save failed: %v", err) 206 + } 207 + 208 + retrieved, err := repo.FindByUserAndType(context.Background(), "did:plc:bob", "followers") 209 + if err != nil { 210 + t.Fatalf("FindByUserAndType failed: %v", err) 211 + } 212 + 213 + if retrieved != nil { 214 + t.Error("expected nil for expired snapshot") 215 + } 216 + } 217 + 218 + func TestSnapshotRepository_FindByUserTypeAndDate(t *testing.T) { 219 + db, cleanup := utils.NewTestDB(t) 220 + defer cleanup() 221 + 222 + repo := &SnapshotRepository{db: db} 223 + if err := repo.Init(context.Background()); err != nil { 224 + t.Fatalf("Init failed: %v", err) 225 + } 226 + 227 + oldSnapshot := &SnapshotModel{ 228 + UserDid: "did:plc:charlie", 229 + SnapshotType: "followers", 230 + TotalCount: 80, 231 + } 232 + oldSnapshot.SetID(GenerateUUID()) 233 + oldSnapshot.SetCreatedAt(time.Now().Add(-48 * time.Hour)) 234 + oldSnapshot.ExpiresAt = time.Now().Add(24 * time.Hour) 235 + 236 + recentSnapshot := &SnapshotModel{ 237 + UserDid: "did:plc:charlie", 238 + SnapshotType: "followers", 239 + TotalCount: 100, 240 + } 241 + recentSnapshot.SetID(GenerateUUID()) 242 + recentSnapshot.SetCreatedAt(time.Now().Add(-12 * time.Hour)) 243 + recentSnapshot.ExpiresAt = time.Now().Add(24 * time.Hour) 244 + 245 + if err := repo.Save(context.Background(), oldSnapshot); err != nil { 246 + t.Fatalf("Save oldSnapshot failed: %v", err) 247 + } 248 + if err := repo.Save(context.Background(), recentSnapshot); err != nil { 249 + t.Fatalf("Save recentSnapshot failed: %v", err) 250 + } 251 + 252 + targetDate := time.Now().Add(-24 * time.Hour) 253 + 254 + retrieved, err := repo.FindByUserTypeAndDate(context.Background(), "did:plc:charlie", "followers", targetDate) 255 + if err != nil { 256 + t.Fatalf("FindByUserTypeAndDate failed: %v", err) 257 + } 258 + 259 + if retrieved == nil { 260 + t.Fatal("expected snapshot, got nil") 261 + } 262 + 263 + if retrieved.TotalCount != 80 { 264 + t.Errorf("expected TotalCount 80 (old snapshot), got %d", retrieved.TotalCount) 265 + } 266 + } 267 + 268 + func TestSnapshotRepository_SaveAndGetEntry(t *testing.T) { 269 + db, cleanup := utils.NewTestDB(t) 270 + defer cleanup() 271 + 272 + repo := &SnapshotRepository{db: db} 273 + if err := repo.Init(context.Background()); err != nil { 274 + t.Fatalf("Init failed: %v", err) 275 + } 276 + 277 + snapshot := &SnapshotModel{ 278 + UserDid: "did:plc:testuser", 279 + SnapshotType: "followers", 280 + TotalCount: 1, 281 + } 282 + 283 + err := repo.Save(context.Background(), snapshot) 284 + if err != nil { 285 + t.Fatalf("Save snapshot failed: %v", err) 286 + } 287 + 288 + entry := &SnapshotEntry{ 289 + SnapshotID: snapshot.ID(), 290 + ActorDid: "did:plc:follower1", 291 + IndexedAt: "2024-01-15T10:00:00Z", 292 + } 293 + 294 + err = repo.SaveEntry(context.Background(), entry) 295 + if err != nil { 296 + t.Fatalf("SaveEntry failed: %v", err) 297 + } 298 + 299 + entries, err := repo.GetEntries(context.Background(), snapshot.ID()) 300 + if err != nil { 301 + t.Fatalf("GetEntries failed: %v", err) 302 + } 303 + 304 + if len(entries) != 1 { 305 + t.Errorf("expected 1 entry, got %d", len(entries)) 306 + } 307 + 308 + if entries[0].ActorDid != "did:plc:follower1" { 309 + t.Errorf("expected ActorDid 'did:plc:follower1', got %s", entries[0].ActorDid) 310 + } 311 + if entries[0].IndexedAt != "2024-01-15T10:00:00Z" { 312 + t.Errorf("expected IndexedAt '2024-01-15T10:00:00Z', got %s", entries[0].IndexedAt) 313 + } 314 + } 315 + 316 + func TestSnapshotRepository_SaveEntries_Batch(t *testing.T) { 317 + db, cleanup := utils.NewTestDB(t) 318 + defer cleanup() 319 + 320 + repo := &SnapshotRepository{db: db} 321 + if err := repo.Init(context.Background()); err != nil { 322 + t.Fatalf("Init failed: %v", err) 323 + } 324 + 325 + snapshot := &SnapshotModel{ 326 + UserDid: "did:plc:testuser", 327 + SnapshotType: "followers", 328 + TotalCount: 3, 329 + } 330 + 331 + err := repo.Save(context.Background(), snapshot) 332 + if err != nil { 333 + t.Fatalf("Save snapshot failed: %v", err) 334 + } 335 + 336 + entries := []*SnapshotEntry{ 337 + { 338 + SnapshotID: snapshot.ID(), 339 + ActorDid: "did:plc:follower1", 340 + IndexedAt: "2024-01-15T10:00:00Z", 341 + }, 342 + { 343 + SnapshotID: snapshot.ID(), 344 + ActorDid: "did:plc:follower2", 345 + IndexedAt: "2024-01-15T11:00:00Z", 346 + }, 347 + { 348 + SnapshotID: snapshot.ID(), 349 + ActorDid: "did:plc:follower3", 350 + IndexedAt: "2024-01-15T12:00:00Z", 351 + }, 352 + } 353 + 354 + err = repo.SaveEntries(context.Background(), entries) 355 + if err != nil { 356 + t.Fatalf("SaveEntries failed: %v", err) 357 + } 358 + 359 + retrieved, err := repo.GetEntries(context.Background(), snapshot.ID()) 360 + if err != nil { 361 + t.Fatalf("GetEntries failed: %v", err) 362 + } 363 + 364 + if len(retrieved) != 3 { 365 + t.Errorf("expected 3 entries, got %d", len(retrieved)) 366 + } 367 + } 368 + 369 + func TestSnapshotRepository_GetActorDids(t *testing.T) { 370 + db, cleanup := utils.NewTestDB(t) 371 + defer cleanup() 372 + 373 + repo := &SnapshotRepository{db: db} 374 + if err := repo.Init(context.Background()); err != nil { 375 + t.Fatalf("Init failed: %v", err) 376 + } 377 + 378 + snapshot := &SnapshotModel{ 379 + UserDid: "did:plc:testuser", 380 + SnapshotType: "followers", 381 + TotalCount: 2, 382 + } 383 + 384 + err := repo.Save(context.Background(), snapshot) 385 + if err != nil { 386 + t.Fatalf("Save snapshot failed: %v", err) 387 + } 388 + 389 + entries := []*SnapshotEntry{ 390 + {SnapshotID: snapshot.ID(), ActorDid: "did:plc:actor1", IndexedAt: "2024-01-15T10:00:00Z"}, 391 + {SnapshotID: snapshot.ID(), ActorDid: "did:plc:actor2", IndexedAt: "2024-01-15T11:00:00Z"}, 392 + } 393 + 394 + err = repo.SaveEntries(context.Background(), entries) 395 + if err != nil { 396 + t.Fatalf("SaveEntries failed: %v", err) 397 + } 398 + 399 + dids, err := repo.GetActorDids(context.Background(), snapshot.ID()) 400 + if err != nil { 401 + t.Fatalf("GetActorDids failed: %v", err) 402 + } 403 + 404 + if len(dids) != 2 { 405 + t.Errorf("expected 2 DIDs, got %d", len(dids)) 406 + } 407 + 408 + didMap := make(map[string]bool) 409 + for _, did := range dids { 410 + didMap[did] = true 411 + } 412 + 413 + if !didMap["did:plc:actor1"] { 414 + t.Error("expected 'did:plc:actor1' in results") 415 + } 416 + if !didMap["did:plc:actor2"] { 417 + t.Error("expected 'did:plc:actor2' in results") 418 + } 419 + } 420 + 421 + func TestSnapshotRepository_Delete_CascadesEntries(t *testing.T) { 422 + db, cleanup := utils.NewTestDB(t) 423 + defer cleanup() 424 + 425 + if _, err := db.Exec("PRAGMA foreign_keys = ON"); err != nil { 426 + t.Fatalf("failed to enable foreign keys: %v", err) 427 + } 428 + 429 + repo := &SnapshotRepository{db: db} 430 + if err := repo.Init(context.Background()); err != nil { 431 + t.Fatalf("Init failed: %v", err) 432 + } 433 + 434 + snapshot := &SnapshotModel{ 435 + UserDid: "did:plc:testuser", 436 + SnapshotType: "followers", 437 + TotalCount: 1, 438 + } 439 + 440 + err := repo.Save(context.Background(), snapshot) 441 + if err != nil { 442 + t.Fatalf("Save snapshot failed: %v", err) 443 + } 444 + 445 + entry := &SnapshotEntry{ 446 + SnapshotID: snapshot.ID(), 447 + ActorDid: "did:plc:follower1", 448 + IndexedAt: "2024-01-15T10:00:00Z", 449 + } 450 + 451 + err = repo.SaveEntry(context.Background(), entry) 452 + if err != nil { 453 + t.Fatalf("SaveEntry failed: %v", err) 454 + } 455 + 456 + err = repo.Delete(context.Background(), snapshot.ID()) 457 + if err != nil { 458 + t.Fatalf("Delete failed: %v", err) 459 + } 460 + 461 + entries, err := repo.GetEntries(context.Background(), snapshot.ID()) 462 + if err != nil { 463 + t.Fatalf("GetEntries failed: %v", err) 464 + } 465 + 466 + if len(entries) != 0 { 467 + t.Errorf("expected 0 entries after cascade delete, got %d", len(entries)) 468 + } 469 + } 470 + 471 + func TestSnapshotRepository_DeleteExpiredSnapshots(t *testing.T) { 472 + db, cleanup := utils.NewTestDB(t) 473 + defer cleanup() 474 + 475 + repo := &SnapshotRepository{db: db} 476 + if err := repo.Init(context.Background()); err != nil { 477 + t.Fatalf("Init failed: %v", err) 478 + } 479 + 480 + freshSnapshot := &SnapshotModel{ 481 + UserDid: "did:plc:user1", 482 + SnapshotType: "followers", 483 + TotalCount: 100, 484 + } 485 + freshSnapshot.SetID(GenerateUUID()) 486 + freshSnapshot.SetCreatedAt(time.Now()) 487 + freshSnapshot.ExpiresAt = time.Now().Add(24 * time.Hour) 488 + 489 + expiredSnapshot := &SnapshotModel{ 490 + UserDid: "did:plc:user1", 491 + SnapshotType: "following", 492 + TotalCount: 50, 493 + } 494 + expiredSnapshot.SetID(GenerateUUID()) 495 + expiredSnapshot.SetCreatedAt(time.Now().Add(-25 * time.Hour)) 496 + expiredSnapshot.ExpiresAt = time.Now().Add(-1 * time.Hour) 497 + 498 + if err := repo.Save(context.Background(), freshSnapshot); err != nil { 499 + t.Fatalf("Save freshSnapshot failed: %v", err) 500 + } 501 + if err := repo.Save(context.Background(), expiredSnapshot); err != nil { 502 + t.Fatalf("Save expiredSnapshot failed: %v", err) 503 + } 504 + 505 + deleted, err := repo.DeleteExpiredSnapshots(context.Background()) 506 + if err != nil { 507 + t.Fatalf("DeleteExpiredSnapshots failed: %v", err) 508 + } 509 + 510 + if deleted != 1 { 511 + t.Errorf("expected 1 deleted snapshot, got %d", deleted) 512 + } 513 + 514 + snapshots, err := repo.List(context.Background()) 515 + if err != nil { 516 + t.Fatalf("List failed: %v", err) 517 + } 518 + 519 + if len(snapshots) != 1 { 520 + t.Errorf("expected 1 snapshot remaining, got %d", len(snapshots)) 521 + } 522 + 523 + if s, ok := snapshots[0].(*SnapshotModel); ok { 524 + if s.SnapshotType != "followers" { 525 + t.Errorf("expected remaining snapshot to be 'followers', got %s", s.SnapshotType) 526 + } 527 + } 528 + } 529 + 530 + func TestSnapshotRepository_IsFresh(t *testing.T) { 531 + db, cleanup := utils.NewTestDB(t) 532 + defer cleanup() 533 + 534 + repo := &SnapshotRepository{db: db} 535 + if err := repo.Init(context.Background()); err != nil { 536 + t.Fatalf("Init failed: %v", err) 537 + } 538 + 539 + snapshot := &SnapshotModel{ 540 + UserDid: "did:plc:testuser", 541 + SnapshotType: "followers", 542 + TotalCount: 100, 543 + } 544 + 545 + err := repo.Save(context.Background(), snapshot) 546 + if err != nil { 547 + t.Fatalf("Save failed: %v", err) 548 + } 549 + 550 + if !snapshot.IsFresh() { 551 + t.Error("newly saved snapshot should be fresh") 552 + } 553 + 554 + expiredSnapshot := &SnapshotModel{ 555 + UserDid: "did:plc:testuser2", 556 + SnapshotType: "followers", 557 + TotalCount: 50, 558 + } 559 + expiredSnapshot.SetID(GenerateUUID()) 560 + expiredSnapshot.SetCreatedAt(time.Now().Add(-25 * time.Hour)) 561 + expiredSnapshot.ExpiresAt = time.Now().Add(-1 * time.Hour) 562 + 563 + if expiredSnapshot.IsFresh() { 564 + t.Error("expired snapshot should not be fresh") 565 + } 566 + } 567 + 568 + func TestSnapshotRepository_Close(t *testing.T) { 569 + db, cleanup := utils.NewTestDB(t) 570 + defer cleanup() 571 + 572 + repo := &SnapshotRepository{db: db} 573 + if err := repo.Init(context.Background()); err != nil { 574 + t.Fatalf("Init failed: %v", err) 575 + } 576 + 577 + err := repo.Close() 578 + if err != nil { 579 + t.Errorf("Close failed: %v", err) 580 + } 581 + }