feat: changes related to QBD connector migration (#119)

* feat: changes related to QBD connector migration

* add tests for cov

* remove schedule, change logger level

* fix webhook callback issues

* add json dumps

apps/fyle/helpers.py

@@ -153,3 +153,15 @@ def assert_valid_request(workspace_id:int, org_id:str):
     workspace = Workspace.objects.get(org_id=org_id)
     if workspace.id != workspace_id:
         raise ValidationError('Workspace mismatch')
+
+
+def validate_webhook_request(org_id: str):
+    """
+    Validate the webhook request by checking the fyle_org_id workspace.
+    """
+    if not org_id:
+        raise ValidationError('Org Id not found')
+
+    workspace = Workspace.objects.filter(org_id=org_id).first()
+    if not workspace:
+        raise ValidationError('Workspace not found for the given Org Id')

apps/workspaces/migrations/0029_workspace_migrated_to_qbd_direct.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.1.14 on 2025-01-15 17:13
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('workspaces', '0028_add_created_at_to_user_sql'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workspace',
+            name='migrated_to_qbd_direct',
+            field=models.BooleanField(default=False, help_text='Migrated to QBD Direct'),
+        ),
+    ]

apps/workspaces/models.py

@@ -27,7 +27,7 @@ class Workspace(models.Model):
     user = models.ManyToManyField(User, help_text='Reference to users table')
     org_id = models.CharField(max_length=255, help_text='org id', unique=True)
     currency = models.CharField(max_length=255, help_text='fyle currency', null=True)
-    
+
     reimbursable_last_synced_at = models.DateTimeField(
         help_text='Datetime when reimbursable expenses were pulled last', 
         null=True
@@ -41,6 +41,12 @@ class Workspace(models.Model):
         max_length=50, choices=ONBOARDING_STATE_CHOICES, default=get_default_onboarding_state,
         help_text='Onboarding status of the workspace', null=True
     )
+
+    migrated_to_qbd_direct = models.BooleanField(
+        default=False,
+        help_text='Migrated to QBD Direct'
+    )
+
     created_at = models.DateTimeField(auto_now_add=True, help_text='Created at datetime')
     updated_at = models.DateTimeField(auto_now=True, help_text='Updated at datetime')
 

apps/workspaces/serializers.py

@@ -214,7 +214,9 @@ def create(self, validated_data):
         )
 
         # Schedule run import export or delete
-        schedule_run_import_export(workspace_id)
+
+        if not advanced_setting.workspace.migrated_to_qbd_direct:
+            schedule_run_import_export(workspace_id)
 
         # Update workspace onboarding state
         workspace = advanced_setting.workspace

apps/workspaces/tasks.py

@@ -1,6 +1,10 @@
+import json
 import logging
+from datetime import datetime, timezone
 
 from django.conf import settings
+from django_q.tasks import async_task
+from django_q.models import Schedule
 from fyle_rest_auth.helpers import get_fyle_admin
 
 from apps.fyle.queue import queue_import_credit_card_expenses, queue_import_reimbursable_expenses
@@ -11,8 +15,10 @@
 )
 from apps.tasks.models import AccountingExport
 from apps.fyle.models import Expense
-from apps.workspaces.models import FyleCredential, ExportSettings, Workspace
+from apps.workspaces.models import FyleCredential, ExportSettings, Workspace, AdvancedSetting
 from fyle_integrations_platform_connector import PlatformConnector
+from apps.fyle.helpers import post_request, validate_webhook_request
+
 
 logger = logging.getLogger(__name__)
 logger.level = logging.INFO
@@ -21,9 +27,13 @@
 def run_import_export(workspace_id: int):
     """
     Run Processes to Generate IIF File
-    
     :param workspace_id: Workspace id
     """
+    workspace = Workspace.objects.get(id=workspace_id)
+    if workspace.migrated_to_qbd_direct:
+        logger.error("Import Export not running since the workspace with id {} is migrated to QBD Connector".format(workspace.id))
+        return
+
     export_settings = ExportSettings.objects.get(workspace_id=workspace_id)
 
     # For Reimbursable Expenses
@@ -72,6 +82,8 @@ def run_import_export(workspace_id: int):
                 elif export_settings.credit_card_expense_export_type == 'JOURNAL_ENTRY':
                     queue_create_journals_iif_file('CCC', workspace_id)
 
