for Startups CRM System in 2026: Real Results

"""
CRM Contact API — FastAPI + SQLAlchemy 2.0 + PostgreSQL
Requirements: pip install fastapi uvicorn sqlalchemy asyncpg pydantic[email]
"""
import asyncio
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Optional

from fastapi import FastAPI, HTTPException, Query, status
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy import (
    Column,
    DateTime,
    Integer,
    String,
    create_engine,
    func,
    select,
    text,
)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

# ---------------------------------------------------------------------------
# Database setup — connection pool tuned for a 2-vCPU RDS instance
# ---------------------------------------------------------------------------
DATABASE_URL = "postgresql+asyncpg://crm_user:changeme@localhost:5432/startup_crm"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,
    echo=False,  # Set True for SQL debug logging
)

async_session_factory = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


class Base(DeclarativeBase):
    """Base class for all ORM models."""
    pass


class Contact(Base):
    __tablename__ = "contacts"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    first_name: Mapped[str] = mapped_column(String(120), nullable=False)
    last_name: Mapped[str] = mapped_column(String(120), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    company: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
    phone: Mapped[Optional[str]] = mapped_column(String(30), nullable=True)
    source: Mapped[str] = mapped_column(String(50), server_default="organic")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
    )


# ---------------------------------------------------------------------------
# Pydantic schemas — strict validation at the API boundary
# ---------------------------------------------------------------------------
class ContactCreate(BaseModel):
    first_name: str = Field(..., min_length=1, max_length=120)
    last_name: str = Field(..., min_length=1, max_length=120)
    email: EmailStr
    company: Optional[str] = Field(None, max_length=255)
    phone: Optional[str] = Field(None, max_length=30)
    source: str = "organic"


class ContactResponse(BaseModel):
    id: int
    first_name: str
    last_name: str
    email: str
    company: Optional[str]
    phone: Optional[str]
    source: str
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}


class PaginatedContacts(BaseModel):
    total: int
    page: int
    per_page: int
    items: list[ContactResponse]


# ---------------------------------------------------------------------------
# Lifespan — create tables on startup (use Alembic in production)
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(title="Startup CRM", version="1.0.0", lifespan=lifespan)


# ---------------------------------------------------------------------------
# Dependency — async DB session with automatic rollback on error
# ---------------------------------------------------------------------------
@asynccontextmanager
async def get_db():
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@app.post("/contacts", response_model=ContactResponse, status_code=status.HTTP_201_CREATED)
async def create_contact(payload: ContactCreate, db: AsyncSession = Depends(get_db)):
    """Create a new contact. Returns 409 if email already exists."""
    existing = await db.execute(select(Contact).where(Contact.email == payload.email))
    if existing.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Email already registered")
    contact = Contact(**payload.model_dump())
    db.add(contact)
    await db.flush()
    await db.refresh(contact)
    return contact


@app.get("/contacts", response_model=PaginatedContacts)
async def list_contacts(
    page: int = Query(1, ge=1),
    per_page: int = Query(25, ge=1, le=100),
    source: Optional[str] = None,
    company: Optional[str] = None,
    db: AsyncSession = Depends(get_db),
):
    """Paginated listing with optional filters on source and company."""
    query = select(Contact)
    if source:
        query = query.where(Contact.source == source)
    if company:
        query = query.where(Contact.company.ilike(f"%{company}%"))
    query = query.order_by(Contact.created_at.desc())

    total_result = await db.execute(select(func.count()).select_from(query.subquery()))
    total = total_result.scalar()

    offset = (page - 1) * per_page
    result = await db.execute(query.offset(offset).limit(per_page))
    items = result.scalars().all()
    return PaginatedContacts(total=total, page=page, per_page=per_page, items=items)


@app.get("/contacts/{contact_id}", response_model=ContactResponse)
async def get_contact(contact_id: int, db: AsyncSession = Depends(get_db)):
    """Fetch a single contact by ID."""
    result = await db.get(Contact, contact_id)
    if not result:
        raise HTTPException(status_code=404, detail="Contact not found")
    return result


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

/**
 * CRM Webhook Processor — Node.js 20 + BullMQ + ioredis
 * Requirements: npm install bullmq ioredis zod stripe tsx
 *
 * This processor subscribes to a Redis-backed queue and handles
 * incoming Stripe webhook events with at-least-once delivery
 * semantics and automatic retry with exponential backoff.
 */
import { Queue, Worker, QueueEvents } from "bullmq";
import { Redis } from "ioredis";
import { z } from "zod";
import crypto from "node:crypto";

