PDS Admin tool make it easier to moderate your PDS with labels
mailqueue
+35
-19
+12
-9
src/handlers/handleNewLabel.ts
+12
-9
src/handlers/handleNewLabel.ts
···
5
5
import type { LibSQLDatabase } from "drizzle-orm/libsql";
6
6
import * as schema from "../db/schema.js";
7
7
import { and, eq } from "drizzle-orm";
8
+
import type PQueue from "p-queue";
8
9
9
10
export const handleNewLabel = async (
10
11
config: LabelerConfig,
11
12
label: Label,
12
13
db: LibSQLDatabase<typeof schema>,
13
14
pdsConfigs: Record<string, PDSConfig>,
15
+
mailQueue: PQueue,
14
16
) => {
15
17
try {
16
18
let targetDid = "";
···
110
112
111
113
// Perform action
112
114
if (labelConfig.action === "notify") {
113
-
//TODO need to prob move this to a queue cause backfill can hit rate limit
114
-
await sendLabelNotification(pdsConfig.notifyEmails, {
115
-
did: targetDid,
116
-
pds: pdsConfig.host,
117
-
label: label.val,
118
-
labeler: config.host,
119
-
negated: label.neg ?? false,
120
-
dateApplied: labledDate,
121
-
});
115
+
mailQueue.add(() =>
116
+
sendLabelNotification(pdsConfig.notifyEmails, {
117
+
did: targetDid,
118
+
pds: pdsConfig.host,
119
+
label: label.val,
120
+
labeler: config.host,
121
+
negated: label.neg ?? false,
122
+
dateApplied: labledDate,
123
+
}).catch((err) => logger.error({ err }, "Error sending label notification email")),
124
+
);
122
125
}
123
126
124
127
return;
+2
-1
src/handlers/lablerSubscriber.ts
+2
-1
src/handlers/lablerSubscriber.ts
···
14
14
db: LibSQLDatabase<typeof schema>,
15
15
queue: PQueue,
16
16
pdsConfigs: Record<string, PDSConfig>,
17
+
mailQueue: PQueue,
17
18
): (() => void) => {
18
19
let cursor = lastCursor;
19
20
if (cursor) {
···
53
54
case "com.atproto.label.subscribeLabels#labels": {
54
55
for (const label of message.labels) {
55
56
queue.add(async () => {
56
-
await handleNewLabel(config, label, db, pdsConfigs);
57
+
await handleNewLabel(config, label, db, pdsConfigs, mailQueue);
57
58
});
58
59
}
59
60
break;
+3
-2
src/index.ts
+3
-2
src/index.ts
···
12
12
13
13
const labelQueue = new PQueue({ concurrency: 2 });
14
14
const identityQueue = new PQueue({ concurrency: 2 });
15
+
const mailQueue = new PQueue({ concurrency: 1 });
15
16
16
17
// Run Drizzle migrations on startup
17
18
migrate(db, { migrationsFolder: process.env.MIGRATIONS_FOLDER ?? "drizzle" });
···
52
53
(cursor) => cursor.labelerId === config.host,
53
54
);
54
55
let lastCursor = lastCursorRow?.cursor ?? undefined;
55
-
return labelerSubscriber(config, lastCursor, db, labelQueue, settings.pds);
56
+
return labelerSubscriber(config, lastCursor, db, labelQueue, settings.pds, mailQueue);
56
57
})
57
58
.filter((x) => x !== null);
58
59
···
74
75
pdsSubscribers.forEach((close) => close());
75
76
76
77
logger.info("Draining the queues...");
77
-
await Promise.all([labelQueue.onIdle(), identityQueue.onIdle()]);
78
+
await Promise.all([labelQueue.onIdle(), identityQueue.onIdle(), mailQueue.onIdle()]);
78
79
79
80
logger.info("Clean shutdown complete.");
80
81
process.exit(0);
+18
-7
src/mailer.ts
+18
-7
src/mailer.ts
···
1
1
import nodemailer from "nodemailer";
2
2
import { Resend } from "resend";
3
+
import { logger } from "./logger.js";
3
4
4
5
const resendApiKey = process.env.RESEND_API_KEY;
5
6
const smtpUrl = process.env.NOTIFY_SMTP_URL;
···
11
12
if (!senderEmail) throw new Error("NOTIFY_SENDER_EMAIL is not set");
12
13
13
14
const resend = resendApiKey ? new Resend(resendApiKey) : null;
14
-
const transporter = !resendApiKey && smtpUrl ? nodemailer.createTransport(smtpUrl) : null;
15
+
const transporter =
16
+
!resendApiKey && smtpUrl ? nodemailer.createTransport(smtpUrl) : null;
15
17
16
18
export const sendLabelNotification = async (
17
19
emails: string[],
···
46
48
text,
47
49
});
48
50
} else {
49
-
await transporter!.sendMail({
50
-
from: senderEmail,
51
-
to: emails.join(", "),
52
-
subject,
53
-
text,
54
-
});
51
+
if (transporter) {
52
+
await transporter.sendMail({
53
+
from: senderEmail,
54
+
to: emails.join(", "),
55
+
subject,
56
+
text,
57
+
});
58
+
} else {
59
+
logger.error(
60
+
{
61
+
error: "No transporter available",
62
+
},
63
+
"Error sending email",
64
+
);
65
+
}
55
66
}
56
67
};