Feature/sdk reporting

.changeset/heavy-pianos-grin.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core': patch
+'@powersync/service-types': patch
+---
+
+sdk reporting

.changeset/honest-sloths-smoke.md

@@ -0,0 +1,8 @@
+---
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core': patch
+'@powersync/service-types': patch
+---
+
+Added sdk reporting to storage

.changeset/smart-mugs-share.md

@@ -0,0 +1,13 @@
+---
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-module-postgres': patch
+'@powersync/service-module-mongodb': patch
+'@powersync/service-core': patch
+'@powersync/service-module-mysql': patch
+'@powersync/service-module-core': patch
+'@powersync/lib-services-framework': patch
+'@powersync/service-types': patch
+---
+
+Reporting mongo storage added to storage engine.

.github/workflows/development_packages_release.yaml

@@ -6,7 +6,7 @@ on: workflow_dispatch
 
 jobs:
   publish_packages:
-    name: Publish Devevelopment Packages
+    name: Publish Development Packages
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

libs/lib-services/src/container.ts

@@ -4,10 +4,10 @@ import { ErrorReporter } from './alerts/definitions.js';
 import { NoOpReporter } from './alerts/no-op-reporter.js';
 import { MigrationManager } from './migrations/MigrationManager.js';
 import {
-  ProbeModule,
-  TerminationHandler,
   createInMemoryProbe,
-  createTerminationHandler
+  createTerminationHandler,
+  ProbeModule,
+  TerminationHandler
 } from './signals/signals-index.js';
 
 export enum ContainerImplementation {
@@ -47,7 +47,6 @@ export type Newable<T> = new (...args: never[]) => T;
  * Identifier used to get and register implementations
  */
 export type ServiceIdentifier<T = unknown> = string | symbol | Newable<T> | Abstract<T> | ContainerImplementation;
-
 const DEFAULT_GENERATORS: ContainerImplementationDefaultGenerators = {
   [ContainerImplementation.REPORTER]: () => NoOpReporter,
   [ContainerImplementation.PROBES]: () => createInMemoryProbe(),

modules/module-mongodb-storage/package.json

@@ -31,9 +31,9 @@
     "@powersync/lib-service-mongodb": "workspace:*",
     "@powersync/lib-services-framework": "workspace:*",
     "@powersync/service-core": "workspace:*",
+    "@powersync/service-types": "workspace:*",
     "@powersync/service-jsonbig": "workspace:*",
     "@powersync/service-sync-rules": "workspace:*",
-    "@powersync/service-types": "workspace:*",
     "bson": "^6.10.3",
     "ix": "^5.0.0",
     "lru-cache": "^10.2.2",

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

@@ -0,0 +1,70 @@
+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.createConnectionReportingCollection();
+
+    await db.connection_report_events.createIndex(
+      {
+        connected_at: 1,
+        jwt_exp: 1,
+        disconnected_at: 1
+      },
+      { name: 'connection_list_index' }
+    );
+
+    await db.connection_report_events.createIndex(
+      {
+        user_id: 1
+      },
+      { name: 'connection_user_id_index' }
+    );
+    await db.connection_report_events.createIndex(
+      {
+        client_id: 1
+      },
+      { name: 'connection_client_id_index' }
+    );
+    await db.connection_report_events.createIndex(
+      {
+        sdk: 1
+      },
+      { name: 'connection_index' }
+    );
+  } 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.connection_report_events.indexExists('connection_list_index')) {
+      await db.connection_report_events.dropIndex('connection_list_index');
+    }
+    if (await db.connection_report_events.indexExists('connection_user_id_index')) {
+      await db.connection_report_events.dropIndex('connection_user_id_index');
+    }
+    if (await db.connection_report_events.indexExists('connection_client_id_index')) {
+      await db.connection_report_events.dropIndex('connection_client_id_index');
+    }
+    if (await db.connection_report_events.indexExists('connection_index')) {
+      await db.connection_report_events.dropIndex('connection_index');
+    }
+    await db.db.dropCollection('connection_report_events');
+  } finally {
+    await db.client.close();
+  }
+};

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

