From 50f87f7c70a18138328890bbae4b4b0cc7f7a532 Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 14:05:06 +0530 Subject: [PATCH 01/11] README with requirements, design and ERD --- forex-mtl/README.md | 156 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 forex-mtl/README.md diff --git a/forex-mtl/README.md b/forex-mtl/README.md new file mode 100644 index 00000000..480bfcc6 --- /dev/null +++ b/forex-mtl/README.md @@ -0,0 +1,156 @@ +# Forex Rate Proxy + +A local HTTP proxy for currency exchange rates. Internal services call this instead of calling +One-Frame directly — it handles caching, rate limiting, error recovery, and all the edge cases +so callers don't have to. + +--- + +## The problem + +One-Frame is the upstream provider of exchange rates. It has one hard constraint that makes +it awkward to use directly: + +> **1,000 requests per day per API token.** + +If your services make 10,000 requests per day (the stated requirement), hitting One-Frame on +every request burns the quota in 6 minutes. This proxy solves that by sitting in front of +One-Frame and serving most requests from an in-memory cache. + +--- + +## Requirements (from the brief) + +1. Return an exchange rate when given two supported currency codes. +2. The rate returned must never be older than **5 minutes**. +3. Support **at least 10,000 successful client requests per day** using a single API token + (which is capped at 1,000 upstream calls per day). + +--- + +## Constraints and what they forced + +### One-Frame returns all pairs in a single request + +The API accepts multiple `pair` query parameters: +``` +GET /rates?pair=USDEUR&pair=USDJPY&pair=GBPAUD&... +``` + +This means there is no extra cost to fetching all 72 pairs (nPr = 9! / 7!) at once versus fetching one. +Every upstream call in this service fetches all 72 pairs and fills the entire cache. +There is no per-pair granularity. + +### The 5-minute freshness ceiling + +Serving stale data is a correctness bug. Serving data that is 4:59 old is fine; 5:01 is not. +The implementation enforces this with a hard check: if a cached entry is ≥ 5 minutes old, it +is never served — the request blocks until a fresh batch has been fetched from One-Frame. + +### One-Frame always returns HTTP 200 + +Even on errors. Quota exhaustion looks like: +```json +{"error": "Quota reached for token ..."} +``` +The client has to inspect the body to distinguish success from failure. This is handled in +`OneFrameHttpClient` — successful responses are JSON arrays; error responses are JSON objects +with an `error` field. + +The One-Frame response also uses `time_stamp` (snake_case with underscore), which is worth +calling out explicitly because it differs from the camelCase convention everywhere else. + +--- + +## Assumptions + +**Single instance.** The service runs as a single process. The cache is in-memory and is not +shared across instances. This is appropriate for a "local proxy" and keeps the deployment +simple. If horizontal scaling were needed, the `CacheAlgebra` interface makes a Redis migration +straightforward. + +**All 9 currencies are always cached together.** There is no logic to cache individual pairs +separately or to prioritise popular pairs. Every cache miss fetches all 72 pairs. This is +deliberate — the marginal cost of fetching 71 extra pairs is zero, and the simplicity is worth it. + +**One-Frame is the only upstream source.** No fallback provider is implemented. If One-Frame +is down and the cache is exhausted, the service returns 502. + +**`maxStaleOnError` defaults to 5 minutes, matching the SLA.** If One-Frame is unreachable and +the cache is older than 5 minutes, the service returns 502. If your system can tolerate slightly +stale rates over an error response during an outage, raise this to e.g. 10 minutes — the service +will then serve cached data up to that age before falling back to 502. + +--- + +## How the cache works + +The cache uses a **Stale-While-Revalidate (SWR)** policy with two TTL boundaries: + +``` +Age of the cached rate What happens +──────────────────────────── ───────────────────────────────────────────────── +0 – 4 min (fresh) Served immediately. No upstream work. +4 – 5 min (stale-valid) Served immediately. Background refresh fires concurrently. +≥ 5 min (expired) Request waits. Fetch completes first. Then respond. +Not in cache (cold) Request waits. Fetch completes first. Then respond. +``` + +The key insight: for the overwhelming majority of requests (anything in the 0–4 min window) +the response latency is just a map lookup. The 4–5 min window means clients are never blocked +by a revalidation — they get a slightly stale value while the cache updates behind them. +Only a cold start or a true expiry blocks the caller. + +### Why not a background polling job? + +An earlier version ran a fiber every 4 minutes to keep the cache warm. It was removed. +A polling job burns upstream quota even when the service has zero traffic — 360 calls per day +regardless. The SWR approach only refreshes when someone actually asks for a rate. Zero traffic, +zero upstream calls. + +### Concurrent cold starts (the thundering herd) + +When the cache is empty and 100 requests arrive simultaneously, without protection all 100 +would race to call One-Frame. The implementation uses a `Deferred` gate (a one-shot promise): + +1. The first request creates the gate and starts the upstream fetch. +2. Every other concurrent request finds the gate and waits. +3. When the first request finishes, all waiters unblock at once with the result already in cache. + +Result: exactly **one** upstream call per burst, regardless of concurrency. + +--- + +## Key decisions + +### In-memory cache, not Redis + +Adding Redis means adding infrastructure, network latency on every cache read, and a new +failure mode (the cache itself can become unavailable). For a single-instance local proxy, +in-memory is correct. The `CacheAlgebra[F]` trait is the only thing a Redis implementation +would need to satisfy. + +### Typed errors, not exceptions + +Every function that can fail returns an `Either` — a value that is either a typed error or +a result. The error type is an exhaustive set of cases: + +- `OneFrameQuotaExceeded` — the daily limit is hit +- `OneFrameUnreachable` — network failure, timeout, or unexpected HTTP status from One-Frame +- `OneFrameLookupFailed` — One-Frame returned an error message we didn't recognise + +The compiler enforces that every call site handles all cases. If a new error is added later, +every unhandled case becomes a compile error — not a runtime crash. + +### `fromString` returns `Either`, not a throw + +The original scaffold had `Currency.fromString` as an unsafe partial pattern match. An unknown +currency code would throw a `MatchError` at runtime. The replacement finds the currency by +looking it up in the `values` list, and returns `Left("Unsupported currency: XYZ")` if not found. +The HTTP layer turns that into a `400 Bad Request` with a JSON error body. + +### Same-currency short-circuit + +`GET /rates?from=USD&to=USD` is a valid request. USD always exchanges to USD at 1.0. +The program layer intercepts this case and returns immediately without touching the cache +or the upstream API. From 7b9701ed7c1c94ab3fa254feab9a3ad9be24adeb Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 14:27:47 +0530 Subject: [PATCH 02/11] Config, domain models, and error types --- forex-mtl/build.sbt | 1 + forex-mtl/project/Dependencies.scala | 1 + forex-mtl/src/main/resources/application.conf | 20 +++++++++++-- .../forex/config/ApplicationConfig.scala | 14 ++++++++++ .../main/scala/forex/domain/Currency.scala | 28 ++++--------------- .../src/main/scala/forex/domain/Rate.scala | 12 ++++++-- .../scala/forex/programs/rates/errors.scala | 18 ++++++++++-- .../scala/forex/services/rates/errors.scala | 4 ++- 8 files changed, 69 insertions(+), 29 deletions(-) diff --git a/forex-mtl/build.sbt b/forex-mtl/build.sbt index 8994026f..dc40e223 100644 --- a/forex-mtl/build.sbt +++ b/forex-mtl/build.sbt @@ -56,6 +56,7 @@ libraryDependencies ++= Seq( Libraries.fs2, Libraries.http4sDsl, Libraries.http4sServer, + Libraries.http4sClient, Libraries.http4sCirce, Libraries.circeCore, Libraries.circeGeneric, diff --git a/forex-mtl/project/Dependencies.scala b/forex-mtl/project/Dependencies.scala index 423210a1..0acb0300 100644 --- a/forex-mtl/project/Dependencies.scala +++ b/forex-mtl/project/Dependencies.scala @@ -27,6 +27,7 @@ object Dependencies { lazy val http4sDsl = http4s("http4s-dsl") lazy val http4sServer = http4s("http4s-blaze-server") + lazy val http4sClient = http4s("http4s-blaze-client") lazy val http4sCirce = http4s("http4s-circe") lazy val circeCore = circe("circe-core") lazy val circeGeneric = circe("circe-generic") diff --git a/forex-mtl/src/main/resources/application.conf b/forex-mtl/src/main/resources/application.conf index b2af6efd..b5caa66a 100644 --- a/forex-mtl/src/main/resources/application.conf +++ b/forex-mtl/src/main/resources/application.conf @@ -1,8 +1,24 @@ app { http { host = "0.0.0.0" - port = 8080 + port = 8081 timeout = 40 seconds } -} + one-frame { + base-uri = "http://localhost:8080" + base-uri = ${?ONE_FRAME_BASE_URI} + + auth-token = "" + auth-token = ${?ONE_FRAME_TOKEN} + + timeout = 10 seconds + } + + cache { + ttl = 5 minutes # Hard freshness ceiling; must not exceed the 5-minute SLA. + soft-ttl = 4 minutes # Serve stale + trigger background revalidation above this age. + max-stale-on-error = 5 minutes # Matches the SLA by default. Raise to e.g. 10 minutes to serve + # slightly stale rates during One-Frame outages instead of 502. + } +} diff --git a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala index eff0fad7..32e21b99 100644 --- a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala +++ b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala @@ -4,6 +4,8 @@ import scala.concurrent.duration.FiniteDuration case class ApplicationConfig( http: HttpConfig, + oneFrame: OneFrameConfig, + cache: CacheConfig ) case class HttpConfig( @@ -11,3 +13,15 @@ case class HttpConfig( port: Int, timeout: FiniteDuration ) + +case class OneFrameConfig( + baseUri: String, + authToken: String, + timeout: FiniteDuration +) + +case class CacheConfig( + ttl: FiniteDuration, + softTtl: FiniteDuration, + maxStaleOnError: FiniteDuration +) diff --git a/forex-mtl/src/main/scala/forex/domain/Currency.scala b/forex-mtl/src/main/scala/forex/domain/Currency.scala index a6f2857d..e180f00f 100644 --- a/forex-mtl/src/main/scala/forex/domain/Currency.scala +++ b/forex-mtl/src/main/scala/forex/domain/Currency.scala @@ -15,28 +15,12 @@ object Currency { case object SGD extends Currency case object USD extends Currency - implicit val show: Show[Currency] = Show.show { - case AUD => "AUD" - case CAD => "CAD" - case CHF => "CHF" - case EUR => "EUR" - case GBP => "GBP" - case NZD => "NZD" - case JPY => "JPY" - case SGD => "SGD" - case USD => "USD" - } + val values: List[Currency] = List(AUD, CAD, CHF, EUR, GBP, NZD, JPY, SGD, USD) - def fromString(s: String): Currency = s.toUpperCase match { - case "AUD" => AUD - case "CAD" => CAD - case "CHF" => CHF - case "EUR" => EUR - case "GBP" => GBP - case "NZD" => NZD - case "JPY" => JPY - case "SGD" => SGD - case "USD" => USD - } + implicit val show: Show[Currency] = Show.fromToString + def fromString(str: String): Either[String, Currency] = + values + .find(_.toString == str.toUpperCase(java.util.Locale.ROOT)) + .toRight(s"Unsupported currency: $str") } diff --git a/forex-mtl/src/main/scala/forex/domain/Rate.scala b/forex-mtl/src/main/scala/forex/domain/Rate.scala index 4a444003..542034f5 100644 --- a/forex-mtl/src/main/scala/forex/domain/Rate.scala +++ b/forex-mtl/src/main/scala/forex/domain/Rate.scala @@ -8,7 +8,15 @@ case class Rate( object Rate { final case class Pair( - from: Currency, - to: Currency + from: Currency, + to: Currency ) + + object Pair { + val allPairs: List[Pair] = for { + from <- Currency.values + to <- Currency.values + if from != to + } yield Pair(from, to) + } } diff --git a/forex-mtl/src/main/scala/forex/programs/rates/errors.scala b/forex-mtl/src/main/scala/forex/programs/rates/errors.scala index 39496b13..af8b96a8 100644 --- a/forex-mtl/src/main/scala/forex/programs/rates/errors.scala +++ b/forex-mtl/src/main/scala/forex/programs/rates/errors.scala @@ -5,11 +5,25 @@ import forex.services.rates.errors.{ Error => RatesServiceError } object errors { sealed trait Error extends Exception + object Error { - final case class RateLookupFailed(msg: String) extends Error + + final case class InvalidCurrency(input: String) extends Error { + override def getMessage: String = s"Unsupported currency: $input" + } + + final case class UpstreamUnavailable(msg: String) extends Error { + override def getMessage: String = msg + } + + final case class SystemError(msg: String) extends Error { + override def getMessage: String = msg + } } def toProgramError(error: RatesServiceError): Error = error match { - case RatesServiceError.OneFrameLookupFailed(msg) => Error.RateLookupFailed(msg) + case RatesServiceError.OneFrameQuotaExceeded => Error.UpstreamUnavailable("One-Frame API quota exceeded for today") + case RatesServiceError.OneFrameUnreachable(cause) => Error.UpstreamUnavailable(s"One-Frame is unreachable: ${cause.getMessage}") + case RatesServiceError.OneFrameLookupFailed(msg) => Error.UpstreamUnavailable(s"One-Frame error: $msg") } } diff --git a/forex-mtl/src/main/scala/forex/services/rates/errors.scala b/forex-mtl/src/main/scala/forex/services/rates/errors.scala index 0584dcf4..a43ad2f8 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/errors.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/errors.scala @@ -4,7 +4,9 @@ object errors { sealed trait Error object Error { + final case object OneFrameQuotaExceeded extends Error + final case class OneFrameUnreachable(cause: Throwable) extends Error final case class OneFrameLookupFailed(msg: String) extends Error } - + } From 3e0280ce91e14aa19a9e5cdab7271d884322b939 Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 17:52:26 +0530 Subject: [PATCH 03/11] Add One-Frame client, SWR cache, live interpreter Made-with: Cursor --- .../clients/oneframe/OneFrameClient.scala | 113 +++++++++++++ .../clients/oneframe/OneFrameProtocol.scala | 26 +++ .../services/rates/cache/CacheEntry.scala | 32 ++++ .../services/rates/cache/RatesCache.scala | 43 +++++ .../rates/interpreters/OneFrameLive.scala | 154 ++++++++++++++++++ 5 files changed, 368 insertions(+) create mode 100644 forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala create mode 100644 forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameProtocol.scala create mode 100644 forex-mtl/src/main/scala/forex/services/rates/cache/CacheEntry.scala create mode 100644 forex-mtl/src/main/scala/forex/services/rates/cache/RatesCache.scala create mode 100644 forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala diff --git a/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala b/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala new file mode 100644 index 00000000..2aac0b54 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala @@ -0,0 +1,113 @@ +package forex.clients.oneframe + +import cats.data.NonEmptyList +import cats.effect.Sync +import cats.instances.either._ +import cats.instances.list._ +import cats.syntax.applicativeError._ +import cats.syntax.either._ +import cats.syntax.functor._ +import cats.syntax.traverse._ +import forex.clients.oneframe.OneFrameProtocol._ +import forex.config.OneFrameConfig +import forex.domain.{ Currency, Price, Rate, Timestamp } +import forex.services.rates.errors.{ Error => ServiceError } +import io.circe.parser.decode +import org.http4s.Header +import org.http4s.Headers +import org.http4s.Method +import org.http4s.Request +import org.http4s.Uri +import org.http4s.client.Client +import org.slf4j.LoggerFactory +import org.typelevel.ci.CIString + +import java.time.OffsetDateTime + +/** Contract for fetching live exchange rates from an upstream provider. + * + * Implementations must be safe to call concurrently. + * A single call supplies all requested pairs in one round-trip. + */ +trait OneFrameClientAlgebra[F[_]] { + def getRates(pairs: NonEmptyList[Rate.Pair]): F[Either[ServiceError, List[Rate]]] +} + +/** Live implementation that talks to the One-Frame HTTP API. + * One-Frame quirks handled here: + * - Always returns HTTP 200; errors are embedded in the JSON body. + * - Auth via a raw `token ` header. + * - Pair encoding: `pair=USDEUR` (concatenated currencies without separators). + */ +class OneFrameHttpClient[F[_]: Sync]( + client: Client[F], + config: OneFrameConfig +) extends OneFrameClientAlgebra[F] { + + private val logger = LoggerFactory.getLogger(getClass) + + override def getRates(pairs: NonEmptyList[Rate.Pair]): F[Either[ServiceError, List[Rate]]] = + client + .expect[String](buildRequest(pairs)) + .map(parseResponse) + .handleErrorWith { err => + Sync[F] + .delay(logger.warn(s"One-Frame HTTP call failed: ${err.getMessage}", err)) + .as(ServiceError.OneFrameUnreachable(err).asLeft) + } + + private def buildRequest(pairs: NonEmptyList[Rate.Pair]): Request[F] = { + val pairParams = pairs.toList + .map(p => s"pair=${Currency.show.show(p.from)}${Currency.show.show(p.to)}") + .mkString("&") + + val uri = Uri.unsafeFromString(s"${config.baseUri}/rates?$pairParams") + + Request[F]( + method = Method.GET, + uri = uri, + headers = Headers(Header.Raw(CIString("token"), config.authToken)) + ) + } + + private def parseResponse(body: String): Either[ServiceError, List[Rate]] = + decode[List[OneFrameRate]](body) match { + case Right(rates) => + rates + .traverse(toRate) + .leftMap(msg => (ServiceError.OneFrameLookupFailed(msg): ServiceError)) + + case Left(_) => + decode[OneFrameErrorResponse](body) match { + case Right(err) if err.error.contains("Quota reached") => + ServiceError.OneFrameQuotaExceeded.asLeft + + case Right(err) => + ServiceError.OneFrameLookupFailed(s"One-Frame error: ${err.error}").asLeft + + case Left(_) => + ServiceError.OneFrameLookupFailed(s"Unparseable One-Frame response: $body").asLeft + } + } + + private def toRate(r: OneFrameRate): Either[String, Rate] = + for { + from <- Currency.fromString(r.from) + to <- Currency.fromString(r.to) + timestamp <- parseTimestamp(r.timeStamp) + } yield Rate(Rate.Pair(from, to), Price(r.price), Timestamp(timestamp)) + + private def parseTimestamp(raw: String): Either[String, OffsetDateTime] = + Either + .catchNonFatal(OffsetDateTime.parse(raw)) + .leftMap(e => s"Invalid timestamp '$raw': ${e.getMessage}") +} + +object OneFrameHttpClient { + + def apply[F[_]: Sync]( + client: Client[F], + config: OneFrameConfig + ): OneFrameClientAlgebra[F] = + new OneFrameHttpClient[F](client, config) +} diff --git a/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameProtocol.scala b/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameProtocol.scala new file mode 100644 index 00000000..1ac6fb19 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameProtocol.scala @@ -0,0 +1,26 @@ +package forex.clients.oneframe + +import io.circe.Decoder +import io.circe.generic.semiauto.deriveDecoder + +/** JSON codecs for the One-Frame API wire format. + * HTTP 200- A successful call returns a JSON array of rate objects + * a failed call (e.g. quota exceeded) returns a JSON object with a single "error" field. + */ +object OneFrameProtocol { + + final case class OneFrameRate( + from: String, + to: String, + bid: BigDecimal, + ask: BigDecimal, + price: BigDecimal, + timeStamp: String + ) + + final case class OneFrameErrorResponse(error: String) + + implicit val oneFrameRateDecoder: Decoder[OneFrameRate] = + Decoder.forProduct6("from", "to", "bid", "ask", "price", "time_stamp")(OneFrameRate.apply) + implicit val oneFrameErrorDecoder: Decoder[OneFrameErrorResponse] = deriveDecoder +} diff --git a/forex-mtl/src/main/scala/forex/services/rates/cache/CacheEntry.scala b/forex-mtl/src/main/scala/forex/services/rates/cache/CacheEntry.scala new file mode 100644 index 00000000..bfd40217 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/rates/cache/CacheEntry.scala @@ -0,0 +1,32 @@ +package forex.services.rates.cache + +import forex.domain.Rate + +import java.time.OffsetDateTime +import java.time.temporal.ChronoUnit +import scala.concurrent.duration.FiniteDuration + +/** An immutable snapshot of a fetched rate together with the wall-clock time it was + * retrieved from One-Frame. + * + * Three staleness levels drive the Stale-While-Revalidate (SWR) policy: + * + * isFresh – age < softTtl → serve immediately, no background work + * needsRevalidation – softTtl ≤ age < ttl → serve immediately AND refresh in background + * isExpired – age ≥ ttl → must fetch synchronously before responding + */ +final case class CacheEntry(rate: Rate, fetchedAt: OffsetDateTime) { + + private def ageMillis(now: OffsetDateTime): Long = + ChronoUnit.MILLIS.between(fetchedAt, now) + + def isFresh(now: OffsetDateTime, softTtl: FiniteDuration): Boolean = + ageMillis(now) < softTtl.toMillis + + def isExpired(now: OffsetDateTime, ttl: FiniteDuration): Boolean = + ageMillis(now) >= ttl.toMillis + + /** True during the softTtl–ttl window: serve the cached value and kick off a background refresh. */ + def needsRevalidation(now: OffsetDateTime, softTtl: FiniteDuration, ttl: FiniteDuration): Boolean = + !isFresh(now, softTtl) && !isExpired(now, ttl) +} diff --git a/forex-mtl/src/main/scala/forex/services/rates/cache/RatesCache.scala b/forex-mtl/src/main/scala/forex/services/rates/cache/RatesCache.scala new file mode 100644 index 00000000..3a6b1900 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/rates/cache/RatesCache.scala @@ -0,0 +1,43 @@ +package forex.services.rates.cache + +import cats.effect.Sync +import cats.effect.concurrent.Ref +import cats.syntax.functor._ +import forex.domain.Rate + +import java.time.OffsetDateTime + +trait CacheAlgebra[F[_]] { + def get(pair: Rate.Pair): F[Option[CacheEntry]] + def putBatch(rates: List[Rate], fetchedAt: OffsetDateTime): F[Unit] + def allKeys: F[Set[Rate.Pair]] +} + +/** TODO(scale): Plug in Redis when running multiple app instances. + * 1- Add a `RedisRatesCache` implementing this `CacheAlgebra`. + * 2- Store each pair under a stable key (for example `rates:USD:EUR`). + * 3- Save `{rate, fetchedAt}` and enforce TTL with Redis expiry. + * 4- Replace `InMemoryRatesCache.create` wiring in `Module`/interpreter assembly. + * 5- Keep `putBatch` semantics atomic via `MULTI/EXEC` (or Lua) to avoid partial writes. + */ +class InMemoryRatesCache[F[_]: Sync] private (state: Ref[F, Map[Rate.Pair, CacheEntry]]) extends CacheAlgebra[F] { + + override def get(pair: Rate.Pair): F[Option[CacheEntry]] = + state.get.map(_.get(pair)) + + override def putBatch(rates: List[Rate], fetchedAt: OffsetDateTime): F[Unit] = + state.update { existing => + rates.foldLeft(existing) { (acc, rate) => + acc.updated(rate.pair, CacheEntry(rate, fetchedAt)) + } + } + + override def allKeys: F[Set[Rate.Pair]] = + state.get.map(_.keySet) +} + +object InMemoryRatesCache { + + def create[F[_]: Sync]: F[InMemoryRatesCache[F]] = + Ref.of[F, Map[Rate.Pair, CacheEntry]](Map.empty).map(new InMemoryRatesCache[F](_)) +} diff --git a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala new file mode 100644 index 00000000..26b7a38c --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala @@ -0,0 +1,154 @@ +package forex.services.rates.interpreters + +import cats.data.NonEmptyList +import cats.effect.concurrent.{ Deferred, Ref } +import cats.effect.{ Concurrent, Resource, Timer } +import cats.syntax.applicative._ +import cats.syntax.applicativeError._ +import cats.syntax.either._ +import cats.syntax.flatMap._ +import cats.syntax.functor._ +import forex.clients.oneframe.{ OneFrameClientAlgebra, OneFrameHttpClient } +import forex.config.ApplicationConfig +import forex.domain.Rate +import forex.services.rates.Algebra +import forex.services.rates.cache.{ CacheAlgebra, InMemoryRatesCache } +import forex.services.rates.errors.{ Error => ServiceError } +import org.http4s.client.Client +import org.slf4j.LoggerFactory + +import java.time.OffsetDateTime + +class OneFrameLive[F[_]: Concurrent: Timer] private ( + oneFrameClient: OneFrameClientAlgebra[F], + cache: CacheAlgebra[F], + fetchGate: Ref[F, Option[Deferred[F, Either[ServiceError, Unit]]]], + config: ApplicationConfig +) extends Algebra[F] { + + private val logger = LoggerFactory.getLogger(getClass) + private val softTtl = config.cache.softTtl + private val ttl = config.cache.ttl + private val maxStaleOnError = config.cache.maxStaleOnError + + override def get(pair: Rate.Pair): F[Either[ServiceError, Rate]] = + nowUtc.flatMap { now => + cache.get(pair).flatMap { cached => + cached match { + case Some(entry) if entry.isFresh(now, softTtl) => + Concurrent[F].delay(logger.debug(s"Cache HIT for $pair")) >> + entry.rate.asRight[ServiceError].pure[F] + + case Some(entry) if entry.needsRevalidation(now, softTtl, ttl) => + Concurrent[F].delay(logger.debug(s"Cache STALE for $pair, revalidating in background")) >> + Concurrent[F] + .start(fetchAllPairsAndPopulateCache) + .as(entry.rate.asRight[ServiceError]) + + case _ => + Concurrent[F].delay(logger.debug(s"Cache MISS for $pair, fetching synchronously")) >> + fetchWithCoalescing.flatMap { + case Right(_) => + cache.get(pair).map { + case Some(entry) => entry.rate.asRight + case None => + ServiceError.OneFrameLookupFailed(s"One-Frame did not return a rate for $pair").asLeft + } + + case Left(err) => + cached match { + case Some(entry) if !entry.isExpired(now, maxStaleOnError) => + Concurrent[F] + .delay( + logger.warn(s"One-Frame unavailable; serving stale cache for $pair") + ) + .as(entry.rate.asRight[ServiceError]) + case _ => + err.asLeft[Rate].pure[F] + } + } + } + } + } + + private val fetchWithCoalescing: F[Either[ServiceError, Unit]] = + Deferred[F, Either[ServiceError, Unit]].flatMap { newGate => + fetchGate + .modify { + case None => (Some(newGate), Left(newGate)) + case Some(existingGate) => (Some(existingGate), Right(existingGate)) + } + .flatMap { + case Left(ourGate) => + fetchAllPairsAndPopulateCache + .flatTap(result => ourGate.complete(result)) + .flatTap(_ => fetchGate.set(None)) + .handleErrorWith { err => + val svcError: ServiceError = ServiceError.OneFrameUnreachable(err) + ourGate.complete(svcError.asLeft) >> + fetchGate.set(None) >> + svcError.asLeft[Unit].pure[F] + } + + case Right(theirGate) => + Concurrent[F].delay(logger.debug("Awaiting in-flight One-Frame fetch")) >> + theirGate.get + } + } + + private def fetchAllPairsAndPopulateCache: F[Either[ServiceError, Unit]] = + NonEmptyList.fromList(Rate.Pair.allPairs) match { + case None => + (ServiceError.OneFrameLookupFailed("No currency pairs defined"): ServiceError).asLeft[Unit].pure[F] + + case Some(pairs) => + Concurrent[F].delay(logger.info(s"Fetching ${pairs.size} pairs from One-Frame")) >> + oneFrameClient.getRates(pairs).flatMap { + case Right(rates) => + nowUtc.flatMap(now => cache.putBatch(rates, now)) >> + Concurrent[F] + .delay(logger.info(s"Cached ${pairs.size} pairs")) + .as(().asRight[ServiceError]) + + case Left(err) => + Concurrent[F] + .delay(logger.warn(s"One-Frame fetch failed: $err")) + .as(err.asLeft[Unit]) + } + } + + private def nowUtc: F[OffsetDateTime] = + Timer[F].clock.realTime(scala.concurrent.duration.MILLISECONDS).map { millis => + OffsetDateTime.ofInstant( + java.time.Instant.ofEpochMilli(millis), + java.time.ZoneOffset.UTC + ) + } +} + +object OneFrameLive { + + def resource[F[_]: Concurrent: Timer]( + httpClient: Client[F], + config: ApplicationConfig + ): Resource[F, Algebra[F]] = + Resource.eval(make(OneFrameHttpClient[F](httpClient, config.oneFrame), config)) + + def make[F[_]: Concurrent: Timer]( + client: OneFrameClientAlgebra[F], + config: ApplicationConfig + ): F[Algebra[F]] = + for { + cache <- InMemoryRatesCache.create[F] + fetchGate <- Ref.of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) + } yield new OneFrameLive[F](client, cache, fetchGate, config) + + def makeWithCache[F[_]: Concurrent: Timer]( + client: OneFrameClientAlgebra[F], + cache: CacheAlgebra[F], + config: ApplicationConfig + ): F[Algebra[F]] = + Ref + .of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) + .map(fetchGate => new OneFrameLive[F](client, cache, fetchGate, config)) +} From a2094d89bdd38c9a096af1ec07401140af27e4f0 Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 18:15:26 +0530 Subject: [PATCH 04/11] Wire HTTP layer, program, and application stream --- forex-mtl/src/main/scala/forex/Main.scala | 16 +++++++++------- .../scala/forex/http/rates/Protocol.scala | 10 +++++++++- .../scala/forex/http/rates/QueryParams.scala | 13 ++++++++----- .../scala/forex/programs/rates/Program.scala | 19 ++++++++++++------- 4 files changed, 38 insertions(+), 20 deletions(-) diff --git a/forex-mtl/src/main/scala/forex/Main.scala b/forex-mtl/src/main/scala/forex/Main.scala index 6dda10a7..50840619 100644 --- a/forex-mtl/src/main/scala/forex/Main.scala +++ b/forex-mtl/src/main/scala/forex/Main.scala @@ -1,16 +1,17 @@ package forex -import scala.concurrent.ExecutionContext import cats.effect._ import forex.config._ import fs2.Stream +import org.http4s.blaze.client.BlazeClientBuilder import org.http4s.blaze.server.BlazeServerBuilder +import scala.concurrent.ExecutionContext + object Main extends IOApp { override def run(args: List[String]): IO[ExitCode] = new Application[IO].stream(executionContext).compile.drain.as(ExitCode.Success) - } class Application[F[_]: ConcurrentEffect: Timer] { @@ -18,11 +19,12 @@ class Application[F[_]: ConcurrentEffect: Timer] { def stream(ec: ExecutionContext): Stream[F, Unit] = for { config <- Config.stream("app") - module = new Module[F](config) + client <- Stream.resource(BlazeClientBuilder[F](ec).withRequestTimeout(config.oneFrame.timeout).resource) + module <- Stream.resource(Module.resource[F](config, client)) _ <- BlazeServerBuilder[F](ec) - .bindHttp(config.http.port, config.http.host) - .withHttpApp(module.httpApp) - .serve + .bindHttp(config.http.port, config.http.host) + .withHttpApp(module.httpApp) + .serve } yield () - + } diff --git a/forex-mtl/src/main/scala/forex/http/rates/Protocol.scala b/forex-mtl/src/main/scala/forex/http/rates/Protocol.scala index 75391f9d..d9571417 100644 --- a/forex-mtl/src/main/scala/forex/http/rates/Protocol.scala +++ b/forex-mtl/src/main/scala/forex/http/rates/Protocol.scala @@ -8,6 +8,8 @@ import io.circe._ import io.circe.generic.extras.Configuration import io.circe.generic.extras.semiauto.deriveConfiguredEncoder +import java.time.format.DateTimeFormatter + object Protocol { implicit val configuration: Configuration = Configuration.default.withSnakeCaseMemberNames @@ -25,7 +27,13 @@ object Protocol { ) implicit val currencyEncoder: Encoder[Currency] = - Encoder.instance[Currency] { show.show _ andThen Json.fromString } + Encoder.instance[Currency](show.show _ andThen Json.fromString) + + implicit val priceEncoder: Encoder[Price] = + Encoder.instance[Price](p => Json.fromBigDecimal(p.value)) + + implicit val timestampEncoder: Encoder[Timestamp] = + Encoder.instance[Timestamp](t => Json.fromString(t.value.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME))) implicit val pairEncoder: Encoder[Pair] = deriveConfiguredEncoder[Pair] diff --git a/forex-mtl/src/main/scala/forex/http/rates/QueryParams.scala b/forex-mtl/src/main/scala/forex/http/rates/QueryParams.scala index b19ed4ce..8677ff0d 100644 --- a/forex-mtl/src/main/scala/forex/http/rates/QueryParams.scala +++ b/forex-mtl/src/main/scala/forex/http/rates/QueryParams.scala @@ -1,15 +1,18 @@ package forex.http.rates +import cats.syntax.either._ import forex.domain.Currency -import org.http4s.QueryParamDecoder -import org.http4s.dsl.impl.QueryParamDecoderMatcher +import org.http4s.{ ParseFailure, QueryParamDecoder } +import org.http4s.dsl.impl.OptionalValidatingQueryParamDecoderMatcher object QueryParams { private[http] implicit val currencyQueryParam: QueryParamDecoder[Currency] = - QueryParamDecoder[String].map(Currency.fromString) + QueryParamDecoder[String].emap { s => + Currency.fromString(s).leftMap(msg => ParseFailure(msg, msg)) + } - object FromQueryParam extends QueryParamDecoderMatcher[Currency]("from") - object ToQueryParam extends QueryParamDecoderMatcher[Currency]("to") + object FromQueryParam extends OptionalValidatingQueryParamDecoderMatcher[Currency]("from") + object ToQueryParam extends OptionalValidatingQueryParamDecoderMatcher[Currency]("to") } diff --git a/forex-mtl/src/main/scala/forex/programs/rates/Program.scala b/forex-mtl/src/main/scala/forex/programs/rates/Program.scala index 528ee1f9..bb6fe358 100644 --- a/forex-mtl/src/main/scala/forex/programs/rates/Program.scala +++ b/forex-mtl/src/main/scala/forex/programs/rates/Program.scala @@ -1,24 +1,29 @@ package forex.programs.rates -import cats.Functor +import cats.Applicative import cats.data.EitherT +import cats.syntax.applicative._ +import cats.syntax.either._ import errors._ import forex.domain._ import forex.services.RatesService -class Program[F[_]: Functor]( +class Program[F[_]: Applicative]( ratesService: RatesService[F] ) extends Algebra[F] { - override def get(request: Protocol.GetRatesRequest): F[Error Either Rate] = - EitherT(ratesService.get(Rate.Pair(request.from, request.to))).leftMap(toProgramError(_)).value - + override def get(request: Protocol.GetRatesRequest): F[Either[Error, Rate]] = { + val pair = Rate.Pair(request.from, request.to) + if (request.from == request.to) + Rate(pair, Price(BigDecimal(1)), Timestamp.now).asRight[Error].pure[F] + else + EitherT(ratesService.get(pair)).leftMap(toProgramError).value + } } object Program { - def apply[F[_]: Functor]( + def apply[F[_]: Applicative]( ratesService: RatesService[F] ): Algebra[F] = new Program[F](ratesService) - } From b24c5fdee93777d16805744c9a8f8c37f4608684 Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 18:52:27 +0530 Subject: [PATCH 05/11] Add request logging and per-IP rate limiting --- forex-mtl/src/main/resources/application.conf | 15 +++--- forex-mtl/src/main/scala/forex/Module.scala | 50 +++++++++--------- .../forex/config/ApplicationConfig.scala | 5 +- .../main/scala/forex/http/RateLimiter.scala | 48 +++++++++++++++++ .../forex/http/rates/RatesHttpRoutes.scala | 51 ++++++++++++++++--- 5 files changed, 129 insertions(+), 40 deletions(-) create mode 100644 forex-mtl/src/main/scala/forex/http/RateLimiter.scala diff --git a/forex-mtl/src/main/resources/application.conf b/forex-mtl/src/main/resources/application.conf index b5caa66a..ae455362 100644 --- a/forex-mtl/src/main/resources/application.conf +++ b/forex-mtl/src/main/resources/application.conf @@ -6,19 +6,18 @@ app { } one-frame { - base-uri = "http://localhost:8080" - base-uri = ${?ONE_FRAME_BASE_URI} - - auth-token = "" - auth-token = ${?ONE_FRAME_TOKEN} - + base-uri = ${?ONE_FRAME_BASE_URI} or "http://localhost:8080" + auth-token = ${?ONE_FRAME_TOKEN} or "" timeout = 10 seconds } cache { ttl = 5 minutes # Hard freshness ceiling; must not exceed the 5-minute SLA. soft-ttl = 4 minutes # Serve stale + trigger background revalidation above this age. - max-stale-on-error = 5 minutes # Matches the SLA by default. Raise to e.g. 10 minutes to serve - # slightly stale rates during One-Frame outages instead of 502. + max-stale-on-error = 5 minutes # Matches SLA. Raise to e.g. 10 minutes to serve during outages instead of 502 + } + + rate-limiter { + max-requests-per-minute = 100 } } diff --git a/forex-mtl/src/main/scala/forex/Module.scala b/forex-mtl/src/main/scala/forex/Module.scala index 3bc47d58..0c1bbed0 100644 --- a/forex-mtl/src/main/scala/forex/Module.scala +++ b/forex-mtl/src/main/scala/forex/Module.scala @@ -1,37 +1,39 @@ package forex -import cats.effect.{ Concurrent, Timer } +import cats.effect.{ Concurrent, Resource, Timer } import forex.config.ApplicationConfig +import forex.http.RateLimiter import forex.http.rates.RatesHttpRoutes -import forex.services._ -import forex.programs._ -import org.http4s._ +import forex.programs.RatesProgram +import forex.services.{ RatesService, RatesServiceFactory } +import org.http4s.{ HttpApp, HttpRoutes } +import org.http4s.client.Client import org.http4s.implicits._ import org.http4s.server.middleware.{ AutoSlash, Timeout } -class Module[F[_]: Concurrent: Timer](config: ApplicationConfig) { - - private val ratesService: RatesService[F] = RatesServices.dummy[F] - - private val ratesProgram: RatesProgram[F] = RatesProgram[F](ratesService) - - private val ratesHttpRoutes: HttpRoutes[F] = new RatesHttpRoutes[F](ratesProgram).routes - +class Module[F[_]: Concurrent: Timer]( + config: ApplicationConfig, + val ratesService: RatesService[F], + rateLimiter: HttpRoutes[F] => HttpRoutes[F] +) { type PartialMiddleware = HttpRoutes[F] => HttpRoutes[F] type TotalMiddleware = HttpApp[F] => HttpApp[F] - private val routesMiddleware: PartialMiddleware = { - { http: HttpRoutes[F] => - AutoSlash(http) - } - } - - private val appMiddleware: TotalMiddleware = { http: HttpApp[F] => - Timeout(config.http.timeout)(http) - } - - private val http: HttpRoutes[F] = ratesHttpRoutes + private val ratesProgram: RatesProgram[F] = RatesProgram[F](ratesService) + private val ratesHttpRoutes: HttpRoutes[F] = new RatesHttpRoutes[F](ratesProgram).routes + private val routesMiddleware: PartialMiddleware = AutoSlash(_) + private val appMiddleware: TotalMiddleware = Timeout(config.http.timeout)(_) + val httpApp: HttpApp[F] = appMiddleware(rateLimiter(routesMiddleware(ratesHttpRoutes)).orNotFound) +} - val httpApp: HttpApp[F] = appMiddleware(routesMiddleware(http).orNotFound) +object Module { + def resource[F[_]: Concurrent: Timer]( + config: ApplicationConfig, + httpClient: Client[F] + ): Resource[F, Module[F]] = + for { + ratesService <- RatesServiceFactory.live[F](httpClient, config) + rateLimiter <- Resource.eval(RateLimiter.middleware[F](config.rateLimiter.maxRequestsPerMinute)) + } yield new Module[F](config, ratesService, rateLimiter) } diff --git a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala index 32e21b99..4b12cab0 100644 --- a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala +++ b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala @@ -5,7 +5,8 @@ import scala.concurrent.duration.FiniteDuration case class ApplicationConfig( http: HttpConfig, oneFrame: OneFrameConfig, - cache: CacheConfig + cache: CacheConfig, + rateLimiter: RateLimiterConfig ) case class HttpConfig( @@ -25,3 +26,5 @@ case class CacheConfig( softTtl: FiniteDuration, maxStaleOnError: FiniteDuration ) + +case class RateLimiterConfig(maxRequestsPerMinute: Int) diff --git a/forex-mtl/src/main/scala/forex/http/RateLimiter.scala b/forex-mtl/src/main/scala/forex/http/RateLimiter.scala new file mode 100644 index 00000000..994cf1f3 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/http/RateLimiter.scala @@ -0,0 +1,48 @@ +package forex.http + +import cats.data.OptionT +import cats.effect.concurrent.Ref +import cats.effect.{ Concurrent, Timer } +import cats.syntax.applicative._ +import cats.syntax.flatMap._ +import cats.syntax.functor._ +import org.http4s.{ Header, HttpRoutes, Response, Status } +import org.typelevel.ci.CIString + +import scala.concurrent.duration.MILLISECONDS + +object RateLimiter { + + def middleware[F[_]: Concurrent: Timer]( + maxRequestsPerMinute: Int + ): F[HttpRoutes[F] => HttpRoutes[F]] = + Ref.of[F, Map[String, (Int, Long)]](Map.empty).map { counter => routes => + HttpRoutes[F] { req => + val ip = req.remoteAddr.map(_.toString).getOrElse("unknown") + OptionT( + for { + now <- Timer[F].clock.realTime(MILLISECONDS) + window = now / 60000L + allowed <- counter.modify { map => + // Evict entries from previous windows on every request, bounding map size + // to the number of unique IPs active within the current minute. + val current = map.filter { case (_, (_, b)) => b == window } + val count = current.get(ip).map(_._1).getOrElse(0) + if (count >= maxRequestsPerMinute) + (current, false) + else + (current + (ip -> (count + 1, window)), true) + } + resp <- if (allowed) routes.run(req).value + else { + val secondsUntilReset = ((window + 1) * 60000L - now) / 1000L + Some( + Response[F](Status.TooManyRequests) + .putHeaders(Header.Raw(CIString("Retry-After"), secondsUntilReset.toString)) + ).pure[F] + } + } yield resp + ) + } + } +} diff --git a/forex-mtl/src/main/scala/forex/http/rates/RatesHttpRoutes.scala b/forex-mtl/src/main/scala/forex/http/rates/RatesHttpRoutes.scala index d91dcffb..d9d7b626 100644 --- a/forex-mtl/src/main/scala/forex/http/rates/RatesHttpRoutes.scala +++ b/forex-mtl/src/main/scala/forex/http/rates/RatesHttpRoutes.scala @@ -1,29 +1,66 @@ package forex.http package rates +import cats.data.Validated.{ Invalid, Valid } import cats.effect.Sync -import cats.syntax.flatMap._ +import cats.implicits._ import forex.programs.RatesProgram +import forex.programs.rates.errors.{ Error => ProgramError } import forex.programs.rates.{ Protocol => RatesProgramProtocol } +import io.circe.Json +import io.circe.syntax._ import org.http4s.HttpRoutes +import org.http4s.circe._ import org.http4s.dsl.Http4sDsl import org.http4s.server.Router +import org.slf4j.LoggerFactory class RatesHttpRoutes[F[_]: Sync](rates: RatesProgram[F]) extends Http4sDsl[F] { import Converters._, QueryParams._, Protocol._ + private val logger = LoggerFactory.getLogger(getClass) + private[http] val prefixPath = "/rates" private val httpRoutes: HttpRoutes[F] = HttpRoutes.of[F] { - case GET -> Root :? FromQueryParam(from) +& ToQueryParam(to) => - rates.get(RatesProgramProtocol.GetRatesRequest(from, to)).flatMap(Sync[F].fromEither).flatMap { rate => - Ok(rate.asGetApiResponse) + case GET -> Root :? FromQueryParam(maybeFrom) +& ToQueryParam(maybeTo) => + (maybeFrom, maybeTo) match { + case (Some(validFrom), Some(validTo)) => + (validFrom, validTo).mapN(RatesProgramProtocol.GetRatesRequest.apply) match { + case Valid(request) => + Sync[F].delay(logger.info(s"Rates lookup: ${request.from}/${request.to}")) >> + rates.get(request).flatMap { + case Right(rate) => + Ok(rate.asGetApiResponse) + + case Left(ProgramError.InvalidCurrency(input)) => + Sync[F].delay(logger.warn(s"Invalid currency: $input")) >> + BadRequest(errorBody(s"Unsupported currency: $input")) + + case Left(ProgramError.UpstreamUnavailable(msg)) => + Sync[F].delay(logger.warn(s"Upstream unavailable: $msg")) >> + BadGateway(errorBody(msg)) + + case Left(ProgramError.SystemError(msg)) => + Sync[F].delay(logger.error(s"Internal error: $msg")) >> + InternalServerError(errorBody(msg)) + } + + case Invalid(parseFailures) => + val message = parseFailures.map(_.sanitized).toList.distinct.mkString(", ") + Sync[F].delay(logger.warn(s"Bad request: $message")) >> + BadRequest(errorBody(message)) + } + + case _ => + Sync[F].delay(logger.warn("Missing from/to parameters")) >> + BadRequest(errorBody("Both 'from' and 'to' query parameters are required")) } } - val routes: HttpRoutes[F] = Router( - prefixPath -> httpRoutes - ) + val routes: HttpRoutes[F] = Router(prefixPath -> httpRoutes) + private def errorBody(message: String): Json = + Json.obj("error" -> message.asJson) } From 9c0887177cda2c8ff1d86e41cd4fb34ebe8e5a6b Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 18:53:50 +0530 Subject: [PATCH 06/11] Upstream failure safety: URI validation and cancel-safe fetch gate --- .../clients/oneframe/OneFrameClient.scala | 17 +++++++++----- .../rates/interpreters/OneFrameLive.scala | 22 ++++++++++++++----- 2 files changed, 28 insertions(+), 11 deletions(-) diff --git a/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala b/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala index 2aac0b54..d79bc79a 100644 --- a/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala +++ b/forex-mtl/src/main/scala/forex/clients/oneframe/OneFrameClient.scala @@ -39,9 +39,10 @@ trait OneFrameClientAlgebra[F[_]] { * - Auth via a raw `token ` header. * - Pair encoding: `pair=USDEUR` (concatenated currencies without separators). */ -class OneFrameHttpClient[F[_]: Sync]( +class OneFrameHttpClient[F[_]: Sync] private ( client: Client[F], - config: OneFrameConfig + config: OneFrameConfig, + baseUri: Uri ) extends OneFrameClientAlgebra[F] { private val logger = LoggerFactory.getLogger(getClass) @@ -61,7 +62,7 @@ class OneFrameHttpClient[F[_]: Sync]( .map(p => s"pair=${Currency.show.show(p.from)}${Currency.show.show(p.to)}") .mkString("&") - val uri = Uri.unsafeFromString(s"${config.baseUri}/rates?$pairParams") + val uri = Uri.unsafeFromString(s"${baseUri.renderString}/rates?$pairParams") Request[F]( method = Method.GET, @@ -108,6 +109,12 @@ object OneFrameHttpClient { def apply[F[_]: Sync]( client: Client[F], config: OneFrameConfig - ): OneFrameClientAlgebra[F] = - new OneFrameHttpClient[F](client, config) + ): F[OneFrameClientAlgebra[F]] = + Sync[F] + .fromEither( + Uri + .fromString(config.baseUri) + .leftMap(e => new IllegalArgumentException(s"Invalid One-Frame base URI '${config.baseUri}': ${e.message}")) + ) + .map(new OneFrameHttpClient[F](client, config, _)) } diff --git a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala index 26b7a38c..dcc9c7d6 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala @@ -1,8 +1,10 @@ package forex.services.rates.interpreters import cats.data.NonEmptyList +import cats.effect.ExitCase import cats.effect.concurrent.{ Deferred, Ref } import cats.effect.{ Concurrent, Resource, Timer } +import cats.effect.syntax.bracket._ import cats.syntax.applicative._ import cats.syntax.applicativeError._ import cats.syntax.either._ @@ -81,13 +83,19 @@ class OneFrameLive[F[_]: Concurrent: Timer] private ( .flatMap { case Left(ourGate) => fetchAllPairsAndPopulateCache - .flatTap(result => ourGate.complete(result)) - .flatTap(_ => fetchGate.set(None)) + .flatTap(ourGate.complete) .handleErrorWith { err => val svcError: ServiceError = ServiceError.OneFrameUnreachable(err) - ourGate.complete(svcError.asLeft) >> - fetchGate.set(None) >> - svcError.asLeft[Unit].pure[F] + ourGate.complete(svcError.asLeft).as(svcError.asLeft[Unit]) + } + .guaranteeCase { + case ExitCase.Canceled => + ourGate + .complete(ServiceError.OneFrameUnreachable(new java.util.concurrent.CancellationException("fetch cancelled")).asLeft) + .attempt + .void >> fetchGate.set(None) + case _ => + fetchGate.set(None) } case Right(theirGate) => @@ -132,7 +140,9 @@ object OneFrameLive { httpClient: Client[F], config: ApplicationConfig ): Resource[F, Algebra[F]] = - Resource.eval(make(OneFrameHttpClient[F](httpClient, config.oneFrame), config)) + Resource.eval( + OneFrameHttpClient[F](httpClient, config.oneFrame).flatMap(make(_, config)) + ) def make[F[_]: Concurrent: Timer]( client: OneFrameClientAlgebra[F], From 011ce36354251b01cd0943fcba6afc6b20a9a90a Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 19:25:09 +0530 Subject: [PATCH 07/11] Add resilience config, health checks, and tests --- forex-mtl/src/main/resources/application.conf | 6 + forex-mtl/src/main/scala/forex/Module.scala | 21 +- .../forex/config/ApplicationConfig.scala | 8 +- .../src/main/scala/forex/config/Config.scala | 11 +- .../src/main/scala/forex/domain/Rate.scala | 6 +- .../main/scala/forex/http/HealthRoutes.scala | 18 + .../src/main/scala/forex/http/RequestId.scala | 24 ++ .../src/main/scala/forex/http/package.scala | 2 +- .../scala/forex/programs/rates/errors.scala | 7 +- .../main/scala/forex/services/package.scala | 2 +- .../forex/services/rates/CircuitBreaker.scala | 78 +++++ .../forex/services/rates/Interpreters.scala | 9 + .../scala/forex/services/rates/algebra.scala | 1 + .../scala/forex/services/rates/errors.scala | 2 +- .../rates/interpreters/OneFrameDummy.scala | 1 + .../clients/oneframe/OneFrameClientSpec.scala | 138 ++++++++ .../scala/forex/domain/CurrencySpec.scala | 85 +++++ .../rates/RatesRoutesIntegrationSpec.scala | 314 ++++++++++++++++++ .../programs/rates/RatesProgramSpec.scala | 96 ++++++ .../services/rates/cache/RatesCacheSpec.scala | 138 ++++++++ 20 files changed, 943 insertions(+), 24 deletions(-) create mode 100644 forex-mtl/src/main/scala/forex/http/HealthRoutes.scala create mode 100644 forex-mtl/src/main/scala/forex/http/RequestId.scala create mode 100644 forex-mtl/src/main/scala/forex/services/rates/CircuitBreaker.scala create mode 100644 forex-mtl/src/test/scala/forex/clients/oneframe/OneFrameClientSpec.scala create mode 100644 forex-mtl/src/test/scala/forex/domain/CurrencySpec.scala create mode 100644 forex-mtl/src/test/scala/forex/http/rates/RatesRoutesIntegrationSpec.scala create mode 100644 forex-mtl/src/test/scala/forex/programs/rates/RatesProgramSpec.scala create mode 100644 forex-mtl/src/test/scala/forex/services/rates/cache/RatesCacheSpec.scala diff --git a/forex-mtl/src/main/resources/application.conf b/forex-mtl/src/main/resources/application.conf index ae455362..aa3fc366 100644 --- a/forex-mtl/src/main/resources/application.conf +++ b/forex-mtl/src/main/resources/application.conf @@ -9,6 +9,12 @@ app { base-uri = ${?ONE_FRAME_BASE_URI} or "http://localhost:8080" auth-token = ${?ONE_FRAME_TOKEN} or "" timeout = 10 seconds + max-retries = 3 + } + + circuit-breaker { + max-failures = 5 + reset-timeout = 60 seconds } cache { diff --git a/forex-mtl/src/main/scala/forex/Module.scala b/forex-mtl/src/main/scala/forex/Module.scala index 0c1bbed0..52a4143d 100644 --- a/forex-mtl/src/main/scala/forex/Module.scala +++ b/forex-mtl/src/main/scala/forex/Module.scala @@ -1,8 +1,9 @@ package forex import cats.effect.{ Concurrent, Resource, Timer } +import cats.syntax.semigroupk._ import forex.config.ApplicationConfig -import forex.http.RateLimiter +import forex.http.{ HealthRoutes, RateLimiter, RequestId } import forex.http.rates.RatesHttpRoutes import forex.programs.RatesProgram import forex.services.{ RatesService, RatesServiceFactory } @@ -19,11 +20,19 @@ class Module[F[_]: Concurrent: Timer]( type PartialMiddleware = HttpRoutes[F] => HttpRoutes[F] type TotalMiddleware = HttpApp[F] => HttpApp[F] - private val ratesProgram: RatesProgram[F] = RatesProgram[F](ratesService) - private val ratesHttpRoutes: HttpRoutes[F] = new RatesHttpRoutes[F](ratesProgram).routes + private val ratesProgram: RatesProgram[F] = RatesProgram[F](ratesService) + private val ratesHttpRoutes: HttpRoutes[F] = new RatesHttpRoutes[F](ratesProgram).routes + private val healthRoutes: HttpRoutes[F] = new HealthRoutes[F](ratesService.isReady).routes + private val allRoutes: HttpRoutes[F] = ratesHttpRoutes <+> healthRoutes private val routesMiddleware: PartialMiddleware = AutoSlash(_) - private val appMiddleware: TotalMiddleware = Timeout(config.http.timeout)(_) - val httpApp: HttpApp[F] = appMiddleware(rateLimiter(routesMiddleware(ratesHttpRoutes)).orNotFound) + private val appMiddleware: TotalMiddleware = Timeout(config.http.timeout)(_) + + val httpApp: HttpApp[F] = + appMiddleware( + rateLimiter( + RequestId.middleware(routesMiddleware(allRoutes)) + ).orNotFound + ) } object Module { @@ -34,6 +43,6 @@ object Module { ): Resource[F, Module[F]] = for { ratesService <- RatesServiceFactory.live[F](httpClient, config) - rateLimiter <- Resource.eval(RateLimiter.middleware[F](config.rateLimiter.maxRequestsPerMinute)) + rateLimiter <- Resource.eval(RateLimiter.middleware[F](config.rateLimiter.maxRequestsPerMinute)) } yield new Module[F](config, ratesService, rateLimiter) } diff --git a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala index 4b12cab0..d64cd39e 100644 --- a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala +++ b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala @@ -6,7 +6,8 @@ case class ApplicationConfig( http: HttpConfig, oneFrame: OneFrameConfig, cache: CacheConfig, - rateLimiter: RateLimiterConfig + rateLimiter: RateLimiterConfig, + circuitBreaker: CircuitBreakerConfig ) case class HttpConfig( @@ -18,7 +19,8 @@ case class HttpConfig( case class OneFrameConfig( baseUri: String, authToken: String, - timeout: FiniteDuration + timeout: FiniteDuration, + maxRetries: Int ) case class CacheConfig( @@ -28,3 +30,5 @@ case class CacheConfig( ) case class RateLimiterConfig(maxRequestsPerMinute: Int) + +case class CircuitBreakerConfig(maxFailures: Int, resetTimeout: FiniteDuration) diff --git a/forex-mtl/src/main/scala/forex/config/Config.scala b/forex-mtl/src/main/scala/forex/config/Config.scala index 0181788e..88e189c4 100644 --- a/forex-mtl/src/main/scala/forex/config/Config.scala +++ b/forex-mtl/src/main/scala/forex/config/Config.scala @@ -8,12 +8,9 @@ import pureconfig.generic.auto._ object Config { - /** - * @param path the property path inside the default configuration - */ - def stream[F[_]: Sync](path: String): Stream[F, ApplicationConfig] = { - Stream.eval(Sync[F].delay( - ConfigSource.default.at(path).loadOrThrow[ApplicationConfig])) - } + /** @param path the property path inside the default configuration + */ + def stream[F[_]: Sync](path: String): Stream[F, ApplicationConfig] = + Stream.eval(Sync[F].delay(ConfigSource.default.at(path).loadOrThrow[ApplicationConfig])) } diff --git a/forex-mtl/src/main/scala/forex/domain/Rate.scala b/forex-mtl/src/main/scala/forex/domain/Rate.scala index 542034f5..a5713ce8 100644 --- a/forex-mtl/src/main/scala/forex/domain/Rate.scala +++ b/forex-mtl/src/main/scala/forex/domain/Rate.scala @@ -8,14 +8,14 @@ case class Rate( object Rate { final case class Pair( - from: Currency, - to: Currency + from: Currency, + to: Currency ) object Pair { val allPairs: List[Pair] = for { from <- Currency.values - to <- Currency.values + to <- Currency.values if from != to } yield Pair(from, to) } diff --git a/forex-mtl/src/main/scala/forex/http/HealthRoutes.scala b/forex-mtl/src/main/scala/forex/http/HealthRoutes.scala new file mode 100644 index 00000000..179c54c4 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/http/HealthRoutes.scala @@ -0,0 +1,18 @@ +package forex.http + +import cats.effect.Sync +import cats.syntax.flatMap._ +import org.http4s.HttpRoutes +import org.http4s.dsl.Http4sDsl + +class HealthRoutes[F[_]: Sync](isReady: F[Boolean]) extends Http4sDsl[F] { + + val routes: HttpRoutes[F] = HttpRoutes.of[F] { + case GET -> Root / "health" / "live" => Ok("OK") + case GET -> Root / "health" / "ready" => + isReady.flatMap { + case true => Ok("OK") + case false => ServiceUnavailable("cache cold") + } + } +} diff --git a/forex-mtl/src/main/scala/forex/http/RequestId.scala b/forex-mtl/src/main/scala/forex/http/RequestId.scala new file mode 100644 index 00000000..d9105da5 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/http/RequestId.scala @@ -0,0 +1,24 @@ +package forex.http + +import cats.data.OptionT +import cats.effect.Sync +import cats.syntax.functor._ +import org.http4s.{ Header, HttpRoutes } +import org.typelevel.ci.CIString + +import java.util.UUID + +object RequestId { + + private val headerName = CIString("X-Request-ID") + + def middleware[F[_]: Sync](routes: HttpRoutes[F]): HttpRoutes[F] = + HttpRoutes[F] { req => + val id = req.headers.get(headerName).map(_.head.value).getOrElse(UUID.randomUUID().toString) + val taggedReq = req.putHeaders(Header.Raw(headerName, id)) + OptionT( + routes.run(taggedReq).value + .map(_.map(_.putHeaders(Header.Raw(headerName, id)))) + ) + } +} diff --git a/forex-mtl/src/main/scala/forex/http/package.scala b/forex-mtl/src/main/scala/forex/http/package.scala index 1ffafa5d..1705f9fd 100644 --- a/forex-mtl/src/main/scala/forex/http/package.scala +++ b/forex-mtl/src/main/scala/forex/http/package.scala @@ -16,6 +16,6 @@ package object http { implicit def enumDecoder[A: EnumerationDecoder]: Decoder[A] = implicitly implicit def jsonDecoder[A <: Product: Decoder, F[_]: Sync]: EntityDecoder[F, A] = jsonOf[F, A] - implicit def jsonEncoder[A <: Product: Encoder, F[_]]: EntityEncoder[F, A] = jsonEncoderOf[F, A] + implicit def jsonEncoder[A <: Product: Encoder, F[_]]: EntityEncoder[F, A] = jsonEncoderOf[F, A] } diff --git a/forex-mtl/src/main/scala/forex/programs/rates/errors.scala b/forex-mtl/src/main/scala/forex/programs/rates/errors.scala index af8b96a8..d9ff2cac 100644 --- a/forex-mtl/src/main/scala/forex/programs/rates/errors.scala +++ b/forex-mtl/src/main/scala/forex/programs/rates/errors.scala @@ -22,8 +22,9 @@ object errors { } def toProgramError(error: RatesServiceError): Error = error match { - case RatesServiceError.OneFrameQuotaExceeded => Error.UpstreamUnavailable("One-Frame API quota exceeded for today") - case RatesServiceError.OneFrameUnreachable(cause) => Error.UpstreamUnavailable(s"One-Frame is unreachable: ${cause.getMessage}") - case RatesServiceError.OneFrameLookupFailed(msg) => Error.UpstreamUnavailable(s"One-Frame error: $msg") + case RatesServiceError.OneFrameQuotaExceeded => Error.UpstreamUnavailable("One-Frame API quota exceeded for today") + case RatesServiceError.OneFrameUnreachable(cause) => + Error.UpstreamUnavailable(s"One-Frame is unreachable: ${cause.getMessage}") + case RatesServiceError.OneFrameLookupFailed(msg) => Error.UpstreamUnavailable(s"One-Frame error: $msg") } } diff --git a/forex-mtl/src/main/scala/forex/services/package.scala b/forex-mtl/src/main/scala/forex/services/package.scala index aed4912d..cab734ea 100644 --- a/forex-mtl/src/main/scala/forex/services/package.scala +++ b/forex-mtl/src/main/scala/forex/services/package.scala @@ -2,5 +2,5 @@ package forex package object services { type RatesService[F[_]] = rates.Algebra[F] - final val RatesServices = rates.Interpreters + final val RatesServiceFactory = rates.Interpreters } diff --git a/forex-mtl/src/main/scala/forex/services/rates/CircuitBreaker.scala b/forex-mtl/src/main/scala/forex/services/rates/CircuitBreaker.scala new file mode 100644 index 00000000..e8fc7f0c --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/rates/CircuitBreaker.scala @@ -0,0 +1,78 @@ +package forex.services.rates + +import cats.data.NonEmptyList +import cats.effect.concurrent.Ref +import cats.effect.{ Concurrent, Timer } +import cats.syntax.applicative._ +import cats.syntax.either._ +import cats.syntax.flatMap._ +import cats.syntax.functor._ +import forex.clients.oneframe.OneFrameClientAlgebra +import forex.domain.Rate +import forex.services.rates.errors.{ Error => ServiceError } + +import scala.concurrent.duration.{ FiniteDuration, MILLISECONDS } + +private sealed trait CBState +private object CBState { + final case class Closed(failures: Int) extends CBState + final case class Open(openedAt: Long) extends CBState + case object HalfOpen extends CBState +} + +class CircuitBreaker[F[_]: Concurrent: Timer] private ( + underlying: OneFrameClientAlgebra[F], + state: Ref[F, CBState], + maxFailures: Int, + resetTimeout: FiniteDuration +) extends OneFrameClientAlgebra[F] { + + override def getRates(pairs: NonEmptyList[Rate.Pair]): F[Either[ServiceError, List[Rate]]] = + now.flatMap { ts => + state.get.flatMap { + case CBState.Open(openedAt) if ts - openedAt >= resetTimeout.toMillis => + state.set(CBState.HalfOpen) >> probe(pairs) + case CBState.Open(_) => + (ServiceError.OneFrameUnreachable(new RuntimeException("circuit breaker open")): ServiceError).asLeft[List[Rate]].pure[F] + case CBState.HalfOpen => + probe(pairs) + case CBState.Closed(_) => + call(pairs) + } + } + + private def call(pairs: NonEmptyList[Rate.Pair]): F[Either[ServiceError, List[Rate]]] = + underlying.getRates(pairs).flatTap { + case Right(_) => + state.update { case CBState.Closed(_) => CBState.Closed(0); case s => s } + case Left(ServiceError.OneFrameUnreachable(_)) => + now.flatMap { ts => + state.modify { + case CBState.Closed(n) if n + 1 >= maxFailures => (CBState.Open(ts), ()) + case CBState.Closed(n) => (CBState.Closed(n + 1), ()) + case s => (s, ()) + } + } + case _ => Concurrent[F].unit + } + + private def probe(pairs: NonEmptyList[Rate.Pair]): F[Either[ServiceError, List[Rate]]] = + underlying.getRates(pairs).flatTap { + case Right(_) => state.set(CBState.Closed(0)) + case Left(ServiceError.OneFrameUnreachable(_)) => now.flatMap(ts => state.set(CBState.Open(ts))) + case _ => Concurrent[F].unit + } + + private def now: F[Long] = Timer[F].clock.realTime(MILLISECONDS) +} + +object CircuitBreaker { + def wrap[F[_]: Concurrent: Timer]( + underlying: OneFrameClientAlgebra[F], + maxFailures: Int, + resetTimeout: FiniteDuration + ): F[OneFrameClientAlgebra[F]] = + Ref + .of[F, CBState](CBState.Closed(0)) + .map(new CircuitBreaker(underlying, _, maxFailures, resetTimeout)) +} diff --git a/forex-mtl/src/main/scala/forex/services/rates/Interpreters.scala b/forex-mtl/src/main/scala/forex/services/rates/Interpreters.scala index e523ffab..99e33f56 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/Interpreters.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/Interpreters.scala @@ -1,8 +1,17 @@ package forex.services.rates import cats.Applicative +import cats.effect.{ Concurrent, Resource, Timer } +import forex.config.ApplicationConfig import interpreters._ +import org.http4s.client.Client object Interpreters { def dummy[F[_]: Applicative]: Algebra[F] = new OneFrameDummy[F]() + + def live[F[_]: Concurrent: Timer]( + httpClient: Client[F], + config: ApplicationConfig + ): Resource[F, Algebra[F]] = + OneFrameLive.resource[F](httpClient, config) } diff --git a/forex-mtl/src/main/scala/forex/services/rates/algebra.scala b/forex-mtl/src/main/scala/forex/services/rates/algebra.scala index 8966dce5..6c4198d0 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/algebra.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/algebra.scala @@ -5,4 +5,5 @@ import errors._ trait Algebra[F[_]] { def get(pair: Rate.Pair): F[Error Either Rate] + def isReady: F[Boolean] } diff --git a/forex-mtl/src/main/scala/forex/services/rates/errors.scala b/forex-mtl/src/main/scala/forex/services/rates/errors.scala index a43ad2f8..71956083 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/errors.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/errors.scala @@ -8,5 +8,5 @@ object errors { final case class OneFrameUnreachable(cause: Throwable) extends Error final case class OneFrameLookupFailed(msg: String) extends Error } - + } diff --git a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameDummy.scala b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameDummy.scala index 37a3f50c..e5a5fa00 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameDummy.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameDummy.scala @@ -12,4 +12,5 @@ class OneFrameDummy[F[_]: Applicative] extends Algebra[F] { override def get(pair: Rate.Pair): F[Error Either Rate] = Rate(pair, Price(BigDecimal(100)), Timestamp.now).asRight[Error].pure[F] + override def isReady: F[Boolean] = true.pure[F] } diff --git a/forex-mtl/src/test/scala/forex/clients/oneframe/OneFrameClientSpec.scala b/forex-mtl/src/test/scala/forex/clients/oneframe/OneFrameClientSpec.scala new file mode 100644 index 00000000..51034fcb --- /dev/null +++ b/forex-mtl/src/test/scala/forex/clients/oneframe/OneFrameClientSpec.scala @@ -0,0 +1,138 @@ +package forex.clients.oneframe + +import cats.data.NonEmptyList +import cats.effect.{ IO, Resource } +import forex.config.OneFrameConfig +import forex.domain.{ Currency, Rate } +import forex.services.rates.errors.{ Error => ServiceError } +import org.http4s._ +import org.http4s.client.Client +import org.scalatest.EitherValues +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec +import org.typelevel.ci.CIString +import scala.concurrent.duration._ + +class OneFrameClientSpec extends AnyWordSpec with Matchers with EitherValues { + + private val config = OneFrameConfig( + baseUri = "http://one-frame", + authToken = "test-token", + timeout = 10.seconds, + maxRetries = 0 + ) + + private val usdEur = Rate.Pair(Currency.USD, Currency.EUR) + private val usdGbp = Rate.Pair(Currency.USD, Currency.GBP) + + private def clientFrom(body: String, status: Status = Status.Ok): Client[IO] = + Client.fromHttpApp[IO](HttpApp.pure(Response[IO](status = status).withEntity(body))) + + private def makeClient(httpClient: Client[IO]): OneFrameClientAlgebra[IO] = + OneFrameHttpClient[IO](httpClient, config).unsafeRunSync() + + private val validBody = + """[{ + | "from":"USD","to":"EUR", + | "bid":0.8532,"ask":0.8540,"price":0.8536, + | "time_stamp":"2024-01-15T10:30:00+00:00" + |}]""".stripMargin + + "OneFrameHttpClient.getRates" should { + + "return a list of rates for a well-formed response" in { + val result = makeClient(clientFrom(validBody)) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + + val rates = result.value + rates should have size 1 + rates.head.pair shouldBe usdEur + rates.head.price.value shouldBe BigDecimal("0.8536") + } + + "parse the ISO-8601 timestamp" in { + val rate = makeClient(clientFrom(validBody)) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + .value + .head + + rate.timestamp.value.getYear shouldBe 2024 + rate.timestamp.value.getDayOfMonth shouldBe 15 + } + + "return QuotaExceeded when One-Frame reports quota exhaustion" in { + val quotaBody = """{"error":"Quota reached for token abc123"}""" + val result = makeClient(clientFrom(quotaBody)) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + + result.left.value shouldBe ServiceError.OneFrameQuotaExceeded + } + + "return OneFrameLookupFailed for other One-Frame error messages" in { + val errorBody = """{"error":"Invalid authentication credentials"}""" + val result = makeClient(clientFrom(errorBody)) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + + result.left.value should matchPattern { case ServiceError.OneFrameLookupFailed(_) => } + } + + "return OneFrameUnreachable when the HTTP client raises an exception" in { + val failingClient = Client[IO](_ => Resource.eval(IO.raiseError(new RuntimeException("connection refused")))) + val result = makeClient(failingClient) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + + result.left.value should matchPattern { case ServiceError.OneFrameUnreachable(_) => } + } + + "return OneFrameLookupFailed for a non-JSON response" in { + val result = makeClient(clientFrom("not json")) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + + result.left.value should matchPattern { case ServiceError.OneFrameLookupFailed(_) => } + } + + "send the auth token in the request header" in { + var capturedToken: Option[String] = None + val capturingApp = HttpApp[IO] { req => + capturedToken = req.headers.get(CIString("token")).map(_.head.value) + IO.pure(Response[IO]().withEntity(validBody)) + } + + makeClient(Client.fromHttpApp(capturingApp)) + .getRates(NonEmptyList.one(usdEur)) + .unsafeRunSync() + + capturedToken shouldBe Some("test-token") + } + + "encode multiple pairs as separate 'pair' query parameters" in { + var capturedUri: Option[Uri] = None + val capturingApp = HttpApp[IO] { req => + capturedUri = Some(req.uri) + IO.pure(Response[IO]().withEntity("[]")) + } + + makeClient(Client.fromHttpApp(capturingApp)) + .getRates(NonEmptyList.of(usdEur, usdGbp)) + .unsafeRunSync() + + val qs = capturedUri.map(_.query.renderString).getOrElse("") + qs should include("pair=USDEUR") + qs should include("pair=USDGBP") + } + + "fail at construction with a clear error for a malformed base URI" in { + val badConfig = config.copy(baseUri = "not a uri !!!") + val result = OneFrameHttpClient[IO](clientFrom(validBody), badConfig).attempt.unsafeRunSync() + + result.left.value shouldBe an[IllegalArgumentException] + result.left.value.getMessage should include("not a uri !!!") + } + } +} diff --git a/forex-mtl/src/test/scala/forex/domain/CurrencySpec.scala b/forex-mtl/src/test/scala/forex/domain/CurrencySpec.scala new file mode 100644 index 00000000..db1669e3 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/domain/CurrencySpec.scala @@ -0,0 +1,85 @@ +package forex.domain + +import org.scalatest.EitherValues +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class CurrencySpec extends AnyWordSpec with Matchers with EitherValues { + + "Currency.fromString" should { + + "parse every supported currency code" in { + val codes = List("AUD", "CAD", "CHF", "EUR", "GBP", "NZD", "JPY", "SGD", "USD") + codes.foreach { code => + Currency.fromString(code).value shouldBe a[Currency] + } + } + + "be case-insensitive" in { + Currency.fromString("usd").value shouldBe Currency.USD + Currency.fromString("Eur").value shouldBe Currency.EUR + Currency.fromString("jPy").value shouldBe Currency.JPY + } + + "return Left for unsupported codes" in { + Currency.fromString("XYZ").left.value should include("XYZ") + Currency.fromString("BTC").left.value should include("BTC") + } + + "return Left for an empty string" in { + Currency.fromString("").isLeft shouldBe true + } + + "return Left for whitespace-only input" in { + Currency.fromString(" ").isLeft shouldBe true + } + } + + "Currency.values" should { + + "contain exactly 9 currencies" in { + Currency.values should have size 9 + } + + "contain no duplicates" in { + Currency.values.distinct.size shouldBe Currency.values.size + } + } + + "Rate.Pair.allPairs" should { + + "contain exactly 72 pairs (9 × 8)" in { + Rate.Pair.allPairs should have size 72 + } + + "never include a pair where from == to" in { + Rate.Pair.allPairs.foreach { pair => + pair.from should not be pair.to + } + } + + "include every ordered combination of supported currencies" in { + val expected = for { + from <- Currency.values + to <- Currency.values + if from != to + } yield Rate.Pair(from, to) + + Rate.Pair.allPairs should contain theSameElementsAs expected + } + } + + "Currency.show" should { + + "produce the ISO 4217 code string" in { + Currency.show.show(Currency.USD) shouldBe "USD" + Currency.show.show(Currency.JPY) shouldBe "JPY" + } + + "round-trip through fromString for all currencies" in { + Currency.values.foreach { currency => + Currency.fromString(Currency.show.show(currency)).value shouldBe currency + } + } + } +} diff --git a/forex-mtl/src/test/scala/forex/http/rates/RatesRoutesIntegrationSpec.scala b/forex-mtl/src/test/scala/forex/http/rates/RatesRoutesIntegrationSpec.scala new file mode 100644 index 00000000..2fcd6aa7 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/http/rates/RatesRoutesIntegrationSpec.scala @@ -0,0 +1,314 @@ +package forex.http.rates + +import cats.effect.{ ContextShift, IO, Timer } +import cats.syntax.parallel._ +import forex.clients.oneframe.OneFrameHttpClient +import forex.config.{ ApplicationConfig, CacheConfig, CircuitBreakerConfig, HttpConfig, OneFrameConfig, RateLimiterConfig } +import forex.domain.{ Currency, Price, Rate, Timestamp } +import forex.http.RateLimiter +import forex.programs.rates.Program +import forex.services.rates.cache.InMemoryRatesCache +import forex.services.rates.interpreters.OneFrameLive +import org.http4s._ +import org.http4s.circe._ +import org.http4s.client.Client +import org.http4s.implicits._ +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec +import java.time.OffsetDateTime +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.ExecutionContext.global +import scala.concurrent.duration._ + +class RatesRoutesIntegrationSpec extends AnyWordSpec with Matchers { + + implicit val cs: ContextShift[IO] = IO.contextShift(global) + implicit val timer: Timer[IO] = IO.timer(global) + + private val testConfig = ApplicationConfig( + http = HttpConfig("0.0.0.0", 8080, 40.seconds), + oneFrame = OneFrameConfig("http://one-frame", "test-token", 10.seconds, maxRetries = 0), + cache = CacheConfig( + ttl = 5.minutes, + softTtl = 4.minutes, + maxStaleOnError = 10.minutes + ), + rateLimiter = RateLimiterConfig(maxRequestsPerMinute = 100), + circuitBreaker = CircuitBreakerConfig(maxFailures = 5, resetTimeout = 60.seconds) + ) + + private def allPairsJsonWithPrice(price: BigDecimal): String = { + val entries = Rate.Pair.allPairs.map { pair => + s"""{"from":"${Currency.show.show(pair.from)}","to":"${Currency.show.show(pair.to)}",""" + + s""""bid":${price - 0.01},"ask":${price + 0.01},"price":$price,"time_stamp":"2024-06-01T12:00:00+00:00"}""" + } + s"[${entries.mkString(",")}]" + } + + private def allPairsJson: String = allPairsJsonWithPrice(BigDecimal("0.85")) + + private def routesUsing(oneFrameApp: HttpApp[IO]): HttpRoutes[IO] = { + val httpClient = Client.fromHttpApp[IO](oneFrameApp) + val oneFrameClient = OneFrameHttpClient[IO](httpClient, testConfig.oneFrame).unsafeRunSync() + val interpreter = OneFrameLive.make[IO](oneFrameClient, testConfig).unsafeRunSync() + val program = Program[IO](interpreter) + new RatesHttpRoutes[IO](program).routes + } + + private def routesWithSeededCache( + oneFrameApp: HttpApp[IO], + seedRates: List[Rate], + fetchedAt: OffsetDateTime + ): IO[HttpRoutes[IO]] = + for { + cache <- InMemoryRatesCache.create[IO] + _ <- cache.putBatch(seedRates, fetchedAt) + httpClient = Client.fromHttpApp[IO](oneFrameApp) + oneFrameClient <- OneFrameHttpClient[IO](httpClient, testConfig.oneFrame) + interpreter <- OneFrameLive.makeWithCache[IO](oneFrameClient, cache, testConfig) + } yield new RatesHttpRoutes[IO](Program[IO](interpreter)).routes + + private def allRatesWith(price: BigDecimal): List[Rate] = + Rate.Pair.allPairs.map(p => Rate(p, Price(price), Timestamp.now)) + + private def requestIO(routes: HttpRoutes[IO], uri: Uri): IO[Response[IO]] = + routes.run(Request[IO](Method.GET, uri)).getOrElse(Response.notFound) + + "GET /rates" when { + + "both currencies are valid and One-Frame responds successfully" should { + + "return 200 with from, to, price, and timestamp" in { + val routes = routesUsing(HttpApp.pure(Response[IO]().withEntity(allPairsJson))) + val response = requestIO(routes, uri"/rates?from=USD&to=EUR").unsafeRunSync() + + response.status shouldBe Status.Ok + + val body = response.as[io.circe.Json].unsafeRunSync() + body.hcursor.get[String]("from").toOption shouldBe Some("USD") + body.hcursor.get[String]("to").toOption shouldBe Some("EUR") + body.hcursor.get[BigDecimal]("price").toOption shouldBe Some(BigDecimal("0.85")) + } + } + + "from == to" should { + + "return 200 with price 1.0 without calling One-Frame" in { + var oneFrameCalled = false + val capturingApp = HttpApp[IO] { _ => + IO { oneFrameCalled = true } *> + IO.pure(Response[IO]().withEntity(allPairsJson)) + } + + val response = requestIO(routesUsing(capturingApp), uri"/rates?from=USD&to=USD").unsafeRunSync() + + response.status shouldBe Status.Ok + oneFrameCalled shouldBe false + + val body = response.as[io.circe.Json].unsafeRunSync() + body.hcursor.get[BigDecimal]("price").toOption shouldBe Some(BigDecimal(1)) + } + } + + "an unrecognised currency code is given" should { + + "return 400 mentioning the bad code" in { + val routes = routesUsing(HttpApp.pure(Response[IO]().withEntity(allPairsJson))) + val response = requestIO(routes, uri"/rates?from=XYZ&to=USD").unsafeRunSync() + + response.status shouldBe Status.BadRequest + + val message = response.as[io.circe.Json].unsafeRunSync().hcursor.get[String]("error").getOrElse("") + message should include("XYZ") + } + } + + "the 'from' parameter is missing" should { + + "return 400" in { + val routes = routesUsing(HttpApp.pure(Response[IO]().withEntity(allPairsJson))) + val response = requestIO(routes, uri"/rates?to=EUR").unsafeRunSync() + response.status shouldBe Status.BadRequest + } + } + + "the 'to' parameter is missing" should { + + "return 400" in { + val routes = routesUsing(HttpApp.pure(Response[IO]().withEntity(allPairsJson))) + val response = requestIO(routes, uri"/rates?from=USD").unsafeRunSync() + response.status shouldBe Status.BadRequest + } + } + + "One-Frame reports quota exhaustion" should { + + "return 502 Bad Gateway" in { + val quotaBody = """{"error":"Quota reached for token test-token"}""" + val response = requestIO( + routesUsing(HttpApp.pure(Response[IO]().withEntity(quotaBody))), + uri"/rates?from=USD&to=EUR" + ).unsafeRunSync() + + response.status shouldBe Status.BadGateway + } + } + + "One-Frame returns an HTTP 500" should { + + "return 502 Bad Gateway" in { + val errorApp = HttpApp.pure[IO](Response[IO](status = Status.InternalServerError)) + val response = requestIO(routesUsing(errorApp), uri"/rates?from=USD&to=EUR").unsafeRunSync() + response.status shouldBe Status.BadGateway + } + } + + "ten concurrent requests arrive with a cold cache" should { + + "make exactly one upstream call (Deferred coalescing)" in { + val callCount = new AtomicInteger(0) + val countingApp = HttpApp[IO] { _ => + IO(callCount.incrementAndGet()) *> + IO.pure(Response[IO]().withEntity(allPairsJson)) + } + + val routes = routesUsing(countingApp) + + List + .fill(10)(requestIO(routes, uri"/rates?from=USD&to=EUR")) + .parSequence + .unsafeRunSync() + + callCount.get() shouldBe 1 + } + } + + "a second request arrives after the cache is warm" should { + + "serve the cached rate without a second One-Frame call" in { + val callCount = new AtomicInteger(0) + val countingApp = HttpApp[IO] { _ => + IO(callCount.incrementAndGet()) *> + IO.pure(Response[IO]().withEntity(allPairsJson)) + } + + val routes = routesUsing(countingApp) + + requestIO(routes, uri"/rates?from=USD&to=EUR").unsafeRunSync() + requestIO(routes, uri"/rates?from=USD&to=EUR").unsafeRunSync() + + callCount.get() shouldBe 1 + } + } + + "the cached rate is between softTtl and hardTtl (stale-while-revalidate)" should { + + "serve the stale rate immediately and trigger a background revalidation" in { + val stalePrice = BigDecimal("0.777") + val freshPrice = BigDecimal("0.999") + val callCount = new AtomicInteger(0) + val countingApp = HttpApp[IO] { _ => + IO(callCount.incrementAndGet()) *> + IO.pure(Response[IO]().withEntity(allPairsJsonWithPrice(freshPrice))) + } + + val staleTime = OffsetDateTime.now().minusSeconds(270) + + val (status, body) = (for { + routes <- routesWithSeededCache(countingApp, allRatesWith(stalePrice), staleTime) + response <- requestIO(routes, uri"/rates?from=USD&to=EUR") + body <- response.as[io.circe.Json] + _ <- IO.sleep(300.millis) + } yield (response.status, body)).unsafeRunSync() + + status shouldBe Status.Ok + body.hcursor.get[BigDecimal]("price").toOption shouldBe Some(stalePrice) + callCount.get() shouldBe 1 + } + } + + "the cached rate has expired past hardTtl and One-Frame responds" should { + + "re-fetch synchronously and return the fresh rate" in { + val oldPrice = BigDecimal("0.111") + val freshPrice = BigDecimal("0.999") + val freshApp = HttpApp.pure[IO](Response[IO]().withEntity(allPairsJsonWithPrice(freshPrice))) + + val expiredTime = OffsetDateTime.now().minusSeconds(330) + + val (status, body) = (for { + routes <- routesWithSeededCache(freshApp, allRatesWith(oldPrice), expiredTime) + response <- requestIO(routes, uri"/rates?from=USD&to=EUR") + body <- response.as[io.circe.Json] + } yield (response.status, body)).unsafeRunSync() + + status shouldBe Status.Ok + body.hcursor.get[BigDecimal]("price").toOption shouldBe Some(freshPrice) + } + } + + "One-Frame is unreachable and the cache is within maxStaleOnError (configured to 10 min)" should { + + "serve the stale rate rather than returning an error" in { + val stalePrice = BigDecimal("0.888") + val failingApp = HttpApp[IO](_ => IO.raiseError(new RuntimeException("connection refused"))) + + val expiredTime = OffsetDateTime.now().minusSeconds(360) + + val (status, price) = (for { + routes <- routesWithSeededCache(failingApp, allRatesWith(stalePrice), expiredTime) + response <- requestIO(routes, uri"/rates?from=USD&to=EUR") + body <- response.as[io.circe.Json] + } yield (response.status, body.hcursor.get[BigDecimal]("price").toOption)).unsafeRunSync() + + status shouldBe Status.Ok + price shouldBe Some(stalePrice) + } + } + + "One-Frame is unreachable and the cache exceeds maxStaleOnError" should { + + "return 502 Bad Gateway" in { + val failingApp = HttpApp[IO](_ => IO.raiseError(new RuntimeException("connection refused"))) + + val tooOldTime = OffsetDateTime.now().minusMinutes(15) + + val status = (for { + routes <- routesWithSeededCache(failingApp, allRatesWith(BigDecimal("0.5")), tooOldTime) + response <- requestIO(routes, uri"/rates?from=USD&to=EUR") + } yield response.status).unsafeRunSync() + + status shouldBe Status.BadGateway + } + } + + "One-Frame is unreachable and the cache is empty" should { + + "return 502 Bad Gateway" in { + val failingApp = HttpApp[IO](_ => IO.raiseError(new RuntimeException("connection refused"))) + val response = requestIO(routesUsing(failingApp), uri"/rates?from=USD&to=EUR").unsafeRunSync() + response.status shouldBe Status.BadGateway + } + } + + "more requests than the per-minute limit arrive from the same IP" should { + + "return 429 Too Many Requests once the limit is exceeded" in { + val limit = 3 + val routes = RateLimiter + .middleware[IO](limit) + .map { rl => + rl(routesUsing(HttpApp.pure(Response[IO]().withEntity(allPairsJson)))) + } + .unsafeRunSync() + + val responses = List.fill(limit + 1) { + requestIO(routes, uri"/rates?from=USD&to=EUR").unsafeRunSync() + } + + responses.take(limit).foreach(_.status shouldBe Status.Ok) + responses.last.status shouldBe Status.TooManyRequests + } + } + } +} diff --git a/forex-mtl/src/test/scala/forex/programs/rates/RatesProgramSpec.scala b/forex-mtl/src/test/scala/forex/programs/rates/RatesProgramSpec.scala new file mode 100644 index 00000000..d9a0131a --- /dev/null +++ b/forex-mtl/src/test/scala/forex/programs/rates/RatesProgramSpec.scala @@ -0,0 +1,96 @@ +package forex.programs.rates + +import cats.effect.IO +import forex.domain.{ Currency, Price, Rate, Timestamp } +import forex.programs.rates.errors.Error +import forex.services.rates.{ Algebra => ServiceAlgebra } +import forex.services.rates.errors.{ Error => ServiceError } +import org.scalatest.EitherValues +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class RatesProgramSpec extends AnyWordSpec with Matchers with EitherValues { + + private val usd = Currency.USD + private val eur = Currency.EUR + private val dummyRate = Rate(Rate.Pair(usd, eur), Price(BigDecimal("0.85")), Timestamp.now) + + private def programWith(serviceResult: Either[ServiceError, Rate]): Algebra[IO] = { + val stubService = new ServiceAlgebra[IO] { + override def get(pair: Rate.Pair): IO[Either[ServiceError, Rate]] = IO.pure(serviceResult) + override def isReady: IO[Boolean] = IO.pure(true) + } + Program[IO](stubService) + } + + "Program.get" when { + + "from == to" should { + + "return price 1.0 without calling the service" in { + var serviceCalled = false + val service = new ServiceAlgebra[IO] { + override def get(pair: Rate.Pair): IO[Either[ServiceError, Rate]] = { + serviceCalled = true + IO.pure(Right(dummyRate)) + } + override def isReady: IO[Boolean] = IO.pure(true) + } + val program = Program[IO](service) + val result = program.get(Protocol.GetRatesRequest(usd, usd)).unsafeRunSync() + + serviceCalled shouldBe false + result.value.price shouldBe Price(BigDecimal(1)) + } + + "return a rate with matching from and to" in { + val program = programWith(Right(dummyRate)) + val result = program.get(Protocol.GetRatesRequest(eur, eur)).unsafeRunSync() + + result.value.pair.from shouldBe eur + result.value.pair.to shouldBe eur + } + } + + "the service returns a rate" should { + + "pass the rate through to the caller" in { + val program = programWith(Right(dummyRate)) + val result = program.get(Protocol.GetRatesRequest(usd, eur)).unsafeRunSync() + + result.value shouldBe dummyRate + } + } + + "the service returns QuotaExceeded" should { + + "map to UpstreamUnavailable" in { + val program = programWith(Left(ServiceError.OneFrameQuotaExceeded)) + val result = program.get(Protocol.GetRatesRequest(usd, eur)).unsafeRunSync() + + result.left.value shouldBe a[Error.UpstreamUnavailable] + } + } + + "the service returns OneFrameUnreachable" should { + + "map to UpstreamUnavailable" in { + val cause = new RuntimeException("connection refused") + val program = programWith(Left(ServiceError.OneFrameUnreachable(cause))) + val result = program.get(Protocol.GetRatesRequest(usd, eur)).unsafeRunSync() + + result.left.value shouldBe a[Error.UpstreamUnavailable] + } + } + + "the service returns OneFrameLookupFailed" should { + + "map to UpstreamUnavailable" in { + val program = programWith(Left(ServiceError.OneFrameLookupFailed("timeout"))) + val result = program.get(Protocol.GetRatesRequest(usd, eur)).unsafeRunSync() + + result.left.value shouldBe a[Error.UpstreamUnavailable] + } + } + } +} diff --git a/forex-mtl/src/test/scala/forex/services/rates/cache/RatesCacheSpec.scala b/forex-mtl/src/test/scala/forex/services/rates/cache/RatesCacheSpec.scala new file mode 100644 index 00000000..a9bf525e --- /dev/null +++ b/forex-mtl/src/test/scala/forex/services/rates/cache/RatesCacheSpec.scala @@ -0,0 +1,138 @@ +package forex.services.rates.cache + +import cats.effect.IO +import forex.domain.{ Currency, Price, Rate, Timestamp } +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import java.time.OffsetDateTime +import scala.concurrent.duration._ + +class RatesCacheSpec extends AnyWordSpec with Matchers { + + val usdEur = Rate.Pair(Currency.USD, Currency.EUR) + val gbpJpy = Rate.Pair(Currency.GBP, Currency.JPY) + val softTtl = 4.minutes + val hardTtl = 5.minutes + + def makeRate(pair: Rate.Pair, price: BigDecimal = BigDecimal("1.23")): Rate = + Rate(pair, Price(price), Timestamp.now) + + def newCache(): InMemoryRatesCache[IO] = + InMemoryRatesCache.create[IO].unsafeRunSync() + + "InMemoryRatesCache.get" should { + + "return None for a pair that has never been stored" in { + val cache = newCache() + val result = cache.get(usdEur).unsafeRunSync() + + result shouldBe None + } + } + + "InMemoryRatesCache.putBatch" should { + + "make a stored pair retrievable via get" in { + val cache = newCache() + val rate = makeRate(usdEur) + val now = OffsetDateTime.now() + + cache.putBatch(List(rate), now).unsafeRunSync() + + val stored = cache.get(usdEur).unsafeRunSync() + stored.map(_.rate) shouldBe Some(rate) + } + + "store multiple pairs in one call" in { + val cache = newCache() + val rateUE = makeRate(usdEur) + val rateGJ = makeRate(gbpJpy) + val now = OffsetDateTime.now() + + cache.putBatch(List(rateUE, rateGJ), now).unsafeRunSync() + + val storedUE = cache.get(usdEur).unsafeRunSync() + val storedGJ = cache.get(gbpJpy).unsafeRunSync() + + storedUE.map(_.rate) shouldBe Some(rateUE) + storedGJ.map(_.rate) shouldBe Some(rateGJ) + } + + "overwrite an older entry with a fresher one" in { + val cache = newCache() + val original = makeRate(usdEur, BigDecimal("0.90")) + val updated = makeRate(usdEur, BigDecimal("0.95")) + val t1 = OffsetDateTime.now().minusMinutes(3) + val t2 = OffsetDateTime.now() + + cache.putBatch(List(original), t1).unsafeRunSync() + cache.putBatch(List(updated), t2).unsafeRunSync() + + val stored = cache.get(usdEur).unsafeRunSync() + stored.map(_.rate.price) shouldBe Some(Price(BigDecimal("0.95"))) + } + } + + "InMemoryRatesCache.allKeys" should { + + "return an empty set for a new cache" in { + val cache = newCache() + cache.allKeys.unsafeRunSync() shouldBe empty + } + + "return every pair that has been stored" in { + val cache = newCache() + val now = OffsetDateTime.now() + + cache.putBatch(List(makeRate(usdEur), makeRate(gbpJpy)), now).unsafeRunSync() + + val keys = cache.allKeys.unsafeRunSync() + keys should contain allOf (usdEur, gbpJpy) + } + } + + "CacheEntry freshness" should { + + "be fresh when age is below softTtl" in { + val now = OffsetDateTime.now() + val entry = CacheEntry(makeRate(usdEur), fetchedAt = now.minusSeconds(10)) + + entry.isFresh(now, softTtl) shouldBe true + entry.needsRevalidation(now, softTtl, hardTtl) shouldBe false + entry.isExpired(now, hardTtl) shouldBe false + } + + "need revalidation when age is between softTtl and hardTtl" in { + val now = OffsetDateTime.now() + val entry = CacheEntry(makeRate(usdEur), fetchedAt = now.minusSeconds(250)) + + entry.isFresh(now, softTtl) shouldBe false + entry.needsRevalidation(now, softTtl, hardTtl) shouldBe true + entry.isExpired(now, hardTtl) shouldBe false + } + + "be expired when age exceeds hardTtl" in { + val now = OffsetDateTime.now() + val entry = CacheEntry(makeRate(usdEur), fetchedAt = now.minusSeconds(310)) + + entry.isFresh(now, softTtl) shouldBe false + entry.needsRevalidation(now, softTtl, hardTtl) shouldBe false + entry.isExpired(now, hardTtl) shouldBe true + } + + "be fresh when just inserted (age zero)" in { + val now = OffsetDateTime.now() + val entry = CacheEntry(makeRate(usdEur), fetchedAt = now) + + entry.isFresh(now, softTtl) shouldBe true + } + + "be expired at exactly the hardTtl boundary" in { + val now = OffsetDateTime.now() + val entry = CacheEntry(makeRate(usdEur), fetchedAt = now.minusMinutes(5)) + + entry.isExpired(now, hardTtl) shouldBe true + } + } +} From 0b18b2b194b0ef8b78cb53b876fb38ceb280ed7f Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 19:44:39 +0530 Subject: [PATCH 08/11] Add structured log helper and integrate upstream logs --- .../main/scala/forex/logging/LogEvent.scala | 17 ++ .../rates/interpreters/OneFrameLive.scala | 210 +++++++++++++----- 2 files changed, 171 insertions(+), 56 deletions(-) create mode 100644 forex-mtl/src/main/scala/forex/logging/LogEvent.scala diff --git a/forex-mtl/src/main/scala/forex/logging/LogEvent.scala b/forex-mtl/src/main/scala/forex/logging/LogEvent.scala new file mode 100644 index 00000000..26fcb663 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/logging/LogEvent.scala @@ -0,0 +1,17 @@ +package forex.logging + +import io.circe.{ Json, JsonObject } + +object LogEvent { + def apply(message: String, fields: (String, Json)*): String = + Json + .fromJsonObject( + JsonObject.fromIterable( + List( + "message" -> Json.fromString(message), + "data" -> Json.fromJsonObject(JsonObject.fromIterable(fields)) + ) + ) + ) + .noSpaces +} diff --git a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala index dcc9c7d6..fc54fbdd 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala @@ -10,21 +10,26 @@ import cats.syntax.applicativeError._ import cats.syntax.either._ import cats.syntax.flatMap._ import cats.syntax.functor._ +import cats.syntax.show._ import forex.clients.oneframe.{ OneFrameClientAlgebra, OneFrameHttpClient } import forex.config.ApplicationConfig -import forex.domain.Rate -import forex.services.rates.Algebra +import forex.domain.{ Currency, Rate } +import forex.logging.LogEvent +import forex.services.rates.{ Algebra, CircuitBreaker } import forex.services.rates.cache.{ CacheAlgebra, InMemoryRatesCache } import forex.services.rates.errors.{ Error => ServiceError } +import io.circe.Json import org.http4s.client.Client import org.slf4j.LoggerFactory import java.time.OffsetDateTime +import scala.concurrent.duration.MILLISECONDS class OneFrameLive[F[_]: Concurrent: Timer] private ( oneFrameClient: OneFrameClientAlgebra[F], cache: CacheAlgebra[F], fetchGate: Ref[F, Option[Deferred[F, Either[ServiceError, Unit]]]], + quotaCounter: Ref[F, (Long, Long)], config: ApplicationConfig ) extends Algebra[F] { @@ -32,43 +37,63 @@ class OneFrameLive[F[_]: Concurrent: Timer] private ( private val softTtl = config.cache.softTtl private val ttl = config.cache.ttl private val maxStaleOnError = config.cache.maxStaleOnError + private val maxRetries = config.oneFrame.maxRetries + + override def isReady: F[Boolean] = cache.allKeys.map(_.nonEmpty) override def get(pair: Rate.Pair): F[Either[ServiceError, Rate]] = nowUtc.flatMap { now => cache.get(pair).flatMap { cached => cached match { case Some(entry) if entry.isFresh(now, softTtl) => - Concurrent[F].delay(logger.debug(s"Cache HIT for $pair")) >> - entry.rate.asRight[ServiceError].pure[F] + Concurrent[F].delay( + logger.debug(LogEvent("rates_lookup", + "from" -> Json.fromString(pair.from.show), + "to" -> Json.fromString(pair.to.show), + "cache_status" -> Json.fromString("HIT") + )) + ) >> entry.rate.asRight[ServiceError].pure[F] case Some(entry) if entry.needsRevalidation(now, softTtl, ttl) => - Concurrent[F].delay(logger.debug(s"Cache STALE for $pair, revalidating in background")) >> - Concurrent[F] - .start(fetchAllPairsAndPopulateCache) - .as(entry.rate.asRight[ServiceError]) + Concurrent[F].delay( + logger.debug(LogEvent("rates_lookup", + "from" -> Json.fromString(pair.from.show), + "to" -> Json.fromString(pair.to.show), + "cache_status" -> Json.fromString("STALE_REVALIDATING") + )) + ) >> Concurrent[F].start(fetchAllPairsAndPopulateCache).as(entry.rate.asRight[ServiceError]) case _ => - Concurrent[F].delay(logger.debug(s"Cache MISS for $pair, fetching synchronously")) >> - fetchWithCoalescing.flatMap { - case Right(_) => - cache.get(pair).map { - case Some(entry) => entry.rate.asRight - case None => - ServiceError.OneFrameLookupFailed(s"One-Frame did not return a rate for $pair").asLeft - } - - case Left(err) => - cached match { - case Some(entry) if !entry.isExpired(now, maxStaleOnError) => - Concurrent[F] - .delay( - logger.warn(s"One-Frame unavailable; serving stale cache for $pair") - ) - .as(entry.rate.asRight[ServiceError]) - case _ => - err.asLeft[Rate].pure[F] - } - } + Concurrent[F].delay( + logger.debug(LogEvent("rates_lookup", + "from" -> Json.fromString(pair.from.show), + "to" -> Json.fromString(pair.to.show), + "cache_status" -> Json.fromString("MISS") + )) + ) >> fetchWithCoalescing.flatMap { + case Right(_) => + cache.get(pair).map { + case Some(entry) => entry.rate.asRight + case None => + ServiceError.OneFrameLookupFailed(s"One-Frame did not return a rate for $pair").asLeft + } + + case Left(err) => + cached match { + case Some(entry) if !entry.isExpired(now, maxStaleOnError) => + Concurrent[F] + .delay( + logger.warn(LogEvent("stale_fallback", + "from" -> Json.fromString(pair.from.show), + "to" -> Json.fromString(pair.to.show), + "error" -> Json.fromString(err.toString) + )) + ) + .as(entry.rate.asRight[ServiceError]) + case _ => + err.asLeft[Rate].pure[F] + } + } } } } @@ -98,35 +123,98 @@ class OneFrameLive[F[_]: Concurrent: Timer] private ( fetchGate.set(None) } - case Right(theirGate) => - Concurrent[F].delay(logger.debug("Awaiting in-flight One-Frame fetch")) >> - theirGate.get + case Right(existingGate) => + Concurrent[F].delay(logger.debug(LogEvent("coalesced_wait"))) >> + existingGate.get } } - private def fetchAllPairsAndPopulateCache: F[Either[ServiceError, Unit]] = + private def fetchAllPairsAndPopulateCache: F[Either[ServiceError, Unit]] = { + def loop(attemptsLeft: Int): F[Either[ServiceError, Unit]] = + doOneUpstreamCall.flatMap { + case Right(_) => ().asRight[ServiceError].pure[F] + case Left(ServiceError.OneFrameUnreachable(_)) if attemptsLeft > 1 => + val attempt = maxRetries - attemptsLeft + 2 + val delayMs = math.min(200L * (1L << attempt), 5000L) + Concurrent[F].delay( + logger.warn(LogEvent("upstream_retry", + "attempt" -> Json.fromInt(maxRetries - attemptsLeft + 2), + "delay_ms" -> Json.fromLong(delayMs), + "retries_left" -> Json.fromInt(attemptsLeft - 1) + )) + ) >> Timer[F].sleep(scala.concurrent.duration.Duration(delayMs, MILLISECONDS)) >> + loop(attemptsLeft - 1) + case left => left.pure[F] + } + loop(maxRetries + 1) + } + + private def doOneUpstreamCall: F[Either[ServiceError, Unit]] = NonEmptyList.fromList(Rate.Pair.allPairs) match { case None => (ServiceError.OneFrameLookupFailed("No currency pairs defined"): ServiceError).asLeft[Unit].pure[F] case Some(pairs) => - Concurrent[F].delay(logger.info(s"Fetching ${pairs.size} pairs from One-Frame")) >> - oneFrameClient.getRates(pairs).flatMap { - case Right(rates) => - nowUtc.flatMap(now => cache.putBatch(rates, now)) >> - Concurrent[F] - .delay(logger.info(s"Cached ${pairs.size} pairs")) - .as(().asRight[ServiceError]) - - case Left(err) => - Concurrent[F] - .delay(logger.warn(s"One-Frame fetch failed: $err")) - .as(err.asLeft[Unit]) - } + Timer[F].clock.realTime(MILLISECONDS).flatMap { startMs => + Concurrent[F].delay( + logger.debug(LogEvent("upstream_fetch_start", "pair_count" -> Json.fromInt(pairs.size))) + ) >> + oneFrameClient.getRates(pairs).flatMap { result => + Timer[F].clock.realTime(MILLISECONDS).flatMap { endMs => + val durationMs = endMs - startMs + result match { + case Right(rates) => + trackQuota >> + nowUtc.flatMap(now => cache.putBatch(rates, now)) >> + Concurrent[F].delay( + logger.info(LogEvent("upstream_fetch_ok", + "pair_count" -> Json.fromInt(rates.size), + "duration_ms" -> Json.fromLong(durationMs) + )) + ).as(().asRight[ServiceError]) + + case Left(err) => + Concurrent[F].delay( + logger.warn(LogEvent("upstream_fetch_error", + "error" -> Json.fromString(err.toString), + "duration_ms" -> Json.fromLong(durationMs) + )) + ).as(err.asLeft[Unit]) + } + } + } + } + } + + private def trackQuota: F[Unit] = + nowUtc.flatMap { now => + val today = now.toLocalDate.toEpochDay + quotaCounter.modify { case (day, count) => + val newCount = if (day == today) count + 1 else 1L + ((today, newCount), newCount) + }.flatMap { + case n if n >= 950 => + Concurrent[F].delay( + logger.error(LogEvent("quota_alert", + "calls_today" -> Json.fromLong(n), + "limit" -> Json.fromInt(1000), + "level" -> Json.fromString("critical") + )) + ) + case n if n >= 800 => + Concurrent[F].delay( + logger.warn(LogEvent("quota_alert", + "calls_today" -> Json.fromLong(n), + "limit" -> Json.fromInt(1000), + "level" -> Json.fromString("warning") + )) + ) + case _ => Concurrent[F].unit + } } private def nowUtc: F[OffsetDateTime] = - Timer[F].clock.realTime(scala.concurrent.duration.MILLISECONDS).map { millis => + Timer[F].clock.realTime(MILLISECONDS).map { millis => OffsetDateTime.ofInstant( java.time.Instant.ofEpochMilli(millis), java.time.ZoneOffset.UTC @@ -140,25 +228,35 @@ object OneFrameLive { httpClient: Client[F], config: ApplicationConfig ): Resource[F, Algebra[F]] = - Resource.eval( - OneFrameHttpClient[F](httpClient, config.oneFrame).flatMap(make(_, config)) - ) + Resource + .eval( + for { + rawClient <- OneFrameHttpClient[F](httpClient, config.oneFrame) + client <- CircuitBreaker.wrap(rawClient, config.circuitBreaker.maxFailures, config.circuitBreaker.resetTimeout) + service <- make(client, config) + } yield service + ) + .flatTap { service => + Resource.eval(service.get(Rate.Pair(Currency.USD, Currency.EUR)).void) + } def make[F[_]: Concurrent: Timer]( client: OneFrameClientAlgebra[F], config: ApplicationConfig ): F[Algebra[F]] = for { - cache <- InMemoryRatesCache.create[F] - fetchGate <- Ref.of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) - } yield new OneFrameLive[F](client, cache, fetchGate, config) + cache <- InMemoryRatesCache.create[F] + fetchGate <- Ref.of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) + quotaCounter <- Ref.of[F, (Long, Long)]((0L, 0L)) + } yield new OneFrameLive[F](client, cache, fetchGate, quotaCounter, config) def makeWithCache[F[_]: Concurrent: Timer]( client: OneFrameClientAlgebra[F], cache: CacheAlgebra[F], config: ApplicationConfig ): F[Algebra[F]] = - Ref - .of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) - .map(fetchGate => new OneFrameLive[F](client, cache, fetchGate, config)) + for { + fetchGate <- Ref.of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) + quotaCounter <- Ref.of[F, (Long, Long)]((0L, 0L)) + } yield new OneFrameLive[F](client, cache, fetchGate, quotaCounter, config) } From f7b889906e48472c4a617bb5967647c5d62b0917 Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 20:13:03 +0530 Subject: [PATCH 09/11] Finalized documentation with implementation --- forex-mtl/README.md | 169 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 169 insertions(+) diff --git a/forex-mtl/README.md b/forex-mtl/README.md index 480bfcc6..8f875412 100644 --- a/forex-mtl/README.md +++ b/forex-mtl/README.md @@ -154,3 +154,172 @@ The HTTP layer turns that into a `400 Bad Request` with a JSON error body. `GET /rates?from=USD&to=USD` is a valid request. USD always exchanges to USD at 1.0. The program layer intercepts this case and returns immediately without touching the cache or the upstream API. + +--- + +## API + +``` +GET /rates?from={CURRENCY}&to={CURRENCY} +``` + +Supported currencies: `AUD CAD CHF EUR GBP JPY NZD SGD USD` + +**Success (200)** +```json +{ + "from": "USD", + "to": "EUR", + "price": 0.8432, + "timestamp": "2026-04-28T05:53:56.522Z" +} +``` + +**Error responses** + +| Status | When | Body | +|--------|------|------| +| 400 | Missing or unrecognised currency | `{"error": "Unsupported currency: XYZ"}` | +| 400 | Missing query parameter | `{"error": "Both 'from' and 'to' query parameters are required"}` | +| 429 | Per-IP rate limit exceeded | `{"error": "..."}` + `Retry-After: ` header | +| 502 | One-Frame unreachable or quota exceeded | `{"error": "One-Frame is unreachable: ..."}` | +| 500 | Unexpected internal error | `{"error": "..."}` | + +--- + +## Compliance + +### Requirement 1 — return a rate for two supported currencies + +```bash +curl 'http://localhost:8081/rates?from=USD&to=EUR' +# {"from":"USD","to":"EUR","price":0.8432,"timestamp":"2026-04-28T05:53:56.522Z"} +``` + +All 72 pairs work. Invalid input returns a descriptive error: + +```bash +curl 'http://localhost:8081/rates?from=BTC&to=USD' +# HTTP 400 — {"error":"Unsupported currency: BTC"} + +curl 'http://localhost:8081/rates?from=USD' +# HTTP 400 — {"error":"Both 'from' and 'to' query parameters are required"} +``` + +### Requirement 2 — rate not older than 5 minutes + +Hard TTL is enforced at the cache layer. `CacheEntry.isExpired` returns `true` at exactly +5 minutes. An expired entry is never served — the request fetches synchronously. The test +suite includes a case that seeds a 5m30s-old entry and asserts the fresh price is returned. + +### Requirement 3 — ≥10,000 client requests/day within 1,000 upstream calls + +``` +60 min/hr ÷ 4 min soft-TTL = 15 upstream calls/hr +15 × 24 hr = 360 upstream calls/day (limit: 1,000) + +Client requests served from cache between refreshes: unlimited +``` + +The 360 figure is the worst case under sustained load. Under zero or low traffic, the SWR +policy means fewer upstream calls. The 640-call headroom also absorbs cache misses on cold +starts and any retries. + +--- + +## Configuration + +| Key | Default | Env var override | Description | +|-----|---------|-----------------|-------------| +| `http.host` | `0.0.0.0` | — | Bind address | +| `http.port` | `8081` | — | Proxy listen port | +| `http.timeout` | `40 seconds` | — | Per-request server timeout | +| `one-frame.base-uri` | `http://localhost:8080` | `ONE_FRAME_BASE_URI` | One-Frame base URL | +| `one-frame.auth-token` | *(empty)* | `ONE_FRAME_TOKEN` | One-Frame auth token | +| `one-frame.timeout` | `10 seconds` | — | Upstream request timeout; prevents infinite wait on hung connections | +| `one-frame.max-retries` | `3` | — | Maximum retry attempts for unreachable upstream calls | +| `circuit-breaker.max-failures` | `5` | — | Consecutive unreachable failures before opening the circuit | +| `circuit-breaker.reset-timeout` | `60 seconds` | — | Wait time before half-open probe after circuit opens | +| `cache.ttl` | `5 minutes` | — | Hard freshness ceiling | +| `cache.soft-ttl` | `4 minutes` | — | SWR revalidation boundary | +| `cache.max-stale-on-error` | `5 minutes` | — | Serve expired cache on One-Frame outage; raise to e.g. 10 min to tolerate stale over 502 | +| `rate-limiter.max-requests-per-minute` | `100` | — | Per-IP request cap. Requests above the limit return `429 Too Many Requests` | + +--- + +## Prerequisites + +Use Java 17 for sbt commands in this project. + +```bash +export JAVA_HOME=/opt/homebrew/opt/openjdk@17 +``` + +On Linux, set `JAVA_HOME` to your Java 17 installation path instead. + +--- + +## Running locally + +**1. Start One-Frame (port 8080)** +```bash +docker run -p 8080:8080 paidyinc/one-frame +``` + +**2. Start the proxy (port 8081)** +```bash +ONE_FRAME_TOKEN=10dc303535874aeccc86a8251e6992f5 sbt run +``` + +**3. Try it** +```bash +curl 'http://localhost:8081/rates?from=USD&to=EUR' +curl 'http://localhost:8081/rates?from=USD&to=USD' # returns 1.0 instantly, no upstream call +curl 'http://localhost:8081/rates?from=XYZ&to=USD' # 400 with error message +``` + +--- + +## Running the tests + +```bash +sbt test +``` + +No Docker or network access required. The test suite runs entirely in-memory using mock +One-Frame backends built inline. 53 tests across 5 specs: + +| Spec | What it covers | +|------|---------------| +| `CurrencySpec` | `fromString` parsing, case sensitivity, `allPairs` size and correctness | +| `RatesCacheSpec` | get/put, SWR freshness boundaries, overwrite behaviour | +| `OneFrameClientSpec` | JSON decoding, quota error detection, auth header, pair encoding, malformed base URI | +| `RatesProgramSpec` | Same-currency short-circuit, error mapping from service to program layer | +| `RatesRoutesIntegrationSpec` | Full stack — 15 end-to-end scenarios including coalescing, SWR, graceful degradation, and rate limiting | + +--- + +## Extending + +### Add a currency + +1. Add a `case object` to `Currency` +2. Add it to `Currency.values` + +`fromString`, `allPairs`, and the cache all update automatically. + +### Swap the cache for Redis + +Not needed for a single instance — in-memory is faster and has no external failure mode. +Redis becomes relevant when running multiple instances that need to share rate state, or +when cache warmth across restarts matters. + +Implement `CacheAlgebra[F]` in `forex/services/rates/cache/` using a Redis client +(e.g. [redis4cats](https://redis4cats.profunktor.dev/)). Pass the new instance to +`OneFrameLive.makeWithCache`. Nothing else in the stack changes. + +### Add a second upstream provider + +Implement `OneFrameClientAlgebra[F]` for the new provider. The `OneFrameLive` interpreter +accepts any implementation of that algebra — switching or wrapping providers is a wiring +change, not a logic change. From 4e970d112724ba37a24d32cbf26fa99b16341afa Mon Sep 17 00:00:00 2001 From: deeparagu Date: Tue, 28 Apr 2026 20:56:44 +0530 Subject: [PATCH 10/11] documentation clean up + logging and circuit break decision --- forex-mtl/README.md | 388 ++++++++++++++++++-------------------------- 1 file changed, 157 insertions(+), 231 deletions(-) diff --git a/forex-mtl/README.md b/forex-mtl/README.md index 8f875412..d630f3d9 100644 --- a/forex-mtl/README.md +++ b/forex-mtl/README.md @@ -1,325 +1,251 @@ # Forex Rate Proxy -A local HTTP proxy for currency exchange rates. Internal services call this instead of calling -One-Frame directly — it handles caching, rate limiting, error recovery, and all the edge cases -so callers don't have to. +A local HTTP proxy for exchange rates. Internal services call this service instead of calling +One-Frame directly. It centralizes caching, resilience, and request shaping. ---- +## Requirements (From the Brief) -## The problem +1. Return an exchange rate for two supported currencies. +2. Returned rate must never be older than **5 minutes**. +3. Support **at least 10,000 successful requests/day** with an API token capped at 1,000 upstream + calls/day. -One-Frame is the upstream provider of exchange rates. It has one hard constraint that makes -it awkward to use directly: +## Assumptions -> **1,000 requests per day per API token.** +- **Single instance**: cache is in-memory and not shared across instances. +- **Full-pair cache updates**: no per-pair caching strategy; every miss/refresh fetches all pairs in a single call. +- **Single upstream**: no secondary provider fallback. +- **`maxStaleOnError` defaults to 5m**: stale serving beyond that requires explicit config increase. -If your services make 10,000 requests per day (the stated requirement), hitting One-Frame on -every request burns the quota in 6 minutes. This proxy solves that by sitting in front of -One-Frame and serving most requests from an in-memory cache. +## Key Decisions and Trade-offs ---- +### In-memory cache over Redis -## Requirements (from the brief) +For single-instance local proxying, in-memory avoids extra infra, network latency, and cache-tier failure modes. The cache is abstracted via `CacheAlgebra[F]` for future Redis migration. -1. Return an exchange rate when given two supported currency codes. -2. The rate returned must never be older than **5 minutes**. -3. Support **at least 10,000 successful client requests per day** using a single API token - (which is capped at 1,000 upstream calls per day). +### Typed domain errors over exceptions ---- +Service calls return `Either` with explicit cases: -## Constraints and what they forced +- `OneFrameQuotaExceeded` +- `OneFrameUnreachable` +- `OneFrameLookupFailed` -### One-Frame returns all pairs in a single request +This keeps failure handling explicit and compiler-checked. -The API accepts multiple `pair` query parameters: -``` -GET /rates?pair=USDEUR&pair=USDJPY&pair=GBPAUD&... -``` +### Safe currency parsing -This means there is no extra cost to fetching all 72 pairs (nPr = 9! / 7!) at once versus fetching one. -Every upstream call in this service fetches all 72 pairs and fills the entire cache. -There is no per-pair granularity. +`Currency.fromString` returns `Either` instead of throwing runtime `MatchError`. +Invalid user input maps to deterministic `400 Bad Request`. -### The 5-minute freshness ceiling +### Same-currency shortcut -Serving stale data is a correctness bug. Serving data that is 4:59 old is fine; 5:01 is not. -The implementation enforces this with a hard check: if a cached entry is ≥ 5 minutes old, it -is never served — the request blocks until a fresh batch has been fetched from One-Frame. +`GET /rates?from=USD&to=USD` returns `1.0` immediately without cache/upstream access. -### One-Frame always returns HTTP 200 +## API -Even on errors. Quota exhaustion looks like: -```json -{"error": "Quota reached for token ..."} ``` -The client has to inspect the body to distinguish success from failure. This is handled in -`OneFrameHttpClient` — successful responses are JSON arrays; error responses are JSON objects -with an `error` field. - -The One-Frame response also uses `time_stamp` (snake_case with underscore), which is worth -calling out explicitly because it differs from the camelCase convention everywhere else. - ---- - -## Assumptions - -**Single instance.** The service runs as a single process. The cache is in-memory and is not -shared across instances. This is appropriate for a "local proxy" and keeps the deployment -simple. If horizontal scaling were needed, the `CacheAlgebra` interface makes a Redis migration -straightforward. +GET /rates?from={CURRENCY}&to={CURRENCY} +``` -**All 9 currencies are always cached together.** There is no logic to cache individual pairs -separately or to prioritise popular pairs. Every cache miss fetches all 72 pairs. This is -deliberate — the marginal cost of fetching 71 extra pairs is zero, and the simplicity is worth it. +Supported currencies: `AUD CAD CHF EUR GBP JPY NZD SGD USD` -**One-Frame is the only upstream source.** No fallback provider is implemented. If One-Frame -is down and the cache is exhausted, the service returns 502. +### Success (200) -**`maxStaleOnError` defaults to 5 minutes, matching the SLA.** If One-Frame is unreachable and -the cache is older than 5 minutes, the service returns 502. If your system can tolerate slightly -stale rates over an error response during an outage, raise this to e.g. 10 minutes — the service -will then serve cached data up to that age before falling back to 502. +```json +{ + "from": "USD", + "to": "EUR", + "price": 0.8432, + "timestamp": "2026-04-28T05:53:56.522Z" +} +``` ---- +### Error responses -## How the cache works +| Status | When | Body | +|--------|------|------| +| 400 | Missing or unrecognized currency | `{"error": "Unsupported currency: XYZ"}` | +| 400 | Missing query parameter | `{"error": "Both 'from' and 'to' query parameters are required"}` | +| 429 | Per-IP rate limit exceeded | `{"error": "..."}` + `Retry-After: ` header | +| 502 | One-Frame unreachable or quota exceeded | `{"error": "One-Frame is unreachable: ..."}` | +| 500 | Unexpected internal error | `{"error": "..."}` | -The cache uses a **Stale-While-Revalidate (SWR)** policy with two TTL boundaries: +### Health endpoints ``` -Age of the cached rate What happens -──────────────────────────── ───────────────────────────────────────────────── -0 – 4 min (fresh) Served immediately. No upstream work. -4 – 5 min (stale-valid) Served immediately. Background refresh fires concurrently. -≥ 5 min (expired) Request waits. Fetch completes first. Then respond. -Not in cache (cold) Request waits. Fetch completes first. Then respond. +GET /health/live # returns 200 OK when process is up +GET /health/ready # returns 200 OK when cache is ready to serve traffic; else 503 SERVICE UNAVAILABLE ``` -The key insight: for the overwhelming majority of requests (anything in the 0–4 min window) -the response latency is just a map lookup. The 4–5 min window means clients are never blocked -by a revalidation — they get a slightly stale value while the cache updates behind them. -Only a cold start or a true expiry blocks the caller. +## Operational and Observability Features + +### Request correlation (`X-Request-ID`) -### Why not a background polling job? +Each request is tagged with `X-Request-ID`: +- If incoming header exists, it is propagated. +- If missing, a UUID is generated and returned in response headers. +This makes request tracing easier across gateway/service logs. -An earlier version ran a fiber every 4 minutes to keep the cache warm. It was removed. -A polling job burns upstream quota even when the service has zero traffic — 360 calls per day -regardless. The SWR approach only refreshes when someone actually asks for a rate. Zero traffic, -zero upstream calls. +### Rate limiting -### Concurrent cold starts (the thundering herd) +Per-IP request limiting is enforced (`rate-limiter.max-requests-per-minute`, default `100`): +- Over-limit requests return `429 Too Many Requests`. +- Response includes `Retry-After` header. -When the cache is empty and 100 requests arrive simultaneously, without protection all 100 -would race to call One-Frame. The implementation uses a `Deferred` gate (a one-shot promise): +### Structured logging -1. The first request creates the gate and starts the upstream fetch. -2. Every other concurrent request finds the gate and waits. -3. When the first request finishes, all waiters unblock at once with the result already in cache. +`OneFrameLive` emits structured JSON log events for key behaviors: +- cache path decisions (`rates_lookup`, `coalesced_wait`) +- upstream lifecycle (`upstream_fetch_start`, `upstream_fetch_ok`, `upstream_fetch_error`, `upstream_retry`) +- fallback/degradation (`stale_fallback`, `quota_alert`) +This supports easier querying/alerting in centralized log systems. -Result: exactly **one** upstream call per burst, regardless of concurrency. +### Retry + circuit breaker ---- +For upstream connectivity failures: +- requests are retried (bounded by `one-frame.max-retries`) +- repeated failures open a circuit breaker (`circuit-breaker.max-failures`) +- breaker transitions back via reset timeout (`circuit-breaker.reset-timeout`) +This prevents hammering an unhealthy upstream and improves recovery behavior. -## Key decisions +## Constraints and Resulting Design -### In-memory cache, not Redis +### One-Frame returns all pairs in one call -Adding Redis means adding infrastructure, network latency on every cache read, and a new -failure mode (the cache itself can become unavailable). For a single-instance local proxy, -in-memory is correct. The `CacheAlgebra[F]` trait is the only thing a Redis implementation -would need to satisfy. +One-Frame accepts multiple `pair` query params, e.g. -### Typed errors, not exceptions +``` +GET /rates?pair=USDEUR&pair=USDJPY&pair=GBPAUD&... +``` -Every function that can fail returns an `Either` — a value that is either a typed error or -a result. The error type is an exhaustive set of cases: +Fetching all 72 pairs costs the same request budget as fetching one pair, so each upstream call +refreshes the full cache. -- `OneFrameQuotaExceeded` — the daily limit is hit -- `OneFrameUnreachable` — network failure, timeout, or unexpected HTTP status from One-Frame -- `OneFrameLookupFailed` — One-Frame returned an error message we didn't recognise +### Hard 5-minute freshness ceiling -The compiler enforces that every call site handles all cases. If a new error is added later, -every unhandled case becomes a compile error — not a runtime crash. +Serving data older than 5 minutes violates the requirement. If a cached value is >= 5 minutes +old, it is not served; request blocks until fresh data is fetched. -### `fromString` returns `Either`, not a throw +### One-Frame always returns HTTP 200 -The original scaffold had `Currency.fromString` as an unsafe partial pattern match. An unknown -currency code would throw a `MatchError` at runtime. The replacement finds the currency by -looking it up in the `values` list, and returns `Left("Unsupported currency: XYZ")` if not found. -The HTTP layer turns that into a `400 Bad Request` with a JSON error body. +Failures are encoded in body payloads, e.g.: -### Same-currency short-circuit +```json +{"error": "Quota reached for token ..."} +``` -`GET /rates?from=USD&to=USD` is a valid request. USD always exchanges to USD at 1.0. -The program layer intercepts this case and returns immediately without touching the cache -or the upstream API. +`OneFrameHttpClient` inspects response body shape (`array` success vs `object.error` failure). +It also maps One-Frame's `time_stamp` field to internal timestamp handling. ---- +## Cache Strategy (SWR) -## API +The cache uses stale-while-revalidate with two TTL boundaries: ``` -GET /rates?from={CURRENCY}&to={CURRENCY} +Age of cached rate Behavior +---------------------------- ---------------------------------------------- +0-4m (fresh) Serve immediately, no upstream call. +4-5m (stale-valid) Serve immediately, refresh in background. +>=5m (expired) Block until synchronous fresh fetch completes. +Cache miss (cold) Block until synchronous fresh fetch completes. ``` -Supported currencies: `AUD CAD CHF EUR GBP JPY NZD SGD USD` +This keeps most reads as in-memory lookups while preserving strict freshness at 5 minutes. -**Success (200)** -```json -{ - "from": "USD", - "to": "EUR", - "price": 0.8432, - "timestamp": "2026-04-28T05:53:56.522Z" -} -``` +### Why no fixed background poller? -**Error responses** +A periodic poller would consume quota even with zero traffic (e.g. 360/day at 4-minute cadence). +SWR refreshes only when there is demand. -| Status | When | Body | -|--------|------|------| -| 400 | Missing or unrecognised currency | `{"error": "Unsupported currency: XYZ"}` | -| 400 | Missing query parameter | `{"error": "Both 'from' and 'to' query parameters are required"}` | -| 429 | Per-IP rate limit exceeded | `{"error": "..."}` + `Retry-After: ` header | -| 502 | One-Frame unreachable or quota exceeded | `{"error": "One-Frame is unreachable: ..."}` | -| 500 | Unexpected internal error | `{"error": "..."}` | +### Concurrent cold-start handling ---- +Concurrent requests during cold/expired state are coalesced via `Deferred`: -## Compliance +1. First request starts upstream fetch. +2. Other requests wait on the same gate. +3. All waiters resume from one completed fetch. -### Requirement 1 — return a rate for two supported currencies +This prevents thundering-herd duplicate upstream calls. -```bash -curl 'http://localhost:8081/rates?from=USD&to=EUR' -# {"from":"USD","to":"EUR","price":0.8432,"timestamp":"2026-04-28T05:53:56.522Z"} -``` +## Compliance Check -All 72 pairs work. Invalid input returns a descriptive error: +### Requirement 1: valid rates for supported currencies ```bash -curl 'http://localhost:8081/rates?from=BTC&to=USD' -# HTTP 400 — {"error":"Unsupported currency: BTC"} - -curl 'http://localhost:8081/rates?from=USD' -# HTTP 400 — {"error":"Both 'from' and 'to' query parameters are required"} +curl 'http://localhost:8081/rates?from=USD&to=EUR' +curl 'http://localhost:8081/rates?from=BTC&to=USD' # HTTP 400 +curl 'http://localhost:8081/rates?from=USD' # HTTP 400 ``` -### Requirement 2 — rate not older than 5 minutes +### Requirement 2: max 5-minute age -Hard TTL is enforced at the cache layer. `CacheEntry.isExpired` returns `true` at exactly -5 minutes. An expired entry is never served — the request fetches synchronously. The test -suite includes a case that seeds a 5m30s-old entry and asserts the fresh price is returned. +Hard TTL enforces expiry at exactly 5 minutes. Expired entries are never served. -### Requirement 3 — ≥10,000 client requests/day within 1,000 upstream calls +### Requirement 3: 10k/day within 1k upstream/day -``` -60 min/hr ÷ 4 min soft-TTL = 15 upstream calls/hr -15 × 24 hr = 360 upstream calls/day (limit: 1,000) +Worst-case under steady traffic: -Client requests served from cache between refreshes: unlimited +``` +60 min/hr / 4 min soft-TTL = 15 upstream calls/hr +15 * 24 hr = 360 upstream calls/day ``` -The 360 figure is the worst case under sustained load. Under zero or low traffic, the SWR -policy means fewer upstream calls. The 640-call headroom also absorbs cache misses on cold -starts and any retries. - ---- +This stays under 1,000/day, leaving headroom for cold starts/retries. ## Configuration | Key | Default | Env var override | Description | -|-----|---------|-----------------|-------------| +|-----|---------|------------------|-------------| | `http.host` | `0.0.0.0` | — | Bind address | | `http.port` | `8081` | — | Proxy listen port | | `http.timeout` | `40 seconds` | — | Per-request server timeout | | `one-frame.base-uri` | `http://localhost:8080` | `ONE_FRAME_BASE_URI` | One-Frame base URL | | `one-frame.auth-token` | *(empty)* | `ONE_FRAME_TOKEN` | One-Frame auth token | -| `one-frame.timeout` | `10 seconds` | — | Upstream request timeout; prevents infinite wait on hung connections | -| `one-frame.max-retries` | `3` | — | Maximum retry attempts for unreachable upstream calls | -| `circuit-breaker.max-failures` | `5` | — | Consecutive unreachable failures before opening the circuit | -| `circuit-breaker.reset-timeout` | `60 seconds` | — | Wait time before half-open probe after circuit opens | +| `one-frame.timeout` | `10 seconds` | — | Upstream request timeout | +| `one-frame.max-retries` | `3` | — | Retry attempts for unreachable upstream | +| `circuit-breaker.max-failures` | `5` | — | Failures before opening circuit | +| `circuit-breaker.reset-timeout` | `60 seconds` | — | Half-open probe wait | | `cache.ttl` | `5 minutes` | — | Hard freshness ceiling | -| `cache.soft-ttl` | `4 minutes` | — | SWR revalidation boundary | -| `cache.max-stale-on-error` | `5 minutes` | — | Serve expired cache on One-Frame outage; raise to e.g. 10 min to tolerate stale over 502 | -| `rate-limiter.max-requests-per-minute` | `100` | — | Per-IP request cap. Requests above the limit return `429 Too Many Requests` | - ---- - -## Prerequisites - -Use Java 17 for sbt commands in this project. +| `cache.soft-ttl` | `4 minutes` | — | SWR revalidation threshold | +| `cache.max-stale-on-error` | `5 minutes` | — | Stale serving window during outages | +| `rate-limiter.max-requests-per-minute` | `100` | — | Per-IP request cap | -```bash -export JAVA_HOME=/opt/homebrew/opt/openjdk@17 -``` - -On Linux, set `JAVA_HOME` to your Java 17 installation path instead. - ---- - -## Running locally - -**1. Start One-Frame (port 8080)** -```bash -docker run -p 8080:8080 paidyinc/one-frame -``` - -**2. Start the proxy (port 8081)** -```bash -ONE_FRAME_TOKEN=10dc303535874aeccc86a8251e6992f5 sbt run -``` +## Local Run -**3. Try it** -```bash -curl 'http://localhost:8081/rates?from=USD&to=EUR' -curl 'http://localhost:8081/rates?from=USD&to=USD' # returns 1.0 instantly, no upstream call -curl 'http://localhost:8081/rates?from=XYZ&to=USD' # 400 with error message -``` +### Prerequisite +Use Java 17: +```bash export JAVA_HOME=/opt/homebrew/opt/openjdk@17``` ---- +### Start One-Frame (8080) +```bash docker run -p 8080:8080 paidyinc/one-frame``` -## Running the tests - -```bash -sbt test -``` +### Start proxy (8081) +```bash ONE_FRAME_TOKEN=10dc303535874aeccc86a8251e6992f5 sbt run``` -No Docker or network access required. The test suite runs entirely in-memory using mock -One-Frame backends built inline. 53 tests across 5 specs: +### Quick check +```bash curl 'http://localhost:8081/rates?from=USD&to=EUR'``` +```bash curl 'http://localhost:8081/rates?from=USD&to=USD'``` -| Spec | What it covers | -|------|---------------| -| `CurrencySpec` | `fromString` parsing, case sensitivity, `allPairs` size and correctness | -| `RatesCacheSpec` | get/put, SWR freshness boundaries, overwrite behaviour | -| `OneFrameClientSpec` | JSON decoding, quota error detection, auth header, pair encoding, malformed base URI | -| `RatesProgramSpec` | Same-currency short-circuit, error mapping from service to program layer | -| `RatesRoutesIntegrationSpec` | Full stack — 15 end-to-end scenarios including coalescing, SWR, graceful degradation, and rate limiting | +## Tests +```bash sbt test``` ---- +The suite runs in-memory (no network dependency) and currently covers: +- `CurrencySpec`: parsing, case handling, pair enumeration correctness. +- `RatesCacheSpec`: cache put/get and SWR freshness boundaries. +- `OneFrameClientSpec`: decoding, quota detection, auth header, malformed URI. +- `RatesProgramSpec`: same-currency shortcut and error mapping. +- `RatesRoutesIntegrationSpec`: end-to-end behavior, coalescing, SWR, degradation, rate limiting. -## Extending +## Extensions ### Add a currency +1. Add a `case object` in `Currency`. +2. Add it to `Currency.values`. +`fromString`, `allPairs`, and cache coverage update naturally from the enumeration. -1. Add a `case object` to `Currency` -2. Add it to `Currency.values` - -`fromString`, `allPairs`, and the cache all update automatically. - -### Swap the cache for Redis - -Not needed for a single instance — in-memory is faster and has no external failure mode. -Redis becomes relevant when running multiple instances that need to share rate state, or -when cache warmth across restarts matters. - -Implement `CacheAlgebra[F]` in `forex/services/rates/cache/` using a Redis client -(e.g. [redis4cats](https://redis4cats.profunktor.dev/)). Pass the new instance to -`OneFrameLive.makeWithCache`. Nothing else in the stack changes. - -### Add a second upstream provider +### Swap cache backend (e.g. Redis) +Implement `CacheAlgebra[F]` in `forex/services/rates/cache/` (e.g. with [redis4cats](https://redis4cats.profunktor.dev/)) and wire it into `OneFrameLive.makeWithCache`. -Implement `OneFrameClientAlgebra[F]` for the new provider. The `OneFrameLive` interpreter -accepts any implementation of that algebra — switching or wrapping providers is a wiring -change, not a logic change. +### Add another upstream provider +Implement `OneFrameClientAlgebra[F]`; `OneFrameLive` already depends on this abstraction, so provider switching is mostly composition/wiring. \ No newline at end of file From ca0d118cabe742d131789804e28dfd75d436c4a8 Mon Sep 17 00:00:00 2001 From: deeparagu Date: Wed, 29 Apr 2026 11:36:49 +0530 Subject: [PATCH 11/11] final pass-through: additional test cases + NEL handling to remove call on every upstream --- forex-mtl/README.md | 33 +++- .../rates/interpreters/OneFrameLive.scala | 66 ++++---- .../services/rates/CircuitBreakerSpec.scala | 137 ++++++++++++++++ .../rates/interpreters/OneFrameLiveSpec.scala | 154 ++++++++++++++++++ 4 files changed, 351 insertions(+), 39 deletions(-) create mode 100644 forex-mtl/src/test/scala/forex/services/rates/CircuitBreakerSpec.scala create mode 100644 forex-mtl/src/test/scala/forex/services/rates/interpreters/OneFrameLiveSpec.scala diff --git a/forex-mtl/README.md b/forex-mtl/README.md index d630f3d9..bd673519 100644 --- a/forex-mtl/README.md +++ b/forex-mtl/README.md @@ -104,10 +104,22 @@ This supports easier querying/alerting in centralized log systems. ### Retry + circuit breaker For upstream connectivity failures: -- requests are retried (bounded by `one-frame.max-retries`) +- requests are retried (bounded by `one-frame.max-retries`) with exponential backoff - repeated failures open a circuit breaker (`circuit-breaker.max-failures`) - breaker transitions back via reset timeout (`circuit-breaker.reset-timeout`) -This prevents hammering an unhealthy upstream and improves recovery behavior. + +Only `OneFrameUnreachable` errors trigger retries and count toward the failure threshold. Quota errors (`OneFrameQuotaExceeded`) pass through immediately and do not open the circuit. + +### Daily quota tracking + +Each successful upstream call increments an in-process counter that resets at UTC midnight. Warning thresholds: + +| Calls today | Log level | Event | +|-------------|-----------|-------| +| ≥ 800 | `WARN` | `quota_alert` with `level: "warning"` | +| ≥ 950 | `ERROR` | `quota_alert` with `level: "critical"` | + +The counter is in-memory and resets on restart. If the service restarts mid-day with 900 calls already made, the counter will not reflect prior usage. Raise `max-stale-on-error` as an operational mitigation if continuity across restarts is required. ## Constraints and Resulting Design @@ -230,12 +242,17 @@ Use Java 17: ## Tests ```bash sbt test``` -The suite runs in-memory (no network dependency) and currently covers: -- `CurrencySpec`: parsing, case handling, pair enumeration correctness. -- `RatesCacheSpec`: cache put/get and SWR freshness boundaries. -- `OneFrameClientSpec`: decoding, quota detection, auth header, malformed URI. -- `RatesProgramSpec`: same-currency shortcut and error mapping. -- `RatesRoutesIntegrationSpec`: end-to-end behavior, coalescing, SWR, degradation, rate limiting. +The suite runs entirely in-memory (no network, no Docker). 66 tests across 7 specs: + +| Spec | What it covers | +|------|----------------| +| `CurrencySpec` | `fromString` parsing, case sensitivity, `allPairs` size and correctness | +| `RatesCacheSpec` | get/put, SWR freshness boundaries, overwrite behaviour | +| `OneFrameClientSpec` | JSON decoding, quota error detection, auth header, pair encoding, malformed base URI | +| `RatesProgramSpec` | Same-currency short-circuit, error mapping from service to program layer | +| `CircuitBreakerSpec` | Closed/Open/HalfOpen state transitions, quota errors not counted as failures, probe on reset | +| `OneFrameLiveSpec` | Retry succeeds after transient failures, quota errors not retried, WARN/ERROR quota alerts at 800/950 | +| `RatesRoutesIntegrationSpec` | Full stack — 15 end-to-end scenarios including coalescing, SWR, graceful degradation, and rate limiting | ## Extensions diff --git a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala index fc54fbdd..d92787e9 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala @@ -38,6 +38,7 @@ class OneFrameLive[F[_]: Concurrent: Timer] private ( private val ttl = config.cache.ttl private val maxStaleOnError = config.cache.maxStaleOnError private val maxRetries = config.oneFrame.maxRetries + private val allPairsNel = NonEmptyList.fromListUnsafe(Rate.Pair.allPairs) override def isReady: F[Boolean] = cache.allKeys.map(_.nonEmpty) @@ -150,39 +151,33 @@ class OneFrameLive[F[_]: Concurrent: Timer] private ( } private def doOneUpstreamCall: F[Either[ServiceError, Unit]] = - NonEmptyList.fromList(Rate.Pair.allPairs) match { - case None => - (ServiceError.OneFrameLookupFailed("No currency pairs defined"): ServiceError).asLeft[Unit].pure[F] + Timer[F].clock.realTime(MILLISECONDS).flatMap { startMs => + Concurrent[F].delay( + logger.debug(LogEvent("upstream_fetch_start", "pair_count" -> Json.fromInt(allPairsNel.size))) + ) >> + oneFrameClient.getRates(allPairsNel).flatMap { result => + Timer[F].clock.realTime(MILLISECONDS).flatMap { endMs => + val durationMs = endMs - startMs + result match { + case Right(rates) => + trackQuota >> + nowUtc.flatMap(now => cache.putBatch(rates, now)) >> + Concurrent[F].delay( + logger.info(LogEvent("upstream_fetch_ok", + "pair_count" -> Json.fromInt(rates.size), + "duration_ms" -> Json.fromLong(durationMs) + )) + ).as(().asRight[ServiceError]) - case Some(pairs) => - Timer[F].clock.realTime(MILLISECONDS).flatMap { startMs => - Concurrent[F].delay( - logger.debug(LogEvent("upstream_fetch_start", "pair_count" -> Json.fromInt(pairs.size))) - ) >> - oneFrameClient.getRates(pairs).flatMap { result => - Timer[F].clock.realTime(MILLISECONDS).flatMap { endMs => - val durationMs = endMs - startMs - result match { - case Right(rates) => - trackQuota >> - nowUtc.flatMap(now => cache.putBatch(rates, now)) >> - Concurrent[F].delay( - logger.info(LogEvent("upstream_fetch_ok", - "pair_count" -> Json.fromInt(rates.size), - "duration_ms" -> Json.fromLong(durationMs) - )) - ).as(().asRight[ServiceError]) - - case Left(err) => - Concurrent[F].delay( - logger.warn(LogEvent("upstream_fetch_error", - "error" -> Json.fromString(err.toString), - "duration_ms" -> Json.fromLong(durationMs) - )) - ).as(err.asLeft[Unit]) - } - } + case Left(err) => + Concurrent[F].delay( + logger.warn(LogEvent("upstream_fetch_error", + "error" -> Json.fromString(err.toString), + "duration_ms" -> Json.fromLong(durationMs) + )) + ).as(err.asLeft[Unit]) } + } } } @@ -259,4 +254,13 @@ object OneFrameLive { fetchGate <- Ref.of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) quotaCounter <- Ref.of[F, (Long, Long)]((0L, 0L)) } yield new OneFrameLive[F](client, cache, fetchGate, quotaCounter, config) + + private[interpreters] def makeWithCacheAndQuota[F[_]: Concurrent: Timer]( + client: OneFrameClientAlgebra[F], + cache: CacheAlgebra[F], + quotaCounter: Ref[F, (Long, Long)], + config: ApplicationConfig + ): F[Algebra[F]] = + Ref.of[F, Option[Deferred[F, Either[ServiceError, Unit]]]](None) + .map(fetchGate => new OneFrameLive[F](client, cache, fetchGate, quotaCounter, config)) } diff --git a/forex-mtl/src/test/scala/forex/services/rates/CircuitBreakerSpec.scala b/forex-mtl/src/test/scala/forex/services/rates/CircuitBreakerSpec.scala new file mode 100644 index 00000000..6ad96964 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/services/rates/CircuitBreakerSpec.scala @@ -0,0 +1,137 @@ +package forex.services.rates + +import cats.effect.{ ContextShift, IO, Timer } +import cats.effect.concurrent.Ref +import cats.syntax.either._ +import cats.syntax.flatMap._ +import forex.clients.oneframe.OneFrameClientAlgebra +import forex.domain.{ Currency, Price, Rate, Timestamp } +import forex.services.rates.errors.{ Error => ServiceError } +import org.scalatest.EitherValues +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.ExecutionContext.global +import scala.concurrent.duration._ + +class CircuitBreakerSpec extends AnyWordSpec with Matchers with EitherValues { + + implicit val cs: ContextShift[IO] = IO.contextShift(global) + implicit val timer: Timer[IO] = IO.timer(global) + + private val pair = Rate.Pair(Currency.USD, Currency.EUR) + private val nel = cats.data.NonEmptyList.one(pair) + private val dummyRate = Rate(pair, Price(BigDecimal("0.85")), Timestamp.now) + private val ok = List(dummyRate).asRight[ServiceError] + private val unreachable = ServiceError.OneFrameUnreachable(new RuntimeException("network error")).asLeft[List[Rate]] + private val quota = ServiceError.OneFrameQuotaExceeded.asLeft[List[Rate]] + + private def fixedClient(result: Either[ServiceError, List[Rate]]): OneFrameClientAlgebra[IO] = + _ => IO.pure(result) + + private def countingClient( + result: Ref[IO, Either[ServiceError, List[Rate]]], + calls: AtomicInteger + ): OneFrameClientAlgebra[IO] = + _ => IO(calls.incrementAndGet()) >> result.get + + private def makeCB( + underlying: OneFrameClientAlgebra[IO], + maxFailures: Int = 2, + resetTimeout: FiniteDuration = 100.millis + ): IO[OneFrameClientAlgebra[IO]] = + CircuitBreaker.wrap[IO](underlying, maxFailures, resetTimeout) + + "CircuitBreaker in Closed state" should { + + "forward successful calls through to the underlying client" in { + val cb = makeCB(fixedClient(ok)).unsafeRunSync() + cb.getRates(nel).unsafeRunSync() shouldBe ok + } + + "stay Closed when failure count is below maxFailures" in { + val calls = new AtomicInteger(0) + val result = Ref.of[IO, Either[ServiceError, List[Rate]]](unreachable).unsafeRunSync() + val cb = makeCB(countingClient(result, calls), maxFailures = 3).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // failure 1 + cb.getRates(nel).unsafeRunSync() // failure 2 — still Closed at maxFailures-1 + + calls.get() shouldBe 2 + } + + "open after exactly maxFailures consecutive unreachable errors" in { + val calls = new AtomicInteger(0) + val result = Ref.of[IO, Either[ServiceError, List[Rate]]](unreachable).unsafeRunSync() + val cb = makeCB(countingClient(result, calls), maxFailures = 2).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // failure 1 + cb.getRates(nel).unsafeRunSync() // failure 2 → Open + cb.getRates(nel).unsafeRunSync() // rejected by Open CB + + calls.get() shouldBe 2 // 3rd call never reached upstream + } + + "not count quota errors as circuit-opening failures" in { + val calls = new AtomicInteger(0) + val result = Ref.of[IO, Either[ServiceError, List[Rate]]](quota).unsafeRunSync() + val cb = makeCB(countingClient(result, calls), maxFailures = 2).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // quota error 1 — should not count + cb.getRates(nel).unsafeRunSync() // quota error 2 — CB should still be Closed + + calls.get() shouldBe 2 // both calls reached upstream (not rejected) + } + } + + "CircuitBreaker in Open state" should { + + "reject calls immediately without reaching the underlying client" in { + val calls = new AtomicInteger(0) + val result = Ref.of[IO, Either[ServiceError, List[Rate]]](unreachable).unsafeRunSync() + val cb = makeCB(countingClient(result, calls), maxFailures = 1).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // failure → Open + val rejected = cb.getRates(nel).unsafeRunSync() // rejected + + calls.get() shouldBe 1 + rejected.left.value shouldBe a[ServiceError.OneFrameUnreachable] + } + + "probe the underlying client after resetTimeout and close on success" in { + val calls = new AtomicInteger(0) + val result = Ref.of[IO, Either[ServiceError, List[Rate]]](unreachable).unsafeRunSync() + val cb = makeCB(countingClient(result, calls), maxFailures = 2, resetTimeout = 50.millis).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // failure 1 + cb.getRates(nel).unsafeRunSync() // failure 2 → Open + + result.set(ok).unsafeRunSync() + IO.sleep(120.millis).unsafeRunSync() // past resetTimeout + + cb.getRates(nel).unsafeRunSync() shouldBe ok // probe → Closed + calls.get() shouldBe 3 + + cb.getRates(nel).unsafeRunSync() shouldBe ok // Closed: passes through normally + calls.get() shouldBe 4 + } + + "reopen if the probe call fails" in { + val calls = new AtomicInteger(0) + val result = Ref.of[IO, Either[ServiceError, List[Rate]]](unreachable).unsafeRunSync() + val cb = makeCB(countingClient(result, calls), maxFailures = 2, resetTimeout = 50.millis).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // failure 1 + cb.getRates(nel).unsafeRunSync() // failure 2 → Open + + IO.sleep(120.millis).unsafeRunSync() + + cb.getRates(nel).unsafeRunSync() // probe → still fails → reOpen + calls.get() shouldBe 3 + + cb.getRates(nel).unsafeRunSync() // rejected again (Open) + calls.get() shouldBe 3 // 4th call never reached upstream + } + } +} diff --git a/forex-mtl/src/test/scala/forex/services/rates/interpreters/OneFrameLiveSpec.scala b/forex-mtl/src/test/scala/forex/services/rates/interpreters/OneFrameLiveSpec.scala new file mode 100644 index 00000000..2cee3522 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/services/rates/interpreters/OneFrameLiveSpec.scala @@ -0,0 +1,154 @@ +package forex.services.rates.interpreters + +import cats.effect.{ ContextShift, IO, Timer } +import cats.effect.concurrent.Ref +import cats.syntax.either._ +import ch.qos.logback.classic.{ Level, Logger => LbLogger } +import ch.qos.logback.classic.spi.ILoggingEvent +import ch.qos.logback.core.read.ListAppender +import forex.clients.oneframe.OneFrameClientAlgebra +import forex.config.{ ApplicationConfig, CacheConfig, CircuitBreakerConfig, HttpConfig, OneFrameConfig, RateLimiterConfig } +import forex.domain.{ Currency, Price, Rate, Timestamp } +import forex.services.rates.cache.InMemoryRatesCache +import forex.services.rates.errors.{ Error => ServiceError } +import org.scalatest.EitherValues +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec +import org.slf4j.LoggerFactory + +import java.time.LocalDate +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.ExecutionContext.global +import scala.concurrent.duration._ +import scala.jdk.CollectionConverters._ + +class OneFrameLiveSpec extends AnyWordSpec with Matchers with EitherValues { + + implicit val cs: ContextShift[IO] = IO.contextShift(global) + implicit val timer: Timer[IO] = IO.timer(global) + + private val usdEur = Rate.Pair(Currency.USD, Currency.EUR) + + private val baseConfig = ApplicationConfig( + http = HttpConfig("0.0.0.0", 8080, 40.seconds), + oneFrame = OneFrameConfig("http://one-frame", "test-token", 10.seconds, maxRetries = 0), + cache = CacheConfig(ttl = 5.minutes, softTtl = 4.minutes, maxStaleOnError = 10.minutes), + rateLimiter = RateLimiterConfig(maxRequestsPerMinute = 100), + circuitBreaker = CircuitBreakerConfig(maxFailures = 5, resetTimeout = 60.seconds) + ) + + private def allRates: List[Rate] = + Rate.Pair.allPairs.map(p => Rate(p, Price(BigDecimal("0.85")), Timestamp.now)) + + private def makeService(client: OneFrameClientAlgebra[IO], config: ApplicationConfig) = + OneFrameLive.make[IO](client, config).unsafeRunSync() + + "OneFrameLive retry" should { + + "retry on OneFrameUnreachable and return success after the third attempt" in { + val attempts = new AtomicInteger(0) + val client: OneFrameClientAlgebra[IO] = _ => IO { + val n = attempts.incrementAndGet() + if (n <= 2) ServiceError.OneFrameUnreachable(new RuntimeException(s"fail $n")).asLeft + else allRates.asRight + } + val config = baseConfig.copy(oneFrame = baseConfig.oneFrame.copy(maxRetries = 2)) + val service = makeService(client, config) + + val result = service.get(usdEur).unsafeRunSync() + + result.isRight shouldBe true + attempts.get() shouldBe 3 + } + + "not retry on quota exhaustion" in { + val attempts = new AtomicInteger(0) + val client: OneFrameClientAlgebra[IO] = _ => IO { + attempts.incrementAndGet() + ServiceError.OneFrameQuotaExceeded.asLeft[List[Rate]] + } + val config = baseConfig.copy(oneFrame = baseConfig.oneFrame.copy(maxRetries = 3)) + val service = makeService(client, config) + + service.get(usdEur).unsafeRunSync() + + attempts.get() shouldBe 1 + } + + "return the final error after all retries are exhausted" in { + val client: OneFrameClientAlgebra[IO] = _ => IO.pure( + ServiceError.OneFrameUnreachable(new RuntimeException("always fails")).asLeft + ) + val config = baseConfig.copy(oneFrame = baseConfig.oneFrame.copy(maxRetries = 1)) + val service = makeService(client, config) + + val result = service.get(usdEur).unsafeRunSync() + + result.left.value shouldBe a[ServiceError.OneFrameUnreachable] + } + } + + "OneFrameLive quota tracking" should { + + def withLogCapture(f: => Unit): List[ILoggingEvent] = { + val appender = new ListAppender[ILoggingEvent]() + appender.start() + val lbLogger = LoggerFactory + .getLogger(classOf[OneFrameLive[IO]]) + .asInstanceOf[LbLogger] + lbLogger.addAppender(appender) + try f + finally { val _ = lbLogger.detachAppender(appender) } + appender.list.asScala.toList + } + + "emit a WARN quota_alert when daily calls reach 800" in { + val todayEpoch = LocalDate.now(java.time.ZoneOffset.UTC).toEpochDay + val counter = Ref.of[IO, (Long, Long)]((todayEpoch, 799L)).unsafeRunSync() + val cache = InMemoryRatesCache.create[IO].unsafeRunSync() + val client: OneFrameClientAlgebra[IO] = _ => IO.pure(allRates.asRight) + + val service = OneFrameLive.makeWithCacheAndQuota[IO](client, cache, counter, baseConfig).unsafeRunSync() + + val events = withLogCapture { + val _ = service.get(usdEur).unsafeRunSync() + } + + val alert = events.find(e => e.getMessage.contains("quota_alert") && e.getMessage.contains("warning")) + alert shouldBe defined + alert.get.getLevel shouldBe Level.WARN + } + + "emit an ERROR quota_alert when daily calls reach 950" in { + val todayEpoch = LocalDate.now(java.time.ZoneOffset.UTC).toEpochDay + val counter = Ref.of[IO, (Long, Long)]((todayEpoch, 949L)).unsafeRunSync() + val cache = InMemoryRatesCache.create[IO].unsafeRunSync() + val client: OneFrameClientAlgebra[IO] = _ => IO.pure(allRates.asRight) + + val service = OneFrameLive.makeWithCacheAndQuota[IO](client, cache, counter, baseConfig).unsafeRunSync() + + val events = withLogCapture { + val _ = service.get(usdEur).unsafeRunSync() + } + + val alert = events.find(e => e.getMessage.contains("quota_alert") && e.getMessage.contains("critical")) + alert shouldBe defined + alert.get.getLevel shouldBe Level.ERROR + } + + "not emit a quota_alert below the warning threshold" in { + val todayEpoch = LocalDate.now(java.time.ZoneOffset.UTC).toEpochDay + val counter = Ref.of[IO, (Long, Long)]((todayEpoch, 500L)).unsafeRunSync() + val cache = InMemoryRatesCache.create[IO].unsafeRunSync() + val client: OneFrameClientAlgebra[IO] = _ => IO.pure(allRates.asRight) + + val service = OneFrameLive.makeWithCacheAndQuota[IO](client, cache, counter, baseConfig).unsafeRunSync() + + val events = withLogCapture { + val _ = service.get(usdEur).unsafeRunSync() + } + + events.exists(_.getMessage.contains("quota_alert")) shouldBe false + } + } +}