Add config and improve stopping

.gitignore

@@ -10,7 +10,6 @@ __pycache__
 /*.egg-info
 /.eggs
 
-node_modules
 profiles
 
 /config.yaml

puppet/.gitignore

@@ -0,0 +1,2 @@
+/node_modules
+/config.json

puppet/README.md

@@ -0,0 +1,7 @@
+### Listen config
+If `type` is `unix`, `path` is the path where to create the socket.
+
+If `type` is `tcp`, `port` and `host` are the host/port where to listen.
+
+### Profile directory
+The `profile_dir` specifies which directory to put chromium user data directories.

puppet/example-config.json

@@ -0,0 +1,7 @@
+{
+	"listen": {
+		"type": "unix",
+		"path": "/var/run/mautrix-amp/puppet.sock"
+	},
+	"profile_dir": "./profiles"
+}

puppet/src/api.js

@@ -21,10 +21,10 @@ import Client from "./client.js"
 import { promisify } from "./util.js"
 
 export default class PuppetAPI {
-	path = "/var/run/mautrix-amp/puppet.sock"
-
-	constructor() {
+	constructor(listenConfig) {
+		this.listenConfig = listenConfig
 		this.server = net.createServer(this.acceptConnection)
+		this.connections = []
 		this.puppets = new Map()
 		this.clients = new Map()
 		this.connIDSequence = 0
@@ -38,25 +38,41 @@ export default class PuppetAPI {
 	acceptConnection = sock => {
 		if (this.stopped) {
 			sock.end()
+			sock.destroy()
 		} else {
-			new Client(this, sock, ++this.connIDSequence).start()
+			const connID = this.connIDSequence++
+			this.connections[connID] = sock
+			new Client(this, sock, connID).start()
 		}
 	}
 
+	async startUnix(socketPath) {
+		try {
+			await fs.promises.access(path.dirname(socketPath))
+		} catch (err) {
+			await fs.promises.mkdir(path.dirname(socketPath), 0o700)
+		}
+		try {
+			await fs.promises.unlink(socketPath)
+		} catch (err) {}
+		await promisify(cb => this.server.listen(socketPath, cb))
+		await fs.promises.chmod(socketPath, 0o700)
+		this.log("Now listening at", socketPath)
+	}
+
+	async startTCP(port, host) {
+		await promisify(cb => this.server.listen(port, host, cb))
+		this.log(`Now listening at ${host || ""}:${port}`)
+	}
+
 	async start() {
 		this.log("Starting server")
 
-		try {
-			await fs.promises.access(path.dirname(this.path))
-		} catch (err) {
-			await fs.promises.mkdir(path.dirname(this.path), 0o700)
+		if (this.listenConfig.type === "unix") {
+			await this.startUnix(this.listenConfig.path)
+		} else if (this.listenConfig.type === "tcp") {
+			await this.startTCP(this.listenConfig.port, this.listenConfig.host)
 		}
-		try {
-			await fs.promises.unlink(this.path)
-		} catch (err) {}
-		await promisify(cb => this.server.listen(this.path, cb))
-		await fs.promises.chmod(this.path, 0o700)
-		this.log("Now listening at", this.path)
 	}
 
 	async stop() {
@@ -64,12 +80,18 @@ export default class PuppetAPI {
 		for (const client of this.clients.values()) {
 			await client.stop("Server is shutting down")
 		}
+		for (const socket of this.connections) {
+			socket.end()
+			socket.destroy()
+		}
 		this.log("Stopping server")
 		await promisify(cb => this.server.close(cb))
-		try {
-			await fs.promises.unlink(this.path)
-		} catch (err) {}
-		this.log("Server stopped")
+		if (this.listenConfig.type === "unix") {
+			try {
+				await fs.promises.unlink(this.listenConfig.path)
+			} catch (err) {}
+		}
+		this.log("Stopping puppets")
 		for (const puppet of this.puppets.values()) {
 			await puppet.stop()
 		}

puppet/src/client.js

@@ -119,7 +119,7 @@ export default class Client {
 			return { started: false, is_logged_in: await this.puppet.isLoggedIn() }
 		}
 		this.log("Opening new puppeteer for", this.userID)
-		this.puppet = new MessagesPuppeteer(this.userID, `./profiles/${this.userID}`, this)
+		this.puppet = new MessagesPuppeteer(this.userID, this)
 		this.manager.puppets.set(this.userID, this.puppet)
 		await this.puppet.start(!!req.debug)
 		return { started: true, is_logged_in: await this.puppet.isLoggedIn() }
@@ -196,6 +196,7 @@ export default class Client {
 			handler = {
 				start: this.handleStart,
 				stop: this.handleStop,
+				disconnect: () => this.stop(),
 				login: () => this.puppet.waitForLogin(),
 				send: req => this.puppet.sendMessage(req.chat_id, req.text),
 				get_chats: () => this.puppet.getRecentChats(),

puppet/src/main.js

@@ -14,10 +14,20 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <https://www.gnu.org/licenses/>.
 import process from "process"
+import fs from "fs"
 
 import PuppetAPI from "./api.js"
+import MessagesPuppeteer from "./puppet.js"
 
-const api = new PuppetAPI()
+let path = process.argv[process.argv.length - 1]
+if (!path.endsWith(".json")) {
+	path = "config.json"
+}
+console.log("Reading config from", path)
+const config = JSON.parse(fs.readFileSync(path).toString())
+MessagesPuppeteer.profileDir = config.profile_dir
+
+const api = new PuppetAPI(config.listen)
 
 function stop() {
 	api.stop().then(() => process.exit(0), err => {

puppet/src/puppet.js

@@ -23,15 +23,16 @@ import TaskQueue from "./taskqueue.js"
 import { sleep } from "./util.js"
 
 export default class MessagesPuppeteer {
+	static profileDir = "./profiles"
 	url = "https://messages.google.com/web/"
 
 	/**
 	 *
 	 * @param {string} id
-	 * @param {string} profilePath
 	 * @param {?Client} [client]
 	 */
-	constructor(id, profilePath, client = null) {
+	constructor(id, client = null) {
+		let profilePath = path.join(MessagesPuppeteer.profileDir, id)
 		if (!profilePath.startsWith("/")) {
 			profilePath = path.join(process.cwd(), profilePath)
 		}