30854254088

Committed 03 Aug 2026 09:21PM UTC coverage: 40.895% (-15.0%) from 55.93%

Build # 30854254088

Build Type

Pull #1260

github

Committed by

web-flow

Commit Message

Merge d93794de5 into e3affdd4f

Pull Request Pull Request #1260: WIP: health sync

Coverage Stats

9980 of 24404 relevant lines covered (40.89%)

1.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.48

/lib/powersync/connector.dart

/*
 * This file is part of wger Workout Manager <https://github.com/wger-project>.
 * Copyright (c) 2020 - 2026 wger Team
 *
 * wger Workout Manager is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

// This file performs setup of the PowerSync database
import 'dart:convert';
import 'dart:io';

import 'package:flutter/foundation.dart';
import 'package:http/http.dart' as http;
import 'package:logging/logging.dart';
import 'package:powersync/powersync.dart';
import 'package:wger/core/error_dialogs.dart';
import 'package:wger/core/exceptions/http_exception.dart';
import 'package:wger/core/helpers.dart';
import 'package:wger/core/network/jwt.dart';
import 'package:wger/powersync/api_client.dart';

final logger = Logger('powersync-django');

/// Thrown for an upload status that should be retried, not discarded such as
/// HTTP status codes 5xx, 408, 429, or an unrecovered 401. Throwing leaves the
/// transaction queued for PowerSync to retry. Carries table/op/status for
/// logging and tests.
class RetryableUploadException implements Exception {
  final String table;
  final UpdateType op;
  final int statusCode;

  RetryableUploadException({
    required this.table,
    required this.op,
    required this.statusCode,
  });

  @override
  String toString() => 'Upload of $op on $table deferred: retryable status $statusCode';
}

/// What the transaction loop does with one upload response.
enum _UploadOutcome {
  /// Accepted: complete the transaction once all ops are ok.
  ok,

  /// Permanently refused: surface it but still complete, so one bad op can't
  /// block the queue.
  reject,

  /// Retryable: throw so PowerSync retries the queued transaction later.
  retry,
}

class DjangoConnector extends PowerSyncBackendConnector {
  final String baseUrl;
  final ApiClient apiClient;

  /// Client for the endpoint liveness probes in [fetchCredentials].
  final http.Client _probeClient;

  /// IDs of CRUD operations that already triggered a user-facing
  /// rejection dialog this session. Without this gate the same
  /// permanent-failure op would re-pop the dialog on every sync tick
  /// (PowerSync keeps re-driving `uploadData` until the transaction
  /// is completed, and our `transaction.complete()` only fires once
  /// the loop finishes, so we'd see the dialog at every iteration).
  /// Resets on app restart.
  final Set<String> _reportedFailedOps = {};

  /// Spacing between repeated "backend unreachable" log lines while the
  /// credential fetch keeps failing, so PowerSync's retry loop doesn't
  /// flood the app logs during a longer outage.
  static const unreachableLogInterval = Duration(minutes: 5);

  /// When the current outage was last logged; null while the backend is
  /// reachable.
  DateTime? _lastUnreachableLogAt;

  /// Endpoint last announced in the logs. [fetchCredentials] runs on every
  /// token refresh, so the endpoint is only logged when it changes.
  String? _lastLoggedEndpoint;

  /// When "no live PowerSync endpoint" was last logged; null while an
  /// endpoint resolves. Throttled like [_logUnreachable], since PowerSync
  /// re-drives [fetchCredentials] on its retry schedule.
  DateTime? _lastNoEndpointLogAt;

  /// The `powersync_url` from the last token response and the endpoint it
  /// resolved to. While the server keeps advertising the same URL, the cached
  /// resolution is reused instead of re-probing on every token refresh (a dead
  /// advertised URL would otherwise cost a probe timeout each time). A failed
  /// resolution is never cached, so retries keep probing. Cleared on app
  /// restart with the connector.
  String? _lastProvidedUrl;
  String? _lastResolvedEndpoint;

  DjangoConnector({required this.baseUrl, required this.apiClient, http.Client? client})
    : _probeClient = client ?? http.Client();

  /// Get a token to authenticate against the PowerSync instance.
  @override
  Future<PowerSyncCredentials?> fetchCredentials() async {
    // See the auth docs here:
    // https://docs.powersync.com/usage/installation/authentication-setup/custom
    final Map<String, dynamic> session;
    try {
      session = await apiClient.getPowersyncToken();
    } on http.ClientException catch (e) {
      // Backend unreachable (offline). Returning null skips this attempt;
      // PowerSync retries on its own schedule. Without this, the raw
      // SocketException stack trace floods the logs on every retry.
      _logUnreachable(e.message);
      return null;
    } on SocketException catch (e) {
      _logUnreachable(e.message);
      return null;
    }
    _logReachableAgain();

    final token = session['token'] as String;
    final payload = decodeJwtPayload(token);
    final provided = session['powersync_url'] as String?;
    String? endpoint;
    if (provided == _lastProvidedUrl && _lastResolvedEndpoint != null) {
      endpoint = _lastResolvedEndpoint;
    } else {
      endpoint = await findLivePowerSyncUrl(
        client: _probeClient,
        serverUrl: baseUrl,
        provided: provided,
      );
      if (endpoint != null) {
        _lastProvidedUrl = provided;
        _lastResolvedEndpoint = endpoint;
      }
    }
    if (endpoint == null) {
      // The wger backend answered (the token fetch above succeeded), so this
      // is a bad or down sync service, not the device being offline.
      // Returning null skips the attempt; PowerSync retries on its own.
      _logNoEndpoint();
      return null;
    }
    _lastNoEndpointLogAt = null;
    if (endpoint != _lastLoggedEndpoint) {
      logger.info('Connecting to PowerSync endpoint $endpoint');
      _lastLoggedEndpoint = endpoint;
    }
    return PowerSyncCredentials(
      endpoint: endpoint,
      token: token,
      userId: payload?['sub']?.toString(),
      expiresAt: jwtExpOnLocalClock(payload),
    );
  }

  /// Logs a skipped credential fetch at INFO, throttled to once per [unreachableLogInterval]
  /// per outage.
  void _logUnreachable(String message) {
    final now = DateTime.now();
    final last = _lastUnreachableLogAt;
    if (last == null || now.difference(last) >= unreachableLogInterval) {
      logger.info('PowerSync credential fetch skipped, backend unreachable: $message');
      _lastUnreachableLogAt = now;
    }
  }

  /// Logs an unresolved PowerSync endpoint at WARNING, throttled to once per
  /// [unreachableLogInterval] per outage.
  void _logNoEndpoint() {
    final now = DateTime.now();
    final last = _lastNoEndpointLogAt;
    if (last == null || now.difference(last) >= unreachableLogInterval) {
      logger.warning('No PowerSync endpoint answered its liveness probe, skipping credentials');
      _lastNoEndpointLogAt = now;
    }
  }

  /// Closes an outage announced by [_logUnreachable], if any.
  void _logReachableAgain() {
    if (_lastUnreachableLogAt != null) {
      logger.info('Backend reachable again, PowerSync credential fetch succeeded');
      _lastUnreachableLogAt = null;
    }
  }

  /// Date-only fields per table.
  ///
  /// PowerSync serialises every SQLite `DateTime` column as an ISO-8601 timestamp
  /// (e.g. `2024-11-01T00:00:00.000Z`), but Django's `DateField` only accepts
  /// `YYYY-MM-DD`. For these columns we strip the time component before uploading.
  ///
  /// Keep in sync with `models.DateField` columns in the Django side.
  /// `auto_now_add=True` fields (e.g. `nutrition_nutritionplan.creation_date`)
  /// are read-only on the serializer and therefore safe to leave out.
  static const Map<String, Set<String>> _dateOnlyFields = {
    'manager_routine': {'start', 'end'},
    'manager_workoutsession': {'date'},
    'nutrition_nutritionplan': {'start', 'end'},
    'gallery_image': {'date'},
  };

  /// Transform a record before sending it to the backend.
  ///
  /// Note that PowerSync hands us `op.opData` as native SQLite primitives only
  /// (`null`, `int`, `double`, `String`, `Uint8List`).
  ///
  ///   * inject the row [id] (PowerSync stores it separately from the
  ///     payload),
  ///   * strip the `_id` suffix from foreign-key column names so the
  ///     Django serializers see `category` / `routine` / etc,
  ///   * trim the time component from date-only fields (see
  ///     [_dateOnlyFields]).
  @visibleForTesting
  Map<String, dynamic> genericTransform(
    String table,
    Map<String, dynamic>? src,
    String id,
  ) => _genericTransform(table, src, id);

  Map<String, dynamic> _genericTransform(
    String table,
    Map<String, dynamic>? src,
    String id,
  ) {
    final out = <String, dynamic>{'id': id};
    if (src == null) {
      return out;
    }

    final dateFields = _dateOnlyFields[table] ?? const <String>{};

    src.forEach((k, v) {
      if (k == 'id') {
        return;
      }
      // Trailing `_id` marks a foreign key, which Django exposes without the
      // suffix (`category_id` -> `category`). `external_id` is a plain value
      // column, not an FK, so it keeps its name.
      final key = (k.endsWith('_id') && k != 'external_id') ? k.substring(0, k.length - 3) : k;
      out[key] = (dateFields.contains(key) && v is String && v.length >= 10)
          ? v.substring(0, 10)
          : v;
    });
    return out;
  }

  // Upload pending changes to Postgres via Django backend
  // this is generic. on the django side we inspect the request and do model-specific operations
  // would it make sense to do api calls here specific to the relevant model? (e.g. put to a todo-specific endpoint)
  @override
  Future<void> uploadData(PowerSyncDatabase database) async {
    final transaction = await database.getNextCrudTransaction();

    if (transaction == null) {
      return;
    }

    await processTransaction(transaction);
  }

  /// Uploads every op in [transaction] and decides its fate: all accepted
  /// completes it; a permanent refusal is surfaced but still completes (so one
  /// bad op can't block the queue); a transient status (5xx, 408, 429, 401) or
  /// an unreachable backend throws, leaving it queued for PowerSync to retry.
  ///
  /// A retry re-sends the whole transaction (at-least-once), so backend handlers
  /// must be idempotent.
  @visibleForTesting
  Future<void> processTransaction(CrudTransaction transaction) async {
    try {
      for (final op in transaction.crud) {
        final record = {
          'table': op.table,
          'data': _genericTransform(op.table, op.opData, op.id),
        };

        // logger.finer('Uploading record $record to server with operation ${op.op}');

        final http.Response response;
        switch (op.op) {
          case UpdateType.put:
            response = await apiClient.upsert(record);
            break;
          case UpdateType.patch:
            response = await apiClient.update(record);
            break;
          case UpdateType.delete:
            response = await apiClient.delete(record);
            break;
        }

        switch (_classifyResponse(response)) {
          case _UploadOutcome.ok:
            break;
          case _UploadOutcome.reject:
            _reportRejection(op, response);
            break;
          case _UploadOutcome.retry:
            throw RetryableUploadException(
              table: op.table,
              op: op.op,
              statusCode: response.statusCode,
            );
        }
      }
      await transaction.complete();
    } on http.ClientException catch (e) {
      // Backend unreachable (offline or down). The transaction stays queued;
      // rethrowing lets PowerSync retry it once the backend is reachable.
      logger.fine('Upload deferred, backend unreachable: ${e.message}');
      rethrow;
    } on SocketException catch (e) {
      logger.fine('Upload deferred, backend unreachable: ${e.message}');
      rethrow;
    } on RetryableUploadException catch (e) {
      // Stays queued for PowerSync to retry. Below severe: a brief server blip
      // is expected to clear on its own.
      logger.warning('Upload deferred: $e');
      rethrow;
    } on Exception catch (e) {
      logger.severe('Error uploading data', e);
      // Error may be retryable, e.g. a temporary server error. Throwing here
      // causes PowerSync to retry this transaction after a delay.
      rethrow;
    }
  }

  /// Classifies a single upload [response] into a [_UploadOutcome].
  _UploadOutcome _classifyResponse(http.Response response) {
    final status = response.statusCode;

    // 2xx is success, unless the backend encoded a permanent rejection as
    // 200 + `{error}` (its anti-retry-storm contract).
    if (status >= 200 && status < 300) {
      return _isErrorBody(response) ? _UploadOutcome.reject : _UploadOutcome.ok;
    }

    // Transient or retryable. 401 lands here because AuthHttpClient already
    // tried to refresh; a 401 still reaching us means the session is gone, so
    // queue the op for re-auth rather than dropping it.
    if (status >= 500 || status == 408 || status == 429 || status == 401) {
      return _UploadOutcome.retry;
    }

    // Any other 4xx is permanent (retry won't help). Expected refusals come as
    // 200 + `{error}`, so a non-200 4xx is genuinely unexpected.
    return _UploadOutcome.reject;
  }

  /// Surfaces a permanently refused op via the global error dialog, once per op
  /// per session (a re-driven transaction would otherwise re-pop it each tick).
  void _reportRejection(CrudEntry op, http.Response response) {
    if (!_reportedFailedOps.add(op.id)) {
      // Already shown for this operation in the current session.
      return;
    }

    final exception = WgerHttpException(
      response,
      source: ExceptionSource.powersync,
      context: {'table': op.table, 'op': op.op.name},
    );
    final ctx = '${op.op.name} ${op.table}';
    // 200 + {error} is the expected contract (warning); other statuses are
    // unexpected (severe).
    if (response.statusCode == 200) {
      logger.warning('Backend rejected $ctx', exception);
    } else {
      logger.severe('Unexpected permanent upload failure: $ctx', exception);
    }
    // Route through the app's central error handler (same entry point as the
    // global FlutterError/PlatformDispatcher handlers).
    handleError(exception, StackTrace.current);
  }

  /// Whether [response]'s body is a JSON object carrying an `{error}` key, the
  /// backend's contract for a permanent rejection on a 200.
  bool _isErrorBody(http.Response response) {
    if (response.body.isEmpty) {
      return false;
    }
    try {
      final decoded = json.decode(response.body);
      return decoded is Map<String, dynamic> && decoded.containsKey('error');
    } on FormatException {
      // Non-JSON body, not a structured rejection.
      return false;
    }
  }
}

1	/*
2	* This file is part of wger Workout Manager <https://github.com/wger-project>.
3	* Copyright (c) 2020 - 2026 wger Team
4	*
5	* wger Workout Manager is free software: you can redistribute it and/or modify
6	* it under the terms of the GNU Affero General Public License as published by
7	* the Free Software Foundation, either version 3 of the License, or
8	* (at your option) any later version.
9	*
10	* This program is distributed in the hope that it will be useful,
11	* but WITHOUT ANY WARRANTY; without even the implied warranty of
12	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	* GNU Affero General Public License for more details.
14	*
15	* You should have received a copy of the GNU Affero General Public License
16	* along with this program. If not, see <http://www.gnu.org/licenses/>.
17	*/
18
19	// This file performs setup of the PowerSync database
20	import 'dart:convert';
21	import 'dart:io';
22
23	import 'package:flutter/foundation.dart';
24	import 'package:http/http.dart' as http;
25	import 'package:logging/logging.dart';
26	import 'package:powersync/powersync.dart';
27	import 'package:wger/core/error_dialogs.dart';
28	import 'package:wger/core/exceptions/http_exception.dart';
29	import 'package:wger/core/helpers.dart';
30	import 'package:wger/core/network/jwt.dart';
31	import 'package:wger/powersync/api_client.dart';
32
33	final logger = Logger('powersync-django');	3✔
34
35	/// Thrown for an upload status that should be retried, not discarded such as
36	/// HTTP status codes 5xx, 408, 429, or an unrecovered 401. Throwing leaves the
37	/// transaction queued for PowerSync to retry. Carries table/op/status for
38	/// logging and tests.
39	class RetryableUploadException implements Exception {
40	final String table;
41	final UpdateType op;
42	final int statusCode;
43
44	RetryableUploadException({	2✔
45	required this.table,
46	required this.op,
47	required this.statusCode,
48	});
49
50	@override	2✔
51	String toString() => 'Upload of $op on $table deferred: retryable status $statusCode';	8✔
52	}
53
54	/// What the transaction loop does with one upload response.
55	enum _UploadOutcome {
56	/// Accepted: complete the transaction once all ops are ok.
57	ok,
58
59	/// Permanently refused: surface it but still complete, so one bad op can't
60	/// block the queue.
61	reject,
62
63	/// Retryable: throw so PowerSync retries the queued transaction later.
64	retry,
65	}
66
67	class DjangoConnector extends PowerSyncBackendConnector {
68	final String baseUrl;
69	final ApiClient apiClient;
70
71	/// Client for the endpoint liveness probes in [fetchCredentials].
72	final http.Client _probeClient;
73
74	/// IDs of CRUD operations that already triggered a user-facing
75	/// rejection dialog this session. Without this gate the same
76	/// permanent-failure op would re-pop the dialog on every sync tick
77	/// (PowerSync keeps re-driving `uploadData` until the transaction
78	/// is completed, and our `transaction.complete()` only fires once
79	/// the loop finishes, so we'd see the dialog at every iteration).
80	/// Resets on app restart.
81	final Set<String> _reportedFailedOps = {};
82
83	/// Spacing between repeated "backend unreachable" log lines while the
84	/// credential fetch keeps failing, so PowerSync's retry loop doesn't
85	/// flood the app logs during a longer outage.
86	static const unreachableLogInterval = Duration(minutes: 5);
87
88	/// When the current outage was last logged; null while the backend is
89	/// reachable.
90	DateTime? _lastUnreachableLogAt;
91
92	/// Endpoint last announced in the logs. [fetchCredentials] runs on every
93	/// token refresh, so the endpoint is only logged when it changes.
94	String? _lastLoggedEndpoint;
95
96	/// When "no live PowerSync endpoint" was last logged; null while an
97	/// endpoint resolves. Throttled like [_logUnreachable], since PowerSync
98	/// re-drives [fetchCredentials] on its retry schedule.
99	DateTime? _lastNoEndpointLogAt;
100
101	/// The `powersync_url` from the last token response and the endpoint it
102	/// resolved to. While the server keeps advertising the same URL, the cached
103	/// resolution is reused instead of re-probing on every token refresh (a dead
104	/// advertised URL would otherwise cost a probe timeout each time). A failed
105	/// resolution is never cached, so retries keep probing. Cleared on app
106	/// restart with the connector.
107	String? _lastProvidedUrl;
108	String? _lastResolvedEndpoint;
109
110	DjangoConnector({required this.baseUrl, required this.apiClient, http.Client? client})	1✔
111	: _probeClient = client ?? http.Client();	1✔
112
113	/// Get a token to authenticate against the PowerSync instance.
114	@override	1✔
115	Future<PowerSyncCredentials?> fetchCredentials() async {
116	// See the auth docs here:
117	// https://docs.powersync.com/usage/installation/authentication-setup/custom
118	final Map<String, dynamic> session;
119	try {
120	session = await apiClient.getPowersyncToken();	2✔
121	} on http.ClientException catch (e) {	×
122	// Backend unreachable (offline). Returning null skips this attempt;
123	// PowerSync retries on its own schedule. Without this, the raw
124	// SocketException stack trace floods the logs on every retry.
125	_logUnreachable(e.message);	×
126	return null;
127	} on SocketException catch (e) {	×
128	_logUnreachable(e.message);	×
129	return null;
130	}
131	_logReachableAgain();	1✔
132
133	final token = session['token'] as String;	1✔
134	final payload = decodeJwtPayload(token);	1✔
135	final provided = session['powersync_url'] as String?;	1✔
136	String? endpoint;
137	if (provided == _lastProvidedUrl && _lastResolvedEndpoint != null) {	2✔
138	endpoint = _lastResolvedEndpoint;	×
139	} else {
140	endpoint = await findLivePowerSyncUrl(	1✔
141	client: _probeClient,	1✔
142	serverUrl: baseUrl,	1✔
143	provided: provided,
144	);
145	if (endpoint != null) {
146	_lastProvidedUrl = provided;	1✔
147	_lastResolvedEndpoint = endpoint;	1✔
148	}
149	}
150	if (endpoint == null) {
151	// The wger backend answered (the token fetch above succeeded), so this
152	// is a bad or down sync service, not the device being offline.
153	// Returning null skips the attempt; PowerSync retries on its own.
154	_logNoEndpoint();	×
155	return null;
156	}
157	_lastNoEndpointLogAt = null;	1✔
158	if (endpoint != _lastLoggedEndpoint) {	2✔
159	logger.info('Connecting to PowerSync endpoint $endpoint');	3✔
160	_lastLoggedEndpoint = endpoint;	1✔
161	}
162	return PowerSyncCredentials(	1✔
163	endpoint: endpoint,
164	token: token,
165	userId: payload?['sub']?.toString(),	2✔
166	expiresAt: jwtExpOnLocalClock(payload),	1✔
167	);
168	}
169
170	/// Logs a skipped credential fetch at INFO, throttled to once per [unreachableLogInterval]
171	/// per outage.
172	void _logUnreachable(String message) {	×
173	final now = DateTime.now();	×
174	final last = _lastUnreachableLogAt;	×
175	if (last == null \|\| now.difference(last) >= unreachableLogInterval) {	×
176	logger.info('PowerSync credential fetch skipped, backend unreachable: $message');	×
177	_lastUnreachableLogAt = now;	×
178	}
179	}
180
181	/// Logs an unresolved PowerSync endpoint at WARNING, throttled to once per
182	/// [unreachableLogInterval] per outage.
183	void _logNoEndpoint() {	×
184	final now = DateTime.now();	×
185	final last = _lastNoEndpointLogAt;	×
186	if (last == null \|\| now.difference(last) >= unreachableLogInterval) {	×
187	logger.warning('No PowerSync endpoint answered its liveness probe, skipping credentials');	×
188	_lastNoEndpointLogAt = now;	×
189	}
190	}
191
192	/// Closes an outage announced by [_logUnreachable], if any.
193	void _logReachableAgain() {	1✔
194	if (_lastUnreachableLogAt != null) {	1✔
195	logger.info('Backend reachable again, PowerSync credential fetch succeeded');	×
196	_lastUnreachableLogAt = null;	×
197	}
198	}
199
200	/// Date-only fields per table.
201	///
202	/// PowerSync serialises every SQLite `DateTime` column as an ISO-8601 timestamp
203	/// (e.g. `2024-11-01T00:00:00.000Z`), but Django's `DateField` only accepts
204	/// `YYYY-MM-DD`. For these columns we strip the time component before uploading.
205	///
206	/// Keep in sync with `models.DateField` columns in the Django side.
207	/// `auto_now_add=True` fields (e.g. `nutrition_nutritionplan.creation_date`)
208	/// are read-only on the serializer and therefore safe to leave out.
209	static const Map<String, Set<String>> _dateOnlyFields = {
210	'manager_routine': {'start', 'end'},
211	'manager_workoutsession': {'date'},
212	'nutrition_nutritionplan': {'start', 'end'},
213	'gallery_image': {'date'},
214	};
215
216	/// Transform a record before sending it to the backend.
217	///
218	/// Note that PowerSync hands us `op.opData` as native SQLite primitives only
219	/// (`null`, `int`, `double`, `String`, `Uint8List`).
220	///
221	/// * inject the row [id] (PowerSync stores it separately from the
222	/// payload),
223	/// * strip the `_id` suffix from foreign-key column names so the
224	/// Django serializers see `category` / `routine` / etc,
225	/// * trim the time component from date-only fields (see
226	/// [_dateOnlyFields]).
227	@visibleForTesting	×
228	Map<String, dynamic> genericTransform(
229	String table,
230	Map<String, dynamic>? src,
231	String id,
232	) => _genericTransform(table, src, id);	×
233
234	Map<String, dynamic> _genericTransform(	1✔
235	String table,
236	Map<String, dynamic>? src,
237	String id,
238	) {
239	final out = <String, dynamic>{'id': id};	1✔
240	if (src == null) {
241	return out;
242	}
243
244	final dateFields = _dateOnlyFields[table] ?? const <String>{};	1✔
245
246	src.forEach((k, v) {	2✔
247	if (k == 'id') {	1✔
248	return;
249	}
250	// Trailing `_id` marks a foreign key, which Django exposes without the
251	// suffix (`category_id` -> `category`). `external_id` is a plain value
252	// column, not an FK, so it keeps its name.
253	final key = (k.endsWith('_id') && k != 'external_id') ? k.substring(0, k.length - 3) : k;	1✔
254	out[key] = (dateFields.contains(key) && v is String && v.length >= 10)	2✔
255	? v.substring(0, 10)	×
256	: v;
257	});
258	return out;
259	}
260
261	// Upload pending changes to Postgres via Django backend
262	// this is generic. on the django side we inspect the request and do model-specific operations
263	// would it make sense to do api calls here specific to the relevant model? (e.g. put to a todo-specific endpoint)
264	@override	×
265	Future<void> uploadData(PowerSyncDatabase database) async {
266	final transaction = await database.getNextCrudTransaction();	×
267
268	if (transaction == null) {
269	return;
270	}
271
272	await processTransaction(transaction);	×
273	}
274
275	/// Uploads every op in [transaction] and decides its fate: all accepted
276	/// completes it; a permanent refusal is surfaced but still completes (so one
277	/// bad op can't block the queue); a transient status (5xx, 408, 429, 401) or
278	/// an unreachable backend throws, leaving it queued for PowerSync to retry.
279	///
280	/// A retry re-sends the whole transaction (at-least-once), so backend handlers
281	/// must be idempotent.
282	@visibleForTesting	1✔
283	Future<void> processTransaction(CrudTransaction transaction) async {
284	try {
285	for (final op in transaction.crud) {	2✔
286	final record = {	1✔
287	'table': op.table,	1✔
288	'data': _genericTransform(op.table, op.opData, op.id),	4✔
289	};
290
291	// logger.finer('Uploading record $record to server with operation ${op.op}');
292
293	final http.Response response;
294	switch (op.op) {	1✔
295	case UpdateType.put:	1✔
296	response = await apiClient.upsert(record);	2✔
297	break;
298	case UpdateType.patch:	1✔
299	response = await apiClient.update(record);	2✔
300	break;
301	case UpdateType.delete:	1✔
302	response = await apiClient.delete(record);	2✔
303	break;
304	}
305
306	switch (_classifyResponse(response)) {	1✔
307	case _UploadOutcome.ok:	1✔
308	break;
309	case _UploadOutcome.reject:	1✔
310	_reportRejection(op, response);	1✔
311	break;
312	case _UploadOutcome.retry:	1✔
313	throw RetryableUploadException(	1✔
314	table: op.table,	1✔
315	op: op.op,	1✔
316	statusCode: response.statusCode,	1✔
317	);
318	}
319	}
320	await transaction.complete();	2✔
321	} on http.ClientException catch (e) {	1✔
322	// Backend unreachable (offline or down). The transaction stays queued;
323	// rethrowing lets PowerSync retry it once the backend is reachable.
324	logger.fine('Upload deferred, backend unreachable: ${e.message}');	4✔
325	rethrow;
326	} on SocketException catch (e) {	1✔
327	logger.fine('Upload deferred, backend unreachable: ${e.message}');	×
328	rethrow;
329	} on RetryableUploadException catch (e) {	1✔
330	// Stays queued for PowerSync to retry. Below severe: a brief server blip
331	// is expected to clear on its own.
332	logger.warning('Upload deferred: $e');	3✔
333	rethrow;
334	} on Exception catch (e) {	×
335	logger.severe('Error uploading data', e);	×
336	// Error may be retryable, e.g. a temporary server error. Throwing here
337	// causes PowerSync to retry this transaction after a delay.
338	rethrow;
339	}
340	}
341
342	/// Classifies a single upload [response] into a [_UploadOutcome].
343	_UploadOutcome _classifyResponse(http.Response response) {	1✔
344	final status = response.statusCode;	1✔
345
346	// 2xx is success, unless the backend encoded a permanent rejection as
347	// 200 + `{error}` (its anti-retry-storm contract).
348	if (status >= 200 && status < 300) {	2✔
349	return _isErrorBody(response) ? _UploadOutcome.reject : _UploadOutcome.ok;	1✔
350	}
351
352	// Transient or retryable. 401 lands here because AuthHttpClient already
353	// tried to refresh; a 401 still reaching us means the session is gone, so
354	// queue the op for re-auth rather than dropping it.
355	if (status >= 500 \|\| status == 408 \|\| status == 429 \|\| status == 401) {	4✔
356	return _UploadOutcome.retry;
357	}
358
359	// Any other 4xx is permanent (retry won't help). Expected refusals come as
360	// 200 + `{error}`, so a non-200 4xx is genuinely unexpected.
361	return _UploadOutcome.reject;
362	}
363
364	/// Surfaces a permanently refused op via the global error dialog, once per op
365	/// per session (a re-driven transaction would otherwise re-pop it each tick).
366	void _reportRejection(CrudEntry op, http.Response response) {	1✔
367	if (!_reportedFailedOps.add(op.id)) {	3✔
368	// Already shown for this operation in the current session.
369	return;
370	}
371
372	final exception = WgerHttpException(	1✔
373	response,
374	source: ExceptionSource.powersync,
375	context: {'table': op.table, 'op': op.op.name},	4✔
376	);
377	final ctx = '${op.op.name} ${op.table}';	4✔
378	// 200 + {error} is the expected contract (warning); other statuses are
379	// unexpected (severe).
380	if (response.statusCode == 200) {	2✔
381	logger.warning('Backend rejected $ctx', exception);	3✔
382	} else {
383	logger.severe('Unexpected permanent upload failure: $ctx', exception);	3✔
384	}
385	// Route through the app's central error handler (same entry point as the
386	// global FlutterError/PlatformDispatcher handlers).
387	handleError(exception, StackTrace.current);	2✔
388	}
389
390	/// Whether [response]'s body is a JSON object carrying an `{error}` key, the
391	/// backend's contract for a permanent rejection on a 200.
392	bool _isErrorBody(http.Response response) {	1✔
393	if (response.body.isEmpty) {	2✔
394	return false;
395	}
396	try {
397	final decoded = json.decode(response.body);	2✔
398	return decoded is Map<String, dynamic> && decoded.containsKey('error');	2✔
399	} on FormatException {	×
400	// Non-JSON body, not a structured rejection.
401	return false;
402	}
403	}
404	}

wger-project / flutter / 30854254088

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous