update project and selectors

core/internal/repository/pgxrepo.go

@@ -2,39 +2,44 @@ package repository
 
 import (
 	"context"
-	"log/slog"
+	"fmt"
 
-	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/zepyrshut/rating-orama/internal/app"
 	"github.com/zepyrshut/rating-orama/internal/sqlc"
 )
 
 type pgxRepository struct {
 	*sqlc.Queries
-	db *pgxpool.Pool
+	pool *pgxpool.Pool
+	app  *app.ExtendedApp
 }
 
-func NewPGXRepo(db *pgxpool.Pool) ExtendedQuerier {
+var _ ExtendedQuerier = &pgxRepository{}
+
+func NewPGXRepo(pgx *pgxpool.Pool, app *app.ExtendedApp) ExtendedQuerier {
 	return &pgxRepository{
-		Queries: sqlc.New(db),
-		db:      db,
+		Queries: sqlc.New(pgx),
+		pool:    pgx,
+		app:     app,
 	}
 }
 
-func (r *pgxRepository) execTx(ctx context.Context, txFunc func(tx pgx.Tx) error) error {
-	slog.Info("starting transaction", "txFunc", txFunc)
-	tx, err := r.db.Begin(ctx)
+func (r *pgxRepository) execTx(ctx context.Context, fn func(*sqlc.Queries) error) error {
+	tx, err := r.pool.Begin(ctx)
 	if err != nil {
-		slog.Error("failed to start transaction", "error", err)
-		return err
-	}
-	defer tx.Rollback(ctx)
-
-	if err := txFunc(tx); err != nil {
-		slog.Error("failed to execute transaction", "error", err)
 		return err
 	}
 
-	slog.Info("committing transaction", "txFunc", txFunc)
+	q := sqlc.New(tx)
+
+	err = fn(q)
+	if err != nil {
+		if rbErr := tx.Rollback(ctx); rbErr != nil {
+			return fmt.Errorf("tx err: %v, rb err: %v", err, rbErr)
+		}
+		return err
+	}
+
 	return tx.Commit(ctx)
 }

core/internal/repository/querier.go

@@ -2,12 +2,12 @@ package repository
 
 import (
 	"context"
-	"github.com/zepyrshut/rating-orama/internal/scraper"
 
+	"github.com/zepyrshut/rating-orama/internal/scraper"
 	"github.com/zepyrshut/rating-orama/internal/sqlc"
 )
 
 type ExtendedQuerier interface {
 	sqlc.Querier
-	CreateTvShowWithEpisodes(ctx context.Context, tvShow sqlc.CreateTVShowParams, episodes []scraper.Episode) ([]sqlc.Episode, error)
+	CreateTvShowWithEpisodesTX(ctx context.Context, tvShow sqlc.CreateTVShowParams, episodes []scraper.Episode) ([]sqlc.Episode, error)
 }

core/internal/repository/tvshow.go

@@ -1,33 +0,0 @@
-package repository
-
-import (
-	"context"
-	"github.com/jackc/pgx/v5"
-	"github.com/zepyrshut/rating-orama/internal/scraper"
-	"github.com/zepyrshut/rating-orama/internal/sqlc"
-)
-
-func (r *pgxRepository) CreateTvShowWithEpisodes(ctx context.Context, tvShow sqlc.CreateTVShowParams, episodes []scraper.Episode) ([]sqlc.Episode, error) {
-	var sqlcEpisodes []sqlc.Episode
-	err := r.execTx(ctx, func(tx pgx.Tx) error {
-		qtx := r.WithTx(tx)
-		tvShow, err := qtx.CreateTVShow(ctx, tvShow)
-		if err != nil {
-			return err
-		}
-
-		for _, episode := range episodes {
-			sqlcEpisodeParams := episode.ToEpisodeParams(tvShow.ID)
-			episode, err := qtx.CreateEpisodes(ctx, sqlcEpisodeParams)
-			if err != nil {
-				return err
-			}
-
-			sqlcEpisodes = append(sqlcEpisodes, episode)
-		}
-
-		return nil
-	})
-
-	return sqlcEpisodes, err
-}

core/internal/repository/tvshow.tx.go

@@ -0,0 +1,34 @@
+package repository
+
+import (
+	"context"
+
+	"github.com/zepyrshut/rating-orama/internal/scraper"
+	"github.com/zepyrshut/rating-orama/internal/sqlc"
+)
+
+func (r *pgxRepository) CreateTvShowWithEpisodesTX(ctx context.Context, tvShow sqlc.CreateTVShowParams, episodes []scraper.Episode) ([]sqlc.Episode, error) {
+	var err error
+	var episodesSqlc []sqlc.Episode
+
+	err = r.execTx(ctx, func(tx *sqlc.Queries) error {
+		tvShow, err := tx.CreateTVShow(ctx, tvShow)
+		if err != nil {
+			return err
+		}
+
+		for _, episode := range episodes {
+			sqlcEpisodeParams := episode.ToEpisodeParams(tvShow.ID)
+			episode, err := tx.CreateEpisodes(ctx, sqlcEpisodeParams)
+			if err != nil {
+				return err
+			}
+
+			episodesSqlc = append(episodesSqlc, episode)
+		}
+
+		return nil
+	})
+
+	return episodesSqlc, err
+}