diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..b0517d25 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +.git +.github +.agents +.codex +coverage.out +README.md diff --git a/.gitignore b/.gitignore index c1643802..9af65526 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,8 @@ build/ .tmp/ .env .env.* +coverage.out +wallet-api +*.exe +.idea/ +.vscode/ diff --git a/README.md b/README.md index 58c62d1a..bd15d156 100644 --- a/README.md +++ b/README.md @@ -1,35 +1,162 @@ -# Wallet Transfer Assignment Repository +# Wallet Transfer Service -This repository is a reusable coding assignment template for evaluating backend engineers on wallet transfers, idempotency, concurrency control, and double-entry ledger design. +A PostgreSQL-backed wallet transfer API written in Go. The implementation uses +Clean Architecture boundaries, stored wallet balances, an immutable +double-entry ledger, transactional idempotency, and row-level locking. -## Included +## Architecture -- `ASSIGNMENT.md` - candidate-facing prompt -- `.github/pull_request_template.md` - required PR structure -- `.github/workflows/ci.yml` - lint, format, test placeholder workflow -- `.github/workflows/sonarqube.yml` - SonarQube pull request analysis -- `.github/copilot-instructions.md` - repository-level Copilot review guidance -- `evaluation_guide.md` - reviewer rubric -- `branch-protection-checklist.md` - GitHub setup checklist +```text +HTTP handler -> transfer service -> repository contracts -> PostgreSQL adapter + | + domain +``` -## Intended use +- `cmd/api`: dependency wiring and HTTP server lifecycle +- `internal/domain`: wallet, transfer, ledger, and state-transition models +- `internal/service`: transfer orchestration and business rules +- `internal/repository`: persistence and transaction contracts +- `internal/repository/postgres`: PostgreSQL implementation +- `internal/handler/http`: request validation and HTTP mapping +- `migrations`: database schema +- `tests/integration`: PostgreSQL behavior and concurrency tests -1. Mark this repository as a GitHub template repository. -2. Create one private repository per candidate from the template. -3. Add the candidate as a collaborator. -4. Ask them to submit via a pull request into `main`. -5. Enable required checks, SonarQube, and Copilot review in GitHub. +Amounts are signed 64-bit integers in the currency's smallest unit. Floating +point values are intentionally not accepted. -## Notes +## Transaction and concurrency model -- Copilot automatic pull request review is configured in GitHub repository or organization settings, not purely through files in the repo. -- The `copilot-instructions.md` file included here provides repository-specific review guidance once Copilot review is enabled. -- The CI workflow is language-agnostic by default and expects you to set the `LINT_CMD`, `FORMAT_CHECK_CMD`, and `TEST_CMD` repository variables or replace the commands directly. +Each new request runs in one PostgreSQL transaction: -## How to Submit Assignment +1. Insert the idempotency key and canonical request hash. +2. Create a `PENDING` transfer and link it to the idempotency record. +3. Lock both wallets with `SELECT ... FOR UPDATE`, ordered by wallet ID. +4. Validate funds, update both stored balances, and insert one `DEBIT` plus one + `CREDIT` ledger entry. +5. Move the transfer to `PROCESSED`, or commit a deterministic business failure + as `FAILED`. -1. **Fork this repository** to your own GitHub account. -2. Complete the assignment described in [`ASSIGNMENT.md`](./ASSIGNMENT.md). -3. **Raise a Pull Request** back to this repository (`main` branch) with your full solution. +The primary key on `idempotency_records.idempotency_key` serializes concurrent +duplicates. A duplicate with the same request hash reads the original terminal +transfer. Reusing the key for a different payload returns `409 Conflict`. +Infrastructure failures roll back the entire transaction and are safe to retry. -Your PR branch should be named: `solution/` (e.g., `solution/jane-doe`). +## Run with Docker Compose + +Requirements: Docker with Compose support. + +```bash +docker compose up --build +``` + +The migration is applied automatically when PostgreSQL initializes. Seed two +wallets: + +```bash +docker compose exec postgres psql -U wallet -d wallet -c \ + "INSERT INTO wallets (id, balance) VALUES ('wallet_1', 1000), ('wallet_2', 250);" +``` + +Create a transfer: + +```bash +curl -i http://localhost:8080/transfers \ + -H "Content-Type: application/json" \ + -d '{ + "idempotencyKey": "abc123", + "fromWalletId": "wallet_1", + "toWalletId": "wallet_2", + "amount": 100 + }' +``` + +Re-send the same request to receive the original result without another debit. +Change the amount while retaining `abc123` to receive `409 Conflict`. + +To delete local data and rerun the initialization migration: + +```bash +docker compose down -v +``` + +## Run locally + +Start PostgreSQL and apply the migration: + +```bash +psql "$DATABASE_URL" -f migrations/001_init.sql +``` + +Set the database URL and run the API: + +```bash +export DATABASE_URL='postgres://wallet:wallet@localhost:5432/wallet?sslmode=disable' +go run ./cmd/api +``` + +On PowerShell: + +```powershell +$env:DATABASE_URL = 'postgres://wallet:wallet@localhost:5432/wallet?sslmode=disable' +go run ./cmd/api +``` + +The API listens on `:8080` by default. Override it with `HTTP_ADDRESS`. + +## API + +### `POST /transfers` + +Request: + +```json +{ + "idempotencyKey": "abc123", + "fromWalletId": "wallet_1", + "toWalletId": "wallet_2", + "amount": 100 +} +``` + +Responses: + +- `201 Created`: transfer is `PROCESSED` +- `400 Bad Request`: invalid input or JSON +- `404 Not Found`: one or both wallets do not exist; transfer is `FAILED` +- `409 Conflict`: idempotency key was used with a different payload +- `422 Unprocessable Entity`: insufficient balance; transfer is `FAILED` +- `500 Internal Server Error`: transaction was rolled back and may be retried + +`GET /healthz` provides a process health endpoint. + +## Tests + +Run unit tests: + +```bash +go test ./internal/... +``` + +Run every test. Integration tests skip when no database URL is configured: + +```bash +go test ./... +``` + +Run PostgreSQL integration and concurrency tests against a dedicated test +database: + +```bash +export INTEGRATION_DATABASE_URL='postgres://wallet:wallet@localhost:5432/wallet_test?sslmode=disable' +go test ./tests/integration -count=1 +``` + +PowerShell: + +```powershell +$env:INTEGRATION_DATABASE_URL = 'postgres://wallet:wallet@localhost:5432/wallet_test?sslmode=disable' +go test ./tests/integration -count=1 +``` + +The integration suite truncates the wallet service tables between tests. Do not +point it at a database containing data you need. diff --git a/cmd/api/main.go b/cmd/api/main.go new file mode 100644 index 00000000..b34db37b --- /dev/null +++ b/cmd/api/main.go @@ -0,0 +1,75 @@ +package main + +import ( + "context" + "errors" + "log/slog" + "net/http" + "os" + "os/signal" + "syscall" + + httpapi "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/handler/http" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/platform/config" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/platform/database" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/platform/logging" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/repository/postgres" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/service" +) + +func main() { + logger := logging.New() + if err := run(logger); err != nil { + logger.Error("application stopped", "error", err) + os.Exit(1) + } +} + +func run(logger *slog.Logger) error { + cfg, err := config.Load() + if err != nil { + return err + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + pool, err := database.OpenPostgres(ctx, cfg.DatabaseURL) + if err != nil { + return err + } + defer pool.Close() + + unitOfWork := postgres.NewUnitOfWork(pool) + transferService := service.NewTransferService(unitOfWork) + handler := httpapi.NewHandler(transferService, logger) + + server := &http.Server{ + Addr: cfg.HTTPAddress, + Handler: handler.Routes(), + ReadTimeout: cfg.ReadTimeout, + WriteTimeout: cfg.WriteTimeout, + IdleTimeout: cfg.IdleTimeout, + } + + serverError := make(chan error, 1) + go func() { + logger.Info("HTTP server started", "address", cfg.HTTPAddress) + serverError <- server.ListenAndServe() + }() + + select { + case err := <-serverError: + if !errors.Is(err, http.ErrServerClosed) { + return err + } + case <-ctx.Done(): + logger.Info("shutdown requested") + shutdownCtx, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + return err + } + } + return nil +} diff --git a/deployments/docker/Dockerfile b/deployments/docker/Dockerfile new file mode 100644 index 00000000..d71559ce --- /dev/null +++ b/deployments/docker/Dockerfile @@ -0,0 +1,15 @@ +FROM golang:1.24-alpine AS build + +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/wallet-api ./cmd/api + +FROM alpine:3.21 + +RUN addgroup -S app && adduser -S app -G app +COPY --from=build /out/wallet-api /usr/local/bin/wallet-api +USER app +EXPOSE 8080 +ENTRYPOINT ["wallet-api"] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 00000000..0a3cc5a1 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,38 @@ +services: + postgres: + image: postgres:17-alpine + environment: + POSTGRES_DB: wallet + POSTGRES_USER: wallet + POSTGRES_PASSWORD: wallet + ports: + - "5432:5432" + volumes: + - postgres_data:/var/lib/postgresql/data + - ./migrations/001_init.sql:/docker-entrypoint-initdb.d/001_init.sql:ro + healthcheck: + test: ["CMD-SHELL", "pg_isready -U wallet -d wallet"] + interval: 2s + timeout: 3s + retries: 15 + + api: + build: + context: . + dockerfile: deployments/docker/Dockerfile + environment: + DATABASE_URL: postgres://wallet:wallet@postgres:5432/wallet?sslmode=disable + HTTP_ADDRESS: :8080 + ports: + - "8080:8080" + depends_on: + postgres: + condition: service_healthy + healthcheck: + test: ["CMD", "wget", "-q", "-O", "-", "http://localhost:8080/healthz"] + interval: 5s + timeout: 3s + retries: 10 + +volumes: + postgres_data: diff --git a/go.mod b/go.mod new file mode 100644 index 00000000..470f2abe --- /dev/null +++ b/go.mod @@ -0,0 +1,14 @@ +module github.com/sumiyapuletipalli/wallet-transfer-assignment + +go 1.24.0 + +require github.com/jackc/pgx/v5 v5.7.5 + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + golang.org/x/crypto v0.37.0 // indirect + golang.org/x/sync v0.13.0 // indirect + golang.org/x/text v0.24.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 00000000..85a678b1 --- /dev/null +++ b/go.sum @@ -0,0 +1,28 @@ +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/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-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.7.5 h1:JHGfMnQY+IEtGM63d+NGMjoRpysB2JBwDr5fsngwmJs= +github.com/jackc/pgx/v5 v5.7.5/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +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.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.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= +golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= +golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610= +golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0= +golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +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/internal/domain/errors.go b/internal/domain/errors.go new file mode 100644 index 00000000..61b23baa --- /dev/null +++ b/internal/domain/errors.go @@ -0,0 +1,12 @@ +package domain + +import "errors" + +var ( + ErrInvalidIdempotencyKey = errors.New("idempotency key is required and must not exceed 255 characters") + ErrInvalidWalletID = errors.New("wallet IDs are required and must not exceed 255 characters") + ErrSameWallet = errors.New("source and destination wallets must differ") + ErrInvalidAmount = errors.New("amount must be greater than zero") + ErrInvalidTransition = errors.New("invalid transfer state transition") + ErrIdempotencyConflict = errors.New("idempotency key was already used with a different request") +) diff --git a/internal/domain/idempotency.go b/internal/domain/idempotency.go new file mode 100644 index 00000000..1468fa4c --- /dev/null +++ b/internal/domain/idempotency.go @@ -0,0 +1,7 @@ +package domain + +type IdempotencyRecord struct { + Key string + RequestHash string + TransferID string +} diff --git a/internal/domain/ledger.go b/internal/domain/ledger.go new file mode 100644 index 00000000..51bd257f --- /dev/null +++ b/internal/domain/ledger.go @@ -0,0 +1,16 @@ +package domain + +type LedgerEntryType string + +const ( + LedgerEntryDebit LedgerEntryType = "DEBIT" + LedgerEntryCredit LedgerEntryType = "CREDIT" +) + +type LedgerEntry struct { + ID string + WalletID string + TransferID string + Type LedgerEntryType + Amount int64 +} diff --git a/internal/domain/transfer.go b/internal/domain/transfer.go new file mode 100644 index 00000000..8733c4df --- /dev/null +++ b/internal/domain/transfer.go @@ -0,0 +1,70 @@ +package domain + +import "time" + +type TransferStatus string + +const ( + TransferStatusPending TransferStatus = "PENDING" + TransferStatusProcessed TransferStatus = "PROCESSED" + TransferStatusFailed TransferStatus = "FAILED" +) + +type FailureCode string + +const ( + FailureCodeInsufficientBalance FailureCode = "INSUFFICIENT_BALANCE" + FailureCodeWalletNotFound FailureCode = "WALLET_NOT_FOUND" +) + +type Transfer struct { + ID string + IdempotencyKey string + FromWalletID string + ToWalletID string + Amount int64 + Status TransferStatus + FailureCode FailureCode + CreatedAt time.Time + UpdatedAt time.Time +} + +func NewPendingTransfer( + id string, + idempotencyKey string, + fromWalletID string, + toWalletID string, + amount int64, + now time.Time, +) *Transfer { + return &Transfer{ + ID: id, + IdempotencyKey: idempotencyKey, + FromWalletID: fromWalletID, + ToWalletID: toWalletID, + Amount: amount, + Status: TransferStatusPending, + CreatedAt: now, + UpdatedAt: now, + } +} + +func (t *Transfer) MarkProcessed(now time.Time) error { + if t.Status != TransferStatusPending { + return ErrInvalidTransition + } + t.Status = TransferStatusProcessed + t.FailureCode = "" + t.UpdatedAt = now + return nil +} + +func (t *Transfer) MarkFailed(code FailureCode, now time.Time) error { + if t.Status != TransferStatusPending || code == "" { + return ErrInvalidTransition + } + t.Status = TransferStatusFailed + t.FailureCode = code + t.UpdatedAt = now + return nil +} diff --git a/internal/domain/transfer_test.go b/internal/domain/transfer_test.go new file mode 100644 index 00000000..73ee1f5c --- /dev/null +++ b/internal/domain/transfer_test.go @@ -0,0 +1,49 @@ +package domain + +import ( + "errors" + "testing" + "time" +) + +func TestTransferStateTransitions(t *testing.T) { + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + transfer := NewPendingTransfer("transfer-1", "key-1", "wallet-1", "wallet-2", 100, now) + + if err := transfer.MarkProcessed(now.Add(time.Second)); err != nil { + t.Fatalf("mark processed: %v", err) + } + if transfer.Status != TransferStatusProcessed { + t.Fatalf("status = %q, want %q", transfer.Status, TransferStatusProcessed) + } + if err := transfer.MarkFailed(FailureCodeInsufficientBalance, now.Add(2*time.Second)); !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("second terminal transition error = %v, want ErrInvalidTransition", err) + } +} + +func TestFailedTransferCannotBeProcessed(t *testing.T) { + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + transfer := NewPendingTransfer("transfer-1", "key-1", "wallet-1", "wallet-2", 100, now) + + if err := transfer.MarkFailed(FailureCodeInsufficientBalance, now.Add(time.Second)); err != nil { + t.Fatalf("mark failed: %v", err) + } + if err := transfer.MarkProcessed(now.Add(2 * time.Second)); !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("second terminal transition error = %v, want ErrInvalidTransition", err) + } +} + +func TestTransferCannotFailWithoutReason(t *testing.T) { + transfer := NewPendingTransfer( + "transfer-1", + "key-1", + "wallet-1", + "wallet-2", + 100, + time.Now(), + ) + + if err := transfer.MarkFailed("", time.Now()); !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("error = %v, want ErrInvalidTransition", err) + } +} diff --git a/internal/domain/wallet.go b/internal/domain/wallet.go new file mode 100644 index 00000000..25a9ae9d --- /dev/null +++ b/internal/domain/wallet.go @@ -0,0 +1,7 @@ +package domain + +// Wallet stores its balance in the currency's smallest unit. +type Wallet struct { + ID string + Balance int64 +} diff --git a/internal/handler/http/handler.go b/internal/handler/http/handler.go new file mode 100644 index 00000000..c793d838 --- /dev/null +++ b/internal/handler/http/handler.go @@ -0,0 +1,170 @@ +package httpapi + +import ( + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "time" + + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/service" +) + +const maxRequestBodyBytes = 1 << 20 + +type TransferUseCase interface { + CreateTransfer(context.Context, service.TransferCommand) (*service.TransferResult, error) +} + +type Handler struct { + transfers TransferUseCase + logger *slog.Logger +} + +func NewHandler(transfers TransferUseCase, logger *slog.Logger) *Handler { + return &Handler{transfers: transfers, logger: logger} +} + +func (h *Handler) Routes() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("POST /transfers", h.createTransfer) + mux.HandleFunc("GET /healthz", h.health) + return h.logging(mux) +} + +type createTransferRequest struct { + IdempotencyKey string `json:"idempotencyKey"` + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` +} + +type transferResponse struct { + TransferID string `json:"transferId"` + IdempotencyKey string `json:"idempotencyKey"` + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` + Status domain.TransferStatus `json:"status"` + FailureCode domain.FailureCode `json:"failureCode,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type errorResponse struct { + Code string `json:"code"` + Message string `json:"message"` +} + +func (h *Handler) createTransfer(writer http.ResponseWriter, request *http.Request) { + request.Body = http.MaxBytesReader(writer, request.Body, maxRequestBodyBytes) + decoder := json.NewDecoder(request.Body) + decoder.DisallowUnknownFields() + + var payload createTransferRequest + if err := decoder.Decode(&payload); err != nil { + writeJSON(writer, http.StatusBadRequest, errorResponse{ + Code: "INVALID_JSON", Message: "request body must be valid JSON", + }) + return + } + if err := ensureSingleJSONValue(decoder); err != nil { + writeJSON(writer, http.StatusBadRequest, errorResponse{ + Code: "INVALID_JSON", Message: "request body must contain one JSON object", + }) + return + } + + result, err := h.transfers.CreateTransfer(request.Context(), service.TransferCommand{ + IdempotencyKey: payload.IdempotencyKey, + FromWalletID: payload.FromWalletID, + ToWalletID: payload.ToWalletID, + Amount: payload.Amount, + }) + if err != nil { + h.writeServiceError(writer, err) + return + } + + statusCode := http.StatusCreated + switch result.Transfer.FailureCode { + case domain.FailureCodeInsufficientBalance: + statusCode = http.StatusUnprocessableEntity + case domain.FailureCodeWalletNotFound: + statusCode = http.StatusNotFound + } + + transfer := result.Transfer + writeJSON(writer, statusCode, transferResponse{ + TransferID: transfer.ID, + IdempotencyKey: transfer.IdempotencyKey, + FromWalletID: transfer.FromWalletID, + ToWalletID: transfer.ToWalletID, + Amount: transfer.Amount, + Status: transfer.Status, + FailureCode: transfer.FailureCode, + CreatedAt: transfer.CreatedAt, + UpdatedAt: transfer.UpdatedAt, + }) +} + +func (h *Handler) writeServiceError(writer http.ResponseWriter, err error) { + switch { + case errors.Is(err, domain.ErrIdempotencyConflict): + writeJSON(writer, http.StatusConflict, errorResponse{ + Code: "IDEMPOTENCY_CONFLICT", + Message: domain.ErrIdempotencyConflict.Error(), + }) + case errors.Is(err, domain.ErrInvalidIdempotencyKey), + errors.Is(err, domain.ErrInvalidWalletID), + errors.Is(err, domain.ErrSameWallet), + errors.Is(err, domain.ErrInvalidAmount): + writeJSON(writer, http.StatusBadRequest, errorResponse{ + Code: "VALIDATION_ERROR", + Message: err.Error(), + }) + default: + h.logger.Error("create transfer failed", "error", err) + writeJSON(writer, http.StatusInternalServerError, errorResponse{ + Code: "INTERNAL_ERROR", + Message: "an internal error occurred", + }) + } +} + +func (h *Handler) health(writer http.ResponseWriter, _ *http.Request) { + writeJSON(writer, http.StatusOK, map[string]string{"status": "ok"}) +} + +func (h *Handler) logging(next http.Handler) http.Handler { + return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + started := time.Now() + next.ServeHTTP(writer, request) + h.logger.Info("request completed", + "method", request.Method, + "path", request.URL.Path, + "duration", time.Since(started), + ) + }) +} + +func ensureSingleJSONValue(decoder *json.Decoder) error { + var extra any + err := decoder.Decode(&extra) + if errors.Is(err, io.EOF) { + return nil + } + if err == nil { + return errors.New("multiple JSON values") + } + return err +} + +func writeJSON(writer http.ResponseWriter, statusCode int, payload any) { + writer.Header().Set("Content-Type", "application/json") + writer.WriteHeader(statusCode) + _ = json.NewEncoder(writer).Encode(payload) +} diff --git a/internal/handler/http/handler_test.go b/internal/handler/http/handler_test.go new file mode 100644 index 00000000..7b598239 --- /dev/null +++ b/internal/handler/http/handler_test.go @@ -0,0 +1,107 @@ +package httpapi + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/service" +) + +func TestCreateTransferHandler(t *testing.T) { + useCase := &stubTransferUseCase{ + result: &service.TransferResult{Transfer: &domain.Transfer{ + ID: "transfer-1", + IdempotencyKey: "key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + Status: domain.TransferStatusProcessed, + CreatedAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), + UpdatedAt: time.Date(2026, 1, 1, 0, 0, 1, 0, time.UTC), + }}, + } + handler := NewHandler(useCase, discardLogger()).Routes() + body := bytes.NewBufferString(`{ + "idempotencyKey":"key-1", + "fromWalletId":"wallet-1", + "toWalletId":"wallet-2", + "amount":100 + }`) + request := httptest.NewRequest(http.MethodPost, "/transfers", body) + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusCreated { + t.Fatalf("status = %d, want %d; body: %s", recorder.Code, http.StatusCreated, recorder.Body.String()) + } + var response transferResponse + if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { + t.Fatalf("decode response: %v", err) + } + if response.TransferID != "transfer-1" || response.Status != domain.TransferStatusProcessed { + t.Fatalf("unexpected response: %#v", response) + } +} + +func TestCreateTransferHandlerMapsIdempotencyConflict(t *testing.T) { + useCase := &stubTransferUseCase{err: domain.ErrIdempotencyConflict} + handler := NewHandler(useCase, discardLogger()).Routes() + request := httptest.NewRequest( + http.MethodPost, + "/transfers", + bytes.NewBufferString(`{ + "idempotencyKey":"key-1", + "fromWalletId":"wallet-1", + "toWalletId":"wallet-2", + "amount":100 + }`), + ) + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusConflict { + t.Fatalf("status = %d, want %d", recorder.Code, http.StatusConflict) + } +} + +func TestCreateTransferHandlerRejectsUnknownFields(t *testing.T) { + handler := NewHandler(&stubTransferUseCase{}, discardLogger()).Routes() + request := httptest.NewRequest( + http.MethodPost, + "/transfers", + bytes.NewBufferString(`{"idempotencyKey":"key-1","unexpected":true}`), + ) + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d", recorder.Code, http.StatusBadRequest) + } +} + +type stubTransferUseCase struct { + result *service.TransferResult + err error +} + +func (s *stubTransferUseCase) CreateTransfer( + context.Context, + service.TransferCommand, +) (*service.TransferResult, error) { + return s.result, s.err +} + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} diff --git a/internal/platform/config/config.go b/internal/platform/config/config.go new file mode 100644 index 00000000..a8e1d913 --- /dev/null +++ b/internal/platform/config/config.go @@ -0,0 +1,51 @@ +package config + +import ( + "fmt" + "os" + "strconv" + "time" +) + +type Config struct { + HTTPAddress string + DatabaseURL string + ShutdownTimeout time.Duration + ReadTimeout time.Duration + WriteTimeout time.Duration + IdleTimeout time.Duration +} + +func Load() (Config, error) { + config := Config{ + HTTPAddress: envOrDefault("HTTP_ADDRESS", ":8080"), + DatabaseURL: os.Getenv("DATABASE_URL"), + ShutdownTimeout: durationEnvOrDefault("SHUTDOWN_TIMEOUT_SECONDS", 10*time.Second), + ReadTimeout: durationEnvOrDefault("HTTP_READ_TIMEOUT_SECONDS", 5*time.Second), + WriteTimeout: durationEnvOrDefault("HTTP_WRITE_TIMEOUT_SECONDS", 10*time.Second), + IdleTimeout: durationEnvOrDefault("HTTP_IDLE_TIMEOUT_SECONDS", 60*time.Second), + } + if config.DatabaseURL == "" { + return Config{}, fmt.Errorf("DATABASE_URL is required") + } + return config, nil +} + +func envOrDefault(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + +func durationEnvOrDefault(name string, fallback time.Duration) time.Duration { + raw := os.Getenv(name) + if raw == "" { + return fallback + } + seconds, err := strconv.Atoi(raw) + if err != nil || seconds <= 0 { + return fallback + } + return time.Duration(seconds) * time.Second +} diff --git a/internal/platform/database/postgres.go b/internal/platform/database/postgres.go new file mode 100644 index 00000000..446d741e --- /dev/null +++ b/internal/platform/database/postgres.go @@ -0,0 +1,24 @@ +package database + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func OpenPostgres(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) { + config, err := pgxpool.ParseConfig(databaseURL) + if err != nil { + return nil, fmt.Errorf("parse database configuration: %w", err) + } + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + return nil, fmt.Errorf("create database pool: %w", err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, fmt.Errorf("ping database: %w", err) + } + return pool, nil +} diff --git a/internal/platform/logging/logging.go b/internal/platform/logging/logging.go new file mode 100644 index 00000000..be782417 --- /dev/null +++ b/internal/platform/logging/logging.go @@ -0,0 +1,12 @@ +package logging + +import ( + "log/slog" + "os" +) + +func New() *slog.Logger { + return slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ + Level: slog.LevelInfo, + })) +} diff --git a/internal/repository/postgres/postgres.go b/internal/repository/postgres/postgres.go new file mode 100644 index 00000000..0ed5e67b --- /dev/null +++ b/internal/repository/postgres/postgres.go @@ -0,0 +1,44 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/repository" +) + +type UnitOfWork struct { + pool *pgxpool.Pool +} + +func NewUnitOfWork(pool *pgxpool.Pool) *UnitOfWork { + return &UnitOfWork{pool: pool} +} + +func (u *UnitOfWork) WithinTransaction( + ctx context.Context, + fn func(repository.Transaction) error, +) error { + tx, err := u.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted}) + if err != nil { + return fmt.Errorf("begin transaction: %w", err) + } + + committed := false + defer func() { + if !committed { + _ = tx.Rollback(context.Background()) + } + }() + + if err := fn(&transaction{tx: tx}); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit transaction: %w", err) + } + committed = true + return nil +} diff --git a/internal/repository/postgres/transaction.go b/internal/repository/postgres/transaction.go new file mode 100644 index 00000000..dbd73279 --- /dev/null +++ b/internal/repository/postgres/transaction.go @@ -0,0 +1,229 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" +) + +type transaction struct { + tx pgx.Tx +} + +func (t *transaction) TryCreateIdempotency( + ctx context.Context, + key string, + requestHash string, +) (bool, error) { + tag, err := t.tx.Exec(ctx, ` + INSERT INTO idempotency_records (idempotency_key, request_hash) + VALUES ($1, $2) + ON CONFLICT (idempotency_key) DO NOTHING + `, key, requestHash) + if err != nil { + return false, err + } + return tag.RowsAffected() == 1, nil +} + +func (t *transaction) FindTransferByIdempotencyKey( + ctx context.Context, + key string, +) (*domain.IdempotencyRecord, *domain.Transfer, error) { + const query = ` + SELECT + i.idempotency_key, + i.request_hash, + i.transfer_id, + t.id, + t.idempotency_key, + t.from_wallet_id, + t.to_wallet_id, + t.amount, + t.status, + COALESCE(t.failure_code, ''), + t.created_at, + t.updated_at + FROM idempotency_records i + JOIN transfers t ON t.id = i.transfer_id + WHERE i.idempotency_key = $1 + ` + + record := &domain.IdempotencyRecord{} + transfer := &domain.Transfer{} + err := t.tx.QueryRow(ctx, query, key).Scan( + &record.Key, + &record.RequestHash, + &record.TransferID, + &transfer.ID, + &transfer.IdempotencyKey, + &transfer.FromWalletID, + &transfer.ToWalletID, + &transfer.Amount, + &transfer.Status, + &transfer.FailureCode, + &transfer.CreatedAt, + &transfer.UpdatedAt, + ) + if err != nil { + return nil, nil, err + } + return record, transfer, nil +} + +func (t *transaction) LinkIdempotencyToTransfer( + ctx context.Context, + key string, + transferID string, +) error { + tag, err := t.tx.Exec(ctx, ` + UPDATE idempotency_records + SET transfer_id = $2 + WHERE idempotency_key = $1 AND transfer_id IS NULL + `, key, transferID) + if err != nil { + return err + } + if tag.RowsAffected() != 1 { + return fmt.Errorf("idempotency record was not linked") + } + return nil +} + +func (t *transaction) CompleteIdempotency(ctx context.Context, key string) error { + tag, err := t.tx.Exec(ctx, ` + UPDATE idempotency_records + SET completed_at = now() + WHERE idempotency_key = $1 + AND transfer_id IS NOT NULL + AND completed_at IS NULL + `, key) + if err != nil { + return err + } + if tag.RowsAffected() != 1 { + return fmt.Errorf("idempotency record was not completed") + } + return nil +} + +func (t *transaction) CreateTransfer(ctx context.Context, transfer *domain.Transfer) error { + _, err := t.tx.Exec(ctx, ` + INSERT INTO transfers ( + id, + idempotency_key, + from_wallet_id, + to_wallet_id, + amount, + status, + failure_code, + created_at, + updated_at + ) + VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), $8, $9) + `, + transfer.ID, + transfer.IdempotencyKey, + transfer.FromWalletID, + transfer.ToWalletID, + transfer.Amount, + transfer.Status, + transfer.FailureCode, + transfer.CreatedAt, + transfer.UpdatedAt, + ) + return err +} + +func (t *transaction) UpdateTransfer(ctx context.Context, transfer *domain.Transfer) error { + tag, err := t.tx.Exec(ctx, ` + UPDATE transfers + SET status = $2, failure_code = NULLIF($3, ''), updated_at = $4 + WHERE id = $1 AND status = 'PENDING' + `, + transfer.ID, + transfer.Status, + transfer.FailureCode, + transfer.UpdatedAt, + ) + if err != nil { + return err + } + if tag.RowsAffected() != 1 { + return domain.ErrInvalidTransition + } + return nil +} + +func (t *transaction) LockWallets( + ctx context.Context, + walletIDs []string, +) ([]domain.Wallet, error) { + rows, err := t.tx.Query(ctx, ` + SELECT id, balance + FROM wallets + WHERE id = ANY($1::text[]) + ORDER BY id + FOR UPDATE + `, walletIDs) + if err != nil { + return nil, err + } + defer rows.Close() + + wallets := make([]domain.Wallet, 0, len(walletIDs)) + for rows.Next() { + var wallet domain.Wallet + if err := rows.Scan(&wallet.ID, &wallet.Balance); err != nil { + return nil, err + } + wallets = append(wallets, wallet) + } + if err := rows.Err(); err != nil { + return nil, err + } + return wallets, nil +} + +func (t *transaction) UpdateWalletBalance( + ctx context.Context, + walletID string, + balance int64, +) error { + tag, err := t.tx.Exec(ctx, ` + UPDATE wallets + SET balance = $2, updated_at = now() + WHERE id = $1 + `, walletID, balance) + if err != nil { + return err + } + if tag.RowsAffected() != 1 { + return fmt.Errorf("wallet %q was not updated", walletID) + } + return nil +} + +func (t *transaction) CreateLedgerEntries( + ctx context.Context, + entries []domain.LedgerEntry, +) error { + batch := &pgx.Batch{} + for _, entry := range entries { + batch.Queue(` + INSERT INTO ledger_entries (id, wallet_id, transfer_id, entry_type, amount) + VALUES ($1, $2, $3, $4, $5) + `, entry.ID, entry.WalletID, entry.TransferID, entry.Type, entry.Amount) + } + + results := t.tx.SendBatch(ctx, batch) + for range entries { + if _, err := results.Exec(); err != nil { + _ = results.Close() + return err + } + } + return results.Close() +} diff --git a/internal/repository/repository.go b/internal/repository/repository.go new file mode 100644 index 00000000..73f86674 --- /dev/null +++ b/internal/repository/repository.go @@ -0,0 +1,28 @@ +package repository + +import ( + "context" + + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" +) + +// UnitOfWork executes fn in one database transaction. Returning an error rolls +// the transaction back; returning nil commits it. +type UnitOfWork interface { + WithinTransaction(ctx context.Context, fn func(Transaction) error) error +} + +type Transaction interface { + TryCreateIdempotency(ctx context.Context, key, requestHash string) (bool, error) + FindTransferByIdempotencyKey(ctx context.Context, key string) (*domain.IdempotencyRecord, *domain.Transfer, error) + LinkIdempotencyToTransfer(ctx context.Context, key, transferID string) error + CompleteIdempotency(ctx context.Context, key string) error + + CreateTransfer(ctx context.Context, transfer *domain.Transfer) error + UpdateTransfer(ctx context.Context, transfer *domain.Transfer) error + + LockWallets(ctx context.Context, walletIDs []string) ([]domain.Wallet, error) + UpdateWalletBalance(ctx context.Context, walletID string, balance int64) error + + CreateLedgerEntries(ctx context.Context, entries []domain.LedgerEntry) error +} diff --git a/internal/service/transfer.go b/internal/service/transfer.go new file mode 100644 index 00000000..6cf30871 --- /dev/null +++ b/internal/service/transfer.go @@ -0,0 +1,237 @@ +package service + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "time" + + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/repository" +) + +type TransferCommand struct { + IdempotencyKey string + FromWalletID string + ToWalletID string + Amount int64 +} + +type TransferResult struct { + Transfer *domain.Transfer + Replayed bool +} + +type IDGenerator func() (string, error) +type Clock func() time.Time + +type TransferService struct { + unitOfWork repository.UnitOfWork + newID IDGenerator + now Clock +} + +func NewTransferService(unitOfWork repository.UnitOfWork) *TransferService { + return NewTransferServiceWithDependencies(unitOfWork, newUUID, time.Now) +} + +func NewTransferServiceWithDependencies( + unitOfWork repository.UnitOfWork, + newID IDGenerator, + now Clock, +) *TransferService { + return &TransferService{unitOfWork: unitOfWork, newID: newID, now: now} +} + +func (s *TransferService) CreateTransfer(ctx context.Context, command TransferCommand) (*TransferResult, error) { + if err := validateCommand(command); err != nil { + return nil, err + } + + requestHash := hashCommand(command) + var result *TransferResult + + err := s.unitOfWork.WithinTransaction(ctx, func(tx repository.Transaction) error { + claimed, err := tx.TryCreateIdempotency(ctx, command.IdempotencyKey, requestHash) + if err != nil { + return fmt.Errorf("claim idempotency key: %w", err) + } + if !claimed { + record, transfer, err := tx.FindTransferByIdempotencyKey(ctx, command.IdempotencyKey) + if err != nil { + return fmt.Errorf("load idempotent result: %w", err) + } + if record.RequestHash != requestHash { + return domain.ErrIdempotencyConflict + } + result = &TransferResult{Transfer: transfer, Replayed: true} + return nil + } + + transferID, err := s.newID() + if err != nil { + return fmt.Errorf("generate transfer ID: %w", err) + } + now := s.timestamp() + transfer := domain.NewPendingTransfer( + transferID, + command.IdempotencyKey, + command.FromWalletID, + command.ToWalletID, + command.Amount, + now, + ) + if err := tx.CreateTransfer(ctx, transfer); err != nil { + return fmt.Errorf("create pending transfer: %w", err) + } + if err := tx.LinkIdempotencyToTransfer(ctx, command.IdempotencyKey, transfer.ID); err != nil { + return fmt.Errorf("link idempotency record: %w", err) + } + + wallets, err := tx.LockWallets(ctx, []string{command.FromWalletID, command.ToWalletID}) + if err != nil { + return fmt.Errorf("lock wallets: %w", err) + } + if len(wallets) != 2 { + if err := transfer.MarkFailed(domain.FailureCodeWalletNotFound, s.timestamp()); err != nil { + return err + } + if err := tx.UpdateTransfer(ctx, transfer); err != nil { + return fmt.Errorf("record missing wallet failure: %w", err) + } + if err := tx.CompleteIdempotency(ctx, command.IdempotencyKey); err != nil { + return fmt.Errorf("complete idempotency record: %w", err) + } + result = &TransferResult{Transfer: transfer} + return nil + } + + walletByID := make(map[string]domain.Wallet, len(wallets)) + for _, wallet := range wallets { + walletByID[wallet.ID] = wallet + } + source := walletByID[command.FromWalletID] + destination := walletByID[command.ToWalletID] + + if source.Balance < command.Amount { + if err := transfer.MarkFailed(domain.FailureCodeInsufficientBalance, s.timestamp()); err != nil { + return err + } + if err := tx.UpdateTransfer(ctx, transfer); err != nil { + return fmt.Errorf("record insufficient balance failure: %w", err) + } + if err := tx.CompleteIdempotency(ctx, command.IdempotencyKey); err != nil { + return fmt.Errorf("complete idempotency record: %w", err) + } + result = &TransferResult{Transfer: transfer} + return nil + } + + if err := tx.UpdateWalletBalance(ctx, source.ID, source.Balance-command.Amount); err != nil { + return fmt.Errorf("debit source wallet: %w", err) + } + if err := tx.UpdateWalletBalance(ctx, destination.ID, destination.Balance+command.Amount); err != nil { + return fmt.Errorf("credit destination wallet: %w", err) + } + + debitID, err := s.newID() + if err != nil { + return fmt.Errorf("generate debit ledger ID: %w", err) + } + creditID, err := s.newID() + if err != nil { + return fmt.Errorf("generate credit ledger ID: %w", err) + } + entries := []domain.LedgerEntry{ + { + ID: debitID, + WalletID: source.ID, + TransferID: transfer.ID, + Type: domain.LedgerEntryDebit, + Amount: command.Amount, + }, + { + ID: creditID, + WalletID: destination.ID, + TransferID: transfer.ID, + Type: domain.LedgerEntryCredit, + Amount: command.Amount, + }, + } + if err := tx.CreateLedgerEntries(ctx, entries); err != nil { + return fmt.Errorf("create ledger entries: %w", err) + } + if err := transfer.MarkProcessed(s.timestamp()); err != nil { + return err + } + if err := tx.UpdateTransfer(ctx, transfer); err != nil { + return fmt.Errorf("mark transfer processed: %w", err) + } + if err := tx.CompleteIdempotency(ctx, command.IdempotencyKey); err != nil { + return fmt.Errorf("complete idempotency record: %w", err) + } + result = &TransferResult{Transfer: transfer} + return nil + }) + if err != nil { + return nil, err + } + return result, nil +} + +func (s *TransferService) timestamp() time.Time { + // PostgreSQL stores timestamps at microsecond precision. Normalizing before + // the first response keeps replayed timestamps byte-for-byte stable. + return s.now().UTC().Truncate(time.Microsecond) +} + +func validateCommand(command TransferCommand) error { + if strings.TrimSpace(command.IdempotencyKey) == "" || len(command.IdempotencyKey) > 255 { + return domain.ErrInvalidIdempotencyKey + } + if strings.TrimSpace(command.FromWalletID) == "" || + strings.TrimSpace(command.ToWalletID) == "" || + len(command.FromWalletID) > 255 || + len(command.ToWalletID) > 255 { + return domain.ErrInvalidWalletID + } + if command.FromWalletID == command.ToWalletID { + return domain.ErrSameWallet + } + if command.Amount <= 0 { + return domain.ErrInvalidAmount + } + return nil +} + +func hashCommand(command TransferCommand) string { + payload := fmt.Sprintf("%d:%s|%d:%s|%d", + len(command.FromWalletID), + command.FromWalletID, + len(command.ToWalletID), + command.ToWalletID, + command.Amount, + ) + sum := sha256.Sum256([]byte(payload)) + return hex.EncodeToString(sum[:]) +} + +func newUUID() (string, error) { + var value [16]byte + if _, err := rand.Read(value[:]); err != nil { + return "", err + } + value[6] = (value[6] & 0x0f) | 0x40 + value[8] = (value[8] & 0x3f) | 0x80 + return fmt.Sprintf( + "%08x-%04x-%04x-%04x-%012x", + value[0:4], + value[4:6], + value[6:8], + value[8:10], + value[10:16], + ), nil +} diff --git a/internal/service/transfer_test.go b/internal/service/transfer_test.go new file mode 100644 index 00000000..658005ac --- /dev/null +++ b/internal/service/transfer_test.go @@ -0,0 +1,285 @@ +package service + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/repository" +) + +func TestCreateTransferProcessesBalancesAndLedger(t *testing.T) { + tx := newFakeTransaction(true) + tx.wallets["wallet-1"] = domain.Wallet{ID: "wallet-1", Balance: 200} + tx.wallets["wallet-2"] = domain.Wallet{ID: "wallet-2", Balance: 50} + service := testService(tx) + + result, err := service.CreateTransfer(context.Background(), TransferCommand{ + IdempotencyKey: "key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + }) + if err != nil { + t.Fatalf("create transfer: %v", err) + } + if result.Transfer.Status != domain.TransferStatusProcessed { + t.Fatalf("status = %q, want PROCESSED", result.Transfer.Status) + } + if got := tx.wallets["wallet-1"].Balance; got != 100 { + t.Fatalf("source balance = %d, want 100", got) + } + if got := tx.wallets["wallet-2"].Balance; got != 150 { + t.Fatalf("destination balance = %d, want 150", got) + } + if len(tx.entries) != 2 { + t.Fatalf("ledger entries = %d, want 2", len(tx.entries)) + } + if tx.entries[0].Type != domain.LedgerEntryDebit || tx.entries[1].Type != domain.LedgerEntryCredit { + t.Fatalf("unexpected ledger entries: %#v", tx.entries) + } +} + +func TestCreateTransferInsufficientBalanceIsTerminalFailure(t *testing.T) { + tx := newFakeTransaction(true) + tx.wallets["wallet-1"] = domain.Wallet{ID: "wallet-1", Balance: 20} + tx.wallets["wallet-2"] = domain.Wallet{ID: "wallet-2", Balance: 50} + + result, err := testService(tx).CreateTransfer(context.Background(), TransferCommand{ + IdempotencyKey: "key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + }) + if err != nil { + t.Fatalf("create transfer: %v", err) + } + if result.Transfer.Status != domain.TransferStatusFailed { + t.Fatalf("status = %q, want FAILED", result.Transfer.Status) + } + if result.Transfer.FailureCode != domain.FailureCodeInsufficientBalance { + t.Fatalf("failure code = %q, want INSUFFICIENT_BALANCE", result.Transfer.FailureCode) + } + if tx.wallets["wallet-1"].Balance != 20 || tx.wallets["wallet-2"].Balance != 50 { + t.Fatal("failed transfer changed wallet balances") + } + if len(tx.entries) != 0 { + t.Fatalf("failed transfer created %d ledger entries", len(tx.entries)) + } +} + +func TestCreateTransferMissingWalletIsTerminalFailure(t *testing.T) { + tx := newFakeTransaction(true) + tx.wallets["wallet-1"] = domain.Wallet{ID: "wallet-1", Balance: 200} + + result, err := testService(tx).CreateTransfer(context.Background(), TransferCommand{ + IdempotencyKey: "key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + }) + if err != nil { + t.Fatalf("create transfer: %v", err) + } + if result.Transfer.Status != domain.TransferStatusFailed { + t.Fatalf("status = %q, want FAILED", result.Transfer.Status) + } + if result.Transfer.FailureCode != domain.FailureCodeWalletNotFound { + t.Fatalf("failure code = %q, want WALLET_NOT_FOUND", result.Transfer.FailureCode) + } + if tx.wallets["wallet-1"].Balance != 200 { + t.Fatal("missing wallet failure changed wallet balance") + } + if len(tx.entries) != 0 { + t.Fatalf("missing wallet failure created %d ledger entries", len(tx.entries)) + } +} + +func TestCreateTransferReturnsOriginalResultForDuplicate(t *testing.T) { + tx := newFakeTransaction(false) + command := TransferCommand{ + IdempotencyKey: "key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + } + original := domain.NewPendingTransfer( + "original-transfer", + command.IdempotencyKey, + command.FromWalletID, + command.ToWalletID, + command.Amount, + time.Now(), + ) + if err := original.MarkProcessed(time.Now()); err != nil { + t.Fatal(err) + } + tx.record = &domain.IdempotencyRecord{ + Key: command.IdempotencyKey, + RequestHash: hashCommand(command), + TransferID: original.ID, + } + tx.existingTransfer = original + + result, err := testService(tx).CreateTransfer(context.Background(), command) + if err != nil { + t.Fatalf("create duplicate transfer: %v", err) + } + if !result.Replayed { + t.Fatal("duplicate result was not marked replayed") + } + if result.Transfer.ID != original.ID { + t.Fatalf("transfer ID = %q, want %q", result.Transfer.ID, original.ID) + } + if len(tx.entries) != 0 { + t.Fatal("duplicate request created ledger entries") + } +} + +func TestCreateTransferRejectsIdempotencyKeyPayloadMismatch(t *testing.T) { + tx := newFakeTransaction(false) + tx.record = &domain.IdempotencyRecord{ + Key: "key-1", + RequestHash: "a-different-request-hash", + TransferID: "original-transfer", + } + tx.existingTransfer = domain.NewPendingTransfer( + "original-transfer", "key-1", "wallet-1", "wallet-2", 50, time.Now(), + ) + + _, err := testService(tx).CreateTransfer(context.Background(), TransferCommand{ + IdempotencyKey: "key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + }) + if !errors.Is(err, domain.ErrIdempotencyConflict) { + t.Fatalf("error = %v, want ErrIdempotencyConflict", err) + } +} + +func TestCreateTransferValidatesCommandBeforeOpeningTransaction(t *testing.T) { + unitOfWork := &fakeUnitOfWork{tx: newFakeTransaction(true)} + service := NewTransferServiceWithDependencies(unitOfWork, sequenceIDs(), fixedClock) + + _, err := service.CreateTransfer(context.Background(), TransferCommand{Amount: -1}) + if !errors.Is(err, domain.ErrInvalidIdempotencyKey) { + t.Fatalf("error = %v, want ErrInvalidIdempotencyKey", err) + } + if unitOfWork.called { + t.Fatal("transaction opened for an invalid command") + } +} + +func testService(tx *fakeTransaction) *TransferService { + return NewTransferServiceWithDependencies( + &fakeUnitOfWork{tx: tx}, + sequenceIDs(), + fixedClock, + ) +} + +func sequenceIDs() IDGenerator { + next := 0 + return func() (string, error) { + next++ + return fmt.Sprintf("id-%d", next), nil + } +} + +func fixedClock() time.Time { + return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) +} + +type fakeUnitOfWork struct { + tx *fakeTransaction + called bool +} + +func (u *fakeUnitOfWork) WithinTransaction( + _ context.Context, + fn func(repository.Transaction) error, +) error { + u.called = true + return fn(u.tx) +} + +type fakeTransaction struct { + claimResult bool + record *domain.IdempotencyRecord + existingTransfer *domain.Transfer + createdTransfer *domain.Transfer + wallets map[string]domain.Wallet + entries []domain.LedgerEntry +} + +func newFakeTransaction(claimResult bool) *fakeTransaction { + return &fakeTransaction{ + claimResult: claimResult, + wallets: make(map[string]domain.Wallet), + } +} + +func (t *fakeTransaction) TryCreateIdempotency(context.Context, string, string) (bool, error) { + return t.claimResult, nil +} + +func (t *fakeTransaction) FindTransferByIdempotencyKey( + context.Context, + string, +) (*domain.IdempotencyRecord, *domain.Transfer, error) { + return t.record, t.existingTransfer, nil +} + +func (t *fakeTransaction) LinkIdempotencyToTransfer(context.Context, string, string) error { + return nil +} + +func (t *fakeTransaction) CompleteIdempotency(context.Context, string) error { + return nil +} + +func (t *fakeTransaction) CreateTransfer(_ context.Context, transfer *domain.Transfer) error { + t.createdTransfer = transfer + return nil +} + +func (t *fakeTransaction) UpdateTransfer(_ context.Context, transfer *domain.Transfer) error { + t.createdTransfer = transfer + return nil +} + +func (t *fakeTransaction) LockWallets( + _ context.Context, + walletIDs []string, +) ([]domain.Wallet, error) { + wallets := make([]domain.Wallet, 0, len(walletIDs)) + for _, id := range walletIDs { + if wallet, ok := t.wallets[id]; ok { + wallets = append(wallets, wallet) + } + } + return wallets, nil +} + +func (t *fakeTransaction) UpdateWalletBalance( + _ context.Context, + walletID string, + balance int64, +) error { + wallet := t.wallets[walletID] + wallet.Balance = balance + t.wallets[walletID] = wallet + return nil +} + +func (t *fakeTransaction) CreateLedgerEntries( + _ context.Context, + entries []domain.LedgerEntry, +) error { + t.entries = append(t.entries, entries...) + return nil +} diff --git a/migrations/001_init.sql b/migrations/001_init.sql new file mode 100644 index 00000000..c3be4ed4 --- /dev/null +++ b/migrations/001_init.sql @@ -0,0 +1,50 @@ +CREATE TABLE wallets ( + id TEXT PRIMARY KEY, + balance BIGINT NOT NULL DEFAULT 0 CHECK (balance >= 0), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE transfers ( + id TEXT PRIMARY KEY, + idempotency_key TEXT NOT NULL UNIQUE, + from_wallet_id TEXT NOT NULL, + to_wallet_id TEXT NOT NULL, + amount BIGINT NOT NULL CHECK (amount > 0), + status TEXT NOT NULL CHECK (status IN ('PENDING', 'PROCESSED', 'FAILED')), + failure_code TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT different_wallets CHECK (from_wallet_id <> to_wallet_id), + CONSTRAINT valid_failure_state CHECK ( + (status = 'FAILED' AND failure_code IS NOT NULL) + OR (status <> 'FAILED' AND failure_code IS NULL) + ) +); + +CREATE TABLE idempotency_records ( + idempotency_key TEXT PRIMARY KEY, + request_hash CHAR(64) NOT NULL, + transfer_id TEXT UNIQUE REFERENCES transfers(id) ON DELETE RESTRICT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ +); + +CREATE TABLE ledger_entries ( + id TEXT PRIMARY KEY, + wallet_id TEXT NOT NULL REFERENCES wallets(id) ON DELETE RESTRICT, + transfer_id TEXT NOT NULL REFERENCES transfers(id) ON DELETE RESTRICT, + entry_type TEXT NOT NULL CHECK (entry_type IN ('DEBIT', 'CREDIT')), + amount BIGINT NOT NULL CHECK (amount > 0), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT one_entry_of_each_type_per_transfer UNIQUE (transfer_id, entry_type) +); + +CREATE INDEX idx_transfers_from_wallet_created_at + ON transfers (from_wallet_id, created_at DESC); + +CREATE INDEX idx_transfers_to_wallet_created_at + ON transfers (to_wallet_id, created_at DESC); + +CREATE INDEX idx_ledger_entries_wallet_created_at + ON ledger_entries (wallet_id, created_at DESC); diff --git a/tests/integration/transfer_test.go b/tests/integration/transfer_test.go new file mode 100644 index 00000000..4af1361a --- /dev/null +++ b/tests/integration/transfer_test.go @@ -0,0 +1,290 @@ +package integration + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/domain" + appPostgres "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/repository/postgres" + "github.com/sumiyapuletipalli/wallet-transfer-assignment/internal/service" +) + +func TestPostgresTransferIdempotencyAndLedger(t *testing.T) { + pool := integrationPool(t) + resetDatabase(t, pool) + seedWallets(t, pool, map[string]int64{"wallet-1": 200, "wallet-2": 50}) + transferService := service.NewTransferService(appPostgres.NewUnitOfWork(pool)) + command := service.TransferCommand{ + IdempotencyKey: "integration-key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + } + + first, err := transferService.CreateTransfer(context.Background(), command) + if err != nil { + t.Fatalf("first transfer: %v", err) + } + duplicate, err := transferService.CreateTransfer(context.Background(), command) + if err != nil { + t.Fatalf("duplicate transfer: %v", err) + } + + if first.Transfer.ID != duplicate.Transfer.ID || !duplicate.Replayed { + t.Fatalf("duplicate did not return original result: first=%#v duplicate=%#v", first, duplicate) + } + assertBalance(t, pool, "wallet-1", 100) + assertBalance(t, pool, "wallet-2", 150) + assertCount(t, pool, "SELECT count(*) FROM transfers", 1) + assertCount(t, pool, "SELECT count(*) FROM ledger_entries", 2) + assertLedgerEntry(t, pool, first.Transfer.ID, "wallet-1", "DEBIT", 100) + assertLedgerEntry(t, pool, first.Transfer.ID, "wallet-2", "CREDIT", 100) + + _, err = transferService.CreateTransfer(context.Background(), service.TransferCommand{ + IdempotencyKey: command.IdempotencyKey, + FromWalletID: command.FromWalletID, + ToWalletID: command.ToWalletID, + Amount: 101, + }) + if !errors.Is(err, domain.ErrIdempotencyConflict) { + t.Fatalf("payload mismatch error = %v, want ErrIdempotencyConflict", err) + } +} + +func TestPostgresFailedTransferIsIdempotentlyReplayed(t *testing.T) { + pool := integrationPool(t) + resetDatabase(t, pool) + seedWallets(t, pool, map[string]int64{"wallet-1": 20, "wallet-2": 50}) + transferService := service.NewTransferService(appPostgres.NewUnitOfWork(pool)) + command := service.TransferCommand{ + IdempotencyKey: "failed-key-1", + FromWalletID: "wallet-1", + ToWalletID: "wallet-2", + Amount: 100, + } + + first, err := transferService.CreateTransfer(context.Background(), command) + if err != nil { + t.Fatalf("first transfer: %v", err) + } + if first.Transfer.Status != domain.TransferStatusFailed { + t.Fatalf("first status = %q, want FAILED", first.Transfer.Status) + } + + if _, err := pool.Exec(context.Background(), "UPDATE wallets SET balance = 200 WHERE id = 'wallet-1'"); err != nil { + t.Fatalf("top up source wallet: %v", err) + } + duplicate, err := transferService.CreateTransfer(context.Background(), command) + if err != nil { + t.Fatalf("duplicate transfer: %v", err) + } + + if first.Transfer.ID != duplicate.Transfer.ID || !duplicate.Replayed { + t.Fatalf("duplicate did not return original failure: first=%#v duplicate=%#v", first, duplicate) + } + if duplicate.Transfer.Status != domain.TransferStatusFailed { + t.Fatalf("duplicate status = %q, want FAILED", duplicate.Transfer.Status) + } + assertBalance(t, pool, "wallet-1", 200) + assertBalance(t, pool, "wallet-2", 50) + assertCount(t, pool, "SELECT count(*) FROM transfers", 1) + assertCount(t, pool, "SELECT count(*) FROM ledger_entries", 0) +} + +func TestPostgresConcurrentDebitsDoNotOverspend(t *testing.T) { + pool := integrationPool(t) + resetDatabase(t, pool) + seedWallets(t, pool, map[string]int64{ + "source": 100, + "destination-1": 0, + "destination-2": 0, + }) + transferService := service.NewTransferService(appPostgres.NewUnitOfWork(pool)) + commands := []service.TransferCommand{ + { + IdempotencyKey: "concurrent-key-1", + FromWalletID: "source", + ToWalletID: "destination-1", + Amount: 80, + }, + { + IdempotencyKey: "concurrent-key-2", + FromWalletID: "source", + ToWalletID: "destination-2", + Amount: 80, + }, + } + + var waitGroup sync.WaitGroup + results := make(chan *service.TransferResult, len(commands)) + errs := make(chan error, len(commands)) + for _, command := range commands { + waitGroup.Add(1) + go func(command service.TransferCommand) { + defer waitGroup.Done() + result, err := transferService.CreateTransfer(context.Background(), command) + if err != nil { + errs <- err + return + } + results <- result + }(command) + } + waitGroup.Wait() + close(results) + close(errs) + + for err := range errs { + t.Errorf("concurrent transfer: %v", err) + } + processed, failed := 0, 0 + for result := range results { + switch result.Transfer.Status { + case domain.TransferStatusProcessed: + processed++ + case domain.TransferStatusFailed: + failed++ + } + } + if processed != 1 || failed != 1 { + t.Fatalf("processed=%d failed=%d, want 1 and 1", processed, failed) + } + assertBalance(t, pool, "source", 20) + assertCount(t, pool, "SELECT count(*) FROM ledger_entries", 2) + assertCount(t, pool, "SELECT count(*) FROM transfers", 2) +} + +func integrationPool(t *testing.T) *pgxpool.Pool { + t.Helper() + databaseURL := os.Getenv("INTEGRATION_DATABASE_URL") + if databaseURL == "" { + t.Skip("INTEGRATION_DATABASE_URL is not set") + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + pool, err := pgxpool.New(ctx, databaseURL) + if err != nil { + t.Fatalf("connect to PostgreSQL: %v", err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + t.Fatalf("ping PostgreSQL: %v", err) + } + t.Cleanup(pool.Close) + applyMigration(t, pool) + return pool +} + +func applyMigration(t *testing.T, pool *pgxpool.Pool) { + t.Helper() + var exists bool + if err := pool.QueryRow( + context.Background(), + "SELECT to_regclass('public.wallets') IS NOT NULL", + ).Scan(&exists); err != nil { + t.Fatalf("check schema: %v", err) + } + if exists { + return + } + path := filepath.Join("..", "..", "migrations", "001_init.sql") + migration, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read migration: %v", err) + } + if _, err := pool.Exec(context.Background(), string(migration)); err != nil { + t.Fatalf("apply migration: %v", err) + } +} + +func resetDatabase(t *testing.T, pool *pgxpool.Pool) { + t.Helper() + _, err := pool.Exec(context.Background(), ` + TRUNCATE TABLE ledger_entries, idempotency_records, transfers, wallets + `) + if err != nil { + t.Fatalf("reset database: %v", err) + } +} + +func seedWallets(t *testing.T, pool *pgxpool.Pool, wallets map[string]int64) { + t.Helper() + for id, balance := range wallets { + if _, err := pool.Exec( + context.Background(), + "INSERT INTO wallets (id, balance) VALUES ($1, $2)", + id, + balance, + ); err != nil { + t.Fatalf("seed wallet %q: %v", id, err) + } + } +} + +func assertBalance(t *testing.T, pool *pgxpool.Pool, walletID string, want int64) { + t.Helper() + var got int64 + if err := pool.QueryRow( + context.Background(), + "SELECT balance FROM wallets WHERE id = $1", + walletID, + ).Scan(&got); err != nil { + t.Fatalf("read balance for %q: %v", walletID, err) + } + if got != want { + t.Fatalf("balance for %q = %d, want %d", walletID, got, want) + } +} + +func assertCount(t *testing.T, pool *pgxpool.Pool, query string, want int) { + t.Helper() + var got int + if err := pool.QueryRow(context.Background(), query).Scan(&got); err != nil { + t.Fatalf("query count: %v", err) + } + if got != want { + t.Fatalf("count = %d, want %d", got, want) + } +} + +func assertLedgerEntry( + t *testing.T, + pool *pgxpool.Pool, + transferID string, + walletID string, + entryType string, + amount int64, +) { + t.Helper() + var got int + if err := pool.QueryRow( + context.Background(), + `SELECT count(*) + FROM ledger_entries + WHERE transfer_id = $1 + AND wallet_id = $2 + AND entry_type = $3 + AND amount = $4`, + transferID, + walletID, + entryType, + amount, + ).Scan(&got); err != nil { + t.Fatalf("query ledger entry: %v", err) + } + if got != 1 { + t.Fatalf("ledger entries for transfer=%q wallet=%q type=%q amount=%d = %d, want 1", + transferID, + walletID, + entryType, + amount, + got, + ) + } +}