Review feedback

Author: simolus3

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

@@ -381,7 +381,7 @@ export class BucketParameterState {
   private readonly staticBuckets: Map<string, ResolvedBucket>;
   private readonly includeDefaultStreams: boolean;
   // Indexed by the client-side id
-  private readonly explicitStreamSubscriptions: Record<string, util.RequestedStreamSubscription>;
+  private readonly explicitStreamSubscriptions: util.RequestedStreamSubscription[];
   private readonly subscribedStreamNames: Set<string>;
   private readonly logger: Logger;
   private cachedDynamicBuckets: ResolvedBucket[] | null = null;
@@ -403,16 +403,16 @@ export class BucketParameterState {
     this.syncParams = new RequestParameters(tokenPayload, request.parameters ?? {});
     this.logger = logger;
 
-    const idToStreamSubscription: Record<string, util.RequestedStreamSubscription> = {};
     const streamsByName: Record<string, RequestedStream[]> = {};
     const subscriptions = request.streams;
+    const explicitStreamSubscriptions: util.RequestedStreamSubscription[] = subscriptions?.subscriptions ?? [];
     if (subscriptions) {
-      for (const subscription of subscriptions.subscriptions) {
-        idToStreamSubscription[subscription.client_id] = subscription;
+      for (let i = 0; i < explicitStreamSubscriptions.length; i++) {
+        const subscription = explicitStreamSubscriptions[i];
 
         const syncRuleStream: RequestedStream = {
           parameters: subscription.parameters ?? {},
-          opaque_id: subscription.client_id
+          opaque_id: i
         };
         if (Object.hasOwn(streamsByName, subscription.stream)) {
           streamsByName[subscription.stream].push(syncRuleStream);
@@ -422,7 +422,7 @@ export class BucketParameterState {
       }
     }
     this.includeDefaultStreams = subscriptions?.include_defaults ?? true;
-    this.explicitStreamSubscriptions = idToStreamSubscription;
+    this.explicitStreamSubscriptions = explicitStreamSubscriptions;
 
     this.querier = syncRules.getBucketParameterQuerier({
       globalParameters: this.syncParams,
@@ -469,7 +469,7 @@ export class BucketParameterState {
           const stream = description.definition;
           return { default: lookupIndex.get(stream)! };
         } else {
-          return reason.subscription;
+          return { sub: reason.subscription };
         }
       })
     };

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

@@ -21,10 +21,6 @@ export const RequestedStreamSubscription = t.object({
    * The defined name of the stream as it appears in sync stream definitions.
    */
   stream: t.string,
-  /**
-   * An opaque textual identifier assigned to this request by the client.
-   */
-  client_id: t.string,
   /**
    * An optional dictionary of parameters to pass to this specific stream.
    */
@@ -151,14 +147,35 @@ export type StreamingSyncLine =
 export type ProtocolOpId = string;
 
 export interface StreamDescription {
+  /**
+   * The name of the stream as it appears in the sync configuration.
+   */
   name: string;
+
+  /**
+   * Whether this stream is subscribed to by default.
+   *
+   * For default streams, this field is still `true` if clients have an explicit subscription to the stream.
+   */
   is_default: boolean;
 }
 
 export interface Checkpoint {
   last_op_id: ProtocolOpId;
   write_checkpoint?: ProtocolOpId;
   buckets: CheckpointBucket[];
+
+  /**
+   * All streams that the client is subscribed to.
+   *
+   * This field has two purposes:
+   *
+   *  1. It allows clients to determine which of their subscriptions actually works. E.g. if a user does
+   *     `db.syncStream('non_existent_stream').subscribe()`, clients don't immediately know that the stream doesn't
+   *     exist. Only after the next `checkpoint` line can they query this field and mark unresolved subscriptions.
+   *. 2. It allows clients to learn which default streams they have been subscribed to. This is relevant for APIs
+   *     listing all streams on the client-side.
+   */
   streams: StreamDescription[];
 }
 
@@ -236,10 +253,13 @@ export type BucketDerivedFromDefaultStream = {
 /**
  * The bucket has been included in a checkpoint because it's part of a stream that a client has explicitly subscribed
  * to.
- *
- * The string is the client id associated with the subscription in {@link RequestedStreamSubscription}.
  */
-export type BucketDerivedFromExplicitSubscription = string;
+export type BucketDerivedFromExplicitSubscription = {
+  /**
+   * The index (into {@link StreamSubscriptionRequest.subscriptions}) of the subscription yielding this bucket.
+   */
+  sub: number;
+};
 
 export interface ClientBucketDescription {
   /**

packages/sync-rules/src/BucketDescription.ts

@@ -45,4 +45,4 @@ export interface ResolvedBucket extends BucketDescription {
   inclusion_reasons: BucketInclusionReason[];
 }
 
-export type BucketInclusionReason = 'default' | { subscription: string };
+export type BucketInclusionReason = 'default' | { subscription: number };

packages/sync-rules/src/SqlSyncRules.ts

@@ -52,7 +52,7 @@ export interface RequestedStream {
    * An opaque id of the stream subscription, used to associate buckets with the stream subscriptions that have caused
    * them to be included.
    */
-  opaque_id: string;
+  opaque_id: number;
 }
 
 export interface GetQuerierOptions {