Merge pull request #1152 from jetstreamapp/feat/record-sync

Feat/record-sync

.github/workflows/ci.yml

@@ -6,6 +6,7 @@ on:
   pull_request:
 
 env:
+  NODE_OPTIONS: '--max_old_space_size=4096'
   LOG_LEVEL: warn
   CONTENTFUL_HOST: cdn.contentful.com
   CONTENTFUL_SPACE: wuv9tl5d77ll
@@ -21,6 +22,7 @@ jobs:
   build-and-test:
     runs-on: ubuntu-latest
     timeout-minutes: 60
+
     steps:
       - uses: actions/checkout@v4
         name: Checkout [master]
@@ -42,14 +44,10 @@ jobs:
       - name: install dependencies
         run: yarn install --frozen-lockfile
 
-      - name: Test all affected projects
-        env:
-          NODE_OPTIONS: '--max_old_space_size=4096'
+      - name: Test all projects
         run: yarn test:all
 
       - name: Build
-        env:
-          NODE_OPTIONS: '--max_old_space_size=4096'
         run: yarn build:ci
 
       - name: Uploading artifacts
@@ -58,9 +56,53 @@ jobs:
           name: dist-artifacts
           path: dist
 
-  # e2e tests only runs if build passes, since it uses production build to run tests
+  test-cron:
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    env:
+      PRISMA_TEST_DB_URI: postgres://postgres:postgres@localhost:5432/postgres
+      JETSTREAM_POSTGRES_DBURI: postgres://postgres:postgres@localhost:5432/postgres
+
+    services:
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: postgres
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+
+    steps:
+      - uses: actions/checkout@v4
+        name: Checkout [master]
+        with:
+          fetch-depth: 0
+
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '20'
+          cache: 'yarn'
+
+      - name: install dependencies
+        run: yarn install --frozen-lockfile
+
+      # Run database migrations
+      - name: Generate database
+        run: yarn db:generate
+
+      - name: Run database migration
+        run: yarn db:migrate
+
+      - name: Test cron-tasks
+        run: yarn test:cron
+
   e2e:
-    needs: build-and-test
     runs-on: ubuntu-latest
     env:
       NX_CLOUD_DISTRIBUTED_EXECUTION: false
@@ -116,11 +158,8 @@ jobs:
       - name: install dependencies
         run: yarn install --frozen-lockfile
 
-      - name: Download artifacts from build
-        uses: actions/download-artifact@v4
-        with:
-          name: dist-artifacts
-          path: dist
+      - name: Build
+        run: yarn build:ci
 
       - name: Install Playwright dependencies
         run: npx playwright install --with-deps

Dockerfile.e2e

@@ -19,7 +19,9 @@ COPY ./prisma ./prisma/
 RUN yarn
 
 # Install other dependencies that were not calculated by nx, but are required
-RUN yarn add dotenv prisma@^3.13.0
+# Install matching version of prisma by extracting @prisma/client version from package.json,
+# and stripping the caret ("^") if present.
+RUN yarn add prisma@$(node -p "require('./package.json')['dependencies']['@prisma/client'].replace('^','')")
 
 # Generate prisma client - ensure that there are no OS differences
 RUN npx prisma generate

apps/api/src/app/controllers/data-sync.controller.ts

@@ -0,0 +1,91 @@
+import { ensureBoolean, REGEX } from '@jetstream/shared/utils';
+import { SyncRecordOperationSchema } from '@jetstream/types';
+import { parseISO } from 'date-fns';
+import { clamp } from 'lodash';
+import { z } from 'zod';
+import * as userSyncDbService from '../db/data-sync.db';
+import { emitRecordSyncEventsToOtherClients, SyncEvent } from '../services/data-sync-broadcast.service';
+import { sendJson } from '../utils/response.handlers';
+import { createRoute } from '../utils/route.utils';
+
+export const routeDefinition = {
+  pull: {
+    controllerFn: () => pull,
+    validators: {
+      query: z.object({
+        updatedAt: z
+          .string()
+          .regex(REGEX.ISO_DATE)
+          .nullish()
+          .transform((val) => (val ? parseISO(val) : null)),
+        limit: z.coerce
+          .number()
+          .int()
+          .optional()
+          .default(userSyncDbService.MAX_PULL)
+          .transform((val) => clamp(val, userSyncDbService.MIN_PULL, userSyncDbService.MAX_PULL)),
+        /**
+         * Used for pagination, if there are more records, this is the last key of the previous page
+         */
+        lastKey: z.string().nullish(),
+      }),
+      hasSourceOrg: false,
+    },
+  },
+  push: {
+    controllerFn: () => push,
+    validators: {
+      query: z.object({
+        clientId: z.string().uuid(),
+        updatedAt: z
+          .string()
+          .regex(REGEX.ISO_DATE)
+          .nullish()
+          .transform((val) => (val ? parseISO(val) : null)),
+        includeAllIfUpdatedAtNull: z
+          .union([z.enum(['true', 'false']), z.boolean()])
+          .optional()
+          .default(false)
+          .transform(ensureBoolean),
+      }),
+      body: SyncRecordOperationSchema.array().max(userSyncDbService.MAX_SYNC),
+      hasSourceOrg: false,
+    },
+  },
+};
+
+/**
+ * Pull changes from server
+ */
+const pull = createRoute(routeDefinition.pull.validators, async ({ user, query }, req, res) => {
+  const { lastKey, updatedAt, limit } = query;
+  const response = await userSyncDbService.findByUpdatedAt({
+    userId: user.id,
+    lastKey,
+    updatedAt,
+    limit,
+  });
+  sendJson(res, response);
+});
+
+/**
+ * Push changes to server and emit to any other clients the user has active
+ */
+const push = createRoute(routeDefinition.push.validators, async ({ user, body: records, query }, req, res) => {
+  const response = await userSyncDbService.syncRecordChanges({
+    updatedAt: query.updatedAt,
+    userId: user.id,
+    records,
+    includeAllIfUpdatedAtNull: query.includeAllIfUpdatedAtNull,
+  });
+
+  const syncEvent: SyncEvent = {
+    clientId: query.clientId,
+    data: { keys: response.records.map(({ key }) => key) },
+    userId: user.id,
+  };
+
+  emitRecordSyncEventsToOtherClients(req.session.id, syncEvent);
+
+  sendJson(res, response);
+});