feat: WIP stripe billing refactor update for 2025

apps/api/src/api-v1/consumers/create-consumer.ts

@@ -0,0 +1,57 @@
+import { createRoute, type OpenAPIHono } from '@hono/zod-openapi'
+
+import type { AuthenticatedEnv } from '@/lib/types'
+import { schema } from '@/db'
+import { upsertConsumer } from '@/lib/billing/upsert-consumer'
+import {
+  openapiAuthenticatedSecuritySchemas,
+  openapiErrorResponse404,
+  openapiErrorResponse409,
+  openapiErrorResponse410,
+  openapiErrorResponses
+} from '@/lib/openapi-utils'
+import { parseZodSchema } from '@/lib/utils'
+
+const route = createRoute({
+  description: 'Creates a new consumer by subscribing a customer to a project.',
+  tags: ['consumers'],
+  operationId: 'createConsumer',
+  method: 'post',
+  path: 'consumers',
+  security: openapiAuthenticatedSecuritySchemas,
+  request: {
+    body: {
+      required: true,
+      content: {
+        'application/json': {
+          schema: schema.consumerInsertSchema
+        }
+      }
+    }
+  },
+  responses: {
+    200: {
+      description: 'A consumer object',
+      content: {
+        'application/json': {
+          schema: schema.consumerSelectSchema
+        }
+      }
+    },
+    ...openapiErrorResponses,
+    ...openapiErrorResponse404,
+    ...openapiErrorResponse409,
+    ...openapiErrorResponse410
+  }
+})
+
+export function registerV1ConsumersCreateConsumer(
+  app: OpenAPIHono<AuthenticatedEnv>
+) {
+  return app.openapi(route, async (c) => {
+    const body = c.req.valid('json')
+    const consumer = await upsertConsumer(c, body)
+
+    return c.json(parseZodSchema(schema.consumerSelectSchema, consumer))
+  })
+}

apps/api/src/api-v1/consumers/get-consumer.ts

