|
| 1 | +import { appendFile, chmod, readFile } from "node:fs/promises"; |
| 2 | +import { parseArgs } from "node:util"; |
| 3 | +import { LoopsApi, LoopsApiError } from "./api"; |
| 4 | +import { type ContactProfile, emailHash } from "./profile"; |
| 5 | +import { importContactUpdate, type RemoteContact } from "./sync-policy"; |
| 6 | + |
| 7 | +type Receipt = { |
| 8 | + hash: string; |
| 9 | + status: "created" | "existing" | "updated" | "invalid"; |
| 10 | + id?: string; |
| 11 | + detail?: unknown; |
| 12 | +}; |
| 13 | +const { values } = parseArgs({ |
| 14 | + options: { |
| 15 | + contacts: { type: "string" }, |
| 16 | + receipt: { type: "string" }, |
| 17 | + team: { type: "string" }, |
| 18 | + apply: { type: "boolean", default: false }, |
| 19 | + limit: { type: "string" }, |
| 20 | + "mailing-list": { type: "string" }, |
| 21 | + "retry-invalid": { type: "boolean", default: false }, |
| 22 | + }, |
| 23 | +}); |
| 24 | +if (!values.contacts || !values.receipt || !values.team) |
| 25 | + throw new Error("Pass --contacts, --receipt and --team"); |
| 26 | +const filename = values.receipt; |
| 27 | +const profiles: ContactProfile[] = JSON.parse( |
| 28 | + await readFile(values.contacts, "utf8"), |
| 29 | +); |
| 30 | +const hashes = new Set<string>(); |
| 31 | +for (const profile of profiles) { |
| 32 | + const hash = emailHash(profile.email); |
| 33 | + if ( |
| 34 | + hashes.has(hash) || |
| 35 | + !profile.subscribed || |
| 36 | + profile.capConsent !== "subscribed" || |
| 37 | + profile.capLifecycleEnabled || |
| 38 | + profile.capOnboardingEligible || |
| 39 | + profile.capLifecycleStage !== "idle" |
| 40 | + ) |
| 41 | + throw new Error( |
| 42 | + "Import must contain unique, positively subscribed, held contacts", |
| 43 | + ); |
| 44 | + hashes.add(hash); |
| 45 | +} |
| 46 | +const prior: Receipt[] = []; |
| 47 | +try { |
| 48 | + for (const line of (await readFile(filename, "utf8")) |
| 49 | + .split("\n") |
| 50 | + .filter(Boolean)) |
| 51 | + prior.push(JSON.parse(line)); |
| 52 | +} catch (error) { |
| 53 | + if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) |
| 54 | + throw error; |
| 55 | +} |
| 56 | +const latest = new Map(prior.map((row) => [row.hash, row])); |
| 57 | +const done = new Set( |
| 58 | + [...latest.values()] |
| 59 | + .filter( |
| 60 | + (row) => |
| 61 | + row.status !== "existing" && |
| 62 | + !(values["retry-invalid"] && row.status === "invalid"), |
| 63 | + ) |
| 64 | + .map((row) => row.hash), |
| 65 | +); |
| 66 | +const limit = values.limit ? Number(values.limit) : profiles.length; |
| 67 | +if (!Number.isSafeInteger(limit) || limit < 1) throw new Error("Invalid limit"); |
| 68 | +const pending = profiles |
| 69 | + .filter((profile) => !done.has(emailHash(profile.email))) |
| 70 | + .slice(0, limit); |
| 71 | +const api = new LoopsApi(process.env.LOOPS_API_KEY ?? "", 150); |
| 72 | +const identity = await api.request<{ success: boolean; teamName: string }>( |
| 73 | + "api-key", |
| 74 | +); |
| 75 | +if (!identity.success || identity.teamName !== values.team) |
| 76 | + throw new Error("Wrong Loops team"); |
| 77 | +const workflows = await api.list<{ id: string }>("workflows"); |
| 78 | +for (const row of workflows) { |
| 79 | + const workflow = await api.request<{ status: string }>(`workflows/${row.id}`); |
| 80 | + if (workflow.status !== "Draft") |
| 81 | + throw new Error("Contact import requires all workflows to remain drafts"); |
| 82 | +} |
| 83 | +const lists = await api.request<{ id: string; name: string }[]>("lists"); |
| 84 | +const listId = |
| 85 | + values["mailing-list"] ?? |
| 86 | + lists.find((list) => list.name === "Product updates and tips")?.id; |
| 87 | +if (listId && !lists.some((list) => list.id === listId)) |
| 88 | + throw new Error("Unknown mailing list"); |
| 89 | +console.log( |
| 90 | + JSON.stringify({ |
| 91 | + team: identity.teamName, |
| 92 | + mode: values.apply ? "create missing contacts" : "dry run", |
| 93 | + total: profiles.length, |
| 94 | + alreadyProcessed: done.size, |
| 95 | + pending: pending.length, |
| 96 | + mailingListId: listId ?? null, |
| 97 | + }), |
| 98 | +); |
| 99 | +if (!values.apply) process.exit(0); |
| 100 | +await appendFile(filename, "", { mode: 0o600 }); |
| 101 | +await chmod(filename, 0o600); |
| 102 | +let next = 0; |
| 103 | +let stopped = false; |
| 104 | +let completed = 0; |
| 105 | +const counts = { created: 0, existing: 0, updated: 0, invalid: 0 }; |
| 106 | +const worker = async () => { |
| 107 | + while (!stopped && next < pending.length) { |
| 108 | + const profile = pending[next++]; |
| 109 | + if (!profile) return; |
| 110 | + const hash = emailHash(profile.email); |
| 111 | + let receipt: Receipt; |
| 112 | + try { |
| 113 | + const result = await api.request<{ success: boolean; id: string }>( |
| 114 | + "contacts/create", |
| 115 | + "POST", |
| 116 | + { ...profile, ...(listId ? { mailingLists: { [listId]: true } } : {}) }, |
| 117 | + ); |
| 118 | + if (!result.success || !result.id) |
| 119 | + throw new Error( |
| 120 | + "Unexpected create response; reconcile before retrying", |
| 121 | + ); |
| 122 | + receipt = { hash, status: "created", id: result.id }; |
| 123 | + } catch (error) { |
| 124 | + if (error instanceof LoopsApiError && error.status === 409) { |
| 125 | + const contacts = await api.request<RemoteContact[]>( |
| 126 | + `contacts/find?email=${encodeURIComponent(profile.email)}`, |
| 127 | + ); |
| 128 | + const contact = contacts[0]; |
| 129 | + if (contacts.length !== 1 || !contact) { |
| 130 | + stopped = true; |
| 131 | + throw new Error( |
| 132 | + "Resolve the existing contact identity before resuming", |
| 133 | + ); |
| 134 | + } |
| 135 | + await api.request( |
| 136 | + "contacts/update", |
| 137 | + "PUT", |
| 138 | + importContactUpdate(profile, contact, listId), |
| 139 | + ); |
| 140 | + receipt = { hash, status: "updated", id: contact.id }; |
| 141 | + } else if (error instanceof LoopsApiError && error.status === 400) |
| 142 | + receipt = { hash, status: "invalid", detail: error.details }; |
| 143 | + else { |
| 144 | + stopped = true; |
| 145 | + throw error; |
| 146 | + } |
| 147 | + } |
| 148 | + await appendFile(filename, `${JSON.stringify(receipt)}\n`); |
| 149 | + counts[receipt.status]++; |
| 150 | + completed++; |
| 151 | + if (completed % 250 === 0) |
| 152 | + console.log( |
| 153 | + JSON.stringify({ processed: done.size + completed, ...counts }), |
| 154 | + ); |
| 155 | + } |
| 156 | +}; |
| 157 | +const results = await Promise.allSettled( |
| 158 | + Array.from({ length: 8 }, async () => { |
| 159 | + try { |
| 160 | + await worker(); |
| 161 | + } catch (error) { |
| 162 | + stopped = true; |
| 163 | + throw error; |
| 164 | + } |
| 165 | + }), |
| 166 | +); |
| 167 | +console.log( |
| 168 | + JSON.stringify({ |
| 169 | + processed: done.size + completed, |
| 170 | + ...counts, |
| 171 | + remaining: profiles.length - done.size - completed, |
| 172 | + }), |
| 173 | +); |
| 174 | +for (const result of results) |
| 175 | + if (result.status === "rejected") { |
| 176 | + console.error( |
| 177 | + result.reason instanceof LoopsApiError |
| 178 | + ? `Import stopped with HTTP ${result.reason.status}; resume with the same receipt` |
| 179 | + : "Import stopped; reconcile the receipt before resuming", |
| 180 | + ); |
| 181 | + process.exitCode = 1; |
| 182 | + } |
0 commit comments