From f04b0188051aa9e4a29d06607f79909545affbde Mon Sep 17 00:00:00 2001
From: Jake Moshenko <jake.moshenko@coreos.com>
Date: Thu, 13 Oct 2016 13:48:35 -0400
Subject: [PATCH] Write our users to Marketo as leads.

---
 app.py                             |   2 +
 config.py                          |   2 +-
 endpoints/api/user.py              |  10 ++-
 endpoints/common.py                |   8 +-
 endpoints/web.py                   |   3 +-
 requirements-nover.txt             |   1 +
 requirements.txt                   |   4 +-
 static/js/services/user-service.js |  20 +++++
 templates/base.html                |   6 ++
 util/asyncwrapper.py               |  63 +++++++++++++
 util/saas/useranalytics.py         | 137 +++++++++++++++++++++++++++++
 11 files changed, 250 insertions(+), 6 deletions(-)
 create mode 100644 util/asyncwrapper.py
 create mode 100644 util/saas/useranalytics.py

diff --git a/app.py b/app.py
index 2ad2fe618..214efb3b6 100644
--- a/app.py
+++ b/app.py
@@ -26,6 +26,7 @@ from data.userevent import UserEventsBuilderModule
 from data.queue import WorkQueue, BuildMetricQueueReporter
 from util import get_app_url
 from util.saas.analytics import Analytics
+from util.saas.useranalytics import UserAnalytics
 from util.saas.exceptionlog import Sentry
 from util.names import urn_generator
 from util.config.oauth import (GoogleOAuthConfig, GithubOAuthConfig, GitLabOAuthConfig,
@@ -177,6 +178,7 @@ storage = Storage(app, metric_queue, instance_keys)
 userfiles = Userfiles(app, storage)
 log_archive = LogArchive(app, storage)
 analytics = Analytics(app)
+user_analytics = UserAnalytics(app)
 billing = Billing(app)
 sentry = Sentry(app)
 build_logs = BuildLogs(app)
diff --git a/config.py b/config.py
index f877afb7e..027f36254 100644
--- a/config.py
+++ b/config.py
@@ -19,7 +19,7 @@ CLIENT_WHITELIST = ['SERVER_HOSTNAME', 'PREFERRED_URL_SCHEME', 'MIXPANEL_KEY',
                     'STRIPE_PUBLISHABLE_KEY', 'ENTERPRISE_LOGO_URL', 'SENTRY_PUBLIC_DSN',
                     'AUTHENTICATION_TYPE', 'REGISTRY_TITLE', 'REGISTRY_TITLE_SHORT',
                     'CONTACT_INFO', 'AVATAR_KIND', 'LOCAL_OAUTH_HANDLER', 'DOCUMENTATION_LOCATION',
-                    'DOCUMENTATION_METADATA', 'SETUP_COMPLETE', 'DEBUG', 'MUNCHKIN_KEY']
+                    'DOCUMENTATION_METADATA', 'SETUP_COMPLETE', 'DEBUG', 'MARKETO_MUNCHKIN_ID']
 
 
 def frontend_visible_config(config_dict):
diff --git a/endpoints/api/user.py b/endpoints/api/user.py
index 7547d014a..c6c75c0be 100644
--- a/endpoints/api/user.py
+++ b/endpoints/api/user.py
@@ -10,7 +10,7 @@ from peewee import IntegrityError
 
 import features
 
-from app import app, billing as stripe, authentication, avatar
+from app import app, billing as stripe, authentication, avatar, user_analytics
 from auth import scopes
 from auth.auth_context import get_authenticated_user
 from auth.permissions import (AdministerOrganizationPermission, CreateRepositoryPermission,
@@ -119,6 +119,14 @@ def user_view(user):
       'tag_expiration': user.removed_tag_expiration_s,
     })
 
