Add connection identity backend support

apps/cloud/drizzle/0020_add_connection_identity_override.sql

@@ -0,0 +1 @@
+ALTER TABLE "connection" ADD COLUMN IF NOT EXISTS "identity_override" json;

apps/cloud/drizzle/meta/_journal.json

@@ -141,6 +141,13 @@
       "when": 1780081200000,
       "tag": "0019_workos_vault_plugin_storage",
       "breakpoints": true
+    },
+    {
+      "idx": 20,
+      "version": "7",
+      "when": 1780124534904,
+      "tag": "0020_add_connection_identity_override",
+      "breakpoints": true
     }
   ]
 }

apps/cloud/src/services/db.schema.test.ts

@@ -18,9 +18,36 @@ import { drizzle } from "drizzle-orm/postgres-js";
 import { Effect } from "effect";
 import postgres from "postgres";
 
+import { collectTables } from "@executor-js/sdk";
+
+import executorConfig from "../../executor.config";
 import * as cloudSchema from "./schema";
 import * as executorSchema from "./executor-schema";
 import { combinedSchema } from "./db";
+import { createDrizzleFumaDb } from "./fuma";
+
+interface GeneratedFumaTable {
+  readonly names: {
+    readonly drizzle: string;
+  };
+  readonly columns: Record<
+    string,
+    {
+      readonly ormName: string;
+      readonly names: {
+        readonly drizzle: string;
+      };
+    }
+  >;
+}
+
+const fumaDbInternals = (
+  value: unknown,
+): { internal: { tables: Record<string, GeneratedFumaTable> } } =>
+  value as { internal: { tables: Record<string, GeneratedFumaTable> } };
+
+const drizzleTableColumns = (value: unknown): Record<string, unknown> =>
+  value as Record<string, unknown>;
 
 describe("combinedSchema", () => {
   it("spreads every cloud + executor schema export", () => {
@@ -64,4 +91,38 @@ describe("combinedSchema", () => {
       ),
     ),
   );
+
+  it.effect("generated drizzle tables expose every executor Fuma column", () =>
+    Effect.acquireRelease(
+      Effect.sync(() => postgres("postgres://u:p@127.0.0.1:1/x", { max: 1 })),
+      (sql) => Effect.promise(() => sql.end({ timeout: 0 })),
+    ).pipe(
+      Effect.flatMap((sql) =>
+        Effect.sync(() => {
+          const db = drizzle(sql, { schema: combinedSchema });
+          const fuma = createDrizzleFumaDb({
+            db,
+            tables: collectTables(executorConfig.plugins({})),
+            namespace: "executor_cloud",
+            provider: "postgresql",
+          });
+          const schemaTables = drizzleTableColumns(combinedSchema);
+          const missingColumns: string[] = [];
+
+          for (const table of Object.values(fumaDbInternals(fuma.db).internal.tables)) {
+            const drizzleTable = drizzleTableColumns(schemaTables[table.names.drizzle]);
+            for (const column of Object.values(table.columns)) {
+              if (drizzleTable[column.names.drizzle] === undefined) {
+                missingColumns.push(
+                  `${table.names.drizzle}.${column.ormName} -> ${column.names.drizzle}`,
+                );
+              }
+            }
+          }
+
+          expect(missingColumns).toEqual([]);
+        }),
+      ),
+    ),
+  );
 });

apps/cloud/src/services/executor-schema.ts

@@ -1,10 +1,7 @@
-import { pgTable, varchar, text, boolean, timestamp, uniqueIndex, json, bigint } from "drizzle-orm/pg-core"
+import { pgTable, text, boolean, timestamp, varchar, uniqueIndex, json, bigint } from "drizzle-orm/pg-core"
 import { createId } from "fumadb/cuid"
 
 export const source = pgTable("source", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   kind: text("kind").notNull(),
   name: text("name").notNull(),
@@ -13,89 +10,90 @@ export const source = pgTable("source", {
   can_refresh: boolean("can_refresh").notNull().default(false),
   can_edit: boolean("can_edit").notNull().default(false),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("source_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const tool = pgTable("tool", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   source_id: text("source_id").notNull(),
   plugin_id: text("plugin_id").notNull(),
   name: text("name").notNull(),
   description: text("description").notNull(),
   input_schema: json("input_schema"),
   output_schema: json("output_schema"),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("tool_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const definition = pgTable("definition", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   source_id: text("source_id").notNull(),
   plugin_id: text("plugin_id").notNull(),
   name: text("name").notNull(),
   schema: json("schema").notNull(),
-  created_at: timestamp("created_at").notNull()
+  created_at: timestamp("created_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("definition_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const secret = pgTable("secret", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   name: text("name").notNull(),
   provider: text("provider").notNull(),
   owned_by_connection_id: text("owned_by_connection_id"),
-  created_at: timestamp("created_at").notNull()
+  created_at: timestamp("created_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("secret_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const connection = pgTable("connection", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   provider: text("provider").notNull(),
   identity_label: text("identity_label"),
   access_token_secret_id: text("access_token_secret_id").notNull(),
   refresh_token_secret_id: text("refresh_token_secret_id"),
   expires_at: bigint("expires_at", { mode: "bigint" }),
   scope: text("scope"),
   provider_state: json("provider_state"),
+  identity_override: json("identity_override"),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("connection_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const oauth2_session = pgTable("oauth2_session", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   strategy: text("strategy").notNull(),
   connection_id: text("connection_id").notNull(),
   token_scope: text("token_scope").notNull(),
   redirect_url: text("redirect_url").notNull(),
   payload: json("payload").notNull(),
   expires_at: bigint("expires_at", { mode: "bigint" }).notNull(),
-  created_at: timestamp("created_at").notNull()
+  created_at: timestamp("created_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("oauth2_session_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const credential_binding = pgTable("credential_binding", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   source_id: text("source_id").notNull(),
   source_scope_id: text("source_scope_id").notNull(),
@@ -106,49 +104,52 @@ export const credential_binding = pgTable("credential_binding", {
   secret_scope_id: text("secret_scope_id"),
   connection_id: text("connection_id"),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("credential_binding_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const plugin_storage = pgTable("plugin_storage", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   plugin_id: text("plugin_id").notNull(),
   collection: text("collection").notNull(),
   key: text("key").notNull(),
   data: json("data").notNull(),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("plugin_storage_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const tool_policy = pgTable("tool_policy", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
-  scope_id: varchar("scope_id", { length: 255 }).notNull(),
   pattern: text("pattern").notNull(),
   action: text("action").notNull(),
   position: text("position").notNull(),
   created_at: timestamp("created_at").notNull(),
-  updated_at: timestamp("updated_at").notNull()
+  updated_at: timestamp("updated_at").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull(),
+  scope_id: varchar("scope_id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("tool_policy_scope_id_id_uidx").on(table.scope_id, table.id)
 ])
 
 export const blob = pgTable("blob", {
-  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
-  id: varchar("id", { length: 255 }).notNull(),
   namespace: text("namespace").notNull(),
   key: text("key").notNull(),
-  value: text("value").notNull()
+  value: text("value").notNull(),
+  row_id: varchar("row_id", { length: 255 }).primaryKey().notNull().$defaultFn(() => createId()),
+  id: varchar("id", { length: 255 }).notNull()
 }, (table) => [
   uniqueIndex("blob_id_uidx").on(table.id)
 ])
 
 export const private_executor_cloud_settings = pgTable("private_executor_cloud_settings", {
   id: varchar("id", { length: 255 }).primaryKey().notNull(),
   version: varchar("version", { length: 255 }).notNull().default("1.0.0")
-})
+})

apps/local/drizzle/0012_connection_identity_override.sql

@@ -0,0 +1 @@
+ALTER TABLE `connection` ADD `identity_override` text;

apps/local/drizzle/meta/_journal.json

@@ -85,6 +85,13 @@
       "when": 1779087600000,
       "tag": "0011_plugin_storage_sources",
       "breakpoints": true
+    },
+    {
+      "idx": 12,
+      "version": "6",
+      "when": 1779998400000,
+      "tag": "0012_connection_identity_override",
+      "breakpoints": true
     }
   ]
 }

apps/local/src/server/executor-schema.ts

@@ -93,6 +93,7 @@ export const connection = sqliteTable(
     expires_at: integer("expires_at"),
     scope: text("scope"),
     provider_state: text("provider_state", { mode: "json" }),
+    identity_override: text("identity_override", { mode: "json" }),
     created_at: integer("created_at", { mode: "timestamp_ms" }).notNull(),
     updated_at: integer("updated_at", { mode: "timestamp_ms" }).notNull(),
   },

apps/local/src/server/sqlite-fumadb.ts

@@ -53,6 +53,15 @@ export const createSqliteFumaDb = async <const TTables extends FumaTables>(
   })) {
     sqlite.exec(statement);
   }
+  const connectionColumns = sqlite
+    .prepare("PRAGMA table_info('connection')")
+    .all() as ReadonlyArray<{ readonly name: string }>;
+  if (
+    connectionColumns.length > 0 &&
+    !connectionColumns.some((column) => column.name === "identity_override")
+  ) {
+    sqlite.exec("ALTER TABLE connection ADD COLUMN identity_override TEXT");
+  }
 
   const latestSchema = fumaSchema({
     version,

packages/core/api/src/connections/api.ts

@@ -3,7 +3,9 @@ import { Schema } from "effect";
 
 import {
   ConnectionId,
+  ConnectionIdentityOverride,
   ConnectionInUseError,
+  ConnectionNotFoundError,
   InternalError,
   ScopeId,
   Usage,
@@ -27,15 +29,34 @@ const ConnectionRefResponse = Schema.Struct({
   identityLabel: Schema.NullOr(Schema.String),
   expiresAt: Schema.NullOr(Schema.Number),
   oauthScope: Schema.NullOr(Schema.String),
+  identityOverride: Schema.NullOr(ConnectionIdentityOverride),
   createdAt: Schema.Number,
   updatedAt: Schema.Number,
 });
 
+export const ConnectionIdentityResponse = Schema.Struct({
+  status: Schema.Literals(["available", "unavailable", "reauth_required", "error"]),
+  source: Schema.Literals(["detected", "manual", "mixed", "unknown"]),
+  subject: Schema.NullOr(Schema.String),
+  email: Schema.NullOr(Schema.String),
+  emailVerified: Schema.NullOr(Schema.Boolean),
+  name: Schema.NullOr(Schema.String),
+  username: Schema.NullOr(Schema.String),
+  picture: Schema.NullOr(Schema.String),
+  message: Schema.NullOr(Schema.String),
+});
+export type ConnectionIdentityResponse = typeof ConnectionIdentityResponse.Type;
+
+const UpdateConnectionIdentityPayload = Schema.Struct({
+  identityOverride: Schema.NullOr(ConnectionIdentityOverride),
+});
+
 // ---------------------------------------------------------------------------
 // Group
 // ---------------------------------------------------------------------------
 
 const ConnectionInUse = ConnectionInUseError.annotate({ httpApiStatus: 409 });
+const ConnectionNotFound = ConnectionNotFoundError.annotate({ httpApiStatus: 404 });
 
 export const ConnectionsApi = HttpApiGroup.make("connections")
   .add(
@@ -58,4 +79,19 @@ export const ConnectionsApi = HttpApiGroup.make("connections")
       success: Schema.Array(Usage),
       error: InternalError,
     }),
+  )
+  .add(
+    HttpApiEndpoint.get("identity", "/scopes/:scopeId/connections/:connectionId/identity", {
+      params: ConnectionParams,
+      success: ConnectionIdentityResponse,
+      error: InternalError,
+    }),
+  )
+  .add(
+    HttpApiEndpoint.patch("updateIdentity", "/scopes/:scopeId/connections/:connectionId/identity", {
+      params: ConnectionParams,
+      payload: UpdateConnectionIdentityPayload,
+      success: ConnectionRefResponse,
+      error: [InternalError, ConnectionNotFound],
+    }),
   );