fix: credit tracking, voice clone TTL, avatar upload ui, asset serving fallback, OAuth encryption, free plan video renders, backlink outreach sprint

backend/services/backlink_outreach_models.py

@@ -106,22 +106,138 @@ class CampaignDetailResponse(BaseModel):
     leads: List[LeadRecord] = Field(default_factory=list)
 
 
+class GenerateEmailRequest(BaseModel):
+    topic: str = Field(..., min_length=2, max_length=500)
+    target_site: Optional[str] = Field(None, description="Target website for guest post pitch")
+    tone: str = Field(default="professional", pattern="^(professional|friendly|casual|formal)$")
+    existing_template_id: Optional[str] = None
+
+
 class GeneratedEmailResponse(BaseModel):
     subject: str
     body: str
 
 
+class PersonalizeEmailRequest(BaseModel):
+    lead_name: str = Field(..., min_length=1, max_length=200)
+    lead_site: str = Field(..., min_length=1, max_length=500)
+    lead_content_topic: str = Field(..., min_length=1, max_length=500)
+    pitch_topic: str = Field(..., min_length=2, max_length=500)
+    existing_body: str = Field(default="", max_length=10000)
+
+
+class SubjectLinesRequest(BaseModel):
+    body: str = Field(..., min_length=10, max_length=10000)
+    count: int = Field(default=5, ge=1, le=10)
+
+
+class SubjectLinesResponse(BaseModel):
+    subjects: list[str]
+
+
+class FollowUpRequest(BaseModel):
+    original_subject: str = Field(..., min_length=1, max_length=500)
+    original_body: str = Field(..., min_length=10, max_length=10000)
+    days_elapsed: int = Field(default=7, ge=1, le=90)
+    reply_context: str = Field(default="", max_length=2000)
+
+
 class OutreachStatusRecord(BaseModel):
     opportunity_url: HttpUrl
     status: str
     notes: Optional[str] = None
 
 
+class SendOutreachRequest(BaseModel):
+    lead_id: str = Field(..., min_length=1)
+    campaign_id: str = Field(..., min_length=1)
+    user_id: str = Field(..., min_length=1)
+    workspace_id: str = Field(default="default")
+    sender_email: str = Field(..., min_length=3)
+    subject: str = Field(..., min_length=1)
+    body: str = Field(..., min_length=1)
+    idempotency_key: str = Field(..., min_length=8)
+    template_id: Optional[str] = Field(None, description="Optional template ID for personalization")
+    template_variables: Optional[dict] = Field(None, description="Variable values for template personalization")
+
+
+class SendOutreachResponse(BaseModel):
+    attempt_id: str
+    status: str
+    policy_allowed: bool
+    policy_reasons: List[str] = Field(default_factory=list)
+
+
+class OutreachAttemptRecord(BaseModel):
+    attempt_id: str
+    lead_id: str
+    campaign_id: str
+    idempotency_key: str
+    sender_email: Optional[str] = None
+    subject: Optional[str] = None
+    status: str = "queued"
+    decision_reason: Optional[str] = None
+    sent_at: Optional[str] = None
+    created_at: Optional[str] = None
+
+
+class OutreachAttemptListResponse(BaseModel):
+    attempts: List[OutreachAttemptRecord]
+    total: int
+
+
+class OutreachReplyRecord(BaseModel):
+    reply_id: str
+    attempt_id: str
+    from_email: Optional[str] = None
+    subject: Optional[str] = None
+    received_at: Optional[str] = None
+    classification: str = "replied"
+    body: Optional[str] = None
+
+
+class OutreachReplyListResponse(BaseModel):
+    replies: List[OutreachReplyRecord]
+    total: int
+
+
+class ScheduleFollowUpRequest(BaseModel):
+    attempt_id: str = Field(..., min_length=1)
+    scheduled_for: str = Field(..., min_length=1)
+    subject: Optional[str] = None
+    body: Optional[str] = None
+
+
+class FollowUpScheduleRecord(BaseModel):
+    schedule_id: str
+    attempt_id: str
+    subject: Optional[str] = None
+    scheduled_for: str
+    sent: bool = False
+
+
+class EmailTemplateRequest(BaseModel):
+    name: str = Field(..., min_length=1)
+    subject_template: str = Field(..., min_length=1)
+    body_template: str = Field(..., min_length=1)
+    variables: Optional[List[str]] = None
+
+
+class EmailTemplateRecord(BaseModel):
+    template_id: str
+    user_id: str
+    name: str
+    subject_template: str
+    body_template: str
+    variables: Optional[List[str]] = None
+    created_at: Optional[str] = None
+
+
 class PolicyValidationRequest(BaseModel):
     user_id: str = Field(..., min_length=1)
     workspace_id: str = Field(..., min_length=1)
     campaign_id: str = Field(..., min_length=1)
-    recipient_email: EmailStr
+    recipient_email: str = Field(..., min_length=1)
     recipient_domain: str
     recipient_region: str = Field(default="unknown")
     legal_basis: str = Field(..., min_length=2)
@@ -135,3 +251,61 @@ class PolicyValidationResponse(BaseModel):
     allowed: bool
     reasons: List[str] = Field(default_factory=list)
     final_status: str
+
+
+# -- Analytics & Reporting Models --
+
+class CampaignAnalyticsResponse(BaseModel):
+    campaign_id: str
+    lead_count: int = 0
+    send_volume: int = 0
+    blocked_count: int = 0
+    reply_count: int = 0
+    response_rate: float = 0.0
+    placement_rate: float = 0.0
+    reply_classification: Dict[str, int] = Field(default_factory=dict)
+
+
+class BacklinkReportingSnapshot(BaseModel):
+    send_volume: int = 0
+    decision_events: int = 0
+    response_rate: float = 0.0
+    placement_conversion: float = 0.0
+
+
+class CampaignVolumePoint(BaseModel):
+    date: str
+    count: int = 0
+
+
+class CampaignVolumeResponse(BaseModel):
+    campaign_id: str
+    days: int = 30
+    volume: List[CampaignVolumePoint] = Field(default_factory=list)
+
+
+class FunnelStage(BaseModel):
+    status: str
+    count: int = 0
+
+
+class ConversionFunnelResponse(BaseModel):
+    campaign_id: str
+    stages: List[FunnelStage] = Field(default_factory=list)
+
+
+class BulkStatusUpdateRequest(BaseModel):
+    lead_ids: List[str] = Field(..., min_length=1)
+    status: str = Field(..., min_length=1)
+    notes: Optional[str] = None
+
+
+class BulkStatusUpdateResponse(BaseModel):
+    updated: int = 0
+    failed: List[str] = Field(default_factory=list)
+
+
+class SuppressionAddRequest(BaseModel):
+    email: str = Field(..., min_length=3)
+    reason: str = Field(default="")
+    domain: str = Field(default="")

backend/services/backlink_outreach_reply_monitor.py

