add users logic

users/internal/storage/postgres/builder.go

@@ -0,0 +1,11 @@
+package postgres
+
+import sq "github.com/Masterminds/squirrel"
+
+// psql is the shared statement builder, fixed to PostgreSQL $N placeholders so
+// no call site repeats PlaceholderFormat(sq.Dollar).
+//
+// Not everything goes through it. Two genuinely set-based statements stay as
+// raw SQL — claimNext (a FOR UPDATE SKIP LOCKED CTE) and expireLeases (CASE
+// logic in the SET) — because a builder would obscure them, not clarify them.
+var psql = sq.StatementBuilder.PlaceholderFormat(sq.Dollar)

users/internal/storage/postgres/integration_test.go

@@ -0,0 +1,109 @@
+//go:build integration
+
+// Integration tests run against a real PostgreSQL instance supplied through
+// TEST_DATABASE_URL, with the userservice migrations already applied. A real DB
+// is required because the guarantees under test — the unique-email constraint
+// mapping to ErrEmailExists, the ck_users_email_lower check — are properties of
+// Postgres, not of the Go code.
+//
+//	docker compose up -d
+//	TEST_DATABASE_URL='postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable' \
+//	  go test -tags=integration ./internal/storage/postgres/ -v
+package postgres
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/emil28092005/SciMesh/users/internal/domain"
+	"github.com/emil28092005/SciMesh/users/internal/usecase"
+)
+
+func testPool(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+	url := os.Getenv("TEST_DATABASE_URL")
+	if url == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	pool, err := pgxpool.New(context.Background(), url)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	t.Cleanup(pool.Close)
+	return pool
+}
+
+// seedUser inserts a user with a unique email and removes it afterwards, so
+// tests stay independent of each other and of leftovers from earlier runs.
+func seedUser(t *testing.T, repo *UserRepo) *domain.User {
+	t.Helper()
+	email := fmt.Sprintf("it-%s@example.com", uuid.NewString())
+	u, err := domain.NewUser(email, "$2a$04$abcdefghijklmnopqrstuv", time.Now().UTC())
+	if err != nil {
+		t.Fatalf("build user: %v", err)
+	}
+	if err := repo.Insert(context.Background(), u); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = repo.pool.Exec(context.Background(), "DELETE FROM users WHERE id = $1", u.ID)
+	})
+	return u
+}
+
+func TestUserRepoInsertAndGet(t *testing.T) {
+	repo := NewUserRepo(testPool(t))
+	ctx := context.Background()
+	want := seedUser(t, repo)
+
+	byEmail, err := repo.GetByEmail(ctx, want.Email)
+	if err != nil {
+		t.Fatalf("GetByEmail: %v", err)
+	}
+	if byEmail.ID != want.ID || byEmail.Email != want.Email || byEmail.Role != domain.RoleUser {
+		t.Errorf("GetByEmail mismatch: %+v", byEmail)
+	}
+
+	byID, err := repo.GetByID(ctx, want.ID)
+	if err != nil {
+		t.Fatalf("GetByID: %v", err)
+	}
+	if byID.Email != want.Email {
+		t.Errorf("GetByID mismatch: %+v", byID)
+	}
+}
+
+func TestUserRepoDuplicateEmail(t *testing.T) {
+	repo := NewUserRepo(testPool(t))
+	existing := seedUser(t, repo)
+
+	// A second user with the same email must hit the unique constraint and map
+	// to the port's sentinel error.
+	dup, err := domain.NewUser(existing.Email, "$2a$04$abcdefghijklmnopqrstuv", time.Now().UTC())
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = repo.Insert(context.Background(), dup)
+	if !errors.Is(err, usecase.ErrEmailExists) {
+		t.Errorf("got %v, want ErrEmailExists", err)
+	}
+}
+
+func TestUserRepoNotFound(t *testing.T) {
+	repo := NewUserRepo(testPool(t))
+	ctx := context.Background()
+
+	if _, err := repo.GetByID(ctx, uuid.New()); !errors.Is(err, usecase.ErrUserNotFound) {
+		t.Errorf("GetByID unknown: got %v, want ErrUserNotFound", err)
+	}
+	if _, err := repo.GetByEmail(ctx, "ghost@example.com"); !errors.Is(err, usecase.ErrUserNotFound) {
+		t.Errorf("GetByEmail unknown: got %v, want ErrUserNotFound", err)
+	}
+}

users/internal/storage/postgres/retry.go

@@ -0,0 +1,82 @@
+package postgres
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+// Transient PostgreSQL failures. Under concurrent claiming these are expected
+// rather than exceptional: two coordinators touching neighbouring rows can
+// deadlock or fail to serialize, and the correct response is to try again.
+const (
+	codeSerializationFailure = "40001"
+	codeDeadlockDetected     = "40P01"
+	codeTooManyConnections   = "53300"
+	codeCannotConnectNow     = "57P03"
+)
+
+// Retry budget: short and bounded. A worker polling for tasks would rather get
+// a fast error and poll again than have its request hang for half a minute.
+const (
+	retryInitialInterval = 50 * time.Millisecond
+	retryMaxInterval     = 1 * time.Second
+	retryMaxElapsedTime  = 5 * time.Second
+)
+
+// isTransient reports whether err is worth retrying.
+//
+// The default is *not* to retry: a constraint violation or a syntax error will
+// fail identically every time, and retrying it only multiplies the damage.
+func isTransient(err error) bool {
+	if err == nil {
+		return false
+	}
+	// A cancelled caller does not want another attempt.
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return false
+	}
+
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		switch pgErr.Code {
+		case codeSerializationFailure, codeDeadlockDetected,
+			codeTooManyConnections, codeCannotConnectNow:
+			return true
+		default:
+			return false
+		}
+	}
+
+	// Connection-level trouble (dropped socket, closed pool). pgconn knows
+	// whether the query could have been executed before the failure — retrying
+	// a maybe-executed write would risk duplicating it.
+	return pgconn.SafeToRetry(err)
+}
+
+// withRetry runs op, retrying only transient database failures with
+// exponential backoff and jitter, and giving up as soon as ctx is done.
+//
+// Jitter matters here: without it, several coordinators that collide once will
+// retry in lockstep and collide again at exactly the same moment.
+func withRetry(ctx context.Context, op func(context.Context) error) error {
+	b := backoff.NewExponentialBackOff()
+	b.InitialInterval = retryInitialInterval
+	b.MaxInterval = retryMaxInterval
+	b.MaxElapsedTime = retryMaxElapsedTime
+	// RandomizationFactor defaults to 0.5, which is the jitter.
+
+	return backoff.Retry(func() error {
+		err := op(ctx)
+		if err == nil {
+			return nil
+		}
+		if !isTransient(err) {
+			return backoff.Permanent(err) // stop now, do not burn the budget
+		}
+		return err
+	}, backoff.WithContext(b, ctx))
+}

users/internal/storage/postgres/retry_test.go

@@ -0,0 +1,98 @@
+package postgres
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+func TestIsTransient(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"serialization failure", &pgconn.PgError{Code: codeSerializationFailure}, true},
+		{"deadlock", &pgconn.PgError{Code: codeDeadlockDetected}, true},
+		{"too many connections", &pgconn.PgError{Code: codeTooManyConnections}, true},
+		// A unique-violation repeats identically forever — retrying is pointless.
+		{"unique violation", &pgconn.PgError{Code: "23505"}, false},
+		{"syntax error", &pgconn.PgError{Code: "42601"}, false},
+		{"context cancelled", context.Canceled, false},
+		{"deadline exceeded", context.DeadlineExceeded, false},
+		{"unknown error", errors.New("boom"), false},
+		// Wrapping must not hide the cause: errors.As walks the chain.
+		{"wrapped deadlock", errors2Wrap(&pgconn.PgError{Code: codeDeadlockDetected}), true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isTransient(tt.err); got != tt.want {
+				t.Errorf("isTransient(%v) = %v, want %v", tt.err, got, tt.want)
+			}
+		})
+	}
+}
+
+func errors2Wrap(err error) error {
+	return errors.Join(errors.New("query failed"), err)
+}
+
+func TestWithRetrySucceedsAfterTransientFailures(t *testing.T) {
+	calls := 0
+	err := withRetry(context.Background(), func(context.Context) error {
+		calls++
+		if calls < 3 {
+			return &pgconn.PgError{Code: codeSerializationFailure}
+		}
+		return nil
+	})
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if calls != 3 {
+		t.Errorf("calls = %d, want 3", calls)
+	}
+}
+
+func TestWithRetryStopsOnPermanentError(t *testing.T) {
+	permanent := &pgconn.PgError{Code: "23505"} // unique violation
+	calls := 0
+
+	err := withRetry(context.Background(), func(context.Context) error {
+		calls++
+		return permanent
+	})
+
+	if !errors.Is(err, permanent) {
+		t.Errorf("err = %v, want the original error", err)
+	}
+	if calls != 1 {
+		t.Errorf("calls = %d, want 1 — a permanent error must not be retried", calls)
+	}
+}
+
+func TestWithRetryHonoursContextCancellation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	calls := 0
+	start := time.Now()
+	err := withRetry(ctx, func(context.Context) error {
+		calls++
+		return &pgconn.PgError{Code: codeDeadlockDetected}
+	})
+
+	if err == nil {
+		t.Fatal("expected an error once the context expired")
+	}
+	// Must abort at the deadline, not run the full 5s retry budget.
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Errorf("took %v, expected to stop at the context deadline", elapsed)
+	}
+}

users/internal/storage/postgres/tx.go

