Develop emax

pipeline/archiver/downloader/downloader.js

@@ -70,38 +70,46 @@ async function download(app) {
         appSavePath = await resolveAPKDir(app);
     } catch (err) {
         await new Promise((resolve) => setTimeout(resolve, 6000));
-        return Promise.reject('Did not have access to resolve dir', err.message);
+        throw new Error(`Did not have access to resolve dir ${err.message}`);
     }
 
     try {
         await downloadApp(app, appSavePath);
     } catch (err) {
         logger.debug('Attempting to remove created dir');
         await fs.rmdir(appSavePath).catch(logger.warning);
-        return Promise.reject('Downloading failed with err:', err.message);
+        throw new Error(`Downloading failed with err: ${err.message}`);
     }
 
-    try {
-        const apkPath = path.join(appSavePath, `${app.app}.apk`);
-
-        if (fs.existsSync(apkPath)) {
-            // Perform a check on apk size
-            await fs.stat(apkPath, async(err, stats) => {
-                if (stats.size == 0 || stats.size == undefined) {
-                    await fs.rmdir(appSavePath).catch(logger.warning);
-                    return Promise.reject('File did not successfully download and is a empty size');
-                }
-
-                await db.updateDownloadedApp(app);
-                return undefined;
-            });
+    const apkPath = path.join(appSavePath, `${app.app}.apk`);
+
+    if (fs.existsSync(apkPath)) {
+        // Perform a check on apk size
+        let stats;
+        try {
+            stats = fs.statSync(apkPath);
+        } catch (err) {
+            throw new Error(`Unable to stat file: ${err.message}`);
+        }
+        if (stats.size == 0 || stats.size == undefined) {
+            try {
+                fs.unlinkSync(apkPath);
+                fs.rmdirSync(appSavePath);
+            } catch (e) {
+                logger.warning(e);
+            }
+            // old version ->  await fs.rmdir(appSavePath).catch(logger.warning);
+            throw new Error('File did not successfully download and is a empty size');
+        }
+
+        try {
+            await db.updateDownloadedApp(app);
+        } catch (err) {
+            // TODO: Maybe do something else? Destroying process as we have apks that
+            // don't exist in db...
+            throw new Error(`Err when updated the downloaded app ${err.message}`);
         }
-    } catch (err) {
-        // TODO: Maybe do something else? Destroying process as we have apks that
-        // don't exist in db...
-        return Promise.reject('Err when updated the downloaded app', err);
     }
-    return undefined;
 }
 
 async function main() {

pipeline/archiver/freezer/freezer.js

@@ -0,0 +1,148 @@
+
+
+const gplay = require('google-play-scraper'),
+      fs = require('fs'),
+      config = require('/etc/xray/config.json'),
+      { Pool, Client } = require('pg'),
+      _ = require('lodash'),
+      path = require('path'),
+      Promise = require('bluebird'),
+      logger = require('../../util/logger'),
+      { exec } = require('child_process');
+
+const region = 'uk';
+
+function walk(dir, callback) {
+    const files = fs.readdirSync(dir);
+    files.map(function(file) {
+	var filepath = path.join(dir, file), 
+	    stats = fs.statSync(filepath);
+	if (stats.isDirectory()) {
+	    walk(filepath, callback);
+	} else if (stats.isFile()) {
+	    callback(filepath, stats);
+	}
+    });
+}
+
+
+function getAnalysis(client, appId, appVersion) { //
+    console.log('getAnalysis ', appId, appVersion);
+    return client.query('select app_versions.app,app_versions.version,app_versions.analyzed,app_analyses.analyzer,app_analyses.analysis from app_versions left outer join app_analyses on app_analyses.id = app_versions.id WHERE app_versions.app = $1 AND app_versions.version = $2', [appId, appVersion]).then((res) => {
+	console.log('!!!!!!!!!!!!!!!!!!!!!!! >>>>>>>>>>>>> result ', res.rows);
+	// console.log(`app query res count: ${res.rowCount}`);
+	return res.rows[0];
+    }).catch((e) => {
+	console.error('Error query ', e);
+	logger.err('Error checking if app exists in the database:', err);
+	throw err;
+    });
+}
+
+// relocation src/dst
+const source = config.freezer.sync && config.freezer.sync.src,
+      dest = config.freezer.sync && config.freezer.sync.dest;
+
+function processAPK(client, dirname) {
+    // returns the path to the apk if apk exists
+    const fns = [];
+    walk(dirname, (f, src_stats) => {
+	fns.push(() => {
+	    if (f.toLowerCase().indexOf('.apk') >= 0 && f.toLowerCase().indexOf('.apk') === f.length - '.apk'.length) {
+		// console.log(`got a candidate! ${f.slice(source.length+1)}`)
+		const apk_path = f.slice(source.length+1),
+		      split = f.split('/'),
+		      version = split[split.length - 2],
+		      pkg = apk_path.slice(0,apk_path.indexOf('/'));
+
+		console.log(`pkg path ${apk_path} - pkg:${pkg}, version:${version}`);
+
+		const winMatch = () => {
+		    // matches already, so no need to sync.
+		    return getAnalysis(client, pkg, version).then((has) => {
+		    	console.log('getAnalysis results ', has);
+			// interpret results.. and if successful then check to delete file!
+			if (!has.analysis) {
+			    console.info('i think we have a false');
+			    // then do analysis here...
+			    // then deletion			    
+			} else {			    
+    			    console.info('i think we have a true');
+			    // just go straight to deletion
+			}
+			return Promise.resolve();
+		    }).catch((ea) => {
+		    	console.error('ERROR with getanalysis', ea);
+		    });
+		}, failMatch = () => {
+		    // failmatch so we first sync if
+		    console.log(`sync ${config.freezer.sync}, ${config.freezer.sync.enable}`);
+		    if (config.freezer.sync && config.freezer.sync.enable) {
+			// first rsync then ...
+			return new Promise((win, fail) => {
+			    logger.info(`running rsync >> /usr/bin/rsync -avz ${path.join(source, pkg)} ${dest}/`);
+			    exec(`/usr/bin/rsync -avz ${path.join(source, pkg)} ${dest}/`, (error, stdout, stderr) => {
+				if (error) {
+				    logger.error(`rsync error: ${error}`);
+				    return fail();
+				}
+				logger.info(`rsync output ${stdout} ${stderr}`);
+				return win();
+			    });
+			});			
+		    }
+		    return Promise.reject('error: sync is not enabled');
+		};
+
+		try {
+		    const dstat =  fs.statSync(path.join(dest, apk_path));
+		    if (dstat && dstat.isFile() && dstat.size === src_stats.size) {
+			return winMatch();
+		    } else {
+			logger.info('exists on cold but mismatch ', path.join(dest, apk_path), ' size ', dstat.size, ' <> ', src_stats.size);
+			return failMatch().then(() => winMatch());
+		    }
+		} catch(e) {
+		    if (e && e.errno === -2) { 
+			logger.info('no such file on dest ', path.join(dest, apk_path));
+			return failMatch().then(() => winMatch());
+		    } else {
+			logger.error(`unknown error ${e}, so skipping ${dest}`);
+			return Promise.resolve();
+		    }
+		}
+	    }
+	});
+    });
+    // now we have a list of things we can reduce
+    return Promise.reduce(fns, (reduction, f) => f(), {});
+};
+
+
+function connect() {
+    const dbCfg = _.extend({}, config.db, config.freezer.db);
+
+    dbCfg.max = 10;
+    dbCfg.idleTimeoutMillis = 30000;
+
+    console.info(`authenticating using u:${dbCfg.user}, p:${dbCfg.password}`);
+    console.log(dbCfg);
+
+    const client = new Client(dbCfg);
+    client.connect();
+    return client;
+}
+
+
+(()=> {
+    const client = connect();
+    client.query('select * from app_versions limit 1;').then(async (res) => {
+	console.log('test query res ', res.rows);
+	fs.readdir(source, (err, items) => {
+	    console.log(`got ${items.length} items`);
+	    Promise.reduce(items, (reductor, dirname) => processAPK(client, path.join(source,dirname)), {});
+	});
+    });
+
+})();
+

pipeline/archiver/retriever/retriever-async.js

@@ -0,0 +1,59 @@
+
+const gplay = require('google-play-scraper');
+const _ = require('lodash');
+const Promise = require('bluebird');
+
+const logger = require('../../util/logger');
+const db = new (require('../../db/db'))('retriever');
+
+const region = 'uk';
+
+/**
+ * Inserts app data into the db using db.js
+ * @param {*The app data json that is to be inserted into the databae.} appData
+ */
+function insertAppData(appData) {
+    // Checking version data - correct version to update date
+    if (!appData.version || appData.version === 'Varies with device') {
+        logger.debug('Version not found defaulting too', appData.updated);
+        // let formatDate = appData.updated.replace(/\s+/g, '').replace(',', '/');
+        const formatDate = new Date(appData.updated).toISOString().substring(0, 10);
+        appData.version = formatDate;
+    }
+
+    // push the app data to the DB
+    return db.insertPlayApp(appData, region);
+}
+
+// TODO Add Permission list to app Data JSON
+function fetchAppData(searchTerm, numberOfApps, perSecond) {
+    return gplay.search({
+        term: searchTerm,
+        num: numberOfApps,
+        throttle: perSecond,
+        country: region,
+        fullDetail: true,
+    }).then((appDatas) => {
+	// TODO: Move this to DB.
+	return Promise.all(_.map(appDatas, (appData) => {
+            logger.debug(`inserting ${appData.title} to the DB`);
+            return db.doesAppExist(appData).then((appExists) => {
+		if (!appExists) {
+		    return insertAppData(appData).catch((err) => logger.err(err));
+		} else {
+		    logger.debug('App already exists', appData.appId);
+		    return Promise.resolve();
+		}
+	    });
+	}));
+    });
+}
+
+(() => {
+    return Promise.reduce(db.getStaleSearchTerms(), (res, dbRow) => {
+        logger.info(`searching for: ${dbRow.search_term}`);
+        return fetchAppData(dbRow.search_term, 60, 1)
+            .then(() => db.updateLastSearchedDate(dbRow.search_term).catch(logger.err))
+	    .catch(logger.err);
+    }, Promise.resolve()).then(() => console.log('doneeezo'));
+})();