Feature flag and permissions change handling via client reset and disconnect

.github/workflows/bin/check_if_any_files_changed.sh

@@ -1,10 +1,16 @@
 #!/usr/bin/bash
 
+set -ex
+
+protoc --version
+git status --porcelain
+git diff -p
+
 for file in $(git status --porcelain | awk '{ print $2 }')
 do
   echo "::error file=$file,title=This generated file doesn't match the protobuf file::Please run \`make update_protobuf\` to update \`$file\` and commit the changes."
 done
 
 if [ -n "`git status --porcelain`" ]; then
   exit 1
-fi
+fi

clients/typescript/src/_generated/protocol/satellite.ts

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-ts_proto. DO NOT EDIT.
 // versions:
 //   protoc-gen-ts_proto  v1.175.0
-//   protoc               v4.25.3
+//   protoc               v5.26.1
 // source: protocol/satellite.proto
 
 /* eslint-disable */
@@ -731,6 +731,21 @@ export interface SatShapeDataEnd {
   $type: "Electric.Satellite.SatShapeDataEnd";
 }
 
+export interface SatClientCommand {
+  $type: "Electric.Satellite.SatClientCommand";
+  resetDatabase?: SatClientCommand_ResetDatabase | undefined;
+}
+
+export interface SatClientCommand_ResetDatabase {
+  $type: "Electric.Satellite.SatClientCommand.ResetDatabase";
+  reason: SatClientCommand_ResetDatabase_Reason;
+}
+
+export enum SatClientCommand_ResetDatabase_Reason {
+  PERMISSIONS_CHANGE = 0,
+  UNRECOGNIZED = -1,
+}
+
 /**
  * represents the client permissions
  * - `Rules`: the global permission rules, defined by the DDLX
@@ -4529,6 +4544,108 @@ export const SatShapeDataEnd = {
 
 messageTypeRegistry.set(SatShapeDataEnd.$type, SatShapeDataEnd);
 
+function createBaseSatClientCommand(): SatClientCommand {
+  return { $type: "Electric.Satellite.SatClientCommand", resetDatabase: undefined };
+}
+
+export const SatClientCommand = {
+  $type: "Electric.Satellite.SatClientCommand" as const,
+
+  encode(message: SatClientCommand, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+    if (message.resetDatabase !== undefined) {
+      SatClientCommand_ResetDatabase.encode(message.resetDatabase, writer.uint32(10).fork()).ldelim();
+    }
+    return writer;
+  },
+
+  decode(input: _m0.Reader | Uint8Array, length?: number): SatClientCommand {
+    const reader = input instanceof _m0.Reader ? input : _m0.Reader.create(input);
+    let end = length === undefined ? reader.len : reader.pos + length;
+    const message = createBaseSatClientCommand();
+    while (reader.pos < end) {
+      const tag = reader.uint32();
+      switch (tag >>> 3) {
+        case 1:
+          if (tag !== 10) {
+            break;
+          }
+
+          message.resetDatabase = SatClientCommand_ResetDatabase.decode(reader, reader.uint32());
+          continue;
+      }
+      if ((tag & 7) === 4 || tag === 0) {
+        break;
+      }
+      reader.skipType(tag & 7);
+    }
+    return message;
+  },
+
+  create<I extends Exact<DeepPartial<SatClientCommand>, I>>(base?: I): SatClientCommand {
+    return SatClientCommand.fromPartial(base ?? ({} as any));
+  },
+  fromPartial<I extends Exact<DeepPartial<SatClientCommand>, I>>(object: I): SatClientCommand {
+    const message = createBaseSatClientCommand();
+    message.resetDatabase = (object.resetDatabase !== undefined && object.resetDatabase !== null)
+      ? SatClientCommand_ResetDatabase.fromPartial(object.resetDatabase)
+      : undefined;
+    return message;
+  },
+};
+
+messageTypeRegistry.set(SatClientCommand.$type, SatClientCommand);
+
+function createBaseSatClientCommand_ResetDatabase(): SatClientCommand_ResetDatabase {
+  return { $type: "Electric.Satellite.SatClientCommand.ResetDatabase", reason: 0 };
+}
+
+export const SatClientCommand_ResetDatabase = {
+  $type: "Electric.Satellite.SatClientCommand.ResetDatabase" as const,
+
+  encode(message: SatClientCommand_ResetDatabase, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+    if (message.reason !== 0) {
+      writer.uint32(8).int32(message.reason);
+    }
+    return writer;
+  },
+
+  decode(input: _m0.Reader | Uint8Array, length?: number): SatClientCommand_ResetDatabase {
+    const reader = input instanceof _m0.Reader ? input : _m0.Reader.create(input);
+    let end = length === undefined ? reader.len : reader.pos + length;
+    const message = createBaseSatClientCommand_ResetDatabase();
+    while (reader.pos < end) {
+      const tag = reader.uint32();
+      switch (tag >>> 3) {
+        case 1:
+          if (tag !== 8) {
+            break;
+          }
+
+          message.reason = reader.int32() as any;
+          continue;
+      }
+      if ((tag & 7) === 4 || tag === 0) {
+        break;
+      }
+      reader.skipType(tag & 7);
+    }
+    return message;
+  },
+
+  create<I extends Exact<DeepPartial<SatClientCommand_ResetDatabase>, I>>(base?: I): SatClientCommand_ResetDatabase {
+    return SatClientCommand_ResetDatabase.fromPartial(base ?? ({} as any));
+  },
+  fromPartial<I extends Exact<DeepPartial<SatClientCommand_ResetDatabase>, I>>(
+    object: I,
+  ): SatClientCommand_ResetDatabase {
+    const message = createBaseSatClientCommand_ResetDatabase();
+    message.reason = object.reason ?? 0;
+    return message;
+  },
+};
+
+messageTypeRegistry.set(SatClientCommand_ResetDatabase.$type, SatClientCommand_ResetDatabase);
+
 function createBaseSatPerms(): SatPerms {
   return { $type: "Electric.Satellite.SatPerms", id: Long.ZERO, userId: "", rules: undefined, roles: [] };
 }

clients/typescript/src/_generated/typeRegistry.ts

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-ts_proto. DO NOT EDIT.
 // versions:
 //   protoc-gen-ts_proto  v1.175.0
-//   protoc               v4.25.3
+//   protoc               v5.26.1
 
 /* eslint-disable */
 import Long from "long";

