Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

22770 - Refactor EFT job so it works with multiple EFT credit invoice links #1677

Merged
merged 3 commits into from
Aug 12, 2024
Merged

22770 - Refactor EFT job so it works with multiple EFT credit invoice links #1677

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8de0a01
Refactor EFT job so it works with multiple EFT credit invoice links
seeker25 Aug 12, 2024
56cda2e
Minor tweaks for unit tests
seeker25 Aug 12, 2024
f6af29e
minor
seeker25 Aug 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions jobs/payment-jobs/invoke_jobs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,8 +136,7 @@ def run(job_name, argument=None):
RoutingSlipTask.adjust_routing_slips()
application.logger.info('<<<< Completed Routing Slip tasks >>>>')
case 'EFT':
overdue_accounts = EFTTask.link_electronic_funds_transfers_cfs()
EFTTask.unlock_overdue_accounts(overdue_accounts)
EFTTask.link_electronic_funds_transfers_cfs()
EFTTask.reverse_electronic_funds_transfers_cfs()
application.logger.info('<<<< Completed EFT tasks >>>>')
case 'EJV_PAYMENT':
Expand Down
253 changes: 147 additions & 106 deletions jobs/payment-jobs/tasks/eft_task.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,12 @@

from flask import current_app

from pay_api.exceptions import BusinessException
from pay_api.models import CfsAccount as CfsAccountModel
from pay_api.models import EFTCredit as EFTCreditModel
from pay_api.models import EFTCreditInvoiceLink as EFTCreditInvoiceLinkModel
from pay_api.models import EFTShortnamesHistorical as EFTShortnameHistoryModel
from pay_api.models import Invoice as InvoiceModel
from pay_api.models import InvoiceReference as InvoiceReferenceModel
from pay_api.models import PaymentAccount as PaymentAccountModel
from pay_api.models import Payment as PaymentModel
from pay_api.models import Receipt as ReceiptModel
from pay_api.models import db
Expand All @@ -42,6 +41,9 @@
class EFTTask: # pylint:disable=too-few-public-methods
"""Task to link electronic funds transfers."""

history_group_ids = set()
overdue_account_ids = {}

@classmethod
def get_eft_credit_invoice_links_by_status(cls, status: str) \
-> List[tuple[InvoiceModel, EFTCreditInvoiceLinkModel, CfsAccountModel]]:
Expand All @@ -51,156 +53,195 @@ def get_eft_credit_invoice_links_by_status(cls, status: str) \
.filter(CfsAccountModel.status == CfsAccountStatus.ACTIVE.value) \
.group_by(CfsAccountModel.account_id).subquery('latest_cfs_account')

query = db.session.query(InvoiceModel, EFTCreditInvoiceLinkModel, CfsAccountModel) \
.join(EFTCreditModel, EFTCreditModel.id == EFTCreditInvoiceLinkModel.eft_credit_id) \
.join(InvoiceModel, InvoiceModel.id == EFTCreditInvoiceLinkModel.invoice_id) \
cil_rollup = db.session.query(func.min(EFTCreditInvoiceLinkModel.id).label('id'),
EFTCreditInvoiceLinkModel.invoice_id,
EFTCreditInvoiceLinkModel.status_code,
EFTCreditInvoiceLinkModel.receipt_number,
func.array_agg(EFTCreditInvoiceLinkModel.id) # pylint: disable=not-callable
.label('link_ids'),
func.sum(EFTCreditInvoiceLinkModel.amount).label('rollup_amount')) \
.filter(EFTCreditInvoiceLinkModel.status_code == status) \
.group_by(EFTCreditInvoiceLinkModel.invoice_id,
EFTCreditInvoiceLinkModel.status_code,
EFTCreditInvoiceLinkModel.receipt_number) \
.subquery()

class EFTCILRollup(db.Model):
"""Here so we can map our subquery tuple to an object, it's only used locally."""

__table__ = cil_rollup
__mapper_args__ = {
'primary_key': [cil_rollup.c.invoice_id, cil_rollup.c.status_code, cil_rollup.c.receipt_number]
}

query = db.session.query(InvoiceModel, CfsAccountModel, EFTCILRollup) \
.join(cil_rollup, InvoiceModel.id == cil_rollup.c.invoice_id) \
.join(CfsAccountModel, CfsAccountModel.account_id == InvoiceModel.payment_account_id) \
.join(latest_cfs_account, CfsAccountModel.id == latest_cfs_account.c.max_id_per_payment_account) \
.options(lazyload('*')) \
.filter(EFTCreditInvoiceLinkModel.status_code == status) \
.filter(InvoiceModel.payment_method_code == PaymentMethod.EFT.value)
.filter(InvoiceModel.payment_method_code == PaymentMethod.EFT.value) \
.filter(InvoiceModel.total == cil_rollup.c.rollup_amount)

