added sum, reset cron

frontend/src/views/features/metered-features/FeatureConfig.tsx

@@ -10,7 +10,7 @@ import { slugify } from "@/utils/formatUtils/formatTextUtils";
 import { cn } from "@/lib/utils";
 import { useHotkeys } from "react-hotkeys-hook";
 import { XIcon } from "lucide-react";
-import { Expression, MeteredConfig } from "@autumn/shared";
+import { AggregateType, Expression, MeteredConfig } from "@autumn/shared";
 import { FeatureType } from "@autumn/shared";
 import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs";
 import { useFeaturesContext } from "../FeaturesContext";
@@ -162,10 +162,24 @@ export function FeatureConfig({
                 <SelectValue placeholder="Select" />
               </SelectTrigger>
               <SelectContent>
-                <SelectItem value="count">COUNT</SelectItem>
+                {Object.values(AggregateType).map((type) => (
+                  <SelectItem key={type} value={type}>
+                    {type.toUpperCase()}
+                  </SelectItem>
+                ))}
               </SelectContent>
             </Select>
           </div>
+          {meteredConfig.aggregate.type == AggregateType.Sum && (
+            <div>
+              <FieldLabel>Property</FieldLabel>
+              <Input
+                placeholder="Property"
+                value={meteredConfig.aggregate.property || ""}
+                onChange={(e) => setAggregate("property", e.target.value)}
+              />
+            </div>
+          )}
         </>
       )}
     </div>

server/package.json

@@ -5,7 +5,7 @@
   "main": "index.js",
   "type": "module",
   "scripts": {
-    "dev": "nodemon --exec tsx src/index.ts --ignore 'src/scripts/*' --ignore 'workflow-dependencies/*'",
+    "dev": "nodemon --no-deprecation --exec tsx src/index.ts",
     "start": "tsx src/index.ts",
     "queue:dev": "tsx watch src/queue.ts",
     "build": "tsc -b",
@@ -13,22 +13,16 @@
     "test": "tsx test.ts",
     "prod:build": "tsc -b && tsc-alias",
     "prod:start": "node dist/src/index.js",
+
     "cron": "tsx src/cron.ts",
-    "clean": "tsx src/clean.ts",
-    "invoiceCron": "tsx src/invoiceCron.ts",
-    "trigger": "npx trigger.dev@latest dev",
-    "trigger:deploy": "npx trigger.dev@latest deploy",
-    "inngest:dev": "npx inngest-cli@latest dev -u http://localhost:8080/api/inngest --poll-interval 5"
+    "cron:start": "tsc -b && tsx src/cron.ts",
+    "clean": "tsx src/clean.ts"
   },
   "author": "",
   "license": "ISC",
   "dependencies": {
-    "@anthropic-ai/sdk": "^0.32.1",
     "@autumn/shared": "*",
-    "@aws-sdk/client-secrets-manager": "^3.726.0",
     "@clerk/express": "^1.3.22",
-    "@clickhouse/client": "^1.10.0",
-    "@infisical/sdk": "^3.0.4",
     "@slack/events-api": "^3.0.1",
     "@slack/web-api": "^7.8.0",
     "@supabase/supabase-js": "^2.46.2",
@@ -59,7 +53,6 @@
     "tsx": "^4.19.2"
   },
   "devDependencies": {
-    "@trigger.dev/build": "^3.3.11",
     "@types/pg": "^8.11.10"
   }
 }

server/scripts/reset.nnb

@@ -0,0 +1,11 @@
+{
+    "cells": [
+        {
+            "language": "typescript",
+            "source": [
+                "import dotenv from \"dotenv\";\nimport axios from \"axios\";\ndotenv.config();\n\nconst apiKey = \"am_test_3ZQ1KsigC9j4bc5u5uxzBabV\"\nconst url = \"http://localhost:8080/v1\"\n\n// Send event\nconst data = {\n\tcustomer_id: \"123\",\n\tevent_name: \"minutes\",\n\tproperties: {\n\t\tlength: 19.5\n\t},\n}\nconst res = await axios.post(`${url}/events`, data,\n\t{\n\t\theaders: {\n\t\t\t\"Content-Type\": \"application/json\",\n\t\t\t\"Authorization\": `Bearer ${apiKey}`\n\t\t}\n\t}\n)"
+            ],
+            "outputs": []
+        }
+    ]
+}

server/scripts/test_setup.nnb

