diff --git a/.changeset/auto-provision-digest.md b/.changeset/auto-provision-digest.md new file mode 100644 index 0000000000..b61aa6782c --- /dev/null +++ b/.changeset/auto-provision-digest.md @@ -0,0 +1,23 @@ +--- +--- + +feat(notifications): daily digest of new auto-provisioned members for org admins + +The consent receipt for the `auto_provision_verified_domain` default. With auto-add on (which it is by default), org membership grows quietly when verified-domain emails sign in — owners had no signal that their seat list was changing. This adds a daily Slack digest that lists the new auto-joined members per org and links to the team page where the owner can review or flip the toggle off. + +## Mechanics + +- New per-org watermark `organizations.last_auto_provision_digest_sent_at`. Migration 437. +- New `findOrgsWithNewAutoProvisionedMembers()` and `listNewAutoProvisionedMembers(orgId, since)` queries find members where `provisioning_source = 'verified_domain'` and `created_at > watermark`. Skips personal workspaces and orgs with `auto_provision_verified_domain = false`. +- `server/src/scheduled/auto-provision-digest.ts` runs the check every 24 hours (5-minute startup delay so it doesn't fire during boot's noisy window). Uses the same Slack DM dispatch helper (`sendToOrgAdmins`) that the seat-request reminder job uses, so multi-admin orgs get a group DM and single-admin orgs get a direct DM. +- Watermark is only updated after a successful Slack delivery. If no admins are mapped to Slack, or delivery fails, the watermark stays put and the next run retries — eventually surfaces the news once an admin's Slack mapping shows up. + +## Tests + +- `server/tests/integration/membership-webhook.test.ts` — 31 → 37 tests. Six new cases cover: candidates filtered to verified-domain since watermark, opt-out flag honored, NULL watermark = beginning of time, members returned chronologically with non-verified-domain sources excluded, and `markAutoProvisionDigestSent` updating the timestamp. + +## Future + +- Email fallback for orgs without Slack mappings (defer — Slack is the primary admin surface today). +- Per-org cadence preference (defer — daily is the right default for a low-volume notification). +- "What's in this digest" preview accessible from the team page (defer — wire into the next admin UI cycle). diff --git a/.changeset/provisioning-source-attribution.md b/.changeset/provisioning-source-attribution.md new file mode 100644 index 0000000000..785d163761 --- /dev/null +++ b/.changeset/provisioning-source-attribution.md @@ -0,0 +1,20 @@ +--- +--- + +feat(membership): track provisioning source on each org membership + +Each row in `organization_memberships` now carries a `provisioning_source` tag identifying how it came to exist: + +- `verified_domain` — `autoLinkByVerifiedDomain` matched the user's email domain to a verified org with auto-provision on +- `invited` — accepted via `POST /:orgId/invitations` or `/members/by-email` Path 1 +- `admin_added` — created via `/members/by-email` Path 2 (admin/owner direct add) +- `webhook` — surfaced by an `organization_membership.created` event with no staged source (e.g. someone added the membership directly in the WorkOS dashboard) +- `null` — pre-existing rows that haven't been touched since this migration + +The originating endpoint stages source + seat_type into `invitation_seat_types` (a new `source` column there). The `organization_membership.created` webhook handler reads both back via `consumeInvitationSeatType` and writes `provisioning_source` on the local cache row. The upsert preserves an existing source on conflict, so a subsequent webhook upsert can't wipe a more-specific origin. + +Sets up the new-member digest in the auto-provision notification feature so org owners can see which auto-joined members showed up via verified-domain vs. were explicitly invited. + +## Migration + +`436_organization_membership_provisioning_source.sql` adds the columns and an index keyed on `(workos_organization_id, provisioning_source, created_at DESC)` for the digest query. diff --git a/server/src/db/membership-db.ts b/server/src/db/membership-db.ts index c6fbfb73f6..50f7bce632 100644 --- a/server/src/db/membership-db.ts +++ b/server/src/db/membership-db.ts @@ -13,6 +13,14 @@ const logger = createLogger('membership-db'); // ── Types ──────────────────────────────────────────────────────────── +/** Tracks how each organization_memberships row came to exist. */ +export type ProvisioningSource = + | 'verified_domain' // autoLinkByVerifiedDomain + | 'invited' // POST /:orgId/invitations or /members/by-email Path 1 + | 'admin_added' // /members/by-email Path 2 direct add + | 'webhook' // organization_membership.created with no staged source + | 'unknown'; + export interface MembershipUpsertParams { user_id: string; organization_id: string; @@ -23,6 +31,13 @@ export interface MembershipUpsertParams { role: string; // raw role slug from WorkOS (e.g. 'member', 'admin', 'owner') seat_type: string; // resolved seat type ('contributor' | 'community_only') has_explicit_seat_type: boolean; + /** + * Provisioning source to record on the local cache row. Only written when + * the row is being inserted (or when the existing row has NULL/'unknown' + * source) — once a membership is tagged, subsequent webhook upserts don't + * overwrite the original attribution. + */ + provisioning_source?: ProvisioningSource; } export interface MembershipUpsertResult { @@ -57,6 +72,7 @@ export async function upsertOrganizationMembership( last_name, role, seat_type, + provisioning_source, synced_at ) VALUES ( $1, $2, $3, $4, $5, $6, @@ -70,7 +86,7 @@ export async function upsertOrganizationMembership( WHEN $7 = '__auto__' THEN 'member' ELSE $7 END, - $8, NOW() + $8, $10, NOW() ) ON CONFLICT (workos_user_id, workos_organization_id) DO UPDATE SET @@ -83,6 +99,9 @@ export async function upsertOrganizationMembership( WHEN $9::boolean THEN EXCLUDED.seat_type ELSE organization_memberships.seat_type END, + -- Don't overwrite an existing attribution; later webhooks would be the + -- 'webhook' source and would otherwise wipe a more specific origin. + provisioning_source = COALESCE(organization_memberships.provisioning_source, EXCLUDED.provisioning_source), synced_at = NOW(), updated_at = NOW() RETURNING role`, @@ -96,6 +115,7 @@ export async function upsertOrganizationMembership( effectiveRole, params.seat_type, params.has_explicit_seat_type, + params.provisioning_source ?? null, ], ); @@ -136,23 +156,28 @@ export async function deleteOrganizationMembership( // ── Invitation seat type ───────────────────────────────────────────── /** - * Consume any pending seat_type assignment from an invitation. - * Returns the seat type if one was found, or null. + * Consume any pending seat_type and provisioning_source staged by the endpoint + * that triggered the membership creation. Returns both fields when a row is + * found; null when no staging row exists. */ export async function consumeInvitationSeatType( organizationId: string, email: string, -): Promise { +): Promise<{ seat_type: string; source: ProvisioningSource | null } | null> { const pool = getPool(); - const result = await pool.query<{ seat_type: string }>( + const result = await pool.query<{ seat_type: string; source: string | null }>( `DELETE FROM invitation_seat_types WHERE workos_organization_id = $1 AND lower(email) = lower($2) - RETURNING seat_type`, + RETURNING seat_type, source`, [organizationId, email], ); - return result.rows[0]?.seat_type ?? null; + if (!result.rows[0]) return null; + return { + seat_type: result.rows[0].seat_type, + source: (result.rows[0].source as ProvisioningSource | null) ?? null, + }; } // ── Successor promotion query ──────────────────────────────────────── @@ -260,6 +285,23 @@ export async function autoLinkByVerifiedDomain( if (userAlreadyMember) return null; + // Stage the provisioning source so the organization_membership.created + // webhook handler can record 'verified_domain' on the local cache row. + // Clear any stale (org, email) staging row first; consumeInvitationSeatType + // matches by (org, email) and we don't want a leftover row from a prior + // failed attempt to win. + const stagingKey = `verified_domain_${orgId}_${userId}`; + await pool.query( + 'DELETE FROM invitation_seat_types WHERE workos_organization_id = $1 AND lower(email) = lower($2)', + [orgId, email], + ); + await pool.query( + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, 'community_only', 'verified_domain') + ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type, source = EXCLUDED.source`, + [stagingKey, orgId, email], + ); + // Always create as member. Auto-promotion to owner for ownerless orgs is // handled atomically by upsertOrganizationMembership when the // organization_membership.created webhook fires — that path uses a NOT EXISTS @@ -280,7 +322,131 @@ export async function autoLinkByVerifiedDomain( logger.info({ userId, orgId }, 'Auto-link skipped: membership already exists in WorkOS'); return { organizationId: orgId, organizationName: orgName, role: 'member' }; } + // Roll back the staging row so a stale 'verified_domain' source can't be + // consumed by an unrelated future invite for the same (org, email) pair. + try { + await pool.query( + 'DELETE FROM invitation_seat_types WHERE workos_invitation_id = $1', + [stagingKey], + ); + } catch (rollbackErr) { + logger.error( + { err: rollbackErr, userId, orgId, email, stagingKey }, + 'CRITICAL: failed to rollback verified_domain staging row after createOrganizationMembership failure — manually delete row to avoid source leak', + ); + } logger.warn({ err, userId, orgId }, 'Failed to auto-link user to organization'); return null; } } + +// ── Auto-provision digest queries ─────────────────────────────────── + +/** + * Row in the auto-provision digest payload — one per newly-auto-joined member + * since the org's last digest watermark. + */ +export interface NewAutoProvisionedMember { + workos_user_id: string; + email: string; + first_name: string | null; + last_name: string | null; + role: string; + seat_type: string; + joined_at: Date; +} + +/** + * Find orgs that have at least one auto-provisioned member since their last + * digest watermark. Returns one row per org (with the org's owner emails and + * the count of new members) so the caller can iterate. + * + * Skip personal workspaces and orgs with auto_provision_verified_domain=false + * (the latter shouldn't have any verified-domain members anyway, but the join + * makes the intent explicit). + */ +export async function findOrgsWithNewAutoProvisionedMembers(): Promise< + Array<{ + workos_organization_id: string; + org_name: string; + last_sent_at: Date | null; + new_member_count: number; + }> +> { + const pool = getPool(); + const result = await pool.query<{ + workos_organization_id: string; + org_name: string; + last_sent_at: Date | null; + new_member_count: string; // pg COUNT comes back as string + }>(` + SELECT + o.workos_organization_id, + o.name AS org_name, + o.last_auto_provision_digest_sent_at AS last_sent_at, + COUNT(om.workos_user_id) AS new_member_count + FROM organizations o + JOIN organization_memberships om + ON om.workos_organization_id = o.workos_organization_id + WHERE om.provisioning_source = 'verified_domain' + AND om.created_at > COALESCE(o.last_auto_provision_digest_sent_at, 'epoch'::timestamptz) + AND COALESCE(o.is_personal, false) = false + AND COALESCE(o.auto_provision_verified_domain, true) = true + GROUP BY o.workos_organization_id, o.name, o.last_auto_provision_digest_sent_at + HAVING COUNT(om.workos_user_id) > 0 + `); + + return result.rows.map(r => ({ + workos_organization_id: r.workos_organization_id, + org_name: r.org_name, + last_sent_at: r.last_sent_at, + new_member_count: parseInt(r.new_member_count, 10), + })); +} + +/** + * List the auto-provisioned members added to a given org since the watermark. + * Used to build the digest body once findOrgsWithNewAutoProvisionedMembers has + * filtered to orgs with non-zero counts. + */ +export async function listNewAutoProvisionedMembers( + organizationId: string, + since: Date | null, +): Promise { + const pool = getPool(); + const sinceTs = since ?? new Date(0); + const result = await pool.query(` + SELECT + workos_user_id, + email, + first_name, + last_name, + role, + seat_type, + created_at AS joined_at + FROM organization_memberships + WHERE workos_organization_id = $1 + AND provisioning_source = 'verified_domain' + AND created_at > $2 + ORDER BY created_at ASC + `, [organizationId, sinceTs]); + + return result.rows; +} + +/** + * Mark the digest as sent for an organization. Called after successful delivery + * so the next run skips the same members. + */ +export async function markAutoProvisionDigestSent( + organizationId: string, + sentAt: Date = new Date(), +): Promise { + const pool = getPool(); + await pool.query( + `UPDATE organizations + SET last_auto_provision_digest_sent_at = $2, updated_at = NOW() + WHERE workos_organization_id = $1`, + [organizationId, sentAt], + ); +} diff --git a/server/src/db/migrations/436_organization_membership_provisioning_source.sql b/server/src/db/migrations/436_organization_membership_provisioning_source.sql new file mode 100644 index 0000000000..98f4cacba3 --- /dev/null +++ b/server/src/db/migrations/436_organization_membership_provisioning_source.sql @@ -0,0 +1,30 @@ +-- Track how each organization membership came to exist. +-- +-- Existing rows are NULL ('unknown'). New rows are tagged at creation by the +-- code path that triggered them: autoLinkByVerifiedDomain → 'verified_domain', +-- /members/by-email Path 1 → 'invited', Path 2 → 'admin_added', +-- POST /:orgId/invitations → 'invited', everything else (sync from WorkOS, +-- webhook for an externally-created membership) → 'webhook'. +-- +-- Used by the new-member digest in the auto-provision notification feature so +-- org owners can see which auto-joined members showed up via verified-domain +-- vs. were explicitly invited. + +ALTER TABLE organization_memberships + ADD COLUMN IF NOT EXISTS provisioning_source VARCHAR(32); + +COMMENT ON COLUMN organization_memberships.provisioning_source IS + 'How this membership was created: verified_domain, invited, admin_added, webhook, unknown'; + +CREATE INDEX IF NOT EXISTS idx_organization_memberships_provisioning_source + ON organization_memberships(workos_organization_id, provisioning_source, created_at DESC) + WHERE provisioning_source IS NOT NULL; + +-- Mirror on the invitation_seat_types staging table so the webhook handler can +-- read the source set by the originating endpoint and apply it to the local +-- cache row when the membership.created event arrives. +ALTER TABLE invitation_seat_types + ADD COLUMN IF NOT EXISTS source VARCHAR(32); + +COMMENT ON COLUMN invitation_seat_types.source IS + 'Provisioning source to apply to the membership when this staging row is consumed'; diff --git a/server/src/db/migrations/437_auto_provision_digest_sent_at.sql b/server/src/db/migrations/437_auto_provision_digest_sent_at.sql new file mode 100644 index 0000000000..734b2c2753 --- /dev/null +++ b/server/src/db/migrations/437_auto_provision_digest_sent_at.sql @@ -0,0 +1,13 @@ +-- Track the last time the auto-provision new-member digest was sent for an +-- organization. The scheduled job uses this to find members auto-joined since +-- the previous digest (no per-member tracking, just a watermark per org). +-- +-- NULL means we've never sent — the first run after this migration deploys +-- will look back at the full window of `verified_domain` provisioning_source +-- members and include all of them in that org's first digest. + +ALTER TABLE organizations + ADD COLUMN IF NOT EXISTS last_auto_provision_digest_sent_at TIMESTAMPTZ; + +COMMENT ON COLUMN organizations.last_auto_provision_digest_sent_at IS + 'Watermark for the auto-provision new-member digest. Updated when the digest is successfully delivered. NULL = never sent.'; diff --git a/server/src/http.ts b/server/src/http.ts index 6d7d416b89..ea338c7425 100644 --- a/server/src/http.ts +++ b/server/src/http.ts @@ -8647,6 +8647,12 @@ ${p.category ? `${p.category}\n` : ''}${publishedUrl}< import('./scheduled/seat-request-reminders.js').then(({ startSeatRequestReminders }) => { startSeatRequestReminders(workos!); }).catch(err => logger.warn({ err }, 'Failed to start seat request reminders')); + + // Daily auto-provision new-member digest for org admins/owners. + // Consent receipt for the auto_provision_verified_domain default. + import('./scheduled/auto-provision-digest.js').then(({ startAutoProvisionDigest }) => { + startAutoProvisionDigest(workos!); + }).catch(err => logger.warn({ err }, 'Failed to start auto-provision digest')); } // Start Luma calendar sync (catches events missed by webhooks) @@ -8701,6 +8707,10 @@ ${p.category ? `${p.category}\n` : ''}${publishedUrl}< stopSeatRequestReminders(); }).catch(() => {}); + import('./scheduled/auto-provision-digest.js').then(({ stopAutoProvisionDigest }) => { + stopAutoProvisionDigest(); + }).catch(() => {}); + import('./luma/sync.js').then(({ stopLumaSync }) => { stopLumaSync(); }).catch(() => {}); diff --git a/server/src/routes/organizations.ts b/server/src/routes/organizations.ts index b6acd21f2c..6009d9b719 100644 --- a/server/src/routes/organizations.ts +++ b/server/src/routes/organizations.ts @@ -2400,11 +2400,11 @@ export function createOrganizationsRouter(): Router { roleSlug: role || 'member', }); - // Persist seat_type intent for when the invitation is accepted + // Persist seat_type intent + provisioning source for when the invitation is accepted await query( - `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type) - VALUES ($1, $2, $3, $4) - ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type`, + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, $4, 'invited') + ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type, source = EXCLUDED.source`, [invitation.id, orgId, email, seatType] ); @@ -2575,10 +2575,10 @@ export function createOrganizationsRouter(): Router { roleSlug: 'member', }); - // Persist seat_type intent for the new invitation + // Persist seat_type intent + provisioning source for the new invitation await query( - `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type) - VALUES ($1, $2, $3, $4)`, + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, $4, 'invited')`, [newInvitation.id, orgId, invitation.email, preservedSeatType] ); @@ -2751,12 +2751,13 @@ export function createOrganizationsRouter(): Router { roleSlug: 'member', }); - // Persist seat_type intent so the webhook handler picks it up when the - // invitee accepts (mirrors the existing /invitations endpoint). + // Persist seat_type intent + provisioning source so the webhook + // handler picks them up when the invitee accepts (mirrors the + // existing /invitations endpoint). await query( - `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type) - VALUES ($1, $2, $3, $4) - ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type`, + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, $4, 'invited') + ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type, source = EXCLUDED.source`, [invitation.id, orgId, normalizedEmail, seatType], ); @@ -2835,9 +2836,9 @@ export function createOrganizationsRouter(): Router { ); const stagingKey = `direct_${orgId}_${targetUserId}`; await query( - `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type) - VALUES ($1, $2, $3, $4) - ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type`, + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, $4, 'admin_added') + ON CONFLICT (workos_invitation_id) DO UPDATE SET seat_type = EXCLUDED.seat_type, source = EXCLUDED.source`, [stagingKey, orgId, normalizedEmail, seatType], ); diff --git a/server/src/routes/workos-webhooks.ts b/server/src/routes/workos-webhooks.ts index 630525e93d..b9c6c25b0b 100644 --- a/server/src/routes/workos-webhooks.ts +++ b/server/src/routes/workos-webhooks.ts @@ -122,10 +122,14 @@ async function upsertMembership( const role = membership.role?.slug || 'member'; - // Consume any pending seat_type assignment from the invitation - const consumedSeatType = await consumeInvitationSeatType(membership.organization_id, userData.email); - const hasExplicitSeatType = consumedSeatType !== null; - const seatType = consumedSeatType || 'community_only'; + // Consume any pending seat_type + provisioning_source staged by the + // endpoint that triggered this membership creation. Falls back to defaults + // when no row was staged (e.g. someone added the membership directly in + // WorkOS rather than through one of our endpoints). + const consumed = await consumeInvitationSeatType(membership.organization_id, userData.email); + const hasExplicitSeatType = consumed !== null; + const seatType = consumed?.seat_type || 'community_only'; + const provisioningSource = consumed?.source || 'webhook'; const { assigned_role } = await upsertOrganizationMembership({ user_id: membership.user_id, @@ -137,6 +141,7 @@ async function upsertMembership( role, seat_type: seatType, has_explicit_seat_type: hasExplicitSeatType, + provisioning_source: provisioningSource, }); // If the DB promoted this member to owner, sync the change to WorkOS diff --git a/server/src/scheduled/auto-provision-digest.ts b/server/src/scheduled/auto-provision-digest.ts new file mode 100644 index 0000000000..3e697c4789 --- /dev/null +++ b/server/src/scheduled/auto-provision-digest.ts @@ -0,0 +1,198 @@ +/** + * Daily digest of new auto-provisioned members for org owners/admins. + * + * Why this exists: with `auto_provision_verified_domain` defaulting to ON, + * the policy quietly grows org membership when a verified-domain email signs + * in. Without a notification, owners have no signal that their seat list is + * changing — exactly the SOC2-flavored finding the adtech-product reviewer + * flagged on the original PR. This is the consent receipt: every org that had + * auto-joined members since its last digest gets a Slack message listing + * them, with a link to the team page where the owner can review or flip the + * toggle off. + * + * Mechanics: + * - Watermark per org (`organizations.last_auto_provision_digest_sent_at`). + * - Once a day, find orgs with at least one verified-domain membership + * created since the watermark. Skip silently if zero. + * - Send a Slack DM/group-DM to the org's admin/owner cluster (matches the + * existing seat-request reminder dispatch pattern). + * - Mark the org as digested only after a successful send so failures retry. + */ + +import { WorkOS } from '@workos-inc/node'; +import { createLogger } from '../logger.js'; +import { + findOrgsWithNewAutoProvisionedMembers, + listNewAutoProvisionedMembers, + markAutoProvisionDigestSent, + type NewAutoProvisionedMember, +} from '../db/membership-db.js'; +import { sendToOrgAdmins, escapeSlackMrkdwn } from '../slack/org-group-dm.js'; +import { getOrgAdminEmails } from '../utils/org-admins.js'; + +const logger = createLogger('auto-provision-digest'); + +const APP_URL = process.env.APP_URL || 'https://agenticadvertising.org'; +// Run once a day. The watermark column is the per-org cooldown — a fresh +// candidate set with non-zero new members triggers a send for that org. +const CHECK_INTERVAL_MS = 24 * 60 * 60 * 1000; +// Initial delay so the job doesn't fire during boot's noisy window. +const INITIAL_DELAY_MS = 5 * 60 * 1000; + +let intervalId: ReturnType | null = null; +let initialTimeoutId: ReturnType | null = null; + +export function startAutoProvisionDigest(workos: WorkOS): void { + if (intervalId || initialTimeoutId) return; + + initialTimeoutId = setTimeout(() => { + initialTimeoutId = null; + runDigest(workos).catch(err => + logger.error({ err }, 'Auto-provision digest failed'), + ); + intervalId = setInterval(() => { + runDigest(workos).catch(err => + logger.error({ err }, 'Auto-provision digest failed'), + ); + }, CHECK_INTERVAL_MS); + }, INITIAL_DELAY_MS); + + logger.info('Auto-provision digest scheduler started'); +} + +export function stopAutoProvisionDigest(): void { + if (initialTimeoutId) { + clearTimeout(initialTimeoutId); + initialTimeoutId = null; + } + if (intervalId) { + clearInterval(intervalId); + intervalId = null; + } +} + +/** + * One pass: find candidate orgs, send a digest to each, mark sent. + * Exported for tests + manual invocation from incident scripts. + */ +export async function runDigest(workos: WorkOS): Promise<{ + candidateOrgs: number; + delivered: number; + skipped: number; + failed: number; +}> { + const candidates = await findOrgsWithNewAutoProvisionedMembers(); + let delivered = 0; + let skipped = 0; + let failed = 0; + + for (const candidate of candidates) { + try { + const members = await listNewAutoProvisionedMembers( + candidate.workos_organization_id, + candidate.last_sent_at, + ); + if (members.length === 0) { + // Race: members were removed between findOrgs and list. Skip silently. + skipped++; + continue; + } + + const adminEmails = await getOrgAdminEmails(workos, candidate.workos_organization_id); + if (adminEmails.length === 0) { + // No admins/owners to notify. Don't update watermark — try again next + // run, in case an admin gets added. + logger.info( + { orgId: candidate.workos_organization_id, memberCount: members.length }, + 'Auto-provision digest: no admins to notify, deferring', + ); + skipped++; + continue; + } + + const message = buildSlackMessage(candidate.org_name, candidate.workos_organization_id, members); + const sent = await sendToOrgAdmins( + candidate.workos_organization_id, + adminEmails, + message, + ); + + if (sent) { + await markAutoProvisionDigestSent(candidate.workos_organization_id); + delivered++; + logger.info( + { + orgId: candidate.workos_organization_id, + memberCount: members.length, + adminEmailCount: adminEmails.length, + }, + 'Auto-provision digest delivered', + ); + } else { + // Slack delivery failed (no admins on Slack, channel issue, etc.). Don't + // mark watermark so the next run retries; future Slack-mapped admins + // will pick it up. + skipped++; + logger.info( + { orgId: candidate.workos_organization_id, memberCount: members.length }, + 'Auto-provision digest: slack delivery skipped', + ); + } + } catch (err) { + failed++; + logger.warn( + { err, orgId: candidate.workos_organization_id }, + 'Auto-provision digest failed for org', + ); + } + } + + if (candidates.length > 0) { + logger.info( + { candidateOrgs: candidates.length, delivered, skipped, failed }, + 'Auto-provision digest pass complete', + ); + } + + return { candidateOrgs: candidates.length, delivered, skipped, failed }; +} + +function buildSlackMessage( + orgName: string, + orgId: string, + members: NewAutoProvisionedMember[], +): { text: string; blocks: any[] } { + const teamUrl = `${APP_URL}/team?org=${orgId}`; + const memberLines = members.map(m => { + const displayName = [m.first_name, m.last_name].filter(Boolean).join(' ').trim(); + const namePart = displayName ? `${escapeSlackMrkdwn(displayName)} (${escapeSlackMrkdwn(m.email)})` : escapeSlackMrkdwn(m.email); + const date = m.joined_at instanceof Date + ? m.joined_at.toISOString().slice(0, 10) + : String(m.joined_at).slice(0, 10); + return `• ${namePart} — joined ${date}`; + }); + + const summary = members.length === 1 + ? `1 person joined *${escapeSlackMrkdwn(orgName)}* via verified-domain auto-add since the last digest.` + : `${members.length} people joined *${escapeSlackMrkdwn(orgName)}* via verified-domain auto-add since the last digest.`; + + return { + text: `${members.length} new auto-joined member${members.length === 1 ? '' : 's'} in ${orgName}`, + blocks: [ + { + type: 'section', + text: { + type: 'mrkdwn', + text: `${summary}\n\n${memberLines.join('\n')}`, + }, + }, + { + type: 'section', + text: { + type: 'mrkdwn', + text: `<${teamUrl}|Review or change roles> · turn auto-add off in the *Verified Domains* card on the same page if you'd rather invite explicitly.`, + }, + }, + ], + }; +} diff --git a/server/tests/integration/membership-webhook.test.ts b/server/tests/integration/membership-webhook.test.ts index 1e8982fed2..606c8ee607 100644 --- a/server/tests/integration/membership-webhook.test.ts +++ b/server/tests/integration/membership-webhook.test.ts @@ -16,6 +16,9 @@ import { findSuccessorForPromotion, setMembershipRole, autoLinkByVerifiedDomain, + findOrgsWithNewAutoProvisionedMembers, + listNewAutoProvisionedMembers, + markAutoProvisionDigestSent, } from '../../src/db/membership-db.js'; import type { WorkOS } from '@workos-inc/node'; import type { Pool } from 'pg'; @@ -264,15 +267,15 @@ describe('Membership webhook DB operations', () => { // ========================================================================= describe('consumeInvitationSeatType', () => { - it('returns and deletes the pending seat type', async () => { + it('returns and deletes the pending seat type and source', async () => { await pool.query( - `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type) - VALUES ($1, $2, $3, $4)`, - ['inv_test_1', TEST_ORG_ID, 'invited@test.com', 'contributor'], + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, $4, $5)`, + ['inv_test_1', TEST_ORG_ID, 'invited@test.com', 'contributor', 'invited'], ); const result = await consumeInvitationSeatType(TEST_ORG_ID, 'invited@test.com'); - expect(result).toBe('contributor'); + expect(result).toEqual({ seat_type: 'contributor', source: 'invited' }); // Should be consumed (deleted) const second = await consumeInvitationSeatType(TEST_ORG_ID, 'invited@test.com'); @@ -280,14 +283,27 @@ describe('Membership webhook DB operations', () => { }); it('matches case-insensitively', async () => { + await pool.query( + `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type, source) + VALUES ($1, $2, $3, $4, $5)`, + ['inv_test_2', TEST_ORG_ID, 'CamelCase@Test.com', 'contributor', 'invited'], + ); + + const result = await consumeInvitationSeatType(TEST_ORG_ID, 'camelcase@test.com'); + expect(result?.seat_type).toBe('contributor'); + expect(result?.source).toBe('invited'); + }); + + it('returns null source when staging row predates the source column', async () => { + // Backward compatibility: rows written before migration 436 have NULL source. await pool.query( `INSERT INTO invitation_seat_types (workos_invitation_id, workos_organization_id, email, seat_type) VALUES ($1, $2, $3, $4)`, - ['inv_test_2', TEST_ORG_ID, 'CamelCase@Test.com', 'contributor'], + ['inv_test_legacy', TEST_ORG_ID, 'legacy@test.com', 'community_only'], ); - const result = await consumeInvitationSeatType(TEST_ORG_ID, 'camelcase@test.com'); - expect(result).toBe('contributor'); + const result = await consumeInvitationSeatType(TEST_ORG_ID, 'legacy@test.com'); + expect(result).toEqual({ seat_type: 'community_only', source: null }); }); it('returns null when no invitation exists', async () => { @@ -571,5 +587,222 @@ describe('Membership webhook DB operations', () => { `DELETE FROM organization_memberships WHERE workos_organization_id = 'org_personal_autolink_user'`, ); }); + + it('stages provisioning_source=verified_domain so the webhook can record it', async () => { + await seedOrgWithVerifiedDomain('autolink.com'); + const workos = makeWorkOSMock(); + + const result = await autoLinkByVerifiedDomain(workos, AUTOLINK_USER, 'matt@autolink.com'); + expect(result).not.toBeNull(); + + // The staging row should now exist for the org+email pair so the + // organization_membership.created webhook can consume it. + const staged = await pool.query<{ seat_type: string; source: string | null }>( + `SELECT seat_type, source FROM invitation_seat_types + WHERE workos_organization_id = $1 AND lower(email) = lower($2)`, + [TEST_AUTOLINK_ORG_ID, 'matt@autolink.com'], + ); + expect(staged.rows[0]).toBeDefined(); + expect(staged.rows[0].seat_type).toBe('community_only'); + expect(staged.rows[0].source).toBe('verified_domain'); + }); + }); + + describe('Auto-provision digest queries', () => { + const DIGEST_ORG_ID = 'org_digest_test'; + const DIGEST_USER_NEW = 'user_digest_new'; + const DIGEST_USER_OLD = 'user_digest_old'; + const DIGEST_USER_INVITED = 'user_digest_invited'; + + beforeEach(async () => { + await pool.query('DELETE FROM organization_memberships WHERE workos_organization_id = $1', [DIGEST_ORG_ID]); + await pool.query('DELETE FROM organizations WHERE workos_organization_id = $1', [DIGEST_ORG_ID]); + await pool.query( + `INSERT INTO organizations (workos_organization_id, name, is_personal, subscription_status, auto_provision_verified_domain, last_auto_provision_digest_sent_at, created_at, updated_at) + VALUES ($1, 'Digest Test Org', false, 'active', true, $2, NOW(), NOW())`, + [DIGEST_ORG_ID, new Date('2026-04-01T00:00:00Z')], + ); + }); + + afterAll(async () => { + await pool.query('DELETE FROM organization_memberships WHERE workos_organization_id = $1', [DIGEST_ORG_ID]); + await pool.query('DELETE FROM organizations WHERE workos_organization_id = $1', [DIGEST_ORG_ID]); + }); + + async function seedMember(opts: { + userId: string; + email: string; + source: 'verified_domain' | 'invited' | 'admin_added'; + createdAt: Date; + }) { + await pool.query( + `INSERT INTO organization_memberships + (workos_user_id, workos_organization_id, email, role, seat_type, provisioning_source, created_at, updated_at, synced_at) + VALUES ($1, $2, $3, 'member', 'community_only', $4, $5, $5, $5)`, + [opts.userId, DIGEST_ORG_ID, opts.email, opts.source, opts.createdAt], + ); + } + + it('finds orgs with new verified_domain members since the watermark', async () => { + // Member joined BEFORE watermark — excluded. + await seedMember({ + userId: DIGEST_USER_OLD, + email: 'old@digest.com', + source: 'verified_domain', + createdAt: new Date('2026-03-15T00:00:00Z'), + }); + // Member joined AFTER watermark — included. + await seedMember({ + userId: DIGEST_USER_NEW, + email: 'new@digest.com', + source: 'verified_domain', + createdAt: new Date('2026-04-15T00:00:00Z'), + }); + // Invited member — wrong source, excluded. + await seedMember({ + userId: DIGEST_USER_INVITED, + email: 'invited@digest.com', + source: 'invited', + createdAt: new Date('2026-04-15T00:00:00Z'), + }); + + const rows = await findOrgsWithNewAutoProvisionedMembers(); + const target = rows.find(r => r.workos_organization_id === DIGEST_ORG_ID); + expect(target).toBeDefined(); + expect(target!.new_member_count).toBe(1); + expect(target!.org_name).toBe('Digest Test Org'); + }); + + it('skips orgs with auto_provision_verified_domain disabled', async () => { + await pool.query( + 'UPDATE organizations SET auto_provision_verified_domain = false WHERE workos_organization_id = $1', + [DIGEST_ORG_ID], + ); + await seedMember({ + userId: DIGEST_USER_NEW, + email: 'new@digest.com', + source: 'verified_domain', + createdAt: new Date('2026-04-15T00:00:00Z'), + }); + + const rows = await findOrgsWithNewAutoProvisionedMembers(); + expect(rows.find(r => r.workos_organization_id === DIGEST_ORG_ID)).toBeUndefined(); + }); + + it('skips orgs with no new members since watermark', async () => { + await seedMember({ + userId: DIGEST_USER_OLD, + email: 'old@digest.com', + source: 'verified_domain', + createdAt: new Date('2026-03-15T00:00:00Z'), // before watermark + }); + + const rows = await findOrgsWithNewAutoProvisionedMembers(); + expect(rows.find(r => r.workos_organization_id === DIGEST_ORG_ID)).toBeUndefined(); + }); + + it('treats NULL watermark as the beginning of time', async () => { + await pool.query( + 'UPDATE organizations SET last_auto_provision_digest_sent_at = NULL WHERE workos_organization_id = $1', + [DIGEST_ORG_ID], + ); + await seedMember({ + userId: DIGEST_USER_NEW, + email: 'new@digest.com', + source: 'verified_domain', + createdAt: new Date('2026-01-01T00:00:00Z'), + }); + + const rows = await findOrgsWithNewAutoProvisionedMembers(); + const target = rows.find(r => r.workos_organization_id === DIGEST_ORG_ID); + expect(target).toBeDefined(); + expect(target!.new_member_count).toBe(1); + expect(target!.last_sent_at).toBeNull(); + }); + + it('lists members chronologically, excluding non-verified-domain sources', async () => { + const t1 = new Date('2026-04-10T00:00:00Z'); + const t2 = new Date('2026-04-12T00:00:00Z'); + const t3 = new Date('2026-04-14T00:00:00Z'); + + await seedMember({ userId: 'u_b', email: 'b@digest.com', source: 'verified_domain', createdAt: t2 }); + await seedMember({ userId: 'u_a', email: 'a@digest.com', source: 'verified_domain', createdAt: t1 }); + await seedMember({ userId: 'u_c', email: 'c@digest.com', source: 'verified_domain', createdAt: t3 }); + await seedMember({ userId: 'u_inv', email: 'inv@digest.com', source: 'invited', createdAt: t2 }); + + const members = await listNewAutoProvisionedMembers(DIGEST_ORG_ID, new Date('2026-04-01T00:00:00Z')); + expect(members.map(m => m.email)).toEqual(['a@digest.com', 'b@digest.com', 'c@digest.com']); + }); + + it('markAutoProvisionDigestSent updates the watermark', async () => { + const sent = new Date('2026-04-26T12:00:00Z'); + await markAutoProvisionDigestSent(DIGEST_ORG_ID, sent); + const row = await pool.query<{ last_auto_provision_digest_sent_at: Date }>( + 'SELECT last_auto_provision_digest_sent_at FROM organizations WHERE workos_organization_id = $1', + [DIGEST_ORG_ID], + ); + expect(row.rows[0].last_auto_provision_digest_sent_at.toISOString()).toBe(sent.toISOString()); + }); + }); + + describe('upsertOrganizationMembership provisioning_source', () => { + it('writes provisioning_source on insert', async () => { + await upsertOrganizationMembership({ + user_id: TEST_USER_1, + organization_id: TEST_ORG_ID, + membership_id: 'om_source_test', + email: 'src@test.com', + first_name: 'Src', + last_name: 'Test', + role: 'member', + seat_type: 'community_only', + has_explicit_seat_type: false, + provisioning_source: 'verified_domain', + }); + + const row = await pool.query<{ provisioning_source: string | null }>( + 'SELECT provisioning_source FROM organization_memberships WHERE workos_user_id = $1 AND workos_organization_id = $2', + [TEST_USER_1, TEST_ORG_ID], + ); + expect(row.rows[0].provisioning_source).toBe('verified_domain'); + }); + + it('preserves an existing provisioning_source on subsequent upserts', async () => { + // Initial insert tags the row. + await upsertOrganizationMembership({ + user_id: TEST_USER_1, + organization_id: TEST_ORG_ID, + membership_id: 'om_pres_1', + email: 'pres@test.com', + first_name: null, + last_name: null, + role: 'member', + seat_type: 'community_only', + has_explicit_seat_type: false, + provisioning_source: 'admin_added', + }); + + // Subsequent webhook upsert with a less-specific source must not overwrite. + await upsertOrganizationMembership({ + user_id: TEST_USER_1, + organization_id: TEST_ORG_ID, + membership_id: 'om_pres_1', + email: 'pres@test.com', + first_name: null, + last_name: null, + role: 'admin', + seat_type: 'community_only', + has_explicit_seat_type: false, + provisioning_source: 'webhook', + }); + + const row = await pool.query<{ provisioning_source: string | null; role: string }>( + 'SELECT provisioning_source, role FROM organization_memberships WHERE workos_user_id = $1 AND workos_organization_id = $2', + [TEST_USER_1, TEST_ORG_ID], + ); + expect(row.rows[0].provisioning_source).toBe('admin_added'); + // Role still updates normally. + expect(row.rows[0].role).toBe('admin'); + }); }); });