// SPDX-FileCopyrightText: 2024 John Livingston
//
// SPDX-License-Identifier: AGPL-3.0-only
import type { RegisterServerOptions } from '@peertube/peertube-types'
import type { RoomConf } from 'xmppjs-chat-bot'
import { getProsodyDomain } from '../prosody/config/domain'
import { listProsodyRooms, updateProsodyRoom } from '../prosody/api/manage-rooms'
import { getChannelInfosById } from '../database/channel'
import { ChannelConfigurationOptions } from '../../../shared/lib/types'
import {
getChannelConfigurationOptions,
channelConfigurationOptionsToBotRoomConf
} from '../configuration/channel/storage'
import { BotConfiguration } from '../configuration/bot'
import { fillVideoCustomFields } from '../custom-fields'
import { videoHasWebchat } from '../../../shared/lib/video'
import * as path from 'path'
import * as fs from 'fs'
let singleton: RoomChannel | undefined
/**
* Class used to request and store some informations about relation between rooms and channels.
*/
class RoomChannel {
protected readonly options: RegisterServerOptions
protected readonly mucDomain: string
protected readonly dataFilePath: string
protected readonly logger: {
debug: (s: string) => void
info: (s: string) => void
warn: (s: string) => void
error: (s: string) => void
}
protected room2Channel: Map = new Map()
protected channel2Rooms: Map> = new Map>()
protected needSync: boolean = false
protected roomConfToUpdate: Map = new Map()
protected syncTimeout: ReturnType | undefined
protected isWriting: boolean = false
constructor (params: {
options: RegisterServerOptions
mucDomain: string
dataFilePath: string
}) {
this.options = params.options
this.mucDomain = params.mucDomain
this.dataFilePath = params.dataFilePath
const logger = params.options.peertubeHelpers.logger
this.logger = {
debug: (s) => logger.debug('[RoomChannel] ' + s),
info: (s) => logger.info('[RoomChannel] ' + s),
warn: (s) => logger.warn('[RoomChannel] ' + s),
error: (s) => logger.error('[RoomChannel] ' + s)
}
}
/**
* Instanciate the singleton
*/
public static async initSingleton (options: RegisterServerOptions): Promise {
const prosodyDomain = await getProsodyDomain(options)
const mucDomain = 'room.' + prosodyDomain
const dataFilePath = path.resolve(
options.peertubeHelpers.plugin.getDataDirectoryPath(),
'room-channel',
mucDomain + '.json'
)
singleton = new RoomChannel({
options,
mucDomain,
dataFilePath
})
return singleton
}
/**
* frees the singleton
*/
public static async destroySingleton (): Promise {
if (!singleton) { return }
singleton.cancelScheduledSync()
await singleton.sync()
singleton.cancelScheduledSync() // in case sync rescheduled... we will lose data, but they could be rebuild later
singleton = undefined
}
/**
* Gets the singleton, or raise an exception if it is too soon.
* @returns the singleton
*/
public static singleton (): RoomChannel {
if (!singleton) {
throw new Error('RoomChannel singleton is not initialized yet')
}
return singleton
}
/**
* Reads data from the room-channel data file.
* @return Returns true if the data where found and valid. If there is no data (or no valid data), returns false.
*/
public async readData (): Promise {
// Reading the data file (see https://livingston.frama.io/peertube-plugin-livechat/technical/data/)
let content: string
try {
content = (await fs.promises.readFile(this.dataFilePath)).toString()
} catch (err) {
this.logger.info('Failed reading room-channel data file (' + this.dataFilePath + '), assuming it does not exists')
return false
}
content ??= '{}'
let data: any
try {
data = JSON.parse(content)
} catch (err) {
this.logger.error('Unable to parse the content of the room-channel data file, will start with an empty database.')
return false
}
// This part must be done atomicly:
return this._readData(data)
}
/**
* _readData is the atomic part of readData:
* once the date are read from disk, object data must be emptied and filled atomicly.
*/
protected _readData (data: any): boolean {
this.room2Channel.clear()
this.channel2Rooms.clear()
this.needSync = true
if (typeof data !== 'object') {
this.logger.error('Invalid room-channel data file content')
return false
}
for (const k in data) {
if (!/^\d+$/.test(k)) {
this.logger.error('Invalid channel ID type, should be a number, dropping')
continue
}
const channelId = parseInt(k)
const rooms = data[k]
if (!Array.isArray(rooms)) {
this.logger.error('Invalid room list for Channel ' + channelId.toString() + ', dropping')
continue
}
const c2r = new Map()
this.channel2Rooms.set(channelId, c2r)
for (const roomJID of rooms) {
if (typeof roomJID !== 'string') {
this.logger.error('Invalid room jid for Channel ' + channelId.toString() + ', dropping')
continue
}
c2r.set(roomJID, true)
this.room2Channel.set(roomJID, channelId)
}
}
return true
}
/**
* Rebuilt the data from scratch.
* Can be used for the initial migration.
* Can also be scheduled daily, or on an admin action (not sure it will be done, at the time of the writing).
*/
public async rebuildData (): Promise {
const data: any = {}
const rooms = await listProsodyRooms(this.options)
const settings = await this.options.settingsManager.getSettings([
'chat-per-live-video',
'chat-all-lives',
'chat-all-non-lives',
'chat-videos-list',
'prosody-room-type'
])
for (const room of rooms) {
let channelId: string | number | undefined
const matches = room.localpart.match(/^channel\.(\d+)$/)
if (matches?.[1]) {
if (settings['prosody-room-type'] !== 'channel') {
this.logger.debug(
`Room ${room.localpart} is a channel-wide room, but prosody-room-type!== channel. Ignoring it`
)
continue
}
channelId = parseInt(matches[1])
if (isNaN(channelId)) {
this.logger.error(`Invalid room JID '${room.localpart}'`)
continue
}
// Checking that channel still exists
const channelInfos = await getChannelInfosById(this.options, channelId)
if (!channelInfos) {
this.logger.debug(
`Ignoring room ${room.localpart}, because channel ${channelId} seems to not exist anymore`
)
continue
}
} else {
if (settings['prosody-room-type'] !== 'video') {
this.logger.debug(
`Room ${room.localpart} is a video-related room, but prosody-room-type!== room. Ignoring it`
)
continue
}
const uuid = room.localpart
const video = await this.options.peertubeHelpers.videos.loadByIdOrUUID(uuid)
if (!video) {
this.logger.debug(
`Ignoring room ${room.localpart}, because video ${uuid} seems to not exist anymore`
)
continue
}
await fillVideoCustomFields(this.options, video)
const hasChat = await videoHasWebchat({
'chat-per-live-video': !!settings['chat-per-live-video'],
'chat-all-lives': !!settings['chat-all-lives'],
'chat-all-non-lives': !!settings['chat-all-non-lives'],
'chat-videos-list': settings['chat-videos-list'] as string
}, video)
if (!hasChat) {
// Either there were never any chat, either it was disabled...
this.logger.debug(`Video ${video.uuid} has no chat, ignoring it during the rebuild`)
continue
}
channelId = video.channelId
}
if (!channelId) {
this.logger.error(`Did not find channelId for ${room.localpart}`)
continue
}
channelId = channelId.toString()
if (!(channelId in data)) {
this.logger.debug(`Room ${room.localpart} is associated to channel ${channelId}`)
data[channelId] = []
}
data[channelId].push(room.localpart)
}
// ************ ATOMIC PART ****************
// The rebuild process can remove some rooms (for example if prosody-room-type is changed),
// So we must mark all previous rooms as to refresh:
// Now we must mark all rooms for conf update.
for (const roomJID of this.room2Channel.keys()) {
this.roomConfToUpdate.set(roomJID, true)
}
// This part must be done atomicly:
this._readData(data)
// Now we must mark all rooms for conf update.
for (const roomJID of this.room2Channel.keys()) {
this.roomConfToUpdate.set(roomJID, true)
}
// ************ END OF ATOMIC PART ****************
await this.sync() // FIXME: or maybe scheduleSync ?
}
/**
* Syncs data to disk.
*/
public async sync (): Promise {
if (!this.needSync) { return }
if (this.isWriting) {
this.logger.info('Already writing, scheduling a new sync')
this.scheduleSync()
return
}
this.logger.info('Syncing...')
this.isWriting = true
const prosodyRoomUpdates = new Map[2]>()
try {
const data = this._serializeData() // must be atomic
this.needSync = false // Note: must be done atomicly with the read
await fs.promises.mkdir(path.dirname(this.dataFilePath), { recursive: true })
await fs.promises.writeFile(this.dataFilePath, JSON.stringify(data))
this.logger.debug('room-channel sync done, must sync room conf now')
// Note: getChannelConfigurationOptions has no cache for now, so we will handle it here
const channelConfigurationOptionsCache = new Map()
const roomJIDs = Array.from(this.roomConfToUpdate.keys())
for (const roomJID of roomJIDs) {
const channelId = this.room2Channel.get(roomJID) // roomJID already normalized, so bypassing getRoomChannelId
if (channelId === undefined) {
// No more channel, must disable room!
this.logger.info(`Room ${roomJID} has no associated channel, ensuring there is no active bot conf`)
await BotConfiguration.singleton().disableRoom(roomJID)
this.roomConfToUpdate.delete(roomJID)
continue
}
// Must write the correct Channel conf for the room.
if (!channelConfigurationOptionsCache.has(channelId)) {
try {
channelConfigurationOptionsCache.set(
channelId,
await getChannelConfigurationOptions(this.options, channelId)
)
} catch (err) {
this.logger.error(err as string)
this.logger.error('Failed reading channel configuration, will assume there is none.')
channelConfigurationOptionsCache.set(
channelId,
null
)
}
}
const channelConfigurationOptions = channelConfigurationOptionsCache.get(channelId)
if (!channelConfigurationOptions) {
// no channel configuration, disabling
this.logger.info(`Room ${roomJID} has not associated channel options, ensuring there is no active bot conf`)
await BotConfiguration.singleton().disableRoom(roomJID)
this.roomConfToUpdate.delete(roomJID)
continue
}
this.logger.info(`Room ${roomJID} has associated channel options, writing it`)
const previousRoomConf = await BotConfiguration.singleton().getRoom(roomJID)
const botConf: RoomConf = Object.assign(
{
local: roomJID,
domain: this.mucDomain
},
channelConfigurationOptionsToBotRoomConf(this.options, channelConfigurationOptions, previousRoomConf)
)
await BotConfiguration.singleton().updateRoom(roomJID, botConf)
// Now we also must update some room metadata on Prosody side (livechat_muc_terms, ...)
// This can be done without waiting for the API call to finish, but we don't want to send thousands of
// API calls at the same time. So storing data in a map, and we well launch it sequentially at the end
prosodyRoomUpdates.set(roomJID, {
livechat_muc_terms: channelConfigurationOptions.terms ?? '' // must pass a string, else wont update
})
this.roomConfToUpdate.delete(roomJID)
}
this.logger.info('Syncing done.')
} catch (err) {
this.logger.error(err as string)
this.logger.error('Syncing failed.')
this.needSync = true
} finally {
this.isWriting = false
}
if (prosodyRoomUpdates.size) {
// Here we don't have to wait.
// If it fails (for example because we are turning off prosody), it is not a big deal.
// Does not worth the cost to wait.
// eslint-disable-next-line @typescript-eslint/no-misused-promises
setTimeout(async () => {
this.logger.info('Syncing done, but still some data to send to Prosody')
for (const [roomJID, data] of prosodyRoomUpdates.entries()) {
try {
await updateProsodyRoom(this.options, roomJID, data)
} catch (err) {
this.logger.error(`Failed updating prosody room info: "${err as string}".`)
}
}
}, 0)
}
}
/**
* Schedules a sync.
* Each times data are modified, we can schedule a sync, but we don't have to wait the file writing to be done.
*/
public scheduleSync (): void {
if (!this.needSync) { return }
if (this.syncTimeout) {
// Already scheduled... nothing to do
this.logger.debug('There is already a sync scheduled, skipping.')
return
}
this.logger.info('Scheduling a new sync...')
this.syncTimeout = setTimeout(() => {
this.syncTimeout = undefined
this.logger.info('Running scheduled sync')
this.sync().then(() => {}, (err) => {
this.logger.error(err)
// We will not re-schedule the sync, to avoid flooding error log if there is an issue with the server
})
}, 100)
}
public cancelScheduledSync (): void {
if (this.syncTimeout) {
clearTimeout(this.syncTimeout)
}
}
/**
* Sets a relation between room and channel id
* @param channelId The channel ID
* @param roomJID The room JID. Can be the local part only, or the full JID.
* In the second case, the domain will be checked.
*/
public link (channelId: number | string, roomJIDParam: string): void {
channelId = parseInt(channelId.toString())
if (isNaN(channelId)) {
this.logger.error('Invalid channelId, we wont link')
return
}
const roomJID = this._canonicJID(roomJIDParam)
if (!roomJID) {
this.logger.error('Invalid room JID, we wont link')
return
}
// First, if the room was linked to another channel, we must unlink.
const previousChannelId = this.room2Channel.get(roomJID)
if (previousChannelId) {
if (this.room2Channel.delete(roomJID)) {
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
const previousRooms = this.channel2Rooms.get(previousChannelId)
if (previousRooms) {
if (previousRooms.delete(roomJID)) {
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
}
}
if (this.room2Channel.get(roomJID) !== channelId) {
this.room2Channel.set(roomJID, channelId)
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
let rooms = this.channel2Rooms.get(channelId)
if (!rooms) {
rooms = new Map()
this.channel2Rooms.set(channelId, rooms)
this.needSync = true
}
if (!rooms.has(roomJID)) {
rooms.set(roomJID, true)
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
this.scheduleSync()
}
/**
* Removes all relations for this room
* @param roomJID the room JID
*/
public removeRoom (roomJIDParam: string): void {
const roomJID = this._canonicJID(roomJIDParam)
if (!roomJID) {
this.logger.error('Invalid room JID, we wont link')
return
}
const channelId = this.room2Channel.get(roomJID)
if (channelId) {
const rooms = this.channel2Rooms.get(channelId)
if (rooms) {
if (rooms.delete(roomJID)) {
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
}
}
if (this.room2Channel.delete(roomJID)) {
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
this.scheduleSync()
}
/**
* Removes all relations for this channel
* @param channelId the channel id
*/
public removeChannel (channelId: number | string): void {
channelId = parseInt(channelId.toString())
if (isNaN(channelId)) {
this.logger.error('Invalid channelId, we wont remove')
return
}
const rooms = this.channel2Rooms.get(channelId)
if (rooms) {
for (const roomJID of rooms.keys()) {
// checking the consistency... only removing if the channel is the current one
if (this.room2Channel.get(roomJID) === channelId) {
this.room2Channel.delete(roomJID)
this.needSync = true
this.roomConfToUpdate.set(roomJID, true)
}
}
}
if (this.channel2Rooms.delete(channelId)) {
this.needSync = true
}
this.scheduleSync()
}
/**
* Get the channel ID for a given room.
* Returns null if not found.
* @param roomJIDParam room JID (local part, or full JID)
*/
public getRoomChannelId (roomJIDParam: string): number | null {
const roomJID = this._canonicJID(roomJIDParam)
if (!roomJID) {
this.logger.error('Invalid room JID: ' + roomJIDParam)
return null
}
return this.room2Channel.get(roomJID) ?? null
}
/**
* Returns room local JID parts for all room linked to given channel.
* @param channelId channel id
* @returns list of room JIDs local parts
*/
public getChannelRoomJIDs (channelId: number | string): string[] {
channelId = parseInt(channelId.toString())
if (isNaN(channelId)) {
this.logger.error('Invalid channelId, we wont link')
return []
}
const rooms = this.channel2Rooms.get(channelId)
if (!rooms) {
return []
}
return Array.from(rooms.keys())
}
/**
* Call this method when the channel configuration options changed, to refresh all files.
* @param channelId channel ID
*/
public refreshChannelConfigurationOptions (channelId: number | string): void {
channelId = parseInt(channelId.toString())
if (isNaN(channelId)) {
this.logger.error('Invalid channelId, we wont link')
return
}
const roomJIDs = this.getChannelRoomJIDs(channelId)
this.needSync = true
for (const roomJID of roomJIDs) {
this.roomConfToUpdate.set(roomJID, true)
}
this.scheduleSync()
}
protected _canonicJID (roomJID: string): string | null {
const splits = roomJID.split('@')
if (splits.length < 2) {
return roomJID
}
if (splits.length > 2) {
this.logger.error('The room JID contains multiple @, not valid')
return null
}
if (splits[1] !== this.mucDomain) {
this.logger.error('The room JID is not on the correct domain')
return null
}
return splits[0]
}
protected _serializeData (): any {
const data: any = {}
this.channel2Rooms.forEach((rooms, channelId) => {
const a: string[] = []
rooms.forEach((_val, roomJID) => {
a.push(roomJID)
})
data[channelId.toString()] = a
})
return data
}
}
export {
RoomChannel
}
// TODO: schedule rebuild every X hours/days