Avoid subscription terminology

Author: simolus3

packages/service-core/src/sync/BucketChecksumState.ts

@@ -228,7 +228,7 @@ export class BucketChecksumState {
       bucketsToFetch = allBuckets;
       this.parameterState.syncRules.bucketDescriptors;
 
-      const subscriptions: util.SubscribedStream[] = [];
+      const subscriptions: util.StreamDescription[] = [];
       for (const desc of this.parameterState.syncRules.bucketDescriptors) {
         if (desc.type == SqlBucketDescriptorType.STREAM && this.parameterState.isSubscribedToStream(desc)) {
           subscriptions.push({
@@ -246,7 +246,7 @@ export class BucketChecksumState {
             ...e,
             ...bucketDescriptionMap.get(e.bucket)!
           })),
-          included_subscriptions: subscriptions
+          streams: subscriptions
         }
       } satisfies util.StreamingSyncCheckpoint;
     }
@@ -355,7 +355,7 @@ export class BucketParameterState {
   private readonly querier: BucketParameterQuerier;
   private readonly staticBuckets: Map<string, BucketDescription>;
   private readonly includeDefaultStreams: boolean;
-  private readonly explicitStreamSubscriptions: Record<string, util.StreamSubscription>;
+  private readonly explicitlyOpenedStreams: Record<string, util.OpenedStream>;
   private readonly logger: Logger;
   private cachedDynamicBuckets: BucketDescription[] | null = null;
   private cachedDynamicBucketSet: Set<string> | null = null;
@@ -376,21 +376,21 @@ export class BucketParameterState {
     this.syncParams = syncParams;
     this.logger = logger;
 
-    const explicitStreamSubscriptions: Record<string, util.StreamSubscription> = {};
+    const explicitlyOpenedStreams: Record<string, util.OpenedStream> = {};
     const subscriptions = request.subscriptions;
     if (subscriptions) {
-      for (const subscription of subscriptions.subscriptions) {
-        explicitStreamSubscriptions[subscription.stream] = subscription;
+      for (const subscription of subscriptions.opened) {
+        explicitlyOpenedStreams[subscription.stream] = subscription;
       }
     }
     this.includeDefaultStreams = subscriptions?.include_defaults ?? true;
-    this.explicitStreamSubscriptions = explicitStreamSubscriptions;
+    this.explicitlyOpenedStreams = explicitlyOpenedStreams;
 
     this.querier = syncRules.getBucketParameterQuerier({
       globalParameters: this.syncParams,
       hasDefaultSubscriptions: this.includeDefaultStreams,
       resolveSubscription(name) {
-        const subscription = explicitStreamSubscriptions[name];
+        const subscription = explicitlyOpenedStreams[name];
         if (subscription) {
           return subscription.parameters ?? {};
         } else {
@@ -408,7 +408,7 @@ export class BucketParameterState {
    * In partiuclar, this can override the priority assigned to a bucket.
    */
   overrideBucketDescription(description: BucketDescription): BucketDescription {
-    const changedPriority = this.explicitStreamSubscriptions[description.definition]?.override_priority;
+    const changedPriority = this.explicitlyOpenedStreams[description.definition]?.override_priority;
     if (changedPriority != null && isValidPriority(changedPriority)) {
       return {
         ...description,
@@ -420,7 +420,7 @@ export class BucketParameterState {
   }
 
   isSubscribedToStream(desc: SqlBucketDescriptor): boolean {
-    return (desc.subscribedToByDefault && this.includeDefaultStreams) || desc.name in this.explicitStreamSubscriptions;
+    return (desc.subscribedToByDefault && this.includeDefaultStreams) || desc.name in this.explicitlyOpenedStreams;
   }
 
   async getCheckpointUpdate(checkpoint: storage.StorageCheckpointUpdate): Promise<CheckpointUpdate> {

packages/service-core/src/util/protocol-types.ts

@@ -14,32 +14,32 @@ export const BucketRequest = t.object({
 export type BucketRequest = t.Decoded<typeof BucketRequest>;
 
 /**
- * An explicit subscription to a defined sync stream made by the client.
+ * A sync steam that a client has expressed interest in by explicitly opening it on the client side.
  */
-export const StreamSubscription = t.object({
+export const OpenedStream = t.object({
   /**
-   * The defined name of the stream as it appears in sync rules.
+   * The defined name of the stream as it appears in sync stream definitions.
    */
   stream: t.string,
   /**
    * An optional dictionary of parameters to pass to this specific stream.
    */
   parameters: t.record(t.any).optional(),
   /**
-   * Set when the client wishes to re-assign a different priority to this subscription.
+   * Set when the client wishes to re-assign a different priority to this stream.
    *
    * Streams and sync rules can also assign a default priority, but clients are allowed to override those. This can be
    * useful when the priority for partial syncs depends on e.g. the current page opened in a client.
    */
   override_priority: t.number.optional()
 });
 
-export type StreamSubscription = t.Decoded<typeof StreamSubscription>;
+export type OpenedStream = t.Decoded<typeof OpenedStream>;
 
 /**
- * An overview of all subscriptions as part of a streaming sync request.
+ * An overview of all opened streams as part of a streaming sync request.
  */
-export const StreamSubscriptions = t.object({
+export const OpenedStreams = t.object({
   /**
    * Whether to sync default streams.
    *
@@ -48,12 +48,12 @@ export const StreamSubscriptions = t.object({
   include_defaults: t.boolean.optional(),
 
   /**
-   * An array of streams the client has subscribed to.
+   * An array of sync streams the client has opened explicitly.
    */
-  subscriptions: t.array(StreamSubscription)
+  opened: t.array(OpenedStream)
 });
 
-export type StreamSubscriptions = t.Decoded<typeof StreamSubscriptions>;
+export type StreamSubscriptions = t.Decoded<typeof OpenedStreams>;
 
 export const StreamingSyncRequest = t.object({
   /**
@@ -92,9 +92,9 @@ export const StreamingSyncRequest = t.object({
   client_id: t.string.optional(),
 
   /**
-   * If the client is aware of stream subscriptions, an array of streams the client is subscribing to.
+   * If the client is aware of streams, an array of streams the client has opened.
    */
-  subscriptions: StreamSubscriptions.optional()
+  subscriptions: OpenedStreams.optional()
 });
 
 export type StreamingSyncRequest = t.Decoded<typeof StreamingSyncRequest>;
@@ -146,7 +146,7 @@ export type StreamingSyncLine =
  */
 export type ProtocolOpId = string;
 
-export interface SubscribedStream {
+export interface StreamDescription {
   name: string;
   is_default: boolean;
 }
@@ -155,7 +155,7 @@ export interface Checkpoint {
   last_op_id: ProtocolOpId;
   write_checkpoint?: ProtocolOpId;
   buckets: BucketChecksumWithDescription[];
-  included_subscriptions: SubscribedStream[];
+  streams: StreamDescription[];
 }
 
 export interface BucketState {