Advanced

import { Effect, Fiber, Exit, Duration } from "effect"

const task = (name: string, ms: number, succeed: boolean) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(ms))
    if (!succeed) {
      return yield* Effect.fail(new Error(`${name} failed`))
      // => Typed failure for this task
    }
    console.log(`${name} completed`)
    return name
  })

const program = Effect.gen(function* () {
  // Fork multiple fibers — they run concurrently immediately
  const fiber1 = yield* Effect.fork(task("task-A", 50, true))
  // => fiber1: Fiber<string, Error> — succeeds after 50ms
  const fiber2 = yield* Effect.fork(task("task-B", 100, false))
  // => fiber2: Fiber<string, Error> — fails after 100ms
  const fiber3 = yield* Effect.fork(task("task-C", 75, true))
  // => fiber3: Fiber<string, Error> — succeeds after 75ms

  // Fiber.await: wait for fiber to complete, returns Exit
  // Unlike Fiber.join, Fiber.await never throws — always returns Exit
  const exit1 = yield* Fiber.await(fiber1)
  // => exit1 is Exit.Success { value: "task-A" }

  Exit.match(exit1, {
    onSuccess: (v) => console.log("fiber1 result:", v),
    // => Output: fiber1 result: task-A
    onFailure: (c) => console.log("fiber1 failed:", c),
  })

  // Fiber.poll: non-blocking check — returns Option<Exit>
  const fiber2Status = yield* Fiber.poll(fiber2)
  // => Option.none() if still running, Option.some(exit) if done
  console.log("fiber2 done?", fiber2Status._id !== "None" ? "yes" : "maybe not")
  // => Output: fiber2 done? maybe not (might still be running at this point)

  // Fiber.awaitAll: wait for multiple fibers simultaneously
  const exits = yield* Fiber.awaitAll([fiber2, fiber3])
  // => exits is [Exit<string, Error>, Exit<string, Error>]
  // => Waits for all fibers; returns their Exit values

  exits.forEach((exit, i) => {
    if (Exit.isSuccess(exit)) console.log(`fiber${i + 2}: success`)
    if (Exit.isFailure(exit)) console.log(`fiber${i + 2}: failure`)
    // => Output: fiber2: failure, fiber3: success
  })
})

Effect.runPromise(program)

import { Effect, FiberRef, Fiber } from "effect"

// Create a FiberRef with an initial value
// FiberRef.make creates a FiberRef scoped to the current fiber
const requestId = FiberRef.unsafeMake("initial-request-id")
// => FiberRef<string> — each fiber gets its own copy

const program = Effect.gen(function* () {
  // Set the current fiber's value
  yield* FiberRef.set(requestId, "req-001")
  // => This fiber's requestId is now "req-001"

  const parentValue = yield* FiberRef.get(requestId)
  console.log("parent:", parentValue)
  // => Output: parent: req-001

  // Fork a child fiber — it inherits the parent's current value
  const childFiber = yield* Effect.fork(
    Effect.gen(function* () {
      const inherited = yield* FiberRef.get(requestId)
      console.log("child inherited:", inherited)
      // => Output: child inherited: req-001
      // => Child gets a COPY of the parent's value

      yield* FiberRef.set(requestId, "req-001-child")
      // => Only this fiber's copy is updated

      const childValue = yield* FiberRef.get(requestId)
      console.log("child updated to:", childValue)
      // => Output: child updated to: req-001-child
    })
  )

  yield* Fiber.join(childFiber)
  // => Wait for child to finish

  // Parent's value is unchanged — child's update did not affect parent
  const parentAfter = yield* FiberRef.get(requestId)
  console.log("parent after child:", parentAfter)
  // => Output: parent after child: req-001
  // => Unchanged — child's FiberRef.set only changed the child's copy
})

Effect.runSync(program)
// => FiberRef is used internally for:
// => - Request correlation IDs (trace context propagation)
// => - Logger context (current log level, metadata)
// => - Transaction boundaries (current DB transaction)

import { Effect, Duration, Either } from "effect"

// raceAll: race multiple effects, return the first to succeed
const servers = [
  { id: "us-east", latency: 80 },
  { id: "eu-west", latency: 40 },
  { id: "ap-south", latency: 120 },
]

const fetchFromServer = (server: { id: string; latency: number }) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(server.latency))
    console.log(`${server.id} responded after ${server.latency}ms`)
    return `response from ${server.id}`
  })

// raceAll: race all servers, take the fastest
const fastest = Effect.raceAll(servers.map(fetchFromServer))
// => The fastest server (eu-west at 40ms) wins
// => Others are interrupted when eu-west responds

Effect.runPromise(fastest).then(result => {
  console.log("fastest server result:", result)
  // => Output:
  // => eu-west responded after 40ms
  // => fastest server result: response from eu-west
})

// Running all effects and collecting both successes and failures
const tasks: Effect.Effect<string, Error, never>[] = [
  Effect.succeed("task-1 ok"),
  Effect.fail(new Error("task-2 failed")),
  Effect.succeed("task-3 ok"),
  Effect.fail(new Error("task-4 failed")),
]

// Use Effect.either to collect results without failing
const allResults = Effect.all(
  tasks.map(t => t.pipe(Effect.either)),
  // => Effect.either wraps each result in Either
  { concurrency: "unbounded" }
  // => Run all concurrently
)

Effect.runPromise(allResults).then(results => {
  const successes = results.filter(Either.isRight).map(e => e.right)
  const failures = results.filter(Either.isLeft).map(e => e.left.message)
  console.log("successes:", successes)
  // => Output: successes: ['task-1 ok', 'task-3 ok']
  console.log("failures:", failures)
  // => Output: failures: ['task-2 failed', 'task-4 failed']
})

import { Effect, Semaphore, Fiber, Duration } from "effect"

// Create a semaphore with N permits
const program = Effect.gen(function* () {
  const semaphore = yield* Semaphore.make(2)
  // => Semaphore with 2 permits
  // => At most 2 fibers can hold a permit simultaneously

  let active = 0
  let maxConcurrent = 0

  const task = (id: number) =>
    semaphore.withPermits(1)(
      // => Acquire 1 permit before executing, release after
      Effect.gen(function* () {
        active++
        maxConcurrent = Math.max(maxConcurrent, active)
        // => Track peak concurrency

        console.log(`task ${id} started (active: ${active})`)
        yield* Effect.sleep(Duration.millis(50))
        // => Simulates work
        console.log(`task ${id} completed`)

        active--
      })
    )

  // Launch 5 tasks concurrently — semaphore limits to 2 at a time
  const fibers = yield* Effect.all(
    [1, 2, 3, 4, 5].map(id => Effect.fork(task(id))),
    { concurrency: "unbounded" }
  )

  yield* Fiber.awaitAll(fibers)
  // => Wait for all tasks to complete

  console.log(`max concurrent: ${maxConcurrent}`)
  // => Output: max concurrent: 2 — never exceeded the permit limit
})

Effect.runPromise(program)
// => Output (order may vary):
// => task 1 started (active: 1)
// => task 2 started (active: 2)
// => task 1 completed
// => task 3 started (active: 2)  ← task 3 starts after task 1 releases
// => task 2 completed
// => task 4 started (active: 2)
// => ...
// => max concurrent: 2

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

// Stream.unfold: generate elements by unfolding state
// unfold(initialState, step) where step returns Option<[element, nextState]>
// Option.none() terminates the stream

// Generate Fibonacci numbers
const fibonacci = Stream.unfold([0, 1] as [number, number], ([a, b]) =>
  Option.some([a, [b, a + b] as [number, number]])
  // => Emit 'a', next state is [b, a+b]
  // => This never returns none() — infinite stream
)
// => Stream<number, never, never> — infinite Fibonacci sequence

// Take the first 10 Fibonacci numbers
const first10 = fibonacci.pipe(Stream.take(10))

Effect.runPromise(Stream.runCollect(first10)).then(chunk => {
  console.log("Fibonacci:", Array.from(chunk))
  // => Output: Fibonacci: [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
})

// Practical: paginated API unfold
interface Page {
  items: string[]
  nextCursor: string | null
}

const fetchPage = (cursor: string | null): Effect.Effect<Page, Error, never> =>
  Effect.succeed({
    items: cursor === null ? ["item1", "item2"] : ["item3", "item4"],
    nextCursor: cursor === null ? "cursor-1" : null,
    // => Page 1: items 1-2 with cursor; Page 2: items 3-4, no next cursor
  })

const paginatedStream = Stream.unfoldEffect(
  null as string | null,
  // => Initial state: null cursor (first page)
  (cursor) =>
    fetchPage(cursor).pipe(
      Effect.map(page => {
        if (page.items.length === 0) return Option.none()
        // => No items: end stream
        return Option.some([
          Stream.fromIterable(page.items),
          // => Emit this page's items as a sub-stream
          page.nextCursor,
          // => Next state: the cursor for the next page
        ] as const)
      }),
    )
)
// => Stream<Stream<string>, Error, never> — stream of page streams

const allItems = paginatedStream.pipe(Stream.flatten())
// => Stream<string, Error, never> — flattened stream of all items

Effect.runPromise(Stream.runCollect(allItems)).then(chunk => {
  console.log("all items:", Array.from(chunk))
  // => Output: all items: ['item1', 'item2', 'item3', 'item4']
})

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

const numbers = Stream.range(1, 11)
// => Stream emitting 1 through 10

// Sink.collectAll: collect all elements into a Chunk
const collected = numbers.pipe(Stream.run(Sink.collectAll()))
// => Effect<Chunk<number>, never, never>

Effect.runPromise(collected).then(chunk => {
  console.log("collected:", Array.from(chunk))
  // => Output: collected: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
})

