forked from
pds.dad/label-watcher
WIP: PDS Admin tool to hopefully make it easier to moderate your PDS
Queue and settings change
+286
-37
+1
.env.example
+1
.env.example
+6
drizzle/0001_faithful_shard.sql
+6
drizzle/0001_faithful_shard.sql
+155
drizzle/meta/0001_snapshot.json
+155
drizzle/meta/0001_snapshot.json
···
1
+
{
2
+
"version": "6",
3
+
"dialect": "sqlite",
4
+
"id": "69fbc25d-9d45-43d4-973b-b6bcd6148bb3",
5
+
"prevId": "2f73eace-efb4-467d-8b3e-0fce14415678",
6
+
"tables": {
7
+
"labeler_cursors": {
8
+
"name": "labeler_cursors",
9
+
"columns": {
10
+
"labeler_id": {
11
+
"name": "labeler_id",
12
+
"type": "text",
13
+
"primaryKey": false,
14
+
"notNull": false,
15
+
"autoincrement": false
16
+
},
17
+
"cursor": {
18
+
"name": "cursor",
19
+
"type": "text",
20
+
"primaryKey": false,
21
+
"notNull": true,
22
+
"autoincrement": false
23
+
}
24
+
},
25
+
"indexes": {
26
+
"labeler_cursors_labeler_id_unique": {
27
+
"name": "labeler_cursors_labeler_id_unique",
28
+
"columns": [
29
+
"labeler_id"
30
+
],
31
+
"isUnique": true
32
+
}
33
+
},
34
+
"foreignKeys": {},
35
+
"compositePrimaryKeys": {},
36
+
"uniqueConstraints": {},
37
+
"checkConstraints": {}
38
+
},
39
+
"labels_applied": {
40
+
"name": "labels_applied",
41
+
"columns": {
42
+
"id": {
43
+
"name": "id",
44
+
"type": "integer",
45
+
"primaryKey": true,
46
+
"notNull": true,
47
+
"autoincrement": true
48
+
},
49
+
"did": {
50
+
"name": "did",
51
+
"type": "text",
52
+
"primaryKey": false,
53
+
"notNull": true,
54
+
"autoincrement": false
55
+
},
56
+
"label": {
57
+
"name": "label",
58
+
"type": "text",
59
+
"primaryKey": false,
60
+
"notNull": true,
61
+
"autoincrement": false
62
+
},
63
+
"action": {
64
+
"name": "action",
65
+
"type": "text",
66
+
"primaryKey": false,
67
+
"notNull": true,
68
+
"autoincrement": false
69
+
},
70
+
"date_applied": {
71
+
"name": "date_applied",
72
+
"type": "integer",
73
+
"primaryKey": false,
74
+
"notNull": true,
75
+
"autoincrement": false
76
+
}
77
+
},
78
+
"indexes": {},
79
+
"foreignKeys": {
80
+
"labels_applied_did_watched_repos_did_fk": {
81
+
"name": "labels_applied_did_watched_repos_did_fk",
82
+
"tableFrom": "labels_applied",
83
+
"tableTo": "watched_repos",
84
+
"columnsFrom": [
85
+
"did"
86
+
],
87
+
"columnsTo": [
88
+
"did"
89
+
],
90
+
"onDelete": "no action",
91
+
"onUpdate": "no action"
92
+
}
93
+
},
94
+
"compositePrimaryKeys": {},
95
+
"uniqueConstraints": {},
96
+
"checkConstraints": {}
97
+
},
98
+
"watched_repos": {
99
+
"name": "watched_repos",
100
+
"columns": {
101
+
"did": {
102
+
"name": "did",
103
+
"type": "text",
104
+
"primaryKey": true,
105
+
"notNull": true,
106
+
"autoincrement": false
107
+
},
108
+
"pds_host": {
109
+
"name": "pds_host",
110
+
"type": "text",
111
+
"primaryKey": false,
112
+
"notNull": true,
113
+
"autoincrement": false
114
+
},
115
+
"active": {
116
+
"name": "active",
117
+
"type": "integer",
118
+
"primaryKey": false,
119
+
"notNull": true,
120
+
"autoincrement": false
121
+
},
122
+
"date_first_seen": {
123
+
"name": "date_first_seen",
124
+
"type": "integer",
125
+
"primaryKey": false,
126
+
"notNull": true,
127
+
"autoincrement": false
128
+
}
129
+
},
130
+
"indexes": {
131
+
"watched_repos_did_unique": {
132
+
"name": "watched_repos_did_unique",
133
+
"columns": [
134
+
"did"
135
+
],
136
+
"isUnique": true
137
+
}
138
+
},
139
+
"foreignKeys": {},
140
+
"compositePrimaryKeys": {},
141
+
"uniqueConstraints": {},
142
+
"checkConstraints": {}
143
+
}
144
+
},
145
+
"views": {},
146
+
"enums": {},
147
+
"_meta": {
148
+
"schemas": {},
149
+
"tables": {},
150
+
"columns": {}
151
+
},
152
+
"internal": {
153
+
"indexes": {}
154
+
}
155
+
}
+7
drizzle/meta/_journal.json
+7
drizzle/meta/_journal.json
+1
package.json
+1
package.json
+23
pnpm-lock.yaml
+23
pnpm-lock.yaml
···
20
20
drizzle-orm:
21
21
specifier: ^0.45.1
22
22
version: 0.45.1(@libsql/client@0.17.0)(kysely@0.22.0)(pg@8.18.0)
23
+
p-queue:
24
+
specifier: ^9.1.0
25
+
version: 9.1.0
23
26
smol-toml:
24
27
specifier: ^1.6.0
25
28
version: 1.6.0
···
574
577
event-target-polyfill@0.0.4:
575
578
resolution: {integrity: sha512-Gs6RLjzlLRdT8X9ZipJdIZI/Y6/HhRLyq9RdDlCsnpxr/+Nn6bU2EFGuC94GjxqhM+Nmij2Vcq98yoHrU8uNFQ==}
576
579
580
+
eventemitter3@5.0.4:
581
+
resolution: {integrity: sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw==}
582
+
577
583
fetch-blob@3.2.0:
578
584
resolution: {integrity: sha512-7yAQpD2UMJzLi1Dqv7qFYnPbaPx7ZfFK6PiIxQ4PfkGPyNyl2Ugx+a/umUonmKqjhM4DnfbMvdX6otXq83soQQ==}
579
585
engines: {node: ^12.20 || >= 14.13}
···
618
624
resolution: {integrity: sha512-dRB78srN/l6gqWulah9SrxeYnxeddIG30+GOqK/9OlLVyLg3HPnr6SqOWTWOXKRwC2eGYCkZ59NNuSgvSrpgOA==}
619
625
engines: {node: ^12.20.0 || ^14.13.1 || >=16.0.0}
620
626
627
+
p-queue@9.1.0:
628
+
resolution: {integrity: sha512-O/ZPaXuQV29uSLbxWBGGZO1mCQXV2BLIwUr59JUU9SoH76mnYvtms7aafH/isNSNGwuEfP6W/4xD0/TJXxrizw==}
629
+
engines: {node: '>=20'}
630
+
631
+
p-timeout@7.0.1:
632
+
resolution: {integrity: sha512-AxTM2wDGORHGEkPCt8yqxOTMgpfbEHqF51f/5fJCmwFC3C/zNcGT63SymH2ttOAaiIws2zVg4+izQCjrakcwHg==}
633
+
engines: {node: '>=20'}
634
+
621
635
partysocket@1.1.13:
622
636
resolution: {integrity: sha512-RNXGzc6j0NISGE84+VTHHtbPwmnzZuOYJm9XZ+en+aZlIA2vC4AfwPlYxAHmGGGko3pQF7xRNhoe7bu1Brej4Q==}
623
637
···
1116
1130
1117
1131
event-target-polyfill@0.0.4: {}
1118
1132
1133
+
eventemitter3@5.0.4: {}
1134
+
1119
1135
fetch-blob@3.2.0:
1120
1136
dependencies:
1121
1137
node-domexception: 1.0.0
···
1162
1178
data-uri-to-buffer: 4.0.1
1163
1179
fetch-blob: 3.2.0
1164
1180
formdata-polyfill: 4.0.10
1181
+
1182
+
p-queue@9.1.0:
1183
+
dependencies:
1184
+
eventemitter3: 5.0.4
1185
+
p-timeout: 7.0.1
1186
+
1187
+
p-timeout@7.0.1: {}
1165
1188
1166
1189
partysocket@1.1.13:
1167
1190
dependencies:
+5
src/db/schema.ts
+5
src/db/schema.ts
···
18
18
dateApplied: integer("date_applied", { mode: "timestamp" }).notNull(),
19
19
});
20
20
21
+
export const labelerCursor = sqliteTable("labeler_cursors", {
22
+
labelerId: text("labeler_id").unique(),
23
+
cursor: text("cursor").notNull(),
24
+
});
25
+
21
26
export const watchedReposRelations = relations(watchedRepos, ({ many }) => ({
22
27
labelsApplied: many(labelsApplied),
23
28
}));
+21
src/handlers/handleNewLabel.ts
+21
src/handlers/handleNewLabel.ts
···
1
+
import type { Label } from "@atcute/atproto/types/label/defs";
2
+
import type { LabelerConfig } from "../types/settings.js";
3
+
4
+
export const handleNewLabel = async (config: LabelerConfig, label: Label) => {
5
+
// TODO: MAKE SURE TO CHECK NEG
6
+
console.log(`From: ${config.host}`);
7
+
8
+
await new Promise((r) => setTimeout(r, 2000));
9
+
10
+
if (config.labels[label.val]) {
11
+
console.log(
12
+
`Listed label found. Performing the action: ${config.labels[label.val]?.action}`,
13
+
);
14
+
console.log("\n");
15
+
}
16
+
console.log("Label from: ", label.src);
17
+
console.log("Label: ", label.val);
18
+
console.log("Label for: ", label.uri);
19
+
console.log("neg:", label.neg);
20
+
console.log("\n");
21
+
};
+33
src/handlers/lablerSubscriber.ts
+33
src/handlers/lablerSubscriber.ts
···
1
+
import { FirehoseSubscription } from "@atcute/firehose";
2
+
import type { LabelerConfig } from "../types/settings.js";
3
+
import { ComAtprotoLabelSubscribeLabels } from "@atcute/atproto";
4
+
import type PQueue from "p-queue";
5
+
import { handleNewLabel } from "./handleNewLabel.js";
6
+
7
+
export const labelerSubscriber = async (
8
+
config: LabelerConfig,
9
+
queue: PQueue,
10
+
) => {
11
+
const subscription = new FirehoseSubscription({
12
+
service: `wss://${config.host}`,
13
+
nsid: ComAtprotoLabelSubscribeLabels.mainSchema,
14
+
});
15
+
16
+
console.log(`Listening to ${config.host}`);
17
+
for await (const message of subscription) {
18
+
switch (message.$type) {
19
+
case "com.atproto.label.subscribeLabels#info": {
20
+
console.log("commit:", message);
21
+
break;
22
+
}
23
+
case "com.atproto.label.subscribeLabels#labels": {
24
+
for (const label of message.labels) {
25
+
queue.add(async () => {
26
+
await handleNewLabel(config, label);
27
+
});
28
+
}
29
+
break;
30
+
}
31
+
}
32
+
}
33
+
};
+27
-37
src/index.ts
+27
-37
src/index.ts
···
1
-
import { FirehoseSubscription } from "@atcute/firehose";
2
-
import { ComAtprotoLabelSubscribeLabels } from "@atcute/atproto";
3
1
import { db } from "./db/index.js";
4
2
import { migrate } from "drizzle-orm/libsql/migrator";
5
3
import { readFileSync } from "node:fs";
6
4
import { parse } from "smol-toml";
7
-
import type { LabelerConfig, Settings } from "./types/settings.js";
5
+
import PQueue from "p-queue";
6
+
import { labelerSubscriber } from "./handlers/lablerSubscriber.js";
7
+
import type { Settings } from "./types/settings.js";
8
+
9
+
const queue = new PQueue({ concurrency: 2 });
8
10
9
11
// TODO
10
12
// 1. Figure out a schema for settings we want. PDSs to watch.Labelers and their Labels
···
18
20
19
21
const settingsFile = readFileSync("./settings.toml", "utf-8");
20
22
21
-
//TODO I really really don't like this unknown to settings. Figure that out later
23
+
//TODO I really really don't like this unknown to settings. Figure that out later. Cause. It does work >.>
22
24
const settings = parse(settingsFile) as unknown as Settings;
23
25
24
26
const labelers = settings.labeler;
25
27
26
-
const labelerSubscriber = async (config: LabelerConfig) => {
27
-
const subscription = new FirehoseSubscription({
28
-
service: `wss://${config.host}`,
29
-
nsid: ComAtprotoLabelSubscribeLabels.mainSchema,
30
-
});
28
+
// --- Graceful shutdown ---
29
+
async function shutdown(signal: string) {
30
+
console.log(`\nReceived ${signal}, shutting down...`);
31
31
32
-
console.log(`Listening to ${config.host}`);
33
-
for await (const message of subscription) {
34
-
switch (message.$type) {
35
-
case "com.atproto.label.subscribeLabels#info": {
36
-
console.log("commit:", message);
37
-
break;
38
-
}
39
-
case "com.atproto.label.subscribeLabels#labels": {
40
-
// repository commit (record creates, updates, deletes)
41
-
for (const label of message.labels) {
42
-
console.log(`From: ${config.host}`);
32
+
// TODO maybe should make sure the websockets close here?
43
33
44
-
if (config.labels[label.val]) {
45
-
console.log(
46
-
`Listed label found. Performing the action: ${config.labels[label.val]?.action}`,
47
-
);
48
-
console.log("\n");
49
-
}
50
-
console.log("Label from: ", label.src);
51
-
console.log("Label: ", label.val);
52
-
console.log("Label for: ", label.uri);
53
-
console.log("\n");
54
-
}
55
-
break;
56
-
}
57
-
}
58
-
}
59
-
};
34
+
// Drain all queues in parallel
35
+
console.log("Draining the queue...");
36
+
await queue.onIdle();
37
+
38
+
console.log("Clean shutdown complete.");
39
+
process.exit(0);
40
+
}
41
+
42
+
process.on("SIGTERM", () => shutdown("SIGTERM"));
43
+
process.on("SIGINT", () => shutdown("SIGINT"));
44
+
45
+
process.on("unhandledRejection", (reason) => {
46
+
console.error("Unhandled rejection:", reason);
47
+
});
60
48
61
49
Promise.all(
62
-
Object.entries(labelers).map(([_, config]) => labelerSubscriber(config)),
50
+
Object.entries(labelers).map(([_, config]) =>
51
+
labelerSubscriber(config, queue),
52
+
),
63
53
);
+7
src/types/settings.ts
+7
src/types/settings.ts
···
1
1
// matches the settings.toml
2
2
3
+
export interface PDSConfig {
4
+
host: string;
5
+
emails: string[];
6
+
pdsAdminPassword: string;
7
+
}
8
+
3
9
export type LabelAction = "notify" | "takedown";
4
10
5
11
export interface LabelConfig {
···
19
25
};
20
26
};
21
27
labeler: Record<string, LabelerConfig>;
28
+
pds: Record<LabelAction, PDSConfig>;
22
29
}