@@ -0,0 +1,48 @@
+{
+    "cells": [
+        {
+            "language": "typescript",
+            "source": [
+                "// SET UP\nimport axios from \"axios\";\nimport stripe from \"stripe\";\nimport { createClient } from \"@supabase/supabase-js\";\nimport dotenv from \"dotenv\";\ndotenv.config({ path: \"../.env.prod\" })\n\nlet API_KEY = \"am_test_3ZjayPZBAT9n7vgruTcstYvk\"\nconst BASE_URL = \"https://api.useautumn.com/v1\"\nconst orgId = \"org_2rzkkRh7r5dBSaBC101QHG9KDgt\"\nconst env = \"sandbox\"\n\nconst proProduct = \"prod_2s2wfNCD1qctYUbq9rjdRSKLklf\"\nconst rechargeProduct = \"prod_2s2wlOpHEkbD7uGo62IR1LvJImc\"\nconst stripeCli = new stripe(process.env.STRIPE_TEST_KEY!)\n\nconst sb = createClient(\n\tprocess.env.SUPABASE_URL!,\n\tprocess.env.SUPABASE_SERVICE_KEY!\n);\n\nlet headers = {\n\t\"Content-Type\": \"application/json\",\n\t\"Authorization\": `Bearer ${API_KEY}`\n}\n\nconst attachPmToCus = async (cusId: string) => {\n\tconst pm = await stripeCli.paymentMethods.create({\n\t\ttype: \"card\",\n\t\tcard: {\n\t\t\ttoken: \"tok_visa\", // This is a special test token that represents a valid card\n\t\t},\n\t});\n\tconst attachRes = await stripeCli.paymentMethods.attach(pm.id, {\n\t\tcustomer: cusId,\n\t});\n}\n\nconst attachProductToCus = async (cusId: string, productId: string, options: any) => {\n\tconst { data } = await axios.post(`${BASE_URL}/attach`, {\n\t\tcustomer_id: cusId,\n\t\tproduct_id: productId,\n\t\toptions: options\n\t}, { headers })\n}\n\nconst createCustomer = async (index: number) => {\n\ttry {\n\n\t\t// Create customer\n\t\tconst { data } = await axios.post(`${BASE_URL}/customers`, {\n\t\t\tid: `user_${index}`,\n\t\t\tname: `User ${index}`,\n\t\t\temail: `user${index}@example.com`\n\t\t}, { headers })\n\n\t\tconst { customer } = data\n\t\t\n\t\t// Attach payment method\n\t\tawait attachPmToCus(customer.processor.id)\n\n\t\tawait attachProductToCus(customer.id, proProduct, [])\n\t\tif (index > NUM_USERS / 2) {\n\t\t\tawait attachProductToCus(customer.id, rechargeProduct, [{\n\t\t\t\tfeature_id: FEATURE_ID,\n\t\t\t\tthreshold: 100\n\t\t\t}])\n\t\t}\n\n\t\tconsole.log(`Customer ${customer.id} created`)\n\n\t} catch (error: any) {\n\t\tconsole.log(\"Error creating customer\", error.response.data)\n\t\t// Customer already exists\n\t}\n\n}\n\nconst deleteCustomer = async (index: number) => {\n\t// 1. Delete all customer events\n\tconst { data: eventsDeleted, error: eventsError } = await sb.from(\"events\").delete().eq(\"customer_id\", `user_${index}`).eq(\"org_id\", orgId).eq(\"env\", env)\n\tif (eventsError) {\n\t\tconsole.log(\"Error deleting events\", eventsError)\n\t}\n\ttry {\n\t\tconst { data } = await axios.delete(`${BASE_URL}/customers/user_${index}`, { headers })\n\t} catch (error: any) {\n\t\tif (error.response.data && error.response.data.code !== \"customer_not_found\") {\n\t\t\tconsole.log(\"Error deleting customer\", error.response.data)\n\t\t}\n\t}\n}\n\n// SET UP\nconst setup = async () => {\n\n\tfor (let i = 0; i < Math.ceil(NUM_USERS / BATCH_SIZE); i++) {\n\t\tconst createCusCalls = []\n\t\tfor (let j = 0; j < BATCH_SIZE; j++) {\n\t\t\tcreateCusCalls.push(createCustomer(i * BATCH_SIZE + j))\n\t\t\tif (i * BATCH_SIZE + j >= NUM_USERS) {\n\t\t\t\tbreak\n\t\t\t}\n\t\t}\n\t\tawait Promise.all(createCusCalls)\n\t\tawait new Promise(resolve => setTimeout(resolve, 1000))\n\t}\n}\n\n// TEARDOWN\nconst teardown = async (numUsers: number) => {\n\tfor (let i = 0; i < Math.ceil(numUsers / DELETE_BATCH_SIZE); i++) {\n\t\tconst deleteCusCalls = []\n\t\tfor (let j = 0; j < DELETE_BATCH_SIZE; j++) {\n\t\t\tdeleteCusCalls.push(deleteCustomer(i * DELETE_BATCH_SIZE + j))\n\t\t\tif (i * DELETE_BATCH_SIZE + j >= numUsers) {\n\t\t\t\tbreak\n\t\t\t}\n\t\t}\n\t\tawait Promise.all(deleteCusCalls)\n\t\tawait new Promise(resolve => setTimeout(resolve, 1000))\n\t}\n}\n\nconst FEATURE_ID = \"credits\"\nconst BATCH_SIZE = 5\nconst DELETE_BATCH_SIZE = 20\nconst res1 = await teardown(200)\nconst NUM_USERS = 200;\n// const res2 = await setup()"
+            ],
+            "outputs": []
+        },
+        {
+            "language": "typescript",
+            "source": [
+                "// Create 1000 customers\nimport dotenv from \"dotenv\";\nimport { createClient } from \"@supabase/supabase-js\";\ndotenv.config({ path: \"../.env\" })\n\nconst orgId = \"org_2s2tZflaVSKGVyFyclgXOVsTZc4\"\nconst env = \"sandbox\"\nconst sb = createClient(\n\tprocess.env.SUPABASE_URL!,\n\tprocess.env.SUPABASE_SERVICE_KEY!\n);\n\nlet customers = []\nfor (let i = 0; i < 1000; i++) {\n\tcustomers.push({\n\t\tid: `user_${i}`,\n\t\tname: `User ${i}`,\n\t\temail: `user${i}@example.com`,\n\t\torg_id: orgId,\n\t\tenv: env,\n\t\tinternal_id: `user_${i}`,\n\t\tcreated_at: Date.now(),\n\t})\n}\n\nconst res4 = await sb.from(\"customers\").insert(customers)\nconsole.log(res4)"
+            ],
+            "outputs": [
+                {
+                    "items": [
+                        {
+                            "mime": "application/vnd.code.notebook.stderr",
+                            "value": [
+                                "(node:70632) [DEP0040] DeprecationWarning: The `punycode` module is deprecated. Please use a userland alternative instead.",
+                                "(Use `node --trace-deprecation ...` to show where the warning was created)",
+                                ""
+                            ]
+                        }
+                    ]
+                },
+                {
+                    "items": [
+                        {
+                            "mime": "application/vnd.code.notebook.stdout",
+                            "value": [
+                                "{",
+                                "  error: null,",
+                                "  data: null,",
+                                "  count: null,",
+                                "  status: 201,",
+                                "  statusText: 'Created'",
+                                "}",
+                                ""
+                            ]
+                        }
+                    ]
+                }
+            ]
+        }
+    ]
+}