+    analytics_metadata = user_analytics.get_user_analytics_metadata(user)
+
+    # This is a sync call, but goes through the async wrapper interface and
+    # returns a Future. By calling with timeout 0 immediately after the method
+    # call, we ensure that if it ever accidentally becomes async it will raise
+    # a TimeoutError.
+    user_response.update(analytics_metadata.result(timeout=0))
+
   user_view_perm = UserReadPermission(user.username)
   if user_view_perm.can():
     user_response.update({
diff --git a/endpoints/common.py b/endpoints/common.py
index 791f987f8..d00c4bc64 100644
--- a/endpoints/common.py
+++ b/endpoints/common.py
@@ -16,7 +16,7 @@ from flask_principal import identity_changed
 import endpoints.decorated # Register the various exceptions via decorators.
 import features
 
-from app import app, oauth_apps, LoginWrappedDBUser
+from app import app, oauth_apps, LoginWrappedDBUser, user_analytics
 from auth import scopes
 from auth.permissions import QuayDeferredPermissionUser
 from config import frontend_visible_config
@@ -114,6 +114,10 @@ def common_login(db_user):
     new_identity = QuayDeferredPermissionUser.for_user(db_user)
     identity_changed.send(app, identity=new_identity)
     session['login_time'] = datetime.datetime.now()
+
+    # Inform our user analytics that we have a new "lead"
+    user_analytics.create_lead(db_user.email, db_user.username)
+
     return True
   else:
     logger.debug('User could not be logged in, inactive?.')
@@ -209,7 +213,7 @@ def render_page_template(name, route_data=None, **kwargs):
                                        vuln_priority_set=json.dumps(PRIORITY_LEVELS),
                                        enterprise_logo=app.config.get('ENTERPRISE_LOGO_URL', ''),
                                        mixpanel_key=app.config.get('MIXPANEL_KEY', ''),
-                                       munchkin_key=app.config.get('MUNCHKIN_KEY', ''),
+                                       munchkin_key=app.config.get('MARKETO_MUNCHKIN_ID', ''),
                                        google_tagmanager_key=app.config.get('GOOGLE_TAGMANAGER_KEY', ''),
                                        google_anaytics_key=app.config.get('GOOGLE_ANALYTICS_KEY', ''),
                                        sentry_public_dsn=app.config.get('SENTRY_PUBLIC_DSN', ''),
diff --git a/endpoints/web.py b/endpoints/web.py
index 4b4d9f4e1..b421f9f86 100644
--- a/endpoints/web.py
+++ b/endpoints/web.py
@@ -11,7 +11,7 @@ from flask_login import current_user
 import features
 
 from app import (app, billing as stripe, build_logs, avatar, signer, log_archive, config_provider,
-                 get_app_url, instance_keys)
+                 get_app_url, instance_keys, user_analytics)
 from auth import scopes
 from auth.auth_context import get_authenticated_user
 from auth.permissions import (AdministerOrganizationPermission, ReadRepositoryPermission,
@@ -390,6 +390,7 @@ def confirm_email():
 
   if new_email:
     send_email_changed(user.username, old_email, new_email)
+    user_analytics.change_email(old_email, new_email)
 
   common_login(user)
 
diff --git a/requirements-nover.txt b/requirements-nover.txt
index f1c093b26..121492bc1 100644
--- a/requirements-nover.txt
+++ b/requirements-nover.txt
@@ -7,6 +7,7 @@
 -e git+https://github.com/coreos/py-bitbucket.git#egg=py-bitbucket
 -e git+https://github.com/coreos/pyapi-gitlab.git@timeout#egg=pyapi-gitlab
 -e git+https://github.com/coreos/resumablehashlib.git#egg=resumablehashlib
+-e git+https://github.com/jepcastelein/marketo-rest-python.git#egg=marketorestpython
 APScheduler==3.0.5
 Flask-Login
 Flask-Mail
diff --git a/requirements.txt b/requirements.txt
index bbbeef29a..b2083ecf1 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -47,6 +47,7 @@ jsonschema==2.5.1
 keystoneauth1==2.4.0
 Mako==1.0.4
 marisa-trie==0.7.2
+-e git+https://github.com/jepcastelein/marketo-rest-python.git@1ba6dfee030b192f0930dd8c3b6d53b52d886c65#egg=marketorestpython-master
 MarkupSafe==0.23
 mixpanel==4.3.0
 mock==2.0.0
@@ -98,7 +99,7 @@ raven==5.12.0
 redis==2.10.5
 redlock==1.2.0
 reportlab==2.7
-requests==2.9.1
+requests==2.11.1
 requests-oauthlib==0.6.1
 -e git+https://github.com/coreos/resumablehashlib.git@b1b631249589b07adf40e0ee545b323a501340b4#egg=resumablehashlib
 semantic-version==2.5.0
@@ -114,6 +115,7 @@ urllib3==1.14
 waitress==0.8.10
 WebOb==1.6.0
 Werkzeug==0.11.5
+wheel==0.24.0
 wrapt==1.10.7
 xhtml2pdf==0.0.6
 xmltodict==0.10.1
diff --git a/static/js/services/user-service.js b/static/js/services/user-service.js
index b205d331f..9f101374f 100644
--- a/static/js/services/user-service.js
+++ b/static/js/services/user-service.js
@@ -53,6 +53,26 @@ function(ApiService, CookieService, $rootScope, Config) {
           }
         }
 
+        if (Config.MARKETO_MUNCHKIN_ID && userResponse['marketo_user_hash']) {
+          associateLeadBody = {'Email': userResponse.email};
+          if (window.Munchkin !== undefined) {
+            try {
+              Munchkin.munchkinFunction(
+                'associateLead',
+                associateLeadBody,
+                userResponse['marketo_user_hash']
+              );
+            } catch (e) {
+            }
+          } else {
+            window.__quay_munchkin_queue.push([
+              'associateLead',
+              associateLeadBody,
+              userResponse['marketo_user_hash']
+            ]);
+          }
+        }
+
         if (window.Raven !== undefined) {
           try {
             Raven.setUser({
diff --git a/templates/base.html b/templates/base.html
index f860cef37..d5f03c88b 100644
--- a/templates/base.html
+++ b/templates/base.html
@@ -75,11 +75,17 @@
     {% if munchkin_key %}
       <script type="text/javascript">
       (function() {
+        window.__quay_munchkin_queue = []
+
         var didInit = false;
         function initMunchkin() {
           if(didInit === false) {
             didInit = true;
             Munchkin.init('{{ munchkin_key }}');
+            window.__quay_munchkin_queue.forEach(function(queue_item) {
+              Munchkin.munchkinFunction.apply(Munchkin, queue_item);
+            });
+            window.__quay_munchkin_queue = [];
           }
         }
         var s = document.createElement('script');
diff --git a/util/asyncwrapper.py b/util/asyncwrapper.py
new file mode 100644
index 000000000..4b11c689a
--- /dev/null
+++ b/util/asyncwrapper.py
@@ -0,0 +1,63 @@
+import queue
+
+from functools import wraps
+
+from concurrent.futures import Executor, Future, CancelledError
+
+
+class AsyncExecutorWrapper(object):
+  """ This class will wrap a syncronous library transparently in a way which
+      will move all calls off to an asynchronous Executor, and will change all
+      returned values to be Future objects.
+  """
+  SYNC_FLAG_FIELD = '__AsyncExecutorWrapper__sync__'
+
+  def __init__(self, delegate, executor):
+    """ Wrap the specified synchronous delegate instance, and submit() all
+        method calls to the specified Executor instance.
+    """
+    self._delegate = delegate
+    self._executor = executor
+
+  def __getattr__(self, attr_name):
+    maybe_callable = getattr(self._delegate, attr_name) # Will raise proper attribute error
+    if callable(maybe_callable):
+      # Build a callable which when executed places the request
+      # onto a queue
+      @wraps(maybe_callable)
+      def wrapped_method(*args, **kwargs):
+        if getattr(maybe_callable, self.SYNC_FLAG_FIELD, False):
+          sync_result = Future()
+          try:
+            sync_result.set_result(maybe_callable(*args, **kwargs))
+          except Exception as ex:
+            sync_result.set_exception(ex)
+          return sync_result
+
+        try:
+          return self._executor.submit(maybe_callable, *args, **kwargs)
+        except queue.Full as ex:
+          queue_full = Future()
+          queue_full.set_exception(ex)
+          return queue_full
+
+      return wrapped_method
+    else:
+      return maybe_callable
+
+  @classmethod
+  def sync(cls, f):
+    """ Annotate the given method to flag it as synchronous so that AsyncExecutorWrapper
+        will return the result immediately without submitting it to the executor.
+    """
+    setattr(f, cls.SYNC_FLAG_FIELD, True)
+    return f
+
+
+class NullExecutor(Executor):
+  """ Executor instance which always returns a Future completed with a
+      CancelledError exception. """
+  def submit(self, _, *args, **kwargs):
+    always_fail = Future()
+    always_fail.set_exception(CancelledError('Null executor always fails.'))
+    return always_fail
diff --git a/util/saas/useranalytics.py b/util/saas/useranalytics.py
new file mode 100644
index 000000000..2feeae647
--- /dev/null
+++ b/util/saas/useranalytics.py
@@ -0,0 +1,137 @@
+import logging
+
+from hashlib import sha1
+
+from concurrent.futures import ThreadPoolExecutor
+from marketorestpython.client import MarketoClient
+
+from util.asyncwrapper import AsyncExecutorWrapper, NullExecutor
+
+logger = logging.getLogger(__name__)
+
+
+class LeadNotFoundException(Exception):
+  pass
+
+
+class _MarketoAnalyticsClient(object):
+  """ User analytics implementation which will report user changes to the
+      Marketo API.
+  """
+  def __init__(self, marketo_client, munchkin_private_key, lead_source):
+    """ Instantiate with the given marketorestpython.client, the Marketo
+        Munchkin Private Key, and the Lead Source that we want to set when we
+        create new lead records in Marketo.
+    """
+    self._marketo = marketo_client
+    self._munchkin_private_key = munchkin_private_key
+    self._lead_source = lead_source
+
+  def create_lead(self, email, username):
+    lead_data = dict(
+      email=email,
+      Quay_Username__c=username,
+      leadSource='Web - Product Trial',
+      Lead_Source_Detail__c=self._lead_source,
+    )
+    self._marketo.create_update_leads(
+      action='createOrUpdate',
+      leads=[lead_data],
+      asyncProcessing=True,
+      lookupField='email',
+    )
+
+  def _find_leads_by_email(self, email):
+    # Fetch the existing user from the database by email
+    found = self._marketo.get_multiple_leads_by_filter_type(
+      filterType='email',
+      filterValues=[email],
+    )
+
+    if not found:
+      raise LeadNotFoundException('No lead found with email: {}'.format(email))
+
+    return found
+
+  def change_email(self, old_email, new_email):
+    found = self._find_leads_by_email(old_email)
+
+    # Update using their user id.
+    updated = [dict(id=lead['id'], email=new_email) for lead in found]
+    self._marketo.create_update_leads(
+      action='updateOnly',
+      leads=updated,
+      asyncProcessing=True,
+      lookupField='id',
+    )
+
+  def change_username(self, email, new_username):
+    found = self._find_leads_by_email(email)
+
+    # Update using their user id.
+    updated = [dict(id=lead['id'], Quay_Username__c=new_username) for lead in found]
+    self._marketo.create_update_leads(
+      action='updateOnly',
+      leads=updated,
+      asyncProcessing=True,
+      lookupField='id',
+    )
+
+  @AsyncExecutorWrapper.sync
+  def get_user_analytics_metadata(self, user_obj):
+    """ Return a list of properties that should be added to the user object to allow
+        analytics associations.
+    """
+    if not self._munchkin_private_key:
+      return dict()
+
+    marketo_user_hash = sha1(self._munchkin_private_key)
+    marketo_user_hash.update(user_obj.email)
+
+    return dict(
+      marketo_user_hash=marketo_user_hash.hexdigest(),
+    )
+
+
+class UserAnalytics(object):
+  def __init__(self, app=None):
+    self.app = app
+    if app is not None:
+      self.state = self.init_app(app)
+    else:
+      self.state = None
+
+  def init_app(self, app):
+    analytics_type = app.config.get('USER_ANALYTICS_TYPE', 'FakeAnalytics')
+
+    marketo_munchkin_id = ''
+    marketo_munchkin_private_key = ''
+    marketo_client_id = ''
+    marketo_client_secret = ''
+    marketo_lead_source = ''
+    executor = NullExecutor()
+
+    if analytics_type == 'Marketo':
+      marketo_munchkin_id = app.config['MARKETO_MUNCHKIN_ID']
+      marketo_munchkin_private_key = app.config['MARKETO_MUNCHKIN_PRIVATE_KEY']
+      marketo_client_id = app.config['MARKETO_CLIENT_ID']
+      marketo_client_secret = app.config['MARKETO_CLIENT_SECRET']
+      marketo_lead_source = app.config['MARKETO_LEAD_SOURCE']
+
+      logger.debug('Initializing marketo with keys: %s %s %s', marketo_munchkin_id,
+                   marketo_client_id, marketo_client_secret)
+
+      executor = ThreadPoolExecutor(max_workers=1)
+
+    marketo_client = MarketoClient(marketo_munchkin_id, marketo_client_id, marketo_client_secret)
+    client_wrapper = _MarketoAnalyticsClient(marketo_client, marketo_munchkin_private_key,
+                                             marketo_lead_source)
+    user_analytics = AsyncExecutorWrapper(client_wrapper, executor)
+
+    # register extension with app
+    app.extensions = getattr(app, 'extensions', {})
+    app.extensions['user_analytics'] = user_analytics
+    return user_analytics
+
+  def __getattr__(self, name):
+    return getattr(self.state, name, None)