merch-api/internal/merch/repository.go

package merch

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"github.com/jackc/pgx/v5/pgxpool"
	"strings"
	"time"
)

type Repository interface {
	// createMerch creates new merch record and inserts extra data if given
	createMerch(ctx context.Context, merch *Merch, extra []ExtraData) error

	// getMany returns list of only main merch record, without origins extra data
	getMany(ctx context.Context, userId int64) ([]merchDTO, error)

	getMerchIdByUuid(ctx context.Context, userId int64, uuid string) (int64, error)
	getMerchUuidMap(ctx context.Context, merchUuids []string) (map[string]int64, error)

	updateMerch(ctx context.Context, userId int64, merch *updateMerchDTO) (*merchDTO, error)
	updateExtraData(ctx context.Context, merchId int64, insertData []ExtraData) ([]ExtraData, error)

	// deleteOneMerchRecord sets deleted_at in merch + extra tables
	deleteOneMerchRecord(ctx context.Context, userId int64, merchUuid string, delTime time.Time) error

	Origins
	Prices
	Tasks
}

type Origins interface {
	createOrigin(ctx context.Context, origin *Origin) error
	getOrigins(ctx context.Context) ([]Origin, error)
	deleteOriginByName(ctx context.Context, name string, deletedAt sql.NullTime) error
}

type Prices interface {
	insertPrices(ctx context.Context, prices []Price) error
}

type Tasks interface {
	getTaskData(ctx context.Context) ([]taskData, error)
}

type repo struct {
	db *pgxpool.Pool
}

func newRepo(db *pgxpool.Pool) Repository {
	return &repo{
		db: db,
	}
}

func (r *repo) createOrigin(ctx context.Context, origin *Origin) error {
	q := `INSERT INTO merch_origins (created_at, deleted_at, name) VALUES ($1, $2, $3)`

	_, err := r.db.Exec(ctx, q, origin.CreatedAt, origin.DeletedAt, origin.Name)
	if err != nil {
		return err
	}

	return nil
}

