Solution/piyushkumar rajput#111
Open
loopinnovates wants to merge 18 commits into
Open
Conversation
…also refactor boilerplate code
… intergrated that in repository
… handling at middleware + transfer lifecycle management with 2 trnsactions + testcases for concurrency safety
…tion go routine added + test cases for same
…maximum concurrent requests
There was a problem hiding this comment.
Pull request overview
This PR introduces a Go + PostgreSQL wallet transfer service with layered architecture (handlers → service → repositories), a SQL schema for wallets/transfers/ledger entries, and a reconciler to resolve transfers left in PENDING.
Changes:
- Adds Postgres schema/migrations plus sqlc-generated query layer for wallets/transfers/ledger entries.
- Implements transfer execution with wallet row locking, plus a periodic worker to reconcile stuck
PENDINGtransfers. - Adds middleware (panic recovery + concurrency limiting), logging/config/factory wiring, and a broad set of unit + DB-backed concurrency tests.
Reviewed changes
Copilot reviewed 57 out of 61 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| sqlc.yml | Configures sqlc generation and type overrides. |
| repository/postgres/wallet.repo.go | Wallet read repository using sqlc queries. |
| repository/postgres/transfer.repo.go | Transfer repository: insert pending, lock wallets, move funds, ledger insert, status transitions. |
| repository/postgres/transfer_scenarios_test.go | Scenario tests (timeouts, hotspot, balance boundary). |
| repository/postgres/transfer_lifecycle_test.go | Lifecycle tests for stuck pending rows + recon end-to-end. |
| repository/postgres/transfer_concurrency_test.go | Concurrency/idempotency/deadlock tests against Postgres. |
| repository/postgres/sqlcgen/wallet.sql.go | sqlc-generated wallet queries. |
| repository/postgres/sqlcgen/transfer.sql.go | sqlc-generated transfer/ledger queries. |
| repository/postgres/sqlcgen/querier.go | sqlc-generated Querier interface. |
| repository/postgres/sqlcgen/models.go | sqlc-generated DB models. |
| repository/postgres/sqlcgen/db.go | sqlc-generated DBTX plumbing + WithTx helpers. |
| repository/postgres/retry.go | Generic retry wrapper for transient DB errors. |
| repository/postgres/queries/wallet.sql | Wallet SQL source for sqlc. |
| repository/postgres/queries/transfer.sql | Transfer SQL source for sqlc (pending list, insert, finalize, ledger). |
| repository/postgres/postgres.go | DB connection + pool configuration helper. |
| pkg/utils/http.util.go | HTTP response helpers + middleware wrapper helper. |
| pkg/utils/http.util_test.go | Tests for HTTP helper functions. |
| pkg/utils/error.go | CustomError type with status code + message. |
| pkg/utils/error_test.go | Tests for CustomError. |
| pkg/utils/config.util.go | Env helpers for string/int. |
| pkg/utils/config.util_test.go | Tests for env helpers. |
| pkg/logger/logger.go | Zerolog-based global logger with runtime-configurable level. |
| pkg/logger/logger_test.go | Logger init tests. |
| pkg/factory/factory.go | Factory wiring for read/write DB replicas. |
| pkg/construct/errors.go | Maps domain/service errors to HTTP-friendly CustomErrors. |
| pkg/construct/errors_test.go | Tests for error mapping. |
| migrations/0001_init.up.sql | Creates wallets, transfers, ledger_entries + enums + constraints/indexes. |
| migrations/0001_init.down.sql | Drops tables/types for rollback. |
| Makefile | Local dev/test/lint/migrate targets + coverage enforcement. |
| internal/worker/pending_transfer_recon.go | Background reconciler for stuck pending transfers. |
| internal/worker/pending_transfer_recon_test.go | Unit tests for recon behavior (skip overlap, continue on errors). |
| internal/testutils/IWalletSvc.go | Mockery-generated wallet service mock. |
| internal/testutils/IWalletRepository.go | Mockery-generated wallet repo mock. |
| internal/testutils/ITransferRepository.go | Mockery-generated transfer repo mock. |
| internal/service/wallet.svc.go | Wallet service: transfer + idempotent replay + balance + history. |
| internal/service/wallet_test.go | Wallet service unit tests using mocks. |
| internal/router/router.go | Gorilla/mux routes + middleware wiring. |
| internal/middleware/recover.middleware.go | Panic recovery middleware. |
| internal/middleware/recover.middleware_test.go | Tests for recovery middleware. |
| internal/middleware/concurrency_limit.middleware.go | Concurrency limiter middleware. |
| internal/middleware/concurrency_limit.middleware_test.go | Tests for concurrency limiter. |
| internal/handler/wallet.handler.go | HTTP handlers for transfers, balances, and transfer history. |
| internal/handler/wallet_test.go | Handler tests (transfer, balance, history). |
| internal/handler/health.go | Basic health endpoint handler. |
| internal/handler/health_test.go | Health handler test. |
| internal/dto/wallet.go | Request/response DTOs for transfer + wallet APIs. |
| internal/domain/wallet.go | Domain Wallet model. |
| internal/domain/transfer.go | Domain Transfer model + statuses. |
| internal/domain/repository.go | Repository interfaces for wallet/transfer. |
| internal/domain/ledger.go | Domain ledger model + entry type enum. |
| internal/domain/errors.go | Domain error definitions. |
| internal/config/config.go | App config from env, including DB + concurrency limits. |
| internal/config/config.constant.go | Env key constants for config. |
| internal/config/config_test.go | Config tests (defaults + overrides). |
| internal/config/.gitkeep | Keeps config directory in VCS. |
| go.sum | Dependency lockfile updates. |
| go.mod | Module definition + dependencies. |
| docker-compose.yml | Local Postgres container for dev/test. |
| cmd/server/main.go | Server bootstrap: config, logger, DB factory, router, recon worker, graceful shutdown. |
| .mockery.yml | Mockery generation configuration. |
| .github/workflows/ci.yml | CI workflow updates (env command config). |
Files not reviewed (2)
- internal/testutils/ITransferRepository.go: Generated file
- internal/testutils/IWalletRepository.go: Generated file
Confirmed, this is a real gap. Line 41 validates FromWalletID, ToWalletID, and Amount but never checks req.IdempotencyKey. The DB column is idempotency_key TEXT NOT NULL UNIQUE (migrations/0001_init.up.sql:17), so an empty string is a legal value on first insert — the first request with "" succeeds, and every *subsequent* unrelated request that also omits the key would either collide on the unique constraint or (worse, depending on how ExecuteTransfer handles conflicts) get treated as a duplicate of an unrelated transfer, defeating idempotency semantics. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Comment on lines
+30
to
+33
| dbUsername := utils.GetEnvStr(usernameKey, "wallet") | ||
| dbPassword := utils.GetEnvStr(passwordKey, "wallet") | ||
| readPGURI := fmt.Sprintf(utils.GetEnvStr(ReadPGURIKey, "postgres://%s:%s@localhost:5432/wallet_transfer?sslmode=disable"), dbUsername, dbPassword) | ||
| writePGURI := fmt.Sprintf(utils.GetEnvStr(WritePGURIKey, "postgres://%s:%s@localhost:5432/wallet_transfer?sslmode=disable"), dbUsername, dbPassword) |
| @@ -0,0 +1,55 @@ | |||
| .PHONY: run build test test-race lint db-up db-down migrate-up migrate-down test-coverage check-test-coverage-html deploy | |||
|
|
|||
| DATABASE_URL ?= postgres://wallet:<password>@localhost:5432/wallet_transfer?sslmode=disable | |||
Comment on lines
+94
to
+99
| ownerName, balance, err := walletSvc.GetWalletBalance(ctx, walletID) | ||
| if err != nil { | ||
| logger.Warn().Err(err).Str("wallet_id", walletID).Msg("wallet balance request failed") | ||
| utils.WriteError(w, construct.MapTransferError(err)) | ||
| return | ||
| } |
Comment on lines
+139
to
+144
| transfers, limit, offset, err := walletSvc.GetTransferHistory(ctx, walletID, limit, offset) | ||
| if err != nil { | ||
| logger.Warn().Err(err).Str("wallet_id", walletID).Msg("transfer history request failed") | ||
| utils.WriteError(w, construct.MapTransferError(err)) | ||
| return | ||
| } |
Comment on lines
+17
to
+19
| func NewWalletRepository(readDB *sql.DB, writeDB *sql.DB) *WalletRepository { | ||
| return &WalletRepository{readQueries: sqlcgen.New(readDB)} | ||
| } |
Author
There was a problem hiding this comment.
No writes were there so i didn't added writeQueries. But if i add then it will be a dead code
Comment on lines
+30
to
+39
| if err := db.Ping(); err != nil { | ||
| logger.Warn().Err(err).Msg("failed to ping db") | ||
| go func() { | ||
| err := db.Close() | ||
| if err != nil { | ||
| logger.Error().Err(err).Msg("error closing db") | ||
| } | ||
| }() | ||
| return nil, err | ||
| } |
| dbPassword := utils.GetEnvStr(passwordKey, "wallet") | ||
| dbHost := utils.GetEnvStr(hostKey, "localhost") | ||
| dbPort := utils.GetEnvInt(portKey, 5432) | ||
| dbName := utils.GetEnvStr(dbNameKey, "wallet") |
… close synchronously
Comment on lines
+102
to
+110
| cfg := config.AppConfig() | ||
|
|
||
| expectedURI := "postgres://wallet:wallet@localhost:5432/wallet?sslmode=disable" | ||
| if cfg.ReadPGURI != expectedURI { | ||
| t.Errorf("expected default ReadPGURI %q, got %q", expectedURI, cfg.ReadPGURI) | ||
| } | ||
| if cfg.WritePGURI != expectedURI { | ||
| t.Errorf("expected default WritePGURI %q, got %q", expectedURI, cfg.WritePGURI) | ||
| } |
Comment on lines
+92
to
+116
| func (r *TransferRepository) ExecuteTransfer(ctx context.Context, idempotencyKey, fromWalletID, toWalletID string, amount float64, timestamp int64) (*domain.Transfer, error) { | ||
| logger.Info(). | ||
| Str("idempotency_key", idempotencyKey). | ||
| Str("from_wallet_id", fromWalletID). | ||
| Str("to_wallet_id", toWalletID). | ||
| Float64("amount", amount). | ||
| Msg("executing transfer") | ||
| return withRetry(ctx, func() (*domain.Transfer, error) { | ||
| return r.executeTransferOnce(ctx, idempotencyKey, fromWalletID, toWalletID, amount, timestamp) | ||
| }) | ||
| } | ||
|
|
||
| func (r *TransferRepository) executeTransferOnce(ctx context.Context, idempotencyKey, fromWalletID, toWalletID string, amount float64, timestamp int64) (*domain.Transfer, error) { | ||
| pending, err := insertPendingTransfer(ctx, r.writeDB, r.writeQueries, idempotencyKey, fromWalletID, toWalletID, amount) | ||
| if err != nil { | ||
| if errors.Is(err, domain.ErrIdempotencyKeyConflict) { | ||
| logger.Warn().Str("idempotency_key", idempotencyKey).Msg("idempotency key conflict while inserting pending transfer") | ||
| } else { | ||
| logger.Warn().Err(err).Str("idempotency_key", idempotencyKey).Msg("failed to insert pending transfer") | ||
| } | ||
| return nil, err | ||
| } | ||
| logger.Debug().Str("transfer_id", pending.ID).Msg("inserted pending transfer") | ||
| return r.finalizePendingTransfer(ctx, pending.ID, fromWalletID, toWalletID, amount) | ||
| } |
| }) | ||
| } | ||
|
|
||
| func (r *TransferRepository) ExecuteTransfer(ctx context.Context, idempotencyKey, fromWalletID, toWalletID string, amount float64, timestamp int64) (*domain.Transfer, error) { |
| }) | ||
| } | ||
|
|
||
| func (r *TransferRepository) executeTransferOnce(ctx context.Context, idempotencyKey, fromWalletID, toWalletID string, amount float64, timestamp int64) (*domain.Transfer, error) { |
Comment on lines
+17
to
+19
| func NewWalletRepository(readDB *sql.DB, writeDB *sql.DB) *WalletRepository { | ||
| return &WalletRepository{readQueries: sqlcgen.New(readDB)} | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
A Go/PostgreSQL wallet transfer service. Wallets and transfers are managed through a layered architecture (handler → service → repository, with domain models in between). Transfers execute in two phases inside a single writer transaction each:
(1) insert a PENDING transfer row keyed on a unique idempotency_key,
(2) lock both wallets in a deterministic order, debit/credit balances, write two ledger entries, and mark the transfer PROCESSED or FAILED.
A background reconciler (internal/worker/pending_transfer_recon.go) sweeps any transfers left stuck in PENDING (e.g. a crash between phase 1 and 2) and resolves them to the same terminal state the original request would have reached. Endpoints: POST /transfers, GET /wallets/{id}/balance, GET /wallets/{id}/transfers, GET /health.
AI disclosure
Tool used: Claude Code.
How I generally use it: For this assignment I used it as a pair-programmer — describing the desired behavior (idempotent transfer endpoint, double-entry ledger, concurrency-safe balance updates) and iterating with it on the schema, repository/service layering, and test scenarios, rather than asking it to generate the whole solution unprompted. I reviewed and adjusted its output at each step (locking strategy, error mapping, retry logic) rather than accepting it verbatim.
All the prompts that you used with the AI:
Schema Design
Three tables (migrations/0001_init.up.sql):
wallets: id UUID PK, owner_name, balance DOUBLE PRECISION NOT NULL DEFAULT 0 CHECK (balance >= 0), created_at/updated_at. The balance >= 0 check is a last-line-of-defense invariant against negative balances.
transfers: id UUID PK, idempotency_key TEXT NOT NULL UNIQUE, from_wallet_id/to_wallet_id (FKs to wallets), amount CHECK (amount > 0), status (transfer_status enum: PENDING/PROCESSED/FAILED, default PENDING), failure_reason, timestamps. CHECK (from_wallet_id <> to_wallet_id) rejects self-transfers at the DB level. Indexes on from_wallet_id and to_wallet_id support history lookups.
ledger_entries: id UUID PK, transfer_id FK, wallet_id FK, entry_type (DEBIT/CREDIT enum), amount CHECK (amount > 0), created_at. UNIQUE (transfer_id, entry_type) enforces exactly one debit and one credit row per transfer — the ledger can never go out of balance for a given transfer. Indexed on wallet_id.
No separate idempotency_records table — the unique constraint on transfers.idempotency_key itself is the dedupe mechanism, which keeps the request/result state co-located with the transfer instead of a second source of truth.
Idempotency Strategy
Idempotency Strategy
idempotency_key is a required field on POST /transfers and is stored as a UNIQUE column on transfers.
On each request, the repository attempts to INSERT a new PENDING transfer row with that key. If a row with the same key already exists, Postgres raises a 23505 unique-violation, which the repository maps to ErrIdempotencyKeyConflict (repository/postgres/transfer.repo.go).
On conflict, the service layer (internal/service/wallet.svc.go: replayIdempotentTransfer) looks up the existing transfer by key and replays its result instead of re-executing the transfer. If the replayed request's from/to/amount don't match the original, it's treated as key reuse with different parameters and rejected — the key is a request fingerprint, not just a dedupe token.
Because the insert of the PENDING row and the wallet locking happen in the same transaction, two concurrent requests with the same key can only have one winner; the loser gets the unique violation and replays the winner's outcome once it resolves.
If a transfer is left in PENDING (process crash between insert and finalize), retries with the same key don't create a duplicate — the finalize step is itself idempotent (see below) and the background reconciler will eventually resolve it to the correct terminal state either way.
Concurrency Strategy
Each transfer finalize step runs inside a single DB transaction and takes row-level locks (SELECT ... FOR UPDATE) on both wallets before reading balances or writing anything (lockWalletPair / lockWallet in transfer.repo.go).
Wallets are locked in a deterministic order (sorted by wallet ID, not by from/to) so that two transfers moving funds in opposite directions between the same pair of wallets always acquire locks in the same order — this is what prevents deadlock (covered by TestExecuteTransfer_ConcurrentReverseTransfers_NoDeadlock).
Balance checks happen after the lock is held, so no other transaction can concurrently read/modify either wallet's balance mid-transfer — this prevents lost updates and double-spending under concurrent debits from the same wallet (TestExecuteTransfer_ConcurrentTransfers_NoLostUpdates).
The finalize step re-checks the transfer's status (still PENDING?) after acquiring the locks, so if two callers race to finalize the same transfer (e.g. the original request and a reconciler sweep), the second one is a no-op that just returns the already-resolved result rather than double-applying the ledger entries.
Insufficient balance is a normal terminal outcome, not an error path with side effects: the transfer is marked FAILED with a reason, still inside the same locked transaction, so no partial debit/credit or a orphaned ledger entry can occur.
Transient DB errors (connection blips, retryable Postgres error classes) are retried with backoff at the repository layer (repository/postgres/retry.go); non-retryable errors (e.g. the idempotency conflict) are not retried, since retrying those would be a correctness issue, not a resiliency one.
A ConcurrencyLimit middleware (internal/middleware/concurrency_limit.middleware.go) bounds the number of in-flight requests server-side to protect against overload, independent of the DB-level locking.
How to Run
How to Test
make test # go test ./... -v
make test-race # same, with -race (important given the concurrency claims above)
make test-coverage # runs tests + enforces the 60% coverage threshold
Key test files to look at for the claims above: repository/postgres/transfer_concurrency_test.go (concurrent transfers, deadlock avoidance, concurrent same-idempotency-key requests, concurrent insufficient-balance races), repository/postgres/transfer_lifecycle_test.go and transfer_scenarios_test.go (state transitions), internal/service/wallet_test.go and internal/handler/wallet_test.go (service/handler behavior with mocked repos).
Tradeoffs / Assumptions
balance is stored as DOUBLE PRECISION rather than a fixed-point/decimal type — fine for this assignment's scope, but a real money-handling system should use an integer minor-unit or NUMERIC column to avoid floating-point rounding drift.
Balance is maintained as a running column on wallets (updated during transfer) rather than derived from ledger_entries on read — faster balance reads, at the cost of needing the row lock discipline above to keep it consistent with the ledger.
The reconciler for stuck PENDING transfers runs on a fixed interval (default 3 minutes) rather than being triggered by a crash-detection signal; this bounds — but doesn't zero out — the window during which a transfer can be visibly PENDING after a crash.
No authentication/authorization layer — out of scope per the assignment brief, which focuses on transactional/concurrency correctness.
Checklist