clients/typescript/src/satellite/client.ts

@@ -29,6 +29,7 @@ import {
   RootClientImpl,
   SatRpcRequest,
   SatInStartReplicationReq_Dialect,
+  SatClientCommand,
 } from '../_generated/protocol/satellite'
 import {
   getObjFromString,
@@ -76,6 +77,7 @@ import {
   ReplicatedRowTransformer,
   DataGone,
   GoneBatchCallback,
+  CommandCallback,
 } from '../util/types'
 import {
   base64,
@@ -142,7 +144,9 @@ type Events = {
     changes: DataGone[],
     ack: () => void
   ) => Promise<void>
+  command: (cmd: SatClientCommand) => Promise<void>
 }
+
 type EventEmitter = AsyncEventEmitter<Events>
 
 export class SatelliteClient implements Client {
@@ -195,6 +199,7 @@ export class SatelliteClient implements Client {
         SatUnsubsDataBegin: (msg) => this.handleUnsubsDataBegin(msg),
         SatUnsubsDataEnd: (msg) => this.handleUnsubsDataEnd(msg),
         SatPerms: (msg) => void msg,
+        SatClientCommand: (msg) => this.handleCommand(msg),
       } satisfies HandlerMapping).map((e) => [getFullTypeName(e[0]), e[1]])
     )
 
@@ -500,6 +505,14 @@ export class SatelliteClient implements Client {
     // TODO: real removeListener implementation, because the old one for txns doesn't work
   }
 
