[Core] ocean core next resync

port_ocean/clients/port/client.py

@@ -13,6 +13,7 @@
     get_internal_http_client,
 )
 from port_ocean.exceptions.clients import KafkaCredentialsNotFound
+from typing import Any
 
 
 class PortClient(
@@ -75,3 +76,13 @@ async def get_org_id(self) -> str:
         handle_status_code(response)
 
         return response.json()["organization"]["id"]
+
+    async def update_integration_state(self, state: dict[str, Any]) -> dict[str, Any]:
+        logger.debug(f"Updating integration state with: {state}")
+        response = await self.client.patch(
+            f"{self.api_url}/integration/{self.integration_identifier}/state",
+            headers=await self.auth.headers(),
+            json=state,
+        )
+        handle_status_code(response)
+        return response.json().get("integration", {})

port_ocean/config/settings.py

@@ -74,7 +74,7 @@ class IntegrationConfiguration(BaseOceanSettings, extra=Extra.allow):
     event_listener: EventListenerSettingsType
     # If an identifier or type is not provided, it will be generated based on the integration name
     integration: IntegrationSettings = IntegrationSettings(type="", identifier="")
-    runtime: Runtime = "OnPrem"
+    runtime: Runtime = Runtime.OnPrem
 
     @root_validator()
     def validate_integration_config(cls, values: dict[str, Any]) -> dict[str, Any]:
@@ -98,8 +98,8 @@ def parse_config(model: Type[BaseModel], config: Any) -> BaseModel:
         return values
 
     @validator("runtime")
-    def validate_runtime(cls, runtime: Literal["OnPrem", "Saas"]) -> Runtime:
-        if runtime == "Saas":
+    def validate_runtime(cls, runtime: Runtime) -> Runtime:
+        if runtime == Runtime.Saas:
             spec = get_spec_file()
             if spec is None:
                 raise ValueError(

port_ocean/core/event_listener/base.py

@@ -5,6 +5,7 @@
 
 from port_ocean.config.base import BaseOceanModel
 from port_ocean.utils.signal import signal_handler
+from port_ocean.context.ocean import ocean
 
 
 class EventListenerEvents(TypedDict):
@@ -36,6 +37,39 @@ def _stop(self) -> None:
         """
         pass
 
+    async def _before_resync(self) -> None:
+        """
+        Can be used for event listeners that need to perform some action before resync.
+        """
+        await ocean.app.update_state_before_scheduled_sync()
+
+    async def _after_resync(self) -> None:
+        """
+        Can be used for event listeners that need to perform some action after resync.
+        """
+        await ocean.app.update_state_after_scheduled_sync()
+
+    async def _on_resync_failure(self, e: Exception) -> None:
+        """
+        Can be used for event listeners that need to handle resync failures.
+        """
+        await ocean.app.update_state_after_scheduled_sync("failed")
+
+    async def _resync(
+        self,
+        resync_args: dict[Any, Any],
+    ) -> None:
+        """
+        Triggers the "on_resync" event.
+        """
+        await self._before_resync()
+        try:
+            await self.events["on_resync"](resync_args)
+            await self._after_resync()
+        except Exception as e:
+            await self._on_resync_failure(e)
+            raise e
+
 
 class EventListenerSettings(BaseOceanModel, extra=Extra.allow):
     type: str

port_ocean/core/event_listener/http.py

@@ -64,6 +64,6 @@ async def _start(self) -> None:
 
         @target_channel_router.post("/resync")
         async def resync() -> None:
-            await self.events["on_resync"]({})
+            await self._resync({})
 
         ocean.app.fast_api_app.include_router(target_channel_router)

port_ocean/core/event_listener/kafka.py

@@ -122,7 +122,7 @@ async def _handle_message(self, raw_msg: Message) -> None:
 
         if "change.log" in topic and message is not None:
             try:
-                await self.events["on_resync"](message)
+                await self._resync(message)
             except Exception as e:
                 _type, _, tb = sys.exc_info()
                 logger.opt(exception=(_type, None, tb)).error(

port_ocean/core/event_listener/once.py

@@ -1,3 +1,4 @@
+import datetime
 import signal
 from typing import Literal, Any
 
@@ -9,6 +10,8 @@
     EventListenerSettings,
 )
 from port_ocean.utils.repeat import repeat_every
+from port_ocean.context.ocean import ocean
+from port_ocean.utils.time import convert_str_to_utc_datetime, convert_to_minutes
 
 
 class OnceEventListenerSettings(EventListenerSettings):
@@ -41,6 +44,93 @@ def __init__(
     ):
         super().__init__(events)
         self.event_listener_config = event_listener_config
+        self.cached_integration: dict[str, Any] | None = None
+
+    async def get_current_integration_cached(self) -> dict[str, Any]:
+        if self.cached_integration:
+            return self.cached_integration
+
+        self.cached_integration = await ocean.port_client.get_current_integration()
+        return self.cached_integration
+
+    async def get_saas_resync_initialization_and_interval(
+        self,
+    ) -> tuple[int | None, datetime.datetime | None]:
+        """
+        Get the scheduled resync interval and the last updated time of the integration config for the saas application.
+        interval is the saas configured resync interval time.
+        start_time is the last updated time of the integration config.
+        return: (interval, start_time)
+        """
+        if not ocean.app.is_saas():
+            return (None, None)
+
+        try:
+            integration = await self.get_current_integration_cached()
+        except Exception:
+            logger.exception("Error occurred while getting current integration")
+            return (None, None)
+
+        interval_str = (
+            integration.get("spec", {})
+            .get("appSpec", {})
+            .get("scheduledResyncInterval")
+        )
+
+        if not interval_str:
+            logger.error("scheduledResyncInterval not found for integration")
+            return (None, None)
+
+        last_updated_saas_integration_config_str = integration.get(
+            "statusInfo", {}
+        ).get("updatedAt")
+
+        # we use the last updated time of the integration config as the start time since in saas application the interval is configured by the user from the portal
+        if not last_updated_saas_integration_config_str:
+            logger.error("updatedAt not found for integration")
+            return (None, None)
+
+        return (
+            convert_to_minutes(interval_str),
+            convert_str_to_utc_datetime(last_updated_saas_integration_config_str),
+        )
+
+    async def _before_resync(self) -> None:
+        if not ocean.app.is_saas():
+            # in case of non-saas, we still want to update the state before and after the resync
+            await super()._before_resync()
+            return
+
+        (interval, start_time) = (
+            await self.get_saas_resync_initialization_and_interval()
+        )
+        await ocean.app.update_state_before_scheduled_sync(interval, start_time)
+
+    async def _after_resync(self) -> None:
+        if not ocean.app.is_saas():
+            # in case of non-saas, we still want to update the state before and after the resync
+            await super()._after_resync()
+            return
+
+        (interval, start_time) = (
+            await self.get_saas_resync_initialization_and_interval()
+        )
+        await ocean.app.update_state_after_scheduled_sync(
+            "completed", interval, start_time
+        )
+
+    async def _on_resync_failure(self, e: Exception) -> None:
+        if not ocean.app.is_saas():
+            # in case of non-saas, we still want to update the state before and after the resync
+            await super()._after_resync()
+            return
+
+        (interval, start_time) = (
+            await self.get_saas_resync_initialization_and_interval()
+        )
+        await ocean.app.update_state_after_scheduled_sync(
+            "failed", interval, start_time
+        )
 
     async def _start(self) -> None:
         """
@@ -53,7 +143,7 @@ async def _start(self) -> None:
         async def resync_and_exit() -> None:
             logger.info("Once event listener started")
             try:
-                await self.events["on_resync"]({})
+                await self._resync({})
             except Exception:
                 # we catch all exceptions here to make sure the application will exit gracefully
                 logger.exception("Error occurred while resyncing")

port_ocean/core/event_listener/polling.py

@@ -49,7 +49,14 @@ def __init__(
     ):
         super().__init__(events)
         self.event_listener_config = event_listener_config
-        self._last_updated_at = None
+
+    def should_resync(self, last_updated_at: str) -> bool:
+        _last_updated_at = ocean.app.last_integration_updated_at
+
+        if _last_updated_at is None:
+            return self.event_listener_config.resync_on_start
+
+        return _last_updated_at != last_updated_at
 
     async def _start(self) -> None:
         """
@@ -69,17 +76,10 @@ async def resync() -> None:
             integration = await ocean.app.port_client.get_current_integration()
             last_updated_at = integration["updatedAt"]
 
-            should_resync = (
-                self._last_updated_at is not None
-                or self.event_listener_config.resync_on_start
-            ) and self._last_updated_at != last_updated_at
-
-            if should_resync:
+            if self.should_resync(last_updated_at):
                 logger.info("Detected change in integration, resyncing")
-                self._last_updated_at = last_updated_at
-                running_task: Task[Any] = get_event_loop().create_task(
-                    self.events["on_resync"]({})  # type: ignore
-                )
+                ocean.app.last_integration_updated_at = last_updated_at
+                running_task: Task[Any] = get_event_loop().create_task(self._resync({}))
                 signal_handler.register(running_task.cancel)
 
                 await running_task

port_ocean/core/models.py

@@ -1,11 +1,14 @@
 from dataclasses import dataclass, field
-from typing import Any, Literal
+from enum import Enum
+from typing import Any
 
 from pydantic import BaseModel
 from pydantic.fields import Field
 
 
-Runtime = Literal["OnPrem", "Saas"]
+class Runtime(Enum):
+    Saas = "Saas"
+    OnPrem = "OnPrem"
 
 
 class Entity(BaseModel):

port_ocean/core/utils.py

@@ -35,12 +35,13 @@ def is_same_entity(first_entity: Entity, second_entity: Entity) -> bool:
 
 
 async def validate_integration_runtime(
-    port_client: PortClient, requested_runtime: Runtime
+    port_client: PortClient,
+    requested_runtime: Runtime,
 ) -> None:
     logger.debug("Validating integration runtime")
     current_integration = await port_client.get_current_integration(should_raise=False)
     current_runtime = current_integration.get("installationType", "OnPrem")
-    if current_integration and current_runtime != requested_runtime:
+    if current_integration and current_runtime != requested_runtime.value:
         raise IntegrationRuntimeException(
             f"Invalid Runtime! Requested to run existing {current_runtime} integration in {requested_runtime} runtime."
         )