[Issue #6814] Create ecs task for setting up cert user (#7021)

## Summary

<!-- Use "Fixes" to automatically close issue upon PR merge. Use "Work
for" when UAT is required. -->
Fixes / Work for #6814 

## Changes proposed

<!-- What was added, updated, or removed in this PR. -->
Added a task/command to create a LegacyCertificate based on a
`tcertificate.currentcertid` and some `Role.role_ids`. Creates a `User`
of `UserType.LEGACY_CERTIFICATE` and assigns it to the agency associated
with the tcertificate along with adding the `AgencyUser` and
`AgencyUserRoles`.

```
make cmd args="task setup-cert-user --cert-id 1001 --role-ids <UUID> --role-ids <UUID>"
```

## Context for reviewers

<!-- Technical or background context, more in-depth details of the
implementation, and anything else you'd like reviewers to know about
that will help them understand the changes in the PR. -->
This will take a tcertificate and create an associated user with roles
attached to it.

## Validation steps

<!-- Manual testing instructions, as well as any helpful references
(screenshots, GIF demos, code examples or output). -->

Author: jakobpederson

api/src/db/migrations/versions/2025_11_17_add_staging_tcertificate_uuid.py

@@ -0,0 +1,50 @@
+"""add_staging_tcertificate_uuid
+
+Revision ID: 15dd300bbcc3
+Revises: 19e80334aaae
+Create Date: 2025-11-17 18:51:44.936793
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "15dd300bbcc3"
+down_revision = "19e80334aaae"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "tcertificates",
+        sa.Column(
+            "tcertificates_id",
+            sa.UUID(),
+            server_default=sa.text("gen_random_uuid()"),
+            nullable=True,
+        ),
+        schema="staging",
+    )
+    op.create_unique_constraint(
+        op.f("tcertificates_tcertificates_id_uniq"),
+        "tcertificates",
+        ["tcertificates_id"],
+        schema="staging",
+    )
+    op.execute(sa.text("UPDATE staging.tcertificates SET tcertificates_id = gen_random_uuid()"))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(
+        op.f("tcertificates_tcertificates_id_uniq"),
+        "tcertificates",
+        schema="staging",
+        type_="unique",
+    )
+    op.drop_column("tcertificates", "tcertificates_id", schema="staging")
+    # ### end Alembic commands ###

api/src/db/models/staging/certificates.py

@@ -1,6 +1,16 @@
+import uuid
+
+from sqlalchemy import text
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
 from src.db.models.legacy_mixin.certificates_mixin import TcertificatesMixin
 from src.db.models.staging.staging_base import StagingBase, StagingParamMixin
 
 
 class Tcertificates(StagingBase, TcertificatesMixin, StagingParamMixin):
     __tablename__ = "tcertificates"
+
+    tcertificates_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID, server_default=text("gen_random_uuid()"), unique=True
+    )

api/src/task/__init__.py

@@ -9,6 +9,7 @@
 import src.task.apply.create_application_submission_task  # noqa: F401 isort:skip
 import src.task.generate_internal_token  # noqa: F401 isort:skip
 import src.task.forms.update_form_task  # noqa: F401 isort:skip
+import src.task.certificates.setup_cert_user_task  # noqa: F401 isort:skip
 import src.task.forms.list_forms_task  # noqa: F401 isort:skip
 import src.task.opportunities.generate_opportunity_sql  # noqa: F401 isort:skip
 import src.task.opportunities.build_automatic_opportunities  # noqa: F401 isort:skip

api/src/task/certificates/__init__.py

@@ -0,0 +1,198 @@
+import logging
+import uuid
+from datetime import date
+from enum import StrEnum
+
+import click
+from sqlalchemy import select
+
+import src.adapters.db.flask_db as flask_db
+import src.util.datetime_util as datetime_util
+from src.adapters import db
+from src.constants.lookup_constants import UserType
+from src.db.models import staging
+from src.db.models.user_models import (
+    Agency,
+    AgencyUser,
+    AgencyUserRole,
+    LegacyCertificate,
+    Role,
+    User,
+)
+from src.task.ecs_background_task import ecs_background_task
+from src.task.task import Task
+from src.task.task_blueprint import task_blueprint
+
+logger = logging.getLogger(__name__)
+
+FUTURE_DATE = date(2050, 1, 1)
+
+
+class SetupCertUserTaskStatus(StrEnum):
+    INVALID_ROLE_IDS = "Invalid role ids"
+    AGENCY_NOT_FOUND = "Agency not found"
+    LEGACY_CERTIFICATE_ALREADY_EXISTS = "LegacyCertificate already exists"
+    SUCCESS = "Success"
+    TCERTIFICATE_IS_EXPIRED = "Tcertificate is expired"
+    TCERTIFICATE_NOT_FOUND = "Tcertificate not found"
+    TCERTIFICATE_IS_MISSING_SERIAL_NUMBER = "Tcertificate is missing serial number"
+
+
+@task_blueprint.cli.command("setup-cert-user", help="Setup the LegacyCertificate and User")
+@click.option("--tcertficates-id", "-t", help="tcertificates_id on Staging Tcertificate")
+@click.option("--role-ids", "-t", help="role_id of role that needs to be added", multiple=True)
+@flask_db.with_db_session()
+@ecs_background_task(task_name="setup-cert-user")
+def setup_cert_user(db_session: db.Session, cert_id: str, role_ids: list[str]) -> None:
+    SetupCertUserTask(db_session, cert_id, role_ids).run_task()
+
+
+class SetupCertUserTask(Task):
+
+    def __init__(self, db_session: db.Session, tcertificates_id: str, role_ids: list[str]):
+        super().__init__(db_session)
+        self.tcertificates_id = tcertificates_id
+        self.role_ids = role_ids
+
+    def run_task(self) -> None:
+        with self.db_session.begin():
+            self.setup_cert()
+
+    def setup_cert(self) -> SetupCertUserTaskStatus:
+        logger.info("setup cert user start")
+        roles = self.get_roles()
+        if roles is None:
+            return SetupCertUserTaskStatus.INVALID_ROLE_IDS
+
+        tcertificate = self.get_tcertificate()
+        if tcertificate is None:
+            logger.warning("Tcertificate not found")
+            return SetupCertUserTaskStatus.TCERTIFICATE_NOT_FOUND
+        if not tcertificate.serial_num:
+            logger.warning("Tcertificate is missing serial number")
+            return SetupCertUserTaskStatus.TCERTIFICATE_IS_MISSING_SERIAL_NUMBER
+        valid_expiration_date = tcertificate.expirationdate or FUTURE_DATE
+        if valid_expiration_date <= datetime_util.get_now_us_eastern_date():
+            logger.warning("Cert is expired")
+            return SetupCertUserTaskStatus.TCERTIFICATE_IS_EXPIRED
+        if self.is_existing_certificate(tcertificate):
+            logger.warning("LegacyCertificate already exists")
+            return SetupCertUserTaskStatus.LEGACY_CERTIFICATE_ALREADY_EXISTS
+
+        agency, related_agencies = self.get_agencies(tcertificate)
+        if agency is None:
+            return SetupCertUserTaskStatus.AGENCY_NOT_FOUND
+
+        else:
+            self.process_cert_user(
+                roles, tcertificate, agency, related_agencies, valid_expiration_date
+            )
+        logger.info("setup cert user complete")
+        return SetupCertUserTaskStatus.SUCCESS
+
+    def process_cert_user(
+        self,
+        roles: list[Role],
+        tcertificate: staging.certificates.Tcertificates,
+        agency: Agency,
+        related_agencies: list[Agency],
+        valid_expiration_date: date,
+    ) -> None:
+        all_agencies = related_agencies + [agency]
+        user = self.create_user_with_agency_roles(all_agencies, roles)
+        legacy_certificate = LegacyCertificate(
+            legacy_certificate_id=uuid.uuid4(),
+            agency=agency,
+            cert_id=tcertificate.currentcertid,
+            expiration_date=valid_expiration_date,
+            serial_number=tcertificate.serial_num,
+            user=user,
+        )
+        self.db_session.add(legacy_certificate)
+
+        logger.info(
+            "Created legacy certificate",
+            extra={
+                "legacy_certificate_id": legacy_certificate.legacy_certificate_id,
+                "user_id": user.user_id,
+                "agency_code": agency.agency_code,
+            },
+        )
+
+    def create_user_with_agency_roles(self, agencies: list[Agency], roles: list[Role]) -> User:
+        user = User(user_id=uuid.uuid4(), user_type=UserType.LEGACY_CERTIFICATE)
+        self.db_session.add(user)
+
+        log_extra = {"user_id": user.user_id}
+        logger.info("Created legacy cert user", extra=log_extra)
+
+        for agency in agencies:
+            agency_user = AgencyUser(user=user, agency=agency)
+            self.db_session.add(agency_user)
+
+            agency_roles = [AgencyUserRole(agency_user=agency_user, role=r) for r in roles]
+
+            self.db_session.add_all(agency_roles)
+
+            logger.info(
+                "Added user to agency",
+                extra=log_extra
+                | {"agency_code": agency.agency_code, "role_ids": [r.role_id for r in roles]},
+            )
+
+        return user
+
+    def get_roles(self) -> list[Role] | None:
+        roles = list(
+            self.db_session.scalars(
+                select(Role).where(Role.role_id.in_([uuid.UUID(r) for r in self.role_ids]))
+            ).all()
+        )
+        if len(self.role_ids) != len(roles):
+            log_extra = {"found_role_ids": [r.role_id for r in roles] if roles else []}
+            logger.warning("Invalid role ids", extra=log_extra)
+            return None
+        return roles
+
+    def get_tcertificate(self) -> staging.certificates.Tcertificates | None:
+        return self.db_session.scalars(
+            select(staging.certificates.Tcertificates).where(
+                staging.certificates.Tcertificates.tcertificates_id
+                == uuid.UUID(self.tcertificates_id)
+            )
+        ).one_or_none()
+
+    def get_agencies(
+        self, tcertificate: staging.certificates.Tcertificates
+    ) -> tuple[Agency | None, list[Agency]]:
+        agency = self.db_session.scalar(
+            select(Agency).where(Agency.agency_code == tcertificate.agencyid)
+        )
+        if not agency:
+            logger.warning("Agency not found")
+            return (None, [])
+        agencies: tuple[Agency | None, list[Agency]] = (agency, [])
+        """
+        If the tcertificate agency has is_multilevel marked as True then:
+        1. fetch every agency that starts with the same prefix as the agency:
+            SELECT * FROM agency WHERE agency_code LIKE '{agency.agency_code}-%'
+        2. add an AgencyUser and AgencyUserRole for every subagency
+        this is to mimic the grants.gov behavior
+        """
+        if agency.is_multilevel_agency:
+            search_pattern = f"{agency.agency_code}-%"
+            agency_query_results = list(
+                self.db_session.scalars(
+                    select(Agency).where(Agency.agency_code.like(search_pattern))
+                ).all()
+            )
+            agencies[1].extend(agency_query_results)
+        return agencies
+
+    def is_existing_certificate(self, tcertificate: staging.certificates.Tcertificates) -> bool:
+        existing_tcertificate = self.db_session.scalars(
+            select(LegacyCertificate.legacy_certificate_id).where(
+                LegacyCertificate.cert_id == tcertificate.currentcertid
+            )
+        ).one_or_none()
+        return existing_tcertificate is not None

api/src/task/certificates/setup_cert_user_task.py

@@ -2940,6 +2940,24 @@ class Meta:
     created_date = factory.Faker("date_time_between", start_date="-2y", end_date="-1y")
 
 
+class StagingTcertificatesFactory(AbstractStagingFactory):
+    class Meta:
+        model = staging.certificates.Tcertificates
+
+    tcertificates_id = Generators.UuidObj
+    previouscertid = factory.Sequence(lambda n: f"{n}")
+    currentcertid = factory.Sequence(lambda n: f"{1000 + n}")
+    orgduns = None
+    orgname = None
+    expirationdate = factory.Faker("date_between", start_date="+3d", end_date="+1y")
+    agencyid = factory.Faker("agency_code")
+    serial_num = factory.Faker("pystr", min_chars=15, max_chars=15)
+    created_date = factory.Faker("date_between", start_date="-2y", end_date="-1y")
+    certemail = factory.Faker("email")
+    creator_id = factory.Faker("email")
+    is_selfsigned = "Y"
+
+
 ###################
 # Extract Factories
 ###################