diff --git a/atproto/client_test.go b/atproto/client_test.go index e772868..c4e4c8a 100644 --- a/atproto/client_test.go +++ b/atproto/client_test.go @@ -493,7 +493,7 @@ func TestBuildClient(t *testing.T) { t.Parallel() identity := &mockAuthClient{apiClient: tt.apiClient, did: tt.did} - result, err := BuildClient[map[string]any](identity, tt.customClient) + result, err := BuildClient(identity, tt.customClient) if tt.wantErr { if err == nil { @@ -646,7 +646,6 @@ func TestFixedPasswordAuth_RefreshUsesPOST(t *testing.T) { ctx := context.Background() err := auth.Refresh(ctx, httpClient, "refresh-token") - if err != nil { t.Fatalf("Refresh failed: %v", err) } @@ -699,7 +698,6 @@ func TestFixedPasswordAuth_IndigoBugCheck(t *testing.T) { fixedAuth := &FixedPasswordAuth{PasswordAuth: pa} err := fixedAuth.Refresh(context.Background(), server.Client(), "refresh-token") - if err != nil { t.Fatalf("FixedPasswordAuth.Refresh failed: %v", err) } diff --git a/atproto/rate_test.go b/atproto/rate_test.go index 5489f98..806f69d 100644 --- a/atproto/rate_test.go +++ b/atproto/rate_test.go @@ -59,7 +59,7 @@ func TestQuotaLimiter_AllowRead(t *testing.T) { t.Fatalf("AllowRead failed: %v", err) } - if kv.incrs == nil || len(kv.incrs) == 0 { + if len(kv.incrs) == 0 { t.Error("Expected kv.IncrByMulti to be called") } if !chargedAt.IsZero() { diff --git a/atproto/repo_test.go b/atproto/repo_test.go index 3af82fd..473ed98 100644 --- a/atproto/repo_test.go +++ b/atproto/repo_test.go @@ -745,7 +745,7 @@ func TestPrepareWrites(t *testing.T) { } func TestApplyWrites_Empty(t *testing.T) { - err := applyWrites[map[string]any](context.Background(), nil, "did:plc:test", "app.bsky.feed.post", []map[string]any{}) + err := applyWrites(context.Background(), nil, "did:plc:test", "app.bsky.feed.post", []map[string]any{}) if err != nil { t.Errorf("applyWrites with empty records should not fail: %v", err) } diff --git a/atproto/testmock.go b/atproto/testmock.go index b68dea4..fab1df0 100644 --- a/atproto/testmock.go +++ b/atproto/testmock.go @@ -47,7 +47,6 @@ func (m *mockClock) Now() time.Time { } type mockKVStoreWithErr struct { - mockKVStore Data map[string]int Err error } diff --git a/flake.nix b/flake.nix index 40074a3..14ed226 100644 --- a/flake.nix +++ b/flake.nix @@ -17,7 +17,7 @@ let lazuli = pkgs.buildGoModule rec { name = "lazuli"; - version = "0.1.3"; + version = "0.1.4"; src = pkgs.nix-gitignore.gitignoreSource [ "*.csv" "*.zip" "*.json" ] ./.; vendorHash = "sha256-MfBPv/L7wHuUGXx4BDd+DFq0RB11KuMHCzPjFv6FMgs="; ldflags = [ diff --git a/main.go b/main.go index f673aab..7aef455 100644 --- a/main.go +++ b/main.go @@ -60,8 +60,6 @@ func main() { } func run() error { - sync.ClientAgent = "lazuli/" + Version - storage, err := cache.NewBoltStorage() if err != nil { return fmt.Errorf("open cache: %w", err) @@ -317,7 +315,7 @@ func (a *App) runRetry(ctx context.Context, cmd *cli.Command) error { continue } - res := sync.PublishBatch(ctx, repoClient, did, []sync.PlayRecord{fr.rec}, a.storage) + res := sync.PublishBatch(ctx, repoClient, did, []sync.PlayRecord{fr.rec}, a.storage, sync.DefaultClientAgent) if res == nil { fmt.Printf("Successfully retried: %s - %s\n", fr.rec.ArtistName(), fr.rec.TrackName) diff --git a/sources/lastfm/lastfm.go b/sources/lastfm/lastfm.go index af858df..5fbe98f 100644 --- a/sources/lastfm/lastfm.go +++ b/sources/lastfm/lastfm.go @@ -129,7 +129,7 @@ func toSync(records []record) []sync.PlayRecord { TrackName: r.Track, Artists: []sync.PlayRecordArtist{{ArtistName: r.Artist, ArtistMbId: r.ArtistMbid}}, PlayedTime: sync.Timestamp{Time: time.Unix(utsSec, 0).UTC()}, - SubmissionClientAgent: sync.ClientAgent, + SubmissionClientAgent: sync.DefaultClientAgent, MusicServiceBaseDomain: sync.MusicServiceLastFM, ReleaseName: r.Album, ReleaseMbId: r.AlbumMbid, diff --git a/sources/spotify/spotify.go b/sources/spotify/spotify.go index 5ef80a5..bac9cf9 100644 --- a/sources/spotify/spotify.go +++ b/sources/spotify/spotify.go @@ -64,7 +64,7 @@ func (Parser) ParseFile(ctx context.Context, r io.Reader) ([]sync.PlayRecord, er TrackName: trackName, Artists: []sync.PlayRecordArtist{{ArtistName: artistName}}, PlayedTime: r.Ts, - SubmissionClientAgent: sync.ClientAgent, + SubmissionClientAgent: sync.DefaultClientAgent, MusicServiceBaseDomain: sync.MusicServiceSpotify, ReleaseName: releaseName, OriginUrl: originUrl, diff --git a/sync/batch_test.go b/sync/batch_test.go index 3efe84b..7d2d83e 100644 --- a/sync/batch_test.go +++ b/sync/batch_test.go @@ -197,6 +197,7 @@ func TestIsTransientError(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() if got := isTransientError(tt.err); got != tt.want { t.Errorf("isTransientError() = %v, want %v", got, tt.want) } @@ -206,80 +207,106 @@ func TestIsTransientError(t *testing.T) { func TestApplyWrites_RateClient(t *testing.T) { ctx := context.Background() + clientAgent := "test-agent" - t.Run("Empty anyRecords(records)", func(t *testing.T) { - limiter := &mockLimiter{} - client := atproto.NewRateClient[any](nil, "did:example:123", limiter) - err := client.ApplyWrites(ctx, "test", nil) - if err != nil { - t.Errorf("ApplyWrites(nil) error = %v", err) - } - }) - - t.Run("Too many records", func(t *testing.T) { - limiter := &mockLimiter{} - client := atproto.NewRateClient[any](&atclient.APIClient{}, "did:example:123", limiter) - recs := make([]any, 201) - err := client.ApplyWrites(ctx, "test", recs) - if err == nil { - t.Fatal("expected error for > 200 records") - } - expected := "too many records in one ApplyWrites call: 201 (max 200)" - if err.Error() != expected { - t.Errorf("expected error %q, got %q", expected, err.Error()) - } - }) - - t.Run("Transient error refunds tokens", func(t *testing.T) { - limiter := &mockLimiter{} - apiClient := atclient.NewAPIClient("https://example.com") - apiClient.Client.Transport = mockRoundTripper(func(req *http.Request) (*http.Response, error) { - return &http.Response{ - StatusCode: 503, - Body: http.NoBody, - }, nil - }) - client := atproto.NewRateClient[PlayRecord](apiClient, "did:example:123", limiter) + tests := []struct { + name string + setupClient func() (*atproto.RateClient[PlayRecord], *mockLimiter) + records []PlayRecord + wantErr bool + wantErrMsg string + wantRefunds int32 + }{ + { + name: "empty records succeeds", + setupClient: func() (*atproto.RateClient[PlayRecord], *mockLimiter) { + limiter := &mockLimiter{} + return atproto.NewRateClient[PlayRecord](nil, "did:example:123", limiter), limiter + }, + records: nil, + wantErr: false, + wantRefunds: 0, + }, + { + name: "too many records fails", + setupClient: func() (*atproto.RateClient[PlayRecord], *mockLimiter) { + limiter := &mockLimiter{} + return atproto.NewRateClient[PlayRecord](&atclient.APIClient{}, "did:example:123", limiter), limiter + }, + records: make([]PlayRecord, 201), + wantErr: true, + wantErrMsg: "too many records in one ApplyWrites call: 201 (max 200)", + wantRefunds: 0, + }, + { + name: "transient error refunds tokens", + setupClient: func() (*atproto.RateClient[PlayRecord], *mockLimiter) { + limiter := &mockLimiter{} + apiClient := atclient.NewAPIClient("https://example.com") + apiClient.Client.Transport = mockRoundTripper(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: 503, + Body: http.NoBody, + }, nil + }) + return atproto.NewRateClient[PlayRecord](apiClient, "did:example:123", limiter), limiter + }, + records: []PlayRecord{{TrackName: "Song 1"}}, + wantErr: true, + wantRefunds: 1, + }, + { + name: "non-transient error does NOT refund", + setupClient: func() (*atproto.RateClient[PlayRecord], *mockLimiter) { + limiter := &mockLimiter{} + apiClient := atclient.NewAPIClient("https://example.com") + apiClient.Client.Transport = mockRoundTripper(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: 400, + Body: http.NoBody, + }, nil + }) + return atproto.NewRateClient[PlayRecord](apiClient, "did:example:123", limiter), limiter + }, + records: []PlayRecord{{TrackName: "Song 1"}}, + wantErr: true, + wantRefunds: 0, + }, + } - err := client.ApplyWrites(ctx, "test", []PlayRecord{{TrackName: "Song 1"}}) - if err == nil { - t.Fatal("expected error") - } - if atomic.LoadInt32(&limiter.refunds) != 1 { - t.Errorf("expected 1 refund, got %d", limiter.refunds) - } - }) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, limiter := tt.setupClient() + err := client.ApplyWrites(ctx, "test", tt.records) + + if tt.wantErr { + if err == nil { + t.Error("expected error, got nil") + } else if tt.wantErrMsg != "" && err.Error() != tt.wantErrMsg { + t.Errorf("error msg = %q, want %q", err.Error(), tt.wantErrMsg) + } + } else if err != nil { + t.Errorf("unexpected error: %v", err) + } - t.Run("Non-transient error does NOT refund", func(t *testing.T) { - limiter := &mockLimiter{} - apiClient := atclient.NewAPIClient("https://example.com") - apiClient.Client.Transport = mockRoundTripper(func(req *http.Request) (*http.Response, error) { - return &http.Response{ - StatusCode: 400, - Body: http.NoBody, - }, nil + if got := atomic.LoadInt32(&limiter.refunds); got != tt.wantRefunds { + t.Errorf("refunds = %d, want %d", got, tt.wantRefunds) + } }) - client := atproto.NewRateClient[PlayRecord](apiClient, "did:example:123", limiter) - - err := client.ApplyWrites(ctx, "test", []PlayRecord{{TrackName: "Song 1"}}) - if err == nil { - t.Fatal("expected error") - } - if atomic.LoadInt32(&limiter.refunds) != 0 { - t.Errorf("expected 0 refunds, got %d", limiter.refunds) - } - }) + } + _ = clientAgent } func TestPublishBatch(t *testing.T) { ctx := context.Background() did := "did:example:123" batch := []PlayRecord{{TrackName: "Song 1"}} + clientAgent := "test-agent" t.Run("Success", func(t *testing.T) { storage := newMockStorage() client := &mockATProtoClient{} - err := PublishBatch(ctx, client, did, batch, storage) + err := PublishBatch(ctx, client, did, batch, storage, clientAgent) if err != nil { t.Fatal(err) } @@ -296,7 +323,7 @@ func TestPublishBatch(t *testing.T) { return expectedErr }, } - err := PublishBatch(ctx, client, did, batch, storage) + err := PublishBatch(ctx, client, did, batch, storage, clientAgent) if !errors.Is(err, expectedErr) { t.Errorf("expected error %v, got %v", expectedErr, err) } @@ -308,7 +335,7 @@ func TestPublishBatch(t *testing.T) { t.Run("Storage failure after ApplyWrites success", func(t *testing.T) { storage := &failingStorage{} client := &mockATProtoClient{} - err := PublishBatch(ctx, client, did, batch, storage) + err := PublishBatch(ctx, client, did, batch, storage, clientAgent) if err == nil || !strings.Contains(err.Error(), "failed to save records") { t.Errorf("expected storage save error, got %v", err) } @@ -326,6 +353,7 @@ func (s *failingStorage) SaveRecords(did string, records map[string][]byte) erro func TestPublish_Iterative(t *testing.T) { ctx := context.Background() did := "did:example:123" + clientAgent := "test-agent" rec1, _ := json.Marshal(PlayRecord{TrackName: "Song 1"}) rec2, _ := json.Marshal(PlayRecord{TrackName: "Song 2"}) @@ -352,6 +380,7 @@ func TestPublish_Iterative(t *testing.T) { BatchSize: 1, ATProtoClient: client, Storage: storage, + ClientAgent: clientAgent, }) if res.SuccessCount != 2 { @@ -376,6 +405,7 @@ func TestPublish_Iterative(t *testing.T) { BatchSize: 1, ATProtoClient: client, Storage: storage, + ClientAgent: clientAgent, }) if res.SuccessCount != 0 { diff --git a/sync/config.go b/sync/config.go index 8015541..dfe98e9 100644 --- a/sync/config.go +++ b/sync/config.go @@ -41,7 +41,7 @@ type Config struct { var DefaultConfig = Config{ RecordType: RecordType, - ClientAgent: ClientAgent, + ClientAgent: DefaultClientAgent, BatchSize: DefaultBatchSize, CrossSourceTolerance: CrossSourceTolerance, CacheTTL: CacheTTL, diff --git a/sync/import_test.go b/sync/import_test.go index a21b7a2..c92ef4f 100644 --- a/sync/import_test.go +++ b/sync/import_test.go @@ -169,6 +169,7 @@ func TestImportE2E(t *testing.T) { ATProtoClient: mockRepo, Storage: storage, Limiter: limiter, + ClientAgent: sync.DefaultClientAgent, } auth := &mockAuthClient{did: did} diff --git a/sync/progress.go b/sync/progress.go index c5f4ad7..53df34c 100644 --- a/sync/progress.go +++ b/sync/progress.go @@ -37,8 +37,8 @@ func NewProgressTracker(total int, limiter RateLimiter) *ProgressTracker { StartTime: time.Now(), LastLogTime: time.Now(), limiter: limiter, - LogInterval: 30 * time.Second, - LogRecordsMetric: 1000, + LogInterval: 5 * time.Second, + LogRecordsMetric: 100, } } @@ -120,7 +120,7 @@ func (t *ProgressTracker) Report() ProgressReport { percent, eta, elapsed, rate := t.Progress() etaStr := "" if eta > 0 { - etaStr = FormatDuration(eta) + etaStr = eta.String() } var w, g int @@ -143,7 +143,7 @@ func (t *ProgressTracker) Report() ProgressReport { resetStr := "" if timeUntilReset > 0 { - resetStr = FormatDuration(timeUntilReset) + resetStr = timeUntilReset.String() } return ProgressReport{ @@ -169,19 +169,3 @@ func formatRate(perMin float64) string { } return fmt.Sprintf("%.0f/min", perMin) } - -func FormatDuration(d time.Duration) string { - if d == 0 { - return "done" - } - hours := int(d.Hours()) - minutes := int(d.Minutes()) % 60 - seconds := int(d.Seconds()) % 60 - if hours > 0 { - return fmt.Sprintf("%dh %dm", hours, minutes) - } - if minutes > 0 { - return fmt.Sprintf("%dm %ds", minutes, seconds) - } - return fmt.Sprintf("%ds", seconds) -} diff --git a/sync/publish.go b/sync/publish.go index 26a97af..70782a1 100644 --- a/sync/publish.go +++ b/sync/publish.go @@ -57,6 +57,7 @@ type PublishOptions struct { ProgressLog func(ProgressReport) Storage cache.Storage Limiter RateLimiter + ClientAgent string } func Publish(ctx context.Context, client AuthClient, opts PublishOptions) PublishResult { @@ -116,11 +117,19 @@ func Publish(ctx context.Context, client AuthClient, opts PublishOptions) Publis } totalSuccess += len(batch) tracker.Increment(len(batch)) + slog.Debug("batch dry run completed", + slog.Int("count", len(batch)), + slog.Int("completed", tracker.Completed)) batch = batch[:0] batchKeys = batchKeys[:0] return nil } + slog.Debug("processing batch", + slog.Int("count", len(batch)), + slog.Int("completed", tracker.Completed), + slog.Int("total", tracker.Total)) + did := client.DID() retryPolicy := retrypolicy.NewBuilder[any](). WithMaxRetries(10). @@ -138,7 +147,7 @@ func Publish(ctx context.Context, client AuthClient, opts PublishOptions) Publis Build() err := failsafe.With(retryPolicy).WithContext(ctx).Run(func() error { - return PublishBatch(ctx, atprotoClient, did, batch, opts.Storage) + return PublishBatch(ctx, atprotoClient, did, batch, opts.Storage, opts.ClientAgent) }) if err != nil { slog.Error("batch failed after retries", @@ -161,6 +170,10 @@ func Publish(ctx context.Context, client AuthClient, opts PublishOptions) Publis totalSuccess += len(batch) tracker.Increment(len(batch)) + slog.Debug("batch published", + slog.Int("count", len(batch)), + slog.Int("completed", tracker.Completed), + slog.Int("total", tracker.Total)) if tracker.ShouldLog() { progressLog(tracker.Report()) @@ -266,12 +279,13 @@ func logResult(success, errors int, startTime time.Time) { slog.String("rate", formatRate(ratePerMinute(success, time.Since(startTime))))) } -func PublishBatch(ctx context.Context, client ATProtoClient, did string, batch []PlayRecord, storage cache.Storage) error { +func PublishBatch(ctx context.Context, client ATProtoClient, did string, batch []PlayRecord, storage cache.Storage, clientAgent string) error { if len(batch) == 0 { return nil } - err := client.ApplyWrites(ctx, RecordType, batch) + atprotoRecords := prepareRecords(batch, clientAgent) + err := client.ApplyWrites(ctx, RecordType, atprotoRecords) if err != nil { slog.Error("batch publish failed", Error(err)) return err @@ -298,11 +312,11 @@ func PublishBatch(ctx context.Context, client ATProtoClient, did string, batch [ return nil } -func prepareRecords(batch []PlayRecord) []PlayRecord { +func prepareRecords(batch []PlayRecord, clientAgent string) []PlayRecord { atprotoRecords := make([]PlayRecord, 0, len(batch)) for _, record := range batch { record.Type = RecordType - record.SubmissionClientAgent = ClientAgent + record.SubmissionClientAgent = clientAgent atprotoRecords = append(atprotoRecords, record) } return atprotoRecords diff --git a/sync/rate_test.go b/sync/rate_test.go index ba33d29..d1c57a3 100644 --- a/sync/rate_test.go +++ b/sync/rate_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/bluesky-social/indigo/atproto/atclient" + "tangled.org/karitham.dev/lazuli/atproto" ) @@ -84,49 +85,45 @@ func (l *testRateLimiter) AllowBulkWrite(ctx context.Context, n int) (time.Time, wCost := n * atproto.WriteOnlyCost gCost := n * atproto.WriteGlobalCost - for { - now := l.clock.now - wKeys, gKeys := l.getAllKeys(now) + now := l.clock.now + wKeys, gKeys := l.getAllKeys(now) - maxWait, err := l.checkQuota(now, wKeys, gKeys, wCost, gCost) - if err != nil { - return now, err - } + maxWait, err := l.checkQuota(now, wKeys, gKeys, wCost, gCost) + if err != nil { + return now, err + } - if maxWait > 0 { - return now, context.DeadlineExceeded - } + if maxWait > 0 { + return now, context.DeadlineExceeded + } - err = l.charge(wKeys, gKeys, wCost, gCost) - if err != nil { - return now, err - } - return now, nil + err = l.charge(wKeys, gKeys, wCost, gCost) + if err != nil { + return now, err } + return now, nil } func (l *testRateLimiter) AllowRead(ctx context.Context) (time.Time, error) { gCost := atproto.ReadGlobalCost - for { - now := l.clock.now - _, gKeys := l.getAllKeys(now) + now := l.clock.now + _, gKeys := l.getAllKeys(now) - maxWait, err := l.checkQuota(now, nil, gKeys, 0, gCost) - if err != nil { - return now, err - } + maxWait, err := l.checkQuota(now, nil, gKeys, 0, gCost) + if err != nil { + return now, err + } - if maxWait > 0 { - return now, context.DeadlineExceeded - } + if maxWait > 0 { + return now, context.DeadlineExceeded + } - err = l.charge(nil, gKeys, 0, gCost) - if err != nil { - return now, err - } - return now, nil + err = l.charge(nil, gKeys, 0, gCost) + if err != nil { + return now, err } + return now, nil } func (l *testRateLimiter) RefundBulkWrite(ctx context.Context, n int, chargedAt time.Time) { @@ -287,6 +284,7 @@ func TestRateLimiter_Weighting(t *testing.T) { func TestRetryExhaustionMarkFailed(t *testing.T) { ctx := context.Background() did := "did:example:123" + clientAgent := "test-agent" storage := newMockStorage() rec1, _ := json.Marshal(PlayRecord{TrackName: "Song 1"}) storage.SaveRecords(did, map[string][]byte{"k1": rec1}) @@ -305,6 +303,7 @@ func TestRetryExhaustionMarkFailed(t *testing.T) { BatchSize: 1, ATProtoClient: client, Storage: storage, + ClientAgent: clientAgent, }) if res.SuccessCount != 0 { diff --git a/sync/record.go b/sync/record.go index 9e24dd1..ddf48b2 100644 --- a/sync/record.go +++ b/sync/record.go @@ -126,9 +126,9 @@ const ( TimeBucketSize = 30 * time.Second MinListenDuration = 30 * time.Second -) -var ClientAgent = "lazuli/dev" + DefaultClientAgent = "lazuli/dev" +) func CreateRecordKey(record PlayRecord) string { return string(syntax.NewTIDFromTime(record.PlayedTime.Time, 0))