Skip to content

Commit ce35f92

Browse files
onlyjackfrostonlyjackfrost
committed
feat(wren-ui): implement dashboard cache refresh background tracker
- Added migration to create `dashboard_cache_refresh` table for tracking cache refresh jobs. - Introduced `DashboardCacheRefreshRepository` for managing cache refresh records. - Implemented `DashboardCacheBackgroundTracker` to periodically refresh dashboard caches based on scheduling. - Updated context and service layers to integrate new cache refresh functionality. - Enhanced GraphQL API to support the new caching features.
1 parent 52ffcb1 commit ce35f92

9 files changed

Lines changed: 364 additions & 5 deletions

File tree

‎wren-ui/migrations/20250423000000_create_dashboard_cache_refresh_table.js‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
exports.up = function (knex) {
2+
return knex.schema.createTable('dashboard_cache_refresh', (table) => {
3+
table.increments('id').primary();
4+
table.string('hash').notNullable();
5+
table.integer('dashboard_id').notNullable();
6+
table.integer('dashboard_item_id').notNullable();
7+
table.timestamp('started_at').notNullable();
8+
table.timestamp('finished_at');
9+
table.string('status').notNullable(); // 'success', 'failed', 'in_progress'
10+
table.text('error_message');
11+
table.timestamps(true, true);
12+
13+
// Foreign keys
14+
table
15+
.foreign('dashboard_id')
16+
.references('id')
17+
.inTable('dashboard')
18+
.onDelete('CASCADE');
19+
table
20+
.foreign('dashboard_item_id')
21+
.references('id')
22+
.inTable('dashboard_item')
23+
.onDelete('CASCADE');
24+
25+
// Indexes
26+
table.index(['dashboard_id', 'created_at']);
27+
table.index(['dashboard_item_id', 'created_at']);
28+
table.index('status');
29+
table.index('hash');
30+
});
31+
};
32+
33+
exports.down = function (knex) {
34+
return knex.schema.dropTable('dashboard_cache_refresh');
35+
};

‎wren-ui/src/apollo/server/backgrounds/dashboardCacheBackgroundTracker.ts‎