+    async_task('apps.workspaces.tasks.async_update_timestamp_in_qbd_direct', workspace_id=workspace_id)
+
 
 def async_update_workspace_name(workspace: Workspace, access_token: str):
     """
@@ -100,3 +112,58 @@ def async_create_admin_subcriptions(workspace_id: int) -> None:
         'webhook_url': '{}/workspaces/{}/fyle/webhook_callback/'.format(settings.API_URL, workspace_id)
     }
     platform.subscriptions.post(payload)
+
+
+def async_handle_webhook_callback(payload: dict) -> None:
+    """
+    Handle webhook callback
+    :param data: data
+    :return: None
+    """
+    logger.info("Received Webhook Callback with payload: %s", payload)
+
+    org_id = payload.get('data', {}).get('org_id')
+    action = payload.get('action')
+    validate_webhook_request(org_id=org_id)
+
+    if action == 'DISABLE_EXPORT':
+        workspace = Workspace.objects.filter(org_id=org_id).first()
+        workspace.migrated_to_qbd_direct = True
+        workspace.updated = datetime.now(timezone.utc)
+        workspace.save(update_fields=['migrated_to_qbd_direct', 'updated_at'])
+
+        Schedule.objects.filter(args=str(workspace.id)).all().delete()
+        adv_settings = AdvancedSetting.objects.filter(workspace_id=workspace.id, schedule_id__isnull=False).first()
+        if adv_settings:
+            adv_settings.schedule_id = None
+            adv_settings.save(update_fields=['schedule_id'])
+
+
+def async_update_timestamp_in_qbd_direct(workspace_id: int) -> None:
+    """
+    Update timestamp in QBD Direct App
+    """
+    workspace = Workspace.objects.get(id=workspace_id)
+
+    payload = {
+        'data': {
+            'org_id': workspace.org_id,
+            'reimbursable_last_synced_at': workspace.reimbursable_last_synced_at.isoformat() if workspace.reimbursable_last_synced_at else None,
+            'ccc_last_synced_at': workspace.ccc_last_synced_at.isoformat() if workspace.ccc_last_synced_at else None
+        },
+        'action': 'UPDATE_LAST_SYNCED_TIMESTAMP'
+    }
+
+    api_url = '{}/workspaces/webhook_callback/'.format(settings.QBD_DIRECT_API_URL)
+
+    try:
+        logger.info('Posting Timestamp Update to QBD Connector with payload: {}'.format(payload))
+        fyle_creds = FyleCredential.objects.filter(workspace_id=workspace.id).first()
+
+        if fyle_creds:
+            refresh_token = fyle_creds.refresh_token
+            post_request(url=api_url, body=json.dumps(payload), refresh_token=refresh_token)
+        else:
+            raise Exception('Auth Token not present for workspace id {}'.format(workspace.id))
+    except Exception as e:
+        logger.error("Failed to sync timestamp to QBD Connector: {}".format(e))

apps/workspaces/urls.py

@@ -20,13 +20,15 @@
     AdvancedSettingView,
     FieldMappingView,
     TriggerExportView,
-    ReadyView
+    ReadyView,
+    WebhookCallbackView
 )
 
 
 urlpatterns = [
     path('', WorkspaceView.as_view(), name='workspaces'),
     path('ready/', ReadyView.as_view(), name='ready'),
+    path('webhook_callback/', WebhookCallbackView.as_view(), name='workspace_webhook_callback'),
     path('<int:workspace_id>/export_settings/', ExportSettingView.as_view(), name='export-settings'),
     path('<int:workspace_id>/advanced_settings/', AdvancedSettingView.as_view(), name='advanced-settings'),
     path('<int:workspace_id>/field_mappings/', FieldMappingView.as_view(), name='field-mappings'),

apps/workspaces/views.py

@@ -1,3 +1,4 @@
+from apps.fyle.exceptions import handle_view_exceptions
 from rest_framework import generics
 from rest_framework.views import Response, status
 from rest_framework.permissions import IsAuthenticated
@@ -111,6 +112,7 @@ def post(self, request, *args, **kwargs):
             }
         )
 
+
 class ReadyView(generics.RetrieveAPIView):
     """
     Ready call to check if the api is ready
@@ -130,3 +132,27 @@ def get(self, request, *args, **kwargs):
             },
             status=status.HTTP_200_OK
         )
+
+
+class WebhookCallbackView(generics.CreateAPIView):
+    """
+    Webhook View
+    """
+    permission_classes = []
+
+    @handle_view_exceptions()
+    def post(self, request, *args, **kwargs):
+        """
+        Webhook post call
+        """
+        async_task(
+            'apps.workspaces.tasks.async_handle_webhook_callback',
+            payload=request.data
+        )
+
+        return Response(
+            status=status.HTTP_200_OK,
+            data={
+                'message': 'Webhook callback received'
+            }
+        )

docker-compose-pipeline.yml

@@ -22,6 +22,7 @@ services:
       FYLE_TOKEN_URI: https://localhost:1234/oauth/token
       FYLE_BASE_URL: https://localhost:1234
       API_URL: http://localhost:8000/api
+      QBD_DIRECT_API_URL: http://localhost:8011/api
       FYLE_CLIENT_ID: client_id
       FYLE_CLIENT_SECRET: client_secret
       DB_NAME: qbd_db

quickbooks_desktop_api/settings.py

@@ -270,6 +270,8 @@
 FYLE_CLIENT_ID = os.environ.get('FYLE_CLIENT_ID')
 FYLE_CLIENT_SECRET = os.environ.get('FYLE_CLIENT_SECRET')
 
