n8n use passed token

application/use_cases/process_job.py

@@ -88,18 +88,25 @@ class ProcessJobUseCase:
                 processing_time=processing_time
             )
 
-            # Get job record to retrieve bearer token
-            job_record = await self.job_repository.get(job_id)
-            bearer_token = job_record.bearer_token if job_record else None
-
-            await self.notification_service.send_job_completion_notification(
-                job_id=job_id,
-                status="completed",
-                processing_time=processing_time,
-                bearer_token=bearer_token
-            )
+            # Send notification if service is available and we have a bearer token
+            if self.notification_service:
+                # Get job record to retrieve bearer token and external job ID
+                job_record = await self.job_repository.get(job_id)
+                bearer_token = job_record.bearer_token if job_record else None
+                external_job_id = job_record.external_job_id if job_record else None
+                
+                if bearer_token:
+                    await self.notification_service.send_job_completion_notification(
+                        job_id=job_id,
+                        status="completed",
+                        processing_time=processing_time,
+                        bearer_token=bearer_token,
+                        external_job_id=external_job_id
+                    )
+                else:
+                    logger.debug(f"Skipping N8N notification for job {job_id} - no bearer token available")
             
-            # Clear bearer token for security after notification is sent
+            # Clear bearer token for security (regardless of notification status)
             await self.job_repository.clear_bearer_token(job_id)
             
             logger.info(f"Job {job_id} completed in {processing_time:.2f} seconds")
@@ -115,29 +122,32 @@ class ProcessJobUseCase:
                 processing_time=processing_time
             )
             
-            # Get job record to retrieve bearer token for failure notification
+            # Send failure notification if service is available and we have a bearer token
+            if self.notification_service:
+                try:
+                    job_record = await self.job_repository.get(job_id)
+                    bearer_token = job_record.bearer_token if job_record else None
+                    external_job_id = job_record.external_job_id if job_record else None
+                    
+                    if bearer_token:
+                        await self.notification_service.send_job_completion_notification(
+                            job_id=job_id,
+                            status="failed",
+                            processing_time=processing_time,
+                            bearer_token=bearer_token,
+                            external_job_id=external_job_id
+                        )
+                    else:
+                        logger.debug(f"Skipping N8N failure notification for job {job_id} - no bearer token available")
+                    
+                except Exception as notify_error:
+                    # Don't let notification failures mask the original job failure
+                    logger.warning(f"Failed to send failure notification for job {job_id}: {notify_error}")
+            
+            # Clear bearer token for security (regardless of notification status)
             try:
-                job_record = await self.job_repository.get(job_id)
-                bearer_token = job_record.bearer_token if job_record else None
-                
-                await self.notification_service.send_job_completion_notification(
-                    job_id=job_id,
-                    status="failed",
-                    processing_time=processing_time,
-                    bearer_token=bearer_token
-                )
-                
-                # Clear bearer token for security after notification is sent
                 await self.job_repository.clear_bearer_token(job_id)
-                
-            except Exception as notify_error:
-                # Don't let notification failures mask the original job failure
-                logger.warning(f"Failed to send failure notification for job {job_id}: {notify_error}")
-                
-                # Still clear the token even if notification failed
-                try:
-                    await self.job_repository.clear_bearer_token(job_id)
-                except Exception:
-                    logger.warning(f"Failed to clear bearer token for failed job {job_id}")
+            except Exception:
+                logger.warning(f"Failed to clear bearer token for failed job {job_id}")
             
             raise

domain/services/notification_service.py

@@ -17,5 +17,6 @@ class NotificationService(Protocol):
                                              job_id: str, 
                                              status: str,
                                              processing_time: float,
-                                             bearer_token: Optional[str] = None) -> NotificationResponse:
+                                             bearer_token: str,
+                                             external_job_id: Optional[str] = None) -> NotificationResponse:
         ...

infrastructure/clients/n8n/models.py

@@ -18,6 +18,6 @@ class WebhooksResponse:
 class N8NClientProtocol(Protocol):
     """Protocol defining the API client interface."""
     
-    async def post_completion_event(self, data: WebhooksRequest, bearer_token: Optional[str] = None) -> WebhooksResponse:
-        """Post to webhooks endpoint with optional client bearer token."""
+    async def post_completion_event(self, data: WebhooksRequest, bearer_token: str, external_job_id: Optional[str] = None) -> WebhooksResponse:
+        """Post to webhooks endpoint with client bearer token for authentication."""
         ...

infrastructure/clients/n8n/n8n_client.py

@@ -21,9 +21,8 @@ class N8NClient(N8NClientProtocol):
         )
     
     def _get_default_headers(self) -> Dict[str, str]:
-        """Get default headers including authentication."""
+        """Get default headers (no authentication - client token will be provided per request)."""
         return {
-            "Authorization": f"Bearer {self.settings.token}",
             "Content-Type": "application/json",
             "Accept": "application/json"
         }
@@ -88,19 +87,26 @@ class N8NClient(N8NClientProtocol):
             self.logger.error(f"Failed to parse JSON response from {url}: {str(e)}")
             raise APIResponseError(f"Invalid JSON response: {str(e)}", response.status_code)
     