@@ -17,7 +17,7 @@ const route = createRoute({
   tags: ['consumers'],
   operationId: 'getConsumer',
   method: 'get',
-  path: 'consumers/{consumersId}',
+  path: 'consumers/{consumerId}',
   security: openapiAuthenticatedSecuritySchemas,
   request: {
     params: consumerIdParamsSchema,

apps/api/src/api-v1/consumers/list-consumers.ts

@@ -14,7 +14,7 @@ import { projectIdParamsSchema } from '../projects/schemas'
 import { paginationAndPopulateConsumerSchema } from './schemas'
 
 const route = createRoute({
-  description: 'Lists consumers (customers) for a project.',
+  description: 'Lists all of the customers for a project.',
   tags: ['consumers'],
   operationId: 'listConsumers',
   method: 'get',

apps/api/src/api-v1/consumers/update-consumer.ts

@@ -0,0 +1,65 @@
+import { createRoute, type OpenAPIHono } from '@hono/zod-openapi'
+
+import type { AuthenticatedEnv } from '@/lib/types'
+import { schema } from '@/db'
+import { upsertConsumer } from '@/lib/billing/upsert-consumer'
+import {
+  openapiAuthenticatedSecuritySchemas,
+  openapiErrorResponse404,
+  openapiErrorResponse409,
+  openapiErrorResponse410,
+  openapiErrorResponses
+} from '@/lib/openapi-utils'
+import { parseZodSchema } from '@/lib/utils'
+
+import { consumerIdParamsSchema } from './schemas'
+
+const route = createRoute({
+  description: "Updates a consumer's subscription to a project.",
+  tags: ['consumers'],
+  operationId: 'updateConsumer',
+  method: 'post',
+  path: 'consumers/{consumerId}',
+  security: openapiAuthenticatedSecuritySchemas,
+  request: {
+    params: consumerIdParamsSchema,
+    body: {
+      required: true,
+      content: {
+        'application/json': {
+          schema: schema.consumerInsertSchema
+        }
+      }
+    }
+  },
+  responses: {
+    200: {
+      description: 'A consumer object',
+      content: {
+        'application/json': {
+          schema: schema.consumerSelectSchema
+        }
+      }
+    },
+    ...openapiErrorResponses,
+    ...openapiErrorResponse404,
+    ...openapiErrorResponse409,
+    ...openapiErrorResponse410
+  }
+})
+
+export function registerV1ConsumersUpdateConsumer(
+  app: OpenAPIHono<AuthenticatedEnv>
+) {
+  return app.openapi(route, async (c) => {
+    const { consumerId } = c.req.valid('param')
+    const body = c.req.valid('json')
+
+    const consumer = await upsertConsumer(c, {
+      ...body,
+      consumerId
+    })
+
+    return c.json(parseZodSchema(schema.consumerSelectSchema, consumer))
+  })
+}

apps/api/src/api-v1/consumers/upsert-consumer.ts

@@ -1,152 +0,0 @@
-import { parseFaasIdentifier } from '@agentic/validators'
-import { createRoute, type OpenAPIHono } from '@hono/zod-openapi'
-
-import type { AuthenticatedEnv } from '@/lib/types'
-import { and, db, eq, schema } from '@/db'
-import { upsertStripeConnectCustomer } from '@/lib/billing/upsert-stripe-connect-customer'
-import { upsertStripeCustomer } from '@/lib/billing/upsert-stripe-customer'
-import { upsertStripePricing } from '@/lib/billing/upsert-stripe-pricing'
-import { upsertStripeSubscription } from '@/lib/billing/upsert-stripe-subscription'
-import {
-  openapiAuthenticatedSecuritySchemas,
-  openapiErrorResponse404,
-  openapiErrorResponse409,
-  openapiErrorResponse410,
-  openapiErrorResponses
-} from '@/lib/openapi-utils'
-import { assert, parseZodSchema, sha256 } from '@/lib/utils'
-
-const route = createRoute({
-  description:
-    'Upserts a consumer (customer), subscribing to a specific deployment within project.',
-  tags: ['consumers'],
-  operationId: 'createConsumer',
-  method: 'post',
-  path: 'consumers',
-  security: openapiAuthenticatedSecuritySchemas,
-  request: {
-    body: {
-      required: true,
-      content: {
-        'application/json': {
-          schema: schema.consumerInsertSchema
-        }
-      }
-    }
-  },
-  responses: {
-    200: {
-      description: 'A consumer object',
-      content: {
-        'application/json': {
-          schema: schema.consumerSelectSchema
-        }
-      }
-    },
-    ...openapiErrorResponses,
-    ...openapiErrorResponse404,
-    ...openapiErrorResponse409,
-    ...openapiErrorResponse410
-  }
-})
-
-export function registerV1ConsumersUpsertConsumer(
-  app: OpenAPIHono<AuthenticatedEnv>
-) {
-  return app.openapi(route, async (c) => {
-    const body = c.req.valid('json')
-    const userId = c.get('userId')
-
-    const parsedIds = parseFaasIdentifier(body.deploymentId)
-    assert(parsedIds, 400, 'Invalid "deploymentId"')
-    const { projectId } = parsedIds
-
-    const [{ user, stripeCustomer }, existing] = await Promise.all([
-      upsertStripeCustomer(c),
-
-      db.query.consumers.findFirst({
-        where: and(
-          eq(schema.consumers.userId, userId),
-          eq(schema.consumers.projectId, projectId)
-        )
-      })
-    ])
-
-    assert(
-      !existing ||
-        !existing.enabled ||
-        existing.plan !== body.plan ||
-        existing.deploymentId !== body.deploymentId,
-      409,
-      `User "${user.email}" already has an active subscription to plan "${body.plan}" for project "${projectId}"`
-    )
-
-    const deployment = await db.query.deployments.findFirst({
-      where: eq(schema.deployments.id, body.deploymentId),
-      with: {
-        project: true
-      }
-    })
-    assert(deployment, 404, `Deployment not found "${body.deploymentId}"`)
-
-    const { project } = deployment
-    assert(
-      project,
-      404,
-      `Project not found "${projectId}" for deployment "${body.deploymentId}"`
-    )
-    assert(
-      deployment.enabled,
-      410,
-      `Deployment has been disabled by its owner "${deployment.id}"`
-    )
-
-    let consumer = existing
-
-    if (consumer) {
-      consumer.plan = body.plan
-      consumer.deploymentId = body.deploymentId
-      ;[consumer] = await db
-        .update(schema.consumers)
-        .set(consumer)
-        .where(eq(schema.consumers.id, consumer.id))
-        .returning()
-    } else {
-      ;[consumer] = await db.insert(schema.consumers).values({
-        ...body,
-        userId,
-        projectId,
-        token: sha256().slice(0, 24),
-        _stripeCustomerId: stripeCustomer.id
-      })
-    }
-
-    assert(consumer, 500, 'Error creating consumer')
-
-    // Ensure that all Stripe pricing resources exist for this deployment
-    await upsertStripePricing({ deployment, project })
-
-    // Ensure that customer and default source are created on the stripe connect account
-    // TODO: is this necessary?
-    // consumer._stripeAccount = project._stripeAccount
-    await upsertStripeConnectCustomer({ stripeCustomer, consumer, project })
-
-    const logger = c.get('logger')
-    logger.info('SUBSCRIPTION', existing ? 'UPDATE' : 'CREATE', {
-      project,
-      deployment,
-      consumer
-    })
-
-    const { subscription, consumer: updatedConsumer } =
-      await upsertStripeSubscription(c, {
-        consumer,
-        user,
-        project,
-        deployment
-      })
-    logger.info('subscription', subscription)
-
-    return c.json(parseZodSchema(schema.consumerSelectSchema, updatedConsumer))
-  })
-}

apps/api/src/api-v1/index.test.ts

@@ -0,0 +1,27 @@
+import type { hc } from 'hono/client'
+import { expectTypeOf, test } from 'vitest'
+
+import type { User } from '@/db'
+
+import type { ApiRoutes } from './index'
+
+type ApiClient = ReturnType<typeof hc<ApiRoutes>>
+
+type GetUserResponse = Awaited<
+  ReturnType<Awaited<ReturnType<ApiClient['users'][':userId']['$get']>>['json']>
+>
+
+test('User types are compatible', async () => {
+  expectTypeOf<GetUserResponse>().toEqualTypeOf<User>()
+
+  // const client = hc<ApiRoutes>('http://localhost:3000/v1')
+
+  // const user = await client.users[':userId'].$post({
+  //   param: {
+  //     userId: '123'
+  //   },
+  //   json: {
+  //     firstName: 'John'
+  //   }
+  // })
+})

apps/api/src/api-v1/index.ts

@@ -5,7 +5,10 @@ import type { AuthenticatedEnv } from '@/lib/types'
 import * as middleware from '@/lib/middleware'
 
 import { registerV1AdminConsumersGetConsumerByToken } from './consumers/admin-get-consumer-by-token'
+import { registerV1ConsumersCreateConsumer } from './consumers/create-consumer'
 import { registerV1ConsumersGetConsumer } from './consumers/get-consumer'
+import { registerV1ProjectsListConsumers } from './consumers/list-consumers'
+import { registerV1ConsumersUpdateConsumer } from './consumers/update-consumer'
 import { registerHealthCheck } from './health-check'
 import { registerV1ProjectsCreateProject } from './projects/create-project'
 import { registerV1ProjectsGetProject } from './projects/get-project'
@@ -50,23 +53,23 @@ const privateRouter = new OpenAPIHono<AuthenticatedEnv>()
 
 registerHealthCheck(publicRouter)
 
-// Users crud
+// Users
 registerV1UsersGetUser(privateRouter)
 registerV1UsersUpdateUser(privateRouter)
 
-// Teams crud
+// Teams
 registerV1TeamsCreateTeam(privateRouter)
 registerV1TeamsListTeams(privateRouter)
 registerV1TeamsGetTeam(privateRouter)
 registerV1TeamsDeleteTeam(privateRouter)
 registerV1TeamsUpdateTeam(privateRouter)
 
-// Team members crud
+// Team members
 registerV1TeamsMembersCreateTeamMember(privateRouter)
 registerV1TeamsMembersUpdateTeamMember(privateRouter)
 registerV1TeamsMembersDeleteTeamMember(privateRouter)
 
-// Projects crud
+// Projects
 registerV1ProjectsCreateProject(privateRouter)
 registerV1ProjectsListProjects(privateRouter)
 registerV1ProjectsGetProject(privateRouter)
@@ -83,8 +86,11 @@ registerV1ProjectsUpdateProject(privateRouter)
 //   require('./projects').read
 // )
 
-// Consumers crud
+// Consumers
 registerV1ConsumersGetConsumer(privateRouter)
+registerV1ConsumersCreateConsumer(privateRouter)
+registerV1ConsumersUpdateConsumer(privateRouter)
+registerV1ProjectsListConsumers(privateRouter)
 
 // Webhook event handlers
 registerV1StripeWebhook(publicRouter)
@@ -123,3 +129,6 @@ export type ApiRoutes =
   | ReturnType<typeof registerV1ProjectsUpdateProject>
   // Consumers
   | ReturnType<typeof registerV1ConsumersGetConsumer>
+  | ReturnType<typeof registerV1ConsumersCreateConsumer>
+  | ReturnType<typeof registerV1ConsumersUpdateConsumer>
+  | ReturnType<typeof registerV1ProjectsListConsumers>

apps/api/src/db/schema/consumer.ts

@@ -1,3 +1,4 @@
+import { validators } from '@agentic/validators'
 import { relations } from '@fisch0920/drizzle-orm'
 import {
   boolean,
@@ -14,6 +15,7 @@ import { users, userSelectSchema } from './user'
 import {
   createInsertSchema,
   createSelectSchema,
+  createUpdateSchema,
   cuid,
   deploymentId,
   id,
@@ -45,7 +47,8 @@ export const consumers = pgTable(
     // API token for this consumer
     token: text().notNull(),
 
-    // The stripe subscription plan this consumer is subscribed to (or 'free' if supported)
+    // The slug of the PricingPlan in the target deployment that this consumer
+    // is subscribed to.
     plan: text(),
 
     // Whether the consumer has made at least one successful API call after
@@ -55,8 +58,8 @@ export const consumers = pgTable(
     // Whether the consumer's subscription is currently active
     enabled: boolean().default(true).notNull(),
 
-    env: text().default('dev').notNull(),
+    // TODO: Re-add coupon support
-    coupon: text(),
+    // coupon: text(),
 
     // only used during initial creation
     source: text(),
@@ -72,23 +75,22 @@ export const consumers = pgTable(
         onDelete: 'cascade'
       }),
 
-    // The specific deployment this user is subscribed to
+    // The specific deployment this user is subscribed to, since pricing can
-    // (since pricing can change across deployment versions)
+    // change across deployment versions)
     deploymentId: deploymentId()
       .notNull()
       .references(() => deployments.id, {
         onDelete: 'cascade'
       }),
 
-    // stripe subscription status (synced via webhooks)
+    // Stripe subscription status (synced via webhooks)
     stripeStatus: text(),
 
+    // Main Stripe Subscription id
     stripeSubscriptionId: stripeId(),
-    stripeSubscriptionBaseItemId: stripeId(),
-    stripeSubscriptionRequestItemId: stripeId(),
 
-    // [metricSlug: string]: string
+    // [lineItemSlug: string]: string
-    stripeSubscriptionMetricItems: jsonb()
+    stripeSubscriptionLineItemIdMap: jsonb()
       .$type<Record<string, string>>()
       .default({})
       .notNull(),
@@ -99,7 +101,6 @@ export const consumers = pgTable(
   },
   (table) => [
     index('consumer_token_idx').on(table.token),
-    index('consumer_env_idx').on(table.env),
     index('consumer_userId_idx').on(table.userId),
     index('consumer_projectId_idx').on(table.projectId),
     index('consumer_deploymentId_idx').on(table.deploymentId),
@@ -131,7 +132,17 @@ export const consumerRelationsSchema: z.ZodType<ConsumerRelationFields> =
   z.enum(['user', 'project', 'deployment'])
 
 export const consumerSelectSchema = createSelectSchema(consumers, {
-  stripeSubscriptionMetricItems: z.record(z.string(), z.string())
+  stripeSubscriptionLineItemIdMap: z.record(z.string(), z.string()),
+
+  deploymentId: (schema) =>
+    schema.refine((id) => validators.deploymentId(id), {
+      message: 'Invalid deployment id'
+    }),
+
+  projectId: (schema) =>
+    schema.refine((id) => validators.projectId(id), {
+      message: 'Invalid project id'
+    })
 })
   .omit({
     _stripeCustomerId: true
@@ -155,12 +166,31 @@ export const consumerSelectSchema = createSelectSchema(consumers, {
   .strip()
   .openapi('Consumer')
 
-export const consumerInsertSchema = createInsertSchema(consumers)
+export const consumerInsertSchema = createInsertSchema(consumers, {
+  deploymentId: (schema) =>
+    schema.refine((id) => validators.deploymentId(id), {
+      message: 'Invalid deployment id'
+    }),
+
+  plan: z.string().nonempty()
+})
   .pick({
     plan: true,
-    env: true,
-    coupon: true,
     source: true,
     deploymentId: true
   })
   .strict()
+
+export const consumerUpdateSchema = createUpdateSchema(consumers, {
+  deploymentId: (schema) =>
+    schema
+      .refine((id) => validators.deploymentId(id), {
+        message: 'Invalid deployment id'
+      })
+      .optional()
+})
+  .pick({
+    plan: true,
+    deploymentId: true
+  })
+  .strict()

apps/api/src/db/schema/deployment.ts

@@ -129,7 +129,7 @@ export const deploymentSelectSchema = createSelectSchema(deployments, {
 
 export const deploymentInsertSchema = createInsertSchema(deployments, {
   id: (schema) =>
-    schema.refine((id) => validators.project(id), {
+    schema.refine((id) => validators.deploymentId(id), {
       message: 'Invalid deployment id'
     }),
 
@@ -138,6 +138,11 @@ export const deploymentInsertSchema = createInsertSchema(deployments, {
       message: 'Invalid deployment hash'
     }),
 
+  projectId: (schema) =>
+    schema.refine((id) => validators.projectId(id), {
+      message: 'Invalid project id'
+    }),
+
   _url: (schema) => schema.url(),
 
   // TODO: should this public resource be decoupled from the internal pricing

apps/api/src/db/schema/project.ts

@@ -217,7 +217,7 @@ export const projectSelectSchema = createSelectSchema(projects, {
 
 export const projectInsertSchema = createInsertSchema(projects, {
   id: (schema) =>
-    schema.refine((id) => validators.project(id), {
+    schema.refine((id) => validators.projectId(id), {
       message: 'Invalid project id'
     }),
 

apps/api/src/db/schema/types.ts

@@ -113,10 +113,10 @@ export type StripeMeterIdMap = z.infer<typeof stripeMeterIdMapSchema>
 
 const commonPricingPlanLineItemSchema = z.object({
   /**
-   * Slugs act as the primary key for metrics. They should be lower and
+   * Slugs act as the primary key for LineItems. They should be lower and
    * kebab-cased ("base", "requests", "image-transformations").
    *
-   * TODO: ensure user-provided custom metrics don't use reserved 'base'
+   * TODO: ensure user-provided custom LineItems don't use reserved 'base'
    * and 'requests' slugs.
    */
   slug: z.union([z.string(), z.literal('base'), z.literal('requests')]),
@@ -152,7 +152,7 @@ export const pricingPlanLineItemSchema = z
         unitLabel: z.string().optional(),
 
         /**
-         * Optional rate limit to enforce for this metric.
+         * Optional rate limit to enforce for this metered LineItem.
          *
          * You can use this, for example, to limit the number of API calls that
          * can be made during a given interval.
@@ -223,16 +223,16 @@ export const pricingPlanLineItemSchema = z
   ])
   .refine((data) => {
     assert(
-      !(data.slug === 'base' && data.usageType !== 'licensed'),
+      data.slug !== 'base' || data.usageType === 'licensed',
       `Invalid pricing plan metric "${data.slug}": "base" pricing plan metrics are reserved for "licensed" usage type.`
     )
 
     assert(
-      !(data.slug === 'requests' && data.usageType !== 'metered'),
+      data.slug !== 'requests' || data.usageType === 'metered',
       `Invalid pricing plan metric "${data.slug}": "requests" pricing plan metrics are reserved for "metered" usage type.`
     )
 
-    return data
+    return true
   })
   .describe(
     'PricingPlanLineItems represent a single line-item in a Stripe Subscription. They map to a Stripe billing `Price` and possibly a corresponding Stripe `Meter` for metered usage.'
@@ -242,7 +242,7 @@ export type PricingPlanLineItem = z.infer<typeof pricingPlanLineItemSchema>
 
 /**
  * Represents the config for a Stripe subscription with one or more
- * PricingPlanLineItems as line-items.
+ * PricingPlanLineItems.
  */
 export const pricingPlanSchema = z
   .object({
@@ -260,26 +260,29 @@ export const pricingPlanSchema = z
     // TODO?
     trialPeriodDays: z.number().nonnegative().optional(),
 
-    metricsMap: z
+    lineItems: z.array(pricingPlanLineItemSchema).nonempty().max(20, {
-      .record(pricingPlanLineItemSlugSchema, pricingPlanLineItemSchema)
+      message:
-      .refine((metricsMap) => {
+        'Stripe Checkout currently supports a max of 20 line-items per subscription.'
-        // Stripe Checkout currently supports a max of 20 line items per
+    })
-        // subscription.
-        return Object.keys(metricsMap).length <= 20
-      })
-      .default({})
   })
   .refine((data) => {
-    if (data.interval === undefined && data.slug !== 'free') {
+    assert(
-      throw new Error(
+      data.interval !== undefined || data.slug === 'free',
-        `Invalid PricingPlan "${data.slug}": non-free pricing plans must have an interval`
+      `Invalid PricingPlan "${data.slug}": non-free pricing plans must have an interval`
-      )
+    )
-    }
 
-    return data
+    const lineItemSlugs = new Set(
+      data.lineItems.map((lineItem) => lineItem.slug)
+    )
+    assert(
+      lineItemSlugs.size === data.lineItems.length,
+      `Invalid PricingPlan "${data.slug}": duplicate line-item slugs`
+    )
+
+    return true
   })
   .describe(
-    'Represents the config for a Stripe subscription with one or more PricingPlanLineItems as line-items.'
+    'Represents the config for a Stripe subscription with one or more PricingPlanLineItems.'
   )
   .openapi('PricingPlan')
 export type PricingPlan = z.infer<typeof pricingPlanSchema>
@@ -294,13 +297,27 @@ export const stripeProductIdMapSchema = z
 export type StripeProductIdMap = z.infer<typeof stripeProductIdMapSchema>
 
 export const pricingPlanMapSchema = z
-  .record(z.string().describe('PricingPlan slug'), pricingPlanSchema)
+  .record(
+    z
+      .string()
+      .describe(
+        'PricingPlan slug ("free", "starter-monthly", "pro-annual", etc)'
+      ),
+    pricingPlanSchema
+  )
   .refine((data) => Object.keys(data).length > 0, {
     message: 'Must contain at least one PricingPlan'
   })
   .describe('Map from PricingPlan slug to PricingPlan')
 export type PricingPlanMap = z.infer<typeof pricingPlanMapSchema>
 
+// TODO
+// export const stripeSubscriptionLineItemIdMapSchema = z
+//   .record(pricingPlanLineItemHashSchema, z.string().describe('Stripe LineItem id'))
+//   .describe('Map from internal PricingPlanLineItem **hash** to Stripe LineItem id')
+//   .openapi('StripeSubscriptionLineItemMap')
+// export type StripeSubscriptionLineItemMap = z.infer<typeof stripeSubscriptionLineItemMapSchema>
+
 // export const couponSchema = z
 //   .object({
 //     // used to uniquely identify this coupon across deployments

apps/api/src/db/schema/user.ts

@@ -93,7 +93,7 @@ export const userInsertSchema = createInsertSchema(users, {
     image: true
   })
   .strict()
-  .refine((user) => {
+  .transform((user) => {
     return {
       ...user,
       emailConfirmToken: sha256(),
@@ -110,7 +110,7 @@ export const userUpdateSchema = createUpdateSchema(users)
     isStripeConnectEnabledByDefault: true
   })
   .strict()
-  .refine((user) => {
+  .transform((user) => {
     return {
       ...user,
       password: user.password ? hashSync(user.password) : undefined

apps/api/src/db/schemas.ts

@@ -13,13 +13,13 @@ export const consumerIdSchema = getCuidSchema('consumer id')
 
 export const projectIdSchema = z
   .string()
-  .refine((id) => validators.project(id), {
+  .refine((id) => validators.projectId(id), {
     message: 'Invalid project id'
   })
 
 export const deploymentIdSchema = z
   .string()
-  .refine((id) => validators.deployment(id), {
+  .refine((id) => validators.deploymentId(id), {
     message: 'Invalid deployment id'
   })
 

apps/api/src/lib/billing/upsert-consumer.ts

@@ -0,0 +1,151 @@
+import { parseFaasIdentifier } from '@agentic/validators'
+
+import { and, db, eq, schema } from '@/db'
+import { assert, sha256 } from '@/lib/utils'
+
+import type { AuthenticatedContext } from '../types'
+import { upsertStripeConnectCustomer } from './upsert-stripe-connect-customer'
+import { upsertStripeCustomer } from './upsert-stripe-customer'
+import { upsertStripePricing } from './upsert-stripe-pricing'
+import { upsertStripeSubscription } from './upsert-stripe-subscription'
+
+export async function upsertConsumer(
+  c: AuthenticatedContext,
+  {
+    plan,
+    deploymentId,
+    consumerId
+  }: {
+    plan?: string
+    deploymentId?: string
+    consumerId?: string
+  }
+) {
+  assert(consumerId || deploymentId, 400)
+  const userId = c.get('userId')
+  let projectId: string | undefined
+
+  if (deploymentId) {
+    const parsedIds = parseFaasIdentifier(deploymentId)
+    assert(parsedIds, 400, 'Invalid "deploymentId"')
+    projectId = parsedIds.projectId
+  }
+
+  if (!consumerId) {
+    assert(projectId, 400, 'Missing required "deploymentId"')
+  }
+
+  const [{ user, stripeCustomer }, existingConsumer] = await Promise.all([
+    upsertStripeCustomer(c),
+
+    db.query.consumers.findFirst({
+      where: consumerId
+        ? eq(schema.consumers.id, consumerId)
+        : and(
+            eq(schema.consumers.userId, userId),
+            eq(schema.consumers.projectId, projectId!)
+          )
+    })
+  ])
+
+  if (consumerId) {
+    assert(existingConsumer, 404, `Consumer not found "${consumerId}"`)
+    assert(existingConsumer.id === consumerId, 403)
+
+    if (projectId) {
+      assert(
+        existingConsumer.projectId === projectId,
+        400,
+        `Deployment "${deploymentId}" does not belong to with consumer "${consumerId}" project "${existingConsumer.projectId}"`
+      )
+    }
+
+    consumerId = existingConsumer.id
+    deploymentId ??= existingConsumer.deploymentId
+    projectId ??= existingConsumer.projectId
+  }
+
+  assert(consumerId)
+  assert(deploymentId)
+  assert(projectId)
+
+  assert(
+    !existingConsumer ||
+      !existingConsumer.enabled ||
+      existingConsumer.plan !== plan ||
+      existingConsumer.deploymentId !== deploymentId,
+    409,
+    `User "${user.email}" already has an active subscription to plan "${plan}" for project "${projectId}"`
+  )
+
+  const deployment = await db.query.deployments.findFirst({
+    where: eq(schema.deployments.id, deploymentId),
+    with: {
+      project: true
+    }
+  })
+  assert(deployment, 404, `Deployment not found "${deploymentId}"`)
+
+  const { project } = deployment
+  assert(
+    project,
+    404,
+    `Project not found "${projectId}" for deployment "${deploymentId}"`
+  )
+  assert(
+    deployment.enabled,
+    410,
+    `Deployment has been disabled by its owner "${deployment.id}"`
+  )
+
+  let consumer = existingConsumer
+
+  if (consumer) {
+    ;[consumer] = await db
+      .update(schema.consumers)
+      .set({
+        plan,
+        deploymentId
+      })
+      .where(eq(schema.consumers.id, consumer.id))
+      .returning()
+  } else {
+    ;[consumer] = await db.insert(schema.consumers).values({
+      plan,
+      userId,
+      deploymentId,
+      projectId,
+      // TODO: refactor / improve token generation
+      token: sha256().slice(0, 24),
+      _stripeCustomerId: stripeCustomer.id
+    })
+  }
+
+  assert(consumer, 500, 'Error creating consumer')
+
+  // Ensure that all Stripe pricing resources exist for this deployment
+  await upsertStripePricing({ deployment, project })
+
+  // Ensure that customer and default source are created on the stripe connect account
+  // TODO: is this necessary?
+  // consumer._stripeAccount = project._stripeAccount
+  await upsertStripeConnectCustomer({ stripeCustomer, consumer, project })
+
+  const logger = c.get('logger')
+  logger.info('SUBSCRIPTION', existingConsumer ? 'UPDATE' : 'CREATE', {
+    project,
+    deployment,
+    consumer
+  })
+
+  const { subscription, consumer: updatedConsumer } =
+    await upsertStripeSubscription(c, {
+      consumer,
+      user,
+      project,
+      deployment
+    })
+  logger.info('subscription', subscription)
+
+  return updatedConsumer
+}

apps/api/src/lib/billing/upsert-stripe-pricing.ts

@@ -259,7 +259,7 @@ export async function upsertStripePricing({
   }
 
   for (const pricingPlan of Object.values(deployment.pricingPlanMap)) {
-    for (const pricingPlanLineItem of Object.values(pricingPlan.metricsMap)) {
+    for (const pricingPlanLineItem of pricingPlan.lineItems) {
       upserts.push(() =>
         upsertStripeResourcesForPricingPlanLineItem({
           pricingPlan,

apps/api/src/lib/try-get-deployment.ts

@@ -0,0 +1,95 @@
+import { parseFaasIdentifier } from '@agentic/validators'
+
+import { db, eq, type RawDeployment, schema } from '@/db'
+
+import type { AuthenticatedContext } from './types'
+import { ensureAuthUser } from './ensure-auth-user'
+import { assert } from './utils'
+
+/**
+ * Attempts to find the Deployment matching the given identifier.
+ *
+ * If the Deployment is not found, throw an HttpError.
+ */
+export async function tryGetDeployment(
+  ctx: AuthenticatedContext,
+  identifier: string,
+  dbQueryOpts: {
+    with?: {
+      user?: true
+      team?: true
+      project?: true
+    }
+  } = {}
+): Promise<RawDeployment> {
+  const user = await ensureAuthUser(ctx)
+
+  const teamMember = ctx.get('teamMember')
+  const namespace = teamMember ? teamMember.teamSlug : user.username
+  const parsedFaas = parseFaasIdentifier(identifier, {
+    namespace
+  })
+  assert(parsedFaas, 400, `Invalid deployment identifier "${identifier}"`)
+
+  const { projectId, deploymentHash, version } = parsedFaas
+
+  if (deploymentHash) {
+    const deploymentId = `${projectId}@${deploymentHash}`
+
+    const deployment = await db.query.deployments.findFirst({
+      ...dbQueryOpts,
+      where: eq(schema.deployments.id, deploymentId)
+    })
+    assert(deployment, 404, `Deployment not found "${deploymentId}"`)
+
+    return deployment
+  } else if (version === 'latest') {
+    const project = await db.query.projects.findFirst({
+      ...dbQueryOpts,
+      where: eq(schema.projects.id, projectId)
+    })
+    assert(project, 404, `Project not found "${projectId}"`)
+    assert(
+      project.lastPublishedDeploymentId,
+      404,
+      'Project has no published deployments'
+    )
+
+    const deployment = await db.query.deployments.findFirst({
+      ...dbQueryOpts,
+      where: eq(schema.deployments.id, project.lastPublishedDeploymentId)
+    })
+    assert(
+      deployment,
+      404,
+      `Deployment not found "${project.lastPublishedDeploymentId}"`
+    )
+
+    return deployment
+  } else if (version === 'dev') {
+    const project = await db.query.projects.findFirst({
+      ...dbQueryOpts,
+      where: eq(schema.projects.id, projectId)
+    })
+    assert(project, 404, `Project not found "${projectId}"`)
+    assert(
+      project.lastDeploymentId,
+      404,
+      'Project has no published deployments'
+    )
+
+    const deployment = await db.query.deployments.findFirst({
+      ...dbQueryOpts,
+      where: eq(schema.deployments.id, project.lastDeploymentId)
+    })
+    assert(
+      deployment,
+      404,
+      `Deployment not found "${project.lastDeploymentId}"`
+    )
+
+    return deployment
+  }
+
+  assert(false, 400, `Invalid Deployment identifier "${identifier}"`)
+}

packages/validators/src/validators.test.ts

@@ -72,40 +72,40 @@ test('deploymentHash failure', () => {
   expect(validators.deploymentHash('012345678')).toBe(false)
 })
 
-test('project success', () => {
+test('projectId success', () => {
-  expect(validators.project('username/project-name')).toBe(true)
+  expect(validators.projectId('username/project-name')).toBe(true)
-  expect(validators.project('a/123')).toBe(true)
+  expect(validators.projectId('a/123')).toBe(true)
 })
 
-test('project failure', () => {
+test('projectId failure', () => {
-  expect(validators.project('aaa//0123')).toBe(false)
+  expect(validators.projectId('aaa//0123')).toBe(false)
-  expect(validators.project('foo@bar')).toBe(false)
+  expect(validators.projectId('foo@bar')).toBe(false)
-  expect(validators.project('abc/1.23')).toBe(false)
+  expect(validators.projectId('abc/1.23')).toBe(false)
-  expect(validators.project('012345678/123@latest')).toBe(false)
+  expect(validators.projectId('012345678/123@latest')).toBe(false)
-  expect(validators.project('foo@dev')).toBe(false)
+  expect(validators.projectId('foo@dev')).toBe(false)
-  expect(validators.project('username/Project-Name')).toBe(false)
+  expect(validators.projectId('username/Project-Name')).toBe(false)
-  expect(validators.project('_/___')).toBe(false)
+  expect(validators.projectId('_/___')).toBe(false)
 })
 
-test('deployment success', () => {
+test('deploymentId success', () => {
-  expect(validators.deployment('username/project-name@01234567')).toBe(true)
+  expect(validators.deploymentId('username/project-name@01234567')).toBe(true)
-  expect(validators.deployment('a/123@01234567')).toBe(true)
+  expect(validators.deploymentId('a/123@01234567')).toBe(true)
 })
 
-test('deployment failure', () => {
+test('deploymentId failure', () => {
-  expect(validators.deployment('username/project-name@012345678')).toBe(false)
+  expect(validators.deploymentId('username/project-name@012345678')).toBe(false)
-  expect(validators.deployment('username/project-name@latest')).toBe(false)
+  expect(validators.deploymentId('username/project-name@latest')).toBe(false)
-  expect(validators.deployment('username/project-name@dev')).toBe(false)
+  expect(validators.deploymentId('username/project-name@dev')).toBe(false)
-  expect(validators.deployment('username/Project-Name@01234567')).toBe(false)
+  expect(validators.deploymentId('username/Project-Name@01234567')).toBe(false)
-  expect(validators.deployment('a/123@0123A567')).toBe(false)
+  expect(validators.deploymentId('a/123@0123A567')).toBe(false)
-  expect(validators.deployment('_/___@012.4567')).toBe(false)
+  expect(validators.deploymentId('_/___@012.4567')).toBe(false)
-  expect(validators.deployment('_/___@01234567')).toBe(false)
+  expect(validators.deploymentId('_/___@01234567')).toBe(false)
-  expect(validators.deployment('aaa//0123@01234567')).toBe(false)
+  expect(validators.deploymentId('aaa//0123@01234567')).toBe(false)
-  expect(validators.deployment('foo@bar@01234567')).toBe(false)
+  expect(validators.deploymentId('foo@bar@01234567')).toBe(false)
-  expect(validators.deployment('abc/1.23@01234567')).toBe(false)
+  expect(validators.deploymentId('abc/1.23@01234567')).toBe(false)
-  expect(validators.deployment('012345678/123@latest')).toBe(false)
+  expect(validators.deploymentId('012345678/123@latest')).toBe(false)
-  expect(validators.deployment('012345678/123@dev')).toBe(false)
+  expect(validators.deploymentId('012345678/123@dev')).toBe(false)
-  expect(validators.deployment('012345678/123@1.0.1')).toBe(false)
+  expect(validators.deploymentId('012345678/123@1.0.1')).toBe(false)
 })
 
 test('serviceName success', () => {

packages/validators/src/validators.ts

@@ -40,11 +40,11 @@ export function deploymentHash(value: string): boolean {
   return !!value && deploymentHashRe.test(value)
 }
 
-export function project(value: string): boolean {
+export function projectId(value: string): boolean {
   return !!value && projectRe.test(value)
 }
 
-export function deployment(value: string): boolean {
+export function deploymentId(value: string): boolean {
   return !!value && deploymentRe.test(value)
 }