Lines changed: 194 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,194 @@
1+
import { getLogger } from '@server/utils';
2+
import {
3+
IDashboardRepository,
4+
IDashboardItemRepository,
5+
IDashboardCacheRefreshRepository,
6+
DashboardCacheRefreshStatus,
7+
} from '@server/repositories';
8+
import {
9+
IProjectService,
10+
IDeployService,
11+
IQueryService,
12+
} from '@server/services';
13+
import { CronExpressionParser } from 'cron-parser';
14+
import { v4 as uuidv4 } from 'uuid';
15+
16+
const logger = getLogger('DashboardCacheBackgroundTracker');
17+
logger.level = 'debug';
18+
19+
export class DashboardCacheBackgroundTracker {
20+
private intervalTime: number;
21+
private dashboardRepository: IDashboardRepository;
22+
private dashboardItemRepository: IDashboardItemRepository;
23+
private dashboardCacheRefreshRepository: IDashboardCacheRefreshRepository;
24+
private projectService: IProjectService;
25+
private deployService: IDeployService;
26+
private queryService: IQueryService;
27+
private runningJobs = new Set<number>();
28+
29+
constructor({
30+
dashboardRepository,
31+
dashboardItemRepository,
32+
dashboardCacheRefreshRepository,
33+
projectService,
34+
deployService,
35+
queryService,
36+
}: {
37+
dashboardRepository: IDashboardRepository;
38+
dashboardItemRepository: IDashboardItemRepository;
39+
dashboardCacheRefreshRepository: IDashboardCacheRefreshRepository;
40+
projectService: IProjectService;
41+
deployService: IDeployService;
42+
queryService: IQueryService;
43+
}) {
44+
this.dashboardRepository = dashboardRepository;
45+
this.dashboardItemRepository = dashboardItemRepository;
46+
this.dashboardCacheRefreshRepository = dashboardCacheRefreshRepository;
47+
this.projectService = projectService;
48+
this.deployService = deployService;
49+
this.queryService = queryService;
50+
this.intervalTime = 60000; // 1 minute
51+
this.start();
52+
}
53+
54+
private start(): void {
55+
logger.info('Dashboard cache background tracker started');
56+
setInterval(() => {
57+
this.checkAndRefreshCaches();
58+
}, this.intervalTime);
59+
}
60+
61+
private async checkAndRefreshCaches(): Promise<void> {
62+
try {
63+
// Get all dashboards with cache enabled
64+
const dashboards = await this.dashboardRepository.findAllBy({
65+
cacheEnabled: true,
66+
});
67+
68+
for (const dashboard of dashboards) {
69+
if (!dashboard.scheduleCron || !dashboard.nextScheduledAt) {
70+
continue;
71+
}
72+
73+
const now = new Date();
74+
const nextScheduledAt = new Date(dashboard.nextScheduledAt);
75+
76+
// Check if it's time to refresh
77+
if (now >= nextScheduledAt) {
78+
logger.info(`Start Refreshing cache for dashboard ${dashboard.id}`);
79+
await this.refreshDashboardCache(dashboard);
80+
logger.info(
81+
`Finished Refreshing cache for dashboard ${dashboard.id}`,
82+
);
83+
}
84+
}
85+
} catch (error) {
86+
logger.error(`Error checking dashboard caches: ${error.message}`);
87+
}
88+
}
89+
90+
private async refreshDashboardCache(dashboard: any): Promise<void> {
91+
if (this.runningJobs.has(dashboard.id)) {
92+
logger.debug(`Dashboard ${dashboard.id} refresh already in progress`);
93+
return;
94+
}
95+
96+
this.runningJobs.add(dashboard.id);
97+
98+
try {
99+
// Get all items for this dashboard
100+
const items = await this.dashboardItemRepository.findAllBy({
101+
dashboardId: dashboard.id,
102+
});
103+
104+
// Get project and deployment info
105+
const project = await this.projectService.getCurrentProject();
106+
const deployment = await this.deployService.getLastDeployment(project.id);
107+
const mdl = deployment.manifest;
108+
const hash = uuidv4();
109+
110+
// Refresh cache for each item
111+
for (const item of items) {
112+
try {
113+
// Create a record for this refresh job
114+
const refreshJob =
115+
await this.dashboardCacheRefreshRepository.createOne({
116+
hash,
117+
dashboardId: dashboard.id,
118+
dashboardItemId: item.id,
119+
startedAt: new Date(),
120+
finishedAt: null,
121+
status: DashboardCacheRefreshStatus.IN_PROGRESS,
122+
errorMessage: null,
123+
});
124+
125+
try {
126+
await this.queryService.preview(item.detail.sql, {
127+
project,
128+
manifest: mdl,
129+
cacheEnabled: true,
130+
refresh: true,
131+
});
132+
133+
// Update the record with success
134+
await this.dashboardCacheRefreshRepository.updateOne(
135+
refreshJob.id,
136+
{
137+
finishedAt: new Date(),
138+
status: DashboardCacheRefreshStatus.SUCCESS,
139+
},
140+
);
141+
} catch (error) {
142+
// Update the record with failure
143+
await this.dashboardCacheRefreshRepository.updateOne(
144+
refreshJob.id,
145+
{
146+
finishedAt: new Date(),
147+
status: DashboardCacheRefreshStatus.FAILED,
148+
errorMessage: error.message,
149+
},
150+
);
151+
logger.debug(
152+
`Error refreshing cache for item ${item.id}: ${error.message}`,
153+
);
154+
}
155+
} catch (error) {
156+
logger.debug(
157+
`Error creating refresh job record for item ${item.id}: ${error.message}`,
158+
);
159+
}
160+
}
161+
162+
// Calculate next scheduled time
163+
const nextScheduledAt = this.calculateNextRunTime(dashboard.scheduleCron);
164+
165+
// Update dashboard with new next scheduled time
166+
await this.dashboardRepository.updateOne(dashboard.id, {
167+
nextScheduledAt,
168+
});
169+
logger.info(
170+
`Next scheduled time for dashboard ${dashboard.id}: ${nextScheduledAt}`,
171+
);
172+
173+
logger.info(`Successfully refreshed cache for dashboard ${dashboard.id}`);
174+
} catch (error) {
175+
logger.error(
176+
`Error refreshing dashboard ${dashboard.id}: ${error.message}`,
177+
);
178+
} finally {
179+
this.runningJobs.delete(dashboard.id);
180+
}
181+
}
182+
183+
private calculateNextRunTime(cronExpression: string): Date | null {
184+
try {
185+
const interval = CronExpressionParser.parse(cronExpression, {
186+
currentDate: new Date(),
187+
});
188+
return interval.next().toDate();
189+
} catch (error) {
190+
logger.error(`Failed to parse cron expression: ${error.message}`);
191+
return null;
192+
}
193+
}
194+
}

