feat: implement billing and subscription schema and migrate credits l…

.github/workflows/be-billing.yml

@@ -0,0 +1,68 @@
+name: BE Billing Docker
+
+on:
+  push:
+    branches: [ main ]
+    paths:
+      - 'apps/backend/billing/**'
+      - '.github/workflows/be-billing.yml'
+  pull_request:
+    branches: [ main ]
+    paths:
+      - 'apps/backend/billing/**'
+      - '.github/workflows/be-billing.yml'
+  workflow_dispatch:
+
+env:
+  REGISTRY: docker.io
+  IMAGE_NAME: ${{ secrets.DOCKERHUB_USERNAME }}/be-billing
+
+jobs:
+  build-and-push:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Log in to Docker Hub
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Extract metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=semver,pattern={{version}}
+            type=raw,value=latest,enable={{is_default_branch}}
+
+      - name: Build and push Docker image
+        id: build
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: ./apps/backend/billing/Dockerfile
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+          platforms: linux/amd64,linux/arm64
+
+      - name: Image digest
+        run: echo ${{ steps.build.outputs.digest }}
+
+      - name: Deploy to Dokploy
+        if: github.event_name != 'pull_request' && secrets.DOKPLOY_BILLING_TOKEN != ''
+        run: curl -X GET "https://dokploy.reloop.sh/api/deploy/${{ secrets.DOKPLOY_BILLING_TOKEN }}"

apps/backend/credits/Dockerfile → apps/backend/billing/Dockerfile

@@ -8,7 +8,7 @@ WORKDIR /app
 RUN apk add --no-cache python3 make g++
 
 COPY . .
-RUN bun --filter=credits install
+RUN bun --filter=billing install
 
 # Stage 2: Build the application
 FROM base AS builder
@@ -17,7 +17,7 @@ ARG NODE_ENV=production
 ENV NODE_ENV=${NODE_ENV}
 COPY --from=deps /app/node_modules ./node_modules
 COPY . .
-RUN bun run --filter=credits build
+RUN bun run --filter=billing build
 
 # Stage 3: Production server
 FROM base AS runner
@@ -27,7 +27,7 @@ ENV NODE_ENV=production
 ENV HOSTNAME=0.0.0.0
 
 # Copy standalone build
-COPY --from=builder /app/apps/backend/credits/dist ./dist
+COPY --from=builder /app/apps/backend/billing/dist ./dist
 COPY --from=builder /app/node_modules ./node_modules
 COPY --from=builder /app/packages ./packages
 COPY --from=builder /app/package.json ./package.json

apps/backend/credits/package.json → apps/backend/billing/package.json