match status:
case EFTCreditInvoiceStatus.PENDING.value:
query = query.filter(InvoiceModel.disbursement_status_code.is_(None))
query = query.filter(InvoiceModel.invoice_status_code.in_([InvoiceStatus.APPROVED.value,
InvoiceStatus.OVERDUE.value]))
case EFTCreditInvoiceStatus.PENDING_REFUND.value:
query = query.filter(InvoiceModel.invoice_status_code == InvoiceStatus.PAID.value)
query = query.filter(or_(InvoiceModel.disbursement_status_code.is_(
None), InvoiceModel.disbursement_status_code == DisbursementStatus.COMPLETED.value))
query = query.filter(InvoiceModel.invoice_status_code == InvoiceStatus.PAID.value)
case _:
pass
return query.order_by(InvoiceModel.payment_account_id, EFTCreditInvoiceLinkModel.id).all()

@classmethod
def get_eft_history_by_group_id(cls, related_group_id: int) -> EFTShortnameHistoryModel:
"""Get EFT short name historical record by related group id."""
return (db.session.query(EFTShortnameHistoryModel)
.filter(EFTShortnameHistoryModel.related_group_link_id == related_group_id)).one_or_none()

@classmethod
def _finalize_shortname_history(cls, group_set: set, invoice_link: EFTCreditInvoiceLinkModel):
"""Finalize EFT short name historical record state."""
if invoice_link.link_group_id is None or invoice_link.link_group_id in group_set:
return

group_set.add(invoice_link.link_group_id)
if history_model := cls.get_eft_history_by_group_id(invoice_link.link_group_id):
history_model.hidden = False
history_model.is_processing = False
history_model.flush()
return query.order_by(InvoiceModel.payment_account_id, cil_rollup.c.invoice_id).all()

@classmethod
def link_electronic_funds_transfers_cfs(cls) -> dict:
"""Replicate linked EFT's as receipts inside of CFS and mark invoices as paid."""
credit_invoice_links = cls.get_eft_credit_invoice_links_by_status(EFTCreditInvoiceStatus.PENDING.value)
overdue_accounts = {}
history_group_ids = set()
for invoice, credit_invoice_link, cfs_account in credit_invoice_links:
target_status = EFTCreditInvoiceStatus.PENDING.value
credit_invoice_links = cls.get_eft_credit_invoice_links_by_status(target_status)
cls.history_group_ids = set()
for invoice, cfs_account, cil_rollup in credit_invoice_links:
try:
current_app.logger.info(f'PayAccount: {invoice.payment_account_id} Id: {credit_invoice_link.id} -'
f' Invoice Id: {invoice.id} - Amount: {credit_invoice_link.amount}')
invoice_reference = InvoiceReferenceModel.find_by_invoice_id_and_status(
credit_invoice_link.invoice_id, InvoiceReferenceStatus.ACTIVE.value
)
invoice_reference.status_code = InvoiceReferenceStatus.COMPLETED.value
invoice_reference.flush()
# Note: Not creating the entire EFT as a receipt because it can be mapped to multiple CFS accounts.
# eft_credit_invoice_links table should reflect exactly what's in CAS.
receipt_number = f'EFTCIL{credit_invoice_link.id}'
CFSService.create_cfs_receipt(
cfs_account=cfs_account,
rcpt_number=receipt_number,
rcpt_date=credit_invoice_link.created_on.strftime('%Y-%m-%d'),
amount=credit_invoice_link.amount,
payment_method=PaymentMethod.EFT.value,
access_token=CFSService.get_token(PaymentSystem.FAS).json().get('access_token'))
CFSService.apply_receipt(cfs_account, receipt_number, invoice_reference.invoice_number)
ReceiptModel(receipt_number=receipt_number,
receipt_amount=credit_invoice_link.amount,
invoice_id=invoice_reference.invoice_id,
receipt_date=datetime.now(tz=timezone.utc)).flush()
PaymentModel(payment_method_code=PaymentMethod.EFT.value,
payment_status_code=PaymentStatus.COMPLETED.value,
payment_system_code=PaymentSystem.PAYBC.value,
invoice_number=invoice.id,
invoice_amount=invoice.total,
payment_account_id=cfs_account.account_id,
payment_date=datetime.now(tz=timezone.utc),
paid_amount=credit_invoice_link.amount,
receipt_number=receipt_number).flush()
current_app.logger.info(f'PayAccount: {invoice.payment_account_id} Id: {cil_rollup.id} -'
f' Invoice Id: {invoice.id} - Amount: {cil_rollup.rollup_amount}')
receipt_number = f'EFTCIL{cil_rollup.id}'
if invoice.invoice_status_code == InvoiceStatus.OVERDUE.value:
overdue_accounts[invoice.payment_account_id] = cfs_account.payment_account

