Continued fixes

all/src/test/scala/spec/AbstractAsyncSpec.scala

@@ -60,7 +60,7 @@ abstract class AbstractAsyncSpec extends AnyWordSpec with Matchers { spec =>
 
   specName should {
     "initialize the database" in {
-      db.init().map(b => b should be(true)).sync()
+      db.init.map(b => b should be(true)).sync()
     }
     "verify the database is empty" in {
       db.people.transaction { implicit transaction =>
@@ -187,7 +187,7 @@ abstract class AbstractAsyncSpec extends AnyWordSpec with Matchers { spec =>
     }
     "prepare a new instance" in {
       db = new DB
-      db.init().map(_ should be(true)).sync()
+      db.init.map(_ should be(true)).sync()
     }
     "query the database to verify records were persisted properly" in {
       db.people.transaction { implicit transaction =>

all/src/test/scala/spec/AirportSpec.scala

@@ -22,7 +22,7 @@ import scala.io.Source
 class AirportSpec extends AnyWordSpec with Matchers {
   "AirportSpec" should {
     "initialize the database" in {
-      DB.init()
+      db.init
     }
     "have two collections" in {
       DB.collections.map(_.name).toSet should be(Set("_backingStore", "Flight", "Airport"))

benchmark/src/main/scala/benchmark/JMHBenchmarks.scala

@@ -34,7 +34,7 @@
 //    new File("benchmarks.json").delete()
 //    FileUtils.deleteDirectory(dbDir)
 //    dbDir.mkdirs()
-//    DB.init()
+//    db.init
 //
 //    val s = sqliteConnection.createStatement()
 //    s.executeUpdate("CREATE TABLE record(id VARCHAR NOT NULL, key TEXT, number INTEGER, PRIMARY KEY (id))")

benchmark/src/main/scala/benchmark/TightLoops.scala

@@ -49,7 +49,7 @@
 //      FileUtils.deleteDirectory(dbDir)
 //      dbDir.mkdirs()
 //
-//      DB.init()
+//      db.init
 //      DB.people.transaction { implicit transaction =>
 //        val insertTime = elapsed(insertRecords())
 //        scribe.info(s"Inserted in $insertTime")

benchmark/src/main/scala/benchmark/bench/impl/LightDB011Bench.scala

@@ -16,7 +16,7 @@
 //object LightDB011Bench extends Bench {
 //  override def name: String = "LightDB 0.11"
 //
-//  override def init(): Unit = DB.init().unsafeRunSync()
+//  override def init(): Unit = db.init.unsafeRunSync()
 //
 //  override protected def insertRecords(iterator: Iterator[LightDB011Bench.P]): Unit = {
 //    fs2.Stream.fromBlockingIterator[IO](iterator.map { p =>

benchmark/src/main/scala/benchmark/bench/impl/LightDBAsyncBench.scala

@@ -16,7 +16,7 @@ import java.sql.ResultSet
 case class LightDBAsyncBench(storeManager: StoreManager) extends Bench { bench =>
   override def name: String = s"LightDB Async ${storeManager.name}"
 
-  override def init(): Unit = DB.init().sync()
+  override def init(): Unit = db.init.sync()
 
   implicit def p2Person(p: P): Person = Person(p.name, p.age, Id(p.id))
 

benchmark/src/main/scala/benchmark/bench/impl/LightDBBench.scala

@@ -17,7 +17,7 @@ import scala.language.implicitConversions
 case class LightDBBench(storeManager: StoreManager) extends Bench { bench =>
   override def name: String = s"LightDB ${storeManager.name}"
 
-  override def init(): Unit = DB.init()
+  override def init(): Unit = db.init
 
   implicit def p2Person(p: P): Person = Person(p.name, p.age, Id(p.id))
 

benchmark/src/main/scala/benchmark/imdb/LightDBImplementation.scala

@@ -20,7 +20,7 @@
 //
 //  override def name: String = "LightDB"
 //
-//  override def init(): Task[Unit] = IO(DB.init())
+//  override def init(): Task[Unit] = IO(db.init)
 //
 //  override def map2TitleAka(map: Map[String, String]): TitleAkaLDB = TitleAkaLDB(
 //    titleId = map.value("titleId"),

benchmark/src/main/scala/benchmark/imdb/ScarangoImplementation.scala

@@ -24,7 +24,7 @@
 //
 //  override def name: String = "Scarango"
 //
-//  override def init(): Task[Unit] = db.init()
+//  override def init(): Task[Unit] = db.init
 //
 //  override def map2TitleAka(map: Map[String, String]): TitleAkaADB = {
 //    val title = map.value("title")

core/src/main/scala/lightdb/LightDB.scala

@@ -83,7 +83,7 @@ trait LightDB extends Initializable with FeatureSupport[DBFeatureKey] {
   override protected def initialize(): Task[Unit] = for {
     _ <- logger.info(s"$name database initializing...")
     _ = backingStore
-    _ <- collections.map(_.init()).tasks
+    _ <- collections.map(_.init).tasks
     // Truncate the database before we do anything if specified
     _ <- truncate().when(truncateOnInit)
     // Determine if this is an uninitialized database
@@ -124,7 +124,7 @@ trait LightDB extends Initializable with FeatureSupport[DBFeatureKey] {
       _collections = c :: _collections
     }
     if (isInitialized) { // Already initialized database, init collection immediately
-      c.init()
+      c.init.sync()
     }
     c
   }
@@ -169,12 +169,13 @@ trait LightDB extends Initializable with FeatureSupport[DBFeatureKey] {
       upgrade.upgrade(this).when(runUpgrade).flatMap { _ =>
         appliedUpgrades.modify { set =>
           set + upgrade.label
-        }
-        val next = doUpgrades(upgrades.tail, dbInitialized, continueBlocking)
-        if (stillBlocking && !continueBlocking) {
-          next.start()
-        } else {
-          next
+        }.flatMap { _ =>
+          val next = doUpgrades(upgrades.tail, dbInitialized, continueBlocking)
+          if (stillBlocking && !continueBlocking) {
+            next.start()
+          } else {
+            next
+          }
         }
       }
     case None => logger.info("Upgrades completed successfully")

core/src/main/scala/lightdb/backup/DatabaseBackup.scala

@@ -18,28 +18,26 @@ object DatabaseBackup {
    * @return the number of records backed up
    */
   def archive(db: LightDB,
-              archive: File = new File("backup.zip")): Task[Int] = {
+              archive: File = new File("backup.zip")): Task[Int] = Task {
     Option(archive.getParentFile).foreach(_.mkdirs())
     if (archive.exists()) archive.delete()
     val out = new ZipOutputStream(new FileOutputStream(archive))
-    try {
-      process(db) {
-        case (collection, stream) => Task {
-          val fileName = s"${collection.name}.jsonl"
-          val entry = new ZipEntry(s"backup/$fileName")
-          out.putNextEntry(entry)
-          stream.map { json =>
-            val line = JsonFormatter.Compact(json)
-            val bytes = s"$line\n".getBytes("UTF-8")
-            out.write(bytes)
-          }.count.guarantee(Task(out.closeEntry()))
-        }.flatten
-      }
-    } finally {
+    process(db) {
+      case (collection, stream) => Task {
+        val fileName = s"${collection.name}.jsonl"
+        val entry = new ZipEntry(s"backup/$fileName")
+        out.putNextEntry(entry)
+        stream.map { json =>
+          val line = JsonFormatter.Compact(json)
+          val bytes = s"$line\n".getBytes("UTF-8")
+          out.write(bytes)
+        }.count.guarantee(Task(out.closeEntry()))
+      }.flatten
+    }.guarantee(Task {
       out.flush()
       out.close()
-    }
-  }
+    })
+  }.flatten
 
   /**
    * Does a full backup of the supplied database to the directory specified

core/src/main/scala/lightdb/collection/Collection.scala

@@ -46,11 +46,11 @@ case class Collection[Doc <: Document[Doc], Model <: DocumentModel[Doc]](name: S
     }
 
     // Give the Model a chance to initialize
-    model.init(this)
-
-    // Verify the data is in-sync
-    verify()
-  }
+    model.init(this).flatMap { _ =>
+      // Verify the data is in-sync
+      verify()
+    }
+  }.flatten.unit
 
   def verify(): Task[Boolean] = store.verify()
 

core/src/main/scala/lightdb/util/Aggregator.scala

@@ -92,6 +92,8 @@ object Aggregator {
         }
         groups += group -> map
       }
+      .drain
+      .sync()
     groups = groups.map {
       case (key, jsonMap) =>
         var map = jsonMap

core/src/main/scala/lightdb/util/Initializable.scala

@@ -8,16 +8,15 @@ import rapid._
  */
 trait Initializable {
   @volatile private var initialized = false
-  private lazy val singleton = initialize().map { _ =>
-    initialized = true
-  }.singleton
-
-  def isInitialized: Boolean = initialized
 
   /**
    * Calls initialize() exactly one time. Safe to call multiple times.
    */
-  final def init(): Task[Unit] = singleton
+  lazy val init: Task[Unit] = initialize().map { _ =>
+    initialized = true
+  }.singleton
+
+  def isInitialized: Boolean = initialized
 
   /**
    * Define initialization functionality here, but never call directly.

core/src/test/scala/spec/AbstractBasicSpec.scala

@@ -69,7 +69,7 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
 
   specName should {
     "initialize the database" in {
-      db.init().succeed
+      db.init.succeed
     }
     "verify the database is empty" in {
       db.people.transaction { implicit transaction =>
@@ -220,7 +220,7 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
       } else {
         db.dispose().flatMap { _ =>
           db = new DB
-          db.init()
+          db.init
         }.succeed
       }
     }
@@ -250,9 +250,12 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
             .should(_.search.words("nica 13", matchEndsWith = true), boost = Some(2.0))
             .should(_.age <=> (10, 15))
           ).search.docs.flatMap { results =>
-            results.list.map { people =>
+            for {
+              people <- results.list
+              scores <- results.scores
+            } yield {
               people.map(_.name) should be(List("Veronica", "Brenda", "Diana", "Greg", "Charlie", "Evan", "Fiona", "Hanna", "Ian", "Jenna", "Kevin", "Mike", "Nancy", "Oscar", "Penny", "Quintin", "Ruth", "Sam", "Tori", "Uba", "Wyatt", "Xena", "Zoey", "Allan"))
-              results.scores should be(List(6.0, 2.0, 2.0, 2.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0))
+              scores should be(List(6.0, 2.0, 2.0, 2.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0))
             }
           }
         }
@@ -399,11 +402,11 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
           .should(_.friends has fiona._id)
         )
         q.toList.map { list =>
-          list.map(_.name).toSet should be(Set("Same"))
+          list.map(_.name).toSet should be(Set("Sam"))
         }
       }
     }
-    /*"do a database backup" in {
+    "do a database backup" in {
       DatabaseBackup.archive(db, new File(s"backups/$specName.zip")).map(_ should be(49))
     }
     "insert a lot more names" in {
@@ -469,7 +472,7 @@ abstract class AbstractBasicSpec extends AsyncWordSpec with AsyncTaskSpec with M
             }
           }
       }
-    }*/
+    }
     "truncate the collection" in {
       db.people.transaction { implicit transaction =>
         db.people.truncate().map(_ should be(CreateRecords + 24))

core/src/test/scala/spec/AbstractFacetSpec.scala

@@ -29,7 +29,7 @@ abstract class AbstractFacetSpec extends AsyncWordSpec with AsyncTaskSpec with M
 
   specName should {
     "initialize the database" in {
-      db.init().succeed
+      db.init.succeed
     }
     "verify the database is empty" in {
       db.entries.transaction { implicit transaction =>

core/src/test/scala/spec/AbstractKeyValueSpec.scala

@@ -53,7 +53,7 @@ abstract class AbstractKeyValueSpec extends AsyncWordSpec with AsyncTaskSpec wit
 
   specName should {
     "initialize the database" in {
-      db.init().succeed
+      db.init.succeed
     }
     "verify the database is empty" in {
       db.users.transaction { implicit transaction =>

core/src/test/scala/spec/AbstractSpatialSpec.scala

@@ -67,7 +67,7 @@ abstract class AbstractSpatialSpec extends AsyncWordSpec with AsyncTaskSpec with
 
   specName should {
     "initialize the database" in {
-      DB.init().succeed
+      DB.init.succeed
     }
     "store three people" in {
       DB.people.transaction { implicit transaction =>

core/src/test/scala/spec/AbstractSpecialCasesSpec.scala

@@ -18,7 +18,7 @@ trait AbstractSpecialCasesSpec extends AsyncWordSpec with AsyncTaskSpec with Mat
 
   specName should {
     "initialize the database" in {
-      DB.init().succeed
+      DB.init.succeed
     }
     "insert a couple SpecialOne instances" in {
       DB.specialOne.t.insert(List(