chore: Move the drift_offline_first to legacy in favor of Serverpod implementation (#32)
This commit is contained in:
Marcelo Mendonça Soares
Marcelo Soares
parent
027aec5ed0
commit
cbc472523b
71 changed files with 13575 additions and 284 deletions
164
legacy/drift_offline_sync/.github/workflows/ci.yaml
vendored
Normal file
164
legacy/drift_offline_sync/.github/workflows/ci.yaml
vendored
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
name: CI
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
- stable-*
|
||||
pull_request:
|
||||
branches:
|
||||
- main
|
||||
- stable-*
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.head_ref || github.ref }}
|
||||
cancel-in-progress: ${{ github.ref != 'refs/heads/main' }}
|
||||
|
||||
jobs:
|
||||
generate:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Dart
|
||||
uses: dart-lang/setup-dart@v1
|
||||
- name: Run pub get
|
||||
run: dart pub get
|
||||
- name: Generate code
|
||||
run: dart run build_runner build --delete-conflicting-outputs
|
||||
- name: Upload generated code
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: generated-code
|
||||
path: |
|
||||
**/*.g.dart
|
||||
**/*.freezed.dart
|
||||
lib/**/*.part.dart
|
||||
|
||||
dart_format:
|
||||
name: dart format
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Dart
|
||||
uses: dart-lang/setup-dart@v1
|
||||
- name: Run pub get
|
||||
run: dart pub get
|
||||
# Skip code generation for formatting check
|
||||
- name: Verify formatting
|
||||
run: dart format --output=none --set-exit-if-changed .
|
||||
|
||||
dart_analyze:
|
||||
needs: generate
|
||||
name: dart analyze
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Dart
|
||||
uses: dart-lang/setup-dart@v1
|
||||
- name: Run pub get
|
||||
run: dart pub get
|
||||
- name: Download generated code
|
||||
uses: actions/download-artifact@v4
|
||||
with:
|
||||
name: generated-code
|
||||
- name: Run lint
|
||||
run: dart analyze .
|
||||
|
||||
tests:
|
||||
needs: generate
|
||||
name: dart test
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
pull-requests: write
|
||||
contents: read
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Dart
|
||||
uses: dart-lang/setup-dart@v1
|
||||
with:
|
||||
sdk: stable
|
||||
- name: Run pub get
|
||||
run: dart pub get
|
||||
- name: Download generated code
|
||||
uses: actions/download-artifact@v4
|
||||
with:
|
||||
name: generated-code
|
||||
- name: Run tests with coverage
|
||||
run: dart test --coverage=coverage --reporter=failures-only
|
||||
- name: Activate coverage package
|
||||
run: dart pub global activate coverage
|
||||
- name: Format coverage
|
||||
run: dart pub global run coverage:format_coverage --lcov --in=coverage --out=coverage/lcov.info --report-on=lib --check-ignore --ignore-files="**/*.g.dart"
|
||||
- name: Generate coverage report
|
||||
run: dart scripts/coverage_report.dart --ci > coverage-results.txt
|
||||
- name: Comment on PR
|
||||
if: github.event_name == 'pull_request'
|
||||
uses: thollander/actions-comment-pull-request@v3
|
||||
with:
|
||||
file-path: coverage-results.txt
|
||||
comment-tag: coverage-results
|
||||
|
||||
benchmarks:
|
||||
needs: generate
|
||||
name: run benchmarks
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
pull-requests: write
|
||||
contents: write
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Dart
|
||||
uses: dart-lang/setup-dart@v1
|
||||
with:
|
||||
sdk: stable
|
||||
- name: Run pub get
|
||||
run: dart pub get
|
||||
- name: Download generated code
|
||||
uses: actions/download-artifact@v4
|
||||
with:
|
||||
name: generated-code
|
||||
- name: Configure Git
|
||||
run: |
|
||||
git config --global user.name "GitHub Actions"
|
||||
git config --global user.email "actions@github.com"
|
||||
- name: Fetch Existing Benchmark Data
|
||||
run: git fetch origin refs/notes/benchmarks:refs/notes/benchmarks || true
|
||||
- name: Run benchmarks
|
||||
run: dart benchmark/run.dart --ci > benchmark-results.txt
|
||||
- name: Comment on PR
|
||||
if: github.event_name == 'pull_request'
|
||||
uses: thollander/actions-comment-pull-request@v3
|
||||
with:
|
||||
file-path: benchmark-results.txt
|
||||
comment-tag: benchmark-results
|
||||
- name: Store Benchmark Data in Git Notes
|
||||
run: |
|
||||
if [ "${{ github.event_name }}" = "pull_request" ]; then
|
||||
COMMIT_SHA=${{ github.event.pull_request.head.sha }}
|
||||
else
|
||||
COMMIT_SHA=${{ github.sha }}
|
||||
fi
|
||||
git notes --ref=benchmarks add -f -F benchmark-results.txt $COMMIT_SHA
|
||||
- name: Push Benchmark Data
|
||||
if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.fork != true
|
||||
run: git push origin refs/notes/benchmarks
|
||||
|
||||
publish_check:
|
||||
needs: generate
|
||||
name: dart publish check
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Dart
|
||||
uses: dart-lang/setup-dart@v1
|
||||
with:
|
||||
sdk: stable
|
||||
- name: Run pub get
|
||||
run: dart pub get
|
||||
- name: Download generated code
|
||||
uses: actions/download-artifact@v4
|
||||
with:
|
||||
name: generated-code
|
||||
- name: Run publish dry run
|
||||
run: dart pub publish --dry-run
|
||||
13
legacy/drift_offline_sync/.gitignore
vendored
Normal file
13
legacy/drift_offline_sync/.gitignore
vendored
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
# https://dart.dev/guides/libraries/private-files
|
||||
# Created by `dart pub`
|
||||
.dart_tool/
|
||||
|
||||
# Avoid committing pubspec.lock for library packages; see
|
||||
# https://dart.dev/guides/libraries/private-files#pubspeclock.
|
||||
pubspec.lock
|
||||
|
||||
.vscode/
|
||||
|
||||
# Coverage reports
|
||||
coverage/
|
||||
coverage-results.txt
|
||||
3
legacy/drift_offline_sync/CHANGELOG.md
Normal file
3
legacy/drift_offline_sync/CHANGELOG.md
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
## 0.0.1
|
||||
|
||||
- Initial version.
|
||||
21
legacy/drift_offline_sync/LICENSE
Normal file
21
legacy/drift_offline_sync/LICENSE
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
MIT License
|
||||
|
||||
Copyright (c) 2025 Marcelo Mendonça Soares
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
90
legacy/drift_offline_sync/README.md
Normal file
90
legacy/drift_offline_sync/README.md
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
# Drift Offline Sync
|
||||
|
||||
[](https://github.com/marcelomendoncasoares)
|
||||
[](https://pub.dev/packages/drift_offline_sync)
|
||||
[](https://pub.dev/packages/drift_offline_sync)
|
||||
[](https://github.com/marcelomendoncasoares/drift_offline_sync/blob/main/LICENSE)
|
||||
[](https://github.com/marcelomendoncasoares/drift_offline_sync/actions/workflows/ci.yaml)
|
||||
|
||||
A plugin to build offline-first applications with Drift. Develop your app as a
|
||||
normal offline app and use this non-invasive plugin to automatically handle the
|
||||
sync process using CRDT (Conflict-free Replicated Data Type). Works
|
||||
out-of-the-box with Drift's existing APIs.
|
||||
|
||||
- [Drift Offline Sync](#drift-offline-sync)
|
||||
- [What is this package?](#what-is-this-package)
|
||||
- [Usage](#usage)
|
||||
- [How does it work?](#how-does-it-work)
|
||||
- [Performance considerations](#performance-considerations)
|
||||
- [License](#license)
|
||||
|
||||
---
|
||||
|
||||
## What is this package?
|
||||
|
||||
This package is an adapter for a Drift database that can transform any offline application into offline-first with multi-device support. It can be plugged into any application developed with an offline database (managed in Drift) to provide full synchronization with a remote server. The only required change is the replacement of the original `Migrator` object in the database declaration by the one provided in this package and that all tables have a UUID as part of its primary key.
|
||||
|
||||
Under the hood, the sync engine uses Conflict-free Replicated Data Types (CRDT) to merge updates and solve merge conflicts. Beyond traditional CRDT approaches, it implements a robust mechanism to preserve invariants of the system (foreign key and unique constraints, for example) and handle migrations. This is what enables this engine to be safely used with traditional relational data instead of sacrificing data integrity and strict typing to use key-value storages.
|
||||
|
||||
## Usage
|
||||
|
||||
Just use the `OfflineSyncMigrator` in the `createMigrator` method of your database class.
|
||||
|
||||
```dart
|
||||
class TodoDb extends _$TodoDb {
|
||||
TodoDb(super.e);
|
||||
|
||||
@override
|
||||
int schemaVersion = 1;
|
||||
|
||||
// This is what needs to be added to the user database class to make it CRDT aware.
|
||||
@override
|
||||
OfflineSyncMigrator createMigrator() => OfflineSyncMigrator(
|
||||
this,
|
||||
userId: userId,
|
||||
nodeId: nodeId,
|
||||
synchronizedTables: [
|
||||
todosTable,
|
||||
categories,
|
||||
users,
|
||||
sharedTodos,
|
||||
],
|
||||
excludeTables: [
|
||||
listing,
|
||||
product,
|
||||
],
|
||||
);
|
||||
}
|
||||
```
|
||||
|
||||
Make sure that all tables have a UUID as part of its primary key - standard practice for distributed systems.
|
||||
|
||||
## How does it work?
|
||||
|
||||
When the database is initialized, the `OfflineSyncMigrator` will create a few control tables and add triggers to all user tables listed for synchronization (defaults to all tables). Each table will receive three triggers: after insert, after update and after delete. The triggers serve to populate the main CRDT control table with all the changes (granular, individualized to field updates and their invariance preserving rules). Each update is marked with an HLC (Hybrid Logical Clock) timestamp that will be used as reference to further merging changes between devices.
|
||||
|
||||
During migrations, as triggers are created expecting the current table schema, they must be removed to avoid errors - like trying to compare a column that was removed. So, after migration is finished, a `regenerate` is called to compare the changes and synchronize the central CRDT registry. In the meantime, changes to the schema (like column addition and deletion) are reflected to the stored data on the central registry.
|
||||
|
||||
Although schema changes does not get propagated through the merging process—there is no CRDT registry for table schemas—this is not a concern, as changes can only be synchronized if the databases match version. So, after migration, all databases will converge to the same schema, with each carrying their pending merge data to the new format. Only then, sync will occur.
|
||||
|
||||
## Performance considerations
|
||||
|
||||
The greater costs of this adapter are:
|
||||
|
||||
- Database size can be doubled or more.
|
||||
- Each insert/update/delete happen twice.
|
||||
- Migrations require a regenerate process that increases time considerably.
|
||||
|
||||
The advantages are:
|
||||
|
||||
- Offline-first sync.
|
||||
- No performance cost on read operations.
|
||||
- Simple offline app development.
|
||||
- Invariant safety - no weird states in the data.
|
||||
- Plug and play to any existing offline app.
|
||||
- No dealing with remote-aware repositories.
|
||||
|
||||
## License
|
||||
|
||||
This package is distributed under the MIT License. See the LICENSE file for
|
||||
more information.
|
||||
132
legacy/drift_offline_sync/analysis_options.yaml
Normal file
132
legacy/drift_offline_sync/analysis_options.yaml
Normal file
|
|
@ -0,0 +1,132 @@
|
|||
include: package:lints/recommended.yaml
|
||||
|
||||
linter:
|
||||
rules:
|
||||
- always_declare_return_types
|
||||
- always_put_required_named_parameters_first
|
||||
- annotate_redeclares
|
||||
- avoid_bool_literals_in_conditional_expressions
|
||||
- avoid_catches_without_on_clauses
|
||||
- avoid_catching_errors
|
||||
- avoid_empty_else
|
||||
- avoid_escaping_inner_quotes
|
||||
- avoid_field_initializers_in_const_classes
|
||||
- avoid_final_parameters
|
||||
- avoid_js_rounded_ints
|
||||
- avoid_positional_boolean_parameters
|
||||
- avoid_private_typedef_functions
|
||||
- avoid_relative_lib_imports
|
||||
- avoid_returning_this
|
||||
- avoid_setters_without_getters
|
||||
- avoid_shadowing_type_parameters
|
||||
- avoid_slow_async_io
|
||||
- avoid_types_as_parameter_names
|
||||
- avoid_unused_constructor_parameters
|
||||
- avoid_void_async
|
||||
- await_only_futures
|
||||
- camel_case_extensions
|
||||
- cancel_subscriptions
|
||||
- cascade_invocations
|
||||
- cast_nullable_to_non_nullable
|
||||
- close_sinks
|
||||
- collection_methods_unrelated_type
|
||||
- combinators_ordering
|
||||
- comment_references
|
||||
- conditional_uri_does_not_exist
|
||||
- curly_braces_in_flow_control_structures
|
||||
- dangling_library_doc_comments
|
||||
- depend_on_referenced_packages
|
||||
- deprecated_consistency
|
||||
- deprecated_member_use_from_same_package
|
||||
- directives_ordering
|
||||
- discarded_futures
|
||||
- document_ignores
|
||||
- empty_catches
|
||||
- eol_at_end_of_file
|
||||
- file_names
|
||||
- hash_and_equals
|
||||
- implicit_call_tearoffs
|
||||
- invalid_case_patterns
|
||||
- library_names
|
||||
- literal_only_boolean_expressions
|
||||
- matching_super_parameters
|
||||
- missing_code_block_language_in_doc_comment
|
||||
- missing_whitespace_between_adjacent_strings
|
||||
- no_adjacent_strings_in_list
|
||||
- no_duplicate_case_values
|
||||
- no_literal_bool_comparisons
|
||||
- no_runtimeType_toString
|
||||
- no_self_assignments
|
||||
- no_wildcard_variable_uses
|
||||
- non_constant_identifier_names
|
||||
- noop_primitive_operations
|
||||
- null_check_on_nullable_type_parameter
|
||||
- omit_local_variable_types
|
||||
- only_throw_errors
|
||||
- package_prefixed_library_names
|
||||
- prefer_const_constructors
|
||||
- prefer_const_declarations
|
||||
- prefer_final_in_for_each
|
||||
- prefer_final_locals
|
||||
- prefer_foreach
|
||||
- prefer_if_elements_to_conditional_expressions
|
||||
- prefer_is_empty
|
||||
- prefer_is_not_empty
|
||||
- prefer_iterable_whereType
|
||||
- prefer_mixin
|
||||
- prefer_null_aware_method_calls
|
||||
- prefer_relative_imports
|
||||
- prefer_single_quotes
|
||||
- prefer_typing_uninitialized_variables
|
||||
- prefer_void_to_null
|
||||
- provide_deprecation_message
|
||||
- public_member_api_docs
|
||||
- require_trailing_commas
|
||||
- sized_box_shrink_expand
|
||||
- sort_constructors_first
|
||||
- sort_unnamed_constructors_first
|
||||
- test_types_in_equals
|
||||
- throw_in_finally
|
||||
- tighten_type_of_initializing_formals
|
||||
- type_annotate_public_apis
|
||||
- type_literal_in_constant_pattern
|
||||
- unawaited_futures
|
||||
- unintended_html_in_doc_comment
|
||||
- unnecessary_await_in_return
|
||||
- unnecessary_breaks
|
||||
- unnecessary_lambdas
|
||||
- unnecessary_library_directive
|
||||
- unnecessary_null_aware_operator_on_extension_on_nullable
|
||||
- unnecessary_null_checks
|
||||
- unnecessary_overrides
|
||||
- unnecessary_parenthesis
|
||||
- unnecessary_raw_strings
|
||||
- unnecessary_statements
|
||||
- unreachable_from_main
|
||||
- unrelated_type_equality_checks
|
||||
- use_colored_box
|
||||
- use_decorated_box
|
||||
- use_enums
|
||||
- use_if_null_to_convert_nulls_to_bools
|
||||
- use_is_even_rather_than_modulo
|
||||
- use_named_constants
|
||||
- use_raw_strings
|
||||
- use_setters_to_change_properties
|
||||
- use_string_buffers
|
||||
- use_string_in_part_of_directives
|
||||
- use_test_throws_matchers
|
||||
- use_to_and_as_if_applicable
|
||||
- valid_regexps
|
||||
- void_checks
|
||||
|
||||
formatter:
|
||||
page_width: 88
|
||||
trailing_commas: preserve
|
||||
|
||||
analyzer:
|
||||
exclude:
|
||||
- "**/*.g.dart"
|
||||
- "**/*.drift.dart"
|
||||
language:
|
||||
strict-casts: true
|
||||
strict-inference: true
|
||||
62
legacy/drift_offline_sync/benchmark/README.md
Normal file
62
legacy/drift_offline_sync/benchmark/README.md
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
# Benchmarks
|
||||
|
||||
This folder contains benchmarks to measure the performance impact of using the `drift_offline_sync` package on database operations.
|
||||
|
||||
## What is Measured
|
||||
|
||||
The benchmarks measure two key metrics:
|
||||
|
||||
1. **Insert/Update/Delete Performance**: Time required to insert/update/delete
|
||||
a large number of rows with many columns.
|
||||
2. **Storage Overhead**: Database file size after inserting a large number of
|
||||
rows with many columns.
|
||||
|
||||
The benchmarks are run in a separate isolate to avoid interference with the main
|
||||
process. A baseline benchmark is run first to establish a reference point and
|
||||
then a CRDT-enabled benchmark is run to measure the performance impact of using
|
||||
the `drift_offline_sync` package.
|
||||
|
||||
## Running the Benchmarks
|
||||
|
||||
To run the benchmarks:
|
||||
|
||||
```bash
|
||||
dart benchmark/run.dart
|
||||
```
|
||||
|
||||
## Accessing Benchmark Results
|
||||
|
||||
When running from the CI, benchmark results are automatically stored as git notes on commits. If running on a pull-request, it is also pasted as a comment that is kept up to date on each push. To access historical benchmark data:
|
||||
|
||||
```bash
|
||||
# Fetch benchmark notes from remote (if not already available)
|
||||
git fetch origin refs/notes/benchmarks:refs/notes/benchmarks
|
||||
|
||||
# View benchmark results for the latest commit
|
||||
git notes --ref=benchmarks show
|
||||
|
||||
# View benchmark results for a specific commit
|
||||
git notes --ref=benchmarks show <commit-sha>
|
||||
|
||||
# View all commits with their benchmark results
|
||||
git log --show-notes=benchmark
|
||||
```
|
||||
|
||||
## Benchmark Details
|
||||
|
||||
### Table Structure
|
||||
|
||||
The benchmarks use `TableWithEveryColumnType` which includes:
|
||||
- Boolean (`aBool`)
|
||||
- DateTime (`aDateTime`)
|
||||
- Text (`aText`)
|
||||
- Integer (`anInt`)
|
||||
- Int64 (`anInt64`)
|
||||
- Real/Double (`aReal`)
|
||||
- Blob (`aBlob`)
|
||||
- Integer Enum (`anIntEnum`)
|
||||
- Text with Converter (`aTextWithConverter`)
|
||||
- UUID (`aUuid`)
|
||||
|
||||
This comprehensive table structure ensures the benchmarks reflect realistic use
|
||||
cases with various data types.
|
||||
93
legacy/drift_offline_sync/benchmark/run.dart
Normal file
93
legacy/drift_offline_sync/benchmark/run.dart
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
import 'utils/benchmark.dart';
|
||||
import 'utils/conversion.dart';
|
||||
import 'utils/results.dart';
|
||||
import 'utils/runner.dart';
|
||||
|
||||
Future<void> main(List<String> args) async {
|
||||
final runningInCI = args.contains('--ci');
|
||||
final rowCountArg = args.where((arg) => arg.startsWith('--rows=')).firstOrNull;
|
||||
final rowCount = rowCountArg == null ? 1000 : int.parse(rowCountArg.substring(7));
|
||||
final rowsString = formatter0.format(rowCount);
|
||||
|
||||
if (!runningInCI) print('\n${'=' * 60}');
|
||||
print('## Performance Benchmark ($rowsString rows)\n');
|
||||
print(
|
||||
[
|
||||
'This benchmark measures the performance impact of using CRDT ',
|
||||
'synchronization on database operations.',
|
||||
].join(runningInCI ? '' : '\n'),
|
||||
);
|
||||
if (!runningInCI) print('-' * 60);
|
||||
|
||||
final benchmarkResults = <BenchmarkResults>[];
|
||||
|
||||
for (final operation in Operation.values) {
|
||||
final baselineResult = await runWithProgress(
|
||||
'Running ${operation.name} benchmark (baseline)',
|
||||
CrdtBenchmark(
|
||||
'${operation.name} (baseline)',
|
||||
crdtEnabled: false,
|
||||
operation: operation,
|
||||
rowCount: rowCount,
|
||||
).report,
|
||||
skipProgress: runningInCI,
|
||||
);
|
||||
|
||||
final crdtResult = await runWithProgress(
|
||||
'Running ${operation.name} benchmark (CRDT)',
|
||||
CrdtBenchmark(
|
||||
'${operation.name} (CRDT)',
|
||||
crdtEnabled: true,
|
||||
operation: operation,
|
||||
rowCount: rowCount,
|
||||
).report,
|
||||
validator: (result) => result.$1 > baselineResult.$1,
|
||||
skipProgress: runningInCI,
|
||||
);
|
||||
|
||||
benchmarkResults.add(
|
||||
BenchmarkResults(
|
||||
operation: operation,
|
||||
baseline: baselineResult,
|
||||
crdt: crdtResult,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
final spuriousBenchmarks = benchmarkResults.where(
|
||||
(result) => result.baseline.$1 > result.crdt.$1,
|
||||
);
|
||||
|
||||
if (spuriousBenchmarks.isNotEmpty) {
|
||||
final leadingText = runningInCI ? '\n> ' : '\n';
|
||||
print(
|
||||
[
|
||||
'$leadingText❌ Baseline time is greater than CRDT time. This is likely due ',
|
||||
'to running with a very low number of rows ($rowsString). This is not ',
|
||||
' a valid benchmark and the time comparison should be ignored.',
|
||||
].join(runningInCI ? '' : '\n'),
|
||||
);
|
||||
|
||||
for (final result in spuriousBenchmarks) {
|
||||
final baselineDelay = result.baseline.$1.toFormattedDuration();
|
||||
final crdtDelay = result.crdt.$1.toFormattedDuration();
|
||||
print(' - ${result.operation.name.toUpperCase()}: $baselineDelay > $crdtDelay');
|
||||
}
|
||||
}
|
||||
|
||||
for (final result in benchmarkResults) {
|
||||
printPerformanceImpact(
|
||||
result,
|
||||
rowCount: rowCount,
|
||||
runningInCI: runningInCI,
|
||||
);
|
||||
}
|
||||
printStorageImpact(
|
||||
benchmarkResults.first,
|
||||
rowCount: rowCount,
|
||||
runningInCI: runningInCI,
|
||||
);
|
||||
|
||||
print('\n${'-' * 60}');
|
||||
print('✅ Benchmark complete!\n');
|
||||
}
|
||||
99
legacy/drift_offline_sync/benchmark/utils/benchmark.dart
Normal file
99
legacy/drift_offline_sync/benchmark/utils/benchmark.dart
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
import 'dart:io';
|
||||
|
||||
import 'package:benchmark_harness/benchmark_harness.dart';
|
||||
import 'package:drift_offline_sync/src/database/triggers.dart';
|
||||
import 'package:path/path.dart' as p;
|
||||
|
||||
import 'database.dart';
|
||||
|
||||
export '../../lib/src/database/triggers.dart' show Operation;
|
||||
|
||||
class CrdtBenchmark extends AsyncBenchmarkBase {
|
||||
CrdtBenchmark(
|
||||
super.name, {
|
||||
required this.crdtEnabled,
|
||||
required this.operation,
|
||||
required this.rowCount,
|
||||
});
|
||||
|
||||
final bool crdtEnabled;
|
||||
final Operation operation;
|
||||
final int rowCount;
|
||||
|
||||
late final CrdtBenchmarkDatabase _db;
|
||||
late final File _dbFile;
|
||||
|
||||
int _lastDatabaseSize = 0;
|
||||
int _lastRowsCount = 0;
|
||||
|
||||
@override
|
||||
Future<void> setup() async {
|
||||
final dbPath = p.join(Directory.systemTemp.path, 'benchmark_$name.db');
|
||||
_dbFile = File(dbPath);
|
||||
_db = CrdtBenchmarkDatabase.createDatabase(
|
||||
_dbFile,
|
||||
crdtEnabled: crdtEnabled,
|
||||
);
|
||||
|
||||
// Force the migration to be applied during setup.
|
||||
await _db.batchInsertTestRows(1);
|
||||
await _db.deleteTestRows(1);
|
||||
await _db.managers.tableWithEveryColumnType.delete();
|
||||
|
||||
if (operation != Operation.insert) {
|
||||
await _db.batchInsertTestRows(rowCount);
|
||||
}
|
||||
}
|
||||
|
||||
/// The default warmup is to invoke the [run] method once, but this pollutes the
|
||||
/// results by persisting data before the run.
|
||||
@override
|
||||
Future<void> warmup() async {}
|
||||
|
||||
@override
|
||||
Future<void> run() async {
|
||||
_lastDatabaseSize = 0;
|
||||
_lastRowsCount = 0;
|
||||
switch (operation) {
|
||||
case Operation.insert:
|
||||
case Operation.update:
|
||||
await _db.upsertTestRows(rowCount);
|
||||
case Operation.delete:
|
||||
await _db.deleteTestRows(rowCount);
|
||||
}
|
||||
}
|
||||
|
||||
@override
|
||||
Future<(double, int)> report() async {
|
||||
final averageMicroseconds = await measure();
|
||||
return (averageMicroseconds, measureStorage());
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> teardown() async {
|
||||
_lastDatabaseSize = _dbFile.lengthSync();
|
||||
_lastRowsCount = await _db.testRowsCount();
|
||||
await _db.close();
|
||||
if (_dbFile.existsSync()) {
|
||||
_dbFile.deleteSync();
|
||||
}
|
||||
}
|
||||
|
||||
int measureStorage() {
|
||||
if (operation == Operation.delete) {
|
||||
return _lastDatabaseSize;
|
||||
}
|
||||
|
||||
if (_lastDatabaseSize == 0) {
|
||||
throw Exception('Last database size is not set.');
|
||||
}
|
||||
if (_lastRowsCount < rowCount) {
|
||||
throw Exception(
|
||||
'Rows count is $_lastRowsCount but expected at least $rowCount.',
|
||||
);
|
||||
}
|
||||
// If the benchmark ran for more than one loop, it can contain a greater number
|
||||
// of rows than the expected. We need to normalize it to the expected total.
|
||||
return _lastDatabaseSize ~/ (_lastRowsCount / rowCount);
|
||||
}
|
||||
}
|
||||
48
legacy/drift_offline_sync/benchmark/utils/conversion.dart
Normal file
48
legacy/drift_offline_sync/benchmark/utils/conversion.dart
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
import 'package:intl/intl.dart';
|
||||
|
||||
final formatter0 = NumberFormat.decimalPatternDigits(decimalDigits: 0);
|
||||
final formatter2 = NumberFormat.decimalPatternDigits(decimalDigits: 2);
|
||||
final formatter3 = NumberFormat.decimalPatternDigits(decimalDigits: 3);
|
||||
|
||||
extension FormattedStorageSize on num {
|
||||
String toFormattedDuration() {
|
||||
var time = this;
|
||||
var unit = 1;
|
||||
|
||||
while (time.abs() > 1000 && unit < Duration.microsecondsPerSecond) {
|
||||
time /= 1000;
|
||||
unit *= 1000;
|
||||
}
|
||||
while (unit > Duration.microsecondsPerMinute && time.abs() > 60) {
|
||||
time /= 60;
|
||||
unit *= 60;
|
||||
}
|
||||
return '${formatter2.format(time)} ${switch (unit) {
|
||||
1 => 'μs',
|
||||
Duration.millisecondsPerSecond => 'ms',
|
||||
Duration.microsecondsPerSecond => 's',
|
||||
Duration.microsecondsPerMinute => 'min',
|
||||
Duration.microsecondsPerHour => 'h',
|
||||
_ => throw Exception('Unexpected time unit: $unit'),
|
||||
}}';
|
||||
}
|
||||
|
||||
String toFormattedStorageSize() {
|
||||
var storageIncrease = toDouble();
|
||||
var storageUnit = 1;
|
||||
|
||||
while (storageIncrease > 1024) {
|
||||
storageIncrease /= 1024;
|
||||
storageUnit *= 1024;
|
||||
}
|
||||
|
||||
final storageUnitString = switch (storageUnit) {
|
||||
1 => 'B',
|
||||
1024 => 'KB',
|
||||
const (1024 * 1024) => 'MB',
|
||||
_ => throw Exception('Unexpected storage unit: $storageUnit'),
|
||||
};
|
||||
|
||||
return '${formatter2.format(storageIncrease)} $storageUnitString';
|
||||
}
|
||||
}
|
||||
95
legacy/drift_offline_sync/benchmark/utils/database.dart
Normal file
95
legacy/drift_offline_sync/benchmark/utils/database.dart
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
import 'dart:io';
|
||||
|
||||
import 'package:drift/drift.dart';
|
||||
import 'package:drift/native.dart';
|
||||
import 'package:drift_offline_sync/drift_offline_sync.dart';
|
||||
import 'package:sqlite3/sqlite3.dart';
|
||||
import 'package:uuid/uuid.dart';
|
||||
|
||||
import '../../test/utils/tables.dart';
|
||||
|
||||
part 'database.g.dart';
|
||||
|
||||
@DriftDatabase(tables: [TableWithEveryColumnType])
|
||||
class CrdtBenchmarkDatabase extends _$CrdtBenchmarkDatabase {
|
||||
CrdtBenchmarkDatabase(super.e, {required this.crdtEnabled}) {
|
||||
driftRuntimeOptions.dontWarnAboutMultipleDatabases = true;
|
||||
}
|
||||
|
||||
static CrdtBenchmarkDatabase createDatabase(
|
||||
File file, {
|
||||
required bool crdtEnabled,
|
||||
}) {
|
||||
if (file.existsSync()) {
|
||||
file.deleteSync();
|
||||
}
|
||||
|
||||
return CrdtBenchmarkDatabase(
|
||||
NativeDatabase.opened(sqlite3.open(file.path)),
|
||||
crdtEnabled: crdtEnabled,
|
||||
);
|
||||
}
|
||||
|
||||
final bool crdtEnabled;
|
||||
|
||||
@override
|
||||
int get schemaVersion => 1;
|
||||
|
||||
@override
|
||||
Migrator createMigrator() => !crdtEnabled
|
||||
? super.createMigrator()
|
||||
: OfflineSyncMigrator(
|
||||
this,
|
||||
userId: const Uuid().v7(),
|
||||
nodeId: const Uuid().v7(),
|
||||
synchronizedTables: [tableWithEveryColumnType],
|
||||
);
|
||||
|
||||
Future<void> batchInsertTestRows(int count) async {
|
||||
final baseTimestamp = DateTime.now();
|
||||
await managers.tableWithEveryColumnType.bulkCreate(
|
||||
(_) => List.generate(count, (i) => _createTestRow(i, baseTimestamp)),
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> upsertTestRows(int count) async {
|
||||
final baseTimestamp = DateTime.now();
|
||||
for (var i = 0; i < count; i++) {
|
||||
await managers.tableWithEveryColumnType.create(
|
||||
(_) => _createTestRow(i, baseTimestamp),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> deleteTestRows(int count) async {
|
||||
for (var i = 0; i < count; i++) {
|
||||
await managers.tableWithEveryColumnType
|
||||
.filter((t) => t.id.equals(RowId(i)))
|
||||
.delete();
|
||||
}
|
||||
}
|
||||
|
||||
Future<int> testRowsCount() async {
|
||||
return managers.tableWithEveryColumnType.count();
|
||||
}
|
||||
|
||||
TableWithEveryColumnTypeCompanion _createTestRow(
|
||||
int i,
|
||||
DateTime baseTimestamp,
|
||||
) {
|
||||
return TableWithEveryColumnTypeCompanion(
|
||||
id: Value(RowId(i)),
|
||||
aBool: Value(i.isEven),
|
||||
aDateTime: Value(baseTimestamp.add(Duration(seconds: i))),
|
||||
aText: Value('Text $i'),
|
||||
anInt: Value(i),
|
||||
anInt64: Value(BigInt.from(i)),
|
||||
aReal: Value(i.toDouble()),
|
||||
aBlob: Value(Uint8List.fromList([i % 256])),
|
||||
anIntEnum: const Value(null),
|
||||
aTextWithConverter: const Value(null),
|
||||
aUuid: const Value(null),
|
||||
);
|
||||
}
|
||||
}
|
||||
1075
legacy/drift_offline_sync/benchmark/utils/database.g.dart
generated
Normal file
1075
legacy/drift_offline_sync/benchmark/utils/database.g.dart
generated
Normal file
File diff suppressed because it is too large
Load diff
64
legacy/drift_offline_sync/benchmark/utils/results.dart
Normal file
64
legacy/drift_offline_sync/benchmark/utils/results.dart
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
import 'benchmark.dart';
|
||||
import 'conversion.dart';
|
||||
|
||||
void printPerformanceImpact(
|
||||
BenchmarkResults benchmarkResults, {
|
||||
required int rowCount,
|
||||
bool runningInCI = false,
|
||||
}) {
|
||||
print(
|
||||
'${runningInCI ? '```' : ''}'
|
||||
'\n📊 ${benchmarkResults.operation.name.toUpperCase()} performance impact:',
|
||||
);
|
||||
final (baselineTime, _) = benchmarkResults.baseline;
|
||||
final (crdtTime, _) = benchmarkResults.crdt;
|
||||
|
||||
final runDelayUs = crdtTime - baselineTime;
|
||||
final slowdown = runDelayUs / baselineTime * 100;
|
||||
|
||||
final baselineDelay = baselineTime.toFormattedDuration();
|
||||
final crdtDelay = crdtTime.toFormattedDuration();
|
||||
final runDelay = runDelayUs.toFormattedDuration();
|
||||
final runDelayPerOperation = (runDelayUs / rowCount).toFormattedDuration();
|
||||
|
||||
print(' Time: $baselineDelay --> $crdtDelay ($runDelay)');
|
||||
print(' CRDT overhead: ${formatter2.format(slowdown)}% slower');
|
||||
print(' Delay per ${benchmarkResults.operation.name}: $runDelayPerOperation');
|
||||
if (runningInCI) print('```');
|
||||
}
|
||||
|
||||
void printStorageImpact(
|
||||
BenchmarkResults benchmarkResults, {
|
||||
required int rowCount,
|
||||
bool runningInCI = false,
|
||||
}) {
|
||||
print('${runningInCI ? '```' : ''}\n💽 Storage impact:');
|
||||
final (_, baselineSize) = benchmarkResults.baseline;
|
||||
final (_, crdtSize) = benchmarkResults.crdt;
|
||||
|
||||
final storageIncrease = crdtSize - baselineSize;
|
||||
final storageIncPercent = storageIncrease / baselineSize * 100;
|
||||
final extraStoragePerRow = storageIncrease / rowCount;
|
||||
final storageIncreaseVolume = storageIncPercent > 100
|
||||
? '(x${formatter2.format(storageIncPercent / 100)})'
|
||||
: '(+${(crdtSize - baselineSize).toFormattedStorageSize()})';
|
||||
print(
|
||||
' Storage size: ${baselineSize.toFormattedStorageSize()} '
|
||||
'--> ${crdtSize.toFormattedStorageSize()} $storageIncreaseVolume',
|
||||
);
|
||||
print(' CRDT overhead: ${formatter2.format(storageIncPercent)}% increase');
|
||||
print(' Extra storage per row: ${extraStoragePerRow.toFormattedStorageSize()}');
|
||||
if (runningInCI) print('```');
|
||||
}
|
||||
|
||||
class BenchmarkResults {
|
||||
const BenchmarkResults({
|
||||
required this.operation,
|
||||
required this.baseline,
|
||||
required this.crdt,
|
||||
});
|
||||
|
||||
final Operation operation;
|
||||
final (double, int) baseline;
|
||||
final (double, int) crdt;
|
||||
}
|
||||
39
legacy/drift_offline_sync/benchmark/utils/runner.dart
Normal file
39
legacy/drift_offline_sync/benchmark/utils/runner.dart
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
import 'dart:io';
|
||||
import 'dart:isolate';
|
||||
|
||||
import 'package:cli_tools/src/logger/helpers/progress.dart';
|
||||
|
||||
Progress? _trackedAnimationInProgress;
|
||||
|
||||
Future<T> runWithProgress<T>(
|
||||
String message,
|
||||
Future<T> Function() runner, {
|
||||
bool Function(T result)? validator,
|
||||
bool skipProgress = false,
|
||||
}) async {
|
||||
if (skipProgress) {
|
||||
return Isolate.run(runner);
|
||||
}
|
||||
|
||||
_stopAnimationInProgress();
|
||||
final progress = Progress(message, stdout);
|
||||
_trackedAnimationInProgress = progress;
|
||||
final result = await Isolate.run(runner);
|
||||
_trackedAnimationInProgress = null;
|
||||
if (validator?.call(result) ?? true) {
|
||||
progress.complete();
|
||||
} else {
|
||||
progress.fail();
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
void _stopAnimationInProgress() {
|
||||
if (_trackedAnimationInProgress != null) {
|
||||
_trackedAnimationInProgress?.stopAnimation();
|
||||
// Since animation modifies the current line we add a new line so that
|
||||
// the next print doesn't end up on the same line.
|
||||
stdout.write('\n');
|
||||
}
|
||||
_trackedAnimationInProgress = null;
|
||||
}
|
||||
17
legacy/drift_offline_sync/build.yaml
Normal file
17
legacy/drift_offline_sync/build.yaml
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
targets:
|
||||
$default:
|
||||
builders:
|
||||
drift_dev:
|
||||
options:
|
||||
store_date_time_values_as_text: true
|
||||
use_sql_column_name_as_json_key: true
|
||||
sql:
|
||||
dialects:
|
||||
- sqlite
|
||||
- postgres
|
||||
sources:
|
||||
include:
|
||||
- $package$
|
||||
- lib/**
|
||||
- test/**
|
||||
- benchmark/**
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
import 'package:drift_offline_sync/drift_offline_sync.dart';
|
||||
|
||||
void main() {
|
||||
print('awesome: $OfflineSyncMigrator');
|
||||
}
|
||||
17
legacy/drift_offline_sync/lib/drift_offline_sync.dart
Normal file
17
legacy/drift_offline_sync/lib/drift_offline_sync.dart
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
/// Support for doing something awesome.
|
||||
///
|
||||
/// More docs go here.
|
||||
library;
|
||||
|
||||
export 'src/crdt/base.dart';
|
||||
export 'src/crdt/crdt.dart';
|
||||
export 'src/database/database.dart';
|
||||
export 'src/database/extensions/convert.dart';
|
||||
export 'src/database/extensions/entry.dart';
|
||||
export 'src/database/extensions/extract.dart';
|
||||
export 'src/database/extensions/sync.dart';
|
||||
export 'src/database/extensions/triggers.dart';
|
||||
export 'src/hlc/converter.dart';
|
||||
export 'src/hlc/exceptions.dart';
|
||||
export 'src/hlc/hlc.dart';
|
||||
export 'src/migrator.dart';
|
||||
109
legacy/drift_offline_sync/lib/src/crdt/base.dart
Normal file
109
legacy/drift_offline_sync/lib/src/crdt/base.dart
Normal file
|
|
@ -0,0 +1,109 @@
|
|||
import 'dart:async';
|
||||
|
||||
import 'package:meta/meta.dart';
|
||||
|
||||
import '../database/database.dart';
|
||||
import '../hlc/hlc.dart';
|
||||
|
||||
/// Base class for CRDT implementations.
|
||||
///
|
||||
/// This class provides the basic functionality for a CRDT implementation.
|
||||
/// It is not meant to be used directly, but rather to be extended by concrete
|
||||
/// implementations.
|
||||
abstract class CrdtBase {
|
||||
/// Creates a new instance of [CrdtBase].
|
||||
CrdtBase({required this.userId, required this.nodeId});
|
||||
|
||||
/// The user ID for the CRDT system.
|
||||
final String userId;
|
||||
|
||||
/// The node ID of the CRDT.
|
||||
final String nodeId;
|
||||
|
||||
/// The current canonical time from the database (in-SQLite HLC state).
|
||||
Future<Hlc> getCanonicalTime();
|
||||
|
||||
/// Merge the [maxMergedHlc] into the canonical time.
|
||||
Future<void> mergeCanonicalTime(Hlc maxMergedHlc);
|
||||
|
||||
/// Returns the last modified timestamp, optionally filtering for or against a
|
||||
/// specific node id. Useful to get "modified since" timestamps for executing
|
||||
/// synchronization.
|
||||
Future<Hlc?> getLastModified({
|
||||
String? onlyNodeId,
|
||||
String? exceptNodeId,
|
||||
});
|
||||
|
||||
/// Get a list of [CrdtDataEntry] with the changes to be merged.
|
||||
///
|
||||
/// Set the filtering parameters to to generate subsets:
|
||||
/// [onlyNodeId] only records set by the specified node.
|
||||
/// [exceptNodeId] only records not set by the specified node.
|
||||
/// [modifiedAfter] records modified after the specified [Hlc].
|
||||
Future<Iterable<CrdtDataEntry>> getChangeset({
|
||||
String? onlyNodeId,
|
||||
String? exceptNodeId,
|
||||
Hlc? modifiedAfter,
|
||||
});
|
||||
|
||||
/// Save the [changeset] to the database.
|
||||
///
|
||||
/// When saving, must be careful to only override values that are older than
|
||||
/// the [Hlc] of the changeset entry. If there are any guardrails to be
|
||||
/// applied to the saved data, they should be applied here.
|
||||
///
|
||||
/// The [receivedHlcs] is a list of HLCs, one for each external node. It must
|
||||
/// be saved separately since it will be used to request updates from external
|
||||
/// nodes on next syncs.
|
||||
///
|
||||
/// It is recommended to use a transaction to ensure the integrity of the
|
||||
/// data and allow for rollback in case of errors.
|
||||
///
|
||||
/// Note that this method is only meant to be used internally by the [merge]
|
||||
/// method and should not be called directly.
|
||||
@protected
|
||||
Future<void> saveChangeset(
|
||||
Iterable<CrdtDataEntry> changeset,
|
||||
Iterable<Hlc> receivedHlcs,
|
||||
);
|
||||
|
||||
/// Merge [changeset] with the local dataset.
|
||||
///
|
||||
/// This operation must be done within a transaction to ensure that no other
|
||||
/// operation modifies the data between the save, invariants preservation and
|
||||
/// the merge of remote HLCs into the canonical time.
|
||||
Future<void> merge(Iterable<CrdtDataEntry> changeset) async {
|
||||
if (changeset.isEmpty) return;
|
||||
|
||||
final receivedHlcs = _extractLastReceivedHlcs(changeset);
|
||||
await saveChangeset(changeset, receivedHlcs);
|
||||
|
||||
if (receivedHlcs.isEmpty) return;
|
||||
await mergeCanonicalTime(receivedHlcs.max);
|
||||
}
|
||||
|
||||
/// Extracts the last received HLCs from the changeset.
|
||||
Iterable<Hlc> _extractLastReceivedHlcs(Iterable<CrdtDataEntry> changeset) {
|
||||
final receivedHlcs = <String, Hlc>{};
|
||||
for (final entry in changeset) {
|
||||
if (entry.hlcTimestamp.nodeId == nodeId) continue;
|
||||
receivedHlcs.addOrUpdate(entry.hlcTimestamp);
|
||||
}
|
||||
return receivedHlcs.values;
|
||||
}
|
||||
}
|
||||
|
||||
extension on Iterable<Hlc> {
|
||||
Hlc get max => reduce((a, b) => a > b ? a : b);
|
||||
}
|
||||
|
||||
extension on Map<String, Hlc> {
|
||||
void addOrUpdate(Hlc hlc) {
|
||||
final registeredHlc = this[hlc.nodeId];
|
||||
if (registeredHlc == null) {
|
||||
this[hlc.nodeId] = hlc;
|
||||
} else if (registeredHlc < hlc) {
|
||||
this[hlc.nodeId] = hlc;
|
||||
}
|
||||
}
|
||||
}
|
||||
190
legacy/drift_offline_sync/lib/src/crdt/crdt.dart
Normal file
190
legacy/drift_offline_sync/lib/src/crdt/crdt.dart
Normal file
|
|
@ -0,0 +1,190 @@
|
|||
import 'package:drift/drift.dart';
|
||||
|
||||
import '../database/database.dart';
|
||||
import '../database/extensions/entry.dart';
|
||||
import '../hlc/converter.dart';
|
||||
import 'base.dart';
|
||||
|
||||
/// A CRDT implementation for offline-first synchronization.
|
||||
class OfflineSyncCrdt extends CrdtBase {
|
||||
/// Creates a new instance of [OfflineSyncCrdt].
|
||||
///
|
||||
/// The [nodeId] must be unique across the system.
|
||||
OfflineSyncCrdt(this._db, {required super.userId, required super.nodeId});
|
||||
|
||||
final CrdtDatabase _db;
|
||||
|
||||
@override
|
||||
Future<Hlc> getCanonicalTime() async {
|
||||
final row = await _db.managers.crdtHlcStateTable
|
||||
.filter((o) => o.userId.equals(userId))
|
||||
.getSingleOrNull();
|
||||
|
||||
return row?.toHlc(nodeId) ?? Hlc.zero(nodeId);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> mergeCanonicalTime(Hlc maxMergedHlc) async {
|
||||
await _db.managers.crdtHlcStateTable.create(
|
||||
(t) => t(
|
||||
userId: userId,
|
||||
lastTimestamp: maxMergedHlc.unixTimestamp,
|
||||
counter: maxMergedHlc.counter,
|
||||
),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
onConflict: DoUpdate.withExcluded(
|
||||
(old, excluded) => CrdtHlcStateTableCompanion.custom(
|
||||
lastTimestamp: CaseWhenExpression(
|
||||
// Combined with the `orElse`, this is the same as a max() function,
|
||||
// but it works for all databases. Otherwise, we would need to check
|
||||
// the database dialect to use either `MAX` or `GREATEST`.
|
||||
cases: [
|
||||
excluded.lastTimestamp
|
||||
.isBiggerOrEqual(old.lastTimestamp)
|
||||
.then(excluded.lastTimestamp),
|
||||
],
|
||||
orElse: old.lastTimestamp,
|
||||
),
|
||||
counter: CaseWhenExpression(
|
||||
cases: [
|
||||
// If one timestamp is greater, its counter should be persisted.
|
||||
excluded.lastTimestamp
|
||||
.isBiggerThan(old.lastTimestamp)
|
||||
.then(excluded.counter),
|
||||
// Otherwise, if the timestamps are equal, the counter should be
|
||||
// the greater counter.
|
||||
(excluded.lastTimestamp.equalsExp(old.lastTimestamp) &
|
||||
excluded.counter.isBiggerThan(old.counter))
|
||||
.then(excluded.counter),
|
||||
],
|
||||
orElse: old.counter,
|
||||
),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
// TODO: Remove this method lately if it is not needed - as it seems it will.
|
||||
Future<Hlc?> getLastModified({String? onlyNodeId, String? exceptNodeId}) async {
|
||||
if (onlyNodeId != null && exceptNodeId != null) {
|
||||
throw ArgumentError('Only one of onlyNodeId or exceptNodeId can be provided.');
|
||||
}
|
||||
|
||||
// NOTE: Although `max` has a better performance for the general use case,
|
||||
// when the column is indexed, `order by` with `limit 1` will be just as fast.
|
||||
// The plus of using the order by approach is the native deserialization of
|
||||
// the Hlc object by Drift.
|
||||
final rowWithMaxLastSyncHlc = await _db.managers.crdtDataTable
|
||||
.filter((o) {
|
||||
if (onlyNodeId != null) {
|
||||
return o.hlcTimestamp.column.contains(onlyNodeId);
|
||||
}
|
||||
if (exceptNodeId != null) {
|
||||
return o.hlcTimestamp.column.contains(exceptNodeId).not();
|
||||
}
|
||||
return const Constant(true);
|
||||
})
|
||||
.orderBy((o) => o.hlcTimestamp.desc())
|
||||
.limit(1)
|
||||
.getSingleOrNull();
|
||||
|
||||
return rowWithMaxLastSyncHlc?.hlcTimestamp;
|
||||
}
|
||||
|
||||
@override
|
||||
Future<Iterable<CrdtDataEntry>> getChangeset({
|
||||
String? onlyNodeId,
|
||||
String? exceptNodeId,
|
||||
Hlc? modifiedAfter,
|
||||
}) async {
|
||||
assert(onlyNodeId == null || exceptNodeId == null);
|
||||
|
||||
// Modified times use the local node id
|
||||
modifiedAfter = modifiedAfter?.copyWith(nodeId: nodeId);
|
||||
|
||||
final query = _db.managers.crdtDataTable.filter((o) {
|
||||
Expression<bool> condition = const Constant(true);
|
||||
|
||||
if (onlyNodeId != null) {
|
||||
condition &= o.hlcTimestamp.column.contains(onlyNodeId);
|
||||
}
|
||||
if (exceptNodeId != null) {
|
||||
condition &= o.hlcTimestamp.column.contains(exceptNodeId).not();
|
||||
}
|
||||
if (modifiedAfter != null) {
|
||||
condition &= o.hlcTimestamp.column.isBiggerThanValue(
|
||||
hlcConverter.toSql(modifiedAfter),
|
||||
);
|
||||
}
|
||||
return condition;
|
||||
});
|
||||
|
||||
return query.get();
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> saveChangeset(
|
||||
Iterable<CrdtDataEntry> changeset,
|
||||
Iterable<Hlc> receivedHlcs,
|
||||
) async {
|
||||
await _db.transaction(() async {
|
||||
await _saveChangeset(changeset);
|
||||
await _applyCompensationRules();
|
||||
await _saveMergedHlcs(receivedHlcs);
|
||||
});
|
||||
}
|
||||
|
||||
/// Save the [changeset] to the database.
|
||||
///
|
||||
/// Should be called within a transaction. Must ensure that only newer values
|
||||
/// are saved to the database.
|
||||
Future<void> _saveChangeset(Iterable<CrdtDataEntry> changeset) async {
|
||||
// Pure inserts; conflict resolution (newer HLC wins) is handled by the
|
||||
// INSTEAD OF INSERT trigger on the CRDT data view.
|
||||
return _db.managers.crdtDataTable.bulkCreate(
|
||||
(_) => changeset.map((e) => e.toCompanion(false)),
|
||||
mode: InsertMode.insert,
|
||||
);
|
||||
}
|
||||
|
||||
/// Applies compensation rules to the CRDT data table to preserve invariants.
|
||||
///
|
||||
/// If the any invariant fails to be preserved, a flag will be set on the
|
||||
/// control table to prevent further merges until the issue is resolved by
|
||||
/// the server and a repair merge is pushed to this node.
|
||||
Future<void> _applyCompensationRules() async {
|
||||
// TODO: Implement this.
|
||||
}
|
||||
|
||||
/// Save the [mergedHlcs] to the database.
|
||||
///
|
||||
/// Should be called within a transaction after the changeset has been saved
|
||||
/// and properly compensated.
|
||||
Future<void> _saveMergedHlcs(Iterable<Hlc> mergedHlcs) async {
|
||||
return _db.managers.crdtMergeHlcTable.bulkCreate(
|
||||
(t) => mergedHlcs.map(
|
||||
(e) => t(
|
||||
userId: userId,
|
||||
nodeId: e.nodeId,
|
||||
lastReceivedHlc: e,
|
||||
),
|
||||
),
|
||||
mode: InsertMode.insertOrReplace,
|
||||
onConflict: DoUpdate.withExcluded(
|
||||
(old, excluded) => CrdtMergeHlcTableCompanion.custom(
|
||||
lastReceivedHlc: excluded.lastReceivedHlc,
|
||||
),
|
||||
where: (old, excluded) =>
|
||||
old.lastReceivedHlc.isSmallerThan(excluded.lastReceivedHlc),
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
extension on Expression<bool> {
|
||||
CaseWhen<bool, T> then<T extends Object>(Expression<T> value) => CaseWhen(
|
||||
this,
|
||||
then: value,
|
||||
);
|
||||
}
|
||||
129
legacy/drift_offline_sync/lib/src/database/database.dart
Normal file
129
legacy/drift_offline_sync/lib/src/database/database.dart
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
import 'package:drift/drift.dart';
|
||||
|
||||
import '../utils/sql_builder.dart';
|
||||
import 'tables/control.dart';
|
||||
import 'tables/data.dart';
|
||||
import 'tables/hlc.dart';
|
||||
import 'tables/merge.dart';
|
||||
|
||||
part 'database.g.dart';
|
||||
|
||||
/// The main database class for CRDT synchronization.
|
||||
@DriftDatabase(
|
||||
tables: [
|
||||
CrdtDataTable,
|
||||
CrdtNormalizedDataTable,
|
||||
CrdtControlTable,
|
||||
CrdtHlcStateTable,
|
||||
CrdtMergeHlcTable,
|
||||
],
|
||||
)
|
||||
class CrdtDatabase extends _$CrdtDatabase {
|
||||
/// Creates a new instance of [CrdtDatabase] with the provided [executor].
|
||||
CrdtDatabase(
|
||||
this.userDatabase, {
|
||||
required this.userId,
|
||||
required this.nodeId,
|
||||
required this.synchronizedTables,
|
||||
}) : super(
|
||||
userDatabase.executor.interceptWith(
|
||||
_EnsureHlcInitialized(userDatabase),
|
||||
),
|
||||
);
|
||||
|
||||
/// The user database to synchronize with.
|
||||
final GeneratedDatabase userDatabase;
|
||||
|
||||
/// The user ID for the CRDT system.
|
||||
final String userId;
|
||||
|
||||
/// The node ID for the CRDT system.
|
||||
final String nodeId;
|
||||
|
||||
/// Tables to be synchronized with CRDT.
|
||||
final List<TableInfo> synchronizedTables;
|
||||
|
||||
/// The SQL builder for the CRDT data table.
|
||||
late final sqlBuilder = CrdtDataSqlBuilder(this);
|
||||
|
||||
/// Whether the database is using the SQLite or Postgres dialect.
|
||||
bool get isPostgres => executor.dialect == SqlDialect.postgres;
|
||||
|
||||
/// Not actually used, but required by the generated code. Will be ignored because
|
||||
/// the [CrdtDatabase] always receives an already opened executor. No migrations
|
||||
/// should ever be run on this database to avoid messing with the user schema.
|
||||
@override
|
||||
int schemaVersion = 1;
|
||||
}
|
||||
|
||||
class _EnsureHlcInitialized extends QueryInterceptor {
|
||||
_EnsureHlcInitialized(this.userDatabase);
|
||||
|
||||
final GeneratedDatabase userDatabase;
|
||||
|
||||
@override
|
||||
Future<bool> ensureOpen(QueryExecutor executor, QueryExecutorUser user) async {
|
||||
return executor.ensureOpen(_EnsureHlcInitializedUser(user, userDatabase));
|
||||
}
|
||||
}
|
||||
|
||||
class _EnsureHlcInitializedUser implements QueryExecutorUser {
|
||||
_EnsureHlcInitializedUser(this.user, this.userDatabase);
|
||||
|
||||
final QueryExecutorUser user;
|
||||
final GeneratedDatabase userDatabase;
|
||||
|
||||
/// This must be set to the [schemaVersion] of the [userDatabase] to ensure
|
||||
/// that the `PRAGMA user_version` is correctly updated after migrations are
|
||||
/// applied. Otherwise, the saved version will match the CRDT database version.
|
||||
@override
|
||||
int get schemaVersion => userDatabase.schemaVersion;
|
||||
|
||||
/// If this method is reached, it means that the CRDT database is being opened
|
||||
/// before the user database. This method will ensure that the user database is
|
||||
/// always opened first. When, on the opposite, the user database is opened before
|
||||
/// the CRDT database, this method will never be called.
|
||||
@override
|
||||
Future<void> beforeOpen(QueryExecutor executor, OpeningDetails details) async {
|
||||
final userDetails = OpeningDetails(
|
||||
details.versionBefore,
|
||||
userDatabase.schemaVersion,
|
||||
);
|
||||
await userDatabase.beforeOpen(executor, userDetails);
|
||||
|
||||
// This means an explicit opt-out from Drift managed migrations. Since the
|
||||
// [userDatabase] will be migrated first, by using its [schemaVersion] on
|
||||
// the [crdtDetails], the CRDT database will skip migrations entirely. If
|
||||
// migrations are needed for the CRDT database in the future, they should
|
||||
// be hooked manually on this method.
|
||||
final crdtDetails = OpeningDetails(
|
||||
userDatabase.schemaVersion,
|
||||
userDatabase.schemaVersion,
|
||||
);
|
||||
await user.beforeOpen(executor, crdtDetails);
|
||||
}
|
||||
}
|
||||
|
||||
/// Extension methods for the [List<TableInfo>] class to find a synchronized table.
|
||||
extension FindSynchronizedTable on List<TableInfo> {
|
||||
/// Gets the table names of the synchronized tables.
|
||||
Iterable<String> get tableNames => map((t) => t.actualTableName);
|
||||
|
||||
/// Gets the synchronized table information for the given table name.
|
||||
///
|
||||
/// If the table is not found, an [ArgumentError] is thrown.
|
||||
TableInfo find<T extends Insertable>([String? tableName]) {
|
||||
if (tableName == null && T == dynamic) {
|
||||
throw ArgumentError('Table name is required when T is dynamic.');
|
||||
}
|
||||
return firstWhere(
|
||||
(t) =>
|
||||
(tableName == null && t is TableInfo<Table, T>) ||
|
||||
t.actualTableName == tableName,
|
||||
orElse: () => throw ArgumentError(
|
||||
'Table ${tableName != null ? '"$tableName"' : 'for type "$T"'} not '
|
||||
'found in synchronized tables.',
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
Some files were not shown because too many files have changed in this diff Show more
Loading…
Reference in a new issue