// Sink.sum: sum all numbers
const total = numbers.pipe(Stream.run(Sink.sum))
// => Effect<number, never, never>

Effect.runPromise(total).then(n => {
  console.log("sum:", n)
  // => Output: sum: 55
})

// Sink.count: count elements
const count = numbers.pipe(Stream.run(Sink.count))

Effect.runPromise(count).then(n => {
  console.log("count:", n)
  // => Output: count: 10
})

// Sink.fold: reduce stream with an initial value and function
const product = numbers.pipe(
  Stream.run(Sink.foldLeft(1, (acc, n) => acc * n))
  // => Multiplies all numbers: 1 * 1 * 2 * 3 * ... * 10 = 3628800
)

Effect.runPromise(product).then(n => {
  console.log("product:", n)
  // => Output: product: 3628800
})

// Custom sink: collect only even numbers
const evenSink: Sink.Sink<Chunk.Chunk<number>, number, never, never, never> =
  Sink.collectAllWhile(n => n % 2 === 0)
  // => Note: collectAllWhile stops on first non-matching element
  // => Better to use Stream.filter before the sink for complex filtering

const evens = Stream.make(2, 4, 3, 6, 8).pipe(
  Stream.filter(n => n % 2 === 0),
  // => Filter in the stream, not the sink
  Stream.run(Sink.collectAll()),
)

Effect.runPromise(evens).then(chunk => {
  console.log("evens:", Array.from(chunk))
  // => Output: evens: [2, 4, 6, 8]
})

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

// Stream.groupBy: partition stream elements by a key
const events = Stream.make(
  { type: "click", userId: "u1" },
  { type: "view", userId: "u2" },
  { type: "click", userId: "u1" },
  { type: "purchase", userId: "u3" },
  { type: "view", userId: "u1" },
)

// Group events by userId, then count events per user
const eventCounts = events.pipe(
  Stream.groupByKey(event => event.userId),
  // => Partitions events into sub-streams by userId
  // => u1 gets: click, click, view
  // => u2 gets: view
  // => u3 gets: purchase
  GroupBy.evaluate((userId, userEvents) =>
    userEvents.pipe(
      Stream.runCollect,
      // => Collect all events for this user
      Effect.map(events => ({ userId, count: events.length })),
      // => Count events for this user
      Stream.fromEffect,
      // => Wrap result as a single-element stream
    )
  ),
)

Effect.runPromise(Stream.runCollect(eventCounts)).then(chunk => {
  console.log("event counts:", Array.from(chunk))
  // => Output: event counts:
  // => [{ userId: 'u1', count: 3 }, { userId: 'u2', count: 1 }, { userId: 'u3', count: 1 }]
})

// Stream.chunks: access the underlying Chunk batches
const chunked = Stream.range(1, 11).pipe(
  Stream.rechunk(3),
  // => Regroup elements into chunks of size 3
  // => Chunks: [1,2,3], [4,5,6], [7,8,9], [10]
  Stream.chunks,
  // => Emit each Chunk as a single stream element
  Stream.map(chunk => ({
    batch: Array.from(chunk),
    size: chunk.length,
  })),
)

Effect.runPromise(Stream.runCollect(chunked)).then(batches => {
  Array.from(batches).forEach(b => console.log("batch:", b.batch, "size:", b.size))
  // => Output:
  // => batch: [1, 2, 3] size: 3
  // => batch: [4, 5, 6] size: 3
  // => batch: [7, 8, 9] size: 3
  // => batch: [10] size: 1
})

import { Schema, Effect, Brand } from "effect"

// Define branded types — same underlying type, different semantic
type UserId = string & Brand.Brand<"UserId">
type OrderId = string & Brand.Brand<"OrderId">

// Schema.brand creates a Schema producing branded types
const UserIdSchema = Schema.String.pipe(
  Schema.brand<"UserId">(),
  // => Schema<UserId, string, never>
  // => Decodes string -> UserId branded type
)

const OrderIdSchema = Schema.String.pipe(
  Schema.brand<"OrderId">(),
  // => Schema<OrderId, string, never>
)

// Use branded types in a struct
const OrderSchema = Schema.Struct({
  id: OrderIdSchema,
  userId: UserIdSchema,
  // => Both are strings in JSON, but distinct types in TypeScript
  total: Schema.Number,
})

type Order = typeof OrderSchema.Type
// => { id: OrderId; userId: UserId; total: number }

// Decoding produces branded types
const rawOrder = { id: "ord-1", userId: "usr-1", total: 99.99 }
Effect.runPromise(Schema.decodeUnknown(OrderSchema)(rawOrder)).then(order => {
  console.log("order.id type check:", typeof order.id)
  // => Output: order.id type check: string
  // => At runtime it's a string; at compile time it's OrderId

  // TypeScript prevents passing an OrderId where UserId is expected:
  // const wrong: UserId = order.id  // => TypeScript error: OrderId is not UserId
  // This prevents a class of bugs where IDs are swapped
})

