clean up weblite and start on pubsub refactor

Author: jacobprall

package-lock.json

@@ -472,7 +472,8 @@ export class Database extends EventEmitter {
             } else {
               // metadata for operations like insert, update, delete?
               const context = this.processContext(results)
-              resolve(context ? context : results)
+              const result = { data: context ? context : results, error: null }
+              resolve(result)
             }
           })
         }

src/drivers/database.ts

@@ -2,10 +2,8 @@ import { Database } from '../drivers/database'
 import { Fetch, fetchWithAuth } from './utils/fetch'
 import { PubSubClient } from './_pubsub/PubSubClient'
 import { WebliteClient } from './weblite/WebliteClient'
-import { StorageClient } from './_storage/StorageClient'
 import { SQLiteCloudDataTypes, SQLiteCloudError } from '../drivers/types'
 import { cleanConnectionString } from './utils'
-import { FunctionsClient } from './_functions/FunctionsClient'
 import { SQLiteCloudClientConfig } from './types'
 import { DEFAULT_HEADERS } from '../drivers/constants'
 
@@ -61,13 +59,6 @@ export class SQLiteCloudClient {
     return await this.weblite.sql(sql, ...values)
   }
 
-  get pubSub() {
-    if (!this._pubSub) {
-      this._pubSub = new PubSubClient(this.db.getConfiguration())
-    }
-    return this._pubSub
-  }
-
   get db() {
     if (!this._db) {
       this._db = new Database(this.connectionString)
@@ -85,18 +76,18 @@ export class SQLiteCloudClient {
     return this._weblite
   }
 
-  get files() {
-      return new StorageClient(this.connectionString, { 
-        customFetch: this.fetch, 
-        headers: this._globalHeaders 
-      })
+  get pubSub() {
+    if (!this._pubSub) {
+      this._pubSub = new PubSubClient(
+        this.db
+      )
+    }
+    return this._pubSub
   }
 
-  get functions() {
-    return new FunctionsClient(this.connectionString, { 
-      customFetch: this.fetch, 
-      headers: this._globalHeaders 
-    })
+  close() {
+    if (this._db) this._db.close()
+    if (this._pubSub) this._pubSub.close()
   }
 }
 

src/packages/SQLiteCloudClient.ts

@@ -1,6 +1,8 @@
 import { SQLiteCloudConnection } from '../../drivers/connection'
 import SQLiteCloudTlsConnection from '../../drivers/connection-tls'
+import { Database } from '../../drivers/database'
 import { SQLiteCloudConfig } from '../../drivers/types'
+import { getDefaultDatabase } from '../utils'
 
 /**
  * PubSubCallback
@@ -33,14 +35,13 @@ export interface ListenOptions {
  * @param close - Close the connection.
  */
 export interface PubSub {
-  listen<T>(options: ListenOptions, callback: PubSubCallback): Promise<T>
+  listen<T>(options: ListenOptions, callback: PubSubCallback): Promise<string>
   unlisten(options: ListenOptions): void
   subscribe(channelName: string, callback: PubSubCallback): Promise<any>
   unsubscribe(channelName: string): void
   create(channelName: string, failIfExists: boolean): Promise<any>
   delete(channelName: string): Promise<any>
   notify(channelName: string, message: string): Promise<any>
-  setPubSubOnly(): Promise<any>
   connected(): boolean
   close(): void
 }
@@ -50,13 +51,15 @@ export interface PubSub {
  */
 export class PubSubClient implements PubSub {
   protected _pubSubConnection: SQLiteCloudConnection | null
-  protected defaultDatabaseName: string
+  protected _queryConnection: Database
+  protected defaultDatabase: string
   protected config: SQLiteCloudConfig
 
-  constructor(config: SQLiteCloudConfig) {
-    this.config = config
+  constructor(conn: Database) {
+    this.config = conn.getConfiguration()
+    this.defaultDatabase = this.config.database ?? ''
+    this._queryConnection = conn
     this._pubSubConnection = null
-    this.defaultDatabaseName = config?.database ?? ''
   }
 
   /**
@@ -66,16 +69,16 @@ export class PubSubClient implements PubSub {
    * @param callback Callback to be called when a message is received
    */
 
-  get pubSubConnection(): SQLiteCloudConnection {
+  private get pubSubConnection(): SQLiteCloudConnection {
     if (!this._pubSubConnection) {
       this._pubSubConnection = new SQLiteCloudTlsConnection(this.config)
     }
     return this._pubSubConnection
   }
 
-  async listen<T>(options: ListenOptions, callback: PubSubCallback): Promise<T> {
-    const _dbName = options.dbName ? options.dbName : this.defaultDatabaseName;
-    const authCommand: string = await this.pubSubConnection.sql`LISTEN ${options.tableName} DATABASE ${_dbName};`
+  async listen<T>(options: ListenOptions, callback: PubSubCallback): Promise<string> {
+    const _dbName = options.dbName ? options.dbName : this.defaultDatabase;
+    const authCommand: string = await this._queryConnection.sql`LISTEN TABLE ${options.tableName} DATABASE ${_dbName};`
 
     return new Promise((resolve, reject) => {
       this.pubSubConnection.sendCommands(authCommand, (error, results) => {
@@ -85,6 +88,7 @@ export class PubSubClient implements PubSub {
         } else {
           // skip results from pubSub auth command
           if (results !== 'OK') {
+            console.log(results)
             callback.call(this, null, results)
           }
           resolve(results)
@@ -97,8 +101,8 @@ export class PubSubClient implements PubSub {
    * Unlisten to a table.
    * @param options Options for the unlisten operation.
    */
-  public unlisten(options: ListenOptions): void {
-    this.pubSubConnection.sql`UNLISTEN ${options.tableName} DATABASE ${options.dbName};`
+  public async unlisten(options: ListenOptions): Promise<any> {
+    return this._queryConnection.sql`UNLISTEN ${options.tableName} DATABASE ${options.dbName};`
   }
 
   /**
@@ -107,7 +111,7 @@ export class PubSubClient implements PubSub {
    * @param callback Callback to be called when a message is received.
    */
   public async subscribe(channelName: string, callback: PubSubCallback): Promise<any> {
-    const authCommand: string = await this.pubSubConnection.sql`LISTEN ${channelName};`
+    const authCommand: string = await this._queryConnection.sql`LISTEN ${channelName};`
 
     return new Promise((resolve, reject) => {
       this.pubSubConnection.sendCommands(authCommand, (error, results) => {
@@ -125,8 +129,8 @@ export class PubSubClient implements PubSub {
    * Unsubscribe (unlisten) from a channel.
    * @param channelName The name of the channel to unsubscribe from.
    */
-  public unsubscribe(channelName: string): void {
-    this.pubSubConnection.sql`UNLISTEN ${channelName};`
+  public async unsubscribe(channelName: string): Promise<void> {
+    return this._queryConnection.sql`UNLISTEN ${channelName};`
   }
 
   /**
@@ -135,7 +139,7 @@ export class PubSubClient implements PubSub {
    * @param failIfExists Raise an error if the channel already exists
    */
   public async create(channelName: string, failIfExists: boolean = true): Promise<any> {
-    return await this.pubSubConnection.sql(
+    return this._queryConnection.sql(
       `CREATE CHANNEL ?${failIfExists ? '' : ' IF NOT EXISTS'};`, channelName
     )
   }
@@ -145,34 +149,14 @@ export class PubSubClient implements PubSub {
    * @param name Channel name
    */
   public async delete(channelName: string): Promise<any> {
-    return await this.pubSubConnection.sql`REMOVE CHANNEL ${channelName};`
+    return this._queryConnection.sql`REMOVE CHANNEL ${channelName};`
   }
 
   /**
    * Send a message to the channel.
    */
-  public notify(channelName: string, message: string): Promise<any> {
-    return this.pubSubConnection.sql`NOTIFY ${channelName} ${message};`
-  }
-
-  // DOUBLE CHECK THIS
-
-  /**
-   * Ask the server to close the connection to the database and
-   * to keep only open the Pub/Sub connection.
-   * Only interaction with Pub/Sub commands will be allowed.
-   */
-  public setPubSubOnly(): Promise<any> {
-    return new Promise((resolve, reject) => {
-      this.pubSubConnection.sendCommands('PUBSUB ONLY;', (error, results) => {
-        if (error) {
-          reject(error)
-        } else {
-          this.pubSubConnection.close()
-          resolve(results)
-        }
-      })
-    })
+  public async notify(channelName: string, message: string) {
+    return await this._queryConnection.sql`NOTIFY ${channelName} ${message};`
   }
 
   /** True if Pub/Sub connection is open. */

src/packages/_pubsub/PubSubClient.ts

@@ -23,7 +23,7 @@ export class StorageClient implements Storage {
     this.filesUrl = getAPIUrl(connectionString, 'files')
     this.webliteSQLUrl = getAPIUrl(connectionString, 'weblite/sql')
     this.fetch = options.customFetch || fetchWithAuth(connectionString)
-    this.headers = { ...DEFAULT_HEADERS, ...options.headers }
+    this.headers = { ...options.headers }
   }
 
   async createBucket(bucket: string) {

src/packages/_storage/StorageClient.ts

@@ -3,10 +3,11 @@ import { SQLiteCloudClient } from '../SQLiteCloudClient'
 
 const DEFAULT_TABLE_NAME = 'albums';
 
-const client = new SQLiteCloudClient(CHINOOK_DATABASE_URL)
+
 
 describe('SQLiteCloudClient test suite', () => {
   it('should be able to create a client', () => {
+    const client = new SQLiteCloudClient(CHINOOK_DATABASE_URL)
     expect(client).toBeDefined()
     expect(client).toBeInstanceOf(SQLiteCloudClient)
   })
@@ -18,8 +19,17 @@ describe('SQLiteCloudClient test suite', () => {
   })
 
   it('should be able to query the database via HTTP', async () => {
+    const client = new SQLiteCloudClient(CHINOOK_DATABASE_URL)
     const { data, error } = await client.sql`SELECT * FROM ${DEFAULT_TABLE_NAME}`;
     expect(data).toBeDefined()
     expect(error).toBeNull()
   })
-})
+
+  it('should be able to query via database connection', async () => {
+    const client = new SQLiteCloudClient(CHINOOK_DATABASE_URL)
+    const { data, error } = await client.db.sql('SELECT * FROM albums')
+    expect(data).toBeDefined()
+    expect(error).toBeNull()
+    client.close()
+  })
+})