Marketer: for Digital Nomads for Professionals

"""
subscriber_store.py — SQLAlchemy models and bulk-import logic.
Requires: sqlalchemy>=2.0, tenacity>=8.2, psycopg2-binary
"""
import logging
from datetime import datetime, timezone
from typing import List

from sqlalchemy import (
    Column,
    DateTime,
    Float,
    Integer,
    String,
    create_engine,
    text,
)
from sqlalchemy.exc import OperationalError, DBAPIError
from sqlalchemy.orm import declarative_base, sessionmaker
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

logger = logging.getLogger(__name__)
Base = declarative_base()


class Subscriber(Base):
    """Represents a single marketing subscriber."""
    __tablename__ = "subscribers"

    id = Column(Integer, primary_key=True, autoincrement=True)
    email = Column(String(320), unique=True, nullable=False, index=True)
    first_name = Column(String(120), nullable=True)
    last_name = Column(String(120), nullable=True)
    timezone = Column(String(64), default="UTC")
    created_at = Column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )
    # Marketing analytics counters
    emails_sent = Column(Integer, default=0)
    emails_opened = Column(Integer, default=0)
    emails_clicked = Column(Integer, default=0)
    last_opened_at = Column(DateTime(timezone=True), nullable=True)

    def __repr__(self) -> str:
        return f""


# ---------- Retry policy ----------
# Exponential back-off: 1 s, 2 s, 4 s … capped at 30 s;
# stop after 5 attempts so import never hangs forever.
@retry(
    retry=retry_if_exception_type((OperationalError, DBAPIError)),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    stop=stop_after_attempt(5),
    reraise=True,
)
def _get_session(engine):
    """Return a SQLAlchemy session with automatic retry on transient errors."""
    Session = sessionmaker(bind=engine)
    return Session()


def bulk_import_subscribers(
    dsn: str,
    subscriber_dicts: List[dict],
    batch_size: int = 500,
) -> dict:
    """
    Insert or update subscribers in batches.

    Parameters
    ----------
    dsn: SQLAlchemy-compatible data-source name.
    subscriber_dicts: List of dicts with keys email, first_name,
                      last_name, timezone.
    batch_size: Number of rows per INSERT … ON CONFLICT batch.

    Returns
    -------
    dict with keys 'inserted', 'updated', 'errors'.
    """
    engine = create_engine(dsn, pool_size=5, max_overflow=10)
    Base.metadata.create_all(engine)

    inserted = 0
    updated = 0
    errors = 0

    session = _get_session(engine)
    try:
        for i in range(0, len(subscriber_dicts), batch_size):
            batch = subscriber_dicts[i : i + batch_size]
            for row in batch:
                try:
                    existing = (
                        session.query(Subscriber)
                        .filter_by(email=row["email"])
                        .one_or_none()
                    )
                    if existing:
                        existing.first_name = row.get("first_name", existing.first_name)
                        existing.timezone = row.get("timezone", existing.timezone)
                        updated += 1
                    else:
                        sub = Subscriber(
                            email=row["email"],
                            first_name=row.get("first_name"),
                            last_name=row.get("last_name"),
                            timezone=row.get("timezone", "UTC"),
                        )
                        session.add(sub)
                        inserted += 1
                except Exception as exc:
                    logger.error("Failed row %s: %s", row.get("email"), exc)
                    errors += 1
            session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()
        engine.dispose()

    return {"inserted": inserted, "updated": updated, "errors": errors}


if __name__ == "__main__":
    # Quick smoke test — run with a local PostgreSQL instance
    sample = [
        {"email": "alice@example.com", "first_name": "Alice", "timezone": "America/New_York"},
        {"email": "bob@example.com", "first_name": "Bob", "timezone": "Asia/Bangkok"},
    ]
    result = bulk_import_subscribers("postgresql://localhost/marketer", sample)
    print(result)

"""
campaign_engine.py — Celery tasks for rendering and sending email campaigns.
Requires: celery>=5.3, boto3>=1.34, mjml-py>=0.0.8, jinja2>=3.1
"""
import hashlib
import logging
from datetime import datetime, timezone

import boto3
from celery import Celery
from jinja2 import Environment, FileSystemLoader
from mjml import to_html  # converts MJML XML → responsive HTML

from subscriber_store import Subscriber  # import from previous example

logger = logging.getLogger(__name__)