// ---------------------------------------------------------------------------
// Redis connection — single shared connection for queues and workers
// ---------------------------------------------------------------------------
const connection = new Redis({
  host: process.env.REDIS_HOST || "127.0.0.1",
  port: parseInt(process.env.REDIS_PORT || "6379", 10),
  password: process.env.REDIS_PASSWORD || undefined,
  maxRetriesPerRequest: 3,
  retryStrategy(times) {
    const delay = Math.min(times * 200, 5000);
    return delay;
  },
});

// ---------------------------------------------------------------------------
// Schema validation — strict typing on every inbound payload
// ---------------------------------------------------------------------------
const InvoicePaidSchema = z.object({
  id: z.string(),
  customer: z.string(),
  customer_email: z.string().email(),
  amount_paid: z.number(),
  currency: z.string(),
  status: z.string(),
  created: z.number(),
  lines: z.object({
    data: z.array(
      z.object({
        description: z.string().optional(),
        amount: z.number(),
        plan: z.object({ name: z.string().optional() }).optional(),
      })
    ),
  }),
});

type InvoicePaidEvent = z.infer;

// ---------------------------------------------------------------------------
// Queue setup with rate limiting and stalled job detection
// ---------------------------------------------------------------------------
const queue = new Queue("crm-webhooks", {
  connection,
  defaultJobOptions: {
    attempts: 5,
    backoff: {
      type: "exponential",
      delay: 2000, // First retry in 2s, then 4s, 8s, 16s, 32s
    },
    removeOnComplete: 1000, // Keep last 1000 completed jobs
    removeOnFail: 5000,    // Keep last 5000 failed jobs for debugging
  },
});

const queueEvents = new QueueEvents("crm-webhooks", { connection });

// Monitor for stalled jobs and log them
queueEvents.on("stalled", ({ jobId }) => {
  console.error(`[CRM] Job ${jobId} stalled — will be retried`);
});

queueEvents.on("failed", ({ jobId, failedReason }) => {
  console.error(`[CRM] Job ${jobId} failed permanently: ${failedReason}`);
});

// ---------------------------------------------------------------------------
// Worker — processes webhook events with idempotency check
// ---------------------------------------------------------------------------
const worker = new Worker(
  "crm-webhooks",
  async (job) => {
    const { event, signature, timestamp } = job.data;

    // Idempotency check — skip if we already processed this event
    const processed = await connection.get(`crm:webhook:stripe:${event.id}`);
    if (processed) {
      console.log(`[CRM] Duplicate event ${event.id}, skipping`);
      return { status: "duplicate", eventId: event.id };
    }

    // Verify Stripe signature to prevent spoofing
    const expectedSig = crypto
      .createHmac("sha256", process.env.STRIPE_WEBHOOK_SECRET!)
      .update(`${timestamp}.${JSON.stringify(event)}`)
      .digest("hex");

    if (expectedSig !== signature) {
      throw new Error(`[CRM] Invalid Stripe signature for event ${event.id}`);
    }

    // Validate payload shape
    const validated = InvoicePaidSchema.parse(event.data);

    // Upsert contact record — this is where you sync to your CRM DB
    await upsertContactFromInvoice(validated);

    // Mark as processed with 48-hour TTL to allow reprocessing edge cases
    await connection.set(
      `crm:webhook:stripe:${event.id}`,
      JSON.stringify({ processedAt: new Date().toISOString() }),
      "EX",
      172800 // 48 hours
    );

    console.log(
      `[CRM] Processed invoice ${validated.id} for ${validated.customer_email} — $${(validated.amount_paid / 100).toFixed(2)}`
    );

    return { status: "processed", eventId: event.id, email: validated.customer_email };
  },
  {
    connection,
    concurrency: 10, // Process up to 10 webhook events concurrently
  }
);

// ---------------------------------------------------------------------------
// Database upsert — replace with your actual ORM / query runner
// ---------------------------------------------------------------------------
async function upsertContactFromInvoice(invoice: InvoicePaidEvent): Promise {
  // In production, use your ORM (Prisma, Drizzle, TypeORM) here.
  // This is a simplified pg client example.
  const { Pool } = await import("pg");
  const pool = new Pool({
    connectionString: process.env.DATABASE_URL,
  });

  const client = await pool.connect();
  try {
    await client.query("BEGIN");

    const upsertQuery = `
      INSERT INTO contacts (email, company, stripe_customer_id, last_payment_amount, last_payment_date)
      VALUES ($1, $2, $3, $4, to_timestamp($5))
      ON CONFLICT (email) DO UPDATE SET
        last_payment_amount = EXCLUDED.last_payment_amount,
        last_payment_date = EXCLUDED.last_payment_date,
        updated_at = NOW()
    `;

    await client.query(upsertQuery, [
      invoice.customer_email,
      invoice.lines.data[0]?.plan?.name() ?? null,
      invoice.customer,
      invoice.amount_paid / 100,
      invoice.created,
    ]);

    await client.query("COMMIT");
  } catch (err) {
    await client.query("ROLLBACK");
    console.error(`[CRM] Upsert failed for ${invoice.customer_email}:`, err);
    throw err;
  } finally {
    client.release();
  }
}

