fix: 固定跨午夜日报刷新基准并补充验收记录
This commit is contained in:
hectorzhao
@@ -45,78 +45,85 @@ export class SignatureAnalyticsService implements OnModuleInit, OnModuleDestroy
|
||||
|
||||
/** Explicit offline backfill only; no controller exposes this write operation. Existing frozen days cannot be overwritten. */
|
||||
async generate(value: string, backfill = false) {
|
||||
const date = analyticsDate(value);
|
||||
if (!mutableDay(date) && !backfill) return { skipped: true };
|
||||
if (date >= todayKey()) throw new Error('日报只生成完整自然日');
|
||||
const startedAt = new Date();
|
||||
const date = analyticsDate(value, startedAt);
|
||||
if (!mutableDay(date, startedAt) && !backfill) return { skipped: true };
|
||||
if (date >= todayKey(startedAt)) throw new Error('日报只生成完整自然日');
|
||||
const businessDate = databaseDay(date);
|
||||
const existing = await this.db.signatureAnalyticsDay.findUnique({ where: { businessDate } });
|
||||
if (backfill && existing?.publishedGenerationId) throw new Error('历史补建不得覆盖已发布日报');
|
||||
await this.db.signatureAnalyticsDay.upsert({ where: { businessDate }, create: { businessDate }, update: {} });
|
||||
return await analyticsJob(this.db, 'daily', date, async (tx, generationId) => {
|
||||
const sourceAsOf = new Date();
|
||||
await tx.signatureAnalyticsGeneration.create({ data: { id: generationId, businessDate, sourceAsOf } });
|
||||
const quality = await new OperationsQualityQueries(tx).signatureQualityLive({ date }, true);
|
||||
const dimensions = await activityDimensions(tx, date);
|
||||
const counts = await activityCounts(tx, dimensions, date);
|
||||
const unreported = await unreportedRows(tx, date);
|
||||
// All candidate rows and manifest publication share this transaction: readers never see half a day.
|
||||
for (let i = 0; i < quality.items.length; i += 250)
|
||||
await tx.signatureQualityDaily.createMany({
|
||||
data: quality.items.slice(i, i + 250).map((row) => ({
|
||||
businessDate,
|
||||
generationId,
|
||||
signatureId: row.signatureId,
|
||||
signatureName: row.signatureName,
|
||||
tenantId: row.tenantId,
|
||||
tenantName: row.tenantName,
|
||||
applicationNames: row.applicationNames ?? '',
|
||||
total: row.total,
|
||||
payload: JSON.parse(JSON.stringify(row)) as Prisma.InputJsonValue,
|
||||
})),
|
||||
return await analyticsJob(
|
||||
this.db,
|
||||
'daily',
|
||||
date,
|
||||
async (tx, generationId) => {
|
||||
const sourceAsOf = new Date();
|
||||
await tx.signatureAnalyticsGeneration.create({ data: { id: generationId, businessDate, sourceAsOf } });
|
||||
const quality = await new OperationsQualityQueries(tx).signatureQualityLive({ date }, true);
|
||||
const dimensions = await activityDimensions(tx, date);
|
||||
const counts = await activityCounts(tx, dimensions, date);
|
||||
const unreported = await unreportedRows(tx, date);
|
||||
// All candidate rows and manifest publication share this transaction: readers never see half a day.
|
||||
for (let i = 0; i < quality.items.length; i += 250)
|
||||
await tx.signatureQualityDaily.createMany({
|
||||
data: quality.items.slice(i, i + 250).map((row) => ({
|
||||
businessDate,
|
||||
generationId,
|
||||
signatureId: row.signatureId,
|
||||
signatureName: row.signatureName,
|
||||
tenantId: row.tenantId,
|
||||
tenantName: row.tenantName,
|
||||
applicationNames: row.applicationNames ?? '',
|
||||
total: row.total,
|
||||
payload: JSON.parse(JSON.stringify(row)) as Prisma.InputJsonValue,
|
||||
})),
|
||||
});
|
||||
for (let i = 0; i < dimensions.length; i += 250)
|
||||
await tx.signatureActivityDaily.createMany({
|
||||
data: dimensions.slice(i, i + 250).map((d) => ({
|
||||
...d,
|
||||
businessDate,
|
||||
generationId,
|
||||
...(counts.get(d.dimensionKey) ?? {
|
||||
submittedAttempts: 0,
|
||||
acceptedBusinessCount: 0,
|
||||
deliveredBusinessCount: 0,
|
||||
}),
|
||||
applicability: 'applicable',
|
||||
})),
|
||||
});
|
||||
for (let i = 0; i < unreported.length; i += 250)
|
||||
await tx.unreportedSignatureDaily.createMany({
|
||||
data: unreported.slice(i, i + 250).map((r) => ({ ...r, businessDate, generationId })),
|
||||
});
|
||||
if (!backfill && !mutableDay(date)) throw new Error('日报已进入冻结区，拒绝跨日发布');
|
||||
if (backfill) {
|
||||
const current = await tx.signatureAnalyticsDay.findUniqueOrThrow({ where: { businessDate } });
|
||||
if (current.publishedGenerationId) throw new Error('已有发布版本，拒绝覆盖');
|
||||
}
|
||||
await tx.signatureAnalyticsDay.update({
|
||||
where: { businessDate },
|
||||
data: {
|
||||
publishedGenerationId: generationId,
|
||||
state: 'ready',
|
||||
error: null,
|
||||
generatedAt: new Date(),
|
||||
sourceAsOf,
|
||||
refreshFor: databaseDay(todayKey(startedAt)),
|
||||
provenance: backfill ? 'backfill-current-source' : 'daily',
|
||||
rowCounts: { quality: quality.items.length, activity: dimensions.length, unreported: unreported.length },
|
||||
},
|
||||
});
|
||||
for (let i = 0; i < dimensions.length; i += 250)
|
||||
await tx.signatureActivityDaily.createMany({
|
||||
data: dimensions.slice(i, i + 250).map((d) => ({
|
||||
...d,
|
||||
businessDate,
|
||||
generationId,
|
||||
...(counts.get(d.dimensionKey) ?? {
|
||||
submittedAttempts: 0,
|
||||
acceptedBusinessCount: 0,
|
||||
deliveredBusinessCount: 0,
|
||||
}),
|
||||
applicability: 'applicable',
|
||||
})),
|
||||
});
|
||||
for (let i = 0; i < unreported.length; i += 250)
|
||||
await tx.unreportedSignatureDaily.createMany({
|
||||
data: unreported.slice(i, i + 250).map((r) => ({ ...r, businessDate, generationId })),
|
||||
});
|
||||
if (!backfill && !mutableDay(date)) throw new Error('日报已进入冻结区，拒绝跨日发布');
|
||||
if (backfill) {
|
||||
const current = await tx.signatureAnalyticsDay.findUniqueOrThrow({ where: { businessDate } });
|
||||
if (current.publishedGenerationId) throw new Error('已有发布版本，拒绝覆盖');
|
||||
}
|
||||
await tx.signatureAnalyticsDay.update({
|
||||
where: { businessDate },
|
||||
data: {
|
||||
publishedGenerationId: generationId,
|
||||
state: 'ready',
|
||||
error: null,
|
||||
generatedAt: new Date(),
|
||||
sourceAsOf,
|
||||
refreshFor: databaseDay(todayKey()),
|
||||
provenance: backfill ? 'backfill-current-source' : 'daily',
|
||||
rowCounts: { quality: quality.items.length, activity: dimensions.length, unreported: unreported.length },
|
||||
},
|
||||
});
|
||||
return {
|
||||
date,
|
||||
generationId,
|
||||
quality: quality.items.length,
|
||||
activity: dimensions.length,
|
||||
unreported: unreported.length,
|
||||
};
|
||||
});
|
||||
return {
|
||||
date,
|
||||
generationId,
|
||||
quality: quality.items.length,
|
||||
activity: dimensions.length,
|
||||
unreported: unreported.length,
|
||||
};
|
||||
},
|
||||
startedAt,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user