@@ -0,0 +1,164 @@
+"""IMAP-based reply monitoring for backlink outreach."""
+
+from __future__ import annotations
+
+import os
+import asyncio
+import imaplib
+import email as email_lib
+from email.utils import parsedate_to_datetime
+from typing import List, Optional
+from loguru import logger
+
+
+IMAP_HOST = os.getenv("IMAP_HOST", "imap.gmail.com")
+IMAP_PORT = int(os.getenv("IMAP_PORT", "993"))
+IMAP_USERNAME = os.getenv("IMAP_USERNAME", "")
+IMAP_PASSWORD = os.getenv("IMAP_PASSWORD", "")
+IMAP_FOLDER = os.getenv("IMAP_FOLDER", "INBOX")
+IMAP_FETCH_LIMIT = int(os.getenv("IMAP_FETCH_LIMIT", "50"))
+
+# Search keywords for auto-classification
+INTERESTED_KEYWORDS = [
+    "interested", "let's discuss", "sounds good", "would love to", "yes",
+    "sure", "tell me more", "looks good", "happy to", "let's do it",
+    "sign me up", "count me in", "proceed", "approved",
+]
+NOT_INTERESTED_KEYWORDS = [
+    "not interested", "unsubscribe", "no thanks", "remove me", "stop",
+    "don't contact", "spam", "not relevant", "no longer interested",
+    "please stop", "do not email",
+]
+OUT_OF_OFFICE_KEYWORDS = [
+    "out of office", "vacation", "on leave", "away from", "return on",
+    "not in the office", "will be back",
+]
+
+
+class BacklinkOutreachReplyMonitor:
+    def __init__(self):
+        self._host = IMAP_HOST
+        self._port = IMAP_PORT
+        self._username = IMAP_USERNAME
+        self._password = IMAP_PASSWORD
+        self._folder = IMAP_FOLDER
+        self._fetch_limit = IMAP_FETCH_LIMIT
+
+    def is_configured(self) -> bool:
+        return bool(self._username and self._password)
+
+    async def poll_replies(self, sent_from_email: str) -> List[dict]:
+        """Poll IMAP inbox for replies to a specific sender address."""
+        if not self.is_configured():
+            logger.warning("IMAP not configured: set IMAP_USERNAME and IMAP_PASSWORD")
+            return []
+
+        loop = asyncio.get_running_loop()
+
+        def _poll() -> List[dict]:
+            try:
+                mail = imaplib.IMAP4_SSL(self._host, self._port)
+                mail.login(self._username, self._password)
+                mail.select(self._folder)
+
+                safe_email = sent_from_email.replace('"', "").replace("\\", "")
+                search_criteria = f'(TO "{safe_email}")'
+                status, message_ids = mail.search(None, search_criteria)
+                if status != "OK":
+                    return []
+
+                ids = message_ids[0].split() if message_ids[0] else []
+                if not ids:
+                    return []
+
+                ids = ids[-self._fetch_limit:]
+
+                replies = []
+                for mid in ids:
+                    status, msg_data = mail.fetch(mid, "(RFC822)")
+                    if status != "OK":
+                        continue
+
+                    raw_email = msg_data[0][1] if msg_data else None
+                    if not raw_email:
+                        continue
+
+                    parsed = email_lib.message_from_bytes(raw_email)
+                    reply = self._parse_reply(parsed)
+                    if reply:
+                        replies.append(reply)
+
+                mail.logout()
+                return replies
+            except imaplib.IMAP4.error as e:
+                logger.error(f"IMAP error: {e}")
+                return []
+            except Exception as e:
+                logger.error(f"Unexpected IMAP error: {e}")
+                return []
+
+        return await loop.run_in_executor(None, _poll)
+
+    def _parse_reply(self, parsed_msg) -> Optional[dict]:
+        try:
+            from_email = parsed_msg.get("From", "")
+            subject = parsed_msg.get("Subject", "")
+            received_at = parsed_msg.get("Date", "")
+
+            # Extract body
+            body = ""
+            if parsed_msg.is_multipart():
+                for part in parsed_msg.walk():
+                    content_type = part.get_content_type()
+                    if content_type == "text/plain":
+                        try:
+                            body = part.get_payload(decode=True).decode("utf-8", errors="ignore")
+                            break
+                        except Exception:
+                            continue
+            else:
+                try:
+                    body = parsed_msg.get_payload(decode=True).decode("utf-8", errors="ignore")
+                except Exception:
+                    body = str(parsed_msg.get_payload())
+
+            classification = self._classify_reply(body, subject)
+
+            # Parse date
+            try:
+                dt = parsedate_to_datetime(received_at)
+                received_at_iso = dt.isoformat() if dt else None
+            except Exception:
+                received_at_iso = None
+
+            return {
+                "from_email": from_email,
+                "subject": subject,
+                "body": body[:5000],
+                "classification": classification,
+                "received_at": received_at_iso,
+            }
+        except Exception as e:
+            logger.error(f"Failed to parse reply: {e}")
+            return None
+
+    @staticmethod
+    def _classify_reply(body: str, subject: str) -> str:
+        text = f"{subject} {body}".lower()
+
+        for kw in OUT_OF_OFFICE_KEYWORDS:
+            if kw in text:
+                return "out_of_office"
+
+        for kw in NOT_INTERESTED_KEYWORDS:
+            if kw in text:
+                return "not_interested"
+
+        for kw in INTERESTED_KEYWORDS:
+            if kw in text:
+                return "interested"
+
+        return "replied"
+
+
+backlink_outreach_reply_monitor = BacklinkOutreachReplyMonitor()

backend/services/backlink_outreach_sender.py

@@ -0,0 +1,90 @@
+"""Email sender for backlink outreach via SMTP."""
+
+from __future__ import annotations
+
+import os
+import ssl
+import smtplib
+import asyncio
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from typing import Optional
+from loguru import logger
+
+
+SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com")
+SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
+SMTP_USERNAME = os.getenv("SMTP_USERNAME", "")
+SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "")
+SMTP_FROM_EMAIL = os.getenv("SMTP_FROM_EMAIL", SMTP_USERNAME)
+SMTP_USE_TLS = os.getenv("SMTP_USE_TLS", "true").lower() in ("true", "1", "yes")
+SMTP_VERIFY_TLS = os.getenv("SMTP_VERIFY_TLS", "true").lower() in ("true", "1", "yes")
+SMTP_SEND_TIMEOUT = int(os.getenv("SMTP_SEND_TIMEOUT", "30"))
+
+
+class BacklinkOutreachSender:
+    def __init__(self):
+        self._host = SMTP_HOST
+        self._port = SMTP_PORT
+        self._username = SMTP_USERNAME
+        self._password = SMTP_PASSWORD
+        self._from_email = SMTP_FROM_EMAIL or SMTP_USERNAME
+        self._use_tls = SMTP_USE_TLS
+        self._verify_tls = SMTP_VERIFY_TLS
+        self._timeout = SMTP_SEND_TIMEOUT
+
+    def is_configured(self) -> bool:
+        return bool(self._username and self._password)
+
+    async def send_email(
+        self,
+        to_email: str,
+        subject: str,
+        body: str,
+        from_email: Optional[str] = None,
+    ) -> bool:
+        if not self.is_configured():
+            logger.error("SMTP not configured: set SMTP_USERNAME and SMTP_PASSWORD")
+            return False
+
+        sender = from_email or self._from_email
+
+        msg = MIMEMultipart("alternative")
+        msg["From"] = sender
+        msg["To"] = to_email
+        msg["Subject"] = subject
+        msg.attach(MIMEText(body, "plain"))
+
+        loop = asyncio.get_running_loop()
+
+        def _send() -> bool:
+            try:
+                tls_context = ssl.create_default_context()
+                if not self._verify_tls:
+                    tls_context.check_hostname = False
+                    tls_context.verify_mode = ssl.CERT_NONE
+                with smtplib.SMTP(self._host, self._port, timeout=self._timeout) as server:
+                    if self._use_tls:
+                        server.starttls(context=tls_context)
+                        server.ehlo()
+                    server.login(self._username, self._password)
+                    server.sendmail(sender, [to_email], msg.as_string())
+                logger.info(f"Email sent to {to_email}: {subject[:60]}")
+                return True
+            except smtplib.SMTPException as e:
+                logger.error(f"SMTP error sending to {to_email}: {e}")
+                return False
+            except Exception as e:
+                logger.error(f"Unexpected error sending to {to_email}: {e}")
+                return False
+
+        return await loop.run_in_executor(None, _send)
+
+    def personalize(self, template: str, variables: dict) -> str:
+        """Replace {placeholder} variables in a template string."""
+        for key, value in variables.items():
+            template = template.replace(f"{{{key}}}", str(value))
+        return template
+
+
+backlink_outreach_sender = BacklinkOutreachSender()

backend/services/backlink_outreach_service.py

@@ -3,24 +3,25 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 import re
 import time
 
 import requests
 from bs4 import BeautifulSoup
 
-from services.backlink_outreach_models import OpportunityContactInfo, OpportunityRecord, PolicyValidationRequest, PolicyValidationResponse
+import csv
+import io
 
+from services.backlink_outreach_models import (
+    OpportunityContactInfo, OpportunityRecord,
+    PolicyValidationRequest, PolicyValidationResponse,
+    SendOutreachRequest, SendOutreachResponse,
+    CampaignVolumeResponse, CampaignVolumePoint,
+    ConversionFunnelResponse, FunnelStage,
+)
+from services.backlink_outreach_storage import BacklinkOutreachStorageService
 
-
-
-# Temporary in-memory control plane until DB wiring is complete
-SUPPRESSION_LIST = set()
-SENT_IDEMPOTENCY_KEYS = set()
-AUDIT_LOGS: list[dict] = []
-SEND_COUNTERS_BY_USER: dict[str, int] = {}
-SEND_COUNTERS_BY_DOMAIN: dict[str, int] = {}
 DEFAULT_USER_DAILY_CAP = 100
 DEFAULT_DOMAIN_DAILY_CAP = 20
 
@@ -140,8 +141,12 @@ class BacklinkOutreachService:
         return min(1.0, 0.35 + (0.13 * hits))
 
 
+    def _get_storage(self) -> BacklinkOutreachStorageService:
+        return BacklinkOutreachStorageService()
+
     def validate_send_policy(self, payload: PolicyValidationRequest) -> PolicyValidationResponse:
         reasons: List[str] = []
+        storage = self._get_storage()
 
         if payload.workspace_id.startswith("new-") and not payload.approved_by_human:
             reasons.append("human_review_required_for_new_workspace")
@@ -149,19 +154,17 @@ class BacklinkOutreachService:
             reasons.append("invalid_legal_basis")
         if payload.recipient_region.lower() in {"eu", "eea"} and payload.legal_basis.lower() != "consent":
             reasons.append("region_requires_explicit_consent")
-        if not payload.unsubscribe_url:
-            reasons.append("unsubscribe_url_required")
+
         if len(payload.sender_identity.strip()) < 3:
             reasons.append("sender_identity_required")
 
-        recipient_key = f"{payload.recipient_email.lower()}::{payload.recipient_domain.lower()}"
-        if recipient_key in SUPPRESSION_LIST:
+        if storage.is_suppressed(str(payload.recipient_email), payload.recipient_domain, user_id=payload.user_id):
             reasons.append("recipient_suppressed")
-        if payload.idempotency_key in SENT_IDEMPOTENCY_KEYS:
+        if storage.check_idempotency(payload.idempotency_key, user_id=payload.user_id):
             reasons.append("duplicate_idempotency_key")
 
-        user_count = SEND_COUNTERS_BY_USER.get(payload.user_id, 0)
-        domain_count = SEND_COUNTERS_BY_DOMAIN.get(payload.recipient_domain.lower(), 0)
+        user_count = storage.get_user_send_count(payload.user_id)
+        domain_count = storage.get_domain_send_count(payload.recipient_domain, user_id=payload.user_id)
         if user_count >= DEFAULT_USER_DAILY_CAP:
             reasons.append("user_daily_cap_exceeded")
         if domain_count >= DEFAULT_DOMAIN_DAILY_CAP:
@@ -170,33 +173,156 @@ class BacklinkOutreachService:
         allowed = len(reasons) == 0
         final_status = "approved" if allowed else "blocked"
 
-        AUDIT_LOGS.append({
-            "event": "policy_check",
-            "user_id": payload.user_id,
-            "campaign_id": payload.campaign_id,
-            "recipient": str(payload.recipient_email),
-            "allowed": allowed,
-            "reasons": reasons,
-            "override": payload.approved_by_human,
-        })
-
-        if allowed:
-            SENT_IDEMPOTENCY_KEYS.add(payload.idempotency_key)
-            SEND_COUNTERS_BY_USER[payload.user_id] = user_count + 1
-            SEND_COUNTERS_BY_DOMAIN[payload.recipient_domain.lower()] = domain_count + 1
+        storage.add_audit_log(
+            event="policy_check",
+            user_id=payload.user_id,
+            campaign_id=payload.campaign_id,
+            recipient=str(payload.recipient_email),
+            allowed=allowed,
+            reasons=reasons,
+            override=payload.approved_by_human,
+        )
 
         return PolicyValidationResponse(allowed=allowed, reasons=reasons, final_status=final_status)
 
-    def get_reporting_snapshot(self) -> Dict[str, Any]:
-        total_decisions = len(AUDIT_LOGS)
-        approved = sum(1 for row in AUDIT_LOGS if row.get("allowed"))
+    EU_DOMAIN_SUFFIXES = (".de", ".fr", ".it", ".es", ".nl", ".be", ".at", ".se", ".dk", ".fi", ".pt", ".ie", ".gr", ".pl", ".cz", ".ro", ".hu", ".bg", ".hr", ".sk", ".si", ".ee", ".lv", ".lt", ".lu", ".mt", ".cy")
+
+    def _infer_region(self, domain: str) -> str:
+        d = domain.lower()
+        if any(d.endswith(s) or d.endswith(s + "/") for s in self.EU_DOMAIN_SUFFIXES):
+            return "eu"
+        if d.endswith(".uk"):
+            return "uk"
+        if d.endswith(".ca"):
+            return "ca"
+        if d.endswith(".au"):
+            return "au"
+        return "unknown"
+
+    def send_outreach(self, request: SendOutreachRequest) -> SendOutreachResponse:
+        storage = self._get_storage()
+        lead = storage.get_lead(request.lead_id, user_id=request.user_id)
+        if not lead:
+            return SendOutreachResponse(attempt_id="", status="failed", policy_allowed=False, policy_reasons=["lead_not_found"])
+
+        domain = lead.get("domain", request.sender_email.split("@")[-1] if "@" in request.sender_email else "unknown")
+        recipient_region = self._infer_region(domain)
+        legal_basis = "consent" if recipient_region == "eu" else "legitimate_interest"
+
+        policy_req = PolicyValidationRequest(
+            user_id=request.user_id,
+            workspace_id=request.workspace_id,
+            campaign_id=request.campaign_id,
+            recipient_email=lead.get("email", ""),
+            recipient_domain=domain,
+            recipient_region=recipient_region,
+            legal_basis=legal_basis,
+            approved_by_human=False,
+            unsubscribe_url=None,
+            sender_identity=request.sender_email,
+            idempotency_key=request.idempotency_key,
+        )
+        policy = self.validate_send_policy(policy_req)
+
+        attempt = storage.add_attempt(
+            lead_id=request.lead_id,
+            campaign_id=request.campaign_id,
+            idempotency_key=request.idempotency_key,
+            sender_email=request.sender_email,
+            subject=request.subject,
+            body=request.body,
+            status="approved" if policy.allowed else "blocked",
+            decision_reason="; ".join(policy.reasons) if policy.reasons else None,
+            user_id=request.user_id,
+        )
+
+        return SendOutreachResponse(
+            attempt_id=attempt.get("attempt_id", ""),
+            status=attempt.get("status", "failed"),
+            policy_allowed=policy.allowed,
+            policy_reasons=policy.reasons,
+        )
+
+    def get_reporting_snapshot(self, user_id: str = "default") -> Dict[str, Any]:
+        storage = self._get_storage()
+        campaigns = storage.list_campaigns(user_id, user_id, limit=100)
+        total_sent = 0
+        total_replied = 0
+        total_placed = 0
+        total_leads = 0
+        for c in campaigns:
+            cid = c["campaign_id"]
+            attempts = storage.list_attempts(cid, limit=10000, user_id=user_id)
+            leads = storage.list_leads_all(cid, user_id=user_id)
+            total_sent += sum(1 for a in attempts if a.get("status") == "sent")
+            total_replied += storage.count_replies(cid, user_id=user_id)
+            total_placed += sum(1 for l in leads if l.get("status") == "placed")
+            total_leads += len(leads)
+        logs = storage.list_audit_logs("", limit=1000, user_id=user_id)
         return {
-            "send_volume": approved,
-            "decision_events": total_decisions,
-            "response_rate": 0.0,
-            "placement_conversion": 0.0,
+            "send_volume": total_sent,
+            "decision_events": len(logs),
+            "response_rate": round(total_replied / total_sent, 4) if total_sent > 0 else 0.0,
+            "placement_conversion": round(total_placed / total_leads, 4) if total_leads > 0 else 0.0,
         }
 
+    def get_campaign_volume(self, campaign_id: str, days: int = 30, user_id: str = "default") -> CampaignVolumeResponse:
+        storage = self._get_storage()
+        points = storage.get_send_volume_by_day(campaign_id, days, user_id=user_id)
+        return CampaignVolumeResponse(
+            campaign_id=campaign_id, days=days,
+            volume=[CampaignVolumePoint(**p) for p in points],
+        )
+
+    def get_campaign_funnel(self, campaign_id: str, user_id: str = "default") -> ConversionFunnelResponse:
+        storage = self._get_storage()
+        stages = storage.get_lead_status_counts(campaign_id, user_id=user_id)
+        return ConversionFunnelResponse(
+            campaign_id=campaign_id,
+            stages=[FunnelStage(**s) for s in stages],
+        )
+
+    CSV_LEAD_FIELDS = ["lead_id", "campaign_id", "domain", "page_title", "email", "status", "discovery_source", "created_at"]
+    CSV_ATTEMPT_FIELDS = ["attempt_id", "lead_id", "campaign_id", "sender_email", "subject", "status", "sent_at", "created_at"]
+    CSV_REPLY_FIELDS = ["reply_id", "attempt_id", "from_email", "subject", "classification", "received_at"]
+
+    @staticmethod
+    def _sanitize_csv_value(value: Any) -> str:
+        s = str(value) if value is not None else ""
+        if s and s[0] in ("=", "+", "-", "@", "\t", "\r"):
+            s = "'" + s
+        return s
+
+    def export_leads_csv(self, campaign_id: str, user_id: str = "default") -> str:
+        storage = self._get_storage()
+        leads = storage.list_leads_all(campaign_id, user_id=user_id)
+        output = io.StringIO()
+        writer = csv.DictWriter(output, fieldnames=self.CSV_LEAD_FIELDS, extrasaction="ignore")
+        writer.writeheader()
+        for row in leads:
+            writer.writerows([{k: self._sanitize_csv_value(v) for k, v in row.items()}])
+        return output.getvalue()
+
+    def export_attempts_csv(self, campaign_id: str, user_id: str = "default") -> str:
+        storage = self._get_storage()
+        attempts = storage.list_attempts_all(campaign_id, user_id=user_id)
+        output = io.StringIO()
+        writer = csv.DictWriter(output, fieldnames=self.CSV_ATTEMPT_FIELDS, extrasaction="ignore")
+        writer.writeheader()
+        for row in attempts:
+            writer.writerows([{k: self._sanitize_csv_value(v) for k, v in row.items()}])
+        return output.getvalue()
+
+    def export_replies_csv(self, campaign_id: str, user_id: str = "default") -> str:
+        storage = self._get_storage()
+        replies = storage.list_replies_all(campaign_id, user_id=user_id)
+        output = io.StringIO()
+        writer = csv.DictWriter(output, fieldnames=self.CSV_REPLY_FIELDS, extrasaction="ignore")
+        writer.writeheader()
+        for row in replies:
+            writer.writerows([{k: self._sanitize_csv_value(v) for k, v in row.items()}])
+        return output.getvalue()
+
     async def deep_discover(self, keyword: str, max_results: int = 15) -> Dict[str, Any]:
         """Enhanced discovery using Exa neural search + DuckDuckGo with full-page scraping."""
         from services.backlink_outreach_scraper import BacklinkOutreachScraper
@@ -212,9 +338,15 @@ class BacklinkOutreachService:
             "typed opportunity records and confidence score",
             "deep webpage scraping + contact-page extraction via Exa",
             "quality scoring and guest-post signal detection",
