diff --git a/README.md b/README.md index 58c62d1a..eaa7e57f 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,252 @@ -# Wallet Transfer Assignment Repository +# Wallet Transfer Service + +A small, reliable service for **wallet-to-wallet transfers** with idempotent request +handling, double-entry ledger recording, and safe concurrent execution. + +> The template/reviewer instructions for this repository are kept further down under +> [Repository Template Notes](#repository-template-notes). + +## What's implemented + +A single HTTP endpoint, `POST /transfers`, that moves money between two wallets +**atomically** and with **exactly-once semantics**. Every requirement from +[`ASSIGNMENT.md`](./ASSIGNMENT.md) is covered: + +| Requirement | How it's implemented | +|---------------------------------|------------------------------------------------------------------------------------------------------------| +| **Create transfer** | `POST /transfers` accepts `{idempotencyKey, fromWalletId, toWalletId, amount}` and returns the result. | +| **Idempotent requests** | The `idempotencyKey` is the PK of `idempotency_records`; the original response is stored and **replayed** on any retry — duplicates never trigger a second transfer. | +| **Wallet balances** | Stored per-wallet balance, updated **inside** the transfer transaction; `CHECK (balance >= 0)` prevents overdraft at the DB. | +| **Double-entry ledger** | Each transfer writes exactly one `DEBIT` + one `CREDIT` row, enforced by a **unique `(transfer_id, type)`** index, so the ledger can never be unbalanced. | +| **Transfer state machine** | `PENDING → PROCESSED` on success, `PENDING → FAILED` on insufficient funds; state is `CHECK`-constrained. | +| **Concurrency safety** | Whole transfer runs in one DB transaction; both wallets are locked with `SELECT ... FOR UPDATE` in a deterministic order to prevent double-spend and deadlocks. | +| **Persistence** | PostgreSQL via GORM; schema in [`migrations/schema.sql`](./migrations/schema.sql). | +| **Layered architecture** | Thin handler → service (business logic) → repository (persistence), with dependency inversion. | +| **Testing** | Unit tests (no DB) + real-Postgres concurrency integration tests proving no double-spend and a balanced ledger. | + +## How a transfer executes + +A single `POST /transfers` call runs the following steps inside **one** database +transaction (see [`internal/service/transfer_service.go`](./internal/service/transfer_service.go)): + +1. **Validate** the request (non-empty IDs/key, positive amount, distinct wallets). + Invalid input is rejected before any DB work. +2. **Replay check** — if the `idempotencyKey` already exists, return its stored + response verbatim and stop (no new side effects). +3. **Lock both wallets** with `SELECT ... FOR UPDATE`, always in sorted ID order, + so concurrent opposing transfers can't deadlock. +4. **Insert the transfer** row in `PENDING` state. +5. **Check funds under the lock.** If insufficient, mark the transfer `FAILED`, + store the response against the key, and return `422` (still replayable). +6. **Move the money** — debit the source, credit the destination. +7. **Write the ledger** — one `DEBIT` + one `CREDIT` row for the transfer. +8. **Mark `PROCESSED`** and **store the response snapshot** against the key. +9. **Commit.** If a concurrent duplicate committed first, this transaction rolls + back on the unique-key violation and the caller replays the winner's response. + +## Architecture + +Clean layered architecture with clear separation of concerns: + +``` +cmd/server process entrypoint, wiring, graceful shutdown +internal/api HTTP handlers: validation + transport mapping (thin) +internal/service business logic: transfer workflow, idempotency, concurrency +internal/repository persistence abstractions (interfaces + error sentinels) + └── postgres GORM/Postgres implementations +internal/model domain entities, states, and DB constraints +internal/config environment configuration +internal/logging leveled logging +migrations/schema.sql authoritative SQL schema +``` + +The `repository` package defines interfaces and the concrete `postgres` package +depends on them (dependency inversion), so the service layer never imports the ORM. + +## Database schema + +| table | purpose | +|-----------------------|----------------------------------------------------------------| +| `wallets` | stored balance per wallet, `CHECK (balance >= 0)` backstop | +| `transfers` | transfer workflow record with a `PENDING/PROCESSED/FAILED` state | +| `ledger_entries` | immutable double-entry rows (one DEBIT + one CREDIT per transfer) | +| `idempotency_records` | maps an idempotency key to its transfer **and stored response** | + +Integrity is enforced at the database, not just in application code: + +- `wallets.balance CHECK (balance >= 0)` — a hard backstop against double spend. +- `transfers.state CHECK (... IN ('PENDING','PROCESSED','FAILED'))` and + `CHECK (from_wallet_id <> to_wallet_id)`. +- `ledger_entries` **unique `(transfer_id, type)`** — guarantees exactly one debit + and one credit per transfer, so the ledger can never be unbalanced. +- Foreign keys tie ledger entries and idempotency records back to real transfers/wallets. +- Indexes on `ledger_entries(wallet_id)`, `transfers(from_wallet_id)`, + `transfers(to_wallet_id)`, and `transfers(state)`. + +Balances are **stored and updated inside the transfer transaction** (not derived on +read) for O(1) balance reads; the ledger remains the immutable source of truth for audit. + +## Idempotency strategy (exactly-once at the API level) + +- Each request carries an `idempotencyKey` (primary key of `idempotency_records`). +- On success **and** on terminal failure, we store a **snapshot of the response** + (JSON) alongside the key, inside the same transaction as the balance changes. +- A duplicate request returns the **exact original result** — including the balances + captured at execution time — even if the wallets have since changed. This answers + "what if the first request committed but the response was lost?": the retry replays + the stored result rather than re-executing. +- Concurrent duplicates race on the unique key. The loser's transaction rolls back + (no orphan transfer, no double side effects) and then **replays the winner's stored + response** instead of returning an error. + +## Concurrency handling + +- The whole transfer runs in a single database transaction. +- Both wallets are locked with `SELECT ... FOR UPDATE` in a **deterministic (sorted) + order** to prevent deadlocks between opposing transfers (`A→B` and `B→A`). +- Insufficient funds is checked under the lock and recorded as a terminal `FAILED` + transfer, so retries of the same key return the same outcome. +- Even if application logic had a bug, the `CHECK (balance >= 0)` constraint would + reject a double spend at the database. + +### Failure / retry semantics + +| scenario | behavior | +|---------------------------------------|------------------------------------------------------| +| duplicate request (same key) | returns original stored response, no new side effects | +| concurrent duplicate | one commits, others replay the same stored response | +| insufficient funds | `FAILED` transfer recorded; `422` returned; replayable | +| unknown wallet | `404`, no transfer created | +| self-transfer / invalid amount | `400`, no transfer created | + +## API + +The service exposes a single, focused endpoint. Everything else (balances, +history, health) is intentionally out of scope — the goal is a correct, +concurrency-safe transactional core, not feature breadth. + +``` +POST /transfers create a transfer (idempotent) +``` + +Create request: + +```json +{ "idempotencyKey": "abc123", "fromWalletId": "wallet_1", "toWalletId": "wallet_2", "amount": 100 } +``` + +Create response (`201`): + +```json +{ + "transferId": "…", "fromWalletId": "wallet_1", "toWalletId": "wallet_2", + "amount": 100, "state": "PROCESSED", "fromWalletBalance": 400, + "toWalletBalance": 200, "createdAt": "…" +} +``` + +## Running locally + +```bash +export DATABASE_URL="postgres://user:pass@localhost:5432/wallet?sslmode=disable" +# optional: LISTEN_ADDR (:8080), LOG_LEVEL (INFO), DB_MAX_OPEN_CONNS, DB_MAX_IDLE_CONNS +go run ./cmd/server +``` + +The service does not create its own schema. `migrations/schema.sql` is the +authoritative DDL and is applied when the database is provisioned (locally, the +Postgres container runs it via `/docker-entrypoint-initdb.d/` on first init). The +service assumes the schema already exists. + +## Running against real Postgres + +`testenv/` spins up a real Postgres (schema applied automatically via the image's +init scripts), seeds two wallets, and starts the service on `:8080`. + +```bash +cd testenv && ./run-local.sh +``` + +Then, from another shell, hit the endpoint: + +```bash +curl -s http://localhost:8080/transfers \ + -H 'Content-Type: application/json' \ + -d '{"idempotencyKey":"demo-1","fromWalletId":"wallet_1","toWalletId":"wallet_2","amount":100}' +``` + +Inspect the resulting state (balances never go negative; the ledger balances): + +```bash +docker compose -f testenv/docker-compose.yml exec -T postgres \ + psql -U wallet -d wallet -c "SELECT wallet_id, balance FROM wallets ORDER BY wallet_id;" +``` + +Press Ctrl+C in the `run-local.sh` shell to stop the service and tear down Postgres. + +### Manual steps (without the script) + +```bash +# 1. start Postgres — schema.sql is applied by the container's init scripts +docker compose -f testenv/docker-compose.yml up -d + +# 2. seed two wallets +docker compose -f testenv/docker-compose.yml exec -T postgres \ + psql -U wallet -d wallet < testenv/seed.sql + +# 3. run the service +DATABASE_URL="postgres://wallet:wallet@localhost:5432/wallet?sslmode=disable" \ + go run ./cmd/server +``` + +## Testing + +```bash +go test ./... +``` + +Unit tests use in-memory mocks and exercise the real `CreateTransfer` logic through +an injectable transaction runner (no database required). They cover: successful +transfer + ledger balancing, duplicate-key replay, **snapshot semantics** (replay +returns the original balances), concurrent-duplicate conflict resolution, insufficient +funds (with replay), self-transfer rejection, and unknown wallet. + +### Concurrency integration tests (real Postgres) + +`test/integration` runs the **real** repositories and service against a live +Postgres and fires many transfers concurrently to prove the two invariants that +matter for a money system: **no double spend** and a **balanced ledger** +(ΣDEBIT = ΣCREDIT). It is gated on `TEST_DATABASE_URL`, so `go test ./...` still +passes with no database available. To run it: + +```bash +# start a throwaway Postgres (or reuse testenv/docker-compose.yml) +docker compose -f testenv/docker-compose.yml up -d + +TEST_DATABASE_URL="postgres://wallet:wallet@localhost:5432/wallet?sslmode=disable" \ + go test ./test/integration/... -v +``` + +- `TestConcurrentDistinctTransfersNoDoubleSpend` — 50 concurrent distinct-key + transfers against a wallet that can only fund 10 of them; asserts exactly 10 + succeed, the balance never goes negative, and the ledger balances. +- `TestConcurrentDuplicatesSingleSideEffect` — 30 concurrent requests sharing one + idempotency key; asserts the money moves exactly once and every caller observes + the same transfer. + +## Assumptions & tradeoffs + +- Amounts are integer minor units (`int64`); no currency/multi-asset handling. +- Balance is stored (not ledger-derived) for fast reads; the ledger stays authoritative. +- A terminal `FAILED` outcome is recorded against the idempotency key, so a given key + always yields one deterministic result (a client must use a new key to retry later). +- The schema is applied at database provisioning time (via the Postgres image's + init scripts locally); the application never issues DDL. In production the same + `migrations/schema.sql` is applied by your DB provisioning/migration step. + +--- + +# Repository Template Notes This repository is a reusable coding assignment template for evaluating backend engineers on wallet transfers, idempotency, concurrency control, and double-entry ledger design. diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 00000000..75125e47 --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,85 @@ +package main + +import ( + "context" + "errors" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/Robustrade/wallet-transfer-assignment/internal/api" + "github.com/Robustrade/wallet-transfer-assignment/internal/config" + "github.com/Robustrade/wallet-transfer-assignment/internal/logging" + pgrepo "github.com/Robustrade/wallet-transfer-assignment/internal/repository/postgres" + "github.com/Robustrade/wallet-transfer-assignment/internal/service" + "gorm.io/driver/postgres" + "gorm.io/gorm" +) + +func main() { + cfg, err := config.LoadConfig() + if err != nil { + log.Fatalf("configuration error: %v", err) + } + + logger := logging.NewLogger(cfg.LogLevel, nil) + logger.Info("service configuration loaded listenAddr=%s", cfg.ListenAddr) + + gormDB, err := gorm.Open(postgres.Open(cfg.DatabaseURL), &gorm.Config{}) + if err != nil { + logger.Error("open database: %v", err) + log.Fatalf("open database: %v", err) + } + + rawDB, err := gormDB.DB() + if err != nil { + logger.Error("database driver setup: %v", err) + log.Fatalf("database driver setup: %v", err) + } + defer rawDB.Close() + + rawDB.SetMaxOpenConns(cfg.DBMaxOpenConns) + rawDB.SetMaxIdleConns(cfg.DBMaxIdleConns) + + walletRepo := pgrepo.NewWalletRepository(gormDB) + transferRepo := pgrepo.NewTransferRepository(gormDB) + ledgerRepo := pgrepo.NewLedgerRepository(gormDB) + idempotencyRepo := pgrepo.NewIdempotencyRepository(gormDB) + transferSvc := service.NewTransferService(gormDB, walletRepo, transferRepo, ledgerRepo, idempotencyRepo, logger) + handler := api.NewTransferHandler(transferSvc) + + server := &http.Server{ + Addr: cfg.ListenAddr, + Handler: handler.Router(), + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + go func() { + logger.Info("starting wallet transfer service addr=%s", cfg.ListenAddr) + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Error("server error: %v", err) + log.Fatalf("server error: %v", err) + } + }() + + <-ctx.Done() + logger.Info("shutdown signal received, draining connections") + + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + logger.Error("graceful shutdown failed: %v", err) + if closeErr := server.Close(); closeErr != nil { + logger.Error("forced close failed: %v", closeErr) + } + } + logger.Info("service stopped") + _ = os.Stdout.Sync() +} diff --git a/go.mod b/go.mod new file mode 100644 index 00000000..d24db92c --- /dev/null +++ b/go.mod @@ -0,0 +1,24 @@ +module github.com/Robustrade/wallet-transfer-assignment + +go 1.22 + +require ( + github.com/jackc/pgconn v1.14.3 + gorm.io/driver/postgres v1.6.0 + gorm.io/gorm v1.31.2 +) + +require ( + github.com/jackc/chunkreader/v2 v2.0.1 // indirect + github.com/jackc/pgio v1.0.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgproto3/v2 v2.3.3 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/pgx/v5 v5.6.0 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/jinzhu/inflection v1.0.0 // indirect + github.com/jinzhu/now v1.1.5 // indirect + golang.org/x/crypto v0.31.0 // indirect + golang.org/x/sync v0.10.0 // indirect + golang.org/x/text v0.21.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 00000000..aa70c9a9 --- /dev/null +++ b/go.sum @@ -0,0 +1,53 @@ +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/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= +github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8= +github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= +github.com/jackc/pgconn v1.14.3 h1:bVoTr12EGANZz66nZPkMInAV/KHD2TxH9npjXXgiB3w= +github.com/jackc/pgconn v1.14.3/go.mod h1:RZbme4uasqzybK2RK5c65VsHxoyaml09lx3tXOcO/VM= +github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE= +github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8= +github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65 h1:DadwsjnMwFjfWc9y5Wi/+Zz7xoE5ALHsRQlOctkOiHc= +github.com/jackc/pgmock v0.0.0-20210724152146-4ad1a8207f65/go.mod h1:5R2h2EEX+qri8jOWMbJCtaPWkrrNc7OHwsp2TCqp7ak= +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/pgproto3/v2 v2.3.3 h1:1HLSx5H+tXR9pW3in3zaztoEwQYRC9SQaYUHjTSUOag= +github.com/jackc/pgproto3/v2 v2.3.3/go.mod h1:WfJCnwN3HIg9Ish/j3sgWXnAfK8A9Y0bwXYU5xKaEdA= +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.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.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= +github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= +github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= +github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= +github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= +github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +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.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +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.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= +golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= +golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= +golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= +golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +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= +gorm.io/driver/postgres v1.6.0 h1:2dxzU8xJ+ivvqTRph34QX+WrRaJlmfyPqXmoGVjMBa4= +gorm.io/driver/postgres v1.6.0/go.mod h1:vUw0mrGgrTK+uPHEhAdV4sfFELrByKVGnaVRkXDhtWo= +gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ= +gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8= +gorm.io/gorm v1.31.2 h1:3o8FXNo9v9S858gil+3LlZA1LkCOzgb4g5BL64FgaCo= +gorm.io/gorm v1.31.2/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs= diff --git a/internal/api/handler.go b/internal/api/handler.go new file mode 100644 index 00000000..97174448 --- /dev/null +++ b/internal/api/handler.go @@ -0,0 +1,85 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + + "github.com/Robustrade/wallet-transfer-assignment/internal/service" +) + +// TransferHandler defines the HTTP handler for wallet transfer operations. +type TransferHandler struct { + service service.TransferService +} + +// NewTransferHandler creates a new HTTP handler with a transfer service dependency. +func NewTransferHandler(service service.TransferService) *TransferHandler { + return &TransferHandler{service: service} +} + +// Router returns the HTTP router for the service endpoints. +func (h *TransferHandler) Router() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/transfers", h.handleCreateTransfer) + return mux +} + +// handleCreateTransfer accepts POST /transfers requests and returns transfer results. +func (h *TransferHandler) handleCreateTransfer(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + h.writeErrorStatus(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + var request TransferRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + h.writeErrorStatus(w, http.StatusBadRequest, "invalid JSON payload") + return + } + + response, err := h.service.CreateTransfer(r.Context(), service.CreateTransferRequest{ + IdempotencyKey: request.IdempotencyKey, + FromWalletID: request.FromWalletID, + ToWalletID: request.ToWalletID, + Amount: request.Amount, + }) + if err != nil { + h.writeError(w, err) + return + } + + h.writeJSON(w, http.StatusCreated, response) +} + +// writeError maps service errors to HTTP status codes and JSON response bodies. +func (h *TransferHandler) writeError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, service.ErrInvalidTransferRequest): + h.writeErrorStatus(w, http.StatusBadRequest, err.Error()) + case errors.Is(err, service.ErrWalletNotFound): + h.writeErrorStatus(w, http.StatusNotFound, err.Error()) + case errors.Is(err, service.ErrInsufficientFunds): + h.writeErrorStatus(w, http.StatusUnprocessableEntity, err.Error()) + default: + h.writeErrorStatus(w, http.StatusInternalServerError, "internal server error") + } +} + +func (h *TransferHandler) writeErrorStatus(w http.ResponseWriter, status int, message string) { + h.writeJSON(w, status, map[string]string{"error": message}) +} + +func (h *TransferHandler) writeJSON(w http.ResponseWriter, status int, payload any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(payload) +} + +// TransferRequest is the JSON payload accepted by POST /transfers. +type TransferRequest struct { + IdempotencyKey string `json:"idempotencyKey"` + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 00000000..912c5507 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,75 @@ +package config + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +const ( + DefaultListenAddr = ":8080" + DefaultLogLevel = "INFO" + DefaultDBMaxOpenConns = 25 + DefaultDBMaxIdleConns = 5 +) + +// Config holds application configuration loaded from the environment. +type Config struct { + DatabaseURL string + ListenAddr string + LogLevel string + DBMaxOpenConns int + DBMaxIdleConns int +} + +// LoadConfig reads environment variables and returns a validated config. +func LoadConfig() (Config, error) { + databaseURL := strings.TrimSpace(os.Getenv("DATABASE_URL")) + if databaseURL == "" { + return Config{}, fmt.Errorf("DATABASE_URL is required") + } + + listenAddr := strings.TrimSpace(os.Getenv("LISTEN_ADDR")) + if listenAddr == "" { + listenAddr = DefaultListenAddr + } + + logLevel := strings.TrimSpace(os.Getenv("LOG_LEVEL")) + if logLevel == "" { + logLevel = DefaultLogLevel + } + + maxOpenConns, err := parseIntEnv("DB_MAX_OPEN_CONNS", DefaultDBMaxOpenConns) + if err != nil { + return Config{}, err + } + + maxIdleConns, err := parseIntEnv("DB_MAX_IDLE_CONNS", DefaultDBMaxIdleConns) + if err != nil { + return Config{}, err + } + + return Config{ + DatabaseURL: databaseURL, + ListenAddr: listenAddr, + LogLevel: logLevel, + DBMaxOpenConns: maxOpenConns, + DBMaxIdleConns: maxIdleConns, + }, nil +} + +func parseIntEnv(key string, defaultValue int) (int, error) { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return defaultValue, nil + } + parsed, err := strconv.Atoi(value) + if err != nil { + return 0, fmt.Errorf("invalid %s: %w", key, err) + } + if parsed < 0 { + return 0, fmt.Errorf("invalid %s: must be >= 0", key) + } + return parsed, nil +} diff --git a/internal/logging/logger.go b/internal/logging/logger.go new file mode 100644 index 00000000..77a225a8 --- /dev/null +++ b/internal/logging/logger.go @@ -0,0 +1,83 @@ +package logging + +import ( + "fmt" + "io" + "log" + "os" + "strings" +) + +const ( + LevelDebug = "DEBUG" + LevelInfo = "INFO" + LevelWarn = "WARN" + LevelError = "ERROR" +) + +// Logger wraps the standard logger with level filtering. +type Logger struct { + base *log.Logger + level string +} + +// NewLogger creates a logger that writes to stdout and honors the given level. +func NewLogger(level string, output io.Writer) *Logger { + if output == nil { + output = os.Stdout + } + return &Logger{ + base: log.New(output, "", log.LstdFlags|log.Lmsgprefix), + level: strings.ToUpper(strings.TrimSpace(level)), + } +} + +// Debug logs a debug message when the enabled level is DEBUG. +func (l *Logger) Debug(msg string, v ...any) { + if l.enabled(LevelDebug) { + l.base.Printf("DEBUG: %s", fmtArgs(msg, v...)) + } +} + +// Info logs an informational message when the enabled level is DEBUG or INFO. +func (l *Logger) Info(msg string, v ...any) { + if l.enabled(LevelInfo) { + l.base.Printf("INFO: %s", fmtArgs(msg, v...)) + } +} + +// Warn logs a warning message at WARN or higher. +func (l *Logger) Warn(msg string, v ...any) { + if l.enabled(LevelWarn) { + l.base.Printf("WARN: %s", fmtArgs(msg, v...)) + } +} + +// Error logs an error message at ERROR. +func (l *Logger) Error(msg string, v ...any) { + if l.enabled(LevelError) { + l.base.Printf("ERROR: %s", fmtArgs(msg, v...)) + } +} + +func (l *Logger) enabled(level string) bool { + switch strings.ToUpper(level) { + case LevelDebug: + return l.level == LevelDebug + case LevelInfo: + return l.level == LevelDebug || l.level == LevelInfo + case LevelWarn: + return l.level == LevelDebug || l.level == LevelInfo || l.level == LevelWarn + case LevelError: + return true + default: + return true + } +} + +func fmtArgs(msg string, v ...any) string { + if len(v) == 0 { + return msg + } + return fmt.Sprintf(msg, v...) +} diff --git a/internal/model/model.go b/internal/model/model.go new file mode 100644 index 00000000..70d27602 --- /dev/null +++ b/internal/model/model.go @@ -0,0 +1,74 @@ +package model + +import ( + "crypto/rand" + "encoding/hex" + "time" +) + +// TransferState defines the lifecycle state of a transfer. +type TransferState string + +const ( + TransferStatePending TransferState = "PENDING" + TransferStateProcessed TransferState = "PROCESSED" + TransferStateFailed TransferState = "FAILED" +) + +// EntryType defines whether a ledger entry is a debit or credit. +type EntryType string + +const ( + EntryTypeDebit EntryType = "DEBIT" + EntryTypeCredit EntryType = "CREDIT" +) + +// Wallet represents the balance and metadata for a wallet. +// The balance CHECK constraint is a database-level backstop against double spend. +type Wallet struct { + WalletID string `gorm:"primaryKey;column:wallet_id"` + Balance int64 `gorm:"column:balance;not null;check:balance >= 0"` + CreatedAt time.Time `gorm:"column:created_at;autoCreateTime"` + UpdatedAt time.Time `gorm:"column:updated_at;autoUpdateTime"` +} + +// Transfer represents a transfer workflow record. +type Transfer struct { + TransferID string `gorm:"primaryKey;column:transfer_id"` + FromWalletID string `gorm:"column:from_wallet_id;not null"` + ToWalletID string `gorm:"column:to_wallet_id;not null"` + Amount int64 `gorm:"column:amount;not null"` + State TransferState `gorm:"column:state;not null"` + CreatedAt time.Time `gorm:"column:created_at;autoCreateTime"` + UpdatedAt time.Time `gorm:"column:updated_at;autoUpdateTime"` +} + +// LedgerEntry records an individual debit or credit for a transfer. +// The unique (transfer_id, type) index guarantees each transfer produces at most +// one debit and one credit, enforcing double-entry integrity at the database. +type LedgerEntry struct { + LedgerEntryID string `gorm:"primaryKey;column:ledger_entry_id"` + WalletID string `gorm:"column:wallet_id;not null;index"` + TransferID string `gorm:"column:transfer_id;not null;uniqueIndex:ux_ledger_transfer_type"` + EntryType EntryType `gorm:"column:type;not null;uniqueIndex:ux_ledger_transfer_type"` + Amount int64 `gorm:"column:amount;not null"` + CreatedAt time.Time `gorm:"column:created_at;autoCreateTime"` +} + +// IdempotencyRecord maps a client idempotency key to a transfer and stores a +// snapshot of the original response. Storing the response guarantees that a +// replayed request returns the exact original result, even if wallet balances +// have since changed, delivering exactly-once semantics at the API level. +type IdempotencyRecord struct { + IdempotencyKey string `gorm:"primaryKey;column:idempotency_key"` + TransferID string `gorm:"column:transfer_id;not null"` + Response string `gorm:"column:response;type:text;not null"` + CreatedAt time.Time `gorm:"column:created_at;autoCreateTime"` +} + +// NewID returns a randomly generated hex identifier. +func NewID() string { + var buffer [16]byte + _, _ = rand.Read(buffer[:]) + return hex.EncodeToString(buffer[:]) +} diff --git a/internal/repository/postgres/idempotency.go b/internal/repository/postgres/idempotency.go new file mode 100644 index 00000000..88751a8b --- /dev/null +++ b/internal/repository/postgres/idempotency.go @@ -0,0 +1,38 @@ +package postgres + +import ( + "context" + + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/Robustrade/wallet-transfer-assignment/internal/repository" + "gorm.io/gorm" +) + +type idempotencyRepository struct { + db *gorm.DB +} + +// NewIdempotencyRepository returns a Postgres-backed IdempotencyRepository. +func NewIdempotencyRepository(db *gorm.DB) repository.IdempotencyRepository { + return &idempotencyRepository{db: db} +} + +func (r *idempotencyRepository) GetByKey(ctx context.Context, tx *gorm.DB, key string) (*model.IdempotencyRecord, error) { + if tx == nil { + tx = r.db.WithContext(ctx) + } + + var record model.IdempotencyRecord + if err := tx.Where("idempotency_key = ?", key).First(&record).Error; err != nil { + return nil, translateError(err) + } + return &record, nil +} + +func (r *idempotencyRepository) Create(ctx context.Context, tx *gorm.DB, record model.IdempotencyRecord) error { + if tx == nil { + tx = r.db.WithContext(ctx) + } + + return translateError(tx.Create(&record).Error) +} diff --git a/internal/repository/postgres/ledger.go b/internal/repository/postgres/ledger.go new file mode 100644 index 00000000..a7e99554 --- /dev/null +++ b/internal/repository/postgres/ledger.go @@ -0,0 +1,25 @@ +package postgres + +import ( + "context" + + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/Robustrade/wallet-transfer-assignment/internal/repository" + "gorm.io/gorm" +) + +type ledgerRepository struct { + db *gorm.DB +} + +// NewLedgerRepository returns a Postgres-backed LedgerRepository. +func NewLedgerRepository(db *gorm.DB) repository.LedgerRepository { + return &ledgerRepository{db: db} +} + +func (r *ledgerRepository) CreateEntries(ctx context.Context, tx *gorm.DB, entries []model.LedgerEntry) error { + if tx == nil { + tx = r.db.WithContext(ctx) + } + return translateError(tx.Create(&entries).Error) +} diff --git a/internal/repository/postgres/transfer.go b/internal/repository/postgres/transfer.go new file mode 100644 index 00000000..3a9ce75c --- /dev/null +++ b/internal/repository/postgres/transfer.go @@ -0,0 +1,55 @@ +package postgres + +import ( + "context" + "time" + + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/Robustrade/wallet-transfer-assignment/internal/repository" + "gorm.io/gorm" +) + +type transferRepository struct { + db *gorm.DB +} + +// NewTransferRepository returns a Postgres-backed TransferRepository. +func NewTransferRepository(db *gorm.DB) repository.TransferRepository { + return &transferRepository{db: db} +} + +func (r *transferRepository) CreateTransfer(ctx context.Context, tx *gorm.DB, transfer model.Transfer) error { + if tx == nil { + tx = r.db.WithContext(ctx) + } + return translateError(tx.Create(&transfer).Error) +} + +func (r *transferRepository) GetByID(ctx context.Context, tx *gorm.DB, transferID string) (*model.Transfer, error) { + if tx == nil { + tx = r.db.WithContext(ctx) + } + + var transfer model.Transfer + if err := tx.Where("transfer_id = ?", transferID).First(&transfer).Error; err != nil { + return nil, translateError(err) + } + return &transfer, nil +} + +func (r *transferRepository) UpdateTransferState(ctx context.Context, tx *gorm.DB, transferID string, state model.TransferState) error { + if tx == nil { + tx = r.db.WithContext(ctx) + } + + result := tx.Model(&model.Transfer{}). + Where("transfer_id = ?", transferID). + Updates(map[string]any{"state": state, "updated_at": time.Now().UTC()}) + if result.Error != nil { + return translateError(result.Error) + } + if result.RowsAffected != 1 { + return repository.ErrNotFound + } + return nil +} diff --git a/internal/repository/postgres/wallet.go b/internal/repository/postgres/wallet.go new file mode 100644 index 00000000..67a39e99 --- /dev/null +++ b/internal/repository/postgres/wallet.go @@ -0,0 +1,65 @@ +package postgres + +import ( + "context" + "errors" + "time" + + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/Robustrade/wallet-transfer-assignment/internal/repository" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +type walletRepository struct { + db *gorm.DB +} + +// NewWalletRepository returns a Postgres-backed WalletRepository. +func NewWalletRepository(db *gorm.DB) repository.WalletRepository { + return &walletRepository{db: db} +} + +func (r *walletRepository) GetWalletForUpdate(ctx context.Context, tx *gorm.DB, walletID string) (*model.Wallet, error) { + if tx == nil { + tx = r.db.WithContext(ctx) + } + + var wallet model.Wallet + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("wallet_id = ?", walletID).First(&wallet).Error; err != nil { + return nil, translateError(err) + } + return &wallet, nil +} + +func (r *walletRepository) UpdateBalance(ctx context.Context, tx *gorm.DB, walletID string, newBalance int64) error { + if tx == nil { + tx = r.db.WithContext(ctx) + } + + result := tx.Model(&model.Wallet{}). + Where("wallet_id = ?", walletID). + Updates(map[string]any{"balance": newBalance, "updated_at": time.Now().UTC()}) + if result.Error != nil { + return translateError(result.Error) + } + if result.RowsAffected != 1 { + return repository.ErrNotFound + } + return nil +} + +// translateError maps GORM/driver errors to repository-level sentinels so that +// callers depend only on the repository abstraction, not on the ORM. +func translateError(err error) error { + if err == nil { + return nil + } + if errors.Is(err, gorm.ErrRecordNotFound) { + return repository.ErrNotFound + } + if repository.IsUniqueViolation(err) { + return repository.ErrUniqueViolation + } + return err +} diff --git a/internal/repository/repository.go b/internal/repository/repository.go new file mode 100644 index 00000000..6ee0a9c9 --- /dev/null +++ b/internal/repository/repository.go @@ -0,0 +1,55 @@ +package repository + +import ( + "context" + "errors" + "strings" + + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/jackc/pgconn" + "gorm.io/gorm" +) + +var ( + ErrNotFound = errors.New("repository: not found") + ErrUniqueViolation = errors.New("repository: unique violation") +) + +// WalletRepository defines persistence operations for wallet state. +type WalletRepository interface { + GetWalletForUpdate(ctx context.Context, tx *gorm.DB, walletID string) (*model.Wallet, error) + UpdateBalance(ctx context.Context, tx *gorm.DB, walletID string, newBalance int64) error +} + +// TransferRepository defines persistence operations for transfer workflow records. +type TransferRepository interface { + CreateTransfer(ctx context.Context, tx *gorm.DB, transfer model.Transfer) error + GetByID(ctx context.Context, tx *gorm.DB, transferID string) (*model.Transfer, error) + UpdateTransferState(ctx context.Context, tx *gorm.DB, transferID string, state model.TransferState) error +} + +// LedgerRepository defines persistence operations for ledger entries. +type LedgerRepository interface { + CreateEntries(ctx context.Context, tx *gorm.DB, entries []model.LedgerEntry) error +} + +// IdempotencyRepository defines persistence operations for idempotency records. +type IdempotencyRepository interface { + GetByKey(ctx context.Context, tx *gorm.DB, key string) (*model.IdempotencyRecord, error) + Create(ctx context.Context, tx *gorm.DB, record model.IdempotencyRecord) error +} + +// IsUniqueViolation returns true when an error is a database unique constraint violation. +func IsUniqueViolation(err error) bool { + if errors.Is(err, ErrUniqueViolation) { + return true + } + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + return pgErr.Code == "23505" + } + if strings.Contains(strings.ToLower(err.Error()), "unique") { + return true + } + return false +} diff --git a/internal/service/transfer_service.go b/internal/service/transfer_service.go new file mode 100644 index 00000000..9bbf5afb --- /dev/null +++ b/internal/service/transfer_service.go @@ -0,0 +1,316 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/Robustrade/wallet-transfer-assignment/internal/logging" + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/Robustrade/wallet-transfer-assignment/internal/repository" + "gorm.io/gorm" +) + +// CreateTransferRequest contains the data required to execute a wallet transfer. +type CreateTransferRequest struct { + IdempotencyKey string + FromWalletID string + ToWalletID string + Amount int64 +} + +// TransferResponse is returned by the transfer service after transfer execution. +type TransferResponse struct { + TransferID string `json:"transferId"` + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` + State model.TransferState `json:"state"` + FromWalletBalance int64 `json:"fromWalletBalance"` + ToWalletBalance int64 `json:"toWalletBalance"` + CreatedAt time.Time `json:"createdAt"` +} + +var ( + ErrInvalidTransferRequest = errors.New("invalid transfer request") + ErrInsufficientFunds = errors.New("insufficient funds") + ErrWalletNotFound = errors.New("wallet not found") + + // errIdempotencyConflict signals that a concurrent duplicate request committed + // first. It is handled internally to replay the stored result and is never + // surfaced to callers. + errIdempotencyConflict = errors.New("idempotency conflict") +) + +// txFunc executes fn inside a database transaction. It is injectable so the +// business logic in CreateTransfer can be unit tested without a live database. +type txFunc func(ctx context.Context, fn func(tx *gorm.DB) error) error + +// TransferService defines the service boundary for transfer business logic. +type TransferService interface { + CreateTransfer(ctx context.Context, request CreateTransferRequest) (TransferResponse, error) +} + +// transferService is the default implementation of TransferService. +type transferService struct { + withTx txFunc + walletRepo repository.WalletRepository + transferRepo repository.TransferRepository + ledgerRepo repository.LedgerRepository + idempotencyRepo repository.IdempotencyRepository + logger *logging.Logger +} + +// NewTransferService builds a TransferService using repository abstractions. +func NewTransferService( + db *gorm.DB, + walletRepo repository.WalletRepository, + transferRepo repository.TransferRepository, + ledgerRepo repository.LedgerRepository, + idempotencyRepo repository.IdempotencyRepository, + logger *logging.Logger, +) TransferService { + return &transferService{ + withTx: func(ctx context.Context, fn func(tx *gorm.DB) error) error { + return db.WithContext(ctx).Transaction(fn) + }, + walletRepo: walletRepo, + transferRepo: transferRepo, + ledgerRepo: ledgerRepo, + idempotencyRepo: idempotencyRepo, + logger: logger, + } +} + +// CreateTransfer validates the request, enforces idempotency, and moves money +// atomically between two wallets while recording a double-entry ledger. +func (s *transferService) CreateTransfer(ctx context.Context, request CreateTransferRequest) (TransferResponse, error) { + if request.IdempotencyKey == "" || request.FromWalletID == "" || request.ToWalletID == "" || request.Amount <= 0 { + return TransferResponse{}, ErrInvalidTransferRequest + } + if request.FromWalletID == request.ToWalletID { + return TransferResponse{}, ErrInvalidTransferRequest + } + + var response TransferResponse + var businessErr error + + err := s.withTx(ctx, func(tx *gorm.DB) error { + // 1. Replay: if we have already processed this key, return the stored + // original result verbatim (exactly-once semantics). + if existing, lookupErr := s.idempotencyRepo.GetByKey(ctx, tx, request.IdempotencyKey); lookupErr == nil { + stored, decodeErr := decodeResponse(existing.Response) + if decodeErr != nil { + return decodeErr + } + response = stored + businessErr = businessErrorFor(stored.State) + return nil + } else if !errors.Is(lookupErr, repository.ErrNotFound) { + return lookupErr + } + + // 2. Lock both wallets FOR UPDATE in a deterministic order to prevent + // deadlocks between opposing concurrent transfers (A->B and B->A). + fromWallet, toWallet, err := s.lockWalletPair(ctx, tx, request.FromWalletID, request.ToWalletID) + if err != nil { + return err + } + + transfer := model.Transfer{ + TransferID: model.NewID(), + FromWalletID: request.FromWalletID, + ToWalletID: request.ToWalletID, + Amount: request.Amount, + State: model.TransferStatePending, + } + if err := s.transferRepo.CreateTransfer(ctx, tx, transfer); err != nil { + return err + } + + // 3. Insufficient funds is a terminal FAILED outcome. It is still recorded + // against the idempotency key so retries return the same result. + if fromWallet.Balance < request.Amount { + if err := s.transferRepo.UpdateTransferState(ctx, tx, transfer.TransferID, model.TransferStateFailed); err != nil { + return err + } + transfer.State = model.TransferStateFailed + resp := buildResponse(&transfer, fromWallet.Balance, toWallet.Balance) + if err := s.persistIdempotency(ctx, tx, request.IdempotencyKey, transfer.TransferID, resp); err != nil { + return err + } + response = resp + businessErr = ErrInsufficientFunds + return nil + } + + // 4. Move the money and record the double-entry ledger. + newFrom := fromWallet.Balance - request.Amount + newTo := toWallet.Balance + request.Amount + if err := s.walletRepo.UpdateBalance(ctx, tx, request.FromWalletID, newFrom); err != nil { + return err + } + if err := s.walletRepo.UpdateBalance(ctx, tx, request.ToWalletID, newTo); err != nil { + return err + } + + entries := []model.LedgerEntry{ + { + LedgerEntryID: model.NewID(), + WalletID: request.FromWalletID, + TransferID: transfer.TransferID, + EntryType: model.EntryTypeDebit, + Amount: request.Amount, + }, + { + LedgerEntryID: model.NewID(), + WalletID: request.ToWalletID, + TransferID: transfer.TransferID, + EntryType: model.EntryTypeCredit, + Amount: request.Amount, + }, + } + if err := s.ledgerRepo.CreateEntries(ctx, tx, entries); err != nil { + return err + } + + if err := s.transferRepo.UpdateTransferState(ctx, tx, transfer.TransferID, model.TransferStateProcessed); err != nil { + return err + } + transfer.State = model.TransferStateProcessed + + resp := buildResponse(&transfer, newFrom, newTo) + if err := s.persistIdempotency(ctx, tx, request.IdempotencyKey, transfer.TransferID, resp); err != nil { + return err + } + response = resp + return nil + }) + + // A concurrent duplicate committed first: roll back this attempt and return + // the winner's stored result instead of an error. + if errors.Is(err, errIdempotencyConflict) { + return s.replayStoredResponse(ctx, request.IdempotencyKey) + } + if err != nil { + s.log().Error("create transfer failed key=%s: %v", request.IdempotencyKey, err) + return TransferResponse{}, err + } + if businessErr != nil { + s.log().Info("transfer not processed key=%s state=%s", request.IdempotencyKey, response.State) + return response, businessErr + } + s.log().Info("transfer processed key=%s transferId=%s amount=%d", request.IdempotencyKey, response.TransferID, response.Amount) + return response, nil +} + +// lockWalletPair acquires FOR UPDATE locks on both wallets in a deterministic +// (sorted) order to avoid deadlocks, then returns them mapped to from/to roles. +func (s *transferService) lockWalletPair(ctx context.Context, tx *gorm.DB, fromID, toID string) (from, to *model.Wallet, err error) { + first, second := fromID, toID + if first > second { + first, second = second, first + } + + w1, err := s.walletRepo.GetWalletForUpdate(ctx, tx, first) + if err != nil { + return nil, nil, walletLookupError(err) + } + w2, err := s.walletRepo.GetWalletForUpdate(ctx, tx, second) + if err != nil { + return nil, nil, walletLookupError(err) + } + + byID := map[string]*model.Wallet{first: w1, second: w2} + return byID[fromID], byID[toID], nil +} + +// persistIdempotency stores the response snapshot under the idempotency key. A +// unique violation means a concurrent duplicate won the race; it is converted +// to errIdempotencyConflict so the caller can replay the stored result. +func (s *transferService) persistIdempotency(ctx context.Context, tx *gorm.DB, key, transferID string, resp TransferResponse) error { + payload, err := json.Marshal(resp) + if err != nil { + return err + } + record := model.IdempotencyRecord{ + IdempotencyKey: key, + TransferID: transferID, + Response: string(payload), + } + if err := s.idempotencyRepo.Create(ctx, tx, record); err != nil { + if repository.IsUniqueViolation(err) { + return errIdempotencyConflict + } + return err + } + return nil +} + +// replayStoredResponse reads a previously committed idempotency result. +func (s *transferService) replayStoredResponse(ctx context.Context, key string) (TransferResponse, error) { + var response TransferResponse + var businessErr error + err := s.withTx(ctx, func(tx *gorm.DB) error { + existing, err := s.idempotencyRepo.GetByKey(ctx, tx, key) + if err != nil { + return err + } + stored, err := decodeResponse(existing.Response) + if err != nil { + return err + } + response = stored + businessErr = businessErrorFor(stored.State) + return nil + }) + if err != nil { + return TransferResponse{}, err + } + return response, businessErr +} + +func (s *transferService) log() *logging.Logger { + if s.logger == nil { + return logging.NewLogger(logging.LevelError, nil) + } + return s.logger +} + +// buildResponse assembles a transport response from a transfer and the balances +// captured at the moment of execution. +func buildResponse(transfer *model.Transfer, fromBalance, toBalance int64) TransferResponse { + return TransferResponse{ + TransferID: transfer.TransferID, + FromWalletID: transfer.FromWalletID, + ToWalletID: transfer.ToWalletID, + Amount: transfer.Amount, + State: transfer.State, + FromWalletBalance: fromBalance, + ToWalletBalance: toBalance, + CreatedAt: transfer.CreatedAt, + } +} + +func decodeResponse(payload string) (TransferResponse, error) { + var response TransferResponse + if err := json.Unmarshal([]byte(payload), &response); err != nil { + return TransferResponse{}, err + } + return response, nil +} + +func businessErrorFor(state model.TransferState) error { + if state == model.TransferStateFailed { + return ErrInsufficientFunds + } + return nil +} + +func walletLookupError(err error) error { + if errors.Is(err, repository.ErrNotFound) { + return ErrWalletNotFound + } + return err +} diff --git a/internal/service/transfer_service_test.go b/internal/service/transfer_service_test.go new file mode 100644 index 00000000..7bd7e960 --- /dev/null +++ b/internal/service/transfer_service_test.go @@ -0,0 +1,349 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + "github.com/Robustrade/wallet-transfer-assignment/internal/repository" + "gorm.io/gorm" +) + +type mockWalletRepo struct { + wallets map[string]*model.Wallet +} + +func (m *mockWalletRepo) GetWalletForUpdate(ctx context.Context, tx *gorm.DB, walletID string) (*model.Wallet, error) { + wallet, ok := m.wallets[walletID] + if !ok { + return nil, repository.ErrNotFound + } + return wallet, nil +} + +func (m *mockWalletRepo) UpdateBalance(ctx context.Context, tx *gorm.DB, walletID string, newBalance int64) error { + wallet, ok := m.wallets[walletID] + if !ok { + return repository.ErrNotFound + } + wallet.Balance = newBalance + wallet.UpdatedAt = time.Now().UTC() + return nil +} + +type mockTransferRepo struct { + transfers map[string]*model.Transfer +} + +func (m *mockTransferRepo) CreateTransfer(ctx context.Context, tx *gorm.DB, transfer model.Transfer) error { + if _, exists := m.transfers[transfer.TransferID]; exists { + return errors.New("duplicate transfer") + } + stored := transfer + stored.CreatedAt = time.Now().UTC() + stored.UpdatedAt = stored.CreatedAt + m.transfers[transfer.TransferID] = &stored + return nil +} + +func (m *mockTransferRepo) GetByID(ctx context.Context, tx *gorm.DB, transferID string) (*model.Transfer, error) { + transfer, ok := m.transfers[transferID] + if !ok { + return nil, repository.ErrNotFound + } + return transfer, nil +} + +func (m *mockTransferRepo) UpdateTransferState(ctx context.Context, tx *gorm.DB, transferID string, state model.TransferState) error { + transfer, ok := m.transfers[transferID] + if !ok { + return repository.ErrNotFound + } + transfer.State = state + transfer.UpdatedAt = time.Now().UTC() + return nil +} + +type mockLedgerRepo struct { + entries []model.LedgerEntry +} + +func (m *mockLedgerRepo) CreateEntries(ctx context.Context, tx *gorm.DB, entries []model.LedgerEntry) error { + m.entries = append(m.entries, entries...) + return nil +} + +type mockIdempotencyRepo struct { + records map[string]*model.IdempotencyRecord +} + +func (m *mockIdempotencyRepo) GetByKey(ctx context.Context, tx *gorm.DB, key string) (*model.IdempotencyRecord, error) { + record, ok := m.records[key] + if !ok { + return nil, repository.ErrNotFound + } + return record, nil +} + +func (m *mockIdempotencyRepo) Create(ctx context.Context, tx *gorm.DB, record model.IdempotencyRecord) error { + if _, exists := m.records[record.IdempotencyKey]; exists { + return repository.ErrUniqueViolation + } + stored := record + m.records[record.IdempotencyKey] = &stored + return nil +} + +// directTx runs the transactional body without a database, so the real +// CreateTransfer business logic is exercised by unit tests. +func directTx(ctx context.Context, fn func(tx *gorm.DB) error) error { + return fn(nil) +} + +func newTestService(walletRepo repository.WalletRepository, idempotencyRepo repository.IdempotencyRepository) (*transferService, *mockTransferRepo, *mockLedgerRepo) { + transferRepo := &mockTransferRepo{transfers: map[string]*model.Transfer{}} + ledgerRepo := &mockLedgerRepo{} + svc := &transferService{ + withTx: directTx, + walletRepo: walletRepo, + transferRepo: transferRepo, + ledgerRepo: ledgerRepo, + idempotencyRepo: idempotencyRepo, + } + return svc, transferRepo, ledgerRepo +} + +func TestCreateTransferSucceeds(t *testing.T) { + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 500}, + "wallet_2": {WalletID: "wallet_2", Balance: 100}, + }} + idempotencyRepo := &mockIdempotencyRepo{records: map[string]*model.IdempotencyRecord{}} + svc, _, ledgerRepo := newTestService(walletRepo, idempotencyRepo) + + response, err := svc.CreateTransfer(context.Background(), CreateTransferRequest{ + IdempotencyKey: "abc123", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 200, + }) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + + if response.State != model.TransferStateProcessed { + t.Fatalf("expected PROCESSED, got %s", response.State) + } + if response.FromWalletBalance != 300 || response.ToWalletBalance != 300 { + t.Fatalf("expected updated balances, got %d and %d", response.FromWalletBalance, response.ToWalletBalance) + } + if len(ledgerRepo.entries) != 2 { + t.Fatalf("expected two ledger entries, got %d", len(ledgerRepo.entries)) + } + if ledgerRepo.entries[0].EntryType != model.EntryTypeDebit || ledgerRepo.entries[1].EntryType != model.EntryTypeCredit { + t.Fatalf("expected debit then credit entries") + } +} + +func TestCreateTransferReturnsSameResultForDuplicateIdempotencyKey(t *testing.T) { + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 500}, + "wallet_2": {WalletID: "wallet_2", Balance: 100}, + }} + idempotencyRepo := &mockIdempotencyRepo{records: map[string]*model.IdempotencyRecord{}} + svc, _, ledgerRepo := newTestService(walletRepo, idempotencyRepo) + + request := CreateTransferRequest{ + IdempotencyKey: "duplicate-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 150, + } + + first, err := svc.CreateTransfer(context.Background(), request) + if err != nil { + t.Fatalf("first transfer failed: %v", err) + } + second, err := svc.CreateTransfer(context.Background(), request) + if err != nil { + t.Fatalf("duplicate transfer failed: %v", err) + } + + if first.TransferID != second.TransferID { + t.Fatalf("expected same transfer id for duplicate request") + } + if len(ledgerRepo.entries) != 2 { + t.Fatalf("expected only two ledger entries, got %d", len(ledgerRepo.entries)) + } +} + +// TestDuplicateReturnsOriginalSnapshot proves exactly-once semantics: a replayed +// request returns the balances captured at execution time, even if the wallets +// have since changed. +func TestDuplicateReturnsOriginalSnapshot(t *testing.T) { + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 500}, + "wallet_2": {WalletID: "wallet_2", Balance: 100}, + }} + idempotencyRepo := &mockIdempotencyRepo{records: map[string]*model.IdempotencyRecord{}} + svc, _, _ := newTestService(walletRepo, idempotencyRepo) + + request := CreateTransferRequest{ + IdempotencyKey: "snapshot-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 100, + } + + first, err := svc.CreateTransfer(context.Background(), request) + if err != nil { + t.Fatalf("first transfer failed: %v", err) + } + + // Simulate later activity changing the wallet balances. + walletRepo.wallets["wallet_1"].Balance = 9999 + walletRepo.wallets["wallet_2"].Balance = 8888 + + replay, err := svc.CreateTransfer(context.Background(), request) + if err != nil { + t.Fatalf("replay failed: %v", err) + } + if replay.FromWalletBalance != first.FromWalletBalance || replay.ToWalletBalance != first.ToWalletBalance { + t.Fatalf("replay should return original snapshot balances, got from=%d to=%d", replay.FromWalletBalance, replay.ToWalletBalance) + } +} + +func TestCreateTransferFailsForInsufficientFunds(t *testing.T) { + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 50}, + "wallet_2": {WalletID: "wallet_2", Balance: 100}, + }} + idempotencyRepo := &mockIdempotencyRepo{records: map[string]*model.IdempotencyRecord{}} + svc, _, ledgerRepo := newTestService(walletRepo, idempotencyRepo) + + request := CreateTransferRequest{ + IdempotencyKey: "insufficient-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 200, + } + + resp, err := svc.CreateTransfer(context.Background(), request) + if !errors.Is(err, ErrInsufficientFunds) { + t.Fatalf("expected insufficient funds error, got %v", err) + } + if resp.State != model.TransferStateFailed { + t.Fatalf("expected FAILED state, got %s", resp.State) + } + if len(ledgerRepo.entries) != 0 { + t.Fatalf("expected no ledger entries for failed transfer, got %d", len(ledgerRepo.entries)) + } + + // Replaying the same key must return the same FAILED outcome. + replay, err := svc.CreateTransfer(context.Background(), request) + if !errors.Is(err, ErrInsufficientFunds) { + t.Fatalf("expected insufficient funds on replay, got %v", err) + } + if replay.TransferID != resp.TransferID { + t.Fatalf("expected same transfer id on replay") + } +} + +func TestCreateTransferRejectsSelfTransfer(t *testing.T) { + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 500}, + }} + idempotencyRepo := &mockIdempotencyRepo{records: map[string]*model.IdempotencyRecord{}} + svc, _, _ := newTestService(walletRepo, idempotencyRepo) + + _, err := svc.CreateTransfer(context.Background(), CreateTransferRequest{ + IdempotencyKey: "self-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_1", + Amount: 100, + }) + if !errors.Is(err, ErrInvalidTransferRequest) { + t.Fatalf("expected invalid request for self transfer, got %v", err) + } +} + +func TestCreateTransferUnknownWalletReturnsNotFound(t *testing.T) { + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 500}, + }} + idempotencyRepo := &mockIdempotencyRepo{records: map[string]*model.IdempotencyRecord{}} + svc, _, _ := newTestService(walletRepo, idempotencyRepo) + + _, err := svc.CreateTransfer(context.Background(), CreateTransferRequest{ + IdempotencyKey: "missing-wallet", + FromWalletID: "wallet_1", + ToWalletID: "wallet_missing", + Amount: 100, + }) + if !errors.Is(err, ErrWalletNotFound) { + t.Fatalf("expected wallet not found, got %v", err) + } +} + +// conflictIdempotencyRepo simulates a concurrent duplicate that committed first: +// the initial pre-check misses, the insert conflicts, and the replay lookup finds +// the winner's stored result. +type conflictIdempotencyRepo struct { + winner *model.IdempotencyRecord + seen bool +} + +func (m *conflictIdempotencyRepo) GetByKey(ctx context.Context, tx *gorm.DB, key string) (*model.IdempotencyRecord, error) { + if !m.seen { + m.seen = true + return nil, repository.ErrNotFound + } + return m.winner, nil +} + +func (m *conflictIdempotencyRepo) Create(ctx context.Context, tx *gorm.DB, record model.IdempotencyRecord) error { + return repository.ErrUniqueViolation +} + +func TestConcurrentDuplicateReplaysWinnerResult(t *testing.T) { + winnerResponse := TransferResponse{ + TransferID: "winner-transfer", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 100, + State: model.TransferStateProcessed, + FromWalletBalance: 400, + ToWalletBalance: 200, + } + payload, err := json.Marshal(winnerResponse) + if err != nil { + t.Fatalf("marshal winner response: %v", err) + } + + walletRepo := &mockWalletRepo{wallets: map[string]*model.Wallet{ + "wallet_1": {WalletID: "wallet_1", Balance: 500}, + "wallet_2": {WalletID: "wallet_2", Balance: 100}, + }} + idempotencyRepo := &conflictIdempotencyRepo{winner: &model.IdempotencyRecord{ + IdempotencyKey: "race-key", + TransferID: winnerResponse.TransferID, + Response: string(payload), + }} + svc, _, _ := newTestService(walletRepo, idempotencyRepo) + + resp, err := svc.CreateTransfer(context.Background(), CreateTransferRequest{ + IdempotencyKey: "race-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 100, + }) + if err != nil { + t.Fatalf("expected conflict to resolve to winner result, got %v", err) + } + if resp.TransferID != winnerResponse.TransferID { + t.Fatalf("expected winner transfer id %q, got %q", winnerResponse.TransferID, resp.TransferID) + } +} diff --git a/migrations/schema.sql b/migrations/schema.sql new file mode 100644 index 00000000..c2f8f346 --- /dev/null +++ b/migrations/schema.sql @@ -0,0 +1,54 @@ +-- Wallet transfer schema. +-- +-- This is the authoritative DDL for the service. It is applied when the database +-- is provisioned, NOT by the application at startup. Locally, the Postgres +-- container runs this file via /docker-entrypoint-initdb.d/ on first init (see +-- testenv/docker-compose.yml). The service assumes the schema already exists. +-- +-- Statements use IF NOT EXISTS so applying the schema is safe to re-run. + +CREATE TABLE IF NOT EXISTS wallets ( + wallet_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 IF NOT EXISTS transfers ( + transfer_id TEXT PRIMARY KEY, + from_wallet_id TEXT NOT NULL REFERENCES wallets(wallet_id), + to_wallet_id TEXT NOT NULL REFERENCES wallets(wallet_id), + amount BIGINT NOT NULL CHECK (amount > 0), + state TEXT NOT NULL CHECK (state IN ('PENDING', 'PROCESSED', 'FAILED')), + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + CONSTRAINT transfers_distinct_wallets CHECK (from_wallet_id <> to_wallet_id) +); + +CREATE INDEX IF NOT EXISTS transfers_from_wallet_idx ON transfers (from_wallet_id); +CREATE INDEX IF NOT EXISTS transfers_to_wallet_idx ON transfers (to_wallet_id); +CREATE INDEX IF NOT EXISTS transfers_state_idx ON transfers (state); + +CREATE TABLE IF NOT EXISTS ledger_entries ( + ledger_entry_id TEXT PRIMARY KEY, + wallet_id TEXT NOT NULL REFERENCES wallets(wallet_id), + transfer_id TEXT NOT NULL REFERENCES transfers(transfer_id), + type TEXT NOT NULL CHECK (type IN ('DEBIT', 'CREDIT')), + amount BIGINT NOT NULL CHECK (amount > 0), + created_at TIMESTAMPTZ NOT NULL +); + +CREATE INDEX IF NOT EXISTS ledger_entries_wallet_idx ON ledger_entries (wallet_id); + +-- Guarantees exactly one DEBIT and one CREDIT per transfer (double-entry integrity). +CREATE UNIQUE INDEX IF NOT EXISTS ux_ledger_transfer_type ON ledger_entries (transfer_id, type); + +CREATE TABLE IF NOT EXISTS idempotency_records ( + idempotency_key TEXT PRIMARY KEY, + transfer_id TEXT NOT NULL REFERENCES transfers(transfer_id), + -- Snapshot of the original response so replayed requests return the exact + -- original result (exactly-once semantics at the API level). + response TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL +); + diff --git a/server b/server new file mode 100644 index 00000000..f460c846 Binary files /dev/null and b/server differ diff --git a/test/integration/concurrency_test.go b/test/integration/concurrency_test.go new file mode 100644 index 00000000..5cb95847 --- /dev/null +++ b/test/integration/concurrency_test.go @@ -0,0 +1,269 @@ +package integration + +// Package integration holds tests that run the real service wiring against a +// live Postgres. They are gated on TEST_DATABASE_URL so `go test ./...` still +// passes with no database available; set the variable to opt in: +// +// TEST_DATABASE_URL="postgres://wallet:wallet@localhost:5432/wallet?sslmode=disable" \ +// go test ./test/integration/... +// +// The tests below hammer POST /transfers-equivalent service calls concurrently +// and assert the two invariants that matter for a money system: no double spend +// and a ledger that always balances (sum of DEBITs == sum of CREDITs). + +import ( + "context" + "os" + "sync" + "testing" + + "github.com/Robustrade/wallet-transfer-assignment/internal/logging" + "github.com/Robustrade/wallet-transfer-assignment/internal/model" + pgrepo "github.com/Robustrade/wallet-transfer-assignment/internal/repository/postgres" + "github.com/Robustrade/wallet-transfer-assignment/internal/service" + "gorm.io/driver/postgres" + "gorm.io/gorm" + gormlogger "gorm.io/gorm/logger" +) + +// openDB connects to the test Postgres, migrates the schema, and truncates all +// tables so each test starts from a clean, isolated state. It skips (not fails) +// when TEST_DATABASE_URL is unset. +func openDB(t *testing.T) *gorm.DB { + t.Helper() + + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + t.Skip("TEST_DATABASE_URL not set; skipping Postgres integration test") + } + + db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{ + Logger: gormlogger.Default.LogMode(gormlogger.Silent), + }) + if err != nil { + t.Fatalf("open test database: %v", err) + } + + if err := db.AutoMigrate( + &model.Wallet{}, + &model.Transfer{}, + &model.LedgerEntry{}, + &model.IdempotencyRecord{}, + ); err != nil { + t.Fatalf("automigrate: %v", err) + } + + if err := db.Exec( + "TRUNCATE ledger_entries, idempotency_records, transfers, wallets RESTART IDENTITY CASCADE", + ).Error; err != nil { + t.Fatalf("truncate: %v", err) + } + + return db +} + +// newService wires the real repositories and transfer service against db. +func newService(db *gorm.DB) service.TransferService { + return service.NewTransferService( + db, + pgrepo.NewWalletRepository(db), + pgrepo.NewTransferRepository(db), + pgrepo.NewLedgerRepository(db), + pgrepo.NewIdempotencyRepository(db), + logging.NewLogger(logging.LevelError, nil), + ) +} + +func seedWallet(t *testing.T, db *gorm.DB, id string, balance int64) { + t.Helper() + if err := db.Create(&model.Wallet{WalletID: id, Balance: balance}).Error; err != nil { + t.Fatalf("seed wallet %s: %v", id, err) + } +} + +func walletBalance(t *testing.T, db *gorm.DB, id string) int64 { + t.Helper() + var wallet model.Wallet + if err := db.Where("wallet_id = ?", id).First(&wallet).Error; err != nil { + t.Fatalf("read wallet %s: %v", id, err) + } + return wallet.Balance +} + +func ledgerSum(t *testing.T, db *gorm.DB, entryType model.EntryType) int64 { + t.Helper() + var total int64 + if err := db.Model(&model.LedgerEntry{}). + Where("type = ?", entryType). + Select("COALESCE(SUM(amount), 0)"). + Scan(&total).Error; err != nil { + t.Fatalf("sum %s ledger entries: %v", entryType, err) + } + return total +} + +func countRows(t *testing.T, db *gorm.DB, dest any, query string, args ...any) int64 { + t.Helper() + var count int64 + if err := db.Model(dest).Where(query, args...).Count(&count).Error; err != nil { + t.Fatalf("count rows: %v", err) + } + return count +} + +// TestConcurrentDistinctTransfersNoDoubleSpend fires many concurrent transfers +// (each with a unique idempotency key) that together demand far more than the +// source wallet holds. Only the funded fraction may succeed; the rest must fail +// safely. The source balance must never go negative and the ledger must balance. +func TestConcurrentDistinctTransfersNoDoubleSpend(t *testing.T) { + db := openDB(t) + svc := newService(db) + + const ( + amount = 100 + concurrency = 50 + initialBalance = 1000 // funds exactly 10 of the 50 transfers + ) + expectedProcessed := int64(initialBalance / amount) + + seedWallet(t, db, "src", initialBalance) + seedWallet(t, db, "dst", 0) + + var ( + wg sync.WaitGroup + mu sync.Mutex + processed int + failed int + ) + for i := 0; i < concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + resp, err := svc.CreateTransfer(context.Background(), service.CreateTransferRequest{ + IdempotencyKey: "distinct-" + model.NewID(), + FromWalletID: "src", + ToWalletID: "dst", + Amount: amount, + }) + mu.Lock() + defer mu.Unlock() + switch { + case err == nil && resp.State == model.TransferStateProcessed: + processed++ + case err == service.ErrInsufficientFunds: + failed++ + default: + t.Errorf("unexpected transfer outcome: state=%s err=%v", resp.State, err) + } + }() + } + wg.Wait() + + if int64(processed) != expectedProcessed { + t.Fatalf("expected %d processed transfers, got %d (failed=%d)", expectedProcessed, processed, failed) + } + + // No double spend: the source drained by exactly the processed amount and + // never went negative; the destination received exactly that amount. + srcBalance := walletBalance(t, db, "src") + dstBalance := walletBalance(t, db, "dst") + if srcBalance < 0 { + t.Fatalf("source balance went negative: %d", srcBalance) + } + if wantSrc := initialBalance - expectedProcessed*amount; srcBalance != wantSrc { + t.Fatalf("source balance: got %d, want %d", srcBalance, wantSrc) + } + if wantDst := expectedProcessed * amount; dstBalance != wantDst { + t.Fatalf("destination balance: got %d, want %d", dstBalance, wantDst) + } + + // The ledger always balances: total DEBIT == total CREDIT, and there are + // exactly two entries per processed transfer. + debits := ledgerSum(t, db, model.EntryTypeDebit) + credits := ledgerSum(t, db, model.EntryTypeCredit) + if debits != credits { + t.Fatalf("ledger not balanced: debits=%d credits=%d", debits, credits) + } + if entries := countRows(t, db, &model.LedgerEntry{}, "1 = 1"); entries != expectedProcessed*2 { + t.Fatalf("expected %d ledger entries, got %d", expectedProcessed*2, entries) + } + + // Every non-processed transfer is durably recorded as FAILED (replayable). + if failedRows := countRows(t, db, &model.Transfer{}, "state = ?", model.TransferStateFailed); failedRows != int64(failed) { + t.Fatalf("expected %d FAILED transfer rows, got %d", failed, failedRows) + } +} + +// TestConcurrentDuplicatesSingleSideEffect fires many concurrent requests that +// all share one idempotency key. Exactly one may take effect; every caller must +// observe the same result and the money must move only once. +func TestConcurrentDuplicatesSingleSideEffect(t *testing.T) { + db := openDB(t) + svc := newService(db) + + const ( + amount = 100 + concurrency = 30 + initialBalance = 1000 + key = "shared-idempotency-key" + ) + + seedWallet(t, db, "src", initialBalance) + seedWallet(t, db, "dst", 0) + + var ( + wg sync.WaitGroup + mu sync.Mutex + transferIDs = map[string]struct{}{} + ) + for i := 0; i < concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + resp, err := svc.CreateTransfer(context.Background(), service.CreateTransferRequest{ + IdempotencyKey: key, + FromWalletID: "src", + ToWalletID: "dst", + Amount: amount, + }) + if err != nil { + t.Errorf("duplicate request returned error: %v", err) + return + } + mu.Lock() + transferIDs[resp.TransferID] = struct{}{} + mu.Unlock() + }() + } + wg.Wait() + + // Every caller saw the same transfer id (one logical result). + if len(transferIDs) != 1 { + t.Fatalf("expected all callers to observe one transfer id, got %d distinct ids", len(transferIDs)) + } + + // The money moved exactly once, regardless of the number of duplicates. + if got, want := walletBalance(t, db, "src"), int64(initialBalance-amount); got != want { + t.Fatalf("source balance: got %d, want %d (money moved more than once)", got, want) + } + if got, want := walletBalance(t, db, "dst"), int64(amount); got != want { + t.Fatalf("destination balance: got %d, want %d", got, want) + } + + // Exactly one transfer, one idempotency record, and two ledger entries exist. + if n := countRows(t, db, &model.Transfer{}, "1 = 1"); n != 1 { + t.Fatalf("expected 1 transfer row, got %d", n) + } + if n := countRows(t, db, &model.IdempotencyRecord{}, "1 = 1"); n != 1 { + t.Fatalf("expected 1 idempotency record, got %d", n) + } + if n := countRows(t, db, &model.LedgerEntry{}, "1 = 1"); n != 2 { + t.Fatalf("expected 2 ledger entries, got %d", n) + } + + debits := ledgerSum(t, db, model.EntryTypeDebit) + credits := ledgerSum(t, db, model.EntryTypeCredit) + if debits != credits || debits != amount { + t.Fatalf("ledger not balanced: debits=%d credits=%d", debits, credits) + } +} diff --git a/testenv/docker-compose.yml b/testenv/docker-compose.yml new file mode 100644 index 00000000..f6aa29a3 --- /dev/null +++ b/testenv/docker-compose.yml @@ -0,0 +1,22 @@ +services: + postgres: + image: postgres:16-alpine + container_name: wallet_pg + environment: + POSTGRES_USER: wallet + POSTGRES_PASSWORD: wallet + POSTGRES_DB: wallet + ports: + - "5432:5432" + # Scripts in /docker-entrypoint-initdb.d/ are executed by the Postgres image + # (in filename order) the first time the data directory is initialized. This + # applies the authoritative schema as part of database provisioning, so the + # service never creates its own tables. In production the same schema.sql is + # applied by your DB provisioning/migration step; this mirrors it locally. + volumes: + - ../migrations/schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro + healthcheck: + test: ["CMD-SHELL", "pg_isready -U wallet -d wallet"] + interval: 2s + timeout: 3s + retries: 30 diff --git a/testenv/run-local.sh b/testenv/run-local.sh new file mode 100644 index 00000000..613157e0 --- /dev/null +++ b/testenv/run-local.sh @@ -0,0 +1,56 @@ +#!/usr/bin/env bash +# +# Run the service against a REAL Postgres. +# +# What it does: +# 1. starts Postgres in Docker (schema.sql is applied automatically by the +# Postgres image via /docker-entrypoint-initdb.d/ on first init) +# 2. waits for the schema to be ready +# 3. seeds two wallets (wallet_1=1000, wallet_2=0) +# 4. starts the service in the foreground on :8080 +# +# Postgres (and this script's foreground service) are torn down on Ctrl+C. +# +# Usage: +# cd testenv && ./run-local.sh +# +# Then, from another shell: +# curl -s http://localhost:8080/transfers \ +# -H 'Content-Type: application/json' \ +# -d '{"idempotencyKey":"demo-1","fromWalletId":"wallet_1","toWalletId":"wallet_2","amount":100}' +# +# Requirements: docker (with compose plugin), go. + +set -euo pipefail +cd "$(dirname "$0")" + +export DATABASE_URL="postgres://wallet:wallet@host.docker.internal:5432/wallet?sslmode=disable" +export LISTEN_ADDR=":8080" +export LOG_LEVEL="INFO" + +cleanup() { + docker compose down -v >/dev/null 2>&1 || true +} +trap cleanup EXIT + +psql_exec() { docker compose exec -T postgres psql -U wallet -d wallet "$@"; } + +echo "==> [1/4] starting postgres (applies schema.sql via init scripts)" +docker compose up -d +until [ "$(docker inspect -f '{{.State.Health.Status}}' wallet_pg 2>/dev/null)" = "healthy" ]; do + sleep 1 +done + +echo "==> [2/4] waiting for schema to be applied" +# pg_isready can report healthy while init scripts are still running, so wait +# until the schema (the wallets table) actually exists before seeding. +until psql_exec -tAc "SELECT to_regclass('public.wallets') IS NOT NULL" 2>/dev/null | grep -q '^t$'; do + sleep 1 +done + +echo "==> [3/4] seeding wallets (wallet_1=1000, wallet_2=0)" +psql_exec < seed.sql >/dev/null + +echo "==> [4/4] starting service on ${LISTEN_ADDR} (Ctrl+C to stop and tear down)" +cd .. +exec go run ./cmd/server diff --git a/testenv/seed.sql b/testenv/seed.sql new file mode 100644 index 00000000..8e9f9121 --- /dev/null +++ b/testenv/seed.sql @@ -0,0 +1,13 @@ +-- Seed data for local endpoint checks. +-- Run AFTER the service has started once, because the service creates the tables +-- via GORM AutoMigrate on startup. +-- +-- wallet_1 starts funded, wallet_2 starts empty. + +INSERT INTO wallets (wallet_id, balance, created_at, updated_at) +VALUES + ('wallet_1', 1000, NOW(), NOW()), + ('wallet_2', 0, NOW(), NOW()) +ON CONFLICT (wallet_id) DO UPDATE + SET balance = EXCLUDED.balance, + updated_at = NOW();