server/src/cron.ts

@@ -1,78 +1,109 @@
 import {
-  BillingInterval,
-  CusEntWithEntitlement,
-  Duration,
+  CusProductSchema,
   EntInterval,
+  FullCustomerEntitlement,
+  FullCustomerEntitlementSchema,
 } from "@autumn/shared";
 import { CustomerEntitlementService } from "./internal/customers/entitlements/CusEntitlementService.js";
 import { createSupabaseClient } from "./external/supabaseUtils.js";
-import { getNextEntitlementReset } from "./utils/timeUtils.js";
+import { SupabaseClient } from "@supabase/supabase-js";
 import dotenv from "dotenv";
-import { format } from "date-fns";
-import pg from "pg";
-
+import { getEntOptions } from "./internal/prices/priceUtils.js";
+import { getNextResetAt } from "./utils/timeUtils.js";
+import chalk from "chalk";
+import { z } from "zod";
 dotenv.config();
 
-export const cronTask = async () => {
-  console.log("\n-----------------------------------\n");
-  console.log("Running cron job");
-  // 1. Query customer_entitlements for all customers with reset_interval < now
-  const sbClient = createSupabaseClient();
+const FullCustomerEntitlementWithProduct = FullCustomerEntitlementSchema.extend(
+  {
+    customer_product: CusProductSchema,
+  }
+);
 
-  let cusEntitlements: CusEntWithEntitlement[] = [];
+type FullCustomerEntitlementWithProduct = z.infer<
+  typeof FullCustomerEntitlementWithProduct
+>;
+
+const resetCustomerEntitlement = async ({
+  sb,
+  cusEnt,
+}: {
+  sb: SupabaseClient;
+  cusEnt: FullCustomerEntitlementWithProduct;
+}) => {
   try {
-    cusEntitlements = await CustomerEntitlementService.getEntitlementsForReset(
-      sbClient
+    console.log("----------------------------------");
+    console.log(`Resetting cusEnt ${cusEnt.id}`);
+    console.log(
+      `Customer: ${chalk.yellowBright(
+        cusEnt.customer_id
+      )}, Feature: ${chalk.yellowBright(cusEnt.entitlement.feature_id)}`
     );
+
+    // 1. Get allowance and quantity
+    const allowance = cusEnt.entitlement.allowance || 0;
+
+    // 2. Quantity is from prices...
+    const entOptions = getEntOptions(
+      cusEnt.customer_product.options,
+      cusEnt.entitlement
+    );
+
+    let quantity = (entOptions && entOptions.quantity) || 1;
+    const newBalance = allowance * quantity;
+
+    console.log(
+      `Allowance: ${chalk.yellow(allowance)} | Quantity: ${chalk.yellow(
+        quantity
+      )} | New Balance: ${chalk.yellow(newBalance)}`
+    );
+
+    // 3. Update the next_reset_at for each entitlement
+    const nextResetAt = getNextResetAt(
+      new Date(cusEnt.next_reset_at!),
+      cusEnt.entitlement.interval as EntInterval
+    );
+
+    await CustomerEntitlementService.update({
+      sb,
+      id: cusEnt.id,
+      updates: {
+        next_reset_at: nextResetAt,
+        balance: newBalance,
+      },
+    });
+
+    console.log(`Successfull`);
+  } catch (error: any) {
+    console.log("Error:", error.message || error);
+  }
+};
+
+export const cronTask = async () => {
+  console.log("RUNNING RESET CRON");
+  // 1. Query customer_entitlements for all customers with reset_interval < now
+  const sb = createSupabaseClient();
+  let cusEntitlements: FullCustomerEntitlement[] = [];
+  try {
+    cusEntitlements = await CustomerEntitlementService.getActiveResetPassed({
+      sb,
+    });
+
+    let resets = [];
+    for (const cusEnt of cusEntitlements) {
+      resets.push(
+        await resetCustomerEntitlement({
+          sb,
+          cusEnt: cusEnt as FullCustomerEntitlementWithProduct,
+        })
+      );
+    }
+
+    await Promise.all(resets);
   } catch (error) {
     console.error("Error getting entitlements for reset:", error);
     return;
   }
-
-  const pgClient = new pg.Client(process.env.SUPABASE_CONNECTION_STRING || "");
-  await pgClient.connect();
-
-  // 2. Update the next_reset_at for each entitlement
-  let updateStatements = ``;
-  for (const cusEnt of cusEntitlements) {
-    if (!cusEnt.next_reset_at) {
-      continue;
-    }
-
-    const nextResetAt = getNextEntitlementReset(
-      new Date(cusEnt.next_reset_at),
-      cusEnt.entitlement.interval as EntInterval
-    ).getTime();
-
-    // TODO: Find price with config->entitlement_id = cusEnt.entitlement_id & get options
-    const resetBalance = cusEnt.entitlement.allowance || 0;
-
-    console.log(
-      "Current reset at:",
-      format(cusEnt.next_reset_at, "yyyy MMM dd HH:mm:ss")
-    );
-
-    console.log("Next reset at:", format(nextResetAt, "yyyy MMM dd HH:mm:ss"));
-
-    updateStatements += `UPDATE customer_entitlements SET
-        next_reset_at = ${nextResetAt},
-        balance = ${resetBalance}
-        WHERE id = '${cusEnt.id}';\n`;
-  }
-
-  console.log(`Resetting ${cusEntitlements.length} entitlements`);
-
-  try {
-    const result = await pgClient.query(updateStatements);
-  } catch (error) {
-    console.error("Error updating entitlements:", error);
-  }
-
-  await pgClient.end();
-  console.log("Finished cron job");
 };
 
-// cronTask();
-
-// Run cron job every 60 seconds
-setInterval(cronTask, 60000);
+cronTask();

server/src/external/awsSecrets.ts

@@ -1,50 +0,0 @@
-import {
-  SecretsManagerClient,
-  PutSecretValueCommand,
-  GetSecretValueCommand,
-  CreateSecretCommand,
-} from "@aws-sdk/client-secrets-manager";
-
-export async function createSecret(secretId: string, secretString: string) {
-  const client = new SecretsManagerClient();
-
-  try {
-    const res = await client.send(
-      new CreateSecretCommand({
-        Name: secretId,
-        SecretString: secretString,
-      })
-    );
-    return res.ARN;
-  } catch (error) {
-    console.error("Error creating secret:", error);
-    return null;
-  }
-}
-
-export async function updateSecret(secretId: string, secretString: string) {
-  const client = new SecretsManagerClient();
-
-  try {
-    const res = await client.send(
-      new PutSecretValueCommand({
-        SecretId: secretId,
-        SecretString: secretString,
-      })
-    );
-    return res.ARN;
-  } catch (error) {
-    console.error("Error updating secret:", error);
-    return null;
-  }
-}
-
-export async function getSecret(secretId: string) {
-  const client = new SecretsManagerClient();
-
-  const response = await client.send(
-    new GetSecretValueCommand({ SecretId: secretId })
-  );
-
-  return response.SecretString;
-}

server/src/external/clickhouse/clickhouseUtils.ts

@@ -1,17 +0,0 @@
-import { createClient } from "@clickhouse/client";
-
-export const createClickhouseCli = () =>
-  createClient({
-    url: process.env.CLICKHOUSE_URL,
-    username: process.env.CLICKHOUSE_USERNAME,
-    password: process.env.CLICKHOUSE_PASSWORD,
-  });
-
-const test = async () => {
-  const clickhouseClient = createClickhouseCli();
-  const rows = await clickhouseClient.query({
-    query: "SELECT 1",
-    format: "JSONEachRow",
-  });
-  console.log("Result: ", await rows.json());
-};

server/src/index.ts

@@ -11,8 +11,6 @@ import { envMiddleware } from "./middleware/envMiddleware.js";
 import pg from "pg";
 
 import { serve } from "inngest/express";
-import { functions } from "./trigger/inngest.js";
-import { inngest } from "./trigger/inngest.js";
 import { initQueue, initWorkers } from "./queue/queue.js";
 
 const app = express();
@@ -57,7 +55,6 @@ const init = async () => {
   app.use("/webhooks", webhooksRouter);
 
   app.use(express.json());
-  app.use("/api/inngest", serve({ client: inngest, functions }));
 
   app.use(mainRouter);
   app.use("/v1", apiRouter);

server/src/internal/api/events/eventRouter.ts

@@ -16,32 +16,12 @@ import { generateId } from "@/utils/genUtils.js";
 
 import { EventService } from "./EventService.js";
 import { StatusCodes } from "http-status-codes";
-import { CustomerEntitlementService } from "../../customers/entitlements/CusEntitlementService.js";
 import { CusService } from "@/internal/customers/CusService.js";
-import { getBelowThresholdPrice } from "../../../trigger/invoiceThresholdUtils.js";
-import { updateBalanceTask } from "@/trigger/updateBalanceTask.js";
 import { Client } from "pg";
-import { inngest } from "@/trigger/inngest.js";
 import { Queue } from "bullmq";
 
 export const eventsRouter = Router();
 
-const getCreditSystemDeduction = (
-  meteredFeatures: Feature[],
-  creditSystem: Feature
-) => {
-  let creditsUpdate = 0;
-  let meteredFeatureIds = meteredFeatures.map((feature) => feature.id);
-
-  for (const schema of creditSystem.config.schema) {
-    if (meteredFeatureIds.includes(schema.metered_feature_id)) {
-      creditsUpdate += (1 / schema.feature_amount) * schema.credit_amount;
-    }
-  }
-
-  return creditsUpdate;
-};
-
 const getEventAndCustomer = async (req: any) => {
   const body = req.body;
   const orgId = req.orgId;
@@ -89,50 +69,6 @@ const getEventAndCustomer = async (req: any) => {
   return { customer, event: newEvent };
 };
 
-const getFeaturesAndCustomerEnts = async ({
-  req,
-  customer,
-  event,
-}: {
-  req: any;
-  customer: Customer;
-  event: Event;
-}) => {
-  const { rows }: { rows: Feature[] } = await req.pg.query(`
-    with features_with_event as (
-      select * from features
-      where config -> 'filters' @> '[{"value": ["${event.event_name}"]}]'::jsonb
-    )
-    
-    select * from features WHERE EXISTS (
-      SELECT 1 FROM jsonb_array_elements(config->'schema') as schema_element WHERE
-      schema_element->>'metered_feature_id' IN (SELECT id FROM features_with_event)
-    )
-    UNION all
-    select * from features_with_event
-    `);
-
-  if (rows.length === 0) {
-    return { customerEntitlements: [], features: [] };
-  }
-
-  let internalFeatureIds = rows.map((feature) => feature.internal_id);
-  const cusEnts = await CustomerEntitlementService.getActiveInFeatureIds({
-    sb: req.sb,
-    internalCustomerId: customer.internal_id,
-    internalFeatureIds: internalFeatureIds as string[],
-  });
-
-  cusEnts.sort((a, b) => {
-    if (a.balance <= 0) return 1;
-    if (b.balance <= 0) return -1;
-
-    return a.created_at - b.created_at;
-  });
-
-  return { customerEntitlements: cusEnts, features: rows };
-};
-
 const getAffectedFeatures = async ({
   pg,
   event,
@@ -179,6 +115,13 @@ eventsRouter.post("", async (req: any, res: any) => {
     });
 
     if (affectedFeatures.length > 0) {
+      let queue: Queue = req.queue;
+      queue.add("update-balance", {
+        customerId: customer.internal_id,
+        customer,
+        features: affectedFeatures,
+        event,
+      });
       // await inngest.send({
       //   name: "autumn/update-balance",
       //   data: {
@@ -199,24 +142,6 @@ eventsRouter.post("", async (req: any, res: any) => {
       //     concurrencyKey: customer.internal_id,
       //   }
       // );
-      let queue: Queue = req.queue;
-      queue.add(
-        "update-balance",
-        {
-          customerId: customer.internal_id,
-          customer,
-          features: affectedFeatures,
-        }
-        // {
-        //   attempts: 10,
-        //   backoff: {
-        //     type: "exponential",
-        //     delay: 1000,
-        //   },
-        // }
-      );
-
-      // console.log("Queued update balance task...");
     } else {
       console.log("No affected features found");
     }

server/src/internal/api/features/featureApiRouter.ts

@@ -3,7 +3,7 @@ import RecaseError, {
   formatZodError,
   handleRequestError,
 } from "@/utils/errorUtils.js";
-import { Feature } from "@autumn/shared";
+import { AggregateType, Feature, FeatureType } from "@autumn/shared";
 import { CreateFeatureSchema } from "@autumn/shared";
 import express from "express";
 import { generateId } from "@/utils/genUtils.js";
@@ -11,20 +11,41 @@ import { ErrCode } from "@/errors/errCodes.js";
 
 export const featureApiRouter = express.Router();
 
+export const validateFeature = (data: any) => {
+  let featureType = data.type;
+  if (featureType == FeatureType.Metered) {
+    // 1. Check if property is provided
+
+    let config = data.config;
+    if (
+      config.aggregate.type == AggregateType.Sum &&
+      !config.aggregate.property
+    ) {
+      throw new RecaseError({
+        message: `Property is required for sum aggregate`,
+        code: ErrCode.InvalidFeature,
+        statusCode: 400,
+      });
+    }
+  }
+
+  try {
+    CreateFeatureSchema.parse(data);
+  } catch (error: any) {
+    throw new RecaseError({
+      message: `Invalid feature: ${formatZodError(error)}`,
+      code: ErrCode.InvalidFeature,
+      statusCode: 400,
+    });
+  }
+};
+
 featureApiRouter.post("", async (req: any, res) => {
   let org = req.org;
   let data = req.body;
 
   try {
-    try {
-      CreateFeatureSchema.parse(data);
-    } catch (error: any) {
-      throw new RecaseError({
-        message: `Invalid feature: ${formatZodError(error)}`,
-        code: ErrCode.InvalidFeature,
-        statusCode: 400,
-      });
-    }
+    validateFeature(data);
 
     let feature: Feature = {
       internal_id: generateId("fe"),

server/src/internal/customers/entitlements/CusEntitlementService.ts

@@ -144,6 +144,22 @@ export class CustomerEntitlementService {
     return data;
   }
 
+  static async getActiveResetPassed({ sb }: { sb: SupabaseClient }) {
+    const { data, error } = await sb
+      .from("customer_entitlements")
+      .select(
+        "*, customer_product:customer_products!inner(*), entitlement:entitlements(*)"
+      )
+      .eq("customer_product.status", "active")
+      .lt("next_reset_at", Date.now());
+
+    if (error) {
+      throw error;
+    }
+
+    return data;
+  }
+
   static async update({
     sb,
     id,

server/src/queue/queue.ts

@@ -45,13 +45,7 @@ const initWorker = (id: number, queue: Queue) => {
   let worker = new Worker(
     "autumn",
     async (job: Job) => {
-      // if (job.name === "update-balance") {
-      //   await runUpdateBalanceTask(job.data);
-      // }
-
-      const { customerId, customer } = job.data;
-      // await runUpdateBalanceTask(job.data);
-      // return;
+      const { customerId } = job.data;
 
       while (!(await acquireLock(customerId, 10000))) {
         // console.log(`Customer ${customer.id} locked by another worker`);

server/src/trigger/inngest.ts

@@ -1,34 +0,0 @@
-import { createSupabaseClient } from "@/external/supabaseUtils.js";
-import { Inngest } from "inngest";
-import { updateCustomerBalance } from "./updateBalanceUtils.js";
-
-// Create a client to send and receive events
-export const inngest = new Inngest({ id: "autumn" });
-
-const updateBalanceTask = inngest.createFunction(
-  { id: "update-balance" },
-  { event: "autumn/update-balance" },
-
-  async ({ event, step, logger }) => {
-    console.log("Beginning inngest update balance task...");
-
-    try {
-      const sb = createSupabaseClient();
-
-      // 1. Update customer balance
-      const { customer, features } = event.data;
-      logger.info("Updating customer balance...");
-      const cusEnts: any = await updateCustomerBalance({
-        sb,
-        customer,
-        features,
-      });
-    } catch (error) {
-      logger.error("Inngest update balance task failed...");
-      logger.error(error);
-    }
-  }
-);
-
-// Create an empty array where we'll export future Inngest functions
-export const functions = [updateBalanceTask];

server/src/trigger/updateBalanceTask.ts

@@ -1,57 +1,197 @@
-import { logger, task } from "@trigger.dev/sdk/v3";
 import { createSupabaseClient } from "@/external/supabaseUtils.js";
-import { updateCustomerBalance } from "./updateBalanceUtils.js";
 import { handleBelowThresholdInvoicing } from "./invoiceThresholdUtils.js";
 import { getBelowThresholdPrice } from "./invoiceThresholdUtils.js";
 
-export const updateBalanceTask = task({
-  id: "update-customer-balance",
+import { AggregateType, Event, Feature } from "@autumn/shared";
+import { CustomerEntitlementService } from "@/internal/customers/entitlements/CusEntitlementService.js";
+import { Customer, FeatureType } from "@autumn/shared";
+import { SupabaseClient } from "@supabase/supabase-js";
 
-  maxDuration: 300, // Stop executing after 300 secs (5 mins) of compute
+// 3. Get customer entitlements and sort
+const getCustomerEntitlements = async ({
+  sb,
+  internalCustomerId,
+  features,
+}: {
+  sb: SupabaseClient;
+  internalCustomerId: string;
+  features: Feature[];
+}) => {
+  const internalFeatureIds = features.map((feature) => feature.internal_id);
+  const cusEnts = await CustomerEntitlementService.getActiveInFeatureIds({
+    sb,
+    internalCustomerId,
+    internalFeatureIds: internalFeatureIds as string[],
+  });
 
-  run: async (payload: any, { ctx }) => {
-    try {
-      const sb = createSupabaseClient();
+  cusEnts.sort((a, b) => {
+    if (a.balance <= 0) return 1;
+    if (b.balance <= 0) return -1;
 
-      // 1. Update customer balance
-      const { customer, features } = payload;
+    return a.created_at - b.created_at;
+  });
 
-      logger.log("Updating customer balance...");
-      const cusEnts: any = await updateCustomerBalance({
-        sb,
-        customer,
-        features,
-      });
+  return cusEnts;
+};
 
-      // 2. Check if there's below threshold price
-      const belowThresholdPrice = await getBelowThresholdPrice({
-        sb,
-        internalCustomerId: customer.internal_id,
-        cusEnts,
-      });
+// 2. Functions to get deduction per feature
+export const getMeteredDeduction = (meteredFeature: Feature, event: Event) => {
+  let config = meteredFeature.config;
+  let aggregate = config.aggregate;
 
-      if (belowThresholdPrice) {
-        console.log("--------------------------------");
-        console.log("Below threshold price found");
-        await handleBelowThresholdInvoicing({
-          sb,
-          internalCustomerId: payload.internalCustomerId,
-          belowThresholdPrice,
-        });
-      }
-    } catch (error) {
-      console.log(`Error updating customer balance: ${error}`);
-      console.log(error);
+  if (aggregate.type == AggregateType.Count) {
+    return 1;
+  }
+
+  if (aggregate.type == AggregateType.Sum) {
+    let property = aggregate.property;
+    let value = event.properties[property] || 0;
+
+    let floatVal = parseFloat(value);
+    if (isNaN(floatVal)) {
+      return 0;
     }
-  },
-});
+
+    return floatVal;
+  }
+
+  return 0;
+};
+
+const getCreditSystemDeduction = ({
+  meteredFeatures,
+  creditSystem,
+  event,
+}: {
+  meteredFeatures: Feature[];
+  creditSystem: Feature;
+  event: Event;
+}) => {
+  let creditsUpdate = 0;
+  let meteredFeatureIds = meteredFeatures.map((feature) => feature.id);
+
+  for (const schema of creditSystem.config.schema) {
+    if (meteredFeatureIds.includes(schema.metered_feature_id)) {
+      let meteredFeature = meteredFeatures.find(
+        (feature) => feature.id === schema.metered_feature_id
+      );
+
+      if (!meteredFeature) {
+        continue;
+      }
+
+      let meteredDeduction = getMeteredDeduction(meteredFeature, event);
+      creditsUpdate +=
+        (meteredDeduction / schema.feature_amount) * schema.credit_amount;
+    }
+  }
+
+  return creditsUpdate;
+};
+
+// 1. Main function to update customer balance
+export const updateCustomerBalance = async ({
+  sb,
+  customer,
+  event,
+  features,
+}: {
+  sb: SupabaseClient;
+  customer: Customer;
+  event: Event;
+  features: Feature[];
+}) => {
+  const cusEnts = await getCustomerEntitlements({
+    sb,
+    internalCustomerId: customer.internal_id,
+    features,
+  });
+
+  if (cusEnts.length === 0 || features.length === 0) {
+    return;
+  }
+
+  // Update customer balance
+  const featureIdToDeduction: any = {};
+  const meteredFeatures = features.filter(
+    (feature) => feature.type === FeatureType.Metered
+  );
+
+  console.log(`   - Customer: ${customer.name} (${customer.internal_id})`);
+  console.log(`   - Features: ${features.map((f) => f.id).join(", ")}`);
+
+  for (const cusEnt of cusEnts) {
+    const internalFeatureId = cusEnt.internal_feature_id;
+    if (featureIdToDeduction[internalFeatureId]) {
+      continue;
+    }
+
+    const feature = features.find(
+      (feature) => feature.internal_id === internalFeatureId
+    );
+
+    // 1. Get metered feature deduction
+    if (feature?.type === FeatureType.Metered) {
+      let deduction = getMeteredDeduction(feature, event);
+
+      featureIdToDeduction[internalFeatureId] = {
+        cusEntId: cusEnt.id,
+        deduction,
+        feature: feature,
+      };
+    }
+
+    // 2. Get credit system deduction
+    if (feature?.type === FeatureType.CreditSystem) {
+      const deduction = getCreditSystemDeduction({
+        meteredFeatures,
+        creditSystem: feature,
+        event,
+      });
+
+      if (deduction) {
+        featureIdToDeduction[internalFeatureId] = {
+          cusEntId: cusEnt.id,
+          deduction: deduction,
+          feature: feature,
+        };
+      }
+    }
+
+    let deduction = featureIdToDeduction[internalFeatureId]?.deduction;
+    let curBalance = cusEnt.balance!;
+
+    if (curBalance === undefined || curBalance === null) {
+      continue;
+    }
+
+    // 3. Update customer balance
+    const { error } = await sb
+      .from("customer_entitlements")
+      .update({ balance: curBalance - deduction })
+      .eq("id", cusEnt.id);
+
+    if (error) {
+      console.error(
+        `   ❌ Failed to update (${feature?.id}: ${deduction}). Error: ${error}`
+      );
+    }
+  }
+
+  let featuresUpdated = Object.values(featureIdToDeduction).map(
+    (obj: any) => `(${obj.feature.id}: ${obj.deduction})`
+  );
+
+  console.log(`   - Deducted ${featuresUpdated}`);
+  return cusEnts;
+};
 
 export const runUpdateBalanceTask = async (payload: any) => {
   try {
     const sb = createSupabaseClient();
 
     // 1. Update customer balance
-    const { customer, features } = payload;
+    const { customer, features, event } = payload;
 
     console.log("--------------------------------");
     console.log("Inside updateBalanceTask...");
@@ -61,6 +201,7 @@ export const runUpdateBalanceTask = async (payload: any) => {
       sb,
       customer,
       features,
+      event,
     });
 
     if (!cusEnts || cusEnts.length === 0) {

server/src/trigger/updateBalanceUtils.ts

@@ -1,130 +0,0 @@
-import { Feature } from "@autumn/shared";
-import { CustomerEntitlementService } from "@/internal/customers/entitlements/CusEntitlementService.js";
-import { Customer, FeatureType } from "@autumn/shared";
-import { SupabaseClient } from "@supabase/supabase-js";
-
-const getCustomerEntitlements = async ({
-  sb,
-  internalCustomerId,
-  features,
-}: {
-  sb: SupabaseClient;
-  internalCustomerId: string;
-  features: Feature[];
-}) => {
-  const internalFeatureIds = features.map((feature) => feature.internal_id);
-  const cusEnts = await CustomerEntitlementService.getActiveInFeatureIds({
-    sb,
-    internalCustomerId,
-    internalFeatureIds: internalFeatureIds as string[],
-  });
-
-  cusEnts.sort((a, b) => {
-    if (a.balance <= 0) return 1;
-    if (b.balance <= 0) return -1;
-
-    return a.created_at - b.created_at;
-  });
-
-  return cusEnts;
-};
-
-const getCreditSystemDeduction = (
-  meteredFeatures: Feature[],
-  creditSystem: Feature
-) => {
-  let creditsUpdate = 0;
-  let meteredFeatureIds = meteredFeatures.map((feature) => feature.id);
-
-  for (const schema of creditSystem.config.schema) {
-    if (meteredFeatureIds.includes(schema.metered_feature_id)) {
-      creditsUpdate += (1 / schema.feature_amount) * schema.credit_amount;
-    }
-  }
-
-  return creditsUpdate;
-};
-
-export const updateCustomerBalance = async ({
-  sb,
-  customer,
-  features,
-}: {
-  sb: SupabaseClient;
-  customer: Customer;
-  features: Feature[];
-}) => {
-  const cusEnts = await getCustomerEntitlements({
-    sb,
-    internalCustomerId: customer.internal_id,
-    features,
-  });
-
-  if (cusEnts.length === 0 || features.length === 0) {
-    return;
-  }
-
-  // Update customer balance
-  const featureIdToDeduction: any = {};
-  const meteredFeatures = features.filter(
-    (feature) => feature.type === FeatureType.Metered
-  );
-
-  console.log(`   - Customer: ${customer.name} (${customer.internal_id})`);
-  console.log(`   - Features: ${features.map((f) => f.id).join(", ")}`);
-
-  for (const cusEnt of cusEnts) {
-    const internalFeatureId = cusEnt.internal_feature_id;
-    if (featureIdToDeduction[internalFeatureId]) {
-      continue;
-    }
-
-    const feature = features.find(
-      (feature) => feature.internal_id === internalFeatureId
-    );
-
-    if (feature?.type === FeatureType.Metered) {
-      featureIdToDeduction[internalFeatureId] = {
-        cusEntId: cusEnt.id,
-        deduction: 1,
-        feature: feature,
-      };
-    }
-
-    if (feature?.type === FeatureType.CreditSystem) {
-      const deduction = getCreditSystemDeduction(meteredFeatures, feature);
-      if (deduction) {
-        featureIdToDeduction[internalFeatureId] = {
-          cusEntId: cusEnt.id,
-          deduction: deduction,
-          feature: feature,
-        };
-      }
-    }
-
-    let deduction = featureIdToDeduction[internalFeatureId]?.deduction;
-    let curBalance = cusEnt.balance!;
-
-    if (curBalance === undefined || curBalance === null) {
-      continue;
-    }
-
-    const { error } = await sb
-      .from("customer_entitlements")
-      .update({ balance: curBalance - deduction })
-      .eq("id", cusEnt.id);
-
-    if (error) {
-      console.error(
-        `   ❌ Failed to update (${feature?.id}: ${deduction}). Error: ${error}`
-      );
-    }
-  }
-
-  let featuresUpdated = Object.values(featureIdToDeduction).map(
-    (obj: any) => `(${obj.feature.id}: ${obj.deduction})`
-  );
-
-  console.log(`   - Deducted ${featuresUpdated}`);
-  return cusEnts;
-};

server/src/utils/timeUtils.ts

@@ -29,3 +29,16 @@ export const getNextEntitlementReset = (
       throw new Error("Invalid duration");
   }
 };
+
+export const getNextResetAt = (
+  curReset: Date | null,
+  interval: EntInterval
+) => {
+  while (true) {
+    const nextReset = getNextEntitlementReset(curReset, interval);
+    if (nextReset.getTime() > Date.now()) {
+      return nextReset.getTime();
+    }
+    curReset = nextReset;
+  }
+};

shared/models/cusModels/cusEntModels/cusEntitlementModels.ts

@@ -3,7 +3,6 @@ import {
   EntitlementSchema,
   EntitlementWithFeatureSchema,
 } from "../../productModels/entitlementModels.js";
-import { BillingInterval } from "../../productModels/fixedPriceModels.js";
 import { FeatureSchema } from "../../featureModels/featureModels.js";
 
 export const CustomerEntitlementSchema = z.object({

shared/models/eventModels/eventModels.ts

@@ -11,7 +11,7 @@ export const EventSchema = z.object({
   event_name: z.string().nonempty(),
 
   // Optional
-  properties: z.record(z.string(), z.any()).optional(),
+  properties: z.record(z.string(), z.any()),
   idempotency_key: z.string().optional(),
   timestamp: z.number().optional(),
 });

shared/models/featureModels/featureModels.ts

@@ -7,6 +7,11 @@ export enum FeatureType {
   CreditSystem = "credit_system",
 }
 
+export enum AggregateType {
+  Count = "count",
+  Sum = "sum",
+}
+
 export const FeatureSchema = z.object({
   internal_id: z.string().optional(),
   org_id: z.string().optional(),