invoice.invoice_status_code = InvoiceStatus.PAID.value
invoice.paid = credit_invoice_link.amount
invoice.payment_date = datetime.now(tz=timezone.utc)
invoice.flush()

credit_invoice_link.status_code = EFTCreditInvoiceStatus.COMPLETED.value
credit_invoice_link.receipt_number = receipt_number
credit_invoice_link.flush()

cls._finalize_shortname_history(history_group_ids, credit_invoice_link)
cls.overdue_account_ids[invoice.payment_account_id] = cfs_account.payment_account
cls._create_receipt_and_invoice(cfs_account, cil_rollup, invoice, receipt_number)
cls._update_cil_and_shortname_history(cil_rollup, receipt_number=receipt_number)
db.session.commit()
except Exception as e: # NOQA # pylint: disable=broad-except
capture_message(
f'Error on linking EFT invoice links in CFS '
f'Account id={invoice.payment_account_id} '
f'EFT Credit invoice Link : {credit_invoice_link.id}'
f'EFT Credit invoice Link : {cil_rollup.id}'
f'ERROR : {str(e)}', level='error')
current_app.logger.error(f'Error Account id={invoice.payment_account_id} - '
f'EFT Credit invoice Link : {credit_invoice_link.id}', exc_info=True)
f'EFT Credit invoice Link : {cil_rollup.id}', exc_info=True)
db.session.rollback()
continue
return overdue_accounts
cls.unlock_overdue_accounts()

@classmethod
def reverse_electronic_funds_transfers_cfs(cls):
"""Reverse electronic funds transfers receipts in CFS and reset invoices."""
credit_invoice_links = cls.get_eft_credit_invoice_links_by_status(EFTCreditInvoiceStatus.PENDING_REFUND.value)
history_group_ids = set()
for invoice, credit_invoice_link, cfs_account in credit_invoice_links:
target_status = EFTCreditInvoiceStatus.PENDING_REFUND.value
credit_invoice_links = cls.get_eft_credit_invoice_links_by_status(target_status)
cls.history_group_ids = set()
for invoice, cfs_account, cil_rollup in credit_invoice_links:
try:
current_app.logger.info(f'PayAccount: {invoice.payment_account_id} Id: {credit_invoice_link.id} -'
f' Invoice Id: {invoice.id} - Amount: {credit_invoice_link.amount}')
receipt_number = credit_invoice_link.receipt_number
invoice_reference = InvoiceReferenceModel.find_by_invoice_id_and_status(
invoice.id, InvoiceReferenceStatus.COMPLETED.value
)
invoice_reference.status_code = InvoiceReferenceStatus.ACTIVE.value
invoice_reference.flush()
CFSService.reverse_rs_receipt_in_cfs(cfs_account, receipt_number, ReverseOperation.VOID.value)
invoice.invoice_status_code = InvoiceStatus.APPROVED.value
invoice.paid = 0
invoice.payment_date = None
invoice.flush()
if payment := PaymentModel.find_payment_for_invoice(invoice.id):
db.session.delete(payment)
for receipt in ReceiptModel.find_all_receipts_for_invoice(invoice.id):
db.session.delete(receipt)
credit_invoice_link.status_code = EFTCreditInvoiceStatus.REFUNDED.value
credit_invoice_link.flush()

cls._finalize_shortname_history(history_group_ids, credit_invoice_link)
current_app.logger.info(f'PayAccount: {invoice.payment_account_id} Id: {cil_rollup.id} -'
f' Invoice Id: {invoice.id} - Amount: {cil_rollup.rollup_amount}')
receipt_number = cil_rollup.receipt_number
cls._rollback_receipt_and_invoice(cfs_account, invoice, receipt_number)
cls._update_cil_and_shortname_history(cil_rollup)
db.session.commit()
except Exception as e: # NOQA # pylint: disable=broad-except
capture_message(
f'Error on reversing EFT invoice links in CFS '
f'Account id={invoice.payment_account_id} '
f'EFT Credit invoice Link : {credit_invoice_link.id}'
f'EFT Credit invoice Link : {cil_rollup.id}'
f'ERROR : {str(e)}', level='error')
current_app.logger.error(f'Error Account id={invoice.payment_account_id} - '
f'EFT Credit invoice Link : {credit_invoice_link.id}', exc_info=True)
f'EFT Credit invoice Link : {cil_rollup.id}', exc_info=True)
db.session.rollback()
continue

@classmethod
def unlock_overdue_accounts(cls, overdue_accounts: dict):
def unlock_overdue_accounts(cls):
"""Check and unlock overdue EFT accounts."""
for payment_account_id in overdue_accounts:
for (payment_account_id, payment_account) in cls.overdue_account_ids.items():
if InvoiceService.has_overdue_invoices(payment_account_id):
continue
payment_account: PaymentAccountModel = overdue_accounts[payment_account_id]
AuthEvent.publish_unlock_account_event(payment_account)