app = Celery(
    "campaigns",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    task_default_retry_delay=60,
    task_max_retries=3,
)

SES_REGION = "us-east-1"
SENDER_EMAIL = "noreply@yourdomain.com"
SENDER_NAME = "Your Newsletter"

ses_client = boto3.client("ses", region_name=SES_REGION)

jinja_env = Environment(
    loader=FileSystemLoader(searchpath="./templates"),
    autoescape=True,
)


def _render_template(template_name: str, context: dict) -> str:
    """Render a Jinja2 template to MJML, then convert to HTML."""
    mjml_template = jinja_env.get_template(f"{template_name}.mjml.j2")
    mjml_source = mjml_template.render(**context)
    response = to_html(mjml_source)
    if response.errors:
        raise ValueError(f"MJML errors: {response.errors}")
    return response.html


def _idempotency_key(campaign_id: int, subscriber_email: str) -> str:
    """Deterministic key so SES deduplicates retries."""
    raw = f"{campaign_id}:{subscriber_email}"
    return hashlib.sha256(raw.encode()).hexdigest()


@app.task(bind=True, max_retries=3)
def send_campaign_email(self, campaign_id: int, subscriber_email: str):
    """
    Send one campaign email to one subscriber.

    Idempotent: uses MessageDeduplicationId so duplicate
    sends within a 5-minute window are silently ignored by SES.
    """
    try:
        # Fetch subscriber from DB
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        engine = create_engine("postgresql://localhost/marketer")
        Session = sessionmaker(bind=engine)
        session = Session()

        subscriber = (
            session.query(Subscriber)
            .filter_by(email=subscriber_email)
            .one_or_none()
        )
        if subscriber is None:
            logger.warning("Subscriber %s not found, skipping.", subscriber_email)
            session.close()
            engine.dispose()
            return {"status": "skipped", "reason": "not_found"}

        # Build personalized context
        context = {
            "first_name": subscriber.first_name or subscriber.email.split("@")[0],
            "campaign_id": campaign_id,
            "unsubscribe_url": f"https://yourdomain.com/unsub?email={subscriber_email}",
        }

        html_body = _render_template(f"campaign_{campaign_id}", context)

        dedup_key = _idempotency_key(campaign_id, subscriber_email)

        ses_client.send_email(
            Source=f"{SENDER_NAME} <{SENDER_EMAIL}>",
            Destination={"ToAddresses": [subscriber.email]},
            Message={
                "Subject": {"Data": "Your Weekly Update", "Charset": "UTF-8"},
                "Body": {"Html": {"Data": html_body, "Charset": "UTF-8"}},
            },
            MessageDeduplicationId=dedup_key,
            MessageGroupId="campaigns",  # required for FIFO config
        )

        # Update stats
        subscriber.emails_sent += 1
        subscriber.last_opened_at = datetime.now(timezone.utc)
        session.commit()

        logger.info("Sent campaign %d to %s", campaign_id, subscriber.email)
        return {"status": "sent", "email": subscriber.email}

    except Exception as exc:
        logger.error("Failed to send to %s: %s", subscriber_email, exc)
        session.rollback()
        raise self.retry(exc=exc, countdown=60)
    finally:
        session.close()
        engine.dispose()

"""
analytics_receiver.py — FastAPI webhook that ingests SES/SNS events
into TimescaleDB for real-time campaign analytics.
Requires: fastapi>=0.109, uvicorn, psycopg2-binary, boto3
"""
import json
import logging
import os
from datetime import datetime, timezone

import boto3
from fastapi import FastAPI, Request, HTTPException
from psycopg2 import sql
from psycopg2.extras import execute_values

logger = logging.getLogger(__name__)
app = FastAPI(title="Marketer Analytics Receiver")

SNS_SNS_PUBLISH_ARN = os.environ.get(
    "SNS_TOPIC_ARN", "arn:aws:sns:us-east-1:123456789:ses-events"
)

# Initialize SNS client for subscription confirmation
sns = boto3.client("sns", region_name="us-east-1")

# ---------- Database helpers ----------
import psycopg2

DB_CONN_STR = os.environ.get(
    "DATABASE_URL",
    "postgresql://postgres:postgres@localhost:5432/marketer",
)


def get_conn():
    """Return a new psycopg2 connection. Callers must close it."""
    return psycopg2.connect(DB_CONN_STR)


