diff options
Diffstat (limited to 'mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/SyncClientsEngineStage.java')
| -rw-r--r-- | mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/SyncClientsEngineStage.java | 691 |
1 files changed, 691 insertions, 0 deletions
diff --git a/mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/SyncClientsEngineStage.java b/mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/SyncClientsEngineStage.java new file mode 100644 index 000000000..04d3e7ce2 --- /dev/null +++ b/mobile/android/services/src/main/java/org/mozilla/gecko/sync/stage/SyncClientsEngineStage.java @@ -0,0 +1,691 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ + +package org.mozilla.gecko.sync.stage; + +import android.accounts.Account; +import android.content.Context; +import android.support.annotation.NonNull; +import android.text.TextUtils; +import android.util.Log; + +import java.io.UnsupportedEncodingException; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.json.simple.JSONArray; +import org.json.simple.JSONObject; +import org.mozilla.gecko.AppConstants; +import org.mozilla.gecko.background.common.log.Logger; +import org.mozilla.gecko.background.fxa.FxAccountClient; +import org.mozilla.gecko.background.fxa.FxAccountClient20; +import org.mozilla.gecko.background.fxa.FxAccountClientException; +import org.mozilla.gecko.fxa.FirefoxAccounts; +import org.mozilla.gecko.fxa.authenticator.AndroidFxAccount; +import org.mozilla.gecko.sync.CommandProcessor; +import org.mozilla.gecko.sync.CommandProcessor.Command; +import org.mozilla.gecko.sync.CryptoRecord; +import org.mozilla.gecko.sync.ExtendedJSONObject; +import org.mozilla.gecko.sync.HTTPFailureException; +import org.mozilla.gecko.sync.NoCollectionKeysSetException; +import org.mozilla.gecko.sync.Utils; +import org.mozilla.gecko.sync.crypto.CryptoException; +import org.mozilla.gecko.sync.crypto.KeyBundle; +import org.mozilla.gecko.sync.delegates.ClientsDataDelegate; +import org.mozilla.gecko.sync.net.AuthHeaderProvider; +import org.mozilla.gecko.sync.net.BaseResource; +import org.mozilla.gecko.sync.net.SyncStorageCollectionRequest; +import org.mozilla.gecko.sync.net.SyncStorageRecordRequest; +import org.mozilla.gecko.sync.net.SyncStorageResponse; +import org.mozilla.gecko.sync.net.WBOCollectionRequestDelegate; +import org.mozilla.gecko.sync.net.WBORequestDelegate; +import org.mozilla.gecko.sync.repositories.NullCursorException; +import org.mozilla.gecko.sync.repositories.android.ClientsDatabaseAccessor; +import org.mozilla.gecko.sync.repositories.android.RepoUtils; +import org.mozilla.gecko.sync.repositories.domain.ClientRecord; +import org.mozilla.gecko.sync.repositories.domain.ClientRecordFactory; +import org.mozilla.gecko.sync.repositories.domain.VersionConstants; + +import ch.boye.httpclientandroidlib.HttpStatus; + +public class SyncClientsEngineStage extends AbstractSessionManagingSyncStage { + private static final String LOG_TAG = "SyncClientsEngineStage"; + + public static final String COLLECTION_NAME = "clients"; + public static final String STAGE_NAME = COLLECTION_NAME; + public static final int CLIENTS_TTL_REFRESH = 604800000; // 7 days in milliseconds. + public static final int MAX_UPLOAD_FAILURE_COUNT = 5; + public static final long NOTIFY_TAB_SENT_TTL_SECS = TimeUnit.SECONDS.convert(1L, TimeUnit.HOURS); // 1 hour + + protected final ClientRecordFactory factory = new ClientRecordFactory(); + protected ClientUploadDelegate clientUploadDelegate; + protected ClientDownloadDelegate clientDownloadDelegate; + + // Be sure to use this safely via getClientsDatabaseAccessor/closeDataAccessor. + protected ClientsDatabaseAccessor db; + + protected volatile boolean shouldWipe; + protected volatile boolean shouldUploadLocalRecord; // Set if, e.g., we received commands or need to refresh our version. + protected final AtomicInteger uploadAttemptsCount = new AtomicInteger(); + protected final List<ClientRecord> modifiedClientsToUpload = new ArrayList<ClientRecord>(); + + protected int getClientsCount() { + return getClientsDatabaseAccessor().clientsCount(); + } + + protected synchronized ClientsDatabaseAccessor getClientsDatabaseAccessor() { + if (db == null) { + db = new ClientsDatabaseAccessor(session.getContext()); + } + return db; + } + + protected synchronized void closeDataAccessor() { + if (db == null) { + return; + } + db.close(); + db = null; + } + + /** + * The following two delegates, ClientDownloadDelegate and ClientUploadDelegate + * are both triggered in a chain, starting when execute() calls + * downloadClientRecords(). + * + * Client records are downloaded using a get() request. Upon success of the + * get() request, the local client record is uploaded. + * + * @author Marina Samuel + * + */ + public class ClientDownloadDelegate extends WBOCollectionRequestDelegate { + + // We use this on each WBO, so lift it out. + final ClientsDataDelegate clientsDelegate = session.getClientsDelegate(); + boolean localAccountGUIDDownloaded = false; + + @Override + public AuthHeaderProvider getAuthHeaderProvider() { + return session.getAuthHeaderProvider(); + } + + @Override + public String ifUnmodifiedSince() { + // TODO last client download time? + return null; + } + + @Override + public void handleRequestSuccess(SyncStorageResponse response) { + + // Hang onto the server's last modified timestamp to use + // in X-If-Unmodified-Since for upload. + session.config.persistServerClientsTimestamp(response.normalizedWeaveTimestamp()); + BaseResource.consumeEntity(response); + + // Wipe the clients table if it still hasn't been wiped but needs to be. + wipeAndStore(null); + + // If we successfully downloaded all records but ours was not one of them + // then reset the timestamp. + if (!localAccountGUIDDownloaded) { + Logger.info(LOG_TAG, "Local client GUID does not exist on the server. Upload timestamp will be reset."); + session.config.persistServerClientRecordTimestamp(0); + } + localAccountGUIDDownloaded = false; + + final int clientsCount; + try { + clientsCount = getClientsCount(); + } finally { + // Close the database to clear cached readableDatabase/writableDatabase + // after we've completed our last transaction (db.store()). + closeDataAccessor(); + } + + Logger.debug(LOG_TAG, "Database contains " + clientsCount + " clients."); + Logger.debug(LOG_TAG, "Server response asserts " + response.weaveRecords() + " records."); + + // TODO: persist the response timestamp to know whether to download next time (Bug 726055). + clientUploadDelegate = new ClientUploadDelegate(); + clientsDelegate.setClientsCount(clientsCount); + + // If we upload remote records, checkAndUpload() will be called upon + // upload success in the delegate. Otherwise call checkAndUpload() now. + if (modifiedClientsToUpload.size() > 0) { + // modifiedClientsToUpload is cleared in uploadRemoteRecords, save what we need here + final List<String> devicesToNotify = new ArrayList<>(); + for (ClientRecord record : modifiedClientsToUpload) { + if (!TextUtils.isEmpty(record.fxaDeviceId)) { + devicesToNotify.add(record.fxaDeviceId); + } + } + + // This method is synchronous, there's no risk of notifying the clients + // before we actually uploaded the records + uploadRemoteRecords(); + + // Notify the clients who got their record written + notifyClients(devicesToNotify); + + return; + } + checkAndUpload(); + } + + private void notifyClients(final List<String> devicesToNotify) { + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final Context context = session.getContext(); + final Account account = FirefoxAccounts.getFirefoxAccount(context); + if (account == null) { + Log.e(LOG_TAG, "Can't notify other clients: no account"); + return; + } + final AndroidFxAccount fxAccount = new AndroidFxAccount(context, account); + final ExtendedJSONObject payload = createNotifyDevicesPayload(); + + final byte[] sessionToken; + try { + sessionToken = fxAccount.getSessionToken(); + } catch (AndroidFxAccount.InvalidFxAState invalidFxAState) { + Log.e(LOG_TAG, "Could not get session token", invalidFxAState); + return; + } + + // API doc : https://github.com/mozilla/fxa-auth-server/blob/master/docs/api.md#post-v1accountdevicesnotify + final FxAccountClient fxAccountClient = new FxAccountClient20(fxAccount.getAccountServerURI(), executor); + fxAccountClient.notifyDevices(sessionToken, devicesToNotify, payload, NOTIFY_TAB_SENT_TTL_SECS, new FxAccountClient20.RequestDelegate<ExtendedJSONObject>() { + @Override + public void handleError(Exception e) { + Log.e(LOG_TAG, "Error while notifying devices", e); + } + + @Override + public void handleFailure(FxAccountClientException.FxAccountClientRemoteException e) { + Log.e(LOG_TAG, "Error while notifying devices", e); + } + + @Override + public void handleSuccess(ExtendedJSONObject result) { + Log.i(LOG_TAG, devicesToNotify.size() + " devices notified"); + } + }); + } + + @NonNull + @SuppressWarnings("unchecked") + private ExtendedJSONObject createNotifyDevicesPayload() { + final ExtendedJSONObject payload = new ExtendedJSONObject(); + payload.put("version", 1); + payload.put("command", "sync:collection_changed"); + final ExtendedJSONObject data = new ExtendedJSONObject(); + final JSONArray collections = new JSONArray(); + collections.add("clients"); + data.put("collections", collections); + payload.put("data", data); + return payload; + } + + @Override + public void handleRequestFailure(SyncStorageResponse response) { + BaseResource.consumeEntity(response); // We don't need the response at all, and any exception handling shouldn't need the response body. + localAccountGUIDDownloaded = false; + + try { + Logger.info(LOG_TAG, "Client upload failed. Aborting sync."); + session.abort(new HTTPFailureException(response), "Client download failed."); + } finally { + // Close the database upon failure. + closeDataAccessor(); + } + } + + @Override + public void handleRequestError(Exception ex) { + localAccountGUIDDownloaded = false; + try { + Logger.info(LOG_TAG, "Client upload error. Aborting sync."); + session.abort(ex, "Failure fetching client record."); + } finally { + // Close the database upon error. + closeDataAccessor(); + } + } + + @Override + public void handleWBO(CryptoRecord record) { + ClientRecord r; + try { + r = (ClientRecord) factory.createRecord(record.decrypt()); + if (clientsDelegate.isLocalGUID(r.guid)) { + Logger.info(LOG_TAG, "Local client GUID exists on server and was downloaded."); + localAccountGUIDDownloaded = true; + handleDownloadedLocalRecord(r); + } else { + // Only need to store record if it isn't our local one. + wipeAndStore(r); + addCommands(r); + } + RepoUtils.logClient(r); + } catch (Exception e) { + session.abort(e, "Exception handling client WBO."); + return; + } + } + + @Override + public KeyBundle keyBundle() { + try { + return session.keyBundleForCollection(COLLECTION_NAME); + } catch (NoCollectionKeysSetException e) { + return null; + } + } + } + + public class ClientUploadDelegate extends WBORequestDelegate { + protected static final String LOG_TAG = "ClientUploadDelegate"; + public Long currentlyUploadingRecordTimestamp; + public boolean currentlyUploadingLocalRecord; + + @Override + public AuthHeaderProvider getAuthHeaderProvider() { + return session.getAuthHeaderProvider(); + } + + private void setUploadDetails(boolean isLocalRecord) { + // Use the timestamp for the whole collection per Sync storage 1.1 spec. + currentlyUploadingRecordTimestamp = session.config.getPersistedServerClientsTimestamp(); + currentlyUploadingLocalRecord = isLocalRecord; + } + + @Override + public String ifUnmodifiedSince() { + Long timestampInMilliseconds = currentlyUploadingRecordTimestamp; + + // It's the first upload so we don't care about X-If-Unmodified-Since. + if (timestampInMilliseconds <= 0) { + return null; + } + + return Utils.millisecondsToDecimalSecondsString(timestampInMilliseconds); + } + + @Override + public void handleRequestSuccess(SyncStorageResponse response) { + Logger.debug(LOG_TAG, "Upload succeeded."); + uploadAttemptsCount.set(0); + + // X-Weave-Timestamp is the modified time of uploaded records. + // Always persist this. + final long responseTimestamp = response.normalizedWeaveTimestamp(); + Logger.trace(LOG_TAG, "Timestamp from header is: " + responseTimestamp); + + if (responseTimestamp == -1) { + final String message = "Response did not contain a valid timestamp."; + session.abort(new RuntimeException(message), message); + return; + } + + BaseResource.consumeEntity(response); + session.config.persistServerClientsTimestamp(responseTimestamp); + + // If we're not uploading our record, we're done here; just + // clean up and finish. + if (!currentlyUploadingLocalRecord) { + // TODO: check failed uploads in body. + clearRecordsToUpload(); + checkAndUpload(); + return; + } + + // If we're processing our record, we have a little more cleanup + // to do. + shouldUploadLocalRecord = false; + session.config.persistServerClientRecordTimestamp(responseTimestamp); + session.advance(); + } + + @Override + public void handleRequestFailure(SyncStorageResponse response) { + int statusCode = response.getStatusCode(); + + // If upload failed because of `ifUnmodifiedSince` then there are new + // commands uploaded to our record. We must download and process them first. + if (!shouldUploadLocalRecord || + statusCode == HttpStatus.SC_PRECONDITION_FAILED || + uploadAttemptsCount.incrementAndGet() > MAX_UPLOAD_FAILURE_COUNT) { + + Logger.debug(LOG_TAG, "Client upload failed. Aborting sync."); + if (!currentlyUploadingLocalRecord) { + modifiedClientsToUpload.clear(); // These will be redownloaded. + } + BaseResource.consumeEntity(response); // The exception thrown should need the response body. + session.abort(new HTTPFailureException(response), "Client upload failed."); + return; + } + Logger.trace(LOG_TAG, "Retrying upload…"); + // Preconditions: + // shouldUploadLocalRecord == true && + // statusCode != 412 && + // uploadAttemptCount < MAX_UPLOAD_FAILURE_COUNT + checkAndUpload(); + } + + @Override + public void handleRequestError(Exception ex) { + Logger.info(LOG_TAG, "Client upload error. Aborting sync."); + session.abort(ex, "Client upload failed."); + } + + @Override + public KeyBundle keyBundle() { + try { + return session.keyBundleForCollection(COLLECTION_NAME); + } catch (NoCollectionKeysSetException e) { + return null; + } + } + } + + @Override + public void execute() throws NoSuchStageException { + // We can be disabled just for this sync. + boolean enabledThisSync = session.isEngineLocallyEnabled(STAGE_NAME); + if (!enabledThisSync) { + // These log messages look best when they match the messages in ServerSyncStage. + Logger.debug(LOG_TAG, "Stage " + STAGE_NAME + " disabled just for this sync."); + Logger.info(LOG_TAG, "Skipping stage " + STAGE_NAME + "."); + session.advance(); + return; + } + + if (shouldDownload()) { + downloadClientRecords(); // Will kick off upload, too… + } else { + // Upload if necessary. + } + } + + @Override + protected void resetLocal() { + // Clear timestamps and local data. + session.config.persistServerClientRecordTimestamp(0L); // TODO: roll these into one. + session.config.persistServerClientsTimestamp(0L); + + session.getClientsDelegate().setClientsCount(0); + try { + getClientsDatabaseAccessor().wipeDB(); + } finally { + closeDataAccessor(); + } + } + + @Override + protected void wipeLocal() throws Exception { + // Nothing more to do. + this.resetLocal(); + } + + @Override + public Integer getStorageVersion() { + return VersionConstants.CLIENTS_ENGINE_VERSION; + } + + protected String getLocalClientVersion() { + return AppConstants.MOZ_APP_VERSION; + } + + @SuppressWarnings("unchecked") + protected JSONArray getLocalClientProtocols() { + final JSONArray protocols = new JSONArray(); + protocols.add(ClientRecord.PROTOCOL_LEGACY_SYNC); + protocols.add(ClientRecord.PROTOCOL_FXA_SYNC); + return protocols; + } + + protected ClientRecord newLocalClientRecord(ClientsDataDelegate delegate) { + final String ourGUID = delegate.getAccountGUID(); + final String ourName = delegate.getClientName(); + + ClientRecord r = new ClientRecord(ourGUID); + r.name = ourName; + r.version = getLocalClientVersion(); + r.protocols = getLocalClientProtocols(); + + r.os = "Android"; + r.application = AppConstants.MOZ_APP_DISPLAYNAME; + r.appPackage = AppConstants.ANDROID_PACKAGE_NAME; + r.device = android.os.Build.MODEL; + r.formfactor = delegate.getFormFactor(); + + Context context = session.getContext(); + final Account account = FirefoxAccounts.getFirefoxAccount(context); + if (account != null) { + final AndroidFxAccount fxAccount = new AndroidFxAccount(context, account); + final String deviceId = fxAccount.getDeviceId(); + if (!TextUtils.isEmpty(deviceId)) { + r.fxaDeviceId = deviceId; + } + } + + return r; + } + + // TODO: Bug 726055 - More considered handling of when to sync. + protected boolean shouldDownload() { + // Ask info/collections whether a download is needed. + return true; + } + + protected boolean shouldUpload() { + if (shouldUploadLocalRecord) { + return true; + } + + long lastUpload = session.config.getPersistedServerClientRecordTimestamp(); // Defaults to 0. + if (lastUpload == 0) { + return true; + } + + if (session.getClientsDelegate().getLastModifiedTimestamp() > lastUpload) { + // Something's changed locally since we last uploaded. + return true; + } + + // Note the opportunity for clock drift problems here. + // TODO: if we track download times, we can use the timestamp of most + // recent download response instead of the current time. + long now = System.currentTimeMillis(); + long age = now - lastUpload; + return age >= CLIENTS_TTL_REFRESH; + } + + protected void handleDownloadedLocalRecord(ClientRecord r) { + session.config.persistServerClientRecordTimestamp(r.lastModified); + + if (!getLocalClientVersion().equals(r.version) || + !getLocalClientProtocols().equals(r.protocols)) { + shouldUploadLocalRecord = true; + } + processCommands(r.commands); + } + + protected void processCommands(JSONArray commands) { + if (commands == null || + commands.size() == 0) { + return; + } + + shouldUploadLocalRecord = true; + CommandProcessor processor = CommandProcessor.getProcessor(); + + for (Object o : commands) { + processor.processCommand(session, new ExtendedJSONObject((JSONObject) o)); + } + } + + @SuppressWarnings("unchecked") + protected void addCommands(ClientRecord record) throws NullCursorException { + Logger.trace(LOG_TAG, "Adding commands to " + record.guid); + List<Command> commands = db.fetchCommandsForClient(record.guid); + + if (commands == null || commands.size() == 0) { + Logger.trace(LOG_TAG, "No commands to add."); + return; + } + + for (Command command : commands) { + JSONObject jsonCommand = command.asJSONObject(); + if (record.commands == null) { + record.commands = new JSONArray(); + } + record.commands.add(jsonCommand); + } + modifiedClientsToUpload.add(record); + } + + @SuppressWarnings("unchecked") + protected void uploadRemoteRecords() { + Logger.trace(LOG_TAG, "In uploadRemoteRecords. Uploading " + modifiedClientsToUpload.size() + " records" ); + + for (ClientRecord r : modifiedClientsToUpload) { + Logger.trace(LOG_TAG, ">> Uploading record " + r.guid + ": " + r.name); + } + + if (modifiedClientsToUpload.size() == 1) { + ClientRecord record = modifiedClientsToUpload.get(0); + Logger.debug(LOG_TAG, "Only 1 remote record to upload."); + Logger.debug(LOG_TAG, "Record last modified: " + record.lastModified); + CryptoRecord cryptoRecord = encryptClientRecord(record); + if (cryptoRecord != null) { + clientUploadDelegate.setUploadDetails(false); + this.uploadClientRecord(cryptoRecord); + } + return; + } + + JSONArray cryptoRecords = new JSONArray(); + for (ClientRecord record : modifiedClientsToUpload) { + Logger.trace(LOG_TAG, "Record " + record.guid + " is being uploaded" ); + + CryptoRecord cryptoRecord = encryptClientRecord(record); + cryptoRecords.add(cryptoRecord.toJSONObject()); + } + Logger.debug(LOG_TAG, "Uploading records: " + cryptoRecords.size()); + clientUploadDelegate.setUploadDetails(false); + this.uploadClientRecords(cryptoRecords); + } + + protected void checkAndUpload() { + if (!shouldUpload()) { + Logger.debug(LOG_TAG, "Not uploading client record."); + session.advance(); + return; + } + + final ClientRecord localClient = newLocalClientRecord(session.getClientsDelegate()); + clientUploadDelegate.setUploadDetails(true); + CryptoRecord cryptoRecord = encryptClientRecord(localClient); + if (cryptoRecord != null) { + this.uploadClientRecord(cryptoRecord); + } + } + + protected CryptoRecord encryptClientRecord(ClientRecord recordToUpload) { + // Generate CryptoRecord from ClientRecord to upload. + final String encryptionFailure = "Couldn't encrypt new client record."; + + try { + CryptoRecord cryptoRecord = recordToUpload.getEnvelope(); + cryptoRecord.keyBundle = clientUploadDelegate.keyBundle(); + if (cryptoRecord.keyBundle == null) { + session.abort(new NoCollectionKeysSetException(), "No collection keys set."); + return null; + } + return cryptoRecord.encrypt(); + } catch (UnsupportedEncodingException e) { + session.abort(e, encryptionFailure + " Unsupported encoding."); + } catch (CryptoException e) { + session.abort(e, encryptionFailure); + } + return null; + } + + public void clearRecordsToUpload() { + try { + getClientsDatabaseAccessor().wipeCommandsTable(); + modifiedClientsToUpload.clear(); + } finally { + closeDataAccessor(); + } + } + + protected void downloadClientRecords() { + shouldWipe = true; + clientDownloadDelegate = makeClientDownloadDelegate(); + + try { + final URI getURI = session.config.collectionURI(COLLECTION_NAME, true); + final SyncStorageCollectionRequest request = new SyncStorageCollectionRequest(getURI); + request.delegate = clientDownloadDelegate; + + Logger.trace(LOG_TAG, "Downloading client records."); + request.get(); + } catch (URISyntaxException e) { + session.abort(e, "Invalid URI."); + } + } + + protected void uploadClientRecords(JSONArray records) { + Logger.trace(LOG_TAG, "Uploading " + records.size() + " client records."); + try { + final URI postURI = session.config.collectionURI(COLLECTION_NAME, false); + final SyncStorageRecordRequest request = new SyncStorageRecordRequest(postURI); + request.delegate = clientUploadDelegate; + request.post(records); + } catch (URISyntaxException e) { + session.abort(e, "Invalid URI."); + } catch (Exception e) { + session.abort(e, "Unable to parse body."); + } + } + + /** + * Upload a client record via HTTP POST to the parent collection. + */ + protected void uploadClientRecord(CryptoRecord record) { + Logger.debug(LOG_TAG, "Uploading client record " + record.guid); + try { + final URI postURI = session.config.collectionURI(COLLECTION_NAME); + final SyncStorageRecordRequest request = new SyncStorageRecordRequest(postURI); + request.delegate = clientUploadDelegate; + request.post(record); + } catch (URISyntaxException e) { + session.abort(e, "Invalid URI."); + } + } + + protected ClientDownloadDelegate makeClientDownloadDelegate() { + return new ClientDownloadDelegate(); + } + + protected void wipeAndStore(ClientRecord record) { + final ClientsDatabaseAccessor db = getClientsDatabaseAccessor(); + if (shouldWipe) { + db.wipeClientsTable(); + shouldWipe = false; + } + if (record != null) { + db.store(record); + } + } +} |