Persistence Store

A Dart port of Store5 - A Kotlin Multiplatform library for building network-resilient applications.

Features

  • Type-Safe Caching: In-memory caching with LRU eviction
  • Source of Truth: Swappable persistence layer (Hive, ObjectBox, Drift, etc.)
  • Request Deduplication: Automatic coalescing of concurrent requests
  • Reactive Streams: Built on Dart Streams and Futures
  • Functional Error Handling: Using FpDart's Either and Option types
  • RxDart Integration: Advanced reactive operators

Core Types

  • ✅ StoreKey - Single, Collection, Page, and Cursor key types
  • ✅ StoreData - Single and Collection data types with insertion strategies
  • ✅ InsertionStrategy - Append, Prepend, Replace
  • ✅ KeyProvider - Bidirectional key conversion
  • ✅ ExperimentalStoreApi - Annotation for experimental features

Response Types

  • ✅ StoreReadResponse - Sealed class hierarchy for responses
    • Initial - Initial state
    • Loading - Loading state with origin tracking
    • Data - Successful data response
    • NoNewData - Empty response from fetcher
    • Error - Exception, Message, and Custom error types
  • ✅ StoreReadResponseOrigin - Origin tracking (Cache, SourceOfTruth, Fetcher, Initial)

Store Interface

  • ✅ Store - Main interface combining read and clear operations
  • ✅ StoreRead - Stream-based reactive data access
  • ✅ StoreClear - Cache clearing operations

Fetcher System

  • ✅ Fetcher - Network data fetching abstraction
    • Factory methods: of, ofStream, ofResult, ofResultStream
    • Fallback support: withFallback, ofStreamWithFallback
  • ✅ FetcherResult - Result type for fetcher operations
    • Data - Successful fetch with optional origin
    • Error - Exception, Message, and Custom error types
    • FpDart integration (Either, Option)

Request Configuration

  • ✅ StoreReadRequest - Flexible request configuration
    • fresh - Skip all caches, fetch fresh data
    • cached - Use cache, optionally refresh
    • localOnly - Never fetch, cache only
    • skipMemory - Skip memory cache, use disk
  • ✅ CacheType - Memory and Disk cache types

Source of Truth

  • ✅ SourceOfTruth - Database abstraction layer
    • Factory methods: of (Future-based), ofFlow (Stream-based)
    • Operations: reader, write, delete, deleteAll
    • Supports reactive database updates
    • Database-agnostic interface
  • ✅ Exception types:
    • SourceOfTruthWriteException - Write operation failures
    • SourceOfTruthReadException - Read operation failures

Validation

  • ✅ Validator - Custom validation for cached items
    • Factory: Validator.by for function-based validation
    • Async validation support
    • Used to determine if cached data should be refreshed

Cache System

  • ✅ Cache - In-memory cache interface
    • Operations: getIfPresent, getOrPut, put, putAll, invalidate, invalidateAll
    • Batch operations: getAllPresent(keys), getAllPresent()
    • Size tracking: size()
  • ✅ CacheBuilder - Fluent builder for cache configuration
    • Size limits: maximumSize(int), weigher(int, Weigher)
    • Time-based expiration: expireAfterWrite(Duration), expireAfterAccess(Duration)
    • Custom time source: ticker(Ticker) for testing
    • Concurrency: concurrencyLevel(int)
  • ✅ InMemoryCache - LRU cache implementation
    • Automatic eviction based on size or weight
    • Time-based expiration (write time and access time)
    • Efficient LRU ordering with LinkedHashMap

Supporting Types

  • ✅ Ticker - Nanosecond-precision time source
  • ✅ Weigher - Custom entry weight calculation
  • ✅ RemovalCause - Reason for cache entry removal (explicit, replaced, expired, size, collected)

Multicast System

  • ✅ Multicaster - Shares one upstream Stream with multiple downstream listeners
    • Lazy upstream activation (starts when first listener subscribes)
    • Automatic cleanup when all listeners cancel
    • Broadcast stream sharing for concurrent requests
    • Optional onEach callback for value inspection
  • ✅ FetcherController - Request deduplication for network fetches
    • Deduplicates concurrent requests for the same key
    • Tracks active fetch operations
    • Operations: getFetcher, cancelFetch, cancelAll
    • Status checks: isActive(key), activeCount

