feat(llc, persistence): add offline support for reactions #2847
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| import 'package:stream_chat/src/client/client.dart'; | ||
| import 'package:stream_chat/src/client/reaction_pending_operation.dart'; | ||
| import 'package:stream_chat/src/core/error/error.dart'; | ||
| import 'package:stream_chat/src/core/models/pending_operation.dart'; | ||
| import 'package:stream_chat/src/core/models/reaction.dart'; | ||
|
|
||
| /// Replays the stored [PendingOperation] queue against the server when the | ||
| /// connection is recovered. | ||
| /// | ||
| /// Replay is at-least-once: an operation is removed from the queue only after | ||
| /// the server accepts or terminally rejects it, so a crash between acceptance | ||
| /// and removal re-sends it on the next recovery. Every operation type handled | ||
| /// by [_replayCallFor] must therefore be idempotent on the server — e.g. | ||
| /// reactions dedupe by (message, type, user). | ||
| class PendingOperationReplayer { | ||
| /// Creates a replayer for [client]'s pending-operation queue. | ||
| PendingOperationReplayer(this._client); | ||
|
|
||
| final StreamChatClient _client; | ||
|
|
||
| // Prevents overlapping replays. | ||
| bool _isReplaying = false; | ||
|
|
||
| /// Replays each stored operation against the server in insertion order. | ||
| Future<void> replay() async { | ||
| final persistence = _client.chatPersistenceClient; | ||
| if (persistence == null) return; | ||
| if (_isReplaying) return; | ||
| _isReplaying = true; | ||
|
|
||
| try { | ||
| final operations = await persistence.getPendingOperations(); | ||
| for (final operation in operations) { | ||
| try { | ||
| final Future<void> Function()? call; | ||
| try { | ||
| call = _replayCallFor(operation); | ||
| } catch (error, stk) { | ||
| // Malformed payload for a known type — can never be replayed. | ||
| _client.logger.warning( | ||
| 'Dropping unreplayable pending operation ${operation.id}', | ||
| error, | ||
| stk, | ||
| ); | ||
| await persistence.deletePendingOperation(operation.id!); | ||
| continue; | ||
| } | ||
|
|
||
| if (call == null) { | ||
| // Unknown type (e.g. persisted by a newer app version) — drop it. | ||
| _client.logger.warning( | ||
| 'Dropping unknown pending operation type "${operation.type}" ' | ||
| '(${operation.id})', | ||
| ); | ||
| await persistence.deletePendingOperation(operation.id!); | ||
| continue; | ||
| } | ||
|
|
||
| try { | ||
| await call(); | ||
| } on StreamChatNetworkError catch (error) { | ||
| // Keep transient failures for the next recovery. | ||
| if (error.isRetriable) continue; | ||
| } | ||
|
|
||
| // Accepted or terminally rejected by the server — drop it. | ||
| await persistence.deletePendingOperation(operation.id!); | ||
| } catch (error, stk) { | ||
| _client.logger.warning( | ||
| 'Error replaying pending operation ${operation.id}', | ||
| error, | ||
| stk, | ||
| ); | ||
| } | ||
| } | ||
| } catch (error, stk) { | ||
| _client.logger.severe( | ||
| 'Error replaying pending operations', | ||
| error, | ||
| stk, | ||
| ); | ||
| } finally { | ||
| _isReplaying = false; | ||
| } | ||
| } | ||
|
|
||
| /// Returns the server call that replays [operation], or `null` if its type | ||
| /// is unknown to this version. | ||
| Future<void> Function()? _replayCallFor(PendingOperation operation) { | ||
| switch (operation.type) { | ||
| case ReactionPendingOperation.addType: | ||
| final targetMessageId = operation.targetMessageId; | ||
| if (targetMessageId == null) { | ||
| throw StateError('Missing targetMessageId for ${operation.type}'); | ||
| } | ||
| final reaction = Reaction.fromJson( | ||
| operation.payload[ReactionPendingOperation.reactionKey] as Map<String, dynamic>, | ||
| ); | ||
| final skipPush = operation.payload[ReactionPendingOperation.skipPushKey] as bool? ?? false; | ||
| final enforceUnique = operation.payload[ReactionPendingOperation.enforceUniqueKey] as bool? ?? false; | ||
| return () => _client.sendReaction( | ||
| targetMessageId, | ||
| reaction, | ||
| skipPush: skipPush, | ||
| enforceUnique: enforceUnique, | ||
| ); | ||
| case ReactionPendingOperation.deleteType: | ||
| final targetMessageId = operation.targetMessageId; | ||
| if (targetMessageId == null) { | ||
| throw StateError('Missing targetMessageId for ${operation.type}'); | ||
| } | ||
| final reactionType = operation.payload[ReactionPendingOperation.reactionTypeKey] as String; | ||
| return () => _client.deleteReaction(targetMessageId, reactionType); | ||
| default: | ||
| // Unknown operation type — cannot be replayed by this version. | ||
| return null; | ||
| } | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| import 'package:stream_chat/src/core/models/pending_operation.dart'; | ||
| import 'package:stream_chat/src/core/models/reaction.dart'; | ||
|
|
||
| /// Builds and identifies the reaction-specific forms of [PendingOperation]. | ||
| abstract class ReactionPendingOperation { | ||
| /// The [PendingOperation.type] discriminator for a reaction add. | ||
| static const addType = 'reaction.add'; | ||
|
|
||
| /// The [PendingOperation.type] discriminator for a reaction delete. | ||
| static const deleteType = 'reaction.delete'; | ||
|
|
||
| /// The [PendingOperation.payload] key holding the serialized reaction of an | ||
| /// add. | ||
| static const reactionKey = 'reaction'; | ||
|
|
||
| /// The [PendingOperation.payload] key holding the `enforce_unique` flag of an | ||
| /// add. | ||
| static const enforceUniqueKey = 'enforce_unique'; | ||
|
|
||
| /// The [PendingOperation.payload] key holding the `skip_push` flag of an add. | ||
| static const skipPushKey = 'skip_push'; | ||
|
|
||
| /// The [PendingOperation.payload] key holding the reaction type of a delete. | ||
| static const reactionTypeKey = 'reaction_type'; | ||
|
|
||
| /// Builds the pending operation recording an optimistic reaction add. | ||
| static PendingOperation add( | ||
| Reaction reaction, { | ||
| required bool skipPush, | ||
| required bool enforceUnique, | ||
| }) => PendingOperation( | ||
| type: addType, | ||
| targetMessageId: reaction.messageId, | ||
| payload: { | ||
| reactionKey: reaction.toJson(), | ||
| enforceUniqueKey: enforceUnique, | ||
| skipPushKey: skipPush, | ||
| }, | ||
| ); | ||
|
|
||
| /// Builds the pending operation recording an optimistic reaction delete. | ||
| static PendingOperation delete({ | ||
| required String messageId, | ||
| required String reactionType, | ||
| }) => PendingOperation( | ||
| type: deleteType, | ||
| targetMessageId: messageId, | ||
| payload: {reactionTypeKey: reactionType}, | ||
| ); | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| import 'package:equatable/equatable.dart'; | ||
|
|
||
| /// {@template pendingOperation} | ||
| /// A durable record of an optimistic mutation awaiting server confirmation | ||
| /// (e.g. a reaction added or removed while offline). | ||
| /// | ||
| /// Operations are replayed at-least-once on reconnect, so the server-side | ||
| /// effect of every operation type must be idempotent. | ||
| /// {@endtemplate} | ||
| class PendingOperation extends Equatable { | ||
| /// {@macro pendingOperation} | ||
| const PendingOperation({ | ||
| required this.type, | ||
| required this.payload, | ||
| this.id, | ||
| this.targetMessageId, | ||
| }); | ||
|
|
||
| /// The database autoincrement id, assigned when the operation is stored; | ||
| /// `null` until then. | ||
| final int? id; | ||
|
|
||
| /// The discriminator persisted in the `type` column, e.g. `reaction.add`. | ||
| final String type; | ||
|
|
||
| /// The id of the message the operation targets, if any. | ||
| final String? targetMessageId; | ||
|
|
||
| /// The operation-specific value fields, stored as JSON. | ||
| final Map<String, dynamic> payload; | ||
|
|
||
| @override | ||
| List<Object?> get props => [ | ||
| type, | ||
| targetMessageId, | ||
| payload, | ||
| ]; | ||
| } | ||
|
Comment on lines
+10
to
+51
Contributor
There was a problem hiding this comment. Suggestion — make As written this is a stringly-typed bag: The DB row should stay generic The repo is on Dart /// Identity of a queued operation: either a persisted row or session-only.
sealed class PendingOperationId extends Equatable {
const PendingOperationId();
}
/// An operation mirrored to persistence, identified by its row id.
final class PersistedOperationId extends PendingOperationId {
const PersistedOperationId(this.value) : assert(value > 0, 'row ids are positive');
final int value;
@override
List<Object?> get props => [value];
}
/// An operation that lives only in memory for this session.
final class SessionOperationId extends PendingOperationId {
const SessionOperationId(this.value);
final int value;
@override
List<Object?> get props => [value];
}
/// {@macro pendingOperation}
sealed class PendingOperation extends Equatable {
const PendingOperation({this.id});
/// Identity of this operation, `null` until it is queued.
final PendingOperationId? id;
/// The discriminator persisted in the `type` column.
String get type;
/// The message this operation targets.
String get targetMessageId;
/// The operation-specific value fields. Must be JSON-encodable.
Map<String, dynamic> toPayload();
/// Returns a copy of this operation with [id] assigned.
PendingOperation withId(PendingOperationId id);
/// Rebuilds a stored operation, or returns `null` when [type] is unknown to
/// this version. Throws [FormatException] on a malformed payload.
static PendingOperation? fromStored({
required PendingOperationId id,
required String type,
required String? targetMessageId,
required Map<String, dynamic> payload,
}) => switch (type) {
AddReactionOperation.opType => AddReactionOperation.fromPayload(id, payload),
DeleteReactionOperation.opType => DeleteReactionOperation.fromPayload(id, payload),
_ => null,
};
}
/// A reaction added while offline, awaiting replay.
final class AddReactionOperation extends PendingOperation {
const AddReactionOperation({
required this.reaction,
this.skipPush = false,
this.enforceUnique = false,
super.id,
});
/// The discriminator persisted for this operation type.
static const opType = 'reaction.add';
/// The reaction to send.
final Reaction reaction;
/// Whether to skip the push notification for the reaction.
final bool skipPush;
/// Whether the reaction replaces the user's existing one.
final bool enforceUnique;
@override
String get type => opType;
@override
String get targetMessageId => reaction.messageId!;
// ... toPayload / fromPayload / withId / props
}
Future<void> Function() _replayCallFor(PendingOperation operation) =>
switch (operation) {
AddReactionOperation(:final targetMessageId, :final reaction, :final skipPush, :final enforceUnique) =>
() => _client.sendReaction(targetMessageId, reaction, skipPush: skipPush, enforceUnique: enforceUnique),
DeleteReactionOperation(:final targetMessageId, :final reactionType) =>
() => _client.deleteReaction(targetMessageId, reactionType),
};What this buys, concretely:
Two smaller things on the current class while it is being touched: Fully understand if the sealed refactor is out of scope for this PR — in that case the minimum I would ask for is making the strictness in
Contributor
There was a problem hiding this comment. Follow-up: React Native already ships exactly this shape, which I think settles the "is this over-engineering?" question. RN's equivalent type is a discriminated union whose payload type is derived from the method being replayed ( export type PendingTask = {
channelId: string;
channelType: string;
messageId: string;
id?: number;
} & (
| { type: 'send-reaction'; payload: Parameters<Channel['sendReaction']> }
| { type: 'delete-reaction'; payload: Parameters<Channel['deleteReaction']> }
| { type: 'delete-message'; payload: Parameters<StreamChat['deleteMessage']> }
);and its executor narrows on the discriminant and spreads the payload, so there are no casts and no defaults to get wrong ( if (task.type === 'send-reaction') return await channel._sendReaction(...task.payload);
if (task.type === 'delete-reaction') return await channel._deleteReaction(...task.payload);Same idea as the sealed hierarchy above: the row stays generic Two extra reasons this matters more here than it looks:
Worth noting RN also carries
Contributor
Author
There was a problem hiding this comment. I was considering such approach as well, but in the end I decided for a flat hierarchy, because if we define the |
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not entirely sure if we should always
rethrowhere, maybe just in the non-retry-able case?