Complete App refactor - #19
Merged
Merged
Conversation
Contributor
|
Warning Review limit reached
Next review available in: 33 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (12)
Note
|
| Layer / File(s) | Summary |
|---|---|
Frontend routing and table flows frontend/web/src/features/routes/..., frontend/web/src/features/edi_headers/..., frontend/web/src/routes/... |
Adds separate inbound/outbound route modals, typed destination and processing-mode handling, route sorting/grouping, EDI-header deletion actions, and updated transaction display/API paths. |
API contracts and repository adapters services/api/src/api/adapters/..., services/api/src/api/ports/..., services/api/src/api/domain/models.py, services/api/src/api/core/services/... |
Adds typed repository ports, SQLAlchemy adapters, transaction DTOs, route validation, partner/header/token persistence, and Unit-of-Work service wiring. |
Database session and schema contracts libs/database/src/database/..., libs/domain/src/domain/... |
Adds global/tenant session types, session validation, parameterized RLS configuration, new EDI metadata fields, updated processing defaults, and domain enums/models. |
Transformation and delivery pipeline libs/pipeline/src/pipeline/..., libs/transformer/src/transformer/... |
Renames translation APIs to transformation APIs, adds inbound/outbound transformation services, introduces webhook/SFTP/AS2 delivery strategies and routing, and publishes idempotent pipeline events. |
Compute and provisioning workers services/workers/compute/..., services/workers/orchestrator/... |
Adds SQS compute processing, provisioning replication, queue adapters, tenant resolution, delivery polling, SSRF URL validation, Vault access, and worker tests. |
Configuration, queue setup, and project guidance Makefile, docker-compose.yml, docker/localstack/..., pyproject.toml, .agents/..., TECHNICAL_DEBT.md |
Updates worker commands, LocalStack volumes and queues, tooling exclusions, development guidance, and technical-debt entries. |
Estimated code review effort: 5 (Critical) | ~120 minutes
Possibly related PRs
- pramodnarayana/soopaedi#18: Shares transformation, delivery-event, queue-routing, and worker SQS flow changes.
- pramodnarayana/soopaedi#16: Shares worker Makefile targets and LocalStack queue/DLQ initialization.
- pramodnarayana/soopaedi#1: Shares database session and row-level security provisioning changes.
Poem
A rabbit hops through queues so bright,
Transforming EDI by moonlit light.
Routes group, workers hum,
New typed contracts drum—
And delivery lands just right.
🚥 Pre-merge checks | ✅ 3 | ❌ 2
❌ Failed checks (1 warning, 1 inconclusive)
| Check name | Status | Explanation | Resolution |
|---|---|---|---|
| Docstring Coverage | Docstring coverage is 25.00% which is insufficient. The required threshold is 80.00%. | Write docstrings for the functions missing them to satisfy the coverage threshold. | |
| Title check | ❓ Inconclusive | The title is very generic and does not clearly identify the main change in the pull request. | Use a more specific title that names the primary refactor area, such as workers, routing, or transformer pipeline changes. |
✅ Passed checks (3 passed)
| Check name | Status | Explanation |
|---|---|---|
| Description Check | ✅ Passed | Check skipped - CodeRabbit’s high-level summary is enabled. |
| Linked Issues check | ✅ Passed | Check skipped because no linked issues were found for this pull request. |
| Out of Scope Changes check | ✅ Passed | Check skipped because no linked issues were found for this pull request. |
✨ Finishing Touches
🧪 Generate unit tests (beta)
- Create PR with unit tests
- Commit unit tests in branch
refactor/complete-app-refactor
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.
Comment @coderabbitai help to get the list of available commands.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 40
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
services/api/src/api/ports/transaction_repository.py (1)
6-70: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAdd
publish_outbox_eventto the data-plane portApiReceiverServicecallsdata_plane.publish_outbox_event(...), butDataPlaneRepositoryPortonly inheritsTransactionRepositoryPort, so this method still isn’t part of the typed contract.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/api/src/api/ports/transaction_repository.py` around lines 6 - 70, Add an async publish_outbox_event method to TransactionRepositoryPort, matching the arguments and return type used by ApiReceiverService and its data-plane repository implementation, so DataPlaneRepositoryPort exposes this call through inheritance. Keep the existing transaction methods unchanged.services/workers/compute/src/compute_worker/worker.py (1)
63-106: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winValidation failures aren't distinguished from transient failures — malformed messages are retried instead of being quarantined.
_process_messageraises plainValueErrorfor missingBody/trace_id/s3_uri(lines 70-71, 79-83), but the outerexcept Exception(line 103) treats these identically to a transient use-case failure: log and leave on the queue. A message that will never parse successfully keeps getting redelivered and reprocessed until (if configured) a DLQ redrive kicks in, wasting worker cycles in the meantime. The siblingSqsOutboxAdapterin this PR already establishes the correct pattern — distinguish permanent vs. transient and delete on permanent failure.🛡️ Proposed fix
+ receipt_handle = sqs_message.get("ReceiptHandle") try: body = sqs_message.get("Body") if not body: - raise ValueError("Message Body is missing") + raise _PermanentMessageError("Message Body is missing") ... if not trace_id or not isinstance(trace_id, str) or not trace_id.strip(): - raise ValueError("Required field 'trace_id' is missing or empty") + raise _PermanentMessageError("Required field 'trace_id' is missing or empty") ... - except Exception as e: + except _PermanentMessageError as e: + logger.error(f"Discarding unprocessable EDI message: {e}") + if receipt_handle: + await sqs_client.delete_message(QueueUrl=self.queue_url, ReceiptHandle=str(receipt_handle)) + except Exception as e: logger.error(f"Failed to process EDI message: {e}") # Message naturally returns to queue for Dead Letter Queue routing🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/workers/compute/src/compute_worker/worker.py` around lines 63 - 106, Update _process_message to distinguish malformed-message validation failures from transient processing errors: catch the validation errors raised for missing or invalid Body, trace_id, and s3_uri, log them as permanent failures, and delete the message using its ReceiptHandle. Keep transient use-case or queue-operation exceptions on the existing retry path, and preserve successful processing and deletion behavior.services/api/src/api/cdc_relay.py (1)
83-94: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAdd a real CDC DLQ target here.
MessageQueueNamehas noCDC_DLQ_QUEUE, and the queue isn’t provisioned indocker/localstack/init-aws.sh, so schema-validation failures will hit this path, fail to quarantine, and return 500 instead. Add the enum entry and provision the DLQ with the other SQS queues.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/api/src/api/cdc_relay.py` around lines 83 - 94, Define CDC_DLQ_QUEUE in the MessageQueueName enum used by queue.send, and provision the corresponding SQS queue in docker/localstack/init-aws.sh alongside the existing queues. Keep the schema-validation quarantine flow in the CDC relay unchanged so invalid events are sent to the newly available DLQ target.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docker/localstack/init-aws.sh`:
- Around line 13-15: Update the queue URL used by the local worker in
run_local_worker.py, specifically the polling configuration near the worker
startup flow, to reference TransformOrchestrationQueue instead of
EdiTransformerQueue. Keep the existing localhost endpoint and account/region
path unchanged.
In `@frontend/web/src/features/routes/components/CreateInboundRouteModal.tsx`:
- Around line 19-22: Update DialogContent in CreateInboundRouteModal to
explicitly set aria-describedby={undefined} when no dialog description is
provided, suppressing the Radix UI accessibility warning while preserving the
existing dialog behavior.
In `@frontend/web/src/features/routes/components/InboundRouteForm.tsx`:
- Around line 58-60: Replace the toUpperCase-based comparisons in the
webhook_id, as2_partner_id, and sftp_partner_id assignments with direct strict
comparisons to the corresponding DestinationType values, matching the filtering
logic used earlier in InboundRouteForm.
In `@libs/database/src/database/connection.py`:
- Around line 98-103: Update the tenant setup in the session context manager
around session.info["session_type"] to use PostgreSQL’s set_config function with
a bound tenant_id parameter instead of interpolating it into the SET LOCAL SQL
string. Preserve the existing transaction-local behavior and yield flow.
In
`@libs/database/src/database/migrations/global/versions/42c7e50a7b1c_global_initial_schema.py`:
- Line 271: Revert the change to the existing migration’s processing_mode
definition, restoring its original server_default. Create a new Alembic
migration that uses op.alter_column to update the processing_mode server
default, including the appropriate upgrade and downgrade behavior.
In `@libs/identity/src/identity/dependencies.py`:
- Around line 180-184: Update the tenant context management around
_tenant_id.set in the dependency flow: capture the returned token, then reset
_tenant_id with that token in the existing finally block after yielding
tenant_session. Preserve the current session lifecycle behavior while ensuring
the ContextVar state is restored on every exit path.
In `@libs/pipeline/src/pipeline/core/delivery/as2.py`:
- Around line 30-59: Ensure all failures in the AS2 delivery flow reach
terminal-status handling: broaden the guard around get_as2_partner,
get_local_as2_partner, and _as2_orchestrator.build so non-ValueError exceptions
also mark the message FAILED and emit completion. Add equivalent protection
around parse_mdn and subsequent MDN field access, preserving the existing
failure status update and _emit_delivery_completed behavior.
In `@libs/pipeline/src/pipeline/core/delivery/sftp.py`:
- Around line 28-30: Ensure every post-claim failure reaches a terminal delivery
state: in libs/pipeline/src/pipeline/core/delivery/sftp.py lines 28-30, move
get_sftp_partner and the delivery body under the existing failure-handling flow
so a missing partner marks the message FAILED and emits DELIVERY_COMPLETED; in
libs/pipeline/src/pipeline/core/delivery/as2.py lines 30-59, widen the
build-step handler to catch Exception; and in lines 78-106, wrap MDN parsing
with equivalent failure handling so any exception marks the message FAILED and
emits the completion event.
In `@libs/pipeline/src/pipeline/core/delivery/webhook.py`:
- Around line 45-47: Update the authentication handling around
partner.get("auth_header_vault_ref") in the webhook delivery flow to fail
immediately when the reference is configured but self.vault is unavailable.
Raise the established configuration or dependency error with clear context
instead of proceeding without an auth token; continue fetching the secret
through self.vault.get_secret when the dependency is present.
In `@libs/pipeline/src/pipeline/core/transformation/inbound.py`:
- Around line 74-108: Move the get_route() lookup that resolves
partnership_id_str to before the transformed_txns loop, then reuse the resolved
partnership_id_str in each save_edi_json() call. Ensure inbound transactions are
persisted with the partnership id rather than the initial None value.
In `@libs/transformer/src/transformer/infrastructure/adapters/bots_adapter.py`:
- Around line 110-112: Update the async transform method to invoke get_raw_ast
via asyncio.to_thread, awaiting its result while preserving the existing
arguments and error handling; add the asyncio import if needed.
In `@Makefile`:
- Around line 43-44: Update the Makefile dev target’s concurrently command to
include both make dev-worker-orchestrator and make dev-worker-compute, while
preserving the existing API and web startup commands and adjusting labels or
colors as needed for all processes.
- Line 78: Update the volume cleanup command in the Makefile to pipe matching
volume names through xargs -r, ensuring docker volume rm runs only when at least
one matching volume exists while preserving the current filtering pattern and
error suppression.
In `@pyproject.toml`:
- Around line 106-115: The coverage exclusion for */adapters/*_repository.py is
too broad and hides business logic such as validation in
outbound_route_repository.py. Remove the blanket repository glob and replace it
with explicit exclusions only for pass-through repository adapters that contain
no branching logic, preserving the existing narrower adapter exclusions.
In `@services/api/src/api/adapters/control_plane_aggregator.py`:
- Around line 14-36: Replace the broken multiple-inheritance composition in
SqlAlchemyControlPlaneAggregator by aligning its constructor and self.session
with the mixins’ GlobalSession contract, and update every delegated repository
initializer accordingly. Resolve the
get_outbound_edi_header_by_trading_partner_id and get_outbound_edi_headers
collisions between SqlAlchemyOutboundRouteRepository and
SqlAlchemyEdiHeaderRepository so only one compatible implementation is exposed.
Update SqlAlchemyOutboxRepository’s session annotation and initialization to
match GlobalSqlAlchemyRepository.
In `@services/api/src/api/adapters/edi_header_repository.py`:
- Around line 29-48: The update_outbound_edi_header method must handle commands
whose fields are all UNSET before constructing the SQL update. Return success
without executing an empty update statement, while preserving the existing
update and rowcount behavior when values contains fields.
In `@services/api/src/api/adapters/inbound_route_repository.py`:
- Around line 28-68: Extract the duplicated webhook, AS2, and SFTP existence
checks from create_inbound_route and update_inbound_route into a shared
_validate_destination helper. Have both methods call this helper with the tenant
and destination identifiers, while preserving the existing tenant scoping and
ValueError messages.
- Around line 169-184: Update the inbound route query around the
transaction_type condition so ordering is applied consistently before
scalars().first(), including when transaction_type is omitted. Ensure the
ordering deterministically prioritizes the generic NULL transaction_type
fallback for omitted types while preserving the existing specific-type matching
behavior.
In `@services/api/src/api/adapters/outbound_route_repository.py`:
- Around line 57-78: Extract the repeated AS2 and SFTP partner ownership checks
from the route create/update flows into shared validation helpers, such as
validate_as2_partner_owned and validate_sftp_partner_owned, and reuse them in
outbound route validation and inbound_route_repository.create_inbound_route.
Preserve the existing AS2 tenant rule allowing tenant_id or 0, the SFTP
tenant_id equality rule, and the current ValueError behavior and messages.
In `@services/api/src/api/adapters/repository.py`:
- Around line 30-44: In the __init__ method, remove the exploratory comment and
the unnecessary hasattr check, and invoke
SqlAlchemyApiTokenRepository.__init__(self, session) directly alongside the
other repository initializers.
In `@services/api/src/api/adapters/sftp_repository.py`:
- Around line 76-98: Update the partner-field assignment logic in the SFTP
partner update flow to distinguish omitted fields from explicitly provided None
values, using UpdateSFTPPartnerCmd’s Pydantic presence-aware data such as
model_dump(exclude_unset=True). Apply updates for explicitly supplied optional
fields, including clearing inbound_remote_path and outbound_remote_path, while
preserving existing values for omitted fields.
In `@services/api/src/api/adapters/transaction_repository.py`:
- Around line 148-160: Update the contains-filter handling in the transaction
repository to escape user-supplied % and _ characters before constructing the
surrounding wildcard pattern, and pass the matching escape parameter to every
ilike call in that branch, including sender, receiver, and optional GS fields.
Preserve the existing OR conditions and contains semantics for literal input.
- Around line 28-44: Update TransactionRepository.publish_outbox_event to use a
caller-provided stable idempotency key, adding it to the method’s inputs and
assigning it to Outbox.idempotency_key instead of generating a new UUID on each
call; preserve the event_id generation and existing record persistence flow.
In `@services/api/src/api/core/services/inbound_route_service.py`:
- Around line 19-49: Update create_inbound_route, update_inbound_route, and
delete_inbound_route to access both repository and outbox operations through
self.uow.control_plane, matching the existing UnitOfWork contract and
RouteService pattern; do not use self.uow.inbound_routes or self.uow.outbox.
In `@services/api/src/api/core/services/outbound_route_service.py`:
- Around line 20-173: The get_trading_partner_name method incorrectly passes
tenant_id=0 for SFTP and webhook lookups. Accept the caller’s tenant_id in this
method and pass it to list_sftp_partners and list_webhooks, while preserving the
existing tenant_id=0 behavior for AS2 lookups.
In `@services/api/src/api/core/services/routing_resolver.py`:
- Around line 92-99: Update the AS2 partner lookup in the routing resolver to
constrain AS2Partner.as2_id by the current tenant_id as well, using the tenant
context already available to the resolver. Preserve the existing name resolution
and AS2 return behavior while ensuring duplicate AS2 IDs across tenants cannot
match or raise.
In `@services/api/src/api/core/services/trading_partner_resolver.py`:
- Around line 1-148: Remove the orphaned TradingPartnerResolverService class and
its module because it references nonexistent UnitOfWork attributes and
duplicates RoutingResolutionService; first verify there are no remaining callers
or imports of TradingPartnerResolverService or trading_partner_resolver,
preserving uow.resolve_trading_partner_name as the active resolution path.
In `@services/api/src/api/domain/models.py`:
- Line 146: Update CreateInboundRouteCmd.processing_mode to use the existing
domain.models.ProcessingMode enum instead of str, preserving TRANSFORM as its
default so API validation only accepts the defined processing modes and remains
consistent with InboundRouteDomainModel.processing_mode.
In `@services/api/src/api/ports/webhook_repository.py`:
- Around line 14-21: Update the update_webhook method signature to use the
established UnsetType/UNSET sentinel for optional name, active, and url fields,
matching the other update ports. Preserve None as a valid explicit clear value
while treating UNSET as “leave unchanged,” and update the implementation and
callers to honor this distinction.
In `@services/api/tests/test_as2_receive.py`:
- Around line 35-43: The test_as2_receive test must validate both POST responses
instead of discarding them. Capture each client.post result and assert the
expected status_code, following the established response-assertion pattern used
by test_webhooks.py; add response-body assertions only if that pattern requires
them.
In `@services/workers/compute/src/compute_worker/main.py`:
- Around line 20-50: Replace MockStoragePort and MockRepositoryPort in main with
the existing concrete S3 storage and database repository adapters so processed
payloads are persisted through real outbound ports. Update the SQSComputeWorker
setup to obtain queue_url and endpoint_url from get_settings() or the
established application configuration instead of hardcoded LocalStack values,
while preserving the existing use-case dependency injection flow.
In `@services/workers/compute/tests/test_worker.py`:
- Around line 78-88: Update test_worker_lifecycle to set worker._running to True
before calling stop(), then assert it becomes False afterward, so the test
verifies the stop() state transition rather than only the initial state.
In `@services/workers/orchestrator/src/worker/adapters/db_outbox.py`:
- Around line 59-61: Replace the global_gen.__anext__() call in the finally
block with global_gen.aclose() to close the async generator without resuming its
success path. Preserve the existing suppression behavior as needed for
generator-close exceptions.
- Around line 47-58: Update the exception handler around the async context
manager’s yield to re-raise the caught exception after performing the existing
permanent/transient status and transaction handling. Preserve the FAILED update
and commit for PermanentProvisioningError, and rollback/logging for transient
errors, but ensure both paths propagate the exception to
ProvisioningWorkerService.
In `@services/workers/orchestrator/src/worker/adapters/db_replication.py`:
- Around line 61-416: Refactor _do_replicate to eliminate the seven row-by-row
upsert loops and their per-row tenant_session.execute calls. Add a reusable
bulk-upsert helper using multi-row insert values and stmt.excluded for update
values, then have each entity block build row dictionaries and invoke it once
with the appropriate model and update columns, preserving each query’s
filtering, logging, and field mappings.
- Around line 34-59: Update replicate_tenant_configuration so both session
__anext__() calls and the replication/commit flow are inside the guarded
error-handling block, with rollback performed when a tenant session was
acquired. Preserve the existing provisioning error wrapping for session-open
failures, but classify non-retryable data or constraint exceptions as
PermanentProvisioningError and reserve TransientProvisioningError for genuinely
transient failures instead of wrapping every _do_replicate() or commit error as
transient.
In `@services/workers/orchestrator/src/worker/adapters/db_tenant.py`:
- Around line 15-25: Replace the manual __anext__() and StopAsyncIteration
cleanup in get_all_tenant_ids with an async for loop over
self.db_router.get_global_session(), executing the query and returning from
inside the loop. Apply the same refactor to the method at
services/workers/orchestrator/src/worker/adapters/db_tenant.py lines 26-44, with
no direct changes required elsewhere.
In `@services/workers/orchestrator/src/worker/adapters/sqs_outbox.py`:
- Around line 38-56: The process_next_event method recreates the SQS client and
resolves the queue URL on every poll. Update SqsOutboxAdapter to retain and
reuse a client across polls, and cache the result of get_queue_url for
subsequent process_next_event calls, while preserving the existing error and
no-event behavior.
In `@services/workers/orchestrator/src/worker/core/service.py`:
- Line 42: Move the asyncio import from its inline location into the
module-level import section, alongside logging, and remove the original inline
import.
In `@services/workers/orchestrator/src/worker/data/main.py`:
- Around line 73-117: Update validate_target_url and
HttpxDeliveryAdapter.deliver so the IP address validated during SSRF checks is
also used for the actual HTTP connection, preventing a second DNS resolution
from reaching a rebound internal address. Return or propagate the vetted address
and configure the request transport/connection to pin it while preserving the
original Host header and URL semantics.
---
Outside diff comments:
In `@services/api/src/api/cdc_relay.py`:
- Around line 83-94: Define CDC_DLQ_QUEUE in the MessageQueueName enum used by
queue.send, and provision the corresponding SQS queue in
docker/localstack/init-aws.sh alongside the existing queues. Keep the
schema-validation quarantine flow in the CDC relay unchanged so invalid events
are sent to the newly available DLQ target.
In `@services/api/src/api/ports/transaction_repository.py`:
- Around line 6-70: Add an async publish_outbox_event method to
TransactionRepositoryPort, matching the arguments and return type used by
ApiReceiverService and its data-plane repository implementation, so
DataPlaneRepositoryPort exposes this call through inheritance. Keep the existing
transaction methods unchanged.
In `@services/workers/compute/src/compute_worker/worker.py`:
- Around line 63-106: Update _process_message to distinguish malformed-message
validation failures from transient processing errors: catch the validation
errors raised for missing or invalid Body, trace_id, and s3_uri, log them as
permanent failures, and delete the message using its ReceiptHandle. Keep
transient use-case or queue-operation exceptions on the existing retry path, and
preserve successful processing and deletion behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: b216ed04-ef7e-4133-8b89-47cfe305aba2
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (143)
MakefileTECHNICAL_DEBT.mddocker-compose.ymldocker/localstack/init-aws.shfrontend/web/src/components/ui/code-viewer.tsxfrontend/web/src/components/ui/data-table.tsxfrontend/web/src/features/edi_headers/components/EdiHeadersTable.tsxfrontend/web/src/features/partners/api/partnersApi.tsfrontend/web/src/features/routes/components/CreateInboundRouteModal.tsxfrontend/web/src/features/routes/components/CreateOutboundRouteModal.tsxfrontend/web/src/features/routes/components/InboundRouteForm.tsxfrontend/web/src/features/routes/components/OutboundRouteForm.tsxfrontend/web/src/features/routes/components/RouteDetails.tsxfrontend/web/src/features/routes/components/RoutesTable.tsxfrontend/web/src/features/routes/hooks/useTenantDestinations.tsfrontend/web/src/features/routes/types.tsfrontend/web/src/features/transactions/components/TransactionTimeline.tsxfrontend/web/src/routes/tenant/routes.tsxfrontend/web/src/routes/tenant/webhooks.tsxlibs/config/src/config/settings.pylibs/database/src/database/base_repository.pylibs/database/src/database/connection.pylibs/database/src/database/migrations/global/versions/42c7e50a7b1c_global_initial_schema.pylibs/database/src/database/migrations/tenant/versions/f966e8446341_tenant_initial_schema.pylibs/database/src/database/models/control_plane.pylibs/database/src/database/models/data_plane.pylibs/database/src/database/models/replicated_mixins.pylibs/domain/src/domain/events.pylibs/domain/src/domain/models.pylibs/identity/docker/docker-compose.ymllibs/identity/src/identity/dependencies.pylibs/pipeline/src/pipeline/adapters/transformer.pylibs/pipeline/src/pipeline/core/delivery/__init__.pylibs/pipeline/src/pipeline/core/delivery/as2.pylibs/pipeline/src/pipeline/core/delivery/base.pylibs/pipeline/src/pipeline/core/delivery/router.pylibs/pipeline/src/pipeline/core/delivery/sftp.pylibs/pipeline/src/pipeline/core/delivery/webhook.pylibs/pipeline/src/pipeline/core/saga.pylibs/pipeline/src/pipeline/core/transformation/__init__.pylibs/pipeline/src/pipeline/core/transformation/inbound.pylibs/pipeline/src/pipeline/core/transformation/outbound.pylibs/pipeline/src/pipeline/ports/transformer.pylibs/pipeline/tests/fakes.pylibs/pipeline/tests/test_delivery_service.pylibs/pipeline/tests/test_delivery_service_as2.pylibs/pipeline/tests/test_pipeline_repository.pylibs/pipeline/tests/test_transform_service.pylibs/pipeline/tests/test_transformer.pylibs/transformer/src/transformer/application/ports.pylibs/transformer/src/transformer/application/use_cases.pylibs/transformer/src/transformer/domain/exceptions.pylibs/transformer/src/transformer/infrastructure/adapters/bots_adapter.pylibs/transformer/tests/application/test_transformer_use_cases.pylibs/transformer/tests/infrastructure/adapters/test_bots_adapter.pypyproject.tomlservices/api/src/api/adapters/api_token_repository.pyservices/api/src/api/adapters/as2_partner_repository.pyservices/api/src/api/adapters/as2_partnership_repository.pyservices/api/src/api/adapters/control_plane_aggregator.pyservices/api/src/api/adapters/data_plane_as2_repository.pyservices/api/src/api/adapters/edi_header_repository.pyservices/api/src/api/adapters/http/dtos.pyservices/api/src/api/adapters/inbound_route_repository.pyservices/api/src/api/adapters/outbound_route_repository.pyservices/api/src/api/adapters/outbox_repository.pyservices/api/src/api/adapters/repository.pyservices/api/src/api/adapters/sftp_repository.pyservices/api/src/api/adapters/tenant_repository.pyservices/api/src/api/adapters/transaction_repository.pyservices/api/src/api/adapters/webhook_repository.pyservices/api/src/api/auth/api_key.pyservices/api/src/api/cdc_relay.pyservices/api/src/api/core/authorization.pyservices/api/src/api/core/services/as2_partner_service.pyservices/api/src/api/core/services/as2_partnership_service.pyservices/api/src/api/core/services/inbound_route_service.pyservices/api/src/api/core/services/outbound_route_service.pyservices/api/src/api/core/services/route_service.pyservices/api/src/api/core/services/routing_resolver.pyservices/api/src/api/core/services/sftp_partner_service.pyservices/api/src/api/core/services/trading_partner_resolver.pyservices/api/src/api/core/services/webhook_service.pyservices/api/src/api/core/uow.pyservices/api/src/api/dependencies.pyservices/api/src/api/domain/models.pyservices/api/src/api/ports/api_token_repository.pyservices/api/src/api/ports/as2_partner_repository.pyservices/api/src/api/ports/as2_partnership_repository.pyservices/api/src/api/ports/control_plane_aggregator.pyservices/api/src/api/ports/data_plane_as2_repository.pyservices/api/src/api/ports/edi_header_repository.pyservices/api/src/api/ports/inbound_route_repository.pyservices/api/src/api/ports/outbound_route_repository.pyservices/api/src/api/ports/outbox_repository.pyservices/api/src/api/ports/repository.pyservices/api/src/api/ports/sftp_repository.pyservices/api/src/api/ports/tenant_repository.pyservices/api/src/api/ports/transaction_repository.pyservices/api/src/api/ports/webhook_repository.pyservices/api/src/api/routers/edi_tools.pyservices/api/src/api/routers/trading_partners/as2.pyservices/api/src/api/routers/transactions.pyservices/api/src/api/services/api_receiver_service.pyservices/api/src/api/services/as2_receiver_service.pyservices/api/tests/api_fakes.pyservices/api/tests/routers/test_edi_tools.pyservices/api/tests/test_api_receiver_service.pyservices/api/tests/test_as2_partner_service.pyservices/api/tests/test_as2_receive.pyservices/api/tests/test_as2_receiver_service.pyservices/api/tests/test_cdc_relay.pyservices/api/tests/test_platform_as2.pyservices/api/tests/test_routers_as2.pyservices/api/tests/test_routers_partners.pyservices/api/tests/test_routers_transactions.pyservices/api/tests/test_webhooks.pyservices/workers/compute/pyproject.tomlservices/workers/compute/src/compute_worker/__init__.pyservices/workers/compute/src/compute_worker/main.pyservices/workers/compute/src/compute_worker/worker.pyservices/workers/compute/tests/test_provision_worker.pyservices/workers/compute/tests/test_worker.pyservices/workers/orchestrator/README.mdservices/workers/orchestrator/pyproject.tomlservices/workers/orchestrator/src/worker/__init__.pyservices/workers/orchestrator/src/worker/adapters/__init__.pyservices/workers/orchestrator/src/worker/adapters/db_outbox.pyservices/workers/orchestrator/src/worker/adapters/db_replication.pyservices/workers/orchestrator/src/worker/adapters/db_tenant.pyservices/workers/orchestrator/src/worker/adapters/sqs_outbox.pyservices/workers/orchestrator/src/worker/adapters/vault.pyservices/workers/orchestrator/src/worker/core/__init__.pyservices/workers/orchestrator/src/worker/core/errors.pyservices/workers/orchestrator/src/worker/core/service.pyservices/workers/orchestrator/src/worker/data/main.pyservices/workers/orchestrator/src/worker/main.pyservices/workers/orchestrator/src/worker/ports/__init__.pyservices/workers/orchestrator/src/worker/ports/outbox.pyservices/workers/orchestrator/src/worker/ports/replication.pyservices/workers/orchestrator/src/worker/ports/tenant.pyservices/workers/orchestrator/src/worker/provision/main.pyservices/workers/orchestrator/tests/test_data_main.py
Comment on lines
+13
to
+15
| awslocal sqs create-queue --queue-name TransformOrchestrationQueue-DLQ | ||
| TRANSFORM_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/TransformOrchestrationQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text) | ||
| awslocal sqs create-queue --queue-name TransformOrchestrationQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$TRANSFORM_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}" |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n "EdiTransformerQueue|TransformQueue|TransformOrchestrationQueue" services/workers libs/config services/api --type py --type shRepository: pramodnarayana/soopaedi
Length of output: 401
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== candidate references ==\n'
rg -n "EdiTransformerQueue|TransformQueue|TransformOrchestrationQueue" \
services services/api libs docker -g '!**/node_modules/**' -g '!**/dist/**' -g '!**/build/**'
printf '\n== init script context ==\n'
cat -n docker/localstack/init-aws.sh | sed -n '1,80p'
printf '\n== likely worker/config files ==\n'
for f in \
services/workers/orchestrator/src/worker/main.py \
services/workers/compute/src/compute_worker/main.py \
services/api/tests/test_cdc_relay.py
do
if [ -f "$f" ]; then
echo "\n--- $f ---"
cat -n "$f" | sed -n '1,220p'
fi
doneRepository: pramodnarayana/soopaedi
Length of output: 13158
Update the local worker queue URL libs/transformer/scripts/run_local_worker.py:46 still polls EdiTransformerQueue, so it will miss messages created on TransformOrchestrationQueue.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docker/localstack/init-aws.sh` around lines 13 - 15, Update the queue URL
used by the local worker in run_local_worker.py, specifically the polling
configuration near the worker startup flow, to reference
TransformOrchestrationQueue instead of EdiTransformerQueue. Keep the existing
localhost endpoint and account/region path unchanged.
| sa.Column("transaction_type", sa.String(length=50), nullable=False), | ||
| sa.Column( | ||
| "processing_mode", sa.String(length=50), server_default="TRANSLATE", nullable=False | ||
| "processing_mode", sa.String(length=50), server_default="TRANSFORM", nullable=False |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Do not modify existing migration files.
Altering an existing migration (such as changing the server_default for processing_mode) is an anti-pattern. If this migration has already been applied to any database environments, Alembic will not run it again. This leads to schema inconsistencies between newly initialized databases and existing ones.
Please revert this change and generate a new migration script using op.alter_column to update the server default.
♻️ Proposed revert to the original state
- "processing_mode", sa.String(length=50), server_default="TRANSFORM", nullable=False
+ "processing_mode", sa.String(length=50), server_default="TRANSLATE", nullable=False📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| "processing_mode", sa.String(length=50), server_default="TRANSFORM", nullable=False | |
| "processing_mode", sa.String(length=50), server_default="TRANSLATE", nullable=False |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@libs/database/src/database/migrations/global/versions/42c7e50a7b1c_global_initial_schema.py`
at line 271, Revert the change to the existing migration’s processing_mode
definition, restoring its original server_default. Create a new Alembic
migration that uses op.alter_column to update the processing_mode server
default, including the appropriate upgrade and downgrade behavior.
Comment on lines
+61
to
+416
| async def _do_replicate(self, tenant_id: int, global_session: Any, tenant_session: Any) -> None: | ||
| # --- AS2 Partners --- | ||
| stmt = ( | ||
| select(GlobalAS2Partner) | ||
| .where((GlobalAS2Partner.tenant_id == tenant_id) | (GlobalAS2Partner.tenant_id == 0)) | ||
| .order_by(GlobalAS2Partner.id) | ||
| ) | ||
| tp_result = await global_session.execute(stmt) | ||
| as2_partners = tp_result.scalars().all() | ||
| logger.info(f"[tenant={tenant_id}] Replicating {len(as2_partners)} AS2 partner(s)") | ||
|
|
||
| for global_tp in as2_partners: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting AS2Partner id={global_tp.id} as2_id={global_tp.as2_id!r}" | ||
| ) | ||
| insert_stmt = ( | ||
| insert(TenantAS2Partner) | ||
| .values( | ||
| id=global_tp.id, | ||
| tenant_id=tenant_id, | ||
| name=global_tp.name, | ||
| as2_id=global_tp.as2_id, | ||
| public_cert_pem=global_tp.public_cert_pem, | ||
| public_cert_vault_ref=global_tp.public_cert_vault_ref, | ||
| private_key_vault_ref=global_tp.private_key_vault_ref, | ||
| prev_public_cert_pem=global_tp.prev_public_cert_pem, | ||
| prev_public_cert_vault_ref=global_tp.prev_public_cert_vault_ref, | ||
| prev_private_key_vault_ref=global_tp.prev_private_key_vault_ref, | ||
| url=global_tp.url, | ||
| active=global_tp.active, | ||
| created_at=global_tp.created_at, | ||
| updated_at=global_tp.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "tenant_id": tenant_id, | ||
| "name": global_tp.name, | ||
| "as2_id": global_tp.as2_id, | ||
| "public_cert_pem": global_tp.public_cert_pem, | ||
| "public_cert_vault_ref": global_tp.public_cert_vault_ref, | ||
| "private_key_vault_ref": global_tp.private_key_vault_ref, | ||
| "prev_public_cert_pem": global_tp.prev_public_cert_pem, | ||
| "prev_public_cert_vault_ref": global_tp.prev_public_cert_vault_ref, | ||
| "prev_private_key_vault_ref": global_tp.prev_private_key_vault_ref, | ||
| "url": global_tp.url, | ||
| "active": global_tp.active, | ||
| "created_at": global_tp.created_at, | ||
| "updated_at": global_tp.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_stmt) | ||
|
|
||
| # --- AS2 Partnerships --- | ||
| ps_stmt = ( | ||
| select(GlobalAS2Partnership) | ||
| .where( | ||
| (GlobalAS2Partnership.tenant_id == tenant_id) | ||
| | (GlobalAS2Partnership.tenant_id == 0) | ||
| ) | ||
| .order_by(GlobalAS2Partnership.id) | ||
| ) | ||
| ps_result = await global_session.execute(ps_stmt) | ||
| as2_partnerships = ps_result.scalars().all() | ||
| logger.info(f"[tenant={tenant_id}] Replicating {len(as2_partnerships)} AS2 partnership(s)") | ||
|
|
||
| for global_ps in as2_partnerships: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting AS2Partnership id={global_ps.id} name={global_ps.name!r}" | ||
| ) | ||
| insert_ps_stmt = ( | ||
| insert(TenantAS2Partnership) | ||
| .values( | ||
| id=global_ps.id, | ||
| tenant_id=tenant_id, | ||
| name=global_ps.name, | ||
| local_partner_id=global_ps.local_partner_id, | ||
| remote_partner_id=global_ps.remote_partner_id, | ||
| credentials_vault_ref=global_ps.credentials_vault_ref, | ||
| mdn_type=global_ps.mdn_type, | ||
| mdn_url=global_ps.mdn_url, | ||
| encryption_algorithm=global_ps.encryption_algorithm, | ||
| signature_algorithm=global_ps.signature_algorithm, | ||
| advanced_flags=global_ps.advanced_flags, | ||
| active=global_ps.active, | ||
| created_at=global_ps.created_at, | ||
| updated_at=global_ps.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "name": global_ps.name, | ||
| "local_partner_id": global_ps.local_partner_id, | ||
| "remote_partner_id": global_ps.remote_partner_id, | ||
| "credentials_vault_ref": global_ps.credentials_vault_ref, | ||
| "mdn_type": global_ps.mdn_type, | ||
| "mdn_url": global_ps.mdn_url, | ||
| "encryption_algorithm": global_ps.encryption_algorithm, | ||
| "signature_algorithm": global_ps.signature_algorithm, | ||
| "advanced_flags": global_ps.advanced_flags, | ||
| "active": global_ps.active, | ||
| "created_at": global_ps.created_at, | ||
| "updated_at": global_ps.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_ps_stmt) | ||
|
|
||
| # --- SFTP Partners --- | ||
| sftp_stmt = ( | ||
| select(GlobalSFTPPartner) | ||
| .where(GlobalSFTPPartner.tenant_id == tenant_id) | ||
| .order_by(GlobalSFTPPartner.id) | ||
| ) | ||
| sftp_result = await global_session.execute(sftp_stmt) | ||
| sftp_partners = sftp_result.scalars().all() | ||
| logger.info(f"[tenant={tenant_id}] Replicating {len(sftp_partners)} SFTP partner(s)") | ||
|
|
||
| for global_sftp in sftp_partners: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting SFTPPartner id={global_sftp.id} name={global_sftp.name!r}" | ||
| ) | ||
| insert_sftp_stmt = ( | ||
| insert(TenantSFTPPartner) | ||
| .values( | ||
| id=global_sftp.id, | ||
| tenant_id=tenant_id, | ||
| name=global_sftp.name, | ||
| host=global_sftp.host, | ||
| port=global_sftp.port, | ||
| username=global_sftp.username, | ||
| inbound_remote_path=global_sftp.inbound_remote_path, | ||
| outbound_remote_path=global_sftp.outbound_remote_path, | ||
| host_key=global_sftp.host_key, | ||
| password_encrypted=global_sftp.password_encrypted, | ||
| credentials_vault_ref=global_sftp.credentials_vault_ref, | ||
| active=global_sftp.active, | ||
| created_at=global_sftp.created_at, | ||
| updated_at=global_sftp.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "name": global_sftp.name, | ||
| "host": global_sftp.host, | ||
| "port": global_sftp.port, | ||
| "username": global_sftp.username, | ||
| "inbound_remote_path": global_sftp.inbound_remote_path, | ||
| "outbound_remote_path": global_sftp.outbound_remote_path, | ||
| "host_key": global_sftp.host_key, | ||
| "password_encrypted": global_sftp.password_encrypted, | ||
| "credentials_vault_ref": global_sftp.credentials_vault_ref, | ||
| "active": global_sftp.active, | ||
| "created_at": global_sftp.created_at, | ||
| "updated_at": global_sftp.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_sftp_stmt) | ||
|
|
||
| # --- Webhooks --- | ||
| wh_stmt = ( | ||
| select(GlobalWebhook) | ||
| .where(GlobalWebhook.tenant_id == tenant_id) | ||
| .order_by(GlobalWebhook.id) | ||
| ) | ||
| wh_result = await global_session.execute(wh_stmt) | ||
| webhooks = wh_result.scalars().all() | ||
| logger.info(f"[tenant={tenant_id}] Replicating {len(webhooks)} webhook(s)") | ||
|
|
||
| for global_wh in webhooks: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting Webhook id={global_wh.id} name={global_wh.name!r}" | ||
| ) | ||
| insert_wh_stmt = ( | ||
| insert(TenantWebhook) | ||
| .values( | ||
| id=global_wh.id, | ||
| tenant_id=tenant_id, | ||
| name=global_wh.name, | ||
| url=global_wh.url, | ||
| auth_header_vault_ref=global_wh.auth_header_vault_ref, | ||
| active=global_wh.active, | ||
| created_at=global_wh.created_at, | ||
| updated_at=global_wh.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "name": global_wh.name, | ||
| "url": global_wh.url, | ||
| "auth_header_vault_ref": global_wh.auth_header_vault_ref, | ||
| "active": global_wh.active, | ||
| "created_at": global_wh.created_at, | ||
| "updated_at": global_wh.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_wh_stmt) | ||
|
|
||
| # --- Inbound Routes --- | ||
| ir_stmt = ( | ||
| select(GlobalInboundRoute) | ||
| .where(GlobalInboundRoute.tenant_id == tenant_id) | ||
| .order_by(GlobalInboundRoute.id) | ||
| ) | ||
| ir_result = await global_session.execute(ir_stmt) | ||
| inbound_routes = ir_result.scalars().all() | ||
| logger.info(f"[tenant={tenant_id}] Replicating {len(inbound_routes)} inbound route(s)") | ||
|
|
||
| for global_ir in inbound_routes: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting InboundRoute id={global_ir.id} name={global_ir.name!r}" | ||
| ) | ||
| insert_ir_stmt = ( | ||
| insert(TenantInboundRoute) | ||
| .values( | ||
| id=global_ir.id, | ||
| tenant_id=tenant_id, | ||
| name=global_ir.name, | ||
| trading_partner_id=global_ir.trading_partner_id, | ||
| isa_sender_id=global_ir.isa_sender_id, | ||
| isa_receiver_id=global_ir.isa_receiver_id, | ||
| gs_sender_id=global_ir.gs_sender_id, | ||
| gs_receiver_id=global_ir.gs_receiver_id, | ||
| transaction_type=global_ir.transaction_type, | ||
| processing_mode=global_ir.processing_mode, | ||
| webhook_id=global_ir.webhook_id, | ||
| as2_partner_id=global_ir.as2_partner_id, | ||
| sftp_partner_id=global_ir.sftp_partner_id, | ||
| active=global_ir.active, | ||
| created_at=global_ir.created_at, | ||
| updated_at=global_ir.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "name": global_ir.name, | ||
| "trading_partner_id": global_ir.trading_partner_id, | ||
| "isa_sender_id": global_ir.isa_sender_id, | ||
| "isa_receiver_id": global_ir.isa_receiver_id, | ||
| "gs_sender_id": global_ir.gs_sender_id, | ||
| "gs_receiver_id": global_ir.gs_receiver_id, | ||
| "transaction_type": global_ir.transaction_type, | ||
| "processing_mode": global_ir.processing_mode, | ||
| "webhook_id": global_ir.webhook_id, | ||
| "as2_partner_id": global_ir.as2_partner_id, | ||
| "sftp_partner_id": global_ir.sftp_partner_id, | ||
| "active": global_ir.active, | ||
| "created_at": global_ir.created_at, | ||
| "updated_at": global_ir.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_ir_stmt) | ||
|
|
||
| # --- Outbound Routes --- | ||
| or_stmt = ( | ||
| select(GlobalOutboundRoute) | ||
| .where(GlobalOutboundRoute.tenant_id == tenant_id) | ||
| .order_by(GlobalOutboundRoute.id) | ||
| ) | ||
| or_result = await global_session.execute(or_stmt) | ||
| outbound_routes = or_result.scalars().all() | ||
| logger.info(f"[tenant={tenant_id}] Replicating {len(outbound_routes)} outbound route(s)") | ||
|
|
||
| for global_or in outbound_routes: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting OutboundRoute id={global_or.id} name={global_or.name!r}" | ||
| ) | ||
| insert_or_stmt = ( | ||
| insert(TenantOutboundRoute) | ||
| .values( | ||
| id=global_or.id, | ||
| tenant_id=tenant_id, | ||
| trading_partner_id=global_or.trading_partner_id, | ||
| name=global_or.name, | ||
| protocol=global_or.protocol, | ||
| as2_partner_id=global_or.as2_partner_id, | ||
| sftp_partner_id=global_or.sftp_partner_id, | ||
| active=global_or.active, | ||
| created_at=global_or.created_at, | ||
| updated_at=global_or.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "trading_partner_id": global_or.trading_partner_id, | ||
| "name": global_or.name, | ||
| "protocol": global_or.protocol, | ||
| "as2_partner_id": global_or.as2_partner_id, | ||
| "sftp_partner_id": global_or.sftp_partner_id, | ||
| "active": global_or.active, | ||
| "created_at": global_or.created_at, | ||
| "updated_at": global_or.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_or_stmt) | ||
|
|
||
| # --- Outbound EDI Headers --- | ||
| oeh_stmt = ( | ||
| select(GlobalOutboundEdiHeader) | ||
| .where(GlobalOutboundEdiHeader.tenant_id == tenant_id) | ||
| .order_by(GlobalOutboundEdiHeader.id) | ||
| ) | ||
| oeh_result = await global_session.execute(oeh_stmt) | ||
| outbound_edi_headers = oeh_result.scalars().all() | ||
| logger.info( | ||
| f"[tenant={tenant_id}] Replicating {len(outbound_edi_headers)} outbound EDI header(s)" | ||
| ) | ||
|
|
||
| for global_oeh in outbound_edi_headers: | ||
| logger.debug( | ||
| f"[tenant={tenant_id}] Upserting OutboundEdiHeader id={global_oeh.id} name={global_oeh.name!r}" | ||
| ) | ||
| insert_oeh_stmt = ( | ||
| insert(TenantOutboundEdiHeader) | ||
| .values( | ||
| id=global_oeh.id, | ||
| tenant_id=tenant_id, | ||
| trading_partner_id=global_oeh.trading_partner_id, | ||
| name=global_oeh.name, | ||
| isa_sender_id=global_oeh.isa_sender_id, | ||
| isa_sender_qualifier=global_oeh.isa_sender_qualifier, | ||
| isa_receiver_id=global_oeh.isa_receiver_id, | ||
| isa_receiver_qualifier=global_oeh.isa_receiver_qualifier, | ||
| gs_sender_id=global_oeh.gs_sender_id, | ||
| gs_receiver_id=global_oeh.gs_receiver_id, | ||
| default_standard=global_oeh.default_standard, | ||
| default_version=global_oeh.default_version, | ||
| transaction_type=global_oeh.transaction_type, | ||
| created_at=global_oeh.created_at, | ||
| updated_at=global_oeh.updated_at, | ||
| ) | ||
| .on_conflict_do_update( | ||
| index_elements=["id"], | ||
| set_={ | ||
| "trading_partner_id": global_oeh.trading_partner_id, | ||
| "name": global_oeh.name, | ||
| "isa_sender_id": global_oeh.isa_sender_id, | ||
| "isa_sender_qualifier": global_oeh.isa_sender_qualifier, | ||
| "isa_receiver_id": global_oeh.isa_receiver_id, | ||
| "isa_receiver_qualifier": global_oeh.isa_receiver_qualifier, | ||
| "gs_sender_id": global_oeh.gs_sender_id, | ||
| "gs_receiver_id": global_oeh.gs_receiver_id, | ||
| "default_standard": global_oeh.default_standard, | ||
| "default_version": global_oeh.default_version, | ||
| "transaction_type": global_oeh.transaction_type, | ||
| "created_at": global_oeh.created_at, | ||
| "updated_at": global_oeh.updated_at, | ||
| }, | ||
| ) | ||
| ) | ||
| await tenant_session.execute(insert_oeh_stmt) |
Contributor
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Seven near-identical select+per-row-upsert blocks; each issues N sequential round trips per entity type.
_do_replicate repeats the same select → loop → insert(...).on_conflict_do_update(...) shape for AS2Partner, AS2Partnership, SFTPPartner, Webhook, InboundRoute, OutboundRoute, and OutboundEdiHeader — each looping row-by-row and awaiting one tenant_session.execute() per row. For tenants with many partners/routes this is an N+1 pattern (N round trips per entity type instead of 1), and the duplication makes it easy for the blocks to silently drift (e.g. inconsistent set_ field lists between blocks).
SQLAlchemy 2.0 Core supports a single bulk upsert per entity type via multi-row insert(...).values([...]) with on_conflict_do_update referencing insert_stmt.excluded.<col> instead of closure-captured row variables — this collapses both the duplication and the N+1 round trips into one generic helper.
♻️ Sketch of a generic bulk-upsert helper
async def _bulk_upsert(
self, session: Any, model: Any, rows: list[dict[str, Any]], update_cols: list[str]
) -> None:
if not rows:
return
stmt = insert(model).values(rows)
stmt = stmt.on_conflict_do_update(
index_elements=["id"],
set_={col: getattr(stmt.excluded, col) for col in update_cols},
)
await session.execute(stmt)Each of the 7 blocks then reduces to building rows from the global query result and calling _bulk_upsert(tenant_session, TenantAS2Partner, rows, [...]) once.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/workers/orchestrator/src/worker/adapters/db_replication.py` around
lines 61 - 416, Refactor _do_replicate to eliminate the seven row-by-row upsert
loops and their per-row tenant_session.execute calls. Add a reusable bulk-upsert
helper using multi-row insert values and stmt.excluded for update values, then
have each entity block build row dictionaries and invoke it once with the
appropriate model and update columns, preserving each query’s filtering,
logging, and field mappings.
Comment on lines
+15
to
+25
| async def get_all_tenant_ids(self) -> list[int]: | ||
| global_gen = self.db_router.get_global_session() | ||
| global_session = await global_gen.__anext__() | ||
| try: | ||
| stmt = select(Tenant.id).join(DatabaseShard) | ||
| result = await global_session.execute(stmt) | ||
| return list(result.scalars().all()) | ||
| finally: | ||
| with contextlib.suppress(StopAsyncIteration): | ||
| await global_gen.__anext__() | ||
|
|
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Prefer async for over manual __anext__() to iterate async generators. Both methods manually drive the get_global_session() async generator using __anext__() and StopAsyncIteration suppression. This is less idiomatic and more error-prone than a simple async for loop, which automatically closes the generator cleanly when exiting the loop.
services/workers/orchestrator/src/worker/adapters/db_tenant.py#L15-L25: refactor toasync for global_session in self.db_router.get_global_session():and return inside the loop.services/workers/orchestrator/src/worker/adapters/db_tenant.py#L26-L44: refactor toasync for global_session in self.db_router.get_global_session():and return inside the loop.
♻️ Proposed refactor for both methods
async def get_all_tenant_ids(self) -> list[int]:
- global_gen = self.db_router.get_global_session()
- global_session = await global_gen.__anext__()
- try:
- stmt = select(Tenant.id).join(DatabaseShard)
- result = await global_session.execute(stmt)
- return list(result.scalars().all())
- finally:
- with contextlib.suppress(StopAsyncIteration):
- await global_gen.__anext__()
+ async for global_session in self.db_router.get_global_session():
+ stmt = select(Tenant.id).join(DatabaseShard)
+ result = await global_session.execute(stmt)
+ return list(result.scalars().all())
+ return []
async def resolve_shard(self, tenant_id: int) -> tuple[str, str]:
if tenant_id in self._cache:
return self._cache[tenant_id]
- global_gen = self.db_router.get_global_session()
- global_session = await global_gen.__anext__()
- try:
- stmt = select(Tenant, DatabaseShard).join(DatabaseShard).where(Tenant.id == tenant_id)
- result = await global_session.execute(stmt)
- row = result.first()
- if not row:
- raise ValueError(f"Tenant {tenant_id} not found in Global DB")
- _, shard_obj = row
- self._cache[tenant_id] = (str(shard_obj.name), str(shard_obj.dsn))
- return self._cache[tenant_id]
- finally:
- with contextlib.suppress(StopAsyncIteration):
- await global_gen.__anext__()
+ async for global_session in self.db_router.get_global_session():
+ stmt = select(Tenant, DatabaseShard).join(DatabaseShard).where(Tenant.id == tenant_id)
+ result = await global_session.execute(stmt)
+ row = result.first()
+ if not row:
+ raise ValueError(f"Tenant {tenant_id} not found in Global DB")
+ _, shard_obj = row
+ self._cache[tenant_id] = (str(shard_obj.name), str(shard_obj.dsn))
+ return self._cache[tenant_id]
+
+ raise ValueError(f"Tenant {tenant_id} not found in Global DB")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| async def get_all_tenant_ids(self) -> list[int]: | |
| global_gen = self.db_router.get_global_session() | |
| global_session = await global_gen.__anext__() | |
| try: | |
| stmt = select(Tenant.id).join(DatabaseShard) | |
| result = await global_session.execute(stmt) | |
| return list(result.scalars().all()) | |
| finally: | |
| with contextlib.suppress(StopAsyncIteration): | |
| await global_gen.__anext__() | |
| async def get_all_tenant_ids(self) -> list[int]: | |
| async for global_session in self.db_router.get_global_session(): | |
| stmt = select(Tenant.id).join(DatabaseShard) | |
| result = await global_session.execute(stmt) | |
| return list(result.scalars().all()) | |
| return [] | |
| async def resolve_shard(self, tenant_id: int) -> tuple[str, str]: | |
| if tenant_id in self._cache: | |
| return self._cache[tenant_id] | |
| async for global_session in self.db_router.get_global_session(): | |
| stmt = select(Tenant, DatabaseShard).join(DatabaseShard).where(Tenant.id == tenant_id) | |
| result = await global_session.execute(stmt) | |
| row = result.first() | |
| if not row: | |
| raise ValueError(f"Tenant {tenant_id} not found in Global DB") | |
| _, shard_obj = row | |
| self._cache[tenant_id] = (str(shard_obj.name), str(shard_obj.dsn)) | |
| return self._cache[tenant_id] | |
| raise ValueError(f"Tenant {tenant_id} not found in Global DB") |
📍 Affects 1 file
services/workers/orchestrator/src/worker/adapters/db_tenant.py#L15-L25(this comment)services/workers/orchestrator/src/worker/adapters/db_tenant.py#L26-L44
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/workers/orchestrator/src/worker/adapters/db_tenant.py` around lines
15 - 25, Replace the manual __anext__() and StopAsyncIteration cleanup in
get_all_tenant_ids with an async for loop over
self.db_router.get_global_session(), executing the query and returning from
inside the loop. Apply the same refactor to the method at
services/workers/orchestrator/src/worker/adapters/db_tenant.py lines 26-44, with
no direct changes required elsewhere.
Comment on lines
+38
to
+56
| class SqsOutboxAdapter(OutboxPort): | ||
| def __init__(self, queue_name: str = MessageQueueName.PROVISIONING_QUEUE): | ||
| self.queue_name = queue_name | ||
| self.endpoint_url = os.environ.get("AWS_ENDPOINT_URL", "http://localhost:4566") | ||
| self.region = "us-east-1" | ||
| self.session = aioboto3.Session() | ||
|
|
||
| @asynccontextmanager | ||
| async def process_next_event(self) -> AsyncIterator[OutboxEvent | None]: | ||
| async with self.session.client( | ||
| "sqs", endpoint_url=self.endpoint_url, region_name=self.region | ||
| ) as sqs: | ||
| try: | ||
| queue_url_response = await sqs.get_queue_url(QueueName=self.queue_name) | ||
| queue_url = queue_url_response["QueueUrl"] | ||
| except Exception as e: | ||
| logger.error(f"Failed to get queue URL for {self.queue_name}: {e}") | ||
| yield None | ||
| return |
Contributor
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
SQS client and queue URL are recreated on every poll cycle.
process_next_event opens a fresh SQS client and calls get_queue_url every time. Reuse the client and cache the queue URL across polls to avoid repeated connection setup and extra API calls.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/workers/orchestrator/src/worker/adapters/sqs_outbox.py` around lines
38 - 56, The process_next_event method recreates the SQS client and resolves the
queue URL on every poll. Update SqsOutboxAdapter to retain and reuse a client
across polls, and cache the result of get_queue_url for subsequent
process_next_event calls, while preserving the existing error and no-event
behavior.
| # the number of simultaneous DB connections to avoid overwhelming the | ||
| # connection pool. Combined with deterministic ORDER BY id in the | ||
| # replication adapter, this eliminates deadlocks while remaining scalable. | ||
| import asyncio |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Move standard library import to the top of the file.
Inline imports are typically reserved for avoiding circular dependencies. Since asyncio is a standard library module, it should be imported at the module level.
♻️ Proposed refactor
- import asyncio
-Then add import asyncio at the top of the file alongside import logging.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/workers/orchestrator/src/worker/core/service.py` at line 42, Move
the asyncio import from its inline location into the module-level import
section, alongside logging, and remove the original inline import.
Comment on lines
+73
to
+117
| def validate_target_url(url: str) -> bool: | ||
| """ | ||
| Validate target URL to prevent SSRF attacks. | ||
| Returns True if URL is safe, False otherwise. | ||
| """ | ||
| try: | ||
| parsed = urlparse(url) | ||
|
|
||
| # Only allow http and https schemes | ||
| if parsed.scheme not in ("http", "https"): | ||
| logger.warning(f"SSRF check failed: invalid scheme {parsed.scheme}") | ||
| return False | ||
|
|
||
| # Reject URLs without a hostname | ||
| if not parsed.hostname: | ||
| logger.warning("SSRF check failed: missing hostname") | ||
| return False | ||
|
|
||
| # Resolve all A/AAAA records for the hostname | ||
| import socket | ||
|
|
||
| try: | ||
| # getaddrinfo returns a list of 5-tuples: (family, type, proto, canonname, sockaddr) | ||
| addr_info = socket.getaddrinfo(parsed.hostname, None) | ||
| except socket.gaierror: | ||
| logger.warning(f"SSRF check failed: could not resolve hostname {parsed.hostname}") | ||
| return False | ||
|
|
||
| for addr in addr_info: | ||
| ip_str = addr[4][0] | ||
| ip = ipaddress.ip_address(ip_str) | ||
| if ( | ||
| ip.is_private | ||
| or ip.is_loopback | ||
| or ip.is_link_local | ||
| or ip.is_reserved | ||
| or ip.is_multicast | ||
| ): | ||
| logger.warning(f"SSRF check failed: resolved to private/internal IP {ip}") | ||
| return False | ||
|
|
||
| return True | ||
| except Exception as e: | ||
| logger.error(f"SSRF validation error: {e}") | ||
| return False |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🔴 Critical | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
ast-grep outline libs/pipeline/src/pipeline/adapters/http.py --items all
rg -n -A15 'class HttpxDeliveryAdapter' libs/pipeline/src/pipeline/adapters/http.pyRepository: pramodnarayana/soopaedi
Length of output: 1045
🏁 Script executed:
#!/bin/bash
sed -n '1,120p' libs/pipeline/src/pipeline/adapters/http.py | cat -nRepository: pramodnarayana/soopaedi
Length of output: 1364
🏁 Script executed:
#!/bin/bash
sed -n '1,140p' libs/pipeline/src/pipeline/adapters/http.py | cat -nRepository: pramodnarayana/soopaedi
Length of output: 1364
Potential DNS-rebinding SSRF gap
validate_target_url() only checks the hostname up front; HttpxDeliveryAdapter.deliver() then posts to the original URL with httpx.AsyncClient, which resolves DNS again at connect time. Pin the vetted IP/transport for the request, or the URL can be rebound to an internal address between validation and connect.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/workers/orchestrator/src/worker/data/main.py` around lines 73 - 117,
Update validate_target_url and HttpxDeliveryAdapter.deliver so the IP address
validated during SSRF checks is also used for the actual HTTP connection,
preventing a second DNS resolution from reaching a rebound internal address.
Return or propagate the vetted address and configure the request
transport/connection to pin it while preserving the original Host header and URL
semantics.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 13
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
services/api/src/api/routers/webhooks/webhook.py (1)
57-72: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAvoid nested UnitOfWork context managers.
The second
async with uow:block is currently nested inside the first one. This can lead to nested session risks, unexpected transaction boundaries, or connection leaks. Exit the first context manager before opening the new session, similar to how it was fixed inservices/api/src/api/routers/webhooks/__init__.py.🐛 Proposed fix to un-nest the session context
- async with uow: - if not uow.control_plane: - raise HTTPException(status_code=500, detail="Control plane not initialized") - partner = await uow.control_plane.get_webhook(tenant_id, _.partner_id) - if not partner or partner.tenant_id != tenant_id: - raise HTTPException(status_code=404, detail="Webhook not found after creation") - - return PartnerResponse( - partner_id=partner.id, - tenant_id=tenant_id, - name=partner.name, - type="WEBHOOK", - status="ACTIVE" if partner.active else "INACTIVE", - active=partner.active, - url=partner.url, - ) + async with uow: + if not uow.control_plane: + raise HTTPException(status_code=500, detail="Control plane not initialized") + partner = await uow.control_plane.get_webhook(tenant_id, _.partner_id) + if not partner or partner.tenant_id != tenant_id: + raise HTTPException(status_code=404, detail="Webhook not found after creation") + + return PartnerResponse( + partner_id=partner.id, + tenant_id=tenant_id, + name=partner.name, + type="WEBHOOK", + status="ACTIVE" if partner.active else "INACTIVE", + active=partner.active, + url=partner.url, + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/api/src/api/routers/webhooks/webhook.py` around lines 57 - 72, Un-nest the UnitOfWork context in the webhook creation flow: complete and exit the existing async with uow block before opening the separate block that calls uow.control_plane.get_webhook. Preserve the control-plane validation, webhook lookup, not-found handling, and PartnerResponse construction while ensuring no async with uow contexts overlap.
♻️ Duplicate comments (1)
services/api/src/api/adapters/inbound_route_repository.py (1)
150-156: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winNon-deterministic row selection when
transaction_typeis omitted.This exact issue was flagged in a previous review but has not been fully resolved. The
order_byclause is still nested inside theif transaction_type:block. Whentransaction_typeis omitted, the query has no deterministic ordering or fallback filtering, meaning.scalars().first()returns an unpredictable row instead of explicitly prioritizing theNULLfallback route.🛠️ Proposed fix
Add an explicit
elsebranch to guarantee that omitted transaction types consistently resolve to the genericNULLfallback.if transaction_type: stmt = stmt.where( or_( InboundRoute.transaction_type == transaction_type, InboundRoute.transaction_type.is_(None), ) ).order_by(InboundRoute.transaction_type.desc().nullslast()) + else: + stmt = stmt.where(InboundRoute.transaction_type.is_(None))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/api/src/api/adapters/inbound_route_repository.py` around lines 150 - 156, Update the transaction-type query logic near the existing transaction_type condition to add an explicit else branch that filters for InboundRoute.transaction_type.is_(None) and orders the results deterministically for the NULL fallback. Preserve the current matching-or-NULL filtering and ordering when transaction_type is provided.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.agents/AGENTS.md:
- Around line 8-23: Add blank lines immediately after the Markdown headings in
AGENTS.md, including “Enterprise Coding Standards (Strictly Enforced),” “Package
Manager,” and “Destructive Commands,” while preserving all existing list
content.
In `@libs/database/tests/test_connection_unit.py`:
- Around line 96-98: Make the tenant ID assertion after the params extraction
unconditional: remove the isinstance(params, dict) guard and directly assert
params contains tenant_id equal to "123", so non-dictionary values fail the test
rather than being skipped.
In `@services/api/src/api/adapters/repository.py`:
- Around line 43-46: Restore the Data Plane repository capabilities by adding
the Outbox, EdiJson, and EdiMessage SQLAlchemy mixins to
SqlAlchemyDataPlaneRepository and initializing each with the session; extend
DataPlaneRepositoryPort with the corresponding port interfaces. Update
services/api/src/api/adapters/repository.py lines 43-46 and
services/api/src/api/ports/repository.py lines 45-46;
services/api/src/api/services/api_receiver_service.py lines 79-102 requires no
direct change because it is corrected by restoring these repository
capabilities.
In `@services/api/src/api/adapters/sftp_repository.py`:
- Around line 77-86: The partial update filtering in the command update flow
must not use dataclasses.asdict(), because its deep copy breaks identity checks
against UNSET. Iterate over the command’s fields or otherwise access their
original values, filter omitted fields by comparing to UNSET, then preserve the
existing password encryption and setattr behavior.
In `@services/api/src/api/auth/api_key.py`:
- Around line 71-76: Update get_tenant_id_from_api_key to type global_session as
GlobalSession, add the GlobalSession import to the module-level imports, and
remove the inline GlobalSession import and cast around
SqlAlchemyApiTokenRepository. Preserve the existing repository call and tenant
lookup behavior.
In `@services/api/src/api/core/services/as2_partner_service.py`:
- Around line 30-35: Remove the entity-ID idempotency_key from every outbox
publish in AS2PartnerService and OutboundRouteService so each event receives a
unique generated key. Apply this to the listed CREATE, UPDATE, DELETE, and
rotate events in services/api/src/api/core/services/as2_partner_service.py at
lines 30-35, 55-60, 73-78, and 96-101, and in
services/api/src/api/core/services/outbound_route_service.py at lines 27-32,
40-45, and 51-56.
In `@services/api/src/api/core/services/outbound_route_service.py`:
- Around line 90-93: Update get_trading_partner_name to use get_as2_partner(...)
for each required partner instead of the batch get_as2_partners_by_ids lookup.
Ensure the helper receives the relevant tenant context and falls back to the
global tenant so global-only partners are found, while preserving the existing
partner-name resolution behavior.
In `@services/api/src/api/core/services/route_service.py`:
- Around line 30-35: Remove the idempotency_key=route_id argument from every
publish_outbox_event call in route_service.py at lines 30-35, 43-48, 54-59,
69-74, 82-87, and 93-98, and in inbound_route_service.py at lines 22-27, 35-40,
and 46-51. Leave the existing event payloads and lifecycle behavior unchanged so
the repository assigns unique keys automatically.
In `@services/api/src/api/core/services/sftp_partner_service.py`:
- Around line 46-55: The update flow around update_sftp_partner must validate
the merged existing and requested credential state before persisting changes:
after loading the current SFTP partner, ensure the resulting password or
credentials_vault_ref remains configured, and reject updates where both are
absent. Preserve the existing update, event publication, and retrieval behavior
for valid requests.
- Around line 27-31: The SFTP partner outbox events currently reuse partner_id
as the idempotency key across operations. In sftp_partner_service.py lines
27-31, update the SFTP_PARTNER_CREATED publish call to use a
creation-operation-specific key; in lines 49-54, provide a distinct stable key
for each update operation while preserving deduplication for repeated attempts.
In `@services/api/src/api/core/services/webhook_service.py`:
- Around line 33-38: Remove the manually supplied idempotency_key from all three
publish_outbox_event calls in webhook_service.py: the create flow at lines
33-38, update flow at lines 62-67, and delete flow at lines 76-81. Let the
repository generate a unique UUID for each lifecycle event while preserving the
existing event types and payloads.
In `@services/api/tests/test_provisioning_core.py`:
- Around line 28-34: Add the `@pytest.fixture` decorator to mock_uow so pytest
registers it as a fixture for dependent fixtures such as as2_partner_service.
Preserve its existing MagicMock setup and return behavior.
In `@TECHNICAL_DEBT.md`:
- Around line 80-82: Add a blank line immediately after the “UnitOfWork
Architecture (Control Plane vs Data Plane Naming)” heading in TECHNICAL_DEBT.md,
before the following paragraph, without changing the content.
---
Outside diff comments:
In `@services/api/src/api/routers/webhooks/webhook.py`:
- Around line 57-72: Un-nest the UnitOfWork context in the webhook creation
flow: complete and exit the existing async with uow block before opening the
separate block that calls uow.control_plane.get_webhook. Preserve the
control-plane validation, webhook lookup, not-found handling, and
PartnerResponse construction while ensuring no async with uow contexts overlap.
---
Duplicate comments:
In `@services/api/src/api/adapters/inbound_route_repository.py`:
- Around line 150-156: Update the transaction-type query logic near the existing
transaction_type condition to add an explicit else branch that filters for
InboundRoute.transaction_type.is_(None) and orders the results deterministically
for the NULL fallback. Preserve the current matching-or-NULL filtering and
ordering when transaction_type is provided.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 7bfcdc2c-eedb-4fec-bcf1-2e0f4499666b
📒 Files selected for processing (68)
.agents/AGENTS.md.agents/skills/architect/SKILL.md.agents/skills/cloud-architect/SKILL.md.agents/skills/devops-engineer/SKILL.md.agents/skills/generalist-programmer/SKILL.md.agents/skills/reviews/SKILL.md.agents/skills/testing/SKILL.mdMakefileTECHNICAL_DEBT.mddocker/localstack/init-aws.shfrontend/web/src/features/routes/components/CreateInboundRouteModal.tsxfrontend/web/src/features/routes/components/CreateRouteModal.tsxfrontend/web/src/features/routes/components/InboundRouteForm.tsxlibs/database/src/database/base_repository.pylibs/database/src/database/connection.pylibs/database/tests/test_connection_unit.pylibs/domain/src/domain/events.pylibs/pipeline/src/pipeline/core/deliver.pylibs/pipeline/src/pipeline/core/delivery/as2.pylibs/pipeline/src/pipeline/core/delivery/sftp.pylibs/pipeline/src/pipeline/core/delivery/webhook.pylibs/pipeline/src/pipeline/core/transformation/inbound.pylibs/pipeline/src/pipeline/core/translate.pylibs/transformer/src/transformer/infrastructure/adapters/bots_adapter.pypyproject.tomlservices/api/src/api/adapters/api_token_repository.pyservices/api/src/api/adapters/edi_header_repository.pyservices/api/src/api/adapters/http/dtos.pyservices/api/src/api/adapters/inbound_route_repository.pyservices/api/src/api/adapters/outbound_route_repository.pyservices/api/src/api/adapters/outbox_repository.pyservices/api/src/api/adapters/repository.pyservices/api/src/api/adapters/sftp_repository.pyservices/api/src/api/adapters/transaction_repository.pyservices/api/src/api/auth/api_key.pyservices/api/src/api/cdc_relay.pyservices/api/src/api/core/services/as2_partner_service.pyservices/api/src/api/core/services/as2_partnership_service.pyservices/api/src/api/core/services/edi_header_service.pyservices/api/src/api/core/services/inbound_route_service.pyservices/api/src/api/core/services/outbound_route_service.pyservices/api/src/api/core/services/route_service.pyservices/api/src/api/core/services/sftp_partner_service.pyservices/api/src/api/core/services/webhook_service.pyservices/api/src/api/core/uow.pyservices/api/src/api/dependencies.pyservices/api/src/api/domain/models.pyservices/api/src/api/ports/outbox_repository.pyservices/api/src/api/ports/repository.pyservices/api/src/api/ports/transaction_repository.pyservices/api/src/api/routers/edi_headers.pyservices/api/src/api/routers/routes.pyservices/api/src/api/routers/trading_partners/as2.pyservices/api/src/api/routers/trading_partners/platform/as2_partners.pyservices/api/src/api/routers/trading_partners/platform/as2_partnerships.pyservices/api/src/api/routers/trading_partners/sftp.pyservices/api/src/api/routers/transactions.pyservices/api/src/api/routers/webhooks/__init__.pyservices/api/src/api/routers/webhooks/webhook.pyservices/api/src/api/services/api_receiver_service.pyservices/api/src/api/services/as2_receiver_service.pyservices/api/tests/api_fakes.pyservices/api/tests/test_api_receiver_service.pyservices/api/tests/test_api_repository.pyservices/api/tests/test_as2_partner_service.pyservices/api/tests/test_as2_receiver_service.pyservices/api/tests/test_provisioning_core.pyservices/worker/tests/test_data_worker.py
💤 Files with no reviewable changes (6)
- frontend/web/src/features/routes/components/CreateRouteModal.tsx
- services/worker/tests/test_data_worker.py
- services/api/tests/test_api_repository.py
- libs/pipeline/src/pipeline/core/deliver.py
- libs/pipeline/src/pipeline/core/translate.py
- services/api/src/api/adapters/http/dtos.py
Comment on lines
+8
to
+23
| # Enterprise Coding Standards (Strictly Enforced) | ||
| - **Hexagonal Architecture**: Keep the domain isolated. Ports and Adapters must strictly separate business logic from external frameworks, APIs, and databases. | ||
| - **SOLID Principles**: Adhere to Single Responsibility, Open/Closed, Liskov Substitution, Interface Segregation, and Dependency Inversion. | ||
| - **Red-Green-Refactor Cycle**: Write failing tests first, make them pass, then refactor to clean up. | ||
| - **Zero Mocks for Pure Logic**: Do not mock pure business logic. Domain models and core logic must be self-contained and testable without external mocks. | ||
| - **Narrow Integration Tests**: Stop writing "forced" unit tests with excessive mocking just to hit coverage limits. Focus on writing Narrow Integration Tests that actually connect to databases/external systems via test harnesses to test real behavior. | ||
| - **No Static Mutable Singletons**: Avoid global state. Use dependency injection to pass dependencies dynamically. | ||
| - **Infra & Business Decoupling**: Infrastructure code (AWS, SQS, DB connections) must never leak into business/domain logic. | ||
| - **No Leakage**: Data transfer objects (DTOs), API models, and ORM models must not leak across their respective boundaries. Map them appropriately. | ||
| - **DRY (Don't Repeat Yourself)**: Avoid code duplication. Extract shared logic into reusable, well-named functions/modules. | ||
|
|
||
| # Package Manager | ||
| - ALWAYS use `pnpm` for frontend/Node.js package management instead of `npm`. Do not use `npm install`. | ||
|
|
||
| # Destructive Commands | ||
| - NEVER use destructive terminal commands like `git checkout`, `git restore`, `git reset`, `git clean`, or `rm -rf` without explicitly asking for and receiving the user's permission first. Always prefer precise code-editing tools for reverting changes. |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Add blank lines below headings for Markdown consistency.
To comply with standard Markdown formatting (MD022), ensure that headings are followed by a blank line.
♻️ Proposed formatting fix
# Enterprise Coding Standards (Strictly Enforced)
+
- **Hexagonal Architecture**: Keep the domain isolated. Ports and Adapters must strictly separate business logic from external frameworks, APIs, and databases.
- **SOLID Principles**: Adhere to Single Responsibility, Open/Closed, Liskov Substitution, Interface Segregation, and Dependency Inversion.
- **Red-Green-Refactor Cycle**: Write failing tests first, make them pass, then refactor to clean up.
- **Zero Mocks for Pure Logic**: Do not mock pure business logic. Domain models and core logic must be self-contained and testable without external mocks.
- **Narrow Integration Tests**: Stop writing "forced" unit tests with excessive mocking just to hit coverage limits. Focus on writing Narrow Integration Tests that actually connect to databases/external systems via test harnesses to test real behavior.
- **No Static Mutable Singletons**: Avoid global state. Use dependency injection to pass dependencies dynamically.
- **Infra & Business Decoupling**: Infrastructure code (AWS, SQS, DB connections) must never leak into business/domain logic.
- **No Leakage**: Data transfer objects (DTOs), API models, and ORM models must not leak across their respective boundaries. Map them appropriately.
- **DRY (Don't Repeat Yourself)**: Avoid code duplication. Extract shared logic into reusable, well-named functions/modules.
# Package Manager
+
- ALWAYS use `pnpm` for frontend/Node.js package management instead of `npm`. Do not use `npm install`.
# Destructive Commands
+
- NEVER use destructive terminal commands like `git checkout`, `git restore`, `git reset`, `git clean`, or `rm -rf` without explicitly asking for and receiving the user's permission first. Always prefer precise code-editing tools for reverting changes.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| # Enterprise Coding Standards (Strictly Enforced) | |
| - **Hexagonal Architecture**: Keep the domain isolated. Ports and Adapters must strictly separate business logic from external frameworks, APIs, and databases. | |
| - **SOLID Principles**: Adhere to Single Responsibility, Open/Closed, Liskov Substitution, Interface Segregation, and Dependency Inversion. | |
| - **Red-Green-Refactor Cycle**: Write failing tests first, make them pass, then refactor to clean up. | |
| - **Zero Mocks for Pure Logic**: Do not mock pure business logic. Domain models and core logic must be self-contained and testable without external mocks. | |
| - **Narrow Integration Tests**: Stop writing "forced" unit tests with excessive mocking just to hit coverage limits. Focus on writing Narrow Integration Tests that actually connect to databases/external systems via test harnesses to test real behavior. | |
| - **No Static Mutable Singletons**: Avoid global state. Use dependency injection to pass dependencies dynamically. | |
| - **Infra & Business Decoupling**: Infrastructure code (AWS, SQS, DB connections) must never leak into business/domain logic. | |
| - **No Leakage**: Data transfer objects (DTOs), API models, and ORM models must not leak across their respective boundaries. Map them appropriately. | |
| - **DRY (Don't Repeat Yourself)**: Avoid code duplication. Extract shared logic into reusable, well-named functions/modules. | |
| # Package Manager | |
| - ALWAYS use `pnpm` for frontend/Node.js package management instead of `npm`. Do not use `npm install`. | |
| # Destructive Commands | |
| - NEVER use destructive terminal commands like `git checkout`, `git restore`, `git reset`, `git clean`, or `rm -rf` without explicitly asking for and receiving the user's permission first. Always prefer precise code-editing tools for reverting changes. | |
| # Enterprise Coding Standards (Strictly Enforced) | |
| - **Hexagonal Architecture**: Keep the domain isolated. Ports and Adapters must strictly separate business logic from external frameworks, APIs, and databases. | |
| - **SOLID Principles**: Adhere to Single Responsibility, Open/Closed, Liskov Substitution, Interface Segregation, and Dependency Inversion. | |
| - **Red-Green-Refactor Cycle**: Write failing tests first, make them pass, then refactor to clean up. | |
| - **Zero Mocks for Pure Logic**: Do not mock pure business logic. Domain models and core logic must be self-contained and testable without external mocks. | |
| - **Narrow Integration Tests**: Stop writing "forced" unit tests with excessive mocking just to hit coverage limits. Focus on writing Narrow Integration Tests that actually connect to databases/external systems via test harnesses to test real behavior. | |
| - **No Static Mutable Singletons**: Avoid global state. Use dependency injection to pass dependencies dynamically. | |
| - **Infra & Business Decoupling**: Infrastructure code (AWS, SQS, DB connections) must never leak into business/domain logic. | |
| - **No Leakage**: Data transfer objects (DTOs), API models, and ORM models must not leak across their respective boundaries. Map them appropriately. | |
| - **DRY (Don't Repeat Yourself)**: Avoid code duplication. Extract shared logic into reusable, well-named functions/modules. | |
| # Package Manager | |
| - ALWAYS use `pnpm` for frontend/Node.js package management instead of `npm`. Do not use `npm install`. | |
| # Destructive Commands | |
| - NEVER use destructive terminal commands like `git checkout`, `git restore`, `git reset`, `git clean`, or `rm -rf` without explicitly asking for and receiving the user's permission first. Always prefer precise code-editing tools for reverting changes. |
🧰 Tools
🪛 markdownlint-cli2 (0.23.0)
[warning] 8-8: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
[warning] 19-19: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
[warning] 22-22: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.agents/AGENTS.md around lines 8 - 23, Add blank lines immediately after the
Markdown headings in AGENTS.md, including “Enterprise Coding Standards (Strictly
Enforced),” “Package Manager,” and “Destructive Commands,” while preserving all
existing list content.
Source: Linters/SAST tools
Comment on lines
+43
to
+46
| class SqlAlchemyDataPlaneRepository(DataPlaneRepositoryPort, SqlAlchemyTransactionRepository): | ||
| def __init__(self, session: TenantSession) -> None: | ||
| self.session = session | ||
|
|
||
| async def create_api_token( | ||
| self, | ||
| tenant_id: int, | ||
| name: str, | ||
| client_id: str, | ||
| secret_hash: str, | ||
| expires_at: object | None = None, | ||
| ) -> UUID: | ||
| import uuid as uuid_module | ||
|
|
||
| token_id = uuid_module.uuid4() | ||
| record = ApiToken( | ||
| id=token_id, | ||
| tenant_id=tenant_id, | ||
| name=name, | ||
| client_id=client_id, | ||
| secret_hash=secret_hash, | ||
| expires_at=expires_at, | ||
| active=False, | ||
| ) | ||
| self.session.add(record) | ||
| await self.session.flush() | ||
| return token_id | ||
|
|
||
| async def list_api_tokens(self, tenant_id: int) -> list[dict[str, Any]]: | ||
| result = await self.session.execute( | ||
| select(ApiToken) | ||
| .where(ApiToken.tenant_id == tenant_id) | ||
| .order_by(ApiToken.created_at.desc()) | ||
| ) | ||
| tokens = result.scalars().all() | ||
| return [ | ||
| { | ||
| "id": str(t.id), | ||
| "name": t.name, | ||
| "client_id": t.client_id, # safe to return; secret_hash is never exposed | ||
| "active": t.active, | ||
| "last_used_at": t.last_used_at.isoformat() if t.last_used_at else None, | ||
| "expires_at": t.expires_at.isoformat() if t.expires_at else None, | ||
| "created_at": t.created_at.isoformat(), | ||
| } | ||
| for t in tokens | ||
| ] | ||
|
|
||
| async def update_api_token( | ||
| self, tenant_id: int, token_id: UUID, name: str | None = None, active: bool | None = None | ||
| ) -> bool: | ||
| values: dict[str, Any] = {} | ||
| if name is not None: | ||
| values["name"] = name | ||
| if active is not None: | ||
| values["active"] = active | ||
|
|
||
| if not values: | ||
| return True | ||
|
|
||
| stmt = ( | ||
| update(ApiToken) | ||
| .where(ApiToken.id == token_id, ApiToken.tenant_id == tenant_id) | ||
| .values(**values) | ||
| ) | ||
| result = await self.session.execute(stmt) | ||
| await self.session.flush() | ||
| return (getattr(result, "rowcount", 0) or 0) > 0 | ||
|
|
||
| async def delete_api_token(self, tenant_id: int, token_id: UUID) -> bool: | ||
| result = await self.session.execute( | ||
| delete(ApiToken).where(ApiToken.id == token_id, ApiToken.tenant_id == tenant_id) | ||
| ) | ||
| await self.session.flush() | ||
| return (getattr(result, "rowcount", 0) or 0) > 0 | ||
|
|
||
| async def get_tenant_id_by_credentials(self, client_id: str, secret_hash: str) -> int | None: | ||
| """ | ||
| Two-step lookup (indexed client_id → hash check → tenant_id). | ||
| Step 1: Find row by client_id (plaintext index — O(1), no full scan). | ||
| Step 2: Verify secret_hash matches (prevents timing attacks via constant-time compare). | ||
| Also updates last_used_at. | ||
| """ | ||
| import hmac | ||
| from datetime import datetime, timedelta | ||
|
|
||
| from sqlalchemy import or_, update | ||
|
|
||
| now = datetime.now(UTC).replace(tzinfo=None) | ||
| result = await self.session.execute( | ||
| select(ApiToken).where( | ||
| ApiToken.client_id == client_id, | ||
| ApiToken.active.is_(True), | ||
| or_(ApiToken.expires_at.is_(None), ApiToken.expires_at > now), | ||
| ) | ||
| ) | ||
| record = result.scalar_one_or_none() | ||
| if not record: | ||
| return None | ||
|
|
||
| # Constant-time comparison prevents timing-based secret enumeration | ||
| if not hmac.compare_digest(record.secret_hash, secret_hash): | ||
| return None | ||
|
|
||
| if not record.last_used_at or record.last_used_at < (now - timedelta(hours=1)): | ||
| await self.session.execute( | ||
| update(ApiToken).where(ApiToken.id == record.id).values(last_used_at=now) | ||
| ) | ||
| return record.tenant_id | ||
| SqlAlchemyTransactionRepository.__init__(self, session) |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | 🏗️ Heavy lift
Missing Data Plane capabilities (Outbox, EdiJson, EdiMessage). The refactor of the Data Plane repository stripped away required capabilities from both the adapter and the port, while services still depend on them, which will cause AttributeError at runtime and type checker failures.
services/api/src/api/adapters/repository.py#L43-L46: Add missing mixins (e.g.,SqlAlchemyOutboxRepository,SqlAlchemyEdiJsonRepository,SqlAlchemyEdiMessageRepository) toSqlAlchemyDataPlaneRepositoryand initialize them in its__init__.services/api/src/api/ports/repository.py#L45-L46: Add the corresponding missing parent port interfaces (e.g.,OutboxRepositoryPort) to theDataPlaneRepositoryPortdefinition.services/api/src/api/services/api_receiver_service.py#L79-L102: This service (and others likeAs2ReceiverService) relies ondata_plane.publish_outbox_eventanddata_plane.create_edi_json, which will safely execute once the repository composition is restored.
📍 Affects 3 files
services/api/src/api/adapters/repository.py#L43-L46(this comment)services/api/src/api/ports/repository.py#L45-L46services/api/src/api/services/api_receiver_service.py#L79-L102
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/api/src/api/adapters/repository.py` around lines 43 - 46, Restore
the Data Plane repository capabilities by adding the Outbox, EdiJson, and
EdiMessage SQLAlchemy mixins to SqlAlchemyDataPlaneRepository and initializing
each with the session; extend DataPlaneRepositoryPort with the corresponding
port interfaces. Update services/api/src/api/adapters/repository.py lines 43-46
and services/api/src/api/ports/repository.py lines 45-46;
services/api/src/api/services/api_receiver_service.py lines 79-102 requires no
direct change because it is corrected by restoring these repository
capabilities.
Comment on lines
+28
to
+34
| def mock_uow(global_repo): | ||
| from unittest.mock import MagicMock | ||
|
|
||
| uow = MagicMock() | ||
| uow.control_plane = global_repo | ||
| return uow | ||
|
|
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Missing @pytest.fixture decorator.
The mock_uow function is injected as a dependency into other fixtures (like as2_partner_service(mock_uow)), but it is missing the @pytest.fixture decorator. Pytest will raise a FixtureNotFound error.
🐛 Proposed fix to add the fixture decorator
+@pytest.fixture
def mock_uow(global_repo):
from unittest.mock import MagicMock
uow = MagicMock()
uow.control_plane = global_repo
return uow📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def mock_uow(global_repo): | |
| from unittest.mock import MagicMock | |
| uow = MagicMock() | |
| uow.control_plane = global_repo | |
| return uow | |
| `@pytest.fixture` | |
| def mock_uow(global_repo): | |
| from unittest.mock import MagicMock | |
| uow = MagicMock() | |
| uow.control_plane = global_repo | |
| return uow |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@services/api/tests/test_provisioning_core.py` around lines 28 - 34, Add the
`@pytest.fixture` decorator to mock_uow so pytest registers it as a fixture for
dependent fixtures such as as2_partner_service. Preserve its existing MagicMock
setup and return behavior.
Comment on lines
+80
to
+82
| ### UnitOfWork Architecture (Control Plane vs Data Plane Naming) | ||
| Currently, the `UnitOfWork` (and its underlying SQL Alchemy repositories) leak infrastructure/deployment boundaries ("Control Plane" and "Data Plane") into domain business logic. We have giant God-objects like `SqlAlchemyControlPlaneRepository` inheriting from 10+ distinct repositories, causing namespace collisions and violating SOLID principles (Single Responsibility Principle). | ||
| **Future Action:** Refactor `UnitOfWork` to remove `control_plane` and `data_plane` concepts from class names and properties. Use Composition to expose distinct Bounded Contexts (e.g., `self.trading_partners`, `self.transactions`, `self.routes`) instead of lumping them into control/data plane buckets. |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Add a blank line below the heading for Markdown consistency.
To comply with standard Markdown formatting (MD022), ensure the heading is followed by a blank line.
♻️ Proposed formatting fix
### UnitOfWork Architecture (Control Plane vs Data Plane Naming)
+
Currently, the `UnitOfWork` (and its underlying SQL Alchemy repositories) leak infrastructure/deployment boundaries ("Control Plane" and "Data Plane") into domain business logic. We have giant God-objects like `SqlAlchemyControlPlaneRepository` inheriting from 10+ distinct repositories, causing namespace collisions and violating SOLID principles (Single Responsibility Principle).📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| ### UnitOfWork Architecture (Control Plane vs Data Plane Naming) | |
| Currently, the `UnitOfWork` (and its underlying SQL Alchemy repositories) leak infrastructure/deployment boundaries ("Control Plane" and "Data Plane") into domain business logic. We have giant God-objects like `SqlAlchemyControlPlaneRepository` inheriting from 10+ distinct repositories, causing namespace collisions and violating SOLID principles (Single Responsibility Principle). | |
| **Future Action:** Refactor `UnitOfWork` to remove `control_plane` and `data_plane` concepts from class names and properties. Use Composition to expose distinct Bounded Contexts (e.g., `self.trading_partners`, `self.transactions`, `self.routes`) instead of lumping them into control/data plane buckets. | |
| ### UnitOfWork Architecture (Control Plane vs Data Plane Naming) | |
| Currently, the `UnitOfWork` (and its underlying SQL Alchemy repositories) leak infrastructure/deployment boundaries ("Control Plane" and "Data Plane") into domain business logic. We have giant God-objects like `SqlAlchemyControlPlaneRepository` inheriting from 10+ distinct repositories, causing namespace collisions and violating SOLID principles (Single Responsibility Principle). | |
| **Future Action:** Refactor `UnitOfWork` to remove `control_plane` and `data_plane` concepts from class names and properties. Use Composition to expose distinct Bounded Contexts (e.g., `self.trading_partners`, `self.transactions`, `self.routes`) instead of lumping them into control/data plane buckets. |
🧰 Tools
🪛 markdownlint-cli2 (0.23.0)
[warning] 80-80: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@TECHNICAL_DEBT.md` around lines 80 - 82, Add a blank line immediately after
the “UnitOfWork Architecture (Control Plane vs Data Plane Naming)” heading in
TECHNICAL_DEBT.md, before the following paragraph, without changing the content.
Source: Linters/SAST tools
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit