Migrate from SQLite to PostgreSQL storage

Remove SQLite storage implementation and replace with PostgreSQL-only solution:
- Add PostgreSQL storage with JSONB support via doobie-postgres-circe
- Remove all SQLite dependencies and code
- Add storage configuration file approach with --storage CLI parameter
- Update CI/CD with PostgreSQL service container

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: stanch

.github/workflows/test.yml

@@ -7,6 +7,21 @@ jobs:
 
     runs-on: ubuntu-latest
 
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_PASSWORD: test_password
+          POSTGRES_USER: test_user
+          POSTGRES_DB: micro_test
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+
     steps:
     - uses: actions/checkout@v2
 

README.md

@@ -32,10 +32,28 @@ sbt +publishLocal && cd ..
 
 To run the tests:
 
+**Prerequisites**: PostgreSQL tests require a local PostgreSQL instance. Start one with Docker:
+
+```bash
+docker run --name micro-postgres-test \
+  -e POSTGRES_PASSWORD=test_password \
+  -e POSTGRES_USER=test_user \
+  -e POSTGRES_DB=micro_test \
+  -p 5432:5432 \
+  -d postgres:15
 ```
+
+Then run the tests:
+
+```bash
 sbt test
 ```
 
+To stop the test database:
+```bash
+docker stop micro-postgres-test && docker rm micro-postgres-test
+```
+
 To build a Docker image for local testing:
 ```
 cd ui

build.sbt

@@ -40,7 +40,9 @@ lazy val dependencies = Seq(
     Dependencies.circeGeneric,
     Dependencies.doobieCore,
     Dependencies.doobieHikari,
-    Dependencies.sqliteJdbc,
+    Dependencies.doobiePostgres,
+    Dependencies.doobiePostgresCirce,
+    Dependencies.postgresqlJdbc,
     Dependencies.specs2,
     Dependencies.specs2CE,
     Dependencies.badRows,

project/Dependencies.scala

@@ -25,7 +25,7 @@ object Dependencies {
     val decline     = "2.4.1"
     val slf4j       = "2.0.17"
     val doobie      = "1.0.0-RC10"
-    val sqlite      = "3.50.3.0"
+    val postgresql  = "42.7.8"
 
     // specs2
     val specs2   = "4.12.2"
@@ -55,11 +55,13 @@ object Dependencies {
   val circeGeneric = "io.circe" %% "circe-generic" % V.circe
 
   // doobie
-  val doobieCore   = "org.tpolecat" %% "doobie-core"   % V.doobie
-  val doobieHikari = "org.tpolecat" %% "doobie-hikari" % V.doobie
+  val doobieCore           = "org.tpolecat" %% "doobie-core"           % V.doobie
+  val doobieHikari         = "org.tpolecat" %% "doobie-hikari"         % V.doobie
+  val doobiePostgres       = "org.tpolecat" %% "doobie-postgres"       % V.doobie
+  val doobiePostgresCirce  = "org.tpolecat" %% "doobie-postgres-circe" % V.doobie
 
-  // sqlite
-  val sqliteJdbc = "org.xerial" % "sqlite-jdbc" % V.sqlite
+  // postgresql
+  val postgresqlJdbc = "org.postgresql" % "postgresql" % V.postgresql
 
   // specs2
   val specs2   = "org.specs2"     %% "specs2-core"                % V.specs2   % Test

src/main/scala/com.snowplowanalytics.snowplow.micro/Configuration.scala

@@ -10,7 +10,7 @@
 
 package com.snowplowanalytics.snowplow.micro
 
-import cats.data.EitherT
+import cats.data.{EitherT, ValidatedNel}
 import cats.effect.IO
 import cats.implicits._
 import com.monovore.decline.{Argument, Opts}
@@ -24,10 +24,11 @@ import com.snowplowanalytics.snowplow.collector.core.{Config => CollectorConfig}
 import com.snowplowanalytics.snowplow.enrich.common.adapters.{CallrailSchemas, CloudfrontAccessLogSchemas, GoogleAnalyticsSchemas, HubspotSchemas, MailchimpSchemas, MailgunSchemas, MandrillSchemas, MarketoSchemas, OlarkSchemas, PagerdutySchemas, PingdomSchemas, SendgridSchemas, StatusGatorSchemas, UnbounceSchemas, UrbanAirshipSchemas, VeroSchemas, AdaptersSchemas => EnrichAdaptersSchemas}
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.{AtomicFields, EnrichmentRegistry}
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.registry.EnrichmentConf
-import com.typesafe.config.{ConfigFactory, ConfigParseOptions, ConfigValueFactory, Config => TypesafeConfig}
+import com.typesafe.config.{ConfigFactory, ConfigParseOptions, Config => TypesafeConfig}
 import fs2.io.file.{Files, Path => FS2Path}
 import io.circe.config.syntax.CirceConfigOps
 import io.circe.generic.semiauto.deriveDecoder
+import io.circe.config.syntax._
 import io.circe.syntax.EncoderOps
 import io.circe.{Decoder, Json, JsonObject}
 import org.http4s.Uri
@@ -36,7 +37,7 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
 
 import java.net.URI
 import java.nio.file.{Path, Paths}
-import scala.concurrent.duration.{Duration, DurationInt, FiniteDuration}
+import scala.concurrent.duration.FiniteDuration
 
 object Configuration {
 
@@ -47,32 +48,30 @@ object Configuration {
     case object Json extends OutputFormat
   }
 
-  sealed trait StorageConfig
-  object StorageConfig {
-    case object None extends StorageConfig
-    case object InMemory extends StorageConfig
-    case class Persistent(path: Path, ttl: FiniteDuration, cleanupInterval: FiniteDuration) extends StorageConfig
+  sealed trait StorageMode
+  object StorageMode {
+    case object None extends StorageMode
+    case object InMemory extends StorageMode
+    case class Persistent(host: String,
+                          port: Int,
+                          database: String,
+                          user: String,
+                          password: String,
+                          ttl: FiniteDuration,
+                          cleanupInterval: FiniteDuration) extends StorageMode
   }
 
   object Cli {
     implicit val uriArgument: Argument[Uri] = Argument.from("uri") { str =>
       Uri.fromString(str).leftMap(_ => s"Invalid URI: $str").toValidatedNel
     }
 
-    implicit val durationArgument: Argument[FiniteDuration] = Argument.from("duration") { str =>
-      Either.catchNonFatal {
-        val d = ConfigValueFactory.fromAnyRef(str).atKey("d").getDuration("d")
-        Duration.fromNanos(d.toNanos)
-      }.leftMap(_ => s"Invalid duration format: $str. Use HOCON format (e.g., 1h, 30m, 7d)")
-        .toValidatedNel
-    }
-
     final case class Config(collector: Option[Path],
                             iglu: Option[Path],
                             outputFormat: OutputFormat,
                             destination: Option[Uri],
                             yauaa: Boolean,
-                            storage: StorageConfig,
+                            storage: StorageMode,
                             auth: Option[Path])
 
     private val collector = Opts.option[Path]("collector-config", "Configuration file for Collector", "c", "config.hocon").orNone
@@ -81,11 +80,7 @@ object Configuration {
     private val outputJson = Opts.flag("output-json", "Output events in JSON format to standard output or HTTP destination (with a separate key for each schema)", "j").orFalse
     private val destination = Opts.option[Uri]("destination", "HTTP(s) URL to send output data to (requires --output-json or --output-tsv)", "d").orNone
     private val noStorage = Opts.flag("no-storage", "Do not store events anywhere, and disable the API (handy if using Micro purely for output)").orFalse
-    private val storagePath = Opts.option[Path]("storage-file", "Path to an SQLite database file for persistent storage", "s").orNone
-    private val storageTtl = Opts.option[FiniteDuration]("storage-ttl", "Time-to-live for events in persistent storage (e.g. 1h, 1d; default: 7d)")
-      .withDefault(7.days)
-    private val storageCleanupInterval = Opts.option[FiniteDuration]("storage-cleanup-interval", "Interval for cleanup of expired events in persistent storage (e.g. 1h, 1d; default: 1h)")
-      .withDefault(1.hour)
+    private val storage = Opts.option[Path]("storage", "Configuration file for PostgreSQL storage", "s", "storage.hocon").orNone
     private val yauaa = Opts.flag("yauaa", "Enable YAUAA user agent enrichment").orFalse
     private val auth = Opts.option[Path]("auth", "Configuration file for authentication", "a", "auth.hocon").orNone
 
@@ -99,22 +94,18 @@ object Configuration {
         case (true, true, _) => "Cannot specify both --output-tsv and --output-json".invalidNel[(OutputFormat, Option[Uri])]
       }
 
-    private val storage = (noStorage, storagePath, storageTtl, storageCleanupInterval)
-      .mapN { (_, _, _, _) }
+    private val storageConfig = (noStorage, storage)
+      .mapN { (_, _) }
       .mapValidated {
-        case (true, _, _, _) =>
-          StorageConfig.None.validNel[String]
-        case (_, _, ttl, _) if ttl.toMinutes < 5 =>
-          "--storage-ttl must be at least 5 minutes (5m)".invalidNel[StorageConfig]
-        case (_, _, _, cleanupInterval) if cleanupInterval.toMinutes < 1 =>
-          "--storage-cleanup-interval must be at least 1 minute (1m)".invalidNel[StorageConfig]
-        case (_, None, _, _) =>
-          StorageConfig.InMemory.validNel[String]
-        case (_, Some(path), ttl, cleanupInterval) =>
-          StorageConfig.Persistent(path, ttl, cleanupInterval).validNel[String]
+        case (true, _) =>
+          StorageMode.None.validNel[String]
+        case (_, None) =>
+          StorageMode.InMemory.validNel[String]
+        case (_, Some(path)) =>
+          parseStorageConfig(path)
       }
 
-    val config: Opts[Config] = (collector, iglu, output, yauaa, storage, auth).mapN {
+    val config: Opts[Config] = (collector, iglu, output, yauaa, storageConfig, auth).mapN {
       case (c, i, (f, d), y, s, a) => Config(c, i, f, d, y, s, a)
     }
   }
@@ -126,6 +117,7 @@ object Configuration {
     val sslCertificatePassword = "MICRO_SSL_CERT_PASSWORD"
     val azureBlobAccount = "MICRO_AZURE_BLOB_ACCOUNT"
     val azureBlobSasToken = "MICRO_AZURE_BLOB_SAS_TOKEN"
+    val postgresqlPassword = "MICRO_POSTGRESQL_PASSWORD"
   }
 
   final case class DummySinkConfig()
@@ -139,13 +131,20 @@ object Configuration {
                               organizationId: String,
                               clientId: String)
 
+  final case class StorageConfig(host: String,
+                                port: Int,
+                                database: String,
+                                user: String,
+                                ttl: FiniteDuration,
+                                cleanupInterval: FiniteDuration)
+
   final case class MicroConfig(collector: CollectorConfig[SinkConfig],
                                iglu: IgluResources,
                                enrichmentsConfig: List[EnrichmentConf],
                                enrichConfig: EnrichConfig,
                                outputFormat: OutputFormat,
                                destination: Option[Uri],
-                               storage: StorageConfig,
+                               storage: StorageMode,
                                auth: Option[AuthConfig])
 
   final case class EnrichValidation(atomicFieldsLimits: AtomicFields)
@@ -159,6 +158,40 @@ object Configuration {
 
   implicit private def logger: Logger[IO] = Slf4jLogger.getLogger[IO]
 
+  private def parseStorageConfig(path: Path): ValidatedNel[String, StorageMode] = {
+    try {
+      val config = ConfigFactory.parseFile(path.toFile)
+      config.as[StorageConfig] match {
+        case Right(storageConfig) =>
+          sys.env.get(EnvironmentVariables.postgresqlPassword) match {
+            case Some(password) =>
+              if (storageConfig.ttl.toMinutes < 5) {
+                "Storage TTL must be at least 5 minutes (5m)".invalidNel[StorageMode]
+              } else if (storageConfig.cleanupInterval.toMinutes < 1) {
+                "Storage cleanup interval must be at least 1 minute (1m)".invalidNel[StorageMode]
+              } else {
+                StorageMode.Persistent(
+                  storageConfig.host,
+                  storageConfig.port,
+                  storageConfig.database,
+                  storageConfig.user,
+                  password,
+                  storageConfig.ttl,
+                  storageConfig.cleanupInterval
+                ).validNel[String]
+              }
+            case None =>
+              s"PostgreSQL password not found in environment variable ${EnvironmentVariables.postgresqlPassword}".invalidNel[StorageMode]
+          }
+        case Left(error) =>
+          s"Failed to parse storage configuration: ${error.show}".invalidNel[StorageMode]
+      }
+    } catch {
+      case e: Exception =>
+        s"Failed to read storage configuration file: ${e.getMessage}".invalidNel[StorageMode]
+    }
+  }
+
   def load(): Opts[EitherT[IO, String, MicroConfig]] = {
     Cli.config.map { cliConfig =>
       for {
@@ -407,4 +440,6 @@ object Configuration {
   }
 
   implicit val authConfigDecoder: Decoder[AuthConfig] = deriveDecoder[AuthConfig]
+
+  implicit val storageConfigDecoder: Decoder[StorageConfig] = deriveDecoder[StorageConfig]
 }

src/main/scala/com.snowplowanalytics.snowplow.micro/EventStorage.scala

@@ -11,9 +11,12 @@
 package com.snowplowanalytics.snowplow.micro
 
 import cats.effect.{IO, Resource}
-import com.snowplowanalytics.snowplow.micro.Configuration.StorageConfig
+import cats.implicits._
+import com.snowplowanalytics.snowplow.micro.Configuration.StorageMode
 import com.snowplowanalytics.snowplow.micro.model.ColumnStatsResponse
 import io.circe.Json
+import org.typelevel.log4cats.Logger
+import org.typelevel.log4cats.slf4j.Slf4jLogger
 
 import scala.collection.mutable
 
@@ -41,16 +44,21 @@ object NoStorage extends EventStorage {
 }
 
 object EventStorage {
-  def create(config: StorageConfig): Resource[IO, EventStorage] = {
+  implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
+
+  def create(config: StorageMode): Resource[IO, EventStorage] = {
     config match {
-      case StorageConfig.None =>
+      case StorageMode.None =>
+        Resource.eval(logger.info("Not storing any events")) *>
         Resource.pure(NoStorage)
-      case StorageConfig.Persistent(path, ttl, cleanupInterval) =>
-        for {
-          storage <- SqliteStorage.file(path.toString)
+      case StorageMode.Persistent(host, port, database, user, password, ttl, cleanupInterval) =>
+        Resource.eval(logger.info(s"Storing events in PostgreSQL at $host:$port/$database")) *>
+        (for {
+          storage <- PostgresqlStorage.create(host, port, database, user, password)
           _ <- storage.scheduleCleanup(ttl, cleanupInterval)
-        } yield storage
-      case StorageConfig.InMemory =>
+        } yield storage)
+      case StorageMode.InMemory =>
+        Resource.eval(logger.info("Storing events in memory")) *>
         Resource.pure(new InMemoryStorage())
     }
   }

src/main/scala/com.snowplowanalytics.snowplow.micro/Main.scala

@@ -20,8 +20,8 @@ object Main
     header =
       s"""Notes:
          |    - By default, events are stored in memory. This mode supports all API endpoints.
-         |    - With --storage-file (persistent storage), /micro/all, /micro/good and /micro/bad endpoints are not supported.
-         |    - With --no-storage (no in-memory or on-disk storage), all /micro/* endpoints are disabled.""".stripMargin,
+         |    - With --storage (PostgreSQL storage), /micro/all, /micro/good and /micro/bad endpoints are not supported.
+         |    - With --no-storage (no in-memory or database storage), all /micro/* endpoints are disabled.""".stripMargin,
     version = BuildInfo.version
   ) {
   override def main: Opts[IO[ExitCode]] = Run.run()