changed document type mongo

Author: EspressoTrip-v2

modules/module-mongodb-storage/src/migrations/db/migrations/1752661449910-sdk-reporting.ts

@@ -0,0 +1,63 @@
+import { migrations } from '@powersync/service-core';
+import * as storage from '../../../storage/storage-index.js';
+import { MongoStorageConfig } from '../../../types/types.js';
+
+export const up: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    await db.createSdkReportingCollection();
+
+    await db.sdk_report_events.createIndex(
+      {
+        connect_at: 1,
+        jwt_exp: 1,
+        disconnect_at: 1
+      },
+      { name: 'connect_at' }
+    );
+
+    await db.sdk_report_events.createIndex(
+      {
+        user_id: 1,
+        sdk: 1,
+        version: 1
+      },
+      { name: 'user_id' }
+    );
+    await db.sdk_report_events.createIndex(
+      {
+        client_id: 1
+      },
+      { name: 'client_id' }
+    );
+  } finally {
+    await db.client.close();
+  }
+};
+
+export const down: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    if (await db.write_checkpoints.indexExists('connect_at')) {
+      await db.write_checkpoints.dropIndex('connect_at');
+    }
+    if (await db.custom_write_checkpoints.indexExists('user_id')) {
+      await db.custom_write_checkpoints.dropIndex('user_id');
+    }
+    if (await db.custom_write_checkpoints.indexExists('client_id')) {
+      await db.custom_write_checkpoints.dropIndex('client_id');
+    }
+    await db.db.dropCollection('sdk_report_events');
+  } finally {
+    await db.client.close();
+  }
+};

modules/module-mongodb-storage/src/storage/MongoReportStorage.ts

@@ -2,6 +2,7 @@ import { mongo } from '@powersync/lib-service-mongodb';
 import { storage } from '@powersync/service-core';
 import { event_types } from '@powersync/service-types';
 import { PowerSyncMongo } from './implementation/db.js';
+import { SdkConnectDocument } from './implementation/models.js';
 
 export class MongoReportStorage implements storage.ReportStorageFactory {
   private readonly client: mongo.MongoClient;
@@ -16,10 +17,10 @@ export class MongoReportStorage implements storage.ReportStorageFactory {
     console.log('MongoReportStorage.scrapeSdkData', data);
   }
 
-  async reportSdkConnect(data: event_types.SdkConnectEventData): Promise<void> {
-    console.log('MongoReportStorage.reportSdkConnect', data);
+  async reportSdkConnect(data: SdkConnectDocument): Promise<void> {
+    await this.db.sdk_report_events.insertOne(data);
   }
-  async reportSdkDisconnect(data: event_types.SdkDisconnectEventData): Promise<void> {
+  async reportSdkDisconnect(data: SdkConnectDocument): Promise<void> {
     console.log('MongoReportStorage.reportSdkDisconnect', data);
   }
   async listCurrentConnections(data: event_types.PaginatedInstanceRequest): Promise<void> {

modules/module-mongodb-storage/src/storage/implementation/db.ts

@@ -12,7 +12,7 @@ import {
   CustomWriteCheckpointDocument,
   IdSequenceDocument,
   InstanceDocument,
-  SdkConnectEventDocument,
+  SdkConnectDocument,
   SourceTableDocument,
   SyncRuleDocument,
   WriteCheckpointDocument
@@ -38,7 +38,7 @@ export class PowerSyncMongo {
   readonly locks: mongo.Collection<lib_mongo.locks.Lock>;
   readonly bucket_state: mongo.Collection<BucketStateDocument>;
   readonly checkpoint_events: mongo.Collection<CheckpointEventDocument>;
-  readonly sdk_connect_events: mongo.Collection<SdkConnectEventDocument>;
+  readonly sdk_report_events: mongo.Collection<SdkConnectDocument>;
 
   readonly client: mongo.MongoClient;
   readonly db: mongo.Db;
@@ -63,7 +63,7 @@ export class PowerSyncMongo {
     this.locks = this.db.collection('locks');
     this.bucket_state = this.db.collection('bucket_state');
     this.checkpoint_events = this.db.collection('checkpoint_events');
-    this.sdk_connect_events = this.db.collection('sdk_connect_events');
+    this.sdk_report_events = this.db.collection('sdk_report_events');
   }
 
   /**
@@ -130,6 +130,20 @@ export class PowerSyncMongo {
       max: 50 // max number of documents
     });
   }
+
+  /**
+   * Only use in migrations and tests.
+   */
+  async createSdkReportingCollection() {
+    const existingCollections = await this.db
+      .listCollections({ name: 'sdk_report_events' }, { nameOnly: false })
+      .toArray();
+    const collection = existingCollections[0];
+    if (collection != null) {
+      return;
+    }
+    await this.db.createCollection('sdk_report_events');
+  }
 }
 
 export function createPowerSyncMongo(config: MongoStorageConfig, options?: lib_mongo.MongoConnectionOptions) {

modules/module-mongodb-storage/src/storage/implementation/models.ts

@@ -1,6 +1,7 @@
 import { InternalOpId, storage } from '@powersync/service-core';
 import { SqliteJsonValue } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
+import { event_types } from '@powersync/service-types';
 
 /**
  * Replica id uniquely identifying a row on the source database.
@@ -214,11 +215,4 @@ export interface InstanceDocument {
   _id: string;
 }
 
-export interface SdkConnectEventDocument {
-  version: string;
-  sdk: string;
-  user_agent: string;
-  client_id: string;
-  user_id: string;
-  jwt_exp: number;
-}
+export interface SdkConnectDocument extends event_types.SdkConnectDocument {}

packages/service-core/src/routes/endpoints/socket-route.ts

@@ -1,5 +1,6 @@
 import { ErrorCode, errors, schema } from '@powersync/lib-services-framework';
 import { RequestParameters } from '@powersync/service-sync-rules';
+import * as bson from 'bson';
 import { serialize } from 'bson';
 
 import * as sync from '../../sync/sync-index.js';
@@ -24,6 +25,7 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
       };
 
       const sdkData: event_types.SdkUserData = {
+        _id: new bson.ObjectId(),
         client_id: params.client_id,
         user_id: context.user_id!,
         user_agent: context.user_agent,

packages/service-core/src/routes/endpoints/sync-stream.ts

@@ -1,7 +1,7 @@
 import { ErrorCode, errors, router, schema } from '@powersync/lib-services-framework';
 import { RequestParameters } from '@powersync/service-sync-rules';
 import { Readable } from 'stream';
-
+import * as bson from 'bson';
 import * as sync from '../../sync/sync-index.js';
 import * as util from '../../util/util-index.js';
 
@@ -34,6 +34,7 @@ export const syncStreamed = routeDefinition({
     };
 
     const sdkData: event_types.SdkUserData = {
+      _id: new bson.ObjectId(),
       client_id: clientId,
       user_id: payload.context.user_id!,
       user_agent: userAgent as string,

packages/service-core/src/storage/ReportStorageFactory.ts

@@ -1,12 +1,8 @@
-import {
-  PaginatedInstanceRequest,
-  SdkConnectEventData,
-  SdkDisconnectEventData
-} from '@powersync/service-types/dist/events.js';
+import { PaginatedInstanceRequest, SdkConnectDocument } from '@powersync/service-types/dist/events.js';
 
 export interface ReportStorageFactory extends AsyncDisposable {
-  reportSdkConnect(data: SdkConnectEventData): Promise<void>;
-  reportSdkDisconnect(data: SdkDisconnectEventData): Promise<void>;
+  reportSdkConnect(data: SdkConnectDocument): Promise<void>;
+  reportSdkDisconnect(data: SdkConnectDocument): Promise<void>;
   listCurrentConnections(data: PaginatedInstanceRequest): Promise<void>;
   scrapeSdkData(data: PaginatedInstanceRequest): Promise<void>;
 }

packages/types/src/events.ts

@@ -1,3 +1,5 @@
+import * as bson from 'bson';
+
 export enum EmitterEngineEvents {
   SDK_CONNECT_EVENT = 'sdk-connect-event',
   SDK_DISCONNECT_EVENT = 'sdk-disconnect-event'
@@ -12,20 +14,33 @@ export type SubscribeEvents = {
 };
 
 export type SdkUserData = {
+  _id: bson.ObjectId;
   client_id?: string;
   user_id: string;
   user_agent?: string;
   jwt_exp: JwtExp;
 };
 
 export type SdkConnectEventData = {
-  connect_at: number;
+  connect_at: Date;
 } & SdkUserData;
 
 export type SdkDisconnectEventData = {
-  disconnect_at: number;
+  disconnect_at: Date;
 } & SdkUserData;
 
+export type SdkConnectDocument = {
+  _id: bson.ObjectId;
+  sdk: string;
+  version: string;
+  user_agent: string;
+  client_id: string;
+  user_id: string;
+  jwt_exp?: Date;
+  connect_at: Date;
+  disconnect_at?: Date;
+};
+
 export type PaginatedInstanceRequest = {
   app_id: string;
   org_id: string;