fix: skip timestamp check for same user

Author: kevinkim-ogp

packages/backend/src/graphql/__tests__/mutations/create-step.itest.ts

@@ -67,6 +67,7 @@ describe('createStep mutation integration tests', async () => {
     testFlow = await testUser.$relatedQuery('flows').insertAndFetch({
       name: 'Test Flow',
       // additional flow properties as needed
+      updatedBy: owner.id,
     })
     testFlowTimestampString = String(new Date(testFlow.updatedAt).getTime())
 
@@ -448,7 +449,7 @@ describe('createStep mutation integration tests', async () => {
     expect((newStep as any).connectionId).toBeNull()
   })
 
-  describe('updatedAt validation', () => {
+  describe('updatedAt and updatedBy validation', () => {
     it('should succeed when input.flow.updatedAt matches flow.updatedAt (timestamp string)', async () => {
       const params = {
         input: {
@@ -469,7 +470,7 @@ describe('createStep mutation integration tests', async () => {
       expect(newStep.key).toBe('newStep')
     })
 
-    it('should throw when input.flow.updatedAt is different from flow.updatedAt (timestamp string)', async () => {
+    it('should succeed when input.flow.updatedAt is different from flow.updatedAt for same user', async () => {
       const futureTimestamp = (
         new Date(testFlow.updatedAt).getTime() + 1000
       ).toString()
@@ -486,23 +487,22 @@ describe('createStep mutation integration tests', async () => {
         },
       }
 
-      await expect(createStep(null, params, context)).rejects.toThrow(
-        BadUserInputError,
-      )
-      await expect(createStep(null, params, context)).rejects.toThrow(
-        REFRESH_PIPE_MESSAGE,
-      )
+      const newStep = await createStep(null, params, context)
+
+      expect(newStep).toBeDefined()
+      expect(newStep.key).toBe('newStep')
     })
 
-    it('should throw when input.flow.updatedAt is different from flow.updatedAt (unix timestamp string)', async () => {
-      const futureUnixTimestamp = Math.floor(
-        (new Date(testFlow.updatedAt).getTime() + 1000) / 1000,
+    it('should throw when input.flow.updatedAt is different from flow.updatedAt for different user', async () => {
+      const futureTimestamp = (
+        new Date(testFlow.updatedAt).getTime() + 1000
       ).toString()
+      context.currentUser = editor
       const params = {
         input: {
           flow: {
             id: testFlow.id,
-            updatedAt: futureUnixTimestamp,
+            updatedAt: futureTimestamp,
           },
           previousStep: { id: existingSteps[0].id },
           key: 'newStep',
@@ -519,10 +519,11 @@ describe('createStep mutation integration tests', async () => {
       )
     })
 
-    it('should throw when input.flow.updatedAt is older than flow.updatedAt (timestamp string)', async () => {
+    it('should throw when input.flow.updatedAt is older than flow.updatedAt for different user', async () => {
       const oldTimestamp = (
         new Date(testFlow.updatedAt).getTime() - 5000
       ).toString()
+      context.currentUser = editor
       const params = {
         input: {
           flow: {
@@ -544,7 +545,7 @@ describe('createStep mutation integration tests', async () => {
       )
     })
 
-    it('should throw when input.flow.updatedAt is an invalid timestamp string', async () => {
+    it('should not throw when input.flow.updatedAt is an invalid timestamp string', async () => {
       const params = {
         input: {
           flow: {
@@ -558,12 +559,10 @@ describe('createStep mutation integration tests', async () => {
         },
       }
 
-      await expect(createStep(null, params, context)).rejects.toThrow(
-        BadUserInputError,
-      )
-      await expect(createStep(null, params, context)).rejects.toThrow(
-        REFRESH_PIPE_MESSAGE,
-      )
+      const newStep = await createStep(null, params, context)
+
+      expect(newStep).toBeDefined()
+      expect(newStep.key).toBe('newStep')
     })
   })
 })

packages/backend/src/graphql/mutations/create-step.ts

@@ -24,7 +24,7 @@ const createStep: MutationResolvers['createStep'] = async (
       })
       .throwIfNotFound()
 
-    flow.assertNotUpdatedSince(input.flow.updatedAt)
+    flow.assertNotUpdatedSince(input.flow.updatedAt, context.currentUser.id)
 
     // if connectionId is specified, verify that the connection exists
     // and the user has the appropriate permissions to use it
@@ -84,7 +84,11 @@ const createStep: MutationResolvers['createStep'] = async (
       })
     }
 
-    const updatedFlow = await flow.patchLastUpdated({ flowId: flow.id, trx })
+    const updatedFlow = await flow.patchLastUpdated({
+      flowId: flow.id,
+      updatedBy: context.currentUser.id,
+      trx,
+    })
 
     return {
       ...step,

packages/backend/src/graphql/mutations/delete-step.ts

@@ -38,7 +38,7 @@ const deleteStep: MutationResolvers['deleteStep'] = async (
       })
 
     const flow = steps[0].flow
-    flow.assertNotUpdatedSince(input.flow.updatedAt)
+    flow.assertNotUpdatedSince(input.flow.updatedAt, context.currentUser.id)
 
     if (!steps.every((step) => step.flowId === steps[0].flowId)) {
       throw new Error('All steps to be deleted must be from the same pipe!')
@@ -114,7 +114,11 @@ const deleteStep: MutationResolvers['deleteStep'] = async (
         .patch({ position: raw(`position - ${steps.length}`) })
     }
 
-    await flow.patchLastUpdated({ flowId: flow.id, trx })
+    await flow.patchLastUpdated({
+      flowId: flow.id,
+      updatedBy: context.currentUser.id,
+      trx,
+    })
 
     return await flow
       .$query(trx)

packages/backend/src/graphql/mutations/update-step.ts

@@ -28,7 +28,10 @@ const updateStep: MutationResolvers['updateStep'] = async (
       throw new BadUserInputError('Step not found')
     }
 
-    step.flow.assertNotUpdatedSince(input.flow.updatedAt)
+    step.flow.assertNotUpdatedSince(
+      input.flow.updatedAt,
+      context.currentUser.id,
+    )
 
     if (input.connection.id) {
       // if connectionId is specified, verify that the connection exists
@@ -121,11 +124,13 @@ const updateStep: MutationResolvers['updateStep'] = async (
     // update the flow's last updated
     const updatedFlow = await step.flow.patchLastUpdated({
       flowId: step.flowId,
+      updatedBy: context.currentUser.id,
       trx,
     })
 
     // do this instead of spreading the updatedStep so that it preserves the Model instance
     updatedStep.flow.updatedAt = updatedFlow.updatedAt
+    updatedStep.flow.updatedBy = context.currentUser.id
     return updatedStep
   })
 

packages/backend/src/models/__tests__/flow.test.ts

@@ -1,9 +1,12 @@
+import { randomUUID } from 'crypto'
 import { describe, expect, it } from 'vitest'
 
 import { BadUserInputError } from '@/errors/graphql-errors/bad-user-input'
 import Flow from '@/models/flow'
 
 describe('flow model', () => {
+  const updatedBy = randomUUID()
+
   describe('assertNotUpdatedSince', () => {
     it('should not throw when timestamps match', () => {
       const flow = new Flow()
@@ -13,7 +16,9 @@ describe('flow model', () => {
 
       const clientUpdatedAt = String(updatedAt.getTime())
 
-      expect(() => flow.assertNotUpdatedSince(clientUpdatedAt)).not.toThrow()
+      expect(() =>
+        flow.assertNotUpdatedSince(clientUpdatedAt, updatedBy),
+      ).not.toThrow()
     })
 
     it('should throw BadUserInputError when timestamps mismatch', () => {
@@ -24,10 +29,10 @@ describe('flow model', () => {
 
       const mismatched = String(updatedAt.getTime() + 1)
 
-      expect(() => flow.assertNotUpdatedSince(mismatched)).toThrowError(
-        BadUserInputError,
-      )
-      expect(() => flow.assertNotUpdatedSince(mismatched)).toThrow(
+      expect(() =>
+        flow.assertNotUpdatedSince(mismatched, updatedBy),
+      ).toThrowError(BadUserInputError)
+      expect(() => flow.assertNotUpdatedSince(mismatched, updatedBy)).toThrow(
         'This Pipe has been edited by another user. Please refresh the page to see the latest changes and try again.',
       )
     })
@@ -38,10 +43,12 @@ describe('flow model', () => {
 
       flow.updatedAt = updatedAt.toISOString()
 
-      expect(() => flow.assertNotUpdatedSince('not-a-number')).toThrowError(
-        BadUserInputError,
-      )
-      expect(() => flow.assertNotUpdatedSince('not-a-number')).toThrow(
+      expect(() =>
+        flow.assertNotUpdatedSince('not-a-number', updatedBy),
+      ).toThrowError(BadUserInputError)
+      expect(() =>
+        flow.assertNotUpdatedSince('not-a-number', updatedBy),
+      ).toThrow(
         'This Pipe has been edited by another user. Please refresh the page to see the latest changes and try again.',
       )
     })

packages/backend/src/models/__tests__/step.itest.ts

@@ -5,6 +5,7 @@ import Step from '@/models/step'
 
 import Execution from '../execution'
 import Flow from '../flow'
+import User from '../user'
 
 const flowId = randomUUID()
 const stepId = randomUUID()
@@ -15,9 +16,14 @@ describe('step model', () => {
   let step: Step
 
   beforeEach(async () => {
+    const user = await User.query().insert({
+      id: randomUUID(),
+      email: '[email protected]',
+    })
     await Flow.query().insert({
       id: flowId,
       name: 'test flow',
+      userId: user.id,
     })
     step = await Step.query()
       .insert({
@@ -171,15 +177,19 @@ describe('step model', () => {
     it('should patch the flow last updated', async () => {
       const flow = await step.$relatedQuery('flow')
       const originalUpdatedAt = flow.updatedAt
-
-      await flow.patchLastUpdated({ flowId: flow.id })
+      const updatedBy = flow.userId
+      await flow.patchLastUpdated({
+        flowId: flow.id,
+        updatedBy,
+      })
 
       const updatedFlow = await step.$relatedQuery('flow')
 
       expect(updatedFlow.updatedAt).toBeDefined()
       expect(new Date(updatedFlow.updatedAt).getTime()).toBeGreaterThan(
         new Date(originalUpdatedAt).getTime(),
       )
+      expect(updatedFlow.updatedBy).toBe(updatedBy)
     })
   })
 })

packages/backend/src/models/flow.ts

@@ -50,6 +50,7 @@ class Flow extends Base {
       userId: { type: 'string', format: 'uuid' },
       remoteWebhookId: { type: 'string' },
       active: { type: 'boolean' },
+      updatedBy: { type: 'string', format: 'uuid' },
 
       config: {
         type: 'object',
@@ -251,19 +252,30 @@ class Flow extends Base {
 
   async patchLastUpdated({
     flowId,
+    updatedBy,
     trx,
   }: {
     flowId: string
+    updatedBy: string
     trx?: Transaction
   }) {
     return await this.$query(trx).patchAndFetchById(flowId, {
       updatedAt: new Date().toISOString(),
+      updatedBy,
     })
   }
 
-  assertNotUpdatedSince(clientUpdatedAt: string) {
+  assertNotUpdatedSince(clientUpdatedAt: string, updatedBy: string) {
     const inputTimestamp = Number(clientUpdatedAt)
     const flowTimestamp = new Date(this.updatedAt).getTime()
+    const flowLastUpdatedBy = this.updatedBy
+
+    // return early if the flow was last updated by the same user
+    // avoid potential issues where its a valid update by the same user
+    // but the client timestamp is not updated due to race condition or network issues
+    if (flowLastUpdatedBy === updatedBy) {
+      return true
+    }
 
     if (isNaN(inputTimestamp) || inputTimestamp !== flowTimestamp) {
       throw new BadUserInputError(