import { Channel, Data, Effect, PubSub } from "effect"

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

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<number>(16)

  // Create a channel that reads arrays of values
  const channel = Channel.fromPubSubArray(pubsub)

  // Publish some values
  yield* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)
  yield* PubSub.publish(pubsub, 3)
  yield* PubSub.publish(pubsub, 4)

  // The channel will output arrays like [1, 2, 3] and [4]
  return channel
})

import { Channel, Effect, PubSub } from "effect"

interface Order {
  readonly id: string
  readonly customerId: string
  readonly items: ReadonlyArray<string>
  readonly total: number
  readonly submittedAt: number
}

const orderBatchProcessor = Effect.gen(function*() {
  const orderPubSub = yield* PubSub.bounded<Order>(100)

  // Create a channel that processes orders in batches
  const orderChannel = Channel.fromPubSubArray(orderPubSub)

  // Transform to process each batch of orders
  const processedChannel = Channel.map(orderChannel, (orderBatch) => {
    const totalRevenue = orderBatch.reduce((sum, order) => sum + order.total, 0)
    const customerCount = new Set(orderBatch.map((order) =>
      order.customerId
    )).size

    return {
      batchSize: orderBatch.length,
      totalRevenue,
      uniqueCustomers: customerCount,
      firstSubmittedAt: Math.min(...orderBatch.map((order) => order.submittedAt)),
      orders: orderBatch
    }
  })

  return processedChannel
})

import { Channel, Effect, PubSub } from "effect"

interface LogEntry {
  readonly timestamp: number
  readonly level: "info" | "warn" | "error"
  readonly message: string
  readonly source: string
}

const logAggregator = Effect.gen(function*() {
  const logPubSub = yield* PubSub.bounded<LogEntry>(500)

  // Create a channel that collects logs in batches
  const logChannel = Channel.fromPubSubArray(logPubSub)

  // Transform to analyze log batches
  const analysisChannel = Channel.map(logChannel, (logBatch) => {
    const errorCount = logBatch.filter((log) => log.level === "error").length
    const warnCount = logBatch.filter((log) => log.level === "warn").length
    const infoCount = logBatch.filter((log) => log.level === "info").length

    const timeRange = {
      start: Math.min(...logBatch.map((log) => log.timestamp)),
      end: Math.max(...logBatch.map((log) => log.timestamp))
    }

    return {
      batchId: `${timeRange.start}-${timeRange.end}`,
      totalEntries: logBatch.length,
      errorCount,
      warnCount,
      infoCount,
      timeRange,
      sources: [...new Set(logBatch.map((log) => log.source))]
    }
  })

  return analysisChannel
})