Monthly billing

.env.example

@@ -3,3 +3,6 @@ DATABASE_URL="YOUR_NOT_SO_SECRET_DATABASE_URL_GOES_HERE"
 LEMON_SQUEEZY_API_KEY="" # Just get it from somewhere
 LEMON_SQUEEZY_STORE_ID="" # Just get it from somewhere
 LEMON_SQUEEZY_VARIANT_ID="" # Just get it from somewhere
+LEMON_SQUEEZY_WEBHOOK_SECRET=
+TEST_API_KEY=
+REDIS_URL=

package.json

@@ -18,11 +18,13 @@
     "@bufbuild/protoc-gen-connect-es": "^0.13.0",
     "@bufbuild/protoc-gen-es": "^2.9.0",
     "@types/bun": "latest",
+    "@types/expr-eval": "^1.1.2",
     "@types/luxon": "^3.7.1",
     "@vitest/ui": "^4.0.3",
     "buf": "bufbuild/buf",
     "drizzle-kit": "^0.31.6",
     "tsx": "^4.20.6",
+    "skills": "^1.3.1",
     "vitest": "^4.0.3"
   },
   "peerDependencies": {
@@ -31,20 +33,20 @@
   "dependencies": {
     "@bufbuild/protobuf": "^2.9.0",
     "@connectrpc/connect": "^2.1.0",
-    "@connectrpc/connect-node": "^2.1.0",
+    "@connectrpc/connect-fastify": "^2.1.1",
+    "@connectrpc/connect-node": "^2.1.1",
     "@connectrpc/validate": "^0.2.0",
     "@lemonsqueezy/lemonsqueezy.js": "^4.0.0",
-    "@libsql/client": "^0.15.15",
-    "@types/expr-eval": "^1.1.2",
+    "bullmq": "^5.75.2",
     "dotenv": "^17.2.3",
     "drizzle-orm": "^0.44.7",
     "expr-eval": "^2.0.2",
+    "fastify": "^5.8.5",
+    "fastify-raw-body": "^5.0.0",
     "luxon": "^3.7.2",
-    "mysql2": "^3.15.3",
     "pino": "^10.1.0",
     "pino-pretty": "^13.1.2",
     "postgres": "^3.4.7",
-    "skills": "^1.3.1",
     "zod": "^4.1.12"
   }
 }

src/errors/internals.ts

@@ -0,0 +1,73 @@
+import { Code, ConnectError } from "@connectrpc/connect";
+
+export enum InternalsErrorType {
+  INVALID_CRON = "INVALID_CRON",
+  QUEUE_CREATION_FAILED = "QUEUE_CREATION_FAILED",
+  VALIDATION_FAILED = "VALIDATION_FAILED",
+  UNKNOWN = "UNKNOWN",
+}
+
+export interface InternalsErrorContext {
+  type: InternalsErrorType;
+  message: string;
+  originalError?: Error;
+  code: Code;
+}
+
+export class InternalsError extends ConnectError {
+  readonly type: InternalsErrorType;
+  readonly originalError?: Error;
+
+  constructor(context: InternalsErrorContext) {
+    super(context.message, context.code);
+    this.name = "InternalsError";
+    this.type = context.type;
+    this.originalError = context.originalError;
+
+    Object.setPrototypeOf(this, InternalsError.prototype);
+  }
+
+  static invalidCron(details?: string, originalError?: Error): InternalsError {
+    return new InternalsError({
+      type: InternalsErrorType.INVALID_CRON,
+      message: details
+        ? `Invalid cron expression: ${details}`
+        : "Invalid cron expression",
+      code: Code.InvalidArgument,
+      originalError,
+    });
+  }
+
+  static queueCreationFailed(
+    details?: string,
+    originalError?: Error
+  ): InternalsError {
+    return new InternalsError({
+      type: InternalsErrorType.QUEUE_CREATION_FAILED,
+      message: details
+        ? `Failed to create queue: ${details}`
+        : "Failed to create queue",
+      code: Code.Internal,
+      originalError,
+    });
+  }
+
+  static validationFailed(details: string, originalError?: Error): InternalsError {
+    return new InternalsError({
+      type: InternalsErrorType.VALIDATION_FAILED,
+      message: `Internals validation failed: ${details}`,
+      code: Code.InvalidArgument,
+      originalError,
+    });
+  }
+
+  static unknown(originalError?: Error): InternalsError {
+    const details = originalError?.message || "No details available";
+    return new InternalsError({
+      type: InternalsErrorType.UNKNOWN,
+      message: `Unexpected internals error: ${details}`,
+      code: Code.Internal,
+      originalError,
+    });
+  }
+}

