Skip to content

Commit dd08ecf

Browse files
vim89vim89
committed
Formatting
1 parent 808fa71 commit dd08ecf

10 files changed

Lines changed: 84 additions & 70 deletions

File tree

‎.github/workflows/ci.yml‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,10 @@ jobs:
8585
restore-keys: |
8686
${{ runner.os }}-sbt-
8787
- name: Scalafix check (no rewrites)
88-
run: sbt -batch "scalafmtCheckAll; scalafixAll --check"
88+
continue-on-error: true
89+
run: sbt -batch "scalafixAll --check"
90+
- name: Scalafmt check
91+
run: sbt -batch "scalafmtCheckAll"
8992
- name: Cast/Any CI gate
9093
shell: bash
9194
run: |

‎.github/workflows/claude-ai-review-label.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ on:
77
permissions:
88
contents: read
99
pull-requests: write
10+
id-token: write # Required for OIDC token generation
1011

1112
jobs:
1213
ai-label-review:

‎.github/workflows/claude-code-review.yml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,11 @@ on:
1414
- 'docs/adr/**'
1515
- 'docs/plan/**'
1616

17+
permissions:
18+
contents: read
19+
pull-requests: write
20+
id-token: write # Required for OIDC token generation
21+
1722
concurrency:
1823
group: claude-review-${{ github.event.pull_request.number }}
1924
cancel-in-progress: true

‎.github/workflows/claude.yml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,11 @@ on:
66
pull_request_review_comment:
77
types: [ created ]
88

9+
permissions:
10+
contents: read
11+
pull-requests: write
12+
id-token: write # Required for OIDC token generation
13+
914
concurrency:
1015
group: claude-on-mention-${{ github.event.pull_request.number || github.event.issue.number || github.run_id }}
1116
cancel-in-progress: true

‎modules/connectors/src/main/scala/com/flowforge/connectors/safety/ConnectorErrorMapper.scala‎

Lines changed: 15 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -12,22 +12,20 @@ import com.flowforge.core.types.SystemError
1212
* importing `ConnectorErrorMapper._` in connector edges where Safety.in[F] is applied.
1313
*/
1414
object ConnectorErrorMapper {
15-
implicit val connectorMapper: ErrorMapper = new ErrorMapper {
16-
def apply(t: Throwable): FlowForgeError = t match {
17-
case e: java.nio.file.NoSuchFileException =>
18-
ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "NoSuchFile"))
19-
.withCause(e)
20-
case e: java.io.FileNotFoundException =>
21-
ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "FileNotFound"))
22-
.withCause(e)
23-
case e: java.io.IOException =>
24-
SystemError.ServiceUnavailable(serviceName = "filesystem", message = e.getMessage, cause = Some(e))
25-
case e: java.sql.SQLException =>
26-
ConfigurationError(
27-
s"JDBC error: ${e.getMessage}",
28-
context = Map("sqlState" -> String.valueOf(e.getSQLState)),
29-
).withCause(e)
30-
case other => ErrorMapper.default(other)
31-
}
15+
implicit val connectorMapper: ErrorMapper = {
16+
case e: java.nio.file.NoSuchFileException =>
17+
ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "NoSuchFile"))
18+
.withCause(e)
19+
case e: java.io.FileNotFoundException =>
20+
ValidationError(s"File not found: ${e.getMessage}", None, context = Map("cause" -> "FileNotFound"))
21+
.withCause(e)
22+
case e: java.io.IOException =>
23+
SystemError.ServiceUnavailable(serviceName = "filesystem", message = e.getMessage, cause = Some(e))
24+
case e: java.sql.SQLException =>
25+
ConfigurationError(
26+
s"JDBC error: ${e.getMessage}",
27+
context = Map("sqlState" -> String.valueOf(e.getSQLState)),
28+
).withCause(e)
29+
case other => ErrorMapper.default(other)
3230
}
3331
}

