perf: cache-only strategy for list views, cache-then-cw for single fetch
- Add data-source hierarchy to opportunity manager (cache-only, cache-then-cw, cw-first) - fetchPages/search/fetchByCompany use cache-only: Redis → DB (no CW calls) - fetchItem uses cache-then-cw by default, cw-first when fresh=true - Add idleTimeout: 255 to Bun.serve to prevent request timeouts - Map CW status 57 (04. Confirmed Quote) to Active equivalency - Add computeCacheTTL algorithm and opportunityCache module
This commit is contained in:
@@ -0,0 +1,166 @@
|
||||
/**
|
||||
* @module computeCacheTTL
|
||||
*
|
||||
* Adaptive Cache TTL Algorithm
|
||||
* ============================
|
||||
*
|
||||
* Determines how long a cached record should live before it must be
|
||||
* re-fetched from the upstream source (e.g. ConnectWise API).
|
||||
*
|
||||
* The algorithm prioritises freshness for records that are actively
|
||||
* being worked on, while avoiding unnecessary API calls for stale or
|
||||
* inactive data.
|
||||
*
|
||||
* ## Spec
|
||||
*
|
||||
* | # | Condition | TTL (ms) | TTL (human) | Rationale |
|
||||
* |---|------------------------------------------------------------------|----------|-------------|--------------------------------------------------------------------|
|
||||
* | 1 | `closedFlag` is `true` | `null` | Do not cache| Closed records are rarely accessed; caching wastes memory. |
|
||||
* | 2 | `expectedCloseDate` OR `lastUpdated` is within the last **5 days**| 30 000 | 30 seconds | High-activity window — data changes frequently and must stay fresh.|
|
||||
* | 3 | `expectedCloseDate` OR `lastUpdated` is within the last **14 days**| 60 000 | 60 seconds | Moderate activity — still relevant, but changes less often. |
|
||||
* | 4 | Everything else (older than 14 days) | 900 000 | 15 minutes | Low activity — safe to serve from cache for longer. |
|
||||
*
|
||||
* ## Evaluation order
|
||||
*
|
||||
* Rules are evaluated **top-to-bottom**; the first matching rule wins.
|
||||
* Rule 2 (5-day window) is a subset of Rule 3 (14-day window), so it
|
||||
* must be checked first.
|
||||
*
|
||||
* ## Inputs
|
||||
*
|
||||
* | Field | Type | Description |
|
||||
* |--------------------|------------------|--------------------------------------------------------------------|
|
||||
* | `closedFlag` | `boolean` | Whether the record is closed / inactive. |
|
||||
* | `expectedCloseDate`| `Date \| null` | The projected close date (future-looking relevance signal). |
|
||||
* | `lastUpdated` | `Date \| null` | The last time the upstream record was modified (backward-looking). |
|
||||
* | `now` | `Date` (optional)| Override for the current timestamp; defaults to `new Date()`. |
|
||||
*
|
||||
* ## Output
|
||||
*
|
||||
* Returns `number | null`:
|
||||
* - A positive integer representing the TTL in **milliseconds**, or
|
||||
* - `null` when the record should **not** be cached at all.
|
||||
*
|
||||
* ## Usage
|
||||
*
|
||||
* ```ts
|
||||
* import { computeCacheTTL } from "../modules/algorithms/computeCacheTTL";
|
||||
*
|
||||
* const ttl = computeCacheTTL({
|
||||
* closedFlag: opportunity.closedFlag,
|
||||
* expectedCloseDate: opportunity.expectedCloseDate,
|
||||
* lastUpdated: opportunity.cwLastUpdated,
|
||||
* });
|
||||
*
|
||||
* if (ttl !== null) {
|
||||
* await redis.set(key, serialised, "PX", ttl);
|
||||
* }
|
||||
* ```
|
||||
*/
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** 30 seconds – TTL for high-activity records (within 5 days). */
|
||||
export const TTL_HIGH_ACTIVITY = 30_000;
|
||||
|
||||
/** 60 seconds – TTL for moderate-activity records (within 14 days). */
|
||||
export const TTL_MODERATE_ACTIVITY = 60_000;
|
||||
|
||||
/** 15 minutes – TTL for low-activity / stale records. */
|
||||
export const TTL_LOW_ACTIVITY = 900_000;
|
||||
|
||||
/** 30 days in milliseconds. */
|
||||
const THIRTY_DAYS_MS = 30 * 24 * 60 * 60 * 1000;
|
||||
|
||||
/** 5 days in milliseconds. */
|
||||
const FIVE_DAYS_MS = 5 * 24 * 60 * 60 * 1000;
|
||||
|
||||
/** 14 days in milliseconds. */
|
||||
const FOURTEEN_DAYS_MS = 14 * 24 * 60 * 60 * 1000;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Input type
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface CacheTTLInput {
|
||||
/** Whether the record is closed / inactive. */
|
||||
closedFlag: boolean;
|
||||
/** When the record was closed — used for recently-closed caching (within 30 days). */
|
||||
closedDate: Date | null;
|
||||
/** The projected close date — serves as a forward-looking relevance signal. */
|
||||
expectedCloseDate: Date | null;
|
||||
/** The date the upstream record was last modified — backward-looking signal. */
|
||||
lastUpdated: Date | null;
|
||||
/**
|
||||
* Override for the current timestamp.
|
||||
* Useful for deterministic testing. Defaults to `new Date()`.
|
||||
*/
|
||||
now?: Date;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Algorithm
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Compute the cache TTL for a record based on its activity signals.
|
||||
*
|
||||
* @param input - The record's activity signals. See {@link CacheTTLInput}.
|
||||
* @returns The TTL in milliseconds, or `null` if the record should not be cached.
|
||||
*
|
||||
* @see Module-level JSDoc for the full spec table and evaluation rules.
|
||||
*/
|
||||
export function computeCacheTTL(input: CacheTTLInput): number | null {
|
||||
const {
|
||||
closedFlag,
|
||||
closedDate,
|
||||
expectedCloseDate,
|
||||
lastUpdated,
|
||||
now = new Date(),
|
||||
} = input;
|
||||
|
||||
const nowMs = now.getTime();
|
||||
|
||||
/**
|
||||
* Check whether a date falls within a window around `now`.
|
||||
*
|
||||
* "Within" means the date is between `now - windowMs` and `now + windowMs`,
|
||||
* allowing both past updates and future-scheduled dates to qualify.
|
||||
*/
|
||||
const isWithinWindow = (date: Date | null, windowMs: number): boolean => {
|
||||
if (!date) return false;
|
||||
const diff = Math.abs(nowMs - date.getTime());
|
||||
return diff <= windowMs;
|
||||
};
|
||||
|
||||
// Rule 1 — Closed records
|
||||
if (closedFlag) {
|
||||
// Rule 1b — Recently closed (within 30 days) → cache at low-activity TTL
|
||||
if (isWithinWindow(closedDate, THIRTY_DAYS_MS)) {
|
||||
return TTL_LOW_ACTIVITY;
|
||||
}
|
||||
// Rule 1a — Closed longer than 30 days → do not cache
|
||||
return null;
|
||||
}
|
||||
|
||||
// Rule 2 — High activity (5 days)
|
||||
if (
|
||||
isWithinWindow(expectedCloseDate, FIVE_DAYS_MS) ||
|
||||
isWithinWindow(lastUpdated, FIVE_DAYS_MS)
|
||||
) {
|
||||
return TTL_HIGH_ACTIVITY;
|
||||
}
|
||||
|
||||
// Rule 3 — Moderate activity (14 days)
|
||||
if (
|
||||
isWithinWindow(expectedCloseDate, FOURTEEN_DAYS_MS) ||
|
||||
isWithinWindow(lastUpdated, FOURTEEN_DAYS_MS)
|
||||
) {
|
||||
return TTL_MODERATE_ACTIVITY;
|
||||
}
|
||||
|
||||
// Rule 4 — Low activity / stale
|
||||
return TTL_LOW_ACTIVITY;
|
||||
}
|
||||
+257
@@ -0,0 +1,257 @@
|
||||
/**
|
||||
* @module opportunityCache
|
||||
*
|
||||
* Redis-backed cache for expensive ConnectWise API data associated
|
||||
* with opportunities.
|
||||
*
|
||||
* ## What is cached
|
||||
*
|
||||
* Each non-closed opportunity may have two cached payloads keyed by
|
||||
* its `cwOpportunityId`:
|
||||
*
|
||||
* - **Activities** (`opp:activities:{cwOpportunityId}`) — the raw
|
||||
* `CWActivity[]` array fetched from `activityCw.fetchByOpportunity()`.
|
||||
* - **Company CW data** (`opp:company-cw:{cw_CompanyId}`) — the hydrated
|
||||
* company / contacts blob set by `CompanyController.hydrateCwData()`.
|
||||
*
|
||||
* TTLs are computed dynamically via {@link computeCacheTTL} so hot
|
||||
* opportunities refresh every 30 s while stale ones live for 15 min.
|
||||
*
|
||||
* ## Background refresh
|
||||
*
|
||||
* {@link refreshOpportunityCache} is designed to be called on a
|
||||
* 30-second interval from `src/index.ts`. It scans all non-closed
|
||||
* DB opportunities, checks which cache keys have expired, and
|
||||
* re-fetches only those from ConnectWise.
|
||||
*/
|
||||
|
||||
import { prisma, redis } from "../../constants";
|
||||
import { activityCw } from "../cw-utils/activities/activities";
|
||||
import { computeCacheTTL } from "../algorithms/computeCacheTTL";
|
||||
import { connectWiseApi } from "../../constants";
|
||||
import { fetchCwCompanyById } from "../cw-utils/fetchCompany";
|
||||
import { events } from "../globalEvents";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Key helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const ACTIVITY_PREFIX = "opp:activities:";
|
||||
const COMPANY_CW_PREFIX = "opp:company-cw:";
|
||||
|
||||
/** Redis key for cached activities by CW opportunity ID. */
|
||||
export const activityCacheKey = (cwOppId: number) =>
|
||||
`${ACTIVITY_PREFIX}${cwOppId}`;
|
||||
|
||||
/** Redis key for cached company CW hydration data by CW company ID. */
|
||||
export const companyCwCacheKey = (cwCompanyId: number) =>
|
||||
`${COMPANY_CW_PREFIX}${cwCompanyId}`;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Read helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Retrieve cached CW activities for an opportunity.
|
||||
*
|
||||
* @returns The parsed `CWActivity[]` or `null` on cache miss.
|
||||
*/
|
||||
export async function getCachedActivities(
|
||||
cwOpportunityId: number,
|
||||
): Promise<any[] | null> {
|
||||
const raw = await redis.get(activityCacheKey(cwOpportunityId));
|
||||
if (!raw) return null;
|
||||
try {
|
||||
return JSON.parse(raw);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve cached company CW hydration data.
|
||||
*
|
||||
* @returns `{ company, defaultContact, allContacts }` or `null` on cache miss.
|
||||
*/
|
||||
export async function getCachedCompanyCwData(
|
||||
cwCompanyId: number,
|
||||
): Promise<{ company: any; defaultContact: any; allContacts: any[] } | null> {
|
||||
const raw = await redis.get(companyCwCacheKey(cwCompanyId));
|
||||
if (!raw) return null;
|
||||
try {
|
||||
return JSON.parse(raw);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Write helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Fetch activities from CW and cache them with the appropriate TTL.
|
||||
*
|
||||
* @returns The raw `CWActivity[]` collection (as plain array).
|
||||
*/
|
||||
export async function fetchAndCacheActivities(
|
||||
cwOpportunityId: number,
|
||||
ttlMs: number,
|
||||
): Promise<any[]> {
|
||||
const collection = await activityCw.fetchByOpportunity(cwOpportunityId);
|
||||
const arr = collection.map((item) => item);
|
||||
await redis.set(
|
||||
activityCacheKey(cwOpportunityId),
|
||||
JSON.stringify(arr),
|
||||
"PX",
|
||||
ttlMs,
|
||||
);
|
||||
return arr;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch company CW data (company, contacts) and cache with the given TTL.
|
||||
*
|
||||
* @returns The hydration blob or `null` if the company doesn't exist in CW.
|
||||
*/
|
||||
export async function fetchAndCacheCompanyCwData(
|
||||
cwCompanyId: number,
|
||||
ttlMs: number,
|
||||
): Promise<{ company: any; defaultContact: any; allContacts: any[] } | null> {
|
||||
const cwCompany = await fetchCwCompanyById(cwCompanyId);
|
||||
if (!cwCompany) return null;
|
||||
|
||||
const contactHref = cwCompany.defaultContact?._info?.contact_href;
|
||||
const defaultContactData = contactHref
|
||||
? await connectWiseApi.get(contactHref)
|
||||
: undefined;
|
||||
|
||||
const allContactsData = await connectWiseApi.get(
|
||||
`${cwCompany._info.contacts_href}&pageSize=1000`,
|
||||
);
|
||||
|
||||
const blob = {
|
||||
company: cwCompany,
|
||||
defaultContact: defaultContactData?.data ?? null,
|
||||
allContacts: allContactsData.data,
|
||||
};
|
||||
|
||||
await redis.set(
|
||||
companyCwCacheKey(cwCompanyId),
|
||||
JSON.stringify(blob),
|
||||
"PX",
|
||||
ttlMs,
|
||||
);
|
||||
|
||||
return blob;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Background refresh
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Refresh the opportunity cache.
|
||||
*
|
||||
* Scans all non-closed opportunities in the database, computes a TTL for each,
|
||||
* checks whether the cache key still exists, and re-fetches from ConnectWise
|
||||
* only for entries that have expired.
|
||||
*
|
||||
* Designed to be called every **30 seconds** from the process entry point.
|
||||
*/
|
||||
export async function refreshOpportunityCache(): Promise<void> {
|
||||
// Include non-closed AND recently-closed (within 30 days) opportunities
|
||||
const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
|
||||
|
||||
const opportunities = await prisma.opportunity.findMany({
|
||||
where: {
|
||||
OR: [
|
||||
{ closedFlag: false },
|
||||
{ closedFlag: true, closedDate: { gte: thirtyDaysAgo } },
|
||||
],
|
||||
},
|
||||
select: {
|
||||
cwOpportunityId: true,
|
||||
closedFlag: true,
|
||||
closedDate: true,
|
||||
expectedCloseDate: true,
|
||||
cwLastUpdated: true,
|
||||
company: { select: { cw_CompanyId: true } },
|
||||
},
|
||||
});
|
||||
|
||||
events.emit("cache:opportunities:refresh:started", {
|
||||
totalOpportunities: opportunities.length,
|
||||
});
|
||||
|
||||
let activitiesRefreshed = 0;
|
||||
let companiesRefreshed = 0;
|
||||
let skipped = 0;
|
||||
|
||||
// Batch-check which activity keys already exist via a pipeline
|
||||
const pipeline = redis.pipeline();
|
||||
for (const opp of opportunities) {
|
||||
pipeline.exists(activityCacheKey(opp.cwOpportunityId));
|
||||
}
|
||||
const existsResults = await pipeline.exec();
|
||||
|
||||
const refreshTasks: Promise<void>[] = [];
|
||||
|
||||
for (let i = 0; i < opportunities.length; i++) {
|
||||
const opp = opportunities[i]!;
|
||||
|
||||
const ttl = computeCacheTTL({
|
||||
closedFlag: opp.closedFlag,
|
||||
closedDate: opp.closedDate,
|
||||
expectedCloseDate: opp.expectedCloseDate,
|
||||
lastUpdated: opp.cwLastUpdated,
|
||||
});
|
||||
|
||||
// Skip closed (ttl === null) — should not happen because of the query filter,
|
||||
// but guard anyway.
|
||||
if (ttl === null) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// existsResults entries are [error, result] tuples
|
||||
const activityExists = existsResults?.[i]?.[1] === 1;
|
||||
|
||||
if (!activityExists) {
|
||||
refreshTasks.push(
|
||||
fetchAndCacheActivities(opp.cwOpportunityId, ttl).then(() => {
|
||||
activitiesRefreshed++;
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
// Also refresh company CW data if the key is missing
|
||||
if (opp.company?.cw_CompanyId) {
|
||||
const cwCompanyId = opp.company.cw_CompanyId;
|
||||
refreshTasks.push(
|
||||
(async () => {
|
||||
const companyExists = await redis.exists(
|
||||
companyCwCacheKey(cwCompanyId),
|
||||
);
|
||||
if (!companyExists) {
|
||||
await fetchAndCacheCompanyCwData(cwCompanyId, ttl);
|
||||
companiesRefreshed++;
|
||||
}
|
||||
})(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Run all refresh tasks concurrently with bounded concurrency
|
||||
const CONCURRENCY = 10;
|
||||
for (let i = 0; i < refreshTasks.length; i += CONCURRENCY) {
|
||||
await Promise.allSettled(refreshTasks.slice(i, i + CONCURRENCY));
|
||||
}
|
||||
|
||||
events.emit("cache:opportunities:refresh:completed", {
|
||||
totalOpportunities: opportunities.length,
|
||||
activitiesRefreshed,
|
||||
companiesRefreshed,
|
||||
skipped,
|
||||
});
|
||||
}
|
||||
@@ -178,6 +178,18 @@ interface EventTypes {
|
||||
staleCount: number;
|
||||
}) => void;
|
||||
|
||||
// Cache Events
|
||||
"cache:opportunities:refresh:started": (data: {
|
||||
totalOpportunities: number;
|
||||
}) => void;
|
||||
"cache:opportunities:refresh:completed": (data: {
|
||||
totalOpportunities: number;
|
||||
activitiesRefreshed: number;
|
||||
companiesRefreshed: number;
|
||||
skipped: number;
|
||||
}) => void;
|
||||
"cache:opportunities:refresh:error": (data: { error: unknown }) => void;
|
||||
|
||||
// ConnectWise User Defined Fields Events
|
||||
"cw:udf:refresh:started": () => void;
|
||||
"cw:udf:refresh:completed": (data: { count: number }) => void;
|
||||
|
||||
Reference in New Issue
Block a user