Checking for errors in the codebase#3
Merged
Conversation
Copilot created this pull request from a session on behalf of
Victor-Kipruto-Rop
July 11, 2026 21:19
View session
Reviewer's GuideAdds backward-compatible helpers and shims around RDS connections, database queries, webhook handling, Kafka DLQ, Kafka producer/consumer, Mpesa transaction handling, metrics, caching, and Safaricom service aliases while tightening webhook validation/metrics and hardening DLQ/DB behavior for missing dependencies and flexible configuration. Sequence diagram for updated C2B webhook confirmation flowsequenceDiagram
actor SafaricomSystem
participant FlaskApp
participant WebhookProcessor
participant WebhookMetrics
participant Validator as validate_c2b_payload
participant KafkaProducer as get_producer
SafaricomSystem->>FlaskApp: POST /webhook/c2b/confirmation
FlaskApp->>FlaskApp: _rate_limited("c2b_confirmation")
alt rate_limited
FlaskApp->>WebhookMetrics: record_request(429, duration)
FlaskApp-->>SafaricomSystem: 429 {status: error, error: rate_limited}
else not_rate_limited
FlaskApp->>FlaskApp: request.get_json(silent=True)
alt invalid_json
FlaskApp->>WebhookMetrics: record_request(400, duration)
FlaskApp-->>SafaricomSystem: 400 {status: error, error: Invalid JSON}
else valid_json
FlaskApp->>WebhookProcessor: _validate_payload(payload)
FlaskApp->>Validator: validate_c2b_payload(payload)
alt invalid_payload
FlaskApp->>WebhookMetrics: record_request(400, duration)
FlaskApp-->>SafaricomSystem: 400 {ResultCode: 1, ResultDesc: Invalid payload}
else valid_payload
FlaskApp->>WebhookProcessor: process_c2b_confirmation(payload)
FlaskApp->>KafkaProducer: get_producer()
alt producer_available
FlaskApp->>KafkaProducer: publish_transaction(payload, key, event_type)
end
FlaskApp->>WebhookMetrics: record_request(200, duration)
FlaskApp-->>SafaricomSystem: 200 {ResultCode: 0, ResultDesc: Accepted}
end
end
end
note over FlaskApp,WebhookMetrics: WebhookMetrics has classmethod record_request used for observability
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Victor-Kipruto-Rop
approved these changes
Jul 11, 2026
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 4 issues, and left some high level feedback:
- Several new helpers in ingestion/db_queries.py (e.g. get_transaction_by_id, get_daily_summary, bulk_insert_transactions, date range/index helpers) bypass the existing connection pool and hard‑code localhost/mpesa_test credentials, which reduces configurability and may break non‑test deployments; consider wiring these through the existing pooling/configuration mechanisms instead of direct psycopg2.connect calls.
- The lightweight kafka shim (top‑level kafka.py) will shadow the real kafka-python package if both are present, potentially altering behavior in environments that expect the full client; if the shim is only intended for CI/local without kafka-python, consider renaming the module or making its use explicitly opt‑in to avoid accidental override.
- DarajaClient now uses the global requests.get/post instead of the existing session object, which means you lose any session-level configuration (timeouts, adapters, connection pooling) and makes it harder to customize behavior; it may be safer to keep using a configured session or wire the session usage through a backward-compatible path rather than ignoring it.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Several new helpers in ingestion/db_queries.py (e.g. get_transaction_by_id, get_daily_summary, bulk_insert_transactions, date range/index helpers) bypass the existing connection pool and hard‑code localhost/mpesa_test credentials, which reduces configurability and may break non‑test deployments; consider wiring these through the existing pooling/configuration mechanisms instead of direct psycopg2.connect calls.
- The lightweight kafka shim (top‑level kafka.py) will shadow the real kafka-python package if both are present, potentially altering behavior in environments that expect the full client; if the shim is only intended for CI/local without kafka-python, consider renaming the module or making its use explicitly opt‑in to avoid accidental override.
- DarajaClient now uses the global requests.get/post instead of the existing session object, which means you lose any session-level configuration (timeouts, adapters, connection pooling) and makes it harder to customize behavior; it may be safer to keep using a configured session or wire the session usage through a backward-compatible path rather than ignoring it.
## Individual Comments
### Comment 1
<location path="ingestion/db_queries.py" line_range="68-77" />
<code_context>
- LIMIT 1
- """
- return DatabaseQueries.execute_query(query, (transaction_id,), fetch_one=True)
+ conn = psycopg2.connect(
+ host="localhost",
+ port=5432,
+ database="mpesa_test",
+ user="test_user",
+ password="test_password",
+ )
+ cur = conn.cursor()
+ cur.execute(
+ """
+ SELECT * FROM stg_c2b_transactions
+ WHERE transaction_id = %s
+ LIMIT 1
+ """,
+ (transaction_id,),
+ )
+ return cur.fetchone()
@staticmethod
</code_context>
<issue_to_address>
**🚨 issue (security):** New helpers open raw psycopg2 connections with hardcoded credentials and never close them.
These helper functions (`get_transaction_by_id`, `get_daily_summary`, `bulk_insert_transactions`, `get_transactions_by_date_range`, `analyze_missing_indexes`, `get_index_usage_stats`) each open a new `psycopg2` connection with hardcoded `localhost/mpesa_test/test_user/test_password`, and do not reliably close the connection or cursor. This sidesteps the existing pooling/config setup and risks connection leaks under load, as well as embedding credentials in code. Please reuse `get_pooled_connection`/existing connection helpers and ensure connections/cursors are always closed (e.g., via context managers), without hardcoded credentials.
</issue_to_address>
### Comment 2
<location path="ingestion/db_queries.py" line_range="123" />
<code_context>
+ def get_daily_summary(transaction_date: str) -> List[Dict]:
</code_context>
<issue_to_address>
**issue (bug_risk):** `get_daily_summary` return shape and fetch mode no longer match the type annotation or previous behavior.
The function is annotated as returning `List[Dict]`, but it now uses `cur.fetchone()` and may return a single dict, a tuple (`result`), or an empty list. This inconsistent return shape breaks the previous "list of row dicts" contract and is likely to confuse callers. Either update the annotation and implementation to always return a single summary dict, or restore the `fetchall()` + row-to-dict behavior so the type and runtime shape are aligned.
</issue_to_address>
### Comment 3
<location path="ingestion/kafka_dlq.py" line_range="371" />
<code_context>
- is_recoverable: bool = True,
-) -> bool:
- """Convenience function to send message to DLQ"""
+def send_to_dlq(*args, **kwargs) -> bool:
+ """Convenience function to send message to DLQ with legacy/new signatures."""
+ if args and isinstance(args[0], dict):
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring `send_to_dlq(*args, **kwargs)` into a typed main API plus a small legacy shim with centralized defaults to make its behavior easier to understand and maintain while preserving backward compatibility.
The complexity concerns around `send_to_dlq(*args, **kwargs)` are valid. You can keep full backward compatibility while making the function easier to reason about by:
1. Extracting explicit, typed helpers for legacy patterns.
2. Centralizing defaults/magic values.
3. Ensuring all paths go through `get_dlq_handler()` instead of creating handlers ad‑hoc.
### 1. Centralize defaults
```python
DLQ_TOPIC = "mpesa-transactions-dlq"
DEFAULT_ORIGINAL_TOPIC = "mpesa-transactions"
UNKNOWN_MESSAGE_ID = "unknown"
UNKNOWN_ERROR_MESSAGE = "unknown"
DEFAULT_FAILURE_REASON = FailureReasonEnum.UNKNOWN_ERROR
DEFAULT_ERROR_REASON_STR = "unknown_error"
```
### 2. Explicit helper for “failed_message dict” legacy usage
```python
def send_dlq_from_failed_message(
failed_message: Dict[str, Any],
*,
topic: str = DLQ_TOPIC,
error_reason: str = DEFAULT_ERROR_REASON_STR,
) -> bool:
handler = get_dlq_handler()
return handler.send_to_dlq(
message_id=failed_message.get("TransID", UNKNOWN_MESSAGE_ID),
original_topic=topic,
original_message=failed_message,
failure_reason=DEFAULT_FAILURE_REASON,
error_message=str(error_reason),
)
```
### 3. Explicit “new” API with typed signature
```python
def send_to_dlq(
message_id: str,
original_topic: str = DEFAULT_ORIGINAL_TOPIC,
original_message: Dict[str, Any] | None = None,
failure_reason: FailureReasonEnum = DEFAULT_FAILURE_REASON,
error_message: str = UNKNOWN_ERROR_MESSAGE,
error_stacktrace: Optional[str] = None,
is_recoverable: bool = True,
) -> bool:
handler = get_dlq_handler()
return handler.send_to_dlq(
message_id=message_id,
original_topic=original_topic,
original_message=original_message or {},
failure_reason=failure_reason,
error_message=error_message,
error_stacktrace=error_stacktrace,
is_recoverable=is_recoverable,
)
```
### 4. Thin compatibility shim that delegates based on shape
You can keep the overloaded entry point but make it a thin router that delegates to the explicit functions above:
```python
def send_to_dlq_legacy(*args, **kwargs) -> bool:
# Legacy dict-only usage
if args and isinstance(args[0], dict):
return send_dlq_from_failed_message(
failed_message=args[0],
topic=kwargs.get("topic", DLQ_TOPIC),
error_reason=kwargs.get("error_reason", DEFAULT_ERROR_REASON_STR),
)
# Legacy positional usage mapped into the explicit API
return send_to_dlq(
message_id=kwargs.get("message_id", args[0] if len(args) > 0 else UNKNOWN_MESSAGE_ID),
original_topic=kwargs.get("original_topic", args[1] if len(args) > 1 else DEFAULT_ORIGINAL_TOPIC),
original_message=kwargs.get("original_message", args[2] if len(args) > 2 else {}),
failure_reason=kwargs.get("failure_reason", args[3] if len(args) > 3 else DEFAULT_FAILURE_REASON),
error_message=kwargs.get("error_message", args[4] if len(args) > 4 else UNKNOWN_ERROR_MESSAGE),
error_stacktrace=kwargs.get("error_stacktrace"),
is_recoverable=kwargs.get("is_recoverable", True),
)
```
Then, either:
- Keep the public name as the explicit API and expose the shim separately, or
- Keep the shim as the public `send_to_dlq` and have it internally call the explicit `send_to_dlq` / `send_dlq_from_failed_message`.
This keeps all current call patterns working but moves the complexity into a small, clearly “compatibility” function, makes the primary API straightforward and typed, and removes duplicated handler initialization and scattered magic values.
</issue_to_address>
### Comment 4
<location path="ingestion/webhook_receiver.py" line_range="27" />
<code_context>
_RATE_STATE: Dict[Tuple[str, str], Tuple[float, int]] = {}
+def validate_c2b_payload(payload: Dict[str, Any]) -> bool:
+ """Backward-compatible C2B payload validator used by tests."""
+ try:
</code_context>
<issue_to_address>
**issue (complexity):** Consider centralizing C2B payload validation in WebhookProcessor and using a small helper to wrap metrics-aware responses so the routes share one validation path and avoid repeated metrics code.
You can reduce the new complexity without changing behavior by (1) centralizing C2B payload validation and (2) wrapping metrics recording so it’s not repeated in each branch.
### 1. Centralize C2B validation
Right now you have:
- `validate_c2b_payload(payload)` (module-level, used in route)
- Inline validation inside `process_c2b_validation`
- `_validate_payload` used before confirmation
You can move the shared validation into `WebhookProcessor` and keep the module-level helper as a thin wrapper used by tests, so all routes rely on a single validation implementation.
```python
def validate_c2b_payload(payload: Dict[str, Any]) -> bool:
"""Backward-compatible C2B payload validator used by tests."""
return WebhookProcessor.validate_c2b_payload(payload)
class WebhookProcessor:
"""Process and validate incoming webhook callbacks."""
@staticmethod
def validate_c2b_payload(payload: Dict[str, Any]) -> bool:
try:
required_fields = ["TransID", "TransAmount", "MSISDN", "TransTime"]
if not all(payload.get(field) for field in required_fields):
return False
datetime.strptime(str(payload.get("TransTime")), "%Y%m%d%H%M%S")
return True
except Exception:
return False
@staticmethod
def process_c2b_validation(payload: Dict[str, Any]) -> Dict[str, str]:
try:
if not WebhookProcessor.validate_c2b_payload(payload):
return {"ResultCode": 1, "ResultDesc": "Missing or invalid fields"}
transaction_id = payload.get("TransID")
amount = payload.get("TransAmount")
phone = payload.get("MSISDN")
logger.info(
"C2B Validation - TxnID: %s, Amount: %s, Phone: %s",
transaction_id,
amount,
phone,
)
if float(amount) > 1000000:
return {"ResultCode": 1, "ResultDesc": "Amount exceeds limit"}
return {"ResultCode": 0, "ResultDesc": "Validation accepted"}
except Exception as e:
logger.error("Validation error: %s", str(e))
return {"ResultCode": 1, "ResultDesc": "Validation failed"}
```
Then in `c2b_confirmation`, reuse the same method instead of a separate validator:
```python
@app.route("/webhook/c2b/confirmation", methods=["POST"])
def c2b_confirmation():
started = time()
try:
if _rate_limited("c2b_confirmation"):
return _metrics_response(429, {"status": "error", "error": "rate_limited"}, started)
payload = request.get_json(silent=True)
if payload is None:
return _metrics_response(400, {"status": "error", "error": "Invalid JSON"}, started)
payload = processor._validate_payload(payload)
logger.debug("Received C2B confirmation callback")
if not processor.validate_c2b_payload(payload):
return _metrics_response(400, {"ResultCode": 1, "ResultDesc": "Invalid payload"}, started)
processor.process_c2b_confirmation(payload)
producer = get_producer()
if producer:
producer.publish_transaction(
payload,
key=payload.get("MSISDN"),
event_type="c2b_confirmation",
)
return _metrics_response(200, {"ResultCode": 0, "ResultDesc": "Accepted"}, started)
except Exception as e:
logger.error("Webhook confirmation error: %s", str(e))
return _metrics_response(500, {"status": "error"}, started)
```
### 2. Wrap metrics recording
You can encapsulate the repeated `WebhookMetrics.record_request(status, time() - started)` plus response construction into a tiny helper. This keeps all branches using metrics consistently and avoids duplication.
```python
def _metrics_response(status: int, body: Dict[str, Any], started: float):
WebhookMetrics.record_request(status, time() - started)
return jsonify(body), status
```
This helper keeps the behavior identical (same status codes and bodies) while making the route easier to read and reducing the chance of missing metrics in a new branch.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+68
to
+77
| conn = psycopg2.connect( | ||
| host="localhost", | ||
| port=5432, | ||
| database="mpesa_test", | ||
| user="test_user", | ||
| password="test_password", | ||
| ) | ||
| cur = conn.cursor() | ||
| cur.execute( | ||
| """ |
There was a problem hiding this comment.
🚨 issue (security): New helpers open raw psycopg2 connections with hardcoded credentials and never close them.
These helper functions (get_transaction_by_id, get_daily_summary, bulk_insert_transactions, get_transactions_by_date_range, analyze_missing_indexes, get_index_usage_stats) each open a new psycopg2 connection with hardcoded localhost/mpesa_test/test_user/test_password, and do not reliably close the connection or cursor. This sidesteps the existing pooling/config setup and risks connection leaks under load, as well as embedding credentials in code. Please reuse get_pooled_connection/existing connection helpers and ensure connections/cursors are always closed (e.g., via context managers), without hardcoded credentials.
| "total_amount": result[0], | ||
| "transaction_count": result[1], | ||
| } | ||
| return result or [] |
There was a problem hiding this comment.
issue (bug_risk): get_daily_summary return shape and fetch mode no longer match the type annotation or previous behavior.
The function is annotated as returning List[Dict], but it now uses cur.fetchone() and may return a single dict, a tuple (result), or an empty list. This inconsistent return shape breaks the previous "list of row dicts" contract and is likely to confuse callers. Either update the annotation and implementation to always return a single summary dict, or restore the fetchall() + row-to-dict behavior so the type and runtime shape are aligned.
| is_recoverable: bool = True, | ||
| ) -> bool: | ||
| """Convenience function to send message to DLQ""" | ||
| def send_to_dlq(*args, **kwargs) -> bool: |
There was a problem hiding this comment.
issue (complexity): Consider refactoring send_to_dlq(*args, **kwargs) into a typed main API plus a small legacy shim with centralized defaults to make its behavior easier to understand and maintain while preserving backward compatibility.
The complexity concerns around send_to_dlq(*args, **kwargs) are valid. You can keep full backward compatibility while making the function easier to reason about by:
- Extracting explicit, typed helpers for legacy patterns.
- Centralizing defaults/magic values.
- Ensuring all paths go through
get_dlq_handler()instead of creating handlers ad‑hoc.
1. Centralize defaults
DLQ_TOPIC = "mpesa-transactions-dlq"
DEFAULT_ORIGINAL_TOPIC = "mpesa-transactions"
UNKNOWN_MESSAGE_ID = "unknown"
UNKNOWN_ERROR_MESSAGE = "unknown"
DEFAULT_FAILURE_REASON = FailureReasonEnum.UNKNOWN_ERROR
DEFAULT_ERROR_REASON_STR = "unknown_error"2. Explicit helper for “failed_message dict” legacy usage
def send_dlq_from_failed_message(
failed_message: Dict[str, Any],
*,
topic: str = DLQ_TOPIC,
error_reason: str = DEFAULT_ERROR_REASON_STR,
) -> bool:
handler = get_dlq_handler()
return handler.send_to_dlq(
message_id=failed_message.get("TransID", UNKNOWN_MESSAGE_ID),
original_topic=topic,
original_message=failed_message,
failure_reason=DEFAULT_FAILURE_REASON,
error_message=str(error_reason),
)3. Explicit “new” API with typed signature
def send_to_dlq(
message_id: str,
original_topic: str = DEFAULT_ORIGINAL_TOPIC,
original_message: Dict[str, Any] | None = None,
failure_reason: FailureReasonEnum = DEFAULT_FAILURE_REASON,
error_message: str = UNKNOWN_ERROR_MESSAGE,
error_stacktrace: Optional[str] = None,
is_recoverable: bool = True,
) -> bool:
handler = get_dlq_handler()
return handler.send_to_dlq(
message_id=message_id,
original_topic=original_topic,
original_message=original_message or {},
failure_reason=failure_reason,
error_message=error_message,
error_stacktrace=error_stacktrace,
is_recoverable=is_recoverable,
)4. Thin compatibility shim that delegates based on shape
You can keep the overloaded entry point but make it a thin router that delegates to the explicit functions above:
def send_to_dlq_legacy(*args, **kwargs) -> bool:
# Legacy dict-only usage
if args and isinstance(args[0], dict):
return send_dlq_from_failed_message(
failed_message=args[0],
topic=kwargs.get("topic", DLQ_TOPIC),
error_reason=kwargs.get("error_reason", DEFAULT_ERROR_REASON_STR),
)
# Legacy positional usage mapped into the explicit API
return send_to_dlq(
message_id=kwargs.get("message_id", args[0] if len(args) > 0 else UNKNOWN_MESSAGE_ID),
original_topic=kwargs.get("original_topic", args[1] if len(args) > 1 else DEFAULT_ORIGINAL_TOPIC),
original_message=kwargs.get("original_message", args[2] if len(args) > 2 else {}),
failure_reason=kwargs.get("failure_reason", args[3] if len(args) > 3 else DEFAULT_FAILURE_REASON),
error_message=kwargs.get("error_message", args[4] if len(args) > 4 else UNKNOWN_ERROR_MESSAGE),
error_stacktrace=kwargs.get("error_stacktrace"),
is_recoverable=kwargs.get("is_recoverable", True),
)Then, either:
- Keep the public name as the explicit API and expose the shim separately, or
- Keep the shim as the public
send_to_dlqand have it internally call the explicitsend_to_dlq/send_dlq_from_failed_message.
This keeps all current call patterns working but moves the complexity into a small, clearly “compatibility” function, makes the primary API straightforward and typed, and removes duplicated handler initialization and scattered magic values.
| _RATE_STATE: Dict[Tuple[str, str], Tuple[float, int]] = {} | ||
|
|
||
|
|
||
| def validate_c2b_payload(payload: Dict[str, Any]) -> bool: |
There was a problem hiding this comment.
issue (complexity): Consider centralizing C2B payload validation in WebhookProcessor and using a small helper to wrap metrics-aware responses so the routes share one validation path and avoid repeated metrics code.
You can reduce the new complexity without changing behavior by (1) centralizing C2B payload validation and (2) wrapping metrics recording so it’s not repeated in each branch.
1. Centralize C2B validation
Right now you have:
validate_c2b_payload(payload)(module-level, used in route)- Inline validation inside
process_c2b_validation _validate_payloadused before confirmation
You can move the shared validation into WebhookProcessor and keep the module-level helper as a thin wrapper used by tests, so all routes rely on a single validation implementation.
def validate_c2b_payload(payload: Dict[str, Any]) -> bool:
"""Backward-compatible C2B payload validator used by tests."""
return WebhookProcessor.validate_c2b_payload(payload)
class WebhookProcessor:
"""Process and validate incoming webhook callbacks."""
@staticmethod
def validate_c2b_payload(payload: Dict[str, Any]) -> bool:
try:
required_fields = ["TransID", "TransAmount", "MSISDN", "TransTime"]
if not all(payload.get(field) for field in required_fields):
return False
datetime.strptime(str(payload.get("TransTime")), "%Y%m%d%H%M%S")
return True
except Exception:
return False
@staticmethod
def process_c2b_validation(payload: Dict[str, Any]) -> Dict[str, str]:
try:
if not WebhookProcessor.validate_c2b_payload(payload):
return {"ResultCode": 1, "ResultDesc": "Missing or invalid fields"}
transaction_id = payload.get("TransID")
amount = payload.get("TransAmount")
phone = payload.get("MSISDN")
logger.info(
"C2B Validation - TxnID: %s, Amount: %s, Phone: %s",
transaction_id,
amount,
phone,
)
if float(amount) > 1000000:
return {"ResultCode": 1, "ResultDesc": "Amount exceeds limit"}
return {"ResultCode": 0, "ResultDesc": "Validation accepted"}
except Exception as e:
logger.error("Validation error: %s", str(e))
return {"ResultCode": 1, "ResultDesc": "Validation failed"}Then in c2b_confirmation, reuse the same method instead of a separate validator:
@app.route("/webhook/c2b/confirmation", methods=["POST"])
def c2b_confirmation():
started = time()
try:
if _rate_limited("c2b_confirmation"):
return _metrics_response(429, {"status": "error", "error": "rate_limited"}, started)
payload = request.get_json(silent=True)
if payload is None:
return _metrics_response(400, {"status": "error", "error": "Invalid JSON"}, started)
payload = processor._validate_payload(payload)
logger.debug("Received C2B confirmation callback")
if not processor.validate_c2b_payload(payload):
return _metrics_response(400, {"ResultCode": 1, "ResultDesc": "Invalid payload"}, started)
processor.process_c2b_confirmation(payload)
producer = get_producer()
if producer:
producer.publish_transaction(
payload,
key=payload.get("MSISDN"),
event_type="c2b_confirmation",
)
return _metrics_response(200, {"ResultCode": 0, "ResultDesc": "Accepted"}, started)
except Exception as e:
logger.error("Webhook confirmation error: %s", str(e))
return _metrics_response(500, {"status": "error"}, started)2. Wrap metrics recording
You can encapsulate the repeated WebhookMetrics.record_request(status, time() - started) plus response construction into a tiny helper. This keeps all branches using metrics consistently and avoids duplication.
def _metrics_response(status: int, body: Dict[str, Any], started: float):
WebhookMetrics.record_request(status, time() - started)
return jsonify(body), statusThis helper keeps the behavior identical (same status codes and bodies) while making the route easier to read and reducing the chance of missing metrics in a new branch.
|
You are seeing this message because GitHub Code Scanning has recently been set up for this repository, or this pull request contains the workflow file for the Code Scanning tool. What Enabling Code Scanning Means:
For more information about GitHub Code Scanning, check out the documentation. |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| @@ -1,4 +1,5 @@ | |||
| fastapi==0.115.14 | |||
| Flask==3.0.3 | |||
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
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.
Pull request created by AI Agent
Summary by Sourcery
Improve compatibility and observability across ingestion, Kafka, and webhook components while adding lightweight shims and helpers for legacy integrations and tests.
New Features:
Bug Fixes:
Enhancements:
Build:
Tests: