github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Endpoint registration and indexing now is only initiated outside of clients

Browse Source
This commit is contained in:
Eric Allam
2023-06-14 16:07:07 +01:00
parent e4837cea9d
commit 817b4ed133
21 changed files with 661 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/four-ligers-perform.md
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/sdk": patch
---
Endpoint registration and indexing now is only initiated outside of clients
apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts
+73
View File
@@ -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);
}
apps/webapp/app/routes/api.v1.endpoints.$environmentId.$endpointSlug.index.$indexHookIdentifier.ts
+218
View File
@@ -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<typeof ParamsSchema> & {
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(),
});
apps/webapp/app/routes/api.v1.endpoints.ts
+2 -2
View File
@@ -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);
apps/webapp/app/services/endpointApi.ts
+51 -50
View File
@@ -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<PongResponse> {
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<RegisterTriggerBody | undefined> {
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}`
);
}
apps/webapp/app/services/endpoints/createEndpoint.server.ts
+21 -10
View File
@@ -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 }
);
apps/webapp/app/services/endpoints/endpointRegistered.server.ts
-87
View File
@@ -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,
}
);
}
}
}
apps/webapp/app/services/endpoints/indexEndpoint.server.ts
+130
View File
@@ -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,
},
});
});
}
}
apps/webapp/app/services/runs/performRunExecution.ts
+10 -2
View File
@@ -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();
apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts
+2 -1
View File
@@ -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({
apps/webapp/app/services/triggers/initializeTrigger.server.ts
+5 -1
View File
@@ -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,
apps/webapp/app/services/worker.server.ts
+15 -5
View File
@@ -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: {
apps/webapp/prisma/migrations/20230614103739_add_deploy_hook_identifier_to_endpoints/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "Endpoint" ADD COLUMN "deployHookIdentifier" TEXT;
apps/webapp/prisma/migrations/20230614110359_rename_deploy_to_index/migration.sql
+10
View File
@@ -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);
apps/webapp/prisma/migrations/20230614122553_create_endpoint_index_model/migration.sql
+28
View File
@@ -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;
apps/webapp/prisma/migrations/20230614125945_add_source_data_to_endpoint_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "EndpointIndex" ADD COLUMN "sourceData" JSONB;
apps/webapp/prisma/migrations/20230614135014_change_endpoint_source_enum/migration.sql
+16
View File
@@ -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;
apps/webapp/prisma/migrations/20230614141902_added_api_to_endpoint_index_source/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "EndpointIndexSource" ADD VALUE 'API';
apps/webapp/prisma/schema.prisma
+27
View File
@@ -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
packages/internal/src/schemas/api.ts
+16 -6
View File
@@ -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<typeof PongResponseSchema>;
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<typeof IndexEndpointResponseSchema>;
export const RawEventSchema = z.object({
id: z.string().default(() => ulid()),
packages/trigger-sdk/src/triggerClient.ts
+26 -22
View File
@@ -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<Trigger<EventSpecification<any>>, any>