/* Copyright 2017 Open Ag Data Alliance
 *
 * Licensed under the Apache License, Version 2.0 (the 'License');
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an 'AS IS' BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

'use strict';

const debug = require('debug');
const trace = debug('webhooks:trace');
const info = debug('webhooks:info');
const error = debug('webhooks:error');
var Promise = require('bluebird');
const uuid = require('uuid');

const {Responder, Requester} = require('../../libs/oada-lib-kafka');
const {users} = require('../../libs/oada-lib-arangodb');
const config = require('./config');
const contentTypes = {
    bookmarks: 'application/vnd.oada.bookmarks.1+json',
    shares: 'application/vnd.oada.shares.1+json'
};

const responder = new Responder(
        config.get('kafka:topics:userRequest'),
        config.get('kafka:topics:httpResponse'),
        'users');
const requester = new Requester(
        config.get('kafka:topics:httpResponse'),
        config.get('kafka:topics:writeRequest'),
        'users');

module.exports = function stopResp() {
    return responder.disconnect();
};

responder.on('request', function handleReq(req) {
    // TODO: Sanitize?
    let user = req.user;

    return users.create(user, true)
        .then(function ensureUserResources(user) {
            // Create empty resources for user
            ['bookmarks', 'shares'].forEach(function ensureResource(res) {
                if (!(user[res] && user[res]._id)) {
                    let resid = 'resources/' + uuid();

                    trace(`Creating ${resid} for ${res} of ${user.id}`);
                    user[res] =
                        requester.send({
                            'url': '/' + resid,
                            'resource_id': '',
                            'path_leftover': '/' + resid,
                            'meta_id': resid + '/_meta',
                            'user_id': user._id,
                            // TODO: What to put for these?
                            //'authorizationid': ,
                            //'client_id': ,
                            'contentType': contentTypes[res],
                            'body': {}
                        }).tap(resp => {
                            if (resp.code !== 'success') {
                                // TODO: Clean up on failure?
                                error(resp.code);
                                let err = new Error(`Failed to create ${res}`);
                                return Promise.reject(err);
                            }
                        }).return({_id: resid});
                }
            });

            return user;
        })
        .props()
        .then(users.update)
        .tap(user => info(`Created user ${user._id}`))
        .then(user => {
            return {
                code: 'success',
                new: true,
                user,
            };
        })
		.catch(users.UniqueConstraintError, () => {
            info(`User ${user} already exists`);
            // TODO: Implement updating users?
            return users.like(user).call('next').then(user => {
                return {
                    code: 'success',
                    new: false,
                    user
                };
            });
        })
        .catch(err => {
            error(err);
            return {code: err.message || 'error'}
        });
});

@@ -840,18 +840,16 @@
 esponder
-, 
 Requeste
@@ -1147,25 +1147,70 @@
 esponder
-(%0A       
+Requester(%7B%0A    requestTopics: %7B%0A        produceTopic:
  config.
@@ -1227,20 +1227,21 @@
 :topics:
-user
+write
 Request'
@@ -1242,32 +1242,46 @@
 quest'),%0A       
+ consumeTopic:
  config.get('kaf
@@ -1314,111 +1314,115 @@
     
+%7D,%0A
     
-'users');%0Aconst requester = new Requester(%0A        config.get('kafka:topics:httpResponse'),%0A       
+respondTopics: %7B%0A        consumeTopic: config.get('kafka:topics:userRequest'),%0A        produceTopic:
  con
@@ -1443,36 +1443,36 @@
 :topics:
-writeRequest
+httpResponse
 '),%0A    
     'use
@@ -1467,23 +1467,44 @@
     
+%7D,%0A
     
-'users'
+group: 'user-handlers'%0A%7D
 );%0A%0A
+%0A
 modu
@@ -1929,20 +1929,23 @@
 ser%5Bres%5D
-.
+%5B'
 _id
+'%5D
 )) %7B%0A   
@@ -2128,21 +2128,21 @@
       re
-quest
+spond
 er.send(
@@ -2396,20 +2396,23 @@
 d': user
-.
+%5B'
 _id
+'%5D
 ,%0A      
@@ -2745,17 +2745,17 @@
 sp.code 
-!
+=
 == 'succ
@@ -2758,24 +2758,119 @@
 success') %7B%0A
+                                return Promise.resolve();%0A                            %7D else %7B%0A
             
@@ -3354,12 +3354,15 @@
 user
-.
+%5B'
 _id
+'%5D
 %7D%60))
@@ -3388,35 +3388,11 @@
  =%3E 
+(
 %7B%0A
-            return %7B%0A    
     
@@ -3424,28 +3424,24 @@
             
-    
 new: true,%0A 
@@ -3439,20 +3439,16 @@
 : true,%0A
-    
         
@@ -3457,38 +3457,35 @@
   user,%0A        
-    %7D;
+%7D))
 %0A        %7D)%0A%09%09.c
@@ -3481,13 +3481,8 @@
     
-%7D)%0A%09%09
 .cat
@@ -3683,39 +3683,11 @@
  =%3E 
+(
 %7B%0A
-                return %7B%0A    
     
@@ -3727,28 +3727,24 @@
             
- 
-   
 new: false,%0A
@@ -3743,20 +3743,16 @@
  false,%0A
-    
         
@@ -3780,28 +3780,10 @@
     
-    %7D;%0A            
 %7D
+)
 );%0A 
@@ -3889,16 +3889,17 @@
 'error'%7D
+;
 %0A       

Switch users microservice to new Kafka class