-    async def post_completion_event(self, data: WebhooksRequest, bearer_token: Optional[str] = None) -> WebhooksResponse:
-        """Post to webhooks endpoint with optional client bearer token."""
+    async def post_completion_event(self, data: WebhooksRequest, bearer_token: str, external_job_id: Optional[str] = None) -> WebhooksResponse:
+        """Post to webhooks endpoint with client bearer token for authentication."""
         from dataclasses import asdict
         payload = asdict(data)
         
         # Extract job_id for header and remove from payload
         job_id = payload.pop("job_id")
-        custom_headers = {"rowID": job_id}
         
-        # Add client bearer token if provided
-        if bearer_token:
-            custom_headers["Authorization"] = f"Bearer {bearer_token}"
-            self.logger.debug(f"Adding client bearer token to N8N request for job {job_id}")
+        # Client bearer token is required for authentication to N8N
+        custom_headers = {
+            "rowID": job_id,
+            "Authorization": f"Bearer {bearer_token}"
+        }
+        
+        # Add external job ID as custom header if provided
+        if external_job_id:
+            custom_headers["externalJobID"] = external_job_id
+            self.logger.debug(f"Sending N8N notification for job {job_id} with external job ID {external_job_id}")
+        else:
+            self.logger.debug(f"Sending N8N notification for job {job_id} using client bearer token")
         
         response_data = await self._make_request("POST", "/lovable-analysis", payload, custom_headers)
         

infrastructure/clients/n8n/settings.py

@@ -6,5 +6,4 @@ from typing import Optional
 class ClientSettings:
     """Settings for the HTTP client."""
     base_url: str
-    token: str
     timeout: Optional[int] = 30

infrastructure/config/settings.py

@@ -88,8 +88,8 @@ class Settings(BaseSettings):
 
     # N8N Configuration
     n8n_base_url: str = Field(default="http://localhost:5678", env="N8N_BASE_URL")
-    n8n_token: str = Field(env="N8N_TOKEN")
     n8n_timeout: int = Field(default=30, env="N8N_TIMEOUT")
+    n8n_enabled: bool = Field(default=True, env="N8N_ENABLED")
     
     # Authentication Configuration
     enforce_authentication: bool = Field(default=True, env="ENFORCE_AUTHENTICATION")

infrastructure/services/container.py

@@ -61,12 +61,16 @@ class ServiceContainer:
         else:
             raise ValueError(f"Unsupported storage type: {settings.storage_type}")
     
-    def _create_notification_service(self) -> N8NNotificationService:
-        """Create N8N notification service."""
-        # Create N8N client settings
+    def _create_notification_service(self) -> Optional[N8NNotificationService]:
+        """Create N8N notification service if enabled."""
+        # Check if N8N is enabled
+        if not settings.n8n_enabled:
+            logging.getLogger(__name__).info("N8N notifications disabled via configuration")
+            return None
+        
+        # Create N8N client settings (no token needed - client tokens are passed per request)
         n8n_settings = ClientSettings(
             base_url=settings.n8n_base_url,
-            token=settings.n8n_token,
             timeout=getattr(settings, 'n8n_timeout', 30)
         )
         
@@ -93,8 +97,10 @@ class ServiceContainer:
     async def shutdown(self):
         """Cleanup services on shutdown."""
         await self.cleanup_service.stop()
-        # Close N8N client if it has async cleanup
-        if hasattr(self.notification_service.n8n_client, 'close'):
+        # Close N8N client if it exists and has async cleanup
+        if (self.notification_service and 
+            hasattr(self.notification_service, 'n8n_client') and
+            hasattr(self.notification_service.n8n_client, 'close')):
             await self.notification_service.n8n_client.close()
 
 # Convenience function for getting services

infrastructure/services/n8n_notification_service.py

@@ -25,19 +25,17 @@ class N8NNotificationService(NotificationService):
                                              job_id: str, 
                                              status: str,
                                              processing_time: float,
-                                             bearer_token: Optional[str] = None) -> NotificationResponse:
-        """Send job completion notification via N8N with optional client bearer token."""
+                                             bearer_token: str,
+                                             external_job_id: Optional[str] = None) -> NotificationResponse:
+        """Send job completion notification via N8N with client bearer token for authentication."""
         try:
             message = f"Job {job_id} {status} in {processing_time:.2f}s"
             request = NotificationRequest(message=message, job_id=job_id)
             
-            # Pass bearer token to N8N client if provided
-            response = await self.n8n_client.post_completion_event(request, bearer_token)
+            # Pass client bearer token and external job ID to N8N client
+            response = await self.n8n_client.post_completion_event(request, bearer_token, external_job_id)
             
-            if bearer_token:
-                logger.debug(f"Sent N8N notification for job {job_id} with client bearer token")
-            else:
-                logger.debug(f"Sent N8N notification for job {job_id} without client bearer token")
+            logger.debug(f"Sent N8N notification for job {job_id} using client bearer token")
                 
             return NotificationResponse(acknowledged=response.acknowledged)