feat: optimization

app/api/server.py

@@ -123,6 +123,8 @@ async def lifespan(app: FastAPI):
     await close_gcs_service()
     from app.services.media_storage_service import close_storage_service
     await close_storage_service()
+    from app.utils.http_utils import close_shared_aiohttp_sessions
+    await close_shared_aiohttp_sessions()
     from app.services.supabase import get_supabase_client
     client = get_supabase_client()
     if client:

app/services/chat_service.py

@@ -22,7 +22,7 @@ from app.config import (
     OPENROUTER_MIMIKA_MODEL,
     get_settings,
 )
-from app.utils.http_utils import create_session_timeout
+from app.utils.http_utils import shared_aiohttp_session
 from app.utils.json_utils import extract_single_json
 from app.utils.schema_utils import generate_schema_prompt, validate_against_schema
 
@@ -78,8 +78,7 @@ async def _call_llm_api(
         "stream": stream,
     }
 
-    timeout = create_session_timeout()
-    async with aiohttp.ClientSession(timeout=timeout) as session:
+    async with shared_aiohttp_session() as session:
         async with session.post(
             f"{base_url}{path}",
             json=payload,
@@ -392,8 +391,7 @@ async def call_meganova(
         }
 
         try:
-            timeout = create_session_timeout()
-            async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with shared_aiohttp_session() as session:
                 async with session.post(
                     f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
                     json=payload,
@@ -474,8 +472,7 @@ async def call_aion_labs(
         logger.info("[aion] Attempt %s/%s", attempt + 1, total_tries)
 
         try:
-            timeout = create_session_timeout()
-            async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with shared_aiohttp_session() as session:
                 async with session.post(
                     f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
                     json=payload,
@@ -728,8 +725,7 @@ async def _stream_meganova(
         }
 
         try:
-            timeout = create_session_timeout()
-            async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with shared_aiohttp_session() as session:
                 async with session.post(
                     f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
                     json=payload,
@@ -777,8 +773,7 @@ async def _stream_meganova_no_redis(
     }
 
     try:
-        timeout = create_session_timeout()
-        async with aiohttp.ClientSession(timeout=timeout) as session:
+        async with shared_aiohttp_session() as session:
             async with session.post(
                 f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
                 json=payload,
@@ -830,8 +825,7 @@ async def _stream_aion_labs(
         }
 
         try:
-            timeout = create_session_timeout()
-            async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with shared_aiohttp_session() as session:
                 async with session.post(
                     f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
                     json=payload,
@@ -876,8 +870,7 @@ async def _stream_aion_labs_no_redis(
         }
 
         try:
-            timeout = create_session_timeout()
-            async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with shared_aiohttp_session() as session:
                 async with session.post(
                     f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
                     json=payload,
@@ -915,8 +908,7 @@ async def _stream_openrouter_mimika(
     }
 
     try:
-        timeout = create_session_timeout()
-        async with aiohttp.ClientSession(timeout=timeout) as session:
+        async with shared_aiohttp_session() as session:
             async with session.post(
                 f"{OPENROUTER_MIMIKA_BASE_URL}{OPENROUTER_MIMIKA_CHAT_PATH}",
                 json=payload,

app/services/csv_analysis_service.py

@@ -9,7 +9,6 @@ import tempfile
 import time
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple, Union
-from urllib.parse import unquote, urlparse
 
 from app.services.code_executor_service import CodeSanitizer
 from app.services.dataset_metadata_service import extract_metadata
@@ -30,17 +29,9 @@ class CSVAnalysisError(Exception):
     pass
 
 
-async def _download_file(url: str) -> bytes:
-    data, _ = await download_url(url, timeout_seconds=_DOWNLOAD_TIMEOUT)
-    return data
-
-
 async def _resolve_source(source: Union[str, bytes]) -> Tuple[bytes, Optional[str]]:
     if isinstance(source, str) and source.lower().startswith(("http://", "https://")):
-        data = await _download_file(source)
-        parsed = urlparse(source)
-        filename = unquote(Path(parsed.path).name) if parsed.path else None
-        return data, filename
+        return await download_url(source, timeout_seconds=_DOWNLOAD_TIMEOUT)
     elif isinstance(source, bytes):
         return source, None
     else:

app/services/dataset_metadata_service.py

@@ -24,9 +24,6 @@ from typing import (
     Tuple,
     Union,
 )
-from urllib.parse import unquote, urlparse
-
-import aiohttp
 import chardet
 import numpy as np
 import openpyxl  # noqa: F401 – needed as engine for .xlsx
@@ -34,6 +31,7 @@ import pandas as pd
 import xlrd  # noqa: F401 – needed as engine for .xls
 
 from app.core.thread_pool import thread_pool
+from app.utils.http_utils import DownloadError, download_url
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -45,7 +43,6 @@ logger.setLevel(logging.DEBUG)
 _SAMPLE_BYTES_FOR_DETECTION: int = 65_536          # 64 KiB for encoding/delimiter sniffing
 _DEFAULT_SAMPLE_ROWS: int = 5
 _MAX_FILE_SIZE: int = 2 * 1024 * 1024 * 1024       # 2 GiB
-_DOWNLOAD_CHUNK_SIZE: int = 256 * 1024              # 256 KiB streaming chunks
 _DEFAULT_TIMEOUT_SECONDS: int = 120
 
 # Magic bytes for binary file-type detection
@@ -392,40 +389,21 @@ async def _download_from_url(
     url: str,
     config: ExtractionConfig,
 ) -> Tuple[bytes, Optional[str]]:
-    """Stream-download a remote file with size guard and timeout."""
-    timeout = aiohttp.ClientTimeout(total=config.timeout_seconds)
+    """Stream-download a remote file with size guard and timeout.
+
+    Delegates to the shared :func:`app.utils.http_utils.download_url` helper
+    and maps its errors onto the dataset-specific exceptions.
+    """
     try:
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.get(url) as resp:
-                if resp.status != 200:
-                    raise FileDownloadError(
-                        f"HTTP {resp.status} when fetching {url}"
-                    )
-
-                content_length = resp.content_length
-                if content_length and content_length > config.max_file_size_bytes:
-                    raise FileTooLargeError(
-                        f"Remote file advertises {content_length} bytes, "
-                        f"limit is {config.max_file_size_bytes}"
-                    )
-
-                chunks: List[bytes] = []
-                total = 0
-                async for chunk in resp.content.iter_chunked(_DOWNLOAD_CHUNK_SIZE):
-                    total += len(chunk)
-                    if total > config.max_file_size_bytes:
-                        raise FileTooLargeError(
-                            f"Download exceeded {config.max_file_size_bytes} bytes"
-                        )
-                    chunks.append(chunk)
-
-                data = b"".join(chunks)
-    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
-        raise FileDownloadError(f"Download failed for {url}: {exc}") from exc
-
-    parsed = urlparse(url)
-    filename = unquote(Path(parsed.path).name) if parsed.path else None
-    return data, filename
+        return await download_url(
+            url,
+            timeout_seconds=config.timeout_seconds,
+            max_size_bytes=config.max_file_size_bytes,
+        )
+    except DownloadError as exc:
+        if exc.is_size_error:
+            raise FileTooLargeError(str(exc)) from exc
+        raise FileDownloadError(str(exc)) from exc
 
 
 def _is_url(source: Any) -> bool:

app/services/gcs_service.py

@@ -15,6 +15,7 @@ from pydantic import BaseModel, ValidationError
 
 from app.config import get_settings
 from app.core.logger import get_logger
+from app.utils.http_utils import SharedAsyncClient
 
 _logger = get_logger(__name__)
 _settings = get_settings()
@@ -101,8 +102,7 @@ class GCSService:
     """Asynchronous, connection-pooled client for the Google Cloud Storage JSON API."""
 
     def __init__(self) -> None:
-        self._client: Optional[httpx.AsyncClient] = None
-        self._client_lock = asyncio.Lock()
+        self._http = SharedAsyncClient(timeout=_settings.gcs_timeout)
         self._token_cache = _TokenCache()
 
     # ------------------------------------------------------------------
@@ -110,25 +110,10 @@ class GCSService:
     # ------------------------------------------------------------------
 
     async def _get_client(self) -> httpx.AsyncClient:
-        if self._client is None or self._client.is_closed:
-            async with self._client_lock:
-                if self._client is None or self._client.is_closed:
-                    self._client = httpx.AsyncClient(
-                        timeout=_settings.gcs_timeout,
-                        follow_redirects=True,
-                        limits=httpx.Limits(
-                            max_connections=100,
-                            max_keepalive_connections=20,
-                            keepalive_expiry=30,
-                        ),
-                    )
-        return self._client
+        return await self._http.get()
 
     async def close(self) -> None:
-        async with self._client_lock:
-            if self._client is not None and not self._client.is_closed:
-                await self._client.aclose()
-            self._client = None
+        await self._http.close()
 
     # ------------------------------------------------------------------
     # Credential resolution (JSON body / JSON string / URL / file / env)

app/services/google_maps_service.py

@@ -12,6 +12,7 @@ from urllib.parse import unquote, urlencode
 import httpx
 
 from app.config import get_settings
+from app.utils.http_utils import SharedAsyncClient
 
 _logger = logging.getLogger(__name__)
 _settings = get_settings()
@@ -75,8 +76,7 @@ class GoogleMapsService:
         self._places_base_url: str = _settings.google_maps_places_base_url
         self._timeout: int = _settings.google_maps_timeout
         self._max_retries: int = _settings.google_maps_max_retries
-        self._client: Optional[httpx.AsyncClient] = None
-        self._client_lock = asyncio.Lock()
+        self._http = SharedAsyncClient(timeout=self._timeout)
 
     # -----------------------------------------------------------------------
     # HTTP client management
@@ -84,26 +84,11 @@ class GoogleMapsService:
 
     async def _get_client(self) -> httpx.AsyncClient:
         """Return the shared connection-pooled AsyncClient, creating it lazily."""
-        if self._client is None or self._client.is_closed:
-            async with self._client_lock:
-                if self._client is None or self._client.is_closed:
-                    self._client = httpx.AsyncClient(
-                        timeout=self._timeout,
-                        follow_redirects=True,
-                        limits=httpx.Limits(
-                            max_connections=100,
-                            max_keepalive_connections=20,
-                            keepalive_expiry=30,
-                        ),
-                    )
-        return self._client
+        return await self._http.get()
 
     async def close(self) -> None:
         """Close the shared AsyncClient and release pooled connections."""
-        async with self._client_lock:
-            if self._client is not None and not self._client.is_closed:
-                await self._client.aclose()
-            self._client = None
+        await self._http.close()
 
     # -----------------------------------------------------------------------
     # Geocoding — still uses the legacy Google Geocoding API

app/services/google_oauth_service.py

@@ -11,6 +11,7 @@ import jwt
 from app.config import get_settings
 from app.core.logger import get_logger
 from app.models.schemas import GoogleOAuthUserInfo
+from app.utils.http_utils import SharedAsyncClient
 
 _logger = get_logger(__name__)
 _settings = get_settings()
@@ -92,8 +93,10 @@ class GoogleOAuthService:
     """Abstraction over the Google OAuth 2.0 / OpenID Connect flow."""
 
     def __init__(self) -> None:
-        self._client: Optional[httpx.AsyncClient] = None
-        self._client_lock = asyncio.Lock()
+        self._http = SharedAsyncClient(
+            timeout=httpx.Timeout(_settings.google_oauth_timeout),
+            follow_redirects=False,
+        )
         self._jwks = GoogleJWKSCache()
 
     # ------------------------------------------------------------------
@@ -101,24 +104,10 @@ class GoogleOAuthService:
     # ------------------------------------------------------------------
 
     async def _get_client(self) -> httpx.AsyncClient:
-        if self._client is None or self._client.is_closed:
-            async with self._client_lock:
-                if self._client is None or self._client.is_closed:
-                    self._client = httpx.AsyncClient(
-                        timeout=httpx.Timeout(_settings.google_oauth_timeout),
-                        limits=httpx.Limits(
-                            max_connections=100,
-                            max_keepalive_connections=20,
-                            keepalive_expiry=30,
-                        ),
-                    )
-        return self._client
+        return await self._http.get()
 
     async def close(self) -> None:
-        async with self._client_lock:
-            if self._client is not None and not self._client.is_closed:
-                await self._client.aclose()
-            self._client = None
+        await self._http.close()
 
     # ------------------------------------------------------------------
     # Authorization URL (Step 1)

app/services/scheduler_service.py

@@ -23,6 +23,7 @@ from croniter import croniter
 from app.config import get_settings
 from app.core.logger import get_logger
 from app.services.supabase import SupabaseClient, get_supabase_client
+from app.utils.http_utils import SharedAsyncClient
 
 logger = get_logger(__name__)
 settings = get_settings()
@@ -329,28 +330,21 @@ class HttpExecutionEngine:
     _MAX_RESPONSE_PREVIEW = 500
 
     def __init__(self) -> None:
-        self._client: httpx.AsyncClient | None = None
-        self._lock = asyncio.Lock()
+        self._http = SharedAsyncClient(
+            timeout=httpx.Timeout(settings.max_http_timeout),
+            limits=httpx.Limits(
+                max_connections=200,
+                max_keepalive_connections=50,
+                keepalive_expiry=30,
+            ),
+        )
 
     async def _get_client(self) -> httpx.AsyncClient:
-        if self._client is None or self._client.is_closed:
-            async with self._lock:
-                if self._client is None or self._client.is_closed:
-                    self._client = httpx.AsyncClient(
-                        follow_redirects=True,
-                        limits=httpx.Limits(
-                            max_connections=200,
-                            max_keepalive_connections=50,
-                            keepalive_expiry=30,
-                        ),
-                        timeout=httpx.Timeout(settings.max_http_timeout),
-                    )
-        return self._client
+        return await self._http.get()
 
     async def close(self) -> None:
-        if self._client and not self._client.is_closed:
-            await self._client.aclose()
-            logger.info("HTTP client closed")
+        await self._http.close()
+        logger.info("HTTP client closed")
 
     async def execute(self, job: dict[str, Any]) -> HttpExecutionResult:
         retry_on_status: list[int] = job.get("retry_on_status") or [429, 500, 502, 503, 504]
@@ -506,6 +500,26 @@ class SchedulerService:
         self._maintenance_task: asyncio.Task[None] | None = None
         self._repo: SchedulerRepository | None = None
 
+    # -----------------------------------------------------------------------
+    # Supabase repository access
+    # -----------------------------------------------------------------------
+
+    @staticmethod
+    def _get_repo() -> SchedulerRepository:
+        """Return a Supabase-backed repository or raise if Supabase is unavailable."""
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+        return SchedulerRepository(client)
+
+    @staticmethod
+    def _get_optional_repo() -> SchedulerRepository | None:
+        """Return a Supabase-backed repository or None if Supabase is unavailable."""
+        client = get_supabase_client()
+        if client is None:
+            return None
+        return SchedulerRepository(client)
+
     # -----------------------------------------------------------------------
     # Redis connection management
     # -----------------------------------------------------------------------
@@ -666,9 +680,8 @@ class SchedulerService:
         logger.info("Scheduler started (mode=%s, instance=%s)", "redis" if self._use_redis else "memory", self._instance_id)
 
         try:
-            client = get_supabase_client()
-            if client:
-                repo = SchedulerRepository(client)
+            repo = self._get_optional_repo()
+            if repo is not None:
                 jobs = await repo.get_active_jobs()
                 restored = 0
                 for job in jobs:
@@ -737,9 +750,8 @@ class SchedulerService:
                         await self._async_redis.ping()
                     except Exception:
                         logger.error("Redis ping failed in maintenance loop")
-                client = get_supabase_client()
-                if client:
-                    repo = SchedulerRepository(client)
+                repo = self._get_optional_repo()
+                if repo is not None:
                     purged = await repo.purge_old_history(settings.history_retention_days)
                     if purged:
                         logger.info("Purged %d old history records", purged)
@@ -896,12 +908,11 @@ class SchedulerService:
             "job_id": job_id, "trigger_reason": trigger_reason.value,
         })
 
-        client = get_supabase_client()
-        if client is None:
+        repo = self._get_optional_repo()
+        if repo is None:
             logger.error("Supabase not available, cannot execute job %s", job_id)
             return
 
-        repo = SchedulerRepository(client)
         job = await repo.get_job_by_id(job_id)
         if not job:
             logger.error("Job not found: %s", job_id)
@@ -1006,11 +1017,7 @@ class SchedulerService:
     # -----------------------------------------------------------------------
 
     async def create_job(self, body: dict[str, Any]) -> dict[str, Any]:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
 
         existing = await repo.get_job_by_name(body["name"])
         if existing:
@@ -1078,11 +1085,7 @@ class SchedulerService:
         return await repo.get_job_by_id(job_id) or created
 
     async def update_job(self, job_id: str, body: dict[str, Any]) -> dict[str, Any]:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
         job = await repo.get_job_by_id(job_id)
         if not job:
             raise KeyError(f"Job '{job_id}' not found")
@@ -1149,11 +1152,7 @@ class SchedulerService:
         return updated
 
     async def delete_job(self, job_id: str) -> None:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
         job = await repo.get_job_by_id(job_id)
         if not job:
             raise KeyError(f"Job '{job_id}' not found")
@@ -1163,11 +1162,7 @@ class SchedulerService:
         logger.info("Job deleted: %s", job_id)
 
     async def hard_delete_job(self, job_id: str) -> None:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
         job = await repo.get_job_by_id(job_id)
         if not job:
             raise KeyError(f"Job '{job_id}' not found")
@@ -1177,11 +1172,7 @@ class SchedulerService:
         logger.info("Job hard deleted: %s", job_id)
 
     async def pause_job(self, job_id: str) -> dict[str, Any]:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
         job = await repo.get_job_by_id(job_id)
         if not job:
             raise KeyError(f"Job '{job_id}' not found")
@@ -1195,11 +1186,7 @@ class SchedulerService:
         return updated or job
 
     async def resume_job(self, job_id: str) -> dict[str, Any]:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
         job = await repo.get_job_by_id(job_id)
         if not job:
             raise KeyError(f"Job '{job_id}' not found")
@@ -1213,11 +1200,7 @@ class SchedulerService:
         return updated or job
 
     async def run_job_now(self, job_id: str) -> None:
-        client = get_supabase_client()
-        if client is None:
-            raise RuntimeError("Supabase not available")
-
-        repo = SchedulerRepository(client)
+        repo = self._get_repo()
         job = await repo.get_job_by_id(job_id)
         if not job:
             raise KeyError(f"Job '{job_id}' not found")
@@ -1234,17 +1217,15 @@ class SchedulerService:
         page: int = 1,
         page_size: int = 20,
     ) -> tuple[list[dict[str, Any]], int]:
-        client = get_supabase_client()
-        if client is None:
+        repo = self._get_optional_repo()
+        if repo is None:
             return [], 0
-        repo = SchedulerRepository(client)
         return await repo.list_jobs(status=status, tags=tags, page=page, page_size=page_size)
 
     async def get_job(self, job_id: str) -> dict[str, Any] | None:
-        client = get_supabase_client()
-        if client is None:
+        repo = self._get_optional_repo()
+        if repo is None:
             return None
-        repo = SchedulerRepository(client)
         return await repo.get_job_by_id(job_id)
 
     async def get_job_history(
@@ -1253,10 +1234,9 @@ class SchedulerService:
         page: int = 1,
         page_size: int = 20,
     ) -> tuple[list[dict[str, Any]], int]:
-        client = get_supabase_client()
-        if client is None:
+        repo = self._get_optional_repo()
+        if repo is None:
             return [], 0
-        repo = SchedulerRepository(client)
         return await repo.get_history(job_id=job_id, page=page, page_size=page_size)
 
     async def get_execution_history(
@@ -1265,18 +1245,15 @@ class SchedulerService:
         page_size: int = 50,
         status: str | None = None,
     ) -> tuple[list[dict[str, Any]], int]:
-        client = get_supabase_client()
-        if client is None:
+        repo = self._get_optional_repo()
+        if repo is None:
             return [], 0
-        repo = SchedulerRepository(client)
         return await repo.get_history_all(page=page, page_size=page_size, status_filter=status)
 
     async def get_metrics(self) -> dict[str, Any]:
-        client = get_supabase_client()
-        if client is None:
+        repo = self._get_optional_repo()
+        if repo is None:
             return {"error": "Supabase not available"}
-
-        repo = SchedulerRepository(client)
         total_jobs = await repo.count_jobs()
         active_jobs = await repo.count_jobs(status=JobStatus.ACTIVE.value)
         paused_jobs = await repo.count_jobs(status=JobStatus.PAUSED.value)

app/services/url_shortener_service.py

@@ -312,6 +312,52 @@ def _fire_click_webhook(webhook_url: str, short_code: str, long_url: str,
         urllib.request.urlopen(req, timeout=5)
 
 
+_shared_loop: Optional[asyncio.AbstractEventLoop] = None
+_shared_loop_lock = threading.Lock()
+
+
+def _run_async(coro: Any) -> Any:
+    """Run a coroutine on a single shared background event loop.
+
+    Previously this module called ``asyncio.run()`` ~25 times per request,
+    creating and tearing down a fresh event loop (with its own thread pool)
+    for every Supabase call. Reusing one long-lived daemon loop avoids that
+    repeated construction and lets connections/executors persist.
+
+    The loop is created inside the worker thread itself so it is bound to the
+    thread that runs it (required on Windows where ``new_event_loop()`` builds
+    a ProactorEventLoop with an I/O completion port).
+    """
+    global _shared_loop
+    with _shared_loop_lock:
+        if _shared_loop is None or _shared_loop.is_closed() or not _shared_loop.is_running():
+            created_loop: Optional[asyncio.AbstractEventLoop] = None
+            thread_error: Optional[BaseException] = None
+            ready = threading.Event()
+
+            def _run_forever() -> None:
+                nonlocal created_loop, thread_error
+                try:
+                    created_loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(created_loop)
+                    ready.set()
+                    created_loop.run_forever()
+                except BaseException as exc:  # pragma: no cover - loop teardown path
+                    thread_error = exc
+                    ready.set()
+
+            threading.Thread(
+                target=_run_forever,
+                name="url-shortener-loop",
+                daemon=True,
+            ).start()
+            ready.wait()
+            if thread_error is not None:
+                raise thread_error
+            _shared_loop = created_loop
+    return asyncio.run_coroutine_threadsafe(coro, _shared_loop).result()
+
+
 # ---------------------------------------------------------------------------
 # Supabase-backed Storage
 # ---------------------------------------------------------------------------
@@ -337,8 +383,7 @@ class Storage:
 
     def create_owner(self, owner_id: str, name: str, api_key_hash: str, plan: str = "free"):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().insert("url_shortener_owners", {
+            _run_async(self._get_client().insert("url_shortener_owners", {
                 "owner_id": owner_id,
                 "name": name,
                 "api_key_hash": api_key_hash,
@@ -347,23 +392,19 @@ class Storage:
             }))
 
     def get_owner_by_key_hash(self, api_key_hash: str) -> Optional[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().find_one("url_shortener_owners", "api_key_hash", api_key_hash))
+        return _run_async(self._get_client().find_one("url_shortener_owners", "api_key_hash", api_key_hash))
 
     def get_owner(self, owner_id: str) -> Optional[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().find_one("url_shortener_owners", "owner_id", owner_id))
+        return _run_async(self._get_client().find_one("url_shortener_owners", "owner_id", owner_id))
 
     def update_owner_plan(self, owner_id: str, plan: str):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().update("url_shortener_owners", "owner_id", owner_id, {"plan": plan}))
+            _run_async(self._get_client().update("url_shortener_owners", "owner_id", owner_id, {"plan": plan}))
 
     # ---------- Links ----------
 
     def code_exists(self, short_code: str) -> bool:
-        import asyncio
-        row = asyncio.run(self._get_client().find_one("url_shortener_links", "short_code", short_code, columns="short_code"))
+        row = _run_async(self._get_client().find_one("url_shortener_links", "short_code", short_code, columns="short_code"))
         return row is not None
 
     def insert_link(self, **kwargs):
@@ -389,36 +430,30 @@ class Storage:
                 "webhook_url": kwargs.get("webhook_url"),
                 "geo_targeting": kwargs.get("geo_targeting"),
             }
-            import asyncio
-            asyncio.run(self._get_client().insert("url_shortener_links", data))
+            _run_async(self._get_client().insert("url_shortener_links", data))
 
     def update_link(self, short_code: str, **updates):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().update("url_shortener_links", "short_code", short_code, updates))
+            _run_async(self._get_client().update("url_shortener_links", "short_code", short_code, updates))
 
     def get_link(self, short_code: str) -> Optional[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().find_one("url_shortener_links", "short_code", short_code))
+        return _run_async(self._get_client().find_one("url_shortener_links", "short_code", short_code))
 
     def list_links_for_owner(self, owner_id: str) -> List[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().select(
+        return _run_async(self._get_client().select(
             "url_shortener_links", eq=("owner_id", owner_id), order=("created_at", True),
         ))
 
     def deactivate_link(self, short_code: str):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().update("url_shortener_links", "short_code", short_code, {"is_active": 0}))
+            _run_async(self._get_client().update("url_shortener_links", "short_code", short_code, {"is_active": 0}))
 
     def record_click(self, short_code: str, referrer: Optional[str] = None,
                      user_agent: Optional[str] = None, ip_address: Optional[str] = None):
         now = _now_iso()
         parsed = _parse_user_agent(user_agent or "")
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().insert("url_shortener_clicks", {
+            _run_async(self._get_client().insert("url_shortener_clicks", {
                 "short_code": short_code,
                 "referrer": referrer,
                 "user_agent": user_agent,
@@ -429,18 +464,17 @@ class Storage:
                 "os": parsed["os"],
                 "clicked_at": now,
             }))
-            link = asyncio.run(self._get_client().find_one("url_shortener_links", "short_code", short_code))
+            link = _run_async(self._get_client().find_one("url_shortener_links", "short_code", short_code))
             if link:
                 current_clicks = (link.get("click_count") or 0) + 1
-                asyncio.run(self._get_client().update(
+                _run_async(self._get_client().update(
                     "url_shortener_links", "short_code", short_code,
                     {"click_count": current_clicks, "last_accessed_at": now},
                 ))
 
     def get_click_analytics(self, short_code: str) -> Dict[str, Any]:
-        import asyncio
         client = self._get_client()
-        all_clicks = asyncio.run(client.select("url_shortener_clicks", eq=("short_code", short_code)))
+        all_clicks = _run_async(client.select("url_shortener_clicks", eq=("short_code", short_code)))
         total_val = len(all_clicks)
         browsers: Dict[str, int] = {}
         devices: Dict[str, int] = {}
@@ -471,14 +505,12 @@ class Storage:
         }
 
     def get_link_count_for_owner(self, owner_id: str) -> int:
-        import asyncio
-        links = asyncio.run(self._get_client().select("url_shortener_links", eq=("owner_id", owner_id)))
+        links = _run_async(self._get_client().select("url_shortener_links", eq=("owner_id", owner_id)))
         return len(links)
 
     def audit(self, owner_id: Optional[str], action: str, short_code: Optional[str], detail: str = ""):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().insert("url_shortener_audit_log", {
+            _run_async(self._get_client().insert("url_shortener_audit_log", {
                 "timestamp": _now_iso(),
                 "owner_id": owner_id,
                 "action": action,
@@ -487,8 +519,7 @@ class Storage:
             }))
 
     def export_links_csv(self, owner_id: str) -> str:
-        import asyncio
-        rows = asyncio.run(self._get_client().select(
+        rows = _run_async(self._get_client().select(
             "url_shortener_links", eq=("owner_id", owner_id), order=("created_at", True),
         ))
         buf = io.StringIO()
@@ -500,8 +531,7 @@ class Storage:
         return buf.getvalue()
 
     def export_clicks_csv(self, short_code: str) -> str:
-        import asyncio
-        rows = asyncio.run(self._get_client().select(
+        rows = _run_async(self._get_client().select(
             "url_shortener_clicks", eq=("short_code", short_code), order=("clicked_at", True),
         ))
         buf = io.StringIO()
@@ -513,15 +543,13 @@ class Storage:
         return buf.getvalue()
 
     def list_links_by_campaign(self, campaign_id: str) -> List[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().select(
+        return _run_async(self._get_client().select(
             "url_shortener_links", eq=("campaign_id", campaign_id), order=("created_at", True),
         ))
 
     def create_campaign(self, campaign_id: str, owner_id: str, name: str, description: Optional[str] = None):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().insert("url_shortener_campaigns", {
+            _run_async(self._get_client().insert("url_shortener_campaigns", {
                 "campaign_id": campaign_id,
                 "owner_id": owner_id,
                 "name": name,
@@ -530,41 +558,35 @@ class Storage:
             }))
 
     def get_campaign(self, campaign_id: str) -> Optional[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().find_one("url_shortener_campaigns", "campaign_id", campaign_id))
+        return _run_async(self._get_client().find_one("url_shortener_campaigns", "campaign_id", campaign_id))
 
     def list_campaigns_for_owner(self, owner_id: str) -> List[dict]:
-        import asyncio
-        return asyncio.run(self._get_client().select(
+        return _run_async(self._get_client().select(
             "url_shortener_campaigns", eq=("owner_id", owner_id), order=("created_at", True),
         ))
 
     def deactivate_campaign(self, campaign_id: str):
         with self._lock:
-            import asyncio
-            asyncio.run(self._get_client().update("url_shortener_campaigns", "campaign_id", campaign_id, {"is_active": 0}))
+            _run_async(self._get_client().update("url_shortener_campaigns", "campaign_id", campaign_id, {"is_active": 0}))
 
     def get_campaign_link_count(self, campaign_id: str) -> int:
-        import asyncio
-        links = asyncio.run(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
+        links = _run_async(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
         return len(links)
 
     def get_campaign_total_clicks(self, campaign_id: str) -> int:
-        import asyncio
-        links = asyncio.run(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
+        links = _run_async(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
         return sum(l.get("click_count") or 0 for l in links)
 
     def get_campaign_analytics(self, campaign_id: str) -> Dict[str, Any]:
-        import asyncio
         client = self._get_client()
-        codes = asyncio.run(client.select("url_shortener_links", eq=("campaign_id", campaign_id)))
+        codes = _run_async(client.select("url_shortener_links", eq=("campaign_id", campaign_id)))
         total = 0
         browsers: Dict[str, int] = {}
         devices: Dict[str, int] = {}
         os_data: Dict[str, int] = {}
         for row in codes:
             code = row["short_code"]
-            clicks = asyncio.run(client.select("url_shortener_clicks", eq=("short_code", code)))
+            clicks = _run_async(client.select("url_shortener_clicks", eq=("short_code", code)))
             total += len(clicks)
             for c in clicks:
                 b = c.get("browser")
@@ -582,9 +604,8 @@ class Storage:
         row = self.get_owner(owner_id)
         if row is None:
             return {}
-        import asyncio
         client = self._get_client()
-        links = asyncio.run(client.select("url_shortener_links", eq=("owner_id", owner_id)))
+        links = _run_async(client.select("url_shortener_links", eq=("owner_id", owner_id)))
         link_count = len(links)
         total_clicks = sum(l.get("click_count") or 0 for l in links)
         active = sum(1 for l in links if l.get("is_active"))
@@ -949,11 +970,10 @@ class URLShortenerService:
             raise LinkNotFoundError(f"No link found for code '{short_code}'")
         if row["owner_id"] != owner_id:
             raise AuthorizationError("You do not own this link")
-        import asyncio
         client = get_supabase_client()
         if client is None:
             return []
-        rows = asyncio.run(client.select(
+        rows = _run_async(client.select(
             "url_shortener_clicks", eq=("short_code", short_code),
         ))
         return list(rows)[:limit]

app/services/web_search_service.py

@@ -7,6 +7,7 @@ from typing import Any, Dict, List, Optional
 import aiohttp
 
 from app.config import get_settings
+from app.utils.http_utils import shared_aiohttp_session
 
 _logger = logging.getLogger(__name__)
 _settings = get_settings()
@@ -60,8 +61,8 @@ class WebSearchService:
         for attempt in range(1 + self._max_retries):
             try:
                 timeout = aiohttp.ClientTimeout(total=self._timeout)
-                async with aiohttp.ClientSession(timeout=timeout) as session:
-                    async with session.get(url, params=params) as resp:
+                async with shared_aiohttp_session() as session:
+                    async with session.get(url, params=params, timeout=timeout) as resp:
                         if resp.status == 403:
                             last_error = (
                                 "SearXNG instance returned 403 Forbidden. "
@@ -145,8 +146,8 @@ class WebSearchService:
         url = f"{self._base_url}/config"
         try:
             timeout = aiohttp.ClientTimeout(total=self._timeout)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.get(url) as resp:
+            async with shared_aiohttp_session() as session:
+                async with session.get(url, timeout=timeout) as resp:
                     if resp.status != 200:
                         return {
                             "success": False,
@@ -184,8 +185,8 @@ class WebSearchService:
         params = {"q": query}
         try:
             timeout = aiohttp.ClientTimeout(total=10)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.get(url, params=params) as resp:
+            async with shared_aiohttp_session() as session:
+                async with session.get(url, params=params, timeout=timeout) as resp:
                     if resp.status != 200:
                         return {"success": False, "error": f"HTTP {resp.status} from SearXNG autocompleter"}
                     data = await resp.json()
@@ -205,8 +206,8 @@ class WebSearchService:
         url = f"{self._base_url}/engine_descriptions.json"
         try:
             timeout = aiohttp.ClientTimeout(total=self._timeout)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.get(url) as resp:
+            async with shared_aiohttp_session() as session:
+                async with session.get(url, timeout=timeout) as resp:
                     if resp.status != 200:
                         return {"success": False, "error": f"HTTP {resp.status}"}
                     data = await resp.json()
@@ -218,8 +219,8 @@ class WebSearchService:
         url = f"{self._base_url}/stats"
         try:
             timeout = aiohttp.ClientTimeout(total=self._timeout)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.get(url) as resp:
+            async with shared_aiohttp_session() as session:
+                async with session.get(url, timeout=timeout) as resp:
                     if resp.status != 200:
                         return {"success": False, "error": f"HTTP {resp.status}"}
                     data = await resp.json()
@@ -231,8 +232,8 @@ class WebSearchService:
         url = f"{self._base_url}/healthz"
         try:
             timeout = aiohttp.ClientTimeout(total=10)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.get(url) as resp:
+            async with shared_aiohttp_session() as session:
+                async with session.get(url, timeout=timeout) as resp:
                     return {
                         "success": resp.status == 200,
                         "status": "healthy" if resp.status == 200 else f"unhealthy (HTTP {resp.status})",

app/utils/http_utils.py

@@ -2,11 +2,13 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from contextlib import asynccontextmanager
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import Any, AsyncIterator, Dict, List, Optional, Tuple
 from urllib.parse import unquote, urlparse
 
 import aiohttp
+import httpx
 
 from app.config import get_settings
 
@@ -14,6 +16,64 @@ _logger = logging.getLogger(__name__)
 _settings = get_settings()
 
 
+class DownloadError(RuntimeError):
+    """Raised when a remote download fails (HTTP error, timeout, or size limit).
+
+    ``is_size_error`` distinguishes size-limit violations so callers can map
+    the failure to the appropriate domain exception.
+    """
+
+    def __init__(self, message: str, *, is_size_error: bool = False) -> None:
+        super().__init__(message)
+        self.is_size_error = is_size_error
+
+
+class SharedAsyncClient:
+    """Thread-safe, lazily-created, connection-pooled ``httpx.AsyncClient``.
+
+    Several services (GCS, Google Maps, Google OAuth, scheduler) previously
+    duplicated the same double-checked-lock initialisation and shutdown logic.
+    This helper centralises that pattern so a single instance is created on
+    first use and safely closed exactly once on shutdown.
+    """
+
+    def __init__(
+        self,
+        *,
+        timeout: Any = 30.0,
+        limits: Optional[httpx.Limits] = None,
+        follow_redirects: bool = True,
+    ) -> None:
+        self._client: Optional[httpx.AsyncClient] = None
+        self._lock = asyncio.Lock()
+        self._timeout = timeout
+        self._limits = limits or httpx.Limits(
+            max_connections=100,
+            max_keepalive_connections=20,
+            keepalive_expiry=30,
+        )
+        self._follow_redirects = follow_redirects
+
+    async def get(self) -> httpx.AsyncClient:
+        """Return the shared client, creating it lazily on first use."""
+        if self._client is None or self._client.is_closed:
+            async with self._lock:
+                if self._client is None or self._client.is_closed:
+                    self._client = httpx.AsyncClient(
+                        timeout=self._timeout,
+                        follow_redirects=self._follow_redirects,
+                        limits=self._limits,
+                    )
+        return self._client
+
+    async def close(self) -> None:
+        """Close the underlying client and release pooled connections."""
+        async with self._lock:
+            if self._client is not None and not self._client.is_closed:
+                await self._client.aclose()
+            self._client = None
+
+
 def create_session_timeout(total_seconds: Optional[float] = None) -> aiohttp.ClientTimeout:
     """Create an aiohttp timeout using the configured request timeout."""
     if total_seconds is None:
@@ -21,6 +81,63 @@ def create_session_timeout(total_seconds: Optional[float] = None) -> aiohttp.Cli
     return aiohttp.ClientTimeout(total=total_seconds)
 
 
+class SharedAiohttpSession:
+    """Reuse a single ``aiohttp.ClientSession`` per event loop.
+
+    aiohttp sessions are bound to the loop they are created on, so this caches
+    one session per event loop. In the long-lived server loop this removes the
+    per-request session creation (and the associated TCP/TLS handshake) from hot
+    paths such as chat completions and web search.
+    """
+
+    def __init__(self) -> None:
+        self._sessions: Dict[int, aiohttp.ClientSession] = {}
+
+    async def get(self, timeout: Optional[aiohttp.ClientTimeout] = None) -> aiohttp.ClientSession:
+        loop = asyncio.get_running_loop()
+        loop_id = id(loop)
+        session = self._sessions.get(loop_id)
+        if session is None or session.closed:
+            session = aiohttp.ClientSession(timeout=timeout or create_session_timeout())
+            self._sessions[loop_id] = session
+        return session
+
+    async def close_all(self) -> None:
+        sessions = list(self._sessions.values())
+        self._sessions.clear()
+        for session in sessions:
+            if session is not None and not session.closed:
+                await session.close()
+
+
+_shared_aiohttp_session = SharedAiohttpSession()
+
+
+async def get_shared_aiohttp_session(
+    timeout: Optional[aiohttp.ClientTimeout] = None,
+) -> aiohttp.ClientSession:
+    """Return the loop-bound shared aiohttp session, created lazily on first use."""
+    return await _shared_aiohttp_session.get(timeout)
+
+
+@asynccontextmanager
+async def shared_aiohttp_session(
+    timeout: Optional[aiohttp.ClientTimeout] = None,
+) -> AsyncIterator[aiohttp.ClientSession]:
+    """Context manager that yields the shared session but does NOT close it on exit.
+
+    Drop-in replacement for ``async with aiohttp.ClientSession(...)`` so call
+    sites keep the same structure while reusing one pooled session per loop.
+    """
+    session = await get_shared_aiohttp_session(timeout)
+    yield session
+
+
+async def close_shared_aiohttp_sessions() -> None:
+    """Close all cached sessions. Call once on application shutdown."""
+    await _shared_aiohttp_session.close_all()
+
+
 async def download_url(
     url: str,
     *,
@@ -36,15 +153,16 @@ async def download_url(
     """
     timeout = aiohttp.ClientTimeout(total=timeout_seconds)
     try:
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.get(url) as resp:
+        session = await get_shared_aiohttp_session()
+        async with session.get(url, timeout=timeout) as resp:
                 if resp.status != 200:
-                    raise RuntimeError(f"HTTP {resp.status} when fetching {url}")
+                    raise DownloadError(f"HTTP {resp.status} when fetching {url}")
 
                 if max_size_bytes and resp.content_length and resp.content_length > max_size_bytes:
-                    raise RuntimeError(
+                    raise DownloadError(
                         f"Remote file advertises {resp.content_length} bytes, "
-                        f"limit is {max_size_bytes}"
+                        f"limit is {max_size_bytes}",
+                        is_size_error=True,
                     )
 
                 chunks: List[bytes] = []
@@ -52,12 +170,15 @@ async def download_url(
                 async for chunk in resp.content.iter_chunked(chunk_size):
                     total += len(chunk)
                     if max_size_bytes and total > max_size_bytes:
-                        raise RuntimeError(f"Download exceeded {max_size_bytes} bytes")
+                        raise DownloadError(
+                            f"Download exceeded {max_size_bytes} bytes",
+                            is_size_error=True,
+                        )
                     chunks.append(chunk)
 
                 data = b"".join(chunks)
     except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
-        raise RuntimeError(f"Download failed for {url}: {exc}") from exc
+        raise DownloadError(f"Download failed for {url}: {exc}") from exc
 
     parsed = urlparse(url)
     filename = unquote(Path(parsed.path).name) if parsed.path else None