update payment system

consumer/src/EmailController.ts

@@ -2,77 +2,56 @@ import { ProjectModel } from "./shared/schema/project/ProjectSchema";
 import { UserModel } from "./shared/schema/UserSchema";
 import { LimitNotifyModel } from "./shared/schema/broker/LimitNotifySchema";
 import { EmailService } from './shared/services/EmailService';
-import { TProjectLimit } from "./shared/schema/project/ProjectsLimits";
 import { EmailServiceHelper } from "./EmailServiceHelper";
+import { TUserLimit } from "./shared/schema/UserLimitSchema";
 
 
-export async function checkLimitsForEmail(projectCounts: TProjectLimit) {
+export async function checkLimitsForEmail(projectCounts: TUserLimit) {
 
-    const project_id = projectCounts.project_id;
-    const hasNotifyEntry = await LimitNotifyModel.findOne({ project_id });
+    const user_id = projectCounts.user_id;
+
+    const hasNotifyEntry = await LimitNotifyModel.findOne({ user_id });
     if (!hasNotifyEntry) {
-        await LimitNotifyModel.create({ project_id, limit1: false, limit2: false, limit3: false })
+        await LimitNotifyModel.create({ user_id, limit1: false, limit2: false, limit3: false })
     }
 
+    const owner = await UserModel.findById(project.owner);
+    if (!owner) return;
+
+    const userName = owner.given_name || owner.name || 'no_name';
+
     if ((projectCounts.visits + projectCounts.events) >= (projectCounts.limit)) {
 
         if (hasNotifyEntry.limit3 === true) return;
 
-        const project = await ProjectModel.findById(project_id);
-        if (!project) return;
-
-        const owner = await UserModel.findById(project.owner);
-        if (!owner) return;
-
         setImmediate(() => {
-            const emailData = EmailService.getEmailServerInfo('limit_max', {
-                target: owner.email,
-                projectName: project.name
-            });
+            const emailData = EmailService.getEmailServerInfo('limit_max', { target: owner.email, userName });
             EmailServiceHelper.sendEmail(emailData);
         });
 
-        await LimitNotifyModel.updateOne({ project_id: projectCounts.project_id }, { limit1: true, limit2: true, limit3: true });
+        await LimitNotifyModel.updateOne({ user_id }, { limit1: true, limit2: true, limit3: true });
 
     } else if ((projectCounts.visits + projectCounts.events) >= (projectCounts.limit * 0.9)) {
 
         if (hasNotifyEntry.limit2 === true) return;
 
-        const project = await ProjectModel.findById(project_id);
-        if (!project) return;
-
-        const owner = await UserModel.findById(project.owner);
-        if (!owner) return;
-
         setImmediate(() => {
-            const emailData = EmailService.getEmailServerInfo('limit_90', {
-                target: owner.email,
-                projectName: project.name
-            });
+            const emailData = EmailService.getEmailServerInfo('limit_90', { target: owner.email, userName });
             EmailServiceHelper.sendEmail(emailData);
         });
 
-        await LimitNotifyModel.updateOne({ project_id: projectCounts.project_id }, { limit1: true, limit2: true, limit3: false });
+        await LimitNotifyModel.updateOne({ user_id }, { limit1: true, limit2: true, limit3: false });
 
     } else if ((projectCounts.visits + projectCounts.events) >= (projectCounts.limit * 0.5)) {
 
         if (hasNotifyEntry.limit1 === true) return;
 
-        const project = await ProjectModel.findById(project_id);
-        if (!project) return;
-
-        const owner = await UserModel.findById(project.owner);
-        if (!owner) return;
-
         setImmediate(() => {
-            const emailData = EmailService.getEmailServerInfo('limit_50', {
-                target: owner.email,
-                projectName: project.name
-            });
+            const emailData = EmailService.getEmailServerInfo('limit_50', { target: owner.email, userName });
             EmailServiceHelper.sendEmail(emailData);
         });
-        
-        await LimitNotifyModel.updateOne({ project_id: projectCounts.project_id }, { limit1: true, limit2: false, limit3: false });
+
+        await LimitNotifyModel.updateOne({ user_id }, { limit1: true, limit2: false, limit3: false });
 
     }
 

consumer/src/LimitChecker.ts

@@ -1,15 +1,15 @@
 
 
-import { ProjectLimitModel } from './shared/schema/project/ProjectsLimits';
 import { MAX_LOG_LIMIT_PERCENT } from './shared/data/broker/Limits';
 import { checkLimitsForEmail } from './EmailController';
+import { UserLimitModel } from './shared/schema/UserLimitSchema';
 
-export async function checkLimits(project_id: string) {
-    const projectLimits = await ProjectLimitModel.findOne({ project_id });
-    if (!projectLimits) return false;
-    const TOTAL_COUNT = projectLimits.events + projectLimits.visits;
-    const COUNT_LIMIT = projectLimits.limit;
+export async function checkLimits(user_id: string) {
+    const userLimits = await UserLimitModel.findOne({ user_id });
+    if (!userLimits) return false;
+    const TOTAL_COUNT = userLimits.events + userLimits.visits;
+    const COUNT_LIMIT = userLimits.limit;
     if ((TOTAL_COUNT) > COUNT_LIMIT * MAX_LOG_LIMIT_PERCENT) return false;
-    await checkLimitsForEmail(projectLimits);
+    await checkLimitsForEmail(userLimits);
     return true;
 }

consumer/src/index.ts

@@ -11,10 +11,9 @@ import { UAParser } from 'ua-parser-js';
 import { checkLimits } from './LimitChecker';
 import express from 'express';
 
-import { ProjectLimitModel } from './shared/schema/project/ProjectsLimits';
 import { ProjectCountModel } from './shared/schema/project/ProjectsCounts';
 import { metricsRouter } from './Metrics';
-
+import { UserLimitModel } from './shared/schema/UserLimitSchema';
 
 const app = express();
 
@@ -27,6 +26,12 @@ main();
 
 const CONSUMER_NAME = `CONSUMER_${process.env.NODE_APP_INSTANCE || 'DEFAULT'}`
 
+
+async function getProjectOwner(pid: string) {
+    const ownerData = await ProjectModel.findOne({ _id: pid }, { owner: 1 });
+    return ownerData.owner;
+}
+
 async function main() {
 
     await RedisStreamService.connect();
@@ -51,18 +56,18 @@ async function processStreamEntry(data: Record<string, string>) {
 
         const { pid, sessionHash } = data;
 
-        const project = await ProjectModel.exists({ _id: pid });
-        if (!project) return;
+        const owner = await getProjectOwner(pid);
+        if (!owner) return;
 
-        const canLog = await checkLimits(pid);
+        const canLog = await checkLimits(owner.toString());
         if (!canLog) return;
 
         if (eventType === 'event') {
-            await process_event(data, sessionHash);
+            await process_event(data, sessionHash, owner.toString());
         } else if (eventType === 'keep_alive') {
-            await process_keep_alive(data, sessionHash);
+            await process_keep_alive(data, sessionHash, owner.toString());
         } else if (eventType === 'visit') {
-            await process_visit(data, sessionHash);
+            await process_visit(data, sessionHash, owner.toString());
         }
 
     } catch (ex: any) {
@@ -75,7 +80,7 @@ async function processStreamEntry(data: Record<string, string>) {
 
 }
 
-async function process_visit(data: Record<string, string>, sessionHash: string) {
+async function process_visit(data: Record<string, string>, sessionHash: string, user_id: string) {
 
     const { pid, ip, website, page, referrer, userAgent, flowHash, timestamp } = data;
 
@@ -105,12 +110,12 @@ async function process_visit(data: Record<string, string>, sessionHash: string)
             created_at: new Date(parseInt(timestamp))
         }),
         ProjectCountModel.updateOne({ project_id: pid }, { $inc: { 'visits': 1 } }, { upsert: true }),
-        ProjectLimitModel.updateOne({ project_id: pid }, { $inc: { 'visits': 1 } })
+        UserLimitModel.updateOne({ user_id }, { $inc: { 'visits': 1 } })
     ]);
 
 }
 
-async function process_keep_alive(data: Record<string, string>, sessionHash: string) {
+async function process_keep_alive(data: Record<string, string>, sessionHash: string, user_id: string) {
 
     const { pid, instant, flowHash, timestamp, website } = data;
 
@@ -137,7 +142,7 @@ async function process_keep_alive(data: Record<string, string>, sessionHash: str
 
 }
 
-async function process_event(data: Record<string, string>, sessionHash: string) {
+async function process_event(data: Record<string, string>, sessionHash: string, user_id: string) {
 
     const { name, metadata, pid, flowHash, timestamp, website } = data;
 
@@ -155,7 +160,7 @@ async function process_event(data: Record<string, string>, sessionHash: string)
             created_at: new Date(parseInt(timestamp))
         }),
         ProjectCountModel.updateOne({ project_id: pid }, { $inc: { 'events': 1 } }, { upsert: true }),
-        ProjectLimitModel.updateOne({ project_id: pid }, { $inc: { 'events': 1 } })
+        UserLimitModel.updateOne({ user_id }, { $inc: { 'events': 1 } })
     ]);
 
 

payments/src/controllers/WebhookController.ts

@@ -1,7 +1,7 @@
 
 import type Event from 'stripe';
 import StripeService from '../services/StripeService';
-import { getPlanFromPrice, PLAN_DATA } from '../shared/data/PLANS';
+import { getPlanFromPrice, getPlanFromTag, PLAN_DATA } from '../shared/data/PLANS';
 import { PremiumModel } from '../shared/schema/PremiumSchema';
 import { UserLimitModel } from '../shared/schema/UserLimitSchema';
 
@@ -29,6 +29,27 @@ async function addSubscriptionToUser(user_id: string, plan: PLAN_DATA, subscript
 
 }
 
+export async function onPaymentFailed(event: Event.InvoicePaymentFailedEvent) {
+
+    
+    if (event.data.object.attempt_count == 0) return { received: true, warn: 'attempt_count = 0' }
+    
+    //TODO: Send emails
+
+    const customer_id = event.data.object.customer as string;
+    const premiumData = await PremiumModel.findOne({ customer_id });
+    if (!premiumData) return { error: 'customer not found' }
+
+    const subscription_id = event.data.object.subscription as string;
+    await StripeService.deleteSubscription(subscription_id);
+
+    const freeSub = await StripeService.createFreeSubscription(customer_id);
+    await PremiumModel.updateOne({ customer_id }, { subscription_id: freeSub.id });
+
+    await addSubscriptionToUser(premiumData.user_id.toString(), getPlanFromTag('FREE'), subscription_id, event.data.object.period_start, event.data.object.period_end);
+
+    return { ok: true }
+}
 
 export async function onPaymentSuccess(event: Event.InvoicePaidEvent) {
 

payments/src/index.ts

@@ -2,13 +2,16 @@ import express from 'express';
 import StripeService from './services/StripeService'
 import { webhookRouter } from './routers/WebhookRouter';
 import { paymentRouter } from './routers/PaymentRouter';
+import { connectDatabase } from './shared/services/DatabaseService';
 
 
 const STRIPE_PRIVATE_KEY = process.env.STRIPE_PRIVATE_KEY;
 const STRIPE_WEBHOOK_SECRET = process.env.STRIPE_WEBHOOK_SECRET;
 const STRIPE_TESTMODE = process.env.STRIPE_TESTMODE === 'true';
+const MONGO_CONNECTION_STRING = process.env.MONGO_CONNECTION_STRING;
 
 StripeService.init(STRIPE_PRIVATE_KEY, STRIPE_WEBHOOK_SECRET, STRIPE_TESTMODE);
+connectDatabase(MONGO_CONNECTION_STRING);
 
 console.log('Stripe started in', STRIPE_TESTMODE ? 'TESTMODE' : 'LIVEMODE');
 

payments/src/routers/WebhookRouter.ts

@@ -21,12 +21,12 @@ webhookRouter.get('/', json(), async (req, res) => {
             const response = await WebhookController.onPaymentSuccess(eventData);
             return sendJson(res, 200, response);
         }
+
+        if (eventData.type === 'invoice.payment_failed') {
+            const response = await WebhookController.onPaymentFailed(eventData);
+            return sendJson(res, 200, response);
+        }
         
-        // if (eventData.type === 'payment_intent.succeeded') return await onPaymentOnetimeSuccess(eventData);
-        // if (eventData.type === 'invoice.payment_failed') return await onPaymentFailed(eventData);
-        // if (eventData.type === 'customer.subscription.deleted') return await onSubscriptionDeleted(eventData);
-        // if (eventData.type === 'customer.subscription.created') return await onSubscriptionCreated(eventData);
-        // if (eventData.type === 'customer.subscription.updated') return await onSubscriptionUpdated(eventData);
 
     } catch (ex) {
         res.status(500).json({ error: ex.message });

payments/src/services/StripeService.ts

@@ -1,5 +1,6 @@
 
 import Stripe from "stripe";
+import { getPlanFromTag } from "../shared/data/PLANS";
 
 
 
@@ -186,22 +187,21 @@ class StripeService {
     //     return subscription;
     // }
 
-    // async createFreeSubscription(customer_id: string) {
-    //     if (this.disabledMode) return;
-    //     if (!this.stripe) throw Error('Stripe not initialized');
+    async createFreeSubscription(customer_id: string) {
+        if (!this.stripe) throw Error('Stripe not initialized');
 
-    //     const FREE_PLAN = getPlanFromTag('FREE');
+        const FREE_PLAN = getPlanFromTag('FREE');
 
-    //     const subscription = await this.stripe.subscriptions.create({
-    //         customer: customer_id,
-    //         items: [
-    //             { price: this.testMode ? FREE_PLAN.PRICE_TEST : FREE_PLAN.PRICE, quantity: 1 }
-    //         ]
-    //     });
+        const subscription = await this.stripe.subscriptions.create({
+            customer: customer_id,
+            items: [
+                { price: this.testMode ? FREE_PLAN.PRICE_TEST : FREE_PLAN.PRICE, quantity: 1 }
+            ]
+        });
 
-    //     return subscription;
+        return subscription;
 
-    // }
+    }
 
 }
 

scripts/consumer/shared.ts

@@ -16,13 +16,13 @@ helper.copy('services/EmailService.ts');
 
 helper.create('schema');
 helper.copy('schema/UserSchema.ts');
+helper.copy('schema/UserLimitSchema.ts');
 
 helper.create('schema/broker');
 helper.copy('schema/broker/LimitNotifySchema.ts');
 
 helper.create('schema/project');
 helper.copy('schema/project/ProjectSchema.ts');
-helper.copy('schema/project/ProjectsLimits.ts');
 helper.copy('schema/project/ProjectsCounts.ts');
 
 helper.create('schema/metrics');

shared_global/schema/broker/LimitNotifySchema.ts

@@ -2,14 +2,14 @@ import { model, Schema, Types } from 'mongoose';
 
 export type TLimitNotify = {
     _id: Schema.Types.ObjectId,
-    project_id: Schema.Types.ObjectId,
+    user_id: Schema.Types.ObjectId,
     limit1: boolean,
     limit2: boolean,
     limit3: boolean
 }
 
 const LimitNotifySchema = new Schema<TLimitNotify>({
-    project_id: { type: Types.ObjectId, index: 1 },
+    user_id: { type: Types.ObjectId, index: 1 },
     limit1: { type: Boolean },
     limit2: { type: Boolean },
     limit3: { type: Boolean }

shared_global/schema/project/ProjectsLimits.ts

@@ -1,26 +0,0 @@
-import { model, Schema, Types } from 'mongoose';
-
-export type TProjectLimit = {
-    _id: Schema.Types.ObjectId,
-    project_id: Schema.Types.ObjectId,
-    events: number,
-    visits: number,
-    ai_messages: number,
-    limit: number,
-    ai_limit: number,
-    billing_expire_at: Date,
-    billing_start_at: Date,
-}
-
-const ProjectLimitSchema = new Schema<TProjectLimit>({
-    project_id: { type: Types.ObjectId, index: true, unique: true },
-    events: { type: Number, required: true, default: 0 },
-    visits: { type: Number, required: true, default: 0 },
-    ai_messages: { type: Number, required: true, default: 0 },
-    limit: { type: Number, required: true },
-    ai_limit: { type: Number, required: true },
-    billing_start_at: { type: Date, required: true },
-    billing_expire_at: { type: Date, required: true },
-});
-
-export const ProjectLimitModel = model<TProjectLimit>('project_limits', ProjectLimitSchema);