Installation

Add this to your pubspec.yaml:

dependencies:
  persistencestore:
    path: ../persistencestore  # Adjust path as needed

Usage

Basic Example

import 'package:persistencestore/persistencestore.dart';

// Define your models
class User extends StoreDataSingle<int> {
  @override
  final int id;
  final String name;
  final String email;
  
  const User(this.id, this.name, this.email);
}

// Create a fetcher
final userFetcher = Fetcher.of<int, User>(
  (userId) async {
    // Fetch from your API
    final response = await api.getUser(userId);
    return User(response.id, response.name, response.email);
  },
  name: 'user-fetcher',
);

// The Store interface
abstract class UserStore implements Store<int, User> {
  // Stream of responses with cache and network
  @override
  Stream<StoreReadResponse<User>> stream(StoreReadRequest<int> request);
  
  @override
  Future<void> clear(int key);
  
  @override
  Future<void> clearAll();
}

// Usage patterns
void example(Store<int, User> store) async {
  // Get fresh data from network
  final user = await store.fresh(123);
  print('User: ${user.name}');
  
  // Get cached data, refresh in background
  final cachedUser = await store.get(123, refresh: true);
  
  // Stream of responses
  store.stream(StoreReadRequest.cached(123, refresh: true))
    .listen((response) {
      switch (response) {
        case StoreReadResponseLoading(:final origin):
          print('Loading from $origin');
        case StoreReadResponseData(:final value):
          print('Got ${value.name}');
        case StoreReadResponseError():
          print('Error: ${response.errorMessageOrNull()}');
        default:
          break;
      }
    });
}

Fetcher Examples

// Simple Future-based fetcher
final fetcher1 = Fetcher.of<String, Data>(
  (key) async => await api.fetch(key),
);

// Stream-based fetcher (e.g., WebSocket)
final fetcher2 = Fetcher.ofStream<String, Data>(
  (key) => websocket.stream(key),
);

// Fetcher with fallback
final fetcher3 = Fetcher.withFallback<String, Data>(
  name: 'primary',
  fetch: (key) async => await primaryApi.fetch(key),
  fallback: Fetcher.of((key) async => await backupApi.fetch(key)),
);

// Manual result handling
final fetcher4 = Fetcher.ofResult<String, Data>(
  (key) async {
    try {
      final data = await api.fetch(key);
      return FetcherResultData(data);
    } catch (e) {
      return FetcherResultErrorException(Exception(e.toString()));
    }
  },
);

Source of Truth Examples

// Future-based (non-reactive) database
final sot1 = SourceOfTruth.of<String, User, User>(
  nonFlowReader: (key) async {
    // Read from database
    return await database.getUser(key);
  },
  writer: (key, user) async {
    // Write to database
    await database.saveUser(key, user);
  },
  delete: (key) async {
    await database.deleteUser(key);
  },
  deleteAll: () async {
    await database.clearUsers();
  },
);

// Stream-based (reactive) database (e.g., Drift, Hive with watch)
final sot2 = SourceOfTruth.ofFlow<String, User, User>(
  reader: (key) {
    // Return reactive stream from database
    return database.watchUser(key);
  },
  writer: (key, user) async {
    await database.saveUser(key, user);
  },
  delete: (key) async {
    await database.deleteUser(key);
  },
);

// Type transformation example (Network -> Local)
final sot3 = SourceOfTruth.of<int, NetworkUser, LocalUser>(
  nonFlowReader: (userId) async {
    final local = await db.getUserById(userId);
    return local;
  },
  writer: (userId, networkUser) async {
    // Transform network model to local model
    final localUser = LocalUser(
      id: networkUser.id,
      name: networkUser.fullName,
      email: networkUser.emailAddress,
      cachedAt: DateTime.now(),
    );
    await db.insertUser(localUser);
  },
);

Validator Examples

