partial of the new queue service

pay-api/requirements.txt

@@ -35,6 +35,10 @@ expiringdict==1.2.2
 flask-jwt-oidc==0.3.0
 flask-marshmallow==0.11.0
 flask-restx==1.1.0
+google-api-core==2.11.0
+google-auth==2.18.1
+google-cloud-pubsub==2.17.0
+googleapis-common-protos==1.59.0
 gunicorn==20.1.0
 hijri-converter==2.2.4
 holidays==0.23
@@ -73,3 +77,4 @@ typing_extensions==4.5.0
 urllib3==1.26.15
 zipp==3.15.0
 -e git+https://github.com/bcgov/sbc-common-components.git#egg=sbc-common-components&subdirectory=python
+git+https://github.com/daxiom/simple-cloudevent.py.git

pay-api/requirements/prod.txt

@@ -30,3 +30,5 @@ Jinja2==3.0.3
 protobuf~=3.19.5
 launchdarkly-server-sdk
 holidays==0.23
+google-auth==2.18.1
+google-cloud-pubsub==2.17.0

pay-api/requirements/repo-libraries.txt

@@ -1 +1,2 @@
 -e git+https://github.com/bcgov/sbc-common-components.git#egg=sbc-common-components&subdirectory=python
+git+https://github.com/daxiom/simple-cloudevent.py.git@0.0.2

pay-api/setup.cfg

@@ -58,6 +58,7 @@ good-names=
     e,
     f,
     u,
+    ce,
     rv,
     logger,
     id,
@@ -78,6 +79,7 @@ good-names=
     e,
     f,
     u,
+    ce,
     rv,
     logger,
     id,

pay-api/src/pay_api/config.py

@@ -130,6 +130,12 @@ class _Config():  # pylint: disable=too-few-public-methods
     NATS_ACCOUNT_CLIENT_NAME = os.getenv('NATS_ACCOUNT_CLIENT_NAME', 'account.events.worker')
     NATS_ACCOUNT_SUBJECT = os.getenv('NATS_ACCOUNT_SUBJECT', 'account.events')
 
+    # GCP PubSub
+    AUDIENCE = os.getenv('AUDIENCE', None)
+    GCP_AUTH_KEY = os.getenv('GCP_AUTH_KEY', None)
+    PUBLISHER_AUDIENCE = os.getenv('PUBLISHER_AUDIENCE', None)
+    TOPIC_NAME = os.getenv('TOPIC_NAME', None)
+
     # Auth API Endpoint
     AUTH_API_ENDPOINT = f'{_get_config("AUTH_API_URL")}/'
 

pay-api/src/pay_api/services/gcp_queue_publisher.py

@@ -0,0 +1,54 @@
+"""This module provides Queue type services."""
+import base64
+import json
+import uuid
+from concurrent.futures import CancelledError
+from concurrent.futures import TimeoutError  # pylint: disable=W0622
+
+from flask import current_app
+from google.auth import jwt
+from google.cloud import pubsub_v1
+from simple_cloudevent import SimpleCloudEvent, to_queue_message
+
+from .invoice import Invoice
+
+
+def publish_to_queue(payload: dict, invoice: Invoice):
+    """Publish a 'COMPLETED' invoice's info to the GCP PubSub Queue."""
+    ce = SimpleCloudEvent()
+    ce.id = payload.get('paymentToken', {}).get('id', str(uuid.uuid4()))
+    ce.source = 'sbc-pay'
+    ce.subject = invoice.business_identifier
+    ce.time = invoice.payment_date
+    ce.type = 'payment'
+    ce.data = payload
+
+    _send_to_queue(to_queue_message(ce))
+
+
+def _send_to_queue(payload: bytes):
+    """Send payload to the queue."""
+    if not ((gcp_auth_key := current_app.config.get('GCP_AUTH_KEY')) and
+            (audience := current_app.config.get('AUDIENCE')) and
+            (topic_name := current_app.config.get('TOPIC_NAME')) and
+            (publisher_audience := current_app.config.get('PUBLISHER_AUDIENCE'))):
+
+        raise Exception('missing setup arguments')  # pylint: disable=W0719
+
+    # get authenticated publisher
+    try:
+        service_account_info = json.loads(base64.b64decode(gcp_auth_key).decode('utf-8'))
+        credentials = jwt.Credentials.from_service_account_info(
+            service_account_info, audience=audience
+        )
+        credentials_pub = credentials.with_claims(audience=publisher_audience)
+        publisher = pubsub_v1.PublisherClient(credentials=credentials_pub)
+    except Exception as error:  # noqa: B902
+        raise Exception('Unable to create a connection', error) from error  # pylint: disable=W0719
+
+    try:
+        future = publisher.publish(topic_name, payload)
+
+        return future.result()
+    except (CancelledError, TimeoutError) as error:
+        raise Exception('Unable to post to queue', error) from error  # pylint: disable=W0719

pay-api/src/pay_api/services/payment_transaction.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import uuid
+from contextlib import suppress
 from datetime import datetime
 from typing import Dict
 
@@ -27,6 +28,7 @@
 from pay_api.models import PaymentTransaction as PaymentTransactionModel
 from pay_api.models import PaymentTransactionSchema
 from pay_api.services.base_payment_system import PaymentSystemService
+from pay_api.services.gcp_queue_publisher import publish_to_queue
 from pay_api.services.invoice import Invoice
 from pay_api.services.invoice_reference import InvoiceReference
 from pay_api.services.payment_account import PaymentAccount
@@ -497,6 +499,12 @@ def publish_status(transaction_dao: PaymentTransactionModel, invoice: Invoice):
 
         try:
             publish_response(payload=payload, subject=get_pay_subject_name(invoice.corp_type_code))
+
+            # First stage in rolling in the new queue services
+            # It'll not block or disrupt any flows (in theory)
+            with suppress(Exception):
+                publish_to_queue(payload=payload, invoice=invoice)
+
         except Exception as e:  # NOQA pylint: disable=broad-except
             current_app.logger.error(e)
             current_app.logger.warning(