Skip to content

Commit c8c0500

Browse files
feat: scheduled per-org analytics report generation with signed download URLs
1 parent c6ab67b commit c8c0500

7 files changed

Lines changed: 248 additions & 0 deletions

File tree

src/app-bootstrap.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ export function bootstrapApp(options: BootstrapOptions = {}) {
7171
app.use('/api/organizations', orgVaultsRouter)
7272
app.use('/api/organizations', orgAnalyticsRouter)
7373
app.use('/api/organizations', orgMembersRouter)
74+
app.use('/api/orgs', orgAnalyticsRouter)
7475
app.use('/api/orgs', orgMembersRouter)
7576
app.use('/api/orgs', notificationPreferencesRouter)
7677
app.use('/api/organizations/:orgId/graphql', graphqlRouter)

src/jobs/handlers.ts

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,13 @@ import {
1515
import { cleanupExpiredSessions } from '../services/session.js'
1616
import { relayOutboxBatch } from '../services/outboxRelay.js'
1717
import { runReindexBatches } from '../services/evidenceReindex.js'
18+
import { renderOrgAnalyticsSnapshot } from '../services/analytics.service.js'
19+
import {
20+
saveOrgReport,
21+
getAllOrgIds,
22+
checkAndIncrementReportQuota,
23+
} from '../services/analyticsReports.js'
24+
import { resolveS3Config, uploadToS3 } from '../services/exportS3.js'
1825
import db from '../db/index.js'
1926

2027
type JobHandlerRegistry = {
@@ -108,6 +115,44 @@ export const createDefaultJobHandlers = (
108115
`scope=${payload.scope} entity=${entity} reason=${reason} attempt=${context.attempt}`,
109116
)
110117
},
118+
'analytics.report.generate': async (payload, context) => {
119+
const s3Config = resolveS3Config()
120+
const orgIds = payload.orgIds ?? getAllOrgIds()
121+
let generated = 0
122+
let skipped = 0
123+
124+
for (const orgId of orgIds) {
125+
if (!checkAndIncrementReportQuota(orgId)) {
126+
logJob('analytics.report.generate', `quota_exceeded orgId=${orgId}`)
127+
skipped++
128+
continue
129+
}
130+
131+
try {
132+
const snapshot = renderOrgAnalyticsSnapshot(orgId)
133+
const json = JSON.stringify(snapshot)
134+
const buf = Buffer.from(json, 'utf8')
135+
const ts = new Date().toISOString().replace(/[:.]/g, '-')
136+
const key = `analytics-reports/${orgId}/${ts}.json`
137+
138+
if (s3Config) {
139+
await uploadToS3(s3Config, key, buf, 'application/json')
140+
saveOrgReport({ orgId, s3Key: key, snapshotAt: snapshot.snapshotAt, sizeBytes: buf.byteLength })
141+
} else {
142+
saveOrgReport({ orgId, localBuffer: buf, snapshotAt: snapshot.snapshotAt, sizeBytes: buf.byteLength })
143+
}
144+
generated++
145+
} catch (err) {
146+
const msg = err instanceof Error ? err.message : String(err)
147+
logJob('analytics.report.generate', `error orgId=${orgId}: ${msg}`)
148+
}
149+
}
150+
151+
logJob(
152+
'analytics.report.generate',
153+
`generated=${generated} skipped=${skipped} attempt=${context.attempt} job_id=${context.jobId}`,
154+
)
155+
},
111156
'export.generate': async (payload, context) => {
112157
await processExportJob(payload.exportJobId, undefined, context.attempt)
113158
logJob(

src/jobs/system.ts

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -194,6 +194,7 @@ export class BackgroundJobSystem {
194194
this.queue.registerHandler('deadline.check', handlers['deadline.check'])
195195
this.queue.registerHandler('oracle.call', handlers['oracle.call'])
196196
this.queue.registerHandler('analytics.recompute', handlers['analytics.recompute'])
197+
this.queue.registerHandler('analytics.report.generate', handlers['analytics.report.generate'])
197198
this.queue.registerHandler('export.generate', handlers['export.generate'])
198199
this.queue.registerHandler('sessions.cleanup', handlers['sessions.cleanup'])
199200
this.queue.registerHandler('outbox.relay', handlers['outbox.relay'])
@@ -301,6 +302,10 @@ export class BackgroundJobSystem {
301302
process.env.SAVED_SEARCH_EVAL_INTERVAL_MS,
302303
15 * 60_000, // 15 minutes
303304
)
305+
const analyticsReportIntervalMs = parsePositiveInteger(
306+
process.env.ANALYTICS_REPORT_INTERVAL_MS,
307+
24 * 60 * 60_000, // 24 hours
308+
)
304309

305310
this.schedulerRegistry.registerJob({
306311
name: 'deadline.check',
@@ -362,5 +367,14 @@ export class BackgroundJobSystem {
362367
this.enqueue('saved-search.evaluate', {})
363368
},
364369
})
370+
371+
this.schedulerRegistry.registerJob({
372+
name: 'analytics.report.generate',
373+
intervalMs: analyticsReportIntervalMs,
374+
immediate: false,
375+
execute: () => {
376+
this.enqueue('analytics.report.generate', {})
377+
},
378+
})
365379
}
366380
}

src/jobs/types.ts

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ export const JOB_TYPES = [
66
'milestone.reminders.deferred',
77
'oracle.call',
88
'analytics.recompute',
9+
'analytics.report.generate',
910
'export.generate',
1011
'vault.reconcile',
1112
'sessions.cleanup',
@@ -54,6 +55,10 @@ export interface AnalyticsRecomputeJobPayload {
5455
reason?: string
5556
}
5657

58+
export interface AnalyticsReportGenerateJobPayload {
59+
orgIds?: string[] // if omitted, runs for all known orgs
60+
}
61+
5762
export interface ExportGenerateJobPayload {
5863
exportJobId: string
5964
}
@@ -88,6 +93,7 @@ export interface JobPayloadByType {
8893
'milestone.reminders.deferred': MilestoneRemindersDeferredJobPayload
8994
'oracle.call': OracleCallJobPayload
9095
'analytics.recompute': AnalyticsRecomputeJobPayload
96+
'analytics.report.generate': AnalyticsReportGenerateJobPayload
9197
'export.generate': ExportGenerateJobPayload
9298
'vault.reconcile': VaultReconcileJobPayload
9399
'sessions.cleanup': SessionsCleanupJobPayload
@@ -176,6 +182,8 @@ export const isPayloadForJobType = (
176182
isOptionalString(payload.entityId) &&
177183
isOptionalString(payload.reason)
178184
)
185+
case 'analytics.report.generate':
186+
return payload.orgIds === undefined || Array.isArray(payload.orgIds)
179187
case 'export.generate':
180188
return isNonEmptyString(payload.exportJobId)
181189
case 'vault.reconcile':

src/routes/orgAnalytics.ts

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,9 @@ import { authenticate } from '../middleware/auth.js'
44
import { requireOrgAccess, requireOrgRole } from '../middleware/orgAuth.js'
55
import { vaults, Vault } from './vaults.js'
66
import { getTeamRollup } from '../services/team.js'
7+
import { getOrgReports } from '../services/analyticsReports.js'
8+
import { resolveS3Config, getExportSignedUrl } from '../services/exportS3.js'
9+
import { parsePaginationParams, paginateArray } from '../utils/pagination.js'
710

811
export const orgAnalyticsRouter = Router()
912

@@ -78,3 +81,53 @@ orgAnalyticsRouter.get(
7881
}
7982
}
8083
)
84+
85+
/**
86+
* GET /api/orgs/:orgId/analytics/reports
87+
*
88+
* List all point-in-time analytics reports for the org. Each entry includes a
89+
* signed, expiring download URL when S3 is configured, or a local download path
90+
* otherwise. Paginated; newest reports appear first. Access is restricted to
91+
* owner/admin members of the org (strict per-org isolation).
92+
*/
93+
orgAnalyticsRouter.get(
94+
'/:orgId/analytics/reports',
95+
authenticate,
96+
requireOrgAccess('owner', 'admin'),
97+
orgAnalyticsRateLimiter,
98+
async (req: Request, res: Response) => {
99+
const { orgId } = req.params
100+
const pagination = parsePaginationParams(req)
101+
const s3Config = resolveS3Config()
102+
103+
const allReports = getOrgReports(orgId)
104+
const paginated = paginateArray(allReports, pagination)
105+
106+
const items = await Promise.all(
107+
paginated.data.map(async (r) => {
108+
let downloadUrl: string | null = null
109+
if (r.s3Key && s3Config) {
110+
try {
111+
downloadUrl = await getExportSignedUrl(s3Config, r.s3Key)
112+
} catch {
113+
// signed URL generation failure is non-fatal; return null
114+
}
115+
} else if (r.localBuffer) {
116+
// In non-S3 mode expose a local download path so clients can retrieve it
117+
downloadUrl = `/api/orgs/${orgId}/analytics/reports/${r.id}/download`
118+
}
119+
120+
return {
121+
id: r.id,
122+
orgId: r.orgId,
123+
snapshotAt: r.snapshotAt,
124+
createdAt: r.createdAt,
125+
sizeBytes: r.sizeBytes,
126+
downloadUrl,
127+
}
128+
}),
129+
)
130+
131+
res.json({ ...paginated, data: items })
132+
}
133+
)

src/services/analytics.service.ts

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -191,4 +191,19 @@ export async function getCapitalAnalytics(period: string = 'all'): Promise<{
191191
export async function updateAnalyticsSummary(orgId?: string): Promise<void> {
192192
await dbUpdateSummary()
193193
await invalidate('analytics:overall', orgId)
194+
}
195+
196+
/**
197+
* Render a point-in-time analytics snapshot for a single org.
198+
* Pulls vault IDs from the in-memory vaults store so it works without a DB.
199+
*/
200+
export function renderOrgAnalyticsSnapshot(orgId: string): OrgVaultAnalytics & { orgId: string; snapshotAt: string } {
201+
// Import lazily to avoid circular deps and to stay hermetic in tests
202+
// eslint-disable-next-line @typescript-eslint/no-var-requires
203+
const { vaults } = require('../routes/vaults.js') as { vaults: Array<{ id: string; orgId?: string }> }
204+
const orgVaultIds = vaults
205+
.filter((v) => v.orgId === orgId)
206+
.map((v) => v.id)
207+
const analytics = getOrgAnalyticsBatched(orgVaultIds)
208+
return { ...analytics, orgId, snapshotAt: utcNow() }
194209
}

src/services/analyticsReports.ts

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
/**
2+
* In-memory store for per-org analytics report records.
3+
*
4+
* Each report entry holds a reference to the S3 key (or a local JSON buffer
5+
* when S3 is not configured) for the rendered snapshot. Signed download URLs
6+
* are generated on demand from the stored S3 key.
7+
*
8+
* Retention: reports older than ANALYTICS_REPORT_RETENTION_DAYS are pruned
9+
* automatically whenever a new report is saved for that org.
10+
*/
11+
12+
const DEFAULT_RETENTION_DAYS =
13+
Number.parseInt(process.env.ANALYTICS_REPORT_RETENTION_DAYS ?? '30', 10) || 30
14+
15+
const DEFAULT_REPORT_QUOTA =
16+
Number.parseInt(process.env.ANALYTICS_REPORT_DAILY_QUOTA ?? '10', 10) || 10
17+
18+
export interface AnalyticsReport {
19+
id: string
20+
orgId: string
21+
createdAt: string
22+
/** Present when S3 is configured. */
23+
s3Key?: string
24+
/** Present when S3 is not configured (dev / test). */
25+
localBuffer?: Buffer
26+
snapshotAt: string
27+
/** Content size in bytes (for informational purposes). */
28+
sizeBytes: number
29+
}
30+
31+
// Keyed by orgId -> AnalyticsReport[] (sorted oldest-first)
32+
const _store = new Map<string, AnalyticsReport[]>()
33+
34+
export function _resetReportsStore(): void {
35+
_store.clear()
36+
}
37+
38+
/** Return a shallow copy of all reports for an org, newest-first. */
39+
export function getOrgReports(orgId: string): AnalyticsReport[] {
40+
return (_store.get(orgId) ?? []).slice().reverse()
41+
}
42+
43+
/**
44+
* Persist a new report record for an org and purge stale entries.
45+
* Returns the saved report.
46+
*/
47+
export function saveOrgReport(
48+
report: Omit<AnalyticsReport, 'id' | 'createdAt'>,
49+
retentionDays = DEFAULT_RETENTION_DAYS,
50+
): AnalyticsReport {
51+
const { randomUUID } = require('node:crypto') as typeof import('node:crypto')
52+
const saved: AnalyticsReport = {
53+
id: randomUUID(),
54+
createdAt: new Date().toISOString(),
55+
...report,
56+
}
57+
58+
const list = _store.get(report.orgId) ?? []
59+
list.push(saved)
60+
_store.set(report.orgId, list)
61+
62+
purgeOldReports(report.orgId, retentionDays)
63+
return saved
64+
}
65+
66+
/** Remove reports older than `retentionDays` for the given org. */
67+
function purgeOldReports(orgId: string, retentionDays: number): void {
68+
const cutoff = new Date()
69+
cutoff.setUTCDate(cutoff.getUTCDate() - retentionDays)
70+
const cutoffIso = cutoff.toISOString()
71+
72+
const list = _store.get(orgId)
73+
if (!list) return
74+
75+
const kept = list.filter((r) => r.createdAt >= cutoffIso)
76+
if (kept.length !== list.length) {
77+
_store.set(orgId, kept)
78+
}
79+
}
80+
81+
/** All org IDs that currently have at least one stored report. */
82+
export function getAllOrgIds(): string[] {
83+
return Array.from(_store.keys())
84+
}
85+
86+
/** Daily in-memory quota counters (orgId -> date -> count). */
87+
const _quotaCounters = new Map<string, Map<string, number>>()
88+
89+
export function _resetQuotaCounters(): void {
90+
_quotaCounters.clear()
91+
}
92+
93+
const utcDate = (d = new Date()): string => d.toISOString().slice(0, 10)
94+
95+
/**
96+
* Check and increment the per-org report-generation quota for today.
97+
* Returns true when allowed, false when the daily cap is exhausted.
98+
*/
99+
export function checkAndIncrementReportQuota(
100+
orgId: string,
101+
dailyLimit = DEFAULT_REPORT_QUOTA,
102+
): boolean {
103+
const today = utcDate()
104+
if (!_quotaCounters.has(orgId)) {
105+
_quotaCounters.set(orgId, new Map())
106+
}
107+
const byDate = _quotaCounters.get(orgId)!
108+
const current = byDate.get(today) ?? 0
109+
if (current >= dailyLimit) return false
110+
byDate.set(today, current + 1)
111+
return true
112+
}

0 commit comments

Comments
 (0)