// Simple validation
final validator1 = Validator.by<User>((user) async {
  return user.name.isNotEmpty && user.email.isNotEmpty;
});

// Time-based validation (TTL)
final validator2 = Validator.by<CachedData>((data) async {
  final age = DateTime.now().difference(data.cachedAt);
  return age.inMinutes < 5; // Valid for 5 minutes
});

// Async validation with API check
final validator3 = Validator.by<Document>((doc) async {
  // Check with server if document version is current
  final serverVersion = await api.getDocumentVersion(doc.id);
  return doc.version >= serverVersion;
});

// Complex validation logic
final validator4 = Validator.by<Product>((product) async {
  // Multiple validation checks
  if (product.price <= 0) return false;
  if (product.stock < 0) return false;
  
  // Check if data is stale
  final age = DateTime.now().difference(product.lastUpdated);
  if (age.inHours > 24) return false;
  
  return true;
});

Cache Examples

// Basic cache with size limit
final cache1 = CacheBuilder<String, User>()
  .maximumSize(100)
  .build();

cache1.put('user:123', User(id: '123', name: 'Alice'));
final user = cache1.getIfPresent('user:123');

// Cache with time-based expiration
final cache2 = CacheBuilder<String, ApiResponse>()
  .expireAfterWrite(Duration(minutes: 5))
  .maximumSize(50)
  .build();

// Cache with access-based expiration (keeps frequently accessed items)
final cache3 = CacheBuilder<String, Document>()
  .expireAfterAccess(Duration(minutes: 10))
  .maximumSize(200)
  .build();

// Cache with custom weigher (e.g., by data size)
final cache4 = CacheBuilder<String, Image>()
  .weigher(
    1024 * 1024 * 10, // 10MB max
    (key, image) => image.sizeInBytes,
  )
  .build();

// Cache with all features
final cache5 = CacheBuilder<int, CachedData>()
  .maximumSize(1000)
  .expireAfterWrite(Duration(minutes: 15))
  .expireAfterAccess(Duration(minutes: 5))
  .concurrencyLevel(8)
  .build();

// Using getOrPut for lazy loading
final data = cache5.getOrPut(42, () {
  // Only called if not in cache
  return fetchDataFromNetwork(42);
});

// Batch operations
cache5.putAll({
  1: data1,
  2: data2,
  3: data3,
});

final subset = cache5.getAllPresent([1, 2, 5]); // Returns {1: data1, 2: data2}

// Invalidation
cache5.invalidate(1); // Remove single entry
cache5.invalidateAll([2, 3]); // Remove multiple entries
cache5.invalidateAll(); // Clear entire cache

Request Deduplication Examples

// Create a fetcher controller for request deduplication
final controller = FetcherController<String, User>();

// Multiple concurrent requests for the same user
// Only one network call will be made
final futures = [
  controller.getFetcher('user:123', () => fetchUserFromApi('123')),
  controller.getFetcher('user:123', () => fetchUserFromApi('123')),
  controller.getFetcher('user:123', () => fetchUserFromApi('123')),
];

// All three will receive the same result from a single fetch
final results = await Future.wait(futures.map((s) => s.first));

// Check active fetches
if (controller.isActive('user:123')) {
  print('Fetch in progress');
}

print('Active fetches: ${controller.activeCount}');

// Cancel a specific fetch
await controller.cancelFetch('user:456');

// Cancel all active fetches
await controller.cancelAll();

// Using Multicaster directly for custom scenarios
final multicaster = Multicaster<int>(
  source: () => expensiveDataStream(),
  onEach: (value) => print('Received: $value'),
);

// Multiple subscribers share the same upstream
final stream1 = multicaster.newDownstream();
final stream2 = multicaster.newDownstream();

await Future.wait([
  stream1.forEach((v) => print('Stream 1: $v')),
  stream2.forEach((v) => print('Stream 2: $v')),
]);

// Clean up
await multicaster.close();

License

MIT

Credits

This is a port of Store5 by the Mobile Native Foundation, originally created by Google and the open-source community.

Libraries

persistencestore
Persistence Store - A Dart port of Store5