edi json, access token, edi pipeline, edi envelope, cdc fix, outbound path testing - #16
Merged
Merged
Conversation
Contributor
|
Warning Review limit reached
Next review available in: 25 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 (3)
Note
|
| Layer / File(s) | Summary |
|---|---|
Development, identity, and CDC infrastructure .agents/AGENTS.md, Makefile, docker-compose.yml, docker/*, infra/zitadel/*, libs/domain/*, libs/identity/* |
Development commands, Debezium proxying, SQS queues, local identity provisioning, domain packaging, and tenant session resolution are updated. |
Database models, migrations, and contracts libs/database/*, libs/config/*, services/api/src/api/domain/*, services/api/src/api/ports/* |
API token, EDI JSON, route, timestamp, tenant-default, storage, and replicated configuration models and contracts are added or reshaped. |
Frontend and API workflows
| Layer / File(s) | Summary |
|---|---|
API token access and route configuration UI frontend/web/src/components/ui/*, frontend/web/src/features/developers/*, frontend/web/src/features/routes/*, frontend/web/src/routes/* |
Token creation, listing, revocation, deletion, credential display, route forms, route metadata, and tenant navigation are added. |
API services and HTTP workflows services/api/src/api/* |
API-key authentication, token lifecycle endpoints, outbound EDI JSON submission, route and partner services, certificate rotation, CDC relay routing, and service wiring are introduced. |
Processing workers
| Layer / File(s) | Summary |
|---|---|
Translation, envelopes, storage, and delivery libs/pipeline/*, libs/transformer/* |
Payload storage is configurable, JSON-to-EDI translation supports X12 and EDIFACT envelopes, metadata extraction is added, and delivery reads persisted payloads. |
Provisioning and worker orchestration services/worker/* |
Typed worker ports, database replication, SQS processing, Vault access, provisioning error handling, and the unified worker entrypoint are added. |
Estimated code review effort: 5 (Critical) | ~120 minutes
Possibly related PRs
- pramodnarayana/soopaedi#1: Related tenant identity and session-routing changes.
- pramodnarayana/soopaedi#2: Related CDC relay and queue-routing changes.
- pramodnarayana/soopaedi#11: Related provisioning service and replication changes.
Poem
A rabbit hops through queues of code,
With tokens tucked in carrot mode.
Routes bloom, envelopes fly,
Workers hum beneath the sky. 🐇
🚥 Pre-merge checks | ✅ 3 | ❌ 2
❌ Failed checks (1 warning, 1 inconclusive)
| Check name | Status | Explanation | Resolution |
|---|---|---|---|
| Docstring Coverage | Docstring coverage is 22.06% 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 related to the PR, but it reads like a keyword list rather than a concise summary of the main change. | Replace it with a short sentence that captures the primary change, e.g. "Add EDI JSON pipeline, API tokens, and route/outbox updates." |
✅ 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
feature/edi-json
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: 52
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
services/api/tests/test_cdc_relay.py (1)
98-113: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winTest names and docstrings contradict the new behavior.
test_cdc_relay_rejects_unknown_table(line 98) andtest_cdc_relay_rejects_missing_trace_id(line 142) both assert HTTP 200 with{"status": "ok"}, yet their names say "rejects" and their docstrings say "fails explicitly" and "must be rejected." The CDC relay now silently accepts and ignores these events instead of rejecting them. Update the test names and docstrings to reflect the actual behavior (e.g.,ignores_unknown_table,skips_missing_trace_id).✏️ Suggested rename and docstring updates
-def test_cdc_relay_rejects_unknown_table(memory_queue: InMemoryQueueAdapter) -> None: - """Tests the CDC relay fails explicitly on unknown table sources to prevent silent drops.""" +def test_cdc_relay_ignores_unknown_table(memory_queue: InMemoryQueueAdapter) -> None: + """Tests the CDC relay silently ignores events from unknown tables to avoid blocking the pipeline."""-def test_cdc_relay_rejects_missing_trace_id(memory_queue: InMemoryQueueAdapter) -> None: - """Payloads without trace_id must be rejected to prevent poison messages in SQS.""" +def test_cdc_relay_skips_missing_trace_id(memory_queue: InMemoryQueueAdapter) -> None: + """Payloads without trace_id are acknowledged but not enqueued to prevent poison messages in SQS."""Also applies to: 142-156
🤖 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_cdc_relay.py` around lines 98 - 113, Rename test_cdc_relay_rejects_unknown_table and test_cdc_relay_rejects_missing_trace_id to reflect that these events are accepted but ignored, such as ignores_unknown_table and skips_missing_trace_id. Update both docstrings to describe the silent ignore/skip behavior while preserving their existing HTTP 200, response, and queue assertions.services/api/src/api/routers/webhooks/webhook.py (1)
36-38: 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy liftSSRF check is vulnerable to DNS rebinding (TOCTOU).
The hostname is resolved and validated at webhook creation time, but the URL (not the IP) is persisted. When the webhook is delivered later, the hostname is resolved again — potentially to a different IP. An attacker controlling DNS can return a public IP during validation and a private/internal IP during delivery, bypassing the SSRF protection entirely.
Additionally,
socket.gethostbynamereturns only one IPv4 address; if the hostname resolves to multiple IPs, only one is validated.Consider one or more of the following mitigations:
- Pin the resolved IP — store the validated IP alongside the URL and use it for the outbound connection (set
Hostheader to the original hostname).- Re-validate at delivery time — repeat the SSRF check in the delivery worker before making the HTTP request.
- Use a custom DNS resolver with caching/pinning to ensure the delivery-time resolution matches the validation-time resolution.
🛡️ Proposed re-validation at delivery time
ip = await anyio.to_thread.run_sync(socket.gethostbyname, hostname) ip_obj = ipaddress.ip_address(ip) -if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_unspecified: +if ( + ip_obj.is_private + or ip_obj.is_loopback + or ip_obj.is_unspecified + or ip_obj.is_link_local + or ip_obj.is_multicast + or ip_obj.is_reserved +): raise HTTPException(status_code=400, detail="Webhook URL must be a public address")Also add
is_link_localexplicitly as defense-in-depth for cloud metadata endpoint (169.254.169.254) protection, sinceis_privatecoverage of this range varies across Python versions.🤖 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 36 - 38, Mitigate DNS rebinding by repeating SSRF validation immediately before the outbound request in the webhook delivery worker, and validate every address returned by DNS rather than only one IPv4 result. Update the shared validation logic around socket.gethostbyname and delivery code to reject private, loopback, unspecified, and explicitly is_link_local addresses, including IPv6 results, before connecting.libs/pipeline/tests/test_delivery_service_as2.py (1)
93-113: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRemove dead
storageandedi_s3_urivariables from AS2 delivery tests.After the refactoring,
make_serviceno longer accepts astorageadapter and_seed_as2_routetakesedi_datadirectly. However, several tests still createInMemoryStorageAdapter()instances andedi_s3_urivariables that are never used. This dead code appears intest_deliver_as2_plain_no_crypto,test_deliver_as2_http_failure_sets_failed_status,test_deliver_as2_null_adapter_is_caught_and_marked_failed,test_deliver_as2_idempotent_claim, andtest_deliver_as2_missing_local_partner_sets_failed.♻️ Example cleanup for test_deliver_as2_plain_no_crypto
async def test_deliver_as2_plain_no_crypto() -> None: """ When no crypto material is configured, the raw EDI payload is transmitted as-is and the AS2 HTTP headers are correctly set. """ # ── Arrange ──────────────────────────────────────────────────────────────── - storage = InMemoryStorageAdapter() repo = InMemoryRepositoryAdapter() as2_adapter = FakeAS2DeliveryAdapter() trace_id = "trace-as2-plain" - edi_s3_uri = f"s3://bucket/edi/{trace_id}/raw.edi" raw_edi = ( b"ISA*00* *00* *ZZ*SENDER " b"*ZZ*RECEIVER *210101*1200*^*00501*000000001*0*P*>~" ) - storage.store[edi_s3_uri] = raw_edi _seed_as2_route(repo, trace_id, raw_edi.decode("utf-8"))🤖 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/pipeline/tests/test_delivery_service_as2.py` around lines 93 - 113, Remove the unused InMemoryStorageAdapter instances and edi_s3_uri variables from test_deliver_as2_plain_no_crypto, test_deliver_as2_http_failure_sets_failed_status, test_deliver_as2_null_adapter_is_caught_and_marked_failed, test_deliver_as2_idempotent_claim, and test_deliver_as2_missing_local_partner_sets_failed; keep each test’s raw EDI data passed directly through _seed_as2_route and preserve the existing delivery assertions.services/api/src/api/routers/trading_partners/platform/as2_partners.py (1)
72-85: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winUse
p.idconsistently and add a null check on the fetched partner.The create endpoint uses
entity.partner_idfor theidfield while the list and update endpoints usep.id/updated_partner.idfrom the fetched record. Additionally, unlike the update endpoint (line 135–136), there's no null check onpafter the fetch — anAttributeErrorwould occur ifget_as2_partnerreturnsNone.🔧 Proposed fix for consistency and safety
await uow.commit() p = await uow.control_plane.get_as2_partner(tenant_id=0, partner_id=entity.partner_id) + + if not p: + raise HTTPException(status_code=404, detail="Partner not found after creation") return AS2TradingPartnerResponse( - id=str(entity.partner_id), + id=str(p.id), name=p.name, as2_id=p.as2_id, is_local=p.is_local, url=p.url, active=p.active, )🤖 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/trading_partners/platform/as2_partners.py` around lines 72 - 85, In the create endpoint, update the fetched partner handling after get_as2_partner to check whether p is None and raise the same appropriate not-found error used by the update endpoint; then populate AS2TradingPartnerResponse.id from p.id instead of entity.partner_id, keeping identifiers consistent with the list and update endpoints.
🤖 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 17-23: Add dead-letter queues and redrive policies for
TranslateQueue, DeliverQueue, and ProvisioningQueue in init-aws.sh, matching the
existing EdiTransformerQueue pattern and using maxReceiveCount=3 so
poison-message handling can be tested locally.
In `@frontend/web/src/components/ui/alert.tsx`:
- Around line 35-57: Update AlertTitle and AlertDescription to use ref types
matching their rendered elements: HTMLHeadingElement for the h5 and
HTMLDivElement for the div. Since the project uses React 19, remove
React.forwardRef and accept ref as a regular prop while preserving the existing
className and prop handling.
In `@frontend/web/src/components/ui/dropdown-menu.tsx`:
- Around line 19-37: Update DropdownMenuSubTrigger and the other dropdown-menu
wrappers to accept ref directly in their component props instead of using
React.forwardRef, matching the React 19 pattern while preserving existing ref
forwarding behavior, props, styling, and exported component APIs.
In `@frontend/web/src/features/developers/api/apiTokenHooks.ts`:
- Around line 43-49: Add an enabled condition to useApiTokensQuery so useQuery
only runs when a valid access token is available, rather than invoking
repo.getApiTokens with the repository’s empty-token fallback. Use the same
authentication state that supplies useRepository, and ensure the query becomes
enabled and fetches once the token is restored.
In `@frontend/web/src/features/developers/components/ApiTokensTable.tsx`:
- Around line 91-97: Update the last_used_at cell renderer in ApiTokensTable to
wrap parseISO and formatDistanceToNow in the same try-catch pattern used by the
created_at column, returning the existing fallback for invalid dates instead of
allowing parsing errors to crash the table.
In `@frontend/web/src/features/developers/components/TokenCredentialsModal.tsx`:
- Around line 25-34: Handle rejected clipboard writes in copyToClipboard by
wrapping navigator.clipboard.writeText in try/catch, and only update
copiedSecret or copiedId state and timers after a successful write. Add
appropriate user feedback for failures, such as displaying an error state or
notification, without allowing an unhandled rejection.
In `@frontend/web/src/features/routes/components/CreateRouteModal.tsx`:
- Around line 14-26: Update the Dialog in CreateRouteModal to use the default
modal behavior by removing modal={false}, preserving focus trapping and dialog
semantics. Keep onPointerDownOutside preventing dismissal for the non-closable
form, and remove the contradictory onFocusOutside prevention unless a specific
accessibility requirement requires it.
In `@infra/zitadel/machinekey.json`:
- Line 1: Remove the committed RSA private key from machinekey.json, immediately
rotate/revoke the Zitadel service-account credential, and purge the file and
secret from all Git history using git filter-repo or BFG before force-pushing.
Add machinekey.json or its containing directory to .gitignore, then update the
runtime configuration to read the replacement credential from an injected
environment variable or secret manager rather than repository files.
In
`@libs/database/src/database/migrations/global/versions/94b943b2920d_global_initial_schema.py`:
- Around line 147-150: Remove the redundant uniqueness definition for
api_tokens.client_id: update the api_tokens table declaration and migration
operations so only either UniqueConstraint("client_id") or the unique index
ix_api_tokens_client_id remains, preferably retaining the unique index and
deleting the table-level constraint.
In `@libs/database/src/database/models/common.py`:
- Around line 40-46: Update the timestamp definitions in the common model mixin
and OutboxMixin to use UTC-aware values: replace datetime.utcnow with
datetime.now(timezone.utc) and configure each corresponding DateTime column with
timezone=True, including OutboxMixin.created_at.
In `@libs/database/src/database/models/control_plane.py`:
- Around line 98-105: Remove the redundant unique index declaration from the
model’s __table_args__, keeping client_id’s existing unique=True constraint;
ensure no separate Index is created for client_id.
In `@libs/database/src/database/models/data_plane.py`:
- Line 165: Remove the inline created_at and updated_at column declarations from
EdiMessage, allowing TimestampMixin to provide them without duplication.
In `@libs/database/src/database/models/replicated_mixins.py`:
- Around line 61-62: Add a matching database-side server_default of false to
every active column with a Python default across AS2PartnerMixin,
AS2PartnershipMixin, SFTPPartnerMixin, WebhookMixin, InboundRouteMixin, and
OutboundRouteMixin. Preserve the existing Python default and Boolean type,
following the established pattern used by OutboundRouteMixin.default_standard
and default_version.
In `@libs/database/src/database/repository.py`:
- Around line 67-77: Update the repository method get_inbound_route to accept a
tenant_id parameter and add InboundRoute.tenant_id == tenant_id to its query
filters alongside the ISA IDs and active condition; update all callers to
provide the appropriate tenant ID.
In `@libs/identity/docker/docker-compose.yml`:
- Line 45: Remove the committed Zitadel machine credential file
infra/zitadel/machinekey.json from version control, remove any related volume
mounts such as the machinekey mapping in the Docker Compose configuration, and
rotate/revoke the exposed credential by generating a replacement.
In `@libs/pipeline/src/pipeline/adapters/transformer.py`:
- Line 38: Remove the unused import datetime and delete the discarded
datetime.datetime.utcnow() call in the affected transformation logic, leaving
the surrounding behavior unchanged.
In `@libs/pipeline/src/pipeline/core/translate.py`:
- Line 108: Remove the unused json.dumps(json_dict).encode("utf-8") statement
and the redundant edi_msg.get("tenant_id") call from the translation logic;
retain the existing tenant_id usage near line 132 and ensure behavior remains
unchanged.
In `@libs/transformer/src/transformer/domain/envelope/edifact.py`:
- Line 71: Replace the naive UTC timestamp created with
datetime.datetime.utcnow() in the envelope timestamp logic with
datetime.datetime.now(datetime.timezone.utc), ensuring the resulting datetime is
timezone-aware and compatible with aware datetime comparisons.
In `@libs/transformer/src/transformer/domain/envelope/x12.py`:
- Around line 110-116: Replace the random generation of ISA13 and GS06 in the
envelope-generation logic with collision-safe monotonic control numbers,
preferably sourced from a database-backed sequence or distributed counter scoped
to the trading partner relationship; preserve the required formatting and ensure
each value is unique across concurrent messages.
- Line 110: Replace the deprecated datetime.datetime.utcnow() call in the
envelope timestamp logic with datetime.datetime.now(datetime.timezone.utc),
ensuring the resulting timestamp is timezone-aware and remains in UTC.
In `@services/api/src/api/adapters/repository.py`:
- Around line 815-843: Update get_tenant_id_by_credentials to restrict the
ApiToken query to tokens whose expires_at is either null or later than the
current UTC time, alongside the existing client_id and active filters, so
expired tokens return None before secret verification.
- Around line 96-112: Update rotate_as2_certificates to raise ValueError
immediately when get_as2_partner returns None, before applying certificate
changes or calling session.flush(); preserve the existing rotation and flush
behavior for a found partner, matching create_as2_partnership’s missing-partner
handling.
In `@services/api/src/api/auth/api_key.py`:
- Around line 87-92: Address the multi-process invalidation gap in
invalidate_token_cache: replace the process-local _token_cache approach with a
shared Redis-backed cache and TTL, add pub/sub invalidation, or remove caching
and rely on the O(1) database lookup. Ensure revoking a token immediately
prevents cached authentication across all workers.
- Around line 62-63: Fix the authentication bypass in the token-cache logic:
update the cache used by the authentication function to store both the tenant_id
and the client_secret hash, compute the provided secret’s hash before checking
the cache, and on cache hits compare hashes using constant-time verification
before returning the tenant_id. Apply the same validation to the cache
write/read paths around the referenced authentication method.
In `@services/api/src/api/cdc_relay.py`:
- Around line 64-72: Handle non-mapping elements in the raw_events loop before
constructing DebeziumUnwrappedEvent, or catch the resulting TypeError alongside
ValidationError. Log the invalid payload and continue so malformed items are
skipped without failing the batch; update the exception handling around the
validated_events append.
- Around line 86-89: Handle malformed string payloads in the event payload
parsing logic by catching JSONDecodeError around json.loads within the CDC relay
handler, logging or recording the invalid payload as appropriate, and
skipping/acknowledging the event instead of allowing the exception to propagate
and trigger retries. Use the surrounding handler’s existing skip/error flow and
symbols to preserve consistent behavior.
In `@services/api/src/api/core/services/route_service.py`:
- Around line 93-199: Extract the duplicated destination type/name selection
from both loops in list_routes into a private helper method, such as
_resolve_destination, accepting a route and the fetched AS2, SFTP, and webhook
name mappings. Have it return the destination type and name, handling inbound
webhook destinations while preserving outbound behavior, then call it from both
loops.
In `@services/api/src/api/domain/models.py`:
- Around line 205-220: Replace CreateApiTokenCmd.expires_at’s Any | None
annotation with datetime | None and import datetime as needed. Propagate the
explicit datetime | None type through the related API token port and adapter
method signatures, replacing object | None or Any | None while preserving
existing behavior.
In `@services/api/src/api/routers/developers/api_tokens.py`:
- Around line 39-42: Replace the placeholder tenant_name logic and remove the
TODO comments by fetching the actual tenant name from the tenant repository
using tenant_id, then pass that value to the token service’s prefix/slug
generation logic. Handle a missing tenant consistently with existing repository
conventions before generating the token.
- Around line 88-101: The hard-delete endpoint currently permits any
authenticated tenant user. Update delete_api_token to add
require_platform_admin() as a dependency, while retaining tenant resolution and
repository injection, so only platform administrators can invoke the
irreversible deletion.
- Around line 44-55: Replace the `"just now"` placeholder in the
`ApiTokenCreatedResponse` returned by the token-creation handler with
`token.created_at` when available; otherwise use the current UTC timestamp in
the format expected by `ApiTokenCreatedResponse.created_at`, ensuring the
response remains type-safe and parseable.
In `@services/api/src/api/routers/routes.py`:
- Around line 119-127: Move UNSET, UpdateInboundRouteCmd, and
UpdateOutboundRouteCmd from the inbound and outbound route handler bodies to the
module-level imports, alongside the existing Create* command imports; remove the
now-redundant inline imports.
- Line 47: Update the docstrings for the route creation methods near the
referenced symbols to remove “directly in the Tenant Data Plane” and describe
routes as being managed in the global/control-plane schema, consistent with the
uow.control_plane implementation and migration.
In `@services/api/src/api/routers/trading_partners/as2.py`:
- Around line 156-159: Update the certificate rotation endpoint’s exception
handling around the shown block: handle ValueError separately with HTTP 400 and
a safe client-facing detail, log unexpected exceptions with context, and return
HTTP 500 without exposing str(e). Make vault.delete_secret cleanup best-effort
by catching and logging cleanup failures so it cannot mask the original
exception; preserve exception chaining where applicable.
In `@services/api/src/api/services/as2_receive_service.py`:
- Around line 292-314: Update _extract_isa_headers to decode pure_edi_bytes
using a one-to-one encoding such as latin-1 instead of ASCII with
errors="ignore", preserving fixed byte-to-character offsets for
element_separator, isa_segment, and sender/receiver extraction.
- Around line 327-336: The inbound-route lookup in the AS2 receive flow is not
sufficiently scoped and may resolve the wrong tenant. Update get_inbound_route
and its call from the AS2 receive service to include the tenant/partnership and
transaction_type keys required by the tenant-scoped unique index, then derive
true_tenant_id only from the fully scoped route result.
In `@services/api/src/api/services/outbound_service.py`:
- Around line 57-80: List payloads currently skip transaction_type inference,
causing extraction to receive an empty type. Update the transaction_type
inference logic in the outbound service to inspect payload[0] when payload is a
non-empty list, applying the same transaction_type, heading, and ST fallback
checks before calling extractor.extract; preserve existing behavior for
dictionary payloads.
In `@services/api/tests/test_api_repository.py`:
- Around line 177-179: Strengthen the tenant-isolation assertion in the relevant
test by removing the standalone `assert "1" in compiled_clean` and the
unverified `tenant_id=` alternative. Assert that the compiled SQL explicitly
binds the tenant value in the predicate, such as `tenant_id=1` or
`tenant_idIN(1,0)`, while preserving whitespace normalization.
In `@services/worker/src/worker/adapters/db_outbox.py`:
- Around line 59-61: Replace the manual global_gen.__anext__() call in the
finally block with await global_gen.aclose(), retaining appropriate suppression
if needed, so the async generator is closed through its standard cleanup
mechanism and reliably releases the session.
In `@services/worker/src/worker/adapters/db_replication.py`:
- Around line 256-289: Add webhook_id to the on_conflict_do_update set_
dictionary in the TenantInboundRoute upsert within the inbound route replication
logic, matching the value already supplied by insert_ir_stmt.values() so
existing tenant records receive updated webhook references.
- Around line 370-406: Replace the global_model.__name__ check in _sync_deletes
with an explicit include_shared: bool parameter, using it to decide whether
tenant_id 0 or NULL records are included. Update every _sync_deletes call site
to pass the appropriate value: true for AS2Partner and AS2Partnership
synchronization, false for other models.
- Around line 33-58: Move acquisition of global_gen and tenant_gen and their
__anext__() calls inside an outer cleanup scope in
replicate_tenant_configuration, using contextlib.aclosing() so each async
generator is closed even when startup or connection acquisition fails. Preserve
tenant rollback/commit behavior, and explicitly finalize the global session
(rollback or equivalent cleanup for its read-only work) before closing both
generators; retain the existing provisioning error handling and replication
flow.
In `@services/worker/src/worker/adapters/db_tenant.py`:
- Line 13: Implement cache invalidation for the cache used by resolve_shard: add
a TTL or explicit cache-clear mechanism, ensure expired entries are refreshed,
and expose invalidation for shard reconfiguration. Update resolve_shard and
related cache access so migrated tenants or changed DSNs cannot remain cached
indefinitely.
In `@services/worker/src/worker/adapters/sqs_outbox.py`:
- Around line 84-98: Both exception branches in the SQS event context manager
currently swallow failures after queue handling. In the exception handler around
yielding the event, retain the existing permanent/transient logging and delete
behavior, then re-raise the original exception so process_next_event() can
detect the failed event.
In `@services/worker/src/worker/adapters/vault.py`:
- Around line 26-36: Update VaultAdapter.get_secret and its nested _fetch
function to fail clearly when the response contains no value or the first value
is not a string, instead of returning an empty string. Raise an appropriate
descriptive error that identifies the invalid or missing Vault secret.
In `@services/worker/src/worker/core/service.py`:
- Around line 35-50: Track whether any tenant replication fails in the global
broadcast branch of process_next_event, while continuing to attempt all tenants;
after the loop, raise TransientProvisioningError if a failure occurred so the
event is retried instead of being marked processed. Preserve the per-tenant
logger.exception diagnostics and avoid swallowing the failure.
In `@services/worker/src/worker/data/main.py`:
- Around line 137-144: Replace the print calls in process_translation with the
module logger: use logger.info for the translation start and success messages,
and logger.exception in the exception handler to preserve the traceback and
contextual trace_id.
- Around line 173-176: Apply the existing SSRF protection used by
HttpxDeliveryAdapter to HttpxAS2DeliveryAdapter by passing validate_target_url
when constructing as2_adapter, and ensure the adapter invokes it before posting
to remote_partner.url; alternatively enforce the same validation during AS2
partner creation.
In `@services/worker/src/worker/main.py`:
- Around line 14-17: Update the concurrent task orchestration in the worker
entrypoint to isolate failures: await data_main and provision_main with
asyncio.gather(..., return_exceptions=True), inspect each result, and log an
error identifying the corresponding task and exception before exiting or
handling the failure. Keep one task’s exception from cancelling its sibling.
In `@services/worker/src/worker/provision/main.py`:
- Around line 17-28: In run_worker, remove or replace the stale SQS
WaitTimeSeconds comment with wording that reflects database outbox polling, and
change the idle asyncio.sleep(0.1) delay to asyncio.sleep(1.0) to reduce
unnecessary database polling while retaining responsive processing.
In `@services/worker/tests/test_provision_worker.py`:
- Around line 9-73: Add tests alongside test_process_next_event_no_event,
test_process_next_event_tenant_specific, and test_process_next_event_global for
the requested edge cases: configure replicate_tenant_configuration to raise
PermanentProvisioningError and assert the message is deleted with the expected
behavior; configure a transient exception and assert the message remains queued;
and provide an event payload without tenant_id, verifying process_next_event
handles it gracefully. Use mocks to assert the relevant SqsOutboxAdapter message
deletion/retention interactions.
In `@TECHNICAL_DEBT.md`:
- Around line 59-60: Add blank lines after the `### No Frontend Test Runner
(Vitest)` heading and the other affected `###` heading around lines 66-67,
placing an empty line before their following content to satisfy MD022.
---
Outside diff comments:
In `@libs/pipeline/tests/test_delivery_service_as2.py`:
- Around line 93-113: Remove the unused InMemoryStorageAdapter instances and
edi_s3_uri variables from test_deliver_as2_plain_no_crypto,
test_deliver_as2_http_failure_sets_failed_status,
test_deliver_as2_null_adapter_is_caught_and_marked_failed,
test_deliver_as2_idempotent_claim, and
test_deliver_as2_missing_local_partner_sets_failed; keep each test’s raw EDI
data passed directly through _seed_as2_route and preserve the existing delivery
assertions.
In `@services/api/src/api/routers/trading_partners/platform/as2_partners.py`:
- Around line 72-85: In the create endpoint, update the fetched partner handling
after get_as2_partner to check whether p is None and raise the same appropriate
not-found error used by the update endpoint; then populate
AS2TradingPartnerResponse.id from p.id instead of entity.partner_id, keeping
identifiers consistent with the list and update endpoints.
In `@services/api/src/api/routers/webhooks/webhook.py`:
- Around line 36-38: Mitigate DNS rebinding by repeating SSRF validation
immediately before the outbound request in the webhook delivery worker, and
validate every address returned by DNS rather than only one IPv4 result. Update
the shared validation logic around socket.gethostbyname and delivery code to
reject private, loopback, unspecified, and explicitly is_link_local addresses,
including IPv6 results, before connecting.
In `@services/api/tests/test_cdc_relay.py`:
- Around line 98-113: Rename test_cdc_relay_rejects_unknown_table and
test_cdc_relay_rejects_missing_trace_id to reflect that these events are
accepted but ignored, such as ignores_unknown_table and skips_missing_trace_id.
Update both docstrings to describe the silent ignore/skip behavior while
preserving their existing HTTP 200, response, and queue assertions.
🪄 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: 7ed0791a-af5c-40c8-9484-cdb72a20959d
⛔ Files ignored due to path filters (3)
frontend/web/pnpm-lock.yamlis excluded by!**/pnpm-lock.yamlinfra/zitadel/terraform.tfstateis excluded by!**/*.tfstateuv.lockis excluded by!**/*.lock
📒 Files selected for processing (133)
.agents/AGENTS.mdMakefileTECHNICAL_DEBT.mddocker-compose.ymldocker/debezium/application-global.propertiesdocker/localstack/init-aws.shfrontend/web/package.jsonfrontend/web/src/components/ui/alert.tsxfrontend/web/src/components/ui/dropdown-menu.tsxfrontend/web/src/components/ui/tabs.tsxfrontend/web/src/features/developers/api/IApiTokenRepository.tsfrontend/web/src/features/developers/api/apiTokenHooks.tsfrontend/web/src/features/developers/api/apiTokensApi.tsfrontend/web/src/features/developers/components/ApiTokensTable.tsxfrontend/web/src/features/developers/components/CreateApiTokenModal.tsxfrontend/web/src/features/developers/components/TokenCredentialsModal.tsxfrontend/web/src/features/developers/types.tsfrontend/web/src/features/partners/components/PartnershipDetails.tsxfrontend/web/src/features/partners/components/PartnershipsTable.tsxfrontend/web/src/features/partners/types.tsfrontend/web/src/features/routes/components/CreateRouteModal.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/types.tsfrontend/web/src/routeTree.gen.tsfrontend/web/src/routes/tenant.tsxfrontend/web/src/routes/tenant/developers.tsxinfra/zitadel/.terraform.lock.hclinfra/zitadel/machinekey.jsoninfra/zitadel/main.tflibs/config/src/config/settings.pylibs/database/src/database/migrations/global/versions/94b943b2920d_global_initial_schema.pylibs/database/src/database/migrations/tenant/versions/0841cfb2afb2_tenant_initial_schema.pylibs/database/src/database/models/common.pylibs/database/src/database/models/control_plane.pylibs/database/src/database/models/data_plane.pylibs/database/src/database/models/replicated_mixins.pylibs/database/src/database/repository.pylibs/domain/README.mdlibs/domain/pyproject.tomllibs/domain/src/domain/__init__.pylibs/domain/src/domain/events.pylibs/identity/docker/docker-compose.ymllibs/identity/src/identity/dependencies.pylibs/identity/tests/test_dependencies.pylibs/pipeline/pyproject.tomllibs/pipeline/src/pipeline/adapters/repository.pylibs/pipeline/src/pipeline/adapters/transformer.pylibs/pipeline/src/pipeline/core/deliver.pylibs/pipeline/src/pipeline/core/metadata_extractor.pylibs/pipeline/src/pipeline/core/translate.pylibs/pipeline/src/pipeline/ports/repository.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_transformer.pylibs/pipeline/tests/test_translation_service.pylibs/transformer/src/transformer/domain/ast_utils.pylibs/transformer/src/transformer/domain/envelope/__init__.pylibs/transformer/src/transformer/domain/envelope/base.pylibs/transformer/src/transformer/domain/envelope/edifact.pylibs/transformer/src/transformer/domain/envelope/x12.pylibs/transformer/src/transformer/domain/envelope_factory.pylibs/transformer/tests/domain_tests/__init__.pylibs/transformer/tests/domain_tests/test_domain_models.pypyproject.tomlscripts/purge_sqs.pyservices/api/pyproject.tomlservices/api/src/api/adapters/http/dtos.pyservices/api/src/api/adapters/repository.pyservices/api/src/api/adapters/vault.pyservices/api/src/api/auth/__init__.pyservices/api/src/api/auth/api_key.pyservices/api/src/api/cdc_relay.pyservices/api/src/api/core/authorization.pyservices/api/src/api/core/provisioning.pyservices/api/src/api/core/services/__init__.pyservices/api/src/api/core/services/api_token_service.pyservices/api/src/api/core/services/as2_partner_service.pyservices/api/src/api/core/services/as2_partnership_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/dependencies.pyservices/api/src/api/domain/models.pyservices/api/src/api/main.pyservices/api/src/api/ports/repository.pyservices/api/src/api/routers/developers/__init__.pyservices/api/src/api/routers/developers/api_tokens.pyservices/api/src/api/routers/edi_json.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/webhooks/webhook.pyservices/api/src/api/services/as2_receive_service.pyservices/api/src/api/services/outbound_service.pyservices/api/tests/api_fakes.pyservices/api/tests/test_api_repository.pyservices/api/tests/test_api_token_service.pyservices/api/tests/test_as2_partner_service.pyservices/api/tests/test_as2_receive_service.pyservices/api/tests/test_cdc_relay.pyservices/api/tests/test_outbound_service.pyservices/api/tests/test_provisioning_core.pyservices/api/tests/test_routers_api_tokens.pyservices/api/tests/test_routers_partners.pyservices/api/tests/test_routers_routes.pyservices/worker/pyproject.tomlservices/worker/src/worker/adapters/__init__.pyservices/worker/src/worker/adapters/db_outbox.pyservices/worker/src/worker/adapters/db_replication.pyservices/worker/src/worker/adapters/db_tenant.pyservices/worker/src/worker/adapters/sqs_outbox.pyservices/worker/src/worker/adapters/vault.pyservices/worker/src/worker/core/__init__.pyservices/worker/src/worker/core/errors.pyservices/worker/src/worker/core/service.pyservices/worker/src/worker/data/main.pyservices/worker/src/worker/main.pyservices/worker/src/worker/ports/__init__.pyservices/worker/src/worker/ports/outbox.pyservices/worker/src/worker/ports/replication.pyservices/worker/src/worker/ports/tenant.pyservices/worker/src/worker/provision/main.pyservices/worker/tests/test_data_worker.pyservices/worker/tests/test_provision_worker.pyservices/worker/tests/test_utils.py
💤 Files with no reviewable changes (2)
- services/worker/tests/test_utils.py
- services/api/src/api/core/provisioning.py
Comment on lines
173
to
+176
| http_adapter = HttpxDeliveryAdapter(validator=validate_target_url) | ||
| sftp_adapter = ParamikoSftpDeliveryAdapter() | ||
| vault_adapter = WorkerVaultAdapter() | ||
| as2_adapter = HttpxAS2DeliveryAdapter() |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP 'class HttpxAS2DeliveryAdapter' --type=py -A25Repository: pramodnarayana/soopaedi
Length of output: 161
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the AS2 adapter and related validation wiring.
rg -n --hidden --glob '!**/.git/**' -e 'HttpxAS2DeliveryAdapter|validate_target_url|AS2' services/worker/src
# Show file list candidates if the adapter lives elsewhere.
git ls-files | rg 'AS2|as2|delivery|worker/data/main\.py'Repository: pramodnarayana/soopaedi
Length of output: 5544
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find where the AS2 delivery adapter is defined and how it handles target URLs.
fd -a '.*\.py$' services/worker/src | xargs rg -n --no-heading -e 'class .*AS2|def .*AS2|validate_target_url|target_url|urlparse|SSRF|allowlist|validator' || trueRepository: pramodnarayana/soopaedi
Length of output: 1400
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Broadly locate AS2-related implementation files and inspect the worker wiring.
files=$(git ls-files | rg 'as2|AS2|worker/data/main\.py')
printf '%s\n' "$files"
# If relevant files exist, show concise outlines / snippets.
for f in $files; do
if [ -f "$f" ]; then
echo "### $f"
wc -l "$f"
sed -n '1,260p' "$f" | rg -n 'HttpxAS2DeliveryAdapter|validate_target_url|target_url|validator|AS2|as2'
fi
doneRepository: pramodnarayana/soopaedi
Length of output: 6376
🌐 Web query:
"HttpxAS2DeliveryAdapter validate_target_url SSRF"
💡 Result:
The term HttpxAS2DeliveryAdapter does not refer to a widely recognized standard library or public vulnerability. It is likely a custom implementation or a specific component within a private or niche codebase, as evidenced by its naming convention often found in internal delivery or webhook processing modules [1][2][3]. In the context of SSRF (Server-Side Request Forgery) protection for HTTP clients like httpx, the validation pattern validate_target_url is a common, though historically precarious, approach to security [4][1][2]. Key technical considerations regarding validate_target_url and SSRF include: 1. The TOCTOU (Time-of-Check to Time-of-Use) Vulnerability: Many implementations of validate_target_url perform URL validation (parsing the scheme, hostname, and checking against a blocklist) as a separate, initial step [4]. Because the actual network request is made later by the HTTP client, an attacker can exploit DNS rebinding to bypass these checks [5][4]. An attacker-controlled domain may resolve to a public IP during the validation check and subsequently to a private or internal IP (e.g., 127.0.0.1, 169.254.169.254) when the client actually connects [4]. 2. Robust Mitigation Strategies: To prevent SSRF effectively, relying solely on a validate_target_url function is generally insufficient [4]. Industry-standard solutions involve: - Using a custom, SSRF-safe httpx transport [6][7][4]. - Resolving the hostname once and performing the validation on the resolved IP addresses [8][4]. - Pinning the connection to the validated IP to ensure the request is only made to the intended, safe destination, thereby closing the DNS rebinding window [6][4]. 3. General HTTPX Security: While httpx is a robust library, it is not inherently immune to SSRF if used without proper transport-level controls [4]. Past security advisories for httpx (such as CVE-2021-41945) have focused on URL parsing and input validation issues, but the primary defense against SSRF in applications using httpx must be implemented by the developer at the transport or network layer [8][4][9][10]. If you are auditing a system containing a HttpxAS2DeliveryAdapter, you should verify whether the validate_target_url function merely parses the URL string or if it enforces secure, atomic DNS resolution and IP validation before the connection is established [4][3].
Citations:
- 1: https://github.com/emiltsoi/hermes-agent-a2a/releases/tag/v3.3.1
- 2: emiltsoi/hermes-agent-a2a@0c44187
- 3: https://github.com/adriannoes/asap-protocol/blob/main/src/asap/transport/webhook.py
- 4: GHSA-r7w7-9xr2-qq2r
- 5: https://github.com/layzhi/safehttpx
- 6: https://github.com/langchain-ai/langchain/blob/02991cb4cf2063d51a07268edafb05fe53de1826/libs/core/langchain_core/_security/_transport.py
- 7: agent-fox-dev/agent-fox@97db774
- 8: https://github.com/Zaczero/pkgs/tree/main/httpx-secure
- 9: https://osv.dev/vulnerability/GHSA-h8pj-cxx2-jfg2
- 10: https://security.archlinux.org/CVE-2021-41945
🏁 Script executed:
#!/bin/bash
set -euo pipefail
for f in \
libs/pipeline/src/pipeline/adapters/as2.py \
libs/pipeline/src/pipeline/ports/as2.py \
libs/pipeline/tests/test_delivery_service_as2.py
do
echo "### $f"
wc -l "$f"
sed -n '1,260p' "$f"
echo
doneRepository: pramodnarayana/soopaedi
Length of output: 12185
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Trace how the AS2 remote URL reaches the adapter.
for f in \
libs/pipeline/src/pipeline/core/deliver.py \
libs/pipeline/src/pipeline/core/as2_orchestrator.py \
services/worker/src/worker/data/main.py
do
echo "### $f"
wc -l "$f"
sed -n '1,260p' "$f"
echo
doneRepository: pramodnarayana/soopaedi
Length of output: 25954
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Search the API layer for AS2 remote_url validation or sanitization.
rg -n --hidden --glob '!**/.git/**' -e 'remote_url|validate_target_url|urlparse|http[s]?://|AS2 partner|AS2Partnership|AS2Partner' services/api/src/api services/api/testsRepository: pramodnarayana/soopaedi
Length of output: 17463
🏁 Script executed:
#!/bin/bash
set -euo pipefail
for f in \
services/api/src/api/adapters/http/dtos.py \
services/api/src/api/domain/models.py \
services/api/src/api/core/services/as2_partner_service.py \
services/api/src/api/routers/trading_partners/platform/as2_partners.py \
services/api/src/api/routers/trading_partners/platform/as2_partnerships.py
do
echo "### $f"
wc -l "$f"
sed -n '1,260p' "$f"
echo
doneRepository: pramodnarayana/soopaedi
Length of output: 39018
Add SSRF validation to AS2 delivery. HttpxAS2DeliveryAdapter posts to remote_partner.url directly, and the AS2 partner schema only uses HttpUrl—that still allows loopback/private hosts. Apply the same SSRF guard here or reject unsafe AS2 endpoints at partner creation time.
🤖 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/worker/src/worker/data/main.py` around lines 173 - 176, Apply the
existing SSRF protection used by HttpxDeliveryAdapter to HttpxAS2DeliveryAdapter
by passing validate_target_url when constructing as2_adapter, and ensure the
adapter invokes it before posting to remote_partner.url; alternatively enforce
the same validation during AS2 partner creation.
Comment on lines
+14
to
+17
| data_task = asyncio.create_task(data_main()) | ||
| provision_task = asyncio.create_task(provision_main()) | ||
|
|
||
| await asyncio.gather(data_task, provision_task) |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
No error isolation between concurrent worker tasks.
asyncio.gather without return_exceptions=True cancels the sibling task on the first exception. A transient crash in data_main will kill provision_main (and vice versa), causing unnecessary downtime for the unaffected component. There is also no logging to identify which task failed before the process exits.
🔧 Suggested fix: isolate failures and log before exiting
async def main() -> None:
logger.info("Starting unified Enterprise EDI Worker (Data + Provisioning)...")
- # Run both the Provisioning and Data worker tasks concurrently
- data_task = asyncio.create_task(data_main())
- provision_task = asyncio.create_task(provision_main())
-
- await asyncio.gather(data_task, provision_task)
+ results = await asyncio.gather(
+ asyncio.create_task(data_main()),
+ asyncio.create_task(provision_main()),
+ return_exceptions=True,
+ )
+
+ for name, result in zip(("data", "provision"), results):
+ if isinstance(result, BaseException):
+ logger.error("Worker '%s' exited with exception", name, exc_info=result)
+ raise result🤖 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/worker/src/worker/main.py` around lines 14 - 17, Update the
concurrent task orchestration in the worker entrypoint to isolate failures:
await data_main and provision_main with asyncio.gather(...,
return_exceptions=True), inspect each result, and log an error identifying the
corresponding task and exception before exiting or handling the failure. Keep
one task’s exception from cancelling its sibling.
Comment on lines
+17
to
28
| async def run_worker(service: ProvisioningWorkerService) -> None: | ||
| logger.info("Started polling Database for PROVISION events") | ||
| while True: | ||
| global_gen = db_router.get_global_session() | ||
| global_session = await global_gen.__anext__() | ||
| try: | ||
| # Find a pending provision event | ||
| stmt = ( | ||
| select(GlobalOutbox) | ||
| .where( | ||
| GlobalOutbox.status == "PENDING", | ||
| GlobalOutbox.event_type.in_( | ||
| [ | ||
| "AS2_PARTNER_CREATED", | ||
| "AS2_PARTNERSHIP_CREATED", | ||
| "AS2_PARTNER_UPDATED", | ||
| "AS2_PARTNERSHIP_UPDATED", | ||
| "AS2_PARTNER_DELETED", | ||
| "AS2_PARTNERSHIP_DELETED", | ||
| "SFTP_PARTNER_CREATED", | ||
| "SFTP_PARTNER_UPDATED", | ||
| "SFTP_PARTNER_DELETED", | ||
| "WEBHOOK_CREATED", | ||
| "WEBHOOK_UPDATED", | ||
| "WEBHOOK_DELETED", | ||
| "INBOUND_ROUTE_CREATED", | ||
| "INBOUND_ROUTE_UPDATED", | ||
| "INBOUND_ROUTE_DELETED", | ||
| "OUTBOUND_ROUTE_CREATED", | ||
| "OUTBOUND_ROUTE_UPDATED", | ||
| "OUTBOUND_ROUTE_DELETED", | ||
| ] | ||
| ), | ||
| ) | ||
| .limit(1) | ||
| .with_for_update(skip_locked=True) | ||
| ) | ||
|
|
||
| result = await global_session.execute(stmt) | ||
| outbox_event = result.scalar_one_or_none() | ||
|
|
||
| if outbox_event: | ||
| payload = outbox_event.payload | ||
| tenant_id = payload.get("tenant_id") | ||
|
|
||
| if tenant_id is None: | ||
| logger.error(f"Missing tenant_id in provision event: {outbox_event.id}") | ||
| outbox_event.status = "FAILED" | ||
| await global_session.commit() | ||
| continue | ||
|
|
||
| logger.info(f"Processing provision event for tenant_id={tenant_id}") | ||
|
|
||
| shard_name, shard_dsn = await resolver.resolve(tenant_id) | ||
| tenant_gen = db_router.get_tenant_session(tenant_id, shard_name, shard_dsn) | ||
| tenant_session = await tenant_gen.__anext__() | ||
| try: | ||
| await replicate_tenant_config(tenant_id, global_session, tenant_session) | ||
|
|
||
| # Mark outbox event as processed | ||
| outbox_event.status = "PROCESSED" | ||
| await global_session.commit() | ||
| except (ValueError, KeyError) as e: | ||
| # Permanent data errors: bad payload or missing key — mark FAILED | ||
| await tenant_session.rollback() | ||
| logger.error(f"Permanent provisioning failure for tenant {tenant_id}: {e}") | ||
| outbox_event.status = "FAILED" | ||
| await global_session.commit() | ||
| except Exception as e: | ||
| # Transient errors (network, DB): leave PENDING for retry | ||
| await tenant_session.rollback() | ||
| logger.exception( | ||
| f"Transient error provisioning tenant {tenant_id}, will retry: {e}" | ||
| ) | ||
| # Do NOT change status — let the poller pick it up again | ||
| finally: | ||
| with contextlib.suppress(StopAsyncIteration): | ||
| await tenant_gen.__anext__() | ||
|
|
||
| processed_event = await service.process_next_event() | ||
| # The SQS receive_message already blocks for up to 5 seconds (WaitTimeSeconds=5) | ||
| # We don't need to sleep here if we didn't process an event, but we can do a tiny yield | ||
| if not processed_event: | ||
| await asyncio.sleep(0.1) | ||
| except Exception as e: | ||
| logger.exception(f"Error polling global outbox: {e}") | ||
| logger.exception(f"Error in provisioning loop: {e}") | ||
| await asyncio.sleep(5) |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Remove or update stale SQS comment and consider a longer idle poll interval.
The comment on lines 22-23 references SQS WaitTimeSeconds=5, but this worker now polls a database outbox via ProvisioningWorkerService, not SQS. The comment is misleading and should be removed or updated.
Additionally, asyncio.sleep(0.1) when idle results in ~10 database polls per second. For a provisioning worker that processes configuration changes (low frequency), a 1-second interval would significantly reduce idle database load without meaningfully increasing provisioning latency.
♻️ Proposed fix
async def run_worker(service: ProvisioningWorkerService) -> None:
logger.info("Started polling Database for PROVISION events")
while True:
try:
processed_event = await service.process_next_event()
- # The SQS receive_message already blocks for up to 5 seconds (WaitTimeSeconds=5)
- # We don't need to sleep here if we didn't process an event, but we can do a tiny yield
if not processed_event:
- await asyncio.sleep(0.1)
+ await asyncio.sleep(1)
except Exception as e:
logger.exception(f"Error in provisioning loop: {e}")
await asyncio.sleep(5)📝 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 run_worker(service: ProvisioningWorkerService) -> None: | |
| logger.info("Started polling Database for PROVISION events") | |
| while True: | |
| global_gen = db_router.get_global_session() | |
| global_session = await global_gen.__anext__() | |
| try: | |
| # Find a pending provision event | |
| stmt = ( | |
| select(GlobalOutbox) | |
| .where( | |
| GlobalOutbox.status == "PENDING", | |
| GlobalOutbox.event_type.in_( | |
| [ | |
| "AS2_PARTNER_CREATED", | |
| "AS2_PARTNERSHIP_CREATED", | |
| "AS2_PARTNER_UPDATED", | |
| "AS2_PARTNERSHIP_UPDATED", | |
| "AS2_PARTNER_DELETED", | |
| "AS2_PARTNERSHIP_DELETED", | |
| "SFTP_PARTNER_CREATED", | |
| "SFTP_PARTNER_UPDATED", | |
| "SFTP_PARTNER_DELETED", | |
| "WEBHOOK_CREATED", | |
| "WEBHOOK_UPDATED", | |
| "WEBHOOK_DELETED", | |
| "INBOUND_ROUTE_CREATED", | |
| "INBOUND_ROUTE_UPDATED", | |
| "INBOUND_ROUTE_DELETED", | |
| "OUTBOUND_ROUTE_CREATED", | |
| "OUTBOUND_ROUTE_UPDATED", | |
| "OUTBOUND_ROUTE_DELETED", | |
| ] | |
| ), | |
| ) | |
| .limit(1) | |
| .with_for_update(skip_locked=True) | |
| ) | |
| result = await global_session.execute(stmt) | |
| outbox_event = result.scalar_one_or_none() | |
| if outbox_event: | |
| payload = outbox_event.payload | |
| tenant_id = payload.get("tenant_id") | |
| if tenant_id is None: | |
| logger.error(f"Missing tenant_id in provision event: {outbox_event.id}") | |
| outbox_event.status = "FAILED" | |
| await global_session.commit() | |
| continue | |
| logger.info(f"Processing provision event for tenant_id={tenant_id}") | |
| shard_name, shard_dsn = await resolver.resolve(tenant_id) | |
| tenant_gen = db_router.get_tenant_session(tenant_id, shard_name, shard_dsn) | |
| tenant_session = await tenant_gen.__anext__() | |
| try: | |
| await replicate_tenant_config(tenant_id, global_session, tenant_session) | |
| # Mark outbox event as processed | |
| outbox_event.status = "PROCESSED" | |
| await global_session.commit() | |
| except (ValueError, KeyError) as e: | |
| # Permanent data errors: bad payload or missing key — mark FAILED | |
| await tenant_session.rollback() | |
| logger.error(f"Permanent provisioning failure for tenant {tenant_id}: {e}") | |
| outbox_event.status = "FAILED" | |
| await global_session.commit() | |
| except Exception as e: | |
| # Transient errors (network, DB): leave PENDING for retry | |
| await tenant_session.rollback() | |
| logger.exception( | |
| f"Transient error provisioning tenant {tenant_id}, will retry: {e}" | |
| ) | |
| # Do NOT change status — let the poller pick it up again | |
| finally: | |
| with contextlib.suppress(StopAsyncIteration): | |
| await tenant_gen.__anext__() | |
| processed_event = await service.process_next_event() | |
| # The SQS receive_message already blocks for up to 5 seconds (WaitTimeSeconds=5) | |
| # We don't need to sleep here if we didn't process an event, but we can do a tiny yield | |
| if not processed_event: | |
| await asyncio.sleep(0.1) | |
| except Exception as e: | |
| logger.exception(f"Error polling global outbox: {e}") | |
| logger.exception(f"Error in provisioning loop: {e}") | |
| await asyncio.sleep(5) | |
| async def run_worker(service: ProvisioningWorkerService) -> None: | |
| logger.info("Started polling Database for PROVISION events") | |
| while True: | |
| try: | |
| processed_event = await service.process_next_event() | |
| if not processed_event: | |
| await asyncio.sleep(1) | |
| except Exception as e: | |
| logger.exception(f"Error in provisioning loop: {e}") | |
| await asyncio.sleep(5) |
🤖 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/worker/src/worker/provision/main.py` around lines 17 - 28, In
run_worker, remove or replace the stale SQS WaitTimeSeconds comment with wording
that reflects database outbox polling, and change the idle asyncio.sleep(0.1)
delay to asyncio.sleep(1.0) to reduce unnecessary database polling while
retaining responsive processing.
Comment on lines
+9
to
+73
| async def test_process_next_event_no_event() -> None: | ||
| mock_tenant = AsyncMock() | ||
| mock_outbox = MagicMock() | ||
| mock_replication = AsyncMock() | ||
|
|
||
| # Setup the async context manager to yield None | ||
| from contextlib import asynccontextmanager | ||
|
|
||
| @asynccontextmanager | ||
| async def fake_cm(): | ||
| yield None | ||
|
|
||
| mock_outbox.process_next_event.return_value = fake_cm() | ||
|
|
||
| svc = ProvisioningWorkerService(mock_tenant, mock_outbox, mock_replication) | ||
| result = await svc.process_next_event() | ||
| assert result is False | ||
| mock_replication.replicate_tenant_configuration.assert_not_called() | ||
|
|
||
|
|
||
| async def test_process_next_event_tenant_specific() -> None: | ||
| mock_tenant = AsyncMock() | ||
| mock_outbox = MagicMock() | ||
| mock_replication = AsyncMock() | ||
|
|
||
| mock_event = MagicMock() | ||
| mock_event.payload = {"tenant_id": 99} | ||
|
|
||
| from contextlib import asynccontextmanager | ||
|
|
||
| @asynccontextmanager | ||
| async def fake_cm(): | ||
| yield mock_event | ||
|
|
||
| mock_outbox.process_next_event.return_value = fake_cm() | ||
|
|
||
| svc = ProvisioningWorkerService(mock_tenant, mock_outbox, mock_replication) | ||
| result = await svc.process_next_event() | ||
| assert result is True | ||
| mock_replication.replicate_tenant_configuration.assert_awaited_once_with(99) | ||
|
|
||
|
|
||
| async def test_process_next_event_global() -> None: | ||
| mock_tenant = AsyncMock() | ||
| mock_outbox = MagicMock() | ||
| mock_replication = AsyncMock() | ||
|
|
||
| mock_event = MagicMock() | ||
| mock_event.payload = {"tenant_id": 0} | ||
|
|
||
| from contextlib import asynccontextmanager | ||
|
|
||
| @asynccontextmanager | ||
| async def fake_cm(): | ||
| yield mock_event | ||
|
|
||
| mock_outbox.process_next_event.return_value = fake_cm() | ||
|
|
||
| mock_tenant.get_all_tenant_ids.return_value = [1, 2, 3] | ||
|
|
||
| svc = ProvisioningWorkerService(mock_tenant, mock_outbox, mock_replication) | ||
| result = await svc.process_next_event() | ||
|
|
||
| assert result is True | ||
| assert mock_replication.replicate_tenant_configuration.call_count == 3 |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Add tests for error and edge-case scenarios.
The three tests cover happy paths only. Missing coverage:
replicate_tenant_configurationraisesPermanentProvisioningError— verify message is deleted and behavior is correct.replicate_tenant_configurationraises a transient exception — verify message is left on queue.payloadmissingtenant_idkey — verify graceful handling.
These paths are critical given the exception-handling logic in SqsOutboxAdapter.
🤖 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/worker/tests/test_provision_worker.py` around lines 9 - 73, Add
tests alongside test_process_next_event_no_event,
test_process_next_event_tenant_specific, and test_process_next_event_global for
the requested edge cases: configure replicate_tenant_configuration to raise
PermanentProvisioningError and assert the message is deleted with the expected
behavior; configure a transient exception and assert the message remains queued;
and provide an event payload without tenant_id, verifying process_next_event
handles it gracefully. Use mocks to assert the relevant SqsOutboxAdapter message
deletion/retention interactions.
Comment on lines
+39
to
+42
| # Get the tenant name from somewhere? For now, we can pass a dummy or fetch it. | ||
| # We should ideally fetch the tenant name to use in the prefix. | ||
| # For now, using a generic placeholder as the slug generator handles it. | ||
| tenant_name = f"tenant{tenant_id}" |
Contributor
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Address the placeholder tenant_name and TODO comments.
The tenant_name = f"tenant{tenant_id}" placeholder produces generic token prefixes (e.g., tenant1-...) instead of using the actual tenant name. The inline comments acknowledge this needs fixing. Would you like me to generate a solution that fetches the tenant name from the tenant repository and passes it to the token service?
🤖 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/developers/api_tokens.py` around lines 39 - 42,
Replace the placeholder tenant_name logic and remove the TODO comments by
fetching the actual tenant name from the tenant repository using tenant_id, then
pass that value to the token service’s prefix/slug generation logic. Handle a
missing tenant consistently with existing repository conventions before
generating the token.
Comment on lines
+44
to
+55
| cmd = CreateApiTokenCmd(name=request.name, expires_at=request.expires_at) | ||
|
|
||
| token = await service.create_token(tenant_id=tenant_id, tenant_name=tenant_name, cmd=cmd) | ||
|
|
||
| return ApiTokenCreatedResponse( | ||
| id=token.id, | ||
| name=token.name, | ||
| client_id=token.client_id, | ||
| client_secret=token.client_secret, | ||
| active=token.active, | ||
| created_at="just now", # This is usually handled by the DB, but since we return the entity immediately, we need a value. Let's just return the current ISO string if needed, or rely on the UI to just know it's new. Actually, we should probably fetch the record or just return a fresh datetime. | ||
| ) |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Replace created_at="just now" placeholder with a real timestamp.
If ApiTokenCreatedResponse.created_at is typed as datetime, FastAPI will raise a ResponseValidationError at runtime when serializing "just now". If it's str, the frontend receives an unparseable value. Either way, this breaks the API contract.
🐛 Proposed fix
+from datetime import UTC, datetime
+
...
return ApiTokenCreatedResponse(
id=token.id,
name=token.name,
client_id=token.client_id,
client_secret=token.client_secret,
active=token.active,
- created_at="just now",
+ created_at=token.created_at if hasattr(token, "created_at") and token.created_at else datetime.now(UTC).isoformat(),
)If the token entity has created_at populated after DB insert, use token.created_at directly. Otherwise, datetime.now(UTC).isoformat() is a safe fallback.
📝 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
| cmd = CreateApiTokenCmd(name=request.name, expires_at=request.expires_at) | |
| token = await service.create_token(tenant_id=tenant_id, tenant_name=tenant_name, cmd=cmd) | |
| return ApiTokenCreatedResponse( | |
| id=token.id, | |
| name=token.name, | |
| client_id=token.client_id, | |
| client_secret=token.client_secret, | |
| active=token.active, | |
| created_at="just now", # This is usually handled by the DB, but since we return the entity immediately, we need a value. Let's just return the current ISO string if needed, or rely on the UI to just know it's new. Actually, we should probably fetch the record or just return a fresh datetime. | |
| ) | |
| from datetime import UTC, datetime | |
| cmd = CreateApiTokenCmd(name=request.name, expires_at=request.expires_at) | |
| token = await service.create_token(tenant_id=tenant_id, tenant_name=tenant_name, cmd=cmd) | |
| return ApiTokenCreatedResponse( | |
| id=token.id, | |
| name=token.name, | |
| client_id=token.client_id, | |
| client_secret=token.client_secret, | |
| active=token.active, | |
| created_at=token.created_at if hasattr(token, "created_at") and token.created_at else datetime.now(UTC).isoformat(), | |
| ) |
🤖 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/developers/api_tokens.py` around lines 44 - 55,
Replace the `"just now"` placeholder in the `ApiTokenCreatedResponse` returned
by the token-creation handler with `token.created_at` when available; otherwise
use the current UTC timestamp in the format expected by
`ApiTokenCreatedResponse.created_at`, ensuring the response remains type-safe
and parseable.
Comment on lines
+88
to
+101
| @router.delete( | ||
| "/{token_id}/hard", | ||
| status_code=status.HTTP_204_NO_CONTENT, | ||
| ) | ||
| async def delete_api_token( | ||
| token_id: UUID, | ||
| tenant_id: int = Depends(get_current_tenant_id), | ||
| repo: ApiTokenRepositoryPort = Depends(get_api_token_repo), | ||
| ) -> None: | ||
| """Permanently delete an API token.""" | ||
| service = ApiTokenService(repo) | ||
| success = await service.delete_token(tenant_id, token_id) | ||
| if not success: | ||
| raise HTTPException(status_code=404, detail="Token not found") |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how other destructive endpoints handle authorization
rg -n 'AuthorizationService\|require_admin\|require_role' --type py services/api/src/api/routers/ | head -20Repository: pramodnarayana/soopaedi
Length of output: 161
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== api_tokens.py outline ==\n'
ast-grep outline services/api/src/api/routers/developers/api_tokens.py --view expanded || true
printf '\n== api_tokens.py relevant lines ==\n'
sed -n '1,180p' services/api/src/api/routers/developers/api_tokens.py | cat -n
printf '\n== dependencies.py auth-related symbols ==\n'
ast-grep outline services/api/src/api/dependencies.py --view expanded || true
printf '\n== search for authz helpers in API routers ==\n'
rg -n 'require_platform_admin|require_admin|require_role|AuthorizationService|Depends\\(' services/api/src/api/routers services/api/src/api/dependencies.py || trueRepository: pramodnarayana/soopaedi
Length of output: 5656
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== dependencies.py excerpt ==\n'
sed -n '96,145p' services/api/src/api/dependencies.py | cat -n
printf '\n== api_token_service excerpt ==\n'
ast-grep outline services/api/src/api/core/services/api_token_service.py --view expanded || true
sed -n '1,220p' services/api/src/api/core/services/api_token_service.py | cat -n
printf '\n== authz usage around token routes ==\n'
rg -n 'require_platform_admin|get_authorization_service|AuthorizationService|api token|delete_token\\(|revoke_token\\(' services/api/src | head -200Repository: pramodnarayana/soopaedi
Length of output: 7919
Gate hard delete with platform-admin auth
/{token_id}/hard only depends on get_current_tenant_id, so any authenticated tenant user can reach the irreversible delete path. require_platform_admin() already exists in services/api/src/api/dependencies.py; use it here if hard delete should be limited to elevated users.
🤖 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/developers/api_tokens.py` around lines 88 - 101,
The hard-delete endpoint currently permits any authenticated tenant user. Update
delete_api_token to add require_platform_admin() as a dependency, while
retaining tenant resolution and repository injection, so only platform
administrators can invoke the irreversible deletion.
Comment on lines
+26
to
+36
| async def get_secret(self, vault_ref: str) -> str: | ||
| # HVAC is synchronous, but we wrap in asyncio.to_thread for the port | ||
| def _fetch() -> str: | ||
| resp = self.client.secrets.kv.v2.read_secret_version( | ||
| path=vault_ref, mount_point=self.mount_point | ||
| ) | ||
| data = resp.get("data", {}).get("data", {}) | ||
| val = next(iter(data.values()), None) | ||
| return val if isinstance(val, str) else "" | ||
|
|
||
| return await asyncio.to_thread(_fetch) |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target file and nearby callers/usages.
git ls-files services/worker/src/worker/adapters/vault.py
ast-grep outline services/worker/src/worker/adapters/vault.py --view expanded
printf '\n## Usages of get_secret\n'
rg -n "get_secret\(" services/worker -S
printf '\n## Read target file with line numbers\n'
cat -n services/worker/src/worker/adapters/vault.pyRepository: pramodnarayana/soopaedi
Length of output: 2028
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target file's surrounding context in smaller chunks if needed.
wc -l services/worker/src/worker/adapters/vault.py
sed -n '1,220p' services/worker/src/worker/adapters/vault.pyRepository: pramodnarayana/soopaedi
Length of output: 1405
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check the implementation of the Vault client interface and secret consumers.
rg -n "mount_point|Vault|secrets.kv.v2.read_secret_version|vault_ref|secret" services/worker/src -SRepository: pramodnarayana/soopaedi
Length of output: 2779
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the worker entrypoints and any secret-consuming code paths.
sed -n '150,230p' services/worker/src/worker/data/main.py
printf '\n## Secret field definitions / consumers\n'
rg -n "credentials_vault_ref|auth_header_vault_ref|public_cert_vault_ref|private_key_vault_ref|prev_public_cert_vault_ref|prev_private_key_vault_ref|get_secret\(" services/worker/src -SRepository: pramodnarayana/soopaedi
Length of output: 4883
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect schema/models for vault reference fields to infer expected secret shapes.
rg -n "credentials_vault_ref|auth_header_vault_ref|public_cert_vault_ref|private_key_vault_ref|prev_public_cert_vault_ref|prev_private_key_vault_ref" services/worker/src -SRepository: pramodnarayana/soopaedi
Length of output: 2110
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find actual call sites and the expected shapes of secrets retrieved from Vault.
rg -n "\.get_secret\(|vault\." services -S
printf '\n## DeliveryService and related secret usage\n'
rg -n "class DeliveryService|get_secret|credentials|auth_header|private_key|public_cert" services/worker/src -SRepository: pramodnarayana/soopaedi
Length of output: 5484
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the implementation and any helper that reads secret values.
ast-grep outline services/worker/src/worker -r --view expanded 2>/dev/null || trueRepository: pramodnarayana/soopaedi
Length of output: 161
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate DeliveryService and secret-related classes.
fd -a "delivery_service.py|service.py|*.py" services/worker/src | sed -n '1,120p'Repository: pramodnarayana/soopaedi
Length of output: 507
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find the DeliveryService definition and inspect nearby secret handling.
service_file=$(rg -l "class DeliveryService" services/worker/src || true)
printf 'SERVICE_FILE=%s\n' "${service_file:-}"
if [ -n "${service_file:-}" ]; then
wc -l "$service_file"
sed -n '1,260p' "$service_file"
fiRepository: pramodnarayana/soopaedi
Length of output: 175
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Search the whole repo for get_secret usage and vault adapter injection.
rg -n "get_secret\(|WorkerVaultAdapter|vault_adapter|vault=" . -SRepository: pramodnarayana/soopaedi
Length of output: 1780
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect references to vault-backed credential fields and their consumers.
rg -n "credentials_vault_ref|auth_header_vault_ref|public_cert_vault_ref|private_key_vault_ref|prev_public_cert_vault_ref|prev_private_key_vault_ref|auth_header|private_key|public_cert" services -SRepository: pramodnarayana/soopaedi
Length of output: 27118
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find the domain/service code that consumes Vault-backed refs and any empty-string handling.
rg -n "vault_ref|get_secret|WorkerVaultAdapter|empty string|not a string|isinstance\(.*str\)" . -SRepository: pramodnarayana/soopaedi
Length of output: 26924
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Look for tests or docs around WorkerVaultAdapter secret semantics.
rg -n "WorkerVaultAdapter|get_secret\(|Vault" tests services/worker -SRepository: pramodnarayana/soopaedi
Length of output: 569
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the DeliveryService implementation.
service_file=$(rg -l "class DeliveryService" services/worker/src . -S || true)
printf 'SERVICE_FILE=%s\n' "${service_file:-}"
if [ -n "${service_file:-}" ]; then
wc -l "$service_file"
sed -n '1,260p' "$service_file"
fiRepository: pramodnarayana/soopaedi
Length of output: 11542
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check if any tests encode the expected behavior for Vault secrets.
rg -n "secret.*string|missing or not a string|VAULT_TOKEN|required in non-dev|read_secret_version" tests services -SRepository: pramodnarayana/soopaedi
Length of output: 1108
Fail fast on missing or non-string Vault values. Returning "" here hides a Vault misconfiguration and turns it into a hard-to-debug auth failure; raise an error or log a warning instead.
🤖 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/worker/src/worker/adapters/vault.py` around lines 26 - 36, Update
VaultAdapter.get_secret and its nested _fetch function to fail clearly when the
response contains no value or the first value is not a string, instead of
returning an empty string. Raise an appropriate descriptive error that
identifies the invalid or missing Vault secret.
… path testing code review
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
services/api/src/api/adapters/repository.py (1)
117-124: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winScope tenant-scoped writes to tenant-owned partners only.
get_as2_partneris shared byupdate_as2_identityandrotate_as2_certificates, so thetenant_id=0fallback lets a tenant mutate global AS2 partners. Keep the shared read path, but use a write-specific lookup or admin-only guard for these mutations.🤖 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 117 - 124, Keep get_as2_partner’s shared read behavior, but prevent update_as2_identity and rotate_as2_certificates from using its tenant_id=0 fallback for writes. Add a write-specific partner lookup restricted to the requesting tenant, or enforce an equivalent admin-only guard, and route both mutation methods through it while preserving global-partner access for permitted reads.services/api/src/api/services/outbound_service.py (1)
116-127: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Nonetransaction_typepassed to envelope builder produces invalid EDI.When
transaction_typecan't be determined (lines 57-74), it remainsNoneand is placed intoroute_configat line 119. The envelope builders callroute_config.get("transaction_type", "UNKNOWN"), which returnsNone(not"UNKNOWN") because the key exists with valueNone. This results inNonebeing used asST01in X12 andUNH02.01in EDIFACT, producing invalid EDI segments. The metadata extractor on line 83 correctly usestransaction_type or "", but the route_config does not apply the same fallback.🔧 Proposed fix
route_config = { "default_standard": route.default_standard, "default_version": route.default_version, - "transaction_type": transaction_type, + "transaction_type": transaction_type or "UNKNOWN", "isa_sender_qualifier": route.isa_sender_qualifier,🤖 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/services/outbound_service.py` around lines 116 - 127, The route_config construction must prevent a missing transaction type from reaching the envelope builders as None. Update the transaction_type value in the route_config assembled by the outbound service to use the same empty-string fallback already applied by the metadata extractor, while preserving determined transaction types.libs/transformer/src/transformer/domain/envelope/edifact.py (1)
76-76: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse monotonic counter for
unb05to match X12 builder and avoid collisions.
random.randint(1, 999999999)for the EDIFACT interchange control reference (UNB05) has the same birthday-paradox collision risk that was fixed in the X12 builder (lines 114-116 ofx12.py). Two interchanges within the same trading partner relationship sharing a UNB05 can cause rejection by the receiving partner. The X12 builder was updated to useint(now.timestamp() * 1000) % 1000000000in this PR, but the EDIFACT builder was not — apply the same approach for consistency.🔧 Proposed fix
- unb05 = str(random.randint(1, 999999999)) + monotonic_counter = int(now.timestamp() * 1000) % 1000000000 + unb05 = str(monotonic_counter)🤖 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/transformer/src/transformer/domain/envelope/edifact.py` at line 76, Replace the random UNB05 generation in the EDIFACT envelope builder with the same monotonic timestamp-based control reference used by the X12 builder, using the existing current-time value and millisecond timestamp modulo 1,000,000,000. Keep the result within the existing nine-digit range and align the implementation with the X12 control-reference logic.services/api/src/api/services/as2_receive_service.py (1)
366-387: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAdd explicit
rollback()on error path for tenant session.If
create_edi_messageorcreate_outbox_eventraises, thefinallyblock only callsasync_gen_tenant.aclose(). Without an explicitrollback(), the session's uncommitted transaction state depends entirely on the generator's cleanup behavior. A failedcommit()also leaves the session in an error state that requires rollback before disposal.🔒️ Proposed fix
try: dp_repo = SqlAlchemyDataPlaneRepository(tenant_session) msg_id = await dp_repo.create_edi_message(tenant_id=true_tenant_id, payload=edi_record) outbox_payload = { "edi_message_id": str(msg_id), "sender_id": as2_msg.as2_from, "receiver_id": as2_msg.as2_to, "status": "RECEIVED", } await dp_repo.create_outbox_event( tenant_id=true_tenant_id, event_type="edi_message.received", payload=outbox_payload, ) await tenant_session.commit() return str(msg_id) + except Exception: + await tenant_session.rollback() + raise finally: await async_gen_tenant.aclose()🤖 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/services/as2_receive_service.py` around lines 366 - 387, Update the tenant-session transaction handling around SqlAlchemyDataPlaneRepository operations so any exception from create_edi_message, create_outbox_event, or commit explicitly rolls back tenant_session before closing async_gen_tenant; preserve the existing commit-and-return path on success.services/api/src/api/core/services/route_service.py (1)
116-130: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winParallelize partner name lookups with
asyncio.gather.The three
awaitcalls for AS2, SFTP, and webhook name resolution are independent. Sequential execution adds unnecessary latency to thelist_routesendpoint.♻️ Proposed refactor
+ import asyncio + + results_tasks = await asyncio.gather( + self.global_repo.get_as2_partners_by_ids(tenant_id, list(as2_ids)) + if as2_ids + else _empty_dict(), + self.global_repo.get_sftp_partners_by_ids(tenant_id, list(sftp_ids)) + if sftp_ids + else _empty_dict(), + self.global_repo.get_webhooks_by_ids(tenant_id, list(webhook_ids)) + if webhook_ids + else _empty_dict(), + ) + as2_names, sftp_names, webhook_names = results_tasksOr more readably with helper coroutines:
+ import asyncio + + async def _or_empty(coro, ids): + return await coro if ids else {} + + as2_names, sftp_names, webhook_names = await asyncio.gather( + _or_empty(self.global_repo.get_as2_partners_by_ids(tenant_id, list(as2_ids)), as2_ids), + _or_empty(self.global_repo.get_sftp_partners_by_ids(tenant_id, list(sftp_ids)), sftp_ids), + _or_empty(self.global_repo.get_webhooks_by_ids(tenant_id, list(webhook_ids)), webhook_ids), + )🤖 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/core/services/route_service.py` around lines 116 - 130, Update the partner name lookups in list_routes to execute the independent AS2, SFTP, and webhook repository calls concurrently via asyncio.gather, while preserving the existing empty-ID fallback dictionaries and assigning each result to its corresponding name mapping.
♻️ Duplicate comments (1)
services/api/src/api/cdc_relay.py (1)
64-72: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winNon-mapping batch element still throws uncaught
TypeError.
DebeziumUnwrappedEvent(**raw_event)raisesTypeError(notValidationError) whenraw_eventis not a dict (e.g., a string or number in a JSON array). Theexcept ValidationErroron line 67 won't catch it, so it escapes uncaught → 500 → Debezium re-delivers the same batch indefinitely (poison loop). This was flagged in a previous review and remains unfixed.🛠️ Proposed fix
for raw_event in raw_events: + if not isinstance(raw_event, dict): + logger.error(f"[CDC Relay] Skipping non-object event: {raw_event!r}") + continue try: validated_events.append(DebeziumUnwrappedEvent(**raw_event)) except ValidationError as e:🤖 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 64 - 72, Update the validation handling in the raw_events loop around DebeziumUnwrappedEvent so non-mapping elements cannot raise an uncaught TypeError. Catch and log the invalid element alongside ValidationError, then continue processing the remaining batch without propagating the exception.
🤖 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 `@libs/database/src/database/repository.py`:
- Around line 67-81: The get_inbound_route method can raise MultipleResultsFound
and may select an unintended cross-tenant route because its query lacks
transaction_type and deterministic selection. Update the InboundRoute lookup to
include the appropriate transaction_type constraint when available, and make the
result deterministic with ordering and a single-row limit while preserving the
existing tenant filtering behavior.
In `@libs/identity/docker/docker-compose.yml`:
- Line 45: Update the volume declaration for the zitadel service so /machinekey
uses a dedicated named volume instead of zitadel_data, and add that new volume
to the compose file’s top-level volumes section while leaving the Postgres data
mount unchanged.
In `@services/api/src/api/adapters/repository.py`:
- Around line 848-851: Throttle the last-used timestamp update in the
authentication flow around the ApiToken update so successful requests do not
write on every authentication. Only update last_used_at when the stored value is
older than the configured interval, preserving the existing tenant_id return
behavior and avoiding unnecessary row updates.
In `@services/worker/src/worker/core/service.py`:
- Around line 47-59: Update the global broadcast result handling around the
errors collection to distinguish PermanentProvisioningError from transient
exceptions. Log permanent failures and continue without adding them to the
retryable errors, while retaining only transient failures in errors so
TransientProvisioningError is raised only when retryable failures remain.
---
Outside diff comments:
In `@libs/transformer/src/transformer/domain/envelope/edifact.py`:
- Line 76: Replace the random UNB05 generation in the EDIFACT envelope builder
with the same monotonic timestamp-based control reference used by the X12
builder, using the existing current-time value and millisecond timestamp modulo
1,000,000,000. Keep the result within the existing nine-digit range and align
the implementation with the X12 control-reference logic.
In `@services/api/src/api/adapters/repository.py`:
- Around line 117-124: Keep get_as2_partner’s shared read behavior, but prevent
update_as2_identity and rotate_as2_certificates from using its tenant_id=0
fallback for writes. Add a write-specific partner lookup restricted to the
requesting tenant, or enforce an equivalent admin-only guard, and route both
mutation methods through it while preserving global-partner access for permitted
reads.
In `@services/api/src/api/core/services/route_service.py`:
- Around line 116-130: Update the partner name lookups in list_routes to execute
the independent AS2, SFTP, and webhook repository calls concurrently via
asyncio.gather, while preserving the existing empty-ID fallback dictionaries and
assigning each result to its corresponding name mapping.
In `@services/api/src/api/services/as2_receive_service.py`:
- Around line 366-387: Update the tenant-session transaction handling around
SqlAlchemyDataPlaneRepository operations so any exception from
create_edi_message, create_outbox_event, or commit explicitly rolls back
tenant_session before closing async_gen_tenant; preserve the existing
commit-and-return path on success.
In `@services/api/src/api/services/outbound_service.py`:
- Around line 116-127: The route_config construction must prevent a missing
transaction type from reaching the envelope builders as None. Update the
transaction_type value in the route_config assembled by the outbound service to
use the same empty-string fallback already applied by the metadata extractor,
while preserving determined transaction types.
---
Duplicate comments:
In `@services/api/src/api/cdc_relay.py`:
- Around line 64-72: Update the validation handling in the raw_events loop
around DebeziumUnwrappedEvent so non-mapping elements cannot raise an uncaught
TypeError. Catch and log the invalid element alongside ValidationError, then
continue processing the remaining batch without propagating the exception.
🪄 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: dd6b9a26-594d-4abe-be12-e90eaa7cf547
📒 Files selected for processing (39)
TECHNICAL_DEBT.mddocker/localstack/init-aws.shfrontend/web/src/components/ui/alert.tsxfrontend/web/src/components/ui/dropdown-menu.tsxfrontend/web/src/features/developers/api/apiTokenHooks.tsfrontend/web/src/features/developers/components/ApiTokensTable.tsxfrontend/web/src/features/developers/components/TokenCredentialsModal.tsxfrontend/web/src/features/routes/components/CreateRouteModal.tsxlibs/database/src/database/migrations/global/versions/94b943b2920d_global_initial_schema.pylibs/database/src/database/models/common.pylibs/database/src/database/models/control_plane.pylibs/database/src/database/models/data_plane.pylibs/database/src/database/models/replicated_mixins.pylibs/database/src/database/repository.pylibs/identity/docker/docker-compose.ymllibs/pipeline/src/pipeline/adapters/http.pylibs/pipeline/src/pipeline/adapters/transformer.pylibs/pipeline/src/pipeline/core/translate.pylibs/pipeline/tests/test_http.pylibs/transformer/src/transformer/domain/envelope/edifact.pylibs/transformer/src/transformer/domain/envelope/x12.pyservices/api/src/api/adapters/http/dtos.pyservices/api/src/api/adapters/repository.pyservices/api/src/api/auth/api_key.pyservices/api/src/api/cdc_relay.pyservices/api/src/api/core/services/route_service.pyservices/api/src/api/domain/models.pyservices/api/src/api/ports/repository.pyservices/api/src/api/routers/routes.pyservices/api/src/api/routers/trading_partners/as2.pyservices/api/src/api/services/as2_receive_service.pyservices/api/src/api/services/outbound_service.pyservices/api/tests/test_cdc_relay.pyservices/worker/src/worker/adapters/db_replication.pyservices/worker/src/worker/adapters/sqs_outbox.pyservices/worker/src/worker/adapters/vault.pyservices/worker/src/worker/core/service.pyservices/worker/src/worker/data/main.pyservices/worker/tests/test_provision_worker.py
💤 Files with no reviewable changes (5)
- libs/pipeline/src/pipeline/adapters/transformer.py
- libs/pipeline/src/pipeline/core/translate.py
- libs/database/src/database/migrations/global/versions/94b943b2920d_global_initial_schema.py
- libs/database/src/database/models/data_plane.py
- libs/database/src/database/models/control_plane.py
… path testing code review
Owner
Author
|
@coderabbitai review |
Contributor
✅ Action performedReview finished.
|
Contributor
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
services/api/src/api/adapters/repository.py (2)
117-124: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftKeep control-plane and data-plane
AS2Partnertenant semantics separate. The production lookup and its test both apply data-plane assumptions to the control-plane model.
services/api/src/api/adapters/repository.py#L117-L124: match shared control-plane partners usingtenant_id IS NULL, not tenant zero.services/api/tests/test_api_repository.py#L189-L204: construct the control-plane model and assert the generated query contains the tenant predicate.🤖 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 117 - 124, The get_as2_partner lookup in services/api/src/api/adapters/repository.py lines 117-124 must treat shared control-plane AS2Partner records as tenant_id IS NULL rather than tenant_id 0; update its tenant predicate accordingly. In services/api/tests/test_api_repository.py lines 189-204, construct the control-plane model and assert the generated query includes the NULL tenant predicate.
843-866: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winUse naive UTC here or migrate
ApiTokentimestamps.api_tokens.last_used_atandapi_tokens.expires_atare still naiveDateTime, sodatetime.now(UTC)can trip timezone-mismatch errors in the lookup/update path oncelast_used_atis set. The smallest schema-compatible fix isdatetime.now(UTC).replace(tzinfo=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/repository.py` around lines 843 - 866, Update the timestamp initialization in the ApiToken lookup/update flow to use a naive UTC datetime via datetime.now(UTC).replace(tzinfo=None), keeping comparisons against expires_at and last_used_at schema-compatible. Preserve the existing expiration filtering and hourly last_used_at update behavior.services/api/src/api/services/outbound_service.py (1)
56-76: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDo not continue with an empty transaction type.
When payload inference fails, Line 119 emits an empty transaction identifier while the persisted record still receives
None. Fall back toroute.transaction_type, then reject the request if no type is available.Proposed fix
if not transaction_type: # Try ST segment directly (for raw transaction payloads) st = first_payload.get("ST", {}) if st: transaction_type = st.get("ST01") + transaction_type = transaction_type or route.transaction_type if not transaction_type: - logger.warning( - "Could not determine transaction_type from payload, metadata extraction might fail." - ) + raise ValueError("Could not determine transaction_type") ... - "transaction_type": transaction_type or "", + "transaction_type": transaction_type,Also applies to: 116-120
🤖 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/services/outbound_service.py` around lines 56 - 76, Update the transaction type resolution in the outbound service after payload inference to fall back to route.transaction_type when no type was found. Before emitting the transaction identifier or persisting the record, reject the request if transaction_type remains unavailable, ensuring neither path proceeds with an empty value.services/api/src/api/cdc_relay.py (1)
51-72: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftQuarantine rejected CDC records before acknowledging them.
Every validation failure is skipped and the endpoint still returns success, permanently dropping the corresponding outbox event. Persist the raw event to a DLQ/quarantine store before continuing; if quarantine fails, return an error so Debezium retries.
Also applies to: 80-94, 110-116
🤖 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 51 - 72, Update the CDC relay validation flow around DebeziumUnwrappedEvent so every rejected raw_event is persisted to the DLQ/quarantine store before continuing. If quarantine succeeds, retain the current skip behavior; if it fails, return an error response instead of acknowledging the batch so Debezium retries. Apply the same quarantine-before-acknowledgment handling to the related failure paths at the indicated locations.
🤖 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 `@libs/database/tests/test_database_repository.py`:
- Around line 62-86: Extend the tests for
InboundRouteRepository.get_inbound_route with a no-match case where
scalar_one_or_none returns None and assert the method returns None. Also add
coverage for tenant_id=None and transaction_type=None to verify optional filters
are handled correctly, reusing the existing mock session setup.
In `@libs/identity/docker/docker-compose.yml`:
- Around line 45-54: Update the Zitadel service environment configuration in the
start-from-init setup to set ZITADEL_FIRSTINSTANCE_MACHINEKEYPATH to the mounted
/machinekey location, ensuring generated machine keys persist in the
zitadel_machinekey volume. Also set
ZITADEL_FIRSTINSTANCE_ORG_MACHINE_MACHINEKEY_TYPE to JSON when the Terraform
machine account requires JSON output.
In `@libs/transformer/src/transformer/domain/envelope/edifact.py`:
- Line 75: Replace the timestamp-and-modulo generation in the UNB05 construction
with a sender-scoped sequence or another collision-resistant identifier that
remains unique for concurrent requests and does not repeat on a short cycle.
Preserve the required UNB05 string format and update the surrounding
envelope-building logic only as needed to obtain the identifier.
In `@services/api/src/api/adapters/repository.py`:
- Around line 516-528: Update get_inbound_route to require and propagate a
tenant_id for inbound route lookups, preventing tenant-less queries from
matching another tenant’s route. If tenant-less lookup must remain supported,
first enforce a globally unique discovery key before delegating to
InboundRouteRepository.get_inbound_route.
In `@services/api/tests/test_api_repository.py`:
- Around line 189-204: Update the test around
SqlAlchemyControlPlaneRepository.get_as2_partner_for_write to return
database.models.control_plane.AS2Partner and verify the executed statement
targets the control-plane model. Inspect mock_session.execute’s statement to
assert predicates for both the requested partner ID and tenant_id, rather than
only asserting the result is non-null.
---
Outside diff comments:
In `@services/api/src/api/adapters/repository.py`:
- Around line 117-124: The get_as2_partner lookup in
services/api/src/api/adapters/repository.py lines 117-124 must treat shared
control-plane AS2Partner records as tenant_id IS NULL rather than tenant_id 0;
update its tenant predicate accordingly. In
services/api/tests/test_api_repository.py lines 189-204, construct the
control-plane model and assert the generated query includes the NULL tenant
predicate.
- Around line 843-866: Update the timestamp initialization in the ApiToken
lookup/update flow to use a naive UTC datetime via
datetime.now(UTC).replace(tzinfo=None), keeping comparisons against expires_at
and last_used_at schema-compatible. Preserve the existing expiration filtering
and hourly last_used_at update behavior.
In `@services/api/src/api/cdc_relay.py`:
- Around line 51-72: Update the CDC relay validation flow around
DebeziumUnwrappedEvent so every rejected raw_event is persisted to the
DLQ/quarantine store before continuing. If quarantine succeeds, retain the
current skip behavior; if it fails, return an error response instead of
acknowledging the batch so Debezium retries. Apply the same
quarantine-before-acknowledgment handling to the related failure paths at the
indicated locations.
In `@services/api/src/api/services/outbound_service.py`:
- Around line 56-76: Update the transaction type resolution in the outbound
service after payload inference to fall back to route.transaction_type when no
type was found. Before emitting the transaction identifier or persisting the
record, reject the request if transaction_type remains unavailable, ensuring
neither path proceeds with an empty value.
🪄 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: ccf81d28-ac1e-437d-aa9b-25c58cb0b38d
📒 Files selected for processing (10)
libs/database/src/database/repository.pylibs/database/tests/test_database_repository.pylibs/identity/docker/docker-compose.ymllibs/transformer/src/transformer/domain/envelope/edifact.pyservices/api/src/api/adapters/repository.pyservices/api/src/api/cdc_relay.pyservices/api/src/api/ports/repository.pyservices/api/src/api/services/outbound_service.pyservices/api/tests/test_api_repository.pyservices/worker/src/worker/core/service.py
… path testing code review
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
libs/transformer/src/transformer/domain/envelope/edifact.py (1)
66-90: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove
import uuidto module level.The
import uuidstatement insidebuild()executes on every call. While Python caches module imports, it's still a code smell and should be at the top of the file with the other imports.♻️ Proposed fix
Move the import to the top of the file (after the existing imports shown in lines 1-7):
+import uuid + class EdifactEnvelopeBuilder(BaseEnvelopeBuilder):And remove the inline import:
# Generation values - import uuid - unb05 = str(uuid.uuid4().int % 1000000000).zfill(9)🤖 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/transformer/src/transformer/domain/envelope/edifact.py` around lines 66 - 90, Move the uuid import to the module-level import section of the EDIFACT envelope module, and remove the inline import from the build classmethod while preserving its existing UUID generation behavior.services/api/src/api/services/as2_receive_service.py (1)
371-381: 🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick winPropagate
trace_idinto the inbound outbox payload
edi_message.receivedis routed through the TRANSLATE path and dropped whentrace_idis missing, so inbound AS2 messages never reach translation. Addedi_record["trace_id"]to this payload, matching the outbound flow.🤖 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/services/as2_receive_service.py` around lines 371 - 381, Update the inbound outbox payload in the AS2 receive flow to include the existing edi_record["trace_id"] value, matching the outbound event payload. Preserve the current edi_message.received event fields and routing behavior while ensuring trace_id is propagated for translation.services/api/src/api/ports/repository.py (1)
80-86: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAlign the no-filters test with the non-null tenant contract
get_inbound_routenow requirestenant_id: int, and theInboundRoutemodel storestenant_idas non-nullable. Updatetest_inbound_route_repository_get_inbound_route_no_filtersto pass a real tenant id, or the test will keep masking the contract mismatch.🤖 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/repository.py` around lines 80 - 86, Update test_inbound_route_repository_get_inbound_route_no_filters to pass a valid integer tenant_id when calling get_inbound_route, matching the required non-null tenant contract and InboundRoute model.
🤖 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 `@libs/database/tests/test_database_repository.py`:
- Around line 104-127: Update
test_inbound_route_repository_get_inbound_route_no_filters to pass a valid
tenant_id such as 1 instead of None, matching
InboundRouteRepository.get_inbound_route’s required tenant_id filter. Rename the
test to indicate that only transaction_type is unfiltered, while preserving the
existing assertions and setup.
---
Outside diff comments:
In `@libs/transformer/src/transformer/domain/envelope/edifact.py`:
- Around line 66-90: Move the uuid import to the module-level import section of
the EDIFACT envelope module, and remove the inline import from the build
classmethod while preserving its existing UUID generation behavior.
In `@services/api/src/api/ports/repository.py`:
- Around line 80-86: Update
test_inbound_route_repository_get_inbound_route_no_filters to pass a valid
integer tenant_id when calling get_inbound_route, matching the required non-null
tenant contract and InboundRoute model.
In `@services/api/src/api/services/as2_receive_service.py`:
- Around line 371-381: Update the inbound outbox payload in the AS2 receive flow
to include the existing edi_record["trace_id"] value, matching the outbound
event payload. Preserve the current edi_message.received event fields and
routing behavior while ensuring trace_id is propagated for translation.
🪄 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: ce566b13-559a-4410-adcd-a772b83c8125
📒 Files selected for processing (11)
libs/database/src/database/repository.pylibs/database/tests/test_database_repository.pylibs/domain/src/domain/events.pylibs/identity/docker/docker-compose.ymllibs/transformer/src/transformer/domain/envelope/edifact.pyservices/api/src/api/adapters/repository.pyservices/api/src/api/cdc_relay.pyservices/api/src/api/ports/repository.pyservices/api/src/api/services/as2_receive_service.pyservices/api/src/api/services/outbound_service.pyservices/api/tests/test_api_repository.py
… path testing code review
This was referenced Jul 14, 2026
Merged
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
New Features
Bug Fixes