def init_db():
    """Create the TimescaleDB hypertable if it does not exist."""
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS email_events (
                    time        TIMESTAMPTZ NOT NULL,
                    subscriber_email TEXT NOT NULL,
                    campaign_id     INTEGER NOT NULL,
                    event_type      TEXT NOT NULL,  -- 'open' or 'click'
                    link            TEXT,
                    user_agent      TEXT
                );
            """)
            # Convert to hypertable (idempotent)
            cur.execute("""
                SELECT create_hypertable(
                    'email_events', 'time', if_not_exists => TRUE
                );
            """)
            conn.commit()
            logger.info("Database initialized successfully.")
    finally:
        conn.close()


@app.on_event("startup")
async def startup():
    init_db()


# ---------- Webhook endpoint ----------
@app.post("/webhook/analytics")
async def receive_event(request: Request):
    """
    Accept SNS → SES event notifications.
    SNS wraps the payload; we unwrap, validate, and batch-insert.
    """
    payload = await request.json()

    # Handle SNS subscription confirmation
    if payload.get("Type") == "SubscriptionConfirmation":
        sns.confirm_subscription(
            TopicArn=SNS_SNS_PUBLISH_ARN,
            Token=payload["Token"],
        )
        return {"status": "subscribed"}

    # Extract actual SES record
    records = payload.get("Records", [])
    if not records:
        raise HTTPException(status_code=400, detail="No records in payload")

    events_to_insert = []
    now = datetime.now(timezone.utc)

    for record in records:
        sns_body = json.loads(record.get("Sns", {}).get("Message", "{}"))
        ses_type = sns_body.get("notificationType")

        if ses_type not in ("Open", "Click"):
            continue  # ignore bounces, deliveries for now

        mail = sns_body.get("mail", {})
        event_type = "open" if ses_type == "Open" else "click"
        link = sns_body.get("click", {}).get("link") if event_type == "click" else None
        user_agent = sns_body.get("open", {}).get("userAgent") or sns_body.get("click", {}).get("userAgent", "")

        events_to_insert.append(
            (now, mail.get("destination", [""])[0], mail.get("messageId", 0), event_type, link, user_agent)
        )

    if not events_to_insert:
        return {"status": "ignored", "reason": "no open/click events"}

    # Batch insert for throughput
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            execute_values(
                cur,
                """
                INSERT INTO email_events (time, subscriber_email, campaign_id, event_type, link, user_agent)
                VALUES %s
                """,
                events_to_insert,
            )
        conn.commit()
        logger.info("Inserted %d events.", len(events_to_insert))
    except Exception as exc:
        conn.rollback()
        logger.error("Insert failed: %s", exc)
        raise HTTPException(status_code=500, detail="Database write failed")
    finally:
        conn.close()

    return {"status": "ok", "inserted": len(events_to_insert)}


# ---------- Query endpoint ----------
@app.get("/stats/campaign/{campaign_id}")
async def campaign_stats(campaign_id: int):
    """Return open and click counts for a single campaign."""
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT event_type, COUNT(*)
                FROM email_events
                WHERE campaign_id = %s
                GROUP BY event_type;
            """, (campaign_id,))
            rows = cur.fetchall()
    finally:
        conn.close()

    stats = {event_type: count for event_type, count in rows}
    return {
        "campaign_id": campaign_id,
        "opens": stats.get("open", 0),
        "clicks": stats.get("click", 0),
    }


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Redis-backed idempotency check for any provider
def is_duplicate(key: str, ttl_seconds: int = 86400) -> bool:
    """Return True if this key was already processed."""
    result = redis_client.set(key, "1", nx=True, ex=ttl_seconds)
    return result is None  # None means key already existed

from mjml import to_html

def render_campaign(template_path: str, context: dict) -> str:
    with open(template_path) as f:
        mjml_source = f.read()
    # Simple placeholder substitution
    for k, v in context.items():
        mjml_source = mjml_source.replace("{{" + k + "}}", str(v))
    result = to_html(mjml_source)
    if result.errors:
        raise RuntimeError(f"MJML compile errors: {result.errors}")
    return result.html

# Enable compression on events older than 7 days
ALTER TABLE email_events SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'subscriber_email, campaign_id',
    timescaledb.compress_orderby = 'time DESC'
);
SELECT add_compression_policy('email_events', INTERVAL '7 days');