matrix-appservice-kakaotalk/node/src/client.js

// matrix-appservice-kakaotalk - A Matrix-KakaoTalk puppeting bridge.
// Copyright (C) 2022 Tulir Asokan, Andrew Ferrazzutti
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// 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 { Long } from "bson"
import { emitLines, promisify } from "./util.js"
import {
	AuthApiClient,
	OAuthApiClient,
	ServiceApiClient,
	TalkClient,
	KnownAuthStatusCode,
	util,
} from "node-kakao"
import chat from "node-kakao/chat"
const { KnownChatType } = chat
/** @typedef {import("node-kakao").OAuthCredential} OAuthCredential */
/** @typedef {import("./clientmanager.js").default} ClientManager} */


class UserClient {
	static #initializing = false

	#talkClient = new TalkClient()
	get talkClient() { return this.#talkClient }

	/** @type {ServiceApiClient} */
	#serviceClient = null
	get serviceClient() { return this.#serviceClient }

	/**
	 * DO NOT CONSTRUCT DIRECTLY. Callers should use {@link UserClient#create} instead.
	 * @param {string} mxid
	 * @param {OAuthCredential} credential
	 */
	constructor(mxid, credential) {
		if (!UserClient.#initializing) {
			throw new Error("Private constructor")
		}
		UserClient.#initializing = false

		this.mxid = mxid
		this.credential = credential
	}

	/**
	 * @param {string} mxid The ID of the associated Matrix user
	 * @param {OAuthCredential} credential The tokens that API calls may use
	 */
	static async create(mxid, credential) {
		this.#initializing = true
		const userClient = new UserClient(mxid, credential)

		userClient.#serviceClient = await ServiceApiClient.create(credential)
		return userClient
	}

	close() {
		this.#talkClient.close()
	}

	/**
	 * TODO Maybe use a "write" method instead
	 * @param {string} command
	 */
	getCmd(command) {
		return `${command}:${this.mxid}`
	}
}

export default class PeerClient {

	/**
	 * @param {ClientManager} manager
	 * @param {import("net").Socket} socket
	 * @param {number} connID
	 * @param {Map<string, UserClient>} userClients
	 */
	constructor(manager, socket, connID) {
		this.manager = manager
		this.socket = socket
		this.connID = connID
		this.stopped = false
		this.notificationID = 0
		this.maxCommandID = 0
		this.peerID = null

		this.userClients = new Map()
	}


	log(...text) {
		if (this.peerID) {
			console.log(`[API/${this.peerID}/${this.connID}]`, ...text)
		} else {
			console.log(`[API/${this.connID}]`, ...text)
		}
	}

	error(...text) {
		if (this.peerID) {
			console.error(`[API/${this.peerID}/${this.connID}]`, ...text)
		} else {
			console.error(`[API/${this.connID}]`, ...text)
		}
	}

	start() {
		this.log("Received connection", this.connID)
		emitLines(this.socket)
		this.socket.on("line", line => this.handleLine(line)
			.catch(err => this.log("Error handling line:", err)))
		this.socket.on("end", this.handleEnd)

		setTimeout(() => {
			if (!this.peerID && !this.stopped) {
				this.log("Didn't receive register request within 3 seconds, terminating")
				this.stop("Register request timeout")
			}
		}, 3000)
	}

	async stop(error = null) {
		if (this.stopped) {
			return
		}
		this.stopped = true
		try {
			await this.#write({ id: --this.notificationID, command: "quit", error })
			await promisify(cb => this.socket.end(cb))
		} catch (err) {
			this.error("Failed to end connection:", err)
			this.socket.destroy(err)
		}
	}

	handleEnd = async () => {
		// TODO Persist clients across bridge disconnections.
		//      But then have to queue received events until bridge acks them!
		this.log("Closing all API clients for", this.peerID)
		for (const userClient of this.userClients.values()) {
			userClient.close()
		}
		this.userClients.clear()

		this.stopped = true
		if (this.peerID && this.manager.clients.get(this.peerID) === this) {
			this.manager.clients.delete(this.peerID)
		}
		this.log(`Connection closed (peer: ${this.peerID})`)
	}

	/**
	 * Write JSON data to the socket.
	 *
	 * @param {object} data - The data to write.
	 * @returns {Promise<void>}
	 */
	#write(data) {
		return promisify(cb => this.socket.write(JSON.stringify(data) + "\n", cb))
	}


	/**
	 * @param {Object} req
	 * @param {string} req.passcode
	 * @param {string} req.uuid
	 * @param {Object} req.form
	 */
	registerDevice = async (req) => {
		const authClient = await this.#createAuthClient(req.uuid)
		return await authClient.registerDevice(req.form, req.passcode, true)
	}

	/**
	 * Log in. If this fails due to not having a device, also request a device passcode.
	 * @param {Object} req
	 * @param {string} req.uuid
	 * @param {Object} req.form
	 * @returns The response of the login attempt, including obtained
	 * credentials for subsequent token-based login. If a required device passcode
	 * request failed, its status is stored here.
	 */
	handleLogin = async (req) => {
		const authClient = await this.#createAuthClient(req.uuid)
		const loginRes = await authClient.login(req.form, true)
		if (loginRes.status === KnownAuthStatusCode.DEVICE_NOT_REGISTERED) {
			const passcodeRes = await authClient.requestPasscode(req.form)
			if (!passcodeRes.success) {
				loginRes.status = passcodeRes.status
			}
		}
		return loginRes
	}

	/**
	 * TODO Consider caching per-user
	 * @param {string} uuid
	 */
	async #createAuthClient(uuid) {
		return await AuthApiClient.create("KakaoTalk Bridge", uuid)
	}