+  subscribeToCommands(callback: CommandCallback): void {
+    this.emitter.on('command', callback)
+  }
+
+  unsubscribeToCommands(callback: CommandCallback): void {
+    this.emitter.removeListener('command', callback)
+  }
+
   enqueueTransaction(transaction: DataTransaction): void {
     if (this.outbound.isReplicating !== ReplicationStatus.ACTIVE) {
       throw new SatelliteError(
@@ -986,6 +999,10 @@ export class SatelliteClient implements Client {
     this.subscriptionsDataCache.shapeDataEnd()
   }
 
+  private handleCommand(msg: SatClientCommand): void {
+    this.emitter.enqueueEmit('command', msg)
+  }
+
   // For now, unsubscribe responses doesn't send any information back
   // It might eventually confirm that the server processed it or was noop.
   private handleUnsubscribeResponse(_msg: SatUnsubsResp): UnsubscribeResponse {

clients/typescript/src/satellite/index.ts

@@ -23,6 +23,7 @@ import {
   DbRecord,
   ReplicatedRowTransformer,
   GoneBatchCallback,
+  CommandCallback,
 } from '../util/types'
 import {
   ShapeRequest,
@@ -116,6 +117,8 @@ export interface Client {
   unsubscribeToOutboundStarted(callback: OutboundStartedCallback): void
   subscribeToError(callback: ErrorCallback): void
   unsubscribeToError(callback: ErrorCallback): void
+  subscribeToCommands(callback: CommandCallback): void
+  unsubscribeToCommands(callback: CommandCallback): void
 
   subscribe(subId: string, shapes: ShapeRequest[]): Promise<SubscribeResponse>
   unsubscribe(subIds: string[]): Promise<UnsubscribeResponse>

clients/typescript/src/satellite/mock.ts

@@ -25,6 +25,7 @@ import {
   GoneBatchCallback,
   DataGone,
   DataChangeType,
+  CommandCallback,
 } from '../util/types'
 import { ElectricConfig } from '../config/index'
 
@@ -217,6 +218,7 @@ export class MockSatelliteClient
   relationsCb?: (relation: Relation) => void
   transactionsCb?: TransactionCallback
   additionalDataCb?: AdditionalDataCallback
+  commandCb?: CommandCallback
 
   outboundStartedCallback?: OutboundStartedCallback
 
@@ -385,6 +387,14 @@ export class MockSatelliteClient
     this.on('error', cb)
   }
 
+  subscribeToCommands(callback: CommandCallback): void {
+    this.commandCb = callback
+  }
+
+  unsubscribeToCommands(_callback: CommandCallback): void {
+    this.commandCb = undefined
+  }
+
   emitSocketClosedError(ev: SocketCloseReason): void {
     this.enqueueEmit('error', new SatelliteError(ev, 'socket closed'))
   }

clients/typescript/src/satellite/process.ts

@@ -3,6 +3,7 @@ import throttle from 'lodash.throttle'
 import {
   SatOpMigrate_Type,
   SatRelation_RelationType,
+  SatClientCommand,
 } from '../_generated/protocol/satellite'
 import { AuthConfig, AuthState } from '../auth/index'
 import { DatabaseAdapter } from '../electric/adapter'
@@ -323,6 +324,9 @@ export class SatelliteProcess implements Satellite {
     const clientGoneBatchCallback = this._applyGoneBatch.bind(this)
     this.client.subscribeToGoneBatch(clientGoneBatchCallback)
 
+    const clientCommandCallback = this._handleCommand.bind(this)
+    this.client.subscribeToCommands(clientCommandCallback)
+
     const clientSubscriptionDataCallback =
       this._handleSubscriptionData.bind(this)
     const clientSubscriptionErrorCallback =
@@ -339,6 +343,7 @@ export class SatelliteProcess implements Satellite {
       this.client.unsubscribeToTransactions(clientTransactionsCallback)
       this.client.unsubscribeToAdditionalData(clientAdditionalDataCallback)
       this.client.unsubscribeToOutboundStarted(clientOutboundStartedCallback)
+      this.client.unsubscribeToCommands(clientCommandCallback)
 
       this.client.unsubscribeToSubscriptionEvents(
         clientSubscriptionDataCallback,
@@ -1564,6 +1569,14 @@ export class SatelliteProcess implements Satellite {
     this._notifyChanges(fakeOplogEntries, 'remote')
   }
 
+  async _handleCommand(cmd: SatClientCommand) {
+    if (cmd.resetDatabase) {
+      await this._resetClientState({
+        keepSubscribedShapes: true,
+      })
+    }
+  }
+
   private async maybeGarbageCollect(
     origin: string,
     commitTimestamp: Date

clients/typescript/src/util/proto.ts

@@ -130,6 +130,7 @@ const msgtypetuples: MappingTuples = {
   SatUnsubsDataBegin: [24, Pb.SatUnsubsDataBegin],
   SatUnsubsDataEnd: [25, Pb.SatUnsubsDataEnd],
   SatPerms: [26, Pb.SatPerms],
+  SatClientCommand: [27, Pb.SatClientCommand],
 }
 
 const msgtypemapping = Object.fromEntries(
@@ -184,6 +185,7 @@ export type SatPbMsg =
   | Pb.SatUnsubsDataBegin
   | Pb.SatUnsubsDataEnd
   | Pb.SatPerms
+  | Pb.SatClientCommand
 
 export type SatPbMsgObj<
   Msg extends { $type: string },
@@ -428,6 +430,12 @@ export function msgToString(message: MessageOfInterest): string {
       return `#SatUnsubsDataEnd{}`
     case 'Electric.Satellite.SatPerms':
       return `#SatPerms{}`
+    case 'Electric.Satellite.SatClientCommand':
+      if (message.resetDatabase) {
+        return `#SatClientCommand{command: #ResetDatabase{reason: ${message.resetDatabase.reason}}}`
+      }
+
+      return `#SatClientCommand{command: <unknown>}`
   }
 }
 

clients/typescript/src/util/types.ts

@@ -5,6 +5,7 @@ import {
   SatOpMigrate_Table,
   SatOpMigrate_Type,
   SatRelation_RelationType,
+  SatClientCommand,
 } from '../_generated/protocol/satellite'
 import { Tag } from '../satellite/oplog'
 
@@ -265,6 +266,7 @@ export type GoneBatchCallback = (
   subscriptionIds: string[],
   changes: DataGone[]
 ) => void | Promise<void>
+export type CommandCallback = (cmd: SatClientCommand) => Promise<void>
 
 export type ConnectivityStatus = 'connected' | 'disconnected'
 export type ConnectivityState = {

clients/typescript/test/satellite/client.test.ts

@@ -809,7 +809,7 @@ test.serial('default and null test', async (t) => {
   })
 })
 
-test.serial('subscription succesful', async (t) => {
+test.serial('subscription successful', async (t) => {
   await connectAndAuth(t.context)
   const { client, server } = t.context
   await startReplication(client, server)
@@ -1310,7 +1310,7 @@ test.serial('client correctly handles additional data messages', async (t) => {
   })
 })
 
-test.serial('unsubscribe successfull', async (t) => {
+test.serial('unsubscribe successful', async (t) => {
   await connectAndAuth(t.context)
   const { client, server } = t.context
   await startReplication(client, server)