// Validating and constructing branded values from user input
const parseUserId = Schema.decodeUnknown(UserIdSchema)
Effect.runPromise(parseUserId("usr-42")).then(id => {
  console.log("UserId:", id)
  // => Output: UserId: usr-42
  // => TypeScript type is UserId, not string
})

import { Effect, RequestResolver, Request } from "effect"

// Define a Request type — the "query" to be batched
interface GetUser extends Request.Request<{ id: string; name: string }, Error> {
  readonly _tag: "GetUser"
  readonly id: string
}

const GetUser = Request.tagged<GetUser>("GetUser")
// => Constructor for GetUser requests

// Database call counter — demonstrates that batching reduces calls
let dbCallCount = 0

// RequestResolver: handles a batch of requests in one call
const UserResolver = RequestResolver.makeBatched(
  (requests: ReadonlyArray<GetUser>) =>
    Effect.gen(function* () {
      dbCallCount++
      console.log(`DB call #${dbCallCount}: fetching ${requests.length} users`)
      // => One DB call for the entire batch of requests

      // Simulate batch database fetch
      const users = requests.map(r => ({
        id: r.id,
        name: `User-${r.id}`,
      }))

      // Resolve each request with its result
      yield* Effect.forEach(requests, (request, i) =>
        Request.succeed(request, users[i]!)
        // => Provides the result for each individual request
      )
    }),
  { concurrency: 1 },
)

// Usage: each fiber requests its own user
const getUser = (id: string) =>
  Effect.request(GetUser({ id }), UserResolver)
// => Effect<{ id: string; name: string }, Error, never>

// Multiple concurrent fibers requesting different users
const program = Effect.gen(function* () {
  // These three requests run in the same "batch window"
  const [user1, user2, user3] = yield* Effect.all(
    [getUser("1"), getUser("2"), getUser("3")],
    { concurrency: "unbounded" }
    // => All three run concurrently — Effect batches them automatically
  )

  console.log("users:", user1.name, user2.name, user3.name)
  // => Output: users: User-1 User-2 User-3
  console.log("total DB calls:", dbCallCount)
  // => Output: total DB calls: 1
  // => Three logical requests, but only ONE database call
})

Effect.runPromise(program)

import { Effect, Metric, Duration } from "effect"

// Metric.counter: tracks cumulative counts
const requestCount = Metric.counter("http.requests.total", {
  description: "Total number of HTTP requests processed",
  // => Metadata for the monitoring system
})
// => Metric<MetricState.Counter, number, never>

// Metric.histogram: tracks value distributions (latencies, sizes)
const requestLatency = Metric.histogram(
  "http.request.duration.ms",
  Metric.Boundaries.linear({ start: 0, width: 10, count: 20 }),
  // => Buckets: 0-10ms, 10-20ms, 20-30ms, ..., 190-200ms, >200ms
)

// Metric.gauge: tracks current value (queue depth, active connections)
const activeConnections = Metric.gauge("db.connections.active", {
  description: "Number of active database connections",
})

// Using metrics in Effects
const handleRequest = Effect.gen(function* () {
  const startTime = Date.now()
  // => Record request start time

  yield* Effect.sleep(Duration.millis(15))
  // => Simulates request processing (15ms)

  const duration = Date.now() - startTime
  // => Calculate actual duration

  yield* requestCount.pipe(Metric.increment)
  // => Increment the request counter by 1

  yield* requestLatency.pipe(Metric.record(duration))
  // => Record the latency in the histogram bucket for ~15ms

  yield* activeConnections.pipe(Metric.set(5))
  // => Set the active connections gauge to 5

  return "response"
})

// Metric.value: read the current metric state
const monitoringProgram = Effect.gen(function* () {
  // Process several requests
  yield* Effect.all(
    Array.from({ length: 5 }, () => handleRequest),
    { concurrency: "unbounded" }
  )

  // Read metric state
  const countState = yield* Metric.value(requestCount)
  console.log("requests processed:", countState.count)
  // => Output: requests processed: 5

  const latencyState = yield* Metric.value(requestLatency)
  console.log("latency count:", latencyState.count)
  // => Output: latency count: 5
})

Effect.runPromise(monitoringProgram)

import { Effect, Duration } from "effect"

// withSpan: wrap an Effect in a named tracing span
const parseRequest = Effect.gen(function* () {
  yield* Effect.sleep(Duration.millis(5))
  // => Simulates parsing work
  return { userId: "u1", action: "create" }
}).pipe(
  Effect.withSpan("parse-request", {
    attributes: { "http.method": "POST", "http.path": "/users" },
    // => Span attributes appear in the trace as key-value pairs
  }),
)
// => Effect<..., never, Tracer>
// => withSpan requires a Tracer service to be provided

const validateUser = (userId: string) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(10))
    // => Simulates validation work
    return { id: userId, valid: true }
  }).pipe(
    Effect.withSpan("validate-user", {
      attributes: { "user.id": userId },
      // => Span attribute for the user being validated
    }),
  )

