feat(trigger): add configuration for trigger provider refresh task

api/configs/feature/__init__.py

@@ -1020,6 +1020,28 @@ class CeleryScheduleTasksConfig(BaseSettings):
         default=0,
     )
 
+    # Trigger provider refresh (simple version)
+    ENABLE_TRIGGER_PROVIDER_REFRESH_TASK: bool = Field(
+        description="Enable trigger provider refresh poller",
+        default=True,
+    )
+    TRIGGER_PROVIDER_REFRESH_INTERVAL: int = Field(
+        description="Trigger provider refresh poller interval in minutes",
+        default=1,
+    )
+    TRIGGER_PROVIDER_REFRESH_BATCH_SIZE: int = Field(
+        description="Max trigger subscriptions to process per tick",
+        default=200,
+    )
+    TRIGGER_PROVIDER_CREDENTIAL_THRESHOLD_SECONDS: int = Field(
+        description="Proactive credential refresh threshold in seconds",
+        default=180,
+    )
+    TRIGGER_PROVIDER_SUBSCRIPTION_THRESHOLD_SECONDS: int = Field(
+        description="Proactive subscription refresh threshold in seconds",
+        default=300,
+    )
+
 
 class PositionConfig(BaseSettings):
     POSITION_PROVIDER_PINS: str = Field(

api/extensions/ext_celery.py

@@ -166,6 +166,12 @@ def init_app(app: DifyApp) -> Celery:
             "task": "schedule.workflow_schedule_task.poll_workflow_schedules",
             "schedule": timedelta(minutes=dify_config.WORKFLOW_SCHEDULE_POLLER_INTERVAL),
         }
+    if dify_config.ENABLE_TRIGGER_PROVIDER_REFRESH_TASK:
+        imports.append("schedule.trigger_provider_refresh_task")
+        beat_schedule["trigger_provider_refresh"] = {
+            "task": "schedule.trigger_provider_refresh_task.trigger_provider_refresh",
+            "schedule": timedelta(minutes=dify_config.TRIGGER_PROVIDER_REFRESH_INTERVAL),
+        }
     celery_app.conf.update(beat_schedule=beat_schedule, imports=imports)
 
     return celery_app

api/schedule/trigger_provider_refresh_task.py

@@ -0,0 +1,21 @@
+import logging
+import time
+
+import app
+
+logger = logging.getLogger(__name__)
+
+
+def _now_ts() -> int:
+    return int(time.time())
+
+
+@app.celery.task(queue="trigger_refresh")
+def trigger_provider_refresh() -> None:
+    """
+    Simple trigger provider refresh task.
+    - Scans due trigger subscriptions in small batches
+    - Refreshes OAuth credentials if needed
+    - Refreshes subscription metadata if needed
+    """
+    pass