09 — Concurrency: Futures, ExecutionContext & Backpressure

import scala.concurrent.{Future, Promise, ExecutionContext}
import scala.concurrent.duration.*
import scala.util.{Success, Failure}

// Future[T] represents an async computation that will produce T or fail with Throwable.
// A Future is either: Incomplete | Completed(Success(T)) | Completed(Failure(ex))
// Once completed, it's IMMUTABLE — you can't change its result.

// Future starts executing IMMEDIATELY upon construction (eager, not lazy)
// on the provided ExecutionContext (thread pool).
def fetchUser(id: Long)(using ec: ExecutionContext): Future[User] = Future {
  httpClient.get(s"/users/$id").as[User]   // runs on ec's thread pool
}

// Transformations return NEW Futures — they don't mutate the original
// map: runs after the source Future completes successfully
val withName: Future[String] = fetchUser(42).map(_.name)

// flatMap: sequencng — the returned Future determines the next async step
val withPosts: Future[(User, List[Post])] = fetchUser(42).flatMap { user =>
  fetchPosts(user.id).map(posts => (user, posts))  // only starts after fetchUser succeeds
}

// for-comprehension is sugar for flatMap/map — sequential, not parallel
val pipeline: Future[String] =
  for
    user  <- fetchUser(42)                // starts immediately
    posts <- fetchPosts(user.id)          // starts AFTER fetchUser completes
    top   <- fetchTopPost(posts)          // starts AFTER fetchPosts completes
  yield top.title
// Each step waits for the previous — total time = sum of all steps

// ❌ ANTI-PATTERN: sequential when you want parallel
for
  users <- fetchUsers()                    // 100ms
  stats <- fetchStats()                    // 100ms — doesn't start until fetchUsers done
yield (users, stats)
// Total: 200ms (sequential)

// ✅ CORRECT: start both immediately, wait for both
val users = fetchUsers()                    // starts NOW
val stats = fetchStats()                    // starts NOW (parallel with users)
val combined: Future[(List[User], Stats)] = users.zip(stats)
// Total: ~100ms (parallel, limited by slowest)

// zip vs zipPar:
//   zip: completes when both complete; if one fails, the result fails
//   In Scala stdlib, Future.zip already runs both in parallel (both are already started)

// Future.sequence — N parallel futures → one Future of List
val futures: List[Future[User]] = userIds.map(fetchUser)  // all start immediately
val allUsers: Future[List[User]] = Future.sequence(futures)  // completes when all complete

// ❌ ANTI-PATTERN: sequence on a HUGE list → overwhelming downstream service
// 10,000 parallel fetches → connection pool exhaustion → timeouts
// ✅ CORRECT: use batching or a semaphore to limit concurrency
def fetchBatched(ids: List[Long], batchSize: Int = 50): Future[List[User]] =
  ids.grouped(batchSize).toList            // List[List[Long]] of 50 each
    .foldLeft(Future.successful(List.empty[User])) { (acc, batch) =>
      acc.flatMap(done => Future.sequence(batch.map(fetchUser)).map(done ++ _))
    }

// Promise is the write-side of a Future. Future is the read-side.
// A Promise can be completed exactly once — success or failure.

// Use case: bridge callback-based APIs to Future-based code
def fromCallback[A](register: (A => Unit, Throwable => Unit) => Unit): Future[A] =
  val p = Promise[A]()
  register(
    value => p.success(value),             // complete the promise with a value
    error => p.failure(error)              // complete the promise with an exception
  )
  p.future                                  // the read-side Future

// Use case: timeout race
def withTimeout[T](f: Future[T], timeout: FiniteDuration)(using ec: ExecutionContext): Future[T] =
  val p = Promise[T]()
  f.onComplete(p.complete)                  // complete p with f's result (success or failure)
  ec.execute(() =>                          // schedule timeout on the same EC
    Thread.sleep(timeout.toMillis)
    p.tryFailure(new TimeoutException(s"timed out after $timeout"))
  )
  p.future

// trySuccess / tryFailure: returns Boolean — true if this call completed the promise
// success / failure: throws if already completed (shouldn't happen, but defensive)

import java.util.concurrent.{Executors, ThreadPoolExecutor, TimeUnit}
import scala.concurrent.ExecutionContext

// ❌ ANTI-PATTERN: using ExecutionContext.global for blocking I/O
// global is a ForkJoinPool with parallelism = CPU cores.
// Blocking calls (Thread.sleep, JDBC, HTTP) consume threads — pool starvation.
implicit val global: ExecutionContext = ExecutionContext.global
Future { Thread.sleep(5000); 42 }  // occupies a thread for 5s — with 8 cores, 8 of these = deadlock

// ✅ CORRECT: separate ECs for CPU-bound vs blocking I/O
val cpuEC: ExecutionContext = ExecutionContext.global   // for CPU-bound (map, filter, compute)

val ioEC: ExecutionContext = ExecutionContext.fromExecutor(
  Executors.newFixedThreadPool(50)  // 50 threads for I/O — higher than CPU count
  // Each thread can block on I/O without starving CPU-bound work
)

// Use the right EC for the right work:
Future { parseHugeJson(data) }(cpuEC)       // CPU-bound → use cpuEC
Future { jdbc.query("SELECT ...") }(ioEC)   // blocking I/O → use ioEC

// blocking { } — tells the EC this code will block, provision extra threads
// ONLY works with ExecutionContext.global (ForkJoinPool manages spare threads)
import scala.concurrent.blocking
Future {
  blocking {
    Thread.sleep(5000)                    // FJP creates a spare thread to compensate
  }
  42
}(global)

