+
+
{{ event.event_type|replace('_', ' ')|title }}
{% if event.status == 'success' %}
{{ _('Success') }}
@@ -89,7 +89,13 @@
{% endif %}
{% if event.message %}
-
{{ event.message }}
+
{{ event.message }}
+ {% endif %}
+ {% if event.event_metadata %}
+
+ {{ _('Details') }}
+ {{ event.event_metadata | tojson }}
+
{% endif %}
{{ event.created_at|user_datetime }}
diff --git a/app/utils/integration_http.py b/app/utils/integration_http.py
new file mode 100644
index 0000000..0761949
--- /dev/null
+++ b/app/utils/integration_http.py
@@ -0,0 +1,57 @@
+"""
+Shared HTTP session for outbound integration calls (retries, timeouts).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, Optional
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_TIMEOUT = (5, 30)
+
+
+def integration_session(
+ total_retries: int = 3,
+ backoff_factor: float = 0.5,
+ timeout: tuple = _DEFAULT_TIMEOUT,
+) -> requests.Session:
+ """
+ Session with retry on 429, 500, 502, 503, 504 for GET/POST/PUT/PATCH/DELETE.
+ """
+ session = requests.Session()
+ retry = Retry(
+ total=total_retries,
+ connect=total_retries,
+ read=total_retries,
+ backoff_factor=backoff_factor,
+ status_forcelist=(429, 500, 502, 503, 504),
+ allowed_methods=frozenset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]),
+ raise_on_status=False,
+ )
+ adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=20)
+ session.mount("http://", adapter)
+ session.mount("https://", adapter)
+ session.request_timeout = timeout # type: ignore[attr-defined]
+ return session
+
+
+def session_request(
+ session: requests.Session,
+ method: str,
+ url: str,
+ *,
+ headers: Optional[Dict[str, str]] = None,
+ params: Optional[Dict[str, Any]] = None,
+ json: Any = None,
+ data: Any = None,
+ timeout: Optional[tuple] = None,
+) -> requests.Response:
+ """Perform request using session's default timeout."""
+ to = timeout or getattr(session, "request_timeout", _DEFAULT_TIMEOUT)
+ return session.request(method.upper(), url, headers=headers, params=params, json=json, data=data, timeout=to)
diff --git a/app/utils/integration_sync_context.py b/app/utils/integration_sync_context.py
new file mode 100644
index 0000000..6d543a6
--- /dev/null
+++ b/app/utils/integration_sync_context.py
@@ -0,0 +1,153 @@
+"""
+Resolve client and actor user for integration sync (especially global integrations).
+
+Per-user integrations use integration.user_id. Global integrations use, in order:
+1. INTEGRATION_SYNC_USER_ID (numeric user id)
+2. First active user with role admin
+3. First active user
+
+Projects are created under a dedicated client (default name "Integration imports"),
+overridable via INTEGRATION_IMPORT_CLIENT_NAME.
+
+External system linkage is stored in Project.custom_fields / Task.custom_fields under
+the key "integration": {"source": "
", "ref": ""}.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_IMPORT_CLIENT_NAME = "Integration imports"
+
+
+def _import_client_name() -> str:
+ raw = (os.getenv("INTEGRATION_IMPORT_CLIENT_NAME") or "").strip()
+ return raw or DEFAULT_IMPORT_CLIENT_NAME
+
+
+def get_or_create_integration_import_client():
+ """Return the shared Client used for imported integration projects; flush only (caller commits)."""
+ from app import db
+ from app.models import Client
+
+ name = _import_client_name()
+ c = Client.query.filter_by(name=name).first()
+ if c:
+ return c
+ c = Client(name=name)
+ db.session.add(c)
+ db.session.flush()
+ return c
+
+
+def resolve_integration_actor_user_id(integration) -> Optional[int]:
+ """
+ User id to use for Task.created_by and similar when syncing.
+ """
+ from app.models import User
+
+ if integration is not None and getattr(integration, "user_id", None) is not None:
+ return integration.user_id
+
+ env_raw = (os.getenv("INTEGRATION_SYNC_USER_ID") or "").strip()
+ if env_raw.isdigit():
+ uid = int(env_raw)
+ from app import db
+
+ u = db.session.get(User, uid)
+ if u and getattr(u, "is_active", True):
+ return uid
+ logger.warning("INTEGRATION_SYNC_USER_ID=%s is missing or inactive", env_raw)
+
+ admin = User.query.filter_by(role="admin", is_active=True).order_by(User.id).first()
+ if admin:
+ return admin.id
+
+ any_user = User.query.filter_by(is_active=True).order_by(User.id).first()
+ return any_user.id if any_user else None
+
+
+def require_sync_context(integration) -> Tuple[int, int]:
+ """
+ Returns (actor_user_id, import_client_id).
+ Raises ValueError with a clear message if no actor user exists.
+ """
+ uid = resolve_integration_actor_user_id(integration)
+ if uid is None:
+ raise ValueError(
+ "No active user found to attribute imported tasks to. "
+ "Create a user or set INTEGRATION_SYNC_USER_ID to a valid user id."
+ )
+ client = get_or_create_integration_import_client()
+ return uid, client.id
+
+
+def find_project_by_integration_ref(client_id: int, source: str, ref: str):
+ from app.models import Project
+
+ for p in Project.query.filter_by(client_id=client_id).all():
+ cf = p.custom_fields if p.custom_fields is not None else {}
+ block = cf.get("integration") if isinstance(cf, dict) else {}
+ if isinstance(block, dict) and block.get("source") == source and block.get("ref") == ref:
+ return p
+ return None
+
+
+def ensure_project_integration_fields(
+ project,
+ *,
+ source: str,
+ ref: str,
+ display_name: str,
+ description: Optional[str] = None,
+) -> None:
+ """Attach integration marker to project custom_fields (mutates in place)."""
+ cf: Dict[str, Any] = dict(project.custom_fields) if isinstance(project.custom_fields, dict) else {}
+ cf["integration"] = {"source": source, "ref": ref}
+ project.custom_fields = cf
+ if display_name and project.name != display_name:
+ project.name = display_name
+ if description is not None:
+ project.description = description
+
+
+def find_task_by_integration_ref(project_id: int, ref: str, source: Optional[str] = None):
+ """Match task by integration ref. If ``source`` is set, require the same integration source."""
+ from app.models import Task
+
+ for t in Task.query.filter_by(project_id=project_id).all():
+ cf = t.custom_fields if t.custom_fields is not None else {}
+ block = cf.get("integration") if isinstance(cf, dict) else {}
+ if not isinstance(block, dict) or block.get("ref") != ref:
+ continue
+ if source is not None and block.get("source") != source:
+ continue
+ return t
+ return None
+
+
+def set_task_integration_ref(task, *, source: str, ref: str, extra: Optional[Dict[str, Any]] = None) -> None:
+ cf: Dict[str, Any] = dict(task.custom_fields) if isinstance(task.custom_fields, dict) else {}
+ block: Dict[str, Any] = {"source": source, "ref": ref}
+ if extra:
+ block.update(extra)
+ cf["integration"] = block
+ task.custom_fields = cf
+
+
+def sync_result_item_count(sync_result: Optional[Dict[str, Any]]) -> int:
+ """Normalize synced_count vs synced_items from connector sync_data return values."""
+ if not sync_result or not isinstance(sync_result, dict):
+ return 0
+ for key in ("synced_count", "synced_items"):
+ v = sync_result.get(key)
+ if v is not None:
+ try:
+ return int(v)
+ except (TypeError, ValueError):
+ continue
+ return 0
diff --git a/app/utils/scheduled_tasks.py b/app/utils/scheduled_tasks.py
index bb7bccf..5bb0407 100644
--- a/app/utils/scheduled_tasks.py
+++ b/app/utils/scheduled_tasks.py
@@ -899,7 +899,16 @@ def sync_integrations():
f"Failed to sync integration {integration.id} ({integration.provider}): {result.get('message')}"
)
- db.session.commit()
+ from app.utils.integration_sync_context import sync_result_item_count
+
+ _n = sync_result_item_count(result)
+ service._log_event(
+ integration.id,
+ "sync",
+ bool(result.get("success")),
+ result.get("message"),
+ ({"synced_count": _n, "synced_items": _n, "trigger": "scheduler"} if _n or result.get("success") else {"trigger": "scheduler"}),
+ )
except Exception as e:
error_msg = f"Error syncing integration {integration.id} ({integration.provider}): {str(e)}"
@@ -907,7 +916,11 @@ def sync_integrations():
logger.error(error_msg, exc_info=True)
integration.last_sync_status = "error"
integration.last_error = str(e)
- db.session.commit()
+ try:
+ service._log_event(integration.id, "sync", False, str(e), {"trigger": "scheduler"})
+ except Exception as log_err:
+ logger.warning("Could not log integration sync failure: %s", log_err)
+ db.session.commit()
logger.info(f"Integration sync completed. Synced {synced_count}/{len(active_integrations)} integrations")
if errors:
diff --git a/tests/test_utils/test_integration_sync_context.py b/tests/test_utils/test_integration_sync_context.py
new file mode 100644
index 0000000..ec89354
--- /dev/null
+++ b/tests/test_utils/test_integration_sync_context.py
@@ -0,0 +1,51 @@
+"""Tests for integration sync helpers."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+pytestmark = [pytest.mark.unit]
+
+
+def test_sync_result_item_count_prefers_synced_count():
+ from app.utils.integration_sync_context import sync_result_item_count
+
+ assert sync_result_item_count({"synced_count": 5, "synced_items": 9}) == 5
+
+
+def test_sync_result_item_count_falls_back_to_synced_items():
+ from app.utils.integration_sync_context import sync_result_item_count
+
+ assert sync_result_item_count({"synced_items": 7}) == 7
+
+
+def test_sync_result_item_count_empty():
+ from app.utils.integration_sync_context import sync_result_item_count
+
+ assert sync_result_item_count({}) == 0
+ assert sync_result_item_count(None) == 0
+
+
+@patch("app.models.Task")
+def test_find_task_by_integration_ref_filters_by_source(MockTask):
+ from app.utils.integration_sync_context import find_task_by_integration_ref
+
+ t_git = MagicMock()
+ t_git.custom_fields = {"integration": {"source": "github", "ref": "same-ref"}}
+ t_jira = MagicMock()
+ t_jira.custom_fields = {"integration": {"source": "jira", "ref": "same-ref"}}
+ MockTask.query.filter_by.return_value.all.return_value = [t_git, t_jira]
+
+ assert find_task_by_integration_ref(42, "same-ref", source="jira") is t_jira
+ assert find_task_by_integration_ref(42, "same-ref", source="github") is t_git
+
+
+@patch("app.models.Task")
+def test_find_task_by_integration_ref_without_source_matches_any(MockTask):
+ from app.utils.integration_sync_context import find_task_by_integration_ref
+
+ first = MagicMock()
+ first.custom_fields = {"integration": {"source": "github", "ref": "r1"}}
+ MockTask.query.filter_by.return_value.all.return_value = [first]
+
+ assert find_task_by_integration_ref(1, "r1") is first