diff --git a/.gitignore b/.gitignore index cc9152f8..c58b0a9c 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,10 @@ target .ensime .ensime_lucene .ensime_cache +.metals +.bloop +.vscode +metals.sbt TAGS \#*# *~ diff --git a/forex-mtl/README.md b/forex-mtl/README.md new file mode 100644 index 00000000..f638bfe6 --- /dev/null +++ b/forex-mtl/README.md @@ -0,0 +1,276 @@ +# Forex Proxy Service + +## Problem Statement + +Build a caching proxy that serves 10,000+ exchange-rate requests per day from an upstream API limited to 1,000 requests per day, with rates no older than 5 minutes. + +### Constraints + +| Constraint | Value | +|---|---| +| One-Frame API rate limit | **1,000 requests/day** per token | +| Max pairs per One-Frame request | **320** (determined by local load testing) | +|Currencies to support | Total 170 currencies | +| Proxy traffic requirement | **10,000+ requests/day** | +| Max rate staleness | **5 minutes** | + +## Problem Analysis + +### Theoretical Minimum + +With a 5-minute cache TTL, rates must be refreshed at least every 5 minutes so total 24 * 60 / 5 = **288 API calls / day minimum**. + +**Can we fetch all potential currency pairs on each batch api calls so all requests will hit cache?** + +No, the maximum batch size is 320 but there are at least potential 170 * 169 = 28730 pairs to fetch. + +**Gap to close:** 288 (minimum) → 1,000 (budget) gives us **712 API calls of headroom** for discovery, long-tail pairs, and traffic bursts. + +### Naive Caching Approach +A naive reactive cache that fetches individual pairs and set cache by pair later on cache miss won't meet the 1000 API calls/ day requirements. + +Each pair will cost 288 apis per day. Only 5 Pairs will over the limit. + + +### Traffic Distribution Assumptions + +We can assume that the pairs between top 10 currencies contribute the 85% traffic. And the rest 15% traffic are the long-tail pairs. + +We can have assumptions below: + +``` +Total requests: 10,000 / day + +Popular pairs (top 10 currencies): + - Currencies: USD, EUR, JPY, GBP, AUD, CNY, CAD, CHF, SGD, NZD + - Pairs: 90 ordered pairs + - Traffic: 85% = 8,500 requests/day + - Avg requests per pair: ~94 requests/pair/day + +Long-tail pairs (all others): + - Unique Pairs: ~150 unique pairs (1500 requests with 10x duplication) + - Avg requests per pair: ~10 requests/pair/day +``` + +## Core Problem Analysis + +### Solution for popular pairs - Easy to Cache + +90 popular pairs with 8,500 requests/day: +``` +90 pairs < 320 batch limit + +Fetch all 90 pairs in ONE batch request per refresh cycle + +API calls: 288 cycles × 1 request = 288 calls/day +Cache hit rate: (8,500 − 288) / 8,500 ≈ 96.6% + +Within each 5-min window: + - First request (any pair) -> triggers fetch of all 90 pairs + - Next ~29 requests → cache hits +``` + +### Long-Tail Pairs - The Bottleneck + +~150 long-tail pairs with 1,500 requests/day: + +**Problem:** Long-tail pairs have MUCH lower cache hit rates: + +``` +150 unique pairs for 1500 requests / day -> ~10 requests/pair/day on average + +Inter-request interval: 1440 min / 10 = 144 minutes >> 5mins cache TTL + +``` + +**Core Problem:** Long-tail pairs have low request frequency relative to 5-minute cache TTL, causing high cache miss rates and excessive API calls. + +## Solution Design: Dynamic popular & long-tail pairs combination caching + +### Key Observation +We're already fetching 90 popular pairs every 5 minutes (288 times/day). The One-Frame API supports **up to 320 pairs per request** + +**Opportunity:** There are **230 spare slots** (320 − 90) in each batch request. Batching long-tail and popular pairs together can save api calls. + + +### Algorithm Design + +Keeping Pairs Fetching State: + - popular_pairs: Fixed set of 90 pairs + - encountered_longtail: Set of fetched long-tail pairs + +On cache miss request: + 1. Fetch: POPULAR_PAIRS & encountered_longtail & requested_pair + 2. Cache all fetched pairs (5-min TTL) + 3. Add requested_pair to encountered_longtail + +Batch size: 90 popular + ~150 dicovered = 240 pairs / request + +240 < 320 limit -> work under constrain + + +### API call budget calculation + +Popular Pairs minimum -> 288 requests / day +Long-tail pairs on first request -> 150 request / day + +Total 438 request / day -> Meet 1000 requests / day + + +## Implementation Options + +### 1. Option 1: Reactive Lazy Batch Fetching +Fetch on cache miss, batch all known pairs (popular pairs & encountered longtail paris & requested pair) together + +``` +pseudo code +if requested_pair in cache + return cache[requested_pair] +else: + batch = popular_pairs + encountered_longtail + requested_pair + rates = fetch_rates_from_api(batch) + cache.setAll(rates, ttl = 5.minutes) + encountered_longtail.add(requested_pair) + return cache[requested_pair] +``` + +Pros: + - Simple implementation and low complexity + - No fetching when no request coming + - Simple architecture & minimal dependencies: only cache + HTTP client + - Lowest API usage with high API quota room + - Self learning patterns automatically + +Cons: + - Thundering herd risk + - No pre-warm cache + - First request for new pair has higher latency + + +### Option 2: Proactive Periodic Background Refresh +Background scheduler refreshes all pairs every 4 minutes. Make request for new pair + +``` +pseudo code +Scheduler every 4 minuts: + batch = popular_pairs + encountered_longtail + requested_pair + rates = fetch_rates_from_api(batch) + cache.setAll(rates, ttl = 5.minutes) + +On request: + if cache[requested_pair]: + return cache[requested_pair] + else: + rate = fetch_single_rate_from_api(requested_pair) + cache.set(requested_pair, rate, ttl=300) + encountered_longtail.add(requested_pair) + return rate +``` + +Pros: +- Popular & encountered pairs always hit cache +- Distributed system ready + +Cons: +- Higher complexity: Background scheduler +- Wasted refreshes if no traffic + +## Final Decision + +Chosen Approach: Option 2 - Reactive Lazy Batch Fetching + +** Rationale:** +1. API useage meets 1000 calls/days + - Potentially only 288 calls / day (Theoretical Minimum) + +2. Simplest architecture: + - No background scheduler + - minimal dependencies + +3. Self-optimizing + - Automatically learns request patterns + - Adapts to changing traffic + +## Architecture + +``` +Client -> RatesHttpRoutes -> Program (cache logic) -> OneFrameLive (HTTP client) + | | + RatesCache One-Frame API + | + CacheClient (in-memory) +``` + +- **OneFrameLive** - Pure HTTP client for the One-Frame API. No caching logic. +- **Program** - Orchestrates cache check, batch refresh, and thundering herd protection. +- **RatesCache** - Domain adapter that maps `Rate.Pair` to cache keys and `Rate` to JSON values. +- **CacheClient** - Generic key-value store with TTL (Redis-like interface). Currently backed by an in-memory `Ref`. + +## Key Design & Considerations + +### Thundering herd protection + +When multiple requests comes in for the same pair and no cache not hit, then there could be multiple API request sent + +Using a `Semaphore(1)` with double-checked locking. Cache hits are never blocked. On cache miss, only one request refreshes while others wait, then find fresh data in cache. + +### Generic cache client interface + +`CacheClient` has a Redis-like interface (`get`, `set` with TTL, `keys`). This makes it straightforward to swap `InMemoryCacheClient` for Redis in a multi-pod deployment. + +### Per-route error handling + +Each route wraps its response in a route-specific error handler keeping error-to-HTTP mapping within the rates module rather than a global handler. This makes the error handling is explict and maintainable + +## Production metrics +- Cache hit rate +- API calls/day +- Batch sizes +- Error rate + + +## Assumptions and Simplifications + +- **Single pod**: The semaphore and in-memory cache are JVM-local. For multiple pods, replace `InMemoryCacheClient` with Redis and use distributed locks (`SETNX`). Redis Lock can be used for distributed lock for thundering herd problem +- **Cache TTL = 300 seconds (5 minutes)**: Matches the requirement that rates should be no older than 5 minutes. +- **Expired keys are included in `keys`**: So they get re-fetched on the next batch refresh rather than being lost. +- **Seed currencies**: The 10 currencies (CNY, AUD, CAD, CHF, EUR, GBP, NZD, JPY, SGD, USD) are pre-fetched on the first request. Additional pairs are added to the cache dynamically when requested. +This list could be a ENV config in production. + +## How to Run + +### Prerequisites + +- JDK 17 +- sbt +- Docker + +### 1. Start the One-Frame API + +```bash +docker run -p 8080:8080 paidyinc/one-frame +``` + +### 2. Start the proxy + +```bash +sbt run +``` + +The proxy starts on `http://localhost:8081`. + +### 3. Query a rate + +```bash +curl 'http://localhost:8081/rates?from=USD&to=JPY' +``` + +Response: +```json +{ + "from": "USD", + "to": "JPY", + "price": 0.123456, + "timestamp": "2026-02-12T02:47:29.605Z" +} +``` 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..a8a09e95 100644 --- a/forex-mtl/src/main/resources/application.conf +++ b/forex-mtl/src/main/resources/application.conf @@ -1,8 +1,13 @@ app { http { host = "0.0.0.0" - port = 8080 + port = 8081 timeout = 40 seconds } + one-frame { + host = "localhost" + port = 8080 + token = "10dc303535874aeccc86a8251e6992f5" + } } diff --git a/forex-mtl/src/main/scala/forex/Main.scala b/forex-mtl/src/main/scala/forex/Main.scala index 6dda10a7..498145df 100644 --- a/forex-mtl/src/main/scala/forex/Main.scala +++ b/forex-mtl/src/main/scala/forex/Main.scala @@ -3,7 +3,10 @@ package forex import scala.concurrent.ExecutionContext import cats.effect._ import forex.config._ +import forex.services.RatesServices +import forex.services.cache.{ InMemoryCacheClient, RatesCache } import fs2.Stream +import org.http4s.blaze.client.BlazeClientBuilder import org.http4s.blaze.server.BlazeServerBuilder object Main extends IOApp { @@ -17,8 +20,12 @@ class Application[F[_]: ConcurrentEffect: Timer] { def stream(ec: ExecutionContext): Stream[F, Unit] = for { - config <- Config.stream("app") - module = new Module[F](config) + config <- Config.stream("app") + client <- BlazeClientBuilder[F](ec).stream + cacheClient <- Stream.eval(InMemoryCacheClient[F]) + ratesCache = RatesCache[F](cacheClient) + ratesService = RatesServices.live[F](client, config.oneFrame) + module <- Stream.eval(Module[F](config, ratesService, ratesCache)) _ <- BlazeServerBuilder[F](ec) .bindHttp(config.http.port, config.http.host) .withHttpApp(module.httpApp) diff --git a/forex-mtl/src/main/scala/forex/Module.scala b/forex-mtl/src/main/scala/forex/Module.scala index 3bc47d58..2b37f4ca 100644 --- a/forex-mtl/src/main/scala/forex/Module.scala +++ b/forex-mtl/src/main/scala/forex/Module.scala @@ -1,19 +1,20 @@ package forex import cats.effect.{ Concurrent, Timer } +import cats.syntax.functor._ import forex.config.ApplicationConfig import forex.http.rates.RatesHttpRoutes -import forex.services._ import forex.programs._ +import forex.services._ +import forex.services.cache.RatesCache import org.http4s._ 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) +class Module[F[_]: Concurrent: Timer] private ( + config: ApplicationConfig, + ratesProgram: RatesProgram[F] +) { private val ratesHttpRoutes: HttpRoutes[F] = new RatesHttpRoutes[F](ratesProgram).routes @@ -35,3 +36,14 @@ class Module[F[_]: Concurrent: Timer](config: ApplicationConfig) { val httpApp: HttpApp[F] = appMiddleware(routesMiddleware(http).orNotFound) } + +object Module { + def apply[F[_]: Concurrent: Timer]( + config: ApplicationConfig, + ratesService: RatesService[F], + ratesCache: RatesCache[F] + ): F[Module[F]] = + for { + ratesProgram <- RatesProgram[F](ratesService, ratesCache) + } yield new Module[F](config, ratesProgram) +} diff --git a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala index eff0fad7..d026241a 100644 --- a/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala +++ b/forex-mtl/src/main/scala/forex/config/ApplicationConfig.scala @@ -4,6 +4,7 @@ import scala.concurrent.duration.FiniteDuration case class ApplicationConfig( http: HttpConfig, + oneFrame: OneFrameConfig, ) case class HttpConfig( @@ -11,3 +12,9 @@ case class HttpConfig( port: Int, timeout: FiniteDuration ) + +case class OneFrameConfig( + host: String, + port: Int, + token: String +) diff --git a/forex-mtl/src/main/scala/forex/domain/Currency.scala b/forex-mtl/src/main/scala/forex/domain/Currency.scala index a6f2857d..6fb47747 100644 --- a/forex-mtl/src/main/scala/forex/domain/Currency.scala +++ b/forex-mtl/src/main/scala/forex/domain/Currency.scala @@ -2,41 +2,17 @@ package forex.domain import cats.Show -sealed trait Currency +case class Currency(code: String) extends AnyVal object Currency { - case object AUD extends Currency - case object CAD extends Currency - case object CHF extends Currency - case object EUR extends Currency - case object GBP extends Currency - case object NZD extends Currency - case object JPY extends Currency - case object SGD extends Currency - case object USD extends Currency + implicit val show: Show[Currency] = Show.show(_.code) - 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" - } - - 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 + def fromString(s: String): Either[String, Currency] = { + val upper = s.trim.toUpperCase + if (upper.matches("[A-Z]{3}")) Right(Currency(upper)) + else Left(s"Invalid currency code: $s") } + val seedCurrencies: List[Currency] = + List("AUD", "CAD", "CHF", "EUR", "GBP", "NZD", "JPY", "SGD", "USD", "CNY").map(Currency(_)) } diff --git a/forex-mtl/src/main/scala/forex/domain/Rate.scala b/forex-mtl/src/main/scala/forex/domain/Rate.scala index 4a444003..d38e2d91 100644 --- a/forex-mtl/src/main/scala/forex/domain/Rate.scala +++ b/forex-mtl/src/main/scala/forex/domain/Rate.scala @@ -11,4 +11,11 @@ object Rate { from: Currency, to: Currency ) + + val seedPairs: List[Pair] = + for { + from <- Currency.seedCurrencies + to <- Currency.seedCurrencies + if from != to + } yield Pair(from, to) } diff --git a/forex-mtl/src/main/scala/forex/http/rates/ErrorHandler.scala b/forex-mtl/src/main/scala/forex/http/rates/ErrorHandler.scala new file mode 100644 index 00000000..70efccf8 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/http/rates/ErrorHandler.scala @@ -0,0 +1,32 @@ +package forex.http +package rates + +import cats.effect.Sync +import cats.syntax.applicative._ +import cats.syntax.applicativeError._ +import forex.programs.rates.errors.{ Error => ProgramError } +import io.circe.Json +import org.http4s.{ Response, Status } +import org.http4s.circe.jsonEncoder +import org.slf4j.LoggerFactory + +object ErrorHandler { + + private val logger = LoggerFactory.getLogger(getClass) + + private def errorResponse[F[_]: Sync](status: Status, msg: String): F[Response[F]] = + Response[F](status).withEntity(Json.obj("error" -> Json.fromString(msg))).pure[F] + + def getRates[F[_]: Sync](result: F[Response[F]]): F[Response[F]] = + result.handleErrorWith { + case e: ProgramError.RateLookupFailed => + logger.error(s"Rate lookup failed: ${e.msg}") + errorResponse(Status.BadGateway, e.msg) + case e: IllegalArgumentException => + logger.warn(s"Invalid input: ${e.getMessage}") + errorResponse(Status.BadRequest, e.getMessage) + case e => + logger.error(s"Unexpected error: ${e.getMessage}", e) + errorResponse(Status.InternalServerError, "Internal server error") + } +} 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..5ca18ffc 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,17 @@ package forex.http.rates import forex.domain.Currency -import org.http4s.QueryParamDecoder -import org.http4s.dsl.impl.QueryParamDecoderMatcher +import org.http4s.{ ParseFailure, QueryParamDecoder } +import org.http4s.dsl.impl.ValidatingQueryParamDecoderMatcher object QueryParams { private[http] implicit val currencyQueryParam: QueryParamDecoder[Currency] = - QueryParamDecoder[String].map(Currency.fromString) + QueryParamDecoder[String].emap(s => + Currency.fromString(s).left.map(msg => ParseFailure(msg, msg)) + ) - object FromQueryParam extends QueryParamDecoderMatcher[Currency]("from") - object ToQueryParam extends QueryParamDecoderMatcher[Currency]("to") + object FromQueryParam extends ValidatingQueryParamDecoderMatcher[Currency]("from") + object ToQueryParam extends ValidatingQueryParamDecoderMatcher[Currency]("to") } 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..d05ad974 100644 --- a/forex-mtl/src/main/scala/forex/http/rates/RatesHttpRoutes.scala +++ b/forex-mtl/src/main/scala/forex/http/rates/RatesHttpRoutes.scala @@ -1,6 +1,7 @@ package forex.http package rates +import cats.data.Validated.Valid import cats.effect.Sync import cats.syntax.flatMap._ import forex.programs.RatesProgram @@ -17,8 +18,15 @@ class RatesHttpRoutes[F[_]: Sync](rates: RatesProgram[F]) extends Http4sDsl[F] { 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) + ErrorHandler.getRates { + (from, to) match { + case (Valid(f), Valid(t)) => + rates.get(RatesProgramProtocol.GetRatesRequest(f, t)) + .flatMap(Sync[F].fromEither(_)) + .flatMap(rate => Ok(rate.asGetApiResponse)) + case _ => + Sync[F].raiseError(new IllegalArgumentException(s"Invalid input")) + } } } 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..40e8587e 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,77 @@ package forex.programs.rates -import cats.Functor -import cats.data.EitherT +import cats.effect.Concurrent +import cats.effect.concurrent.Semaphore +import cats.syntax.either._ +import cats.syntax.flatMap._ +import cats.syntax.functor._ import errors._ import forex.domain._ import forex.services.RatesService +import forex.services.cache.RatesCache +import org.slf4j.LoggerFactory -class Program[F[_]: Functor]( - ratesService: RatesService[F] +class Program[F[_]: Concurrent]( + ratesService: RatesService[F], + cache: RatesCache[F], + lock: Semaphore[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 + private val logger = LoggerFactory.getLogger(getClass) -} + override def get(request: Protocol.GetRatesRequest): F[Error Either Rate] = { + val pair = Rate.Pair(request.from, request.to) + for { + cached <- cache.get(pair) + result <- cached match { + case Some(rate) => Concurrent[F].pure(rate.asRight[Error]) + case None => withLock(pair) + } + } yield result + } -object Program { + private def withLock(pair: Rate.Pair): F[Error Either Rate] = + lock.withPermit { + for { + cached <- cache.get(pair) // double-check after acquiring lock + result <- cached match { + case Some(rate) => + logger.debug(s"Cache hit after lock for ${pair.from.code}/${pair.to.code}") + Concurrent[F].pure(rate.asRight[Error]) + case None => + logger.info(s"Cache miss for ${pair.from.code}/${pair.to.code}, refreshing") + refresh(pair) + } + } yield result + } - def apply[F[_]: Functor]( - ratesService: RatesService[F] - ): Algebra[F] = new Program[F](ratesService) + private def refresh(pair: Rate.Pair): F[Error Either Rate] = + for { + currentKeys <- cache.keys + pairsToFetch = if (currentKeys.isEmpty) + (Rate.seedPairs.toSet + pair).toList + else + (currentKeys + pair).toList + result <- ratesService.get(pairsToFetch) + out <- result match { + case Right(fetched) => + cache.put(fetched).map { _ => + fetched.get(pair) + .toRight(Error.RateLookupFailed(s"Pair not in response: ${pair.from.code}${pair.to.code}")) + } + case Left(err) => + logger.error(s"Refresh failed: ${err.msg}") + Concurrent[F].pure(Error.RateLookupFailed(err.msg).asLeft[Rate]) + } + } yield out +} +object Program { + def apply[F[_]: Concurrent]( + ratesService: RatesService[F], + cache: RatesCache[F] + ): F[Algebra[F]] = + Semaphore[F](1).map { lock => + new Program[F](ratesService, cache, lock) + } } 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..40d9c519 100644 --- a/forex-mtl/src/main/scala/forex/programs/rates/errors.scala +++ b/forex-mtl/src/main/scala/forex/programs/rates/errors.scala @@ -1,15 +1,12 @@ package forex.programs.rates -import forex.services.rates.errors.{ Error => RatesServiceError } - object errors { - sealed trait Error extends Exception + sealed trait Error extends Exception { + def msg: String + } object Error { final case class RateLookupFailed(msg: String) extends Error } - def toProgramError(error: RatesServiceError): Error = error match { - case RatesServiceError.OneFrameLookupFailed(msg) => Error.RateLookupFailed(msg) - } } diff --git a/forex-mtl/src/main/scala/forex/services/cache/CacheClient.scala b/forex-mtl/src/main/scala/forex/services/cache/CacheClient.scala new file mode 100644 index 00000000..488e921e --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/cache/CacheClient.scala @@ -0,0 +1,7 @@ +package forex.services.cache + +trait CacheClient[F[_]] { + def get(key: String): F[Option[String]] + def set(key: String, value: String, seconds: Long = 300): F[Unit] + def keys: F[Set[String]] +} diff --git a/forex-mtl/src/main/scala/forex/services/cache/InMemoryCacheClient.scala b/forex-mtl/src/main/scala/forex/services/cache/InMemoryCacheClient.scala new file mode 100644 index 00000000..4294764a --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/cache/InMemoryCacheClient.scala @@ -0,0 +1,37 @@ +package forex.services.cache + +import cats.effect.{ Clock, Sync } +import cats.effect.concurrent.Ref +import cats.syntax.flatMap._ +import cats.syntax.functor._ + +import java.util.concurrent.TimeUnit + +class InMemoryCacheClient[F[_]: Sync: Clock]( + ref: Ref[F, Map[String, InMemoryCacheClient.Entry]] +) extends CacheClient[F] { + import InMemoryCacheClient._ + + override def get(key: String): F[Option[String]] = + for { + now <- Clock[F].monotonic(TimeUnit.SECONDS) + entries <- ref.get + } yield entries.get(key).filter(e => now < e.expiresAt).map(_.value) + + override def set(key: String, value: String, seconds: Long = 300): F[Unit] = + Clock[F].monotonic(TimeUnit.SECONDS).flatMap { now => + ref.update(_ + (key -> Entry(value, now + seconds))) + } + + override def keys: F[Set[String]] = + ref.get.map(_.keySet) +} + +object InMemoryCacheClient { + private[cache] case class Entry(value: String, expiresAt: Long) + + def apply[F[_]: Sync: Clock]: F[CacheClient[F]] = + Ref.of[F, Map[String, Entry]](Map.empty).map { ref => + new InMemoryCacheClient[F](ref) + } +} diff --git a/forex-mtl/src/main/scala/forex/services/cache/RatesCache.scala b/forex-mtl/src/main/scala/forex/services/cache/RatesCache.scala new file mode 100644 index 00000000..c7888569 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/cache/RatesCache.scala @@ -0,0 +1,59 @@ +package forex.services.cache + +import cats.Monad +import cats.instances.list._ +import cats.syntax.foldable._ +import cats.syntax.functor._ +import forex.domain._ +import io.circe.{ Decoder, Encoder, HCursor, Json } +import io.circe.syntax._ + +class RatesCache[F[_]: Monad](cache: CacheClient[F]) { + import RatesCache._ + + def get(pair: Rate.Pair): F[Option[Rate]] = + cache.get(pairToKey(pair)).map(_.flatMap(parseRate(pair, _))) + + def keys: F[Set[Rate.Pair]] = + cache.keys.map(_.flatMap(keyToPair)) + + def put(rates: Map[Rate.Pair, Rate], seconds: Long = 300): F[Unit] = + rates.toList.traverse_ { case (pair, rate) => + cache.set(pairToKey(pair), rateToValue(rate), seconds) + } +} + +object RatesCache { + def apply[F[_]: Monad](cache: CacheClient[F]): RatesCache[F] = + new RatesCache[F](cache) + + private val keyPrefix = "rate:" + + private def pairToKey(pair: Rate.Pair): String = + s"$keyPrefix${pair.from.code}:${pair.to.code}" + + private def keyToPair(key: String): Option[Rate.Pair] = + key.stripPrefix(keyPrefix).split(":") match { + case Array(from, to) => Some(Rate.Pair(Currency(from), Currency(to))) + case _ => None + } + + private case class RateValue(price: BigDecimal, timestamp: String) + + private implicit val rateValueEncoder: Encoder[RateValue] = + (v: RateValue) => Json.obj("price" -> v.price.asJson, "timestamp" -> v.timestamp.asJson) + + private implicit val rateValueDecoder: Decoder[RateValue] = + (c: HCursor) => for { + price <- c.downField("price").as[BigDecimal] + timestamp <- c.downField("timestamp").as[String] + } yield RateValue(price, timestamp) + + private def rateToValue(rate: Rate): String = + RateValue(rate.price.value, rate.timestamp.value.toString).asJson.noSpaces + + private def parseRate(pair: Rate.Pair, value: String): Option[Rate] = + io.circe.parser.decode[RateValue](value).toOption.map { rv => + Rate(pair, Price(rv.price), Timestamp(java.time.OffsetDateTime.parse(rv.timestamp))) + } +} 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..cf046344 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,14 @@ package forex.services.rates import cats.Applicative +import cats.effect.Sync +import forex.config.OneFrameConfig import interpreters._ +import org.http4s.client.Client object Interpreters { def dummy[F[_]: Applicative]: Algebra[F] = new OneFrameDummy[F]() + + def live[F[_]: Sync](client: Client[F], config: OneFrameConfig): Algebra[F] = + OneFrameLive[F](client, 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..e108df62 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/algebra.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/algebra.scala @@ -4,5 +4,5 @@ import forex.domain.Rate import errors._ trait Algebra[F[_]] { - def get(pair: Rate.Pair): F[Error Either Rate] + def get(pairs: List[Rate.Pair]): F[Error Either Map[Rate.Pair, Rate]] } 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..dad707b9 100644 --- a/forex-mtl/src/main/scala/forex/services/rates/errors.scala +++ b/forex-mtl/src/main/scala/forex/services/rates/errors.scala @@ -2,7 +2,9 @@ package forex.services.rates object errors { - sealed trait Error + sealed trait Error extends Exception { + def msg: String + } object 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..49094c7f 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 @@ -9,7 +9,7 @@ import forex.services.rates.errors._ 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 get(pairs: List[Rate.Pair]): F[Error Either Map[Rate.Pair, Rate]] = + pairs.map(p => p -> Rate(p, Price(BigDecimal(100)), Timestamp.now)).toMap.asRight[Error].pure[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..915ed1b3 --- /dev/null +++ b/forex-mtl/src/main/scala/forex/services/rates/interpreters/OneFrameLive.scala @@ -0,0 +1,86 @@ +package forex.services.rates.interpreters + +import cats.effect.Sync +import cats.syntax.applicativeError._ +import cats.syntax.either._ +import cats.syntax.functor._ +import forex.config.OneFrameConfig +import forex.domain._ +import forex.services.rates.Algebra +import forex.services.rates.errors._ +import io.circe.{ Decoder, HCursor } +import org.http4s.{ Header, Method, Query, Request, Uri } +import org.http4s.client.Client +import org.typelevel.ci.CIString + +import java.time.OffsetDateTime +import org.slf4j.LoggerFactory + +class OneFrameLive[F[_]: Sync]( + client: Client[F], + config: OneFrameConfig +) extends Algebra[F] { + import OneFrameLive._ + + private val logger = LoggerFactory.getLogger(getClass) + + override def get(pairs: List[Rate.Pair]): F[Error Either Map[Rate.Pair, Rate]] = { + val baseUri = Uri.unsafeFromString(s"http://${config.host}:${config.port}/rates") + val query = Query.fromPairs(pairs.map(p => "pair" -> s"${p.from.code}${p.to.code}"): _*) + val uri = baseUri.copy(query = query) + val request = Request[F](Method.GET, uri).withHeaders(Header.Raw(CIString("token"), config.token)) + + client.expect[String](request).attempt.map { + case Right(body) => + io.circe.parser.decode[OneFrameError](body) match { + case Right(err) => + logger.error(s"One-Frame returned error: ${err.error}") + Error.OneFrameLookupFailed(err.error).asLeft[Map[Rate.Pair, Rate]] + case Left(_) => + io.circe.parser.decode[List[OneFrameRate]](body) match { + case Right(rates) => + logger.info(s"Fetched ${rates.size} rates from One-Frame") + rates.map { r => + val p = Rate.Pair(r.from, r.to) + p -> Rate(p, Price(r.price), Timestamp(r.timeStamp)) + }.toMap.asRight[Error] + case Left(err) => + logger.error(s"Failed to parse One-Frame response: ${err.getMessage}") + Error.OneFrameLookupFailed(s"Failed to parse response: ${err.getMessage}").asLeft[Map[Rate.Pair, Rate]] + } + } + case Left(err) => + logger.error(s"One-Frame API call failed: ${err.getMessage}", err) + Error.OneFrameLookupFailed(s"API call failed: ${err.getMessage}").asLeft[Map[Rate.Pair, Rate]] + } + } +} + +object OneFrameLive { + def apply[F[_]: Sync](client: Client[F], config: OneFrameConfig): Algebra[F] = + new OneFrameLive[F](client, config) + + private[interpreters] case class OneFrameError(error: String) + + private[interpreters] implicit val oneFrameErrorDecoder: Decoder[OneFrameError] = + Decoder.forProduct1("error")(OneFrameError.apply) + + private[interpreters] case class OneFrameRate( + from: Currency, + to: Currency, + price: BigDecimal, + timeStamp: OffsetDateTime + ) + + private[interpreters] implicit val currencyDecoder: Decoder[Currency] = + Decoder.decodeString.emap(Currency.fromString) + + private[interpreters] implicit val oneFrameRateDecoder: Decoder[OneFrameRate] = + (c: HCursor) => + for { + from <- c.downField("from").as[Currency] + to <- c.downField("to").as[Currency] + price <- c.downField("price").as[BigDecimal] + timeStamp <- c.downField("time_stamp").as[String].map(OffsetDateTime.parse) + } yield OneFrameRate(from, to, price, timeStamp) +} 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..de685e29 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/domain/CurrencySpec.scala @@ -0,0 +1,32 @@ +package forex.domain + +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class CurrencySpec extends AnyWordSpec with Matchers { + + "Currency.fromString" should { + "accept valid 3-letter currency codes" in { + Currency.fromString("USD") shouldBe Right(Currency("USD")) + } + + "be case-insensitive" in { + Currency.fromString("usd") shouldBe Right(Currency("USD")) + } + + "trim whitespace" in { + Currency.fromString(" JPY ") shouldBe Right(Currency("JPY")) + } + + "reject codes that are not 3 letters" in { + Currency.fromString("ABCD").isLeft shouldBe true + Currency.fromString("AB").isLeft shouldBe true + Currency.fromString("").isLeft shouldBe true + } + + "reject codes with non-alpha characters" in { + Currency.fromString("12A").isLeft shouldBe true + Currency.fromString("U$D").isLeft shouldBe true + } + } +} diff --git a/forex-mtl/src/test/scala/forex/programs/rates/ProgramSpec.scala b/forex-mtl/src/test/scala/forex/programs/rates/ProgramSpec.scala new file mode 100644 index 00000000..6058dc91 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/programs/rates/ProgramSpec.scala @@ -0,0 +1,90 @@ +package forex.programs.rates + +import java.util.concurrent.atomic.AtomicInteger +import cats.effect.{ ContextShift, IO, Timer } +import cats.instances.list._ +import cats.syntax.parallel._ +import forex.domain._ +import forex.services.cache.{ InMemoryCacheClient, RatesCache } +import forex.services.rates.{ Algebra => RatesAlgebra } +import forex.services.rates.errors.{ Error => ServiceError } +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import java.time.OffsetDateTime +import scala.concurrent.ExecutionContext + +class ProgramSpec extends AnyWordSpec with Matchers { + + implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global) + implicit val timer: Timer[IO] = IO.timer(ExecutionContext.global) + + private val usdJpy = Rate.Pair(Currency("USD"), Currency("JPY")) + + private val testRate = Rate( + usdJpy, + Price(BigDecimal("110.5")), + Timestamp(OffsetDateTime.parse("2026-01-01T00:00:00Z")) + ) + + private def setup( + response: ServiceError Either Map[Rate.Pair, Rate] + ): IO[(Algebra[IO], RatesCache[IO])] = + for { + cacheClient <- InMemoryCacheClient[IO] + cache = RatesCache[IO](cacheClient) + service: RatesAlgebra[IO] = (_: List[Rate.Pair]) => IO.pure(response) + program <- Program[IO](service, cache) + } yield (program, cache) + + "Program.get" should { + "fetch from service on cache miss and cache the result" in { + val (result, cached) = (for { + pc <- setup(Right(Map(usdJpy -> testRate))) + (prog, cache) = pc + result <- prog.get(Protocol.GetRatesRequest(Currency("USD"), Currency("JPY"))) + cached <- cache.get(usdJpy) + } yield (result, cached)).unsafeRunSync() + + result shouldBe Right(testRate) + cached shouldBe Some(testRate) + } + + "include seed pairs in first fetch" in { + var fetchedPairs: List[Rate.Pair] = Nil // scalafix:ok + val trackingService: RatesAlgebra[IO] = (pairs: List[Rate.Pair]) => { + fetchedPairs = pairs + IO.pure(Right(pairs.map(p => p -> testRate.copy(pair = p)).toMap)) + } + + (for { + cacheClient <- InMemoryCacheClient[IO] + cache = RatesCache[IO](cacheClient) + program <- Program[IO](trackingService, cache) + _ <- program.get(Protocol.GetRatesRequest(Currency("USD"), Currency("JPY"))) + } yield ()).unsafeRunSync() + + fetchedPairs.size should be >= Rate.seedPairs.size + fetchedPairs should contain(usdJpy) + } + + "only call service once for concurrent requests on cache miss" in { + val callCount = new AtomicInteger(0) + val slowService: RatesAlgebra[IO] = (pairs: List[Rate.Pair]) => { + callCount.incrementAndGet() + IO.sleep(scala.concurrent.duration.FiniteDuration(100, "ms")) *> + IO.pure(Right(pairs.map(p => p -> testRate.copy(pair = p)).toMap)) + } + + val count = (for { + cacheClient <- InMemoryCacheClient[IO] + cache = RatesCache[IO](cacheClient) + program <- Program[IO](slowService, cache) + req = program.get(Protocol.GetRatesRequest(Currency("USD"), Currency("JPY"))) + _ <- (1 to 5).toList.parTraverse(_ => req) + } yield callCount.get()).unsafeRunSync() + + count shouldBe 1 + } + } +} diff --git a/forex-mtl/src/test/scala/forex/services/cache/InMemoryCacheClientSpec.scala b/forex-mtl/src/test/scala/forex/services/cache/InMemoryCacheClientSpec.scala new file mode 100644 index 00000000..fd3017e5 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/services/cache/InMemoryCacheClientSpec.scala @@ -0,0 +1,58 @@ +package forex.services.cache + +import cats.effect.{ ContextShift, IO, Timer } +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import scala.concurrent.ExecutionContext + +class InMemoryCacheClientSpec extends AnyWordSpec with Matchers { + + implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global) + implicit val timer: Timer[IO] = IO.timer(ExecutionContext.global) + + def createClient: IO[CacheClient[IO]] = InMemoryCacheClient[IO] + + "InMemoryCacheClient" should { + "store and retrieve a value" in { + val result = (for { + client <- createClient + _ <- client.set("key1", "value1") + v <- client.get("key1") + } yield v).unsafeRunSync() + + result shouldBe Some("value1") + } + + "return None for missing key" in { + val result = (for { + client <- createClient + v <- client.get("missing") + } yield v).unsafeRunSync() + + result shouldBe None + } + + "return None for expired key" in { + val result = (for { + client <- createClient + _ <- client.set("key1", "value1", 0) + _ <- IO.sleep(scala.concurrent.duration.FiniteDuration(50, "ms")) + v <- client.get("key1") + } yield v).unsafeRunSync() + + result shouldBe None + } + + "include expired keys in keys" in { + val result = (for { + client <- createClient + _ <- client.set("key1", "value1", 0) + _ <- IO.sleep(scala.concurrent.duration.FiniteDuration(50, "ms")) + ks <- client.keys + } yield ks).unsafeRunSync() + + result should contain("key1") + } + } +} 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..9c0329d1 --- /dev/null +++ b/forex-mtl/src/test/scala/forex/services/rates/interpreters/OneFrameLiveSpec.scala @@ -0,0 +1,60 @@ +package forex.services.rates.interpreters + +import cats.effect.IO +import forex.config.OneFrameConfig +import forex.domain.{ Currency, Rate } +import org.http4s.{ HttpApp, Response, Status } +import org.http4s.client.Client +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class OneFrameLiveSpec extends AnyWordSpec with Matchers { + + private val config = OneFrameConfig("localhost", 8080, "test-token") + private val usdJpy = Rate.Pair(Currency("USD"), Currency("JPY")) + + private def clientReturning(body: String, status: Status = Status.Ok): Client[IO] = + Client.fromHttpApp[IO](HttpApp.pure(Response[IO](status).withEntity(body))) + + "OneFrameLive" should { + "parse a successful response" in { + val json = + """[{"from":"USD","to":"JPY","bid":0.61,"ask":0.82,"price":0.71,"time_stamp":"2026-01-01T00:00:00Z"}]""" + val service = OneFrameLive[IO](clientReturning(json), config) + val result = service.get(List(usdJpy)).unsafeRunSync() + + result.isRight shouldBe true + val rates = result.getOrElse(fail()) + rates should contain key usdJpy + rates(usdJpy).price.value shouldBe BigDecimal("0.71") + } + + "return error when One-Frame returns an error body" in { + val json = """{"error":"Invalid Currency Pair"}""" + val service = OneFrameLive[IO](clientReturning(json), config) + val result = service.get(List(usdJpy)).unsafeRunSync() + + result.isLeft shouldBe true + result.left.getOrElse(fail()).msg shouldBe "Invalid Currency Pair" + } + + "return error when response body is malformed" in { + val service = OneFrameLive[IO](clientReturning("not json at all"), config) + val result = service.get(List(usdJpy)).unsafeRunSync() + + result.isLeft shouldBe true + result.left.getOrElse(fail()).msg should include("Failed to parse response") + } + + "return error when HTTP call fails" in { + val failingClient = Client.fromHttpApp[IO]( + HttpApp[IO](_ => IO.raiseError(new RuntimeException("connection refused"))) + ) + val service = OneFrameLive[IO](failingClient, config) + val result = service.get(List(usdJpy)).unsafeRunSync() + + result.isLeft shouldBe true + result.left.getOrElse(fail()).msg should include("connection refused") + } + } +}