// recover: transform a failure into a success (catch + fallback value)
val withDefault: Future[Int] = riskyFuture.recover {
  case _: TimeoutException => 0            // timeout → default to 0
  case _: ConnectionRefused => -1          // connection refused → -1
  // other exceptions propagate as Failure
}

// recoverWith: transform a failure into a new Future (retry/fallback computation)
val withRetry: Future[String] = fetch(url).recoverWith {
  case _: TimeoutException => fetch(fallbackUrl)  // try different URL on timeout
}

// fallbackTo: try another Future if the first fails
val withFallback = primary.fetch().fallbackTo(secondary.fetch())
// ⚠️ fallbackTo swallows the original exception — you lose error diagnostics

// ❌ ANTI-PATTERN: using try/catch inside a Future
Future {
  try riskyOp()
  catch case e: Exception => default       // catches ALL exceptions including fatal ones!
}
// ✅ CORRECT: use recover — only catches non-fatal, composable
riskyFuture.recover { case _: NonFatal => default }

import scala.concurrent.{Future, ExecutionContext}
import scala.concurrent.duration.*
import scala.util.{Success, Failure, Random}

// Production-grade retry: exponential backoff + jitter + max retries + circuit breaking
def retryWithBackoff[T](
  operation: () => Future[T],
  maxRetries: Int = 3,
  initialDelay: FiniteDuration = 100.millis,
  maxDelay: FiniteDuration = 10.seconds,
  jitter: Double = 0.2                     // ±20% jitter to avoid thundering herd
)(using ec: ExecutionContext): Future[T] =

  def backoff(attempt: Int): FiniteDuration =
    val base = initialDelay * math.pow(2, attempt - 1).toLong  // 100ms, 200ms, 400ms...
    val capped = base.min(maxDelay)
    val jittered = capped * (1.0 + (Random.nextDouble() * 2 - 1) * jitter)
    jittered.asInstanceOf[FiniteDuration]

  def attempt(n: Int): Future[T] =
    operation().recoverWith {
      case ex if n < maxRetries =>
        val delay = backoff(n)
        Future.sleep(delay).flatMap(_ => attempt(n + 1))  // schedule retry after delay
      case ex =>
        Future.failed(ex)                  // exhausted retries → propagate original error
    }

  attempt(1)

// Usage:
retryWithBackoff(() => httpClient.post(url, payload), maxRetries = 5)
  .map(response => println(s"Success: $response"))
  .recover { case ex => println(s"All retries exhausted: $ex") }

import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
import java.util.concurrent.ConcurrentHashMap

// ❌ ANTI-PATTERN: shared var in Futures — data race
var counter = 0
val futures = (1 to 1000).map(_ => Future { counter += 1 })
Future.sequence(futures).map(_ => counter)  // → some number < 1000 (lost updates!)
// counter += 1 is read-modify-write: 3 instructions, non-atomic, interleaved across threads

// ✅ CORRECT: AtomicInteger for counter-like state
val atomicCounter = new AtomicLong(0)
val futures2 = (1 to 1000).map(_ => Future { atomicCounter.incrementAndGet() })
Future.sequence(futures2).map(_ => atomicCounter.get())  // → 1000 (guaranteed)

// ✅ CORRECT: CAS (compare-and-swap) for conditional updates
val state = new AtomicReference[String]("initial")
Future {
  var success = false
  while !success do
    val current = state.get()
    val updated = transform(current)
    success = state.compareAndSet(current, updated)  // atomic: only sets if still == current
}

// ✅ CORRECT: ConcurrentHashMap for shared maps (lock-free reads, striped writes)
val cache = new ConcurrentHashMap[String, Array[Byte]]()
cache.put("key", data)                     // thread-safe
cache.computeIfAbsent("key", k => fetch(k)) // atomic compute — fetch only if missing

// ✅ CORRECT: for Scala-native concurrent collections
val concurrentMap = scala.collection.concurrent.TrieMap.empty[String, Int]
concurrentMap("key") = 42                  // lock-free, thread-safe

import scala.util.{Try, Success, Failure}

// Try is the synchronous analog of Future — wraps a computation that may throw.
// Same API as Future: map, flatMap, filter, recover, fold

val result: Try[Int] = Try {
  "42".toInt                              // Success(42)
}
val failed: Try[Int] = Try {
  "abc".toInt                             // Failure(NumberFormatException)
}

// Chaining with for-comprehension — same as Future/Option
val pipeline: Try[Int] =
  for
    n <- Try("100".toInt)
    d <- Try("4".toInt)
    if d != 0                              // guard → throws if false? No, returns Failure
  yield n / d

// fold — handle both cases in one call
val message: String = pipeline.fold(
  ex => s"Error: ${ex.getMessage}",
  value => s"Result: $value"
)

// toOption / toEither — convert to other error types
val asOption: Option[Int] = result.toOption
val asEither: Either[Throwable, Int] = result.toEither

// ❌ ANTI-PATTERN: using Try for expected control flow
// Try is for wrapping truly exceptional code (JDBC, parsing). For expected errors,
// use Either from the start — it's explicit, not exception-based.

def findUser(id: Long): Future[Option[User]] =
  if id < 0 then Future.successful(None)    // no async needed — immediate completion
  else db.find(id)                          // actual async DB call

implicit val ec: ExecutionContext = ExecutionContext.global

val f1 = Future { Thread.sleep(10000); "done" }   // blocks for 10s
val f2 = Future { 42 }                             // should be instant

f2.map(_ * 2).foreach(println)                     // when does this print?