Effect TS

// TypeScript: async generator
async function* doubles(arr: number[]) {
  for (const n of arr) {
    yield n * 2
  }
}
// Consuming:
for await (const n of doubles([1, 2, 3])) {
  if (n > 4) console.log(n)
}
// Problems:
// - No typed errors (throw anything, catch... string?)
// - No dependency injection
// - Composing generators is awkward (yield* in generators is clunky)
// - No built-in concurrency, buffering, or grouping
// - Can't easily merge two generators or zip them

import { Stream, Effect } from "effect"

// ── Creating streams ──
const fromValues = Stream.make(1, 2, 3)
const fromArray  = Stream.fromIterable([1, 2, 3])
const fromEffect = Stream.fromEffect(Effect.succeed(42))
const naturals   = Stream.unfold(1, (n) => Option.some([n, n + 1]))
//                                         [emit, nextSeed]

// ── Transforming ──
const pipeline = Stream.fromIterable([1, 2, 3, 4, 5]).pipe(
  Stream.map((n) => n * 2),           // [2, 4, 6, 8, 10]
  Stream.filter((n) => n > 4),        // [6, 8, 10]
  Stream.tap((n) => Effect.log(`Processing: ${n}`))
)

// ── flatMap: each element → sub-stream ──
const expanded = Stream.make(1, 2, 3).pipe(
  Stream.flatMap((n) => Stream.make(n, n * 10))
)
// [1, 10, 2, 20, 3, 30]

// ── scan: running totals ──
const runningSum = Stream.fromIterable([1, 2, 3, 4]).pipe(
  Stream.scan(0, (acc, n) => acc + n)
)
// [0, 1, 3, 6, 10]

// ── Combining streams ──
const a = Stream.make(1, 2)
const b = Stream.make(3, 4)
const sequential  = Stream.concat(a, b)       // [1, 2, 3, 4]
const interleaved = Stream.merge(a, b)         // async interleaving
const paired      = Stream.zip(a, b)           // [[1,3], [2,4]]

// TypeScript: consuming an async iterable
async function collectAll<T>(gen: AsyncIterable<T>): Promise<T[]> {
  const result: T[] = []
  for await (const item of gen) {
    result.push(item) // all in memory
  }
  return result
}

async function reduce<T, U>(
  gen: AsyncIterable<T>,
  seed: U,
  f: (acc: U, item: T) => U
): Promise<U> {
  let acc = seed
  for await (const item of gen) {
    acc = f(acc, item)
  }
  return acc
}
// Problems:
// - Must write these helpers yourself
// - No built-in batching/grouping
// - No typed errors in the pipeline
// - No "take first N" without manual break logic

import { Stream, Sink, Effect } from "effect"

const numbers = Stream.range(1, 100)

// ── Simple runners ──
const all   = Stream.runCollect(numbers)        // Chunk of 1..100
const sum   = Stream.runFold(numbers, 0, (a, b) => a + b)  // 5050
const log   = Stream.runForEach(numbers, (n) => Effect.log(`Got: ${n}`))

// ── Custom Sinks ──
const first = Stream.run(numbers, Sink.head())
// Effect<Option<number>, never, never>

// Collect first 5 elements:
const firstFive = Stream.run(numbers, Sink.take(5))
// Effect<Chunk<number>, never, never>

// Fold until sum exceeds 50:
const partial = Stream.run(
  numbers,
  Sink.fold(0, (sum) => sum <= 50, (sum, n) => sum + n)
)
// Stops consuming when sum > 50, leftovers stay unconsumed

// ── Batching with grouped ──
const batched = Stream.range(1, 20).pipe(
  Stream.grouped(5),                 // Chunk of 5 elements each
  Stream.runForEach((batch) =>
    Effect.log(`Processing batch of ${batch.length}`)
  )
)

// ── groupedWithin: size OR time ──
const adaptive = someEventStream.pipe(
  Stream.groupedWithin(100, "5 seconds"),
  // batch up to 100 items, OR flush every 5 seconds
  Stream.runForEach((batch) => writeToDB(batch))
)

// TypeScript: concurrent processing with async iterables
async function* processAll(urls: string[]) {
  // Sequential — no built-in concurrency for async generators
  for (const url of urls) {
    try {
      const res = await fetch(url)
      yield await res.json()
    } catch (e) {
      // Error kills the whole generator? Swallow it? Re-throw?
      console.error(e) // usually swallow and continue... fragile
    }
  }
}
// Problems:
// - No parallel processing without Promise.all (loses streaming)
// - Error handling is ad-hoc — no typed recovery
// - No cancellation/switchMap pattern
// - No guaranteed cleanup (finalizers)
// - No built-in retry with backoff

import { Stream, Effect, Schedule } from "effect"

// ── Concurrent flatMap ──
const fetchUser = (id: number) =>
  Stream.fromEffect(Effect.tryPromise(() =>
    fetch(`/api/users/${id}`).then((r) => r.json())
  ))

const users = Stream.fromIterable([1, 2, 3, 4, 5]).pipe(
  Stream.flatMap(fetchUser, { concurrency: 3 })
  // 3 fetches in parallel at a time
)

// ── switchMap: cancel previous on new ──
const searchResults = searchTermStream.pipe(
  Stream.flatMap(
    (term) => Stream.fromEffect(searchAPI(term)),
    { switch: true }
    // new search term cancels the in-flight request
  )
)

// ── Error recovery ──
const resilient = Stream.fromEffect(
  Effect.tryPromise(() => fetch("/api/data"))
).pipe(
  Stream.catchAll((error) =>
    Stream.make({ fallback: true, error: String(error) })
  )
)

// ── Retry with backoff ──
const withRetry = Stream.fromEffect(
  Effect.tryPromise(() => fetch("/api/flaky"))
).pipe(
  Stream.retry(Schedule.exponential("1 second").pipe(
    Schedule.compose(Schedule.recurs(3))
  ))
)

// ── Resource management ──
const fileLines = Stream.scoped(
  Effect.acquireRelease(
    Effect.sync(() => openFile("data.csv")),
    (handle) => Effect.sync(() => handle.close())
  )
).pipe(
  Stream.flatMap((handle) => Stream.fromIterable(handle.readLines()))
)

// ── Finalizer: guaranteed cleanup ──
const withCleanup = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.finalizer(
    Effect.log("Stream finished — cleaning up")
  ))
)