import { Int, Item, InputFailure, TransformFailure, OutputFailure, Input, Transform, Output } from "./contracts.eff"

const convert = (item: Item) =>
  try { { id: item.id, value: Transform.apply(item), recovered: false } } catch {
    TransformFailure(error) =>
      if (error.kind === "Recoverable") { { id: item.id, value: item.value, recovered: true } }
      else { fail(error) }
  }

const process = (input: Int, output: Int) => {
  const incoming = bufferStream(unfold(0, index => Input.next(input, index)), { capacity: 2 })
  const converted = mapEffect(incoming, item => convert(item), { concurrency: 2 })
  const batches = transduceStream(converted, takeSink(2))
  runStream(batches, sinkForEach(batch =>
    if ((batch |> fold(0, (count, _item) => count + 1)) > 0) { Output.write(output, batch) } else {}))
}

const withOutput = (input: Int) =>
  using (consumer = {
      const output = Output.open()
      {
        run: () => process(input, output),
        close: () => Output.close(output)
      }
    }) {
      consumer.run()
    } finally {
      consumer.close()
    }

export const runProcessor = (): void
  fails InputFailure | TransformFailure | OutputFailure
  uses Input.open | Input.next | Input.close | Transform.apply | Output.open | Output.write | Output.close =>
  using (source = {
      const input = Input.open()
      {
        run: () => withOutput(input),
        close: () => Input.close(input)
      }
    }) {
      source.run()
    } finally {
      source.close()
    }