add stripe (#121)

* add some stripe stuff

* more stripe stuff

* more stripe things

* more stripr stuff

* more stripe stuff

* more stripe stuff

* add more stuff

* add more stripe stuff

* more stuff

* fix types

apps/web/src/server/api/root.ts

@@ -7,6 +7,7 @@ import { adminRouter } from "./routers/admin";
 import { contactsRouter } from "./routers/contacts";
 import { campaignRouter } from "./routers/campaign";
 import { templateRouter } from "./routers/template";
+import { billingRouter } from "./routers/billing";
 
 /**
  * This is the primary router for your server.
@@ -22,6 +23,7 @@ export const appRouter = createTRPCRouter({
   contacts: contactsRouter,
   campaign: campaignRouter,
   template: templateRouter,
+  billing: billingRouter,
 });
 
 // export type definition of API

apps/web/src/server/api/routers/billing.ts

@@ -0,0 +1,82 @@
+import { DailyEmailUsage, EmailUsageType } from "@prisma/client";
+import { TRPCError } from "@trpc/server";
+import { format } from "date-fns";
+import { z } from "zod";
+
+import {
+  apiKeyProcedure,
+  createTRPCRouter,
+  teamProcedure,
+} from "~/server/api/trpc";
+import {
+  createCheckoutSessionForTeam,
+  getManageSessionUrl,
+} from "~/server/billing/payments";
+import { db } from "~/server/db";
+
+export const billingRouter = createTRPCRouter({
+  createCheckoutSession: teamProcedure.mutation(async ({ ctx }) => {
+    return (await createCheckoutSessionForTeam(ctx.team.id)).url;
+  }),
+
+  getManageSessionUrl: teamProcedure.mutation(async ({ ctx }) => {
+    return await getManageSessionUrl(ctx.team.id);
+  }),
+
+  getThisMonthUsage: teamProcedure.query(async ({ ctx }) => {
+    const isoStartDate = format(new Date(), "yyyy-MM-01"); // First day of current month
+    const today = format(new Date(), "yyyy-MM-dd");
+
+    const [monthUsage, dayUsage] = await Promise.all([
+      // Get month usage
+      db.$queryRaw<Array<{ type: EmailUsageType; sent: number }>>`
+        SELECT 
+          type,
+          SUM(sent)::integer AS sent
+        FROM "DailyEmailUsage"
+        WHERE "teamId" = ${ctx.team.id}
+        AND "date" >= ${isoStartDate}
+        GROUP BY "type"
+      `,
+      // Get today's usage
+      db.$queryRaw<Array<{ type: EmailUsageType; sent: number }>>`
+        SELECT 
+          type,
+          SUM(sent)::integer AS sent
+        FROM "DailyEmailUsage"
+        WHERE "teamId" = ${ctx.team.id}
+        AND "date" = ${today}
+        GROUP BY "type"
+      `,
+    ]);
+
+    return {
+      month: monthUsage,
+      day: dayUsage,
+    };
+  }),
+
+  getSubscriptionDetails: teamProcedure.query(async ({ ctx }) => {
+    const subscription = await db.subscription.findFirst({
+      where: { teamId: ctx.team.id },
+      orderBy: { status: "asc" },
+    });
+
+    return subscription;
+  }),
+
+  updateBillingEmail: teamProcedure
+    .input(
+      z.object({
+        billingEmail: z.string().email(),
+      })
+    )
+    .mutation(async ({ ctx, input }) => {
+      const { billingEmail } = input;
+
+      await db.team.update({
+        where: { id: ctx.team.id },
+        data: { billingEmail },
+      });
+    }),
+});

apps/web/src/server/api/routers/email.ts

@@ -101,8 +101,6 @@ export const emailRouter = createTRPCRouter({
         ORDER BY "date" ASC
       `;
 
-      console.log({ result });
-
       // Fill in any missing dates with 0 values
       const filledResult: DailyEmailUsage[] = [];
       const endDateObj = new Date();

apps/web/src/server/billing/payments.ts

@@ -0,0 +1,162 @@
+import Stripe from "stripe";
+import { env } from "~/env";
+import { db } from "../db";
+
+function getStripe() {
+  if (!env.STRIPE_SECRET_KEY) {
+    throw new Error("STRIPE_SECRET_KEY is not set");
+  }
+
+  return new Stripe(env.STRIPE_SECRET_KEY);
+}
+
+async function createCustomerForTeam(teamId: number) {
+  const stripe = getStripe();
+  const customer = await stripe.customers.create({ metadata: { teamId } });
+
+  await db.team.update({
+    where: { id: teamId },
+    data: {
+      stripeCustomerId: customer.id,
+      billingEmail: customer.email,
+    },
+  });
+
+  return customer;
+}
+
+export async function createCheckoutSessionForTeam(teamId: number) {
+  const team = await db.team.findUnique({
+    where: { id: teamId },
+  });
+
+  if (!team) {
+    throw new Error("Team not found");
+  }
+
+  if (team.isActive && team.plan !== "FREE") {
+    throw new Error("Team is already active");
+  }
+
+  const stripe = getStripe();
+
+  let customerId = team.stripeCustomerId;
+
+  if (!customerId) {
+    const customer = await createCustomerForTeam(teamId);
+    customerId = customer.id;
+  }
+
+  if (!env.STRIPE_BASIC_PRICE_ID || !customerId) {
+    throw new Error("Stripe prices are not set");
+  }
+
+  const session = await stripe.checkout.sessions.create({
+    mode: "subscription",
+    customer: customerId,
+    line_items: [
+      {
+        price: env.STRIPE_BASIC_PRICE_ID,
+      },
+    ],
+    success_url: `${env.NEXTAUTH_URL}/payments?success=true&session_id={CHECKOUT_SESSION_ID}`,
+    cancel_url: `${env.NEXTAUTH_URL}/settings/billing`,
+    metadata: {
+      teamId,
+    },
+    client_reference_id: teamId.toString(),
+  });
+
+  return session;
+}
+
+function getPlanFromPriceId(priceId: string) {
+  if (priceId === env.STRIPE_BASIC_PRICE_ID) {
+    return "BASIC";
+  }
+
+  return "FREE";
+}
+
+export async function getManageSessionUrl(teamId: number) {
+  const team = await db.team.findUnique({
+    where: { id: teamId },
+  });
+
+  if (!team) {
+    throw new Error("Team not found");
+  }
+
+  if (!team.stripeCustomerId) {
+    throw new Error("Team has no Stripe customer ID");
+  }
+
+  const stripe = getStripe();
+
+  const subscriptions = await stripe.billingPortal.sessions.create({
+    customer: team.stripeCustomerId,
+    return_url: `${env.NEXTAUTH_URL}`,
+  });
+
+  return subscriptions.url;
+}
+
+export async function syncStripeData(customerId: string) {
+  const stripe = getStripe();
+
+  const team = await db.team.findUnique({
+    where: { stripeCustomerId: customerId },
+  });
+
+  if (!team) {
+    return;
+  }
+
+  const subscriptions = await stripe.subscriptions.list({
+    customer: customerId,
+    limit: 1,
+    status: "all",
+    expand: ["data.default_payment_method"],
+  });
+
+  const subscription = subscriptions.data[0];
+
+  if (!subscription) {
+    return;
+  }
+
+  await db.subscription.upsert({
+    where: { id: subscription.id },
+    update: {
+      status: subscription.status,
+      priceId: subscription.items.data[0]?.price?.id || "",
+      currentPeriodEnd: new Date(subscription.current_period_end * 1000),
+      currentPeriodStart: new Date(subscription.current_period_start * 1000),
+      cancelAtPeriodEnd: subscription.cancel_at
+        ? new Date(subscription.cancel_at * 1000)
+        : null,
+      paymentMethod: JSON.stringify(subscription.default_payment_method),
+      teamId: team.id,
+    },
+    create: {
+      id: subscription.id,
+      status: subscription.status,
+      priceId: subscription.items.data[0]?.price?.id || "",
+      currentPeriodEnd: new Date(subscription.current_period_end * 1000),
+      currentPeriodStart: new Date(subscription.current_period_start * 1000),
+      cancelAtPeriodEnd: subscription.cancel_at
+        ? new Date(subscription.cancel_at * 1000)
+        : null,
+      paymentMethod: JSON.stringify(subscription.default_payment_method),
+      teamId: team.id,
+    },
+  });
+
+  await db.team.update({
+    where: { id: team.id },
+    data: {
+      plan: getPlanFromPriceId(subscription.items.data[0]?.price?.id || ""),
+      isActive: subscription.status === "active",
+    },
+  });
+}

apps/web/src/server/billing/usage.ts

@@ -0,0 +1,22 @@
+import Stripe from "stripe";
+import { env } from "~/env";
+import { getUsageTimestamp } from "~/lib/usage";
+
+const METER_EVENT_NAME = "unsend_usage";
+
+export async function sendUsageToStripe(customerId: string, usage: number) {
+  const stripe = new Stripe(env.STRIPE_SECRET_KEY!, {
+    apiVersion: "2025-01-27.acacia",
+  });
+
+  const meterEvent = await stripe.billing.meterEvents.create({
+    event_name: METER_EVENT_NAME,
+    payload: {
+      value: usage.toString(),
+      stripe_customer_id: customerId,
+    },
+    timestamp: getUsageTimestamp(),
+  });
+
+  return meterEvent;
+}

apps/web/src/server/jobs/usage-job.ts

@@ -0,0 +1,90 @@
+import { Queue, Worker } from "bullmq";
+import { db } from "~/server/db";
+import { env } from "~/env";
+import { getUsageDate, getUsageUinits } from "~/lib/usage";
+import { sendUsageToStripe } from "~/server/billing/usage";
+import { getRedis } from "~/server/redis";
+import { DEFAULT_QUEUE_OPTIONS } from "../queue/queue-constants";
+
+const USAGE_QUEUE_NAME = "usage-reporting";
+
+const usageQueue = new Queue(USAGE_QUEUE_NAME, {
+  connection: getRedis(),
+});
+
+// Process usage reporting jobs
+const worker = new Worker(
+  USAGE_QUEUE_NAME,
+  async () => {
+    // Get all teams with stripe customer IDs
+    const teams = await db.team.findMany({
+      where: {
+        stripeCustomerId: {
+          not: null,
+        },
+      },
+      include: {
+        dailyEmailUsages: {
+          where: {
+            // Get yesterday's date by subtracting 1 day from today
+            date: {
+              equals: getUsageDate(),
+            },
+          },
+        },
+      },
+    });
+
+    // Process each team
+    for (const team of teams) {
+      if (!team.stripeCustomerId) continue;
+
+      const transactionUsage = team.dailyEmailUsages
+        .filter((usage) => usage.type === "TRANSACTIONAL")
+        .reduce((sum, usage) => sum + usage.sent, 0);
+
+      const marketingUsage = team.dailyEmailUsages
+        .filter((usage) => usage.type === "MARKETING")
+        .reduce((sum, usage) => sum + usage.sent, 0);
+
+      const totalUsage = getUsageUinits(marketingUsage, transactionUsage);
+
+      try {
+        await sendUsageToStripe(team.stripeCustomerId, totalUsage);
+        console.log(
+          `[Usage Reporting] Reported usage for team ${team.id}, date: ${getUsageDate()}, usage: ${totalUsage}`
+        );
+      } catch (error) {
+        console.error(
+          `[Usage Reporting] Failed to report usage for team ${team.id}:`,
+          error instanceof Error ? error.message : error
+        );
+      }
+    }
+  },
+  {
+    connection: getRedis(),
+  }
+);
+
+// Schedule job to run daily
+await usageQueue.upsertJobScheduler(
+  "daily-usage-report",
+  {
+    pattern: "0 0 * * *", // Run every day at 12 AM
+    tz: "UTC",
+  },
+  {
+    opts: {
+      ...DEFAULT_QUEUE_OPTIONS,
+    },
+  }
+);
+
+worker.on("completed", (job) => {
+  console.log(`[Usage Reporting] Job ${job.id} completed`);
+});
+
+worker.on("failed", (job, err) => {
+  console.error(`[Usage Reporting] Job ${job?.id} failed:`, err);
+});