	// TODO Wrapper for per-user commands

	/**
	 * Checked lookup of a UserClient for a given mxid.
	 * @param {string} mxid
	 * @returns {UserClient}
	 */
	#getUser(mxid) {
		const userClient = this.userClients.get(mxid)
		if (userClient === undefined) {
			throw new Error(`Could not find user ${mxid}`)
		}
		return userClient
	}

	/**
	 * Get the service client for the specified user ID, or create
	 * and return a new service client if no user ID is provided.
	 * @param {string} mxid
	 * @param {OAuthCredential} oauth_credential
	 */
	async #getServiceClient(mxid, oauth_credential) {
		return this.userClients.get(mxid)?.serviceClient ||
			await ServiceApiClient.create(oauth_credential)
	}
	
	/**
	 * @param {Object} req
	 * @param {OAuthCredential} req.oauth_credential
	 */
	handleRenew = async (req) => {
		const oAuthClient = await OAuthApiClient.create()
		return await oAuthClient.renew(req.oauth_credential)
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 * @param {OAuthCredential} req.oauth_credential
	 */
	handleStart = async (req) => {
		// TODO Don't re-login if possible. But must still return a LoginResult!
		{
			const oldUserClient = this.userClients.get(req.mxid)
			if (oldUserClient !== undefined) {
				oldUserClient.close()
				this.userClients.delete(req.mxid)
			}
		}

		const userClient = await UserClient.create(req.mxid, req.oauth_credential)
		const res = await userClient.talkClient.login(req.oauth_credential)
		if (!res.success) return res

		this.userClients.set(req.mxid, userClient)
		return res
	}

	startListen = async (req) => {
		const userClient = this.#getUser(req.mxid)

		userClient.talkClient.on("chat", (data, channel) => {
			this.log(`Received message ${data.chat.logId} in channel ${channel.channelId}`)
			return this.#write({
				id: --this.notificationID,
				command: userClient.getCmd("message"),
				//is_sequential: true, // TODO Make sequential per user & channel (if it isn't already)
				chatlog: data.chat,
				channelId: channel.channelId,
				channelType: channel.info.type,
			})
		})

		/* TODO Many more listeners
		userClient.talkClient.on("chat_read", (chat, channel, reader) => {
			this.log(`chat_read in channel ${channel.channelId}`)
			//chat.logId
		})
		*/

		return this.#voidCommandResult
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 * @param {OAuthCredential} req.oauth_credential
	 */
	getOwnProfile = async (req) => {
		const serviceClient = await this.#getServiceClient(req.mxid, req.oauth_credential)
		return await serviceClient.requestMyProfile()
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 * @param {OAuthCredential} req.oauth_credential
	 * @param {Long} req.user_id
	 */
	getProfile = async (req) => {
		const serviceClient = await this.#getServiceClient(req.mxid, req.oauth_credential)
		return await serviceClient.requestProfile(req.user_id)
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 * @param {Long} req.channel_id
	 */
	getPortalChannelInfo = async (req) => {
		const userClient = this.#getUser(req.mxid)
		const talkChannel = userClient.talkClient.channelList.get(req.channel_id)

		const res = await talkChannel.updateAll()
		if (!res.success) return res

		return this.#makeCommandResult({
			name: talkChannel.getDisplayName(),
			participants: Array.from(talkChannel.getAllUserInfo()),
			// TODO Image
		})
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 * @param {Long} req.channel_id
	 * @param {Long?} req.sync_from
	 */
	getChats = async (req) => {
		const userClient = this.#getUser(req.mxid)
		const talkChannel = userClient.talkClient.channelList.get(req.channel_id)

		return await talkChannel.getChatListFrom(req.sync_from)
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 * @param {Long} req.channel_id
	 * @param {string} req.text
	 */
	sendMessage = async (req) => {
		const userClient = this.#getUser(req.mxid)
		const talkChannel = userClient.talkClient.channelList.get(req.channel_id)

		return await talkChannel.sendChat({
			type: KnownChatType.TEXT,
			text: req.text,
		})
	}

	/**
	 * @param {Object} req
	 * @param {string} req.mxid
	 */
	handleStop = async (req) => {
		this.#getUser(req.mxid).close()
		this.userClients.delete(req.mxid)
		return this.#voidCommandResult
	}

	#makeCommandResult(result) {
		return {
			success: true,
			status: 0,
			result: result
		}
	}

	#voidCommandResult = {
		success: true,
		status: 0,
	}

	handleUnknownCommand = () => {
		throw new Error("Unknown command")
	}

	/**
	 * @param {Object} req
	 * @param {string} req.peer_id
	 */
	handleRegister = async (req) => {
		this.peerID = req.peer_id
		this.log(`Registered socket ${this.connID} -> ${this.peerID}`)
		if (this.manager.clients.has(this.peerID)) {
			const oldClient = this.manager.clients.get(this.peerID)
			this.log(`Terminating previous socket ${oldClient.connID} for ${this.peerID}`)
			await oldClient.stop("Socket replaced by new connection")
		}
		this.manager.clients.set(this.peerID, this)
		return { client_exists: this.authClient !== null }
	}

	async handleLine(line) {
		if (this.stopped) {
			this.log("Ignoring line, client is stopped")
			return
		}
		let req
		try {
			req = JSON.parse(line)
		} catch (err) {
			this.log("Non-JSON request:", line)
			return
		}
		if (!req.command || !req.id) {
			this.log("Invalid request:", line)
			return
		}
		if (req.id <= this.maxCommandID) {
			this.log("Ignoring old request", req.id)
			return
		}
		if (req.command != "is_connected") {
			this.log("Received request", req.id, "with command", req.command)
		}
		this.maxCommandID = req.id
		let handler
		if (!this.peerID) {
			if (req.command !== "register") {
				this.log("First request wasn't a register request, terminating")
				await this.stop("Invalid first request")
				return
			} else if (!req.peer_id) {
				this.log("Register request didn't contain ID, terminating")
				await this.stop("Invalid register request")
				return
			}
			handler = this.handleRegister
		} else {
			handler = {
				// TODO Subclass / object for KakaoTalk-specific handlers?
				start: this.handleStart,
				stop: this.handleStop,
				disconnect: () => this.stop(),
				login: this.handleLogin,
				renew: this.handleRenew,
				generate_uuid: util.randomAndroidSubDeviceUUID,
				register_device: this.registerDevice,
				start_listen: this.startListen,
				get_own_profile: this.getOwnProfile,
				get_portal_channel_info: this.getPortalChannelInfo,
				get_chats: this.getChats,
				get_profile: this.getProfile,
				send_message: this.sendMessage,
				//is_connected: async () => ({ is_connected: !await this.puppet.isDisconnected() }),
			}[req.command] || this.handleUnknownCommand
		}
		const resp = { id: req.id }
		delete req.id
		delete req.command
		req = typeify(req)
		resp.command = "response"
		try {
			resp.response = await handler(req)
		} catch (err) {
			if (err.isAxiosError) {
				resp.response = {
					success: false,
					status: err.response.status,
				}
			} else {
				resp.command = "error"
				resp.error = err.toString()
				this.log(`Error handling request ${resp.id} ${err.stack}`)
			}
		}
		await this.#write(resp)
	}
}

/**
 * Recursively scan an object to check if any of its sub-objects
 * should be converted into instances of a specified class.
 * @param obj The object to be scanned & updated.
 * @returns The converted object.
 */
function typeify(obj) {
	if (!(obj instanceof Object)) {
		return obj
	}
	const converterFunc = TYPE_MAP.get(obj.__type__)
	if (converterFunc !== undefined) {
		return converterFunc(obj)
	}
	for (const key in obj) {
		obj[key] = typeify(obj[key])
	}
	return obj
}

// TODO Add more if needed
const TYPE_MAP = new Map([
	["Long", (obj) => new Long(obj.low, obj.high, obj.unsigned)],
])