SQLite CRDT + Session Extension

main.dart

import 'dart:ffi';
import 'dart:io';
import 'dart:typed_data';

import 'package:sqlite3/sqlite3.dart';
import 'package:sqlite3/src/ffi/implementation.dart';

void main() {
  final lib = DynamicLibrary.open('cr-sqlite/core/dist/sqlite3');
  final sqlite3 = FfiSqlite3(lib);
  // final sqlite3 = sqlite3Native;
  print('Using sqlite3 ${sqlite3.version}');

  final crSqlite = SqliteExtension.inLibrary(
    DynamicLibrary.process(),
    'sqlite3_crsqlite_init',
  );
  sqlite3.ensureExtensionLoaded(crSqlite);

  // Create a new in-memory database. To use a database backed by a file, you
  // can replace this with sqlite3.open(yourFilePath).
  final db = sqlite3.openInMemory() as DB;
  final db2 = sqlite3.openInMemory() as DB;

  var session = db.createSession(CreateSessionOptions(table: 'changes'));

  print('session: ${session.runtimeType}');

  // Create a table and insert some data
  db.init();
  db2.init();

  // Prepare a statement to run it multiple times:
  final stmt = db.prepare('INSERT INTO artists (name) VALUES (?)');
  stmt
    ..execute(['The Beatles'])
    ..execute(['Led Zeppelin'])
    ..execute(['The Who'])
    ..execute(['Nirvana']);

  // Dispose a statement when you don't need it anymore to clean up resources.
  stmt.dispose();

  db.commit();

  // print('changes: ${db.select('SELECT * FROM changes')}');

  // print(db.selectChanges());
  var changeset = session.changeset(); // or patchset
  print('changeset: ${changeset.lengthInBytes} bytes');

  // apply changes
  db2.apply(changeset);
  session.close();

  // save to ./changeset.bin or send it to another peer
  File('./changeset.bin').writeAsBytes(changeset);

  // query the database using a simple select statement
  final result = db2.select('SELECT * FROM artists');
  for (final row in result) {
    print('cs: Artist[id: ${row['id']}, name: ${row['name']}]');
  }

  // You can run select statements with PreparedStatement.select, or directly
  // on the database:
  final ResultSet resultSet = db.select(
    'SELECT * FROM artists WHERE name LIKE ?',
    ['The %'],
  );

  // You can iterate on the result set in multiple ways to retrieve Row objects
  // one by one.
  for (final Row row in resultSet) {
    print('Artist[id: ${row['id']}, name: ${row['name']}]');
  }

  // Register a custom function we can invoke from sql:
  db.createFunction(
    functionName: 'dart_version',
    argumentCount: const AllowedArgumentCount(0),
    function: (args) => Platform.version,
  );
  print(db.select('SELECT dart_version()'));

  // Don't forget to dispose the database to avoid memory leaks
  db2.close();
  db.close();
}

extension type DB(Database db) implements Database {
  void init() {
    db.execute('''
    CREATE TABLE artists (
      id INTEGER NOT NULL PRIMARY KEY,
      name TEXT NOT NULL DEFAULT ''
    );
    CREATE TABLE changes (
      id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
      "pk" TEXT NOT NULL,
      "table" TEXT NOT NULL,
      "cid" INTEGER NOT NULL,
      "val" TEXT NOT NULL,
      "col_version" INTEGER NOT NULL,
      "db_version" INTEGER NOT NULL,
      "site_id" INTEGER NOT NULL,
      "cl" TEXT NOT NULL,
      "seq" INTEGER NOT NULL
    );
    ''');
    db.execute("select crsql_as_crr('artists');");
  }

  List<Map<String, dynamic>> selectChanges() {
    final result = db.select('''
    SELECT "table", "pk", "cid", "val", "col_version", "db_version", "site_id", "cl", "seq" FROM crsql_changes;
    ''');
    return result;
  }

  void _toChanges() {
    db.execute(_sync('changes', 'crsql_changes'));
  }

  void _fromChanges() {
    db.execute(_sync('crsql_changes', 'changes'));
  }

  static String _sync(String from, String to) {
    return '''
    INSERT INTO $from ("table", "pk", "cid", "val", "col_version", "db_version", "site_id", "cl", "seq")
    SELECT "table", "pk", "cid", "val", "col_version", "db_version", "site_id", "cl", "seq" FROM $to;
    ''';
  }

  void commit() {
    _toChanges();
  }

  void apply(Uint8List bytes) {
    db.applyChangeset(bytes);
    _fromChanges();
    db.execute('DELETE FROM changes;');
  }

  void close() {
    db.execute('select crsql_finalize();');
    db.dispose();
  }
}