@@ -1,5 +1,5 @@
 {
-	"name": "credits",
+	"name": "billing",
 	"main": "src/index.ts",
 	"type": "module",
 	"scripts": {

apps/backend/credits/src/credits.config.ts → apps/backend/billing/src/billing.config.ts

@@ -1,7 +1,7 @@
 import "dotenv/config";
 
-export const creditsConfig = {
-	port: Number(process.env.CREDITS_PORT) || 8023,
+export const billingConfig = {
+	port: Number(process.env.BILLING_PORT) || 8023,
 	nodeEnv: process.env.NODE_ENV || "development",
 	initialCredits: Number(process.env.INITIAL_CREDITS) || 100,
 };

apps/backend/billing/src/index.ts

@@ -0,0 +1,116 @@
+import "dotenv/config";
+import cors from "@elysiajs/cors";
+import { openapi } from "@elysiajs/openapi";
+import { logger } from "@reloop/logger";
+import { Elysia, t } from "elysia";
+import { billingConfig } from "./billing.config";
+import { loader } from "./loader";
+import { db } from "@reloop/db/client";
+import {
+	billingInvoice,
+	creditLedger,
+	plan,
+	subscription,
+} from "@reloop/db/schema";
+import { and, desc, eq, sql } from "drizzle-orm";
+
+const port = billingConfig.port;
+
+const app = new Elysia({ prefix: "/api/billing", name: "Billing Service" })
+	.use(cors({ origin: "*" }))
+	.use(
+		openapi({
+			documentation: {
+				info: {
+					title: "Billing Service",
+					version: "1.0.0",
+				},
+			},
+		}),
+	)
+	.get("/balance/:orgId", async ({ params }) => {
+		const result = await db.query.subscription.findFirst({
+			where: (s, { and, eq }) => and(
+				eq(s.organizationId, params.orgId),
+				eq(s.status, "active")
+			),
+			with: {
+				plan: true,
+			},
+		});
+
+		return result || { error: "No active subscription found" };
+	}, {
+		params: t.Object({
+			orgId: t.String(),
+		}),
+	})
+	.get("/transactions/:orgId", async ({ params }) => {
+		return await db.query.creditLedger.findMany({
+			where: eq(creditLedger.organizationId, params.orgId),
+			orderBy: [desc(creditLedger.createdAt)],
+			limit: 50,
+		});
+	}, {
+		params: t.Object({
+			orgId: t.String(),
+		}),
+	})
+	.get("/invoices/:orgId", async ({ params }) => {
+		return await db.query.billingInvoice.findMany({
+			where: eq(billingInvoice.organizationId, params.orgId),
+			orderBy: [desc(billingInvoice.createdAt)],
+		});
+	}, {
+		params: t.Object({
+			orgId: t.String(),
+		}),
+	})
+	.post("/topup", async ({ body }) => {
+		const { organizationId, amount, reason, metadata } = body;
+
+		await db.transaction(async (tx) => {
+			const activeSub = await tx.query.subscription.findFirst({
+				where: (s, { and, eq }) => and(
+					eq(s.organizationId, organizationId),
+					eq(s.status, "active")
+				),
+			});
+
+			if (!activeSub) throw new Error("No active subscription");
+
+			await tx
+				.update(subscription)
+				.set({
+					creditsRemaining: sql`${subscription.creditsRemaining} + ${amount}`,
+					updatedAt: new Date(),
+				})
+				.where(eq(subscription.id, activeSub.id));
+
+			await tx.insert(creditLedger).values({
+				organizationId,
+				subscriptionId: activeSub.id,
+				entryType: "manual_adjustment",
+				delta: amount,
+				balanceAfter: activeSub.creditsRemaining + amount,
+				reason: reason || "Manual top-up",
+			});
+		});
+
+		return { success: true };
+	}, {
+		body: t.Object({
+			organizationId: t.String(),
+			amount: t.Number(),
+			reason: t.Optional(t.String()),
+			metadata: t.Optional(t.Record(t.String(), t.Any())),
+		}),
+	})
+	.onStart(async () => {
+		await loader();
+	})
+	.listen(port, () => {
+		logger.info(`Billing Server is running on http://localhost:${port}/api/billing`);
+	});
+
+export type App = typeof app;

apps/backend/billing/src/loader.ts

@@ -0,0 +1,130 @@
+import { bus, BusEvent } from "@reloop/bus";
+import { db } from "@reloop/db/client";
+import {
+	billingInvoice,
+	creditLedger,
+	emailSend,
+	plan,
+	subscription,
+} from "@reloop/db/schema";
+import { logger } from "@reloop/logger";
+import { and, eq, sql } from "drizzle-orm";
+import { billingConfig } from "./billing.config";
+
+export async function loader() {
+	logger.info("Initializing Billing Service Subscribers...");
+
+	// Handle Organization Created - Initialize subscription and credits
+	await bus.subscribe(BusEvent.ORGANIZATION_CREATED, async (payload) => {
+		logger.info({ organizationId: payload.id }, "Handling ORGANIZATION_CREATED");
+		try {
+			await db.transaction(async (tx) => {
+				// 1. Get or create a default "Free" plan
+				let defaultPlan = await tx.query.plan.findFirst({
+					where: eq(plan.name, "Free"),
+				});
+
+				if (!defaultPlan) {
+					[defaultPlan] = await tx
+						.insert(plan)
+						.values({
+							name: "Free",
+							monthlyCredits: billingConfig.initialCredits,
+							basePriceUsd: "0",
+							isActive: true,
+						})
+						.returning();
+				}
+
+				// 2. Create subscription
+				const now = new Date();
+				const nextMonth = new Date(now);
+				nextMonth.setMonth(nextMonth.getMonth() + 1);
+
+				const [newSub] = await tx
+					.insert(subscription)
+					.values({
+						organizationId: payload.id,
+						planId: defaultPlan.id,
+						status: "active",
+						creditsRemaining: defaultPlan.monthlyCredits,
+						currentPeriodStart: now,
+						currentPeriodEnd: nextMonth,
+					})
+					.onConflictDoNothing()
+					.returning();
+
+				if (newSub) {
+					// 3. Log initial credits in ledger
+					await tx.insert(creditLedger).values({
+						organizationId: payload.id,
+						subscriptionId: newSub.id,
+						entryType: "credit_purchased",
+						delta: defaultPlan.monthlyCredits,
+						balanceAfter: defaultPlan.monthlyCredits,
+						reason: "Initial free plan quota",
+					});
+				}
+			});
+			logger.info({ organizationId: payload.id }, "Initialized subscription for new organization");
+		} catch (error) {
+			logger.error({ error, organizationId: payload.id }, "Failed to initialize subscription");
+		}
+	});
+
+	// Handle Email Sent - Deduct credits
+	await bus.subscribe(BusEvent.EMAIL_SENT, async (payload) => {
+		logger.info({ organizationId: payload.organizationId, count: payload.recipientCount }, "Handling EMAIL_SENT");
+		try {
+			await db.transaction(async (tx) => {
+				// 1. Find active subscription
+				const activeSub = await tx.query.subscription.findFirst({
+					where: (s, { and, eq }) => and(
+						eq(s.organizationId, payload.organizationId),
+						eq(s.status, "active")
+					),
+				});
+
+				if (!activeSub) {
+					logger.warn({ organizationId: payload.organizationId }, "No active subscription found for credit deduction");
+					return;
+				}
+
+				// 2. Create email_send record for billing audit
+				const [sendRecord] = await tx.insert(emailSend).values({
+					organizationId: payload.organizationId,
+					subscriptionId: activeSub.id,
+					recipientEmail: "multiple@recipients.info", // simplified for batch events
+					countedInCredits: true,
+					creditsConsumed: payload.recipientCount,
+					status: "sent",
+					sentAt: new Date(),
+				}).returning();
+
+				// 3. Update subscription counters
+				await tx
+					.update(subscription)
+					.set({
+						creditsUsed: sql`${subscription.creditsUsed} + ${payload.recipientCount}`,
+						creditsRemaining: sql`${subscription.creditsRemaining} - ${payload.recipientCount}`,
+						updatedAt: new Date(),
+					})
+					.where(eq(subscription.id, activeSub.id));
+
+				// 4. Log in credit ledger
+				await tx.insert(creditLedger).values({
+					organizationId: payload.organizationId,
+					subscriptionId: activeSub.id,
+					entryType: "email_sent",
+					delta: -payload.recipientCount,
+					balanceAfter: activeSub.creditsRemaining - payload.recipientCount,
+					reason: `Sent email with ${payload.recipientCount} recipients`,
+					referenceId: sendRecord.id,
+				});
+			});
+			logger.info({ organizationId: payload.organizationId }, "Deducted credits and updated ledger");
+		} catch (error) {
+			logger.error({ error, organizationId: payload.organizationId }, "Failed to deduct credits");
+		}
+	});
+}