From fd59c1810d211e0af75420eaca72bd0bb8713e7a Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Mon, 14 Sep 2026 18:45:36 -0700 Subject: [PATCH] Add MemberDirectory seam and cloud membership mirror tables --- .../drizzle/0018_member_directory_mirror.sql | 30 + apps/cloud/drizzle/meta/0018_snapshot.json | 1724 +++++++++++++++++ apps/cloud/drizzle/meta/_journal.json | 7 + .../account/org-api-key-revoke.node.test.ts | 17 +- apps/cloud/src/api/layers.ts | 4 +- .../api/protected-api-key-auth.node.test.ts | 17 +- .../src/api/protected-jwt-auth.node.test.ts | 17 +- apps/cloud/src/auth/member-directory.ts | 181 ++ .../src/auth/org-api-key-auth.node.test.ts | 17 +- .../src/auth/org-selector-auth.node.test.ts | 17 +- .../auth/workos-callback-state.node.test.ts | 17 +- .../cloud/src/auth/workos-mirror.node.test.ts | 701 +++++++ apps/cloud/src/auth/workos-mirror.ts | 519 +++++ apps/cloud/src/db/org-deletion.test.ts | 2 + apps/cloud/src/db/schema.ts | 133 +- .../src/extensions/billing/route.node.test.ts | 17 +- .../src/auth/member-directory.test.ts | 134 ++ .../src/auth/member-directory.ts | 223 +++ .../core/api/src/admin/member-directory.ts | 50 + packages/core/api/src/server.ts | 11 + .../core/api/src/server/member-directory.ts | 142 ++ 21 files changed, 3952 insertions(+), 28 deletions(-) create mode 100644 apps/cloud/drizzle/0018_member_directory_mirror.sql create mode 100644 apps/cloud/drizzle/meta/0018_snapshot.json create mode 100644 apps/cloud/src/auth/member-directory.ts create mode 100644 apps/cloud/src/auth/workos-mirror.node.test.ts create mode 100644 apps/cloud/src/auth/workos-mirror.ts create mode 100644 apps/host-selfhost/src/auth/member-directory.test.ts create mode 100644 apps/host-selfhost/src/auth/member-directory.ts create mode 100644 packages/core/api/src/admin/member-directory.ts create mode 100644 packages/core/api/src/server/member-directory.ts diff --git a/apps/cloud/drizzle/0018_member_directory_mirror.sql b/apps/cloud/drizzle/0018_member_directory_mirror.sql new file mode 100644 index 0000000000..21a08b7147 --- /dev/null +++ b/apps/cloud/drizzle/0018_member_directory_mirror.sql @@ -0,0 +1,30 @@ +CREATE TABLE "membership_tombstones" ( + "membership_id" text PRIMARY KEY NOT NULL, + "account_id" text NOT NULL, + "organization_id" text NOT NULL, + "deleted_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +CREATE TABLE "workos_sync" ( + "id" text PRIMARY KEY NOT NULL, + "cursor" text, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +ALTER TABLE "accounts" ADD COLUMN "email" text;--> statement-breakpoint +ALTER TABLE "accounts" ADD COLUMN "first_name" text;--> statement-breakpoint +ALTER TABLE "accounts" ADD COLUMN "last_name" text;--> statement-breakpoint +ALTER TABLE "accounts" ADD COLUMN "avatar_url" text;--> statement-breakpoint +ALTER TABLE "accounts" ADD COLUMN "workos_updated_at" timestamp with time zone;--> statement-breakpoint +ALTER TABLE "accounts" ADD COLUMN "last_sign_in_at" timestamp with time zone;--> statement-breakpoint +ALTER TABLE "memberships" ADD COLUMN "membership_id" text;--> statement-breakpoint +ALTER TABLE "memberships" ADD COLUMN "role" text DEFAULT 'member' NOT NULL;--> statement-breakpoint +ALTER TABLE "memberships" ADD COLUMN "status" text DEFAULT 'active' NOT NULL;--> statement-breakpoint +ALTER TABLE "memberships" ADD COLUMN "workos_updated_at" timestamp with time zone;--> statement-breakpoint +ALTER TABLE "memberships" ADD COLUMN "deleted_at" timestamp with time zone;--> statement-breakpoint +ALTER TABLE "membership_tombstones" ADD CONSTRAINT "membership_tombstones_account_id_accounts_id_fk" FOREIGN KEY ("account_id") REFERENCES "public"."accounts"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "membership_tombstones" ADD CONSTRAINT "membership_tombstones_organization_id_organizations_id_fk" FOREIGN KEY ("organization_id") REFERENCES "public"."organizations"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +CREATE INDEX "membership_tombstones_organization_id_idx" ON "membership_tombstones" USING btree ("organization_id");--> statement-breakpoint +CREATE INDEX "accounts_email_lower_idx" ON "accounts" USING btree (lower("email"));--> statement-breakpoint +CREATE UNIQUE INDEX "memberships_membership_id_unique" ON "memberships" USING btree ("membership_id");--> statement-breakpoint +CREATE INDEX "memberships_organization_id_idx" ON "memberships" USING btree ("organization_id"); \ No newline at end of file diff --git a/apps/cloud/drizzle/meta/0018_snapshot.json b/apps/cloud/drizzle/meta/0018_snapshot.json new file mode 100644 index 0000000000..5f29623ce9 --- /dev/null +++ b/apps/cloud/drizzle/meta/0018_snapshot.json @@ -0,0 +1,1724 @@ +{ + "id": "fe5ddc71-31c5-4144-a879-11ea71a63735", + "prevId": "42251aa3-ae24-4010-ac65-9f41e26cdc20", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.accounts": { + "name": "accounts", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "first_name": { + "name": "first_name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "last_name": { + "name": "last_name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "avatar_url": { + "name": "avatar_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "workos_updated_at": { + "name": "workos_updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "last_sign_in_at": { + "name": "last_sign_in_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "accounts_email_lower_idx": { + "name": "accounts_email_lower_idx", + "columns": [ + { + "expression": "lower(\"email\")", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.membership_tombstones": { + "name": "membership_tombstones", + "schema": "", + "columns": { + "membership_id": { + "name": "membership_id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "membership_tombstones_organization_id_idx": { + "name": "membership_tombstones_organization_id_idx", + "columns": [ + { + "expression": "organization_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "membership_tombstones_account_id_accounts_id_fk": { + "name": "membership_tombstones_account_id_accounts_id_fk", + "tableFrom": "membership_tombstones", + "tableTo": "accounts", + "columnsFrom": ["account_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "membership_tombstones_organization_id_organizations_id_fk": { + "name": "membership_tombstones_organization_id_organizations_id_fk", + "tableFrom": "membership_tombstones", + "tableTo": "organizations", + "columnsFrom": ["organization_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.memberships": { + "name": "memberships", + "schema": "", + "columns": { + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "membership_id": { + "name": "membership_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'member'" + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'active'" + }, + "workos_updated_at": { + "name": "workos_updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "memberships_membership_id_unique": { + "name": "memberships_membership_id_unique", + "columns": [ + { + "expression": "membership_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "memberships_organization_id_idx": { + "name": "memberships_organization_id_idx", + "columns": [ + { + "expression": "organization_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "memberships_account_id_accounts_id_fk": { + "name": "memberships_account_id_accounts_id_fk", + "tableFrom": "memberships", + "tableTo": "accounts", + "columnsFrom": ["account_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "memberships_organization_id_organizations_id_fk": { + "name": "memberships_organization_id_organizations_id_fk", + "tableFrom": "memberships", + "tableTo": "organizations", + "columnsFrom": ["organization_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "memberships_account_id_organization_id_pk": { + "name": "memberships_account_id_organization_id_pk", + "columns": ["account_id", "organization_id"] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.organizations": { + "name": "organizations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "slug": { + "name": "slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "organizations_slug_unique": { + "name": "organizations_slug_unique", + "columns": [ + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workos_sync": { + "name": "workos_sync", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "cursor": { + "name": "cursor", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.artifact": { + "name": "artifact", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "code": { + "name": "code", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "bindings": { + "name": "bindings", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "preview": { + "name": "preview", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "artifact_uidx": { + "name": "artifact_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.blob": { + "name": "blob", + "schema": "", + "columns": { + "namespace": { + "name": "namespace", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "blob_id_uidx": { + "name": "blob_id_uidx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.connection": { + "name": "connection", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "template": { + "name": "template", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "item_ids": { + "name": "item_ids", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "credential_write": { + "name": "credential_write", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "identity_label": { + "name": "identity_label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "last_health": { + "name": "last_health", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "tools_synced_at": { + "name": "tools_synced_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "oauth_client": { + "name": "oauth_client", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_client_owner": { + "name": "oauth_client_owner", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_item_id": { + "name": "refresh_item_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "oauth_scope": { + "name": "oauth_scope", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_token_url": { + "name": "oauth_token_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "provider_state": { + "name": "provider_state", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "connection_uidx": { + "name": "connection_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.definition": { + "name": "definition", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "connection": { + "name": "connection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "schema": { + "name": "schema", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "definition_uidx": { + "name": "definition_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "connection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.integration": { + "name": "integration", + "schema": "", + "columns": { + "slug": { + "name": "slug", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "config": { + "name": "config", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "health_check": { + "name": "health_check", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "config_revised_at": { + "name": "config_revised_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "can_remove": { + "name": "can_remove", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "can_refresh": { + "name": "can_refresh", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "integration_uidx": { + "name": "integration_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_client": { + "name": "oauth_client", + "schema": "", + "columns": { + "slug": { + "name": "slug", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "authorization_url": { + "name": "authorization_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token_url": { + "name": "token_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "grant": { + "name": "grant", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_id": { + "name": "client_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_secret_item_id": { + "name": "client_secret_item_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "credential_write": { + "name": "credential_write", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "token_endpoint_auth_method": { + "name": "token_endpoint_auth_method", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "resource": { + "name": "resource", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_kind": { + "name": "origin_kind", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_integration": { + "name": "origin_integration", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_issuer": { + "name": "origin_issuer", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_redirect_uri": { + "name": "origin_redirect_uri", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "oauth_client_uidx": { + "name": "oauth_client_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_session": { + "name": "oauth_session", + "schema": "", + "columns": { + "state": { + "name": "state", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "client_slug": { + "name": "client_slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "integration": { + "name": "integration", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "template": { + "name": "template", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "redirect_url": { + "name": "redirect_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "pkce_verifier": { + "name": "pkce_verifier", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "identity_label": { + "name": "identity_label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "payload": { + "name": "payload", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "oauth_session_uidx": { + "name": "oauth_session_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "state", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_storage": { + "name": "plugin_storage", + "schema": "", + "columns": { + "plugin_id": { + "name": "plugin_id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "collection": { + "name": "collection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "data": { + "name": "data", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "plugin_storage_uidx": { + "name": "plugin_storage_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "plugin_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "collection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.private_executor_cloud_settings": { + "name": "private_executor_cloud_settings", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "version": { + "name": "version", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true, + "default": "'1.0.0'" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.subject": { + "name": "subject", + "schema": "", + "columns": { + "external_id": { + "name": "external_id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "last_seen_at": { + "name": "last_seen_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "subject_uidx": { + "name": "subject_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "external_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tool": { + "name": "tool", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "connection": { + "name": "connection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "input_schema": { + "name": "input_schema", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "output_schema": { + "name": "output_schema", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "annotations": { + "name": "annotations", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "tool_uidx": { + "name": "tool_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "connection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tool_policy": { + "name": "tool_policy", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "pattern": { + "name": "pattern", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "action": { + "name": "action", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "position": { + "name": "position", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "tool_policy_uidx": { + "name": "tool_policy_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/apps/cloud/drizzle/meta/_journal.json b/apps/cloud/drizzle/meta/_journal.json index 375397ceca..421c19a0ab 100644 --- a/apps/cloud/drizzle/meta/_journal.json +++ b/apps/cloud/drizzle/meta/_journal.json @@ -127,6 +127,13 @@ "when": 1788287088210, "tag": "0017_lush_thunderbolts", "breakpoints": true + }, + { + "idx": 18, + "version": "7", + "when": 1789570778982, + "tag": "0018_member_directory_mirror", + "breakpoints": true } ] } diff --git a/apps/cloud/src/account/org-api-key-revoke.node.test.ts b/apps/cloud/src/account/org-api-key-revoke.node.test.ts index 5b96f01f74..736c1adafa 100644 --- a/apps/cloud/src/account/org-api-key-revoke.node.test.ts +++ b/apps/cloud/src/account/org-api-key-revoke.node.test.ts @@ -37,6 +37,19 @@ const MEMBER = "user_member"; const ORG_KEY = "key_org_1"; const USER_KEY = "key_user_1"; const createdAt = new Date("2026-01-01T00:00:00.000Z"); + +// The mirror's account row as `ensureAccount` mints it: id only, profile +// columns unfilled until a WorkOS user payload arrives. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt, +}); const orgHeaders = { [ORG_SELECTOR_HEADER]: ORG }; const session = (accountId: string) => ({ @@ -82,8 +95,8 @@ const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt }), - getAccount: async (id: string) => ({ id, createdAt }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, slug: org.id, diff --git a/apps/cloud/src/api/layers.ts b/apps/cloud/src/api/layers.ts index 6e73261264..039d983e55 100644 --- a/apps/cloud/src/api/layers.ts +++ b/apps/cloud/src/api/layers.ts @@ -6,6 +6,7 @@ import { makeProtectedApiLayer, requestScopedMiddleware } from "@executor-js/api import { SessionAuthLive } from "../auth/middleware-live"; import { UserStoreService } from "../auth/context"; +import { WorkOsMirror } from "../auth/workos-mirror"; import { CloudAuthPublicHandlers, CloudSessionAuthHandlers, @@ -25,12 +26,13 @@ import { CoreSharedServices } from "../auth/workos"; const DbLive = DbService.Live; const UserStoreLive = UserStoreService.Live.pipe(Layer.provide(DbLive)); +const WorkOsMirrorLive = WorkOsMirror.Live.pipe(Layer.provide(DbLive)); // Per-request layer. Anything that opens an I/O object (postgres.js socket, // fetch stream readers, anything backed by a `Writable`) MUST live here — // `provideRequestScoped` rebuilds it per request so Cloudflare Workers' // I/O isolation is satisfied. See `api.request-scope.test.ts`. -export const RequestScopedServicesLive = Layer.mergeAll(DbLive, UserStoreLive); +export const RequestScopedServicesLive = Layer.mergeAll(DbLive, UserStoreLive, WorkOsMirrorLive); // Boot-scoped layer. Built once at worker boot, reused across requests. // Safe for config, in-memory caches, the global tracer provider, and diff --git a/apps/cloud/src/api/protected-api-key-auth.node.test.ts b/apps/cloud/src/api/protected-api-key-auth.node.test.ts index d92ffe723e..e87512d285 100644 --- a/apps/cloud/src/api/protected-api-key-auth.node.test.ts +++ b/apps/cloud/src/api/protected-api-key-auth.node.test.ts @@ -8,6 +8,19 @@ import { resolveProtectedPrincipal } from "./protected"; const createdAt = new Date("2026-01-01T00:00:00.000Z"); +// The mirror's account row as `ensureAccount` mints it: id only, profile +// columns unfilled until a WorkOS user payload arrives. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt, +}); + const stubApiKeys = Layer.succeed(ApiKeyService)({ validate: (value: string) => Effect.succeed( @@ -50,8 +63,8 @@ const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt }), - getAccount: async (id: string) => ({ id, createdAt }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, slug: `org-slug-${org.id}`, diff --git a/apps/cloud/src/api/protected-jwt-auth.node.test.ts b/apps/cloud/src/api/protected-jwt-auth.node.test.ts index b330ecdd70..b374810372 100644 --- a/apps/cloud/src/api/protected-jwt-auth.node.test.ts +++ b/apps/cloud/src/api/protected-jwt-auth.node.test.ts @@ -9,6 +9,19 @@ import { WorkOSClient, type WorkOSClientService } from "../auth/workos"; import { resolveProtectedPrincipal } from "./protected"; const createdAt = new Date("2026-01-01T00:00:00.000Z"); + +// The mirror's account row as `ensureAccount` mints it: id only, profile +// columns unfilled until a WorkOS user payload arrives. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt, +}); const issuer = "https://test-authkit.example.com"; const audience = "client_test_audience"; @@ -67,8 +80,8 @@ const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt }), - getAccount: async (id: string) => ({ id, createdAt }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, slug: `org-slug-${org.id}`, diff --git a/apps/cloud/src/auth/member-directory.ts b/apps/cloud/src/auth/member-directory.ts new file mode 100644 index 0000000000..01174d88f3 --- /dev/null +++ b/apps/cloud/src/auth/member-directory.ts @@ -0,0 +1,181 @@ +// --------------------------------------------------------------------------- +// Cloud's `MemberDirectory`: the shared read seam over the LOCAL membership +// mirror (`memberships` join `accounts`, db/schema.ts), never over WorkOS. +// +// The mirror is written by `WorkOsMirror` (login, write-through, the Events +// API reconciler); this file only reads it. A row without a `membership_id` +// was never written by a feeder — it predates the mirror — and is not +// reported: the directory answers only for memberships it actually knows the +// WorkOS identity of, and every feeder fills the id in on its next pass. +// A deleted membership is never dropped, it is TOMBSTONED (`status = +// 'inactive'`, `deleted_at` set) so a feeder replaying a payload of the +// deleted membership cannot bring it back; every read here filters on status, +// active + pending unless the caller names the statuses it wants. +// +// Per-request layer: it holds the request's postgres socket via `DbService`. +// --------------------------------------------------------------------------- + +import { and, asc, eq, ilike, inArray, isNotNull, or, sql } from "drizzle-orm"; +import { Effect, Layer } from "effect"; + +import { + DEFAULT_MEMBER_STATUSES, + MemberDirectory, + MemberDirectoryError, + normalizeMemberSearch, + type DirectoryMember, + type MemberDirectoryShape, + type MemberQuery, +} from "@executor-js/api/server"; + +import { accounts, memberships } from "../db/schema"; +import { DbService, type DrizzleDb } from "../db/db"; +import { tryPromiseService, withServiceLogging } from "./errors"; + +// Escape LIKE wildcards in a user-typed search term so `_` and `%` match +// themselves. Same treatment `org-deletion.ts` gives an org id prefix. +const escapeLike = (value: string): string => value.replace(/[\\%_]/g, "\\$&"); + +// The display name the seam reports: first + last, or nothing. Computed in SQL +// too (below) so the search term matches what the caller sees. +const displayName = (firstName: string | null, lastName: string | null): string | null => + [firstName, lastName].filter(Boolean).join(" ") || null; + +const makeService = (db: DrizzleDb): MemberDirectoryShape => { + const read = (op: string, fn: () => Promise) => + withServiceLogging( + `member_directory.${op}`, + () => + new MemberDirectoryError({ + message: `Failed to read the member directory (${op})`, + }), + tryPromiseService(fn), + ); + + // One projection for every read, so the seam's row shape is built in exactly + // one place. `membershipId` is non-null by the `known` predicate below. + const select = () => + db + .select({ + accountId: memberships.accountId, + membershipId: memberships.membershipId, + organizationId: memberships.organizationId, + role: memberships.role, + status: memberships.status, + email: accounts.email, + firstName: accounts.firstName, + lastName: accounts.lastName, + avatarUrl: accounts.avatarUrl, + lastSignInAt: accounts.lastSignInAt, + }) + .from(memberships) + .innerJoin(accounts, eq(accounts.id, memberships.accountId)) + .$dynamic(); + + type Row = Awaited>[number]; + + const toMember = (row: Row): DirectoryMember | null => + row.membershipId === null + ? null + : { + accountId: row.accountId, + membershipId: row.membershipId, + organizationId: row.organizationId, + email: row.email, + name: displayName(row.firstName, row.lastName), + avatarUrl: row.avatarUrl, + role: row.role, + status: row.status, + lastActiveAt: row.lastSignInAt === null ? null : row.lastSignInAt.getTime(), + }; + + const toMembers = (rows: readonly Row[]): DirectoryMember[] => { + const members: DirectoryMember[] = []; + for (const row of rows) { + const member = toMember(row); + if (member !== null) members.push(member); + } + return members; + }; + + const known = isNotNull(memberships.membershipId); + + const members = (organizationId: string, query: MemberQuery = {}) => + read("members", async () => { + const term = normalizeMemberSearch(query.search); + const pattern = term === undefined ? undefined : `%${escapeLike(term)}%`; + let statement = select() + .where( + and( + eq(memberships.organizationId, organizationId), + inArray(memberships.status, query.statuses ?? DEFAULT_MEMBER_STATUSES), + known, + pattern === undefined + ? undefined + : or( + ilike(accounts.email, pattern), + ilike(sql`concat_ws(' ', ${accounts.firstName}, ${accounts.lastName})`, pattern), + ), + ), + ) + .orderBy(asc(accounts.email), asc(memberships.accountId)); + if (query.limit !== undefined) statement = statement.limit(query.limit); + if (query.offset !== undefined) statement = statement.offset(query.offset); + return toMembers(await statement); + }); + + return { + membership: (accountId, organizationId, statuses = DEFAULT_MEMBER_STATUSES) => + read("membership", async () => { + const rows = await select() + .where( + and( + eq(memberships.accountId, accountId), + eq(memberships.organizationId, organizationId), + inArray(memberships.status, statuses), + known, + ), + ) + .limit(1); + const row = rows[0]; + return row === undefined ? null : toMember(row); + }), + + members, + + membersById: (organizationId, accountIds, statuses = DEFAULT_MEMBER_STATUSES) => + accountIds.length === 0 + ? Effect.succeed(new Map()) + : read("membersById", async () => { + const rows = await select().where( + and( + eq(memberships.organizationId, organizationId), + inArray(memberships.accountId, accountIds), + inArray(memberships.status, statuses), + known, + ), + ); + return new Map(toMembers(rows).map((member) => [member.accountId, member])); + }), + + findByEmail: (organizationId, email, statuses = DEFAULT_MEMBER_STATUSES) => + read("findByEmail", async () => { + const rows = await select() + .where( + and( + eq(memberships.organizationId, organizationId), + eq(sql`lower(${accounts.email})`, email), + inArray(memberships.status, statuses), + known, + ), + ) + .limit(1); + const row = rows[0]; + return row === undefined ? null : toMember(row); + }), + }; +}; + +/** The cloud `MemberDirectory` over the per-request `DbService`. */ +export const cloudMemberDirectoryLayer: Layer.Layer = + Layer.effect(MemberDirectory)(Effect.map(DbService.asEffect(), ({ db }) => makeService(db))); diff --git a/apps/cloud/src/auth/org-api-key-auth.node.test.ts b/apps/cloud/src/auth/org-api-key-auth.node.test.ts index ec87511c3b..624641644f 100644 --- a/apps/cloud/src/auth/org-api-key-auth.node.test.ts +++ b/apps/cloud/src/auth/org-api-key-auth.node.test.ts @@ -13,6 +13,19 @@ import { isPlatformAuth, resolveApiKeyPrincipal, resolveBearerAuth } from "./wor const createdAt = new Date("2026-01-01T00:00:00.000Z"); +// The mirror's account row as `ensureAccount` mints it: id only, profile +// columns unfilled until a WorkOS user payload arrives. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt, +}); + const stubApiKeys = Layer.succeed(ApiKeyService)({ validate: (value: string) => { if (value === "valid_org_key") { @@ -65,8 +78,8 @@ const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt }), - getAccount: async (id: string) => ({ id, createdAt }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, slug: `org-slug-${org.id}`, diff --git a/apps/cloud/src/auth/org-selector-auth.node.test.ts b/apps/cloud/src/auth/org-selector-auth.node.test.ts index ead56fb893..aa9f11c024 100644 --- a/apps/cloud/src/auth/org-selector-auth.node.test.ts +++ b/apps/cloud/src/auth/org-selector-auth.node.test.ts @@ -16,6 +16,19 @@ import { WorkOSClient, type WorkOSClientService } from "./workos"; const createdAt = new Date("2026-01-01T00:00:00.000Z"); +// The mirror's account row as `ensureAccount` mints it: id only, profile +// columns unfilled until a WorkOS user payload arrives. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt, +}); + // user_session belongs to BOTH orgs; the URL selects which one a request hits. const MEMBER = "user_session"; const SESSION_ORG = "org_session"; @@ -67,8 +80,8 @@ const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt }), - getAccount: async (id: string) => ({ id, createdAt }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), // Slug is minted at insert now — the stub returns a slugged row. upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, diff --git a/apps/cloud/src/auth/workos-callback-state.node.test.ts b/apps/cloud/src/auth/workos-callback-state.node.test.ts index d4f8970735..83fd945257 100644 --- a/apps/cloud/src/auth/workos-callback-state.node.test.ts +++ b/apps/cloud/src/auth/workos-callback-state.node.test.ts @@ -57,12 +57,25 @@ const stubWorkOS = Layer.succeed( }), ); +// A bare account row, as `ensureAccount` mints it before any WorkOS profile +// has been mirrored onto it. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt: new Date(), +}); + const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt: new Date() }), - getAccount: async (id: string) => ({ id, createdAt: new Date() }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, slug: org.id, diff --git a/apps/cloud/src/auth/workos-mirror.node.test.ts b/apps/cloud/src/auth/workos-mirror.node.test.ts new file mode 100644 index 0000000000..61e7357399 --- /dev/null +++ b/apps/cloud/src/auth/workos-mirror.node.test.ts @@ -0,0 +1,701 @@ +// --------------------------------------------------------------------------- +// The cloud membership mirror: `WorkOsMirror` (writes) + the cloud +// `MemberDirectory` (reads), against the real PGlite Postgres every cloud +// unit test runs on (scripts/test-globalsetup.ts), through the same +// `DbService.Live` the request path uses. +// +// What this pins: +// - an older WorkOS payload never overwrites a newer row (replay-safe) +// - a delete tombstones the row (inactive, stamped with the deletion time): +// a stale OR equal-timestamp upsert cannot resurrect it, a newer one +// reactivates it, and every default read treats the tombstone as no +// membership +// - a delete of a membership or user the mirror has not seen yet leaves +// the tombstone behind, so the backfill's older payload cannot insert +// the row live afterwards +// - a delete of a membership id the row does NOT hold (the member's row +// still carries the id it was replaced from) is recorded all the same, +// so a later, newer payload of the deleted id cannot take the row over +// - a deleted user takes no membership at all, however the payload is +// stamped: the account tombstone refuses the insert by identity +// - a delete with no WorkOS instant keeps the row's own WorkOS stamp, so +// a replacement membership WorkOS created meanwhile is not refused, +// while the removed membership's own payload still is +// - the cursor advances only by compare-and-set (one owner per stream) +// - `members` searches email AND name case-insensitively, pages stably +// - `findByEmail` ignores the casing WorkOS stored +// - a membership arriving before its user still holds (FK via ensureAccount) +// --------------------------------------------------------------------------- + +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Layer } from "effect"; + +import { MemberDirectory } from "@executor-js/api/server"; + +import { DbService } from "../db/db"; +import { cloudMemberDirectoryLayer } from "./member-directory"; +import { UserStoreService } from "./context"; +import { WorkOsMirror, type WorkOsMirrorMembership, type WorkOsMirrorUser } from "./workos-mirror"; + +const DbLive = DbService.Live; +const Services = Layer.mergeAll( + WorkOsMirror.Live, + cloudMemberDirectoryLayer, + UserStoreService.Live, +).pipe(Layer.provideMerge(DbLive)); + +const run = (body: Effect.Effect) => + Effect.runPromise(body.pipe(Effect.provide(Services), Effect.scoped)); + +const at = (iso: string) => new Date(iso); +const T1 = at("2026-01-01T00:00:00.000Z"); +const T2 = at("2026-01-02T00:00:00.000Z"); +const T3 = at("2026-01-03T00:00:00.000Z"); +const T4 = at("2026-01-04T00:00:00.000Z"); + +// Every test mints its own org so the shared test database never couples +// them; ids are synthetic placeholders, never real identities. +const freshOrg = () => + Effect.gen(function* () { + const id = `org_${crypto.randomUUID().replaceAll("-", "")}`; + const store = yield* UserStoreService; + yield* store.use("upsertOrganization", (s) => s.upsertOrganization({ id, name: "Mirror Org" })); + return id; + }); + +const user = (id: string, overrides: Partial = {}): WorkOsMirrorUser => ({ + id, + email: `${id}@placeholder.test`, + firstName: null, + lastName: null, + avatarUrl: null, + lastSignInAt: null, + updatedAt: T1, + ...overrides, +}); + +const membership = ( + organizationId: string, + accountId: string, + overrides: Partial = {}, +): WorkOsMirrorMembership => ({ + id: `om_${accountId}_${organizationId}`, + accountId, + organizationId, + role: "member", + status: "active", + updatedAt: T1, + ...overrides, +}); + +describe("WorkOsMirror upserts", () => { + it("ignores a user payload older than the stored row, accepts a newer one", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + yield* mirror.upsertMembership(membership(org, id)); + + const first = yield* mirror.upsertUser(user(id, { firstName: "Ada", updatedAt: T2 })); + const stale = yield* mirror.upsertUser(user(id, { firstName: "Stale", updatedAt: T1 })); + const afterStale = yield* directory.membership(id, org); + const newer = yield* mirror.upsertUser(user(id, { firstName: "Newer", updatedAt: T3 })); + const afterNewer = yield* directory.membership(id, org); + return { first, stale, newer, afterStale, afterNewer }; + }), + ); + expect(result.first).toBe(true); + expect(result.stale, "an older payload is reported as not written").toBe(false); + expect(result.afterStale?.name, "and left the newer row untouched").toBe("Ada"); + expect(result.newer).toBe(true); + expect(result.afterNewer?.name).toBe("Newer"); + }); + + it("ignores a membership payload older than the stored row", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + yield* mirror.upsertUser(user(id)); + yield* mirror.upsertMembership(membership(org, id, { role: "admin", updatedAt: T2 })); + const stale = yield* mirror.upsertMembership( + membership(org, id, { + role: "member", + status: "inactive", + updatedAt: T1, + }), + ); + const row = yield* directory.membership(id, org); + const equal = yield* mirror.upsertMembership( + membership(org, id, { role: "member", updatedAt: T2 }), + ); + const afterEqual = yield* directory.membership(id, org); + return { stale, row, equal, afterEqual }; + }), + ); + expect(result.stale).toBe(false); + expect(result.row?.role).toBe("admin"); + expect(result.row?.status).toBe("active"); + // Equal timestamps are accepted: the feeders replay the same payload and + // must converge, not stall. + expect(result.equal).toBe(true); + expect(result.afterEqual?.role).toBe("member"); + }); + + it("mints the account row when a membership arrives before its user", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + const written = yield* mirror.upsertMembership(membership(org, id)); + const bare = yield* directory.membership(id, org); + // The bare row has no timestamp, so the first user payload — even an + // "old" one — fills it. + yield* mirror.upsertUser(user(id, { firstName: "Late", updatedAt: T1 })); + const filled = yield* directory.membership(id, org); + return { written, bare, filled }; + }), + ); + expect(result.written).toBe(true); + expect(result.bare).not.toBeNull(); + expect(result.bare?.email).toBeNull(); + expect(result.filled?.name).toBe("Late"); + }); + + it("tombstones a deleted membership so a stale upsert cannot resurrect it, and a newer one can", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + const membershipId = `om_${id}_${org}`; + yield* mirror.upsertUser(user(id)); + yield* mirror.upsertMembership(membership(org, id, { id: membershipId, updatedAt: T1 })); + + const ref = { id: membershipId, accountId: id, organizationId: org }; + const removed = yield* mirror.deleteMembership(ref, T2); + const removedAgain = yield* mirror.deleteMembership(ref, T2); + const byDefault = yield* directory.membership(id, org); + const listed = yield* directory.members(org); + const asInactive = yield* directory.membership(id, org, ["inactive"]); + + // A feeder that fetched the membership BEFORE the deletion (login, + // backfill) writes it after: the guard refuses it. + const stale = yield* mirror.upsertMembership( + membership(org, id, { id: membershipId, updatedAt: T1 }), + ); + const afterStale = yield* directory.membership(id, org, ["inactive"]); + // An active payload stamped the SAME instant as the deletion: the + // tombstone wins, a deletion at T is never undone by a payload at T. + const equal = yield* mirror.upsertMembership( + membership(org, id, { id: membershipId, updatedAt: T2 }), + ); + const afterEqual = yield* directory.membership(id, org, ["inactive"]); + // The member re-added in WorkOS: a payload newer than the deletion, + // under a new membership id (a deleted id is never reused). + const readded = yield* mirror.upsertMembership( + membership(org, id, { id: `${membershipId}_2`, updatedAt: T3 }), + ); + const afterReadd = yield* directory.membership(id, org); + // The OLD membership's deletion, replayed after the re-add: the + // newer row, under its new id, stands. + const lateDelete = yield* mirror.deleteMembership(ref, T2); + const afterLateDelete = yield* directory.membership(id, org); + // The same deletion stamped AFTER the replacement (a removal Executor + // made whose clock was read once WorkOS had answered, by which time + // the member had been re-added): the row is another membership, so + // the timestamp does not make it deletable. + const lateDeleteNewerStamp = yield* mirror.deleteMembership(ref, T4); + const afterLateDeleteNewerStamp = yield* directory.membership(id, org); + return { + removed, + removedAgain, + byDefault, + listed, + asInactive, + stale, + afterStale, + equal, + afterEqual, + readded, + afterReadd, + lateDelete, + afterLateDelete, + lateDeleteNewerStamp, + afterLateDeleteNewerStamp, + }; + }), + ); + expect(result.removed).toBe(true); + expect(result.removedAgain, "a replayed delete changes nothing").toBe(false); + expect(result.byDefault, "a tombstone reads as no membership").toBeNull(); + expect(result.listed, "and is not listed").toEqual([]); + expect(result.asInactive, "but is still there when asked for").toMatchObject({ + status: "inactive", + lastActiveAt: null, + }); + expect(result.stale, "an upsert older than the deletion is refused").toBe(false); + expect(result.afterStale?.status).toBe("inactive"); + expect(result.equal, "an upsert stamped AT the deletion is refused too").toBe(false); + expect(result.afterEqual?.status).toBe("inactive"); + expect(result.readded, "an upsert newer than the deletion reactivates").toBe(true); + expect(result.afterReadd?.status).toBe("active"); + expect(result.lateDelete, "a replayed deletion of the OLD id is refused").toBe(false); + expect(result.afterLateDelete?.status).toBe("active"); + expect( + result.lateDeleteNewerStamp, + "a deletion of the OLD id stamped after the replacement is refused too: identity, not time", + ).toBe(false); + expect(result.afterLateDeleteNewerStamp).toMatchObject({ + status: "active", + membershipId: `om_${result.afterLateDeleteNewerStamp?.accountId}_${result.afterLateDeleteNewerStamp?.organizationId}_2`, + }); + }); + + it("keeps the row's own WorkOS stamp on a delete with no instant, so a replacement created meanwhile is accepted and the removed payload is not", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + const membershipId = `om_${id}_${org}`; + const ref = { id: membershipId, accountId: id, organizationId: org }; + yield* mirror.upsertUser(user(id)); + yield* mirror.upsertMembership(membership(org, id, { id: membershipId, updatedAt: T1 })); + + // Executor removes the member holding no WorkOS instant for it, so + // the tombstone keeps T1, the last state WorkOS reported for it — + // never the local clock, which is long past T2 here. + const removed = yield* mirror.deleteMembership(ref, null); + const removedAgain = yield* mirror.deleteMembership(ref, null); + const tombstone = yield* directory.membership(id, org, ["inactive"]); + // A login that fetched the membership before the removal: refused. + const stale = yield* mirror.upsertMembership( + membership(org, id, { id: membershipId, updatedAt: T1 }), + ); + // The member re-added in WorkOS while the removal was in flight, + // under a new id and stamped before any local clock could have + // stamped the tombstone: accepted. + const replaced = yield* mirror.upsertMembership( + membership(org, id, { id: `${membershipId}_2`, updatedAt: T2 }), + ); + const afterReplace = yield* directory.membership(id, org); + // A tombstone minted for a row the mirror never held has no stamp + // to keep; it is still a tombstone. + const other = `user_${crypto.randomUUID()}`; + const minted = yield* mirror.deleteMembership( + { id: `om_${other}_${org}`, accountId: other, organizationId: org }, + null, + ); + const mintedRow = yield* directory.membership(other, org, ["inactive"]); + return { + membershipId, + removed, + removedAgain, + tombstone, + stale, + replaced, + afterReplace, + minted, + mintedRow, + }; + }), + ); + expect(result.removed).toBe(true); + expect(result.removedAgain, "a repeated removal changes nothing").toBe(false); + expect(result.tombstone?.status).toBe("inactive"); + expect(result.stale, "the pre-removal payload is refused").toBe(false); + expect(result.replaced, "a replacement newer than the row's stamp is accepted").toBe(true); + expect(result.afterReplace).toMatchObject({ + status: "active", + membershipId: `${result.membershipId}_2`, + }); + expect(result.minted, "a row the mirror never held is still tombstoned").toBe(true); + expect(result.mintedRow?.status).toBe("inactive"); + }); + + it("tombstones a membership the mirror has not seen, so a later older payload cannot insert it live", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + const membershipId = `om_${id}_${org}`; + const readdedId = `${membershipId}_2`; + const ref = { id: membershipId, accountId: id, organizationId: org }; + + // The reconciler applies the deletion before the backfill has + // inserted the row (the user is unknown too). + const removed = yield* mirror.deleteMembership(ref, T2); + const removedAgain = yield* mirror.deleteMembership(ref, T2); + const tombstone = yield* directory.membership(id, org, ["inactive"]); + // The backfill, listing WorkOS as it was before the deletion, now + // writes the membership: refused, the tombstone stands. + const backfilled = yield* mirror.upsertMembership( + membership(org, id, { id: membershipId, updatedAt: T1 }), + ); + const afterBackfill = yield* directory.membership(id, org); + // The user payload still fills the bare account row the tombstone + // minted, so the inactive row reads with its profile. + yield* mirror.upsertUser(user(id, { firstName: "Late", updatedAt: T1 })); + const profiled = yield* directory.membership(id, org, ["inactive"]); + // Re-added in WorkOS later under a NEW membership id. + const readded = yield* mirror.upsertMembership( + membership(org, id, { id: readdedId, updatedAt: T3 }), + ); + const afterReadd = yield* directory.membership(id, org); + return { + readdedId, + removed, + removedAgain, + tombstone, + backfilled, + afterBackfill, + profiled, + readded, + afterReadd, + }; + }), + ); + expect(result.removed, "the delete leaves a tombstone behind").toBe(true); + expect(result.removedAgain).toBe(false); + expect(result.tombstone).toMatchObject({ status: "inactive", email: null }); + expect(result.backfilled, "the pre-deletion payload is refused").toBe(false); + expect(result.afterBackfill, "and the member is not live").toBeNull(); + expect(result.profiled?.name).toBe("Late"); + expect(result.readded).toBe(true); + expect(result.afterReadd).toMatchObject({ + status: "active", + membershipId: result.readdedId, + }); + }); + + it("records a delete whose id the row does not hold, so a newer payload of that id cannot take the row over", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + const membershipA = `om_${id}_${org}_a`; + const membershipB = `om_${id}_${org}_b`; + yield* mirror.upsertUser(user(id)); + // The mirror holds A (a stale listing). In WorkOS, A was already + // replaced by B (stamped T2), and B is then deleted (T3) before any + // feeder wrote B here. + yield* mirror.upsertMembership(membership(org, id, { id: membershipA, updatedAt: T1 })); + const refB = { id: membershipB, accountId: id, organizationId: org }; + const deletedB = yield* mirror.deleteMembership(refB, T3); + const deletedBAgain = yield* mirror.deleteMembership(refB, T3); + const rowAfterDelete = yield* directory.membership(id, org); + // A delayed scan, listed before B's deletion, now writes B: stamped + // after A and under another id, exactly what the row guard lets + // through — the ledger refuses it. + const lateB = yield* mirror.upsertMembership( + membership(org, id, { id: membershipB, updatedAt: T2 }), + ); + const afterLateB = yield* directory.membership(id, org); + // A's own deletion, applied later, tombstones the row it holds. + const deletedA = yield* mirror.deleteMembership( + { id: membershipA, accountId: id, organizationId: org }, + T3, + ); + const afterDeleteA = yield* directory.membership(id, org); + // The member re-added in WorkOS under a third id: accepted. + const readded = yield* mirror.upsertMembership( + membership(org, id, { id: `${membershipB}_c`, updatedAt: T4 }), + ); + const afterReadd = yield* directory.membership(id, org); + return { + deletedB, + deletedBAgain, + rowAfterDelete, + lateB, + afterLateB, + deletedA, + afterDeleteA, + readded, + afterReadd, + membershipA, + }; + }), + ); + expect(result.deletedB, "the delete is recorded even though the row holds another id").toBe( + true, + ); + expect(result.deletedBAgain, "a replayed delete records nothing new").toBe(false); + expect( + result.rowAfterDelete?.membershipId, + "the row under A is not B's to tombstone and stands", + ).toBe(result.membershipA); + expect(result.lateB, "the newer payload of the deleted id is refused: identity, not time").toBe( + false, + ); + expect(result.afterLateB?.membershipId).toBe(result.membershipA); + expect(result.deletedA).toBe(true); + expect(result.afterDeleteA, "A's deletion tombstones the row").toBeNull(); + expect(result.readded, "a replacement under a fresh id reactivates").toBe(true); + expect(result.afterReadd?.status).toBe("active"); + }); + + it("tombstones every membership of a deleted user and clears the profile, keeping the account row", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const directory = yield* MemberDirectory; + const orgA = yield* freshOrg(); + const orgB = yield* freshOrg(); + const id = `user_${crypto.randomUUID()}`; + yield* mirror.upsertUser(user(id, { firstName: "Gone", updatedAt: T1 })); + yield* mirror.upsertMembership(membership(orgA, id)); + yield* mirror.upsertMembership(membership(orgB, id)); + + const deleted = yield* mirror.deleteUser(id, T2); + const deletedAgain = yield* mirror.deleteUser(id, T2); + const inA = yield* directory.membership(id, orgA); + const inB = yield* directory.membership(id, orgB, ["inactive"]); + // A stale user payload cannot restore the profile, nor can one + // stamped at the deletion itself. No membership of the deleted user + // is written again — not one stamped after the deletion under a new + // id, the payload a timestamp guard would let through: the user is + // gone, and WorkOS never reuses the id. + const staleUser = yield* mirror.upsertUser(user(id, { firstName: "Back", updatedAt: T1 })); + const equalUser = yield* mirror.upsertUser(user(id, { firstName: "Same", updatedAt: T2 })); + const rejoined = yield* mirror.upsertMembership( + membership(orgA, id, { id: `om_${id}_${orgA}_2`, updatedAt: T3 }), + ); + const afterRejoin = yield* directory.membership(id, orgA, ["inactive"]); + // A user the mirror has never seen: the delete mints the account + // tombstone, the backfill's older profile cannot fill it afterwards, + // and a membership of the user the mirror has never seen — no row + // for the membership guard to judge — is refused by the account + // tombstone alone, whatever it is stamped. + const unseen = `user_${crypto.randomUUID()}`; + const unknown = yield* mirror.deleteUser(unseen, T2); + const unseenProfile = yield* mirror.upsertUser( + user(unseen, { firstName: "Ghost", updatedAt: T1 }), + ); + const unseenMembership = yield* mirror.upsertMembership( + membership(orgA, unseen, { updatedAt: T3 }), + ); + const unseenRow = yield* directory.membership(unseen, orgA, [ + "active", + "pending", + "inactive", + ]); + return { + deleted, + deletedAgain, + inA, + inB, + staleUser, + equalUser, + rejoined, + afterRejoin, + unknown, + unseenProfile, + unseenMembership, + unseenRow, + }; + }), + ); + expect(result.deleted).toBe(true); + expect(result.deletedAgain, "a replayed delete changes nothing").toBe(false); + expect(result.inA, "the user's memberships are tombstoned").toBeNull(); + expect(result.inB).toMatchObject({ + status: "inactive", + email: null, + name: null, + }); + expect(result.staleUser).toBe(false); + expect(result.equalUser, "a profile stamped AT the deletion is refused").toBe(false); + expect( + result.rejoined, + "a membership of a deleted user is refused however it is stamped: identity, not time", + ).toBe(false); + expect(result.afterRejoin).toMatchObject({ + status: "inactive", + name: null, + }); + expect(result.unknown, "deleting an unseen user leaves a tombstone").toBe(true); + expect(result.unseenProfile, "which the older profile cannot fill").toBe(false); + expect( + result.unseenMembership, + "and a membership the mirror never held is not inserted for the deleted user", + ).toBe(false); + expect(result.unseenRow).toBeNull(); + }); +}); + +describe("WorkOsMirror cursor", () => { + it("advances only by compare-and-set", async () => { + const result = await run( + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + // The cursor is instance-wide; read whatever a previous test left so + // this test's expectations are relative, not absolute. + const before = yield* mirror.getCursor(); + const first = yield* mirror.setCursor(before, "event_1"); + const wrongPrev = yield* mirror.setCursor(before === null ? "event_0" : null, "event_x"); + const afterWrong = yield* mirror.getCursor(); + const right = yield* mirror.setCursor("event_1", "event_2"); + const after = yield* mirror.getCursor(); + return { first, wrongPrev, afterWrong, right, after }; + }), + ); + expect(result.first).toBe(true); + expect(result.wrongPrev, "a run holding a stale prev cannot move the cursor").toBe(false); + expect(result.afterWrong).toBe("event_1"); + expect(result.right).toBe(true); + expect(result.after).toBe("event_2"); + }); +}); + +describe("cloud MemberDirectory", () => { + const seed = (org: string) => + Effect.gen(function* () { + const mirror = yield* WorkOsMirror; + const ids = { + ada: `user_${crypto.randomUUID()}`, + grace: `user_${crypto.randomUUID()}`, + linus: `user_${crypto.randomUUID()}`, + gone: `user_${crypto.randomUUID()}`, + }; + yield* mirror.upsertUser( + user(ids.ada, { + email: "Ada.Lovelace@Placeholder.test", + firstName: "Ada", + lastName: "Lovelace", + lastSignInAt: T2, + }), + ); + yield* mirror.upsertUser( + user(ids.grace, { + email: "grace@placeholder.test", + firstName: "Grace", + lastName: "Hopper", + }), + ); + yield* mirror.upsertUser( + user(ids.linus, { + email: "linus@placeholder.test", + firstName: "Linus", + lastName: null, + }), + ); + yield* mirror.upsertUser(user(ids.gone, { email: "gone@placeholder.test" })); + yield* mirror.upsertMembership(membership(org, ids.ada, { role: "admin" })); + yield* mirror.upsertMembership(membership(org, ids.grace, { status: "pending" })); + yield* mirror.upsertMembership(membership(org, ids.linus)); + yield* mirror.upsertMembership(membership(org, ids.gone, { status: "inactive" })); + return ids; + }); + + it("lists active + pending members by default, ordered by email, and pages stably", async () => { + const result = await run( + Effect.gen(function* () { + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const ids = yield* seed(org); + const all = yield* directory.members(org); + const page1 = yield* directory.members(org, { limit: 2, offset: 0 }); + const page2 = yield* directory.members(org, { limit: 2, offset: 2 }); + const inactive = yield* directory.members(org, { + statuses: ["inactive"], + }); + return { ids, all, page1, page2, inactive }; + }), + ); + expect(result.all.map((m) => m.email)).toEqual([ + "Ada.Lovelace@Placeholder.test", + "grace@placeholder.test", + "linus@placeholder.test", + ]); + expect(result.all.find((m) => m.accountId === result.ids.ada)).toMatchObject({ + role: "admin", + status: "active", + name: "Ada Lovelace", + lastActiveAt: T2.getTime(), + }); + expect(result.all.find((m) => m.accountId === result.ids.linus)?.name).toBe("Linus"); + expect([...result.page1, ...result.page2].map((m) => m.accountId)).toEqual( + result.all.map((m) => m.accountId), + ); + expect(result.inactive.map((m) => m.accountId)).toEqual([result.ids.gone]); + }); + + it("searches email and name case-insensitively, escaping LIKE wildcards", async () => { + const result = await run( + Effect.gen(function* () { + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const ids = yield* seed(org); + const byEmail = yield* directory.members(org, { search: "LOVELACE@" }); + const byName = yield* directory.members(org, { + search: " grace hop ", + }); + const nothing = yield* directory.members(org, { search: "nobody" }); + const blank = yield* directory.members(org, { search: " " }); + const wildcard = yield* directory.members(org, { search: "%" }); + return { ids, byEmail, byName, nothing, blank, wildcard }; + }), + ); + expect(result.byEmail.map((m) => m.accountId)).toEqual([result.ids.ada]); + expect(result.byName.map((m) => m.accountId)).toEqual([result.ids.grace]); + expect(result.nothing).toEqual([]); + expect(result.blank.length, "a blank term is no filter").toBe(3); + expect(result.wildcard, "a literal % matches nothing rather than everything").toEqual([]); + }); + + it("resolves a normalized email regardless of stored casing, and batches by id", async () => { + const result = await run( + Effect.gen(function* () { + const directory = yield* MemberDirectory; + const org = yield* freshOrg(); + const other = yield* freshOrg(); + const ids = yield* seed(org); + const found = yield* directory.findByEmail(org, "ada.lovelace@placeholder.test"); + const inactive = yield* directory.findByEmail(org, "gone@placeholder.test"); + const inactiveAsked = yield* directory.findByEmail(org, "gone@placeholder.test", [ + "inactive", + ]); + const wrongOrg = yield* directory.findByEmail(other, "ada.lovelace@placeholder.test"); + const batch = yield* directory.membersById(org, [ids.ada, ids.gone, "user_unknown"]); + const batchAll = yield* directory.membersById( + org, + [ids.ada, ids.gone], + ["active", "pending", "inactive"], + ); + const empty = yield* directory.membersById(org, []); + return { + ids, + found, + inactive, + inactiveAsked, + wrongOrg, + batch, + batchAll, + empty, + }; + }), + ); + expect(result.found?.accountId).toBe(result.ids.ada); + expect(result.inactive, "an inactive member is not found by default").toBeNull(); + expect(result.inactiveAsked?.status, "but is when asked for").toBe("inactive"); + expect(result.wrongOrg).toBeNull(); + expect([...result.batch.keys()], "a batch excludes inactive by default").toEqual([ + result.ids.ada, + ]); + expect([...result.batchAll.keys()].sort()).toEqual([result.ids.ada, result.ids.gone].sort()); + expect(result.empty.size).toBe(0); + }); +}); diff --git a/apps/cloud/src/auth/workos-mirror.ts b/apps/cloud/src/auth/workos-mirror.ts new file mode 100644 index 0000000000..f3869e430a --- /dev/null +++ b/apps/cloud/src/auth/workos-mirror.ts @@ -0,0 +1,519 @@ +// --------------------------------------------------------------------------- +// WorkOsMirror — the WRITE side of cloud's local membership mirror. +// +// WorkOS owns users and organization memberships. This service keeps the +// `accounts` / `memberships` rows (db/schema.ts) in step with it so the read +// side (`auth/member-directory.ts`, the cloud `MemberDirectory`) never has to +// ask WorkOS. Three feeders write through it: the login callback (user + +// memberships already in hand), Executor-initiated changes (write-through), +// and the WorkOS Events API reconciler (dashboard-side changes, replayed in +// order from a persisted cursor). +// +// Every write is idempotent and out-of-order safe. Both upserts carry the +// WorkOS `updatedAt` of their payload and refuse to overwrite a row whose +// stored `workos_updated_at` is newer, so a replayed or late-arriving event +// can never regress the mirror. A delete never drops the row: it TOMBSTONES +// it (`status = 'inactive'`, `workos_updated_at` = the deletion time) — and +// INSERTS the tombstone when the row is not there yet — so a feeder that +// fetched the membership before the deletion and writes it after (a login, +// the backfill) is refused by that same guard instead of reinstating access; +// only a payload newer than the deletion (the member re-added in WorkOS) +// reactivates it. At an EQUAL timestamp the tombstone wins: a live row +// accepts a payload stamped the same instant (feeders replay the same +// payload and must converge), a tombstone does not, so a deletion at T is +// never undone by an active payload at T. Every stamp is on WorkOS's clock: +// a deletion replayed from the events stream is stamped with the event's +// time; one Executor made itself — WorkOS answers a delete with no time — +// is stamped with the `updatedAt` WorkOS last reported for the membership, +// or keeps the stamp the row already holds. Never a local clock: read after +// WorkOS answered, it can post-date a replacement membership WorkOS created +// for the same member meanwhile, and a tombstone stamped with it would +// refuse that replacement for good. The row tombstone can only speak for the +// membership id the row happens to hold, so every delete ALSO records the +// deleted WorkOS id in `membership_tombstones`, a ledger keyed by that id +// alone: WorkOS never reuses a deleted `om_…` id, so no membership write +// naming a recorded id is ever current, however it is stamped. That is what +// covers the case the row cannot: membership A replaced by B in WorkOS before +// the mirror saw either, B then deleted — the delete finds a row holding A +// (not B's to tombstone) and would otherwise leave nothing behind, and a +// later scan that still lists B would insert it live, stamped after A. With +// the ledger the delete is recorded whatever the row holds, and the scan's +// payload is refused by identity. A deleted USER is protected by identity +// the same way: `deleteUser` leaves the account row behind as a tombstone +// (profile cleared, stamped with the deletion), and no membership naming +// that account is ever written again, however the payload is stamped — +// WorkOS never reuses a user id, and a membership the mirror had not seen +// has no row of its own for a timestamp guard to refuse the insert against. +// The cursor advances only by compare-and-set, so two reconciler runs cannot +// both own the stream. +// +// Per-request layer shape, like `UserStoreService`: it holds the request's +// postgres socket, so it is rebuilt per request (`RequestScopedServicesLive`) +// and never shared across Workers requests. +// --------------------------------------------------------------------------- + +import { and, eq, isNotNull, isNull, lt, ne, or, sql } from "drizzle-orm"; +import type { AnyPgColumn } from "drizzle-orm/pg-core"; +import { Context, Effect, Layer, Schema } from "effect"; + +import { type MemberStatus } from "@executor-js/api/server"; + +import { accounts, membershipTombstones, memberships, workosSync } from "../db/schema"; +import { DbService, type DrizzleDb } from "../db/db"; +import { + USER_STORE_FAILURE_REASONS, + tryPromiseService, + userStoreReasonFromCause, + withServiceLogging, +} from "./errors"; + +/** A WorkOS user, as the mirror stores it. `updatedAt` is WorkOS's own. */ +export interface WorkOsMirrorUser { + readonly id: string; + readonly email: string; + readonly firstName: string | null; + readonly lastName: string | null; + readonly avatarUrl: string | null; + readonly lastSignInAt: Date | null; + readonly updatedAt: Date; +} + +/** + * A WorkOS organization membership, as the mirror stores it. `id` is the + * WorkOS `om_…`; `accountId` the WorkOS user id; `updatedAt` is WorkOS's own. + * The organization row must already be mirrored (`upsertOrganization`) — a + * membership of an unknown org is a `query` failure, not a silent skip. + */ +export interface WorkOsMirrorMembership { + readonly id: string; + readonly accountId: string; + readonly organizationId: string; + readonly role: string; + readonly status: MemberStatus; + readonly updatedAt: Date; +} + +/** + * What identifies a membership to a delete: the WorkOS `om_…` id AND the + * (account, organization) pair it belongs to. The pair is the row's key, + * and a delete must be able to mint the row as a tombstone when the mirror + * has not seen the membership yet — an id alone cannot. + */ +export type WorkOsMirrorMembershipRef = Pick< + WorkOsMirrorMembership, + "id" | "accountId" | "organizationId" +>; + +/** + * The public failure of every mirror write: which call, and how it failed + * (classified from the driver cause the same way `UserStoreError` is). + */ +export class WorkOsMirrorError extends Schema.TaggedErrorClass()( + "WorkOsMirrorError", + { + operation: Schema.String, + reason: Schema.Literals(USER_STORE_FAILURE_REASONS), + }, + { httpApiStatus: 500 }, +) { + override get message(): string { + return `workos mirror ${this.operation} failed: ${this.reason}`; + } +} + +export interface WorkOsMirrorShape { + /** + * Insert or refresh a user row. `false` when the payload was refused and + * the row left untouched: the stored row is newer than `updatedAt`, or is + * a deletion tombstone stamped at `updatedAt` or later. + */ + readonly upsertUser: (user: WorkOsMirrorUser) => Effect.Effect; + /** + * Insert or refresh a membership row, minting the bare account row first so + * the foreign key holds when the membership arrives before its user. `false` + * when the payload was refused: the stored row is newer than `updatedAt`, + * or is a tombstone stamped at `updatedAt` or later — or the membership id + * is recorded in the deletion ledger (`membership_tombstones`, written by + * `deleteMembership`), whatever the payload is stamped: WorkOS never + * reuses a deleted `om_…` id, so no payload naming a recorded id is + * current — or the account is a deletion tombstone (`deleteUser`), + * whatever the payload is stamped: WorkOS never reuses a user id either, + * so a deleted user has no memberships to mirror. + */ + readonly upsertMembership: ( + membership: WorkOsMirrorMembership, + ) => Effect.Effect; + /** + * Tombstone the membership as deleted: the row stays (or is minted, when + * the mirror has not seen the membership yet), `inactive`, carrying the + * deleted WorkOS id and stamped so that only a payload newer than the + * stamp can reactivate the row. The stamp is on WorkOS's clock, so it + * orders against every payload the mirror is fed: `deletedAt` is an + * instant at or after the membership's last reported state — a deletion + * event's `createdAt`, or the `updatedAt` WorkOS last reported for the + * membership when Executor deletes it (WorkOS answers a delete with no + * time) — or `null` when the caller holds no WorkOS instant at all: the + * row then keeps the stamp it holds, the last state WorkOS reported for + * this membership. Never a local clock: read after WorkOS answered, it + * can post-date a replacement membership WorkOS created for the same + * member meanwhile, and a tombstone stamped with it would refuse that + * replacement for good. Every payload of the deleted membership a feeder + * could have fetched is at or before the stamp (a tombstone wins a tie), + * and every replacement is after it. A row the mirror does not hold and + * no instant to stamp it with takes the current time: there is no WorkOS + * stamp to keep. Matches the row by the membership's IDENTITY, never by + * timestamp: `false` when the row already carries this tombstone or a + * later one (a replayed delete), or holds ANOTHER membership id — the + * member re-added in WorkOS under a new id, which stands whether the + * replacement was mirrored before or after this delete was stamped. In + * EVERY case the deleted id is recorded in the deletion ledger + * (`membership_tombstones`) — a row holding another id is left alone, but + * the id this delete names is still refused to every later write, so a + * membership the mirror never held under its own id (replaced and deleted + * in WorkOS before the mirror saw it) cannot be inserted live by a scan + * that listed it before the deletion. `true` when the delete changed + * anything: the row was tombstoned, or the id was newly recorded. The + * organization row must already be mirrored, as for `upsertMembership`. + */ + readonly deleteMembership: ( + membership: WorkOsMirrorMembershipRef, + deletedAt: Date | null, + ) => Effect.Effect; + /** + * Tombstone a deleted WorkOS user: every membership of the account is + * tombstoned as by `deleteMembership`, and the account row is kept (it + * anchors foreign keys) — or minted, when the mirror has not seen the user + * yet — with its profile cleared and stamped `deletedAt`, so a stale user + * payload cannot restore it. `false` when the account already carries + * this tombstone or a later one (a replayed delete). + */ + readonly deleteUser: ( + accountId: string, + deletedAt: Date, + ) => Effect.Effect; + /** The id of the last WorkOS event applied, or `null` before the first run. */ + readonly getCursor: () => Effect.Effect; + /** + * Compare-and-set the cursor: advance to `next` only if it still reads + * `prev` (`null` = no cursor yet). `false` means another run moved it first + * — the caller must stop, it no longer owns the stream. + */ + readonly setCursor: ( + prev: string | null, + next: string, + ) => Effect.Effect; +} + +// The one events stream the reconciler follows. A row id rather than a +// singleton table so a second stream (another WorkOS environment, a replay) +// can be added without a schema change. +const EVENTS_CURSOR_ID = "events"; + +// A tombstone keeps the row, marks it `inactive`, and moves its timestamp to +// the deletion time — never backwards, so a row that somehow carries a newer +// WorkOS timestamp keeps it and the upsert guard stays at least as strict. +// With no instant at all (`deletedAt` null, see `deleteMembership`) the row +// keeps the stamp it holds, and a row that has none takes the current time +// — the only instant known. (A raw `sql` +// fragment binds the Date without the column's driver mapping, so it is +// passed as ISO text and cast.) +const noEarlierThan = (column: AnyPgColumn, at: Date | null) => + at === null + ? sql`coalesce(${column}, now())` + : sql`greatest(${column}, ${at.toISOString()}::timestamptz)`; + +const tombstone = (deletedAt: Date | null) => ({ + status: "inactive" as const, + workosUpdatedAt: noEarlierThan(memberships.workosUpdatedAt, deletedAt), +}); + +// Which stored membership rows a payload stamped `updatedAt` may overwrite: +// a row with no stamp (predating the mirror), any row stamped earlier, and a +// LIVE row stamped the same instant — feeders replay the same payload and +// must converge, not stall. An `inactive` row stamped the same instant is +// not overwritten: a deletion (or deactivation) at T beats an active payload +// at T, so an equal-timestamp payload can never undo a tombstone. +const membershipAcceptsPayload = (updatedAt: Date) => + or( + isNull(memberships.workosUpdatedAt), + lt(memberships.workosUpdatedAt, updatedAt), + and(eq(memberships.workosUpdatedAt, updatedAt), ne(memberships.status, "inactive")), + ); + +// A deleted user's account row, as `deleteUser` leaves it: the profile is +// cleared (every WorkOS user payload carries an email, so a stamped row with +// none was written by `deleteUser`) and the stamp is the deletion time. A row +// minted bare by `ensureAccount` has no stamp either, and is not a tombstone. +const accountIsTombstone = and(isNull(accounts.email), isNotNull(accounts.workosUpdatedAt)); + +// Whether membership `id` is in the deletion ledger: a WorkOS id a delete +// has named, which never returns. Judged by identity alone — the ledger +// records WHEN for the record, not for ordering. +const membershipIdDeleted = async (db: DrizzleDb, id: string): Promise => { + const rows = await db + .select({ membershipId: membershipTombstones.membershipId }) + .from(membershipTombstones) + .where(eq(membershipTombstones.membershipId, id)); + return rows.length > 0; +}; + +// The same rule as for a membership row, for an account row. A tombstone +// takes no payload stamped at or before the deletion; a bare row takes any. +const accountAcceptsPayload = (updatedAt: Date) => + or( + isNull(accounts.workosUpdatedAt), + lt(accounts.workosUpdatedAt, updatedAt), + and(eq(accounts.workosUpdatedAt, updatedAt), isNotNull(accounts.email)), + ); + +// A delete is applied unless the row already carries this tombstone or a +// later one: a replayed deletion changes nothing and reports so. With no +// instant given, any tombstone the row carries is this one or later. +const notTombstonedSince = (deletedAt: Date | null) => + or( + ne(memberships.status, "inactive"), + isNull(memberships.workosUpdatedAt), + deletedAt === null ? undefined : lt(memberships.workosUpdatedAt, deletedAt), + ); + +// Which (account, organization) row a delete of membership `id` at +// `deletedAt` may tombstone: the row carrying THIS id — whatever its stamp, +// a deleted id is never reused — or a row with no id at all (written before +// the mirror recorded WorkOS ids; the delete fills the id in). Never a row +// under ANOTHER id: that is a different membership of the same account and +// organization, the member re-added in WorkOS after this one was removed, +// and it stands however the two are stamped. Its stamp cannot be trusted to +// order them: a removal Executor makes is stamped by a local clock, and one +// that reads the clock after WorkOS answered can post-date a replacement +// WorkOS created while the request was in flight. Identity orders them, +// timestamps do not. A row already carrying this tombstone or a later one is +// left alone (`notTombstonedSince`) so a replayed delete reports `false`. +const membershipDeletableBy = (id: string, deletedAt: Date | null) => + and( + or(isNull(memberships.membershipId), eq(memberships.membershipId, id)), + notTombstonedSince(deletedAt), + ); + +const makeService = (db: DrizzleDb): WorkOsMirrorShape => { + const run = (op: string, fn: () => Promise) => + withServiceLogging( + `workos_mirror.${op}`, + (failure) => + new WorkOsMirrorError({ + operation: op, + reason: userStoreReasonFromCause(failure), + }), + tryPromiseService(fn), + ); + + // Over `db` or a transaction handle (drizzle's is a `PgDatabase` too). + const ensureAccount = (on: DrizzleDb, id: string) => + on.insert(accounts).values({ id }).onConflictDoNothing({ target: accounts.id }); + + return { + upsertUser: (user) => + run("upsertUser", async () => { + const written = await db + .insert(accounts) + .values({ + id: user.id, + email: user.email, + firstName: user.firstName, + lastName: user.lastName, + avatarUrl: user.avatarUrl, + lastSignInAt: user.lastSignInAt, + workosUpdatedAt: user.updatedAt, + }) + .onConflictDoUpdate({ + target: accounts.id, + set: { + email: user.email, + firstName: user.firstName, + lastName: user.lastName, + avatarUrl: user.avatarUrl, + lastSignInAt: user.lastSignInAt, + workosUpdatedAt: user.updatedAt, + }, + setWhere: accountAcceptsPayload(user.updatedAt), + }) + .returning({ id: accounts.id }); + return written.length > 0; + }), + + upsertMembership: (membership) => + run("upsertMembership", async () => { + await ensureAccount(db, membership.accountId); + // Never for a DELETED user. The row guard below orders a payload + // against the membership row it would overwrite; a membership the + // mirror has not seen yet has no row, so the guard cannot refuse + // the INSERT — and a feeder that fetched the membership before the + // user was deleted and writes it after (a stalled login list, the + // backfill's older listing) would insert it live. The account + // tombstone is the one row a deleted user always leaves behind, so + // it is consulted first, by identity: WorkOS never reuses a user + // id, so no payload naming a tombstoned account is ever current. + const deleted = await db + .select({ id: accounts.id }) + .from(accounts) + .where(and(eq(accounts.id, membership.accountId), accountIsTombstone)); + if (deleted.length > 0) return false; + // Never under a DELETED membership id. The row guard below can only + // refuse against the id the row holds; a delete of THIS id that + // found the row under another id (see the header) left only the + // ledger entry behind, and that is what refuses the payload here. + if (await membershipIdDeleted(db, membership.id)) return false; + const written = await db + .insert(memberships) + .values({ + accountId: membership.accountId, + organizationId: membership.organizationId, + membershipId: membership.id, + role: membership.role, + status: membership.status, + workosUpdatedAt: membership.updatedAt, + }) + .onConflictDoUpdate({ + target: [memberships.accountId, memberships.organizationId], + set: { + membershipId: membership.id, + role: membership.role, + status: membership.status, + workosUpdatedAt: membership.updatedAt, + }, + setWhere: membershipAcceptsPayload(membership.updatedAt), + }) + .returning({ accountId: memberships.accountId }); + return written.length > 0; + }), + + // An upsert, like the membership write it guards against: a delete the + // mirror sees before the membership itself (the reconciler ahead of the + // backfill) must leave the tombstone behind, or the later, older payload + // would insert the row live. + deleteMembership: (membership, deletedAt) => + run("deleteMembership", () => + db.transaction(async (tx) => { + await ensureAccount(tx, membership.accountId); + // The ledger entry FIRST, whatever the row holds: this is the one + // record of the deletion that does not depend on the row carrying + // the deleted id. A replayed delete finds it there (`DO NOTHING`) + // and records nothing new. + const recorded = await tx + .insert(membershipTombstones) + .values({ + membershipId: membership.id, + accountId: membership.accountId, + organizationId: membership.organizationId, + ...(deletedAt === null ? {} : { deletedAt }), + }) + .onConflictDoNothing({ target: membershipTombstones.membershipId }) + .returning({ membershipId: membershipTombstones.membershipId }); + const tombstoned = await tx + .insert(memberships) + .values({ + accountId: membership.accountId, + organizationId: membership.organizationId, + membershipId: membership.id, + status: "inactive", + workosUpdatedAt: deletedAt ?? new Date(), + }) + .onConflictDoUpdate({ + target: [memberships.accountId, memberships.organizationId], + set: { membershipId: membership.id, ...tombstone(deletedAt) }, + setWhere: membershipDeletableBy(membership.id, deletedAt), + }) + .returning({ accountId: memberships.accountId }); + return recorded.length > 0 || tombstoned.length > 0; + }), + ), + + deleteUser: (accountId, deletedAt) => + run("deleteUser", async () => { + // The account tombstone FIRST, then the memberships. A membership + // write checks the account tombstone before it inserts + // (`upsertMembership`), so a write racing this delete either sees + // the tombstone and refuses, or has landed before it and is caught + // by the membership tombstoning below. In the other order a write + // between the two statements would slip through live. + // + // The account tombstone: profile cleared, stamped with the deletion. + // Minted when absent, for the same reason as the membership one. + const cleared = await db + .insert(accounts) + .values({ id: accountId, workosUpdatedAt: deletedAt }) + .onConflictDoUpdate({ + target: accounts.id, + set: { + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: noEarlierThan(accounts.workosUpdatedAt, deletedAt), + }, + // Applied unless the row already carries this tombstone or a + // later one (a replayed delete). + setWhere: or( + isNotNull(accounts.email), + isNull(accounts.workosUpdatedAt), + lt(accounts.workosUpdatedAt, deletedAt), + ), + }) + .returning({ id: accounts.id }); + await db + .update(memberships) + .set(tombstone(deletedAt)) + .where(and(eq(memberships.accountId, accountId), notTombstonedSince(deletedAt))); + return cleared.length > 0; + }), + + getCursor: () => + run("getCursor", async () => { + const rows = await db + .select({ cursor: workosSync.cursor }) + .from(workosSync) + .where(eq(workosSync.id, EVENTS_CURSOR_ID)); + // No row yet is the same state as a row with no cursor: nothing applied. + return rows[0]?.cursor ?? null; + }), + + setCursor: (prev, next) => + run("setCursor", async () => { + const now = new Date(); + if (prev === null) { + // First advance: mint the row, or claim an existing row that still + // has no cursor. A row that already carries one belongs to another + // run and is left alone. + const written = await db + .insert(workosSync) + .values({ id: EVENTS_CURSOR_ID, cursor: next, updatedAt: now }) + .onConflictDoUpdate({ + target: workosSync.id, + set: { cursor: next, updatedAt: now }, + setWhere: isNull(workosSync.cursor), + }) + .returning({ id: workosSync.id }); + return written.length > 0; + } + const written = await db + .update(workosSync) + .set({ cursor: next, updatedAt: now }) + .where(and(eq(workosSync.id, EVENTS_CURSOR_ID), eq(workosSync.cursor, prev))) + .returning({ id: workosSync.id }); + return written.length > 0; + }), + }; +}; + +export class WorkOsMirror extends Context.Service()( + "@executor-js/cloud/WorkOsMirror", +) { + static Live = Layer.effect(this)(Effect.map(DbService.asEffect(), ({ db }) => makeService(db))); +} + +/** + * A FRESH `WorkOsMirror` layer (new layer value per call), for a service built + * once but invoked across many Workers requests — the same reason + * `makeUserStoreLayer` exists. See [[makeDbLayer]]. + */ +export const makeWorkOsMirrorLayer = (): Layer.Layer => + Layer.effect(WorkOsMirror)(Effect.map(DbService.asEffect(), ({ db }) => makeService(db))); diff --git a/apps/cloud/src/db/org-deletion.test.ts b/apps/cloud/src/db/org-deletion.test.ts index 86faa45bb9..979f0b24b3 100644 --- a/apps/cloud/src/db/org-deletion.test.ts +++ b/apps/cloud/src/db/org-deletion.test.ts @@ -200,7 +200,9 @@ const NOT_ORG_OWNED: Record = { // cascades from its FK; `accounts` deliberately outlives the org. organizations: "the identity row itself, deleted directly", memberships: "cascades from the organizations FK", + membership_tombstones: "cascades from the organizations FK", accounts: "shared across orgs — deliberately survives", + workos_sync: "the WorkOS Events API cursor, instance-wide and not org-scoped", }; const countTenantRows = async (db: DrizzleDb, tenant: string): Promise => { diff --git a/apps/cloud/src/db/schema.ts b/apps/cloud/src/db/schema.ts index fb1bd79987..ef18c48248 100644 --- a/apps/cloud/src/db/schema.ts +++ b/apps/cloud/src/db/schema.ts @@ -2,22 +2,54 @@ // Cloud-specific identity & multi-tenancy tables // --------------------------------------------------------------------------- // -// AuthKit owns the canonical user/membership data. We mirror minimally: +// AuthKit owns the canonical user/membership data. We mirror it locally: // -// - `accounts` — login identity (foreign key anchor for created_by, etc.) +// - `accounts` — login identity + profile (foreign key anchor for +// created_by, etc.; email/name/avatar for member lists) // - `organizations` — billing entity, scoping root for all domain data -// - `memberships` — which accounts belong to which organizations +// - `memberships` — which accounts belong to which organizations, with +// the WorkOS role and status +// - `workos_sync` — the WorkOS Events API cursor the reconciler resumes from // -// We do NOT mirror invitations or user profile data — those stay in WorkOS -// and are queried via API when needed. +// The mirror is fed by login (the callback has the user + memberships in +// hand), write-through on every Executor-initiated change, and the WorkOS +// Events API (dashboard-side changes). It is the read path for membership and +// member lists — WorkOS is a write target and an event source, never a +// per-request read. Invitations are NOT mirrored; they stay live in WorkOS. +// +// `workos_updated_at` on `accounts` and `memberships` is the WorkOS +// `updatedAt` of the payload that last wrote the row. Every upsert is guarded +// on it, so feeders can be replayed and reordered without an older payload +// clobbering a newer one. -import { pgTable, primaryKey, text, timestamp, uniqueIndex } from "drizzle-orm/pg-core"; +import { sql } from "drizzle-orm"; +import { index, pgTable, primaryKey, text, timestamp, uniqueIndex } from "drizzle-orm/pg-core"; -/** Login identity. The `id` is the WorkOS user ID. */ -export const accounts = pgTable("accounts", { - id: text("id").primaryKey(), - createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), -}); +/** + * Login identity + mirrored WorkOS profile. The `id` is the WorkOS user ID. + * Profile columns are nullable because a row can be minted by `ensureAccount` + * (an api-key path, a membership arriving before its user event) with nothing + * but the id; the next user payload fills them in. + */ +export const accounts = pgTable( + "accounts", + { + id: text("id").primaryKey(), + email: text("email"), + firstName: text("first_name"), + lastName: text("last_name"), + avatarUrl: text("avatar_url"), + /** WorkOS `updatedAt` of the user payload that last wrote this row. */ + workosUpdatedAt: timestamp("workos_updated_at", { withTimezone: true }), + lastSignInAt: timestamp("last_sign_in_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + }, + (t) => ({ + // `findByEmail` and the search filter compare lower-cased; the index + // matches that expression so the lookup stays indexed. + emailLowerIdx: index("accounts_email_lower_idx").on(sql`lower(${t.email})`), + }), +); /** * Organization (billing entity, scoping root). The `id` is the WorkOS @@ -41,9 +73,26 @@ export const organizations = pgTable( ); /** - * Account ↔ organization link. Lets us answer "which workspaces does this - * account belong to?" without a WorkOS round-trip, and gives future + * Account ↔ organization link, mirroring the WorkOS organization membership. + * Answers "which workspaces does this account belong to?" and "is this caller + * an active member with which role?" without a WorkOS round-trip, and gives * per-(account, organization) data a foreign key to point at. + * + * `membershipId` is the WorkOS `om_…` id — nullable only because rows written + * before the mirror existed carry none; every feeder sets it. `role` is the + * WorkOS role slug as issued (`admin` / `member`); `status` is the WorkOS + * membership status (`active` / `pending` / `inactive`). A membership deleted + * in WorkOS is never dropped here: it is tombstoned as `inactive` with + * `deleted_at` set, so a feeder replaying an older payload cannot resurrect it. + * + * `deleted_at` means "the membership under `membership_id` was DELETED in + * WorkOS" — an identity fact, not a timestamp to order payloads by. WorkOS + * never reuses a deleted `om_…` id, so any payload naming that id is stale + * however it is stamped, and a payload naming a DIFFERENT id for the same + * (account, organization) is the member re-added: a replacement, ordered by + * `workos_updated_at` like every other write. Distinct from `status = + * 'inactive'` with `deleted_at` null, which is a membership WorkOS + * deactivated but still holds and can reactivate under the same id. */ export const memberships = pgTable( "memberships", @@ -54,9 +103,67 @@ export const memberships = pgTable( organizationId: text("organization_id") .notNull() .references(() => organizations.id, { onDelete: "cascade" }), + membershipId: text("membership_id"), + role: text("role").notNull().default("member"), + status: text("status", { enum: ["active", "pending", "inactive"] }) + .notNull() + .default("active"), + /** WorkOS `updatedAt` of the membership payload that last wrote this row. */ + workosUpdatedAt: timestamp("workos_updated_at", { withTimezone: true }), + /** + * When the membership under `membership_id` was deleted in WorkOS, or null + * while WorkOS still holds it (whatever its `status`). Set once; cleared + * only when a replacement membership (another id) takes the row over. + */ + deletedAt: timestamp("deleted_at", { withTimezone: true }), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), }, (t) => ({ pk: primaryKey({ columns: [t.accountId, t.organizationId] }), + membershipIdUnique: uniqueIndex("memberships_membership_id_unique").on(t.membershipId), + organizationIdx: index("memberships_organization_id_idx").on(t.organizationId), + }), +); + +/** + * Every WorkOS membership id (`om_…`) a DELETE has named, keyed by that id + * alone. The `memberships` row is keyed by (account, organization) and holds + * ONE membership id, so a row-level tombstone can only record the deletion of + * the id the row happens to carry: a delete of membership B arriving while + * the row still holds an older membership A of the same pair (A replaced by + * B in WorkOS before the mirror saw either, B then deleted) has no row to + * tombstone, and a later payload of B — stamped after A, under another id — + * would take the row over live. This ledger is what the mirror consults + * instead: a delete always records the id here, whatever the row holds, and + * no membership write ever names a recorded id again, however it is stamped + * — WorkOS never reuses a deleted `om_…` id. `deleted_at` records when; it + * orders nothing. Rows cascade with their account and organization. + */ +export const membershipTombstones = pgTable( + "membership_tombstones", + { + membershipId: text("membership_id").primaryKey(), + accountId: text("account_id") + .notNull() + .references(() => accounts.id, { onDelete: "cascade" }), + organizationId: text("organization_id") + .notNull() + .references(() => organizations.id, { onDelete: "cascade" }), + deletedAt: timestamp("deleted_at", { withTimezone: true }).notNull().defaultNow(), + }, + (t) => ({ + organizationIdx: index("membership_tombstones_organization_id_idx").on(t.organizationId), }), ); + +/** + * The WorkOS Events API cursor. One row per stream (`id` names the stream; + * the reconciler uses `"events"`), holding the id of the last event applied. + * Advanced only by compare-and-set, so two concurrent reconciler runs cannot + * both believe they own the stream: the loser's CAS fails and it stops. + */ +export const workosSync = pgTable("workos_sync", { + id: text("id").primaryKey(), + cursor: text("cursor"), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), +}); diff --git a/apps/cloud/src/extensions/billing/route.node.test.ts b/apps/cloud/src/extensions/billing/route.node.test.ts index dc2a5a7316..b25ec53124 100644 --- a/apps/cloud/src/extensions/billing/route.node.test.ts +++ b/apps/cloud/src/extensions/billing/route.node.test.ts @@ -7,6 +7,19 @@ import { resolveBillingOrganization } from "./route"; const createdAt = new Date("2026-01-01T00:00:00.000Z"); +// The mirror's account row as `ensureAccount` mints it: id only, profile +// columns unfilled until a WorkOS user payload arrives. +const bareAccount = (id: string) => ({ + id, + email: null, + firstName: null, + lastName: null, + avatarUrl: null, + workosUpdatedAt: null, + lastSignInAt: null, + createdAt, +}); + const MEMBER = "user_session"; const SESSION_ORG = "org_session"; const URL_ORG = "org_url"; @@ -37,8 +50,8 @@ const stubUsers = Layer.succeed(UserStoreService)({ use: (_op, fn) => Effect.promise(() => fn({ - ensureAccount: async (id: string) => ({ id, createdAt }), - getAccount: async (id: string) => ({ id, createdAt }), + ensureAccount: async (id: string) => bareAccount(id), + getAccount: async (id: string) => bareAccount(id), upsertOrganization: async (org: { id: string; name: string }) => ({ ...org, slug: org.id, diff --git a/apps/host-selfhost/src/auth/member-directory.test.ts b/apps/host-selfhost/src/auth/member-directory.test.ts new file mode 100644 index 0000000000..1671adb8da --- /dev/null +++ b/apps/host-selfhost/src/auth/member-directory.test.ts @@ -0,0 +1,134 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { afterAll, describe, expect, it } from "@effect/vitest"; +import { Effect, Layer } from "effect"; + +import { MemberDirectory } from "@executor-js/api/server"; + +// The self-host `MemberDirectory` over REAL Better Auth: the org plugin's +// `member` rows joined to `user` rows through Better Auth's own adapter, the +// same read `mcp/auth.ts` makes for an OAuth token's role. +// +// Members are created server-side (`createUser` + `addMember`, no session — +// the same calls the bootstrap seed makes), so this pins the adapter read +// itself rather than the sign-up flow. + +process.env.EXECUTOR_DATA_DIR = mkdtempSync(join(tmpdir(), "eh-member-directory-")); +process.env.BETTER_AUTH_SECRET = "member-directory-secret-0123456789-abcdefghij"; +process.env.EXECUTOR_BOOTSTRAP_ADMIN_EMAIL = "owner@placeholder.test"; +process.env.EXECUTOR_BOOTSTRAP_ADMIN_PASSWORD = "owner-pass-123456"; + +const { makeSelfHostApp } = await import("../app"); +const { BetterAuth } = await import("./better-auth"); +const { betterAuthMemberDirectoryLayer } = await import("./member-directory"); + +const app = await makeSelfHostApp(); +afterAll(() => app.closeDb()); + +const { auth, organizationId } = app.betterAuth; +const directoryLayer = betterAuthMemberDirectoryLayer.pipe( + Layer.provide(Layer.succeed(BetterAuth)(app.betterAuth)), +); +const run = (body: Effect.Effect) => + Effect.runPromise(body.pipe(Effect.provide(directoryLayer))); + +const addMember = async (email: string, name: string, role: "admin" | "member") => { + const created = await auth.api.createUser({ + body: { email, name, password: "pw-12345678" }, + }); + await auth.api.addMember({ + body: { userId: created.user.id, role, organizationId }, + }); + return created.user.id; +}; + +// Better Auth lower-cases the email it stores; the mixed case here proves the +// directory does not depend on that. +const ada = await addMember("Ada.Lovelace@Placeholder.test", "Ada Lovelace", "admin"); +const grace = await addMember("grace@placeholder.test", "Grace Hopper", "member"); +const linus = await addMember("linus@placeholder.test", "Linus", "member"); +// A user who is NOT a member of the org: must never be reported. +const outsider = await auth.api.createUser({ + body: { + email: "outsider@placeholder.test", + name: "Outsider", + password: "pw-12345678", + }, +}); + +describe("self-host MemberDirectory", () => { + it("reports the org's members with Better Auth roles, ordered by email", async () => { + const members = await run( + Effect.flatMap(MemberDirectory.asEffect(), (d) => d.members(organizationId)), + ); + const emails = members.map((m) => m.email); + expect(emails).toEqual([ + "ada.lovelace@placeholder.test", + "grace@placeholder.test", + "linus@placeholder.test", + "owner@placeholder.test", + ]); + const first = members.find((m) => m.accountId === ada); + expect(first).toMatchObject({ + organizationId, + role: "admin", + status: "active", + name: "Ada Lovelace", + lastActiveAt: null, + }); + expect(first?.membershipId, "membershipId is the member ROW id, not the user id").not.toBe(ada); + expect(members.find((m) => m.email === "owner@placeholder.test")?.role).toBe("owner"); + expect(members.some((m) => m.accountId === outsider.user.id)).toBe(false); + }); + + it("searches email and name case-insensitively and pages stably", async () => { + const result = await run( + Effect.gen(function* () { + const d = yield* MemberDirectory; + return { + byEmail: yield* d.members(organizationId, { search: "LOVELACE@" }), + byName: yield* d.members(organizationId, { search: " grace hop " }), + nothing: yield* d.members(organizationId, { search: "nobody" }), + page1: yield* d.members(organizationId, { limit: 2, offset: 0 }), + page2: yield* d.members(organizationId, { limit: 2, offset: 2 }), + inactive: yield* d.members(organizationId, { + statuses: ["inactive"], + }), + all: yield* d.members(organizationId), + }; + }), + ); + expect(result.byEmail.map((m) => m.accountId)).toEqual([ada]); + expect(result.byName.map((m) => m.accountId)).toEqual([grace]); + expect(result.nothing).toEqual([]); + // Two pages of two, concatenated, are the whole ordered list. + const paged = [...result.page1, ...result.page2].map((m) => m.accountId); + expect(paged).toEqual(result.all.map((m) => m.accountId)); + expect(result.page1.length + result.page2.length).toBe(4); + expect(result.inactive, "Better Auth members are always active").toEqual([]); + }); + + it("resolves one membership, a batch by id, and an email in any casing", async () => { + const result = await run( + Effect.gen(function* () { + const d = yield* MemberDirectory; + return { + one: yield* d.membership(grace, organizationId), + oneInactive: yield* d.membership(grace, organizationId, ["inactive"]), + none: yield* d.membership(outsider.user.id, organizationId), + batch: yield* d.membersById(organizationId, [ada, linus, outsider.user.id, "nobody"]), + byEmail: yield* d.findByEmail(organizationId, "ada.lovelace@placeholder.test"), + unknown: yield* d.findByEmail(organizationId, "outsider@placeholder.test"), + }; + }), + ); + expect(result.one?.role).toBe("member"); + expect(result.oneInactive, "Better Auth members are always active").toBeNull(); + expect(result.none).toBeNull(); + expect([...result.batch.keys()].sort()).toEqual([ada, linus].sort()); + expect(result.byEmail?.accountId).toBe(ada); + expect(result.unknown, "a user with no membership is not a member").toBeNull(); + }); +}); diff --git a/apps/host-selfhost/src/auth/member-directory.ts b/apps/host-selfhost/src/auth/member-directory.ts new file mode 100644 index 0000000000..4c84841174 --- /dev/null +++ b/apps/host-selfhost/src/auth/member-directory.ts @@ -0,0 +1,223 @@ +// --------------------------------------------------------------------------- +// Self-host's `MemberDirectory`: the shared read seam over Better Auth's +// organization `member` table joined with `user`. +// +// Reads go through Better Auth's OWN adapter (`auth.$context` → `adapter`) +// rather than `auth.api.listMembers`, for the same reason `mcp/auth.ts` reads +// the membership row that way: the adapter needs no session headers, so the +// HTTP, MCP, and admin planes can all resolve membership through this one code +// path. Better Auth members carry no status (an invitation is not a member), +// so every member reports `"active"`; roles are the plugin's own slugs +// (`owner` / `admin` / `member`), verbatim. +// +// The adapter has no case-insensitive predicate and no join filter, so search +// and email matching run in memory over the org's member list — a single-org +// self-host instance is small, and one code path answering every read is +// worth more than an indexed lookup here. +// --------------------------------------------------------------------------- + +import { Effect, Layer, Schema } from "effect"; + +import { + DEFAULT_MEMBER_STATUSES, + MemberDirectory, + MemberDirectoryError, + normalizeAdminUserEmail, + normalizeMemberSearch, + type DirectoryMember, + type MemberDirectoryShape, + type MemberQuery, + type MemberStatus, +} from "@executor-js/api/server"; + +import { BetterAuth, type BetterAuthHandle } from "./better-auth"; + +// What the adapter hands back is untyped (`findMany` trusts its caller), so +// each row is decoded at this boundary. Extra columns are dropped. +const MemberRow = Schema.Struct({ + id: Schema.String, + userId: Schema.String, + organizationId: Schema.String, + role: Schema.String, +}); +const UserRow = Schema.Struct({ + id: Schema.String, + email: Schema.String, + name: Schema.NullishOr(Schema.String), + image: Schema.NullishOr(Schema.String), +}); +const decodeMemberRows = Schema.decodeUnknownEffect(Schema.Array(MemberRow)); +const decodeUserRows = Schema.decodeUnknownEffect(Schema.Array(UserRow)); + +// The adapter caps every `findMany` at 100 rows unless told otherwise, so +// reads page explicitly until a short page. `in` predicates are chunked so a +// large id list never overruns SQLite's bound-parameter limit. +const PAGE_SIZE = 500; +const IN_CHUNK = 200; + +type BetterAuthAdapter = Awaited["adapter"]; +type AdapterWhere = Parameters[0]["where"]; + +const byEmailThenAccount = (a: DirectoryMember, b: DirectoryMember): number => { + // Nulls sort last, matching Postgres's default ASC ordering on cloud. + if (a.email !== b.email) { + if (a.email === null) return 1; + if (b.email === null) return -1; + return a.email < b.email ? -1 : 1; + } + return a.accountId < b.accountId ? -1 : a.accountId > b.accountId ? 1 : 0; +}; + +const makeService = (adapter: BetterAuthAdapter): MemberDirectoryShape => { + const read = (op: string, fn: () => Promise): Effect.Effect => + Effect.tryPromise(fn).pipe( + Effect.tapCause((cause) => Effect.logError(`member_directory.${op} failed`, cause)), + Effect.mapError( + () => + new MemberDirectoryError({ + message: `Failed to read the member directory (${op})`, + }), + ), + Effect.withSpan(`member_directory.${op}`), + ); + + const undecodable = (op: string) => () => + new MemberDirectoryError({ + message: `Undecodable member directory row (${op})`, + }); + + const memberRows = (op: string, where: AdapterWhere) => + Effect.gen(function* () { + const rows: (typeof MemberRow)["Type"][] = []; + for (let offset = 0; ; offset += PAGE_SIZE) { + const page = yield* read(op, () => + adapter.findMany({ + model: "member", + where, + limit: PAGE_SIZE, + offset, + }), + ).pipe(Effect.flatMap(decodeMemberRows), Effect.mapError(undecodable(op))); + rows.push(...page); + if (page.length < PAGE_SIZE) return rows; + } + }); + + const userRows = (op: string, userIds: readonly string[]) => + Effect.gen(function* () { + const users = new Map(); + for (let start = 0; start < userIds.length; start += IN_CHUNK) { + const ids = userIds.slice(start, start + IN_CHUNK); + const page = yield* read(op, () => + adapter.findMany({ + model: "user", + where: [{ field: "id", operator: "in", value: [...ids] }], + limit: ids.length, + }), + ).pipe(Effect.flatMap(decodeUserRows), Effect.mapError(undecodable(op))); + for (const user of page) users.set(user.id, user); + } + return users; + }); + + // Every read: the member rows matching `where`, joined to their users. A + // member whose user row is gone is not reported — there is no principal + // behind it to name. + const load = (op: string, where: AdapterWhere) => + Effect.gen(function* () { + const rows = yield* memberRows(op, where); + const users = yield* userRows( + op, + rows.map((row) => row.userId), + ); + const members: DirectoryMember[] = []; + for (const row of rows) { + const user = users.get(row.userId); + if (user === undefined) continue; + members.push({ + accountId: row.userId, + membershipId: row.id, + organizationId: row.organizationId, + email: user.email, + name: user.name ?? null, + avatarUrl: user.image ?? null, + role: row.role, + status: "active", + lastActiveAt: null, + }); + } + return members; + }); + + const orgWhere = (organizationId: string): AdapterWhere => [ + { field: "organizationId", value: organizationId }, + ]; + + const matches = (member: DirectoryMember, term: string): boolean => + (member.email !== null && member.email.toLowerCase().includes(term)) || + (member.name !== null && member.name.toLowerCase().includes(term)); + + // Every Better Auth member is active; a query for other statuses only has + // nothing to report. + const reportsActive = (statuses: readonly MemberStatus[]) => statuses.includes("active"); + + return { + membership: (accountId, organizationId, statuses = DEFAULT_MEMBER_STATUSES) => + !reportsActive(statuses) + ? Effect.succeed(null) + : load("membership", [ + { field: "userId", value: accountId }, + { field: "organizationId", value: organizationId }, + ]).pipe(Effect.map((members) => members[0] ?? null)), + + members: (organizationId, query: MemberQuery = {}) => + Effect.gen(function* () { + if (!reportsActive(query.statuses ?? DEFAULT_MEMBER_STATUSES)) return []; + const term = normalizeMemberSearch(query.search); + const all = yield* load("members", orgWhere(organizationId)); + const matched = term === undefined ? all : all.filter((m) => matches(m, term)); + matched.sort(byEmailThenAccount); + const start = query.offset ?? 0; + const end = query.limit === undefined ? undefined : start + query.limit; + return matched.slice(start, end); + }), + + membersById: (organizationId, accountIds, statuses = DEFAULT_MEMBER_STATUSES) => + Effect.gen(function* () { + const found = new Map(); + if (!reportsActive(statuses)) return found; + for (let start = 0; start < accountIds.length; start += IN_CHUNK) { + const ids = accountIds.slice(start, start + IN_CHUNK); + const members = yield* load("membersById", [ + { field: "organizationId", value: organizationId }, + { field: "userId", operator: "in", value: [...ids] }, + ]); + for (const member of members) found.set(member.accountId, member); + } + return found; + }), + + findByEmail: (organizationId, email, statuses = DEFAULT_MEMBER_STATUSES) => + !reportsActive(statuses) + ? Effect.succeed(null) + : load("findByEmail", orgWhere(organizationId)).pipe( + Effect.map( + (members) => + members.find( + (member) => + member.email !== null && normalizeAdminUserEmail(member.email) === email, + ) ?? null, + ), + ), + }; +}; + +/** The self-host `MemberDirectory` over the boot-scoped Better Auth handle. */ +export const betterAuthMemberDirectoryLayer: Layer.Layer = + Layer.effect(MemberDirectory)( + Effect.gen(function* () { + const { auth } = yield* BetterAuth; + const { adapter } = yield* Effect.promise(() => auth.$context); + return MemberDirectory.of(makeService(adapter)); + }), + ); diff --git a/packages/core/api/src/admin/member-directory.ts b/packages/core/api/src/admin/member-directory.ts new file mode 100644 index 0000000000..e90ad89346 --- /dev/null +++ b/packages/core/api/src/admin/member-directory.ts @@ -0,0 +1,50 @@ +// --------------------------------------------------------------------------- +// The admin users plane's directory, derived from the shared `MemberDirectory` +// seam — so each host's `AdminUsersProvider` no longer carries its own +// identity join and email resolver. +// --------------------------------------------------------------------------- + +import { Effect } from "effect"; + +import { MemberStatus, type MemberDirectoryShape } from "../server/member-directory"; +import type { AdminUserDirectory, AdminUserIdentity } from "./reads"; + +/** + * Both directions of the admin plane's directory over one org's + * {@link MemberDirectoryShape}. + * + * `identities` is one batched `membersById` read for the page of ids (never a + * lookup per user); a member the org does not hold reports absent identity. + * `resolveEmail` receives the already-normalized email the contract promises + * and answers with the host principal id, or `null` when no member has it. + * + * Both read ANY membership status, not the directory's active + pending + * default: this plane reports footprint, not current access. A member who was + * removed while their connections remain must still be named on the users + * page and findable by the address an operator has for them. + * + * Both fail with `MemberDirectoryError`, which the shared reads treat as a + * decorative-join outage (identities) or surface as a failed read (resolve). + */ +export const adminUserDirectoryFromMembers = ( + directory: MemberDirectoryShape, + organizationId: string, +): AdminUserDirectory => ({ + identities: (externalIds) => + directory.membersById(organizationId, externalIds, MemberStatus.literals).pipe( + Effect.map((members) => { + const identities = new Map(); + for (const [accountId, member] of members) { + identities.set(accountId, { + email: member.email, + displayName: member.name, + }); + } + return identities; + }), + ), + resolveEmail: (email) => + directory + .findByEmail(organizationId, email, MemberStatus.literals) + .pipe(Effect.map((member) => (member === null ? null : member.accountId))), +}); diff --git a/packages/core/api/src/server.ts b/packages/core/api/src/server.ts index 104d841c39..b5974de98d 100644 --- a/packages/core/api/src/server.ts +++ b/packages/core/api/src/server.ts @@ -106,6 +106,17 @@ export { type IdentityProviderShape, type IdentityFailure, } from "./server/identity"; +export { + MemberDirectory, + MemberDirectoryError, + MemberStatus, + DEFAULT_MEMBER_STATUSES, + normalizeMemberSearch, + type DirectoryMember, + type MemberQuery, + type MemberDirectoryShape, +} from "./server/member-directory"; +export { adminUserDirectoryFromMembers } from "./admin/member-directory"; export { makeExecutionStackMiddleware, textFailureStrategy, diff --git a/packages/core/api/src/server/member-directory.ts b/packages/core/api/src/server/member-directory.ts new file mode 100644 index 0000000000..50bef4a342 --- /dev/null +++ b/packages/core/api/src/server/member-directory.ts @@ -0,0 +1,142 @@ +// --------------------------------------------------------------------------- +// MemberDirectory — the ONE shared READ seam over "who belongs to this org". +// +// Sits beside `IdentityProvider` (./identity.ts) as the second provider-neutral +// auth surface. `IdentityProvider` answers "who is calling"; this answers "who +// is a member, with what role and status" — the question every member list, +// admin users page, seat count, and per-request membership check asks. Cloud +// (WorkOS) implements it over a LOCAL mirror of WorkOS users + memberships +// (fed by login, write-through, and the WorkOS Events API); self-host (Better +// Auth) implements it over its own `member` + `user` tables. Shared code +// consumes only this tag and never learns which host it is on. +// +// Read-only by design. Writes stay host-specific: cloud writes go to WorkOS +// and are mirrored back; self-host writes go through Better Auth's org plugin. +// Invitations are NOT members and are not reported here. +// --------------------------------------------------------------------------- + +import { Context, Effect, Schema } from "effect"; + +/** + * Membership lifecycle as the host stores it. `pending` is a member who has + * not completed joining (cloud: an accepted-but-unactivated WorkOS membership); + * `inactive` is a member who keeps their row but must not be granted access — + * a deactivated member, or (on cloud) the TOMBSTONE of a deleted membership: + * the mirror never drops a membership row, it marks it inactive with the + * deletion time, so a feeder replaying an older payload cannot resurrect it. + * Every read excludes `inactive` unless the caller names it in `statuses`; + * anything that grants access must additionally require `active`. + */ +export const MemberStatus = Schema.Literals(["active", "pending", "inactive"]); +export type MemberStatus = typeof MemberStatus.Type; + +/** + * One member of one organization, as the directory reports it. + * + * `accountId` is the host principal id — the SAME id space `IdentityProvider` + * binds as `Principal.accountId` and the subject table records in + * `external_id` (cloud: the WorkOS `user_…`; self-host: the Better Auth + * `user.id`). `membershipId` is the host's membership ROW id (`om_…` on cloud, + * `member.id` on self-host) and joins to nothing outside the host; it is + * carried for host-specific writes (remove, change role), never as a join key. + */ +export interface DirectoryMember { + readonly accountId: string; + readonly membershipId: string; + readonly organizationId: string; + readonly email: string | null; + readonly name: string | null; + readonly avatarUrl: string | null; + /** The host's role slug as stored (`"admin"` | `"member"` | `"owner"` …), not normalized. */ + readonly role: string; + readonly status: MemberStatus; + /** Epoch ms of the member's last sign-in, when the host records it. */ + readonly lastActiveAt: number | null; +} + +/** + * Filter + paging for {@link MemberDirectoryShape.members}. + * + * `search` is a case-insensitive substring match over email and name; the + * adapter trims + lower-cases it (the same rule `normalizeEmail` applies to + * emails) and an empty term is no filter. `statuses` defaults to active + + * pending. Results are ordered by email then `accountId` so paging is stable. + */ +export interface MemberQuery { + readonly search?: string; + readonly limit?: number; + readonly offset?: number; + readonly statuses?: readonly MemberStatus[]; +} + +export interface MemberDirectoryShape { + /** + * One account's membership in one org, or `null` when it holds none among + * `statuses` (default: active + pending, so a tombstoned membership reads + * as no membership). + */ + readonly membership: ( + accountId: string, + organizationId: string, + statuses?: readonly MemberStatus[], + ) => Effect.Effect; + /** The org's members matching `query` (see {@link MemberQuery} for defaults). */ + readonly members: ( + organizationId: string, + query?: MemberQuery, + ) => Effect.Effect; + /** + * The org's members among `accountIds` with a status in `statuses` + * (default: active + pending), keyed by `accountId`. Ids the org holds no + * such membership for are simply absent. One read for the whole batch — + * never a lookup per id. + */ + readonly membersById: ( + organizationId: string, + accountIds: readonly string[], + statuses?: readonly MemberStatus[], + ) => Effect.Effect, MemberDirectoryError>; + /** + * The org's member with this email among `statuses` (default: active + + * pending). `email` arrives ALREADY normalized (trimmed + lower-cased) and + * is compared against the normalized directory value, so casing never + * decides the answer on either host. + */ + readonly findByEmail: ( + organizationId: string, + email: string, + statuses?: readonly MemberStatus[], + ) => Effect.Effect; +} + +export class MemberDirectory extends Context.Service()( + "@executor-js/api/MemberDirectory", +) {} + +/** + * The directory could not be read (storage fault, undecodable row). Flat + * message only: the cause is logged by the adapter and deliberately not echoed + * to a caller. + */ +export class MemberDirectoryError extends Schema.TaggedErrorClass()( + "MemberDirectoryError", + { message: Schema.String }, +) {} + +/** + * The search-term normalization every adapter applies: trim + lower-case, the + * same rule `normalizeEmail` applies to emails. `undefined` means no filter, + * including for a blank term. + */ +export const normalizeMemberSearch = (search: string | undefined): string | undefined => { + if (search === undefined) return undefined; + const term = search.trim().toLowerCase(); + return term.length === 0 ? undefined : term; +}; + +/** + * The statuses every read reports when the caller names none: the members + * who hold or are joining the org. `inactive` — deactivated or tombstoned — + * is never reported by default. + */ +export const DEFAULT_MEMBER_STATUSES: readonly MemberStatus[] = ["active", "pending"];