Separate classes from client code

apps/webapp/app/routes/api.v1.accounts.$accountId.connections.$clientSlug/CreateExternalConnectionService.server.ts

@@ -0,0 +1,49 @@
+import { CreateExternalConnectionBody } from "@trigger.dev/core";
+import { PrismaClientOrTransaction, prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { integrationAuthRepository } from "~/services/externalApis/integrationAuthRepository.server";
+
+export class CreateExternalConnectionService {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    accountIdentifier: string,
+    clientSlug: string,
+    environment: AuthenticatedEnvironment,
+    payload: CreateExternalConnectionBody
+  ) {
+    const externalAccount = await this.#prismaClient.externalAccount.upsert({
+      where: {
+        environmentId_identifier: {
+          environmentId: environment.id,
+          identifier: accountIdentifier,
+        },
+      },
+      create: {
+        environmentId: environment.id,
+        organizationId: environment.organizationId,
+        identifier: accountIdentifier,
+      },
+      update: {},
+    });
+
+    const integration = await this.#prismaClient.integration.findUniqueOrThrow({
+      where: {
+        organizationId_slug: {
+          organizationId: environment.organizationId,
+          slug: clientSlug,
+        },
+      },
+    });
+
+    return await integrationAuthRepository.createConnectionFromToken({
+      externalAccount: externalAccount,
+      integration,
+      token: payload,
+    });
+  }
+}

apps/webapp/app/routes/api.v1.accounts.$accountId.connections.$clientSlug.ts → apps/webapp/app/routes/api.v1.accounts.$accountId.connections.$clientSlug/route.ts

@@ -1,15 +1,10 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import {
-  CreateExternalConnectionBody,
-  CreateExternalConnectionBodySchema,
-  ErrorWithStackSchema,
-} from "@trigger.dev/core";
+import { CreateExternalConnectionBodySchema, ErrorWithStackSchema } from "@trigger.dev/core";
 import { z } from "zod";
 import { generateErrorMessage } from "zod-error";
-import { PrismaClientOrTransaction, prisma } from "~/db.server";
-import { AuthenticatedEnvironment, authenticateApiRequest } from "~/services/apiAuth.server";
-import { integrationAuthRepository } from "~/services/externalApis/integrationAuthRepository.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { CreateExternalConnectionService } from "./CreateExternalConnectionService.server";
 
 const ParamsSchema = z.object({
   accountId: z.string(),
@@ -67,48 +62,3 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ message: parsedError.data.message }, { status: 500 });
   }
 }
-
-class CreateExternalConnectionService {
-  #prismaClient: PrismaClientOrTransaction;
-
-  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
-  public async call(
-    accountIdentifier: string,
-    clientSlug: string,
-    environment: AuthenticatedEnvironment,
-    payload: CreateExternalConnectionBody
-  ) {
-    const externalAccount = await this.#prismaClient.externalAccount.upsert({
-      where: {
-        environmentId_identifier: {
-          environmentId: environment.id,
-          identifier: accountIdentifier,
-        },
-      },
-      create: {
-        environmentId: environment.id,
-        organizationId: environment.organizationId,
-        identifier: accountIdentifier,
-      },
-      update: {},
-    });
-
-    const integration = await this.#prismaClient.integration.findUniqueOrThrow({
-      where: {
-        organizationId_slug: {
-          organizationId: environment.organizationId,
-          slug: clientSlug,
-        },
-      },
-    });
-
-    return await integrationAuthRepository.createConnectionFromToken({
-      externalAccount: externalAccount,
-      integration,
-      token: payload,
-    });
-  }
-}

apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier.ts → apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier/TriggerEndpointIndexHookService.server.ts

