feat: activate license

app.py

@@ -8,7 +8,10 @@
 from quart_cors import cors
 from werkzeug.exceptions import HTTPException
 
-from lemon import check_signing_secret, parse_event, dispatch_event
+from lemon import check_signing_secret, \
+    parse_event, \
+    dispatch_event, \
+    activate_license as activate_license_internal
 from logger import logger
 from mongo.db import convert_id_to_str_in_json, \
     convert_fields_to_datetime_in_json
@@ -22,7 +25,10 @@
     setup_subscription_payments, \
     find_latest_subscription, \
     convert_subscription_to_response
-from mongo.users import User, setup_users, upsert_user
+from mongo.users import User, \
+    setup_users, \
+    find_user_by_token, \
+    upsert_user
 from oauth import generate_user_token, \
     decrypt_user_token, \
     upsert_user_from_google_oauth
@@ -176,23 +182,17 @@ async def check_latest_subscription():
 
 
 # ?user_token=str   required.
-# &store_id=str     required.
-# &product_id=str   required.
 # &license_key=str  required.
 # &test_mode=bool   optional; default is `false`.
 @app.get('/api/licenses/latest')
 async def check_latest_license():
     user_token = _parse_str_from_dict(request.args, 'user_token')
-    store_id = _parse_str_from_dict(request.args, 'store_id')
-    product_id = _parse_str_from_dict(request.args, 'product_id')
     license_key = _parse_str_from_dict(request.args, 'license_key')
     test_mode = request.args.get('test_mode', False, bool)
 
     res = await find_latest_license(
         user_id=decrypt_user_token(user_token).user_id,
-        store_id=store_id,
-        product_id=product_id,
-        key=license_key,
+        license_key=license_key,
         test_mode=test_mode,
     )
 
@@ -206,6 +206,7 @@ async def check_latest_license():
 #   'user_token':    required; str.
 #   'license_key':   required; str.
 #   'instance_name': optional; str.
+#   'test_mode':     optional; bool, default is `false`.
 # }
 @app.post('/api/licenses/activate')
 async def activate_license():
@@ -215,10 +216,27 @@ async def activate_license():
 
     user_token = _parse_str_from_dict(body, 'user_token')
     license_key = _parse_str_from_dict(body, 'license_key')
-    instance_name = _parse_str_from_dict(body, 'instance_name', required=False)
+    test_mode = bool(body.get('test_mode', False))
 
-    # TODO
-    return {}
+    instance_name = _parse_str_from_dict(
+        data=body,
+        key='instance_name',
+        default=str(int(time.time())),  # activate timestamp in seconds.
+        required=False,
+    )
+
+    # Check the user ownership.
+    res = await find_latest_license(
+        user_id=decrypt_user_token(user_token).user_id,
+        license_key=license_key,
+        test_mode=test_mode,
+    )
+
+    if not res:
+        abort(404, 'license not found')
+
+    res = await activate_license_internal(license_key, instance_name)
+    return await convert_license_to_response(res)
 
 
 def _parse_str_from_dict(

lemon.py

@@ -1,16 +1,22 @@
 import hashlib
 import hmac
+import json
+
+import httpx
 
 from enum import unique
 
 from quart import abort
 from strenum import StrEnum
 from werkzeug.datastructures import Headers
 
+from logger import logger
 from mongo.licenses import insert_license
 from mongo.orders import insert_order
 from mongo.subscriptions import insert_subscription, insert_subscription_payment
-from rds import get_str_from_rds, LEMONSQUEEZY_SIGNING_SECRET
+from rds import get_str_from_rds, \
+    LEMONSQUEEZY_SIGNING_SECRET, \
+    LEMONSQUEEZY_API_KEY
 
 
 # https://docs.lemonsqueezy.com/help/webhooks#event-types
@@ -71,3 +77,72 @@ async def dispatch_event(event: Event, body: dict):
         await insert_license(body)
     else:
         abort(400, f'unsupported event, event={str(event)}')
+
+
+# https://docs.lemonsqueezy.com/help/licensing/license-api#post-v1-licenses-activate
+#
+# FIXME (Matthew Lee)
+# API calls are rate limited to 60 / minute,
+# but the rate limited http code is undocumented,
+# don't know whether it is 429 or not for now until we reach and log it.
+async def activate_license(
+    license_key: str,
+    instance_name: str,
+    api_key: str = '',
+) -> dict:
+    if not api_key:
+        api_key = get_str_from_rds(LEMONSQUEEZY_API_KEY)
+
+    headers = {
+        'Accept': 'application/json',
+        'Authorization': f'Bearer {api_key}',
+    }
+
+    data = {
+        'license_key': license_key,
+        'instance_name': instance_name,
+    }
+
+    transport = httpx.AsyncHTTPTransport(retries=2)
+    client = httpx.AsyncClient(transport=transport)
+
+    try:
+        # Content-Type must be 'application/x-www-form-urlencoded'.
+        response = await client.post(
+            url='https://api.lemonsqueezy.com/v1/licenses/activate',
+            headers=headers,
+            data=data,
+            timeout=10,
+            follow_redirects=True,
+        )
+    finally:
+        await client.aclose()
+
+    if not response.is_success:
+        abort(response.status_code, response.text)
+
+    # Automatically .aclose() if the response body is read to completion.
+    data: dict = response.json()
+    logger.info(
+        f'activate license, '
+        f'license_key={license_key}, '
+        f'instance_name={instance_name}, '
+        f'body={json.dumps(data)}'
+    )
+
+    # The `data` structure is not similar to webhooks request,
+    # so we retrieve license again for later code reusing.
+    return await retrieve_license(license_key, api_key)
+
+
+async def retrieve_license(license_key: str, api_key: str = '') -> dict:
+    if not api_key:
+        api_key = get_str_from_rds(LEMONSQUEEZY_API_KEY)
+
+    headers = {
+        'Accept': 'application/json',
+        'Authorization': f'Bearer {api_key}',
+    }
+
+    # TODO (Matthew Lee) ...
+    pass

mongo/licenses.py

@@ -46,19 +46,20 @@ async def insert_license(license: dict):
     find_latest_license.cache_clear()
 
 
+# https://docs.lemonsqueezy.com/api/license-keys#retrieve-a-license-key
+#
+# Based on the upper API specs,
+# the `license_key` is unique in all stores,
+# so we just need to check the user ownership.
 @alru_cache()
 async def find_latest_license(
     user_id: str,
-    store_id: str,
-    product_id: str,
     license_key: str,
     test_mode: bool = False,
 ) -> Optional[dict]:
     cursor = licenses \
         .find({
             'meta.custom_data.user_id': user_id,
-            'data.attributes.store_id': store_id,
-            'data.attributes.product_id': product_id,
             'data.attributes.key': license_key,
             'data.attributes.test_mode': test_mode,
         }) \

oauth.py

@@ -57,7 +57,7 @@ def generate_user_token(user_id: str, timestamp: int, secret: str = '') -> str:
 
 
 # https://onboardbase.com/blog/aes-encryption-decryption/
-def decrypt_user_token(token: str, secret: str = '') -> Optional[TokenInfo]:
+def decrypt_user_token(token: str, secret: str = '') -> TokenInfo:
     secret = secret.strip()
     if not secret:
         secret = get_str_from_rds(LEMONSQUEEZY_SIGNING_SECRET)

tests/test_oauth.py

@@ -30,6 +30,5 @@ def test_decrypt_user_token():
         secret=_LEMONSQUEEZY_SIGNING_SECRET,
     )
 
-    assert info
     assert info.user_id == user_id
     assert info.generate_timestamp == timestamp