Calling invalidate credentials over comlink, so that worker informs tab's remote connector about invalidation.

Author: Chriztiaan

packages/common/src/client/sync/stream/AbstractRemote.ts

@@ -15,6 +15,7 @@ export type BSONImplementation = typeof BSON;
 
 export type RemoteConnector = {
   fetchCredentials: () => Promise<PowerSyncCredentials | null>;
+  invalidateCredentials?: () => void;
 };
 
 const POWERSYNC_TRAILING_SLASH_MATCH = /\/+$/;
@@ -182,6 +183,7 @@ export abstract class AbstractRemote {
    */
   invalidateCredentials() {
     this.credentials = null;
+    this.connector.invalidateCredentials?.();
   }
 
   getUserAgent() {
@@ -396,8 +398,7 @@ export abstract class AbstractRemote {
         syncQueueRequestSize, // The initial N amount
         {
           onError: (e) => {
-            if (e.message.includes('PSYNC_S2101')) {
-              this.logger.error('PSYNC_S2101 - 401 Unauthorized');
+            if (e.message.includes('Authorization failed') || e.message.includes('401')) {
               this.invalidateCredentials();
             }
             // Don't log closed as an error

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -687,9 +687,6 @@ The next upload iteration will be delayed.`);
 
               await this.delayRetry();
               return;
-            } else if (remaining_seconds < 30) {
-              // Pre-emptively refresh the token
-              await this.options.remote.prefetchCredentials();
             }
             this.triggerCrudUpload();
           } else {

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -31,6 +31,10 @@ class SharedSyncClientProvider extends AbstractSharedSyncClientProvider {
     return Comlink.transfer(port, [port]);
   }
 
+  invalidateCredentials() {
+    this.options.remote.invalidateCredentials();
+  }
+
   async fetchCredentials(): Promise<PowerSyncCredentials | null> {
     const credentials = await this.options.remote.getCredentials();
     if (credentials == null) {

packages/web/src/worker/sync/AbstractSharedSyncClientProvider.ts

@@ -5,6 +5,7 @@ import type { PowerSyncCredentials, SyncStatusOptions } from '@powersync/common'
  */
 export abstract class AbstractSharedSyncClientProvider {
   abstract fetchCredentials(): Promise<PowerSyncCredentials | null>;
+  abstract invalidateCredentials(): void;
   abstract uploadCrud(): Promise<void>;
   abstract statusChanged(status: SyncStatusOptions): void;
   abstract getDBWorkerPort(): Promise<MessagePort>;

packages/web/src/worker/sync/SharedSyncImplementation.ts

@@ -308,6 +308,15 @@ export class SharedSyncImplementation
       adapter: new SqliteBucketStorage(this.dbAdapter!, new Mutex(), this.logger),
       remote: new WebRemote(
         {
+          invalidateCredentials: async () => {
+            const lastPort = this.ports[this.ports.length - 1];
+            try {
+              this.logger.log('calling the last port client provider to invalidate credentials');
+              lastPort.clientProvider.invalidateCredentials();
+            } catch (ex) {
+              this.logger.error('error invalidating credentials', ex);
+            }
+          },
           fetchCredentials: async () => {
             const lastPort = this.ports[this.ports.length - 1];
             return new Promise(async (resolve, reject) => {