ClientX/_opt/pbUtils.js

// _opt/pbutils.js
// Polyfill global EventSource for PocketBase realtime in Node.js (using CommonJS require)
import { createRequire } from 'module';
const require = createRequire(import.meta.url);
const { EventSource } = require('eventsource');
if (typeof global.EventSource === 'undefined') {
  global.EventSource = EventSource;
}

/**
 * PocketBase utilities module - extends PocketBase client with useful shortcuts
 */

/**
 * Initializes the PocketBase utilities module
 * @param {Object} client - Discord client with attached PocketBase instance
 * @param {Object} config - Client configuration
 */
export const init = async (client, config) => {
    const { pb, logger } = client;

    logger.info('[module:pbUtils] Initializing PocketBase utilities module');

    // Attach utility methods to the pb object
    extendPocketBase(client, pb, logger);

   // Add connection state handling
   setupConnectionHandling(pb, logger);
  
  // Subscribe to real-time message queue events and re-emit via client
  try {
    pb.collection('message_queue').subscribe('*', (e) => {
      client.emit('message_queue_event', e.action, e.record);
      logger.debug(`PubSub event: ${e.action} on message_queue: ${JSON.stringify(e.record)}`);
    });
    logger.info('[module:pbUtils] Subscribed to PocketBase message_queue realtime events');
  } catch (error) {
    logger.error(`[module:pbUtils] Failed to subscribe to message_queue realtime: ${error.message}`);
  }

// end of init()

  logger.info('PocketBase utilities module initialized');
};

/**
 * Register a handler for incoming message_queue pub/sub events.
 * Other modules can import and use this to react to external messages.
 * @param {import('discord.js').Client} client - The Discord client
 * @param {(action: string, record: object) => void} handler - Callback for each event
 */
export function onMessageQueueEvent(client, handler) {
  client.on('message_queue_event', (action, record) => {
    try {
      handler(action, record);
    } catch (err) {
      client.logger.error(`[module:pbUtils] Error in message_queue handler: ${err.message}`);
    }
  });
}

/**
 * Extends the PocketBase instance with utility methods
 * @param {Object} pb - PocketBase instance
 * @param {Object} logger - Winston logger
 */
/**
 * Adds utility methods to the PocketBase client.
 * @param {import('discord.js').Client} client - The Discord client
 * @param {object} pb - The PocketBase instance
 * @param {object} logger - Logger instance
 */
