Effect TS

// TypeScript: manual concurrency with Promise.all
async function fetchAllUsers(ids: string[]) {
  // No concurrency limit — fires everything at once
  const users = await Promise.all(ids.map(id => fetchUser(id)))
  return users
}

// Want a concurrency limit? DIY with chunking or p-limit
import pLimit from "p-limit"
const limit = pLimit(5)
const users = await Promise.all(
  ids.map(id => limit(() => fetchUser(id)))
)

// Cancellation? AbortController per request, wire it yourself
const controller = new AbortController()
const result = await fetch(url, { signal: controller.signal })
// To cancel: controller.abort() — but who cleans up?

import { Effect, Fiber } from "effect"

const program = Effect.gen(function* () {
  // Fork a background task — returns immediately
  const fiber = yield* Effect.fork(longRunningJob)

  // Do other work concurrently...
  yield* processItems()

  // Wait for the background task
  const result = yield* Fiber.join(fiber)

  // ── High-level concurrency (preferred) ──

  // Run 10 tasks with concurrency limit (like p-limit but built-in)
  yield* Effect.forEach(urls, fetchUrl, { concurrency: 10 })

  // Run all effects concurrently, collect results
  const [a, b, c] = yield* Effect.all(
    [taskA, taskB, taskC],
    { concurrency: "unbounded" }
  )

  // Race — first to finish wins, others are interrupted
  const fastest = yield* Effect.race(fetchFromCDN, fetchFromOrigin)
})

// TypeScript: manual coordination patterns

// Channel-like behavior? No built-in. Use arrays + polling
const queue: string[] = []
const enqueue = (item: string) => queue.push(item)
const dequeue = () => queue.shift() // undefined if empty — no waiting

// One-shot signal? Wrap a Promise with external resolve
let resolve: (value: string) => void
const deferred = new Promise<string>(r => { resolve = r })
resolve!("done") // easy to misuse, no type safety on completion

// Concurrency limit? p-limit or hand-rolled semaphore
import pLimit from "p-limit"
const limit = pLimit(3)
await Promise.all(tasks.map(t => limit(() => t())))

// Pub/sub? EventEmitter — no backpressure, no type safety
import { EventEmitter } from "events"
const emitter = new EventEmitter()
emitter.on("data", (msg) => { /* hope msg is the right type */ })
emitter.emit("data", { anything: "goes" })

import { Effect, Queue, Deferred, Fiber, PubSub } from "effect"

// ── Queue: producer/consumer channel ──
const producerConsumer = Effect.gen(function* () {
  const queue = yield* Queue.bounded<string>(10)

  // Producer fiber — pushes items
  const producer = yield* Effect.fork(
    Effect.forEach(
      ["a", "b", "c"],
      (item) => Queue.offer(queue, item),
      { concurrency: 1 }
    )
  )

  // Consumer — takes items one by one
  const first = yield* Queue.take(queue)  // suspends until available
  const batch = yield* Queue.takeUpTo(queue, 10)  // grab remaining

  yield* Fiber.join(producer)
  yield* Queue.shutdown(queue)
})

// ── Deferred: one-shot signal between fibers ──
const signaling = Effect.gen(function* () {
  const deferred = yield* Deferred.make<string, never>()

  // Worker waits for the signal
  const worker = yield* Effect.fork(
    Effect.gen(function* () {
      const value = yield* Deferred.await(deferred)
      yield* Effect.log(`Got: ${value}`)
    })
  )

  // Main fiber completes the deferred after some work
  yield* Effect.sleep("1 second")
  yield* Deferred.succeed(deferred, "go!")
  yield* Fiber.join(worker)
})

// ── Semaphore: limit concurrent access ──
const rateLimited = Effect.gen(function* () {
  const semaphore = yield* Effect.makeSemaphore(3)

  // Only 3 API calls run at a time
  yield* Effect.forEach(
    urls,
    (url) => semaphore.withPermits(1)(fetchUrl(url)),
    { concurrency: "unbounded" }
  )
})