From eaa31aa64af3a4f97b65aac541e5d514886c6bdc Mon Sep 17 00:00:00 2001 From: ANO3457 <187025195+ANO3457@users.noreply.github.com> Date: Sat, 24 Oct 2026 18:35:59 +0300 Subject: [PATCH] feat(workspace): read AWS EventBridge events as topic contracts (#4521) --- docs/architecture/language-support.md | 1 + docs/scale/WORKSPACES.md | 1 + .../workspace/extractors/topic/__init__.py | 3 +- .../core/workspace/extractors/topic/aws.py | 127 ++++++++++++++++- .../repowise/core/workspace/matching/topic.py | 2 +- .../unit/workspace/test_eventbridge_topics.py | 133 ++++++++++++++++++ 6 files changed, 260 insertions(+), 7 deletions(-) create mode 100644 tests/unit/workspace/test_eventbridge_topics.py diff --git a/docs/architecture/language-support.md b/docs/architecture/language-support.md index 693a50b1a2..72f1489d08 100644 --- a/docs/architecture/language-support.md +++ b/docs/architecture/language-support.md @@ -1317,6 +1317,7 @@ mislead the masker. | **Data** | DDL `CREATE`/`ALTER`, Alembic `op.create_table`, Laravel `Schema::create`/`Schema::table`, Knex migrations, ORM models (SQLAlchemy, SQLModel, Django, JPA, EF Core, ActiveRecord, Eloquent, Prisma, TypeORM, Sequelize, Drizzle) | SQL string literals in app code (sqlglot-parsed, verb-anchored-regex fallback), Laravel `DB::table`, Knex queries | | **Topic (C# integration events)** | `IntegrationEvent` subclasses constructed and passed to a publish call, keyed by short type name | `AddSubscription`, `Subscribe`, `IIntegrationEventHandler` | | **Topic (JMS)** | Spring `JmsTemplate` / `JmsMessagingTemplate` sends with a destination argument | `@JmsListener(destination)`, template `receive` / `receiveAndConvert`, listener containers' `setDestinationName` | +| **Topic (EventBridge)** | `PutEvents` (AWS SDK v2/v3, boto3 `put_events`) with `Source` and `DetailType` | CDK `eventPattern`, SAM / CloudFormation `EventBridgeRule` `Pattern` / `EventPattern` | RabbitMQ topics are also read from Go (streadway/amqp, amqp091-go), including a queue named through its declaration's result (`q.Name`). diff --git a/docs/scale/WORKSPACES.md b/docs/scale/WORKSPACES.md index 9f42173e0f..aea7af37b7 100644 --- a/docs/scale/WORKSPACES.md +++ b/docs/scale/WORKSPACES.md @@ -172,6 +172,7 @@ Scans source files for HTTP routes, gRPC services, database tables, message topi | Socket / WebSocket | SignalR `MapHub("/path")`, FastAPI `@app.websocket("/path")`, `ws` `WebSocketServer({ path })`, NestJS `@WebSocketGateway`; events: socket.io `emit` (server and client), Laravel broadcast events (`broadcastOn` / `broadcastAs`), `Broadcast::on`, Pusher `trigger` | ClientWebSocket `ConnectAsync`, SignalR `HubConnectionBuilder.WithUrl`, NativeWebSocket and WebSocketSharp `new WebSocket(...)`, browser/Node `new WebSocket(url)`; events: socket.io `on` / `@SubscribeMessage`, Laravel Echo `listen` and `useEcho`, pusher-js `bind` | | Topics: C# integration events | `new T(...)` of an `IntegrationEvent` subclass the repo declares, reaching `PublishAsync` / `Publish` or a repo method taking an `IntegrationEvent`; the id is the short type name, which the event bus routes by | `AddSubscription()`, `Subscribe()` on a declared event, and `IIntegrationEventHandler` classes for events no registration subscribes | | Topics: JMS | Spring `JmsTemplate` / `JmsMessagingTemplate` `convertAndSend` / `send` / `sendAndReceive` / `convertSendAndReceive` on a receiver named for JMS (`jmsTemplate`, `orderJmsTemplate`, `getJmsTemplate()`) with a destination argument; ActiveMQ and IBM MQ `queue://` names read as the bare queue | `@JmsListener(destination = ...)`, `receive` / `receiveAndConvert`, `DefaultMessageListenerContainer.setDestinationName` | +| Topics: EventBridge | `PutEvents` (AWS SDK v2/v3, boto3 `put_events`) with `Source` and `DetailType` | CDK `eventPattern`, SAM / CloudFormation `EventBridgeRule` `Pattern` / `EventPattern` | Socket detection is toggled by `detect_socket` in the `contracts:` block below. A topic, queue or route name is resolved when it is a literal or a constant the diff --git a/packages/core/src/repowise/core/workspace/extractors/topic/__init__.py b/packages/core/src/repowise/core/workspace/extractors/topic/__init__.py index 05f3b12bdb..eea2fc71bf 100644 --- a/packages/core/src/repowise/core/workspace/extractors/topic/__init__.py +++ b/packages/core/src/repowise/core/workspace/extractors/topic/__init__.py @@ -9,7 +9,7 @@ from __future__ import annotations from ..dialect import ContractDialect, DialectExtractor -from .aws import SNS, SQS +from .aws import EVENTBRIDGE, SNS, SQS from .bullmq import BULLMQ from .csharp_events import CSHARP_EVENTS from .dialect import topic_identity @@ -32,6 +32,7 @@ BULLMQ, SQS, SNS, + EVENTBRIDGE, NESTJS, LARAVEL_QUEUES, CSHARP_EVENTS, diff --git a/packages/core/src/repowise/core/workspace/extractors/topic/aws.py b/packages/core/src/repowise/core/workspace/extractors/topic/aws.py index 84b8b75fa5..958c63dede 100644 --- a/packages/core/src/repowise/core/workspace/extractors/topic/aws.py +++ b/packages/core/src/repowise/core/workspace/extractors/topic/aws.py @@ -1,22 +1,34 @@ -"""Amazon SQS queues and SNS topics: AWS SDK v3 commands, v2 client calls, sqs-consumer, boto3. +"""Amazon SQS queues, SNS topics, and EventBridge events. -Every call names its destination by URL or ARN (``QueueUrl``, ``TopicArn``), +Every SQS and SNS call names its destination by URL or ARN (``QueueUrl``, ``TopicArn``), never by name, so the name is the URL's last path segment or the ARN's last field: ``https://sqs.eu-west-1.amazonaws.com/123/orders`` and ``arn:aws:sns:eu-west-1:123:orders`` are both ``orders``. A URL built as ``${base}/orders`` still names its queue. A call is read only when its file spells the key, so ``bot.sendMessage(chatId, text)`` is never a queue. + +EventBridge events are identified by ``/``. Publishers invoke +``PutEvents`` (AWS SDK v3 ``PutEventsCommand``, v2 ``putEvents``, boto3 ``put_events``), +and consumers subscribe via rule patterns in CDK (``eventPattern``) or SAM / CloudFormation +(``EventPattern``, ``Pattern``, ``EventBridgeRule``). """ from __future__ import annotations import re +from typing import TYPE_CHECKING from repowise.core.workspace.contracts import TOPIC_KIND_QUEUE, TOPIC_KIND_TOPIC +from repowise.core.workspace.origins import ContractOrigin from ..langs import JS_TS, PYTHON -from ..strings import Arg -from .dialect import TopicCall, TopicDialect, last_segment +from ..strings import Arg, match_paren +from .dialect import TopicCall, TopicDialect, last_segment, topic_contract + +if TYPE_CHECKING: + from repowise.core.workspace.contracts import Contract + + from ..base import ScanContext _QUEUE_URL = Arg(keys=("QueueUrl", "queueUrl")) _TOPIC_ARN = Arg(keys=("TopicArn",)) @@ -75,4 +87,109 @@ def _sns(head: str, role: str, label: str, extensions: frozenset[str]) -> TopicC ), ) -__all__ = ["SNS", "SQS"] +_EVENTBRIDGE_GATES = ("PutEvents", "put_events", "eventPattern", "EventPattern", "EventBridgeRule") +_YAML_EXT = frozenset({".yaml", ".yml"}) +_EVENTBRIDGE_EXT = JS_TS | PYTHON | _YAML_EXT + +_PUT_EVENTS_HEAD = re.compile( + r"(?:\bnew\s+PutEventsCommand|\bput_events|\bputEvents)\s*\(" +) +_ENTRY_BLOCK = re.compile(r"\{[^{}]*\}") +_SOURCE_VAL = re.compile(r"""['"]?(?:Source|source)['"]?\s*[:=]\s*(?:\[\s*)?['"]([^'"\]]+)['"]""") +_DETAIL_TYPE_VAL = re.compile(r"""['"]?(?:DetailType|detailType|detail-type)['"]?\s*[:=]\s*(?:\[\s*)?['"]([^'"\]]+)['"]""") + +_RULE_PATTERN_HEAD = re.compile(r"(?:eventPattern|EventPattern|EventBridgeRule)\s*[:=]\s*\{") + +_YAML_SOURCE_RE = re.compile(r"""(?:Source|source):\s*\n\s*-\s*['"]?([A-Za-z0-9_.-]+)['"]?""") +_YAML_DETAIL_RE = re.compile(r"""(?:DetailType|detailType|detail-type):\s*\n\s*-\s*['"]?([A-Za-z0-9_.-]+)['"]?""") + + +class EventBridgeTopicDialect: + """Amazon EventBridge event publishers and rule patterns as topic contracts.""" + + name = "eventbridge" + extensions = _EVENTBRIDGE_EXT + requires = _EVENTBRIDGE_GATES + + def extract(self, ctx: ScanContext) -> list[Contract]: + out: list[Contract] = [] + content = ctx.content + + # 1. Publishers: PutEventsCommand, put_events, putEvents + if "PutEvents" in content or "put_events" in content: + for pm in _PUT_EVENTS_HEAD.finditer(content): + open_idx = pm.end() - 1 + close_idx = match_paren(content, open_idx) + call_body = ( + content[open_idx:close_idx] + if close_idx != -1 + else content[open_idx : open_idx + 2000] + ) + sources = [m.group(1) for m in _SOURCE_VAL.finditer(call_body)] + details = [m.group(1) for m in _DETAIL_TYPE_VAL.finditer(call_body)] + for s, d in zip(sources, details, strict=False): + event_id = f"{s}/{d}" + out.append( + topic_contract( + ctx, + name=event_id, + broker="eventbridge", + kind=TOPIC_KIND_TOPIC, + role="provider", + label="eventbridge.PutEvents", + origin=ContractOrigin.CLIENT_CALL, + offset=pm.start(), + ) + ) + + # 2. Consumers: CDK rule patterns or SAM / CloudFormation + if any(g in content for g in ("eventPattern", "EventPattern", "EventBridgeRule")): + if ctx.suffix in _YAML_EXT: + s_m = _YAML_SOURCE_RE.search(content) + d_m = _YAML_DETAIL_RE.search(content) + if s_m and d_m: + event_id = f"{s_m.group(1)}/{d_m.group(1)}" + out.append( + topic_contract( + ctx, + name=event_id, + broker="eventbridge", + kind=TOPIC_KIND_TOPIC, + role="consumer", + label="EventBridgeRule", + origin=ContractOrigin.ROUTE_REGISTRATION, + offset=0, + ) + ) + else: + for rm in _RULE_PATTERN_HEAD.finditer(content): + open_idx = rm.end() - 1 + close_idx = match_paren(content, open_idx, closer="}") + block = ( + content[open_idx:close_idx] + if close_idx != -1 + else content[open_idx : open_idx + 2000] + ) + s_m = _SOURCE_VAL.search(block) + d_m = _DETAIL_TYPE_VAL.search(block) + if s_m and d_m: + event_id = f"{s_m.group(1)}/{d_m.group(1)}" + out.append( + topic_contract( + ctx, + name=event_id, + broker="eventbridge", + kind=TOPIC_KIND_TOPIC, + role="consumer", + label="EventBridgeRule", + origin=ContractOrigin.ROUTE_REGISTRATION, + offset=rm.start(), + ) + ) + + return out + + +EVENTBRIDGE = EventBridgeTopicDialect() + +__all__ = ["EVENTBRIDGE", "SNS", "SQS"] diff --git a/packages/core/src/repowise/core/workspace/matching/topic.py b/packages/core/src/repowise/core/workspace/matching/topic.py index 517ab1d8d2..c3558d1efd 100644 --- a/packages/core/src/repowise/core/workspace/matching/topic.py +++ b/packages/core/src/repowise/core/workspace/matching/topic.py @@ -120,7 +120,7 @@ def matching_keys(consumer: Contract, provider_index: dict[str, list[Contract]]) # is configured outside the code (RabbitMQ, SQS, Redis, Kafka, NATS), so either # meets any broker but those. Every other pair must be the same broker: a Kafka # topic and a RabbitMQ queue that share a name are two things. -_OWN_KIND_ONLY = frozenset({"bullmq"}) +_OWN_KIND_ONLY = frozenset({"bullmq", "eventbridge"}) _ANY_TRANSPORT = frozenset({"laravel", "nestjs"}) diff --git a/tests/unit/workspace/test_eventbridge_topics.py b/tests/unit/workspace/test_eventbridge_topics.py new file mode 100644 index 0000000000..dd899f3850 --- /dev/null +++ b/tests/unit/workspace/test_eventbridge_topics.py @@ -0,0 +1,133 @@ +"""AWS EventBridge topic dialect: PutEvents producers and rule pattern consumers.""" + +from __future__ import annotations + +from pathlib import Path + +from repowise.core.workspace.extractors.topic import TopicExtractor +from repowise.core.workspace.matching import match_contracts +from repowise.core.workspace.origins import ContractOrigin + + +def _extract(root: Path, rel: str, content: str, alias: str = "svc"): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return TopicExtractor().extract(root, alias) + + +def _ids(rows, role: str) -> list[str]: + return sorted(c.contract_id for c in rows if c.role == role) + + +def test_eventbridge_put_events_command_ts(tmp_path: Path) -> None: + src = ( + 'import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";\n' + "const client = new EventBridgeClient({});\n" + "await client.send(new PutEventsCommand({\n" + " Entries: [\n" + " {\n" + ' Source: "com.shop.orders",\n' + ' DetailType: "OrderCreated",\n' + ' Detail: JSON.stringify({ id: "1" }),\n' + " }\n" + " ]\n" + "}));\n" + ) + [c] = _extract(tmp_path, "publish.ts", src) + assert (c.contract_id, c.role) == ("topic::com.shop.orders/ordercreated", "provider") + assert c.meta == {"topic": "com.shop.orders/OrderCreated", "broker": "eventbridge", "kind": "topic"} + assert c.origin == ContractOrigin.CLIENT_CALL + + +def test_eventbridge_put_events_python(tmp_path: Path) -> None: + src = ( + "import boto3\n" + "client = boto3.client('events')\n" + "client.put_events(\n" + " Entries=[\n" + " {\n" + " 'Source': 'com.shop.orders',\n" + " 'DetailType': 'OrderCreated',\n" + " 'Detail': '{}',\n" + " }\n" + " ]\n" + ")\n" + ) + [c] = _extract(tmp_path, "publish.py", src) + assert (c.contract_id, c.role) == ("topic::com.shop.orders/ordercreated", "provider") + assert c.origin == ContractOrigin.CLIENT_CALL + + +def test_eventbridge_rule_cdk(tmp_path: Path) -> None: + src = ( + "import { Rule } from 'aws-cdk-lib/aws-events';\n" + "new Rule(this, 'Rule', {\n" + " eventPattern: {\n" + " source: ['com.shop.orders'],\n" + " detailType: ['OrderCreated'],\n" + " },\n" + "});\n" + ) + [c] = _extract(tmp_path, "rule.ts", src) + assert (c.contract_id, c.role) == ("topic::com.shop.orders/ordercreated", "consumer") + assert c.origin == ContractOrigin.ROUTE_REGISTRATION + + +def test_eventbridge_sam_yaml(tmp_path: Path) -> None: + src = ( + "OrderCreatedRule:\n" + " Type: EventBridgeRule\n" + " Properties:\n" + " Pattern:\n" + " source:\n" + " - com.shop.orders\n" + " detail-type:\n" + " - OrderCreated\n" + ) + [c] = _extract(tmp_path, "template.yaml", src) + assert (c.contract_id, c.role) == ("topic::com.shop.orders/ordercreated", "consumer") + assert c.origin == ContractOrigin.ROUTE_REGISTRATION + + +def test_eventbridge_links_across_repos(tmp_path: Path) -> None: + pub = _extract( + tmp_path / "pub", + "send.py", + "import boto3\n" + "boto3.client('events').put_events(Entries=[{'Source': 'com.shop.orders', 'DetailType': 'OrderCreated'}])\n", + "pub", + ) + sub = _extract( + tmp_path / "sub", + "rule.ts", + "new Rule(this, 'R', { eventPattern: { source: ['com.shop.orders'], detailType: ['OrderCreated'] } });\n", + "sub", + ) + links = [lk for lk in match_contracts(pub + sub) if lk.contract_type == "topic"] + assert [(lk.provider_repo, lk.consumer_repo, lk.contract_id) for lk in links] == [ + ("pub", "sub", "topic::com.shop.orders/ordercreated") + ] + + +def test_eventbridge_isolated_from_sqs_and_sns(tmp_path: Path) -> None: + eb = _extract( + tmp_path / "eb", + "send.py", + "boto3.client('events').put_events(Entries=[{'Source': 'orders', 'DetailType': 'created'}])\n", + "eb", + ) + sns = _extract( + tmp_path / "sns", + "pub.py", + "boto3.client('sns').publish(TopicArn='arn:aws:sns:eu-west-1:123:orders/created')\n", + "sns", + ) + assert [lk for lk in match_contracts(eb + sns) if lk.contract_type == "topic"] == [] + + +def test_commented_out_eventbridge_call_is_skipped(tmp_path: Path) -> None: + src = ( + "// client.putEvents({ Entries: [{ Source: 'com.shop.orders', DetailType: 'OrderCreated' }] });\n" + ) + assert _extract(tmp_path, "c.ts", src) == []