@@ -0,0 +1,83 @@
+// Package postgres implements the usecase repository ports on PostgreSQL.
+// SQL and pgx types never escape this package.
+package postgres
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// querier is satisfied by both *pgxpool.Pool and pgx.Tx, letting every
+// repository method run identically inside or outside a transaction.
+type querier interface {
+	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
+	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+	SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
+}
+
+// txKey is an unexported struct type, so no other package can collide with it
+// or reach the transaction we stash in the context.
+type txKey struct{}
+
+// TxManager implements usecase.TxManager.
+type TxManager struct {
+	pool *pgxpool.Pool
+}
+
+func NewTxManager(pool *pgxpool.Pool) *TxManager {
+	return &TxManager{pool: pool}
+}
+
+// WithinTx runs fn inside one transaction, committing on success and rolling
+// back on any error or panic.
+//
+// The transaction travels in the context rather than in fn's signature, which
+// is what lets the usecase layer express "do these repository calls atomically"
+// without its port ever mentioning pgx.
+// Retrying happens here, around the whole transaction, and deliberately not
+// inside the repositories. Once Postgres aborts a transaction with a
+// serialization failure or deadlock, every further statement in it fails too —
+// replaying a single query would accomplish nothing. The unit of retry is
+// Begin → fn → Commit.
+//
+// This is safe because fn re-reads its rows (via GetForUpdate) on each attempt,
+// so a retry starts from the current state rather than stale entities.
+func (m *TxManager) WithinTx(ctx context.Context, fn func(ctx context.Context) error) error {
+	if _, ok := ctx.Value(txKey{}).(pgx.Tx); ok {
+		// Already inside a transaction — join it. Retrying here would be wrong
+		// twice over: the outer transaction owns the retry, and re-running fn
+		// alone cannot undo what the outer one already wrote.
+		return fn(ctx)
+	}
+
+	return withRetry(ctx, func(ctx context.Context) error {
+		return m.runTx(ctx, fn)
+	})
+}
+
+func (m *TxManager) runTx(ctx context.Context, fn func(ctx context.Context) error) error {
+	tx, err := m.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	// Rollback after a successful Commit is a no-op, so this defer is safe and
+	// also covers the panic path.
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	if err := fn(context.WithValue(ctx, txKey{}, tx)); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+// conn returns the transaction bound to ctx, or the pool when there is none.
+func conn(ctx context.Context, pool *pgxpool.Pool) querier {
+	if tx, ok := ctx.Value(txKey{}).(pgx.Tx); ok {
+		return tx
+	}
+	return pool
+}

users/internal/storage/postgres/user_repo.go

@@ -0,0 +1,85 @@
+package postgres
+
+import (
+	"context"
+	"errors"
+
+	sq "github.com/Masterminds/squirrel"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/emil28092005/SciMesh/users/internal/domain"
+	"github.com/emil28092005/SciMesh/users/internal/usecase"
+)
+
+// uniqueViolation is PostgreSQL's SQLSTATE for a unique-constraint breach.
+const uniqueViolation = "23505"
+
+var userColumns = []string{"id", "email", "password_hash", "role", "created_at", "updated_at"}
+
+// UserRepo implements usecase.UserRepository on PostgreSQL.
+type UserRepo struct {
+	pool *pgxpool.Pool
+}
+
+func NewUserRepo(pool *pgxpool.Pool) *UserRepo {
+	return &UserRepo{pool: pool}
+}
+
+func (r *UserRepo) Insert(ctx context.Context, u *domain.User) error {
+	sql, args, err := psql.Insert("users").
+		Columns(userColumns...).
+		Values(u.ID, u.Email, u.PasswordHash, string(u.Role), u.CreatedAt, u.UpdatedAt).
+		ToSql()
+	if err != nil {
+		return err
+	}
+	if _, err := conn(ctx, r.pool).Exec(ctx, sql, args...); err != nil {
+		// A concurrent insert of the same email surfaces as a unique violation
+		// on uq_users_email; translate it to the port's sentinel so the use
+		// case never sees a driver type.
+		if isUniqueViolation(err) {
+			return usecase.ErrEmailExists
+		}
+		return err
+	}
+	return nil
+}
+
+func (r *UserRepo) GetByEmail(ctx context.Context, email string) (*domain.User, error) {
+	return r.getBy(ctx, sq.Eq{"email": email})
+}
+
+func (r *UserRepo) GetByID(ctx context.Context, id uuid.UUID) (*domain.User, error) {
+	return r.getBy(ctx, sq.Eq{"id": id})
+}
+
+func (r *UserRepo) getBy(ctx context.Context, pred sq.Sqlizer) (*domain.User, error) {
+	sql, args, err := psql.Select(userColumns...).From("users").Where(pred).ToSql()
+	if err != nil {
+		return nil, err
+	}
+	return scanUser(conn(ctx, r.pool).QueryRow(ctx, sql, args...))
+}
+
+func scanUser(row pgx.Row) (*domain.User, error) {
+	var (
+		u    domain.User
+		role string
+	)
+	if err := row.Scan(&u.ID, &u.Email, &u.PasswordHash, &role, &u.CreatedAt, &u.UpdatedAt); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, usecase.ErrUserNotFound
+		}
+		return nil, err
+	}
+	u.Role = domain.Role(role)
+	return &u, nil
+}
+
+func isUniqueViolation(err error) bool {
+	var pgErr *pgconn.PgError
+	return errors.As(err, &pgErr) && pgErr.Code == uniqueViolation
+}