+            "DB-backed policy validation with suppression & idempotency",
+            "outreach attempt recording + status lifecycle",
+            "SMTP email sending via backlink_outreach_sender",
+            "IMAP reply polling with auto-classification",
+            "follow-up scheduling with sent tracking",
+            "email template CRUD + AI generation (llm_text_gen)",
+            "personalized send via template variables",
         ]
         planned = [
-            "email sending automation + response tracking",
             "follow-up orchestration and campaign analytics",
         ]
         return {

backend/services/backlink_outreach_storage.py

@@ -2,13 +2,18 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, date
 from uuid import uuid4
 from typing import List, Optional
-from sqlalchemy import text as sql_text
+from sqlalchemy import text as sql_text, func as sa_func
 
 from services.database import get_session_for_user
-from models.backlink_outreach_models import Base, BacklinkCampaign, BacklinkLead
+from models.backlink_outreach_models import (
+    Base, BacklinkCampaign, BacklinkLead,
+    OutreachAttempt, OutreachReply, FollowUpSchedule, EmailTemplate,
+    SuppressedRecipient, SentIdempotencyKey, AuditLogEntry,
+    SendCounterUser, SendCounterDomain,
+)
 
 
 class BacklinkOutreachStorageService:
@@ -29,11 +34,14 @@ class BacklinkOutreachStorageService:
     def _migrate_lead_columns(self, db) -> None:
         """Add new columns to backlink_leads if they don't exist (dev migration)."""
         try:
+            valid_columns = {"url", "page_title", "snippet", "confidence_score", "discovery_source", "notes"}
             for col in self._NEW_LEAD_COLUMNS:
+                if col not in valid_columns:
+                    continue
+                safe_col = col.replace('"', "").replace(";", "")
                 db.execute(sql_text(
-                    f"ALTER TABLE backlink_leads ADD COLUMN IF NOT EXISTS {col} TEXT"
+                    f"ALTER TABLE backlink_leads ADD COLUMN IF NOT EXISTS \"{safe_col}\" TEXT"
                 ))
-            # confidence_score is Float, add separately
             db.execute(sql_text(
                 "ALTER TABLE backlink_leads ADD COLUMN IF NOT EXISTS confidence_score FLOAT DEFAULT 0.0"
             ))
@@ -198,6 +206,7 @@ class BacklinkOutreachStorageService:
     def update_lead_status(
         self, lead_id: str, user_id: str, status: str, notes: Optional[str] = None
     ) -> Optional[dict]:
+        self._ensure_tables(user_id)
         db = get_session_for_user(user_id)
         if not db:
             return None
@@ -229,3 +238,696 @@ class BacklinkOutreachStorageService:
             "notes": lead.notes,
             "created_at": lead.created_at.isoformat() if lead.created_at else None,
         }
+
+    # -- Outreach Attempt CRUD --
+
+    def add_attempt(
+        self,
+        lead_id: str,
+        campaign_id: str,
+        idempotency_key: str,
+        sender_email: str = "",
+        subject: str = "",
+        body: str = "",
+        status: str = "queued",
+        decision_reason: Optional[str] = None,
+        user_id: str = "default",
+    ) -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            attempt = OutreachAttempt(
+                id=f"att_{uuid4().hex[:16]}",
+                lead_id=lead_id,
+                campaign_id=campaign_id,
+                idempotency_key=idempotency_key,
+                sender_email=sender_email,
+                subject=subject,
+                body=body,
+                status=status,
+                decision_reason=decision_reason,
+                created_at=datetime.utcnow(),
+            )
+            db.add(attempt)
+            db.commit()
+            return self._attempt_to_dict(attempt)
+        finally:
+            db.close()
+
+    def list_attempts(self, campaign_id: str, limit: int = 50, user_id: str = "default") -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(OutreachAttempt)
+                .filter(OutreachAttempt.campaign_id == campaign_id)
+                .order_by(OutreachAttempt.created_at.desc())
+                .limit(limit)
+                .all()
+            )
+            return [self._attempt_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    def update_attempt_status(self, attempt_id: str, status: str, decision_reason: Optional[str] = None, user_id: str = "default") -> Optional[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return None
+        try:
+            attempt = db.query(OutreachAttempt).filter(OutreachAttempt.id == attempt_id).first()
+            if not attempt:
+                return None
+            attempt.status = status
+            if decision_reason is not None:
+                attempt.decision_reason = decision_reason
+            if status == "sent":
+                attempt.sent_at = datetime.utcnow()
+            db.commit()
+            return self._attempt_to_dict(attempt)
+        finally:
+            db.close()
+
+    @staticmethod
+    def _attempt_to_dict(attempt) -> dict:
+        return {
+            "attempt_id": attempt.id,
+            "lead_id": attempt.lead_id,
+            "campaign_id": attempt.campaign_id,
+            "idempotency_key": attempt.idempotency_key,
+            "sender_email": attempt.sender_email or "",
+            "subject": attempt.subject or "",
+            "status": attempt.status,
+            "decision_reason": attempt.decision_reason,
+            "sent_at": attempt.sent_at.isoformat() if attempt.sent_at else None,
+            "created_at": attempt.created_at.isoformat() if attempt.created_at else None,
+        }
+
+    def find_attempt_by_from_email(self, from_email: str, user_id: str = "default") -> Optional[str]:
+        """Find the most recent attempt_id for a given sender email (lead)."""
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return None
+        try:
+            from sqlalchemy import desc
+            attempt = (
+                db.query(OutreachAttempt)
+                .join(BacklinkLead, OutreachAttempt.lead_id == BacklinkLead.id)
+                .filter(BacklinkLead.email == from_email)
+                .order_by(desc(OutreachAttempt.created_at))
+                .first()
+            )
+            return attempt.id if attempt else None
+        finally:
+            db.close()
+
+    # -- Outreach Reply CRUD --
+
+    def reply_exists(self, from_email: str, subject: str, user_id: str = "default") -> bool:
+        """Check if a reply with this from_email+subject already exists."""
+        db = get_session_for_user(user_id)
+        if not db:
+            return False
+        try:
+            exists = (
+                db.query(OutreachReply.id)
+                .filter(OutreachReply.from_email == from_email, OutreachReply.subject == subject)
+                .first()
+            )
+            return exists is not None
+        finally:
+            db.close()
+
+    def add_reply(
+        self,
+        attempt_id: str,
+        from_email: str = "",
+        subject: str = "",
+        body: str = "",
+        classification: str = "replied",
+        user_id: str = "default",
+    ) -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            reply = OutreachReply(
+                id=f"rep_{uuid4().hex[:16]}",
+                attempt_id=attempt_id,
+                from_email=from_email,
+                subject=subject,
+                body=body,
+                classification=classification,
+                received_at=datetime.utcnow(),
+            )
+            db.add(reply)
+            db.commit()
+            return self._reply_to_dict(reply)
+        finally:
+            db.close()
+
+    def list_replies(self, campaign_id: str, limit: int = 50, user_id: str = "default") -> List[dict]:
+        """List replies by joining through attempts to filter by campaign."""
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(OutreachReply)
+                .join(OutreachAttempt, OutreachReply.attempt_id == OutreachAttempt.id)
+                .filter(OutreachAttempt.campaign_id == campaign_id)
+                .order_by(OutreachReply.received_at.desc())
+                .limit(limit)
+                .all()
+            )
+            return [self._reply_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    @staticmethod
+    def _reply_to_dict(reply) -> dict:
+        return {
+            "reply_id": reply.id,
+            "attempt_id": reply.attempt_id,
+            "from_email": reply.from_email or "",
+            "subject": reply.subject or "",
+            "received_at": reply.received_at.isoformat() if reply.received_at else None,
+            "classification": reply.classification,
+            "body": reply.body or "",
+        }
+
+    # -- Follow-Up Schedule CRUD --
+
+    def schedule_followup(
+        self,
+        attempt_id: str,
+        scheduled_for: str,
+        subject: str = "",
+        body: str = "",
+        user_id: str = "default",
+    ) -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            sched = FollowUpSchedule(
+                id=f"fu_{uuid4().hex[:16]}",
+                attempt_id=attempt_id,
+                subject=subject or None,
+                body=body or None,
+                scheduled_for=datetime.fromisoformat(scheduled_for) if isinstance(scheduled_for, str) else scheduled_for,
+                sent=False,
+            )
+            db.add(sched)
+            db.commit()
+            return self._followup_to_dict(sched)
+        finally:
+            db.close()
+
+    def list_followups(self, campaign_id: str, limit: int = 50, user_id: str = "default") -> List[dict]:
+        """List follow-ups by joining through attempts to filter by campaign."""
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(FollowUpSchedule)
+                .join(OutreachAttempt, FollowUpSchedule.attempt_id == OutreachAttempt.id)
+                .filter(OutreachAttempt.campaign_id == campaign_id)
+                .order_by(FollowUpSchedule.scheduled_for.asc())
+                .limit(limit)
+                .all()
+            )
+            return [self._followup_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    def mark_followup_sent(self, schedule_id: str, user_id: str = "default") -> Optional[dict]:
+        db = get_session_for_user(user_id)
+        if not db:
+            return None
+        try:
+            sched = db.query(FollowUpSchedule).filter(FollowUpSchedule.id == schedule_id).first()
+            if not sched:
+                return None
+            sched.sent = True
+            db.commit()
+            return self._followup_to_dict(sched)
+        finally:
+            db.close()
+
+    @staticmethod
+    def _followup_to_dict(sched) -> dict:
+        return {
+            "schedule_id": sched.id,
+            "attempt_id": sched.attempt_id,
+            "subject": sched.subject or "",
+            "scheduled_for": sched.scheduled_for.isoformat() if sched.scheduled_for else None,
+            "sent": sched.sent,
+        }
+
+    # -- Email Template CRUD --
+
+    def create_template(
+        self,
+        user_id: str,
+        name: str,
+        subject_template: str,
+        body_template: str,
+        variables: Optional[List[str]] = None,
+    ) -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            tmpl = EmailTemplate(
+                id=f"tpl_{uuid4().hex[:16]}",
+                user_id=user_id,
+                name=name,
+                subject_template=subject_template,
+                body_template=body_template,
+                variables=",".join(variables) if variables else None,
+                created_at=datetime.utcnow(),
+            )
+            db.add(tmpl)
+            db.commit()
+            return self._template_to_dict(tmpl)
+        finally:
+            db.close()
+
+    def list_templates(self, user_id: str, limit: int = 50) -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(EmailTemplate)
+                .filter(EmailTemplate.user_id == user_id)
+                .order_by(EmailTemplate.created_at.desc())
+                .limit(limit)
+                .all()
+            )
+            return [self._template_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    def get_template(self, template_id: str, user_id: str) -> Optional[dict]:
+        db = get_session_for_user(user_id)
+        if not db:
+            return None
+        try:
+            tmpl = (
+                db.query(EmailTemplate)
+                .filter(EmailTemplate.id == template_id, EmailTemplate.user_id == user_id)
+                .first()
+            )
+            if not tmpl:
+                return None
+            return self._template_to_dict(tmpl)
+        finally:
+            db.close()
+
+    def delete_template(self, template_id: str, user_id: str) -> bool:
+        db = get_session_for_user(user_id)
+        if not db:
+            return False
+        try:
+            tmpl = (
+                db.query(EmailTemplate)
+                .filter(EmailTemplate.id == template_id, EmailTemplate.user_id == user_id)
+                .first()
+            )
+            if not tmpl:
+                return False
+            db.delete(tmpl)
+            db.commit()
+            return True
+        finally:
+            db.close()
+
+    @staticmethod
+    def _template_to_dict(tmpl) -> dict:
+        return {
+            "template_id": tmpl.id,
+            "user_id": tmpl.user_id,
+            "name": tmpl.name,
+            "subject_template": tmpl.subject_template,
+            "body_template": tmpl.body_template,
+            "variables": tmpl.variables.split(",") if tmpl.variables else [],
+            "created_at": tmpl.created_at.isoformat() if tmpl.created_at else None,
+        }
+
+    # -- Suppression List --
+
+    def add_suppressed(self, email: str, user_id: str = "default", domain: str = "", reason: str = "") -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            entry = SuppressedRecipient(
+                id=f"sup_{uuid4().hex[:16]}",
+                email=email.lower(),
+                domain=domain.lower() if domain else email.split("@")[-1].lower(),
+                reason=reason,
+                user_id=user_id,
+                created_at=datetime.utcnow(),
+            )
+            db.add(entry)
+            db.commit()
+            return {"id": entry.id, "email": entry.email, "reason": entry.reason}
+        finally:
+            db.close()
+
+    def is_suppressed(self, email: str, domain: str = "", user_id: str = "default") -> bool:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return False
+        try:
+            email_lower = email.lower()
+            domain_lower = domain.lower() if domain else email.split("@")[-1].lower()
+            exists = (
+                db.query(SuppressedRecipient.id)
+                .filter(
+                    (SuppressedRecipient.email == email_lower) |
+                    (SuppressedRecipient.domain == domain_lower)
+                )
+                .first()
+            )
+            return exists is not None
+        finally:
+            db.close()
+
+    def list_suppressed(self, user_id: str = "default", limit: int = 100) -> List[dict]:
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(SuppressedRecipient)
+                .order_by(SuppressedRecipient.created_at.desc())
+                .limit(limit)
+                .all()
+            )
+            return [{"id": r.id, "email": r.email, "domain": r.domain, "reason": r.reason, "created_at": r.created_at.isoformat() if r.created_at else None} for r in rows]
+        finally:
+            db.close()
+
+    # -- Idempotency --
+
+    def check_idempotency(self, idempotency_key: str, user_id: str = "default") -> bool:
+        """Returns True if key already exists (duplicate)."""
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return False
+        try:
+            exists = (
+                db.query(SentIdempotencyKey.id)
+                .filter(SentIdempotencyKey.idempotency_key == idempotency_key)
+                .first()
+            )
+            return exists is not None
+        finally:
+            db.close()
+
+    def mark_idempotency(self, idempotency_key: str, user_id: str = "default") -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            entry = SentIdempotencyKey(
+                id=f"idm_{uuid4().hex[:16]}",
+                idempotency_key=idempotency_key,
+                user_id=user_id,
+                created_at=datetime.utcnow(),
+            )
+            db.add(entry)
+            db.commit()
+            return {"idempotency_key": idempotency_key}
+        finally:
+            db.close()
+
+    # -- Send Counters --
+
+    def _today(self) -> date:
+        return date.today()
+
+    def increment_user_send_counter(self, user_id: str) -> int:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return 0
+        try:
+            today = self._today()
+            row_id = f"scu_{uuid4().hex[:16]}"
+            db.execute(sql_text(
+                "INSERT INTO backlink_send_counters_user (id, user_id, date, count) "
+                "VALUES (:id, :uid, :dt, 1) "
+                "ON CONFLICT (user_id, date) DO UPDATE SET count = count + 1"
+            ), {"id": row_id, "uid": user_id, "dt": today})
+            db.commit()
+            result = db.query(SendCounterUser.count).filter(
+                SendCounterUser.user_id == user_id, SendCounterUser.date == today
+            ).first()
+            return result[0] if result else 0
+        finally:
+            db.close()
+
+    def get_user_send_count(self, user_id: str) -> int:
+        db = get_session_for_user(user_id)
+        if not db:
+            return 0
+        try:
+            today = self._today()
+            row = (
+                db.query(SendCounterUser.count)
+                .filter(SendCounterUser.user_id == user_id, SendCounterUser.date == today)
+                .first()
+            )
+            return row[0] if row else 0
+        finally:
+            db.close()
+
+    def increment_domain_send_counter(self, domain: str, user_id: str = "default") -> int:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return 0
+        try:
+            today = self._today()
+            domain_lower = domain.lower()
+            row_id = f"scd_{uuid4().hex[:16]}"
+            db.execute(sql_text(
+                "INSERT INTO backlink_send_counters_domain (id, domain, date, count) "
+                "VALUES (:id, :dom, :dt, 1) "
+                "ON CONFLICT (domain, date) DO UPDATE SET count = count + 1"
+            ), {"id": row_id, "dom": domain_lower, "dt": today})
+            db.commit()
+            result = db.query(SendCounterDomain.count).filter(
+                SendCounterDomain.domain == domain_lower, SendCounterDomain.date == today
+            ).first()
+            return result[0] if result else 0
+        finally:
+            db.close()
+
+    def get_domain_send_count(self, domain: str, user_id: str = "default") -> int:
+        db = get_session_for_user(user_id)
+        if not db:
+            return 0
+        try:
+            today = self._today()
+            row = (
+                db.query(SendCounterDomain.count)
+                .filter(SendCounterDomain.domain == domain.lower(), SendCounterDomain.date == today)
+                .first()
+            )
+            return row[0] if row else 0
+        finally:
+            db.close()
+
+    # -- Audit Log --
+
+    def add_audit_log(
+        self,
+        event: str,
+        user_id: str,
+        campaign_id: str = "",
+        recipient: str = "",
+        allowed: bool = False,
+        reasons: Optional[List[str]] = None,
+        override: bool = False,
+    ) -> dict:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            raise RuntimeError("Database session unavailable")
+        try:
+            entry = AuditLogEntry(
+                id=f"aud_{uuid4().hex[:16]}",
+                user_id=user_id,
+                campaign_id=campaign_id or None,
+                event=event,
+                recipient=recipient or None,
+                allowed=allowed,
+                reasons=";".join(reasons) if reasons else None,
+                override=override,
+                created_at=datetime.utcnow(),
+            )
+            db.add(entry)
+            db.commit()
+            return {"id": entry.id, "event": entry.event, "allowed": entry.allowed}
+        finally:
+            db.close()
+
+    def list_audit_logs(self, campaign_id: Optional[str] = None, limit: int = 100, user_id: str = "default") -> List[dict]:
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            q = db.query(AuditLogEntry)
+            if campaign_id:
+                q = q.filter(AuditLogEntry.campaign_id == campaign_id)
+            rows = q.order_by(AuditLogEntry.created_at.desc()).limit(limit).all()
+            return [
+                {
+                    "id": r.id,
+                    "event": r.event,
+                    "recipient": r.recipient,
+                    "allowed": r.allowed,
+                    "reasons": r.reasons.split(";") if r.reasons else [],
+                    "override": r.override,
+                    "created_at": r.created_at.isoformat() if r.created_at else None,
+                }
+                for r in rows
+            ]
+        finally:
+            db.close()
+
+    # -- Analytics --
+
+    def get_send_volume_by_day(self, campaign_id: str, days: int = 30, user_id: str = "default") -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            from datetime import timedelta
+            cutoff = datetime.utcnow() - timedelta(days=days)
+            rows = (
+                db.query(sa_func.date(OutreachAttempt.sent_at).label("date"), sa_func.count(OutreachAttempt.id).label("count"))
+                .filter(OutreachAttempt.campaign_id == campaign_id, OutreachAttempt.status == "sent", OutreachAttempt.sent_at >= cutoff)
+                .group_by(sa_func.date(OutreachAttempt.sent_at))
+                .order_by(sa_func.date(OutreachAttempt.sent_at).asc())
+                .all()
+            )
+            return [{"date": str(r.date), "count": r.count} for r in rows]
+        finally:
+            db.close()
+
+    def get_lead_status_counts(self, campaign_id: str, user_id: str = "default") -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(BacklinkLead.status, sa_func.count(BacklinkLead.id).label("count"))
+                .filter(BacklinkLead.campaign_id == campaign_id)
+                .group_by(BacklinkLead.status)
+                .order_by(BacklinkLead.status.asc())
+                .all()
+            )
+            return [{"status": r.status, "count": r.count} for r in rows]
+        finally:
+            db.close()
+
+    def list_attempts_all(self, campaign_id: str, user_id: str = "default") -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(OutreachAttempt)
+                .filter(OutreachAttempt.campaign_id == campaign_id)
+                .order_by(OutreachAttempt.created_at.desc())
+                .all()
+            )
+            return [self._attempt_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    def list_replies_all(self, campaign_id: str, user_id: str = "default") -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(OutreachReply)
+                .join(OutreachAttempt, OutreachReply.attempt_id == OutreachAttempt.id)
+                .filter(OutreachAttempt.campaign_id == campaign_id)
+                .order_by(OutreachReply.received_at.desc())
+                .all()
+            )
+            return [self._reply_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    def count_replies(self, campaign_id: str, user_id: str = "default") -> int:
+        db = get_session_for_user(user_id)
+        if not db:
+            return 0
+        try:
+            return (
+                db.query(OutreachReply.id)
+                .join(OutreachAttempt, OutreachReply.attempt_id == OutreachAttempt.id)
+                .filter(OutreachAttempt.campaign_id == campaign_id)
+                .count()
+            )
+        finally:
+            db.close()
+
+    def list_leads_all(self, campaign_id: str, user_id: str = "default") -> List[dict]:
+        self._ensure_tables(user_id)
+        db = get_session_for_user(user_id)
+        if not db:
+            return []
+        try:
+            rows = (
+                db.query(BacklinkLead)
+                .filter(BacklinkLead.campaign_id == campaign_id)
+                .order_by(BacklinkLead.created_at.desc())
+                .all()
+            )
+            return [self._lead_to_dict(r) for r in rows]
+        finally:
+            db.close()
+
+    # -- Policy Helpers (composite checks) --
+
+    def get_lead(self, lead_id: str, user_id: str = "default") -> Optional[dict]:
+        db = get_session_for_user(user_id)
+        if not db:
+            return None
+        try:
+            lead = db.query(BacklinkLead).filter(BacklinkLead.id == lead_id).first()
+            if not lead:
+                return None
+            return self._lead_to_dict(lead)
+        finally:
+            db.close()

backend/services/backlink_outreach_template_generator.py

@@ -0,0 +1,307 @@
+"""AI-powered outreach email template generation."""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import List, Optional
+from loguru import logger
+
+from services.llm_providers.main_text_generation import llm_text_gen
+
+
+SYSTEM_PROMPT = """You are an expert outreach copywriter specializing in guest post and backlink pitch emails.
+Write concise, personalized outreach emails that get high response rates.
+Follow these rules:
+- Be specific about why you're reaching out (mention their content)
+- Keep it under 200 words
+- Include a clear call to action
+- Sound human, not templated
+- Never use spammy phrases
+- Output ONLY valid JSON with "subject" and "body" keys"""
+
+SUBJECT_LINES_PROMPT = """You are an expert email subject line writer.
+Given an outreach email body, generate subject lines that are:
+- Intriguing but not clickbait
+- Personalized when possible
+- Under 60 characters
+- Varied in style (question, curiosity, value-prop)
+Output ONLY valid JSON with a "subjects" key containing an array of strings."""
+
+FOLLOW_UP_PROMPT = """You are an expert outreach copywriter.
+Write a polite follow-up email for a guest post pitch that hasn't received a response.
+Rules:
+- Reference the original email without repeating it verbatim
+- Keep it shorter than the original (under 100 words)
+- Add a new angle or piece of value
+- Include a clear call to action
+- Sound human and respectful, never pushy
+- Output ONLY valid JSON with "subject" and "body" keys"""
+
+PERSONALIZATION_PROMPT = """You are an expert outreach personalization specialist.
+Given a lead's information and a draft outreach email, personalize it for that specific lead.
+Rules:
+- Mention their specific content or website
+- Reference something relevant from their site
+- Keep the core pitch but make it feel custom-written
+- Under 200 words
+- Output ONLY valid JSON with "subject" and "body" keys"""
+
+
+def generate_outreach_email(
+    topic: str,
+    target_site: Optional[str] = None,
+    tone: str = "professional",
+    user_id: str = "default",
+    existing_body: Optional[str] = None,
+) -> dict:
+    """Generate an outreach email using the LLM.
+
+    Args:
+        topic: The topic/keyword to pitch.
+        target_site: Optional target website name/URL.
+        tone: professional, friendly, casual, or formal.
+        user_id: Clerk user ID for subscription check.
+        existing_body: If provided, rewrite/improve this existing template.
+
+    Returns:
+        dict with "subject" and "body" keys.
+    """
+    if existing_body:
+        prompt = (
+            f"Rewrite and improve the following outreach email for a {tone} tone. "
+            f"Topic: {topic}. "
+            f"{f'Target website: {target_site}. ' if target_site else ''}"
+            f"Keep the core message but make it more effective. "
+            f"Original email:\n\n{existing_body}\n\n"
+            f"Return ONLY valid JSON with 'subject' and 'body' keys."
+        )
+    else:
+        prompt = (
+            f"Write a {tone} outreach email for a guest post opportunity about: {topic}. "
+            f"{f'We are pitching this to: {target_site}. ' if target_site else ''}"
+            f"Mention specific value the guest post would bring to their audience. "
+            f"Return ONLY valid JSON with 'subject' and 'body' keys."
+        )
+
+    try:
+        raw = llm_text_gen(
+            prompt=prompt,
+            system_prompt=SYSTEM_PROMPT,
+            user_id=user_id,
+            temperature=0.7,
+        )
+
+        result = _parse_json_response(raw)
+        if result:
+            return result
+
+        return _fallback_extract(raw, topic)
+
+    except Exception as e:
+        logger.error(f"Failed to generate outreach email: {e}")
+        return {
+            "subject": f"Guest post opportunity: {topic}",
+            "body": f"Hi there,\n\nI came across your site and I'd love to contribute a guest post about {topic}. "
+                     f"Please let me know if you're open to submissions.\n\nBest regards",
+        }
+
+
+def generate_personalized_email(
+    lead_name: str,
+    lead_site: str,
+    lead_content_topic: str,
+    pitch_topic: str,
+    existing_body: str = "",
+    user_id: str = "default",
+) -> dict:
+    """Personalize an outreach email for a specific lead.
+
+    Args:
+        lead_name: Contact name or site owner name.
+        lead_site: The lead's website URL.
+        lead_content_topic: Topic of relevant content on their site.
+        pitch_topic: The topic we want to pitch.
+        existing_body: Optional draft to personalize further.
+        user_id: Clerk user ID for subscription check.
+
+    Returns:
+        dict with "subject" and "body" keys.
+    """
+    if existing_body:
+        prompt = (
+            f"Personalize this outreach email for {lead_name} from {lead_site}. "
+            f"They have content about '{lead_content_topic}'. "
+            f"We want to pitch: {pitch_topic}. "
+            f"Mention something specific about their content on {lead_content_topic} "
+            f"to show we've done our research. "
+            f"Draft email to personalize:\n\n{existing_body}\n\n"
+            f"Return ONLY valid JSON with 'subject' and 'body' keys."
+        )
+    else:
+        prompt = (
+            f"Write a personalized outreach email to {lead_name} at {lead_site}. "
+            f"They have published content about '{lead_content_topic}'. "
+            f"We want to pitch a guest post about: {pitch_topic}. "
+            f"Reference their article on {lead_content_topic} and explain how our pitch "
+            f"would provide value to their audience. "
+            f"Return ONLY valid JSON with 'subject' and 'body' keys."
+        )
+
+    try:
+        raw = llm_text_gen(
+            prompt=prompt,
+            system_prompt=PERSONALIZATION_PROMPT,
+            user_id=user_id,
+            temperature=0.7,
+        )
+        result = _parse_json_response(raw)
+        if result:
+            return result
+        return _fallback_extract(raw, pitch_topic)
+    except Exception as e:
+        logger.error(f"Failed to personalize email: {e}")
+        return {"subject": f"Question about your content on {lead_content_topic}", "body": existing_body or f"Hi {lead_name},\n\nI enjoyed your article about {lead_content_topic}..."}
+
+
+def generate_subject_lines(
+    body: str,
+    count: int = 5,
+    user_id: str = "default",
+) -> List[str]:
+    """Generate subject line suggestions for an email body.
+
+    Args:
+        body: The email body to generate subject lines for.
+        count: Number of subject lines to generate.
+        user_id: Clerk user ID for subscription check.
+
+    Returns:
+        List of subject line strings.
+    """
+    prompt = (
+        f"Generate {count} subject lines for the following outreach email. "
+        f"Make them varied in style and optimized for open rates.\n\n"
+        f"Email body:\n{body}\n\n"
+        f"Return ONLY valid JSON with a 'subjects' key containing an array of strings."
+    )
+
+    try:
+        raw = llm_text_gen(
+            prompt=prompt,
+            system_prompt=SUBJECT_LINES_PROMPT,
+            user_id=user_id,
+            temperature=0.8,
+        )
+        if raw:
+            text = raw.strip()
+            if text.startswith("```"):
+                text = re.sub(r"^```(?:json)?\s*", "", text)
+                text = re.sub(r"\s*```$", "", text)
+            try:
+                data = json.loads(text)
+                if isinstance(data, dict) and "subjects" in data and isinstance(data["subjects"], list):
+                    return [s.strip() for s in data["subjects"][:count]]
+            except json.JSONDecodeError:
+                pass
+        lines = [l.strip("- ").strip() for l in raw.strip().split("\n") if l.strip() and not l.strip().startswith("```")]
+        return [l for l in lines if len(l) > 10][:count]
+    except Exception as e:
+        logger.error(f"Failed to generate subject lines: {e}")
+        return [f"Guest post opportunity", f"Question about your content", f"Collaboration idea"]
+
+
+def generate_follow_up(
+    original_subject: str,
+    original_body: str,
+    days_elapsed: int = 7,
+    reply_context: str = "",
+    user_id: str = "default",
+) -> dict:
+    """Generate a follow-up email for an outreach that hasn't received a response.
+
+    Args:
+        original_subject: Subject line of the original email.
+        original_body: Body of the original email.
+        days_elapsed: Number of days since the original was sent.
+        reply_context: If the recipient replied, context of their reply.
+        user_id: Clerk user ID for subscription check.
+
+    Returns:
+        dict with "subject" and "body" keys.
+    """
+    if reply_context:
+        prompt = (
+            f"The recipient replied with: '{reply_context}'. "
+            f"Write a follow-up email that addresses their response and keeps the conversation moving. "
+            f"Original subject: {original_subject}.\n\n"
+            f"Original email:\n{original_body}\n\n"
+            f"Return ONLY valid JSON with 'subject' and 'body' keys."
+        )
+    else:
+        prompt = (
+            f"Write a polite follow-up email. {days_elapsed} days have passed since the original email. "
+            f"Do not apologize for following up. Add a new piece of value or angle. "
+            f"Original subject: {original_subject}.\n\n"
+            f"Original email:\n{original_body}\n\n"
+            f"Return ONLY valid JSON with 'subject' and 'body' keys."
+        )
+
+    try:
+        raw = llm_text_gen(
+            prompt=prompt,
+            system_prompt=FOLLOW_UP_PROMPT,
+            user_id=user_id,
+            temperature=0.7,
+        )
+        result = _parse_json_response(raw)
+        if result:
+            return result
+        return _fallback_extract(raw, original_subject)
+    except Exception as e:
+        logger.error(f"Failed to generate follow-up: {e}")
+        return {
+            "subject": f"Re: {original_subject}",
+            "body": f"Hi there,\n\nI wanted to follow up on my previous email. "
+                    f"I'd love to hear your thoughts when you have a moment.\n\nBest regards",
+        }
+
+
+def _parse_json_response(raw: str) -> Optional[dict]:
+    """Try to parse JSON from LLM response, handling markdown fences."""
+    if not raw:
+        return None
+
+    text = raw.strip()
+
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:json)?\s*", "", text)
+        text = re.sub(r"\s*```$", "", text)
+
+    try:
+        data = json.loads(text)
+        if isinstance(data, dict) and "subject" in data and "body" in data:
+            return {"subject": data["subject"].strip(), "body": data["body"].strip()}
+    except json.JSONDecodeError:
+        pass
+
+    return None
+
+
+def _fallback_extract(raw: str, topic: str) -> dict:
+    """Fallback: try to extract subject line and body from unstructured text."""
+    lines = [l.strip() for l in raw.strip().split("\n") if l.strip()]
+    subject = topic
+    body_lines = []
+
+    for i, line in enumerate(lines):
+        lower = line.lower()
+        if lower.startswith("subject") or lower.startswith("subject:"):
+            subject = line.split(":", 1)[-1].strip()
+        elif lower.startswith("body") or lower.startswith("body:"):
+            body_lines.append(line.split(":", 1)[-1].strip())
+        else:
+            body_lines.append(line)
+
+    body = "\n".join(body_lines) if body_lines else raw
+    return {"subject": subject, "body": body}

backend/services/integrations/oauth_callback_utils.py

@@ -0,0 +1,79 @@
+"""
+Shared OAuth callback utilities for Wix and WordPress integrations.
+
+Provides hardened postMessage-based HTML callback generation, origin
+validation, and string sanitization used across OAuth callback routes.
+"""
+
+import json
+import os
+from typing import Any, Optional
+from urllib.parse import urlparse
+
+
+def sanitize_string(value: Any, max_len: int = 500) -> str:
+    if value is None:
+        return ""
+    return " ".join(str(value).split())[:max_len]
+
+
+def sanitize_error(error: Exception, max_len: int = 500) -> str:
+    return sanitize_string(error, max_len)
+
+
+def normalize_origin(url: Optional[str]) -> Optional[str]:
+    if not url:
+        return None
+    parsed = urlparse(url.strip())
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        return None
+    return f"{parsed.scheme}://{parsed.netloc}"
+
+
+def trusted_frontend_origin() -> Optional[str]:
+    origins_env = os.getenv("OAUTH_CALLBACK_ALLOWED_ORIGINS", "")
+    configured = [
+        origin
+        for origin in (normalize_origin(o) for o in origins_env.split(",") if o.strip())
+        if origin is not None
+    ]
+    if configured:
+        return configured[0]
+    return normalize_origin(os.getenv("FRONTEND_URL"))
+
+
+def build_oauth_callback_html(
+    payload: dict,
+    title: str,
+    heading: str,
+    message: str,
+) -> str:
+    trusted_origin = trusted_frontend_origin()
+    payload_json = json.dumps(payload)
+    target_origin_json = json.dumps(trusted_origin or "")
+    heading_html = heading.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+    message_html = message.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+    return f"""
+    <!DOCTYPE html>
+    <html>
+    <head><title>{title}</title></head>
+    <body>
+      <h1>{heading_html}</h1>
+      <p>{message_html}</p>
+      <script>
+        (function() {{
+          var payload = {payload_json};
+          var targetOrigin = {target_origin_json};
+          var destination = window.opener || window.parent;
+          if (destination && targetOrigin) {{
+            try {{
+              destination.postMessage(payload, targetOrigin);
+              window.close();
+              return;
+            }} catch (_e) {{}}
+          }}
+        }})();
+      </script>
+    </body>
+    </html>
+    """

