Skip to content

Commit aa14679

Browse files
committed
Merge pull request #961
2 parents a3c908d + c8c0500 commit aa14679

7 files changed

Lines changed: 255 additions & 48 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: 61 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
1-
import { orgAnalyticsRateLimiter } from "../middleware/rateLimiter.js";
2-
import { Router, Request, Response, NextFunction } from "express";
3-
import { authenticate } from "../middleware/auth.js";
4-
import { requireOrgAccess, requireOrgRole } from "../middleware/orgAuth.js";
5-
import { vaults, Vault } from "./vaults.js";
6-
import { getTeamRollup } from "../services/team.js";
7-
import { getCohortRetention } from "../services/analytics.service.js";
1+
import { orgAnalyticsRateLimiter } from '../middleware/rateLimiter.js'
2+
import { Router, Request, Response } from 'express'
3+
import { authenticate } from '../middleware/auth.js'
4+
import { requireOrgAccess, requireOrgRole } from '../middleware/orgAuth.js'
5+
import { vaults, Vault } from './vaults.js'
6+
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'
810

911
export const orgAnalyticsRouter = Router();
1012

@@ -112,5 +114,55 @@ orgAnalyticsRouter.get(
112114
} catch {
113115
res.status(500).json({ error: "Failed to generate team rollup" });
114116
}
115-
},
116-
);
117+
}
118+
)
119+
120+
/**
121+
* GET /api/orgs/:orgId/analytics/reports
122+
*
123+
* List all point-in-time analytics reports for the org. Each entry includes a
124+
* signed, expiring download URL when S3 is configured, or a local download path
125+
* otherwise. Paginated; newest reports appear first. Access is restricted to
126+
* owner/admin members of the org (strict per-org isolation).
127+
*/
128+
orgAnalyticsRouter.get(
129+
'/:orgId/analytics/reports',
130+
authenticate,
131+
requireOrgAccess('owner', 'admin'),
132+
orgAnalyticsRateLimiter,
133+
async (req: Request, res: Response) => {
134+
const { orgId } = req.params
135+
const pagination = parsePaginationParams(req)
136+
const s3Config = resolveS3Config()
137+
138+
const allReports = getOrgReports(orgId)
139+
const paginated = paginateArray(allReports, pagination)
140+
141+
const items = await Promise.all(
142+
paginated.data.map(async (r) => {
143+
let downloadUrl: string | null = null
144+
if (r.s3Key && s3Config) {
145+
try {
146+
downloadUrl = await getExportSignedUrl(s3Config, r.s3Key)
147+
} catch {
148+
// signed URL generation failure is non-fatal; return null
149+
}
150+
} else if (r.localBuffer) {
151+
// In non-S3 mode expose a local download path so clients can retrieve it
152+
downloadUrl = `/api/orgs/${orgId}/analytics/reports/${r.id}/download`
153+
}
154+
155+
return {
156+
id: r.id,
157+
orgId: r.orgId,
158+
snapshotAt: r.snapshotAt,
159+
createdAt: r.createdAt,
160+
sizeBytes: r.sizeBytes,
161+
downloadUrl,
162+
}
163+
}),
164+
)
165+
166+
res.json({ ...paginated, data: items })
167+
}
168+
)

src/services/analytics.service.ts

Lines changed: 14 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -224,46 +224,21 @@ export async function getCapitalAnalytics(period: string = "all"): Promise<{
224224
}
225225

226226
export async function updateAnalyticsSummary(orgId?: string): Promise<void> {
227-
await dbUpdateSummary();
228-
await invalidate("analytics:overall", orgId);
227+
await dbUpdateSummary()
228+
await invalidate('analytics:overall', orgId)
229229
}
230230

231231
/**
232-
* Retrieves monthly vault creation cohort metrics and retention trends
233-
* from the vault_cohort_retention materialized view.
234-
* @param {any} db The database knex connection instance passed from the router/service layer
235-
* @param {number} [range] Optional filter specifying the number of past months to fetch
232+
* Render a point-in-time analytics snapshot for a single org.
233+
* Pulls vault IDs from the in-memory vaults store so it works without a DB.
236234
*/
237-
export async function getCohortRetention(
238-
db: any,
239-
range?: number,
240-
): Promise<any[]> {
241-
let query = db("vault_cohort_retention")
242-
.select(
243-
"cohort_month",
244-
"total",
245-
"completed",
246-
"failed",
247-
"active",
248-
"median_days_to_complete",
249-
)
250-
.orderBy("cohort_month", "desc");
251-
252-
if (range && range > 0) {
253-
query = query.limit(range);
254-
}
255-
256-
const rows = await query;
257-
258-
return rows.map((row: any) => ({
259-
...row,
260-
cohort_month:
261-
row.cohort_month instanceof Date
262-
? row.cohort_month.toISOString().split("T")[0]
263-
: row.cohort_month,
264-
median_days_to_complete:
265-
row.median_days_to_complete !== null
266-
? parseFloat(parseFloat(row.median_days_to_complete).toFixed(1))
267-
: null,
268-
}));
269-
}
235+
export function renderOrgAnalyticsSnapshot(orgId: string): OrgVaultAnalytics & { orgId: string; snapshotAt: string } {
236+
// Import lazily to avoid circular deps and to stay hermetic in tests
237+
// eslint-disable-next-line @typescript-eslint/no-var-requires
238+
const { vaults } = require('../routes/vaults.js') as { vaults: Array<{ id: string; orgId?: string }> }
239+
const orgVaultIds = vaults
240+
.filter((v) => v.orgId === orgId)
241+
.map((v) => v.id)
242+
const analytics = getOrgAnalyticsBatched(orgVaultIds)
243+
return { ...analytics, orgId, snapshotAt: utcNow() }
244+
}

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)