"use strict";
var EXPORTED_SYMBOLS = ["Sync"];
var {Service: Service} = require("service");
var {Observers: Observers} = require("observers");
var {Auth: Auth, LOGIN_STATES: LOGIN_STATES} = require("auth");
var {Protobuf: Protobuf} = require("protobuf");
var MAX_SQL_ITEMS = 512;
var Sync = {
init: function Sync_init() {
this._API = require("api").API;
this.database.execQuerySpinningly(this._queries.FLUSH_ALL_RECORDS);
Observers.add("ybar:esync:auth:changed",this);
}
,
finalize: function Sync_finalize() {
this._API.stopAllRequests();
this._API = null;
Observers.remove("ybar:esync:auth:changed",this);
this._closeDatabase();
}
,
observe: function Sync_observe(topic, subject, data) {
switch (topic) {
case "yabar:esync:auth:changed":
this._closeDatabase();
break;
}

}
,
commit: function Sync_commit() {
var result = this.database.execQuerySpinningly(this._queries.SELECT_UNPROCESSED_RECORDS);
if (! result.length)
return;
var processQueryOnLIst = (function processQueryOnLIst(list, queryType) {
for (let i = 0, length = Math.ceil(list.length / MAX_SQL_ITEMS);i < length;i++) {
let listPart = list.slice(i * MAX_SQL_ITEMS,(i + 1) * MAX_SQL_ITEMS);
this.database.execQuerySpinningly(this._queries[queryType](listPart),listPart);
}

}
).bind(this,result.map(function (row) row.id));
processQueryOnLIst("SET_PROCESSING");
var commit = Protobuf.getCommitMessage(result);
if (! commit)
{
processQueryOnLIst("SET_UNPROCESSING");
return;
}

this._API.command(commit,(function (xhr) {
var sendedData = result.map(function (row) {
var data = JSON.parse(row.data);
data.engine = row.engine;
data.deleted = ! ! row.on_delete;
return data;
}
);
var protobufMessage = this._getIncomingProtobufMessage(xhr.response);
if (! protobufMessage)
{
processQueryOnLIst("SET_UNPROCESSING");
require("scheduler").Scheduler.plan("sync");
return;
}

require("scheduler").Scheduler.cleanupPlansData();
this._processCommitIncoming(protobufMessage,sendedData);
processQueryOnLIst("FLUSH_UPDATED_RECORDS");
}
).bind(this),(function (err, xhr) {
processQueryOnLIst("SET_UNPROCESSING");
this._logger.debug("Commit error: " + err);
if (this._processHTTPError(err,xhr.status))
require("scheduler").Scheduler.plan("sync"); else
require("scheduler").Scheduler.cleanupPlansData();
}
).bind(this));
}
,
update: function Sync_update(engineNames) {
function notifyEngines(state) {
if (! Service.enabled)
return;
engineNames.forEach(function (engineName) {
var engine = this.get(engineName);
if (engine)
engine.emit("update",state);
}
,Service.engineManager);
}

var commit = Protobuf.getUpdatesMessage(engineNames);
this._API.command(commit,(function (xhr) {
var protobufMessage = this._getIncomingProtobufMessage(xhr.response);
if (! protobufMessage)
{
notifyEngines("error");
require("scheduler").Scheduler.plan("update",engineNames);
return;
}

require("scheduler").Scheduler.cleanupPlansData();
this._processUpdateIncoming(protobufMessage);
notifyEngines("finish");
}
).bind(this),(function (err, xhr) {
this._logger.debug("Update error: " + err);
notifyEngines("error");
if (this._processHTTPError(err,xhr.status))
require("scheduler").Scheduler.plan("update",engineNames); else
require("scheduler").Scheduler.cleanupPlansData();
}
).bind(this));
}
,
drop: function Sync_drop() {
this._API.drop(this._onDrop.bind(this));
}
,
_onDrop: function Sync__onDrop() {
if (! Service.enabled)
return;
this._logger.debug("Drop");
var registeredEngineNames = Service.engineManager.registeredEngines;
Auth.logout();
Service.disable();
this._removeDatabase();
var settings = NativeAPI.Settings;
registeredEngineNames.forEach(function (engineName) {
settings.setValue("engine." + engineName + ".token","");
}
);
settings.setValue("xmpp.token","");
settings.setValue("store.birthday","");
}
,
_processCommitIncoming: function Sync__processCommitIncoming(protobufMessage, sendedData) {
if (! protobufMessage)
return false;
var commit = protobufMessage.commit;
var entries = commit && commit.EntryResponse;
if (! entries)
return false;
var entriesByEngine = {
};
Array.forEach(entries,function (entry, index) {
if (entry.response_type !== Protobuf.sync_pb.CommitResponse.ResponseType.SUCCESS)
return;
var {engineName: engineName, data: data} = Protobuf.parseProtobufEntry(entry,sendedData[index]);
this._logger.debug("_processCommitIncoming for '",engineName,"'\n",JSON.stringify(data,null,"\t").replace(/("(?:username|password)_value":\s*")[^"]+(",?)/g,"$1*****$2"));
if (! engineName || ! data)
return;
data.originator_cache_guid = Auth.token.guid;
entriesByEngine[engineName] = entriesByEngine[engineName] || [];
entriesByEngine[engineName].push(data);
}
,this);
const engineManager = Service.engineManager;
for(let [engineName, entries] in Iterator(entriesByEngine)) {
let engine = engineManager.get(engineName);
if (engine)
engine.setListData(entries);
}

return true;
}
,
_processUpdateIncoming: function Sync__processUpdateIncoming(protobufMessage) {
if (! protobufMessage)
return false;
if (protobufMessage.store_birthday)
Auth.storeBirthday = protobufMessage.store_birthday;
var updates = protobufMessage.get_updates;
if (! updates)
return false;
var entriesByEngine = {
};
var entries = updates.entries;
var guid = Auth.token.guid;
Array.forEach(entries,function (entry) {
if (! entry)
return;
var {engineName: engineName, data: data} = Protobuf.parseProtobufEntry(entry);
if (! engineName || ! data)
return;
entriesByEngine[engineName] = entriesByEngine[engineName] || [];
entriesByEngine[engineName].push(data);
}
,this);
const engineManager = Service.engineManager;
for(let [engineName, entries] in Iterator(entriesByEngine)) {
let engine = engineManager.get(engineName);
if (engine)
engine.setListData(entries);
}

if (updates.new_progress_marker)
{
Array.forEach(updates.new_progress_marker,function (marker) {
var engineName = Protobuf.getEngineNameById(marker.data_type_id);
if (! engineName)
return;
var engine = engineManager.get(engineName);
if (! engine)
return;
engine.token = marker.token.join("");
}
,this);
}

return true;
}
,
_getIncomingProtobufMessage: function Sync__getIncomingProtobufMessage(serverResponse) {
var protobufMessage;
try {
protobufMessage = Protobuf.parseBlob(serverResponse);
}
catch (e) {
this._logger.error("Error parse commit response");
this._logger.debug(e);
return null;
}

this._logger.debug("_incomingProtobufMessage: ",protobufMessage);
var error = protobufMessage.error;
var code = error && error.error_type || protobufMessage.error_code;
var action = error && error.action || null;
if (this._processProtobufError(code,action))
return null;
return protobufMessage;
}
,
_processHTTPError: function Sync_processError(error, status) {
if (status === 401)
{
if (Auth.authorized)
{
Auth.logout();
Auth.state = LOGIN_STATES.EXPIRED;
return false;
}

}

if (error)
return true;
return false;
}
,
_processProtobufError: function Sync__processProtobofError(code, currentAction) {
this._logger.debug("Error response: type: ",code,", action: ",currentAction);
var type = Protobuf.sync_pb.SyncEnums.ErrorType;
switch (code) {
case type.SUCCESS:
break;
case type.NOT_MY_BIRTHDAY:
break;
case type.THROTTLED:
break;
case type.CLEAR_PENDING:
break;
case type.TRANSIENT_ERROR:
return true;
case type.ACCESS_DENIED:

case type.USER_NOT_ACTIVATED:

case type.AUTH_EXPIRED:

case type.AUTH_INVALID:

case type.MIGRATION_DONE:

case type.UNKNOWN:
break;
default:
break;
}

var action = Protobuf.sync_pb.SyncEnums.Action;
switch (currentAction) {
case action.UPGRADE_CLIENT:

case action.ENABLE_SYNC_ON_ACCOUNT:

case action.STOP_AND_RESTART_SYNC:
break;
case action.CLEAR_USER_DATA_AND_RESYNC:
this._clearDataAndResync();
return true;
case action.DISABLE_SYNC_ON_CLIENT:
this._onDrop();
return true;
case action.UNKNOWN_ACTION:

default:
break;
}

return false;
}
,
_clearDataAndResync: function Sync__clearDataAndResync() {
Auth.storeBirthday = "";
var engineManager = Service.engineManager;
var settings = NativeAPI.Settings;
var registeredEngines = engineManager.registeredEngines;
engineManager.finalize();
registeredEngines.forEach(function (engineName) {
settings.setValue("engine." + engineName + ".token","");
this.database.execQuerySpinningly("DROP TABLE IF EXISTS " + engineName + ";");
}
,this);
settings.setValue("xmpp.token","");
engineManager.init();
}
,
get database() {
if (! this._db)
{
let syncDBFile = NativeAPI.Files.getPackageStorage(true);
syncDBFile.append(this.dbName);
this._db = NativeAPI.Database.createInstance(syncDBFile,this._queries.INIT_RECORD_TABLE);
this._db.execQuerySpinningly("SELECT * FROM sqlite_master WHERE type = 'trigger';").forEach(function (trigger) {
this._db.execQuerySpinningly("DROP TRIGGER IF EXISTS " + trigger.name + ";");
}
,this);
}

return this._db;
}
,
get dbName() {
var username = Auth.token.username;
if (! username)
return null;
return encodeURIComponent(username) + ".sync.sqlite";
}
,
get _logger() {
delete this._logger;
return this._logger = NativeAPI.logger.getLogger("Sync");
}
,
_protobuf: null,
_API: null,
_db: null,
_closeDatabase: function Sync__closeDatabase() {
if (this._db)
{
this._db.close();
this._db = null;
}

}
,
_removeDatabase: function Sync__removeDatabase() {
var file = this.database.storageFile;
this.database.close(function () file.remove(false));
this._db = null;
}
,
_queries: {
INIT_RECORD_TABLE: ["CREATE TABLE IF NOT EXISTS record ( ", "id INTEGER PRIMARY KEY, ", "engine TEXT, ", "data TEXT, ", "on_delete BOOLEAN, ", "processing BOOLEAN ", ");"].join(""),
SET_PROCESSING: function SET_PROCESSING(list) "UPDATE record SET processing = 1 WHERE id IN ( " + list.map(function () "?").join(", ") + ");",
SET_UNPROCESSING: function SET_UNPROCESSING(list) "UPDATE record SET processing = 0 WHERE id IN ( " + list.map(function () "?").join(", ") + ");",
SET_ALL_AS_UNPROCESSING: "UPDATE record SET processing = 0 WHERE processing = 1;",
SELECT_UNPROCESSED_RECORDS: "SELECT * FROM record WHERE processing = 0;",
FLUSH_UPDATED_RECORDS: function FLUSH_UPDATED_RECORDS(list) "DELETE FROM record WHERE id IN ( " + list.map(function () "?").join(", ") + ");",
FLUSH_ALL_RECORDS: "DELETE FROM record;"}};