const extendPocketBase = (client, pb, logger) => {
	// ===== COLLECTION OPERATIONS =====

	/**
	 * Get a single record with better error handling
	 * @param {string} collection - Collection name
	 * @param {string} id - Record ID
	 * @param {Object} options - Additional options
	 * @returns {Promise<Object>} The record or null
	 */
	pb.getOne = async (collection, id, options = {}) => {
		try {
			return await pb.collection(collection).getOne(id, options);
		} catch (error) {
			if (error.status === 404) {
				return null;
			}
			logger.error(`Failed to get record ${id} from ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Creates a record with validation and error handling
	 * @param {string} collection - Collection name
	 * @param {Object} data - Record data
	 * @returns {Promise<Object>} Created record
	 */
	pb.createOne = async (collection, data) => {
		try {
			return await pb.collection(collection).create(data);
		} catch (error) {
			logger.error(`Failed to create record in ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Updates a record with better error handling
	 * @param {string} collection - Collection name
	 * @param {string} id - Record ID
	 * @param {Object} data - Record data
	 * @returns {Promise<Object>} Updated record
	 */
	pb.updateOne = async (collection, id, data) => {
		try {
			return await pb.collection(collection).update(id, data);
		} catch (error) {
			logger.error(`Failed to update record ${id} in ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Deletes a record with better error handling
	 * @param {string} collection - Collection name
	 * @param {string} id - Record ID
	 * @returns {Promise<boolean>} Success status
	 */
    pb.deleteOne = async (collection, id) => {
		try {
			await pb.collection(collection).delete(id);
			return true;
		} catch (error) {
			if (error.status === 404) {
				logger.warn(`Record ${id} not found in ${collection} for deletion`);
				return false;
			}
			logger.error(`Failed to delete record ${id} from ${collection}: ${error.message}`);
			throw error;
		}
	};
    
    /**
     * Convenience: publish a message into the "message_queue" collection,
     * with source/destination validation.
     * @param {string} source - origin (client id or 'external')
     * @param {string} destination - target client id
     * @param {string} dataType - message type
     * @param {object} data - JSON-serializable payload
     * @returns {Promise<Object>} created record
     */
    pb.publishMessage = async (source, destination, dataType, data) => {
        // Valid sources: all configured clients + 'external'
        const validSources = client.config.clients.map(c => c.id).concat('external');
        if (!validSources.includes(source)) throw new Error(`Invalid message source: ${source}`);
        // Valid destinations: all configured clients
        const validDest = client.config.clients.map(c => c.id);
        if (!validDest.includes(destination)) throw new Error(`Invalid message destination: ${destination}`);
        return await pb.collection('message_queue').create({ source, destination, dataType, data: JSON.stringify(data) });
    };

	/**
	 * Upsert - creates or updates a record based on whether it exists
	 * @param {string} collection - Collection name
	 * @param {string} id - Record ID or null for new record
	 * @param {Object} data - Record data
	 * @returns {Promise<Object>} Created/updated record
	 */
	pb.upsert = async (collection, id, data) => {
		if (id) {
			const exists = await pb.getOne(collection, id);
			if (exists) {
				return await pb.updateOne(collection, id, data);
			}
		}
		return await pb.createOne(collection, data);
	};

	// ===== QUERY SHORTCUTS =====

	/**
	 * Get first record matching a filter
	 * @param {string} collection - Collection name
	 * @param {string} filter - Filter query
	 * @param {Object} options - Additional options
	 * @returns {Promise<Object>} First matching record or null
	 */
	pb.getFirst = async (collection, filter, options = {}) => {
		try {
			const result = await pb.collection(collection).getList(1, 1, {
				filter,
				...options
			});

			return result.items.length > 0 ? result.items[0] : null;
		} catch (error) {
			if (error.status === 404) {
				return null;
			}
			logger.error(`Failed to get first record from ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Get all records from a collection (handles pagination)
	 * @param {string} collection - Collection name
	 * @param {Object} options - Query options
	 * @returns {Promise<Array>} Array of records
	 */
	pb.getAll = async (collection, options = {}) => {
		const records = [];
		const pageSize = options.pageSize || 200;
		let page = 1;

		try {
			while (true) {
				const result = await pb.collection(collection).getList(page, pageSize, options);
				records.push(...result.items);

				if (records.length >= result.totalItems) {
					break;
				}

				page++;
			}

			return records;
		} catch (error) {
			logger.error(`Failed to get all records from ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Count records matching a filter
	 * @param {string} collection - Collection name
	 * @param {string} filter - Filter query
	 * @returns {Promise<number>} Count of matching records
	 */
	pb.count = async (collection, filter = '') => {
		try {
			const result = await pb.collection(collection).getList(1, 1, {
				filter,
				fields: 'id'
			});

			return result.totalItems;
		} catch (error) {
			logger.error(`Failed to count records in ${collection}: ${error.message}`);
			throw error;
		}
	};

    // ===== BATCH OPERATIONS =====

	/**
	 * Perform batch create
	 * @param {string} collection - Collection name
	 * @param {Array<Object>} items - Array of items to create
	 * @returns {Promise<Array>} Created records
	 */
	pb.batchCreate = async (collection, items) => {
		if (!items || items.length === 0) {
			return [];
		}

		const results = [];

		try {
			// Process in chunks to avoid rate limits
			const chunkSize = 50;

			for (let i = 0; i < items.length; i += chunkSize) {
				const chunk = items.slice(i, i + chunkSize);
				const promises = chunk.map(item => pb.createOne(collection, item));
				const chunkResults = await Promise.all(promises);
				results.push(...chunkResults);
			}

			return results;
		} catch (error) {
			logger.error(`Failed batch create in ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Perform batch update
	 * @param {string} collection - Collection name
	 * @param {Array<Object>} items - Array of items with id field
	 * @returns {Promise<Array>} Updated records
	 */
	pb.batchUpdate = async (collection, items) => {
		if (!items || items.length === 0) {
			return [];
		}

		const results = [];

		try {
			// Process in chunks to avoid rate limits
			const chunkSize = 50;

			for (let i = 0; i < items.length; i += chunkSize) {
				const chunk = items.slice(i, i + chunkSize);
				const promises = chunk.map(item => {
					const { id, ...data } = item;
					return pb.updateOne(collection, id, data);
				});
				const chunkResults = await Promise.all(promises);
				results.push(...chunkResults);
			}

			return results;
		} catch (error) {
			logger.error(`Failed batch update in ${collection}: ${error.message}`);
			throw error;
		}
	};

	/**
	 * Perform batch delete
	 * @param {string} collection - Collection name
	 * @param {Array<string>} ids - Array of record IDs to delete
	 * @returns {Promise<Array>} Results of deletion operations
	 */
	pb.batchDelete = async (collection, ids) => {
		if (!ids || ids.length === 0) {
			return [];
		}

		const results = [];

		try {
			// Process in chunks to avoid rate limits
			const chunkSize = 50;

			for (let i = 0; i < ids.length; i += chunkSize) {
				const chunk = ids.slice(i, i + chunkSize);
				const promises = chunk.map(id => pb.deleteOne(collection, id));
				const chunkResults = await Promise.all(promises);
				results.push(...chunkResults);
			}

			return results;
		} catch (error) {
			logger.error(`Failed batch delete in ${collection}: ${error.message}`);
			throw error;
		}
	};
    /**
     * Delete a message in the "message_queue" collection by its record ID.
     * @param {string} id - Record ID to delete.
     * @returns {Promise<boolean>} True if deleted or not found, false on error.
     */
    pb.deleteMessageQueue = async (id) => {
        return await pb.deleteOne('message_queue', id);
    };

	// ===== PUB/SUB OPERATIONS =====

	/**
	 * Publish a message into the "message_queue" collection.
	 * @param {string} source - Origin identifier for the message.
	 * @param {string} destination - Target identifier (e.g. channel or client ID).
	 * @param {string} dataType - A short string describing the type of data.
	 * @param {object} data - The payload object to deliver.
	 * @returns {Promise<object>} The created message_queue record.
	 */
	pb.publishMessage = async (source, destination, dataType, data) => {
		try {
			return await pb.collection('message_queue').create({
				source,
				destination,
				dataType,
				data: JSON.stringify(data)
			});
		} catch (error) {
			logger.error(`Failed to publish message to message_queue: ${error.message}`);
			throw error;
		}
	};

	// ===== CACHE MANAGEMENT =====

	// Simple in-memory cache
	pb.cache = {
		_store: new Map(),
		_ttls: new Map(),

		/**
		 * Get a value from cache
		 * @param {string} key - Cache key
		 * @returns {*} Cached value or undefined
		 */
		get(key) {
			if (this._ttls.has(key) && this._ttls.get(key) < Date.now()) {
				this.delete(key);
				return undefined;
			}
			return this._store.get(key);
		},

		/**
		 * Set a value in cache
		 * @param {string} key - Cache key
		 * @param {*} value - Value to store
		 * @param {number} ttlSeconds - Time to live in seconds
		 */
		set(key, value, ttlSeconds = 300) {
			this._store.set(key, value);
			if (ttlSeconds > 0) {
				this._ttls.set(key, Date.now() + (ttlSeconds * 1000));
			}
		},

		/**
		 * Delete a value from cache
		 * @param {string} key - Cache key
		 */
		delete(key) {
			this._store.delete(key);
			this._ttls.delete(key);
		},

		/**
		 * Clear all cache
		 */
		clear() {
			this._store.clear();
			this._ttls.clear();
		}
	};

	/**
	 * Get a record with caching
	 * @param {string} collection - Collection name
	 * @param {string} id - Record ID
	 * @param {number} ttlSeconds - Cache TTL in seconds
	 * @returns {Promise<Object>} Record or null
	 */
	pb.getCached = async (collection, id, ttlSeconds = 60) => {
		const cacheKey = `${collection}:${id}`;
		const cached = pb.cache.get(cacheKey);

		if (cached !== undefined) {
			return cached;
		}

		const record = await pb.getOne(collection, id);
		pb.cache.set(cacheKey, record, ttlSeconds);

		return record;
	};

	/**
	 * Get list with caching
	 * @param {string} collection - Collection name
	 * @param {Object} options - Query options
	 * @param {number} ttlSeconds - Cache TTL in seconds
	 * @returns {Promise<Object>} List result
	 */
	pb.getListCached = async (collection, options = {}, ttlSeconds = 30) => {
		const cacheKey = `${collection}:list:${JSON.stringify(options)}`;
		const cached = pb.cache.get(cacheKey);

		if (cached !== undefined) {
			return cached;
		}

		const { page = 1, perPage = 50, ...restOptions } = options;
		const result = await pb.collection(collection).getList(page, perPage, restOptions);
		pb.cache.set(cacheKey, result, ttlSeconds);

		return result;
	};
};

/**
 * Setup connection state handling
 * @param {Object} pb - PocketBase instance
 * @param {Object} logger - Winston logger
 */
const setupConnectionHandling = (pb, logger) => {
	// Add connection state tracking
	pb.isConnected = true;
	pb.lastSuccessfulAuth = null;

	// Add auto-reconnect and token refresh
	pb.authStore.onChange(() => {
		pb.isConnected = pb.authStore.isValid;

		if (pb.isConnected) {
			pb.lastSuccessfulAuth = new Date();
			logger.info('PocketBase authentication successful');
		} else {
			logger.warn('PocketBase auth token expired or invalid');
		}
	});

	// Helper to check health and reconnect if needed
	pb.ensureConnection = async () => {
		if (!pb.isConnected || !pb.authStore.isValid) {
			try {
				logger.info('Reconnecting to PocketBase...');
				// Attempt to refresh the auth if we have a refresh token
				if (pb.authStore.token && pb.authStore.model?.id) {
					await pb.admins.authRefresh();
				} else if (pb._config.username && pb._config.password) {
					// Fall back to full re-authentication if credentials available
					await pb.admins.authWithPassword(
						pb._config.username,
						pb._config.password
					);
				} else {
					logger.error('No credentials available to reconnect PocketBase');
					pb.isConnected = false;
					return false;
				}

				pb.isConnected = true;
				pb.lastSuccessfulAuth = new Date();
				logger.info('Successfully reconnected to PocketBase');

				return true;
			} catch (error) {
				logger.error(`Failed to reconnect to PocketBase: ${error.message}`);
				pb.isConnected = false;
				return false;
			}
		}

		return true;
	};

	// Store credentials for reconnection
	pb._config = pb._config || {};
	// Ensure only if env provided
	if (process.env.SHARED_POCKETBASE_USERNAME && process.env.SHARED_POCKETBASE_PASSWORD) {
		pb._config.username = process.env.SHARED_POCKETBASE_USERNAME;
		pb._config.password = process.env.SHARED_POCKETBASE_PASSWORD;
	}

	// Heartbeat function to check connection periodically
	const heartbeatInterval = setInterval(async () => {
		try {
			// Simple health check
			await pb.health.check();
			pb.isConnected = true;
		} catch (error) {
			logger.warn(`PocketBase connection issue: ${error.message}`);
			pb.isConnected = false;
			await pb.ensureConnection();
		}
	}, 5 * 60 * 1000); // Check every 5 minutes

	// Clean up on client disconnect
	pb.cleanup = () => {
		clearInterval(heartbeatInterval);
	};
};