Merge pull request #3014 from omnivore-app/feature/delete-user-cron

feat: create an api for the cronjob trigger which cleans up the soft deleted users

.github/workflows/run-tests.yaml

@@ -65,8 +65,9 @@ jobs:
           yarn install --frozen-lockfile
       - name: Database Migration
         run: |
+          psql -h localhost -p ${{ job.services.postgres.ports[5432] }} -U postgres -c "CREATE USER app_user WITH ENCRYPTED PASSWORD 'app_pass';"
           yarn workspace @omnivore/db migrate
-          psql -h localhost -p ${{ job.services.postgres.ports[5432] }} -U postgres -c "CREATE USER app_user WITH ENCRYPTED PASSWORD 'app_pass';GRANT omnivore_user to app_user;"
+          psql -h localhost -p ${{ job.services.postgres.ports[5432] }} -U postgres -c "GRANT omnivore_user to app_user;"
         env:
           PG_HOST: localhost
           PG_PORT: ${{ job.services.postgres.ports[5432] }}

packages/api/src/repository/index.ts

@@ -2,6 +2,7 @@ import * as httpContext from 'express-http-context2'
 import { EntityManager, EntityTarget, Repository } from 'typeorm'
 import { appDataSource } from '../data_source'
 import { Claims } from '../resolvers/types'
+import { SetClaimsRole } from '../utils/dictionary'
 
 export const getColumns = <T>(repository: Repository<T>): (keyof T)[] => {
   return repository.metadata.columns.map(
@@ -12,8 +13,10 @@ export const getColumns = <T>(repository: Repository<T>): (keyof T)[] => {
 export const setClaims = async (
   manager: EntityManager,
   uid = '00000000-0000-0000-0000-000000000000',
-  dbRole = 'omnivore_user'
+  userRole = 'user'
 ): Promise<unknown> => {
+  const dbRole =
+    userRole === SetClaimsRole.ADMIN ? 'omnivore_admin' : 'omnivore_user'
   return manager.query('SELECT * from omnivore.set_claims($1, $2)', [
     uid,
     dbRole,

packages/api/src/routers/svc/user.ts

@@ -0,0 +1,70 @@
+/* eslint-disable @typescript-eslint/no-unsafe-member-access */
+/* eslint-disable @typescript-eslint/no-unsafe-assignment */
+import cors from 'cors'
+import express from 'express'
+import { LessThan } from 'typeorm'
+import { StatusType } from '../../entity/user'
+import { readPushSubscription } from '../../pubsub'
+import { deleteUsers } from '../../services/user'
+import { corsConfig } from '../../utils/corsConfig'
+import { logger } from '../../utils/logger'
+
+type CleanupMessage = {
+  subDays: number
+}
+
+const isCleanupMessage = (obj: any): obj is CleanupMessage =>
+  'subDays' in obj && !isNaN(obj.subDays)
+
+const getCleanupMessage = (msgStr: string): CleanupMessage => {
+  try {
+    const obj = JSON.parse(msgStr) as unknown
+    if (isCleanupMessage(obj)) {
+      return obj
+    }
+  } catch (err) {
+    console.log('error deserializing event: ', { msgStr, err })
+  }
+
+  return {
+    subDays: 0, // default to 0
+  }
+}
+
+export function userServiceRouter() {
+  const router = express.Router()
+
+  router.post('/prune', cors<express.Request>(corsConfig), async (req, res) => {
+    logger.info('prune soft deleted users')
+
+    const { message: msgStr, expired } = readPushSubscription(req)
+
+    if (!msgStr) {
+      return res.status(200).send('Bad Request')
+    }
+
+    if (expired) {
+      logger.info('discarding expired message')
+      return res.status(200).send('Expired')
+    }
+
+    const cleanupMessage = getCleanupMessage(msgStr)
+    const subTime = cleanupMessage.subDays * 1000 * 60 * 60 * 24 // convert days to milliseconds
+
+    try {
+      const result = await deleteUsers({
+        status: StatusType.Deleted,
+        updatedAt: LessThan(new Date(Date.now() - subTime)), // subDays ago
+      })
+      logger.info('prune result', result)
+
+      return res.sendStatus(200)
+    } catch (error) {
+      logger.error('error prune users', error)
+
+      return res.sendStatus(500)
+    }
+  })
+
+  return router
+}

packages/api/src/server.ts

@@ -31,6 +31,7 @@ import { newsletterServiceRouter } from './routers/svc/newsletters'
 // import { remindersServiceRouter } from './routers/svc/reminders'
 import { rssFeedRouter } from './routers/svc/rss_feed'
 import { uploadServiceRouter } from './routers/svc/upload'
+import { userServiceRouter } from './routers/svc/user'
 import { webhooksServiceRouter } from './routers/svc/webhooks'
 import { textToSpeechRouter } from './routers/text_to_speech'
 import { userRouter } from './routers/user_router'
@@ -121,6 +122,7 @@ export const createApp = (): {
   app.use('/svc/pubsub/webhooks', webhooksServiceRouter())
   app.use('/svc/pubsub/integrations', integrationsServiceRouter())
   app.use('/svc/pubsub/rss-feed', rssFeedRouter())
+  app.use('/svc/pubsub/user', userServiceRouter())
   // app.use('/svc/reminders', remindersServiceRouter())
   app.use('/svc/email-attachment', emailAttachmentRouter())
 

packages/api/src/services/user.ts

@@ -1,6 +1,8 @@
+import { DeepPartial, FindOptionsWhere, In } from 'typeorm'
 import { StatusType, User } from '../entity/user'
 import { authTrx } from '../repository'
 import { userRepository } from '../repository/user'
+import { SetClaimsRole } from '../utils/dictionary'
 
 export const deleteUser = async (userId: string) => {
   await authTrx(
@@ -20,6 +22,28 @@ export const updateUser = async (userId: string, update: Partial<User>) => {
   )
 }
 
-export const findUser = async (id: string): Promise<User | null> => {
+export const findActiveUser = async (id: string): Promise<User | null> => {
   return userRepository.findOneBy({ id, status: StatusType.Active })
 }
+
+export const findUsersById = async (ids: string[]): Promise<User[]> => {
+  return userRepository.findBy({ id: In(ids) })
+}
+
+export const deleteUsers = async (criteria: FindOptionsWhere<User>) => {
+  return authTrx(
+    async (t) => t.getRepository(User).delete(criteria),
+    undefined,
+    undefined,
+    SetClaimsRole.ADMIN
+  )
+}
+
+export const createUsers = async (users: DeepPartial<User>[]) => {
+  return authTrx(
+    async (t) => t.getRepository(User).save(users),
+    undefined,
+    undefined,
+    SetClaimsRole.ADMIN
+  )
+}

packages/api/test/resolvers/user.test.ts

@@ -6,7 +6,7 @@ import {
   UpdateUserProfileErrorCode,
 } from '../../src/generated/graphql'
 import { findProfile } from '../../src/services/profile'
-import { deleteUser, findUser } from '../../src/services/user'
+import { deleteUser, findActiveUser } from '../../src/services/user'
 import { hashPassword } from '../../src/utils/auth'
 import { createTestUser } from '../db'
 import { generateFakeUuid, graphqlRequest, request } from '../util'
@@ -98,7 +98,7 @@ describe('User API', () => {
 
       it('updates user and responds with status code 200', async () => {
         const response = await graphqlRequest(query, authToken).expect(200)
-        const user = await findUser(response.body.data.updateUser.user.id)
+        const user = await findActiveUser(response.body.data.updateUser.user.id)
         expect(user?.name).to.eql(name)
       })
     })

packages/api/test/routers/user.test.ts

@@ -0,0 +1,65 @@
+import { expect } from 'chai'
+import 'mocha'
+import { In } from 'typeorm'
+import { StatusType } from '../../src/entity/user'
+import {
+  createUsers,
+  deleteUsers,
+  findUsersById,
+} from '../../src/services/user'
+import { request } from '../util'
+
+describe('User Service Router', () => {
+  const token = process.env.PUBSUB_VERIFICATION_TOKEN || ''
+
+  describe('prune', () => {
+    let toDeleteUserIds: string[] = []
+
+    before(async () => {
+      // create test users
+      const users = await createUsers([
+        {
+          name: 'user_1',
+          email: 'user_1@omnivore.app',
+          status: StatusType.Deleted,
+          updatedAt: new Date(Date.now() - 1000 * 60 * 60 * 24 * 2), // 2 days ago
+          source: 'GOOGLE',
+          sourceUserId: '123',
+        },
+        {
+          name: 'user_2',
+          email: 'user_2@omnivore.app',
+          status: StatusType.Deleted,
+          updatedAt: new Date(Date.now() - 1000 * 60 * 60 * 24 * 2), // 2 days ago
+          source: 'GOOGLE',
+          sourceUserId: '456',
+        },
+      ])
+      toDeleteUserIds = users.map((u) => u.id)
+    })
+
+    after(async () => {
+      // delete test users
+      await deleteUsers({ id: In(toDeleteUserIds) })
+    })
+
+    it('prunes soft deleted users a day ago', async () => {
+      const data = {
+        message: {
+          data: Buffer.from(
+            JSON.stringify({ subDays: 1 }) // 1 day ago
+          ).toString('base64'),
+          publishTime: new Date().toISOString(),
+        },
+      }
+
+      await request
+        .post('/svc/pubsub/user/prune?token=' + token)
+        .send(data)
+        .expect(200)
+
+      const deletedUsers = await findUsersById(toDeleteUserIds)
+      expect(deletedUsers.length).to.equal(0)
+    })
+  })
+})

packages/db/migrations/0141.do.add_index_for_cleanup_to_user.sql

@@ -0,0 +1,9 @@
+-- Type: DO
+-- Name: add_index_for_cleanup_to_user
+-- Description: Add index of status and updated_at to omnivore.user table for cleanup of deleted users
+
+BEGIN;
+
+CREATE INDEX IF NOT EXISTS user_status_updated_at_idx ON omnivore.user (status, updated_at);
+
+COMMIT;

packages/db/migrations/0141.undo.add_index_for_cleanup_to_user.sql

@@ -0,0 +1,9 @@
+-- Type: UNDO
+-- Name: add_index_for_cleanup_to_user
+-- Description: Add index of status and updated_at to omnivore.user table for cleanup of deleted users
+
+BEGIN;
+
+DROP INDEX IF EXISTS user_status_updated_at_idx;
+
+COMMIT;

packages/db/migrations/0142.do.create_omnivore_admin_role.sql

@@ -0,0 +1,19 @@
+-- Type: DO
+-- Name: create_omnivore_admin_role
+-- Description: Create omnivore_admin role with admin permissions
+
+BEGIN;
+
+CREATE ROLE omnivore_admin;
+
+GRANT omnivore_admin TO app_user;
+
+GRANT ALL PRIVILEGES ON SCHEMA omnivore TO omnivore_admin;
+GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA omnivore TO omnivore_admin;
+
+CREATE POLICY user_admin_policy on omnivore.user
+    FOR ALL
+    TO omnivore_admin
+    USING (true);
+
+COMMIT;

packages/db/migrations/0142.undo.create_omnivore_admin_role.sql

@@ -0,0 +1,16 @@
+-- Type: UNDO
+-- Name: create_omnivore_admin_role
+-- Description: Create omnivore_admin role with admin permissions
+
+BEGIN;
+
+DROP POLICY user_admin_policy ON omnivore.user;
+
+REVOKE ALL PRIVILEGES on omnivore.user from omnivore_admin;
+REVOKE ALL PRIVILEGES on SCHEMA omnivore from omnivore_admin;
+
+DROP OWNED BY omnivore_admin;
+
+DROP ROLE IF EXISTS omnivore_admin;
+
+COMMIT;