const persistToDatabase = (data: object) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(30))
    // => Simulates database write
    return "record-created"
  }).pipe(
    Effect.withSpan("persist-to-database", {
      attributes: { "db.operation": "INSERT", "db.table": "users" },
    }),
  )

// The full operation creates a trace tree:
// [handleCreateUser] (root span)
//   [parse-request] (child span, 5ms)
//   [validate-user] (child span, 10ms)
//   [persist-to-database] (child span, 30ms)
const handleCreateUser = Effect.gen(function* () {
  const request = yield* parseRequest
  // => Creates a child span "parse-request"

  const validated = yield* validateUser(request.userId)
  // => Creates a child span "validate-user"

  const result = yield* persistToDatabase(validated)
  // => Creates a child span "persist-to-database"

  return result
}).pipe(
  Effect.withSpan("handle-create-user"),
  // => Root span for this operation — contains all child spans
)

// Provide a tracer to capture spans
// In production: use @effect/opentelemetry with OTLP export
Effect.runPromise(
  handleCreateUser.pipe(
    Effect.provide(
      (await import("@effect/opentelemetry")).NodeSdk.layer(() => ({
        resource: { serviceName: "my-service" },
      })) as any
    )
  )
).catch(() => console.log("tracing infrastructure not available in this example"))

import { Effect, ManagedRuntime, Context, Layer, Logger, LogLevel } from "effect"

// Define application services
interface AppConfig { readonly environment: string }
interface AppLogger { readonly log: (msg: string) => void }

const AppConfig = Context.GenericTag<AppConfig>("AppConfig")
const AppLogger = Context.GenericTag<AppLogger>("AppLogger")

// Production layers
const ProdConfig = Layer.succeed(AppConfig, { environment: "production" })

const ProdLogger = Layer.effect(
  AppLogger,
  Effect.gen(function* () {
    const config = yield* AppConfig
    return {
      log: (msg) => console.log(`[${config.environment}] ${msg}`),
    }
  }),
)

// The application layer — combines all services
const AppLayer = ProdLogger.pipe(Layer.provide(ProdConfig))

// ManagedRuntime: builds the layer once, reuses it for many runs
const runtime = ManagedRuntime.make(AppLayer)
// => ManagedRuntime<AppConfig | AppLogger, never>

// Use the runtime to run an Effect (e.g., in an HTTP handler)
const handleRequest = (path: string) =>
  Effect.gen(function* () {
    const logger = yield* AppLogger
    logger.log(`Handling request: ${path}`)
    // => Uses the AppLogger from the runtime's service graph

    return `response for ${path}`
  })

// Run multiple Effects using the same runtime
async function main() {
  // Run effects using the pre-built runtime
  const result1 = await runtime.runPromise(handleRequest("/users"))
  console.log("result1:", result1)
  // => Output:
  // => [production] Handling request: /users
  // => result1: response for /users

  const result2 = await runtime.runPromise(handleRequest("/orders"))
  console.log("result2:", result2)
  // => Output:
  // => [production] Handling request: /orders
  // => result2: response for /orders

  // Dispose when shutting down — runs all finalizers
  await runtime.dispose()
  // => Triggers shutdown: closes connections, flushes logs, etc.
  console.log("runtime disposed")
}

main()

import { Effect, Logger, LogLevel, Runtime } from "effect"

// The default Logger outputs human-readable text
// Replace it for structured JSON logging in production

// Logger.json: outputs each log entry as a JSON object
const jsonLogger = Logger.json
// => Outputs: {"message":"...","level":"INFO","timestamp":"...","fiberId":"..."}

// Logger.logfmt: logfmt format for structured text
const logfmtLogger = Logger.logfmt
// => Outputs: timestamp=... level=INFO message="..."

// Logger.replace: replace the default Logger with a custom one
const withJsonLogging = (program: Effect.Effect<unknown>) =>
  program.pipe(
    Effect.provide(Logger.replace(Logger.defaultLogger, jsonLogger)),
    // => Replaces the default logger with JSON logger
    // => All Effect.log* calls now output JSON
  )

// Logger.withMinimumLogLevel: filter by minimum log level
const withProductionLogLevel = (program: Effect.Effect<unknown>) =>
  program.pipe(
    Effect.provide(Logger.minimumLogLevel(LogLevel.Warning)),
    // => Only WARNING and above are emitted
    // => DEBUG and INFO are filtered out — reduces log volume
  )

// Combining: JSON logger + production log level
const withProductionLogging = (program: Effect.Effect<unknown>) =>
  program.pipe(
    Effect.provide(
      Logger.replace(Logger.defaultLogger, jsonLogger).pipe(
        Layer.provide(Logger.minimumLogLevel(LogLevel.Warning))
      ) as any
    ),
  )

// Test the different loggers
const program = Effect.gen(function* () {
  yield* Effect.logDebug("Debug: connection pool initialized")
  // => Filtered out at WARNING level
  yield* Effect.log("Info: request received")
  // => Filtered out at WARNING level
  yield* Effect.logWarning("Warning: high memory usage")
  // => Emitted: important operational signal
  yield* Effect.logError("Error: database connection failed")
  // => Emitted: critical error
})

// Run with production logging configuration
Effect.runSync(
  program.pipe(
    Effect.provide(Logger.minimumLogLevel(LogLevel.Warning))
  )
)
// => Output (WARNING and above only):
// => timestamp=... level=WARNING message="Warning: high memory usage"
// => timestamp=... level=ERROR message="Error: database connection failed"

import { Effect, Cause, Exit, Data } from "effect"

class InvalidArgument extends Data.TaggedError("InvalidArgument")<{
  readonly message: string
}> {}

// Effect.die: fail with an unexpected defect (not a typed error)
// Use when something impossible has occurred — indicates a bug
const parsePositiveNumber = (n: number): Effect.Effect<number, InvalidArgument, never> => {
  if (!Number.isFinite(n)) {
    // => Not-finite numbers indicate a bug in the caller — this is a defect
    return Effect.die(new Error(`Expected finite number, got ${n}`))
    // => die: puts error in the defect channel, not the typed error channel
    // => Unlike fail: this cannot be caught with catchTag or catchAll
  }
  if (n <= 0) {
    return Effect.fail(new InvalidArgument({ message: `${n} is not positive` }))
    // => fail: puts error in typed error channel — caller is expected to handle this
  }
  return Effect.succeed(n)
}

// Typed error vs defect: different handling
const withTypedError = parsePositiveNumber(-5)
const exit1 = Effect.runSyncExit(withTypedError)

if (Exit.isFailure(exit1)) {
  const isTypedError = Cause.isFailType(exit1.cause)
  console.log("is typed error:", isTypedError)
  // => Output: is typed error: true
  // => catchAll and catchTag CAN catch this
}

const withDefect = parsePositiveNumber(Infinity)
const exit2 = Effect.runSyncExit(withDefect)

if (Exit.isFailure(exit2)) {
  const isDefect = Cause.isDieType(exit2.cause)
  console.log("is defect:", isDefect)
  // => Output: is defect: true
  // => catchAll does NOT catch defects — they are bugs, not expected failures
}

// Effect.sandbox: expose defects as typed errors for recovery
const recovered = withDefect.pipe(
  Effect.sandbox,
  // => Converts defects into typed errors so catchAll can handle them
  Effect.catchAll(cause => {
    if (Cause.isDieType(cause)) {
      console.log("caught defect — this is a bug!")
      return Effect.succeed(-1)
      // => Recovers from defect (only do this at top-level error boundaries)
    }
    return Effect.fail(cause)
  }),
  Effect.unsandbox,
  // => Restores the original Cause structure
)

import { Effect, Fiber, Duration } from "effect"

// Without yielding: a tight loop in a fiber can starve other fibers
const heavyComputation = (n: number): Effect.Effect<number, never, never> => {
  if (n <= 0) return Effect.succeed(0)
  // => Base case: stop recursion
  return Effect.gen(function* () {
    // Simulate CPU-intensive work
    let result = 0
    for (let i = 0; i < 10000; i++) {
      result += Math.sqrt(i) * n   // => CPU work
    }

    // Effect.yieldNow: give other fibers a chance to run
    if (n % 100 === 0) {
      yield* Effect.yieldNow()
      // => Yields control to the scheduler every 100 iterations
      // => Other fibers (logging, health checks) can run here
    }

    return result + (yield* heavyComputation(n - 1))
  })
}

// Monitor fiber: should run regularly, not be starved
const monitor = Effect.gen(function* () {
  let ticks = 0
  while (ticks < 5) {
    yield* Effect.sleep(Duration.millis(1))
    // => Tries to tick every 1ms
    ticks++
    console.log(`monitor tick ${ticks}`)
  }
})

const program = Effect.gen(function* () {
  // Fork the monitor — it should run concurrently with the computation
  const monitorFiber = yield* Effect.fork(monitor)

  // Run heavy computation — yields periodically so monitor can tick
  const result = yield* heavyComputation(200)
  console.log("computation result:", result)

  yield* Fiber.join(monitorFiber)
})

Effect.runPromise(program)
// => monitor ticks appear between computation yields
// => Without yieldNow, monitor might not run until computation completes

import { Effect, Scope, Fiber, Duration, Layer, ManagedRuntime } from "effect"

// Simulate a service with in-flight work
const processRequest = (id: number) =>
  Effect.gen(function* () {
    console.log(`request ${id}: started`)
    yield* Effect.sleep(Duration.millis(100))
    // => Simulates request processing time
    console.log(`request ${id}: completed`)
    return `response-${id}`
  })

