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

@trigger.dev/supabase: You can now trigger on multiple database events in the same job

Browse Source
This commit is contained in:
Eric Allam
2023-08-07 17:27:49 +01:00
parent 4ca758a3de
commit 3cf3eaff48
5 changed files with 314 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/mean-bags-check.md
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/supabase": patch
---
You can now trigger on multiple database events in the same job
docs/integrations/apis/supabase/client.mdx
+7 -7
View File
@@ -38,8 +38,8 @@ client.defineJob({
supabase,
},
run: async (payload, io, ctx) => {
const { data: users, error } = await io.supabase.runTask("find-users", async (db) => {
return db.from("users").select("*");
const { data: todos, error } = await io.supabase.runTask("find-todos", async (db) => {
return db.from("todos").select("*");
});
},
});
@@ -60,8 +60,8 @@ client.defineJob({
supabase,
},
run: async (payload, io, ctx) => {
const users = await io.supabase.runTask("find-users", async (db) => {
const { data, error } = await db.from("users").select("*");
const todos = await io.supabase.runTask("find-todos", async (db) => {
const { data, error } = await db.from("todos").select("*");
if (error) throw error;
@@ -100,15 +100,15 @@ client.defineJob({
supabase,
},
run: async (payload, io, ctx) => {
const users = await io.supabase.runTask("find-users", async (db) => {
const { data, error } = await db.from("users").select("*");
const todos = await io.supabase.runTask("find-todos", async (db) => {
const { data, error } = await db.from("todos").select("*");
if (error) throw error;
return data;
});
// users is now typed as User[] instead of any[]
// todos is now typed as Todo[] instead of any[]
},
});
```
docs/integrations/apis/supabase/management.mdx
+28 -17
View File
@@ -126,7 +126,7 @@ client.defineJob({
id: "supabase-trigger",
name: "Supabase Trigger",
trigger: db.onInserted({
table: "users",
table: "todos",
}),
run: async (payload, io, ctx) => {
// payload is the database webhook body (see https://supabase.com/docs/guides/database/webhooks#payload)
@@ -137,20 +137,6 @@ client.defineJob({
You can add additional filters to the trigger by passing a `filter` object:
```ts
client.defineJob({
id: "supabase-trigger",
name: "Supabase Trigger",
trigger: db.onUpdated({
table: "users",
filter: {
country: ["USA", "Canada"], // This will only trigger the job if the user.country is USA or Canada
},
}),
run: async (payload, io, ctx) => {
// payload is the database webhook body (see https://supabase.com/docs/guides/database/webhooks#payload)
},
});
client.defineJob({
id: "supabase-trigger",
name: "Supabase Trigger",
@@ -172,6 +158,31 @@ client.defineJob({
});
```
You can also listen for multiple different events using the `on` trigger:
```ts
client.defineJob({
id: "supabase-trigger",
name: "Supabase Trigger",
trigger: db.on({
table: "todos",
events: ["INSERT", "UPDATE"] // Trigger on both insert and update events
filter: {
record: {
is_completed: [false],
},
},
}),
run: async (payload, io, ctx) => {
if (payload.type === "INSERT") {
// payload will be typed as the INSERT payload
} else {
// payload will be typed as the UPDATE payload
}
},
});
```
<Note>
We will only create at most 1 database webhook per table, to limit resource usage when writing to
your database. This means we cannot support scoping updated triggers to specific columns.
@@ -196,10 +207,10 @@ client.defineJob({
id: "supabase-trigger",
name: "Supabase Trigger",
trigger: db.onUpdated({
table: "users",
table: "todos",
}),
run: async (payload, io, ctx) => {
// payload.record and payload.old_record are now correctly typed to match the users table
// payload.record and payload.old_record are now correctly typed to match the todos table
},
});
```
examples/job-catalog/src/supabase.ts
+91 -8
View File
@@ -1,14 +1,19 @@
import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
import { TriggerClient } from "@trigger.dev/sdk";
import { createExpressServer } from "@trigger.dev/express";
import { z } from "zod";
import { SupabaseManagement } from "@trigger.dev/supabase";
import { Supabase, SupabaseManagement } from "@trigger.dev/supabase";
const supabaseManagement = new SupabaseManagement({
id: "supabase-management",
apiKey: process.env["SUPABASE_API_KEY"]!,
});
const db = supabaseManagement.db(process.env["SUPABASE_ID"]!);
const triggers = supabaseManagement.db<Database>(process.env["SUPABASE_ID"]!);
const supabase = new Supabase({
id: "supabase",
supabaseKey: process.env["SUPABASE_SERVICE_ROLE_KEY"]!,
supabaseUrl: process.env["SUPABASE_URL"]!,
});
export const client = new TriggerClient({
id: "job-catalog",
@@ -24,8 +29,8 @@ client.defineJob({
id: "supabase-management-example-1",
name: "Supabase Management Example 1",
version: "0.1.0",
trigger: db.onInserted({
table: "users",
trigger: triggers.onInserted({
table: "todos",
}),
run: async (payload, io, ctx) => {},
});
@@ -34,8 +39,86 @@ client.defineJob({
id: "supabase-management-example-2",
name: "Supabase Management Example 2",
version: "0.1.0",
trigger: db.onUpdated({
table: "users",
trigger: triggers.onUpdated({
table: "todos",
}),
run: async (payload, io, ctx) => {},
});
client.defineJob({
id: "supabase-management-example-on",
name: "Supabase Management Example On",
version: "0.1.0",
trigger: triggers.on({
table: "todos",
events: ["INSERT", "UPDATE"],
}),
integrations: {
supabase,
},
run: async (payload, io, ctx) => {
const user = await io.supabase.runTask("fetch-user", async (db) => {
const { data, error } = await db.auth.admin.getUserById(payload.record.user_id);
if (error) {
throw error;
}
return data.user;
});
return user;
},
});
export type Json = string | number | boolean | null | { [key: string]: Json | undefined } | Json[];
export interface Database {
public: {
Tables: {
todos: {
Row: {
id: number;
inserted_at: string;
is_complete: boolean | null;
task: string | null;
user_id: string;
};
Insert: {
id?: number;
inserted_at?: string;
is_complete?: boolean | null;
task?: string | null;
user_id: string;
};
Update: {
id?: number;
inserted_at?: string;
is_complete?: boolean | null;
task?: string | null;
user_id?: string;
};
Relationships: [
{
foreignKeyName: "todos_user_id_fkey";
columns: ["user_id"];
referencedRelation: "users";
referencedColumns: ["id"];
},
];
};
};
Views: {
[_ in never]: never;
};
Functions: {
[_ in never]: never;
};
Enums: {
[_ in never]: never;
};
CompositeTypes: {
[_ in never]: never;
};
};
}
integrations/supabase/src/management/index.ts
+183 -2
View File
@@ -34,6 +34,91 @@ class SupabaseDatabase<Database = any> {
private projectRef: string
) {}
/**
* The function `on` creates a trigger for when a record is inserted, updated, or deleted on a
* specific table in a database schema.
* @param params - The `params` parameter is an object that contains the following properties:
* @param params.table - The `table` property is a string that specifies the name of the table
* that the trigger will be created for.
* @param params.events - The `events` property is an array of events that specifies the events
* that the trigger will be called for. The events that can be specified are `INSERT`, `UPDATE`, or `DELETE`.
* By default, the trigger will be called for all events.
* @param params.schema - The `schema` property is a string that specifies the name of the schema
* that the trigger will be created for. If the schema is not specified, the default schema will
* be used. (public)
* @param params.filter - The `filter` property is an object that specifies the filter that will
* be used to determine if the trigger should be called. If the filter is not specified, the
* trigger will be called for all records.
*
* @example
*
* ```ts
* const supabase = new SupabaseManagement({ id: "supabase" });
* const database = supabase.database<Database>("https://<project-id>.supabase.co");
*
* client.defineJob({
* trigger: database.on({
* table: "todos",
* events: ["INSERTED", "UPDATED"],
* schema: "public",
* filter: {
* record: { is_completed: [false] },
* },
* }),
* })
* ```
*/
on<
SchemaName extends string & keyof Database = "public" extends keyof Database
? "public"
: string & keyof Database,
Schema extends GenericSchema = Database[SchemaName] extends GenericSchema
? Database[SchemaName]
: any,
TTableName extends string & keyof Schema["Tables"] = string & keyof Schema["Tables"],
TTable extends Schema["Tables"][TTableName] = Schema["Tables"][TTableName],
TEvents extends WebhookEvents[] = ["INSERT", "UPDATE", "DELETE"],
>(params: { table: TTableName; events?: TEvents; schema?: SchemaName; filter?: EventFilter }) {
return createTrigger<Prettify<UnionPayloads<TEvents, TTableName, SchemaName, TTable["Row"]>>>(
this.integration.source,
{
event: params.events ?? ["INSERT", "UPDATE", "DELETE"],
projectRef: this.projectRef,
...params,
}
);
}
/**
* The function `onInserted` creates a trigger for when a new record is inserted into a specific
* table in a database schema.
* @param params - The `params` parameter is an object that contains the following properties:
* @param params.table - The `table` property is a string that specifies the name of the table
* that the trigger will be created for.
* @param params.schema - The `schema` property is a string that specifies the name of the schema
* that the trigger will be created for. If the schema is not specified, the default schema will
* be used. (public)
* @param params.filter - The `filter` property is an object that specifies the filter that will
* be used to determine if the trigger should be called. If the filter is not specified, the
* trigger will be called for all records.
*
* @example
*
* ```ts
* const supabase = new SupabaseManagement({ id: "supabase" });
* const database = supabase.database<Database>("https://<project-id>.supabase.co");
*
* client.defineJob({
* trigger: database.onInserted({
* table: "todos",
* schema: "public",
* filter: {
* record: { is_completed: [false] },
* },
* }),
* })
* ```
*/
onInserted<
SchemaName extends string & keyof Database = "public" extends keyof Database
? "public"
@@ -57,6 +142,37 @@ class SupabaseDatabase<Database = any> {
});
}
/**
* The function `onUpdated` creates a trigger for when a new record is updated on a specific
* table in a database schema.
* @param params - The `params` parameter is an object that contains the following properties:
* @param params.table - The `table` property is a string that specifies the name of the table
* that the trigger will be created for.
* @param params.schema - The `schema` property is a string that specifies the name of the schema
* that the trigger will be created for. If the schema is not specified, the default schema will
* be used. (public)
* @param params.filter - The `filter` property is an object that specifies the filter that will
* be used to determine if the trigger should be called. If the filter is not specified, the
* trigger will be called for all records.
*
* @example
*
* ```ts
* const supabase = new SupabaseManagement({ id: "supabase" });
* const database = supabase.database<Database>("https://<project-id>.supabase.co");
*
* client.defineJob({
* trigger: database.onUpdated({
* table: "todos",
* schema: "public",
* filter: {
* record: { completed: [true] },
* old_record: { completed: [false] },
* },
* }),
* })
* ```
*/
onUpdated<
SchemaName extends string & keyof Database = "public" extends keyof Database
? "public"
@@ -80,6 +196,36 @@ class SupabaseDatabase<Database = any> {
});
}
/**
* The function `onDeleted` creates a trigger for when a new record is deleted from a specific
* table in a database schema.
* @param params - The `params` parameter is an object that contains the following properties:
* @param params.table - The `table` property is a string that specifies the name of the table
* that the trigger will be created for.
* @param params.schema - The `schema` property is a string that specifies the name of the schema
* that the trigger will be created for. If the schema is not specified, the default schema will
* be used. (public)
* @param params.filter - The `filter` property is an object that specifies the filter that will
* be used to determine if the trigger should be called. If the filter is not specified, the
* trigger will be called for all records.
*
* @example
*
* ```ts
* const supabase = new SupabaseManagement({ id: "supabase" });
* const database = supabase.database<Database>("https://<project-id>.supabase.co");
*
* client.defineJob({
* trigger: database.onDeleted({
* table: "todos",
* schema: "public",
* filter: {
* old_record: { is_completed: [true] },
* },
* }),
* })
* ```
*/
onDeleted<
SchemaName extends string & keyof Database = "public" extends keyof Database
? "public"
@@ -175,9 +321,44 @@ type WebhookEventSource = ReturnType<typeof createWebhookEventSource>;
type WebhookEvents = "INSERT" | "UPDATE" | "DELETE";
type WebhookEventPayloads<
TTableName extends string,
TSchemaName extends string = "public",
TRecord = any,
> = {
INSERT: {
table: TTableName;
record: Prettify<TRecord>;
type: "INSERT";
schema: TSchemaName;
old_record: null;
};
UPDATE: {
table: TTableName;
record: Prettify<TRecord>;
type: "UPDATE";
schema: TSchemaName;
old_record: Prettify<TRecord>;
};
DELETE: {
table: TTableName;
record: null;
type: "DELETE";
schema: TSchemaName;
old_record: Prettify<TRecord>;
};
};
type UnionPayloads<
T extends WebhookEvents[],
TTableName extends string,
TSchemaName extends string = "public",
TRecord = any,
> = WebhookEventPayloads<TTableName, TSchemaName, TRecord>[T[number]];
function createTrigger<TEvent extends any>(
source: WebhookEventSource,
params: { event: WebhookEvents; filter?: EventFilter } & {
params: { event: WebhookEvents | WebhookEvents[]; filter?: EventFilter } & {
projectRef: string;
table: string;
schema?: string;
@@ -190,7 +371,7 @@ function createTrigger<TEvent extends any>(
icon: "supabase",
filter: {
...params.filter,
type: [params.event],
type: typeof params.event === "string" ? [params.event] : params.event,
schema: [params.schema ?? "public"],
},
properties: [],