@@ -0,0 +1,195 @@
+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 { logger } from '@powersync/lib-services-framework';
+
+export class MongoReportStorage implements storage.ReportStorage {
+  private readonly client: mongo.MongoClient;
+  public readonly db: PowerSyncMongo;
+
+  constructor(db: PowerSyncMongo) {
+    this.client = db.client;
+    this.db = db;
+  }
+  async deleteOldConnectionData(data: event_types.DeleteOldConnectionData): Promise<void> {
+    const { date } = data;
+    const result = await this.db.connection_report_events.deleteMany({
+      connected_at: { $lt: date },
+      $or: [
+        { disconnected_at: { $exists: true } },
+        { jwt_exp: { $lt: new Date() }, disconnected_at: { $exists: false } }
+      ]
+    });
+    if (result.deletedCount > 0) {
+      logger.info(
+        `TTL from ${date.toISOString()}: ${result.deletedCount} MongoDB documents have been removed from connection_report_events.`
+      );
+    }
+  }
+
+  async getClientConnectionReports(
+    data: event_types.ClientConnectionReportRequest
+  ): Promise<event_types.ClientConnectionReport> {
+    const { start, end } = data;
+    const result = await this.db.connection_report_events
+      .aggregate<event_types.ClientConnectionReport>([
+        {
+          $match: {
+            connected_at: { $lte: end, $gte: start }
+          }
+        },
+        this.connectionsFacetPipeline(),
+        this.connectionsProjectPipeline()
+      ])
+      .toArray();
+    return result[0];
+  }
+
+  async reportClientConnection(data: event_types.ClientConnectionBucketData): Promise<void> {
+    const updateFilter = this.updateDocFilter(data.user_id, data.client_id!);
+    await this.db.connection_report_events.findOneAndUpdate(
+      updateFilter,
+      {
+        $set: data,
+        $unset: {
+          disconnected_at: ''
+        }
+      },
+      {
+        upsert: true
+      }
+    );
+  }
+  async reportClientDisconnection(data: event_types.ClientDisconnectionEventData): Promise<void> {
+    const { connected_at, user_id, client_id } = data;
+    await this.db.connection_report_events.findOneAndUpdate(
+      {
+        client_id,
+        user_id,
+        connected_at
+      },
+      {
+        $set: {
+          disconnected_at: data.disconnected_at
+        },
+        $unset: {
+          jwt_exp: ''
+        }
+      }
+    );
+  }
+  async getConnectedClients(data: event_types.ClientConnectionsRequest): Promise<event_types.ClientConnectionReport> {
+    const timeframeFilter = this.listConnectionsDateRange(data);
+    const result = await this.db.connection_report_events
+      .aggregate<event_types.ClientConnectionReport>([
+        {
+          $match: {
+            disconnected_at: { $exists: false },
+            jwt_exp: { $gt: new Date() },
+            ...timeframeFilter
+          }
+        },
+        this.connectionsFacetPipeline(),
+        this.connectionsProjectPipeline()
+      ])
+      .toArray();
+    return result[0];
+  }
+
+  async [Symbol.asyncDispose]() {
+    // No-op
+  }
+
+  private parseJsDate(date: Date) {
+    const year = date.getFullYear();
+    const month = date.getMonth();
+    const today = date.getDate();
+    const day = date.getDay();
+    return {
+      year,
+      month,
+      today,
+      day,
+      parsedDate: date
+    };
+  }
+
+  private connectionsFacetPipeline() {
+    return {
+      $facet: {
+        unique_users: [
+          {
+            $group: {
+              _id: '$user_id'
+            }
+          },
+          {
+            $count: 'count'
+          }
+        ],
+        sdk_versions_array: [
+          {
+            $group: {
+              _id: '$sdk',
+              total: { $sum: 1 },
+              client_ids: { $addToSet: '$client_id' },
+              user_ids: { $addToSet: '$user_id' }
+            }
+          },
+          {
+            $project: {
+              _id: 0,
+              sdk: '$_id',
+              users: { $size: '$user_ids' },
+              clients: { $size: '$client_ids' }
+            }
+          },
+          {
+            $sort: {
+              sdk: 1
+            }
+          }
+        ]
+      }
+    };
+  }
+
+  private connectionsProjectPipeline() {
+    return {
+      $project: {
+        users: { $ifNull: [{ $arrayElemAt: ['$unique_users.count', 0] }, 0] },
+        sdks: '$sdk_versions_array'
+      }
+    };
+  }
+
+  private updateDocFilter(userId: string, clientId: string) {
+    const { year, month, today } = this.parseJsDate(new Date());
+    const nextDay = today + 1;
+    return {
+      user_id: userId,
+      client_id: clientId,
+      connected_at: {
+        // Need to create a new date here to sett the time to 00:00:00
+        $gte: new Date(year, month, today),
+        $lt: new Date(year, month, nextDay)
+      }
+    };
+  }
+
+  private listConnectionsDateRange(data: event_types.ClientConnectionsRequest) {
+    const { range } = data;
+    if (!range) {
+      return undefined;
+    }
+    const endDate = data.range?.end ? new Date(data.range.end) : new Date();
+    const startDate = new Date(range.start);
+    return {
+      connected_at: {
+        $lte: endDate,
+        $gte: startDate
+      }
+    };
+  }
+}

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

@@ -4,8 +4,9 @@ import { POWERSYNC_VERSION, storage } from '@powersync/service-core';
 import { MongoStorageConfig } from '../../types/types.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
 import { PowerSyncMongo } from './db.js';
+import { MongoReportStorage } from '../MongoReportStorage.js';
 
-export class MongoStorageProvider implements storage.BucketStorageProvider {
+export class MongoStorageProvider implements storage.StorageProvider {
   get type() {
     return lib_mongo.MONGO_CONNECTION_TYPE;
   }
@@ -37,15 +38,19 @@ export class MongoStorageProvider implements storage.BucketStorageProvider {
     await client.connect();
 
     const database = new PowerSyncMongo(client, { database: resolvedConfig.storage.database });
-    const factory = new MongoBucketStorage(database, {
+    const syncStorageFactory = new MongoBucketStorage(database, {
       // TODO currently need the entire resolved config due to this
       slot_name_prefix: resolvedConfig.slot_name_prefix
     });
+
+    // Storage factory for reports
+    const reportStorageFactory = new MongoReportStorage(database);
     return {
-      storage: factory,
+      storage: syncStorageFactory,
+      reportStorage: reportStorageFactory,
       shutDown: async () => {
         shuttingDown = true;
-        await factory[Symbol.asyncDispose]();
+        await syncStorageFactory[Symbol.asyncDispose]();
         await client.close();
       },
       tearDown: () => {

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

@@ -0,0 +1,22 @@
+import { TestStorageOptions } from '@powersync/service-core';
+import { connectMongoForTests } from './util.js';
+import { MongoReportStorage } from '../MongoReportStorage.js';
+
+export type MongoTestStorageOptions = {
+  url: string;
+  isCI: boolean;
+};
+
+export const MongoTestReportStorageFactoryGenerator = (factoryOptions: MongoTestStorageOptions) => {
+  return async (options?: TestStorageOptions) => {
+    const db = connectMongoForTests(factoryOptions.url, factoryOptions.isCI);
+
+    await db.createConnectionReportingCollection();
+
+    if (!options?.doNotClear) {
+      await db.clear();
+    }
+
+    return new MongoReportStorage(db);
+  };
+};