func (r *repo) getOrigins(ctx context.Context) ([]Origin, error) {
	q := `SELECT * FROM merch_origins WHERE deleted_at IS NULL`

	rows, err := r.db.Query(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var origins []Origin
	for rows.Next() {
		var o Origin
		if err = rows.Scan(&o.Id, &o.CreatedAt, &o.DeletedAt, &o.Name); err != nil {
			return nil, err
		}
		origins = append(origins, o)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return origins, nil
}

func (r *repo) deleteOriginByName(ctx context.Context, name string, deletedAt sql.NullTime) error {
	q := `UPDATE merch_origins SET deleted_at = $1 WHERE name = $2`

	_, err := r.db.Exec(ctx, q, deletedAt.Time, name)
	if err != nil {
		return err
	}

	return nil
}

//Merch crud

func (r *repo) createMerch(ctx context.Context, merch *Merch, extra []ExtraData) error {
	tx, err := r.db.Begin(ctx)
	if err != nil {
		return err
	}

	qMerch := `INSERT INTO merch (created_at, updated_at, deleted_at, merch_uuid, user_id, name)
				VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`

	var merchId int64
	err = tx.
		QueryRow(ctx, qMerch, merch.CreatedAt, merch.UpdatedAt, merch.DeletedAt, merch.MerchUuid, merch.UserId, merch.Name).
		Scan(&merchId)
	if err != nil {
		tx.Rollback(ctx)
		return err
	}

	if extra == nil {
		return tx.Commit(ctx)
	}

	countArgs := 1
	var insertFields []string
	var insertArgs []interface{}

	for _, item := range extra {
		insertFields = append(insertFields, fmt.Sprintf("($%v, $%v, $%v, $%v, $%v, $%v)",
			countArgs, countArgs+1, countArgs+2, countArgs+3, countArgs+4, countArgs+5))

		insertArgs = append(insertArgs, item.CreatedAt, item.UpdatedAt, item.DeletedAt, merchId, item.OriginId, item.URL)

		countArgs += 6
	}

	qExtra := fmt.Sprintf(
		"INSERT INTO merch_extra_data (created_at, updated_at, deleted_at, merch_id, origin_id, url) VALUES %v",
		strings.Join(insertFields, ","))

	_, err = tx.Exec(ctx, qExtra, insertArgs...)
	if err != nil {
		tx.Rollback(ctx)
		return err
	}

	return tx.Commit(ctx)
}

func (r *repo) getMany(ctx context.Context, userId int64) ([]merchDTO, error) {
	q := `SELECT created_at, updated_at, merch_uuid, name FROM merch WHERE deleted_at IS NULL AND user_id = $1`

	rows, err := r.db.Query(ctx, q, userId)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var result []merchDTO
	for rows.Next() {
		var m merchDTO
		if err = rows.Scan(&m.CreatedAt, &m.UpdatedAt, &m.MerchUuid, &m.Name); err != nil {
			return nil, err
		}
		result = append(result, m)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

func (r *repo) getMerchIdByUuid(ctx context.Context, userId int64, uuid string) (int64, error) {
	q := `SELECT id FROM merch WHERE deleted_at IS NULL AND merch_uuid = $1 AND user_id = $2`

	var merchId int64
	if err := r.db.QueryRow(ctx, q, uuid, userId).Scan(&merchId); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, nil
		} else {
			return 0, err
		}
	}

	return merchId, nil
}

func (r *repo) getMerchUuidMap(ctx context.Context, merchUuids []string) (map[string]int64, error) {
	q := `SELECT merch_uuid, id FROM merch WHERE deleted_at IS NULL AND merch_uuid = ANY($1)`

	rows, err := r.db.Query(ctx, q, merchUuids)
	if err != nil {
		return nil, err
	}

	var merchUuidMap map[string]int64
	for rows.Next() {
		var (
			uuid string
			id   int64
		)
		if err = rows.Scan(&uuid, &id); err != nil {
			rows.Close()
			return nil, err
		}
		merchUuidMap[uuid] = id
	}

	rows.Close()
	if err = rows.Err(); err != nil {
		return nil, err
	}

	return merchUuidMap, nil
}

func (r *repo) deleteOneMerchRecord(ctx context.Context, userId int64, merchUuid string, delTime time.Time) error {
	tx, err := r.db.Begin(ctx)
	if err != nil {
		return err
	}

	var merch_id int64
	qMerch := `UPDATE merch SET deleted_at = $1 WHERE merch_uuid = $2 AND user_id = $3 RETURNING id`
	if err = tx.QueryRow(ctx, qMerch, delTime, merchUuid, userId).Scan(&merch_id); err != nil {
		tx.Rollback(ctx)
		return err
	}

	if merch_id != 0 {
		qExtra := `UPDATE merch_extra_data SET deleted_at = $1 WHERE merch_id = $2`
		_, err = tx.Exec(ctx, qExtra, delTime, merch_id)
	}

	return tx.Commit(ctx)
}

func (r *repo) updateMerch(ctx context.Context, userId int64, merch *updateMerchDTO) (*merchDTO, error) {
	q := `UPDATE merch SET name = $1 WHERE merch_uuid = $2 AND user_id = $3
          RETURNING created_at, updated_at, merch_uuid, name`

	var result merchDTO
	if err := r.db.
		QueryRow(ctx, q, merch.Name, merch.MerchUuid, userId).
		Scan(&result.CreatedAt, &result.UpdatedAt, &result.MerchUuid, &result.Name); err != nil {
		return nil, err
	}

	return &result, nil
}

func (r *repo) updateExtraData(ctx context.Context, merchId int64, insertData []ExtraData) ([]ExtraData, error) {
	q := `
		INSERT INTO merch_extra_data (merch_id, origin_id, url, created_at, updated_at)
		SELECT $1, src.origin_id, src.new_url, src.created_at, src.updated_at
		FROM UNNEST(
			$2::text[],
			$3::bigint[],
			$4::timestamptz[],
			$5::timestamptz[])
		AS src(new_url, origin_id, created_at, updated_at)
		ON CONFLICT (merch_id, origin_id) DO UPDATE SET url = COALESCE(NULLIF(EXCLUDED.url, ''), merch_extra_data.url), updated_at = NOW()
		RETURNING created_at, updated_at, merch_id, origin_id, url;
		`

	var (
		urls      []string
		origins   []int64
		createdAt []time.Time
		updatedAt []sql.NullTime
	)

	for _, data := range insertData {
		urls = append(urls, data.URL)
		origins = append(origins, data.OriginId)
		createdAt = append(createdAt, data.CreatedAt)
		updatedAt = append(updatedAt, data.UpdatedAt)
	}

	rows, err := r.db.Query(ctx, q, merchId, urls, origins, createdAt, updatedAt)
	if err != nil {
		return nil, err
	}

	var result []ExtraData
	for rows.Next() {
		var m ExtraData
		if err = rows.Scan(&m.CreatedAt, &m.UpdatedAt, &m.MerchId, &m.OriginId, &m.URL); err != nil {
			return nil, err
		}
		result = append(result, m)
	}

	rows.Close() //must be before rows.Err check!
	if err = rows.Err(); err != nil {
		return nil, err
	}

	return result, nil
}

func (r *repo) insertPrices(ctx context.Context, prices []Price) error {
	q := `
		INSERT INTO merch_prices (created_at, updated_at, merch_id, origin_id, price)
		SELECT $1, $2, src.merch_id, src.origin_id, src.price
		FROM UNNEST(
		     $3::bigint[],
		     $4::bigint[],
		     $5::int[]
		) AS src (merch_id, origin_id, price)
		`

	var (
		merchIds    []int64
		originIds   []int64
		priceValues []int
	)

	for _, price := range prices {
		merchIds = append(merchIds, price.MerchId)
		originIds = append(originIds, price.OriginId)
		priceValues = append(priceValues, price.Price)
	}

	_, err := r.db.Exec(ctx, q, merchIds, originIds, priceValues)
	if err != nil {
		return err
	}

	return nil
}

func (r *repo) getTaskData(ctx context.Context) ([]taskData, error) {
	q := `SELECT m.merch_uuid, med.url, mo.name
		  FROM merch AS m
		  JOIN merch_extra_data AS med ON m.id = med.merch_id
		  JOIN merch_origins AS mo ON mo.id = med.origin_id
		  WHERE m.deleted_at IS NULL
		  AND med.deleted_at IS NULL
		  AND mo.deleted_at IS NULL
		  `
	rows, err := r.db.Query(ctx, q)
	if err != nil {
		return nil, err
	}

	var result []taskData
	for rows.Next() {
		var t taskData
		if err = rows.Scan(&t.MerchUuid, &t.Url, &t.OriginName); err != nil {
			return nil, err
		}
		result = append(result, t)
	}

	rows.Close()
	if err = rows.Err(); err != nil {
		return nil, err
	}

	return result, nil
}