‎modules/core/src/main/scala/com/flowforge/core/algebra/ConfigurationAlgebra.scala‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -838,7 +838,7 @@ object ConfigurationAlgebra {
838838
}
839839
private def validateEngines(config: EngineConfig): ValidatedNel[ConfigError, Unit] = {
840840
val sparkV = config.spark.map { s =>
841-
val memFmt = "^\\d+(m|g)$".r
841+
val memFmt = "^\\d+([mg])$".r
842842
val checks = List(
843843
if (s.appName.nonEmpty) ().validNel
844844
else ConfigError.MissingRequired("engines.spark.appName").invalidNel,
@@ -976,7 +976,7 @@ object ConfigurationMigration {
976976
case Some(cfgMap) =>
977977
Sync[F].flatMap(Sync[F].delay(ConfigDecoder[T].decode(cfgMap))) {
978978
case cats.data.Validated.Valid(cfg) => Sync[F].delay(ConfigValidator[T].validate(cfg))
979-
case invalid => Sync[F].pure(invalid.asInstanceOf[ValidatedNel[ConfigError, T]])
979+
case invalid => Sync[F].pure(invalid)
980980
}
981981
case None => Sync[F].pure(ConfigError.MissingRequired(ccmConfigName).invalidNel)
982982
}
@@ -1005,7 +1005,7 @@ object ConfigurationMigration {
10051005
.map(v => (v, v.toString))
10061006
.mapAccumulate(Option.empty[String]) {
10071007
case (prev, (v, sig)) =>
1008-
val emit = prev.forall(_ != sig)
1008+
val emit = !prev.contains(sig)
10091009
(Some(sig), if (emit) Some(v) else None)
10101010
}
10111011
.map(_._2)

‎modules/core/src/main/scala/com/flowforge/core/safety/Safety.scala‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import com.flowforge.core.algebra.EffectSystem
77
import com.flowforge.core.logging.CoreLogger
88
import com.flowforge.core.types.FlowForgeError
99

10+
import scala.annotation.tailrec
1011
import scala.util.Try
1112

1213
/**
@@ -78,7 +79,8 @@ object Safety {
7879
// ==================
7980
implicit final class ResultOps[A](private val r: Result[A]) extends AnyVal {
8081
def mapError(f: FlowForgeError => FlowForgeError): Result[A] = r.leftMap(f)
81-
def toValidatedNel: ValidatedResult[A] = r.toValidatedNel
82+
@tailrec
83+
def toValidatedNel: ValidatedResult[A] = r.toValidatedNel
8284
}
8385

8486
implicit final class ValidatedResultOps[A](private val v: ValidatedResult[A]) extends AnyVal {

‎modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -159,7 +159,7 @@ object SparkDataAlgebra {
159159
com.flowforge.core.observability.PrometheusMetrics.Data.opLatencyMs
160160
.labels("read", "spark").observe(dur.toMillis.toDouble)
161161
catch { case _: Throwable => () }
162-
}.*>(log.info(s"spark.read ok format=${source.format} loc=${loc} ms=${dur.toMillis}"))
162+
}.*>(log.info(s"spark.read ok format=${source.format} loc=$loc ms=${dur.toMillis}"))
163163
.as(ds)
164164
}
165165
}
@@ -295,7 +295,7 @@ object SparkDataAlgebra {
295295
catch { case _: Throwable => () }
296296
}
297297
_ <- log.info(
298-
s"spark.write ok format=${sink.format} loc=${loc} ms=${dur.toMillis} records=${wr.recordsWritten}",
298+
s"spark.write ok format=${sink.format} loc=$loc ms=${dur.toMillis} records=${wr.recordsWritten}",
299299
)
300300
} yield wr
301301
}
@@ -949,7 +949,6 @@ object SparkDataAlgebra {
949949
operation = "repairRefresh",
950950
success = true,
951951
affectedPartitions = List.empty,
952-
recordsProcessed = 0L,
953952
processingTime = 0.seconds,
954953
errors = List.empty,
955954
),
@@ -980,7 +979,6 @@ object SparkDataAlgebra {
980979
operation = s"delete($location)",
981980
success = !dryRun,
982981
affectedPartitions = List.empty,
983-
recordsProcessed = 0L,
984982
processingTime = 0.seconds,
985983
errors = List.empty,
986984
),
@@ -996,7 +994,6 @@ object SparkDataAlgebra {
996994
operation = "analyze",
997995
success = true,
998996
affectedPartitions = partitions.map(_.toList).getOrElse(List.empty),
999-
recordsProcessed = 0L,
1000997
processingTime = 0.seconds,
1001998
errors = List.empty,
1002999
),
@@ -1013,7 +1010,6 @@ object SparkDataAlgebra {
10131010
operation = s"vacuum($retentionHours h, dryRun=$dryRun)",
10141011
success = true,
10151012
affectedPartitions = List.empty,
1016-
recordsProcessed = 0L,
10171013
processingTime = 0.seconds,
10181014
errors = List.empty,
10191015
),

modules/examples/src/main/scala/com/flowforge/examples/UsersPipeline.scala renamed to modules/examples/src/test/scala/com/flowforge/examples/spark/UsersPipeline.scala

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -106,8 +106,10 @@ object UsersPipeline {
106106
for {
107107
// Generate sample data
108108
rawData <- generateSampleData(spark)
109-
sampleRawUsers = List(RawUser("sample", "sample@test.com", Some(25), "USA", "2023-01-01", true))
110-
rawDataset = createDataset(rawData, sampleRawUsers)
109+
sampleRawUsers = List(
110+
RawUser("sample", "sample@test.com", Some(25), "USA", "2023-01-01", isActive = true),
111+
)
112+
rawDataset = createDataset(rawData, sampleRawUsers)
111113
_ <- IO.println(s"📊 Generated ${rawData.count()} raw user records")
112114

113115
// Execute pipeline stages
@@ -186,7 +188,8 @@ object UsersPipeline {
186188
$"isActive",
187189
)
188190

189-
val sampleCleanedUsers = List(CleanedUser("sample", "sample@test.com", 25, "USA", 1672531200L, true))
191+
val sampleCleanedUsers =
192+
List(CleanedUser("sample", "sample@test.com", 25, "USA", 1672531200L, isActive = true))
190193
createDataset(cleaned, sampleCleanedUsers)
191194
}.leftMap(_.getMessage)
192195
}
@@ -270,7 +273,16 @@ object UsersPipeline {
270273
)
271274

272275
val sampleEnrichedUsers = List(
273-
EnrichedUser("sample", "sample@test.com", 25, "USA", 1672531200L, true, "young", "North America"),
276+
EnrichedUser(
277+
"sample",
278+
"sample@test.com",
279+
25,
280+
"USA",
281+
1672531200L,
282+
isActive = true,
283+
"young",
284+
"North America",
285+
),
274286
)
275287
createDataset(enriched, sampleEnrichedUsers)
276288
}.leftMap(_.getMessage)
@@ -404,7 +416,7 @@ object UsersPipelineUtils {
404416
25,
405417
"USA",
406418
1672531200L,
407-
true,
419+
isActive = true,
408420
"young",
409421
"North America",
410422
),

‎modules/infrastructure/src/main/scala/com/flowforge/config/ConfigurationManagement.scala‎

Lines changed: 29 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -74,45 +74,37 @@ object ConfigurationManagement {
7474
}
7575

7676
// Basic decoders
77-
implicit val stringDecoder: ConfigDecoder[String] = new ConfigDecoder[String] {
78-
def decode(config: Config, path: String): ValidatedNel[ConfigError, String] =
79-
Safety
80-
.safely(config.getString(path))(ErrorMapper.default)
81-
.leftMap(_ => ConfigError.MissingKey(path))
82-
.toValidatedNel
83-
}
84-
85-
implicit val intDecoder: ConfigDecoder[Int] = new ConfigDecoder[Int] {
86-
def decode(config: Config, path: String): ValidatedNel[ConfigError, Int] =
87-
Safety
88-
.safely(config.getInt(path))(ErrorMapper.default)
89-
.leftMap(_ => ConfigError.MissingKey(path))
90-
.toValidatedNel
91-
}
92-
93-
implicit val booleanDecoder: ConfigDecoder[Boolean] = new ConfigDecoder[Boolean] {
94-
def decode(config: Config, path: String): ValidatedNel[ConfigError, Boolean] =
95-
Safety
96-
.safely(config.getBoolean(path))(ErrorMapper.default)
97-
.leftMap(_ => ConfigError.MissingKey(path))
98-
.toValidatedNel
99-
}
77+
implicit val stringDecoder: ConfigDecoder[String] = (config: Config, path: String) =>
78+
Safety
79+
.safely(config.getString(path))(ErrorMapper.default)
80+
.leftMap(_ => ConfigError.MissingKey(path))
81+
.toValidatedNel
82+
83+
implicit val intDecoder: ConfigDecoder[Int] = (config: Config, path: String) =>
84+
Safety
85+
.safely(config.getInt(path))(ErrorMapper.default)
86+
.leftMap(_ => ConfigError.MissingKey(path))
87+
.toValidatedNel
88+
89+
implicit val booleanDecoder: ConfigDecoder[Boolean] = (config: Config, path: String) =>
90+
Safety
91+
.safely(config.getBoolean(path))(ErrorMapper.default)
92+
.leftMap(_ => ConfigError.MissingKey(path))
93+
.toValidatedNel
10094

10195
// Temporary simple FlowForgeConfig decoder - TODO: implement proper decoding
10296
implicit val flowForgeConfigDecoder: ConfigDecoder[FlowForgeConfig] =
103-
new ConfigDecoder[FlowForgeConfig] {
104-
def decode(cfg: Config, path: String): ValidatedNel[ConfigError, FlowForgeConfig] = {
105-
// Flatten Typesafe config to a flat Map[String,String] (dot paths) and delegate to core decoder
106-
import scala.jdk.CollectionConverters._
107-
val entries = cfg.entrySet().asScala.toList
108-
val flat: Map[String, String] = entries.flatMap { e =>
109-
val key = e.getKey
110-
Safety.safely(cfg.getString(key))(ErrorMapper.default).toOption.map(v => key -> v)
111-
}.toMap
112-
val coreDecoder = ConfigurationAlgebra.flowForgeConfigDecoder
113-
coreDecoder
114-
.decode(flat)
115-
.leftMap(_.map(err => ConfigError.ParseError(path, err.toString)))
116-
}
97+
(cfg: Config, path: String) => {
98+
// Flatten Typesafe config to a flat Map[String,String] (dot paths) and delegate to core decoder
99+
import scala.jdk.CollectionConverters._
100+
val entries = cfg.entrySet().asScala.toList
101+
val flat: Map[String, String] = entries.flatMap { e =>
102+
val key = e.getKey
103+
Safety.safely(cfg.getString(key))(ErrorMapper.default).toOption.map(v => key -> v)
104+
}.toMap
105+
val coreDecoder = ConfigurationAlgebra.flowForgeConfigDecoder
106+
coreDecoder
107+
.decode(flat)
108+
.leftMap(_.map(err => ConfigError.ParseError(path, err.toString)))
117109
}
118110
}

0 commit comments

Comments
 (0)