// ---------------------------------------------------------------------------
// HTTP endpoint to receive Stripe webhooks and enqueue for processing
// ---------------------------------------------------------------------------
import express from "express";

const app = express();
app.use(express.raw({ type: "application/json" }));

app.post("/webhooks/stripe", async (req, res) => {
  try {
    const event = JSON.parse(req.body.toString());

    // Filter to only invoice.paid events — discard the rest immediately
    if (event.type !== "invoice.paid") {
      return res.status(200).json({ received: false, reason: "not_invoice_paid" });
    }

    // Enqueue for async processing — the webhook returns 200 immediately
    await queue.add(
      "stripe-invoice-paid",
      {
        event: event.data,
        signature: req.headers["stripe-signature"],
        timestamp: req.headers["stripe-timestamp"],
      },
      {
        attempts: 5,
        delay: 1000, // Initial 1s delay before first attempt
      }
    );

    res.status(200).json({ received: true });
  } catch (err) {
    console.error("[CRM] Webhook ingestion error:", err);
    res.status(500).json({ error: "Internal server error" });
  }
});

app.listen(3001, () => {
  console.log("[CRM] Webhook processor listening on port 3001");
});

"""
CRM Analytics Pipeline — Python 3.12 + DuckDB + psycopg3
Requirements: pip install duckdb psycopg[binary] pandas tenacity

Reads directly from the CRM Postgres database, computes funnel
metrics, and writes aggregated results back for dashboarding.
"""
import duckdb
import pandas as pd
from datetime import datetime, timedelta
from tenacity import retry, stop_after_attempt, wait_exponential
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
POSTGRES_URL = "postgresql://crm_user:changeme@localhost:5432/startup_crm"
DAYS_BACK = 90
OUTPUT_TABLE = "analytics_weekly_funnel"


# ---------------------------------------------------------------------------
# Resilient connection with exponential backoff
# ---------------------------------------------------------------------------
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    reraise=True,
)
def create_duckdb_connection():
    """Open a DuckDB connection with Postgres scanning enabled."""
    con = duckdb.connect(":memory:", config={"threads": 4})
    con.execute(f"INSTALL postgres;")
    con.execute(f"LOAD postgres;")
    con.execute(f"""
        ATTACH '{POSTGRES_URL}' AS pg
        (TYPE postgres, READ_ONLY true);
    """)
    logger.info("Connected to Postgres via DuckDB")
    return con


# ---------------------------------------------------------------------------
# Core funnel computation
# ---------------------------------------------------------------------------
def compute_weekly_funnel(con: duckdb.DuckDBPyConnection) -> pd.DataFrame:
    """
    Computes a weekly funnel from contact creation → first interaction → deal won.
    Returns a DataFrame with columns:
        week_start, contacts_created, interactions, deals_won, conversion_rate
    """
    query = f"""
    WITH weeks AS (
        SELECT
            generate_series(
                date_trunc('week', CURRENT_DATE - INTERVAL '{DAYS_BACK} days'),
                date_trunc('week', CURRENT_DATE),
                INTERVAL '1 week'
            )::date AS week_start
    ),
    created AS (
        SELECT
            date_trunc('week', created_at)::date AS week_start,
            COUNT(*) AS contacts_created
        FROM pg.public.contacts
        WHERE created_at >= CURRENT_DATE - INTERVAL '{DAYS_BACK} days'
        GROUP BY 1
    ),
    interactions AS (
        SELECT
            date_trunc('week', i.created_at)::date AS week_start,
            COUNT(DISTINCT i.contact_id) AS interaction_contacts
        FROM pg.public.interactions i
        WHERE i.created_at >= CURRENT_DATE - INTERVAL '{DAYS_BACK} days'
        GROUP BY 1
    ),
    deals AS (
        SELECT
            date_trunc('week', closed_at)::date AS week_start,
            COUNT(*) AS deals_won
        FROM pg.public.deals
        WHERE closed_at >= CURRENT_DATE - INTERVAL '{DAYS_BACK} days'
          AND stage = 'closed_won'
        GROUP BY 1
    )
    SELECT
        w.week_start,
        COALESCE(c.contacts_created, 0) AS contacts_created,
        COALESCE(i.interaction_contacts, 0) AS interactions,
        COALESCE(d.deals_won, 0) AS deals_won,
        ROUND(
            100.0 * COALESCE(d.deals_won, 0) / NULLIF(COALESCE(c.contacts_created, 0), 0),
            2
        ) AS conversion_rate
    FROM weeks w
    LEFT JOIN created c ON w.week_start = c.week_start
    LEFT JOIN interactions i ON w.week_start = i.week_start
    LEFT JOIN deals d ON w.week_start = d.week_start
    ORDER BY w.week_start;
    """
    result = con.execute(query).fetchdf()
    logger.info(f"Funnel computed: {len(result)} weeks, {result['contacts_created'].sum()} total contacts")
    return result