+QBD_DIRECT_API_URL = os.environ.get('QBD_DIRECT_API_URL')
+
 SENDGRID_API_KEY = os.environ.get('SENDGRID_API_KEY')
 SENDGRID_FROM_EMAIL = os.environ.get('SENDGRID_EMAIL_FROM')
 CORS_ORIGIN_ALLOW_ALL = True

quickbooks_desktop_api/tests/settings.py

@@ -248,5 +248,7 @@
 FYLE_CLIENT_ID = os.environ.get('FYLE_CLIENT_ID')
 FYLE_CLIENT_SECRET = os.environ.get('FYLE_CLIENT_SECRET')
 
+QBD_DIRECT_API_URL = os.environ.get('QBD_DIRECT_API_URL')
+
 SENDGRID_API_KEY = os.environ.get('SENDGRID_API_KEY')
 SENDGRID_FROM_EMAIL = os.environ.get('SENDGRID_EMAIL_FROM')

tests/test_workspaces/test_tasks.py

@@ -1,7 +1,12 @@
 import pytest
+
+from rest_framework.exceptions import ValidationError
+
 from apps.fyle.models import Expense
 from apps.workspaces.models import Workspace
 from apps.workspaces.tasks import (
+    async_handle_webhook_callback,
+    async_update_timestamp_in_qbd_direct,
     run_import_export,
     async_update_workspace_name,
     async_create_admin_subcriptions
@@ -35,7 +40,7 @@ def test_run_import_export_bill_ccp(
 
     tasks = OrmQ.objects.all()
 
-    assert tasks.count() == 2
+    assert tasks.count() == 3
 
 
 @pytest.mark.django_db(databases=['default'], transaction=True)
@@ -61,7 +66,7 @@ def test_run_import_export_journal_journal(
 
     tasks = OrmQ.objects.all()
 
-    assert tasks.count() == 2
+    assert tasks.count() == 3
 
 
 @pytest.mark.django_db(databases=['default'])
@@ -126,3 +131,101 @@ def test_async_create_admin_subcriptions_2(
 
     mock_api.side_effect = Exception('Error')
     reverse('webhook-callback', kwargs={'workspace_id': workspace_id})
+
+
+def test_handle_webhook_callback_case_1(db, create_temp_workspace):
+    """
+    Test handle webhook callback
+    Case: Valid Payload
+    """
+    workspace = Workspace.objects.first()
+
+    assert workspace.migrated_to_qbd_direct is False
+
+    payload = {
+        'data': {
+            'org_id': workspace.org_id
+        },
+        'action': 'DISABLE_EXPORT'
+    }
+
+    async_handle_webhook_callback(payload=payload)
+
+    workspace.refresh_from_db()
+
+    assert workspace.migrated_to_qbd_direct is True
+
+
+def test_handle_webhook_callback_case_2(db, create_temp_workspace):
+    """
+    Test handle webhook callback
+    Case: Invalid Payload, org_id not present
+    """
+    workspace = Workspace.objects.first()
+
+    assert workspace.migrated_to_qbd_direct is False
+
+    payload = {
+        'data': {
+            'org_id': None
+        },
+        'action': 'DISABLE_EXPORT'
+    }
+
+    try:
+        async_handle_webhook_callback(payload=payload)
+    except ValidationError as e:
+        assert str(e.detail[0]) == 'Org Id not found'
+
+
+def test_handle_webhook_callback_case_3(db, create_temp_workspace):
+    """
+    Test handle webhook callback
+    Case: Invalid Payload, org_id does not match in workspace
+    """
+    workspace = Workspace.objects.first()
+
+    assert workspace.migrated_to_qbd_direct is False
+
+    payload = {
+        'data': {
+            'org_id': 'org123'
+        },
+        'action': 'DISABLE_EXPORT'
+    }
+
+    try:
+        async_handle_webhook_callback(payload=payload)
+    except ValidationError as e:
+        assert str(e.detail[0]) == 'Workspace not found for the given Org Id'
+
+
+def test_async_update_timestamp_in_qbd_direct_case_1(
+    db,
+    mocker,
+    create_temp_workspace,
+    add_fyle_credentials
+):
+    """
+    Test update_timestamp_in_qbd_direct
+    Case: Valid Payload and token
+    """
+    workspace = Workspace.objects.first()
+    post_request = mocker.patch('apps.workspaces.tasks.post_request')
+
+    async_update_timestamp_in_qbd_direct(workspace.id)
+
+    post_request.assert_called_once()
+
+
+def test_async_update_timestamp_in_qbd_direct_case_2(db, create_temp_workspace):
+    """
+    Test update_timestamp_in_qbd_direct
+    Case: Token not present
+    """
+    workspace = Workspace.objects.first()
+
+    try:
+        async_update_timestamp_in_qbd_direct(workspace.id)
+    except Exception as e:
+        assert str(e.detail[0]) == 'Auth Token not present for workspace id {}'.format(workspace.id)