[Core] feat: add live_event kafka metadata (#2936)

# Description

What:
Added type: "live-event" metadata when sending live events (webhooks) to
lakehouse, allowing
  differentiation between sync data and live event data in Snowflake.

  Why:
To enable downstream consumers to distinguish between data ingested via
full sync vs real-time
  webhook events, enabling different processing logic or analytics.

  How:

  Summary

  - Added type field to lakehouse request body from Ocean live events
- Propagated type through lakehouse-writer pipeline into kafka_metadata
in Snowflake


## Type of change

Please leave one option from the following and delete the rest:

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] New Integration (non-breaking change which adds a new integration)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Non-breaking change (fix of existing functionality that will not
change current behavior)
- [ ] Documentation (added/updated documentation)

<h4> All tests should be run against the port production
environment(using a testing org). </h4>

### Core testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Resync finishes successfully
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Scheduled resync able to abort existing resync and start a new one
- [ ] Tested with at least 2 integrations from scratch
- [ ] Tested with Kafka and Polling event listeners
- [ ] Tested deletion of entities that don't pass the selector


### Integration testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Completed a full resync from a freshly installed integration and
it completed successfully
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Resync finishes successfully
- [ ] If new resource kind is added or updated in the integration, add
example raw data, mapping and expected result to the `examples` folder
in the integration directory.
- [ ] If resource kind is updated, run the integration with the example
data and check if the expected result is achieved
- [ ] If new resource kind is added or updated, validate that
live-events for that resource are working as expected
- [ ] Docs PR link [here](#)

### Preflight checklist

- [ ] Handled rate limiting
- [ ] Handled pagination
- [ ] Implemented the code in async
- [ ] Support Multi account

## Screenshots

Include screenshots from your environment showing how the resources of
the integration will look.

## API Documentation

Provide links to the API documentation used for this integration.

Author: IdansPort

CHANGELOG.md

@@ -7,6 +7,12 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+## 0.38.15 (2026-03-15)
+
+### Improvements
+
+- Add type:live_event in body to signify live_event ingested data in lakehouse
+
 ## 0.38.14 (2026-03-15)
 
 ### Improvements

port_ocean/clients/port/mixins/integrations.py

@@ -283,20 +283,25 @@ async def post_integration_raw_data(
         sync_id: str,
         kind: str,
         operation: LakehouseOperation = LakehouseOperation.UPSERT,
+        data_type: str | None = None,
     ) -> None:
         logger.debug(
             "starting POST raw data request", raw_data=raw_data, operation=operation
         )
         headers = await self.auth.headers()
 
+        body: dict[str, Any] = {
+            "items": raw_data,
+            "extractionTimestamp": int(datetime.now().timestamp() * 1000),
+            "operation": operation.value,
+        }
+        if data_type is not None:
+            body["type"] = data_type
+
         response = await self.client.post(
             f"{self.auth.ingest_url}/lake/write/integration-type/{quote_plus(self.auth.integration_type)}/integration/{quote_plus(self.integration_identifier)}/sync/{quote_plus(sync_id)}/kind/{quote_plus(kind)}",
             headers=headers,
-            json={
-                "items": raw_data,
-                "extractionTimestamp": int(datetime.now().timestamp() * 1000),
-                "operation": operation.value,
-            },
+            json=body,
         )
         handle_port_status_code(response, should_raise=False, should_log=True)
         logger.debug("Finished POST raw data request")

port_ocean/core/integrations/mixins/live_events.py

@@ -118,6 +118,7 @@ async def _send_webhook_raw_data_to_lakehouse(
                             event_id,
                             kind,
                             operation=LakehouseOperation.UPSERT,
+                            data_type="live-event",
                         )
                     except Exception as e:
                         logger.warning(
@@ -138,6 +139,7 @@ async def _send_webhook_raw_data_to_lakehouse(
                             event_id,
                             kind,
                             operation=LakehouseOperation.DELETE,
+                            data_type="live-event",
                         )
                     except Exception as e:
                         logger.warning(

port_ocean/core/integrations/mixins/sync_raw.py

@@ -424,7 +424,7 @@ async def _register_in_batches(
 
             if lakehouse_data_enabled and raw_results:
                 await ocean.port_client.post_integration_raw_data(
-                    raw_results, event.id, resource_config.kind
+                    raw_results, event.id, resource_config.kind, data_type="resync"
                 )
 
             logger.info(
@@ -467,7 +467,7 @@ async def _register_in_batches(
                     batch_index += 1
                     if lakehouse_data_enabled:
                         await ocean.port_client.post_integration_raw_data(
-                            items, event.id, resource_config.kind
+                            items, event.id, resource_config.kind, data_type="resync"
                         )
                     number_of_raw_results += len(items)
                     if send_raw_data_examples_amount > 0:

port_ocean/tests/clients/port/mixins/test_integrations_lakehouse.py

@@ -171,3 +171,55 @@ async def test_post_integration_raw_data_extraction_timestamp(
         assert "extractionTimestamp" in expected_json
         assert isinstance(expected_json["extractionTimestamp"], int)
         assert expected_json["extractionTimestamp"] > 0
+
+
+async def test_post_integration_raw_data_with_data_type(
+    lakehouse_integration_client: IntegrationClientMixin,
+) -> None:
+    """Test post_integration_raw_data with data_type parameter for live events."""
+    raw_data = [{"name": "test-entity"}]
+    sync_id = "webhook-event-123"
+    kind = "repository"
+
+    with patch("port_ocean.clients.port.mixins.integrations.handle_port_status_code"):
+        await lakehouse_integration_client.post_integration_raw_data(
+            raw_data,
+            sync_id,
+            kind,
+            operation=LakehouseOperation.UPSERT,
+            data_type="live-event",
+        )
+
+        lakehouse_integration_client.client.post.assert_called_once()
+        call_args = lakehouse_integration_client.client.post.call_args
+
+        expected_json = call_args[1]["json"]
+        assert expected_json["items"] == raw_data
+        assert expected_json["operation"] == "upsert"
+        assert expected_json["type"] == "live-event"
+
+
+async def test_post_integration_raw_data_with_resync_data_type(
+    lakehouse_integration_client: IntegrationClientMixin,
+) -> None:
+    """Test post_integration_raw_data with data_type parameter for resync."""
+    raw_data = [{"name": "test-entity"}]
+    sync_id = "resync-123"
+    kind = "repository"
+
+    with patch("port_ocean.clients.port.mixins.integrations.handle_port_status_code"):
+        await lakehouse_integration_client.post_integration_raw_data(
+            raw_data,
+            sync_id,
+            kind,
+            operation=LakehouseOperation.UPSERT,
+            data_type="resync",
+        )
+
+        lakehouse_integration_client.client.post.assert_called_once()
+        call_args = lakehouse_integration_client.client.post.call_args
+
+        expected_json = call_args[1]["json"]
+        assert expected_json["items"] == raw_data
+        assert expected_json["operation"] == "upsert"
+        assert expected_json["type"] == "resync"

port_ocean/tests/core/handlers/mixins/test_live_events_lakehouse.py

@@ -211,6 +211,7 @@ async def test_send_webhook_raw_data_to_lakehouse_enabled_upsert(
             "test-event-id",
             "repository",
             operation=LakehouseOperation.UPSERT,
+            data_type="live-event",
         )
 
 
@@ -251,6 +252,7 @@ async def test_send_webhook_raw_data_to_lakehouse_enabled_delete(
             "test-event-id",
             "repository",
             operation=LakehouseOperation.DELETE,
+            data_type="live-event",
         )
 
 
@@ -356,12 +358,14 @@ async def test_send_webhook_raw_data_to_lakehouse_both_operations(
                     "test-event-id",
                     "repository",
                     operation=LakehouseOperation.UPSERT,
+                    data_type="live-event",
                 ),
                 call(
                     delete_data,
                     "test-event-id",
                     "repository",
                     operation=LakehouseOperation.DELETE,
+                    data_type="live-event",
                 ),
             ]
         )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "port-ocean"
-version = "0.38.14"
+version = "0.38.15"
 description = "Port Ocean is a CLI tool for managing your Port projects."
 readme = "README.md"
 homepage = "https://app.getport.io"