98 lines
4.2 KiB
TypeScript
98 lines
4.2 KiB
TypeScript
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
|
import { Prisma } from '@prisma/client';
|
|
import { PrismaService } from '../prisma/prisma.service';
|
|
import { addDays, startOfDay, todayKey } from '../signature-analytics/analytics-date';
|
|
import { sourceFacts } from './home-source';
|
|
import { pruneHomeVersions } from './home-retention';
|
|
|
|
@Injectable()
|
|
export class HomeProjection implements OnModuleInit, OnModuleDestroy {
|
|
private timer?: NodeJS.Timeout;
|
|
private readonly logger = new Logger(HomeProjection.name);
|
|
constructor(private readonly db: PrismaService) {}
|
|
onModuleInit() {
|
|
if (
|
|
process.env.NODE_ENV === 'test' ||
|
|
process.env.HOME_DASHBOARD_ENABLED === 'false' ||
|
|
(process.env.CMPP_PROCESS_ROLE && process.env.CMPP_PROCESS_ROLE !== 'api')
|
|
)
|
|
return;
|
|
this.timer = setInterval(() => void this.run(), 10_000);
|
|
this.timer.unref();
|
|
void this.run();
|
|
}
|
|
onModuleDestroy() {
|
|
if (this.timer) clearInterval(this.timer);
|
|
}
|
|
private async run() {
|
|
try {
|
|
await this.tick();
|
|
} catch (error) {
|
|
this.logger.error('首页统计投影失败', error instanceof Error ? error.stack : String(error));
|
|
await this.db.homeProjectionState
|
|
.update({ where: { id: 'home' }, data: { lastError: '统计更新失败，等待重试' } })
|
|
.catch(() => undefined);
|
|
}
|
|
}
|
|
async tick(now = new Date()) {
|
|
return this.db.$transaction(
|
|
async (tx) => {
|
|
const [lock] = await tx.$queryRaw<
|
|
Array<{ locked: boolean }>
|
|
>`SELECT pg_try_advisory_xact_lock(17100917) AS locked`;
|
|
if (!lock.locked) return { busy: true };
|
|
const date = todayKey(now),
|
|
first = addDays(date, -3);
|
|
const state = await tx.homeProjectionState.findUniqueOrThrow({ where: { id: 'home' } });
|
|
if (state.seededDay !== date) {
|
|
await tx.$executeRaw`INSERT INTO "HomeProjectionDirty"("messageRecordId") SELECT id FROM "SmsMessageRecord"
|
|
WHERE "queuedAt">=${startOfDay(first)} AND "queuedAt"<${startOfDay(addDays(date, 1))} ON CONFLICT DO NOTHING`;
|
|
}
|
|
const work = await tx.$queryRaw<
|
|
Array<{ messageRecordId: string }>
|
|
>`SELECT "messageRecordId" FROM "HomeProjectionDirty" ORDER BY "enqueuedAt","messageRecordId" LIMIT 500 FOR UPDATE SKIP LOCKED`;
|
|
const ids = work.map((w) => w.messageRecordId),
|
|
version = state.version + 1;
|
|
const sources = await sourceFacts(tx, ids);
|
|
const prior = await tx.homeMessageFact.findMany({ where: { messageRecordId: { in: ids }, toVersion: null } });
|
|
for (const id of ids) {
|
|
const next = sources.find((s) => s.message.id === id);
|
|
const old = prior.find((p) => p.messageRecordId === id);
|
|
const day = next ? todayKey(next.message.queuedAt) : '';
|
|
const payload = next && day >= first && day <= date ? next.fact : null;
|
|
if (old && old.queuedDay === day && JSON.stringify(old.payload) === JSON.stringify(payload)) continue;
|
|
if (old)
|
|
await tx.homeMessageFact.update({
|
|
where: { messageRecordId_fromVersion: { messageRecordId: id, fromVersion: old.fromVersion } },
|
|
data: { toVersion: version },
|
|
});
|
|
if (payload)
|
|
await tx.homeMessageFact.create({
|
|
data: {
|
|
messageRecordId: id,
|
|
fromVersion: version,
|
|
queuedDay: day,
|
|
payload: payload as unknown as Prisma.InputJsonValue,
|
|
},
|
|
});
|
|
}
|
|
await tx.homeProjectionDirty.deleteMany({ where: { messageRecordId: { in: ids } } });
|
|
const remaining = await tx.homeProjectionDirty.count();
|
|
await tx.homeProjectionState.update({
|
|
where: { id: 'home' },
|
|
data: {
|
|
version,
|
|
seededDay: date,
|
|
initialized: (state.seededDay === date && state.initialized) || remaining === 0,
|
|
updatedAt: now,
|
|
lastError: null,
|
|
},
|
|
});
|
|
if (remaining === 0) await pruneHomeVersions(tx, now, addDays(first, -1), version);
|
|
return { remaining, version };
|
|
},
|
|
{ timeout: 60_000, isolationLevel: Prisma.TransactionIsolationLevel.RepeatableRead },
|
|
);
|
|
}
|
|
}
|