|
| 1 | +import type { PrismaClientOrTransaction } from "~/db.server"; |
| 2 | +import type { AttioUserSync, AttioWorkspaceSync } from "./attio.server"; |
| 3 | +import { logger } from "./logger.server"; |
| 4 | + |
| 5 | +const ATTIO_API = "https://api.attio.com/v2"; |
| 6 | + |
| 7 | +class AttioRequestError extends Error { |
| 8 | + // 5xx/429 are transient (the worker retries); tagged warn so they don't page Sentry. |
| 9 | + readonly transient: boolean; |
| 10 | + readonly logLevel?: "warn"; |
| 11 | + |
| 12 | + constructor( |
| 13 | + object: string, |
| 14 | + readonly status: number, |
| 15 | + readonly code: string | undefined, |
| 16 | + body: string |
| 17 | + ) { |
| 18 | + super(`Attio assert ${object} failed with status ${status}: ${body.slice(0, 500)}`); |
| 19 | + this.transient = status >= 500 || status === 429; |
| 20 | + if (this.transient) this.logLevel = "warn"; |
| 21 | + } |
| 22 | +} |
| 23 | + |
| 24 | +export type AttioClientOptions = { |
| 25 | + apiKey: string; |
| 26 | + // Marks records written from non-production environments. |
| 27 | + isTest: boolean; |
| 28 | + baseUrl?: string; |
| 29 | +}; |
| 30 | + |
| 31 | +export class AttioClient { |
| 32 | + constructor(private readonly options: AttioClientOptions) {} |
| 33 | + |
| 34 | + // Create-or-update by unique attribute; returns the record id. Throws on failure so the worker retries. |
| 35 | + async #assert( |
| 36 | + object: string, |
| 37 | + matchingAttribute: string, |
| 38 | + values: Record<string, unknown> |
| 39 | + ): Promise<string> { |
| 40 | + const baseUrl = this.options.baseUrl ?? ATTIO_API; |
| 41 | + const url = `${baseUrl}/objects/${object}/records?matching_attribute=${matchingAttribute}`; |
| 42 | + const response = await fetch(url, { |
| 43 | + method: "PUT", |
| 44 | + headers: { |
| 45 | + Authorization: `Bearer ${this.options.apiKey}`, |
| 46 | + "Content-Type": "application/json", |
| 47 | + }, |
| 48 | + body: JSON.stringify({ data: { values } }), |
| 49 | + }); |
| 50 | + |
| 51 | + if (!response.ok) { |
| 52 | + const body = await response.text(); |
| 53 | + throw new AttioRequestError(object, response.status, errorCode(body), body); |
| 54 | + } |
| 55 | + |
| 56 | + const recordId = ((await response.json()) as any).data?.id?.record_id; |
| 57 | + if (typeof recordId !== "string") { |
| 58 | + throw new Error(`Attio assert ${object}: response missing data.id.record_id`); |
| 59 | + } |
| 60 | + return recordId; |
| 61 | + } |
| 62 | + |
| 63 | + // The org's creator as an Attio admin user. Returns undefined when Attio rejects the user, |
| 64 | + // so the workspace is still written; transient failures throw so the job retries. |
| 65 | + async #assertAdminUser(userId: string, email: string | undefined) { |
| 66 | + const values = { user_id: userId, role: "Admin", is_test: this.options.isTest }; |
| 67 | + try { |
| 68 | + return await this.#assert("users", "user_id", values); |
| 69 | + } catch (error) { |
| 70 | + if (!(error instanceof AttioRequestError) || error.transient) throw error; |
| 71 | + let rejection = error; |
| 72 | + |
| 73 | + // No Attio user yet, and creating one requires the email. |
| 74 | + if (rejection.code === "missing_value" && email) { |
| 75 | + try { |
| 76 | + return await this.#assert("users", "user_id", { |
| 77 | + ...values, |
| 78 | + primary_email_address: email, |
| 79 | + }); |
| 80 | + } catch (createError) { |
| 81 | + if (!(createError instanceof AttioRequestError) || createError.transient) { |
| 82 | + throw createError; |
| 83 | + } |
| 84 | + rejection = createError; |
| 85 | + } |
| 86 | + |
| 87 | + // A concurrent user sync may have created the user meanwhile. |
| 88 | + if (rejection.code === "uniqueness_conflict") { |
| 89 | + try { |
| 90 | + return await this.#assert("users", "user_id", values); |
| 91 | + } catch (retryError) { |
| 92 | + if (!(retryError instanceof AttioRequestError) || retryError.transient) { |
| 93 | + throw retryError; |
| 94 | + } |
| 95 | + } |
| 96 | + } |
| 97 | + } |
| 98 | + |
| 99 | + logger.warn("Attio admin user not synced, writing the workspace without it", { |
| 100 | + userId, |
| 101 | + error: rejection.message, |
| 102 | + }); |
| 103 | + return undefined; |
| 104 | + } |
| 105 | + } |
| 106 | + |
| 107 | + async upsertWorkspace(payload: AttioWorkspaceSync, adminEmail?: string) { |
| 108 | + const adminRecordId = await this.#assertAdminUser(payload.adminUserId, adminEmail); |
| 109 | + |
| 110 | + await this.#assert("workspaces", "workspace_id", { |
| 111 | + workspace_id: payload.orgId, |
| 112 | + name: payload.title, |
| 113 | + org_slug: payload.slug, |
| 114 | + company_size: payload.companySize ?? undefined, |
| 115 | + email_domain: domainFromEmail(adminEmail), |
| 116 | + signup_date: toDate(payload.createdAt), |
| 117 | + plan: "Free", |
| 118 | + account_status: "Active", |
| 119 | + is_test: this.options.isTest, |
| 120 | + users: adminRecordId |
| 121 | + ? [{ target_object: "users", target_record_id: adminRecordId }] |
| 122 | + : undefined, |
| 123 | + }); |
| 124 | + } |
| 125 | + |
| 126 | + async upsertUser(payload: AttioUserSync) { |
| 127 | + await this.#assert("users", "user_id", { |
| 128 | + user_id: payload.userId, |
| 129 | + primary_email_address: payload.email, |
| 130 | + marketing_opt_in: payload.marketingEmails, |
| 131 | + referral_source: payload.referralSource ?? undefined, |
| 132 | + signup_date: toDate(payload.createdAt), |
| 133 | + is_test: this.options.isTest, |
| 134 | + }); |
| 135 | + } |
| 136 | +} |
| 137 | + |
| 138 | +// Attio `date` attributes want a bare YYYY-MM-DD value. |
| 139 | +function toDate(date: Date): string { |
| 140 | + return date.toISOString().slice(0, 10); |
| 141 | +} |
| 142 | + |
| 143 | +// Attio error bodies look like {"code":"missing_value",...}. |
| 144 | +function errorCode(body: string): string | undefined { |
| 145 | + try { |
| 146 | + const code = (JSON.parse(body) as { code?: unknown } | null)?.code; |
| 147 | + return typeof code === "string" ? code : undefined; |
| 148 | + } catch { |
| 149 | + return undefined; |
| 150 | + } |
| 151 | +} |
| 152 | + |
| 153 | +// Domain from an email; the cloud-side matcher normalizes it further. |
| 154 | +function domainFromEmail(email: string | undefined): string | undefined { |
| 155 | + return email?.split("@")[1]?.toLowerCase().trim() || undefined; |
| 156 | +} |
| 157 | + |
| 158 | +// Writes the org's workspace, looking up its admin's email for the user record and domain. |
| 159 | +export async function syncAttioWorkspace( |
| 160 | + payload: AttioWorkspaceSync, |
| 161 | + client: AttioClient, |
| 162 | + db: PrismaClientOrTransaction |
| 163 | +) { |
| 164 | + const admin = await db.user.findFirst({ |
| 165 | + where: { id: payload.adminUserId }, |
| 166 | + select: { email: true }, |
| 167 | + }); |
| 168 | + await client.upsertWorkspace(payload, admin?.email); |
| 169 | +} |
0 commit comments