package me.betanow
package pipeline

import org.scalatest.funsuite.AnyFunSuite
import play.api.libs.json.Json
import source.SourceFactory
import sink.SinkFactory
import transformation.{FieldsTransformation, FilterTransformation, TransformTransformation}

import java.nio.file.{Files, Path}

class PipelineIntegrationTest extends AnyFunSuite {
  private def withFiles(check: (Path, Path) => Unit): Unit = {
    val directory = Files.createTempDirectory("scala-pipeline-test-")
    val input = directory.resolve("input.csv")
    val output = directory.resolve("output.json")
    try {
      Files.writeString(input,
        "name,age,city,email\n" +
          " Alice ,30, BORDEAUX ,alice@exemple.com\n" +
          " Bob ,34, PARIS ,bob@exemple.com\n" +
          " Charlie ,16, LYON ,charlie@exemple.com\n")
      check(input, output)
    } finally {
      Files.deleteIfExists(output)
      Files.deleteIfExists(input)
      Files.deleteIfExists(directory)
    }
  }

  test("CSV to JSON pipeline filters, transforms and writes the expected records") {
    withFiles { (input, output) =>
      val pipeline = for {
        data <- new SourceFactory()(input.toString)
        adults <- new FilterTransformation()
          .filterWhere(row => row("age").asInstanceOf[Int] >= 18)
          .transform(data)
        cleaned <- new TransformTransformation()
          .trimFields("name", "city")
          .toUpperCase("name")
          .toLowerCase("city")
          .mapField("email", _.replace("exemple.com", "example.com"))
          .transform(adults)
        selected <- new FieldsTransformation()
          .renameFields("name", "fullName")
          .selectFields("fullName", "age", "city", "email")
          .transform(cleaned)
        written <- new SinkFactory()(output.toString, selected)
      } yield written

      pipeline.fold(error => fail("Pipeline failed", error), identity)
      assert(Files.exists(output))
      val actual = Json.parse(Files.readString(output))
      val expected = Json.parse("""[
        {"fullName":"ALICE","age":30,"city":"bordeaux","email":"alice@example.com"},
        {"fullName":"BOB","age":34,"city":"paris","email":"bob@example.com"}
      ]""")
      assert(actual === expected)
    }
  }

  test("A failed transformation stops the pipeline before any output is written") {
    withFiles { (input, output) =>
      val pipeline = for {
        data <- new SourceFactory()(input.toString)
        transformed <- new TransformTransformation()
          .mapField("email", _ => throw new IllegalArgumentException("Invalid email"))
          .transform(data)
        written <- new SinkFactory()(output.toString, transformed)
      } yield written

      pipeline match {
        case Left(error: IllegalArgumentException) => assert(error.getMessage === "Invalid email")
        case other => fail(s"Expected the transformation error, got $other")
      }
      assert(!Files.exists(output))
    }
  }
}