‎wren-ui/src/apollo/server/backgrounds/index.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1-
export * from './recommend-question';
2-
export * from './chart';
31
export * from './adjustmentBackgroundTracker';
2+
export * from './textBasedAnswerBackgroundTracker';
3+
export * from './chart';
4+
export * from './recommend-question';
5+
export * from './dashboardCacheBackgroundTracker';

‎wren-ui/src/apollo/server/repositories/dashboardCacheRefreshRepository.ts‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
import { Knex } from 'knex';
2+
import { BaseRepository, IBasicRepository } from './baseRepository';
3+
4+
export enum DashboardCacheRefreshStatus {
5+
IN_PROGRESS = 'in_progress',
6+
SUCCESS = 'success',
7+
FAILED = 'failed',
8+
}
9+
10+
export interface DashboardCacheRefresh {
11+
id: number;
12+
hash: string;
13+
dashboardId: number;
14+
dashboardItemId: number;
15+
startedAt: Date;
16+
finishedAt: Date | null;
17+
status: DashboardCacheRefreshStatus;
18+
errorMessage: string | null;
19+
createdAt: Date;
20+
updatedAt: Date;
21+
}
22+
23+
export interface IDashboardCacheRefreshRepository
24+
extends IBasicRepository<DashboardCacheRefresh> {
25+
findLatestByDashboardId(
26+
dashboardId: number,
27+
): Promise<DashboardCacheRefresh | null>;
28+
findLatestByDashboardItemId(
29+
dashboardItemId: number,
30+
): Promise<DashboardCacheRefresh | null>;
31+
findInProgressByDashboardId(
32+
dashboardId: number,
33+
): Promise<DashboardCacheRefresh[]>;
34+
findInProgressByDashboardItemId(
35+
dashboardItemId: number,
36+
): Promise<DashboardCacheRefresh[]>;
37+
findByHash(hash: string): Promise<DashboardCacheRefresh | null>;
38+
}
39+
40+
export class DashboardCacheRefreshRepository
41+
extends BaseRepository<DashboardCacheRefresh>
42+
implements IDashboardCacheRefreshRepository
43+
{
44+
constructor(knexPg: Knex) {
45+
super({ knexPg, tableName: 'dashboard_cache_refresh' });
46+
}
47+
48+
public async findLatestByDashboardId(
49+
dashboardId: number,
50+
): Promise<DashboardCacheRefresh | null> {
51+
const result = await this.knex
52+
.table(this.tableName)
53+
.where({ dashboardId })
54+
.orderBy('createdAt', 'desc')
55+
.first();
56+
return result ? this.transformFromDBData(result) : null;
57+
}
58+
59+
public async findLatestByDashboardItemId(
60+
dashboardItemId: number,
61+
): Promise<DashboardCacheRefresh | null> {
62+
const result = await this.knex
63+
.table(this.tableName)
64+
.where({ dashboardItemId })
65+
.orderBy('createdAt', 'desc')
66+
.first();
67+
return result ? this.transformFromDBData(result) : null;
68+
}
69+
70+
public async findInProgressByDashboardId(
71+
dashboardId: number,
72+
): Promise<DashboardCacheRefresh[]> {
73+
const results = await this.knex
74+
.table(this.tableName)
75+
.where({
76+
dashboardId,
77+
status: DashboardCacheRefreshStatus.IN_PROGRESS,
78+
})
79+
.orderBy('createdAt', 'desc');
80+
return results.map((result) => this.transformFromDBData(result));
81+
}
82+
83+
public async findInProgressByDashboardItemId(
84+
dashboardItemId: number,
85+
): Promise<DashboardCacheRefresh[]> {
86+
const results = await this.knex
87+
.table(this.tableName)
88+
.where({
89+
dashboardItemId,
90+
status: DashboardCacheRefreshStatus.IN_PROGRESS,
91+
})
92+
.orderBy('createdAt', 'desc');
93+
return results.map((result) => this.transformFromDBData(result));
94+
}
95+
96+
public async findByHash(hash: string): Promise<DashboardCacheRefresh | null> {
97+
const result = await this.knex
98+
.table(this.tableName)
99+
.where({ hash })
100+
.first();
101+
return result ? this.transformFromDBData(result) : null;
102+
}
103+
}

