Graphical PDS migrator for AT Protocol
knotbin.com
1/**
2 * Test environment utilities for setting up virtual PDS instances
3 */
4
5import { Agent } from "@atproto/api";
6import { TestBsky, TestPds, TestPlc } from "@atproto/dev-env";
7import { ComAtprotoServerCreateAccount } from "@atproto/api";
8import { SMTPServer, SMTPServerAddress } from "smtp-server";
9import * as cheerio from "cheerio";
10import consola from "consola";
11
12export interface TestPDSConfig {
13 sourcePds: TestPds;
14 targetPds: TestPds;
15 plc: TestPlc;
16}
17
18const PLC_PORT = 2580;
19const PDS_A_PORT = 2583;
20const PDS_B_PORT = 2584;
21const SMTP_PORT = 2525;
22
23export type TestAccount = ComAtprotoServerCreateAccount.OutputSchema & {
24 agent: Agent;
25};
26
27export interface Email {
28 to: SMTPServerAddress[];
29 from: false | SMTPServerAddress;
30 subject?: string;
31 body: string;
32 verificationCode?: string;
33 timestamp: Date;
34}
35
36/**
37 * Create a test environment with virtual PDS instances
38 */
39export class TestEnvironment {
40 sourcePds: TestPds;
41 targetPds: TestPds;
42 plc: TestPlc;
43 smtp!: SMTPServer;
44 private mailbox: Email[] = [];
45 private codeWaiters: Map<string, (code: string) => void> = new Map();
46
47 private constructor(
48 sourcePds: TestPds,
49 targetPds: TestPds,
50 plc: TestPlc,
51 ) {
52 this.sourcePds = sourcePds;
53 this.targetPds = targetPds;
54 this.plc = plc;
55 }
56
57 static async create(): Promise<TestEnvironment> {
58 const plc = await TestPlc.create({
59 port: PLC_PORT,
60 });
61
62 const pds = await TestEnvironment.setupMockPDS(plc.url);
63 const env = new TestEnvironment(
64 pds.sourcePds,
65 pds.targetPds,
66 plc,
67 );
68
69 env.smtp = await env.createSMTPServer();
70
71 return env;
72 }
73
74 /**
75 * Get all emails in the mailbox
76 */
77 getMail(): Email[] {
78 return [...this.mailbox];
79 }
80
81 /**
82 * Clear all emails from the mailbox
83 */
84 clearMail(): void {
85 this.mailbox = [];
86 }
87
88 /**
89 * Wait for a verification code to arrive in an email
90 * @param timeoutMs - Timeout in milliseconds
91 * @returns Promise that resolves with the verification code
92 */
93 awaitMail(timeoutMs: number = 10000): Promise<string> | string {
94 // First, check if a verification code is already present in the mailbox
95 const existing = this.mailbox.find((email) => email.verificationCode);
96 if (existing && existing.verificationCode) {
97 return existing.verificationCode;
98 }
99
100 // Otherwise, wait for a new code to arrive
101 return new Promise((resolve, reject) => {
102 const timeout = setTimeout(() => {
103 this.codeWaiters.delete(waiterId);
104 reject(
105 new Error(
106 `Timeout waiting for verification code after ${timeoutMs}ms`,
107 ),
108 );
109 }, timeoutMs);
110
111 const waiterId = Date.now().toString();
112 this.codeWaiters.set(waiterId, (code: string) => {
113 clearTimeout(timeout);
114 this.codeWaiters.delete(waiterId);
115 resolve(code);
116 });
117 });
118 }
119
120 private createSMTPServer(): Promise<SMTPServer> {
121 return new Promise((resolve) => {
122 const server = new SMTPServer({
123 // disable STARTTLS to allow authentication in clear text mode
124 disabledCommands: ["STARTTLS", "AUTH"],
125 allowInsecureAuth: true,
126 hideSTARTTLS: true,
127 onData: (stream, session, callback) => {
128 let emailData = "";
129 stream.on("data", (chunk) => emailData += chunk.toString("utf8"));
130 stream.on("end", () => {
131 const $ = cheerio.load(emailData);
132 const codeEl = $("code").first();
133 const code = codeEl.length ? codeEl.text() : undefined;
134
135 const email: Email = {
136 to: session.envelope.rcptTo,
137 from: session.envelope.mailFrom,
138 body: emailData,
139 verificationCode: code,
140 timestamp: new Date(),
141 };
142
143 this.mailbox.push(email);
144
145 if (code) {
146 consola.info(`Verification code arrived: ${code}`);
147 // Notify all waiting promises
148 this.codeWaiters.forEach((resolve) => resolve(code));
149 this.codeWaiters.clear();
150 } else {
151 consola.info("No <code> found in email.");
152 }
153 });
154
155 callback();
156 },
157 });
158
159 server.listen(SMTP_PORT, () => {
160 consola.info(`SMTP server listening on port ${SMTP_PORT}`);
161 resolve(server);
162 });
163 });
164 }
165
166 private static async setupMockPDS(plcUrl: string) {
167 const sourcePds = await TestPds.create({
168 didPlcUrl: plcUrl,
169 port: PDS_A_PORT,
170 inviteRequired: false,
171 devMode: true,
172 emailSmtpUrl: `smtp://localhost:${SMTP_PORT}`,
173 emailFromAddress: `noreply@localhost:${SMTP_PORT}`,
174 bskyAppViewDid: "did:web:api.bsky.app",
175 });
176
177 const targetPds = await TestPds.create({
178 didPlcUrl: plcUrl,
179 port: PDS_B_PORT,
180 inviteRequired: false,
181 acceptingImports: true,
182 devMode: true,
183 emailSmtpUrl: `smtp://localhost:${SMTP_PORT}`,
184 emailFromAddress: `noreply@localhost:${SMTP_PORT}`,
185 bskyAppViewDid: "did:web:api.bsky.app",
186 });
187
188 return {
189 sourcePds,
190 targetPds,
191 };
192 }
193
194 async cleanup() {
195 try {
196 await this.sourcePds.close();
197 await this.targetPds.close();
198 await this.plc.close();
199 await new Promise<void>((resolve) => {
200 this.smtp.close(resolve);
201 });
202 } catch (error) {
203 console.error("Error during cleanup:", error);
204 }
205 }
206}