backend/services/integrations/wix_oauth.py

@@ -8,7 +8,7 @@ import sqlite3
 from typing import Optional, Dict, Any, List
 from datetime import datetime, timedelta
 from loguru import logger
-
+from cryptography.fernet import Fernet, InvalidToken
 
 from services.database import get_user_db_path
 
@@ -17,6 +17,66 @@ class WixOAuthService:
     
     def __init__(self, db_path: Optional[str] = None):
         self.db_path = db_path
+        self.token_encryption_key = (
+            os.getenv("WIX_TOKEN_ENCRYPTION_KEY")
+            or os.getenv("OAUTH_TOKEN_ENCRYPTION_KEY")
+        )
+        self._fernet = self._initialize_fernet()
+        self._migration_done: set = set()
+
+    def _initialize_fernet(self) -> Optional[Fernet]:
+        if not self.token_encryption_key:
+            logger.error("Wix token encryption key is not configured.")
+            return None
+        try:
+            return Fernet(self.token_encryption_key.encode("utf-8"))
+        except Exception:
+            logger.error("Wix token encryption key is invalid.")
+            return None
+
+    def _encrypt_token(self, token: Optional[str]) -> Optional[str]:
+        if not token:
+            return None
+        if not self._fernet:
+            raise ValueError("Token encryption is unavailable: missing/invalid managed key")
+        return self._fernet.encrypt(token.encode("utf-8")).decode("utf-8")
+
+    def _decrypt_token(self, token_blob: Optional[str]) -> Optional[str]:
+        if not token_blob:
+            return None
+        if not self._fernet:
+            raise ValueError("Token decryption is unavailable: missing/invalid managed key")
+        return self._fernet.decrypt(token_blob.encode("utf-8")).decode("utf-8")
+
+    def _is_likely_encrypted_blob(self, value: Optional[str]) -> bool:
+        return bool(value and value.startswith("gAAAAA"))
+
+    def _migrate_plaintext_tokens_if_needed(self, conn: sqlite3.Connection, user_id: str) -> None:
+        if not self._fernet or user_id in self._migration_done:
+            return
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT id, access_token, refresh_token FROM wix_oauth_tokens WHERE user_id = ?",
+            (user_id,),
+        )
+        rows = cursor.fetchall()
+        migrated = 0
+        for token_id, access_token, refresh_token in rows:
+            needs_access = access_token and not self._is_likely_encrypted_blob(access_token)
+            needs_refresh = refresh_token and not self._is_likely_encrypted_blob(refresh_token)
+            if not (needs_access or needs_refresh):
+                continue
+            enc_access = self._encrypt_token(access_token) if needs_access else access_token
+            enc_refresh = self._encrypt_token(refresh_token) if needs_refresh else refresh_token
+            cursor.execute(
+                "UPDATE wix_oauth_tokens SET access_token = ?, refresh_token = ?, updated_at = datetime('now') WHERE id = ? AND user_id = ?",
+                (enc_access, enc_refresh, token_id, user_id),
+            )
+            migrated += 1
+        if migrated:
+            conn.commit()
+            logger.info(f"Wix OAuth token migration completed for user {user_id}; rows migrated={migrated}")
+        self._migration_done.add(user_id)
     
     def _get_db_path(self, user_id: str) -> str:
         if self.db_path:
@@ -173,13 +233,16 @@ class WixOAuthService:
             if expires_in:
                 expires_at = datetime.now() + timedelta(seconds=expires_in)
             
+            encrypted_access = self._encrypt_token(access_token)
+            encrypted_refresh = self._encrypt_token(refresh_token) if refresh_token else None
+
             with sqlite3.connect(db_path) as conn:
                 cursor = conn.cursor()
                 cursor.execute('''
                     INSERT INTO wix_oauth_tokens 
                     (user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id)
                     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-                ''', (user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id))