‎wren-ui/src/apollo/server/repositories/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,3 +17,4 @@ export * from './dashboardItemRepository';
1717
export * from './sqlPairRepository';
1818
export * from './askingTaskRepository';
1919
export * from './instructionRepository';
20+
export * from './dashboardCacheRefreshRepository';

‎wren-ui/src/apollo/server/services/dashboardService.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -445,7 +445,8 @@ export class DashboardService implements IDashboardService {
445445
}
446446

447447
if (schedule.frequency === ScheduleFrequencyEnum.CUSTOM) {
448-
// can not less than 10 minute
448+
// can not less than 10 minutes, skip if is local
449+
if (process.env.NODE_ENV === 'development') return;
449450
const baseInterval = CronExpressionParser.parse(schedule.cron, {
450451
currentDate: new Date(),
451452
});

‎wren-ui/src/apollo/server/types/context.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
IDashboardItemRepository,
1919
ISqlPairRepository,
2020
IInstructionRepository,
21+
IDashboardCacheRefreshRepository,
2122
} from '@server/repositories';
2223
import {
2324
IQueryService,
@@ -33,6 +34,7 @@ import { ITelemetry } from '@server/telemetry/telemetry';
3334
import {
3435
ProjectRecommendQuestionBackgroundTracker,
3536
ThreadRecommendQuestionBackgroundTracker,
37+
DashboardCacheBackgroundTracker,
3638
} from '@server/backgrounds';
3739
import { ISqlPairService } from '../services/sqlPairService';
3840

@@ -71,8 +73,10 @@ export interface IContext {
7173
dashboardItemRepository: IDashboardItemRepository;
7274
sqlPairRepository: ISqlPairRepository;
7375
instructionRepository: IInstructionRepository;
76+
dashboardCacheRefreshRepository: IDashboardCacheRefreshRepository;
7477

7578
// background trackers
7679
projectRecommendQuestionBackgroundTracker: ProjectRecommendQuestionBackgroundTracker;
7780
threadRecommendQuestionBackgroundTracker: ThreadRecommendQuestionBackgroundTracker;
81+
dashboardCacheBackgroundTracker: DashboardCacheBackgroundTracker;
7882
}

0 commit comments

Comments
 (0)