@@ -1,74 +1,10 @@
-import { ActionFunctionArgs, LoaderFunctionArgs, json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import type { PrismaClient } from "~/db.server";
 import { $transaction, prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { workerQueue } from "~/services/worker.server";
-import { safeJsonParse } from "~/utils/json";
 import { RuntimeEnvironmentType } from "~/database-types";
-
-const ParamsSchema = z.object({
-  environmentId: z.string(),
-  endpointSlug: z.string(),
-  indexHookIdentifier: z.string(),
-});
-
-export async function loader({ params }: LoaderFunctionArgs) {
-  const parsedParams = ParamsSchema.safeParse(params);
-
-  if (!parsedParams.success) {
-    return {
-      status: 400,
-      json: {
-        error: "Invalid params",
-      },
-    };
-  }
-
-  const { environmentId, endpointSlug, indexHookIdentifier } = parsedParams.data;
-
-  const service = new TriggerEndpointIndexHookService();
-
-  await service.call({
-    environmentId,
-    endpointSlug,
-    indexHookIdentifier,
-  });
-
-  return json({
-    ok: true,
-  });
-}
-
-export async function action({ request, params }: ActionFunctionArgs) {
-  const parsedParams = ParamsSchema.safeParse(params);
-
-  if (!parsedParams.success) {
-    return {
-      status: 400,
-      json: {
-        error: "Invalid params",
-      },
-    };
-  }
-
-  const { environmentId, endpointSlug, indexHookIdentifier } = parsedParams.data;
-
-  const body = await request.text();
-
-  const service = new TriggerEndpointIndexHookService();
-
-  await service.call({
-    environmentId,
-    endpointSlug,
-    indexHookIdentifier,
-    body: body ? safeJsonParse(body) : undefined,
-  });
-
-  return json({
-    ok: true,
-  });
-}
+import type { ParamsSchema } from "./route";
 
 type TriggerEndpointDeployHookOptions = z.infer<typeof ParamsSchema> & {
   body?: any;

apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier/route.ts

@@ -0,0 +1,67 @@
+import { ActionFunctionArgs, LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { safeJsonParse } from "~/utils/json";
+import { TriggerEndpointIndexHookService } from "./TriggerEndpointIndexHookService.server";
+
+export const ParamsSchema = z.object({
+  environmentId: z.string(),
+  endpointSlug: z.string(),
+  indexHookIdentifier: z.string(),
+});
+
+export async function loader({ params }: LoaderFunctionArgs) {
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return {
+      status: 400,
+      json: {
+        error: "Invalid params",
+      },
+    };
+  }
+
+  const { environmentId, endpointSlug, indexHookIdentifier } = parsedParams.data;
+
+  const service = new TriggerEndpointIndexHookService();
+
+  await service.call({
+    environmentId,
+    endpointSlug,
+    indexHookIdentifier,
+  });
+
+  return json({
+    ok: true,
+  });
+}
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return {
+      status: 400,
+      json: {
+        error: "Invalid params",
+      },
+    };
+  }
+
+  const { environmentId, endpointSlug, indexHookIdentifier } = parsedParams.data;
+
+  const body = await request.text();
+
+  const service = new TriggerEndpointIndexHookService();
+
+  await service.call({
+    environmentId,
+    endpointSlug,
+    indexHookIdentifier,
+    body: body ? safeJsonParse(body) : undefined,
+  });
+
+  return json({
+    ok: true,
+  });
+}

apps/webapp/app/routes/api.v1.runs.$runId.logs/CreateRunLogService.server.ts

@@ -0,0 +1,20 @@
+import type { LogMessage } from "@trigger.dev/core";
+import type { PrismaClient } from "@trigger.dev/database";
+import { prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+
+export class CreateRunLogService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(environment: AuthenticatedEnvironment, runId: string, logMessage: LogMessage) {
+    // @ts-ignore
+    logger.debug(logMessage.message, logMessage.data ?? {});
+
+    return logMessage;
+  }
+}

apps/webapp/app/routes/api.v1.runs.$runId.logs.ts → apps/webapp/app/routes/api.v1.runs.$runId.logs/route.ts

@@ -1,13 +1,9 @@
-import type { Organization, RuntimeEnvironment } from "@trigger.dev/database";
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import type { LogMessage } from "@trigger.dev/core";
 import { LogMessageSchema } from "@trigger.dev/core";
 import { z } from "zod";
-import type { PrismaClient } from "~/db.server";
-import { prisma } from "~/db.server";
-import { authenticateApiRequest, AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { logger } from "~/services/logger.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { CreateRunLogService } from "./CreateRunLogService.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -53,18 +49,3 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: "Something went wrong" }, { status: 500 });
   }
 }
-
-export class CreateRunLogService {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
-  public async call(environment: AuthenticatedEnvironment, runId: string, logMessage: LogMessage) {
-    // @ts-ignore
-    logger.debug(logMessage.message, logMessage.data ?? {});
-
-    return logMessage;
-  }
-}

apps/webapp/app/routes/api.v1.runs.$runId.statuses.$id/SetStatusService.server.ts

@@ -0,0 +1,71 @@
+import {
+  StatusUpdate,
+  StatusHistory,
+  StatusHistorySchema,
+  StatusUpdateState,
+  StatusUpdateData,
+} from "@trigger.dev/core";
+import { PrismaClient } from "@trigger.dev/database";
+import { prisma, $transaction } from "~/db.server";
+
+export class SetStatusService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(runId: string, id: string, status: StatusUpdate) {
+    const statusRecord = await $transaction(this.#prismaClient, async (tx) => {
+      const existingStatus = await tx.jobRunStatusRecord.findUnique({
+        where: {
+          runId_key: {
+            runId,
+            key: id,
+          },
+        },
+      });
+
+      const history: StatusHistory = [];
+      const historyResult = StatusHistorySchema.safeParse(existingStatus?.history);
+      if (historyResult.success) {
+        history.push(...historyResult.data);
+      }
+      if (existingStatus) {
+        history.push({
+          label: existingStatus.label,
+          state: (existingStatus.state ?? undefined) as StatusUpdateState,
+          data: (existingStatus.data ?? undefined) as StatusUpdateData,
+        });
+      }
+
+      const updatedStatus = await tx.jobRunStatusRecord.upsert({
+        where: {
+          runId_key: {
+            runId,
+            key: id,
+          },
+        },
+        create: {
+          key: id,
+          runId,
+          //this shouldn't ever use the id in reality, as the SDK makess it compulsory on the first call
+          label: status.label ?? id,
+          state: status.state,
+          data: status.data as any,
+          history: [],
+        },
+        update: {
+          label: status.label,
+          state: status.state,
+          data: status.data as any,
+          history: history as any[],
+        },
+      });
+
+      return updatedStatus;
+    });
+
+    return statusRecord;
+  }
+}