+                ''', (user_id, encrypted_access, encrypted_refresh, token_type, expires_at, expires_in, scope, site_id, member_id))
                 conn.commit()
                 logger.info(f"Wix OAuth: Token inserted into database for user {user_id}")
             
@@ -200,6 +263,7 @@ class WixOAuthService:
                 return []
                 
             with sqlite3.connect(db_path) as conn:
+                self._migrate_plaintext_tokens_if_needed(conn, user_id)
                 cursor = conn.cursor()
                 cursor.execute('''
                     SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at
@@ -210,10 +274,29 @@ class WixOAuthService:
                 
                 tokens = []
                 for row in cursor.fetchall():
+                    access_token_val = row[1]
+                    refresh_token_val = row[2]
+                    try:
+                        decrypted_access = (
+                            self._decrypt_token(access_token_val)
+                            if self._is_likely_encrypted_blob(access_token_val)
+                            else access_token_val
+                        )
+                    except InvalidToken:
+                        logger.error(f"Failed to decrypt Wix access token for user {user_id}, token_id={row[0]}")
+                        continue
+                    try:
+                        decrypted_refresh = (
+                            self._decrypt_token(refresh_token_val)
+                            if self._is_likely_encrypted_blob(refresh_token_val)
+                            else refresh_token_val
+                        )
+                    except InvalidToken:
+                        decrypted_refresh = None
                     tokens.append({
                         "id": row[0],
-                        "access_token": row[1],
-                        "refresh_token": row[2],
+                        "access_token": decrypted_access,
+                        "refresh_token": decrypted_refresh,
                         "token_type": row[3],
                         "expires_at": row[4],
                         "expires_in": row[5],
@@ -248,9 +331,9 @@ class WixOAuthService:
                 }
 
             with sqlite3.connect(db_path) as conn:
+                self._migrate_plaintext_tokens_if_needed(conn, user_id)
                 cursor = conn.cursor()
                 
-                # Get all tokens (active and expired)
                 cursor.execute('''
                     SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at, is_active
                     FROM wix_oauth_tokens
@@ -263,10 +346,29 @@ class WixOAuthService:
                 expired_tokens = []
                 
                 for row in cursor.fetchall():
+                    access_token_val = row[1]
+                    refresh_token_val = row[2]
+                    try:
+                        decrypted_access = (
+                            self._decrypt_token(access_token_val)
+                            if self._is_likely_encrypted_blob(access_token_val)
+                            else access_token_val
+                        )
+                    except InvalidToken:
+                        decrypted_access = None
+                    try:
+                        decrypted_refresh = (
+                            self._decrypt_token(refresh_token_val)
+                            if self._is_likely_encrypted_blob(refresh_token_val)
+                            else refresh_token_val
+                        )
+                    except InvalidToken:
+                        decrypted_refresh = None
+
                     token_data = {
                         "id": row[0],
-                        "access_token": row[1],
-                        "refresh_token": row[2],
+                        "access_token": decrypted_access,
+                        "refresh_token": decrypted_refresh,
                         "token_type": row[3],
                         "expires_at": row[4],
                         "expires_in": row[5],
@@ -331,34 +433,46 @@ class WixOAuthService:
         user_id: str,
         access_token: str,
         refresh_token: Optional[str] = None,
-        expires_in: Optional[int] = None
+        expires_in: Optional[int] = None,
+        token_id: Optional[int] = None
     ) -> bool:
         """Update tokens for a user (e.g., after refresh)."""
         try:
-            # Ensure DB initialized for this user
             self._init_db(user_id)
             db_path = self._get_db_path(user_id)
 
             expires_at = None
             if expires_in:
                 expires_at = datetime.now() + timedelta(seconds=expires_in)
+
+            encrypted_access = self._encrypt_token(access_token)
+            encrypted_refresh = self._encrypt_token(refresh_token) if refresh_token else None
             
             with sqlite3.connect(db_path) as conn:
+                self._migrate_plaintext_tokens_if_needed(conn, user_id)
                 cursor = conn.cursor()
-                if refresh_token:
-                    cursor.execute('''
-                        UPDATE wix_oauth_tokens 
-                        SET access_token = ?, refresh_token = ?, expires_at = ?, expires_in = ?, 
-                            is_active = TRUE, updated_at = datetime('now')
-                        WHERE user_id = ? AND refresh_token = ?
-                    ''', (access_token, refresh_token, expires_at, expires_in, user_id, refresh_token))
+                if token_id:
+                    if encrypted_refresh:
+                        cursor.execute('''
+                            UPDATE wix_oauth_tokens 
+                            SET access_token = ?, refresh_token = ?, expires_at = ?, expires_in = ?, 
+                                is_active = TRUE, updated_at = datetime('now')
+                            WHERE user_id = ? AND id = ?
+                        ''', (encrypted_access, encrypted_refresh, expires_at, expires_in, user_id, token_id))
+                    else:
+                        cursor.execute('''
+                            UPDATE wix_oauth_tokens 
+                            SET access_token = ?, expires_at = ?, expires_in = ?, 
+                                is_active = TRUE, updated_at = datetime('now')
+                            WHERE user_id = ? AND id = ?
+                        ''', (encrypted_access, expires_at, expires_in, user_id, token_id))
                 else:
                     cursor.execute('''
                         UPDATE wix_oauth_tokens 
                         SET access_token = ?, expires_at = ?, expires_in = ?, 
                             is_active = TRUE, updated_at = datetime('now')
                         WHERE user_id = ? AND id = (SELECT id FROM wix_oauth_tokens WHERE user_id = ? ORDER BY created_at DESC LIMIT 1)
-                    ''', (access_token, expires_at, expires_in, user_id, user_id))
+                    ''', (encrypted_access, expires_at, expires_in, user_id, user_id))
                 conn.commit()
                 logger.info(f"Wix OAuth: Tokens updated for user {user_id}")
             

backend/services/research/trends/google_trends_service.py

@@ -343,7 +343,7 @@ class GoogleTrendsService:
         logger.info(
             f"[Trends] ===== DONE analyze_trends ===== total={total_ms}ms "
             f"iot={len(interest_over_time)} ibr={len(interest_by_region)} "
-            f"rt_top={rt_top} rq_top={rq_top}"
+            f"rt_top={len(related_topics.get('top', []))} rq_top={len(related_queries.get('top', []))}"
         )
 
         result = {

backend/services/subscription/preflight_validator.py

@@ -548,9 +548,11 @@ def validate_video_generation_operations(
 def validate_scene_animation_operation(
     pricing_service: PricingService,
     user_id: str,
+    scene_count: int = 1,
 ) -> None:
     """
     Validate the per-scene animation workflow before API calls.
+    Validates that the user has sufficient credits for *all* scenes in the batch.
     """
     try:
         operations_to_validate = [
@@ -560,6 +562,7 @@ def validate_scene_animation_operation(
                 'actual_provider_name': 'wavespeed',
                 'operation_type': 'scene_animation',
             }
+            for _ in range(scene_count)
         ]
 
         can_proceed, message, error_details = pricing_service.check_comprehensive_limits(
@@ -581,9 +584,8 @@ def validate_scene_animation_operation(
                 }
             )
 
-        logger.info(f"[Pre-flight Validator] ✅ Scene animation validated for user {user_id}")
-        # Validation passed - no return needed (function raises HTTPException if validation fails)
-        
+        logger.info(f"[Pre-flight Validator] ✅ Scene animation validated for user {user_id} ({scene_count} scene(s))")
+
     except HTTPException:
         raise
     except Exception as e:
@@ -730,9 +732,11 @@ def validate_video_generation_operations(
 def validate_scene_animation_operation(
     pricing_service: PricingService,
     user_id: str,
+    scene_count: int = 1,
 ) -> None:
     """
     Validate the per-scene animation workflow before API calls.
+    Validates that the user has sufficient credits for *all* scenes in the batch.
     """
     try:
         operations_to_validate = [
@@ -742,6 +746,7 @@ def validate_scene_animation_operation(
                 'actual_provider_name': 'wavespeed',
                 'operation_type': 'scene_animation',
             }
+            for _ in range(scene_count)
         ]
 
         can_proceed, message, error_details = pricing_service.check_comprehensive_limits(
@@ -763,7 +768,7 @@ def validate_scene_animation_operation(
                 }
             )
 
-        logger.info(f"[Pre-flight Validator] ✅ Scene animation validated for user {user_id}")
+        logger.info(f"[Pre-flight Validator] ✅ Scene animation validated for user {user_id} ({scene_count} scene(s))")
 
     except HTTPException:
         raise

backend/services/subscription/pricing_service.py

@@ -566,10 +566,10 @@ class PricingService:
                 "firecrawl_calls_limit": 0,  # DISABLED: Firecrawl not in Free tier
                 "stability_calls_limit": 3,  # 3 images - enough to try the product
                 "exa_calls_limit": 10,  # 10 research queries - enough to try the product
-                "video_calls_limit": 0,  # DISABLED: Video generation not in Free tier
+                "video_calls_limit": 2,  # 2 video renders - try podcast video on Free
                 "image_edit_calls_limit": 5,  # 5 image edits - enough to try the product
                 "audio_calls_limit": 5,  # 5 audio clips - enough to try the product
-                "wavespeed_calls_limit": 0,  # DISABLED: WaveSpeed not included in Free tier
+                "wavespeed_calls_limit": 0,  # 0 = unlimited for Free; video controlled via video_calls_limit
                 "gemini_tokens_limit": 50000,
                 "openai_tokens_limit": 0,  # DISABLED
                 "anthropic_tokens_limit": 0,  # DISABLED