fix(runtime): Persist database in local environment

When running locally, persist DBs to the filesystem so that data is present between restarts.

packages/celest/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 - chore: Remove assumptions about where the project is deployed
 - chore: Allow connecting to locally running libSQL servers
+- fix: Persist database in local environment
 
 ## 1.0.0
 

packages/celest/build.yaml

@@ -0,0 +1,5 @@
+targets:
+  $default:
+    sources:
+      include:
+        - 'test/runtime/data/**'

packages/celest/dart_test.yaml

@@ -0,0 +1,2 @@
+tags:
+  e2e: {}

packages/celest/lib/src/runtime/data/connect.dart

@@ -25,9 +25,10 @@ Future<Database> connect<Database extends GeneratedDatabase>(
   required Database Function(QueryExecutor) factory,
   required env hostnameVariable,
   required secret tokenSecret,
+  String? path,
 }) async {
   if (context.environment == Environment.local) {
-    final executor = await inMemoryExecutor();
+    final executor = await localExecutor(name: name, path: path);
     return _checkConnection(factory(executor));
   }
   final host = context.get(hostnameVariable);

packages/celest/lib/src/runtime/data/connect.io.dart

@@ -1,5 +1,44 @@
+import 'dart:io';
+import 'dart:isolate';
+
 import 'package:drift/drift.dart';
 import 'package:drift/native.dart';
+import 'package:logging/logging.dart';
+import 'package:path/path.dart' as p;
+
+final Logger _logger = Logger('Celest.Data');
 
 /// A [QueryExecutor] for an in-memory database.
 Future<QueryExecutor> inMemoryExecutor() async => NativeDatabase.memory();
+
+/// A [QueryExecutor] with local persistence.
+Future<QueryExecutor> localExecutor({
+  required String name,
+  String? path,
+}) async {
+  if (path == null) {
+    final packageConfig = await Isolate.packageConfig;
+    if (packageConfig == null) {
+      _logger.warning(
+        'Failed to determine package config path. '
+        'Falling back to in-memory database.',
+      );
+      return inMemoryExecutor();
+    }
+    path = p.join(
+      p.dirname(p.fromUri(packageConfig)),
+      'celest',
+      name,
+    );
+  }
+  _logger.info('Opening database at $path');
+  final databaseFile = File(path);
+  if (!databaseFile.existsSync()) {
+    _logger.info('Database does not exist. Creating new database.');
+  }
+  return NativeDatabase(
+    databaseFile,
+    cachePreparedStatements: true,
+    enableMigrations: true,
+  );
+}

packages/celest/lib/src/runtime/data/connect.web.dart

@@ -8,3 +8,12 @@ Future<QueryExecutor> inMemoryExecutor() async {
   sqlite3.registerVirtualFileSystem(InMemoryFileSystem(), makeDefault: true);
   return WasmDatabase.inMemory(sqlite3);
 }
+
+/// A [QueryExecutor] with local persistence.
+Future<QueryExecutor> localExecutor({
+  required String name,
+  String? path,
+}) async {
+  // TODO(dnys1): We don't have a use case for this yet.
+  return inMemoryExecutor();
+}

packages/celest/pubspec.yaml

@@ -29,6 +29,7 @@ dependencies:
   http_sfv: ^0.1.0
   logging: ^1.2.0
   meta: ^1.11.0
+  path: ^1.9.0
   platform: ^3.1.5
   shelf: ^1.4.1
   shelf_router: ^1.1.4
@@ -40,8 +41,10 @@ dependencies:
   x509: ^0.2.4
 
 dev_dependencies:
+  build_runner: ^2.4.13
   celest_lints:
     path: ../celest_lints
+  drift_dev: '>=2.21.0 <2.22.0'
   pub_semver: ^2.1.4
   stream_transform: ^2.1.0
   test: ^1.25.1

packages/celest/test/runtime/data/connect_test.dart

@@ -2,70 +2,120 @@
 @Tags(['e2e'])
 library;
 
+import 'dart:async';
+import 'dart:convert';
 import 'dart:io';
+import 'dart:isolate';
 
 import 'package:celest/src/config/config_values.dart';
 import 'package:celest/src/core/context.dart';
 import 'package:celest/src/runtime/data/connect.dart';
-import 'package:drift/drift.dart';
+import 'package:drift/native.dart';
 import 'package:drift_hrana/drift_hrana.dart';
+import 'package:path/path.dart' as p;
 import 'package:platform/platform.dart';
-import 'package:test/fake.dart';
 import 'package:test/test.dart';
 
+import 'test_database.dart';
+
 Future<int> _findOpenPort() async {
   final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
   final port = server.port;
   await server.close();
   return port;
 }
 
-Future<Uri> _startLibSqlServer() async {
+Future<Uri> _startSqld() async {
   final port = await _findOpenPort();
   final process = await Process.start('turso', ['dev', '--port', '$port']);
   addTearDown(process.kill);
+  final running = Completer<void>();
+  process.stdout
+      .transform(utf8.decoder)
+      .transform(const LineSplitter())
+      .listen((event) {
+    if (event.contains('sqld listening on')) {
+      running.complete();
+    }
+  });
+  await running.future.timeout(
+    const Duration(seconds: 5),
+    onTimeout: () {
+      throw StateError('Failed to start sqld');
+    },
+  );
   return Uri.parse('ws://localhost:$port');
 }
 
 void main() {
   group('connect', () {
-    test('allows local libSQL URIs', () async {
-      final host = await _startLibSqlServer();
+    test('allows local sqld URIs', () async {
+      final host = await _startSqld();
       final platform = FakePlatform(
         environment: {
           'CELEST_DATABASE_HOST': host.toString(),
         },
       );
       Context.current.put(env.environment, 'production');
       Context.current.put(ContextKey.platform, platform);
-      await connect(
+      final database = await connect(
         Context.current,
         name: 'test',
         factory: expectAsync1((executor) {
           expect(executor, isA<HranaDatabase>());
-          return _FakeDatabase();
+          return TestDatabase(executor);
         }),
         hostnameVariable: const env('CELEST_DATABASE_HOST'),
         tokenSecret: const secret('CELEST_DATABASE_TOKEN'),
       );
+      await database.close();
     });
   });
-}
 
-final class _FakeDatabase extends Fake implements GeneratedDatabase {
-  @override
-  Selectable<QueryRow> customSelect(
-    String query, {
-    List<Variable<Object>> variables = const [],
-    Set<ResultSetImplementation<dynamic, dynamic>> readsFrom = const {},
-  }) {
-    return _FakeSelectable();
-  }
-}
+  group('localExecutor', () {
+    test('uses package config when path=null', () async {
+      final platform = FakePlatform();
+      Context.current.put(env.environment, 'local');
+      Context.current.put(ContextKey.platform, platform);
+      final database = await connect(
+        Context.current,
+        name: 'test',
+        factory: expectAsync1((executor) {
+          expect(executor, isA<NativeDatabase>());
+          return TestDatabase(executor);
+        }),
+        hostnameVariable: const env('CELEST_DATABASE_HOST'),
+        tokenSecret: const secret('CELEST_DATABASE_TOKEN'),
+      );
+      addTearDown(database.close);
+
+      final packageConfig = await Isolate.packageConfig;
+      final file = File.fromUri(packageConfig!.resolve('./celest/test.db'));
+      expect(file.existsSync(), isTrue);
+    });
+
+    test('path != null', () async {
+      final tmpDir = await Directory.systemTemp.createTemp('celest_test');
+      addTearDown(() => tmpDir.delete(recursive: true));
 
-final class _FakeSelectable extends Fake implements Selectable<Never> {
-  @override
-  Future<List<Never>> get() {
-    return Future.value([]);
-  }
+      final platform = FakePlatform();
+      Context.current.put(env.environment, 'local');
+      Context.current.put(ContextKey.platform, platform);
+      final database = await connect(
+        Context.current,
+        name: 'test',
+        factory: expectAsync1((executor) {
+          expect(executor, isA<NativeDatabase>());
+          return TestDatabase(executor);
+        }),
+        hostnameVariable: const env('CELEST_DATABASE_HOST'),
+        tokenSecret: const secret('CELEST_DATABASE_TOKEN'),
+        path: p.join(tmpDir.path, 'test.db'),
+      );
+      addTearDown(database.close);
+
+      final file = File.fromUri(tmpDir.uri.resolve('./test.db'));
+      expect(file.existsSync(), isTrue);
+    });
+  });
 }

packages/celest/test/runtime/data/test_database.dart

@@ -0,0 +1,16 @@
+import 'package:drift/drift.dart';
+
+part 'test_database.g.dart';
+
+class Items extends Table {
+  IntColumn get id => integer().autoIncrement()();
+  TextColumn get name => text().nullable()();
+}
+
+@DriftDatabase(tables: [Items])
+class TestDatabase extends _$TestDatabase {
+  TestDatabase(super.e);
+
+  @override
+  int get schemaVersion => 1;
+}