@classmethod
def _get_eft_history_by_group_id(cls, related_group_id: int) -> EFTShortnameHistoryModel:
"""Get EFT short name historical record by related group id."""
return (db.session.query(EFTShortnameHistoryModel)
.filter(EFTShortnameHistoryModel.related_group_link_id == related_group_id)).one_or_none()

@classmethod
def _finalize_shortname_history(cls, group_set: set, invoice_link: EFTCreditInvoiceLinkModel):
"""Finalize EFT short name historical record state."""
if invoice_link.link_group_id is None or invoice_link.link_group_id in group_set:
return

group_set.add(invoice_link.link_group_id)
if history_model := cls._get_eft_history_by_group_id(invoice_link.link_group_id):
history_model.hidden = False
history_model.is_processing = False
history_model.flush()

@classmethod
def _update_cil_and_shortname_history(cls, cil_rollup, receipt_number=None):
"""Update electronic invoice links."""
cils = db.session.query(EFTCreditInvoiceLinkModel).filter(
EFTCreditInvoiceLinkModel.id.in_(cil_rollup.link_ids)).all()
for cil in cils:
cil.status_code = EFTCreditInvoiceStatus.COMPLETED.value if receipt_number \
else EFTCreditInvoiceStatus.REFUNDED.value
cil.receipt_number = receipt_number or cil.receipt_number
cil.flush()
cls._finalize_shortname_history(cls.history_group_ids, cil)

@classmethod
def _create_receipt_and_invoice(cls,
cfs_account: CfsAccountModel,
cil_rollup: EFTCreditInvoiceLinkModel,
invoice: InvoiceModel,
receipt_number: str):
"""Create receipt in CFS and marks invoice as paid, with payment and receipt rows."""
if not (invoice_reference := InvoiceReferenceModel.find_by_invoice_id_and_status(
cil_rollup.invoice_id, InvoiceReferenceStatus.ACTIVE.value
)):
raise BusinessException(f'No invoice reference found for invoice id: {invoice.id}')
invoice_reference.status_code = InvoiceReferenceStatus.COMPLETED.value
invoice_reference.flush()
# Note: Not creating the entire EFT as a receipt because it can be mapped to multiple CFS accounts.
# eft_credit_invoice_links table should reflect exactly what's in CAS.
CFSService.create_cfs_receipt(
cfs_account=cfs_account,
rcpt_number=receipt_number,
rcpt_date=datetime.now(tz=timezone.utc).strftime('%Y-%m-%d'),
amount=cil_rollup.rollup_amount,
payment_method=PaymentMethod.EFT.value,
access_token=CFSService.get_token(PaymentSystem.FAS).json().get('access_token'))
CFSService.apply_receipt(cfs_account, receipt_number, invoice_reference.invoice_number)
ReceiptModel(receipt_number=receipt_number,
receipt_amount=cil_rollup.rollup_amount,
invoice_id=invoice_reference.invoice_id,
receipt_date=datetime.now(tz=timezone.utc)).flush()
PaymentModel(payment_method_code=PaymentMethod.EFT.value,
payment_status_code=PaymentStatus.COMPLETED.value,
payment_system_code=PaymentSystem.PAYBC.value,
invoice_number=invoice.id,
invoice_amount=invoice.total,
payment_account_id=cfs_account.account_id,
payment_date=datetime.now(tz=timezone.utc),
paid_amount=cil_rollup.rollup_amount,
receipt_number=receipt_number).flush()
invoice.invoice_status_code = InvoiceStatus.PAID.value
invoice.paid = cil_rollup.rollup_amount
invoice.payment_date = datetime.now(tz=timezone.utc)
invoice.flush()

@classmethod
def _rollback_receipt_and_invoice(cls, cfs_account: CfsAccountModel, invoice: InvoiceModel, receipt_number: str):
"""Rollback receipt in CFS and reset invoice status."""
if not (invoice_reference := InvoiceReferenceModel.find_by_invoice_id_and_status(
invoice.id, InvoiceReferenceStatus.COMPLETED.value
)):
raise BusinessException(f'No invoice reference found for invoice id: {invoice.id}')
CFSService.reverse_rs_receipt_in_cfs(cfs_account, receipt_number, ReverseOperation.VOID.value)
invoice_reference.status_code = InvoiceReferenceStatus.ACTIVE.value
invoice_reference.flush()
invoice.invoice_status_code = InvoiceStatus.APPROVED.value
invoice.paid = 0
invoice.payment_date = None
invoice.flush()
if payment := PaymentModel.find_payment_for_invoice(invoice.id):
db.session.delete(payment)
for receipt in ReceiptModel.find_all_receipts_for_invoice(invoice.id):
db.session.delete(receipt)
Loading
Loading