From a3db1a1e679a9757503f677924ea34b602e3bacb Mon Sep 17 00:00:00 2001 From: Efremenko Arhip Date: Sun, 26 Jul 2026 14:38:24 +0300 Subject: [PATCH] add users logic --- coordinator/cmd/coordinator/main.go | 2 +- coordinator/go.mod | 1 + coordinator/go.sum | 2 + coordinator/internal/authctx/authctx.go | 37 +++ coordinator/internal/domain/job.go | 6 +- coordinator/internal/infra/config.go | 12 + .../internal/storage/postgres/job_repo.go | 8 +- coordinator/internal/token/verifier.go | 58 +++++ coordinator/internal/token/verifier_test.go | 79 ++++++ .../internal/transport/http/middleware.go | 45 +++- coordinator/internal/transport/http/server.go | 9 +- .../internal/transport/http/server_test.go | 2 +- coordinator/internal/usecase/job.go | 7 + coordinator/internal/usecase/ownership.go | 39 +++ .../internal/usecase/ownership_test.go | 66 +++++ coordinator/internal/usecase/reduce.go | 3 + coordinator/internal/usecase/upload.go | 1 + .../migrations/0011_job_owner.down.sql | 6 + coordinator/migrations/0011_job_owner.up.sql | 14 ++ users/.dockerignore | 14 ++ users/.env.example | 32 +++ users/.gitignore | 6 + users/.golangci.yml | 54 ++++ users/ARCHITECTURE.md | 144 +++++++++++ users/Dockerfile | 52 ++++ users/Makefile | 96 +++++++ users/README.md | 230 +++++++++++++++++ users/api/requests.http | 234 ++++++++++++++++++ users/cmd/userservice/main.go | 68 +++++ users/docker-compose.yml | 81 ++++++ users/go.mod | 24 ++ users/go.sum | 45 ++++ users/internal/auth/jwt.go | 67 +++++ users/internal/auth/jwt_test.go | 74 ++++++ users/internal/auth/password.go | 40 +++ users/internal/auth/password_test.go | 30 +++ users/internal/domain/errors.go | 11 + users/internal/domain/user.go | 80 ++++++ users/internal/domain/user_test.go | 62 +++++ users/internal/infra/clock.go | 13 + users/internal/infra/config.go | 151 +++++++++++ users/internal/infra/db.go | 65 +++++ users/internal/infra/logging.go | 65 +++++ users/internal/infra/server.go | 74 ++++++ users/internal/memstore/memstore.go | 66 +++++ users/internal/storage/postgres/builder.go | 11 + .../storage/postgres/integration_test.go | 109 ++++++++ users/internal/storage/postgres/retry.go | 82 ++++++ users/internal/storage/postgres/retry_test.go | 98 ++++++++ users/internal/storage/postgres/tx.go | 83 +++++++ users/internal/storage/postgres/user_repo.go | 85 +++++++ users/internal/transport/http/dto.go | 41 +++ users/internal/transport/http/errors.go | 61 +++++ users/internal/transport/http/handlers.go | 85 +++++++ users/internal/transport/http/middleware.go | 133 ++++++++++ users/internal/transport/http/server.go | 41 +++ users/internal/transport/http/server_test.go | 167 +++++++++++++ users/internal/usecase/errors.go | 18 ++ users/internal/usecase/login.go | 42 ++++ users/internal/usecase/ports.go | 41 +++ users/internal/usecase/register.go | 56 +++++ users/internal/usecase/usecase_test.go | 133 ++++++++++ users/migrations/0001_users.down.sql | 6 + users/migrations/0001_users.up.sql | 26 ++ users/scripts/smoke.sh | 52 ++++ 65 files changed, 3626 insertions(+), 19 deletions(-) create mode 100644 coordinator/internal/authctx/authctx.go create mode 100644 coordinator/internal/token/verifier.go create mode 100644 coordinator/internal/token/verifier_test.go create mode 100644 coordinator/internal/usecase/ownership.go create mode 100644 coordinator/internal/usecase/ownership_test.go create mode 100644 coordinator/migrations/0011_job_owner.down.sql create mode 100644 coordinator/migrations/0011_job_owner.up.sql create mode 100644 users/.dockerignore create mode 100644 users/.env.example create mode 100644 users/.gitignore create mode 100644 users/.golangci.yml create mode 100644 users/ARCHITECTURE.md create mode 100644 users/Dockerfile create mode 100644 users/Makefile create mode 100644 users/README.md create mode 100644 users/api/requests.http create mode 100644 users/cmd/userservice/main.go create mode 100644 users/docker-compose.yml create mode 100644 users/go.mod create mode 100644 users/go.sum create mode 100644 users/internal/auth/jwt.go create mode 100644 users/internal/auth/jwt_test.go create mode 100644 users/internal/auth/password.go create mode 100644 users/internal/auth/password_test.go create mode 100644 users/internal/domain/errors.go create mode 100644 users/internal/domain/user.go create mode 100644 users/internal/domain/user_test.go create mode 100644 users/internal/infra/clock.go create mode 100644 users/internal/infra/config.go create mode 100644 users/internal/infra/db.go create mode 100644 users/internal/infra/logging.go create mode 100644 users/internal/infra/server.go create mode 100644 users/internal/memstore/memstore.go create mode 100644 users/internal/storage/postgres/builder.go create mode 100644 users/internal/storage/postgres/integration_test.go create mode 100644 users/internal/storage/postgres/retry.go create mode 100644 users/internal/storage/postgres/retry_test.go create mode 100644 users/internal/storage/postgres/tx.go create mode 100644 users/internal/storage/postgres/user_repo.go create mode 100644 users/internal/transport/http/dto.go create mode 100644 users/internal/transport/http/errors.go create mode 100644 users/internal/transport/http/handlers.go create mode 100644 users/internal/transport/http/middleware.go create mode 100644 users/internal/transport/http/server.go create mode 100644 users/internal/transport/http/server_test.go create mode 100644 users/internal/usecase/errors.go create mode 100644 users/internal/usecase/login.go create mode 100644 users/internal/usecase/ports.go create mode 100644 users/internal/usecase/register.go create mode 100644 users/internal/usecase/usecase_test.go create mode 100644 users/migrations/0001_users.down.sql create mode 100644 users/migrations/0001_users.up.sql create mode 100755 users/scripts/smoke.sh diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 46b077c..2206bec 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -110,7 +110,7 @@ func run() error { // pool.Ping backs /health: readiness means the database answers, not just // that the process is alive. - api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, pool.Ping) + api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, cfg.JWTSecret, pool.Ping) err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken)) // Shutdown order matters, and defers alone cannot express it (they run diff --git a/coordinator/go.mod b/coordinator/go.mod index 4fc8bcc..3b6d8e7 100644 --- a/coordinator/go.mod +++ b/coordinator/go.mod @@ -5,6 +5,7 @@ go 1.22 require ( github.com/Masterminds/squirrel v1.5.4 github.com/cenkalti/backoff/v4 v4.3.0 + github.com/golang-jwt/jwt/v5 v5.2.1 github.com/google/uuid v1.6.0 github.com/jackc/pgx/v5 v5.6.0 github.com/joho/godotenv v1.5.1 diff --git a/coordinator/go.sum b/coordinator/go.sum index d9c880a..8eea60b 100644 --- a/coordinator/go.sum +++ b/coordinator/go.sum @@ -5,6 +5,8 @@ github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyY github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/golang-jwt/jwt/v5 v5.2.1 h1:OuVbFODueb089Lh128TAcimifWaLhJwVflnrgM17wHk= +github.com/golang-jwt/jwt/v5 v5.2.1/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= diff --git a/coordinator/internal/authctx/authctx.go b/coordinator/internal/authctx/authctx.go new file mode 100644 index 0000000..74bf86e --- /dev/null +++ b/coordinator/internal/authctx/authctx.go @@ -0,0 +1,37 @@ +// Package authctx carries the authenticated requester across the transport and +// use-case layers without either one importing the other. The HTTP middleware +// stamps a Requester after verifying a user's JWT; the job use cases read it to +// record ownership and to enforce that a non-admin only touches their own jobs. +package authctx + +import ( + "context" + + "github.com/google/uuid" +) + +// Requester is the identity behind a request, derived from a verified JWT. +// A request authenticated only by the shared worker/service token carries no +// Requester at all (From returns ok=false), which is how worker traffic and +// legacy unauthenticated-user traffic stay owner-less. +type Requester struct { + UserID uuid.UUID + Role string +} + +// IsAdmin reports whether the requester may act on any user's jobs. +func (r Requester) IsAdmin() bool { return r.Role == "admin" } + +type ctxKey struct{} + +// With returns a copy of ctx carrying r. +func With(ctx context.Context, r Requester) context.Context { + return context.WithValue(ctx, ctxKey{}, r) +} + +// From returns the requester stamped by the middleware, or ok=false when the +// request was not authenticated as a user. +func From(ctx context.Context) (Requester, bool) { + r, ok := ctx.Value(ctxKey{}).(Requester) + return r, ok +} diff --git a/coordinator/internal/domain/job.go b/coordinator/internal/domain/job.go index 16ec023..67ad075 100644 --- a/coordinator/internal/domain/job.go +++ b/coordinator/internal/domain/job.go @@ -19,7 +19,11 @@ const ( // Job is one user submission that fans out into one or more tasks. type Job struct { - ID uuid.UUID + ID uuid.UUID + // OwnerID is the userservice user who submitted the job (JWT `sub`). nil + // when the job was created without user authentication. Not a foreign key: + // users live in a separate service/database. + OwnerID *uuid.UUID Workload string InputURI string // external input URI; empty for uploaded datasets InputArtifactID *uuid.UUID // uploaded input artifact; nil for URI submissions diff --git a/coordinator/internal/infra/config.go b/coordinator/internal/infra/config.go index 7d6db0e..37f8d41 100644 --- a/coordinator/internal/infra/config.go +++ b/coordinator/internal/infra/config.go @@ -26,6 +26,12 @@ type Config struct { Token string // Local operator UI credential. Empty disables the embedded UI entirely. UIToken string + // Shared HS256 secret used to verify userservice-issued JWTs. When set, a + // submitter may authenticate with a JWT (in addition to workers using the + // shared token) and their jobs are stamped with owner_id. Empty disables + // user-JWT auth entirely — the pre-userservice behaviour. Must match the + // userservice's JWT_SECRET. + JWTSecret string // Minimum log level: debug, info, warn, error. LogLevel string @@ -80,6 +86,7 @@ func LoadConfig() (Config, error) { // former name, still honoured so existing .env files keep working. Token: getEnv("COORDINATOR_TOKEN", os.Getenv("WORKER_AUTH_TOKEN")), UIToken: os.Getenv("UI_AUTH_TOKEN"), + JWTSecret: os.Getenv("JWT_SECRET"), LogLevel: getEnv("LOG_LEVEL", "info"), LogFile: os.Getenv("LOG_FILE"), StorageDir: getEnv("COORDINATOR_STORAGE_DIR", "./data"), @@ -100,6 +107,11 @@ func LoadConfig() (Config, error) { if cfg.UIToken != "" && cfg.Token != "" && cfg.UIToken == cfg.Token { return Config{}, fmt.Errorf("UI_AUTH_TOKEN must differ from the worker auth token") } + // A short secret makes the HMAC brute-forceable; refuse a weak one rather + // than verify tokens against it. + if cfg.JWTSecret != "" && len(cfg.JWTSecret) < 32 { + return Config{}, fmt.Errorf("JWT_SECRET must be at least 32 bytes") + } var err error if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil { diff --git a/coordinator/internal/storage/postgres/job_repo.go b/coordinator/internal/storage/postgres/job_repo.go index d889a23..eb859f2 100644 --- a/coordinator/internal/storage/postgres/job_repo.go +++ b/coordinator/internal/storage/postgres/job_repo.go @@ -28,14 +28,15 @@ var _ usecase.JobRepository = (*JobRepo)(nil) var jobColumns = []string{ "id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at", "input_artifact_id", "result_artifact_id", "error_code", "error_message", "reducer_started_at", + "owner_id", } // Insert runs inside the caller's transaction, alongside the job's tasks — that // is what makes "all tasks or none" hold. func (r *JobRepo) Insert(ctx context.Context, j *domain.Job) error { sql, args, err := psql.Insert("jobs"). - Columns("id", "workload", "input_uri", "parameters", "status", "created_at"). - Values(j.ID, j.Workload, j.InputURI, jsonbOrEmpty(j.Parameters), string(j.Status), j.CreatedAt). + Columns("id", "workload", "input_uri", "parameters", "status", "created_at", "owner_id"). + Values(j.ID, j.Workload, j.InputURI, jsonbOrEmpty(j.Parameters), string(j.Status), j.CreatedAt, j.OwnerID). ToSql() if err != nil { return err @@ -59,7 +60,8 @@ func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) { ) err = conn(ctx, r.pool).QueryRow(ctx, sql, args...).Scan( &j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt, - &j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt) + &j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt, + &j.OwnerID) if errors.Is(err, pgx.ErrNoRows) { return nil, domain.ErrJobNotFound } diff --git a/coordinator/internal/token/verifier.go b/coordinator/internal/token/verifier.go new file mode 100644 index 0000000..a366dcb --- /dev/null +++ b/coordinator/internal/token/verifier.go @@ -0,0 +1,58 @@ +// Package token verifies the HS256 JWTs minted by the userservice. The +// coordinator only ever *verifies* — it never issues — so this is a deliberately +// small counterpart to the userservice's issuer. Verification is local: the +// shared secret is enough, with no runtime call back to the userservice. +package token + +import ( + "fmt" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" +) + +// Claims is the subset of a userservice token the coordinator cares about. +type Claims struct { + UserID uuid.UUID + Role string +} + +// Verifier checks tokens against the shared HS256 secret. +type Verifier struct { + secret []byte +} + +// NewVerifier returns a Verifier, or nil when secret is empty — a nil Verifier +// means user-JWT auth is disabled and only the shared service token is accepted. +func NewVerifier(secret string) *Verifier { + if secret == "" { + return nil + } + return &Verifier{secret: []byte(secret)} +} + +type claims struct { + Role string `json:"role"` + jwt.RegisteredClaims +} + +// Verify checks the signature and expiry and returns the identity. It pins the +// algorithm to HMAC, rejecting a token that asks for "none" or an RS256 public +// key — the classic algorithm-substitution attack. +func (v *Verifier) Verify(raw string) (Claims, error) { + var c claims + _, err := jwt.ParseWithClaims(raw, &c, func(t *jwt.Token) (any, error) { + if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok { + return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"]) + } + return v.secret, nil + }) + if err != nil { + return Claims{}, err + } + id, err := uuid.Parse(c.Subject) + if err != nil { + return Claims{}, fmt.Errorf("token subject is not a uuid: %w", err) + } + return Claims{UserID: id, Role: c.Role}, nil +} diff --git a/coordinator/internal/token/verifier_test.go b/coordinator/internal/token/verifier_test.go new file mode 100644 index 0000000..238d150 --- /dev/null +++ b/coordinator/internal/token/verifier_test.go @@ -0,0 +1,79 @@ +package token + +import ( + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" +) + +const secret = "coordinator-verify-secret-32-bytes!!" + +func sign(t *testing.T, method jwt.SigningMethod, key any, sub, role string, exp time.Time) string { + t.Helper() + tok := jwt.NewWithClaims(method, claims{ + Role: role, + RegisteredClaims: jwt.RegisteredClaims{ + Subject: sub, + ExpiresAt: jwt.NewNumericDate(exp), + }, + }) + raw, err := tok.SignedString(key) + if err != nil { + t.Fatalf("sign: %v", err) + } + return raw +} + +func TestNewVerifierNilWhenNoSecret(t *testing.T) { + if NewVerifier("") != nil { + t.Error("empty secret must yield a nil verifier (auth disabled)") + } +} + +func TestVerifyRoundTrip(t *testing.T) { + v := NewVerifier(secret) + id := uuid.New() + raw := sign(t, jwt.SigningMethodHS256, []byte(secret), id.String(), "admin", time.Now().Add(time.Hour)) + + claims, err := v.Verify(raw) + if err != nil { + t.Fatalf("verify: %v", err) + } + if claims.UserID != id { + t.Errorf("UserID = %v, want %v", claims.UserID, id) + } + if claims.Role != "admin" { + t.Errorf("Role = %q, want admin", claims.Role) + } +} + +func TestVerifyRejectsExpired(t *testing.T) { + v := NewVerifier(secret) + raw := sign(t, jwt.SigningMethodHS256, []byte(secret), uuid.New().String(), "user", time.Now().Add(-time.Minute)) + if _, err := v.Verify(raw); err == nil { + t.Error("expired token accepted") + } +} + +func TestVerifyRejectsWrongSecret(t *testing.T) { + raw := sign(t, jwt.SigningMethodHS256, []byte(secret), uuid.New().String(), "user", time.Now().Add(time.Hour)) + if _, err := NewVerifier("another-secret-also-at-least-32-byte").Verify(raw); err == nil { + t.Error("token verified under the wrong secret") + } +} + +func TestVerifyRejectsNoneAlg(t *testing.T) { + raw := sign(t, jwt.SigningMethodNone, jwt.UnsafeAllowNoneSignatureType, uuid.New().String(), "admin", time.Now().Add(time.Hour)) + if _, err := NewVerifier(secret).Verify(raw); err == nil { + t.Error("none-signed token accepted") + } +} + +func TestVerifyRejectsNonUUIDSubject(t *testing.T) { + raw := sign(t, jwt.SigningMethodHS256, []byte(secret), "not-a-uuid", "user", time.Now().Add(time.Hour)) + if _, err := NewVerifier(secret).Verify(raw); err == nil { + t.Error("non-uuid subject accepted") + } +} diff --git a/coordinator/internal/transport/http/middleware.go b/coordinator/internal/transport/http/middleware.go index 2808313..bfa0dea 100644 --- a/coordinator/internal/transport/http/middleware.go +++ b/coordinator/internal/transport/http/middleware.go @@ -9,6 +9,9 @@ import ( "net/http" "strings" "time" + + "github.com/emil28092005/SciMesh/coordinator/internal/authctx" + tokenpkg "github.com/emil28092005/SciMesh/coordinator/internal/token" ) type ctxKey string @@ -41,25 +44,45 @@ func newRequestID() string { // withAuth enforces the shared bearer token every worker presents. // An empty token disables the check (local development only). -func withAuth(token string) func(http.Handler) http.Handler { +// withAuth authenticates a request one of two ways. Workers (and legacy +// submitters) present the shared service token. When user-JWT auth is enabled +// (verifier != nil), a submitter may instead present a userservice JWT; on +// success the requester is stamped into the context so the job use cases can +// record owner_id and enforce ownership. An empty token with no verifier +// disables auth entirely (dev only). +func withAuth(token string, verifier *tokenpkg.Verifier) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if token == "" { + if token == "" && verifier == nil { next.ServeHTTP(w, r) return } presented := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") - // Constant-time compare: a byte-by-byte early exit would let an - // attacker recover the token by timing responses. - if subtle.ConstantTimeCompare([]byte(presented), []byte(token)) != 1 { - w.Header().Set("WWW-Authenticate", "Bearer") - writeJSON(w, http.StatusUnauthorized, errorResponse{ - Error: "unauthorized", - RequestID: requestIDFrom(r.Context()), - }) + + // Shared service token: constant-time compare so a byte-by-byte + // early exit cannot leak the token through response timing. + if token != "" && subtle.ConstantTimeCompare([]byte(presented), []byte(token)) == 1 { + next.ServeHTTP(w, r) return } - next.ServeHTTP(w, r) + + // Otherwise try a user JWT, if that path is configured. + if verifier != nil && presented != "" { + if claims, err := verifier.Verify(presented); err == nil { + ctx := authctx.With(r.Context(), authctx.Requester{ + UserID: claims.UserID, + Role: claims.Role, + }) + next.ServeHTTP(w, r.WithContext(ctx)) + return + } + } + + w.Header().Set("WWW-Authenticate", "Bearer") + writeJSON(w, http.StatusUnauthorized, errorResponse{ + Error: "unauthorized", + RequestID: requestIDFrom(r.Context()), + }) }) } } diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 9eb56e2..b4e5a64 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -9,6 +9,7 @@ import ( "net/http" "time" + tokenpkg "github.com/emil28092005/SciMesh/coordinator/internal/token" "github.com/emil28092005/SciMesh/coordinator/internal/usecase" ) @@ -40,19 +41,23 @@ type Server struct { requestTimeout time.Duration heartbeatInterval time.Duration maxUploadBytes int64 + // verifier validates userservice JWTs. nil disables user-JWT auth, leaving + // only the shared service token — the pre-userservice behaviour. + verifier *tokenpkg.Verifier // ready probes downstream dependencies (the database) for /health. Kept as // a func so the transport layer never imports pgx. ready func(context.Context) error } func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval time.Duration, - maxUploadBytes int64, ready func(context.Context) error) *Server { + maxUploadBytes int64, jwtSecret string, ready func(context.Context) error) *Server { return &Server{ uc: uc, log: log, requestTimeout: requestTimeout, heartbeatInterval: heartbeatInterval, maxUploadBytes: maxUploadBytes, + verifier: tokenpkg.NewVerifier(jwtSecret), ready: ready, } } @@ -99,7 +104,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler { mux.Handle("/", chain(protected, withRequestID, // outermost: every response gets an ID, withAccessLog(s.log), // including the 401s below - withAuth(token), + withAuth(token, s.verifier), )) return mux } diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index e43086e..e9463b3 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -68,7 +68,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur if err != nil { t.Fatalf("register test worker: %v", err) } - srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, ready) + srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, "", ready) ts := httptest.NewServer(srv.Handler(token, configuredUIToken)) t.Cleanup(ts.Close) return &env{ts: ts, blobs: blobs, workerID: worker.ID.String()} diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go index eda4b49..2404cfe 100644 --- a/coordinator/internal/usecase/job.go +++ b/coordinator/internal/usecase/job.go @@ -53,6 +53,7 @@ func (uc *CreateJob) Execute(ctx context.Context, in CreateJobInput) (*domain.Jo if err != nil { return nil, err } + job.OwnerID = ownerFromContext(ctx) err = uc.tx.WithinTx(ctx, func(ctx context.Context) error { if err := uc.jobs.Insert(ctx, job); err != nil { @@ -97,6 +98,9 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error if err != nil { return err } + if err := authorizeJobAccess(ctx, job); err != nil { + return err + } if job.Status == domain.JobCancelled { return nil } @@ -132,6 +136,9 @@ func (uc *GetJobStatus) Execute(ctx context.Context, jobID uuid.UUID) (domain.Jo if err != nil { return domain.JobProgress{}, err } + if err := authorizeJobAccess(ctx, job); err != nil { + return domain.JobProgress{}, err + } counts, err := uc.tasks.CountByStatus(ctx, jobID) if err != nil { return domain.JobProgress{}, err diff --git a/coordinator/internal/usecase/ownership.go b/coordinator/internal/usecase/ownership.go new file mode 100644 index 0000000..c2d0105 --- /dev/null +++ b/coordinator/internal/usecase/ownership.go @@ -0,0 +1,39 @@ +package usecase + +import ( + "context" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/authctx" + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +// ownerFromContext returns the authenticated user id to stamp on a new job, or +// nil when the request was not authenticated as a user — worker or legacy +// traffic, or user-JWT auth disabled. A nil owner is stored as NULL. +func ownerFromContext(ctx context.Context) *uuid.UUID { + if r, ok := authctx.From(ctx); ok { + id := r.UserID + return &id + } + return nil +} + +// authorizeJobAccess enforces that a non-admin user may only act on their own +// job. It returns ErrJobNotFound — not a 403 — on a mismatch, so the response +// never reveals that another user's job exists. +// +// Requests with no authenticated user (worker/legacy traffic, or JWT auth +// disabled) are not restricted here: the shared service token already gated +// them, and worker endpoints legitimately operate across all jobs. +func authorizeJobAccess(ctx context.Context, job *domain.Job) error { + r, ok := authctx.From(ctx) + if !ok || r.IsAdmin() { + return nil + } + if job.OwnerID == nil || *job.OwnerID != r.UserID { + return domain.ErrJobNotFound + } + return nil +} diff --git a/coordinator/internal/usecase/ownership_test.go b/coordinator/internal/usecase/ownership_test.go new file mode 100644 index 0000000..fd4297b --- /dev/null +++ b/coordinator/internal/usecase/ownership_test.go @@ -0,0 +1,66 @@ +package usecase + +import ( + "context" + "errors" + "testing" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/coordinator/internal/authctx" + "github.com/emil28092005/SciMesh/coordinator/internal/domain" +) + +func TestOwnerFromContext(t *testing.T) { + if ownerFromContext(context.Background()) != nil { + t.Error("no requester must yield a nil owner") + } + id := uuid.New() + ctx := authctx.With(context.Background(), authctx.Requester{UserID: id, Role: "user"}) + got := ownerFromContext(ctx) + if got == nil || *got != id { + t.Errorf("owner = %v, want %v", got, id) + } +} + +func TestAuthorizeJobAccess(t *testing.T) { + owner := uuid.New() + other := uuid.New() + job := &domain.Job{ID: uuid.New(), OwnerID: &owner} + + ctxOf := func(id uuid.UUID, role string) context.Context { + return authctx.With(context.Background(), authctx.Requester{UserID: id, Role: role}) + } + + cases := []struct { + name string + ctx context.Context + wantErr bool + }{ + {"no requester (worker/legacy) allowed", context.Background(), false}, + {"owner allowed", ctxOf(owner, "user"), false}, + {"admin allowed", ctxOf(other, "admin"), false}, + {"non-owner denied", ctxOf(other, "user"), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := authorizeJobAccess(tc.ctx, job) + if tc.wantErr { + if !errors.Is(err, domain.ErrJobNotFound) { + t.Errorf("got %v, want ErrJobNotFound", err) + } + } else if err != nil { + t.Errorf("unexpected error: %v", err) + } + }) + } +} + +func TestAuthorizeJobAccessNilOwner(t *testing.T) { + // A legacy job with no owner must not be readable by an arbitrary user. + job := &domain.Job{ID: uuid.New(), OwnerID: nil} + ctx := authctx.With(context.Background(), authctx.Requester{UserID: uuid.New(), Role: "user"}) + if err := authorizeJobAccess(ctx, job); !errors.Is(err, domain.ErrJobNotFound) { + t.Errorf("got %v, want ErrJobNotFound", err) + } +} diff --git a/coordinator/internal/usecase/reduce.go b/coordinator/internal/usecase/reduce.go index e73f575..c1faf71 100644 --- a/coordinator/internal/usecase/reduce.go +++ b/coordinator/internal/usecase/reduce.go @@ -126,6 +126,9 @@ func (uc *GetJobResult) Execute(ctx context.Context, jobID uuid.UUID) (*domain.A if err != nil { return nil, nil, err } + if err := authorizeJobAccess(ctx, job); err != nil { + return nil, nil, err + } if job.Status != domain.JobCompleted || job.ResultArtifactID == nil { return nil, nil, domain.ErrArtifactNotFound } diff --git a/coordinator/internal/usecase/upload.go b/coordinator/internal/usecase/upload.go index ad394db..4bd5caf 100644 --- a/coordinator/internal/usecase/upload.go +++ b/coordinator/internal/usecase/upload.go @@ -43,6 +43,7 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su if err != nil { return SubmitDatasetResult{}, err } + job.OwnerID = ownerFromContext(ctx) // Everything written to blob storage, so a failed transaction can undo it. var putKeys []string diff --git a/coordinator/migrations/0011_job_owner.down.sql b/coordinator/migrations/0011_job_owner.down.sql new file mode 100644 index 0000000..32ffe4f --- /dev/null +++ b/coordinator/migrations/0011_job_owner.down.sql @@ -0,0 +1,6 @@ +BEGIN; + +DROP INDEX IF EXISTS ix_jobs_owner; +ALTER TABLE jobs DROP COLUMN IF EXISTS owner_id; + +COMMIT; diff --git a/coordinator/migrations/0011_job_owner.up.sql b/coordinator/migrations/0011_job_owner.up.sql new file mode 100644 index 0000000..64c9785 --- /dev/null +++ b/coordinator/migrations/0011_job_owner.up.sql @@ -0,0 +1,14 @@ +BEGIN; + +-- Who submitted this job. Equals users.id from the userservice, taken from the +-- JWT `sub` claim. NOT a foreign key: users live in a separate service/database, +-- so integrity is guaranteed by the signed token, not by the DB. +-- +-- Nullable because rows created before auth existed have no owner; new inserts +-- must supply it (enforced in the app, not the schema, during the MVP). +ALTER TABLE jobs ADD COLUMN owner_id uuid; + +-- "List my jobs" / "admin filters by owner" scans by owner. +CREATE INDEX ix_jobs_owner ON jobs (owner_id); + +COMMIT; diff --git a/users/.dockerignore b/users/.dockerignore new file mode 100644 index 0000000..a4d0d5c --- /dev/null +++ b/users/.dockerignore @@ -0,0 +1,14 @@ +# Keep the build context small and never bake secrets or local state into an image. +.env +.git +.gitignore +*.md +Makefile +docker-compose.yml +Dockerfile +.dockerignore + +# Local build artifacts +/coordinator +/bin/ +*.out diff --git a/users/.env.example b/users/.env.example new file mode 100644 index 0000000..4c1e596 --- /dev/null +++ b/users/.env.example @@ -0,0 +1,32 @@ +# Copy to .env and adjust. All settings are read from the environment. + +COORDINATOR_ADDR=:8080 +DATABASE_URL=postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable + +# Shared bearer token every worker must present. Leave empty to disable auth (dev only). +WORKER_AUTH_TOKEN=change-me + +# Optional local operator UI. Use a separate value; never reuse the worker token. +# When empty, /ui is disabled. +UI_AUTH_TOKEN= + +# Logging. LOG_LEVEL: debug|info|warn|error. LOG_FILE empty = stdout only; +# set a path to also write a size-rotated file (kept across restarts). +LOG_LEVEL=info +# LOG_FILE=./logs/coordinator.log + +# Directory where artifact bytes are stored. +COORDINATOR_STORAGE_DIR=./data +# Upper bound on an uploaded dataset or artifact body (bytes). Default 1 GiB. +MAX_UPLOAD_BYTES=1073741824 + +# Optional tuning (defaults shown). +DB_MAX_CONNS=10 +# How long to keep retrying the initial DB connection while Postgres boots. +DB_CONNECT_TIMEOUT=30s +REQUEST_TIMEOUT=15s +LEASE_DURATION=2m +DEFAULT_MAX_ATTEMPTS=3 +REAPER_INTERVAL=30s +# A worker silent longer than this is marked offline by the reaper. +WORKER_OFFLINE_AFTER=1m diff --git a/users/.gitignore b/users/.gitignore new file mode 100644 index 0000000..c8c4f7b --- /dev/null +++ b/users/.gitignore @@ -0,0 +1,6 @@ +/coordinator +/bin/ +.env +*.out +/logs/ +/data/ diff --git a/users/.golangci.yml b/users/.golangci.yml new file mode 100644 index 0000000..428bfed --- /dev/null +++ b/users/.golangci.yml @@ -0,0 +1,54 @@ +version: "2" + +run: + timeout: 3m + +linters: + # "standard" = errcheck, govet, ineffassign, staticcheck, unused. + default: standard + enable: + # Catches `err == ErrFoo` where errors.Is is required. Directly relevant + # here: domain exposes sentinel errors that use cases may wrap with %w. + - errorlint + # Returning nil after checking a non-nil error — a silent bug factory. + - nilerr + # http.Get/Do without a context: every outbound call must be cancellable. + - noctx + # Unclosed response bodies leak connections. + - bodyclose + # Common security mistakes (weak crypto, unhandled file perms). + - gosec + # Style and naming consistency. + - revive + - misspell + - unconvert + + settings: + errcheck: + # Deferred Close/Rollback are intentionally ignored in a few places + # (rollback after commit is a documented no-op). + check-type-assertions: true + revive: + rules: + - name: exported + disabled: true # internal packages need no exported-symbol comments + gosec: + excludes: + - G404 # math/rand is fine for jitter; nothing here is security-sensitive + + exclusions: + rules: + # Tests may skip error checks and use long literals freely. + - path: _test\.go + linters: + - errcheck + - gosec + +formatters: + enable: + - gofmt + - goimports + settings: + goimports: + local-prefixes: + - github.com/emil28092005/SciMesh/coordinator diff --git a/users/ARCHITECTURE.md b/users/ARCHITECTURE.md new file mode 100644 index 0000000..77b6976 --- /dev/null +++ b/users/ARCHITECTURE.md @@ -0,0 +1,144 @@ +# Архитектура координатора + +Карта кода. Читать сверху вниз: сначала «где что лежит», потом «как проходит +запрос», в конце — «куда добавлять новое». + +--- + +## 1. Четыре слоя + +``` + infra конфиг, пул БД, часы, HTTP-сервер, reaper ← драйверы + transport HTTP-хендлеры ← входящее: кто зовёт нас + storage репозитории на SQL ← исходящее: кого зовём мы + usecase операции + ПОРТЫ (интерфейсы) ← прикладные правила + domain Task, Job и их инварианты ← бизнес-правила + + ┌── transport ──┐ + domain ◄── usecase ◄┤ ├◄── infra + └── storage ────┘ +``` + +`transport` и `storage` — один и тот же слой (в книгах он зовётся «адаптеры»), +просто разделённый по направлению: транспорт принимает запросы снаружи, storage +обращается наружу сам. Так путь к файлу говорит о его роли, а не о категории. + +**Единственное правило:** зависимости идут только внутрь. `domain` не импортирует +ничего из проекта. `usecase` видит только `domain`. `transport` и `storage` не +знают друг о друге. + +Проверить в любой момент: + +```sh +go list -f '{{range .Imports}}{{.}}{{"\n"}}{{end}}' ./internal/domain | grep internal +# пусто = правило соблюдено +``` + +--- + +## 2. Где что лежит + +| Файл | Что внутри | Строк | +| --- | --- | --- | +| `domain/task.go` | `Task` и **все** переходы состояний: аренда, завершение, провал, истечение | ~245 | +| `domain/job.go` | `Job`, разбиение на чанки, вывод статуса из счётчиков задач | ~107 | +| `domain/errors.go` | Нарушения бизнес-правил (`ErrLeaseConflict`, `ErrStaleAttempt`, …) | ~18 | +| `usecase/ports.go` | **Порты**: `TaskRepository`, `JobRepository`, `TxManager`, `Clock` | ~79 | +| `usecase/task.go` | Операции над задачей: claim, renew, complete, fail, expire | ~200 | +| `usecase/job.go` | Операции над job: create, status, results, stitch | ~180 | +| `usecase/dto.go` | Входные структуры юзкейсов | ~51 | +| `transport/http/server.go` | Роутер и сборка middleware | ~60 | +| `transport/http/handlers.go` | По хендлеру на эндпоинт | ~180 | +| `transport/http/dto.go` | JSON-форматы запросов и ответов | ~118 | +| `transport/http/middleware.go` | request-ID, access-лог, bearer-авторизация | ~103 | +| `transport/http/errors.go` | Маппинг доменных ошибок в HTTP-коды | ~55 | +| `storage/postgres/task_repo.go` | SQL по задачам, включая атомарный claim | ~109 | +| `storage/postgres/job_repo.go` | SQL по job'ам | ~39 | +| `storage/postgres/tx.go` | `TxManager`: транзакция через контекст | ~65 | +| `infra/*.go` | Конфиг, пул, часы, сервер, reaper | ~240 | +| `cmd/coordinator/main.go` | **Composition root** — единственное место со всеми конкретными типами | ~73 | + +--- + +## 3. Трасса запроса: `POST /tasks/claim` + +Как воркер получает задачу. Четыре остановки, по одной на слой: + +``` + ① transport/http/handlers.go → handleClaim + разбирает JSON, отдаёт usecase.ClaimTaskInput + │ + ▼ + ② usecase/task.go → ClaimTask.Execute + сначала подчищает протухшие аренды, потом просит одну задачу + через ПОРТ TaskRepository (реализацию не знает) + │ + ▼ + ③ usecase/ports.go → TaskRepository.ClaimNext + контракт: «атомарно выдай одну задачу» + │ + ▼ + ④ storage/postgres/task_repo.go → claimNextSQL + SELECT ... FOR UPDATE SKIP LOCKED + UPDATE одним запросом +``` + +Обратно поднимается `*domain.Task`, юзкейс сужает его до `domain.ClaimedTask` +(воркеру не отдаём `version`, `max_attempts` и чужие ошибки), хендлер +превращает в JSON. Пустая очередь — это `nil, nil` на шаге ② и `204` на ①. + +**Трасса `POST /tasks/{id}/result`** такая же, но с одним отличием: решение +принимает **сущность**, а не юзкейс. + +``` + handlers.go → CompleteTask.Execute → tx.WithinTx( + GetForUpdate → task.CompleteWith(...) ←── ЗДЕСЬ правила + │ (чужая аренда? устаревший + Update ←─────────────┘ attempt? повтор того же + syncJobStatus манифеста?) + ) +``` + +--- + +## 4. Куда добавлять новое + +| Хочу… | Правлю | +| --- | --- | +| новое бизнес-правило (когда задачу можно повторить) | `domain/task.go` + тест рядом | +| новую операцию (отменить job) | `usecase/job.go` + порт в `ports.go`, если нужен новый запрос к БД | +| новый HTTP-эндпоинт | `transport/http/handlers.go` + маршрут в `server.go` + DTO в `dto.go` | +| новый SQL-запрос | `storage/postgres/*_repo.go` | +| новую настройку | `infra/config.go` + `.env.example` | +| поменять код ответа на ошибку | `transport/http/errors.go` | + +**Правило при сомнении:** если код можно описать фразой «когда X, то Y» без +упоминания HTTP, SQL и конфигов — это `domain`. Если он оркеструет несколько +шагов и транзакцию — `usecase`. Если знает про JSON — `transport`, про SQL — `storage`. + +--- + +## 5. Три вещи, которые надо понять один раз + +**Порты объявляет потребитель.** `TaskRepository` описан в `usecase/ports.go`, а +реализован в `storage/postgres`. Поэтому `usecase` не импортирует `storage` — +стрелка зависимости смотрит внутрь, хотя вызов на рантайме идёт наружу. + +**Транзакция едет в контексте.** `TxManager.WithinTx` кладёт `pgx.Tx` в контекст +по неэкспортируемому ключу; репозитории достают её через `conn(ctx, pool)`. +Благодаря этому юзкейс говорит «сделай это атомарно», ни разу не упомянув pgx. + +**Атомарный claim нельзя разложить на шаги.** `ClaimNext` — один SQL-запрос, +потому что `SELECT` + отдельный `UPDATE` вернул бы гонку, при которой одну +задачу выдают двум воркерам. Поэтому `ClaimTask.Execute` выглядит тонким: там +нечего оркестровать, вся гарантия — внутри запроса. + +--- + +## 6. Что уже работает, а что заглушка + +Работает: слои и проводка, роутинг, авторизация, access-лог, маппинг ошибок, +транзакции, graceful shutdown, миграции, **весь domain с 12 юнит-тестами без БД**. + +Заглушки (`ErrNotImplemented` → HTTP 501): методы репозиториев. SQL для двух +главных операций уже написан в `task_repo.go` — `claimNextSQL` и +`expireLeasesSQL`, осталось их подключить. diff --git a/users/Dockerfile b/users/Dockerfile new file mode 100644 index 0000000..ffe514e --- /dev/null +++ b/users/Dockerfile @@ -0,0 +1,52 @@ +# syntax=docker/dockerfile:1 +# +# Requires BuildKit (the RUN --mount cache lines below). Docker 23+ enables it +# by default when the buildx plugin is present; install `docker-buildx` if a +# build fails with "the --mount option requires BuildKit". + +# --- build stage ---------------------------------------------------------- +FROM golang:1.24-alpine AS build + +WORKDIR /src + +# Copy manifests first: this layer stays cached until dependencies actually +# change, so editing Go sources does not re-download the module graph. +COPY go.mod go.sum ./ +RUN --mount=type=cache,target=/go/pkg/mod go mod download + +COPY . . + +# The cache mounts persist the module cache and the compiler's build cache +# *across* builds, so a rebuild after a code edit recompiles only what changed +# instead of the whole dependency tree. +# +# CGO_ENABLED=0 produces a fully static binary, so the runtime image needs no +# libc. -trimpath strips local paths; -s -w drop the symbol table and DWARF. +RUN --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + CGO_ENABLED=0 GOOS=linux go build \ + -trimpath -ldflags="-s -w" \ + -o /out/userservice ./cmd/userservice + +# --- runtime stage -------------------------------------------------------- +FROM alpine:3.20 + +# ca-certificates for outbound TLS; wget backs the container healthcheck. +RUN apk add --no-cache ca-certificates wget \ + && adduser -D -H -u 10001 userservice \ + # Pre-create the log dir owned by the non-root user. A named volume mounted + # here inherits this ownership from the image, so the process can write to it — + # a host bind mount, owned by root, could not. + && mkdir -p /var/log/scimesh \ + && chown -R userservice:userservice /var/log/scimesh + +COPY --from=build /out/userservice /usr/local/bin/userservice + +# Never run as root: a compromised process should not own the container. +USER userservice + +EXPOSE 8081 + +# Exec form, not shell: the binary becomes PID 1 and receives SIGTERM directly, +# which is what its graceful shutdown depends on. +ENTRYPOINT ["/usr/local/bin/userservice"] diff --git a/users/Makefile b/users/Makefile new file mode 100644 index 0000000..8f5c701 --- /dev/null +++ b/users/Makefile @@ -0,0 +1,96 @@ +.DEFAULT_GOAL := help + +.PHONY: help build run test test-integration vet lint tidy check migrate-up migrate-down up down down-clean logs ps psql smoke + +# `check` uses its own Compose project and host ports so it never touches a +# developer's local PostgreSQL or the normal `make up` stack. +CHECK_PROJECT ?= scimesh-users-check +CHECK_POSTGRES_PORT ?= 55433 +CHECK_USERSERVICE_PORT ?= 18081 +CHECK_HOST ?= http://localhost:$(CHECK_USERSERVICE_PORT) +CHECK_DATABASE_URL ?= postgres://scimesh:scimesh@localhost:$(CHECK_POSTGRES_PORT)/scimesh_users?sslmode=disable +CHECK_COMPOSE = POSTGRES_PORT=$(CHECK_POSTGRES_PORT) USERSERVICE_PORT=$(CHECK_USERSERVICE_PORT) docker compose -p $(CHECK_PROJECT) + +help: + @printf '%s\n' \ + 'SciMesh userservice commands:' \ + ' make up / make down Start or stop the userservice stack (Postgres + migrate + service).' \ + ' make check One command: vet, lint, race tests, integration, smoke (needs Docker).' \ + ' make test / make vet Run Go verification.' \ + ' make smoke Exercise the live API against a running service.' + +# --- build / run --------------------------------------------------------- +build: + go build ./... + +run: + go run ./cmd/userservice + +test: + go test ./... + +# Needs a running PostgreSQL with the migrations applied: +# make test-integration TEST_DATABASE_URL='postgres://...' +test-integration: + TEST_DATABASE_URL="$(TEST_DATABASE_URL)" go test -tags=integration ./... -v + +vet: + go vet ./... + +# Runs golangci-lint without installing it system-wide. +LINT_VERSION := v2.12.2 +lint: + @command -v golangci-lint >/dev/null 2>&1 \ + && golangci-lint run --build-tags=integration ./... \ + || go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(LINT_VERSION) run --build-tags=integration ./... + +tidy: + go mod tidy + +# One command for a reviewer: unit tests + vet + lint, then the stack up and the +# integration suite and the end-to-end smoke test. Needs Docker. +check: vet lint + go test -race ./... + $(CHECK_COMPOSE) up -d --build + @echo "waiting for the userservice to be ready..." + @attempt=0; until curl -fsS "$(CHECK_HOST)/health" >/dev/null; do \ + attempt=$$((attempt + 1)); \ + if [ $$attempt -ge 30 ]; then $(CHECK_COMPOSE) logs userservice; exit 1; fi; \ + sleep 1; \ + done + TEST_DATABASE_URL="$(CHECK_DATABASE_URL)" \ + go test -tags=integration ./internal/storage/postgres/ -v + HOST="$(CHECK_HOST)" ./scripts/smoke.sh + @echo "\nall checks passed ✓" + +# --- migrations ---------------------------------------------------------- +# Requires the golang-migrate CLI and DATABASE_URL, e.g.: +# export DATABASE_URL='postgres://scimesh:scimesh@localhost:5433/scimesh_users?sslmode=disable' +migrate-up: + migrate -path migrations -database "$(DATABASE_URL)" up + +migrate-down: + migrate -path migrations -database "$(DATABASE_URL)" down 1 + +# --- docker -------------------------------------------------------------- +up: + docker compose up -d --build + +down: + docker compose down + +down-clean: + docker compose down -v + +logs: + docker compose logs -f userservice + +ps: + docker compose ps + +psql: + docker compose exec postgres psql -U scimesh -d scimesh_users + +# --- api ------------------------------------------------------------------ +smoke: + ./scripts/smoke.sh diff --git a/users/README.md b/users/README.md new file mode 100644 index 0000000..7240b99 --- /dev/null +++ b/users/README.md @@ -0,0 +1,230 @@ +# SciMesh Coordinator + +Durable task-queue server for SciMesh, in Go on PostgreSQL. It owns all database +access; workers talk to it only over HTTP and never receive DB credentials. + +Built as a **modular monolith following Clean Architecture** — one binary, four +layers, dependencies pointing strictly inward. See +`docs/database-integration-task.md` and `docs/worker-daemon-task.md` in the repo +root for the full contract. + +## Layers + +``` + infra config, pgxpool, http.Server, clock ← frameworks & drivers + transport http handlers ← inbound: who calls us + storage sql repositories ← outbound: who we call + usecase business operations + PORTS ← application rules + domain Task, Job + their invariants ← enterprise rules + + ┌── transport ──┐ + domain ◄── usecase ◄┤ ├◄── infra + └── storage ────┘ +``` + +`transport` and `storage` are one layer — the "interface adapters" ring — split +by direction rather than by category, so a file's path tells you its role. + +The rule that matters: **source dependencies point only inward**. `domain` +imports nothing from this module; `usecase` sees only `domain`; `transport` and +`storage` know nothing of each other. Verify it at any time with: + +```sh +go list -f '{{range .Imports}}{{.}}{{"\n"}}{{end}}' ./internal/domain | grep internal # must be empty +``` + +## Layout + +``` +coordinator/ + cmd/coordinator/main.go # composition root: the only place with concrete types + internal/ + domain/ # entities + rules, no I/O + task.go Task, lease/complete/fail/expire transitions + job.go Job, chunk fan-out, status derivation + errors.go business-rule violations + usecase/ # one type per operation, dependencies injected + ports.go TaskRepository, JobRepository, TxManager, Clock + dto.go use-case boundary inputs + task.go claim, renew, complete, fail, expire + job.go create, status, results, stitch + transport/http/ # routing, DTOs, middleware, error mapping + storage/postgres/ # SQL behind the ports; TxManager via context + infra/ # config.go db.go clock.go server.go + migrations/ # golang-migrate SQL, run as an explicit command +``` + +A full map — file-by-file table, a request traced through every layer, and a +"where do I add X" guide — lives in [ARCHITECTURE.md](ARCHITECTURE.md). + +## Quickstart + +### With Docker (nothing to install but Docker) + +```sh +make up # Postgres → migrations → coordinator +curl localhost:8080/health +make logs # follow the coordinator +make down # stop (add down-clean to drop the DB volume) +``` + +To enable the local operator UI, set a separate credential before starting: + +```sh +UI_AUTH_TOKEN='local-ui-secret' make up +# Open http://localhost:8080/ui and use any username with this value as password. +``` + +The UI is disabled by default and never accepts the worker bearer token. +The **control room** shows live workers, recent runs, shard state/attempts, +safe failures, coordinator artifacts, and the final CSV for completed +similarity-search jobs. The job page follows the real stages: TSV accepted → +shards execute → workers return CSVs → `reducing` → final deterministic global +top-k result. It polls only its own coordinator read-model and never controls +or exposes worker processes. + +For a hands-on run, open `/ui`, choose **New similarity search**, select a +small ChEMBL-style TSV, then leave one or more `scimesh-worker` processes +running in separate terminals. The detail page updates every two seconds and +stops polling after a completed, failed, or cancelled job. Use **Preview CSV** +to inspect a bounded first page of a partial or completed final result before +downloading it. The UI never exposes source datasets or shard inputs; partial +CSVs remain available only as diagnostics. + +### One-command manual demo + +From the repository root, create the Python environment once, then start a +self-contained UI demo with two local reference workers: + +```sh +python3 -m venv .venv +.venv/bin/pip install -e '.[dev]' +make demo-ui +``` + +This uses a separate Docker project and ports `18080` (coordinator) and +`55432` (PostgreSQL), so it does not conflict with the normal stack. Open +`http://localhost:18080/ui`, use username `operator` and password +`demo-ui-secret`, upload a small ChEMBL TSV, and observe the workers process +it. Change the worker count with `make demo-ui WORKERS=3`; stop all demo +services and workers with `make demo-down`. + +The job page shows a live **Processing speed** graph in completed shards per +minute. It uses the coordinator snapshots observed by the open browser tab, so +it is a transparent local-session measurement rather than a persisted metric. +Use **Preview CSV** before downloading a partial diagnostic or completed final +result. Run `make help` from either the repository root or this directory for +the full list of demo commands. + +`up` starts three services in order: Postgres waits until `pg_isready` passes, a +one-shot `migrate` container applies the schema and exits, and only then does the +coordinator start — so it never queries a database that has no tables. + +> **Needs BuildKit.** The Dockerfile uses `RUN --mount=type=cache` to reuse the +> Go module and compiler caches between builds. If the build fails with +> *"the --mount option requires BuildKit"*, install the buildx plugin — +> `pacman -S docker-buildx` on Arch, `apt install docker-buildx-plugin` on Debian. + +### Locally, against your own Postgres + +```sh +cp .env.example .env # then edit DATABASE_URL / WORKER_AUTH_TOKEN + # it is loaded automatically — no export needed + +make tidy # fetch deps (needs network once) +make migrate-up # apply schema (needs the migrate CLI) +make run # start the server +``` + +## Configuration + +Settings come from the environment. A `.env` file is loaded at startup via +`godotenv` as a local-dev convenience (override its path with `ENV_FILE`): + +- a missing `.env` is not an error — production injects real env vars; +- **real environment variables always win** over the file, so an orchestrator's + values are never shadowed by a stale `.env` baked into an image. + +See `.env.example`; only `DATABASE_URL` is required. + +## Endpoints + +| Method | Path | Purpose | +| ------ | ---------------------------------- | --------------------------------------------- | +| POST | `/workers/register` | Register a worker, get its id | +| POST | `/jobs` | Create job + tasks from chunk URIs | +| POST | `/jobs/upload` | Upload a dataset; coordinator chunks it | +| GET | `/jobs/{job_id}` | Aggregate job progress | +| POST | `/tasks/claim` | Atomically lease one task (`204` if none) | +| GET | `/tasks/{task_id}/input` | Download the task's input shard | +| POST | `/tasks/{task_id}/heartbeat` | Renew the caller's lease (→ `running`) | +| PUT | `/tasks/{task_id}/artifacts/{name}`| Upload a partial-result artifact | +| POST | `/tasks/{task_id}/result` | Complete with an artifact id (idempotent) | +| POST | `/tasks/{task_id}/failure` | Record failure / retryable state | +| GET | `/artifacts/{artifact_id}/download`| Download an artifact by id | +| GET | `/health` | Readiness incl. database (unauthenticated) | + +The full contract is in [`docs/api-contract.md`](../docs/api-contract.md) and +[`docs/openapi.yaml`](../docs/openapi.yaml); a worker-author guide is in +[`docs/building-workers.md`](../docs/building-workers.md). + +## Poking the API + +Two ways, both checked in: + +```sh +make smoke # every endpoint, asserted; non-zero exit on failure +``` + +`api/requests.http` runs the same calls one at a time from an editor with a REST +client (VSCodium/VS Code "REST Client", JetBrains HTTP Client). Later requests +reuse ids captured from earlier responses, so it doubles as API documentation. + +## Status + +Works end to end: a worker registers, a dataset is uploaded and chunked into +shard tasks (or a job is created from chunk URIs), tasks are leased one at a +time, downloaded, heartbeated (`leased → running`), completed via uploaded +result artifacts, and reflected in job progress. A reaper reclaims expired +leases and marks silent workers offline. + +Done: schema + migrations, atomic claim (`FOR UPDATE SKIP LOCKED`), optimistic +concurrency, result/failure paths, lease expiry, worker registry + liveness, +artifact storage, dataset upload + chunking, request-size limits. + +Still stubbed: `StitchJob.Execute` — merging per-chunk top-k into the final CSV +is workload semantics that belongs to the Python side (reducer). + +## Tests + +Unit tests need **no database** — domain rules, use-case orchestration (over +in-memory `internal/memstore`), and HTTP handlers (via `httptest`): + +```sh +make test # go test ./... +make vet +make lint +go test -race ./... +``` + +Integration tests run against a **real PostgreSQL** (the spec forbids mocks +here — they verify `FOR UPDATE SKIP LOCKED`, optimistic concurrency, rollback): + +```sh +docker compose up -d +make test-integration TEST_DATABASE_URL='postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable' +``` + +CI (`.github/workflows/coordinator.yml`) runs vet, gofmt, race tests, lint, and +the integration suite against a Postgres service on every push and PR. + +For the complete local verification, including an isolated Docker PostgreSQL +and the HTTP smoke flow, run: + +```sh +make check +``` + +It uses Compose project `scimesh-check` and ports `55432`/`18080` by default, +so it does not connect to a PostgreSQL already running on `5432`. Override +`CHECK_POSTGRES_PORT`, `CHECK_COORDINATOR_PORT`, or `CHECK_PROJECT` if needed. diff --git a/users/api/requests.http b/users/api/requests.http new file mode 100644 index 0000000..6c50a59 --- /dev/null +++ b/users/api/requests.http @@ -0,0 +1,234 @@ +# SciMesh Coordinator — API requests +# +# Runnable from any editor with a REST client (VSCodium/VS Code "REST Client", +# JetBrains HTTP Client). Click "Send Request" above each block, top to bottom: +# later requests reuse ids captured from earlier responses. +# +# Start the stack first: docker compose up -d + +@host = http://localhost:8080 +@token = change-me +@worker = worker-1 + +### Readiness — the only unauthenticated endpoint (probes the database) +GET {{host}}/health + +### Auth check — no token must be rejected with 401 +POST {{host}}/tasks/claim +Content-Type: application/json + +{ "worker_id": "{{worker}}" } + +### 0. Register a worker (201) +# @name register +POST {{host}}/workers/register +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "name": "lab-worker-01", + "capabilities": ["similarity_search"], + "cpu_count": 8, + "memory_mb": 16384 +} + +@workerId = {{register.response.body.worker_id}} + +### 0b. Upload a dataset — the coordinator splits it into shard tasks (201) +# Text fields first, the file part last (it is streamed, not buffered). +# @name uploadJob +POST {{host}}/jobs/upload +Authorization: Bearer {{token}} +Content-Type: multipart/form-data; boundary=----scimesh + +------scimesh +Content-Disposition: form-data; name="workload" + +similarity_search +------scimesh +Content-Disposition: form-data; name="parameters" + +{"top_k":10} +------scimesh +Content-Disposition: form-data; name="chunk_rows" + +2 +------scimesh +Content-Disposition: form-data; name="file"; filename="chembl.tsv" +Content-Type: text/tab-separated-values + +id smiles +A CC +B CCC +C CCCC +D CCCCC +------scimesh-- + +### Download a task's input shard (200) — taskId must be a shard task from an +### uploaded job (claim one first; its input.uri is /tasks/{id}/input). +GET {{host}}/tasks/{{taskId}}/input +Authorization: Bearer {{token}} + +### 1. Create a job and its chunks (201) +# The coordinator splits the submission into one task per chunk, transactionally. +# @name createJob +POST {{host}}/jobs +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "workload": "similarity_search", + "input_uri": "s3://chembl/full.sdf", + "parameters": { "top_k": 10 }, + "chunks": [ + { "chunk_index": 0, "input_uri": "s3://chembl/shard-0.sdf", "input_sha256": "aaa", "max_attempts": 3 }, + { "chunk_index": 1, "input_uri": "s3://chembl/shard-1.sdf", "input_sha256": "bbb", "max_attempts": 3 }, + { "chunk_index": 2, "input_uri": "s3://chembl/shard-2.sdf", "input_sha256": "ccc", "max_attempts": 3 } + ] +} + +@jobId = {{createJob.response.body.id}} + +### 2. Claim a task (200, or 204 when the queue is empty) +# Each call leases a different task; run it repeatedly to see chunk_index advance. +# @name claim +POST {{host}}/tasks/claim +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "capabilities": ["similarity_search"], + "max_concurrency": 1 +} + +@taskId = {{claim.response.body.task_id}} +@attempt = {{claim.response.body.attempt}} + +### 3. Heartbeat — renew the lease while the task is still running (200) +POST {{host}}/tasks/{{taskId}}/heartbeat +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}} +} + +### 3a. Upload a partial-result artifact (200) — while the task is leased +# Identity travels in headers per the contract; the body is streamed as-is. +# @name uploadArtifact +PUT {{host}}/tasks/{{taskId}}/artifacts/result.csv +Authorization: Bearer {{token}} +Content-Type: text/csv +X-Worker-ID: {{worker}} +X-Task-Attempt: {{attempt}} + +query,match,score +CHEMBL25,CHEMBL139,0.87 + +@artifactId = {{uploadArtifact.response.body.artifact_id}} + +### 3b. Download the artifact by id (200) +GET {{host}}/artifacts/{{artifactId}}/download +Authorization: Bearer {{token}} + +### 3c. Upload a second artifact — used by the conflict check below (200) +# @name uploadArtifact2 +PUT {{host}}/tasks/{{taskId}}/artifacts/secondary.csv +Authorization: Bearer {{token}} +Content-Type: text/csv +X-Worker-ID: {{worker}} +X-Task-Attempt: {{attempt}} + +query,match,score +CHEMBL25,CHEMBL521,0.42 + +@artifactId2 = {{uploadArtifact2.response.body.artifact_id}} + +### 4. Submit the result, referencing the uploaded artifact (200) +POST {{host}}/tasks/{{taskId}}/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}}, + "result": { "artifact_id": "{{artifactId}}", "content_type": "text/csv" }, + "metrics": { "elapsed_ms": 1234, "candidates": 50000 } +} + +### 4a. Replay the same result — must be idempotent (200, not 409) +POST {{host}}/tasks/{{taskId}}/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}}, + "result": { "artifact_id": "{{artifactId}}" } +} + +### 4b. A different artifact for the same task — conflict (409) +POST {{host}}/tasks/{{taskId}}/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}}, + "result": { "artifact_id": "{{artifactId2}}" } +} + +### 4c. Another worker submitting for this task — conflict (409) +POST {{host}}/tasks/{{taskId}}/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "impostor", + "attempt": {{attempt}}, + "result": { "artifact_id": "{{artifactId}}" } +} + +### 5. Report a failure instead (200) +# retryable=true returns the task to the queue while attempts remain; +# retryable=false fails it terminally. +POST {{host}}/tasks/{{taskId}}/failure +Authorization: Bearer {{token}} +Content-Type: application/json + +{ + "worker_id": "{{worker}}", + "attempt": {{attempt}}, + "error_code": "download_failed", + "error_message": "checksum mismatch on shard", + "retryable": true +} + +### 6. Job progress (200) +GET {{host}}/jobs/{{jobId}} +Authorization: Bearer {{token}} + +### --- error cases ------------------------------------------------------- + +### Malformed UUID in the path (400) +POST {{host}}/tasks/not-a-uuid/result +Authorization: Bearer {{token}} +Content-Type: application/json + +{ "worker_id": "{{worker}}", "attempt": 1, "result_uri": "s3://x", "result_sha256": "x" } + +### Unknown field in the body (400) — a misspelled key must not pass silently +POST {{host}}/tasks/claim +Authorization: Bearer {{token}} +Content-Type: application/json + +{ "worker_ID": "{{worker}}" } + +### Unknown job (404) +GET {{host}}/jobs/00000000-0000-0000-0000-000000000000 +Authorization: Bearer {{token}} + +### Stitching is not implemented yet (501) +# Any endpoint whose use case is still a stub answers 501. diff --git a/users/cmd/userservice/main.go b/users/cmd/userservice/main.go new file mode 100644 index 0000000..fbb0095 --- /dev/null +++ b/users/cmd/userservice/main.go @@ -0,0 +1,68 @@ +// Command userservice runs the SciMesh authentication service: it registers +// users, verifies logins, and issues the HS256 JWTs the coordinator trusts. +package main + +import ( + "context" + "fmt" + nethttp "net/http" + "os" + "os/signal" + "syscall" + + "github.com/emil28092005/SciMesh/users/internal/auth" + "github.com/emil28092005/SciMesh/users/internal/infra" + "github.com/emil28092005/SciMesh/users/internal/storage/postgres" + apihttp "github.com/emil28092005/SciMesh/users/internal/transport/http" + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "fatal:", err) + os.Exit(1) + } +} + +func run() error { + // Cancelled on SIGINT/SIGTERM so the HTTP server drains in-flight requests + // instead of dropping them. + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + cfg, err := infra.LoadConfig() + if err != nil { + return err + } + + log, closer, err := infra.NewLogger(cfg) + if err != nil { + return err + } + defer func() { _ = closer.Close() }() + + pool, err := infra.NewPool(ctx, cfg, log) + if err != nil { + return err + } + defer pool.Close() + + // Adapters implementing the usecase ports. + users := postgres.NewUserRepo(pool) + hasher := auth.NewHasher(cfg.BcryptCost) + clock := infra.NewClock() + issuer := auth.NewIssuer(cfg.JWTSecret, cfg.TokenTTL, clock.Now) + + uc := apihttp.UseCases{ + Register: usecase.NewRegister(users, hasher, clock), + Login: usecase.NewLogin(users, hasher, issuer), + Users: users, + } + + handler := apihttp.NewServer(log, uc, issuer) + // A blanket per-request deadline: bcrypt is bounded, so anything slower is a + // stuck handler we want to shed rather than hold a connection open. + handler = nethttp.TimeoutHandler(handler, cfg.RequestTimeout, `{"error":"request timeout"}`) + + return infra.RunServer(ctx, log, cfg.Addr, handler) +} diff --git a/users/docker-compose.yml b/users/docker-compose.yml new file mode 100644 index 0000000..f7e155b --- /dev/null +++ b/users/docker-compose.yml @@ -0,0 +1,81 @@ +# A self-contained stack for the userservice: its own PostgreSQL (a separate +# database from the coordinator's — different bounded context), a one-shot +# migration step, and the service. The project name and host ports differ from +# the coordinator's so both stacks can run side by side on one machine. +name: scimesh-users + +services: + postgres: + image: postgres:16-alpine + environment: + POSTGRES_USER: ${POSTGRES_USER:-scimesh} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-scimesh} + POSTGRES_DB: ${POSTGRES_DB:-scimesh_users} + ports: + # 5433 on the host, so it never clashes with the coordinator's 5432. + - "${POSTGRES_PORT:-5433}:5432" + volumes: + - pgdata:/var/lib/postgresql/data + healthcheck: + # Everything else waits on this, so the check must prove the server + # accepts queries — not merely that the port is open. + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-scimesh} -d ${POSTGRES_DB:-scimesh_users}"] + interval: 5s + timeout: 3s + retries: 10 + start_period: 5s + + # One-shot: applies migrations, then exits. Schema changes stay an explicit + # deployment step — the service binary never migrates on startup. + migrate: + image: migrate/migrate:v4.17.1 + depends_on: + postgres: + condition: service_healthy + volumes: + - ./migrations:/migrations:ro + command: + - -path=/migrations + - -database=postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh_users}?sslmode=disable + - up + restart: on-failure + + userservice: + build: + context: . + depends_on: + postgres: + condition: service_healthy + # Start only once the schema exists, otherwise the first query fails. + migrate: + condition: service_completed_successfully + environment: + USERSERVICE_ADDR: ":8081" + # Host is the service name: compose resolves it on the project network. + DATABASE_URL: postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh_users}?sslmode=disable + # MUST match the coordinator's JWT secret so it can verify these tokens. + JWT_SECRET: ${JWT_SECRET:-dev-secret-change-me-at-least-32-bytes} + JWT_TTL: ${JWT_TTL:-24h} + DB_MAX_CONNS: "10" + REQUEST_TIMEOUT: "15s" + LOG_LEVEL: ${LOG_LEVEL:-info} + # Logs are teed to stdout (docker logs) and this rotated file on a named + # volume, so they survive a rebuild. + LOG_FILE: /var/log/scimesh/userservice.log + ports: + - "${USERSERVICE_PORT:-8081}:8081" + # A named volume (not a host bind mount): it inherits the image's directory + # ownership, so the non-root process can write to it. + volumes: + - userservice_logs:/var/log/scimesh + healthcheck: + test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8081/health"] + interval: 10s + timeout: 3s + retries: 3 + start_period: 5s + restart: unless-stopped + +volumes: + pgdata: + userservice_logs: diff --git a/users/go.mod b/users/go.mod new file mode 100644 index 0000000..86fce7a --- /dev/null +++ b/users/go.mod @@ -0,0 +1,24 @@ +module github.com/emil28092005/SciMesh/users + +go 1.22 + +require ( + github.com/Masterminds/squirrel v1.5.4 + github.com/cenkalti/backoff/v4 v4.3.0 + github.com/golang-jwt/jwt/v5 v5.3.1 + github.com/google/uuid v1.6.0 + github.com/jackc/pgx/v5 v5.6.0 + github.com/joho/godotenv v1.5.1 + golang.org/x/crypto v0.17.0 + gopkg.in/natefinch/lumberjack.v2 v2.2.1 +) + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect + github.com/jackc/puddle/v2 v2.2.1 // indirect + github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect + github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect + golang.org/x/sync v0.1.0 // indirect + golang.org/x/text v0.14.0 // indirect +) diff --git a/users/go.sum b/users/go.sum new file mode 100644 index 0000000..9de4344 --- /dev/null +++ b/users/go.sum @@ -0,0 +1,45 @@ +github.com/Masterminds/squirrel v1.5.4 h1:uUcX/aBc8O7Fg9kaISIUsHXdKuqehiXAMQTYX8afzqM= +github.com/Masterminds/squirrel v1.5.4/go.mod h1:NNaOrjSoIDfDA40n7sr2tPNZRfjzjA400rg+riTZj10= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY= +github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw= +github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= +github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 h1:SOEGU9fKiNWd/HOJuq6+3iTQz8KNCLtVX6idSoTLdUw= +github.com/lann/builder v0.0.0-20180802200727-47ae307949d0/go.mod h1:dXGbAdH5GtBTC4WfIxhKZfyBF/HBFgRZSWwZ9g/He9o= +github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 h1:P6pPBnrTSX3DEVR4fDembhRWSsG5rVo6hYhAB/ADZrk= +github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0/go.mod h1:vmVJ0l/dxyfGW6FmdpVm2joNMFikkuWg0EoCKLGUMNw= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k= +golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= +gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/users/internal/auth/jwt.go b/users/internal/auth/jwt.go new file mode 100644 index 0000000..1c029c5 --- /dev/null +++ b/users/internal/auth/jwt.go @@ -0,0 +1,67 @@ +package auth + +import ( + "fmt" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/users/internal/domain" +) + +// Claims is the payload of a signed token. Subject (from RegisteredClaims) is +// the user id — it becomes the coordinator's jobs.owner_id; Role drives +// authorization. Both services verify this token locally with the shared HS256 +// secret, so no runtime call back to the userservice is ever needed. +type Claims struct { + Role domain.Role `json:"role"` + jwt.RegisteredClaims +} + +// Issuer signs and verifies tokens with a shared HS256 secret. +type Issuer struct { + secret []byte + ttl time.Duration + now func() time.Time +} + +// NewIssuer builds an Issuer. now defaults to time.Now when nil; tests inject a +// fixed clock to make expiry deterministic. +func NewIssuer(secret string, ttl time.Duration, now func() time.Time) Issuer { + if now == nil { + now = time.Now + } + return Issuer{secret: []byte(secret), ttl: ttl, now: now} +} + +// Issue returns a signed token for the user, valid for the configured TTL. +func (i Issuer) Issue(userID uuid.UUID, role domain.Role) (string, error) { + now := i.now() + claims := Claims{ + Role: role, + RegisteredClaims: jwt.RegisteredClaims{ + Subject: userID.String(), + IssuedAt: jwt.NewNumericDate(now), + ExpiresAt: jwt.NewNumericDate(now.Add(i.ttl)), + }, + } + return jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(i.secret) +} + +// Verify checks the signature and expiry and returns the claims. It pins the +// algorithm to HMAC, rejecting a token that asks for "none" or an RS256 public +// key — the classic algorithm-substitution attack against naive verifiers. +func (i Issuer) Verify(token string) (*Claims, error) { + var claims Claims + _, err := jwt.ParseWithClaims(token, &claims, func(t *jwt.Token) (any, error) { + if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok { + return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"]) + } + return i.secret, nil + }) + if err != nil { + return nil, err + } + return &claims, nil +} diff --git a/users/internal/auth/jwt_test.go b/users/internal/auth/jwt_test.go new file mode 100644 index 0000000..c3d14b1 --- /dev/null +++ b/users/internal/auth/jwt_test.go @@ -0,0 +1,74 @@ +package auth + +import ( + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/users/internal/domain" +) + +const testSecret = "test-secret-at-least-32-bytes-long!!" + +func TestIssueVerifyRoundTrip(t *testing.T) { + iss := NewIssuer(testSecret, time.Hour, nil) + id := uuid.New() + + token, err := iss.Issue(id, domain.RoleAdmin) + if err != nil { + t.Fatalf("issue: %v", err) + } + + claims, err := iss.Verify(token) + if err != nil { + t.Fatalf("verify: %v", err) + } + if claims.Subject != id.String() { + t.Errorf("sub = %q, want %q", claims.Subject, id.String()) + } + if claims.Role != domain.RoleAdmin { + t.Errorf("role = %q, want admin", claims.Role) + } +} + +func TestVerifyRejectsExpired(t *testing.T) { + // Negative TTL: the token is already expired when issued. + iss := NewIssuer(testSecret, -time.Minute, nil) + token, _ := iss.Issue(uuid.New(), domain.RoleUser) + + if _, err := iss.Verify(token); err == nil { + t.Error("expired token accepted") + } +} + +func TestVerifyRejectsWrongSecret(t *testing.T) { + token, _ := NewIssuer(testSecret, time.Hour, nil).Issue(uuid.New(), domain.RoleUser) + + other := NewIssuer("another-secret-also-32-bytes-long!!!", time.Hour, nil) + if _, err := other.Verify(token); err == nil { + t.Error("token verified under the wrong secret") + } +} + +func TestVerifyRejectsNoneAlgorithm(t *testing.T) { + // Forge a token signed with "none" — the classic algorithm-substitution + // attack. A verifier that trusts the header's alg would accept it. + tok := jwt.NewWithClaims(jwt.SigningMethodNone, Claims{ + Role: domain.RoleAdmin, + RegisteredClaims: jwt.RegisteredClaims{ + Subject: uuid.New().String(), + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour)), + }, + }) + raw, err := tok.SignedString(jwt.UnsafeAllowNoneSignatureType) + if err != nil { + t.Fatalf("sign none: %v", err) + } + + iss := NewIssuer(testSecret, time.Hour, nil) + if _, err := iss.Verify(raw); err == nil { + t.Error("none-signed token accepted") + } +} diff --git a/users/internal/auth/password.go b/users/internal/auth/password.go new file mode 100644 index 0000000..db790f4 --- /dev/null +++ b/users/internal/auth/password.go @@ -0,0 +1,40 @@ +// Package auth holds the cryptographic adapters — password hashing and JWT +// signing/verification. They implement use-case ports and keep bcrypt and the +// JWT library out of the domain and use-case layers. +package auth + +import "golang.org/x/crypto/bcrypt" + +// Hasher turns plaintext passwords into storable hashes and checks them back. +type Hasher struct { + cost int +} + +// NewHasher builds a Hasher. A cost of 0 uses bcrypt's default work factor. +func NewHasher(cost int) Hasher { + if cost == 0 { + cost = bcrypt.DefaultCost + } + return Hasher{cost: cost} +} + +// Hash returns the bcrypt hash of password. The salt and the cost are embedded +// in the returned string, so nothing else needs to be stored alongside it. +// +// bcrypt silently ignores input past 72 bytes; the use case rejects longer +// passwords before reaching here so a truncated tail never becomes a security +// surprise. +func (h Hasher) Hash(password string) (string, error) { + b, err := bcrypt.GenerateFromPassword([]byte(password), h.cost) + if err != nil { + return "", err + } + return string(b), nil +} + +// Compare reports whether password matches the stored hash. It returns a +// non-nil error (bcrypt.ErrMismatchedHashAndPassword) on any mismatch, which +// the caller collapses into a generic authentication failure. +func (h Hasher) Compare(hash, password string) error { + return bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) +} diff --git a/users/internal/auth/password_test.go b/users/internal/auth/password_test.go new file mode 100644 index 0000000..2094f3b --- /dev/null +++ b/users/internal/auth/password_test.go @@ -0,0 +1,30 @@ +package auth + +import "testing" + +func TestHashAndCompare(t *testing.T) { + h := NewHasher(0) // default cost + + hash, err := h.Hash("correct horse battery staple") + if err != nil { + t.Fatalf("hash: %v", err) + } + if hash == "correct horse battery staple" { + t.Fatal("hash must not equal the plaintext") + } + if err := h.Compare(hash, "correct horse battery staple"); err != nil { + t.Errorf("correct password rejected: %v", err) + } + if err := h.Compare(hash, "wrong password"); err == nil { + t.Error("wrong password accepted") + } +} + +func TestHashSaltsEachTime(t *testing.T) { + h := NewHasher(0) + a, _ := h.Hash("same") + b, _ := h.Hash("same") + if a == b { + t.Error("two hashes of the same password must differ (random salt)") + } +} diff --git a/users/internal/domain/errors.go b/users/internal/domain/errors.go new file mode 100644 index 0000000..39bd39a --- /dev/null +++ b/users/internal/domain/errors.go @@ -0,0 +1,11 @@ +package domain + +import "errors" + +// Domain validation errors. They describe an entity that cannot be constructed, +// independent of storage or transport, and the HTTP layer maps them to 400. +var ( + ErrEmptyEmail = errors.New("email is required") + ErrInvalidEmail = errors.New("email is not a valid address") + ErrEmptyPasswordHash = errors.New("password hash is required") +) diff --git a/users/internal/domain/user.go b/users/internal/domain/user.go new file mode 100644 index 0000000..45db392 --- /dev/null +++ b/users/internal/domain/user.go @@ -0,0 +1,80 @@ +package domain + +import ( + "net/mail" + "strings" + "time" + + "github.com/google/uuid" +) + +type Role string + +const ( + RoleAdmin Role = "admin" + RoleUser Role = "user" +) + +func (r Role) Valid() bool { + switch r { + case RoleAdmin, RoleUser: + return true + default: + return false + } +} + +type User struct { + ID uuid.UUID + Email string + PasswordHash string + Role Role + CreatedAt time.Time + UpdatedAt time.Time +} + +// NewUser builds a freshly registered account. It normalises the email and +// enforces every invariant a row must satisfy, so an invalid User cannot be +// constructed. The caller supplies the already-hashed password — hashing is an +// adapter's job, not the domain's. +// +// Registration always produces a plain user; promotion to admin is a manual, +// out-of-band operation, never something a request can trigger. +func NewUser(email, passwordHash string, now time.Time) (*User, error) { + email = NormalizeEmail(email) + if err := validateEmail(email); err != nil { + return nil, err + } + if passwordHash == "" { + return nil, ErrEmptyPasswordHash + } + return &User{ + ID: uuid.New(), + Email: email, + PasswordHash: passwordHash, + Role: RoleUser, + CreatedAt: now, + UpdatedAt: now, + }, nil +} + +// NormalizeEmail lower-cases and trims an address so that "Bob@X.com " and +// "bob@x.com" resolve to the same account. Every lookup and every insert must +// pass through here, matching the ck_users_email_lower database constraint. +func NormalizeEmail(email string) string { + return strings.ToLower(strings.TrimSpace(email)) +} + +func validateEmail(email string) error { + if email == "" { + return ErrEmptyEmail + } + // A minimal shape check, not full RFC 5322: real deliverability is proven by + // sending mail, not by a regex. mail.ParseAddress also accepts the + // "Name " form, so we insist the parsed address equals the input. + addr, err := mail.ParseAddress(email) + if err != nil || addr.Address != email { + return ErrInvalidEmail + } + return nil +} diff --git a/users/internal/domain/user_test.go b/users/internal/domain/user_test.go new file mode 100644 index 0000000..9894cd2 --- /dev/null +++ b/users/internal/domain/user_test.go @@ -0,0 +1,62 @@ +package domain + +import ( + "errors" + "testing" + "time" + + "github.com/google/uuid" +) + +func TestNewUserNormalisesAndValidates(t *testing.T) { + now := time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + + u, err := NewUser(" Bob@Example.COM ", "hashed", now) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if u.Email != "bob@example.com" { + t.Errorf("email not normalised: got %q", u.Email) + } + if u.Role != RoleUser { + t.Errorf("new user must default to RoleUser, got %q", u.Role) + } + if u.ID == uuid.Nil { + t.Error("new user must get an id") + } + if !u.CreatedAt.Equal(now) || !u.UpdatedAt.Equal(now) { + t.Error("timestamps not set from clock") + } +} + +func TestNewUserRejectsBadInput(t *testing.T) { + now := time.Now() + cases := []struct { + name string + email string + hash string + wantErr error + }{ + {"empty email", "", "h", ErrEmptyEmail}, + {"no domain", "bob", "h", ErrInvalidEmail}, + {"name form", "Bob ", "h", ErrInvalidEmail}, + {"empty hash", "bob@x.com", "", ErrEmptyPasswordHash}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := NewUser(tc.email, tc.hash, now) + if !errors.Is(err, tc.wantErr) { + t.Errorf("got %v, want %v", err, tc.wantErr) + } + }) + } +} + +func TestRoleValid(t *testing.T) { + if !RoleUser.Valid() || !RoleAdmin.Valid() { + t.Error("user and admin must be valid") + } + if Role("root").Valid() { + t.Error("unknown role must be invalid") + } +} diff --git a/users/internal/infra/clock.go b/users/internal/infra/clock.go new file mode 100644 index 0000000..eedbcde --- /dev/null +++ b/users/internal/infra/clock.go @@ -0,0 +1,13 @@ +// Clock: the real implementation of the usecase.Clock port. It lives out here +// because reading the system clock is infrastructure; tests substitute a fixed one. +package infra + +import "time" + +type System struct{} + +func NewClock() System { return System{} } + +// Now returns UTC so every timestamp the coordinator writes is comparable +// regardless of the host's timezone. +func (System) Now() time.Time { return time.Now().UTC() } diff --git a/users/internal/infra/config.go b/users/internal/infra/config.go new file mode 100644 index 0000000..410ee4b --- /dev/null +++ b/users/internal/infra/config.go @@ -0,0 +1,151 @@ +// Config: userservice settings, read only from the environment, so the same +// binary behaves identically in CI, local, and prod. +package infra + +import ( + "errors" + "fmt" + "io/fs" + "math" + "os" + "strconv" + "time" + + "github.com/joho/godotenv" +) + +// defaultEnvFile is loaded by LoadConfig unless ENV_FILE points elsewhere. +const defaultEnvFile = ".env" + +type Config struct { + // HTTP listen address, e.g. ":8081". + Addr string + // PostgreSQL connection string (pgx format / libpq URL). + DatabaseURL string + + // Shared HS256 secret used to sign JWTs. The coordinator verifies tokens + // with this same secret, so the two values MUST match. This is the only + // secret shared between the services. + JWTSecret string + // How long an issued token stays valid. + TokenTTL time.Duration + // bcrypt work factor. 0 falls back to the library default (currently 10). + BcryptCost int + + // Minimum log level: debug, info, warn, error. + LogLevel string + // Path to a rotated log file. Empty logs to stdout only. + LogFile string + + // Connection pool upper bound. + DBMaxConns int32 + // How long to keep retrying the initial database connection at startup + // before giving up. Covers a Postgres container that is still booting. + DBConnectTimeout time.Duration + // Per-request timeout applied to every handler. + RequestTimeout time.Duration +} + +// LoadConfig reads the environment and fails fast on anything required-but- +// missing or malformed, so a misconfigured process never limps along half-wired. +// +// A .env file (path overridable via ENV_FILE) is loaded first as a local-dev +// convenience. It only fills variables the environment does not already define. +func LoadConfig() (Config, error) { + envFile := os.Getenv("ENV_FILE") + if envFile == "" { + envFile = defaultEnvFile + } + // godotenv.Load never overwrites variables already present in the + // environment, so an orchestrator's values always beat the file. A missing + // file is expected in production, where env vars are injected directly. + if err := godotenv.Load(envFile); err != nil && !errors.Is(err, fs.ErrNotExist) { + return Config{}, fmt.Errorf("load env file %q: %w", envFile, err) + } + + cfg := Config{ + Addr: getEnv("USERSERVICE_ADDR", ":8081"), + DatabaseURL: os.Getenv("DATABASE_URL"), + JWTSecret: os.Getenv("JWT_SECRET"), + LogLevel: getEnv("LOG_LEVEL", "info"), + LogFile: os.Getenv("LOG_FILE"), + TokenTTL: 24 * time.Hour, + DBMaxConns: 10, + DBConnectTimeout: 30 * time.Second, + RequestTimeout: 15 * time.Second, + } + + if cfg.DatabaseURL == "" { + return Config{}, fmt.Errorf("DATABASE_URL is required") + } + if cfg.JWTSecret == "" { + return Config{}, fmt.Errorf("JWT_SECRET is required") + } + // A short secret makes the HMAC brute-forceable; refuse to start with one. + if len(cfg.JWTSecret) < 32 { + return Config{}, fmt.Errorf("JWT_SECRET must be at least 32 bytes") + } + + var err error + if cfg.TokenTTL, err = getEnvDuration("JWT_TTL", cfg.TokenTTL); err != nil { + return Config{}, err + } + if cfg.BcryptCost, err = getEnvInt("BCRYPT_COST", cfg.BcryptCost); err != nil { + return Config{}, err + } + if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil { + return Config{}, err + } + if cfg.DBConnectTimeout, err = getEnvDuration("DB_CONNECT_TIMEOUT", cfg.DBConnectTimeout); err != nil { + return Config{}, err + } + if cfg.RequestTimeout, err = getEnvDuration("REQUEST_TIMEOUT", cfg.RequestTimeout); err != nil { + return Config{}, err + } + + return cfg, nil +} + +func getEnv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func getEnvInt(key string, def int) (int, error) { + v := os.Getenv(key) + if v == "" { + return def, nil + } + n, err := strconv.Atoi(v) + if err != nil { + return 0, fmt.Errorf("%s: %w", key, err) + } + return n, nil +} + +func getEnvInt32(key string, def int32) (int32, error) { + n, err := getEnvInt(key, int(def)) + if err != nil { + return 0, err + } + // On 64-bit builds int is wider than int32, so an oversized value would + // wrap silently — DB_MAX_CONNS=2147483648 becoming a negative pool size. + if n < math.MinInt32 || n > math.MaxInt32 { + return 0, fmt.Errorf("%s: %d is out of range for int32", key, n) + } + return int32(n), nil +} + +func getEnvDuration(key string, def time.Duration) (time.Duration, error) { + v := os.Getenv(key) + if v == "" { + return def, nil + } + d, err := time.ParseDuration(v) + if err != nil { + return 0, fmt.Errorf("%s: %w", key, err) + } + return d, nil +} diff --git a/users/internal/infra/db.go b/users/internal/infra/db.go new file mode 100644 index 0000000..d1135cd --- /dev/null +++ b/users/internal/infra/db.go @@ -0,0 +1,65 @@ +// DB: the PostgreSQL connection pool. +package infra + +import ( + "context" + "log/slog" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/jackc/pgx/v5/pgxpool" +) + +// NewPool builds the single shared pool. The caller owns its lifetime and must +// Close() it on shutdown. +func NewPool(ctx context.Context, cfg Config, log *slog.Logger) (*pgxpool.Pool, error) { + poolCfg, err := pgxpool.ParseConfig(cfg.DatabaseURL) + if err != nil { + return nil, err + } + poolCfg.MaxConns = cfg.DBMaxConns + + pool, err := pgxpool.NewWithConfig(ctx, poolCfg) + if err != nil { + return nil, err + } + // pgxpool.New is lazy, so a ping is needed to actually reach the server. + // It is retried because at startup — especially under docker-compose, where + // the coordinator can boot before Postgres is accepting connections — a + // service should wait for its database rather than crash-loop. + if err := pingWithRetry(ctx, pool, cfg.DBConnectTimeout, log); err != nil { + pool.Close() + return nil, err + } + return pool, nil +} + +// pingWithRetry waits for the database to accept connections, backing off +// between attempts until the budget elapses or ctx is cancelled. +// +// Unlike the transaction retry in storage/postgres, this retries *any* ping +// error: at startup a "connection refused" is the expected, retryable state, +// not an anomaly. +func pingWithRetry(ctx context.Context, pool *pgxpool.Pool, budget time.Duration, log *slog.Logger) error { + b := backoff.NewExponentialBackOff() + b.InitialInterval = 200 * time.Millisecond + b.MaxInterval = 3 * time.Second + b.MaxElapsedTime = budget + + attempt := 0 + return backoff.RetryNotify( + func() error { + // A bounded per-attempt timeout so one hung dial cannot eat the + // whole budget in a single try. + pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + return pool.Ping(pingCtx) + }, + backoff.WithContext(b, ctx), + func(err error, next time.Duration) { + attempt++ + log.Warn("database not ready, retrying", + "attempt", attempt, "retry_in", next.String(), "err", err) + }, + ) +} diff --git a/users/internal/infra/logging.go b/users/internal/infra/logging.go new file mode 100644 index 0000000..b5f49e3 --- /dev/null +++ b/users/internal/infra/logging.go @@ -0,0 +1,65 @@ +package infra + +import ( + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strings" + + "gopkg.in/natefinch/lumberjack.v2" +) + +// NewLogger builds the process logger. +// +// It always writes JSON to stdout, so `docker logs` and any 12-factor log +// collector keep working. When LogFile is set it *also* writes to a +// size-rotated file, so logs survive a container rebuild instead of vanishing +// with the previous stdout stream. Rotation is delegated to lumberjack rather +// than hand-rolled. +// +// The returned Closer flushes and closes the file; call it on shutdown. +func NewLogger(cfg Config) (*slog.Logger, io.Closer, error) { + opts := &slog.HandlerOptions{Level: parseLevel(cfg.LogLevel)} + + var ( + out io.Writer = os.Stdout + closer io.Closer = noopCloser{} + ) + + if cfg.LogFile != "" { + if err := os.MkdirAll(filepath.Dir(cfg.LogFile), 0o750); err != nil { + return nil, nil, fmt.Errorf("create log directory: %w", err) + } + rotator := &lumberjack.Logger{ + Filename: cfg.LogFile, + MaxSize: 50, // megabytes before a rotation + MaxBackups: 5, // keep this many rotated files + MaxAge: 30, // days + Compress: true, + } + // Tee to both: the console stays live while the file is the durable copy. + out = io.MultiWriter(os.Stdout, rotator) + closer = rotator + } + + return slog.New(slog.NewJSONHandler(out, opts)), closer, nil +} + +func parseLevel(s string) slog.Level { + switch strings.ToLower(strings.TrimSpace(s)) { + case "debug": + return slog.LevelDebug + case "warn", "warning": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return slog.LevelInfo + } +} + +type noopCloser struct{} + +func (noopCloser) Close() error { return nil } diff --git a/users/internal/infra/server.go b/users/internal/infra/server.go new file mode 100644 index 0000000..fc49a4c --- /dev/null +++ b/users/internal/infra/server.go @@ -0,0 +1,74 @@ +// Server: the HTTP listener and the background lease reaper, both shut down +// cleanly on a signal. +package infra + +import ( + "context" + "errors" + "log/slog" + "net/http" + "time" +) + +const shutdownGrace = 15 * time.Second + +// Run serves handler until ctx is cancelled, then drains in-flight requests. +func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.Handler) error { + srv := &http.Server{ + Addr: addr, + Handler: handler, + ReadHeaderTimeout: 5 * time.Second, + } + + // Buffered so this goroutine can exit even when nobody reads the channel + // (the ctx.Done branch below) — an unbuffered send would leak it forever. + errCh := make(chan error, 1) + go func() { + log.Info("userservice listening", "addr", addr) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + select { + case err := <-errCh: + return err + case <-ctx.Done(): + log.Info("shutdown signal received") + } + + // A fresh context: ctx is already cancelled, and reusing it would abort the + // very requests we are trying to let finish. + shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGrace) + defer cancel() + return srv.Shutdown(shutdownCtx) +} + +// RunReaper periodically reclaims tasks whose lease elapsed, so a worker that +// died without a heartbeat cannot strand its task in 'leased' forever. +// RunPeriodic invokes fn on an interval until ctx is done, logging how many rows +// each tick affected. It backs the background reapers (expired leases, offline +// workers) — each is a set-based UPDATE that is safe to run repeatedly and +// concurrently across coordinators. +func RunPeriodic(ctx context.Context, log *slog.Logger, name string, interval time.Duration, + fn func(context.Context) (int64, error)) { + + t := time.NewTicker(interval) + defer t.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-t.C: + n, err := fn(ctx) + if err != nil { + log.Debug(name+" skipped", "err", err) + continue + } + if n > 0 { + log.Info(name, "count", n) + } + } + } +} diff --git a/users/internal/memstore/memstore.go b/users/internal/memstore/memstore.go new file mode 100644 index 0000000..8ca2f41 --- /dev/null +++ b/users/internal/memstore/memstore.go @@ -0,0 +1,66 @@ +// Package memstore provides in-memory implementations of the usecase ports for +// fast, deterministic tests that need no database. +package memstore + +import ( + "context" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/users/internal/domain" + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +// UserRepo is an in-memory usecase.UserRepository. It stores copies, so callers +// mutating a returned user cannot corrupt the store. +type UserRepo struct { + mu sync.Mutex + byID map[uuid.UUID]domain.User + byEmail map[string]uuid.UUID +} + +func NewUserRepo() *UserRepo { + return &UserRepo{ + byID: make(map[uuid.UUID]domain.User), + byEmail: make(map[string]uuid.UUID), + } +} + +func (r *UserRepo) Insert(_ context.Context, u *domain.User) error { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.byEmail[u.Email]; ok { + return usecase.ErrEmailExists + } + r.byID[u.ID] = *u + r.byEmail[u.Email] = u.ID + return nil +} + +func (r *UserRepo) GetByEmail(_ context.Context, email string) (*domain.User, error) { + r.mu.Lock() + defer r.mu.Unlock() + id, ok := r.byEmail[email] + if !ok { + return nil, usecase.ErrUserNotFound + } + u := r.byID[id] + return &u, nil +} + +func (r *UserRepo) GetByID(_ context.Context, id uuid.UUID) (*domain.User, error) { + r.mu.Lock() + defer r.mu.Unlock() + u, ok := r.byID[id] + if !ok { + return nil, usecase.ErrUserNotFound + } + return &u, nil +} + +// Clock is a fixed usecase.Clock for deterministic tests. +type Clock struct{ T time.Time } + +func (c Clock) Now() time.Time { return c.T } diff --git a/users/internal/storage/postgres/builder.go b/users/internal/storage/postgres/builder.go new file mode 100644 index 0000000..0025347 --- /dev/null +++ b/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) diff --git a/users/internal/storage/postgres/integration_test.go b/users/internal/storage/postgres/integration_test.go new file mode 100644 index 0000000..d60e27b --- /dev/null +++ b/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) + } +} diff --git a/users/internal/storage/postgres/retry.go b/users/internal/storage/postgres/retry.go new file mode 100644 index 0000000..8f0ae86 --- /dev/null +++ b/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)) +} diff --git a/users/internal/storage/postgres/retry_test.go b/users/internal/storage/postgres/retry_test.go new file mode 100644 index 0000000..bdf77ec --- /dev/null +++ b/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) + } +} diff --git a/users/internal/storage/postgres/tx.go b/users/internal/storage/postgres/tx.go new file mode 100644 index 0000000..b5ccc1c --- /dev/null +++ b/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 +} diff --git a/users/internal/storage/postgres/user_repo.go b/users/internal/storage/postgres/user_repo.go new file mode 100644 index 0000000..b170fcf --- /dev/null +++ b/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 +} diff --git a/users/internal/transport/http/dto.go b/users/internal/transport/http/dto.go new file mode 100644 index 0000000..51b05e5 --- /dev/null +++ b/users/internal/transport/http/dto.go @@ -0,0 +1,41 @@ +package http + +import ( + "time" + + "github.com/emil28092005/SciMesh/users/internal/domain" +) + +// registerRequest / loginRequest are the JSON bodies clients POST. Kept separate +// from the domain so the wire format can evolve without touching the entity. +type registerRequest struct { + Email string `json:"email"` + Password string `json:"password"` +} + +type loginRequest struct { + Email string `json:"email"` + Password string `json:"password"` +} + +// userResponse is the public view of a user. It never carries the password hash. +type userResponse struct { + ID string `json:"id"` + Email string `json:"email"` + Role string `json:"role"` + CreatedAt string `json:"created_at"` +} + +type loginResponse struct { + Token string `json:"token"` + User userResponse `json:"user"` +} + +func toUserResponse(u *domain.User) userResponse { + return userResponse{ + ID: u.ID.String(), + Email: u.Email, + Role: string(u.Role), + CreatedAt: u.CreatedAt.UTC().Format(time.RFC3339), + } +} diff --git a/users/internal/transport/http/errors.go b/users/internal/transport/http/errors.go new file mode 100644 index 0000000..5239bba --- /dev/null +++ b/users/internal/transport/http/errors.go @@ -0,0 +1,61 @@ +package http + +import ( + "encoding/json" + "errors" + "log/slog" + "net/http" + + "github.com/emil28092005/SciMesh/users/internal/domain" + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +// maxJSONBody caps a request body. Credentials are tiny; anything larger is a +// mistake or an attack, so reject it before allocating. +const maxJSONBody = 1 << 20 // 1 MiB + +type errorResponse struct { + Error string `json:"error"` + RequestID string `json:"request_id,omitempty"` +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +// writeError maps a domain or use-case error to an HTTP status and a safe +// message, logging only genuine server faults (5xx). Client errors (4xx) are +// expected and stay out of the error log. +func writeError(w http.ResponseWriter, r *http.Request, log *slog.Logger, err error) { + status, msg := statusForError(err) + if status >= http.StatusInternalServerError { + log.Error("request failed", + "err", err, + "request_id", requestIDFrom(r.Context()), + "path", r.URL.Path, + ) + } + writeJSON(w, status, errorResponse{Error: msg, RequestID: requestIDFrom(r.Context())}) +} + +func statusForError(err error) (int, string) { + switch { + case errors.Is(err, usecase.ErrEmailExists): + return http.StatusConflict, "email already registered" + case errors.Is(err, usecase.ErrInvalidCredentials): + return http.StatusUnauthorized, "invalid email or password" + case errors.Is(err, usecase.ErrUserNotFound): + return http.StatusNotFound, "user not found" + case errors.Is(err, usecase.ErrPasswordTooShort): + return http.StatusBadRequest, "password must be at least 8 characters" + case errors.Is(err, usecase.ErrPasswordTooLong): + return http.StatusBadRequest, "password must be at most 72 bytes" + case errors.Is(err, domain.ErrEmptyEmail), errors.Is(err, domain.ErrInvalidEmail): + return http.StatusBadRequest, "email is not a valid address" + default: + // Don't leak internals; the real error is in the log under request_id. + return http.StatusInternalServerError, "internal error" + } +} diff --git a/users/internal/transport/http/handlers.go b/users/internal/transport/http/handlers.go new file mode 100644 index 0000000..055d609 --- /dev/null +++ b/users/internal/transport/http/handlers.go @@ -0,0 +1,85 @@ +package http + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +// Handlers holds the use cases each endpoint drives. +type Handlers struct { + register *usecase.Register + login *usecase.Login + users usecase.UserRepository + log *slog.Logger +} + +// handleHealth is an unauthenticated liveness probe for the container and load +// balancer. +func (h *Handlers) handleHealth(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +// handleRegister creates an account. It returns 201 with the public user view, +// 409 if the email is taken, or 400 on a malformed body / weak password. +func (h *Handlers) handleRegister(w http.ResponseWriter, r *http.Request) { + var req registerRequest + if !decodeJSON(w, r, &req) { + return + } + u, err := h.register.Execute(r.Context(), req.Email, req.Password) + if err != nil { + writeError(w, r, h.log, err) + return + } + writeJSON(w, http.StatusCreated, toUserResponse(u)) +} + +// handleLogin verifies credentials and returns a signed token plus the user. +func (h *Handlers) handleLogin(w http.ResponseWriter, r *http.Request) { + var req loginRequest + if !decodeJSON(w, r, &req) { + return + } + token, u, err := h.login.Execute(r.Context(), req.Email, req.Password) + if err != nil { + writeError(w, r, h.log, err) + return + } + writeJSON(w, http.StatusOK, loginResponse{Token: token, User: toUserResponse(u)}) +} + +// handleMe returns the caller's own account, proving the token works end to end. +// It reads the user id the JWT middleware stashed in the context. +func (h *Handlers) handleMe(w http.ResponseWriter, r *http.Request) { + id, ok := userIDFrom(r.Context()) + if !ok { + unauthorized(w, r) + return + } + u, err := h.users.GetByID(r.Context(), id) + if err != nil { + writeError(w, r, h.log, err) + return + } + writeJSON(w, http.StatusOK, toUserResponse(u)) +} + +// decodeJSON reads a size-capped JSON body into dst, rejecting unknown fields. +// It writes a 400 and returns false on any problem, so callers can `if +// !decodeJSON(...) { return }`. +func decodeJSON(w http.ResponseWriter, r *http.Request, dst any) bool { + r.Body = http.MaxBytesReader(w, r.Body, maxJSONBody) + dec := json.NewDecoder(r.Body) + dec.DisallowUnknownFields() + if err := dec.Decode(dst); err != nil { + writeJSON(w, http.StatusBadRequest, errorResponse{ + Error: "invalid JSON body", + RequestID: requestIDFrom(r.Context()), + }) + return false + } + return true +} diff --git a/users/internal/transport/http/middleware.go b/users/internal/transport/http/middleware.go new file mode 100644 index 0000000..4d6ef66 --- /dev/null +++ b/users/internal/transport/http/middleware.go @@ -0,0 +1,133 @@ +package http + +import ( + "context" + "crypto/rand" + "encoding/hex" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/users/internal/auth" +) + +type ctxKey string + +const ( + requestIDKey ctxKey = "request_id" + userIDKey ctxKey = "user_id" + roleKey ctxKey = "role" +) + +// withRequestID stamps every request with an ID for correlated logs and error +// bodies. It wraps the auth middleware rather than the other way round, so even +// a rejected request carries an ID the caller can quote in a bug report. +func withRequestID(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + id := newRequestID() + w.Header().Set("X-Request-ID", id) + next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), requestIDKey, id))) + }) +} + +func requestIDFrom(ctx context.Context) string { + if v, ok := ctx.Value(requestIDKey).(string); ok { + return v + } + return "" +} + +func newRequestID() string { + var b [8]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} + +// tokenVerifier is the slice of auth.Issuer the JWT middleware needs. Taking an +// interface keeps the middleware testable with a stub verifier. +type tokenVerifier interface { + Verify(token string) (*auth.Claims, error) +} + +// withJWT verifies the Bearer token and stashes the caller's id and role in the +// request context. It rejects any request without a valid, unexpired HS256 +// token — this is what protects endpoints that act on a specific user. +func withJWT(v tokenVerifier) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + if raw == "" { + unauthorized(w, r) + return + } + claims, err := v.Verify(raw) + if err != nil { + unauthorized(w, r) + return + } + id, err := uuid.Parse(claims.Subject) + if err != nil { + unauthorized(w, r) + return + } + ctx := context.WithValue(r.Context(), userIDKey, id) + ctx = context.WithValue(ctx, roleKey, claims.Role) + next.ServeHTTP(w, r.WithContext(ctx)) + }) + } +} + +func unauthorized(w http.ResponseWriter, r *http.Request) { + w.Header().Set("WWW-Authenticate", "Bearer") + writeJSON(w, http.StatusUnauthorized, errorResponse{ + Error: "unauthorized", + RequestID: requestIDFrom(r.Context()), + }) +} + +// userIDFrom returns the authenticated caller's id, set by withJWT. +func userIDFrom(ctx context.Context) (uuid.UUID, bool) { + id, ok := ctx.Value(userIDKey).(uuid.UUID) + return id, ok +} + +// statusRecorder captures the status code for the access log. +type statusRecorder struct { + http.ResponseWriter + status int +} + +func (s *statusRecorder) WriteHeader(code int) { + s.status = code + s.ResponseWriter.WriteHeader(code) +} + +// withAccessLog records one structured line per request — the minimum needed to +// debug a distributed system after the fact. +func withAccessLog(log *slog.Logger) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} + next.ServeHTTP(rec, r) + log.Info("request", + "request_id", requestIDFrom(r.Context()), + "method", r.Method, + "path", r.URL.Path, + "status", rec.status, + "duration_ms", time.Since(start).Milliseconds(), + ) + }) + } +} + +// chain applies middleware so that the first argument is the outermost layer. +func chain(h http.Handler, mw ...func(http.Handler) http.Handler) http.Handler { + for i := len(mw) - 1; i >= 0; i-- { + h = mw[i](h) + } + return h +} diff --git a/users/internal/transport/http/server.go b/users/internal/transport/http/server.go new file mode 100644 index 0000000..58a2a67 --- /dev/null +++ b/users/internal/transport/http/server.go @@ -0,0 +1,41 @@ +// Package http exposes the userservice over HTTP: registration, login, and a +// token-protected /me. It owns routing, request decoding, and error mapping; +// business rules live in the usecase layer. +package http + +import ( + "log/slog" + "net/http" + + "github.com/emil28092005/SciMesh/users/internal/auth" + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +// UseCases bundles the application services the handlers drive. +type UseCases struct { + Register *usecase.Register + Login *usecase.Login + Users usecase.UserRepository +} + +// NewServer wires the routes and the middleware stack and returns the handler. +// The issuer verifies tokens for the protected /me route. +func NewServer(log *slog.Logger, uc UseCases, issuer auth.Issuer) http.Handler { + h := &Handlers{ + register: uc.Register, + login: uc.Login, + users: uc.Users, + log: log, + } + + mux := http.NewServeMux() + // Method-aware patterns (Go 1.22+): a GET to /register is a 405, not a match. + mux.HandleFunc("GET /health", h.handleHealth) + mux.HandleFunc("POST /register", h.handleRegister) + mux.HandleFunc("POST /login", h.handleLogin) + // /me proves a token round-trips; it sits behind JWT auth. + mux.Handle("GET /me", chain(http.HandlerFunc(h.handleMe), withJWT(issuer))) + + // Outermost first: every request gets an ID and an access-log line. + return chain(mux, withRequestID, withAccessLog(log)) +} diff --git a/users/internal/transport/http/server_test.go b/users/internal/transport/http/server_test.go new file mode 100644 index 0000000..f74eb0f --- /dev/null +++ b/users/internal/transport/http/server_test.go @@ -0,0 +1,167 @@ +package http_test + +import ( + "bytes" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/emil28092005/SciMesh/users/internal/auth" + "github.com/emil28092005/SciMesh/users/internal/memstore" + apihttp "github.com/emil28092005/SciMesh/users/internal/transport/http" + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +const secret = "server-test-secret-32-bytes-long!!!!" + +func newTestServer() http.Handler { + users := memstore.NewUserRepo() + hasher := auth.NewHasher(4) + clk := memstore.Clock{T: time.Date(2026, 7, 26, 0, 0, 0, 0, time.UTC)} + // Real clock for the issuer so tokens are valid at verification time. + issuer := auth.NewIssuer(secret, time.Hour, nil) + + uc := apihttp.UseCases{ + Register: usecase.NewRegister(users, hasher, clk), + Login: usecase.NewLogin(users, hasher, issuer), + Users: users, + } + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + return apihttp.NewServer(log, uc, issuer) +} + +func do(t *testing.T, h http.Handler, method, path, token string, body any) *httptest.ResponseRecorder { + t.Helper() + var buf bytes.Buffer + if body != nil { + if err := json.NewEncoder(&buf).Encode(body); err != nil { + t.Fatal(err) + } + } + req := httptest.NewRequest(method, path, &buf) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +func TestRegisterThenLoginThenMe(t *testing.T) { + h := newTestServer() + creds := map[string]string{"email": "flow@example.com", "password": "password123"} + + // Register -> 201 + rec := do(t, h, http.MethodPost, "/register", "", creds) + if rec.Code != http.StatusCreated { + t.Fatalf("register: got %d, body %s", rec.Code, rec.Body) + } + + // Login -> 200 with a token + rec = do(t, h, http.MethodPost, "/login", "", creds) + if rec.Code != http.StatusOK { + t.Fatalf("login: got %d, body %s", rec.Code, rec.Body) + } + var lr struct { + Token string `json:"token"` + User struct { + Email string `json:"email"` + Role string `json:"role"` + } `json:"user"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &lr); err != nil { + t.Fatal(err) + } + if lr.Token == "" || lr.User.Email != "flow@example.com" || lr.User.Role != "user" { + t.Fatalf("unexpected login body: %+v", lr) + } + + // /me with the token -> 200, same user + rec = do(t, h, http.MethodGet, "/me", lr.Token, nil) + if rec.Code != http.StatusOK { + t.Fatalf("me: got %d, body %s", rec.Code, rec.Body) + } + var me struct { + Email string `json:"email"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &me); err != nil { + t.Fatal(err) + } + if me.Email != "flow@example.com" { + t.Errorf("me email = %q", me.Email) + } +} + +func TestRegisterDuplicate(t *testing.T) { + h := newTestServer() + creds := map[string]string{"email": "dup@example.com", "password": "password123"} + _ = do(t, h, http.MethodPost, "/register", "", creds) + + rec := do(t, h, http.MethodPost, "/register", "", creds) + if rec.Code != http.StatusConflict { + t.Errorf("duplicate register: got %d, want 409", rec.Code) + } +} + +func TestRegisterValidation(t *testing.T) { + h := newTestServer() + cases := []struct { + name string + body map[string]string + }{ + {"weak password", map[string]string{"email": "a@b.com", "password": "short"}}, + {"bad email", map[string]string{"email": "nope", "password": "password123"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rec := do(t, h, http.MethodPost, "/register", "", tc.body) + if rec.Code != http.StatusBadRequest { + t.Errorf("got %d, want 400", rec.Code) + } + }) + } +} + +func TestRegisterRejectsUnknownFields(t *testing.T) { + h := newTestServer() + rec := do(t, h, http.MethodPost, "/register", "", map[string]string{ + "email": "a@b.com", "password": "password123", "role": "admin", + }) + if rec.Code != http.StatusBadRequest { + t.Errorf("unknown field must be rejected: got %d", rec.Code) + } +} + +func TestLoginWrongPassword(t *testing.T) { + h := newTestServer() + _ = do(t, h, http.MethodPost, "/register", "", map[string]string{ + "email": "x@example.com", "password": "password123", + }) + rec := do(t, h, http.MethodPost, "/login", "", map[string]string{ + "email": "x@example.com", "password": "wrongpass1", + }) + if rec.Code != http.StatusUnauthorized { + t.Errorf("got %d, want 401", rec.Code) + } +} + +func TestMeRequiresToken(t *testing.T) { + h := newTestServer() + if rec := do(t, h, http.MethodGet, "/me", "", nil); rec.Code != http.StatusUnauthorized { + t.Errorf("no token: got %d, want 401", rec.Code) + } + if rec := do(t, h, http.MethodGet, "/me", "garbage.token.here", nil); rec.Code != http.StatusUnauthorized { + t.Errorf("bad token: got %d, want 401", rec.Code) + } +} + +func TestHealth(t *testing.T) { + h := newTestServer() + if rec := do(t, h, http.MethodGet, "/health", "", nil); rec.Code != http.StatusOK { + t.Errorf("health: got %d", rec.Code) + } +} diff --git a/users/internal/usecase/errors.go b/users/internal/usecase/errors.go new file mode 100644 index 0000000..4e8131d --- /dev/null +++ b/users/internal/usecase/errors.go @@ -0,0 +1,18 @@ +package usecase + +import "errors" + +var ( + // Repository-contract errors, returned by UserRepository implementations. + ErrEmailExists = errors.New("email already registered") + ErrUserNotFound = errors.New("user not found") + + // Use-case errors surfaced to the transport layer. + // + // ErrInvalidCredentials is deliberately returned for both an unknown email + // and a wrong password, so an attacker cannot use the response to learn + // which emails are registered. + ErrInvalidCredentials = errors.New("invalid email or password") + ErrPasswordTooShort = errors.New("password too short") + ErrPasswordTooLong = errors.New("password too long") +) diff --git a/users/internal/usecase/login.go b/users/internal/usecase/login.go new file mode 100644 index 0000000..7c57c3d --- /dev/null +++ b/users/internal/usecase/login.go @@ -0,0 +1,42 @@ +package usecase + +import ( + "context" + "errors" + + "github.com/emil28092005/SciMesh/users/internal/domain" +) + +// Login verifies credentials and issues a signed token. +type Login struct { + users UserRepository + hasher PasswordHasher + tokens TokenIssuer +} + +func NewLogin(users UserRepository, hasher PasswordHasher, tokens TokenIssuer) *Login { + return &Login{users: users, hasher: hasher, tokens: tokens} +} + +// Execute returns a signed token and the user on success. It returns +// ErrInvalidCredentials for both an unknown email and a wrong password so the +// two cases are indistinguishable to a caller probing for valid accounts. +func (l *Login) Execute(ctx context.Context, email, password string) (string, *domain.User, error) { + u, err := l.users.GetByEmail(ctx, domain.NormalizeEmail(email)) + if err != nil { + if errors.Is(err, ErrUserNotFound) { + return "", nil, ErrInvalidCredentials + } + return "", nil, err + } + + if err := l.hasher.Compare(u.PasswordHash, password); err != nil { + return "", nil, ErrInvalidCredentials + } + + token, err := l.tokens.Issue(u.ID, u.Role) + if err != nil { + return "", nil, err + } + return token, u, nil +} diff --git a/users/internal/usecase/ports.go b/users/internal/usecase/ports.go new file mode 100644 index 0000000..3f75323 --- /dev/null +++ b/users/internal/usecase/ports.go @@ -0,0 +1,41 @@ +// Package usecase holds the application logic — registration and login — plus +// the ports (interfaces) it depends on. The concrete adapters (PostgreSQL, +// bcrypt, JWT) are injected from cmd, so this package never imports them. +package usecase + +import ( + "context" + "time" + + "github.com/google/uuid" + + "github.com/emil28092005/SciMesh/users/internal/domain" +) + +// UserRepository persists and looks up users. Implementations return the +// sentinel errors in errors.go so the use cases can react without knowing about +// SQL or driver types. +type UserRepository interface { + // Insert stores a new user, returning ErrEmailExists if the email is taken. + Insert(ctx context.Context, u *domain.User) error + // GetByEmail returns the user with the (normalised) email, or ErrUserNotFound. + GetByEmail(ctx context.Context, email string) (*domain.User, error) + // GetByID returns the user with id, or ErrUserNotFound. + GetByID(ctx context.Context, id uuid.UUID) (*domain.User, error) +} + +// PasswordHasher hashes and verifies passwords. The bcrypt adapter satisfies it. +type PasswordHasher interface { + Hash(password string) (string, error) + Compare(hash, password string) error +} + +// TokenIssuer mints a signed access token for an authenticated user. +type TokenIssuer interface { + Issue(userID uuid.UUID, role domain.Role) (string, error) +} + +// Clock reads the current time; a fake one makes tests deterministic. +type Clock interface { + Now() time.Time +} diff --git a/users/internal/usecase/register.go b/users/internal/usecase/register.go new file mode 100644 index 0000000..1eef927 --- /dev/null +++ b/users/internal/usecase/register.go @@ -0,0 +1,56 @@ +package usecase + +import ( + "context" + + "github.com/emil28092005/SciMesh/users/internal/domain" +) + +const ( + // minPasswordLen is a floor, not a policy engine — enough to reject the + // obviously weak without pretending to measure real strength. + minPasswordLen = 8 + // maxPasswordLen is bcrypt's hard input limit: it ignores bytes past 72, so + // accepting a longer password would silently hash only its prefix. + maxPasswordLen = 72 +) + +// Register creates a new account: it validates the password, hashes it, builds +// the domain user, and persists it. +type Register struct { + users UserRepository + hasher PasswordHasher + clk Clock +} + +func NewRegister(users UserRepository, hasher PasswordHasher, clk Clock) *Register { + return &Register{users: users, hasher: hasher, clk: clk} +} + +// Execute registers email/password and returns the persisted user. The returned +// user carries no plaintext password, only its hash. +func (r *Register) Execute(ctx context.Context, email, password string) (*domain.User, error) { + if len(password) < minPasswordLen { + return nil, ErrPasswordTooShort + } + if len(password) > maxPasswordLen { + return nil, ErrPasswordTooLong + } + + hash, err := r.hasher.Hash(password) + if err != nil { + return nil, err + } + + // NewUser normalises the email and enforces its shape; it returns a domain + // validation error the transport layer maps to 400. + u, err := domain.NewUser(email, hash, r.clk.Now()) + if err != nil { + return nil, err + } + + if err := r.users.Insert(ctx, u); err != nil { + return nil, err + } + return u, nil +} diff --git a/users/internal/usecase/usecase_test.go b/users/internal/usecase/usecase_test.go new file mode 100644 index 0000000..2e470b8 --- /dev/null +++ b/users/internal/usecase/usecase_test.go @@ -0,0 +1,133 @@ +package usecase_test + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/emil28092005/SciMesh/users/internal/auth" + "github.com/emil28092005/SciMesh/users/internal/domain" + "github.com/emil28092005/SciMesh/users/internal/memstore" + "github.com/emil28092005/SciMesh/users/internal/usecase" +) + +const secret = "usecase-test-secret-32-bytes-long!!!" + +func newFixtures() (*usecase.Register, *usecase.Login, *memstore.UserRepo) { + users := memstore.NewUserRepo() + hasher := auth.NewHasher(4) // low cost keeps tests fast + clk := memstore.Clock{T: time.Date(2026, 7, 26, 0, 0, 0, 0, time.UTC)} + // The issuer uses the real clock (nil): token expiry is validated against + // wall-clock time, so a fixed issue-time would make tokens instantly stale. + issuer := auth.NewIssuer(secret, time.Hour, nil) + + reg := usecase.NewRegister(users, hasher, clk) + login := usecase.NewLogin(users, hasher, issuer) + return reg, login, users +} + +func TestRegisterSuccess(t *testing.T) { + reg, _, users := newFixtures() + + u, err := reg.Execute(context.Background(), "Alice@Example.com", "password123") + if err != nil { + t.Fatalf("register: %v", err) + } + if u.Email != "alice@example.com" { + t.Errorf("email not normalised: %q", u.Email) + } + if u.Role != domain.RoleUser { + t.Errorf("role = %q, want user", u.Role) + } + if strings.Contains(u.PasswordHash, "password123") { + t.Error("password stored in cleartext") + } + if _, err := users.GetByEmail(context.Background(), "alice@example.com"); err != nil { + t.Errorf("user not persisted: %v", err) + } +} + +func TestRegisterDuplicateEmail(t *testing.T) { + reg, _, _ := newFixtures() + ctx := context.Background() + + if _, err := reg.Execute(ctx, "dup@example.com", "password123"); err != nil { + t.Fatalf("first register: %v", err) + } + _, err := reg.Execute(ctx, "Dup@example.com", "password123") // different case, same email + if !errors.Is(err, usecase.ErrEmailExists) { + t.Errorf("got %v, want ErrEmailExists", err) + } +} + +func TestRegisterPasswordPolicy(t *testing.T) { + reg, _, _ := newFixtures() + ctx := context.Background() + + if _, err := reg.Execute(ctx, "a@b.com", "short"); !errors.Is(err, usecase.ErrPasswordTooShort) { + t.Errorf("short password: got %v", err) + } + long := strings.Repeat("x", 73) + if _, err := reg.Execute(ctx, "a@b.com", long); !errors.Is(err, usecase.ErrPasswordTooLong) { + t.Errorf("long password: got %v", err) + } +} + +func TestRegisterInvalidEmail(t *testing.T) { + reg, _, _ := newFixtures() + _, err := reg.Execute(context.Background(), "not-an-email", "password123") + if !errors.Is(err, domain.ErrInvalidEmail) { + t.Errorf("got %v, want ErrInvalidEmail", err) + } +} + +func TestLoginSuccess(t *testing.T) { + reg, login, _ := newFixtures() + ctx := context.Background() + if _, err := reg.Execute(ctx, "user@example.com", "password123"); err != nil { + t.Fatal(err) + } + + token, u, err := login.Execute(ctx, "User@Example.com", "password123") + if err != nil { + t.Fatalf("login: %v", err) + } + if token == "" { + t.Error("empty token") + } + if u.Email != "user@example.com" { + t.Errorf("wrong user returned: %q", u.Email) + } + + // The token must verify and carry this user's id. + claims, err := auth.NewIssuer(secret, time.Hour, nil).Verify(token) + if err != nil { + t.Fatalf("issued token does not verify: %v", err) + } + if claims.Subject != u.ID.String() { + t.Errorf("token sub = %q, want %q", claims.Subject, u.ID.String()) + } +} + +func TestLoginWrongPassword(t *testing.T) { + reg, login, _ := newFixtures() + ctx := context.Background() + if _, err := reg.Execute(ctx, "user@example.com", "password123"); err != nil { + t.Fatal(err) + } + + _, _, err := login.Execute(ctx, "user@example.com", "wrongpass1") + if !errors.Is(err, usecase.ErrInvalidCredentials) { + t.Errorf("got %v, want ErrInvalidCredentials", err) + } +} + +func TestLoginUnknownEmailIsIndistinguishable(t *testing.T) { + _, login, _ := newFixtures() + _, _, err := login.Execute(context.Background(), "ghost@example.com", "password123") + if !errors.Is(err, usecase.ErrInvalidCredentials) { + t.Errorf("unknown email must return ErrInvalidCredentials, got %v", err) + } +} diff --git a/users/migrations/0001_users.down.sql b/users/migrations/0001_users.down.sql new file mode 100644 index 0000000..86f518c --- /dev/null +++ b/users/migrations/0001_users.down.sql @@ -0,0 +1,6 @@ +BEGIN; + +DROP TABLE IF EXISTS users; +DROP TYPE IF EXISTS user_role; + +COMMIT; diff --git a/users/migrations/0001_users.up.sql b/users/migrations/0001_users.up.sql new file mode 100644 index 0000000..f0a76f1 --- /dev/null +++ b/users/migrations/0001_users.up.sql @@ -0,0 +1,26 @@ +BEGIN; + +-- Static permission bundles. Roles rarely change, so the role→permission +-- mapping lives in code (auth middleware), not in a table. New role = deploy. +CREATE TYPE user_role AS ENUM ('user','admin'); + +-- One human account. The id is the stable identity that ends up in the JWT +-- `sub` claim; the coordinator stores it as jobs.owner_id. +CREATE TABLE users ( + id uuid PRIMARY KEY, + -- Login handle. App lowercases before insert/lookup, so uniqueness is + -- case-insensitive in practice. + email text NOT NULL, + -- Output of bcrypt/argon2. The salt and cost parameters are embedded in + -- this string, so there is NO separate salt column to store. + password_hash text NOT NULL, + role user_role NOT NULL DEFAULT 'user', + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + + -- No two accounts share a login. + CONSTRAINT uq_users_email UNIQUE (email), + CONSTRAINT ck_users_email_lower CHECK (email = lower(email)) +); + +COMMIT; diff --git a/users/scripts/smoke.sh b/users/scripts/smoke.sh new file mode 100755 index 0000000..d2e3614 --- /dev/null +++ b/users/scripts/smoke.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# End-to-end smoke test against a running userservice. Exercises the full auth +# flow and exits non-zero on the first unexpected status. +# +# HOST=http://localhost:8081 ./scripts/smoke.sh +set -euo pipefail + +HOST="${HOST:-http://localhost:8081}" +EMAIL="smoke-$(date +%s)-$RANDOM@example.com" +PASSWORD="password123" + +pass() { printf ' ok %s\n' "$1"; } +fail() { printf ' FAIL %s\n' "$1" >&2; exit 1; } + +# expect METHOD PATH WANT_STATUS [JSON_BODY] [BEARER] +# Prints the response body to stdout so callers can parse it. +expect() { + local method="$1" path="$2" want="$3" body="${4:-}" token="${5:-}" + local args=(-s -o /tmp/smoke_body -w '%{http_code}' -X "$method" "$HOST$path") + [ -n "$body" ] && args+=(-H 'Content-Type: application/json' -d "$body") + [ -n "$token" ] && args+=(-H "Authorization: Bearer $token") + local code + code="$(curl "${args[@]}")" + if [ "$code" != "$want" ]; then + printf 'body: %s\n' "$(cat /tmp/smoke_body)" >&2 + fail "$method $path -> $code (want $want)" + fi + pass "$method $path -> $code" + cat /tmp/smoke_body +} + +echo "smoke: $HOST (user $EMAIL)" + +expect GET /health 200 >/dev/null + +expect POST /register 201 "{\"email\":\"$EMAIL\",\"password\":\"$PASSWORD\"}" >/dev/null +expect POST /register 409 "{\"email\":\"$EMAIL\",\"password\":\"$PASSWORD\"}" >/dev/null +expect POST /register 400 "{\"email\":\"$EMAIL\",\"password\":\"short\"}" >/dev/null + +# Login and capture the token (extract the "token" JSON string field). +login_body="$(expect POST /login 200 "{\"email\":\"$EMAIL\",\"password\":\"$PASSWORD\"}")" +TOKEN="$(printf '%s' "$login_body" | sed -n 's/.*"token":"\([^"]*\)".*/\1/p')" +[ -n "$TOKEN" ] || fail "login returned no token" +pass "captured token" + +expect POST /login 401 "{\"email\":\"$EMAIL\",\"password\":\"wrongpass1\"}" >/dev/null + +expect GET /me 200 "" "$TOKEN" >/dev/null +expect GET /me 401 "" >/dev/null +expect GET /me 401 "" "not-a-token" >/dev/null + +echo "smoke: all checks passed ✓"