diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a9f40283..c8c9605a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,10 +14,7 @@ jobs: name: lint-format-test runs-on: [actions_runner_dev_new] env: - LINT_CMD: ${{ vars.LINT_CMD }} - FORMAT_CHECK_CMD: ${{ vars.FORMAT_CHECK_CMD }} - TEST_CMD: ${{ vars.TEST_CMD }} - CGO_ENABLED: ${{ vars.CGO_ENABLED }} + CGO_ENABLED: 1 steps: - name: Checkout uses: actions/checkout@v4 @@ -33,32 +30,20 @@ jobs: with: go-version: '1.24' - - name: Install golangci-lint - run: | - curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $HOME/.local/bin v1.64.8 echo "$HOME/.local/bin" >> $GITHUB_PATH - - - name: Validate commands configured - shell: bash - run: | - set -euo pipefail - test -n "${LINT_CMD:-}" || { echo "Missing repository variable LINT_CMD"; exit 1; } - test -n "${FORMAT_CHECK_CMD:-}" || { echo "Missing repository variable FORMAT_CHECK_CMD"; exit 1; } - test -n "${TEST_CMD:-}" || { echo "Missing repository variable TEST_CMD"; exit 1; } - - name: Run lint shell: bash run: | set -euo pipefail - eval "$LINT_CMD" + go vet ./... - name: Run format check shell: bash run: | set -euo pipefail - eval "$FORMAT_CHECK_CMD" + test -z "$(gofmt -l .)" - name: Run tests shell: bash run: | set -euo pipefail - eval "$TEST_CMD" + go test ./... diff --git a/AI_USAGE.md b/AI_USAGE.md new file mode 100644 index 00000000..3d1100eb --- /dev/null +++ b/AI_USAGE.md @@ -0,0 +1,12 @@ +# AI Usage Note + +Tool used: OpenAI Codex in the Codex desktop app. + +How I used it: I used Codex as a pair-programming assistant to inspect the assignment, scaffold the Go project, implement the layered service, write behavioral tests, and prepare documentation. I reviewed the generated design and code decisions while keeping the implementation focused on the assignment requirements. + +Prompts used in this session: + +1. "connect with my github" +2. "we will do this in go make branch as Wallet Transfer Assignment Repository ... Your PR branch should be named: solution/ (e.g., solution/jane-doe)." + +Transcript note: The full interactive transcript is available in the Codex desktop thread for this work. If a plain-text transcript is required outside Codex, export this thread or provide these prompts with the pull request. diff --git a/README.md b/README.md index 58c62d1a..1e4b27f6 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,68 @@ # Wallet Transfer Assignment Repository -This repository is a reusable coding assignment template for evaluating backend engineers on wallet transfers, idempotency, concurrency control, and double-entry ledger design. +This branch contains a Go implementation of the wallet transfer assignment. It provides a small HTTP service with GORM-backed SQLite persistence, idempotent transfer creation, stored wallet balances, safe transfer state transitions, and double-entry ledger entries. + +## Running locally + +Requirements: + +- Go 1.24 +- CGO-enabled SQLite build support. On Linux, install `gcc`; on macOS, Xcode command line tools are enough. + +Run tests: + +```sh +go test ./... +``` + +Run the service: + +```sh +go run ./cmd/server +``` + +By default the service listens on `:8080` and uses `wallets.db`. Override with: + +```sh +HTTP_ADDR=:9090 DB_PATH=/tmp/wallets.db go run ./cmd/server +``` + +Seed wallets are created if absent: + +- `wallet_1` with balance `1000` +- `wallet_2` with balance `1000` +- `wallet_3` with balance `1000` + +Create a transfer: + +```sh +curl -X POST http://localhost:8080/transfers \ + -H 'Content-Type: application/json' \ + -d '{"idempotencyKey":"abc123","fromWalletId":"wallet_1","toWalletId":"wallet_2","amount":100}' +``` + +Read data: + +```sh +curl http://localhost:8080/wallets/wallet_1 +curl http://localhost:8080/transfers/ +``` + +## Implementation notes + +- Architecture is split into handler, service, repository, domain, and port packages. +- GORM repository models are used with SQLite persistence to keep the assignment self-contained. +- Transfers are executed inside a database transaction. +- Idempotency records store the key, canonical request hash, and original transfer id. +- Duplicate requests with the same key and payload return the original result without side effects. +- Duplicate requests with the same key and different payload return a conflict. +- Wallet debits use a conditional update to prevent overspending. +- Processed transfers write exactly two ledger entries: one `DEBIT` and one `CREDIT`. +- Failed business transfers are durable and do not change balances or write ledger entries. + +See [docs/DESIGN.md](./docs/DESIGN.md) for the detailed design, failure modes, and testing strategy. + +## Original template content ## Included diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 00000000..60896123 --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,63 @@ +package main + +import ( + "context" + "log" + "net/http" + "os" + "time" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/handler" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/repository/sqlite" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/service" + gormsqlite "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func main() { + dbPath := getenv("DB_PATH", "wallets.db") + addr := getenv("HTTP_ADDR", ":8080") + + db, err := gorm.Open(gormsqlite.Open(sqlite.DSN(dbPath)), &gorm.Config{}) + if err != nil { + log.Fatalf("open database: %v", err) + } + sqlDB, err := db.DB() + if err != nil { + log.Fatalf("get database handle: %v", err) + } + defer sqlDB.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + store := sqlite.NewStore(db) + if err := store.Migrate(ctx); err != nil { + log.Fatalf("migrate database: %v", err) + } + + if err := store.SeedWallets(ctx, map[string]int64{ + "wallet_1": 1000, + "wallet_2": 1000, + "wallet_3": 1000, + }); err != nil { + log.Fatalf("seed wallets: %v", err) + } + + svc := service.NewTransferService(store) + mux := http.NewServeMux() + handler.RegisterRoutes(mux, svc) + + log.Printf("wallet transfer service listening on %s", addr) + if err := http.ListenAndServe(addr, mux); err != nil { + log.Fatalf("serve: %v", err) + } +} + +func getenv(key, fallback string) string { + value := os.Getenv(key) + if value == "" { + return fallback + } + return value +} diff --git a/docs/DESIGN.md b/docs/DESIGN.md new file mode 100644 index 00000000..f44d3bf6 --- /dev/null +++ b/docs/DESIGN.md @@ -0,0 +1,96 @@ +# Wallet Transfer Service Design + +## Problem statement + +The service exposes wallet-to-wallet transfers with API-level exactly-once behavior for a caller-supplied `idempotencyKey`. A successful transfer debits one wallet, credits another wallet, and records a balanced double-entry ledger. A failed business transfer is recorded as `FAILED` without partial balance or ledger side effects. + +## API contract + +### `POST /transfers` + +Request: + +```json +{ + "idempotencyKey": "abc123", + "fromWalletId": "wallet_1", + "toWalletId": "wallet_2", + "amount": 100 +} +``` + +Responses: + +- `201 Created` for a newly processed transfer. +- `200 OK` for a replay of the same idempotency key and identical payload. +- `409 Conflict` if an existing idempotency key is reused with a different payload. +- `422 Unprocessable Entity` for a recorded business failure such as insufficient funds. +- `400 Bad Request` for validation errors. + +The response includes transfer state, ledger entries for processed transfers, and a `replay` flag. + +### Optional read APIs + +- `GET /wallets/{id}` returns a wallet and its stored balance. +- `GET /transfers/{id}` returns the transfer and its ledger entries. +- `GET /healthz` returns service health. + +## Persistence + +SQLite is used for local portability, with GORM repository models and `AutoMigrate` owning table creation. The schema has: + +- `wallets`: stored balance with `CHECK (balance >= 0)`. +- `transfers`: immutable transfer request data, state, and failure reason. +- `ledger_entries`: double-entry rows with `DEBIT` and `CREDIT` types and a uniqueness constraint on `(transfer_id, type)`. +- `idempotency_records`: unique idempotency key, canonical request hash, and transfer id. + +Amounts are represented as integer minor units. Floating point money is intentionally avoided. + +## Idempotency behavior + +The idempotency key is inserted in the same database transaction as the transfer. The service stores a SHA-256 hash of the canonical request payload excluding the idempotency key. On duplicate requests: + +- Same key and same payload returns the original transfer result. +- Same key and different payload returns `409 Conflict`. +- Duplicate requests never create additional transfers or ledger entries. + +## Transaction and concurrency strategy + +Each transfer runs in a serializable GORM transaction. The SQLite store uses a single open database connection and a busy timeout, so writes are serialized and duplicate/concurrent requests wait for the active writer instead of observing partial state. + +The debit operation uses a guarded update: + +```sql +UPDATE wallets +SET balance = balance - ? +WHERE id = ? AND balance >= ? +``` + +This is the final no-overspend guard. If it affects no rows, the transfer is marked `FAILED` and no ledger entry is written. + +## State transitions + +Transfers start as `PENDING` and can only move to `PROCESSED` or `FAILED`. Repository updates enforce this with `WHERE state = 'PENDING'`, preventing retries from mutating terminal transfers. + +## Failure modes + +- Validation errors are rejected before opening the transaction. +- Missing wallets and insufficient funds become durable `FAILED` transfer records. +- Unexpected database errors roll back the whole transaction. +- Processed transfers always have exactly two ledger entries: one debit and one credit. + +## Observability + +The service logs startup configuration and exposes `/healthz`. In a larger production system I would add request IDs, structured logs, metrics for transfer outcomes, and latency histograms. + +## Testing strategy + +Tests use a real temporary SQLite database and verify: + +- successful transfer execution, +- stored balances, +- double-entry ledger balancing, +- idempotent replay, +- idempotency key conflict, +- insufficient funds failure without side effects, +- concurrent transfers from the same wallet without overspending. diff --git a/go.mod b/go.mod new file mode 100644 index 00000000..b883871f --- /dev/null +++ b/go.mod @@ -0,0 +1,8 @@ +module github.com/vardaanmittal09/wallet-transfer-assignment + +go 1.24 + +require ( + gorm.io/driver/sqlite v1.6.0 + gorm.io/gorm v1.30.0 +) diff --git a/internal/domain/errors.go b/internal/domain/errors.go new file mode 100644 index 00000000..f6edd135 --- /dev/null +++ b/internal/domain/errors.go @@ -0,0 +1,15 @@ +package domain + +import "errors" + +var ( + ErrAmountMustBePositive = errors.New("amount must be positive") + ErrIdempotencyKeyRequired = errors.New("idempotency key is required") + ErrWalletIDRequired = errors.New("wallet ids are required") + ErrSameWalletTransfer = errors.New("source and destination wallets must differ") + ErrIdempotencyConflict = errors.New("idempotency key was already used with a different request") + ErrWalletNotFound = errors.New("wallet not found") + ErrInsufficientFunds = errors.New("insufficient funds") + ErrTransferNotFound = errors.New("transfer not found") + ErrInvalidTransferTransition = errors.New("invalid transfer state transition") +) diff --git a/internal/domain/models.go b/internal/domain/models.go new file mode 100644 index 00000000..555180af --- /dev/null +++ b/internal/domain/models.go @@ -0,0 +1,60 @@ +package domain + +import "time" + +type TransferState string + +const ( + TransferPending TransferState = "PENDING" + TransferProcessed TransferState = "PROCESSED" + TransferFailed TransferState = "FAILED" +) + +type LedgerEntryType string + +const ( + LedgerDebit LedgerEntryType = "DEBIT" + LedgerCredit LedgerEntryType = "CREDIT" +) + +type Transfer struct { + ID string + IdempotencyKey string + FromWalletID string + ToWalletID string + Amount int64 + State TransferState + FailureReason string + CreatedAt time.Time + UpdatedAt time.Time +} + +type Wallet struct { + ID string + Balance int64 + CreatedAt time.Time + UpdatedAt time.Time +} + +type LedgerEntry struct { + ID int64 + WalletID string + TransferID string + Type LedgerEntryType + Amount int64 + CreatedAt time.Time +} + +type IdempotencyRecord struct { + Key string + RequestHash string + TransferID string + CreatedAt time.Time +} + +func (t Transfer) CanTransitionTo(next TransferState) bool { + if t.State != TransferPending { + return false + } + return next == TransferProcessed || next == TransferFailed +} diff --git a/internal/handler/http.go b/internal/handler/http.go new file mode 100644 index 00000000..3fede910 --- /dev/null +++ b/internal/handler/http.go @@ -0,0 +1,103 @@ +package handler + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/domain" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/service" +) + +type transferServiceAdapter struct { + service *service.TransferService +} + +func RegisterRoutes(mux *http.ServeMux, svc *service.TransferService) { + adapter := transferServiceAdapter{service: svc} + mux.HandleFunc("POST /transfers", adapter.createTransfer) + mux.HandleFunc("GET /transfers/", adapter.getTransfer) + mux.HandleFunc("GET /wallets/", adapter.getWallet) + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) + }) +} + +func (a transferServiceAdapter) createTransfer(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + + var input service.CreateTransferInput + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { + writeError(w, http.StatusBadRequest, "invalid JSON request body") + return + } + + result, err := a.service.CreateTransfer(r.Context(), input) + if err != nil { + writeDomainError(w, err) + return + } + status := http.StatusCreated + if result.Replay { + status = http.StatusOK + } + if result.State == domain.TransferFailed { + status = http.StatusUnprocessableEntity + } + writeJSON(w, status, result) +} + +func (a transferServiceAdapter) getTransfer(w http.ResponseWriter, r *http.Request) { + transferID := strings.TrimPrefix(r.URL.Path, "/transfers/") + if transferID == "" { + writeError(w, http.StatusNotFound, "transfer not found") + return + } + result, err := a.service.GetTransfer(r.Context(), transferID) + if err != nil { + writeDomainError(w, err) + return + } + writeJSON(w, http.StatusOK, result) +} + +func (a transferServiceAdapter) getWallet(w http.ResponseWriter, r *http.Request) { + walletID := strings.TrimPrefix(r.URL.Path, "/wallets/") + if walletID == "" { + writeError(w, http.StatusNotFound, "wallet not found") + return + } + wallet, err := a.service.GetWallet(r.Context(), walletID) + if err != nil { + writeDomainError(w, err) + return + } + writeJSON(w, http.StatusOK, wallet) +} + +func writeDomainError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, domain.ErrIdempotencyConflict): + writeError(w, http.StatusConflict, err.Error()) + case errors.Is(err, domain.ErrAmountMustBePositive), + errors.Is(err, domain.ErrIdempotencyKeyRequired), + errors.Is(err, domain.ErrWalletIDRequired), + errors.Is(err, domain.ErrSameWalletTransfer): + writeError(w, http.StatusBadRequest, err.Error()) + case errors.Is(err, domain.ErrWalletNotFound), errors.Is(err, domain.ErrTransferNotFound): + writeError(w, http.StatusNotFound, err.Error()) + default: + writeError(w, http.StatusInternalServerError, "internal server error") + } +} + +func writeError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, map[string]string{"error": message}) +} + +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/internal/port/transfer.go b/internal/port/transfer.go new file mode 100644 index 00000000..ba1aa85d --- /dev/null +++ b/internal/port/transfer.go @@ -0,0 +1,28 @@ +package port + +import ( + "context" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/domain" +) + +type TransferStore interface { + WithinTransfer(ctx context.Context, fn func(ctx context.Context, tx TransferTx) error) error + GetWallet(ctx context.Context, walletID string) (domain.Wallet, error) + GetTransfer(ctx context.Context, transferID string) (domain.Transfer, error) + GetLedgerEntries(ctx context.Context, transferID string) ([]domain.LedgerEntry, error) +} + +type TransferTx interface { + GetIdempotencyRecord(ctx context.Context, key string) (domain.IdempotencyRecord, bool, error) + InsertIdempotencyRecord(ctx context.Context, record domain.IdempotencyRecord) error + InsertTransfer(ctx context.Context, transfer domain.Transfer) error + GetTransfer(ctx context.Context, transferID string) (domain.Transfer, error) + GetWallet(ctx context.Context, walletID string) (domain.Wallet, error) + DebitWallet(ctx context.Context, walletID string, amount int64) (bool, error) + CreditWallet(ctx context.Context, walletID string, amount int64) error + InsertLedgerEntry(ctx context.Context, entry domain.LedgerEntry) error + MarkTransferProcessed(ctx context.Context, transferID string) error + MarkTransferFailed(ctx context.Context, transferID string, reason string) error + GetLedgerEntries(ctx context.Context, transferID string) ([]domain.LedgerEntry, error) +} diff --git a/internal/repository/sqlite/models.go b/internal/repository/sqlite/models.go new file mode 100644 index 00000000..0b674e1c --- /dev/null +++ b/internal/repository/sqlite/models.go @@ -0,0 +1,103 @@ +package sqlite + +import ( + "time" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/domain" +) + +type walletModel struct { + ID string `gorm:"primaryKey;type:text"` + Balance int64 `gorm:"not null;check:balance >= 0"` + CreatedAt time.Time + UpdatedAt time.Time +} + +func (walletModel) TableName() string { + return "wallets" +} + +type transferModel struct { + ID string `gorm:"primaryKey;type:text"` + IdempotencyKey string `gorm:"not null;uniqueIndex;type:text"` + FromWalletID string `gorm:"not null;type:text;check:from_wallet_id <> to_wallet_id"` + ToWalletID string `gorm:"not null;type:text"` + Amount int64 `gorm:"not null;check:amount > 0"` + State string `gorm:"not null;check:state IN ('PENDING','PROCESSED','FAILED')"` + FailureReason string `gorm:"not null;default:''"` + CreatedAt time.Time + UpdatedAt time.Time +} + +func (transferModel) TableName() string { + return "transfers" +} + +type ledgerEntryModel struct { + ID int64 `gorm:"primaryKey;autoIncrement"` + WalletID string `gorm:"not null;type:text;index"` + Wallet walletModel `gorm:"foreignKey:WalletID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:RESTRICT;"` + TransferID string `gorm:"not null;type:text;index;uniqueIndex:idx_ledger_transfer_type"` + Transfer transferModel `gorm:"foreignKey:TransferID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:RESTRICT;"` + Type string `gorm:"not null;type:text;check:type IN ('DEBIT','CREDIT');uniqueIndex:idx_ledger_transfer_type"` + Amount int64 `gorm:"not null;check:amount > 0"` + CreatedAt time.Time +} + +func (ledgerEntryModel) TableName() string { + return "ledger_entries" +} + +type idempotencyRecordModel struct { + Key string `gorm:"primaryKey;type:text"` + RequestHash string `gorm:"not null;type:text"` + TransferID string `gorm:"not null;type:text;uniqueIndex"` + CreatedAt time.Time +} + +func (idempotencyRecordModel) TableName() string { + return "idempotency_records" +} + +func walletToDomain(model walletModel) domain.Wallet { + return domain.Wallet{ + ID: model.ID, + Balance: model.Balance, + CreatedAt: model.CreatedAt, + UpdatedAt: model.UpdatedAt, + } +} + +func transferToDomain(model transferModel) domain.Transfer { + return domain.Transfer{ + ID: model.ID, + IdempotencyKey: model.IdempotencyKey, + FromWalletID: model.FromWalletID, + ToWalletID: model.ToWalletID, + Amount: model.Amount, + State: domain.TransferState(model.State), + FailureReason: model.FailureReason, + CreatedAt: model.CreatedAt, + UpdatedAt: model.UpdatedAt, + } +} + +func ledgerEntryToDomain(model ledgerEntryModel) domain.LedgerEntry { + return domain.LedgerEntry{ + ID: model.ID, + WalletID: model.WalletID, + TransferID: model.TransferID, + Type: domain.LedgerEntryType(model.Type), + Amount: model.Amount, + CreatedAt: model.CreatedAt, + } +} + +func idempotencyRecordToDomain(model idempotencyRecordModel) domain.IdempotencyRecord { + return domain.IdempotencyRecord{ + Key: model.Key, + RequestHash: model.RequestHash, + TransferID: model.TransferID, + CreatedAt: model.CreatedAt, + } +} diff --git a/internal/repository/sqlite/store.go b/internal/repository/sqlite/store.go new file mode 100644 index 00000000..4a4545ec --- /dev/null +++ b/internal/repository/sqlite/store.go @@ -0,0 +1,230 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/domain" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/port" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +type Store struct { + db *gorm.DB +} + +type Tx struct { + db *gorm.DB +} + +func NewStore(db *gorm.DB) *Store { + if sqlDB, err := db.DB(); err == nil { + sqlDB.SetMaxOpenConns(1) + } + return &Store{db: db} +} + +func DSN(path string) string { + if strings.Contains(path, "?") { + return path + "&_busy_timeout=5000&_foreign_keys=on" + } + return path + "?_busy_timeout=5000&_foreign_keys=on" +} + +func (s *Store) Migrate(ctx context.Context) error { + if err := s.db.WithContext(ctx).Exec(`PRAGMA foreign_keys = ON`).Error; err != nil { + return fmt.Errorf("enable foreign keys: %w", err) + } + if err := s.db.WithContext(ctx).AutoMigrate( + &walletModel{}, + &transferModel{}, + &ledgerEntryModel{}, + &idempotencyRecordModel{}, + ); err != nil { + return fmt.Errorf("auto migrate: %w", err) + } + return nil +} + +func (s *Store) SeedWallets(ctx context.Context, wallets map[string]int64) error { + for id, balance := range wallets { + err := s.db.WithContext(ctx). + Clauses(clause.OnConflict{DoNothing: true}). + Create(&walletModel{ID: id, Balance: balance}). + Error + if err != nil { + return err + } + } + return nil +} + +func (s *Store) GetWallet(ctx context.Context, walletID string) (domain.Wallet, error) { + var wallet walletModel + if err := s.db.WithContext(ctx).First(&wallet, "id = ?", walletID).Error; err != nil { + return domain.Wallet{}, mapNotFound(err, domain.ErrWalletNotFound) + } + return walletToDomain(wallet), nil +} + +func (s *Store) GetTransfer(ctx context.Context, transferID string) (domain.Transfer, error) { + var transfer transferModel + if err := s.db.WithContext(ctx).First(&transfer, "id = ?", transferID).Error; err != nil { + return domain.Transfer{}, mapNotFound(err, domain.ErrTransferNotFound) + } + return transferToDomain(transfer), nil +} + +func (s *Store) GetLedgerEntries(ctx context.Context, transferID string) ([]domain.LedgerEntry, error) { + return getLedgerEntries(ctx, s.db, transferID) +} + +func (s *Store) WithinTransfer(ctx context.Context, fn func(ctx context.Context, tx port.TransferTx) error) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + return fn(ctx, &Tx{db: tx}) + }, &sql.TxOptions{Isolation: sql.LevelSerializable}) +} + +func (t *Tx) GetIdempotencyRecord(ctx context.Context, key string) (domain.IdempotencyRecord, bool, error) { + var record idempotencyRecordModel + if err := t.db.WithContext(ctx).First(&record, "key = ?", key).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return domain.IdempotencyRecord{}, false, nil + } + return domain.IdempotencyRecord{}, false, err + } + return idempotencyRecordToDomain(record), true, nil +} + +func (t *Tx) InsertIdempotencyRecord(ctx context.Context, record domain.IdempotencyRecord) error { + return t.db.WithContext(ctx).Create(&idempotencyRecordModel{ + Key: record.Key, + RequestHash: record.RequestHash, + TransferID: record.TransferID, + }).Error +} + +func (t *Tx) InsertTransfer(ctx context.Context, transfer domain.Transfer) error { + return t.db.WithContext(ctx).Create(&transferModel{ + ID: transfer.ID, + IdempotencyKey: transfer.IdempotencyKey, + FromWalletID: transfer.FromWalletID, + ToWalletID: transfer.ToWalletID, + Amount: transfer.Amount, + State: string(transfer.State), + }).Error +} + +func (t *Tx) GetTransfer(ctx context.Context, transferID string) (domain.Transfer, error) { + var transfer transferModel + if err := t.db.WithContext(ctx).First(&transfer, "id = ?", transferID).Error; err != nil { + return domain.Transfer{}, mapNotFound(err, domain.ErrTransferNotFound) + } + return transferToDomain(transfer), nil +} + +func (t *Tx) GetWallet(ctx context.Context, walletID string) (domain.Wallet, error) { + var wallet walletModel + if err := t.db.WithContext(ctx).First(&wallet, "id = ?", walletID).Error; err != nil { + return domain.Wallet{}, mapNotFound(err, domain.ErrWalletNotFound) + } + return walletToDomain(wallet), nil +} + +func (t *Tx) DebitWallet(ctx context.Context, walletID string, amount int64) (bool, error) { + result := t.db.WithContext(ctx). + Model(&walletModel{}). + Where("id = ? AND balance >= ?", walletID, amount). + Updates(map[string]any{ + "balance": gorm.Expr("balance - ?", amount), + "updated_at": gorm.Expr("CURRENT_TIMESTAMP"), + }) + return result.RowsAffected == 1, result.Error +} + +func (t *Tx) CreditWallet(ctx context.Context, walletID string, amount int64) error { + result := t.db.WithContext(ctx). + Model(&walletModel{}). + Where("id = ?", walletID). + Updates(map[string]any{ + "balance": gorm.Expr("balance + ?", amount), + "updated_at": gorm.Expr("CURRENT_TIMESTAMP"), + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return domain.ErrWalletNotFound + } + return nil +} + +func (t *Tx) InsertLedgerEntry(ctx context.Context, entry domain.LedgerEntry) error { + return t.db.WithContext(ctx).Create(&ledgerEntryModel{ + WalletID: entry.WalletID, + TransferID: entry.TransferID, + Type: string(entry.Type), + Amount: entry.Amount, + }).Error +} + +func (t *Tx) MarkTransferProcessed(ctx context.Context, transferID string) error { + result := t.db.WithContext(ctx). + Model(&transferModel{}). + Where("id = ? AND state = ?", transferID, domain.TransferPending). + Updates(map[string]any{ + "state": string(domain.TransferProcessed), + "updated_at": gorm.Expr("CURRENT_TIMESTAMP"), + }) + return requireOneRow(result, domain.ErrInvalidTransferTransition) +} + +func (t *Tx) MarkTransferFailed(ctx context.Context, transferID string, reason string) error { + result := t.db.WithContext(ctx). + Model(&transferModel{}). + Where("id = ? AND state = ?", transferID, domain.TransferPending). + Updates(map[string]any{ + "state": string(domain.TransferFailed), + "failure_reason": reason, + "updated_at": gorm.Expr("CURRENT_TIMESTAMP"), + }) + return requireOneRow(result, domain.ErrInvalidTransferTransition) +} + +func (t *Tx) GetLedgerEntries(ctx context.Context, transferID string) ([]domain.LedgerEntry, error) { + return getLedgerEntries(ctx, t.db, transferID) +} + +func getLedgerEntries(ctx context.Context, db *gorm.DB, transferID string) ([]domain.LedgerEntry, error) { + var models []ledgerEntryModel + if err := db.WithContext(ctx).Where("transfer_id = ?", transferID).Order("id").Find(&models).Error; err != nil { + return nil, err + } + + entries := make([]domain.LedgerEntry, 0, len(models)) + for _, model := range models { + entries = append(entries, ledgerEntryToDomain(model)) + } + return entries, nil +} + +func mapNotFound(err error, notFound error) error { + if errors.Is(err, gorm.ErrRecordNotFound) { + return notFound + } + return err +} + +func requireOneRow(result *gorm.DB, errWhenMissing error) error { + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return errWhenMissing + } + return nil +} diff --git a/internal/service/transfer.go b/internal/service/transfer.go new file mode 100644 index 00000000..98d2735f --- /dev/null +++ b/internal/service/transfer.go @@ -0,0 +1,249 @@ +package service + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/domain" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/port" +) + +type CreateTransferInput struct { + IdempotencyKey string `json:"idempotencyKey"` + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` +} + +type TransferResult struct { + ID string `json:"id"` + IdempotencyKey string `json:"idempotencyKey"` + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` + State domain.TransferState `json:"state"` + FailureReason string `json:"failureReason,omitempty"` + LedgerEntries []domain.LedgerEntry `json:"ledgerEntries,omitempty"` + WalletBalances map[string]int64 `json:"walletBalances,omitempty"` + Replay bool `json:"replay"` +} + +type TransferService struct { + store port.TransferStore +} + +func NewTransferService(store port.TransferStore) *TransferService { + return &TransferService{store: store} +} + +func (s *TransferService) CreateTransfer(ctx context.Context, input CreateTransferInput) (TransferResult, error) { + input = normalize(input) + if err := validate(input); err != nil { + return TransferResult{}, err + } + + requestHash, err := hashRequest(input) + if err != nil { + return TransferResult{}, err + } + + var result TransferResult + err = s.store.WithinTransfer(ctx, func(ctx context.Context, tx port.TransferTx) error { + record, exists, err := tx.GetIdempotencyRecord(ctx, input.IdempotencyKey) + if err != nil { + return err + } + if exists { + if record.RequestHash != requestHash { + return domain.ErrIdempotencyConflict + } + transfer, err := tx.GetTransfer(ctx, record.TransferID) + if err != nil { + return err + } + ledger, err := tx.GetLedgerEntries(ctx, record.TransferID) + if err != nil { + return err + } + result = transferResult(transfer, ledger, nil, true) + return nil + } + + transfer := domain.Transfer{ + ID: newID("tr"), + IdempotencyKey: input.IdempotencyKey, + FromWalletID: input.FromWalletID, + ToWalletID: input.ToWalletID, + Amount: input.Amount, + State: domain.TransferPending, + } + if err := tx.InsertTransfer(ctx, transfer); err != nil { + return err + } + if err := tx.InsertIdempotencyRecord(ctx, domain.IdempotencyRecord{ + Key: input.IdempotencyKey, + RequestHash: requestHash, + TransferID: transfer.ID, + }); err != nil { + return err + } + + fromWallet, err := tx.GetWallet(ctx, input.FromWalletID) + if err != nil { + if errors.Is(err, domain.ErrWalletNotFound) { + return failTransfer(ctx, tx, transfer, "source wallet not found", &result) + } + return err + } + toWallet, err := tx.GetWallet(ctx, input.ToWalletID) + if err != nil { + if errors.Is(err, domain.ErrWalletNotFound) { + return failTransfer(ctx, tx, transfer, "destination wallet not found", &result) + } + return err + } + + ok, err := tx.DebitWallet(ctx, input.FromWalletID, input.Amount) + if err != nil { + return err + } + if !ok { + return failTransfer(ctx, tx, transfer, domain.ErrInsufficientFunds.Error(), &result) + } + if err := tx.CreditWallet(ctx, input.ToWalletID, input.Amount); err != nil { + return err + } + + debit := domain.LedgerEntry{ + WalletID: input.FromWalletID, + TransferID: transfer.ID, + Type: domain.LedgerDebit, + Amount: input.Amount, + } + credit := domain.LedgerEntry{ + WalletID: input.ToWalletID, + TransferID: transfer.ID, + Type: domain.LedgerCredit, + Amount: input.Amount, + } + if err := tx.InsertLedgerEntry(ctx, debit); err != nil { + return err + } + if err := tx.InsertLedgerEntry(ctx, credit); err != nil { + return err + } + if err := tx.MarkTransferProcessed(ctx, transfer.ID); err != nil { + return err + } + + ledger, err := tx.GetLedgerEntries(ctx, transfer.ID) + if err != nil { + return err + } + balances := map[string]int64{ + input.FromWalletID: fromWallet.Balance - input.Amount, + input.ToWalletID: toWallet.Balance + input.Amount, + } + transfer.State = domain.TransferProcessed + result = transferResult(transfer, ledger, balances, false) + return nil + }) + + return result, err +} + +func (s *TransferService) GetWallet(ctx context.Context, walletID string) (domain.Wallet, error) { + return s.store.GetWallet(ctx, walletID) +} + +func (s *TransferService) GetTransfer(ctx context.Context, transferID string) (TransferResult, error) { + transfer, err := s.store.GetTransfer(ctx, transferID) + if err != nil { + return TransferResult{}, err + } + ledger, err := s.store.GetLedgerEntries(ctx, transferID) + if err != nil { + return TransferResult{}, err + } + return transferResult(transfer, ledger, nil, false), nil +} + +func normalize(input CreateTransferInput) CreateTransferInput { + input.IdempotencyKey = strings.TrimSpace(input.IdempotencyKey) + input.FromWalletID = strings.TrimSpace(input.FromWalletID) + input.ToWalletID = strings.TrimSpace(input.ToWalletID) + return input +} + +func validate(input CreateTransferInput) error { + if input.IdempotencyKey == "" { + return domain.ErrIdempotencyKeyRequired + } + if input.FromWalletID == "" || input.ToWalletID == "" { + return domain.ErrWalletIDRequired + } + if input.FromWalletID == input.ToWalletID { + return domain.ErrSameWalletTransfer + } + if input.Amount <= 0 { + return domain.ErrAmountMustBePositive + } + return nil +} + +func hashRequest(input CreateTransferInput) (string, error) { + payload := struct { + FromWalletID string `json:"fromWalletId"` + ToWalletID string `json:"toWalletId"` + Amount int64 `json:"amount"` + }{ + FromWalletID: input.FromWalletID, + ToWalletID: input.ToWalletID, + Amount: input.Amount, + } + body, err := json.Marshal(payload) + if err != nil { + return "", err + } + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]), nil +} + +func failTransfer(ctx context.Context, tx port.TransferTx, transfer domain.Transfer, reason string, result *TransferResult) error { + if err := tx.MarkTransferFailed(ctx, transfer.ID, reason); err != nil { + return err + } + transfer.State = domain.TransferFailed + transfer.FailureReason = reason + *result = transferResult(transfer, nil, nil, false) + return nil +} + +func transferResult(transfer domain.Transfer, ledger []domain.LedgerEntry, balances map[string]int64, replay bool) TransferResult { + return TransferResult{ + ID: transfer.ID, + IdempotencyKey: transfer.IdempotencyKey, + FromWalletID: transfer.FromWalletID, + ToWalletID: transfer.ToWalletID, + Amount: transfer.Amount, + State: transfer.State, + FailureReason: transfer.FailureReason, + LedgerEntries: ledger, + WalletBalances: balances, + Replay: replay, + } +} + +func newID(prefix string) string { + var bytes [16]byte + if _, err := rand.Read(bytes[:]); err != nil { + panic(fmt.Sprintf("generate id: %v", err)) + } + return prefix + "_" + hex.EncodeToString(bytes[:]) +} diff --git a/internal/service/transfer_test.go b/internal/service/transfer_test.go new file mode 100644 index 00000000..6880e944 --- /dev/null +++ b/internal/service/transfer_test.go @@ -0,0 +1,247 @@ +package service_test + +import ( + "context" + "errors" + "fmt" + "path/filepath" + "sync" + "testing" + + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/domain" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/repository/sqlite" + "github.com/vardaanmittal09/wallet-transfer-assignment/internal/service" + gormsqlite "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestCreateTransferProcessesLedgerAndBalances(t *testing.T) { + ctx := context.Background() + db, svc := newTestService(t, map[string]int64{"wallet_1": 500, "wallet_2": 25}) + + result, err := svc.CreateTransfer(ctx, service.CreateTransferInput{ + IdempotencyKey: "key-1", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 125, + }) + if err != nil { + t.Fatalf("create transfer: %v", err) + } + + if result.State != domain.TransferProcessed { + t.Fatalf("state = %s, want %s", result.State, domain.TransferProcessed) + } + if len(result.LedgerEntries) != 2 { + t.Fatalf("ledger entries = %d, want 2", len(result.LedgerEntries)) + } + assertWalletBalance(t, db, "wallet_1", 375) + assertWalletBalance(t, db, "wallet_2", 150) + assertBalancedLedger(t, result.LedgerEntries, 125) +} + +func TestCreateTransferIsIdempotent(t *testing.T) { + ctx := context.Background() + db, svc := newTestService(t, map[string]int64{"wallet_1": 500, "wallet_2": 25}) + input := service.CreateTransferInput{ + IdempotencyKey: "same-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 125, + } + + first, err := svc.CreateTransfer(ctx, input) + if err != nil { + t.Fatalf("first create: %v", err) + } + second, err := svc.CreateTransfer(ctx, input) + if err != nil { + t.Fatalf("second create: %v", err) + } + + if !second.Replay { + t.Fatal("second result should be marked as replay") + } + if first.ID != second.ID { + t.Fatalf("transfer ids differ: %s != %s", first.ID, second.ID) + } + assertWalletBalance(t, db, "wallet_1", 375) + assertWalletBalance(t, db, "wallet_2", 150) + assertTableCount(t, db, "transfers", 1) + assertTableCount(t, db, "ledger_entries", 2) +} + +func TestIdempotencyKeyRejectsDifferentPayload(t *testing.T) { + ctx := context.Background() + _, svc := newTestService(t, map[string]int64{"wallet_1": 500, "wallet_2": 25, "wallet_3": 10}) + + _, err := svc.CreateTransfer(ctx, service.CreateTransferInput{ + IdempotencyKey: "same-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 125, + }) + if err != nil { + t.Fatalf("first create: %v", err) + } + + _, err = svc.CreateTransfer(ctx, service.CreateTransferInput{ + IdempotencyKey: "same-key", + FromWalletID: "wallet_1", + ToWalletID: "wallet_3", + Amount: 125, + }) + if !errors.Is(err, domain.ErrIdempotencyConflict) { + t.Fatalf("err = %v, want %v", err, domain.ErrIdempotencyConflict) + } +} + +func TestInsufficientFundsCreatesFailedTransferWithoutLedgerSideEffects(t *testing.T) { + ctx := context.Background() + db, svc := newTestService(t, map[string]int64{"wallet_1": 100, "wallet_2": 25}) + + result, err := svc.CreateTransfer(ctx, service.CreateTransferInput{ + IdempotencyKey: "too-large", + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: 125, + }) + if err != nil { + t.Fatalf("create transfer: %v", err) + } + + if result.State != domain.TransferFailed { + t.Fatalf("state = %s, want %s", result.State, domain.TransferFailed) + } + if result.FailureReason != domain.ErrInsufficientFunds.Error() { + t.Fatalf("failure reason = %q", result.FailureReason) + } + assertWalletBalance(t, db, "wallet_1", 100) + assertWalletBalance(t, db, "wallet_2", 25) + assertTableCount(t, db, "ledger_entries", 0) +} + +func TestConcurrentTransfersDoNotOverspend(t *testing.T) { + ctx := context.Background() + db, svc := newTestService(t, map[string]int64{"wallet_1": 1000, "wallet_2": 0}) + + const attempts = 20 + const amount = 75 + + var wg sync.WaitGroup + results := make(chan service.TransferResult, attempts) + errs := make(chan error, attempts) + for i := 0; i < attempts; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + result, err := svc.CreateTransfer(ctx, service.CreateTransferInput{ + IdempotencyKey: fmt.Sprintf("concurrent-%02d", i), + FromWalletID: "wallet_1", + ToWalletID: "wallet_2", + Amount: amount, + }) + if err != nil { + errs <- err + return + } + results <- result + }(i) + } + wg.Wait() + close(results) + close(errs) + + for err := range errs { + t.Fatalf("concurrent create returned error: %v", err) + } + + var processed int + var failed int + for result := range results { + switch result.State { + case domain.TransferProcessed: + processed++ + case domain.TransferFailed: + failed++ + default: + t.Fatalf("unexpected state: %s", result.State) + } + } + + if processed != 13 || failed != 7 { + t.Fatalf("processed=%d failed=%d, want processed=13 failed=7", processed, failed) + } + assertWalletBalance(t, db, "wallet_1", 25) + assertWalletBalance(t, db, "wallet_2", 975) + assertTableCount(t, db, "ledger_entries", processed*2) +} + +func newTestService(t *testing.T, wallets map[string]int64) (*gorm.DB, *service.TransferService) { + t.Helper() + + dbPath := filepath.Join(t.TempDir(), "wallets.db") + db, err := gorm.Open(gormsqlite.Open(sqlite.DSN(dbPath)), &gorm.Config{}) + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { + sqlDB, err := db.DB() + if err == nil { + _ = sqlDB.Close() + } + }) + + store := sqlite.NewStore(db) + if err := store.Migrate(context.Background()); err != nil { + t.Fatalf("migrate: %v", err) + } + if err := store.SeedWallets(context.Background(), wallets); err != nil { + t.Fatalf("seed wallets: %v", err) + } + return db, service.NewTransferService(store) +} + +func assertWalletBalance(t *testing.T, db *gorm.DB, walletID string, want int64) { + t.Helper() + + var got int64 + if err := db.Table("wallets").Select("balance").Where("id = ?", walletID).Scan(&got).Error; err != nil { + t.Fatalf("get wallet balance: %v", err) + } + if got != want { + t.Fatalf("%s balance = %d, want %d", walletID, got, want) + } +} + +func assertTableCount(t *testing.T, db *gorm.DB, table string, want int) { + t.Helper() + + var got int64 + if err := db.Table(table).Count(&got).Error; err != nil { + t.Fatalf("count %s: %v", table, err) + } + if got != int64(want) { + t.Fatalf("%s count = %d, want %d", table, got, want) + } +} + +func assertBalancedLedger(t *testing.T, entries []domain.LedgerEntry, amount int64) { + t.Helper() + + var debit int64 + var credit int64 + for _, entry := range entries { + switch entry.Type { + case domain.LedgerDebit: + debit += entry.Amount + case domain.LedgerCredit: + credit += entry.Amount + default: + t.Fatalf("unexpected ledger entry type: %s", entry.Type) + } + } + if debit != amount || credit != amount { + t.Fatalf("debit=%d credit=%d, want %d each", debit, credit, amount) + } +}