src/events/RawEvents/Metadata.ts

@@ -0,0 +1,21 @@
+import type { MetadataEvent, MetadataEventData } from "../../interface/event/Event";
+import { DateTime } from "luxon";
+
+export class Metadata implements MetadataEvent {
+  public reported_timestamp: DateTime;
+  public readonly type = "METADATA" as const;
+
+  constructor(public data: MetadataEventData) {
+    this.reported_timestamp = DateTime.utc();
+  }
+
+  serialize() {
+    return {
+      SQL: {
+        type: this.type,
+        reported_timestamp: this.reported_timestamp,
+        data: this.data,
+      },
+    };
+  }
+}

src/factory/EventStorageAdapterFactory.ts

@@ -1,12 +1,6 @@
 import type { EventKind } from "../interface/event/Event.ts";
 import { PostgresAdapter } from "../storage/adapter/postgres/postgres.ts";
 
-/**
- * StorageAdapterFactory - Facade for the new SQL adapter factory
- *
- * Maintains backward compatibility while delegating to the new
- * dependency-injected SQL adapter factory
- */
 export class StorageAdapterFactory {
   /**
    * Get the appropriate storage adapter for a given event
@@ -29,6 +23,9 @@ export class StorageAdapterFactory {
       case "ADD_KEY": {
         return new PostgresAdapter();
       }
+      case "METADATA": {
+        return new PostgresAdapter();
+      }
       default: {
         throw new Error(`Unknown event type: ${RequestType}`);
       }

src/interceptors/connectInterceptors.ts

@@ -0,0 +1,8 @@
+import type { Interceptor } from "@connectrpc/connect";
+import { createValidateInterceptor } from "@connectrpc/validate";
+import { loggingInterceptor } from "./logging.ts";
+import { authInterceptor } from "./auth.ts";
+
+export function createConnectInterceptors(): Interceptor[] {
+  return [loggingInterceptor(), createValidateInterceptor(), authInterceptor()];
+}

src/interface/event/Event.ts

@@ -27,6 +27,11 @@ export type PaymentEventData = {
   creditAmount: number;
 };
 
+export type MetadataEventData = {
+  payment_cron: string;
+  payment_webhook: string | null;
+};
+
 /**
  * Event kind discriminator
  */
@@ -35,6 +40,7 @@ export type EventKind =
   | "AI_TOKEN_USAGE"
   | "ADD_KEY"
   | "PAYMENT"
+  | "METADATA";
 
 /**
  * Mapping of event kinds to their data structures
@@ -44,6 +50,7 @@ export type EventDataMap = {
   AI_TOKEN_USAGE: AITokenUsageEventData;
   ADD_KEY: AddKeyEventData;
   PAYMENT: PaymentEventData;
+  METADATA: MetadataEventData;
 };
 
 /**
@@ -75,6 +82,7 @@ type SqlRecordMap = {
   SDK_CALL: SqlRecordWithUserId<"SDK_CALL">;
   AI_TOKEN_USAGE: SqlRecordWithUserId<"AI_TOKEN_USAGE">;
   PAYMENT: SqlRecordWithUserId<"PAYMENT">;
+  METADATA: BaseSqlRecord<"METADATA">;
 };
 
 /**
@@ -125,3 +133,8 @@ export interface AddKeyEvent extends Event<"ADD_KEY"> {}
 export interface PaymentEvent extends Event<"PAYMENT"> {
   readonly userId: UserId;
 }
+
+/**
+ * Metadata Event
+ */
+export interface MetadataEvent extends Event<"METADATA"> {}

src/interface/storage/Storage.ts

@@ -9,7 +9,7 @@ export interface StorageAdapter {
 
   add(
     serialized: SerializedEvent<EventKind>,
-    apiKeyId: string
+    apiKeyId?: string
   ): Promise<{ id: string } | void>;
   price(userID: UserId, event_type: EventKind): Promise<number>;
 }

src/queues/onboarding.ts

@@ -0,0 +1,39 @@
+import { Queue, type RepeatOptions } from "bullmq";
+import { DateTime } from "luxon";
+import { getRedisConnection } from "../storage/db/redis.ts";
+
+export interface OnboardingJobData {
+  cronExpression: string;
+  createdAt: string;
+}
+
+let onboardingQueue: Queue<OnboardingJobData> | null = null;
+
+export function getOnboardingQueue(): Queue<OnboardingJobData> {
+  if (!onboardingQueue) {
+    onboardingQueue = new Queue<OnboardingJobData>("onboarding", {
+      connection: getRedisConnection(),
+    });
+  }
+  return onboardingQueue;
+}
+
+export async function addOnboardingCronJob(
+  cronExpression: string
+): Promise<void> {
+  const repeatOptions: RepeatOptions = {
+    pattern: cronExpression,
+  };
+
+  const queue = getOnboardingQueue();
+  await queue.add(
+    `onboarding-${cronExpression}`,
+    {
+      cronExpression,
+      createdAt: DateTime.utc().toISO(),
+    },
+    {
+      repeat: repeatOptions,
+    }
+  );
+}

src/routes/gRPC/registerRoutes.ts

@@ -0,0 +1,23 @@
+import type { ConnectRouter } from "@connectrpc/connect";
+import { EventService } from "../../gen/event/v1/event_pb.ts";
+import { AuthService } from "../../gen/auth/v1/auth_pb.ts";
+import { PaymentService } from "../../gen/payment/v1/payment_pb.ts";
+import { registerEvent } from "./events/registerEvent.ts";
+import { streamEvents } from "./events/streamEvents.ts";
+import { createAPIKey } from "./auth/createAPIKey.ts";
+import { createCheckoutLink } from "./payment/createCheckoutLink.ts";
+
+export function registerGrpcRoutes(router: ConnectRouter): void {
+  router.service(AuthService, {
+    createAPIKey,
+  });
+
+  router.service(EventService, {
+    registerEvent,
+    streamEvents,
+  });
+
+  router.service(PaymentService, {
+    createCheckoutLink,
+  });
+}

src/routes/http/api/onboarding.ts

@@ -0,0 +1,80 @@
+import type { FastifyRequest, FastifyReply } from "fastify";
+import { ZodError } from "zod";
+import {
+  onboardingCronSchema,
+  type OnboardingCronSchemaType,
+} from "../../../zod/internals.ts";
+import { addOnboardingCronJob } from "../../../queues/onboarding.ts";
+import {
+  createWideEventBuilder,
+  generateRequestId,
+} from "../../../context/requestContext.ts";
+import { logger } from "../../../errors/logger.ts";
+import { StorageAdapterFactory } from "../../../factory/index.ts";
+import { Metadata } from "../../../events/RawEvents/Metadata.ts";
+
+export async function handleOnboarding(
+  request: FastifyRequest,
+  reply: FastifyReply
+): Promise<{ crons: string[] }> {
+  const builder = createWideEventBuilder(
+    generateRequestId(),
+    request.method,
+    request.url
+  );
+
+  try {
+    const body = await request.body;
+    const validated = onboardingCronSchema.parse(body);
+
+    const crons: string[] = [];
+
+    for (const cronExpression of validated.crons) {
+      await addOnboardingCronJob(cronExpression);
+      crons.push(cronExpression);
+    }
+
+    const webhookUrl = validated.webhookUrl && validated.webhookUrl !== ""
+      ? validated.webhookUrl
+      : null;
+
+    const metadataEvent = new Metadata({
+      payment_cron: crons.join(","),
+      payment_webhook: webhookUrl,
+    });
+
+    const adapter = await StorageAdapterFactory.getEventStorageAdapter(
+      metadataEvent.type
+    );
+    await adapter.add(metadataEvent.serialize());
+
+    builder.setSuccess(200).addContext({
+      cronCount: crons.length,
+    });
+
+    reply.code(201);
+    return { crons };
+  } catch (error) {
+    if (error instanceof ZodError) {
+      const issues = error.issues
+        .map((issue) => `${issue.path.join(".")}: ${issue.message}`)
+        .join("; ");
+      builder.setError(400, {
+        type: "ValidationError",
+        message: issues,
+      });
+      reply.code(400);
+      return { crons: [] };
+    }
+
+    const err = error instanceof Error ? error : new Error(String(error));
+    builder.setError(500, {
+      type: "InternalError",
+      message: err.message,
+    });
+    reply.code(500);
+    return { crons: [] };
+  } finally {
+    logger.emit(builder.build());
+  }
+}

src/routes/http/api/registerApiRoutes.ts

@@ -0,0 +1,16 @@
+import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify";
+import { handleOnboarding } from "./onboarding.ts";
+
+export async function registerApiRoutes(
+  server: ReturnType<typeof import("fastify")["fastify"]>
+): Promise<void> {
+  server.post(
+    "/api/v1/internals/onboarding",
+    async (
+      request: FastifyRequest,
+      reply: FastifyReply
+    ) => {
+      return handleOnboarding(request, reply);
+    }
+  );
+}