# ---------------------------------------------------------------------------
# Persist results back to Postgres
# ---------------------------------------------------------------------------
def persist_results(con: duckdb.DuckDBPyConnection, df: pd.DataFrame) -> int:
    """Write the funnel DataFrame to analytics_weekly_funnel table."""
    # Create table if needed (idempotent)
    con.execute(f"""
        CREATE TABLE IF NOT EXISTS {OUTPUT_TABLE} (
            week_start DATE PRIMARY KEY,
            contacts_created INTEGER NOT NULL,
            interactions INTEGER NOT NULL,
            deals_won INTEGER NOT NULL,
            conversion_rate DOUBLE NOT NULL,
            computed_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
        );
    """)

    # Use INSERT ... ON CONFLICT for idempotent upsert
    from duckdb import CatalogException
    try:
        # DuckDB's pandas integration handles batch inserts efficiently
        con.execute(f"""
            INSERT INTO {OUTPUT_TABLE}
            (week_start, contacts_created, interactions, deals_won, conversion_rate)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT (week_start) DO UPDATE SET
                contacts_created = EXCLUDED.contacts_created,
                interactions = EXCLUDED.interactions,
                deals_won = EXCLUDED.deals_won,
                conversion_rate = EXCLUDED.conversion_rate,
                computed_at = NOW();
        """)
    except CatalogException as e:
        logger.error(f"Failed to persist results: {e}")
        raise

    row_count = len(df)
    logger.info(f"Persisted {row_count} rows to {OUTPUT_TABLE}")
    return row_count


# ---------------------------------------------------------------------------
# Main pipeline entry point
# ---------------------------------------------------------------------------
def main():
    """Run the full analytics pipeline with error handling."""
    logger.info("Starting CRM analytics pipeline")
    start_time = datetime.now(timezone.utc)

    con = None
    try:
        con = create_duckdb_connection()
        funnel_df = compute_weekly_funnel(con)

        if funnel_df.empty:
            logger.warning("No data returned — check source tables")
            return

        rows = persist_results(con, funnel_df)

        elapsed = (datetime.now(timezone.utc) - start_time).total_seconds()
        logger.info(f"Pipeline complete: {rows} rows in {elapsed:.2f}s")

        # Print summary to stdout for CI / cron job monitoring
        print(f"\n{'='*60}")
        print(f"  CRM Weekly Funnel — Last {DAYS_BACK} Days")
        print(f"{'='*60}")
        print(funnel_df.to_string(index=False))
        print(f"{'='*60}\n")

    except duckdb.Error as e:
        logger.error(f"DuckDB query error: {e}")
        raise SystemExit(1)
    except Exception as e:
        logger.error(f"Pipeline failed: {e}")
        raise SystemExit(2)
    finally:
        if con:
            con.close()
            logger.info("DuckDB connection closed")


if __name__ == "__main__":
    main()

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select
from models import Contact

engine = create_async_engine("postgresql+asyncpg://localhost/crm", pool_size=15)
async_session = async_sessionmaker(engine, expire_on_commit=False)

@asynccontextmanager
async def get_session():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

async def get_contacts_by_company(company_name: str, session: AsyncSession = Depends(get_session)):
    result = await session.execute(
        select(Contact).where(Contact.company.ilike(f"%{company_name}%"))
    )
    return result.scalars().all()

// BullMQ deduplication using the Stripe event ID as the job ID
await queue.add(
  "stripe-invoice-paid",
  { event: event.data, timestamp: Date.now() },
  { jobId: event.id } // BullMQ skips if this jobId already exists
);

import duckdb
con = duckdb.connect(":memory:")
con.execute("INSTALL postgres; LOAD postgres;")
con.execute("ATTACH 'postgresql://localhost/crm' AS pg (TYPE postgres);")
result = con.execute("""
    SELECT date_trunc('week', created_at)::date AS week,
           COUNT(*) AS new_contacts
    FROM pg.public.contacts
    WHERE created_at > CURRENT_DATE - INTERVAL '90 days'
    GROUP BY 1 ORDER BY 1
""").fetchdf()
print(result)