// The application: accepts requests until interrupted
const server = Effect.gen(function* () {
  // Register shutdown cleanup
  yield* Effect.addFinalizer(() =>
    Effect.sync(() => console.log("server: shutdown complete"))
    // => Runs when the server's scope is closed
  )

  console.log("server: started, accepting requests")

  // Fork in-flight requests
  const fibers = yield* Effect.all(
    [1, 2, 3].map(id => Effect.fork(processRequest(id))),
    { concurrency: "unbounded" }
  )

  // Wait for all in-flight requests to complete
  yield* Fiber.awaitAll(fibers)
  console.log("server: all requests completed")
})

// Run the server — Ctrl+C or programmatic interrupt triggers shutdown
const runtime = ManagedRuntime.make(Layer.empty)

async function runServer() {
  // Register OS signal handler
  process.once("SIGTERM", () => {
    console.log("SIGTERM received — shutting down")
    runtime.dispose()
      .then(() => process.exit(0))
    // => dispose() runs all finalizers, waits for scope to close
  })

  await runtime.runPromise(server.pipe(Effect.scoped))
  // => scoped: server's scope is tied to this run's lifetime
}

// Programmatic graceful shutdown demonstration
Effect.runPromise(
  Effect.scoped(server).pipe(
    Effect.timeout(Duration.millis(500)),
    // => Timeout for demonstration — in production, wait for real signal
    Effect.catchAll(() => Effect.succeed("timeout — shutdown triggered"))
  )
).then(result => console.log("shutdown result:", result))

import { Effect, Ref, Schedule, Duration, Data } from "effect"

class CircuitOpenError extends Data.TaggedError("CircuitOpenError")<{
  readonly service: string
}> {}

type CircuitState = "closed" | "open" | "half-open"

// Build a circuit breaker around a service call
const makeCircuitBreaker = (
  serviceName: string,
  failureThreshold: number,
  cooldown: Duration.Duration,
) =>
  Effect.gen(function* () {
    const state = yield* Ref.make<CircuitState>("closed")
    // => Circuit starts closed (requests allowed)
    const failureCount = yield* Ref.make(0)
    // => Counts consecutive failures

    const call = <A, E>(operation: Effect.Effect<A, E, never>) =>
      Effect.gen(function* () {
        const currentState = yield* Ref.get(state)
        // => Check circuit state before calling

        if (currentState === "open") {
          return yield* Effect.fail(new CircuitOpenError({ service: serviceName }))
          // => Circuit is open: reject immediately without calling service
        }

        // Execute the operation
        const result = yield* operation.pipe(
          Effect.tapError((_e) =>
            Effect.gen(function* () {
              const count = yield* Ref.updateAndGet(failureCount, n => n + 1)
              // => Increment failure counter

              if (count >= failureThreshold) {
                yield* Ref.set(state, "open")
                // => Too many failures: open the circuit
                console.log(`${serviceName}: circuit opened after ${count} failures`)

                // Schedule automatic recovery attempt
                yield* Effect.fork(
                  Effect.sleep(cooldown).pipe(
                    Effect.andThen(Ref.set(state, "half-open")),
                    // => After cooldown, allow one probe request
                  )
                )
              }
            })
          ),
          Effect.tap(() => Ref.set(failureCount, 0)),
          // => Success: reset failure counter
        )

        return result
      })

    return { call }
  })

// Usage
const program = Effect.gen(function* () {
  const cb = yield* makeCircuitBreaker("payment-service", 2, Duration.millis(50))

  // Simulate calls: first two fail, triggering the circuit breaker
  for (let i = 1; i <= 4; i++) {
    const result = yield* cb.call(
      i <= 2
        ? Effect.fail(new Error("service down"))
        : Effect.succeed(`response-${i}`)
    ).pipe(Effect.either)
    console.log(`call ${i}:`, result._tag)
    // => call 1: Left (failure)
    // => call 2: Left (failure) → circuit opens
    // => call 3: Left (CircuitOpenError) → circuit is open, rejected immediately
    // => call 4: Left (CircuitOpenError) → still open
  }
})

Effect.runPromise(program)

import { Effect, Duration } from "effect"

// A workload that creates many concurrent operations
const processItem = (id: number) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(50))
    // => Simulates work
    return `processed-${id}`
  })

// Without concurrency limit: all 20 items run simultaneously
const unconstrained = Effect.all(
  Array.from({ length: 20 }, (_, i) => processItem(i)),
  { concurrency: "unbounded" }
  // => 20 concurrent operations — may overwhelm downstream services
)

// Effect.withConcurrency: limit concurrency for a scope of effects
// All Effect.fork calls within this scope are limited to N concurrent fibers
const constrained = Effect.all(
  Array.from({ length: 20 }, (_, i) => processItem(i)),
  { concurrency: "unbounded" }
).pipe(
  Effect.withConcurrency(5),
  // => Even though we said "unbounded" above, this scope limits to 5
  // => At most 5 items process simultaneously
)

