From 817b4ed1333f4893f727f8cb3d08ec970b307f89 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 14 Jun 2023 16:07:07 +0100 Subject: [PATCH] Endpoint registration and indexing now is only initiated outside of clients --- .changeset/four-ligers-perform.md | 5 + .../api.v1.endpoints.$endpointSlug.index.ts | 73 ++++++ ...endpointSlug.index.$indexHookIdentifier.ts | 218 ++++++++++++++++++ apps/webapp/app/routes/api.v1.endpoints.ts | 4 +- apps/webapp/app/services/endpointApi.ts | 101 ++++---- .../endpoints/createEndpoint.server.ts | 31 ++- .../endpoints/endpointRegistered.server.ts | 87 ------- .../endpoints/indexEndpoint.server.ts | 130 +++++++++++ .../app/services/runs/performRunExecution.ts | 12 +- .../deliverHttpSourceRequest.server.ts | 3 +- .../triggers/initializeTrigger.server.ts | 6 +- apps/webapp/app/services/worker.server.ts | 20 +- .../migration.sql | 2 + .../migration.sql | 10 + .../migration.sql | 28 +++ .../migration.sql | 2 + .../migration.sql | 16 ++ .../migration.sql | 2 + apps/webapp/prisma/schema.prisma | 27 +++ packages/internal/src/schemas/api.ts | 22 +- packages/trigger-sdk/src/triggerClient.ts | 48 ++-- 21 files changed, 661 insertions(+), 186 deletions(-) create mode 100644 .changeset/four-ligers-perform.md create mode 100644 apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts create mode 100644 apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier.ts delete mode 100644 apps/webapp/app/services/endpoints/endpointRegistered.server.ts create mode 100644 apps/webapp/app/services/endpoints/indexEndpoint.server.ts create mode 100644 apps/webapp/prisma/migrations/20230614103739_add_deploy_hook_identifier_to_endpoints/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230614110359_rename_deploy_to_index/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230614122553_create_endpoint_index_model/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230614125945_add_source_data_to_endpoint_index/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230614135014_change_endpoint_source_enum/migration.sql create mode 100644 apps/webapp/prisma/migrations/20230614141902_added_api_to_endpoint_index_source/migration.sql diff --git a/.changeset/four-ligers-perform.md b/.changeset/four-ligers-perform.md new file mode 100644 index 000000000..ae3ae5f62 --- /dev/null +++ b/.changeset/four-ligers-perform.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Endpoint registration and indexing now is only initiated outside of clients diff --git a/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts b/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts new file mode 100644 index 000000000..bf212c5fc --- /dev/null +++ b/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts @@ -0,0 +1,73 @@ +import { ActionArgs, json } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { prisma } from "~/db.server"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; +import { IndexEndpointService } from "~/services/endpoints/indexEndpoint.server"; +import { logger } from "~/services/logger"; + +const ParamsSchema = z.object({ + endpointSlug: z.string(), +}); + +const BodySchema = z.object({ + reason: z.string().optional(), + data: z.any().optional(), +}); + +export async function action({ request, params }: ActionArgs) { + // Ensure this is a POST request + if (request.method.toUpperCase() !== "POST") { + return { status: 405, body: "Method Not Allowed" }; + } + + const parsedParams = ParamsSchema.safeParse(params); + + if (!parsedParams.success) { + return json({ error: "Invalid params" }, { status: 400 }); + } + + // Next authenticate the request + const authenticatedEnv = await authenticateApiRequest(request); + + if (!authenticatedEnv) { + logger.info("Invalid or missing api key", { url: request.url }); + + return json({ error: "Invalid or Missing API key" }, { status: 401 }); + } + + const { endpointSlug } = parsedParams.data; + + const endpoint = await prisma.endpoint.findUnique({ + where: { + environmentId_slug: { + environmentId: authenticatedEnv.id, + slug: endpointSlug, + }, + }, + }); + + if (!endpoint) { + logger.info("Endpoint not found", { url: request.url }); + + return json({ error: "Endpoint not found" }, { status: 404 }); + } + + const body = await request.json(); + + const parsedBody = BodySchema.safeParse(body); + + if (!parsedBody.success) { + return json({ error: "Invalid body" }, { status: 400 }); + } + + const service = new IndexEndpointService(); + + const { data, ...index } = await service.call( + endpoint.id, + "API", + parsedBody.data.reason, + parsedBody.data.data + ); + + return json(index); +} diff --git a/apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier.ts b/apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier.ts new file mode 100644 index 000000000..8cb697670 --- /dev/null +++ b/apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier.ts @@ -0,0 +1,218 @@ +import { ActionArgs, LoaderArgs, json } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { PrismaClient, prisma } from "~/db.server"; +import { logger } from "~/services/logger"; +import { workerQueue } from "~/services/worker.server"; +import { safeJsonParse } from "~/utils/json"; + +const ParamsSchema = z.object({ + environmentId: z.string(), + endpointSlug: z.string(), + indexHookIdentifier: z.string(), +}); + +export async function loader({ params }: LoaderArgs) { + 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 }: ActionArgs) { + 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, + }); +} + +type TriggerEndpointDeployHookOptions = z.infer & { + body?: any; +}; + +export class TriggerEndpointIndexHookService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call({ + environmentId, + endpointSlug, + indexHookIdentifier, + body, + }: TriggerEndpointDeployHookOptions) { + logger.debug("TriggerEndpointIndexHookService.call", { + environmentId, + endpointSlug, + indexHookIdentifier, + body, + }); + + const endpoint = await this.#prismaClient.endpoint.findUnique({ + where: { + environmentId_slug: { + environmentId, + slug: endpointSlug, + }, + }, + }); + + if (!endpoint) { + throw new Error("Endpoint not found"); + } + + if (endpoint.indexingHookIdentifier !== indexHookIdentifier) { + throw new Error("Index hook identifier is invalid"); + } + + const reason = parseReasonFromBody(body); + + // Index the endpoint in 5 seconds from now + await workerQueue.enqueue( + "indexEndpoint", + { + id: endpoint.id, + source: "HOOK", + reason, + sourceData: body, + }, + { + runAt: new Date(Date.now() + 5000), + } + ); + } +} + +function parseReasonFromBody(body: any): string | undefined { + const vercelDeployment = VercelDeploymentWebhookSchema.safeParse(body); + + if (!vercelDeployment.success) { + return; + } + + const { payload, type } = vercelDeployment.data; + + if (type !== "deployment.succeeded") { + return; + } + + const githubMeta = VercelDeploymentGithubMetaSchema.safeParse( + payload.deployment.meta + ); + + if (!githubMeta.success) { + return `Vercel project ${payload.deployment.name} was deployed to ${payload.deployment.url}`; + } + + return `"${githubMeta.data.githubCommitMessage}" was deployed from ${ + githubMeta.data.githubCommitRef + } (${githubMeta.data.githubCommitSha.slice(0, 7)}) to ${ + payload.deployment.name + }`; +} + +// Example payload: https://jsonhero.io/j/fhIwXEFmi7qa +const VercelDeploymentWebhookSchema = z.object({ + id: z.string(), + payload: z.object({ + user: z.object({ + id: z.string(), + }), + team: z.object({ + id: z.string(), + }), + deployment: z.object({ + id: z.string(), + meta: z.record(z.any()), + name: z.string(), + url: z.string(), + inspectorUrl: z.string(), + }), + links: z.object({ + deployment: z.string(), + project: z.string(), + }), + name: z.string(), + plan: z.string(), + project: z.object({ + id: z.string(), + }), + regions: z.array(z.string()), + target: z.string(), + type: z.string(), + url: z.string(), + }), + createdAt: z.number(), + type: z.enum([ + "deployment.succeeded", + "deployment.failed", + "deployment.ready", + "deployment.created", + "deployment.error", + "deployment.canceled", + ]), +}); + +const VercelDeploymentGithubMetaSchema = z.object({ + githubCommitAuthorName: z.string(), + githubCommitMessage: z.string(), + githubCommitOrg: z.string(), + githubCommitRef: z.string(), + githubCommitRepo: z.string(), + githubCommitSha: z.string(), + githubDeployment: z.string(), + githubOrg: z.string(), + githubRepo: z.string(), + githubRepoOwnerType: z.string(), + githubCommitRepoId: z.string(), + githubRepoId: z.string(), + githubRepoVisibility: z.string(), + githubCommitAuthorLogin: z.string(), + branchAlias: z.string(), +}); diff --git a/apps/webapp/app/routes/api.v1.endpoints.ts b/apps/webapp/app/routes/api.v1.endpoints.ts index 5a09dd1c5..b304f7101 100644 --- a/apps/webapp/app/routes/api.v1.endpoints.ts +++ b/apps/webapp/app/routes/api.v1.endpoints.ts @@ -7,7 +7,7 @@ import { logger } from "~/services/logger"; const BodySchema = z.object({ url: z.string(), - name: z.string(), + id: z.string(), }); export async function action({ request }: ActionArgs) { @@ -42,7 +42,7 @@ export async function action({ request }: ActionArgs) { const endpoint = await service.call({ environment: authenticatedEnv, url: body.data.url, - name: body.data.name, + id: body.data.id, }); return json(endpoint); diff --git a/apps/webapp/app/services/endpointApi.ts b/apps/webapp/app/services/endpointApi.ts index 037d87757..7e08af8fa 100644 --- a/apps/webapp/app/services/endpointApi.ts +++ b/apps/webapp/app/services/endpointApi.ts @@ -1,6 +1,7 @@ import { ApiEventLog, HttpSourceRequest, + PongResponse, PreprocessRunBody, PreprocessRunResponseSchema, RegisterTriggerBody, @@ -10,7 +11,7 @@ import { import { DeliverEventResponseSchema, ErrorWithStackSchema, - GetEndpointDataResponseSchema, + IndexEndpointResponseSchema, HttpSourceResponseSchema, PongResponseSchema, RunJobResponseSchema, @@ -27,34 +28,42 @@ export class EndpointApiError extends Error { // TODO: this should work with tunnelling export class EndpointApi { - #apiKey: string; - #url: string; + constructor( + private apiKey: string, + private url: string, + private id: string + ) {} - constructor(apiKey: string, url: string) { - this.#apiKey = apiKey; - this.#url = url; - } - - async ping() { - const response = await safeFetch(this.#url, { + async ping(): Promise { + const response = await safeFetch(this.url, { method: "POST", headers: { "Content-Type": "application/json", - "x-trigger-api-key": this.#apiKey, + "x-trigger-api-key": this.apiKey, + "x-trigger-endpoint-id": this.id, "x-trigger-action": "PING", }, }); if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); + return { + ok: false, + error: `Could not connect to endpoint ${this.url}`, + }; + } + + if (response.status === 401) { + return { + ok: false, + error: `Trigger API key is invalid`, + }; } if (!response.ok) { - throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` - ); + return { + ok: false, + error: `Could not connect to endpoint ${this.url}. Status code: ${response.status}`, + }; } const anyBody = await response.json(); @@ -66,57 +75,53 @@ export class EndpointApi { return PongResponseSchema.parse(anyBody); } - async getEndpointData() { - const response = await safeFetch(this.#url, { - method: "POSt", + async indexEndpoint() { + const response = await safeFetch(this.url, { + method: "POST", headers: { Accept: "application/json", - "x-trigger-api-key": this.#apiKey, - "x-trigger-action": "GET_ENDPOINT_DATA", + "x-trigger-api-key": this.apiKey, + "x-trigger-action": "INDEX_ENDPOINT", }, }); if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); + throw new Error(`Could not connect to endpoint ${this.url}`); } if (!response.ok) { throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` + `Could not connect to endpoint ${this.url}. Status code: ${response.status}` ); } const anyBody = await response.json(); - logger.debug("getEndpointData() response from endpoint", { + logger.debug("indexEndpoint() response from endpoint", { body: anyBody, }); - return GetEndpointDataResponseSchema.parse(anyBody); + return IndexEndpointResponseSchema.parse(anyBody); } async deliverEvent(event: ApiEventLog) { - const response = await safeFetch(this.#url, { + const response = await safeFetch(this.url, { method: "POST", headers: { "Content-Type": "application/json", - "x-trigger-api-key": this.#apiKey, + "x-trigger-api-key": this.apiKey, "x-trigger-action": "DELIVER_EVENT", }, body: JSON.stringify(event), }); if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); + throw new Error(`Could not connect to endpoint ${this.url}`); } if (!response.ok) { throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` + `Could not connect to endpoint ${this.url}. Status code: ${response.status}` ); } @@ -130,11 +135,11 @@ export class EndpointApi { } async executeJobRequest(options: RunJobBody) { - const response = await safeFetch(this.#url, { + const response = await safeFetch(this.url, { method: "POST", headers: { "content-type": "application/json", - "x-trigger-api-key": this.#apiKey, + "x-trigger-api-key": this.apiKey, "x-trigger-action": "EXECUTE_JOB", }, body: JSON.stringify(options), @@ -147,11 +152,11 @@ export class EndpointApi { } async preprocessRunRequest(options: PreprocessRunBody) { - const response = await safeFetch(this.#url, { + const response = await safeFetch(this.url, { method: "POST", headers: { "Content-Type": "application/json", - "x-trigger-api-key": this.#apiKey, + "x-trigger-api-key": this.apiKey, "x-trigger-action": "PREPROCESS_RUN", }, body: JSON.stringify(options), @@ -164,18 +169,18 @@ export class EndpointApi { id: string, params: any ): Promise { - const response = await safeFetch(this.#url, { + const response = await safeFetch(this.url, { method: "POST", headers: { "Content-Type": "application/json", - "x-trigger-api-key": this.#apiKey, + "x-trigger-api-key": this.apiKey, "x-trigger-action": "INITIALIZE_TRIGGER", }, body: JSON.stringify({ id, params }), }); if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); + throw new Error(`Could not connect to endpoint ${this.url}`); } if (!response.ok) { @@ -189,9 +194,7 @@ export class EndpointApi { } throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` + `Could not connect to endpoint ${this.url}. Status code: ${response.status}` ); } @@ -212,11 +215,11 @@ export class EndpointApi { data: any; request: HttpSourceRequest; }) { - const response = await safeFetch(this.#url, { + const response = await safeFetch(this.url, { method: "POST", headers: { "Content-Type": "application/octet-stream", - "x-trigger-api-key": this.#apiKey, + "x-trigger-api-key": this.apiKey, "x-trigger-action": "DELIVER_HTTP_SOURCE_REQUEST", "x-ts-key": options.key, "x-ts-secret": options.secret, @@ -231,14 +234,12 @@ export class EndpointApi { }); if (!response) { - throw new Error(`Could not connect to endpoint ${this.#url}`); + throw new Error(`Could not connect to endpoint ${this.url}`); } if (!response.ok) { throw new Error( - `Could not connect to endpoint ${this.#url}. Status code: ${ - response.status - }` + `Could not connect to endpoint ${this.url}. Status code: ${response.status}` ); } diff --git a/apps/webapp/app/services/endpoints/createEndpoint.server.ts b/apps/webapp/app/services/endpoints/createEndpoint.server.ts index f0eee815a..e92384b50 100644 --- a/apps/webapp/app/services/endpoints/createEndpoint.server.ts +++ b/apps/webapp/app/services/endpoints/createEndpoint.server.ts @@ -1,10 +1,14 @@ -import type { Organization, RuntimeEnvironment } from ".prisma/client"; -import { $transaction, PrismaClient } from "~/db.server"; -import { prisma } from "~/db.server"; +import { customAlphabet } from "nanoid"; +import { $transaction, prisma, PrismaClient } from "~/db.server"; import { AuthenticatedEnvironment } from "../apiAuth.server"; import { EndpointApi } from "../endpointApi"; import { workerQueue } from "../worker.server"; +const indexingHookIdentifier = customAlphabet( + "0123456789abcdefghijklmnopqrstuvxyz", + 10 +); + export class CreateEndpointService { #prismaClient: PrismaClient; @@ -15,21 +19,26 @@ export class CreateEndpointService { public async call({ environment, url, - name, + id, }: { environment: AuthenticatedEnvironment; url: string; - name: string; + id: string; }) { - const client = new EndpointApi(environment.apiKey, url); - await client.ping(); + const client = new EndpointApi(environment.apiKey, url, id); + + const pong = await client.ping(); + + if (!pong.ok) { + throw new Error(pong.error); + } return await $transaction(this.#prismaClient, async (tx) => { const endpoint = await tx.endpoint.upsert({ where: { environmentId_slug: { environmentId: environment.id, - slug: name, + slug: id, }, }, create: { @@ -48,8 +57,9 @@ export class CreateEndpointService { id: environment.projectId, }, }, - slug: name, + slug: id, url, + indexingHookIdentifier: indexingHookIdentifier(), }, update: { url, @@ -58,9 +68,10 @@ export class CreateEndpointService { // Kick off process to fetch the jobs for this endpoint await workerQueue.enqueue( - "endpointRegistered", + "indexEndpoint", { id: endpoint.id, + source: "INTERNAL", }, { tx } ); diff --git a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts b/apps/webapp/app/services/endpoints/endpointRegistered.server.ts deleted file mode 100644 index 973e13f72..000000000 --- a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts +++ /dev/null @@ -1,87 +0,0 @@ -import type { PrismaClient } from "~/db.server"; -import { prisma } from "~/db.server"; -import { EndpointApi } from "../endpointApi"; -import { workerQueue } from "../worker.server"; - -export class EndpointRegisteredService { - #prismaClient: PrismaClient; - - constructor(prismaClient: PrismaClient = prisma) { - this.#prismaClient = prismaClient; - } - - public async call(id: string) { - const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ - where: { - id, - }, - include: { - environment: true, - }, - }); - - // Make a request to the endpoint to fetch a list of jobs - const client = new EndpointApi(endpoint.environment.apiKey, endpoint.url); - - const { jobs, sources, dynamicTriggers, dynamicSchedules } = - await client.getEndpointData(); - - const queueName = `endpoint-${endpoint.id}`; - - for (const job of jobs) { - if (!job.enabled) { - continue; - } - - await workerQueue.enqueue( - "registerJob", - { - job, - endpointId: endpoint.id, - }, - { - queueName, - } - ); - } - - for (const source of sources) { - await workerQueue.enqueue( - "registerSource", - { - source, - endpointId: endpoint.id, - }, - { - queueName, - } - ); - } - - for (const dynamicTrigger of dynamicTriggers) { - await workerQueue.enqueue( - "registerDynamicTrigger", - { - dynamicTrigger, - endpointId: endpoint.id, - }, - { - queueName, - } - ); - } - - for (const dynamicSchedule of dynamicSchedules) { - await workerQueue.enqueue( - "registerDynamicSchedule", - { - dynamicSchedule, - endpointId: endpoint.id, - }, - { - queueName, - } - ); - } - } -} diff --git a/apps/webapp/app/services/endpoints/indexEndpoint.server.ts b/apps/webapp/app/services/endpoints/indexEndpoint.server.ts new file mode 100644 index 000000000..b3b7d1c70 --- /dev/null +++ b/apps/webapp/app/services/endpoints/indexEndpoint.server.ts @@ -0,0 +1,130 @@ +import { $transaction, PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import { EndpointApi } from "../endpointApi"; +import { workerQueue } from "../worker.server"; +import type { EndpointIndexSource } from ".prisma/client"; + +export class IndexEndpointService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call( + id: string, + source: EndpointIndexSource = "INTERNAL", + reason?: string, + sourceData?: any + ) { + const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ + where: { + id, + }, + include: { + environment: true, + }, + }); + + // Make a request to the endpoint to fetch a list of jobs + const client = new EndpointApi( + endpoint.environment.apiKey, + endpoint.url, + endpoint.slug + ); + + const { jobs, sources, dynamicTriggers, dynamicSchedules } = + await client.indexEndpoint(); + + const queueName = `endpoint-${endpoint.id}`; + + const indexStats = { + jobs: 0, + sources: 0, + dynamicTriggers: 0, + dynamicSchedules: 0, + }; + + return await $transaction(this.#prismaClient, async (tx) => { + for (const job of jobs) { + if (!job.enabled) { + continue; + } + + indexStats.jobs++; + + await workerQueue.enqueue( + "registerJob", + { + job, + endpointId: endpoint.id, + }, + { + queueName, + } + ); + } + + for (const source of sources) { + indexStats.sources++; + + await workerQueue.enqueue( + "registerSource", + { + source, + endpointId: endpoint.id, + }, + { + queueName, + } + ); + } + + for (const dynamicTrigger of dynamicTriggers) { + indexStats.dynamicTriggers++; + + await workerQueue.enqueue( + "registerDynamicTrigger", + { + dynamicTrigger, + endpointId: endpoint.id, + }, + { + queueName, + } + ); + } + + for (const dynamicSchedule of dynamicSchedules) { + indexStats.dynamicSchedules++; + + await workerQueue.enqueue( + "registerDynamicSchedule", + { + dynamicSchedule, + endpointId: endpoint.id, + }, + { + queueName, + } + ); + } + + return await tx.endpointIndex.create({ + data: { + endpointId: endpoint.id, + stats: indexStats, + data: { + jobs, + sources, + dynamicTriggers, + dynamicSchedules, + }, + source, + sourceData, + reason, + }, + }); + }); + } +} diff --git a/apps/webapp/app/services/runs/performRunExecution.ts b/apps/webapp/app/services/runs/performRunExecution.ts index e29e6e184..67940f25f 100644 --- a/apps/webapp/app/services/runs/performRunExecution.ts +++ b/apps/webapp/app/services/runs/performRunExecution.ts @@ -58,7 +58,11 @@ export class PerformRunExecutionService { async #executePreprocessing(execution: FoundRunExecution) { const { run } = execution; - const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); + const client = new EndpointApi( + run.environment.apiKey, + run.endpoint.url, + run.endpoint.slug + ); const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId }); const startedAt = new Date(); @@ -188,7 +192,11 @@ export class PerformRunExecutionService { async #executeJob(execution: FoundRunExecution) { const { run } = execution; - const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); + const client = new EndpointApi( + run.environment.apiKey, + run.endpoint.url, + run.endpoint.slug + ); const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId }); const startedAt = new Date(); diff --git a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts index 5ab630426..d1311be1e 100644 --- a/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts +++ b/apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts @@ -58,7 +58,8 @@ export class DeliverHttpSourceRequestService { const clientApi = new EndpointApi( httpSourceRequest.environment.apiKey, - httpSourceRequest.endpoint.url + httpSourceRequest.endpoint.url, + httpSourceRequest.endpoint.slug ); const { response, events } = await clientApi.deliverHttpSourceRequest({ diff --git a/apps/webapp/app/services/triggers/initializeTrigger.server.ts b/apps/webapp/app/services/triggers/initializeTrigger.server.ts index a4ec076ba..37c388f45 100644 --- a/apps/webapp/app/services/triggers/initializeTrigger.server.ts +++ b/apps/webapp/app/services/triggers/initializeTrigger.server.ts @@ -49,7 +49,11 @@ export class InitializeTriggerService { }, }); - const clientApi = new EndpointApi(environment.apiKey, endpoint.url); + const clientApi = new EndpointApi( + environment.apiKey, + endpoint.url, + endpoint.slug + ); const registerMetadata = await clientApi.initializeTrigger( dynamicTrigger.slug, diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index b953a46a9..867f2cc05 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -1,7 +1,7 @@ import { z } from "zod"; import { env } from "~/env.server"; import { ZodWorker } from "~/platform/zodWorker.server"; -import { EndpointRegisteredService } from "./endpoints/endpointRegistered.server"; +import { IndexEndpointService } from "./endpoints/indexEndpoint.server"; import { apiAuthenticationRepository } from "./externalApis/apiAuthenticationRepository.server"; import { RegisterJobService } from "./jobs/registerJob.server"; import { StartRunService } from "./runs/startRun.server"; @@ -31,7 +31,12 @@ import { PerformRunExecutionService } from "./runs/performRunExecution"; const workerCatalog = { organizationCreated: z.object({ id: z.string() }), - endpointRegistered: z.object({ id: z.string() }), + indexEndpoint: z.object({ + id: z.string(), + source: z.enum(["MANUAL", "API", "INTERNAL", "HOOK"]).optional(), + sourceData: z.any().optional(), + reason: z.string().optional(), + }), scheduleEmail: DeliverEmailSchema, githubAppInstallationDeleted: z.object({ id: z.string() }), githubPush: z.object({ @@ -285,12 +290,17 @@ function getWorkerQueue() { // TODO: implement }, }, - endpointRegistered: { + indexEndpoint: { queueName: "internal-queue", handler: async (payload, job) => { - const service = new EndpointRegisteredService(); + const service = new IndexEndpointService(); - await service.call(payload.id); + await service.call( + payload.id, + payload.source, + payload.reason, + payload.sourceData + ); }, }, deliverEvent: { diff --git a/apps/webapp/prisma/migrations/20230614103739_add_deploy_hook_identifier_to_endpoints/migration.sql b/apps/webapp/prisma/migrations/20230614103739_add_deploy_hook_identifier_to_endpoints/migration.sql new file mode 100644 index 000000000..d79f97cf1 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230614103739_add_deploy_hook_identifier_to_endpoints/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Endpoint" ADD COLUMN "deployHookIdentifier" TEXT; diff --git a/apps/webapp/prisma/migrations/20230614110359_rename_deploy_to_index/migration.sql b/apps/webapp/prisma/migrations/20230614110359_rename_deploy_to_index/migration.sql new file mode 100644 index 000000000..8ccf40964 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230614110359_rename_deploy_to_index/migration.sql @@ -0,0 +1,10 @@ +/* + Warnings: + + - You are about to drop the column `deployHookIdentifier` on the `Endpoint` table. All the data in the column will be lost. + +*/ +-- AlterTable +ALTER TABLE "Endpoint" DROP COLUMN "deployHookIdentifier", +ADD COLUMN "indexingHookIdentifier" TEXT, +ADD COLUMN "lastIndexedAt" TIMESTAMP(3); diff --git a/apps/webapp/prisma/migrations/20230614122553_create_endpoint_index_model/migration.sql b/apps/webapp/prisma/migrations/20230614122553_create_endpoint_index_model/migration.sql new file mode 100644 index 000000000..d99a9f22a --- /dev/null +++ b/apps/webapp/prisma/migrations/20230614122553_create_endpoint_index_model/migration.sql @@ -0,0 +1,28 @@ +/* + Warnings: + + - You are about to drop the column `lastIndexedAt` on the `Endpoint` table. All the data in the column will be lost. + +*/ +-- CreateEnum +CREATE TYPE "EndpointIndexSource" AS ENUM ('MANUAL', 'ENDPOINT_INITIATED', 'HOOK'); + +-- AlterTable +ALTER TABLE "Endpoint" DROP COLUMN "lastIndexedAt"; + +-- CreateTable +CREATE TABLE "EndpointIndex" ( + "id" TEXT NOT NULL, + "endpointId" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + "source" "EndpointIndexSource" NOT NULL DEFAULT 'MANUAL', + "reason" TEXT, + "data" JSONB NOT NULL, + "stats" JSONB NOT NULL, + + CONSTRAINT "EndpointIndex_pkey" PRIMARY KEY ("id") +); + +-- AddForeignKey +ALTER TABLE "EndpointIndex" ADD CONSTRAINT "EndpointIndex_endpointId_fkey" FOREIGN KEY ("endpointId") REFERENCES "Endpoint"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20230614125945_add_source_data_to_endpoint_index/migration.sql b/apps/webapp/prisma/migrations/20230614125945_add_source_data_to_endpoint_index/migration.sql new file mode 100644 index 000000000..f011778a3 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230614125945_add_source_data_to_endpoint_index/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "EndpointIndex" ADD COLUMN "sourceData" JSONB; diff --git a/apps/webapp/prisma/migrations/20230614135014_change_endpoint_source_enum/migration.sql b/apps/webapp/prisma/migrations/20230614135014_change_endpoint_source_enum/migration.sql new file mode 100644 index 000000000..053e25374 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230614135014_change_endpoint_source_enum/migration.sql @@ -0,0 +1,16 @@ +/* + Warnings: + + - The values [ENDPOINT_INITIATED] on the enum `EndpointIndexSource` will be removed. If these variants are still used in the database, this will fail. + +*/ +-- AlterEnum +BEGIN; +CREATE TYPE "EndpointIndexSource_new" AS ENUM ('MANUAL', 'INTERNAL', 'HOOK'); +ALTER TABLE "EndpointIndex" ALTER COLUMN "source" DROP DEFAULT; +ALTER TABLE "EndpointIndex" ALTER COLUMN "source" TYPE "EndpointIndexSource_new" USING ("source"::text::"EndpointIndexSource_new"); +ALTER TYPE "EndpointIndexSource" RENAME TO "EndpointIndexSource_old"; +ALTER TYPE "EndpointIndexSource_new" RENAME TO "EndpointIndexSource"; +DROP TYPE "EndpointIndexSource_old"; +ALTER TABLE "EndpointIndex" ALTER COLUMN "source" SET DEFAULT 'MANUAL'; +COMMIT; diff --git a/apps/webapp/prisma/migrations/20230614141902_added_api_to_endpoint_index_source/migration.sql b/apps/webapp/prisma/migrations/20230614141902_added_api_to_endpoint_index_source/migration.sql new file mode 100644 index 000000000..872efe8a9 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230614141902_added_api_to_endpoint_index_source/migration.sql @@ -0,0 +1,2 @@ +-- AlterEnum +ALTER TYPE "EndpointIndexSource" ADD VALUE 'API'; diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index 841892bf0..a9eae8b1e 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -291,15 +291,42 @@ model Endpoint { createdAt DateTime @default(now()) updatedAt DateTime @updatedAt + indexingHookIdentifier String? + jobVersions JobVersion[] jobRuns JobRun[] httpRequestDeliveries HttpSourceRequestDelivery[] dynamictriggers DynamicTrigger[] sources TriggerSource[] + indexings EndpointIndex[] @@unique([environmentId, slug]) } +model EndpointIndex { + id String @id @default(cuid()) + + endpoint Endpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade, onUpdate: Cascade) + endpointId String + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + source EndpointIndexSource @default(MANUAL) + sourceData Json? + reason String? + + data Json + stats Json +} + +enum EndpointIndexSource { + MANUAL + API + INTERNAL + HOOK +} + model Job { id String @id @default(cuid()) slug String diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index d248b0a29..e88472f0c 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -116,10 +116,22 @@ export type HttpSourceRequestHeaders = z.output< typeof HttpSourceRequestHeadersSchema >; -export const PongResponseSchema = z.object({ - message: z.literal("PONG"), +export const PongSuccessResponseSchema = z.object({ + ok: z.literal(true), }); +export const PongErrorResponseSchema = z.object({ + ok: z.literal(false), + error: z.string(), +}); + +export const PongResponseSchema = z.discriminatedUnion("ok", [ + PongSuccessResponseSchema, + PongErrorResponseSchema, +]); + +export type PongResponse = z.infer; + export const QueueOptionsSchema = z.object({ name: z.string(), maxConcurrent: z.number().optional(), @@ -162,16 +174,14 @@ export type DynamicTriggerEndpointMetadata = z.infer< typeof DynamicTriggerEndpointMetadataSchema >; -export const GetEndpointDataResponseSchema = z.object({ +export const IndexEndpointResponseSchema = z.object({ jobs: z.array(JobMetadataSchema), sources: z.array(SourceMetadataSchema), dynamicTriggers: z.array(DynamicTriggerEndpointMetadataSchema), dynamicSchedules: z.array(RegisterDynamicSchedulePayloadSchema), }); -export type GetEndpointDataResponse = z.infer< - typeof GetEndpointDataResponseSchema ->; +export type IndexEndpointResponse = z.infer; export const RawEventSchema = z.object({ id: z.string().default(() => ulid()), diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 39a4c9df9..a23689450 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -1,6 +1,6 @@ import { ErrorWithStackSchema, - GetEndpointDataResponse, + IndexEndpointResponse, HandleTriggerSource, HttpSourceRequestHeadersSchema, InitializeTriggerBodySchema, @@ -142,14 +142,36 @@ export class TriggerClient { switch (action) { case "PING": { + const endpointId = request.headers.get("x-trigger-endpoint-id"); + + if (!endpointId) { + return { + status: 200, + body: { + ok: false, + message: "Missing endpoint ID", + }, + }; + } + + if (this.id !== endpointId) { + return { + status: 200, + body: { + ok: false, + message: `Endpoint ID mismatch error. Expected ${this.id}, got ${endpointId}`, + }, + }; + } + return { status: 200, body: { - message: "PONG", + ok: true, }, }; } - case "GET_ENDPOINT_DATA": { + case "INDEX_ENDPOINT": { // if the x-trigger-job-id header is set, we return the job with that id const jobId = request.headers.get("x-trigger-job-id"); @@ -171,7 +193,7 @@ export class TriggerClient { }; } - const body: GetEndpointDataResponse = { + const body: IndexEndpointResponse = { jobs: Object.values(this.#registeredJobs).map((job) => job.toJSON()), sources: Object.values(this.#registeredSources), dynamicTriggers: Object.values(this.#registeredDynamicTriggers).map( @@ -194,16 +216,6 @@ export class TriggerClient { body, }; } - case "INITIALIZE": { - await this.listen(); - - return { - status: 200, - body: { - message: "Initialized", - }, - }; - } case "INITIALIZE_TRIGGER": { const json = await request.json(); const body = InitializeTriggerBodySchema.safeParse(json); @@ -527,14 +539,6 @@ export class TriggerClient { return this.#options.apiKey ?? process.env.TRIGGER_API_KEY; } - async listen() { - // Register the endpoint - await this.#client.registerEndpoint({ - url: this.url, - name: this.id, - }); - } - async #preprocessRun( body: PreprocessRunBody, job: Job>, any>