feat: add transactional InTx to playback and watchlist repos

internal/playback/repository/repository.go

@@ -2,16 +2,37 @@ package repository
 
 import (
 	"context"
+	"database/sql"
 	"mal/internal/db"
 	"mal/internal/domain"
 )
 
 type playbackRepository struct {
+	sqlDB   *sql.DB
 	queries *db.Queries
 }
 
-func NewPlaybackRepository(queries *db.Queries) domain.PlaybackRepository {
-	return &playbackRepository{queries: queries}
+func NewPlaybackRepository(sqlDB *sql.DB, queries *db.Queries) domain.PlaybackRepository {
+	return &playbackRepository{sqlDB: sqlDB, queries: queries}
+}
+
+func (r *playbackRepository) InTx(ctx context.Context, fn func(ctx context.Context, repo domain.PlaybackRepository) error) error {
+	if r.sqlDB == nil {
+		return fn(ctx, r)
+	}
+
+	tx, err := r.sqlDB.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+
+	txRepo := &playbackRepository{sqlDB: nil, queries: r.queries.WithTx(tx)}
+	if err := fn(ctx, txRepo); err != nil {
+		_ = tx.Rollback()
+		return err
+	}
+
+	return tx.Commit()
 }
 
 func (r *playbackRepository) GetWatchListEntry(ctx context.Context, params db.GetWatchListEntryParams) (db.WatchListEntry, error) {

internal/playback/service/service.go

@@ -301,38 +301,41 @@ func (s *playbackService) BuildWatchData(ctx context.Context, animeID int, title
 }
 
 func (s *playbackService) CompleteAnime(ctx context.Context, userID string, animeID int64) error {
-	entry, err := s.repo.GetWatchListEntry(ctx, db.GetWatchListEntryParams{
-		UserID:  userID,
-		AnimeID: animeID,
-	})
-	if err != nil || entry.Status != "completed" {
-		_, err = s.repo.UpsertWatchListEntry(ctx, db.UpsertWatchListEntryParams{
-			ID:                 uuid.New().String(),
+	if err := s.repo.InTx(ctx, func(txCtx context.Context, repo domain.PlaybackRepository) error {
+		entry, err := repo.GetWatchListEntry(txCtx, db.GetWatchListEntryParams{
+			UserID:  userID,
+			AnimeID: animeID,
+		})
+		if err != nil || entry.Status != "completed" {
+			_, err = repo.UpsertWatchListEntry(txCtx, db.UpsertWatchListEntryParams{
+				ID:                 uuid.New().String(),
+				UserID:             userID,
+				AnimeID:            animeID,
+				Status:             "completed",
+				CurrentEpisode:     sql.NullInt64{Valid: false},
+				CurrentTimeSeconds: 0,
+			})
+			if err != nil {
+				return err
+			}
+		}
+
+		if err := repo.DeleteContinueWatchingEntry(txCtx, db.DeleteContinueWatchingEntryParams{
+			UserID:  userID,
+			AnimeID: animeID,
+		}); err != nil {
+			return err
+		}
+		return repo.SaveWatchProgress(txCtx, db.SaveWatchProgressParams{
 			UserID:             userID,
 			AnimeID:            animeID,
-			Status:             "completed",
 			CurrentEpisode:     sql.NullInt64{Valid: false},
 			CurrentTimeSeconds: 0,
 		})
-		if err != nil {
-			return err
-		}
+	}); err != nil {
+		return err
 	}
 
-	if err := s.repo.DeleteContinueWatchingEntry(ctx, db.DeleteContinueWatchingEntryParams{
-		UserID:  userID,
-		AnimeID: animeID,
-	}); err != nil {
-		return err
-	}
-	if err := s.repo.SaveWatchProgress(ctx, db.SaveWatchProgressParams{
-		UserID:             userID,
-		AnimeID:            animeID,
-		CurrentEpisode:     sql.NullInt64{Valid: false},
-		CurrentTimeSeconds: 0,
-	}); err != nil {
-		return err
-	}
 	if err := s.auditSvc.Record(ctx, domain.AuditEvent{
 		UserID:       userID,
 		Action:       "watch_completed",