// Measure the difference
const measureTime = <A>(label: string, effect: Effect.Effect<A>) =>
  Effect.gen(function* () {
    const start = Date.now()
    const result = yield* effect
    const elapsed = Date.now() - start
    console.log(`${label}: ${elapsed}ms`)
    return result
  })

Effect.runPromise(
  Effect.all({
    unconstrained: measureTime("unconstrained", unconstrained),
    constrained: measureTime("constrained (5)", constrained),
  })
).then(() => console.log("done"))
// => Output:
// => unconstrained: ~50ms (all 20 run in parallel, limited by single batch)
// => constrained (5): ~200ms (4 batches of 5, each taking 50ms)

import { Effect, Duration } from "effect"

// Functions that wrap their work in spans
const checkInventory = (itemId: string) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(10))
    return { itemId, available: true, quantity: 50 }
  }).pipe(
    Effect.withSpan("check-inventory", {
      attributes: { "item.id": itemId },
    }),
  )

const reserveItem = (itemId: string, qty: number) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(20))
    return `reservation-${itemId}-${qty}`
  }).pipe(
    Effect.withSpan("reserve-item", {
      attributes: { "item.id": itemId, "item.quantity": qty },
    }),
  )

const chargePayment = (amount: number) =>
  Effect.gen(function* () {
    yield* Effect.sleep(Duration.millis(50))
    return `charge-${amount}`
  }).pipe(
    Effect.withSpan("charge-payment", {
      attributes: { "payment.amount": amount },
    }),
  )

// Root span: the complete checkout operation
// Creates a trace tree:
// [checkout] (root, ~80ms total)
//   [check-inventory] (child, 10ms)
//   [reserve-item] (child, 20ms)
//   [charge-payment] (child, 50ms)
const checkout = (itemId: string, qty: number, amount: number) =>
  Effect.gen(function* () {
    // Span attributes are accessible in the trace — no need to log them
    const inventory = yield* checkInventory(itemId)
    // => Creates "check-inventory" span as child of "checkout"

    if (!inventory.available) {
      return yield* Effect.fail("item not available")
    }

    const reservation = yield* reserveItem(itemId, qty)
    // => Creates "reserve-item" span as child of "checkout"

    const charge = yield* chargePayment(amount)
    // => Creates "charge-payment" span as child of "checkout"

    return { reservation, charge }
  }).pipe(
    Effect.withSpan("checkout", {
      attributes: {
        "order.item": itemId,
        "order.quantity": qty,
        "order.amount": amount,
      },
    }),
  )

// In production: spans are exported to Jaeger, Zipkin, or OTLP
// In development: spans appear in console when using a dev tracer
Effect.runPromise(checkout("item-1", 2, 49.99))
  .then(result => console.log("checkout result:", result))

import { Stream, Effect, Scope } from "effect"
import { NodeFileSystem } from "@effect/platform-node"
import { FileSystem } from "@effect/platform"

// Reading a file as a stream of lines — memory-efficient for large files
const processLargeFile = (path: string) =>
  Effect.gen(function* () {
    const fs = yield* FileSystem.FileSystem
    // => Requires FileSystem service from context

    // Open the file as a readable stream
    const stream = yield* fs.readFileString(path).pipe(
      Effect.map(content => {
        // => For demonstration: simulate line-by-line processing
        const lines = content.split("\n")
        return Stream.fromIterable(lines)
      }),
    )
    // => In production: use fs.readFile and rechunk for true streaming

    return stream
  })

// More realistic: Stream that simulates line-by-line file processing
const simulatedFileStream = (lineCount: number): Stream.Stream<string, never, never> =>
  Stream.unfold(0, (lineNum) => {
    if (lineNum >= lineCount) return undefined as any
    // => End of file
    return [
      `line ${lineNum + 1}: data-${Math.random().toFixed(4)}`,
      // => Simulated file line content
      lineNum + 1,
      // => Next state: incremented line number
    ] as const
  }).pipe(
    Stream.mapBefore(v => v)
    // => Stream.unfold signature note: returns Option<[element, state]>
  )

// Process file in batches for efficient bulk operations
const batchProcessFile = Stream.unfold(0 as number, (n) =>
  n >= 100 ? undefined as any : [`line-${n}`, n + 1] as const
).pipe(
  Stream.rechunk(10),
  // => Group into batches of 10 lines
  Stream.mapChunksEffect(chunk =>
    Effect.gen(function* () {
      const lines = Array.from(chunk)
      console.log(`processing batch of ${lines.length} lines`)
      // => Process each batch: parse, validate, write to DB
      yield* Effect.sleep(0)
      // => Yields to scheduler — allows other fibers to run between batches
      return chunk
    })
  ),
  Stream.runCollect,
  // => Collect all results — in production, use a Sink that writes to a DB
)

Effect.runPromise(batchProcessFile).then(all => {
  console.log(`processed ${all.length} total lines`)
  // => Output: processed 100 total lines
})