Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1156 from botpress/ya-data-retention
feat(storage): implement expiration date for user attributes
- Loading branch information
Showing
12 changed files
with
305 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| import { Table } from 'core/database/interfaces' | ||
|
|
||
| export class DataRetentionTable extends Table { | ||
| name: string = 'data_retention' | ||
|
|
||
| async bootstrap() { | ||
| let created = false | ||
|
|
||
| await this.knex.createTableIfNotExists(this.name, table => { | ||
| table.text('channel').notNullable() | ||
| table.text('user_id').notNullable() | ||
| table.text('field_path').notNullable() | ||
| table.timestamp('expiry_date').notNullable() | ||
| table.timestamp('created_on').notNullable() | ||
| created = true | ||
| }) | ||
| return created | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| import { Logger } from 'botpress/sdk' | ||
| import { UserRepository } from 'core/repositories' | ||
| import { inject, injectable, tagged } from 'inversify' | ||
| import _ from 'lodash' | ||
| import { Memoize } from 'lodash-decorators' | ||
|
|
||
| import { BotpressConfig } from '../../config/botpress.config' | ||
| import { ConfigProvider } from '../../config/config-loader' | ||
| import { TYPES } from '../../types' | ||
| import { Janitor } from '../janitor' | ||
|
|
||
| import { DataRetentionService } from './service' | ||
|
|
||
| @injectable() | ||
| export class DataRetentionJanitor extends Janitor { | ||
| private BATCH_SIZE = 250 | ||
|
|
||
| constructor( | ||
| @inject(TYPES.Logger) | ||
| @tagged('name', 'RetentionJanitor') | ||
| protected logger: Logger, | ||
| @inject(TYPES.ConfigProvider) private configProvider: ConfigProvider, | ||
| @inject(TYPES.DataRetentionService) private dataRetentionService: DataRetentionService, | ||
| @inject(TYPES.UserRepository) private userRepo: UserRepository | ||
| ) { | ||
| super(logger) | ||
| } | ||
|
|
||
| @Memoize() | ||
| private async getBotpressConfig(): Promise<BotpressConfig> { | ||
| return this.configProvider.getBotpressConfig() | ||
| } | ||
|
|
||
| protected async getInterval(): Promise<string> { | ||
| const config = await this.getBotpressConfig() | ||
| return (config.dataRetention && config.dataRetention.janitorInterval) || '15m' | ||
| } | ||
|
|
||
| protected async runTask(): Promise<void> { | ||
| let expired = await this.dataRetentionService.getExpired(this.BATCH_SIZE) | ||
|
|
||
| while (expired.length > 0) { | ||
| await Promise.mapSeries(expired, async ({ channel, user_id, field_path }) => { | ||
| const { result: user } = await this.userRepo.getOrCreate(channel, user_id) | ||
|
|
||
| await this.userRepo.updateAttributes(channel, user.id, _.omit(user.attributes, field_path)) | ||
| await this.dataRetentionService.delete(channel, user_id, field_path) | ||
| }) | ||
|
|
||
| if (expired.length >= this.BATCH_SIZE) { | ||
| expired = await this.dataRetentionService.getExpired(this.BATCH_SIZE) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| import { RetentionPolicy } from 'core/config/botpress.config' | ||
| import { ConfigProvider } from 'core/config/config-loader' | ||
| import Database from 'core/database' | ||
| import { TYPES } from 'core/types' | ||
| import diff from 'deep-diff' | ||
| import { inject, injectable } from 'inversify' | ||
| import _ from 'lodash' | ||
| import moment from 'moment' | ||
| import ms from 'ms' | ||
|
|
||
| import { getPaths } from './util' | ||
|
|
||
| export interface ExpiredData { | ||
| channel: string | ||
| user_id: string | ||
| field_path: string | ||
| } | ||
|
|
||
| @injectable() | ||
| export class DataRetentionService { | ||
| private readonly tableName = 'data_retention' | ||
| private policies: RetentionPolicy | undefined | ||
| private DELETED_ATTR = 'D' | ||
|
|
||
| constructor( | ||
| @inject(TYPES.ConfigProvider) private configProvider: ConfigProvider, | ||
| @inject(TYPES.Database) private database: Database | ||
| ) {} | ||
|
|
||
| async initialize(): Promise<void> { | ||
| const config = await this.configProvider.getBotpressConfig() | ||
| this.policies = config.dataRetention && config.dataRetention.policies | ||
| } | ||
|
|
||
| hasPolicy(): boolean { | ||
| return !_.isEmpty(this.policies) | ||
| } | ||
|
|
||
| async updateExpirationForChangedFields( | ||
| channel: string, | ||
| user_id: string, | ||
| beforeAttributes: any, | ||
| afterAttributes: any | ||
| ) { | ||
| const differences = diff(getPaths(beforeAttributes), getPaths(afterAttributes)) | ||
| if (!differences || !this.policies) { | ||
| return | ||
| } | ||
|
|
||
| const changedPaths = _.flatten(differences.filter(diff => diff.kind != this.DELETED_ATTR).map(diff => diff.path)) | ||
| if (!changedPaths.length) { | ||
| return | ||
| } | ||
|
|
||
| for (const field in this.policies) { | ||
| if (changedPaths.indexOf(field) > -1) { | ||
| const expiry = moment() | ||
| .add(ms(this.policies[field]), 'ms') | ||
| .toDate() | ||
|
|
||
| if (await this.get(channel, user_id, field)) { | ||
| await this.update(channel, user_id, field, expiry) | ||
| } else { | ||
| await this.insert(channel, user_id, field, expiry) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private async get(channel: string, user_id: string, field_path: string) { | ||
| return await this.database | ||
| .knex(this.tableName) | ||
| .where({ channel, user_id, field_path }) | ||
| .limit(1) | ||
| .select('expiry_date') | ||
| .first() | ||
| } | ||
|
|
||
| private async insert(channel: string, user_id: string, field_path: string, expiry_date: Date) { | ||
| await this.database.knex(this.tableName).insert({ | ||
| channel, | ||
| user_id, | ||
| field_path, | ||
| expiry_date: this.database.knex.date.format(expiry_date), | ||
| created_on: this.database.knex.date.now() | ||
| }) | ||
| } | ||
|
|
||
| private async update(channel: string, user_id: string, field_path: string, expiry_date: Date) { | ||
| await this.database | ||
| .knex(this.tableName) | ||
| .update({ expiry_date: this.database.knex.date.format(expiry_date) }) | ||
| .where({ channel, user_id, field_path }) | ||
| } | ||
|
|
||
| async delete(channel: string, user_id: string, field_path: string): Promise<void> { | ||
| await this.database | ||
| .knex(this.tableName) | ||
| .where({ channel, user_id, field_path }) | ||
| .del() | ||
| } | ||
|
|
||
| async getExpired(batchSize): Promise<ExpiredData[]> { | ||
| return await this.database | ||
| .knex(this.tableName) | ||
| .andWhere(this.database.knex.date.isBefore('expiry_date', new Date())) | ||
| .select('channel', 'user_id', 'field_path') | ||
| .limit(batchSize) | ||
| } | ||
| } |
Oops, something went wrong.